""" Optimus Research Center - Inter-Lab Collaboration Module (Hardened & Traceable) Implements formal contract exchange between Lab 1 (Algorithms) and Lab 2 (AI), ensuring strict provenance, non-arbitrary experiment linking, measurable evaluation criteria, and rejection/blocking of unverified transfers. """ import json from datetime import datetime from typing import Dict, Any, List, Optional from sqlalchemy.orm import Session from app.core.database import SessionLocal from app.core import models from app.core.scientific_memory import ScientificMemory class InterLabCoordinator: """Manages the verifiable transfer protocol between Algorithm Lab and AI Lab.""" 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 propose_technique( self, title: str, source_lab: str, target_lab: str, technique_name: str, technique_version: str, payload: Dict[str, Any], origin_experiment_id: int ) -> models.InterLabCollaboration: """ Step 1 & 2: Lab publishes a structured optimization technique or model architecture strictly linked to an existing origin experiment ID with benchmarked metrics. """ db = self._get_db() try: # Verify origin experiment exists physically origin_exp = db.query(models.Experiment).filter(models.Experiment.id == origin_experiment_id).first() if not origin_exp: raise ValueError( f"Invalid origin_experiment_id {origin_experiment_id}: Experiment does not exist in database." ) # Record origin_experiment_id in payload payload_copy = dict(payload) payload_copy["origin_experiment_id"] = origin_experiment_id payload_copy["origin_experiment_name"] = origin_exp.name collab = models.InterLabCollaboration( title=title, source_lab=source_lab, target_lab=target_lab, technique_name=technique_name, technique_version=technique_version, payload_json=json.dumps(payload_copy), evaluation_metrics_json="{}", status="PROPOSED", feedback=f"Awaiting measurable evaluation by {target_lab}. Origin verified (Exp #{origin_experiment_id})." ) db.add(collab) db.commit() db.refresh(collab) return collab finally: if self._db is None: db.close() def submit_evaluation( self, collaboration_id: int, evaluation_experiment_id: int, evaluation_metrics: Dict[str, Any], baseline_metrics: Dict[str, Any], primary_metric_name: str = "accuracy", custom_feedback: Optional[str] = None ) -> models.InterLabCollaboration: """ Step 6, 7 & 8: Target lab executes comparative benchmark and submits formal evaluation. STRICT TRACEABILITY RULES: 1. evaluation_experiment_id MUST exist in the database. 2. Acceptance is strictly bound to measurable criteria (candidate metric >= baseline metric). 3. If origin or evaluation experiment cannot be verified, transfer is set to BLOCKED. """ db = self._get_db() try: collab = db.query(models.InterLabCollaboration).filter( models.InterLabCollaboration.id == collaboration_id ).first() if not collab: raise ValueError(f"Collaboration #{collaboration_id} not found.") # Validate evaluation experiment exists eval_exp = db.query(models.Experiment).filter(models.Experiment.id == evaluation_experiment_id).first() if not eval_exp: collab.status = "BLOCKED" collab.feedback = f"BLOCKED: Evaluation experiment #{evaluation_experiment_id} does not exist in scientific memory." db.commit() db.refresh(collab) return collab payload = json.loads(collab.payload_json or "{}") origin_exp_id = payload.get("origin_experiment_id") if not origin_exp_id: collab.status = "BLOCKED" collab.feedback = "BLOCKED: Origin experiment metadata is missing or corrupted." db.commit() db.refresh(collab) return collab origin_exp = db.query(models.Experiment).filter(models.Experiment.id == origin_exp_id).first() if not origin_exp: collab.status = "BLOCKED" collab.feedback = f"BLOCKED: Origin experiment #{origin_exp_id} could not be validated." db.commit() db.refresh(collab) return collab # Measurable criteria evaluation c_val = evaluation_metrics.get(primary_metric_name) b_val = baseline_metrics.get(primary_metric_name) if c_val is None or b_val is None: collab.status = "BLOCKED" collab.feedback = f"BLOCKED: Primary metric '{primary_metric_name}' missing from evaluation or baseline payload." db.commit() db.refresh(collab) return collab is_improvement = float(c_val) > float(b_val) new_status = "ACCEPTED" if is_improvement else "REJECTED" delta = round(float(c_val) - float(b_val), 4) auto_feedback = ( f"Empirical evaluation on Exp #{evaluation_experiment_id} verified: " f"Candidate {primary_metric_name}={c_val} vs Baseline={b_val} (delta: {delta:+0.4f}). " f"Status: {new_status}." ) if custom_feedback: auto_feedback += f" Notes: {custom_feedback}" collab.evaluation_metrics_json = json.dumps({ "candidate": evaluation_metrics, "baseline": baseline_metrics, "primary_metric": primary_metric_name, "delta": delta, "is_improvement": is_improvement, "evaluation_experiment_id": evaluation_experiment_id, "origin_experiment_id": origin_exp_id }) collab.status = new_status collab.feedback = auto_feedback collab.updated_at = datetime.utcnow() db.commit() db.refresh(collab) # Record shared discovery into scientific evidence strictly linked to eval_exp memory = ScientificMemory(db) memory.record_evidence( experiment_id=eval_exp.id, hypothesis_id=eval_exp.hypothesis_id, title=f"Collaboration Result: {collab.technique_name} v{collab.technique_version}", conclusion=auto_feedback, confidence_level="VERY_HIGH" if is_improvement else "MEDIUM", limitations="Measured on reproducible cross-validation split with non-arbitrary experiment linkage.", sources=[ {"type": "ORIGIN_EXPERIMENT", "id": origin_exp_id, "name": origin_exp.name}, {"type": "EVALUATION_EXPERIMENT", "id": eval_exp.id, "name": eval_exp.name}, {"type": "COLLABORATION_CONTRACT", "id": collab.id} ] ) return collab finally: if self._db is None: db.close() def execute_and_submit_evaluation( self, collaboration_id: int, primary_metric_name: str = "accuracy" ) -> models.InterLabCollaboration: """ Executes an empirical evaluation in the target receiving laboratory, persists the real evaluation experiment, and records the formal collaboration outcome. """ db = self._get_db() try: collab = db.query(models.InterLabCollaboration).filter( models.InterLabCollaboration.id == collaboration_id ).first() if not collab: raise ValueError(f"Collaboration #{collaboration_id} not found.") payload = json.loads(collab.payload_json or "{}") origin_exp_id = payload.get("origin_experiment_id") if not origin_exp_id: collab.status = "BLOCKED" collab.feedback = "BLOCKED: Origin experiment metadata is missing or corrupted." db.commit() db.refresh(collab) return collab if collab.target_lab == "AI_LAB": from app.lab_ai.service import AILabService ai_svc = AILabService(db) proj = db.query(models.ResearchProject).filter( models.ResearchProject.lab_type == "CROSS_LAB" ).first() if not proj: proj = models.ResearchProject( title="Cross-Lab Optimization Initiative", description="Collaborative research bridging Algorithm & AI Labs", lab_type="CROSS_LAB", domain="COLLABORATION" ) db.add(proj) db.commit() db.refresh(proj) tuned_dim = int(payload.get("tuned_hidden_dim", payload.get("best_hidden_dim", 48))) tuned_epochs = int(payload.get("tuned_epochs", 30)) eval_result = ai_svc.train_and_evaluate_model( project_id=proj.id, hypothesis_id=None, model_name=f"tuned_{collab.technique_name.lower()}", architecture_type="NEURAL_NETWORK_MLP", hidden_dim=tuned_dim, epochs=tuned_epochs ) feedback = ( f"Technique {collab.technique_name} evaluated empirically on neural architecture. " f"Achieved accuracy: {eval_result['candidate_metrics']['accuracy']} " f"vs baseline: {eval_result['baseline_metrics']['accuracy']}. Verdict: {eval_result['audit']['verdict']}." ) return self.submit_evaluation( collaboration_id=collaboration_id, evaluation_experiment_id=eval_result["experiment_id"], evaluation_metrics=eval_result["candidate_metrics"], baseline_metrics=eval_result["baseline_metrics"], primary_metric_name="accuracy", custom_feedback=feedback ) elif collab.target_lab == "ALGORITHM_LAB": from app.lab_algorithms.service import AlgorithmLabService alg_svc = AlgorithmLabService(db) proj = db.query(models.ResearchProject).filter( models.ResearchProject.lab_type == "CROSS_LAB" ).first() if not proj: proj = models.ResearchProject( title="Cross-Lab Algorithm Optimization Initiative", description="Collaborative research bridging Algorithm & AI Labs", lab_type="CROSS_LAB", domain="COLLABORATION" ) db.add(proj) db.commit() db.refresh(proj) cands = payload.get("candidate_algorithms") or [] approved_algorithms = { "simulated_annealing", "differential_evolution", "genetic_algorithm", "bayesian_surrogate", "simulatedannealing", "differentialevolution", "geneticalgorithm", "bayesiansurrogate" } clean_cands = [ c for c in cands if c.lower().replace(" ", "_").replace("-", "") in approved_algorithms ] if not clean_cands: clean_cands = ["GeneticAlgorithm", "SimulatedAnnealing"] elif len(clean_cands) == 1: clean_cands = ["SimulatedAnnealing", clean_cands[0]] prob = payload.get("benchmark_problem", "sphere") dimension = int(payload.get("dimension", 5)) eval_res = alg_svc.run_optimization_experiment( project_id=proj.id, hypothesis_id=None, algorithm_names=clean_cands, problem_type=prob, dim=dimension, random_seed=42 ) bench = eval_res["benchmark"] runs = bench.get("runs", []) base_fit = runs[0]["metrics"].get("best_fitness", 0.0) if runs else bench.get("baseline_fitness", 0.0) if len(runs) > 1: cand_fit = max(r["metrics"]["best_fitness"] for r in runs[1:]) else: cand_fit = base_fit return self.submit_evaluation( collaboration_id=collaboration_id, evaluation_experiment_id=eval_res["experiment_id"], evaluation_metrics={"fitness": cand_fit}, baseline_metrics={"fitness": base_fit}, primary_metric_name="fitness", custom_feedback=f"Target Algorithm Lab executed benchmark experiment #{eval_res['experiment_id']}." ) else: collab.status = "BLOCKED" collab.feedback = f"BLOCKED: Unsupported target laboratory '{collab.target_lab}'." db.commit() db.refresh(collab) return collab finally: if self._db is None: db.close() def list_collaborations(self) -> List[models.InterLabCollaboration]: db = self._get_db() try: return db.query(models.InterLabCollaboration).order_by( models.InterLabCollaboration.id.desc() ).all() finally: if self._db is None: db.close()