"""Prospective E09 model-by-edit-interface compatibility experiment.""" 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, CONVERSATION_TOKEN_LIMIT, MAX_MODEL_CALLS, TEST_OUTPUT_CHARS, _assistant_message, _compact_conversation, _failure_validation, _build_task_retrieval, _parse_tool_arguments, _search_schema, _trim, _usage_totals, LiveToolHarness, TaskRetrieval, ) from .lm_studio_embeddings import LMStudioEmbeddingClient from .lm_studio import LMStudioClient, LMStudioError, LMStudioTransportError from .lm_studio_management import ( LMStudioResidencyManager, LMStudioServer, ResidencyTransition, ) from .pilot import research_code_revision, retrieval_metrics from .repair_experiment import ( PatchOutputError, isolated_git_tree, validate_generated_patch, ) from .repository import GitSnapshot from .retrieval import SQLiteEmbeddingCache from .specs import ( EmbeddingSpec, EditInterfaceSpec, ExperimentSpec, HarnessSpec, ModelSpec, RepositorySpec, TaskSpec, load_embeddings, load_edit_interfaces, load_experiments, load_harnesses, load_models, load_repositories, load_task_split, load_tasks, ) from .study2_experiment import _RuntimeLease, tokenizer_for from .telemetry import EventWriter, RunIdentity, load_completed_or_archive_incomplete class ProtocolExperimentError(RuntimeError): """Raised when E09 cannot preserve its frozen protocol.""" def protocol_system_prompt( language: str, interface: EditInterfaceSpec, retrieval_harness: HarnessSpec | None = None, ) -> str: label = {"go": "Go", "python": "Python"}.get(language, language) localization = ( "The exact candidate production file names are provided." if retrieval_harness is None or retrieval_harness.control == "oracle_file" else "The repository is larger than the context window; use the available search action or actions to locate relevant production files." ) navigation = "" if retrieval_harness is not None and retrieval_harness.control == "none": query = ( "Exactly one repository search action is permitted; use its returned paths for reads." if retrieval_harness.query_policy == "one_shot" else "Repeated focused repository searches are permitted for iterative reformulation." ) channel = ( "One fused search tool is available." if retrieval_harness.interface == "unified" else "Separate exact, lexical, syntax, dense, and graph search tools are available." ) navigation = f" {query} {channel} Search results use {retrieval_harness.packing}." return f"""You are a coding agent repairing one issue in a controlled {label} repository. {localization} Read source before editing, make the smallest correct production change, run an allowed public test when useful, and call finish.{navigation} Rules: - Hidden tests, gold code, gold symbols, line locations, and the gold patch are unavailable. - Never create, delete, or edit test files. - Exactly one edit protocol is available: {interface.prompt_contract} - Do not call edit tools that are not exposed. - Tool errors are observations; correct the request rather than claiming success. - Normal assistant text never changes the worktree; only an accepted edit call is durable. - Stay within the model, tool, test, and context budgets. """ class ProtocolWorkspace(AgentWorkspace): """Add structured edit operations while retaining the shared final-diff evaluator.""" def _edit_path(self, path: str) -> tuple[str, Path]: from .live_agent_experiment import _safe_relative_path safe = _safe_relative_path(path) if safe not in self.tracked_paths: raise ValueError( f"edit may modify only tracked {self.language_name} source files: {safe}" ) if self.is_test_path(safe): raise ValueError(f"test edits are forbidden: {safe}") target = self.tree / safe if not target.is_file(): raise ValueError(f"edit target is not an existing file: {safe}") return safe, target def _remember(self, path: str, target: Path) -> None: if path not in self.original: self.original[path] = target.read_text(encoding="utf-8", errors="replace") def replace_text(self, path: str, old_text: str, new_text: str) -> dict[str, Any]: started = time.monotonic() if not isinstance(old_text, str) or not old_text: raise ValueError("old_text must be non-empty exact source text") if not isinstance(new_text, str): raise ValueError("new_text must be a string") safe, target = self._edit_path(path) current = target.read_text(encoding="utf-8", errors="replace") matches = current.count(old_text) if matches != 1: raise ValueError(f"old_text must occur exactly once; observed {matches} matches") self._remember(safe, target) target.write_text(current.replace(old_text, new_text, 1), encoding="utf-8") self.edited_paths.add(safe) return { "accepted": True, "paths": (safe,), "returncode": 0, "stdout": "", "stderr": "", "elapsed_seconds": time.monotonic() - started, "old_text_sha256": sha256(old_text.encode()).hexdigest(), "new_text_sha256": sha256(new_text.encode()).hexdigest(), } def write_file(self, path: str, content: str) -> dict[str, Any]: started = time.monotonic() if not isinstance(content, str) or not content: raise ValueError("content must be the non-empty complete replacement file") safe, target = self._edit_path(path) self._remember(safe, target) target.write_text(content, encoding="utf-8") self.edited_paths.add(safe) return { "accepted": True, "paths": (safe,), "returncode": 0, "stdout": "", "stderr": "", "elapsed_seconds": time.monotonic() - started, "content_sha256": sha256(content.encode()).hexdigest(), "content_characters": len(content), } class ProtocolToolHarness: def __init__(self, interface: EditInterfaceSpec, workspace: ProtocolWorkspace): self.interface = interface self.workspace = workspace self.read_paths: list[str] = [] self.search_paths: list[str] = [] self.finished = False self.finish_summary = "" def execute(self, name: str, arguments: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: if name == "read_file": result = self.workspace.read_file( str(arguments.get("path", "")), int(arguments.get("line_start", 1)), int(arguments["line_end"]) if arguments.get("line_end") is not None else None, ) self.read_paths.append(result["path"]) return result, result if name != self.interface.edit_tool and name in { "apply_patch", "replace_text", "write_file", }: raise ValueError( f"{name} is unavailable in {self.interface.interface_id}; " f"use {self.interface.edit_tool}" ) if name == "apply_patch" and self.interface.edit_tool == name: result = self.workspace.apply_patch(str(arguments.get("patch", ""))) return self._compact_edit(result), result if name == "replace_text" and self.interface.edit_tool == name: result = self.workspace.replace_text( str(arguments.get("path", "")), arguments.get("old_text", ""), arguments.get("new_text", ""), ) return self._compact_edit(result), result if name == "write_file" and self.interface.edit_tool == name: result = self.workspace.write_file( str(arguments.get("path", "")), arguments.get("content", "") ) return self._compact_edit(result), result if name == "run_tests": result = self.workspace.run_tests(str(arguments.get("command", ""))) compact = { **result, "stdout": _trim(str(result.get("stdout", "")), TEST_OUTPUT_CHARS), "stderr": _trim(str(result.get("stderr", "")), TEST_OUTPUT_CHARS), } return compact, result if name == "finish": self.finished = True self.finish_summary = str(arguments.get("summary", "")) result = {"accepted": True, "message": "agent finished"} return result, result raise ValueError(f"unknown tool: {name}") @staticmethod def _compact_edit(result: dict[str, Any]) -> dict[str, Any]: return { "accepted": bool(result.get("accepted")), "paths": result.get("paths", ()), "returncode": result.get("returncode"), "stdout": _trim(str(result.get("stdout", "")), 2_000), "stderr": _trim(str(result.get("stderr", "")), 4_000), "elapsed_seconds": result.get("elapsed_seconds", 0.0), } class RetrievalProtocolToolHarness(ProtocolToolHarness): """Combine one fixed retrieval stack with one E09-gated structured edit tool.""" def __init__( self, interface: EditInterfaceSpec, workspace: ProtocolWorkspace, harness: HarnessSpec, retrieval: TaskRetrieval, residency: LMStudioResidencyManager, model: ModelSpec, embedding: EmbeddingSpec, transition_callback: Any, ): super().__init__(interface, workspace) self.harness = harness self.live = LiveToolHarness( harness, retrieval, workspace, residency, model, embedding, transition_callback, ) def execute(self, name: str, arguments: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: if name == "search_code" or name.startswith("search_"): compact, raw = self.live.execute(name, arguments) self.search_paths[:] = self.live.search_paths return compact, raw result = super().execute(name, arguments) return result def protocol_tool_definitions( interface: EditInterfaceSpec, task: TaskSpec, retrieval_harness: HarnessSpec | None = None, ) -> list[dict[str, Any]]: language = {"go": "Go", "python": "Python"}.get(task.language, task.language) shared: list[dict[str, Any]] = [] if retrieval_harness is not None and retrieval_harness.control == "none": if retrieval_harness.interface == "unified": shared.append( _search_schema( "search_code", "Search the repository using this treatment's fixed retrieval stack.", ) ) else: shared.extend( [ _search_schema("search_exact", "Literal identifier, substring, and path-term search."), _search_schema("search_lexical", "BM25 code search with fuzzy path/name matching."), _search_schema("search_syntax", "Tree-sitter declaration and symbol search."), _search_schema("search_dense", "Code-embedding semantic search."), _search_schema("search_graph", "Fused retrieval followed by one static graph hop."), ] ) shared.extend([ { "type": "function", "function": { "name": "read_file", "description": f"Read at most 200 numbered lines from a listed tracked {language} file.", "parameters": { "type": "object", "properties": { "path": {"type": "string"}, "line_start": {"type": "integer", "minimum": 1}, "line_end": {"type": "integer", "minimum": 1}, }, "required": ["path"], "additionalProperties": False, }, }, } ]) if interface.edit_tool == "apply_patch": edit = { "type": "function", "function": { "name": "apply_patch", "description": ( "Apply raw standard unified-diff text with a/ and b/ file paths. " "Do not use *** Begin Patch or other wrapper markers." ), "parameters": { "type": "object", "properties": {"patch": {"type": "string"}}, "required": ["patch"], "additionalProperties": False, }, }, } elif interface.edit_tool == "replace_text": edit = { "type": "function", "function": { "name": "replace_text", "description": ( "Replace one exact, uniquely occurring source span in an existing production file. " "Copy old_text exactly from read_file." ), "parameters": { "type": "object", "properties": { "path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}, }, "required": ["path", "old_text", "new_text"], "additionalProperties": False, }, }, } else: edit = { "type": "function", "function": { "name": "write_file", "description": ( "Replace an existing production file with complete new contents. " "The content must be the entire file, not a fragment." ), "parameters": { "type": "object", "properties": { "path": {"type": "string"}, "content": {"type": "string"}, }, "required": ["path", "content"], "additionalProperties": False, }, }, } shared.append(edit) shared.extend( [ { "type": "function", "function": { "name": "run_tests", "description": f"Run one frozen public {language} test command.", "parameters": { "type": "object", "properties": { "command": { "type": "string", "enum": list( dict.fromkeys( (*task.fail_to_pass_tests, *task.pass_to_pass_tests) ) ), } }, "required": ["command"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "finish", "description": "Finish after the best accepted edit has been made.", "parameters": { "type": "object", "properties": {"summary": {"type": "string"}}, "required": ["summary"], "additionalProperties": False, }, }, }, ] ) return shared def _identity( experiment: ExperimentSpec, task: TaskSpec, interface: EditInterfaceSpec, model: ModelSpec, revision: str, seed: int = 0, context_budget: int | None = None, retrieval_harness: HarnessSpec | None = None, ) -> RunIdentity: treatment_id = ( interface.interface_id if retrieval_harness is None else f"{retrieval_harness.harness_id}__{interface.interface_id}" ) treatment_hash = ( interface.config_hash if retrieval_harness is None else sha256( f"{retrieval_harness.config_hash}:{interface.config_hash}".encode() ).hexdigest() ) return RunIdentity( experiment_id=experiment.experiment_id, task_id=task.task_id, harness_id=treatment_id, harness_hash=treatment_hash, model_id=model.model_id, model_key=model.expected_inference_key, model_config_hash=model.config_hash, context_budget=( experiment.context_budgets[0] if context_budget is None else int(context_budget) ), seed=seed, repetition=0, repository_sha=task.base_commit, code_revision=revision, ) def run_protocol_cell( root: Path, repository: Path, experiment: ExperimentSpec, task: TaskSpec, interface: EditInterfaceSpec, model: ModelSpec, residency: LMStudioResidencyManager, server: LMStudioServer, revision: str, retrieval_harness: HarnessSpec | None = None, retrieval: TaskRetrieval | None = None, embedding: EmbeddingSpec | None = None, seed: int = 0, context_budget: int | None = None, ) -> dict[str, Any]: if (retrieval_harness is None) != (retrieval is None): raise ProtocolExperimentError("retrieval harness and index must be supplied together") if retrieval_harness is not None and embedding is None: raise ProtocolExperimentError("retrieval protocol cell requires embedding metadata") identity = _identity( experiment, task, interface, model, revision, seed=seed, context_budget=context_budget, retrieval_harness=retrieval_harness, ) completed = load_completed_or_archive_incomplete(root / "results", identity) if completed is not None: return completed tokenizer = tokenizer_for(model) 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), "agent_runtime": resolved.to_dict(), "tokenizer_path": str(tokenizer.path), "tokenizer_sha256": tokenizer.sha256, "initial_residency_transition": initial_transition.to_dict(), } if embedding is not None: resolved_model["embedding_model"] = asdict(embedding) resolved_treatment: dict[str, Any] = asdict(interface) if retrieval_harness is not None: resolved_treatment = { "retrieval_harness": asdict(retrieval_harness), "edit_interface": asdict(interface), "combined_treatment_id": identity.harness_id, } with isolated_git_tree(repository, task.base_commit, task.repository_url) as tree, EventWriter( root / "results", identity, resolved_treatment, resolved_model ) as writer: workspace = ProtocolWorkspace(tree, tracked_paths, task, experiment.max_test_runs) transitions = [initial_transition.to_dict()] responses: list[dict[str, Any]] = [] protocol_violations: list[str] = [] tool_counts: dict[str, int] = {} edit_attempts = 0 edit_acceptances = 0 tool_call_count = 0 model_elapsed = 0.0 finished_reason = "model_turn_budget" cell_started = time.monotonic() def record_transition(transition: ResidencyTransition) -> None: value = transition.to_dict() transitions.append(value) writer.emit("resource_sample", {"kind": "model_residency_transition", **value}) if retrieval_harness is not None: assert retrieval is not None and embedding is not None tools: ProtocolToolHarness = RetrievalProtocolToolHarness( interface, workspace, retrieval_harness, retrieval, residency, model, embedding, record_transition, ) else: tools = ProtocolToolHarness(interface, workspace) definitions = protocol_tool_definitions(interface, task, retrieval_harness) task_prompt = f"ISSUE:\n{task.statement}" oracle_names = retrieval_harness is None or retrieval_harness.control == "oracle_file" if oracle_names: task_prompt += ( "\n\nORACLE PRODUCTION FILE LOCATIONS (names only):\n" + "\n".join(task.gold_files) ) messages: list[dict[str, Any]] = [ { "role": "system", "content": protocol_system_prompt( task.language, interface, retrieval_harness ), }, {"role": "user", "content": task_prompt}, ] writer.emit( "run_started", { "confirmatory": True, "blinded": True, "task_config_hash": task.config_hash, "oracle_file_names_only": oracle_names, "edit_interface": asdict(interface), "retrieval_harness": ( asdict(retrieval_harness) if retrieval_harness is not None else None ), "tool_names": [item["function"]["name"] for item in definitions], "budgets": { "model_calls": MAX_MODEL_CALLS, "tool_calls": experiment.max_tool_calls, "test_runs": experiment.max_test_runs, "timeout_seconds": experiment.timeout_seconds, }, }, ) for model_turn in range(1, MAX_MODEL_CALLS + 1): if time.monotonic() - cell_started > experiment.timeout_seconds: finished_reason = "cell_timeout" break before_tokens, after_tokens = _compact_conversation(messages, tokenizer) if before_tokens != after_tokens: writer.emit( "resource_sample", { "kind": "context_compaction", "before_tokens": before_tokens, "after_tokens": after_tokens, }, ) if after_tokens > CONVERSATION_TOKEN_LIMIT: finished_reason = "context_budget_exhausted" break transition = residency.ensure_exclusive( model.expected_inference_key, model.context_length ) record_transition(transition) call_started = time.monotonic() try: _, active = client.resolve() response = client.chat_completions( active.inference_key, messages, tools=definitions, max_tokens=model.max_tokens, seed=identity.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 ) record_transition(transition) _, active = client.resolve() response = client.chat_completions( active.inference_key, messages, tools=definitions, max_tokens=model.max_tokens, seed=identity.seed, ) elapsed = time.monotonic() - call_started model_elapsed += elapsed responses.append(response) writer.write_artifact( f"model_response_{model_turn:02d}.json", json.dumps(response, indent=2, sort_keys=True) + "\n", ) writer.emit( "model_call", { "turn": model_turn, "elapsed_seconds": elapsed, "usage": response.get("usage", {}), "finish_reason": response.get("choices", [{}])[0].get("finish_reason"), "input_conversation_tokens": after_tokens, }, ) assistant, calls = _assistant_message(response) messages.append(assistant) if not calls: finished_reason = "assistant_stop_without_finish" if assistant.get("content"): protocol_violations.append("assistant stopped without calling finish") break for call in calls: if tool_call_count >= experiment.max_tool_calls: finished_reason = "tool_budget_exhausted" break tool_call_count += 1 call_id = str(call.get("id") or f"tool-{tool_call_count}") parsed_name = str(call.get("function", {}).get("name", "invalid_tool")) is_assigned_edit = parsed_name == interface.edit_tool if is_assigned_edit: edit_attempts += 1 try: name, arguments = _parse_tool_arguments(call) compact_result, raw_result = tools.execute(name, arguments) is_error = False if name == interface.edit_tool and raw_result.get("accepted"): edit_acceptances += 1 except (ValueError, json.JSONDecodeError, PatchOutputError) as exc: name = parsed_name compact_result = {"error": str(exc)} raw_result = compact_result is_error = True protocol_violations.append(f"{name}: {exc}") tool_counts[name] = tool_counts.get(name, 0) + 1 messages.append( { "role": "tool", "tool_call_id": call_id, "content": json.dumps(compact_result, sort_keys=True, default=str), } ) writer.emit( "tool_call", { "tool_call_id": call_id, "name": name, "arguments_sha256": sha256( json.dumps( call.get("function", {}).get("arguments", ""), sort_keys=True, ).encode() ).hexdigest(), "is_error": is_error, "result": raw_result, }, ) if (name == "search_code" or name.startswith("search_")) and not is_error: for candidate in raw_result.get("results", []): writer.emit("retrieval_candidate", candidate) elif name == "read_file" and not is_error: writer.emit("file_read", raw_result) elif name == interface.edit_tool: writer.emit("edit", raw_result) elif name == "run_tests" and not is_error: writer.emit("test_run", raw_result) if tools.finished: finished_reason = "finish_tool" break if tools.finished or finished_reason == "tool_budget_exhausted": break patch = workspace.final_patch() validation = ( validate_generated_patch( root, repository, task, patch, preserve_git_metadata=True ) if patch else _failure_validation("empty_patch") ) edited_paths = tuple(sorted(workspace.edited_paths)) usage = _usage_totals(responses) elapsed = time.monotonic() - cell_started model_patch_apply = validation.get("model_patch_apply") applicable = bool( isinstance(model_patch_apply, dict) and model_patch_apply.get("returncode") == 0 ) accepted_edit_cell = edit_acceptances > 0 and bool(patch) final = { "run_id": identity.run_id, "experiment_id": experiment.experiment_id, "task_id": task.task_id, "harness_id": identity.harness_id, "retrieval_harness_id": ( retrieval_harness.harness_id if retrieval_harness is not None else None ), "edit_interface_id": interface.interface_id, "edit_tool": interface.edit_tool, "model_id": model.model_id, "accepted_edit_cell": accepted_edit_cell, "edit_attempts": edit_attempts, "edit_acceptances": edit_acceptances, "edit_attempt_acceptance_rate": ( edit_acceptances / edit_attempts if edit_attempts else 0.0 ), "applicable_final_patch": applicable, "resolved_at_1": validation["resolved_at_1"], "failure_stage": validation["failure_stage"], "patch_applied": applicable, "fail_to_pass": validation["fail_to_pass"], "pass_to_pass": validation["pass_to_pass"], "modified_files": edited_paths, "exact_modified_file_match": set(edited_paths) == set(task.gold_files), "gold_file_modified_recall": retrieval_metrics( edited_paths, task.gold_files )["file_recall_at_10"], "read_localization_metrics": retrieval_metrics( tuple(dict.fromkeys(tools.read_paths)), task.gold_files ), "search_localization_metrics": retrieval_metrics( tuple(dict.fromkeys(tools.search_paths)), task.gold_files ), "finished_reason": finished_reason, "finish_summary": tools.finish_summary, "protocol_violations": protocol_violations, "model_calls": len(responses), "tool_calls": tool_call_count, "tool_counts": 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, "trajectory_sha256": sha256( json.dumps(messages, sort_keys=True, separators=(",", ":")).encode() ).hexdigest(), "patch_sha256": sha256(patch.encode()).hexdigest() if patch else None, "residency_transitions": transitions, "dense_index_stats": retrieval.dense_index_stats if retrieval is not None else {}, "test_results": validation["tests"], } writer.write_artifact( "messages.json", json.dumps(messages, indent=2, sort_keys=True) + "\n" ) writer.write_artifact("model.patch", 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", "accepted_edit_cell": accepted_edit_cell, "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 = [ item for item in repositories.values() if item.repository_url == task.repository_url ] if len(matches) != 1: raise ProtocolExperimentError( 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" / "E09_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, "planned_cells": 540, "completed_cells": len(rows), "accepted_edit_cells": sum( bool(item["accepted_edit_cell"]) for item in 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_protocol_experiment( root: Path, experiment_id: str = "E09", task_filter: set[str] | None = None, interface_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 != "protocol_interface": raise ProtocolExperimentError("protocol runner requires mode=protocol_interface") task_catalog = load_tasks(root) interface_catalog = load_edit_interfaces(root) model_catalog = load_models(root) repository_catalog = load_repositories(root) 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 ] interfaces = [ interface_catalog[item] for item in experiment.edit_interface_ids if interface_filter is None or item in interface_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 interfaces or not models: raise ProtocolExperimentError("protocol filters selected an empty execution block") if any(task.validation_status != "end_to_end_ready" for task in tasks): raise ProtocolExperimentError("E09 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, ) rows: list[dict[str, Any]] = [] task_summaries: list[dict[str, Any]] = [] runtime = _RuntimeLease(server, residency, stop_server_when_complete) with runtime as server_state: for task_index, task in enumerate(tasks): repository_spec = _repository_for_task(repository_catalog, task) repository = (root / repository_spec.local_path).resolve() GitSnapshot(repository).verify_commit(task.base_commit) model_offset = task_index % len(models) model_order = models[model_offset:] + models[:model_offset] task_rows: list[dict[str, Any]] = [] interface_orders: dict[str, list[str]] = {} for model in model_order: canonical_model_index = models.index(model) offset = (task_index + canonical_model_index) % len(interfaces) interface_order = interfaces[offset:] + interfaces[:offset] interface_orders[model.model_id] = [ item.interface_id for item in interface_order ] for interface in interface_order: row = run_protocol_cell( root, repository, experiment, task, interface, model, residency, server, revision, ) rows.append(row) task_rows.append(row) _write_progress(root, experiment, revision, rows, task_summaries) task_summaries.append( { "task_id": task.task_id, "repository_id": repository_spec.repository_id, "language": task.language, "model_order": [item.model_id for item in model_order], "interface_orders": interface_orders, "cells": len(task_rows), "accepted_edit_cells": sum( bool(item["accepted_edit_cell"]) for item in task_rows ), "resolved": sum(bool(item["resolved_at_1"]) for item in task_rows), } ) _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), "accepted_edit_count": sum(bool(item["accepted_edit_cell"]) for item in 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_path = ( root / "results" / "reports" / f"E09_{revision[:12]}_{int(time.time())}.json" ) report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text( json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) return {**report, "report_path": str(report_path)} def _write_retrieval_progress( root: Path, experiment: ExperimentSpec, revision: str, planned_cells: int, 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, "planned_cells": planned_cells, "completed_cells": len(rows), "accepted_edit_cells": sum(bool(item["accepted_edit_cell"]) for item in 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_retrieval_protocol_experiment( root: Path, experiment_id: str = "E10", task_filter: set[str] | None = None, harness_filter: set[str] | None = None, model_filter: set[str] | None = None, stop_server_when_complete: bool = True, ) -> dict[str, Any]: """Run fresh retrieval treatments with E09 gate-selected edit interfaces.""" revision = research_code_revision(root) experiment = load_experiments(root)[experiment_id] if experiment_id != "E10" or experiment.mode != "protocol_interface": raise ProtocolExperimentError("retrieval protocol runner requires E10") task_catalog = load_tasks(root) harness_catalog = load_harnesses(root) interface_catalog = load_edit_interfaces(root) model_catalog = load_models(root) repository_catalog = load_repositories(root) embedding = load_embeddings(root)[experiment.embedding_id] gate = json.loads( (root / "configs" / "gates" / "E09_model_interface_gate.json").read_text( encoding="utf-8" ) ) selected_interfaces = { str(model_id): str(interface_id) for model_id, interface_id in gate["selected"].items() } 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 harness_filter is None or item in harness_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 harnesses or not models: raise ProtocolExperimentError("E10 filters selected an empty execution block") if any(task.validation_status != "end_to_end_ready" for task in tasks): raise ProtocolExperimentError("E10 includes a task without hidden-test validation") if any(model.model_id not in selected_interfaces for model in models): raise ProtocolExperimentError("E10 model has no frozen E09 gate selection") if set(item.interface_id for item in interface_catalog.values()) != {"P001", "P002", "P003"}: raise ProtocolExperimentError("edit-interface catalog drifted") planned_cells = len(tasks) * len(harnesses) * len(models) 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" 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_offset = task_index % len(models) model_order = models[model_offset:] + models[:model_offset] task_rows: list[dict[str, Any]] = [] harness_orders: dict[str, list[str]] = {} for model in model_order: interface = interface_catalog[selected_interfaces[model.model_id]] canonical_model_index = models.index(model) offset = (task_index + canonical_model_index) % len(harnesses) harness_order = harnesses[offset:] + harnesses[:offset] harness_orders[model.model_id] = [item.harness_id for item in harness_order] for harness in harness_order: row = run_protocol_cell( root, repository, experiment, task, interface, model, residency, server, revision, retrieval_harness=harness, retrieval=retrieval, embedding=embedding, ) rows.append(row) task_rows.append(row) _write_retrieval_progress( root, experiment, revision, planned_cells, rows, task_summaries ) task_summaries.append( { "task_id": task.task_id, "repository_id": repository_spec.repository_id, "language": task.language, "model_order": [item.model_id for item in model_order], "harness_orders": harness_orders, "gate_interfaces": { model.model_id: selected_interfaces[model.model_id] for model in models }, "embedding_index_transition": index_transition.to_dict(), "index_elapsed_seconds": index_elapsed, "dense_index_stats": retrieval.dense_index_stats, "cells": len(task_rows), "accepted_edit_cells": sum( bool(item["accepted_edit_cell"]) for item in task_rows ), "resolved": sum(bool(item["resolved_at_1"]) for item in task_rows), } ) _write_retrieval_progress( root, experiment, revision, planned_cells, rows, task_summaries ) if len(rows) != planned_cells: raise ProtocolExperimentError( f"E10 finalized {len(rows)}/{planned_cells} selected cells" ) report = { "experiment_id": experiment.experiment_id, "code_revision": revision, "planned_cells": planned_cells, "server_lifecycle": server_state, "server_stop": runtime.stop_state, "run_count": len(rows), "accepted_edit_count": sum(bool(item["accepted_edit_cell"]) for item in 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_path = ( root / "results" / "reports" / f"E10_{revision[:12]}_{int(time.time())}.json" ) report_path.parent.mkdir(parents=True, exist_ok=True) 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_study4_ancillary( root: Path, manifest_path: Path, stop_server_when_complete: bool = True, ) -> dict[str, Any]: """Run a frozen sparse Study 4 reliability or context-scarcity manifest.""" revision = research_code_revision(root) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) experiment_id = str(manifest.get("experiment_id", "")) if experiment_id not in {"E11", "E12"}: raise ProtocolExperimentError("Study 4 ancillary manifest must target E11 or E12") experiment = load_experiments(root)[experiment_id] if experiment.mode != "protocol_interface": raise ProtocolExperimentError("Study 4 ancillary requires protocol_interface mode") cells = manifest.get("cells") seeds = tuple(int(item) for item in manifest.get("seeds", [])) temperature = float(manifest.get("temperature", -1.0)) top_p = float(manifest.get("top_p", -1.0)) if not isinstance(cells, list) or not cells or not seeds: raise ProtocolExperimentError("ancillary manifest must contain cells and seeds") expected = ( (6, (0, 1, 2), 0.2, {65536}) if experiment_id == "E11" else (12, (0,), 0.0, {16384, 65536}) ) contexts = {int(item["context_budget"]) for item in cells} if (len(cells), seeds, temperature, contexts) != expected: raise ProtocolExperimentError( f"{experiment_id} manifest does not match the frozen reduced design" ) if top_p != 1.0: raise ProtocolExperimentError("ancillary top_p must equal 1.0") task_catalog = load_tasks(root) harness_catalog = load_harnesses(root) interface_catalog = load_edit_interfaces(root) model_catalog = load_models(root) repository_catalog = load_repositories(root) embedding = load_embeddings(root)[experiment.embedding_id] gate = json.loads( (root / "configs" / "gates" / "E09_model_interface_gate.json").read_text( encoding="utf-8" ) )["selected"] normalized: list[dict[str, Any]] = [] identities: set[tuple[str, str, str, int]] = set() for raw in cells: cell = { "task_id": str(raw["task_id"]), "harness_id": str(raw["harness_id"]), "model_id": str(raw["model_id"]), "context_budget": int(raw["context_budget"]), } key = ( cell["task_id"], cell["harness_id"], cell["model_id"], cell["context_budget"], ) if key in identities: raise ProtocolExperimentError(f"duplicate ancillary cell: {key}") identities.add(key) if cell["harness_id"] not in {"H000", "H007"}: raise ProtocolExperimentError("ancillary cells must be non-oracle H000/H007") if cell["model_id"] not in gate: raise ProtocolExperimentError("ancillary model lacks a frozen E09 gate") normalized.append(cell) planned_cells = len(normalized) * len(seeds) server = LMStudioServer(port=1234) residency = LMStudioResidencyManager( model_catalog[normalized[0]["model_id"]].base_url, model_catalog[normalized[0]["model_id"]].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, Any]]] = {} for cell in normalized: grouped.setdefault(cell["task_id"], []).append(cell) progress_path = root / "results" / "reports" / f"{experiment_id}_progress.json" 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) snapshot.verify_commit(task.base_commit) 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: base_model = model_catalog[cell["model_id"]] interface = interface_catalog[str(gate[cell["model_id"]])] harness = harness_catalog[cell["harness_id"]] for seed in seeds: model = replace( base_model, temperature=temperature, top_p=top_p, seed=seed, context_length=cell["context_budget"], ) row = run_protocol_cell( root, repository, experiment, task, interface, model, residency, server, revision, retrieval_harness=harness, retrieval=retrieval, embedding=embedding, seed=seed, context_budget=cell["context_budget"], ) rows.append(row) progress_path.parent.mkdir(parents=True, exist_ok=True) progress_path.write_text( json.dumps( { "schema_version": 1, "experiment_id": experiment_id, "code_revision": revision, "manifest_sha256": sha256( manifest_path.read_bytes() ).hexdigest(), "planned_cells": planned_cells, "completed_cells": len(rows), "accepted_edit_cells": sum( bool(item["accepted_edit_cell"]) for item in rows ), "resolved_cells": sum( bool(item["resolved_at_1"]) for item in rows ), "rows": rows, }, indent=2, sort_keys=True, ) + "\n", encoding="utf-8", ) if len(rows) != planned_cells: raise ProtocolExperimentError( f"{experiment_id} finalized {len(rows)}/{planned_cells} cells" ) report = { "schema_version": 1, "experiment_id": experiment_id, "code_revision": revision, "manifest_path": str(manifest_path), "manifest_sha256": sha256(manifest_path.read_bytes()).hexdigest(), "server_lifecycle": server_state, "server_stop": runtime.stop_state, "run_count": len(rows), "accepted_edit_count": sum(bool(item["accepted_edit_cell"]) for item in rows), "resolved_count": sum(bool(item["resolved_at_1"]) for item in rows), "final_residency_transition": runtime.final_transition, "cleanup_errors": runtime.cleanup_errors, "rows": rows, } report_path = ( root / "results" / "reports" / f"{experiment_id}_{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)}