"""Controlled experiment execution and report publication for the dashboard.""" from __future__ import annotations import hashlib import json from datetime import datetime, timezone from dataclasses import asdict, dataclass, replace from pathlib import Path from time import perf_counter from types import CodeType, ModuleType from typing import Any, Callable, Protocol, Sequence, cast from src.research.data_v2 import prepare_research_data_v2 from src.research.experiment_recorder import ExperimentRecorder from src.research.experiment_summary import ( write_detailed_experiment_summary, write_statistics_artifact, ) from src.research.protocol_registry import ( OPERATIONAL_RUNNERS, PreregistrationError, protocol_by_id, protocol_catalog, validate_operational_protocol, ) from src.research.registry import ResearchRegistry from src.research_assistant.airr import AIRRPipeline from src.research_assistant.assistant import AnalysisBackend from .models import JSONValue class WorkflowValidationError(ValueError): """Raised when a workflow submission is not scientifically traceable.""" EPISTEMIC_LAYERS: dict[str, JSONValue] = { "ui_state": "operator/dashboard progress only; not a scientific result", "data": "DATA artifacts and deterministic statistics are authoritative run outputs", "evid": "not_created; requires semantic validation, clean freeze and human review", "interpretation": "post-hoc AI or human interpretation; never execution input", } class _ScientificRunLike(Protocol): """Typed read boundary used only for post-run execution validation.""" seed: int condition: str metrics: dict[str, Any] def write_experiment_summary( research_root: Path, experiment_id: str, ai_report: dict[str, object], ) -> str: """Write the canonical detailed, data-first experiment summary.""" return write_detailed_experiment_summary(research_root, experiment_id, ai_report) @dataclass(frozen=True, slots=True) class ExperimentWorkflow: """Validated, reproducible input for one controlled experiment run.""" experiment_id: str question_id: str hypothesis_id: str title: str conditions: str ticks: int notes: str protocol: str seeds: tuple[int, ...] class ExperimentWorkflowService: """Run a fixed controller action and publish its research artifacts. The service deliberately accepts a controller callback, never commands or model-generated code. This keeps language-model assistance outside the causal path of the experiment. """ def __init__( self, research_root: Path, ai_backend: AnalysisBackend | None = None, ) -> None: self._research_root = research_root self._ai_backend = ai_backend def catalog(self) -> dict[str, JSONValue]: """Return registry entries suitable for workflow selection.""" registry = ResearchRegistry(self._research_root / "registry").load_all() return { "questions": cast( JSONValue, [ {"id": question.id, "label": question.question} for question in registry.questions.values() ], ), "hypotheses": cast( JSONValue, [ { "id": hypothesis.id, "question_id": hypothesis.research_question, "label": hypothesis.hypothesis, } for hypothesis in registry.hypotheses.values() ], ), "protocols": cast( JSONValue, [ { "id": "science_suite_v1", "label": "Science Suite v1 (RQ-gesteuerter Runner + DATA + Manifest)", }, { "id": "science_all_v1", "label": "Science ALL v1 (alle Science-Suite-Protokolle)", }, { "id": "science_time_v1", "label": "Science TIME v1 (DATA + Manifest)", }, {"id": "science_5d_v1", "label": "Science 5D v1 (DATA + Manifest)"}, { "id": "learning_operator_v1", "label": "Operator Learning v1 (DATA + Manifest + Report)", }, { "id": "runtime_ticks_v1", "label": "Runtime-Ticks (Laufprotokoll)", }, { "id": "stdp_pair_timing_v1", "label": "STDP Pair-Timing v1 (registriert)", }, *protocol_catalog(self._research_root), ], ), "next_experiment_id": self._next_experiment_id(), } def run_batch( self, body: dict[str, object], *, run_ticks: Callable[[int], object] | None = None, before: Callable[[], dict[str, int]] | None = None, after: Callable[[], dict[str, int]] | None = None, ) -> dict[str, object]: """Run selected registered protocols and publish one aggregate report.""" protocol_ids_value = body.get("protocols") if not isinstance(protocol_ids_value, list) or not protocol_ids_value: raise WorkflowValidationError("protocols must contain at least one protocol id") protocol_ids = [item for item in protocol_ids_value if isinstance(item, str) and item] if not protocol_ids: raise WorkflowValidationError("protocols must contain valid protocol selections") catalog = { str(item["id"]): item for item in protocol_catalog(self._research_root) if isinstance(item.get("id"), str) } exploratory: dict[str, tuple[str, str]] = {} known_protocols: list[str] = [] for protocol_id in protocol_ids: if protocol_id.startswith("exploratory:"): parts = protocol_id.split(":", 2) if len(parts) != 3 or not parts[1] or not parts[2]: raise WorkflowValidationError(f"Invalid exploratory selection: {protocol_id}") exploratory[protocol_id] = (parts[1], parts[2]) else: known_protocols.append(protocol_id) unknown = [protocol_id for protocol_id in known_protocols if protocol_id not in catalog] if unknown: raise WorkflowValidationError(f"Unknown operational protocols: {', '.join(unknown)}") batch_id_value = body.get("batch_id") batch_id = str(batch_id_value).strip() if isinstance(batch_id_value, str) else "" if not batch_id: stamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") batch_id = f"EXP-BATCH-{stamp}" if not batch_id.startswith("EXP-"): raise WorkflowValidationError("batch_id must use the EXP-* convention") ticks = body.get("ticks", 1000) if not isinstance(ticks, int) or isinstance(ticks, bool) or ticks < 1: raise WorkflowValidationError("ticks must be a positive integer") seeds = body.get("seeds", "42-44") options_value = body.get("protocol_options", {}) protocol_options = options_value if isinstance(options_value, dict) else {} title_prefix = str(body.get("title_prefix") or "Experiment workflow").strip() conditions = str(body.get("conditions") or "Registered protocol conditions").strip() notes = str(body.get("notes") or "").strip() results: list[dict[str, object]] = [] for index, protocol_id in enumerate(protocol_ids, start=1): option_value = protocol_options.get(protocol_id) option = option_value if isinstance(option_value, dict) else {} child_ticks = option.get("ticks", ticks) child_seeds = option.get("seeds", seeds) if not isinstance(child_ticks, int) or isinstance(child_ticks, bool) or child_ticks < 1: raise WorkflowValidationError(f"Invalid ticks for {protocol_id}") if protocol_id in exploratory: question_id, hypothesis_id = exploratory[protocol_id] protocol = "runtime_ticks_v1" title = f"{title_prefix}: exploratory {question_id}" else: contract = catalog[protocol_id] question_id = str(contract["research_question"]) hypothesis_id = str(contract["hypothesis"]) protocol = protocol_id title = f"{title_prefix}: {protocol_id}" experiment_id = f"{batch_id}-{index:02d}" child = { "experiment_id": experiment_id, "question_id": question_id, "hypothesis_id": hypothesis_id, "title": title, "conditions": conditions, "ticks": child_ticks, "seeds": child_seeds, "notes": notes, "protocol": protocol, "exploratory": protocol_id in exploratory, } try: if protocol_id in exploratory: if run_ticks is None or before is None or after is None: raise WorkflowValidationError( "Exploratory batch runs require an attached runtime controller." ) result = self.run(child, run_ticks, before(), after) summary = write_experiment_summary( self._research_root, experiment_id, {"status": "unavailable", "reason": "Exploratory runtime run; AI post-hoc analysis not required.", "scientific_evidence": False}, ) result["summary"] = summary result["research_run_mode"] = "EXPLORATORY" result["test_run"] = True else: result = self.run_science(child) results.append({"protocol": protocol_id, "ticks": child_ticks, "seeds": child_seeds, "status": "completed", **result}) except Exception as exc: results.append({ "protocol": protocol_id, "ticks": child_ticks, "seeds": child_seeds, "experiment_id": experiment_id, "status": "failed", "error": f"{type(exc).__name__}: {exc}", }) workflow_root = self._research_root / "workflows" workflow_root.mkdir(parents=True, exist_ok=True) report_path = workflow_root / f"{batch_id}.json" completed = sum(item["status"] == "completed" for item in results) failed = len(results) - completed report = { "workflow_id": batch_id, "created_at": datetime.now(timezone.utc).isoformat(), "protocols": protocol_ids, "requested_ticks": ticks, "seeds": seeds, "completed": completed, "failed": failed, "results": results, } report_path.write_text(json.dumps(report, indent=2, ensure_ascii=True, default=str) + "\n", encoding="utf-8") markdown_path = report_path.with_suffix(".md") markdown_path.write_text( "\n".join([ f"# {batch_id}: Experiment workflow", "", f"- Requested ticks: `{ticks}`", f"- Seeds: `{seeds}`", f"- Completed: `{completed}`", f"- Failed: `{failed}`", "", "## Protocol results", *[f"- `{item['protocol']}`: **{item['status']}**" + (f" — {item['error']}" if item.get("error") else "") for item in results], "", "The workflow report is an aggregate index; each completed protocol keeps its own manifest, raw data, statistics and report.", ]) + "\n", encoding="utf-8", ) return { "workflow_id": batch_id, "report": f"workflows/{batch_id}.json", "report_markdown": f"workflows/{batch_id}.md", "completed": completed, "failed": failed, "results": results, } def run_science( self, body: dict[str, object], *, seeds: tuple[int, ...] | None = None ) -> dict[str, object]: """Execute the RQ-compatible science runner and persist complete artifacts.""" science_body = dict(body) science_body.setdefault("protocol", "science_suite_v1") workflow = self._validate(science_body) runner_name = self._science_runner(science_body, workflow) effective_seeds = seeds if seeds is not None else workflow.seeds workflow = replace(workflow, seeds=effective_seeds) output_dir = self._research_root / "experiments" / workflow.experiment_id if (output_dir / "manifest.json").exists(): raise WorkflowValidationError( f"Experiment '{workflow.experiment_id}' already exists." ) from src.research import experiment_suite config_path = ( self._research_root.parent / "configs" / "learning_experiment.yaml" ) if not config_path.exists(): config_path = Path("configs/learning_experiment.yaml") config_path = config_path.resolve() config = _load_yaml(config_path) config_digest = _sha256_file(config_path) started = perf_counter() if hasattr(experiment_suite, runner_name): runner_module: ModuleType = experiment_suite else: from src.research import followup_experiments runner_module = followup_experiments runner = getattr(runner_module, runner_name) runner_source_value = getattr(runner_module, "__file__", None) if not isinstance(runner_source_value, str): raise WorkflowValidationError( f"Cannot resolve source path for runner module {runner_module.__name__}." ) runner_source = runner_source_value runtime_runner_digest, source_runner_digest = ( _assert_loaded_callable_matches_source( runner, Path(runner_source).resolve(), runner_name ) ) summary_runtime_digest, summary_source_digest = ( _assert_loaded_callable_matches_source( write_detailed_experiment_summary, Path(write_detailed_experiment_summary.__code__.co_filename).resolve(), "write_detailed_experiment_summary", ) ) tick_aware_runners = { "run_all", "run_ping", "run_ping_v2", "run_5d", "run_time", "run_temporal", "run_recurrence_map", "run_replication", "run_5d_matched", "run_regulation_recovery", "run_temporal_order", "run_performance_profile", "run_recurrence_scale", "run_sustained_stability", } if runner_name in tick_aware_runners: runs = runner(config, seeds=effective_seeds, ticks=workflow.ticks) else: runs = runner(config, seeds=effective_seeds) runs = [replace(run, experiment_id=workflow.experiment_id) for run in runs] duration = perf_counter() - started tick_validation = self._validate_tick_execution( runner_name, workflow.ticks, effective_seeds, cast(Sequence[_ScientificRunLike], runs), ) recorder = ExperimentRecorder(workflow.experiment_id, output_dir=output_dir) data_path = output_dir / "DATA" / "runs.json" data_path.parent.mkdir(parents=True, exist_ok=True) serialized_runs = [asdict(run) for run in runs] # Compute statistics from the complete in-memory observations first. Large # per-tick traces are then moved to compressed sidecars so runs.json remains # reviewable without discarding raw observations. statistics_path = write_statistics_artifact(output_dir, serialized_runs) statistics_payload = json.loads(statistics_path.read_text(encoding="utf-8")) data_v2 = prepare_research_data_v2( output_dir, serialized_runs, statistics_payload ) trace_paths = list(data_v2.raw_paths) data_path.write_text( json.dumps(serialized_runs, indent=2, sort_keys=True, default=list) + "\n", encoding="utf-8", ) recorder.record_research_links([workflow.question_id], [workflow.hypothesis_id]) operational_protocol = protocol_by_id(self._research_root, workflow.protocol) if operational_protocol is not None: preregistration = validate_operational_protocol( self._research_root, question_id=workflow.question_id, hypothesis_id=workflow.hypothesis_id, protocol_id=workflow.protocol, seed_count=len(effective_seeds), ) recorder.record_research_run_mode( str(preregistration.get("mode", "EXPLORATORY")).upper() ) recorder.record_config(str(config_path), config_digest) recorder.record_simulation_params( seed=effective_seeds[0], ticks=workflow.ticks, seeds=list(effective_seeds), protocol=workflow.protocol, ) recorder.record_artifact("data", "DATA/runs.json") recorder.record_artifact("data_index", "DATA/runs_index.json") recorder.record_artifact("ai_packet", "analysis/ai_packet.json") recorder.record_artifact("ai_packet_digest", "analysis/ai_packet_digest.json") recorder.record_artifact("statistics", "analysis/statistics.json") recorder.record_artifact("workflow", "workflow.json") recorder.record_artifact("report", "report.md") recorder.record_results( run_count=len(runs), protocol=workflow.protocol, runner=runner_name, ticks_requested=workflow.ticks, tick_contract=tick_validation, ) report_path = output_dir / "report.md" report_path.write_text( self._render_science_report( workflow, len(runs), duration, runner_name, tick_validation ), encoding="utf-8", ) workflow_path = output_dir / "workflow.json" workflow_path.write_text( json.dumps( { "experiment_id": workflow.experiment_id, "research_question": workflow.question_id, "hypothesis": workflow.hypothesis_id, "title": workflow.title, "conditions": workflow.conditions, "ticks": workflow.ticks, "ticks_contract": "minimum requested observation window", "seeds": list(effective_seeds), "protocol": workflow.protocol, "resolved_runner": runner_name, "tick_validation": tick_validation, "notes": workflow.notes, "execution": "registered experiment_suite runner", "assistant_policy": "AI is post-hoc interpretation only.", "epistemic_layers": EPISTEMIC_LAYERS, }, indent=2, ensure_ascii=True, ) + "\n", encoding="utf-8", ) recorder.record_provenance_digests( code_digest=_sha256_file(Path(experiment_suite.__file__)), config_digest=config_digest, prompt_digest=hashlib.sha256(b"NO_PROMPT").hexdigest(), data_digest=_sha256_files( [ data_path, data_v2.raw_index_path, data_v2.ai_packet_path, data_v2.ai_packet_digest_path, *trace_paths, ], output_dir, ), ) recorder.record_runtime(duration).mark_completed().save() ai_report = self._append_ai_report(workflow.experiment_id) manifest_path = output_dir / "manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) artifacts = manifest.setdefault("artifacts", {}) artifacts["statistics"] = str(statistics_path.relative_to(output_dir)).replace( "\\", "/" ) manifest["execution_contract"] = { "resolved_runner": runner_name, "ticks_requested": workflow.ticks, "tick_validation": tick_validation, "question_id": workflow.question_id, "hypothesis_id": workflow.hypothesis_id, "protocol": workflow.protocol, "runtime_runner_bytecode_sha256": runtime_runner_digest, "source_runner_bytecode_sha256": source_runner_digest, "runtime_summary_bytecode_sha256": summary_runtime_digest, "source_summary_bytecode_sha256": summary_source_digest, "source_runtime_consistency": "MATCH", } manifest["epistemic_layers"] = EPISTEMIC_LAYERS artifacts["raw_run_index"] = "DATA/runs_index.json" artifacts["current_run"] = "DATA/current_run.json" artifacts["ai_packet"] = "analysis/ai_packet.json" artifacts["ai_packet_digest"] = "analysis/ai_packet_digest.json" if operational_protocol is not None: artifacts["preregistration"] = str(operational_protocol["preregistration"]) if ai_report.get("status") == "generated": artifacts["ai_report_json"] = str(ai_report["json"]) artifacts["ai_report_markdown"] = str(ai_report["markdown"]) manifest_path.write_text( json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) summary_path = write_experiment_summary( self._research_root, workflow.experiment_id, ai_report ) return { "experiment_id": workflow.experiment_id, "manifest": f"experiments/{workflow.experiment_id}/manifest.json", "report": f"experiments/{workflow.experiment_id}/report.md", "workflow": f"experiments/{workflow.experiment_id}/workflow.json", "statistics": f"experiments/{workflow.experiment_id}/analysis/statistics.json", "ai_packet": f"experiments/{workflow.experiment_id}/analysis/ai_packet.json", "raw_run_index": f"experiments/{workflow.experiment_id}/DATA/runs_index.json", "data_id": f"DATA-{workflow.experiment_id}", "ai_report": ai_report, "summary": summary_path, "result": { "run_count": len(runs), "duration_seconds": duration, "runner": runner_name, "ticks_requested": workflow.ticks, "seeds_executed": list(effective_seeds), "tick_validation": tick_validation, }, "epistemic_layers": EPISTEMIC_LAYERS, } @staticmethod def _science_runner(body: dict[str, object], workflow: ExperimentWorkflow) -> str: """Resolve execution from protocol and registered research question. Generated experiment IDs are labels only. They must never silently select a scientific runner, because that previously allowed TEMP questions to be executed as PING recurrence experiments. """ protocol = str(body.get("protocol") or workflow.protocol) question_id = workflow.question_id operational_runner = OPERATIONAL_RUNNERS.get(protocol) if operational_runner is not None: return operational_runner if protocol == "science_all_v1": return "run_all" if protocol == "science_time_v1": if not question_id.startswith("RQ-TIME-"): raise WorkflowValidationError( "science_time_v1 requires a RQ-TIME-* research question." ) return "run_time" if protocol == "science_5d_v1": if not question_id.startswith("RQ-5D-"): raise WorkflowValidationError( "science_5d_v1 requires a RQ-5D-* research question." ) return "run_5d" if protocol != "science_suite_v1": raise WorkflowValidationError( f"Protocol '{protocol}' is not a Science Suite runner protocol." ) rq_runners = ( ("RQ-PING-", "run_ping"), ("RQ-TEMP-", "run_temporal"), ("RQ-TIME-", "run_time"), ("RQ-5D-", "run_5d"), ("RQ-STDP-", "run_stdp"), ("RQ-REG-", "run_regulation"), ) for prefix, runner_name in rq_runners: if question_id.startswith(prefix): return runner_name if question_id == "RQ-SNN-001": # Keep the UI workflow executable, but never relabel an impulse-response # PING run as primary evidence for long-term stability. The complete # suite is diagnostic only; _semantic_status deliberately keeps this RQ # at MISMATCH until a dedicated sustained-activity protocol exists. return "run_all" if question_id == "RQ-SNN-002": return "run_ping" if question_id == "RQ-SNN-005": return "run_learning_repeat" raise WorkflowValidationError( f"No science runner is registered for research question '{question_id}'. " "Select a compatible registered protocol instead of falling back to PING." ) @staticmethod def _validate_tick_execution( runner_name: str, requested_ticks: int, seeds: tuple[int, ...], runs: Sequence[_ScientificRunLike], ) -> dict[str, object]: """Verify that tick-aware runners really respected the requested window.""" exact_window_runners = { "run_ping", "run_ping_v2", "run_5d", "run_temporal", "run_recurrence_map", "run_replication", "run_5d_matched", "run_regulation_recovery", "run_temporal_order", "run_performance_profile", "run_recurrence_scale", } if runner_name in exact_window_runners: observed_ints: list[int] = [] for run in runs: value = run.metrics.get("ticks_executed") if ( isinstance(value, bool) or not isinstance(value, int) or value < requested_ticks ): raise WorkflowValidationError( f"Tick contract violated: runner {runner_name} did not execute at least {requested_ticks} ticks in every run." ) observed_ints.append(value) if not observed_ints: raise WorkflowValidationError( f"Tick contract violated: runner {runner_name} produced no runs." ) return { "status": "SATISFIED", "mode": "minimum_per_run", "requested_ticks": requested_ticks, "observed_min": min(observed_ints), "observed_max": max(observed_ints), } if runner_name == "run_sustained_stability": observed_ints = [ run.metrics.get("ticks_executed") for run in runs if isinstance(run.metrics.get("ticks_executed"), int) ] if len(observed_ints) != len(runs) or not observed_ints: raise WorkflowValidationError( "Sustained stability runner produced incomplete tick observations." ) if min(observed_ints) < requested_ticks: raise WorkflowValidationError( "Sustained stability runner did not execute the requested tick budget in every run." ) if len(runs) != len(seeds) * 2: raise WorkflowValidationError( "Sustained stability runner must produce control and treatment runs for every seed." ) return { "status": "SATISFIED", "mode": "minimum_per_run_with_control_and_treatment", "requested_ticks": requested_ticks, "observed_min": min(observed_ints), "observed_max": max(observed_ints), "run_count": len(runs), } if runner_name == "run_time": missing_seeds: list[int] = [] for seed in seeds: if not any( run.seed == seed and run.metrics.get("ticks") == requested_ticks for run in runs ): missing_seeds.append(seed) if missing_seeds: raise WorkflowValidationError( "TIME tick ladder does not contain the requested terminal tick count " f"for seeds: {missing_seeds}." ) return { "status": "SATISFIED", "mode": "terminal_tick_per_seed", "requested_ticks": requested_ticks, } if runner_name == "run_all": tick_groups = ("ping:", "temporal:", "5d:") relevant = [ run for run in runs if any(str(run.condition).startswith(prefix) for prefix in tick_groups) ] if not relevant or any( not isinstance(run.metrics.get("ticks_executed"), int) or run.metrics["ticks_executed"] < requested_ticks for run in relevant ): raise WorkflowValidationError( "Science ALL tick contract violated in a tick-aware subgroup." ) for seed in seeds: if not any( run.seed == seed and str(run.condition).startswith("time:") and run.metrics.get("ticks") == requested_ticks for run in runs ): raise WorkflowValidationError( f"Science ALL TIME subgroup lacks requested tick terminal for seed {seed}." ) return { "status": "SATISFIED", "mode": "mixed_protocol_tick_contract", "requested_ticks": requested_ticks, } return { "status": "NOT_APPLICABLE", "mode": "protocol_defined_internal_trials", "requested_ticks": requested_ticks, } def _append_ai_report(self, experiment_id: str) -> dict[str, object]: """Generate AIRR only after completion, or expose unavailable explicitly.""" if self._ai_backend is None: return {"status": "unavailable", "reason": "AI backend not configured"} try: report = AIRRPipeline(self._research_root).analyze( experiment_id, self._ai_backend ) except Exception as exc: return { "status": "failed", "error": type(exc).__name__, "message": str(exc), } return { "status": "generated", "report_id": report.report_id, "json": f"experiments/{experiment_id}/reports/{report.report_id}.json", "markdown": f"experiments/{experiment_id}/reports/{report.report_id}.md", "human_review": "PENDING", "scientific_evidence": False, } def run( self, body: dict[str, object], run_ticks: Callable[[int], object], before: dict[str, int], after: Callable[[], dict[str, int]], ) -> dict[str, object]: """Execute a bounded tick run and write manifest, plan, and report.""" workflow = self._validate(body) output_dir = self._research_root / "experiments" / workflow.experiment_id manifest_path = output_dir / "manifest.json" if manifest_path.exists(): raise WorkflowValidationError( f"Experiment '{workflow.experiment_id}' already exists." ) output_dir.mkdir(parents=True, exist_ok=False) relative_root = Path("experiments") / workflow.experiment_id plan_path = output_dir / "workflow.json" plan_path.write_text( json.dumps( { "title": workflow.title, "research_question": workflow.question_id, "hypothesis": workflow.hypothesis_id, "conditions": workflow.conditions, "ticks": workflow.ticks, "protocol": workflow.protocol, "seeds": list(workflow.seeds), "notes": workflow.notes, "execution": "controller.step", "assistant_policy": "No AI-generated input is executed or used as evidence.", "epistemic_layers": EPISTEMIC_LAYERS, }, indent=2, ensure_ascii=True, ) + "\n", encoding="utf-8", ) recorder = ExperimentRecorder(workflow.experiment_id, output_dir=output_dir) recorder.record_research_links( [workflow.question_id], [workflow.hypothesis_id] ).record_simulation_params( ticks=workflow.ticks, seed=workflow.seeds[0], seeds=list(workflow.seeds), protocol=workflow.protocol, ).record_artifact( "workflow", str(relative_root / plan_path.name).replace("\\", "/") ) data_path = output_dir / "DATA" / "runs.json" data_path.parent.mkdir(parents=True, exist_ok=True) started = perf_counter() try: run_ticks(workflow.ticks) except Exception as exc: recorder.record_runtime_error( tick=before["tick"], phase="controller.step", exception_type=type(exc).__name__, message=str(exc), fatal=True, ).mark_failed() recorder.record_runtime(perf_counter() - started).save() raise runtime = after() duration = perf_counter() - started observed_ticks = runtime["tick"] - before["tick"] if observed_ticks < workflow.ticks: recorder.record_runtime_error( tick=runtime["tick"], phase="tick_contract", exception_type="TickContractViolation", message=f"Requested {workflow.ticks}, observed {observed_ticks} ticks.", fatal=True, ).mark_failed() recorder.record_runtime(duration).save() raise WorkflowValidationError( f"Runtime tick contract violated: requested {workflow.ticks}, observed {observed_ticks}." ) recorder.record_results( passed=True, start=before, end=runtime, observed_ticks=observed_ticks, ticks_requested=workflow.ticks, tick_contract="SATISFIED", ).record_artifact("data", "DATA/runs.json").record_runtime(duration).mark_completed().save() data_path.write_text( json.dumps( [{ "experiment_id": workflow.experiment_id, "condition": "exploratory_runtime_ticks", "seed": workflow.seeds[0], "metrics": { "ticks_requested": workflow.ticks, "ticks_executed": observed_ticks, "start_tick": before["tick"], "end_tick": runtime["tick"], "neurons": runtime["neurons"], "synapses": runtime["synapses"], }, "runtime_error": None, }], indent=2, ensure_ascii=True, ) + "\n", encoding="utf-8", ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["epistemic_layers"] = EPISTEMIC_LAYERS manifest_path.write_text( json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) report_path = output_dir / "report.md" report_path.write_text( self._render_report(workflow, before, runtime, duration), encoding="utf-8" ) return { "experiment_id": workflow.experiment_id, "manifest": str(relative_root / manifest_path.name).replace("\\", "/"), "report": str(relative_root / report_path.name).replace("\\", "/"), "result": {"start": before, "end": runtime, "duration_seconds": duration}, "epistemic_layers": EPISTEMIC_LAYERS, } def _validate(self, body: dict[str, object]) -> ExperimentWorkflow: def required(name: str) -> str: value = body.get(name) if not isinstance(value, str) or not value.strip(): raise WorkflowValidationError(f"Missing '{name}'.") return value.strip() experiment_id_value = body.get("experiment_id", "") if not isinstance(experiment_id_value, str): raise WorkflowValidationError("Experiment ID must be text.") experiment_id = experiment_id_value.strip() or self._next_experiment_id() if not experiment_id.startswith("EXP-") or len(experiment_id) > 64: raise WorkflowValidationError( "Experiment ID must use the EXP-* convention." ) ticks = body.get("ticks") protocol = str(body.get("protocol") or "runtime_ticks_v1") tick_limit = 1_000_000 if protocol == "science_time_v1" else 100_000 if ( not isinstance(ticks, int) or isinstance(ticks, bool) or not 1 <= ticks <= tick_limit ): raise WorkflowValidationError( f"Ticks must be an integer between 1 and {tick_limit}." ) seeds = self._parse_seeds(body.get("seeds")) question_id = required("question_id") hypothesis_id = required("hypothesis_id") registry = ResearchRegistry(self._research_root / "registry").load_all() question = registry.questions.get(question_id) hypothesis = registry.hypotheses.get(hypothesis_id) if question is None: raise WorkflowValidationError(f"Unknown research question '{question_id}'.") exploratory_runtime = protocol == "runtime_ticks_v1" and body.get("exploratory") is True if exploratory_runtime and hypothesis_id == "EXPLORATORY-UNSPECIFIED": hypothesis = None elif hypothesis is None or hypothesis.research_question != question.id: raise WorkflowValidationError( "The selected hypothesis does not belong to the research question." ) if protocol_by_id(self._research_root, protocol) is not None: try: validate_operational_protocol( self._research_root, question_id=question_id, hypothesis_id=hypothesis_id, protocol_id=protocol, seed_count=len(seeds), ) except PreregistrationError as exc: raise WorkflowValidationError(str(exc)) from exc return ExperimentWorkflow( experiment_id=experiment_id, question_id=question_id, hypothesis_id=hypothesis_id, title=required("title"), conditions=required("conditions"), ticks=ticks, notes=str(body.get("notes", "")).strip(), protocol=protocol, seeds=seeds, ) @staticmethod def _parse_seeds(value: object) -> tuple[int, ...]: """Parse comma-separated seeds and compact ranges from the runner UI.""" if value is None or value == "": return (42, 43, 44) if isinstance(value, str): tokens: list[object] = [ token.strip() for token in value.split(",") if token.strip() ] elif isinstance(value, (list, tuple)): tokens = list(cast(list[object] | tuple[object, ...], value)) else: raise WorkflowValidationError("Seeds must be comma-separated integers.") parsed: list[int] = [] for token in tokens: if isinstance(token, bool): raise WorkflowValidationError("Seeds must be non-negative integers.") text = str(token).strip() if "-" in text and text.count("-") == 1: start_text, end_text = (part.strip() for part in text.split("-")) if start_text.isdigit() and end_text.isdigit(): start, end = int(start_text), int(end_text) if end < start or end - start > 63: raise WorkflowValidationError( "Seed ranges must contain 64 values or fewer." ) parsed.extend(range(start, end + 1)) continue if not text.isdigit(): raise WorkflowValidationError("Seeds must be non-negative integers.") parsed.append(int(text)) unique = tuple(dict.fromkeys(parsed)) if not unique or len(unique) > 64: raise WorkflowValidationError("Provide between 1 and 64 unique seeds.") return unique def _next_experiment_id(self) -> str: """Return the first available generated experiment identifier.""" experiments_dir = self._research_root / "experiments" for number in range(1, 10_000): experiment_id = f"EXP-GEN-{number:04d}" if not (experiments_dir / experiment_id).exists(): return experiment_id raise WorkflowValidationError("No generated experiment IDs are available.") @staticmethod def _render_science_report( workflow: ExperimentWorkflow, run_count: int, duration: float, runner_name: str, tick_validation: dict[str, object], ) -> str: return "\n".join( [ f"# {workflow.experiment_id}: {workflow.title}", "", "## Forschungszuordnung", f"Forschungsfrage: `{workflow.question_id}`", f"Hypothese: `{workflow.hypothesis_id}`", f"Aufgeloester Runner: `{runner_name}`", "", "## Protokoll und Einstellungen", f"Protokoll: `{workflow.protocol}`", f"Angeforderte Mindest-Ticks: `{workflow.ticks}`", f"Seeds: `{', '.join(str(seed) for seed in workflow.seeds)}`", f"Tick-Vertrag: `{json.dumps(tick_validation, ensure_ascii=False, sort_keys=True)}`", "", "## Bedingungen", workflow.conditions, f"Runs: {run_count}; Dauer: {duration:.6f} s", "", "## Daten und Statistik", "Kompakte Run-Projektion: `DATA/runs.json`", "Unveränderlicher Rohdatenindex: `DATA/runs_index.json`", "KI-Eingabepaket: `analysis/ai_packet.json` (hart begrenzt)", "Deterministische deskriptive Statistik: `analysis/statistics.json`", "Die Summary verbindet Rohdaten, Formeln, Einzelruns, Bedingungen, Reproduzierbarkeit und AIRR ohne KI-generierte Statistik.", "", "## Epistemische Ebenen", "UI-Zustand: Dashboard-Steuerung und Fortschritt; kein wissenschaftliches Ergebnis.", "DATA: Rohdaten, Run-Index und deterministische Statistik.", "EVID: nicht erzeugt; Clean Freeze, semantische Zuordnung und Human Review erforderlich.", "Interpretation: nachgelagerte KI-/Human-Interpretation; keine Ausfuehrungseingabe.", "", "## Evidenzstatus", "DATA, Manifest, Workflow und deterministische Statistik sind erzeugt. Wissenschaftliche EVID entsteht erst nach passender semantischer Zuordnung, Clean Freeze und Human Review.", "", "## Hinweise", workflow.notes or "Keine.", "", ] ) @staticmethod def _render_report( workflow: ExperimentWorkflow, before: dict[str, int], after: dict[str, int], duration: float, ) -> str: observed_ticks = after["tick"] - before["tick"] return "\n".join( [ f"# {workflow.experiment_id}: {workflow.title}", "", "## Forschungsfrage", workflow.question_id, "", "## Hypothese", workflow.hypothesis_id, "", "## Bedingungen", workflow.conditions, "", "## Ausfuehrung", f"Controller: `step({workflow.ticks})`", "Ausfuehrungsmodus: kontrollierter Runtime-Lauf", f"Dauer: {duration:.6f} s", "", "## Ergebnis", f"Tick: {before['tick']} -> {after['tick']}", f"Neuronen: {before['neurons']} -> {after['neurons']}", f"Synapsen: {before['synapses']} -> {after['synapses']}", f"Angeforderte Ticks: {workflow.ticks}", f"Beobachtete Ticks: {observed_ticks}", f"Tick-Vertrag: {'SATISFIED' if observed_ticks >= workflow.ticks else 'VIOLATED'}", "", "## Epistemische Ebenen", "UI-Zustand ist nur Dashboard-Steuerung und Fortschritt.", "DATA sind Manifest, Workflow und gemessene Laufwerte.", "EVID wurde nicht erzeugt.", "Interpretation bleibt nachgelagert und hat keine Ausfuehrungsautoritaet.", "", "## Reproduzierbarkeit", "Git-Commit, Laufzeitumgebung und Runtime-Parameter stehen im Manifest.", "", "## Evidenzstatus", "Keine EVID erzeugt. Fuer wissenschaftliche Evidenz sind ein sauberer Source-Freeze, ein registriertes Protokoll, kontrollierte unabhaengige Variablen und definierte Messgroessen erforderlich.", "", "## Hinweise", workflow.notes or "Keine.", "", "KI-Ausgaben sind weder Ausfuehrungseingabe noch Evidenz. Eine finale Antwort auf die Forschungsfrage bleibt einer menschlichen wissenschaftlichen Bewertung vorbehalten.", "", ] ) def _find_named_code(code: CodeType, name: str) -> CodeType | None: if code.co_name == name: return code for value in code.co_consts: if not isinstance(value, CodeType): continue found = _find_named_code(value, name) if found is not None: return found return None def _code_digest(code: CodeType) -> str: """Hash executable code semantics while ignoring path/debug metadata. Raw marshal bytes include interpreter- and compilation-specific metadata that can differ between an imported function and the same freshly compiled source. The source-freeze gate needs semantic identity instead: bytecode, exception tables, signature/flags, referenced names and recursively hashed constants. """ digest = hashlib.sha256() def update_value(value: object) -> None: if isinstance(value, CodeType): digest.update(b"code\0") update_code(value) return if isinstance(value, tuple): digest.update(b"tuple\0") tuple_items = cast(tuple[object, ...], value) for item in tuple_items: update_value(item) digest.update(b"\0") return if isinstance(value, frozenset): digest.update(b"frozenset\0") set_items = cast(frozenset[object], value) encoded_items = sorted( f"{type(item).__qualname__}:{item!r}".encode( "utf-8", errors="backslashreplace" ) for item in set_items ) for item in encoded_items: digest.update(item) digest.update(b"\0") return if isinstance(value, bytes): digest.update(b"bytes\0") digest.update(value) return digest.update(type(value).__qualname__.encode("utf-8")) digest.update(b":") digest.update(repr(value).encode("utf-8", errors="backslashreplace")) def update_code(item: CodeType) -> None: for number in ( item.co_argcount, item.co_posonlyargcount, item.co_kwonlyargcount, item.co_nlocals, item.co_stacksize, item.co_flags, ): digest.update(str(number).encode("ascii")) digest.update(b"\0") digest.update(item.co_code) digest.update(b"\0") digest.update(item.co_exceptiontable) digest.update(b"\0") for values in ( item.co_names, item.co_varnames, item.co_freevars, item.co_cellvars, ): update_value(tuple(values)) digest.update(b"\0") update_value(tuple(item.co_consts)) update_code(code) return digest.hexdigest() def _assert_loaded_callable_matches_source( function: Callable[..., object], source_path: Path, function_name: str ) -> tuple[str, str]: """Block scientific runs when a long-lived process still holds stale code.""" runtime_digest = _code_digest(function.__code__) compiled = compile( source_path.read_text(encoding="utf-8"), str(source_path), "exec" ) source_code = _find_named_code(compiled, function_name) if source_code is None: raise WorkflowValidationError( f"Cannot verify runtime/source consistency for {function_name}." ) source_digest = _code_digest(source_code) if runtime_digest != source_digest: raise WorkflowValidationError( f"Running process contains stale code for {function_name}. Restart the Brain-5D " "dashboard/runtime before creating a scientific experiment." ) return runtime_digest, source_digest def _sha256_files(paths: Sequence[Path], root: Path) -> str: """Hash a set of artifacts using experiment-relative names for portability.""" digest = hashlib.sha256() for path in sorted(paths, key=lambda item: item.relative_to(root).as_posix()): digest.update(path.relative_to(root).as_posix().encode("utf-8")) digest.update(b"\0") digest.update(path.read_bytes()) digest.update(b"\0") return digest.hexdigest() def _load_yaml(path: Path) -> dict[str, Any]: import yaml loaded = yaml.safe_load(path.read_text(encoding="utf-8")) if not isinstance(loaded, dict): raise WorkflowValidationError("Science configuration must be a mapping.") return cast(dict[str, Any], loaded) def _sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest()