"""Prospective E08 multi-repository live-agent experiment. Component harness cells reuse the audited E07 engine with generalized inputs. A001 is a three-stage Agentless-style controlled adaptation; A002 uses the same engine with a controlled SWE-agent-style tool interface. """ from __future__ import annotations from dataclasses import asdict, replace from hashlib import sha256 import json from pathlib import Path import resource import time from typing import Any, Sequence from .live_agent_experiment import ( AgentWorkspace, TaskRetrieval, _build_task_retrieval, _failure_validation, _usage_totals, run_live_agent_cell, ) from .lm_studio import LMStudioClient, LMStudioTransportError from .lm_studio_embeddings import LMStudioEmbeddingClient from .lm_studio_management import LMStudioResidencyManager, LMStudioServer from .pilot import research_code_revision, retrieval_metrics from .repair_experiment import ( PatchOutputError, extract_unified_diff, isolated_git_tree, validate_generated_patch, ) from .repository import GitSnapshot from .retrieval import SQLiteEmbeddingCache from .specs import ( AgentSystemSpec, EmbeddingSpec, ExperimentSpec, ModelSpec, RepositorySpec, TaskSpec, load_agent_systems, load_embeddings, load_experiments, load_harnesses, load_models, load_repositories, load_task_split, load_tasks, ) from .telemetry import EventWriter, RunIdentity, load_completed_or_archive_incomplete from .tokenization import QwenTokenCounter AGENTLESS_INDEX_TOKENS = 42_000 AGENTLESS_SOURCE_TOKENS = 48_000 TOKENIZER_PATHS = { "M002": Path.home() / ".lmstudio/models/lmstudio-community/Qwen3.6-35B-A3B-MLX-4bit/tokenizer.json", "M003": Path.home() / ".lmstudio/models/mlx-community/gpt-oss-20b-MXFP4-Q8/tokenizer.json", "M004": Path.home() / ".lmstudio/models/lmstudio-community/Qwen3-Coder-30B-A3B-Instruct-MLX-4bit/tokenizer.json", } class Study2ExperimentError(RuntimeError): """Raised when E08 cannot preserve its preregistered protocol.""" class _RuntimeLease: """Guarantee exclusive-model cleanup on success and on infrastructure errors.""" def __init__( self, server: LMStudioServer, residency: LMStudioResidencyManager, stop_server: bool, ): self.server = server self.residency = residency self.stop_server = stop_server self.server_state: dict[str, Any] | None = None self.final_transition: dict[str, Any] | None = None self.stop_state: dict[str, Any] | None = None self.cleanup_errors: list[str] = [] def __enter__(self) -> dict[str, Any]: self.server_state = self.server.ensure_running() return self.server_state def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: try: self.final_transition = self.residency.unload_all().to_dict() except Exception as cleanup_error: self.cleanup_errors.append(f"unload_all: {cleanup_error}") if self.stop_server: try: status = self.server.status() if status["running"]: self.stop_state = self.server.stop() else: self.stop_state = { "action": "already_stopped", "status": status, } except Exception as cleanup_error: self.cleanup_errors.append(f"server_stop: {cleanup_error}") if exc_type is None and self.cleanup_errors: raise Study2ExperimentError( "runtime cleanup failed: " + "; ".join(self.cleanup_errors) ) def tokenizer_for(model: ModelSpec) -> QwenTokenCounter: try: path = TOKENIZER_PATHS[model.model_id] except KeyError as exc: raise Study2ExperimentError(f"no frozen tokenizer for {model.model_id}") from exc if not path.exists(): raise Study2ExperimentError(f"frozen tokenizer is unavailable: {path}") return QwenTokenCounter(path) def _identity( experiment: ExperimentSpec, task: TaskSpec, system: AgentSystemSpec, model: ModelSpec, revision: str, seed: int, repetition: int, ) -> RunIdentity: return RunIdentity( experiment_id=experiment.experiment_id, task_id=task.task_id, harness_id=system.system_id, harness_hash=system.config_hash, model_id=model.model_id, model_key=model.expected_inference_key, model_config_hash=model.config_hash, context_budget=experiment.context_budgets[0], seed=seed, repetition=repetition, repository_sha=task.base_commit, code_revision=revision, ) def _assistant_content(response: dict[str, Any]) -> str: try: content = response["choices"][0]["message"]["content"] except (KeyError, IndexError, TypeError) as exc: raise Study2ExperimentError("agentless stage returned no assistant message") from exc if not isinstance(content, str): raise Study2ExperimentError("agentless stage assistant content is not text") return content def _json_object(content: str) -> dict[str, Any]: stripped = content.strip() fence = chr(96) * 3 if stripped.startswith(fence): stripped = stripped.strip(chr(96)) if "\n" in stripped: stripped = stripped.split("\n", 1)[1] start, end = stripped.find("{"), stripped.rfind("}") if start < 0 or end < start: raise ValueError("response contains no JSON object") value = json.loads(stripped[start : end + 1]) if not isinstance(value, dict): raise ValueError("response JSON is not an object") return value def _select_files(content: str, tracked_paths: Sequence[str]) -> tuple[str, ...]: allowed = set(tracked_paths) selected: list[str] = [] try: value = _json_object(content) raw = value.get("files", []) if isinstance(raw, list): selected.extend(str(item) for item in raw if str(item) in allowed) except (ValueError, json.JSONDecodeError): pass if not selected: selected.extend(path for path in tracked_paths if path in content) return tuple(dict.fromkeys(selected))[:10] def _pack_blocks( tokenizer: QwenTokenCounter, blocks: Sequence[tuple[str, str]], budget: int, ) -> tuple[str, tuple[str, ...], int]: selected: list[str] = [] paths: list[str] = [] used = 0 for path, block in blocks: count = tokenizer.count(block) if count > budget and not selected: block = block[: budget * 3] count = tokenizer.count(block) if used + count > budget: continue selected.append(block) paths.append(path) used += count return "\n".join(selected), tuple(paths), used def _repository_index( retrieval: TaskRetrieval, tracked_paths: Sequence[str], tokenizer: QwenTokenCounter, ) -> tuple[str, int]: blocks: list[tuple[str, str]] = [] for path in sorted(tracked_paths): symbols = retrieval.graph.by_path.get(path, ()) names = ", ".join(item.name for item in symbols[:30]) line = path if not names else f"{path} :: {names}" blocks.append((path, line + "\n")) text, _, tokens = _pack_blocks(tokenizer, blocks, AGENTLESS_INDEX_TOKENS) return text, tokens def _source_context( tree: Path, paths: Sequence[str], tokenizer: QwenTokenCounter, ) -> tuple[str, tuple[str, ...], int]: blocks: list[tuple[str, str]] = [] for path in paths: target = tree / path if target.is_file(): blocks.append( ( path, f"\n--- FILE: {path} ---\n" + target.read_text(encoding="utf-8", errors="replace") + "\n", ) ) return _pack_blocks(tokenizer, blocks, AGENTLESS_SOURCE_TOKENS) def run_agentless_cell( root: Path, repository: Path, experiment: ExperimentSpec, task: TaskSpec, system: AgentSystemSpec, model: ModelSpec, embedding: EmbeddingSpec, retrieval: TaskRetrieval, residency: LMStudioResidencyManager, server: LMStudioServer, tokenizer: QwenTokenCounter, revision: str, seed: int = 0, repetition: int = 0, ) -> dict[str, Any]: if system.system_id != "A001": raise Study2ExperimentError("agentless cell requires A001") identity = _identity(experiment, task, system, model, revision, seed, repetition) completed = load_completed_or_archive_incomplete(root / "results", identity) if completed is not None: return completed initial_transition = residency.ensure_exclusive( model.expected_inference_key, model.context_length ) client = LMStudioClient(model, timeout_seconds=experiment.timeout_seconds) _, resolved = client.resolve() suffixes = {"go": (".go",), "python": (".py",)} tracked_paths = GitSnapshot(repository).tracked_paths( task.base_commit, suffixes[task.language] ) resolved_model = { "agent_model": asdict(model), "embedding_model": asdict(embedding), "agent_runtime": resolved.to_dict(), "tokenizer_path": str(tokenizer.path), "tokenizer_sha256": tokenizer.sha256, "initial_residency_transition": initial_transition.to_dict(), } with isolated_git_tree(repository, task.base_commit, task.repository_url) as tree, EventWriter( root / "results", identity, asdict(system), resolved_model ) as writer: workspace = AgentWorkspace(tree, tracked_paths, task, system.max_test_runs) transitions: list[dict[str, Any]] = [initial_transition.to_dict()] responses: list[dict[str, Any]] = [] stage_records: list[dict[str, Any]] = [] protocol_violations: list[str] = [] model_elapsed = 0.0 selected_paths: tuple[str, ...] = () patch = "" finished_reason = "completed_three_stages" cell_started = time.monotonic() def call_stage(name: str, prompt: str, max_tokens: int) -> str: nonlocal model_elapsed transition = residency.ensure_exclusive( model.expected_inference_key, model.context_length ) transitions.append(transition.to_dict()) writer.emit( "resource_sample", {"kind": "model_residency_transition", **transition.to_dict()}, ) messages = [ { "role": "system", "content": ( "You are one stage in a controlled Agentless-style coding " "pipeline. Follow the requested output schema exactly. Hidden " "tests and gold changes are unavailable." ), }, {"role": "user", "content": prompt}, ] started = time.monotonic() try: _, active = client.resolve() response = client.chat_completions( active.inference_key, messages, max_tokens=max_tokens, seed=seed, ) except LMStudioTransportError as first_error: recovery = server.ensure_running() writer.emit( "resource_sample", {"kind": "server_recovery", "error": str(first_error), "recovery": recovery}, ) transition = residency.ensure_exclusive( model.expected_inference_key, model.context_length ) transitions.append(transition.to_dict()) _, active = client.resolve() response = client.chat_completions( active.inference_key, messages, max_tokens=max_tokens, seed=seed, ) elapsed = time.monotonic() - started model_elapsed += elapsed responses.append(response) content = _assistant_content(response) record = { "stage": name, "prompt": prompt, "response": content, "elapsed_seconds": elapsed, "usage": response.get("usage", {}), "prompt_tokens_local": tokenizer.count(prompt), } stage_records.append(record) turn = len(stage_records) writer.write_artifact( f"model_response_{turn:02d}.json", json.dumps(response, indent=2, sort_keys=True) + "\n", ) writer.emit( "model_call", { "turn": turn, "stage": name, "elapsed_seconds": elapsed, "usage": response.get("usage", {}), "input_conversation_tokens": record["prompt_tokens_local"], }, ) return content writer.emit( "run_started", { "confirmatory": True, "blinded": True, "task_config_hash": task.config_hash, "stages": ["file_localization", "line_localization", "repair"], "budgets": { "model_calls": system.model_calls, "tool_calls": system.max_tool_calls, "test_runs": system.max_test_runs, "timeout_seconds": experiment.timeout_seconds, }, }, ) index_tokens = 0 source_tokens = 0 try: index, index_tokens = _repository_index(retrieval, tracked_paths, tokenizer) stage1 = call_stage( "file_localization", f"""ISSUE: {task.statement} REPOSITORY FILE/SYMBOL INDEX: {index} Select at most 10 likely production files. Return only JSON: {{"files": ["path/from/index"], "rationale": "brief"}}""", max_tokens=2_048, ) selected_paths = _select_files(stage1, tracked_paths) if not selected_paths: protocol_violations.append("A001 file localization selected no valid paths") finished_reason = "file_localization_failure" raise ValueError("no valid localized files") source, included_paths, source_tokens = _source_context( tree, selected_paths, tokenizer ) selected_paths = included_paths stage2 = call_stage( "line_localization", f"""ISSUE: {task.statement} CANDIDATE SOURCE: {source} Identify the exact functions or line regions that require change. Return only JSON: {{"locations": [{{"path": "...", "symbol_or_lines": "...", "reason": "..."}}]}}""", max_tokens=2_048, ) stage3 = call_stage( "repair", f"""ISSUE: {task.statement} LOCALIZATION: {stage2} CANDIDATE SOURCE: {source} Return only a standard unified diff. Modify production files only, do not add tests, and make the smallest correct change.""", max_tokens=model.max_tokens, ) patch = extract_unified_diff( {"choices": [{"message": {"content": stage3}}]} ) apply_result = workspace.apply_patch(patch) writer.emit("edit", apply_result) if not apply_result["accepted"]: finished_reason = "patch_apply_failure" elif system.max_test_runs: command = task.pass_to_pass_tests[0] public_test = workspace.run_tests(command) writer.emit("test_run", public_test) except (ValueError, PatchOutputError, json.JSONDecodeError) as exc: protocol_violations.append(f"A001 protocol: {exc}") if not patch: finished_reason = "protocol_failure" final_patch = workspace.final_patch() if final_patch: validation = validate_generated_patch( root, repository, task, final_patch, preserve_git_metadata=True, ) else: validation = _failure_validation("empty_patch") edited_paths = tuple(sorted(workspace.edited_paths)) usage = _usage_totals(responses) elapsed = time.monotonic() - cell_started final = { "run_id": identity.run_id, "experiment_id": experiment.experiment_id, "task_id": task.task_id, "harness_id": system.system_id, "resolved_at_1": validation["resolved_at_1"], "failure_stage": validation["failure_stage"], "patch_applied": bool( validation.get("model_patch_apply") and validation["model_patch_apply"].get("returncode") == 0 ), "fail_to_pass": validation["fail_to_pass"], "pass_to_pass": validation["pass_to_pass"], "modified_files": edited_paths, "localization_metrics": retrieval_metrics(edited_paths, task.gold_files), "search_localization_metrics": retrieval_metrics( selected_paths, task.gold_files ), "read_localization_metrics": retrieval_metrics( selected_paths, task.gold_files ), "finished_reason": finished_reason, "finish_summary": "", "protocol_violations": protocol_violations, "model_calls": len(responses), "tool_calls": 0, "tool_counts": {}, "test_runs": len(workspace.test_runs), "usage": usage, "elapsed_seconds": elapsed, "model_elapsed_seconds": model_elapsed, "model_switch_count": sum( not item.get("reused") for item in transitions ), "model_switch_seconds": sum( float(item["elapsed_seconds"]) for item in transitions if not item.get("reused") ), "peak_process_rss_platform_units": resource.getrusage( resource.RUSAGE_SELF ).ru_maxrss, "dense_index_stats": retrieval.dense_index_stats, "agentless_index_tokens": index_tokens, "agentless_source_tokens": source_tokens, "trajectory_sha256": sha256( json.dumps(stage_records, sort_keys=True).encode() ).hexdigest(), "patch_sha256": sha256(final_patch.encode()).hexdigest() if final_patch else None, "residency_transitions": transitions, "test_results": validation["tests"], } writer.write_artifact( "messages.json", json.dumps(stage_records, indent=2, sort_keys=True) + "\n" ) writer.write_artifact("model.patch", final_patch) writer.write_artifact( "validation.json", json.dumps(validation, indent=2, sort_keys=True) + "\n" ) writer.write_artifact( "final_metrics.json", json.dumps(final, indent=2, sort_keys=True) + "\n" ) writer.emit( "run_finished", { "status": "completed", "resolved_at_1": validation["resolved_at_1"], "failure_stage": validation["failure_stage"], "finished_reason": finished_reason, }, ) return final def _repository_for_task( repositories: dict[str, RepositorySpec], task: TaskSpec, ) -> RepositorySpec: matches = [ repository for repository in repositories.values() if repository.repository_url == task.repository_url ] if len(matches) != 1: raise Study2ExperimentError( f"{task.task_id} has no unique repository registry entry" ) return matches[0] def _write_progress( root: Path, experiment: ExperimentSpec, revision: str, rows: Sequence[dict[str, Any]], task_summaries: Sequence[dict[str, Any]], ) -> Path: path = root / "results" / "reports" / f"{experiment.experiment_id}_progress.json" path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps( { "schema_version": 1, "experiment_id": experiment.experiment_id, "code_revision": revision, "completed_cells": len(rows), "resolved_cells": sum(bool(item["resolved_at_1"]) for item in rows), "task_summaries": task_summaries, "rows": rows, }, indent=2, sort_keys=True, ) + "\n", encoding="utf-8", ) return path def run_study2_experiment( root: Path, experiment_id: str = "E08", task_filter: set[str] | None = None, treatment_filter: set[str] | None = None, model_filter: set[str] | None = None, stop_server_when_complete: bool = True, ) -> dict[str, Any]: revision = research_code_revision(root) experiment = load_experiments(root)[experiment_id] if experiment.mode != "study2_live_agent": raise Study2ExperimentError("Study 2 runner requires mode=study2_live_agent") harness_catalog = load_harnesses(root) system_catalog = load_agent_systems(root) model_catalog = load_models(root) repository_catalog = load_repositories(root) task_catalog = load_tasks(root) embedding = load_embeddings(root)[experiment.embedding_id] split = load_task_split( root / "tasks" / "splits" / f"{experiment.task_split}.txt" ) tasks = [ task_catalog[item] for item in split if task_filter is None or item in task_filter ] harnesses = [ harness_catalog[item] for item in experiment.harness_ids if treatment_filter is None or item in treatment_filter ] systems = [ system_catalog[item] for item in experiment.agent_system_ids if treatment_filter is None or item in treatment_filter ] models = [ model_catalog[item] for item in experiment.model_ids if model_filter is None or item in model_filter ] if not tasks or not models or not (harnesses or systems): raise Study2ExperimentError("Study 2 filters selected an empty execution block") if any(task.validation_status != "end_to_end_ready" for task in tasks): raise Study2ExperimentError("Study 2 includes a task without hidden-test validation") server = LMStudioServer(port=1234) residency = LMStudioResidencyManager( models[0].base_url, models[0].api_token_env, timeout_seconds=experiment.timeout_seconds, ) embedding_client = LMStudioEmbeddingClient( embedding, timeout_seconds=experiment.timeout_seconds ) cache_path = root / "indexes" / "embeddings" / f"{embedding.config_hash}.sqlite3" treatments: list[tuple[str, Any]] = [ *((item.harness_id, item) for item in harnesses), *((item.system_id, item) for item in systems), ] rows: list[dict[str, Any]] = [] task_summaries: list[dict[str, Any]] = [] runtime = _RuntimeLease(server, residency, stop_server_when_complete) with runtime as server_state, SQLiteEmbeddingCache(cache_path, embedding) as cache: for task_index, task in enumerate(tasks): repository_spec = _repository_for_task(repository_catalog, task) repository = (root / repository_spec.local_path).resolve() snapshot = GitSnapshot(repository) snapshot.verify_commit(task.base_commit) index_transition = residency.ensure_exclusive( embedding.model_key, embedding.loaded_context_length ) embedding_client.resolve() index_started = time.monotonic() retrieval = _build_task_retrieval( snapshot, task, embedding, embedding_client, cache ) index_elapsed = time.monotonic() - index_started model_order = models if task_index % 2 == 0 else list(reversed(models)) offset = task_index % len(treatments) treatment_order = treatments[offset:] + treatments[:offset] task_rows: list[dict[str, Any]] = [] for model in model_order: tokenizer = tokenizer_for(model) for treatment_id, treatment in treatment_order: if treatment_id.startswith("H"): row = run_live_agent_cell( root, repository, experiment, task, treatment, model, embedding, retrieval, residency, server, tokenizer, revision, seed=experiment.seeds[0], repetition=0, preserve_git_metadata=True, ) elif treatment_id == "A001": row = run_agentless_cell( root, repository, experiment, task, treatment, model, embedding, retrieval, residency, server, tokenizer, revision, seed=experiment.seeds[0], repetition=0, ) elif treatment_id == "A002": row = run_live_agent_cell( root, repository, experiment, task, harness_catalog["H000"], model, embedding, retrieval, residency, server, tokenizer, revision, agent_system=treatment, seed=experiment.seeds[0], repetition=0, preserve_git_metadata=True, ) else: raise Study2ExperimentError( f"unsupported Study 2 treatment: {treatment_id}" ) rows.append(row) task_rows.append(row) _write_progress(root, experiment, revision, rows, task_summaries) summary = { "task_id": task.task_id, "repository_id": repository_spec.repository_id, "language": task.language, "model_order": [item.model_id for item in model_order], "treatment_order": [item[0] for item in treatment_order], "embedding_index_transition": index_transition.to_dict(), "index_elapsed_seconds": index_elapsed, "dense_index_stats": retrieval.dense_index_stats, "cells": len(task_rows), "resolved": sum(bool(item["resolved_at_1"]) for item in task_rows), } task_summaries.append(summary) _write_progress(root, experiment, revision, rows, task_summaries) report = { "experiment_id": experiment.experiment_id, "code_revision": revision, "server_lifecycle": server_state, "server_stop": runtime.stop_state, "run_count": len(rows), "resolved_count": sum(bool(item["resolved_at_1"]) for item in rows), "task_summaries": task_summaries, "final_residency_transition": runtime.final_transition, "cleanup_errors": runtime.cleanup_errors, "rows": rows, } report_dir = root / "results" / "reports" report_dir.mkdir(parents=True, exist_ok=True) report_path = report_dir / f"E08_{revision[:12]}_{int(time.time())}.json" report_path.write_text( json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) return {**report, "report_path": str(report_path)} def run_study2_reliability( root: Path, manifest_path: Path | None = None, stop_server_when_complete: bool = True, ) -> dict[str, Any]: """Run the 24 frozen non-oracle cells under the stochastic sensitivity profile.""" revision = research_code_revision(root) experiment = load_experiments(root)["E08"] path = manifest_path or root / "configs" / "reliability" / "E08_repeat_cells.json" value = json.loads(path.read_text(encoding="utf-8")) cells = value.get("cells", []) seeds = tuple(int(item) for item in value.get("seeds", [])) temperature = float(value.get("temperature", -1.0)) top_p = float(value.get("top_p", -1.0)) if len(cells) != 24 or seeds != (0, 1, 2): raise Study2ExperimentError( "reliability manifest must freeze 24 cells and seeds 0,1,2" ) if temperature != 0.2 or top_p != 1.0: raise Study2ExperimentError( "reliability manifest must freeze temperature=0.2 and top_p=1.0" ) harness_catalog = load_harnesses(root) system_catalog = load_agent_systems(root) model_catalog = load_models(root) repository_catalog = load_repositories(root) task_catalog = load_tasks(root) embedding = load_embeddings(root)[experiment.embedding_id] server = LMStudioServer(port=1234) residency = LMStudioResidencyManager( model_catalog["M002"].base_url, model_catalog["M002"].api_token_env, timeout_seconds=experiment.timeout_seconds, ) embedding_client = LMStudioEmbeddingClient( embedding, timeout_seconds=experiment.timeout_seconds ) cache_path = root / "indexes" / "embeddings" / f"{embedding.config_hash}.sqlite3" rows: list[dict[str, Any]] = [] grouped: dict[str, list[dict[str, str]]] = {} for raw in cells: if not isinstance(raw, dict): raise Study2ExperimentError("reliability cell must be an object") cell = {key: str(raw[key]) for key in ("task_id", "treatment_id", "model_id")} grouped.setdefault(cell["task_id"], []).append(cell) runtime = _RuntimeLease(server, residency, stop_server_when_complete) with runtime as server_state, SQLiteEmbeddingCache(cache_path, embedding) as cache: for task_id, task_cells in grouped.items(): task = task_catalog[task_id] repository_spec = _repository_for_task(repository_catalog, task) repository = (root / repository_spec.local_path).resolve() snapshot = GitSnapshot(repository) index_transition = residency.ensure_exclusive( embedding.model_key, embedding.loaded_context_length ) embedding_client.resolve() retrieval = _build_task_retrieval( snapshot, task, embedding, embedding_client, cache ) for cell in task_cells: treatment_id = cell["treatment_id"] base_model = model_catalog[cell["model_id"]] model = replace(base_model, temperature=temperature, top_p=top_p) tokenizer = tokenizer_for(model) for seed in seeds: if treatment_id.startswith("H"): row = run_live_agent_cell( root, repository, experiment, task, harness_catalog[treatment_id], model, embedding, retrieval, residency, server, tokenizer, revision, seed=seed, repetition=1, preserve_git_metadata=True, ) elif treatment_id == "A001": row = run_agentless_cell( root, repository, experiment, task, system_catalog[treatment_id], model, embedding, retrieval, residency, server, tokenizer, revision, seed=seed, repetition=1, ) elif treatment_id == "A002": row = run_live_agent_cell( root, repository, experiment, task, harness_catalog["H000"], model, embedding, retrieval, residency, server, tokenizer, revision, agent_system=system_catalog[treatment_id], seed=seed, repetition=1, preserve_git_metadata=True, ) else: raise Study2ExperimentError( f"unknown reliability treatment: {treatment_id}" ) rows.append(row) progress = root / "results" / "reports" / "E08_reliability_progress.json" progress.parent.mkdir(parents=True, exist_ok=True) progress.write_text( json.dumps( { "schema_version": 1, "code_revision": revision, "completed_cells": len(rows), "planned_cells": 72, "generation_profile": { "temperature": temperature, "top_p": top_p, "seeds": seeds, }, "rows": rows, "last_index_transition": index_transition.to_dict(), }, indent=2, sort_keys=True, ) + "\n", encoding="utf-8", ) report = { "experiment_id": "E08_reliability", "code_revision": revision, "server_lifecycle": server_state, "server_stop": runtime.stop_state, "run_count": len(rows), "resolved_count": sum(bool(item["resolved_at_1"]) for item in rows), "generation_profile": { "temperature": temperature, "top_p": top_p, "seeds": seeds, }, "final_residency_transition": runtime.final_transition, "cleanup_errors": runtime.cleanup_errors, "rows": rows, } report_path = ( root / "results" / "reports" / f"E08_reliability_{revision[:12]}_{int(time.time())}.json" ) report_path.write_text( json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) return {**report, "report_path": str(report_path)}