""" Optimus Research Center - Scientific Coordinator & Task Worker Fulfills Defect 3: Complete research and experimental execution circuit. 1. Register research request. 2. Validate requirements. 3. Create experimental plan. 4. Select corresponding lab. 5. Assign tasks. 6. Execute permitted scientific operations (Category A trusted internal operations). 7. Record real progress. 8. Evaluate results. 9. Guard evidence with real SHA-256 checksums. 10. Finalize with correct status. Features: - Controlled worker thread (start/stop) - NOT auto-run on Windows boot. - Budget enforcement and graceful cancellation. - Failure recovery and limited retries. - Rejection/blocking of untrusted code without sandbox (SANDBOX_UNAVAILABLE). """ import json import time import threading import hashlib import traceback from datetime import datetime from typing import Dict, Any, List, Optional, Callable import numpy as np from sqlalchemy.orm import Session from sqlalchemy import desc from app.core.database import SessionLocal from app.core import models from app.core.sandbox import ExecutionPolicy from app.core.scientific_memory import ScientificMemory class ScientificCoordinator: """Manages experimental work pipeline, queue states, and execution engine.""" VALID_STATES = { "PENDING", "RUNNING", "COMPLETED", "FAILED", "CANCELLED", "BLOCKED", "INSUFFICIENT_EVIDENCE", "UNSUPPORTED_ANALYSIS", "UNSUPPORTED_TASK_TYPE", "SANDBOX_UNAVAILABLE", "NEEDS_MODEL", "PENDING_SAFE_ENVIRONMENT" } TERMINAL_STATES = { "COMPLETED", "FAILED", "CANCELLED", "BLOCKED", "INSUFFICIENT_EVIDENCE", "UNSUPPORTED_ANALYSIS", "UNSUPPORTED_TASK_TYPE", "SANDBOX_UNAVAILABLE", "NEEDS_MODEL", "PENDING_SAFE_ENVIRONMENT" } NON_RETRYABLE_STATES = { "CANCELLED", "BLOCKED", "INSUFFICIENT_EVIDENCE", "UNSUPPORTED_ANALYSIS", "UNSUPPORTED_TASK_TYPE", "SANDBOX_UNAVAILABLE", "NEEDS_MODEL", "PENDING_SAFE_ENVIRONMENT" } AUTHORIZED_TASK_TYPES = { "ALGORITHM_OPTIMIZATION", "ALGORITHM_BENCHMARK", "ALGORITHM_PREDICTIVE", "ALGORITHM_INVESTIGATION", "AI_TRAINING", "AI_TRAIN", "AI_ROBOTICS_SIM", "AI_ARCHITECTURE_INVESTIGATION", "QUANTITATIVE_RESEARCH", "TECH_RESEARCH", "DATASET_ANALYSIS", "INTER_LAB_COLLABORATION", "DISCOVERY_CYCLE" } def __init__(self, db: Optional[Session] = None): self._db = db def _get_db(self) -> Session: return self._db if self._db is not None else SessionLocal() def submit_task( self, task_type: str, payload: Dict[str, Any], priority: int = 5, execution_budget_seconds: int = 120, max_retries: int = 2 ) -> models.CoordinationTask: """Enqueues a new experimental task.""" db = self._get_db() try: task = models.CoordinationTask( task_type=task_type, priority=max(1, min(10, priority)), status="PENDING", payload_json=json.dumps(payload), result_json="{}", progress_percent=0.0, retry_count=0, max_retries=max_retries, execution_budget_seconds=execution_budget_seconds ) db.add(task) db.commit() db.refresh(task) return task finally: if self._db is None: db.close() def register_research_request( self, title: str, lab_type: str, task_type: str, payload: Dict[str, Any], priority: int = 5, execution_budget_seconds: int = 180 ) -> Dict[str, Any]: """ Registers an end-to-end scientific research request. Creates project, initial hypothesis, experimental plan, and task queue entry. """ db = self._get_db() try: lab_type_upper = lab_type.upper() if lab_type_upper not in {"ALGORITHM", "AI", "CROSS_LAB"}: raise ValueError(f"Invalid lab_type '{lab_type}'. Must be ALGORITHM, AI, or CROSS_LAB.") # Connect ResearchService for structured planning and evidence evaluation from app.core.ai_provider import ResearchService, get_ai_provider provider_override = payload.get("ai_provider") research_svc = ResearchService(provider=get_ai_provider(provider_override)) # Separate administrative fields from scientific fields project_title = title requested_task_type = task_type scientific_question = ( payload.get("question") or payload.get("research_question") or title ) research_domain = ( payload.get("sector") or payload.get("domain") or "GENERAL" ) hypothesis_statement = payload.get("hypothesis_statement") document_paths = payload.get("document_paths") or [] sources = payload.get("sources") or [] documents = payload.get("documents") or [] analysis_type = payload.get("analysis_type") criteria = payload.get("criteria") requires_generative = payload.get("requires_generative", False) # Phase 2: Formulate plan & evaluate evidence through ResearchService research_plan = research_svc.plan_research_inquiry( problem_statement=scientific_question, lab_type=lab_type_upper, domain=research_domain, documents=documents, sources=sources, document_paths=document_paths, analysis_type=analysis_type, criteria=criteria, requested_task_type=requested_task_type, requires_generative=requires_generative ) # If request needs an external generative model that is not configured: if research_plan.get("status") == "NEEDS_MODEL": return { "project_id": None, "hypothesis_id": None, "task_id": None, "task_type": task_type, "status": "NEEDS_MODEL", "reason": research_plan.get("reason", "Generative AI model required but not configured."), "plan": None, "hypothesis_origin": "NONE" } if research_plan.get("status") in {"CONTRADICTORY_CRITERIA", "UNSUPPORTED_ANALYSIS"}: return { "project_id": None, "hypothesis_id": None, "task_id": None, "task_type": task_type, "status": research_plan.get("status"), "reason": research_plan.get("reason", f"Research request halted: {research_plan.get('status')}."), "plan": research_plan, "hypothesis_origin": "NONE" } # Align effective task type effective_task_type = research_plan.get("authorized_task_type") or requested_task_type # Create or resolve project project = db.query(models.ResearchProject).filter( models.ResearchProject.title == project_title ).first() if not project: project = models.ResearchProject( title=project_title, description=f"Automated research program for {scientific_question}", lab_type=lab_type_upper, domain=research_domain ) db.add(project) db.commit() db.refresh(project) # Formulate initial hypothesis hyp_statement = ( hypothesis_statement or research_plan.get("hypothesis") or f"Empirical evaluation of {effective_task_type} on '{scientific_question}' will demonstrate measurable advantage over baseline." ) confidence_score = research_plan.get("justified_confidence_score", 0.5) hyp = models.Hypothesis( project_id=project.id, title=f"Hypothesis: {scientific_question[:80]}", statement=hyp_statement, rationale=payload.get("rationale", f"Formulated via ResearchService ({research_plan.get('hypothesis_origin', 'HEURISTIC')})."), status="PROPOSED", confidence_score=confidence_score ) db.add(hyp) db.commit() db.refresh(hyp) # Inject project_id and hypothesis_id into payload task_payload = dict(payload) task_payload["project_id"] = project.id task_payload["hypothesis_id"] = hyp.id task_payload["lab_type"] = lab_type_upper task_payload["question"] = scientific_question task_payload["research_question"] = scientific_question task_payload["sector"] = research_domain task_payload["domain"] = research_domain task_payload["research_plan"] = research_plan task_payload["hypothesis_origin"] = research_plan.get("hypothesis_origin", "HEURISTIC") if document_paths and "document_paths" not in task_payload: task_payload["document_paths"] = document_paths if sources and "sources" not in task_payload: task_payload["sources"] = sources if analysis_type and "analysis_type" not in task_payload: task_payload["analysis_type"] = analysis_type if criteria and "criteria" not in task_payload: task_payload["criteria"] = criteria task = self.submit_task( task_type=effective_task_type, payload=task_payload, priority=priority, execution_budget_seconds=execution_budget_seconds ) return { "project_id": project.id, "hypothesis_id": hyp.id, "task_id": task.id, "task_type": task.task_type, "status": task.status, "plan": { "problem_statement": scientific_question, "authorized_task_type": effective_task_type, "milestone_1": "Validation of parameters and dataset", "milestone_2": "Empirical candidate execution", "milestone_3": "Audit against baseline metrics", "milestone_4": "Evidence generation and memory indexing", "research_plan_details": research_plan }, "hypothesis_origin": research_plan.get("hypothesis_origin", "HEURISTIC"), "hypothesis_provenance": research_plan.get("hypothesis_provenance"), "justified_confidence_score": confidence_score } finally: if self._db is None: db.close() def get_next_pending_task(self) -> Optional[models.CoordinationTask]: """Fetches the highest priority pending task (priority 1 is highest).""" db = self._get_db() try: return db.query(models.CoordinationTask).filter( models.CoordinationTask.status == "PENDING" ).order_by( models.CoordinationTask.priority.asc(), models.CoordinationTask.created_at.asc() ).first() finally: if self._db is None: db.close() def claim_next_pending_task(self) -> Optional[models.CoordinationTask]: """ Atomically selects and claims the highest priority PENDING task in SQLite. Immediately transitions status to RUNNING to prevent multiple workers from processing it concurrently. """ db = self._get_db() try: candidate = db.query(models.CoordinationTask).filter( models.CoordinationTask.status == "PENDING" ).order_by( models.CoordinationTask.priority.asc(), models.CoordinationTask.created_at.asc() ).first() if not candidate: return None # Perform atomic reservation conditional on status still being PENDING rows_updated = db.query(models.CoordinationTask).filter( models.CoordinationTask.id == candidate.id, models.CoordinationTask.status == "PENDING" ).update({ "status": "RUNNING", "started_at": datetime.utcnow(), "progress_percent": 10.0 }) db.commit() if rows_updated == 0: # Race condition: another thread/worker claimed this task in between return None db.refresh(candidate) return candidate finally: if self._db is None: db.close() def update_task_status( self, task_id: int, status: str, progress: Optional[float] = None, result: Optional[Dict[str, Any]] = None, error_log: Optional[str] = None ) -> Optional[models.CoordinationTask]: """Updates task state with transition validation.""" if status not in self.VALID_STATES: raise ValueError(f"Invalid task status '{status}'. Must be one of {self.VALID_STATES}") db = self._get_db() try: task = db.query(models.CoordinationTask).filter(models.CoordinationTask.id == task_id).first() if not task: return None task.status = status if progress is not None: task.progress_percent = max(0.0, min(100.0, progress)) if result is not None: task.result_json = json.dumps(result) if error_log is not None: task.error_log = error_log if status == "RUNNING" and not task.started_at: task.started_at = datetime.utcnow() elif status in self.TERMINAL_STATES: task.completed_at = datetime.utcnow() db.commit() db.refresh(task) return task finally: if self._db is None: db.close() def cancel_task(self, task_id: int, reason: str = "User cancelled") -> bool: """Gracefully cancels a pending or running task.""" db = self._get_db() try: task = db.query(models.CoordinationTask).filter(models.CoordinationTask.id == task_id).first() if not task: return False if task.status in self.TERMINAL_STATES: return False # Already terminal task.status = "CANCELLED" task.completed_at = datetime.utcnow() task.error_log = f"Cancelled: {reason}" db.commit() return True finally: if self._db is None: db.close() def is_cancelled(self, task_id: int) -> bool: """Checks if a task was flagged as cancelled by a user.""" db = self._get_db() try: task = db.query(models.CoordinationTask).filter(models.CoordinationTask.id == task_id).first() return task.status == "CANCELLED" if task else True finally: if self._db is None: db.close() def recover_interrupted_tasks(self) -> List[Dict[str, Any]]: """ Scans for orphaned RUNNING tasks from prior crashes or forced shutdowns. Explicitly distinguishes: 1. Failure reconciliation (RECONCILED_AS_FAILED, requeued=False, resumed=False): Default safe action when no verified, ownership-bound, idempotent checkpoint exists. Never misrepresented as task resumption. 2. Checkpoint requeue (REQUEUED, requeued=True, resumed=False): Only when the task payload contains a verified checkpoint with matching SHA-256 state digest, matching task_id/task_type ownership, and explicit idempotent=True protection. Note: Re-queuing to PENDING is labeled REQUEUED (resumed=False); actual state restoration and continuation only occur when the target engine restores the checkpoint during execution. """ db = self._get_db() recovered = [] try: orphans = db.query(models.CoordinationTask).filter( models.CoordinationTask.status == "RUNNING" ).all() for t in orphans: payload_obj = {} try: payload_obj = json.loads(t.payload_json or "{}") except Exception: payload_obj = {} checkpoint = payload_obj.get("checkpoint") if isinstance(payload_obj, dict) else None checkpoint_valid = False ownership_valid = False idempotent_allowed = False if ( isinstance(checkpoint, dict) and checkpoint.get("resumable") is True and isinstance(checkpoint.get("state"), dict) ): expected_hash = checkpoint.get("checkpoint_sha256") computed_hash = hashlib.sha256( json.dumps(checkpoint["state"], sort_keys=True).encode("utf-8") ).hexdigest() hash_matches = bool(expected_hash and expected_hash == computed_hash) ck_task_id = checkpoint.get("task_id") ck_task_type = checkpoint.get("task_type") ownership_valid = ( (ck_task_id is None or int(ck_task_id) == int(t.id)) and (ck_task_type is None or str(ck_task_type) == str(t.task_type)) ) idempotent_allowed = bool(checkpoint.get("idempotent") is True) checkpoint_valid = bool(hash_matches and ownership_valid and idempotent_allowed) if checkpoint_valid: t.status = "PENDING" t.error_log = ( "CHECKPOINT_REQUEUED: Verified reproducible and idempotent checkpoint found " f"(SHA-256: {checkpoint['checkpoint_sha256'][:16]}...). Task re-queued as PENDING " "(requeued=True, resumed=False until engine restores state)." ) action_taken = "REQUEUED" requeued_flag = True resumed_flag = False new_state = "PENDING" else: t.status = "FAILED" t.completed_at = datetime.utcnow() t.error_log = ( "CRASH_RECOVERY (RECONCILED_NOT_RESUMED): Process was interrupted during execution " "(system shutdown or unexpected termination). Reconciled safely as FAILED (not resumed)." ) action_taken = "RECONCILED_AS_FAILED" requeued_flag = False resumed_flag = False new_state = "FAILED" recovered.append({ "task_id": t.id, "task_type": t.task_type, "previous_status": "RUNNING", "new_status": new_state, "recovery_action": action_taken, "requeued": requeued_flag, "resumed": resumed_flag, "checkpoint_verified": checkpoint_valid, "ownership_verified": ownership_valid, "idempotent_verified": idempotent_allowed }) # Audit log log = models.AuditLog( event_type="SYSTEM_RECOVERY", severity="WARNING", details_json=json.dumps({ "task_id": t.id, "task_type": t.task_type, "action": action_taken, "requeued": requeued_flag, "resumed": resumed_flag, "checkpoint_verified": checkpoint_valid }) ) db.add(log) # Also check Experiments that were stuck in RUNNING orphaned_exps = db.query(models.Experiment).filter( models.Experiment.status == "RUNNING" ).all() for exp in orphaned_exps: exp.status = "FAILED" exp.completed_at = datetime.utcnow() exp.error_message = "CRASH_RECOVERY (RECONCILED_NOT_RESUMED): Experiment interrupted during shutdown." db.commit() return recovered finally: if self._db is None: db.close() def process_next_task(self, task_id: Optional[int] = None) -> Optional[Dict[str, Any]]: """ Executes a task through the full 10-step circuit. If task_id is specified, executes that task after validating it is executable. Otherwise atomically claims the next highest priority PENDING task. Returns execution summary or None if queue is empty. """ db = self._get_db() try: if task_id is not None: task = db.query(models.CoordinationTask).filter(models.CoordinationTask.id == task_id).first() if not task: return None if task.status in self.TERMINAL_STATES: return { "task_id": task_id, "status": task.status, "error": f"Task {task_id} is already in terminal state '{task.status}' and cannot be re-executed." } if task.status == "PENDING": # Claim atomically rows = db.query(models.CoordinationTask).filter( models.CoordinationTask.id == task.id, models.CoordinationTask.status == "PENDING" ).update({ "status": "RUNNING", "started_at": datetime.utcnow(), "progress_percent": 10.0 }) db.commit() if rows == 0: return { "task_id": task_id, "status": "ALREADY_CLAIMED", "error": "Task was claimed by another worker." } db.refresh(task) else: task = self.claim_next_pending_task() if not task: return None task_id = task.id task_type = task.task_type start_wall_time = time.time() budget_seconds = task.execution_budget_seconds or 120 # Step 1: Mark RUNNING and initial progress self.update_task_status(task_id, "RUNNING", progress=15.0) # Step 2: Validate requirements & payload try: payload = json.loads(task.payload_json or "{}") except Exception as e: err_msg = f"Invalid task payload JSON: {str(e)}" self.update_task_status(task_id, "FAILED", error_log=err_msg) return {"task_id": task_id, "status": "FAILED", "error": err_msg} # Check if task requests untrusted code execution (Category B) if "custom_code" in payload or "untrusted_code" in payload: code_to_check = payload.get("custom_code") or payload.get("untrusted_code") sec_eval = ExecutionPolicy.execute_untrusted_code( code_to_check, action_name=f"task_{task_id}_{task_type}" ) self.update_task_status( task_id, status=sec_eval["status"], error_log=f"Blocked: {sec_eval['reason']}. Preserved as candidate.", result=sec_eval ) return {"task_id": task_id, "status": sec_eval["status"], "audit": sec_eval} # Defecto D: Explicit check against authorized task types if task_type not in self.AUTHORIZED_TASK_TYPES: err_msg = ( f"UNSUPPORTED_TASK_TYPE: Task type '{task_type}' has no authorized scientific executor. " f"Execution rejected without simulated completion." ) diag_result = { "task_id": task_id, "task_type": task_type, "status": "UNSUPPORTED_TASK_TYPE", "reason": err_msg, "authorized_task_types": sorted(list(self.AUTHORIZED_TASK_TYPES)), "submitted_payload": payload, "timestamp": datetime.utcnow().isoformat() } self.update_task_status( task_id, status="UNSUPPORTED_TASK_TYPE", progress=0.0, result=diag_result, error_log=err_msg ) audit = models.AuditLog( event_type="UNSUPPORTED_TASK_REJECTED", severity="WARNING", details_json=json.dumps({"task_id": task_id, "task_type": task_type, "reason": err_msg}) ) db.add(audit) db.commit() return diag_result # Step 3: Experimental Plan & Lab Selection lab_type = payload.get("lab_type", "ALGORITHM") self.update_task_status(task_id, "RUNNING", progress=25.0) # Check budget / cancellation before heavy computation if self.is_cancelled(task_id): return {"task_id": task_id, "status": "CANCELLED"} if (time.time() - start_wall_time) > budget_seconds: self.update_task_status(task_id, "FAILED", error_log="Execution budget exceeded before dispatch.") return {"task_id": task_id, "status": "FAILED", "error": "BUDGET_EXCEEDED"} # Step 4 & 5: Dispatch to corresponding lab result_data = None evidence_hash = None try: # Category A: Trusted operations if task_type in {"ALGORITHM_OPTIMIZATION", "ALGORITHM_BENCHMARK"}: from app.lab_algorithms.service import AlgorithmLabService alg_service = AlgorithmLabService(db) project_id = payload.get("project_id", 1) hyp_id = payload.get("hypothesis_id") algs = payload.get("algorithm_names", ["SimulatedAnnealing", "DifferentialEvolution"]) problem = payload.get("problem_type", "rastrigin") dim = payload.get("dimension", 5) self.update_task_status(task_id, "RUNNING", progress=50.0) exp_res = alg_service.run_optimization_experiment( project_id=project_id, hypothesis_id=hyp_id, algorithm_names=algs, problem_type=problem, dim=dim ) self.update_task_status(task_id, "RUNNING", progress=80.0) result_data = exp_res evidence_hash = exp_res.get("report_sha256") or exp_res.get("evidence", {}).get("sha256_fingerprint") elif task_type in {"AI_TRAINING", "AI_TRAIN"}: from app.lab_ai.service import AILabService ai_service = AILabService(db) project_id = payload.get("project_id", 1) hyp_id = payload.get("hypothesis_id") model_name = payload.get("model_name", "task_model") arch = payload.get("architecture_type", "NEURAL_NETWORK_MLP") hidden_dim = payload.get("hidden_dim", 32) epochs = payload.get("epochs", 20) self.update_task_status(task_id, "RUNNING", progress=50.0) train_res = ai_service.train_and_evaluate_model( project_id=project_id, hypothesis_id=hyp_id, model_name=model_name, architecture_type=arch, hidden_dim=hidden_dim, epochs=epochs ) self.update_task_status(task_id, "RUNNING", progress=80.0) result_data = train_res evidence_hash = ( train_res.get("artifact_sha256") or train_res.get("report_sha256") or train_res.get("evidence", {}).get("sha256_fingerprint") ) elif task_type == "ALGORITHM_PREDICTIVE": from app.lab_algorithms.service import AlgorithmLabService alg_service = AlgorithmLabService(db) project_id = payload.get("project_id", 1) hyp_id = payload.get("hypothesis_id") algs = payload.get("algorithm_names", ["ridge_regression", "gradient_boosting", "knn_regressor"]) n_samples = payload.get("n_samples", 200) n_features = payload.get("n_features", 10) seed = payload.get("random_seed", 42) self.update_task_status(task_id, "RUNNING", progress=50.0) pred_res = alg_service.run_predictive_experiment( project_id=project_id, hypothesis_id=hyp_id, algorithm_names=algs, n_samples=n_samples, n_features=n_features, random_seed=seed ) self.update_task_status(task_id, "RUNNING", progress=80.0) result_data = pred_res evidence_hash = pred_res.get("report_sha256") elif task_type == "AI_ROBOTICS_SIM": from app.lab_ai.service import AILabService ai_service = AILabService(db) project_id = payload.get("project_id", 1) hyp_id = payload.get("hypothesis_id") sim_type = payload.get("simulation_type", "kalman_filter") steps = payload.get("steps", 80) self.update_task_status(task_id, "RUNNING", progress=50.0) sim_res = ai_service.run_robotics_simulation( project_id=project_id, simulation_type=sim_type, hypothesis_id=hyp_id, steps=steps ) self.update_task_status(task_id, "RUNNING", progress=80.0) result_data = sim_res evidence_hash = sim_res.get("report_sha256") elif task_type == "QUANTITATIVE_RESEARCH": from app.lab_algorithms.financial import QuantitativeResearchEngine from app.lab_algorithms.service import AlgorithmLabService QuantitativeResearchEngine.assert_no_live_trading() train_ratio = payload.get("train_ratio", 0.7) strategy = payload.get("strategy", payload.get("strategy_name", "MOMENTUM_RSI_MACD")) ticker = payload.get("ticker", "BENCHMARK_SPY") self.update_task_status(task_id, "RUNNING", progress=50.0) quant_res = QuantitativeResearchEngine.walk_forward_backtest( strategy_name=strategy, train_ratio=train_ratio, ticker=ticker ) alg_service = AlgorithmLabService(db) fin_exp = alg_service.run_financial_research( project_id=payload.get("project_id", 1), ticker=ticker ) quant_res["experiment_id"] = fin_exp.get("experiment_id") quant_res["report_id"] = fin_exp.get("report_id") self.update_task_status(task_id, "RUNNING", progress=80.0) result_data = quant_res evidence_hash = hashlib.sha256( json.dumps(quant_res.get("backtest", {}), sort_keys=True).encode("utf-8") ).hexdigest() elif task_type in {"TECH_RESEARCH", "DATASET_ANALYSIS"}: from app.lab_algorithms.tech_research import GlobalTechResearchManager tech_mgr = GlobalTechResearchManager(db) raw_sector = payload.get("sector") or payload.get("domain") or "GENERAL" sector = raw_sector.upper() if raw_sector.upper() in GlobalTechResearchManager.SECTORS else "GENERAL" question = payload.get("question") or payload.get("research_question") or "General technical assessment" hyp_statement = payload.get("hypothesis_statement") docs = payload.get("documents") or payload.get("sources") or [] doc_paths = payload.get("document_paths") or [] analysis_type = payload.get("analysis_type") criteria = payload.get("criteria") proj_id = payload.get("project_id") hyp_id = payload.get("hypothesis_id") self.update_task_status(task_id, "RUNNING", progress=50.0) tech_res = tech_mgr.create_tech_inquiry( sector=sector, research_question=question, hypothesis_statement=hyp_statement, sources=docs, document_paths=doc_paths, analysis_type=analysis_type, criteria=criteria, project_id=proj_id, hypothesis_id=hyp_id ) self.update_task_status(task_id, "RUNNING", progress=80.0) result_data = tech_res evidence_hash = ( tech_res.get("synthesis", {}).get("sha256_evidence_hash") or tech_res.get("evidence", {}).get("sha256_fingerprint") ) elif task_type == "ALGORITHM_INVESTIGATION": from app.lab_algorithms.service import AlgorithmLabService alg_service = AlgorithmLabService(db) self.update_task_status(task_id, "RUNNING", progress=40.0) inv_res = alg_service.investigate_new_algorithm_flow( problem_description=payload.get("problem_description", "Algorithm investigation"), candidate_algorithms=payload.get("candidate_algorithms", ["SimulatedAnnealing", "DifferentialEvolution"]), parameter_grid=payload.get("parameter_grid"), benchmark_problem=payload.get("benchmark_problem", "sphere"), dimension=payload.get("dimension", 5), project_id=payload.get("project_id"), proposed_custom_code=payload.get("proposed_custom_code") ) self.update_task_status(task_id, "RUNNING", progress=85.0) result_data = inv_res evidence_hash = inv_res.get("evidence_sha256") elif task_type == "AI_ARCHITECTURE_INVESTIGATION": from app.lab_ai.service import AILabService ai_service = AILabService(db) self.update_task_status(task_id, "RUNNING", progress=40.0) inv_res = ai_service.investigate_ai_architecture_flow( target_objective=payload.get("target_objective", "AI architecture investigation"), candidate_architectures=payload.get("candidate_architectures", ["NEURAL_NETWORK_MLP", "TREE_ENSEMBLE"]), domain=payload.get("domain", "GENERAL"), hidden_dims=payload.get("hidden_dims"), epochs=payload.get("epochs", 25), project_id=payload.get("project_id"), proposed_custom_code=payload.get("proposed_custom_code") ) self.update_task_status(task_id, "RUNNING", progress=85.0) result_data = inv_res evidence_hash = inv_res.get("evidence_sha256") elif task_type == "DISCOVERY_CYCLE": self.update_task_status(task_id, "RUNNING", progress=30.0) cycle_res = self.run_discovery_cycle( problem_statement=payload.get("problem_statement") or payload.get("question", "Discovery investigation"), lab_type=payload.get("lab_type", "ALGORITHM"), domain=payload.get("domain", "OPTIMIZATION"), benchmark_target=payload.get("benchmark_target", "sphere"), dimension=payload.get("dimension", 3), max_rounds=payload.get("max_rounds", 2), execution_budget_seconds=budget_seconds, project_id=payload.get("project_id"), hypothesis_id=payload.get("hypothesis_id"), parameter_space=payload.get("parameter_space"), candidate_architectures=payload.get("candidate_architectures"), force_no_improvement=payload.get("force_no_improvement", False), checkpoint=payload.get("checkpoint"), task_id=task_id ) self.update_task_status(task_id, "RUNNING", progress=90.0) result_data = cycle_res evidence_hash = cycle_res.get("evidence_sha256") elif task_type == "INTER_LAB_COLLABORATION": from app.core.collaboration import InterLabCoordinator collab_coord = InterLabCoordinator(db) collab_id = payload.get("collaboration_id") eval_exp_id = payload.get("evaluation_experiment_id") eval_metrics = payload.get("evaluation_metrics") baseline_metrics = payload.get("baseline_metrics") if not collab_id: result_data = { "status": "BLOCKED", "reason": "Unverifiable collaboration: Missing collaboration_id.", "task_id": task_id } elif eval_exp_id and eval_metrics and baseline_metrics: self.update_task_status(task_id, "RUNNING", progress=60.0) collab_obj = collab_coord.submit_evaluation( collaboration_id=collab_id, evaluation_experiment_id=eval_exp_id, evaluation_metrics=eval_metrics, baseline_metrics=baseline_metrics, primary_metric_name=payload.get("primary_metric_name", "accuracy"), custom_feedback=payload.get("custom_feedback") ) result_data = { "status": "COMPLETED" if collab_obj.status in {"ACCEPTED", "INTEGRATED"} else ("BLOCKED" if collab_obj.status == "BLOCKED" else "REJECTED"), "collaboration_id": collab_obj.id, "collaboration_status": collab_obj.status, "feedback": collab_obj.feedback } else: # Target lab executes real evaluation experiment self.update_task_status(task_id, "RUNNING", progress=60.0) collab_obj = collab_coord.execute_and_submit_evaluation( collaboration_id=collab_id, primary_metric_name=payload.get("primary_metric_name", "accuracy") ) result_data = { "status": "COMPLETED" if collab_obj.status in {"ACCEPTED", "INTEGRATED"} else ("BLOCKED" if collab_obj.status == "BLOCKED" else "REJECTED"), "collaboration_id": collab_obj.id, "collaboration_status": collab_obj.status, "feedback": collab_obj.feedback } # Step 6: Budget verification elapsed = time.time() - start_wall_time if elapsed > budget_seconds: self.update_task_status( task_id, "FAILED", error_log=f"Execution budget exceeded: {elapsed:.2f}s > {budget_seconds}s" ) return {"task_id": task_id, "status": "FAILED", "error": "BUDGET_EXCEEDED"} # Step 7: Check cancellation before committing completion if self.is_cancelled(task_id): return {"task_id": task_id, "status": "CANCELLED"} # Step 8 & 9: Evidence & Final status final_status = "COMPLETED" if isinstance(result_data, dict) and result_data.get("status") in { "INSUFFICIENT_EVIDENCE", "UNSUPPORTED_ANALYSIS", "BLOCKED", "FAILED", "NEEDS_MODEL", "UNSUPPORTED_TASK_TYPE", "SANDBOX_UNAVAILABLE", "PENDING_SAFE_ENVIRONMENT" }: final_status = result_data["status"] progress_val = 100.0 if final_status == "COMPLETED" else 50.0 self.update_task_status( task_id, status=final_status, progress=progress_val, result=result_data ) ret_summary = { "task_id": task_id, "task_type": task_type, "status": final_status, "elapsed_seconds": round(elapsed, 4), "evidence_hash": evidence_hash, "execution_control": { "requested_budget_seconds": budget_seconds, "elapsed_seconds": round(elapsed, 4), "budget_enforced_strictly": False, "budget_exceeded": elapsed > budget_seconds }, "result": result_data } if isinstance(result_data, dict): if "experiment_id" in result_data: ret_summary["experiment_id"] = result_data["experiment_id"] if "evidence_level" in result_data: ret_summary["evidence_level"] = result_data["evidence_level"] if "confidence_level" in result_data: ret_summary["confidence_level"] = result_data["confidence_level"] if "resumed_from_checkpoint" in result_data: ret_summary["resumed_from_checkpoint"] = result_data["resumed_from_checkpoint"] ret_summary["restored_round"] = result_data.get("restored_round", 0) return ret_summary except Exception as exec_err: tb = traceback.format_exc() is_non_retryable = ( isinstance(exec_err, (ValueError, PermissionError, FileNotFoundError)) or "SecurityViolation" in type(exec_err).__name__ ) db_task = db.query(models.CoordinationTask).filter(models.CoordinationTask.id == task_id).first() if not is_non_retryable and db_task and db_task.retry_count < db_task.max_retries: db_task.retry_count += 1 db_task.status = "PENDING" db_task.error_log = f"Retry {db_task.retry_count}/{db_task.max_retries} scheduled. Prior error: {str(exec_err)}" db.commit() return {"task_id": task_id, "status": "RETRY_SCHEDULED", "retry_count": db_task.retry_count} else: self.update_task_status( task_id, status="FAILED", progress=0.0, error_log=f"Execution failed: {str(exec_err)}\n{tb}" ) return {"task_id": task_id, "status": "FAILED", "error": str(exec_err)} finally: if self._db is None: db.close() def list_tasks(self, limit: int = 50) -> List[models.CoordinationTask]: db = self._get_db() try: return db.query(models.CoordinationTask).order_by( desc(models.CoordinationTask.id) ).limit(limit).all() finally: if self._db is None: db.close() def run_discovery_cycle( self, problem_statement: str, lab_type: str = "ALGORITHM", domain: str = "OPTIMIZATION", benchmark_target: str = "sphere", dimension: int = 3, max_rounds: int = 2, execution_budget_seconds: int = 120, project_id: Optional[int] = None, hypothesis_id: Optional[int] = None, parameter_space: Optional[Dict[str, Any]] = None, candidate_architectures: Optional[List[str]] = None, force_no_improvement: bool = False, checkpoint: Optional[Dict[str, Any]] = None, task_id: Optional[int] = None ) -> Dict[str, Any]: """ Executes an iterative, evidence-based improvement loop: PROPOSE -> EXECUTE -> MEASURE -> COMPARE -> REFINE -> CONSERVE RESULTS. Executes a maximum of 2 bounded rounds with decision history persisted to SQLite. Supports verified Round-1 checkpoint serialization and deterministic state restoration. """ import time import hashlib start_time = time.time() db = self._get_db() try: lab_type_upper = lab_type.upper() if lab_type_upper not in {"ALGORITHM", "AI"}: raise ValueError(f"Unsupported discovery lab_type '{lab_type}'. Must be 'ALGORITHM' or 'AI'.") # 1. Resolve or create Project project = None if project_id: project = db.query(models.ResearchProject).filter(models.ResearchProject.id == project_id).first() if not project: raise ValueError(f"INVALID_PROJECT_ID: ResearchProject with id {project_id} does not exist.") if not project: project = models.ResearchProject( title=f"Discovery Program: {problem_statement[:60]}", description=f"Automated experimental discovery cycle for {problem_statement}", lab_type=lab_type_upper, domain=domain ) db.add(project) db.commit() db.refresh(project) # 2. Formulate or reuse single Hypothesis hyp = None if hypothesis_id is not None: hyp = db.query(models.Hypothesis).filter( models.Hypothesis.id == hypothesis_id, models.Hypothesis.project_id == project.id ).first() if not hyp: raise ValueError( f"INVALID_HYPOTHESIS_ID: Hypothesis with id {hypothesis_id} does not exist or does not belong to project {project.id}." ) else: hyp_statement = ( f"Under problem '{problem_statement}', iterative candidate parameter refinement " f"will identify an optimized configuration superior to baseline on {benchmark_target}." ) hyp = models.Hypothesis( project_id=project.id, title=f"Discovery Hypothesis: {problem_statement[:60]}", statement=hyp_statement, rationale=f"Systematic {max_rounds}-round evidence discovery loop on {lab_type_upper} Lab.", status="PROPOSED", confidence_score=0.60 ) db.add(hyp) db.commit() db.refresh(hyp) # 3. Create Main Discovery Experiment from app.core.scientific_memory import ScientificMemory memory = ScientificMemory(db) exp = memory.record_experiment( project_id=project.id, hypothesis_id=hyp.id, name=f"Discovery Cycle: {problem_statement[:60]}", experiment_type="EXPERIMENTAL_DISCOVERY_CYCLE", config={ "problem_statement": problem_statement, "lab_type": lab_type_upper, "benchmark_target": benchmark_target, "dimension": dimension, "max_rounds": max_rounds, "budget_seconds": execution_budget_seconds } ) exp.status = "RUNNING" db.commit() decision_history = [] overall_winner = None rounds_completed = 0 resumed_from_checkpoint = False restored_round = 0 round_1_checkpoint = None # ------------------------------------------------------------- # ROUND 1: Initial Exploration (or Verified Checkpoint Restoration) # ------------------------------------------------------------- if (time.time() - start_time) > execution_budget_seconds: return { "project_id": project.id, "hypothesis_id": hyp.id, "experiment_id": exp.id, "status": "BUDGET_EXCEEDED", "reason": "Execution budget exceeded before Round 1 dispatch." } if checkpoint is not None: if not isinstance(checkpoint, dict) or checkpoint.get("resumable") is not True or checkpoint.get("idempotent") is not True: raise ValueError("INVALID_CHECKPOINT: Checkpoint must specify resumable=True and idempotent=True.") ck_state = checkpoint.get("state") if not isinstance(ck_state, dict): raise ValueError("INVALID_CHECKPOINT: Checkpoint state dictionary is missing.") expected_ck_hash = checkpoint.get("checkpoint_sha256") computed_ck_hash = hashlib.sha256( json.dumps(ck_state, sort_keys=True).encode("utf-8") ).hexdigest() if not expected_ck_hash or expected_ck_hash != computed_ck_hash: raise ValueError("INVALID_CHECKPOINT_HASH: Checkpoint SHA-256 verification failed.") ck_tid = checkpoint.get("task_id") if ck_tid is not None and task_id is not None and int(ck_tid) != int(task_id): raise ValueError(f"CHECKPOINT_OWNERSHIP_MISMATCH: Checkpoint belongs to task {ck_tid}, not {task_id}.") ck_ttype = checkpoint.get("task_type") if ck_ttype is not None and str(ck_ttype) != "DISCOVERY_CYCLE": raise ValueError(f"CHECKPOINT_TYPE_MISMATCH: Checkpoint task_type '{ck_ttype}' != 'DISCOVERY_CYCLE'.") if ( ck_state.get("round_completed") != 1 or ck_state.get("lab_type") != lab_type_upper or ck_state.get("benchmark_target") != benchmark_target or int(ck_state.get("dimension", -1)) != int(dimension) ): raise ValueError("CHECKPOINT_PARAMETER_MISMATCH: Checkpoint parameters do not match active cycle.") if lab_type_upper != "ALGORITHM": raise ValueError("UNSUPPORTED_CHECKPOINT_LAB: Only ALGORITHM discovery checkpoints support state restoration.") # Restore Round 1 state deterministically from verified checkpoint r1_best = ck_state["r1_best"] r1_best_config = ck_state["r1_best_config"] r1_metric = float(ck_state["r1_metric"]) baseline_metric = float(ck_state["baseline_metric"]) baseline_name = ck_state["baseline_name"] summary_1 = ck_state["summary_1"] decision_history.append(dict(ck_state["decision_history_r1"])) overall_winner = dict(ck_state["overall_winner_r1"]) rounds_completed = 1 resumed_from_checkpoint = True restored_round = 1 round_1_checkpoint = checkpoint else: rounds_completed += 1 if lab_type_upper == "ALGORITHM": from app.lab_algorithms.service import AlgorithmLabService alg_svc = AlgorithmLabService(db) r1_cands = ["SimulatedAnnealing", "DifferentialEvolution", "GeneticAlgorithm"] r1_grid = parameter_space.get("round_1") if (parameter_space and "round_1" in parameter_space) else None r1_res = alg_svc.investigate_new_algorithm_flow( problem_description=problem_statement, candidate_algorithms=r1_cands, parameter_grid=r1_grid, benchmark_problem=benchmark_target, dimension=dimension, project_id=project.id, hypothesis_id=hyp.id ) r1_best = r1_res["best_candidate"] r1_best_config = r1_res.get("best_candidate_config", r1_best) r1_metric = r1_res["mean_top_fitness"] baseline_metric = r1_res["baseline_fitness"] baseline_name = r1_res["baseline_algorithm"] summary_1 = r1_res["candidate_summary"] r1_improved = (r1_metric > baseline_metric) and not force_no_improvement if r1_improved: r2_decision = f"Improvement detected (+{r1_metric - baseline_metric:.4f}). Proceeding to Round 2 focalized tuning around {r1_best}." else: r2_decision = "No candidate surpassed baseline in Round 1. Round 2 focalized exploration scheduled with alternative parameter schedule." r1_details = summary_1.get(r1_best_config) or summary_1.get(r1_best, {}) r1_candidates_count = r1_res.get("candidates_count") or len([k for k in summary_1 if k != r1_best or k == r1_best_config]) r1_runs_count = r1_res.get("physical_runs_count") or (r1_candidates_count * 3) r1_families = r1_res.get("algorithm_families") or candidate_algorithms r1_configs = r1_res.get("candidate_configurations") or [r1_best_config] decision_history.append({ "round": 1, "type": "BROAD_EXPLORATION", "candidates_evaluated": r1_candidates_count, "physical_runs_count": r1_runs_count, "algorithm_families": r1_families, "candidate_configurations": r1_configs, "best_candidate": r1_best, "best_candidate_config": r1_best_config, "winning_parameters": r1_details.get("parameters", {}), "candidate_summary": summary_1, "best_metric": r1_metric, "baseline_metric": baseline_metric, "decision": r2_decision }) overall_winner = { "name": r1_best, "config": r1_best_config, "metric": r1_metric, "round": 1, "details": r1_details } r1_ck_state = { "round_completed": 1, "lab_type": lab_type_upper, "benchmark_target": benchmark_target, "dimension": dimension, "r1_best": r1_best, "r1_best_config": r1_best_config, "r1_metric": r1_metric, "baseline_metric": baseline_metric, "baseline_name": baseline_name, "summary_1": summary_1, "decision_history_r1": decision_history[0], "overall_winner_r1": overall_winner } r1_ck_sha256 = hashlib.sha256( json.dumps(r1_ck_state, sort_keys=True).encode("utf-8") ).hexdigest() round_1_checkpoint = { "resumable": True, "idempotent": True, "task_id": task_id, "task_type": "DISCOVERY_CYCLE", "state": r1_ck_state, "checkpoint_sha256": r1_ck_sha256 } else: # AI Lab from app.lab_ai.service import AILabService ai_svc = AILabService(db) r1_archs = candidate_architectures or ["NEURAL_NETWORK_MLP", "TREE_ENSEMBLE"] r1_res = ai_svc.investigate_ai_architecture_flow( target_objective=problem_statement, candidate_architectures=r1_archs, domain=domain, hidden_dims=[16, 32], epochs=10, project_id=project.id, hypothesis_id=hyp.id, defer_test_evaluation=True ) r1_best_obj = r1_res["best_candidate"] r1_best = r1_best_obj["model_name"] r1_best_arch = r1_best_obj.get("architecture") if not r1_best_arch: r1_best_arch = "TREE_ENSEMBLE" if "tree" in r1_best.lower() else "NEURAL_NETWORK_MLP" r1_metric = r1_best_obj["val_accuracy"] baseline_metric = r1_best_obj["baseline_val_accuracy"] baseline_name = "LogisticRegression_Baseline" r1_improved = (r1_metric > baseline_metric) and not force_no_improvement r2_decision = f"Val accuracy {r1_metric:.4f} vs baseline {baseline_metric:.4f}. Round 2 tuning scheduled for winning family {r1_best_arch}." decision_history.append({ "round": 1, "type": "BROAD_EXPLORATION", "candidates_evaluated": r1_res["evaluations_count"], "physical_runs_count": r1_res["evaluations_count"], "algorithm_families": r1_archs, "candidate_configurations": [c["model_name"] for c in r1_res.get("candidate_evaluations", [])], "candidate_summary": {c["model_name"]: dict(c) for c in r1_res.get("candidate_evaluations", [])}, "best_candidate": r1_best, "architecture": r1_best_arch, "best_metric": r1_metric, "baseline_metric": baseline_metric, "decision": r2_decision }) overall_winner = { "name": r1_best, "metric": r1_metric, "architecture": r1_best_arch, "round": 1, "details": r1_best_obj, "trained_model": r1_res.get("trained_model"), "test_data_ref": r1_res.get("test_data_ref") } # ------------------------------------------------------------- # ROUND 2: Focalized Refinement (if max_rounds >= 2 and budget allows) # ------------------------------------------------------------- remaining_time = execution_budget_seconds - (time.time() - start_time) if max_rounds >= 2 and remaining_time > 1.0 and not force_no_improvement: rounds_completed += 1 if lab_type_upper == "ALGORITHM": from app.lab_algorithms.service import AlgorithmLabService alg_svc = AlgorithmLabService(db) winning_alg = summary_1.get(r1_best_config, {}).get("algorithm", r1_best) winner_details = summary_1.get(r1_best_config) or summary_1.get(r1_best, {}) winning_params = winner_details.get("parameters", {}) if "round_2" in (parameter_space or {}): r2_grid = parameter_space["round_2"] refine_rationale = f"User-specified parameter schedule for Round 2 around {winning_alg}." else: if winning_alg == "GeneticAlgorithm": pop = int(winning_params.get("population_size", 30)) gen = int(winning_params.get("generations", 25)) mut = float(winning_params.get("mutation_rate", 0.15)) cross = float(winning_params.get("crossover_rate", 0.8)) r2_grid = { "GeneticAlgorithm": [ {"population_size": max(10, int(pop * 0.8)), "generations": max(5, int(gen * 1.1)), "mutation_rate": round(max(0.01, mut * 0.8), 3), "crossover_rate": cross}, {"population_size": pop, "generations": max(5, int(gen * 1.2)), "mutation_rate": round(min(0.5, mut * 1.2), 3), "crossover_rate": min(0.95, round(cross + 0.05, 2))}, {"population_size": int(pop * 1.2), "generations": max(5, int(gen * 1.3)), "mutation_rate": round(mut, 3), "crossover_rate": cross} ] } refine_rationale = f"Perturbing population around {pop} (±20%), generations around {gen} (+10-30%), mutation around {mut}." elif winning_alg == "DifferentialEvolution": pop = int(winning_params.get("population_size", 30)) gen = int(winning_params.get("generations", 25)) f = float(winning_params.get("mutation_factor", 0.7)) cr = float(winning_params.get("crossover_rate", 0.8)) r2_grid = { "DifferentialEvolution": [ {"population_size": max(10, int(pop * 0.8)), "generations": max(5, int(gen * 1.1)), "mutation_factor": round(max(0.1, f - 0.1), 2), "crossover_rate": round(min(0.95, cr + 0.05), 2)}, {"population_size": pop, "generations": max(5, int(gen * 1.2)), "mutation_factor": round(min(1.2, f + 0.1), 2), "crossover_rate": round(max(0.1, cr - 0.05), 2)}, {"population_size": int(pop * 1.2), "generations": max(5, int(gen * 1.3)), "mutation_factor": f, "crossover_rate": cr} ] } refine_rationale = f"Perturbing mutation_factor F around {f} (±0.1) and crossover_rate CR around {cr} (±0.05)." else: # SimulatedAnnealing temp = float(winning_params.get("initial_temp", 50.0)) cooling = float(winning_params.get("cooling_rate", 0.92)) iters = int(winning_params.get("iterations_per_temp", 20)) r2_grid = { "SimulatedAnnealing": [ {"initial_temp": round(temp * 0.85, 1), "cooling_rate": round(min(0.99, cooling + 0.02), 3), "iterations_per_temp": int(iters * 1.2)}, {"initial_temp": round(temp * 1.15, 1), "cooling_rate": round(min(0.99, cooling + 0.01), 3), "iterations_per_temp": int(iters * 1.4)}, {"initial_temp": temp, "cooling_rate": round(max(0.70, cooling - 0.02), 3), "iterations_per_temp": iters} ] } refine_rationale = f"Perturbing initial_temp around {temp} (±15%) and cooling_rate around {cooling} (+0.01-0.02)." r2_res = alg_svc.investigate_new_algorithm_flow( problem_description=f"Refinement: {problem_statement}", candidate_algorithms=[winning_alg], parameter_grid=r2_grid, benchmark_problem=benchmark_target, dimension=dimension, project_id=project.id, hypothesis_id=hyp.id ) r2_best = r2_res["best_candidate"] r2_best_config = r2_res.get("best_candidate_config", r2_best) r2_metric = r2_res["mean_top_fitness"] summary_2 = r2_res["candidate_summary"] r2_beat_r1 = r2_metric > r1_metric r2_candidates_count = r2_res.get("candidates_count") or len([k for k in summary_2 if k != r2_best or k == r2_best_config]) r2_runs_count = r2_res.get("physical_runs_count") or (r2_candidates_count * 3) r2_families = r2_res.get("algorithm_families") or [winning_alg] r2_configs = r2_res.get("candidate_configurations") or [] decision_history.append({ "round": 2, "type": "FOCALIZED_REFINEMENT", "previous_winning_algorithm": winning_alg, "previous_winning_config": r1_best_config, "previous_winning_params": winning_params, "previous_best_metric": r1_metric, "refinement_rationale": refine_rationale, "candidates_evaluated": r2_candidates_count, "physical_runs_count": r2_runs_count, "algorithm_families": r2_families, "candidate_configurations": r2_configs, "candidate_summary": summary_2, "best_candidate": r2_best, "best_candidate_config": r2_best_config, "best_metric": r2_metric, "decision": f"Round 2 refinement completed on {winning_alg}. Gain over Round 1: {r2_beat_r1} (delta: {r2_metric - r1_metric:+.4f})." + ("" if r2_beat_r1 else " Retaining Round 1 winner.") }) if r2_beat_r1: overall_winner = { "name": r2_best, "config": r2_best_config, "metric": r2_metric, "round": 2, "details": summary_2.get(r2_best_config) or summary_2.get(r2_best, {}) } else: # AI Lab from app.lab_ai.service import AILabService ai_svc = AILabService(db) winning_arch = overall_winner.get("architecture") if not winning_arch: winning_arch = "TREE_ENSEMBLE" if "tree" in r1_best.lower() else "NEURAL_NETWORK_MLP" if winning_arch == "TREE_ENSEMBLE": r2_archs = ["TREE_ENSEMBLE"] h_dim = overall_winner["details"].get("hidden_dim", 32) r2_hidden_dims = sorted(list({max(5, int(h_dim * 0.75)), h_dim, int(h_dim * 1.25), int(h_dim * 1.5)})) refine_rationale = f"Winning architecture from Round 1 was TREE_ENSEMBLE. Exploring tree estimators around n_estimators={h_dim} -> {r2_hidden_dims}." r2_epochs = 10 else: r2_archs = ["NEURAL_NETWORK_MLP"] h_dim = overall_winner["details"].get("hidden_dim", 32) r2_hidden_dims = sorted(list({max(8, int(h_dim * 0.75)), h_dim, int(h_dim * 1.5), int(h_dim * 2.0)})) refine_rationale = f"Winning architecture from Round 1 was NEURAL_NETWORK_MLP. Exploring hidden dimensions around hidden_dim={h_dim} -> {r2_hidden_dims}." r2_epochs = 15 r2_res = ai_svc.investigate_ai_architecture_flow( target_objective=f"Refinement: {problem_statement}", candidate_architectures=r2_archs, domain=domain, hidden_dims=r2_hidden_dims, epochs=r2_epochs, project_id=project.id, hypothesis_id=hyp.id, defer_test_evaluation=True ) r2_best_obj = r2_res["best_candidate"] r2_best = r2_best_obj["model_name"] r2_metric = r2_best_obj["val_accuracy"] r2_beat_r1 = r2_metric > r1_metric decision_history.append({ "round": 2, "type": "FOCALIZED_REFINEMENT", "previous_winning_architecture": winning_arch, "previous_winner_arch": winning_arch, "refined_family": winning_arch, "previous_winning_config": r1_best, "previous_best_metric": r1_metric, "refinement_rationale": refine_rationale, "candidates_evaluated": r2_res["evaluations_count"], "physical_runs_count": r2_res["evaluations_count"], "algorithm_families": [winning_arch], "candidate_configurations": [c["model_name"] for c in r2_res.get("candidate_evaluations", [])], "candidate_summary": {c["model_name"]: dict(c) for c in r2_res.get("candidate_evaluations", [])}, "best_candidate": r2_best, "best_metric": r2_metric, "decision": f"Round 2 refinement completed on {winning_arch}. Gain over Round 1: {r2_beat_r1} (delta: {r2_metric - r1_metric:+.4f})." + ("" if r2_beat_r1 else " Retaining Round 1 winner.") }) if r2_beat_r1: overall_winner = { "name": r2_best, "metric": r2_metric, "architecture": winning_arch, "round": 2, "details": r2_best_obj, "trained_model": r2_res.get("trained_model"), "test_data_ref": r2_res.get("test_data_ref") } # ------------------------------------------------------------- # Final Held-Out Test Evaluation (AI Lab only, strictly unpolluted) # ------------------------------------------------------------- if lab_type_upper == "AI": winner_model = overall_winner.get("trained_model") winner_arch = overall_winner.get("architecture") test_data = overall_winner.get("test_data_ref") if winner_model is not None and test_data is not None: X_test = test_data["X_test"] y_test = test_data["y_test"] X_train = test_data["X_train"] if winner_arch == "TREE_ENSEMBLE": test_preds = winner_model.predict(X_test) final_test_acc = float(np.mean(test_preds == y_test)) else: from sklearn.preprocessing import StandardScaler import torch sc = StandardScaler().fit(X_train) X_te_scaled = sc.transform(X_test) winner_model.eval() with torch.no_grad(): te_out = winner_model(torch.tensor(X_te_scaled, dtype=torch.float32)) te_preds = torch.argmax(te_out, dim=1).numpy() final_test_acc = float(np.mean(te_preds == y_test)) overall_winner["test_accuracy"] = round(final_test_acc, 4) if "details" in overall_winner and isinstance(overall_winner["details"], dict): overall_winner["details"]["test_accuracy"] = round(final_test_acc, 4) overall_winner.pop("trained_model", None) overall_winner.pop("test_data_ref", None) # ------------------------------------------------------------- # Conclusion & Scientific Evidence Persistence # ------------------------------------------------------------- final_metric = overall_winner["metric"] if force_no_improvement: final_verdict = "NO_IMPROVEMENT" final_status = "COMPLETED" confidence_level = "LOW" limitations = f"Control-flagged test condition (force_no_improvement=True). Completed {rounds_completed} rounds on synthetic benchmark." conclusion = ( f"Discovery cycle on '{problem_statement}' completed {rounds_completed} rounds. " f"No candidate configuration was able to improve over baseline {baseline_name} " f"({baseline_metric:.4f}). Verdict: NO_IMPROVEMENT." ) elif final_metric > baseline_metric: is_r1_conserved = (rounds_completed > 1 and overall_winner["round"] == 1) final_verdict = "IMPROVEMENT (ROUND_1_WINNER_PRESERVED)" if is_r1_conserved else "IMPROVEMENT" final_status = "COMPLETED" confidence_level = "MODERATE_SYNTHETIC" limitations = ( f"Empirical improvement validated across random seeds on synthetic benchmark topology. " f"No statistical significance hypothesis test performed; external production validity not claimed. " f"Completed {rounds_completed} bounded rounds." ) if is_r1_conserved: conclusion = ( f"Discovery cycle on '{problem_statement}' completed {rounds_completed} rounds. " f"Winner '{overall_winner['name']}' from Round 1 CONSERVED as overall winner (Round 2 did not demonstrate superior metric). " f"Metric {final_metric:.4f} vs baseline {baseline_name} {baseline_metric:.4f} (delta: {final_metric - baseline_metric:+.4f})." ) else: conclusion = ( f"Discovery cycle on '{problem_statement}' completed {rounds_completed} rounds. " f"Winner '{overall_winner['name']}' in Round {overall_winner['round']} demonstrated measurable improvement: " f"metric {final_metric:.4f} vs baseline {baseline_name} {baseline_metric:.4f} (delta: {final_metric - baseline_metric:+.4f})." ) else: final_verdict = "NO_IMPROVEMENT" final_status = "COMPLETED" confidence_level = "MODERATE_SYNTHETIC" limitations = f"Candidates matched or regressed against baseline. Evaluated on synthetic benchmark topology across {rounds_completed} rounds." conclusion = ( f"Discovery cycle on '{problem_statement}' completed {rounds_completed} rounds. " f"Evaluated candidates matched or regressed against baseline {baseline_name} " f"({baseline_metric:.4f}). Verdict: NO_IMPROVEMENT." ) ev_payload = json.dumps({ "cycle_id": f"proj_{project.id}_hyp_{hyp.id}_exp_{exp.id}", "problem_statement": problem_statement, "lab_type": lab_type_upper, "rounds_completed": rounds_completed, "decision_history": decision_history, "overall_winner": {k: v for k, v in overall_winner.items() if k not in {"trained_model", "test_data_ref"}}, "baseline_metric": baseline_metric, "final_verdict": final_verdict }, sort_keys=True) ev_hash = hashlib.sha256(ev_payload.encode("utf-8")).hexdigest() memory.record_evidence( experiment_id=exp.id, hypothesis_id=hyp.id, title=f"Discovery Cycle Evidence: {overall_winner['name']}", conclusion=conclusion, confidence_level=confidence_level, limitations=limitations ) exp.status = final_status exp.completed_at = datetime.utcnow() db.commit() ret_dict = { "project_id": project.id, "hypothesis_id": hyp.id, "experiment_id": exp.id, "status": final_status, "verdict": final_verdict, "rounds_completed": rounds_completed, "resumed_from_checkpoint": resumed_from_checkpoint, "restored_round": restored_round, "round_1_checkpoint": round_1_checkpoint, "decision_history": decision_history, "best_candidate": overall_winner, "baseline_name": baseline_name, "baseline_metric": baseline_metric, "final_metric": final_metric, "evidence_sha256": ev_hash, "confidence_level": confidence_level, "limitations": limitations, "conclusion": conclusion } if lab_type_upper == "AI": ret_dict["held_out_test_accuracy"] = overall_winner.get("test_accuracy") return ret_dict finally: if self._db is None: db.close() class ScientificTaskWorker: """ Controlled background thread worker for consuming coordination tasks. Can be started, stopped, and audited explicitly. NOT configured for automated Windows boot execution. """ _instance = None _lock = threading.Lock() def __init__(self, poll_interval: float = 1.0): self.poll_interval = poll_interval self._thread: Optional[threading.Thread] = None self._stop_event = threading.Event() self._active_task_id: Optional[int] = None self._processed_count: int = 0 self._last_error: Optional[str] = None @classmethod def get_instance(cls) -> "ScientificTaskWorker": with cls._lock: if cls._instance is None: cls._instance = ScientificTaskWorker() return cls._instance def is_running(self) -> bool: return self._thread is not None and self._thread.is_alive() def start(self) -> bool: """Starts the background worker thread if not already running.""" with self._lock: if self.is_running(): return False self._stop_event.clear() self._thread = threading.Thread( target=self._worker_loop, name="ScientificCoordinatorWorkerThread", daemon=True ) self._thread.start() return True def stop(self, timeout: float = 5.0) -> bool: """ Signals the worker to terminate and waits for completion. Returns True if the worker thread stopped within timeout. Returns False if the worker thread is still alive after timeout, preserving the reference. """ with self._lock: if not self.is_running(): return False self._stop_event.set() if self._thread: self._thread.join(timeout=timeout) if self._thread.is_alive(): # Thread did not stop within timeout; preserve reference and report incomplete stop return False self._thread = None self._active_task_id = None return True def status(self) -> Dict[str, Any]: return { "is_running": self.is_running(), "active_task_id": self._active_task_id, "processed_count": self._processed_count, "last_error": self._last_error } def _worker_loop(self): coord = ScientificCoordinator() while not self._stop_event.is_set(): try: # Atomically claim the next pending task claimed_task = coord.claim_next_pending_task() if claimed_task: self._active_task_id = claimed_task.id res = coord.process_next_task(task_id=claimed_task.id) self._processed_count += 1 self._active_task_id = None else: self._stop_event.wait(self.poll_interval) except Exception as e: self._last_error = f"{type(e).__name__}: {str(e)}" self._active_task_id = None self._stop_event.wait(self.poll_interval)