"""E07 live search/read/edit/test agent experiment. Unlike E03, this runner does not prepack evidence. The fixed local Qwen model chooses and invokes repository tools over an isolated base-commit worktree. """ from __future__ import annotations from dataclasses import asdict, dataclass import difflib from hashlib import sha256 import json from pathlib import Path import resource import time from typing import Any, Callable, Sequence from .components import Candidate from .fusion import reciprocal_rank_fusion from .lm_studio import LMStudioClient, LMStudioTransportError from .lm_studio_embeddings import LMStudioEmbeddingClient from .lm_studio_management import ( LMStudioManagementError, LMStudioResidencyManager, LMStudioServer, ResidencyTransition, ) from .pilot import research_code_revision, retrieval_metrics from .repair_experiment import ( PatchOutputError, _apply_patch, isolated_source_tree, isolated_git_tree, modified_paths, run_test_command, validate_generated_patch, ) from .repository import GitSnapshot, SourceChunk, chunk_snapshot from .retrieval import ( BM25FuzzyRetriever, DenseRetriever, ExactRetriever, SQLiteEmbeddingCache, ) from .specs import ( AgentSystemSpec, EmbeddingSpec, ExperimentSpec, HarnessSpec, ModelSpec, TaskSpec, load_embeddings, load_experiments, load_harnesses, load_models, load_task_split, load_tasks, ) from .syntax_index import SymbolGraph, SyntaxRetriever, parse_snapshot from .syntax_index import parse_source_file from .telemetry import EventWriter, RunIdentity, load_completed_or_archive_incomplete from .tokenization import QwenTokenCounter MAX_MODEL_CALLS = 12 SEARCH_LIMIT = 5 SEARCH_SNIPPET_LINES = 24 SEARCH_SNIPPET_CHARS = 1_200 READ_MAX_LINES = 200 READ_MAX_CHARS = 16_000 TEST_OUTPUT_CHARS = 12_000 CONVERSATION_TOKEN_LIMIT = 58_000 LIVE_AGENT_SYSTEM = """You are a coding agent repairing one issue in a large Go repository. The repository is much larger than the context window. Use the provided tools to locate the implementation, read exact source, apply the smallest correct production-code patch, and run an allowed public test when useful. End by calling finish. Rules: - Never invent file contents; read before editing. - Hidden tests, gold patches, and gold symbols are unavailable. - Do not add or modify test files. - apply_patch accepts a standard unified diff with a/ and b/ paths. - Tool errors are observations: correct the request instead of pretending it succeeded. - Stay within the tool and test budgets. Prefer focused queries and reads. - The only durable edit is one accepted by apply_patch. Text in a normal assistant message is not an edit. """ def live_agent_system(language: str, swe_agent_style: bool = False) -> str: if language == "go" and not swe_agent_style: return LIVE_AGENT_SYSTEM label = {"go": "Go", "python": "Python"}.get(language, language) interface = ( "Use the controlled interactive agent-computer interface to find files, search text, " "read source, edit with a patch, and run an allowed test." if swe_agent_style else "Use the provided retrieval tools to locate the implementation, read exact source, " "apply the smallest correct production-code patch, and run an allowed public test when useful." ) return f"""You are a coding agent repairing one issue in a large {label} repository. The repository is much larger than the context window. {interface} End by calling finish. Rules: - Never invent file contents; read before editing. - Hidden tests, gold patches, and gold symbols are unavailable. - Do not add or modify test files. - apply_patch accepts a standard unified diff with a/ and b/ paths. - Tool errors are observations: correct the request instead of pretending it succeeded. - Stay within the tool and test budgets. Prefer focused queries and reads. - The only durable edit is one accepted by apply_patch. Text in a normal assistant message is not an edit. """ class LiveAgentExperimentError(RuntimeError): """Raised when E07 infrastructure cannot preserve its frozen protocol.""" @dataclass(slots=True) class TaskRetrieval: chunks: tuple[SourceChunk, ...] exact: ExactRetriever lexical: BM25FuzzyRetriever syntax: SyntaxRetriever graph: SymbolGraph dense: DenseRetriever dense_index_stats: dict[str, Any] def _safe_relative_path(value: str) -> str: path = Path(value) if not value or path.is_absolute() or ".." in path.parts: raise ValueError(f"unsafe repository path: {value!r}") return path.as_posix() def _trim(value: str, limit: int) -> str: if len(value) <= limit: return value return value[:limit] + f"\n...[truncated {len(value) - limit} characters]" class AgentWorkspace: def __init__(self, tree: Path, tracked_paths: Sequence[str], task: TaskSpec, max_test_runs: int): self.tree = tree self.tracked_paths = set(tracked_paths) self.task = task self.max_test_runs = max_test_runs self.test_runs: list[dict[str, Any]] = [] self.original: dict[str, str] = {} self.edited_paths: set[str] = set() self.patch_attempts = 0 @property def language_name(self) -> str: return {"go": "Go", "python": "Python"}.get(self.task.language, self.task.language) def is_test_path(self, path: str) -> bool: if self.task.language == "go": return path.endswith("_test.go") if self.task.language == "python": return path.startswith("tests/") or Path(path).name.startswith("test_") return False def read_file(self, path: str, line_start: int = 1, line_end: int | None = None) -> dict[str, Any]: safe = _safe_relative_path(path) if safe not in self.tracked_paths: raise ValueError( f"path is not a tracked {self.language_name} source file at the frozen base commit: {safe}" ) target = self.tree / safe text = target.read_text(encoding="utf-8", errors="replace") lines = text.splitlines() start = max(int(line_start), 1) requested_end = len(lines) if line_end is None else int(line_end) end = min(max(requested_end, start), len(lines), start + READ_MAX_LINES - 1) numbered = "\n".join( f"{number:>6}: {lines[number - 1]}" for number in range(start, end + 1) ) return { "path": safe, "line_start": start, "line_end": end, "total_lines": len(lines), "content": _trim(numbered, READ_MAX_CHARS), } def apply_patch(self, patch: str) -> dict[str, Any]: if not isinstance(patch, str) or not patch.strip(): raise ValueError("patch must be non-empty unified-diff text") paths = modified_paths(patch) for path in paths: safe = _safe_relative_path(path) if safe not in self.tracked_paths: raise ValueError( f"patch may modify only tracked {self.language_name} source files: {safe}" ) if self.is_test_path(safe): raise ValueError(f"test edits are forbidden: {safe}") for path in paths: if path not in self.original: self.original[path] = (self.tree / path).read_text( encoding="utf-8", errors="replace" ) self.patch_attempts += 1 patch_path = self.tree.parent / f"agent-edit-{self.patch_attempts:02d}.patch" patch_path.write_text(patch.rstrip() + "\n", encoding="utf-8") result = _apply_patch(self.tree, patch_path) if result["returncode"] == 0: self.edited_paths.update(paths) return {**result, "paths": paths, "accepted": result["returncode"] == 0} def run_tests(self, command: str) -> dict[str, Any]: if command not in set((*self.task.fail_to_pass_tests, *self.task.pass_to_pass_tests)): raise ValueError( "command is not in the frozen public-test allowlist: " + repr(command) ) if len(self.test_runs) >= self.max_test_runs: raise ValueError(f"test budget exhausted ({self.max_test_runs})") result = run_test_command(self.tree, command) self.test_runs.append(result) return result def final_patch(self) -> str: blocks: list[str] = [] for path in sorted(self.edited_paths): before = self.original[path].splitlines(keepends=True) after = (self.tree / path).read_text( encoding="utf-8", errors="replace" ).splitlines(keepends=True) blocks.extend( difflib.unified_diff( before, after, fromfile=f"a/{path}", tofile=f"b/{path}", lineterm="\n", ) ) patch = "".join(blocks) return patch if not patch or patch.endswith("\n") else patch + "\n" def _candidate_record(candidate: Candidate) -> dict[str, Any]: lines = candidate.text.splitlines() selected = lines[:SEARCH_SNIPPET_LINES] snippet = "\n".join( f"{candidate.line_start + offset:>6}: {line}" for offset, line in enumerate(selected) ) return { "path": candidate.path, "line_start": candidate.line_start, "line_end": min(candidate.line_end, candidate.line_start + len(selected) - 1), "source": candidate.source, "score": candidate.score, "symbol": candidate.symbol, "snippet": _trim(snippet, SEARCH_SNIPPET_CHARS), } def _unique_candidates(candidates: Sequence[Candidate], limit: int) -> tuple[Candidate, ...]: seen: set[str] = set() result: list[Candidate] = [] for candidate in candidates: if candidate.path in seen: continue seen.add(candidate.path) result.append(candidate) if len(result) >= limit: break return tuple(result) class LiveToolHarness: def __init__( self, harness: HarnessSpec, retrieval: TaskRetrieval, workspace: AgentWorkspace, residency: LMStudioResidencyManager, model: ModelSpec, embedding: EmbeddingSpec, transition_callback: Callable[[ResidencyTransition], None], ): self.harness = harness self.retrieval = retrieval self.workspace = workspace self.residency = residency self.model = model self.embedding = embedding self.transition_callback = transition_callback self.search_paths: list[str] = [] self.read_paths: list[str] = [] self.finished = False self.finish_summary = "" self.search_call_count = 0 def _packed_records(self, candidates: Sequence[Candidate]) -> list[dict[str, Any]]: """Render search observations according to the immutable packing treatment. Live studies before Study 5 always returned ranked snippets. Study 5 prospectively operationalizes the catalogued packing field at the search observation boundary while retaining the same ranked candidate paths. """ if self.harness.packing == "ranked_snippets": return [_candidate_record(item) for item in candidates] records: list[dict[str, Any]] = [] for candidate in candidates: path = candidate.path source = (self.workspace.tree / path).read_text( encoding="utf-8", errors="replace" ) base = { "path": path, "line_start": candidate.line_start, "line_end": candidate.line_end, "source": candidate.source, "score": candidate.score, "symbol": candidate.symbol, } if self.harness.packing == "whole_files": observation = _trim(source, 12_000) else: symbols = parse_source_file(path, source, self.workspace.task.language) if self.harness.packing == "skeletons": observation = "\n".join( f"{item.kind} {item.name} lines {item.line_start}-{item.line_end}: " f"{item.signature}" for item in symbols ) elif self.harness.packing == "role_summaries": kinds: dict[str, list[str]] = {} for item in symbols: kinds.setdefault(item.kind, []).append(item.name) observation = "\n".join( f"{kind}: {', '.join(names[:40])}" for kind, names in sorted(kinds.items()) ) if not observation: observation = "No indexed declarations; read the file for details." else: # guarded by HarnessSpec validation raise ValueError(f"unsupported packing strategy: {self.harness.packing}") records.append({**base, "packing": self.harness.packing, "snippet": observation}) return records def _begin_search(self) -> None: if self.harness.query_policy == "one_shot" and self.search_call_count >= 1: raise ValueError( "one-shot query policy permits exactly one repository search; " "use read_file on an observed path" ) self.search_call_count += 1 def _dense(self, query: str, limit: int) -> Sequence[Candidate]: transition = self.residency.ensure_exclusive( self.embedding.model_key, self.embedding.loaded_context_length ) self.transition_callback(transition) return self.retrieval.dense.retrieve(query, limit) def _base_rankings(self, query: str, include_dense: bool) -> list[Sequence[Candidate]]: rankings: list[Sequence[Candidate]] = [] if self.harness.exact_search: rankings.append(self.retrieval.exact.retrieve(query, 50)) if self.harness.lexical: rankings.append(self.retrieval.lexical.retrieve(query, 50)) if self.harness.syntax == "tree_sitter": rankings.append(self.retrieval.syntax.retrieve(query, 50)) if include_dense and self.harness.dense: rankings.append(self._dense(query, 50)) return rankings def unified_search(self, query: str) -> tuple[Candidate, ...]: if self.harness.control != "none": raise ValueError("this control harness has no search capability") rankings = self._base_rankings(query, include_dense=True) if not rankings: return () if self.harness.fusion == "rrf" and len(rankings) >= 2: candidates = reciprocal_rank_fusion(rankings, limit=50) elif self.harness.harness_id == "H003" and len(rankings) == 2: # H003 is dense-primary with literal-search backfill; it does not # introduce the RRF treatment used by H007-H011. candidates = _unique_candidates((*rankings[-1], *rankings[0]), 50) else: candidates = tuple(rankings[-1]) if self.harness.graph_hops: candidates = self.retrieval.graph.expand( candidates, self.harness.graph_hops, limit=50 ) return _unique_candidates(candidates, SEARCH_LIMIT) def specialized_search(self, name: str, query: str) -> tuple[Candidate, ...]: if name == "search_exact": values = self.retrieval.exact.retrieve(query, 50) elif name == "search_lexical": values = self.retrieval.lexical.retrieve(query, 50) elif name == "search_syntax": values = self.retrieval.syntax.retrieve(query, 50) elif name == "search_dense": values = self._dense(query, 50) elif name == "search_graph": rankings = self._base_rankings(query, include_dense=True) fused = reciprocal_rank_fusion(rankings, limit=50) values = self.retrieval.graph.expand(fused, 1, limit=50) else: raise ValueError(f"unknown specialized search tool: {name}") return _unique_candidates(values, SEARCH_LIMIT) def execute(self, name: str, arguments: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: if name == "search_code": self._begin_search() candidates = self.unified_search(str(arguments.get("query", ""))) records = self._packed_records(candidates) self.search_paths.extend(item.path for item in candidates) return {"results": records}, {"query": arguments.get("query"), "results": records} if name.startswith("search_"): self._begin_search() candidates = self.specialized_search(name, str(arguments.get("query", ""))) records = self._packed_records(candidates) self.search_paths.extend(item.path for item in candidates) return {"results": records}, {"query": arguments.get("query"), "results": records} 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 == "apply_patch": result = self.workspace.apply_patch(str(arguments.get("patch", ""))) compact = { "accepted": result["accepted"], "paths": result["paths"], "returncode": result["returncode"], "stdout": _trim(result["stdout"], 2_000), "stderr": _trim(result["stderr"], 4_000), "elapsed_seconds": result["elapsed_seconds"], } return compact, 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}") class SWEAgentStyleToolHarness: """Controlled search/read/edit/test interface inspired by SWE-agent's ACI.""" def __init__(self, retrieval: TaskRetrieval, workspace: AgentWorkspace): self.retrieval = retrieval self.workspace = workspace self.search_paths: list[str] = [] self.read_paths: list[str] = [] self.finished = False self.finish_summary = "" def _find_files(self, query: str) -> tuple[Candidate, ...]: terms = tuple(item.lower() for item in query.split() if item.strip()) scored: list[tuple[int, str]] = [] for path in sorted(self.workspace.tracked_paths): lowered = path.lower() score = sum(term in lowered for term in terms) if score: scored.append((score, path)) scored.sort(key=lambda item: (-item[0], item[1])) values: list[Candidate] = [] for score, path in scored[:SEARCH_LIMIT]: text = (self.workspace.tree / path).read_text( encoding="utf-8", errors="replace" ) selected = text.splitlines()[:SEARCH_SNIPPET_LINES] values.append( Candidate( path=path, line_start=1, line_end=max(len(selected), 1), text="\n".join(selected), source="find_files", score=float(score), ) ) return tuple(values) def execute(self, name: str, arguments: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: if name == "find_files": candidates = self._find_files(str(arguments.get("query", ""))) records = [_candidate_record(item) for item in candidates] self.search_paths.extend(item.path for item in candidates) return {"results": records}, {"query": arguments.get("query"), "results": records} if name == "search_text": candidates = _unique_candidates( self.retrieval.exact.retrieve(str(arguments.get("query", "")), 50), SEARCH_LIMIT, ) records = [_candidate_record(item) for item in candidates] self.search_paths.extend(item.path for item in candidates) return {"results": records}, {"query": arguments.get("query"), "results": records} 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 == "apply_patch": result = self.workspace.apply_patch(str(arguments.get("patch", ""))) compact = { "accepted": result["accepted"], "paths": result["paths"], "returncode": result["returncode"], "stdout": _trim(result["stdout"], 2_000), "stderr": _trim(result["stderr"], 4_000), "elapsed_seconds": result["elapsed_seconds"], } return compact, 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}") def _search_schema(name: str, description: str) -> dict[str, Any]: return { "type": "function", "function": { "name": name, "description": description, "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Focused code search query"} }, "required": ["query"], "additionalProperties": False, }, }, } def tool_definitions(harness: HarnessSpec, task: TaskSpec) -> list[dict[str, Any]]: tools: list[dict[str, Any]] = [] language = {"go": "Go", "python": "Python"}.get(task.language, task.language) if harness.control == "none" and harness.interface == "unified": tools.append(_search_schema("search_code", "Search the repository using this harness's fixed retrieval stack.")) elif harness.control == "none" and harness.interface == "specialized": tools.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", "Full fused retrieval followed by one static graph hop."), ] ) tools.extend( [ { "type": "function", "function": { "name": "read_file", "description": f"Read a bounded line range from a known 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, }, }, }, { "type": "function", "function": { "name": "apply_patch", "description": f"Apply a unified diff to production {language} files in the isolated worktree.", "parameters": { "type": "object", "properties": {"patch": {"type": "string"}}, "required": ["patch"], "additionalProperties": False, }, }, }, { "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 patch has been applied.", "parameters": { "type": "object", "properties": {"summary": {"type": "string"}}, "required": ["summary"], "additionalProperties": False, }, }, }, ] ) return tools def swe_agent_style_tool_definitions(task: TaskSpec) -> list[dict[str, Any]]: tools = [ _search_schema("find_files", "Find tracked source files by path/name terms."), _search_schema("search_text", "Literal or regular-expression search over source text."), ] tools.extend( item for item in tool_definitions(load_harnesses()["H000"], task) if item["function"]["name"] != "search_code" ) return tools def _parse_tool_arguments(call: dict[str, Any]) -> tuple[str, dict[str, Any]]: function = call.get("function", {}) name = function.get("name") raw = function.get("arguments", "{}") if not isinstance(name, str) or not name: raise ValueError("tool call has no function name") if isinstance(raw, dict): arguments = raw elif isinstance(raw, str): arguments = json.loads(raw) else: raise ValueError("tool arguments must be JSON text or an object") if not isinstance(arguments, dict): raise ValueError("tool arguments must decode to an object") return name, arguments def _compact_conversation( messages: list[dict[str, Any]], tokenizer: QwenTokenCounter ) -> tuple[int, int]: before = tokenizer.count(json.dumps(messages, ensure_ascii=False, sort_keys=True)) if before <= CONVERSATION_TOKEN_LIMIT: return before, before # Preserve the system/task, assistant decisions, tool-call arguments, and the # two newest tool outputs. Older bulky observations are replaced deterministically. tool_indices = [index for index, item in enumerate(messages) if item.get("role") == "tool"] for index in tool_indices[:-2]: content = str(messages[index].get("content", "")) if len(content) > 240: messages[index]["content"] = json.dumps( {"notice": "older tool output compacted", "original_sha256": sha256(content.encode()).hexdigest()} ) current = tokenizer.count(json.dumps(messages, ensure_ascii=False, sort_keys=True)) if current <= CONVERSATION_TOKEN_LIMIT: return before, current return before, tokenizer.count(json.dumps(messages, ensure_ascii=False, sort_keys=True)) def _assistant_message(response: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]: try: message = response["choices"][0]["message"] except (KeyError, IndexError, TypeError) as exc: raise LiveAgentExperimentError("chat completion has no assistant message") from exc if not isinstance(message, dict): raise LiveAgentExperimentError("chat completion assistant message is malformed") result: dict[str, Any] = { "role": "assistant", "content": message.get("content") if isinstance(message.get("content"), str) else "", } calls = message.get("tool_calls", []) if calls is None: calls = [] if not isinstance(calls, list) or not all(isinstance(item, dict) for item in calls): raise LiveAgentExperimentError("assistant tool_calls field is malformed") if calls: result["tool_calls"] = calls return result, calls def _usage_totals(responses: Sequence[dict[str, Any]]) -> dict[str, int]: keys = ("prompt_tokens", "completion_tokens", "total_tokens") return { key: sum( int(response.get("usage", {}).get(key, 0) or 0) for response in responses if isinstance(response.get("usage", {}), dict) ) for key in keys } def _failure_validation(stage: str) -> dict[str, Any]: return { "hidden_test_patch_apply": None, "model_patch_apply": None, "tests": [], "fail_to_pass": False, "pass_to_pass": False, "resolved_at_1": False, "failure_stage": stage, } def _identity( experiment: ExperimentSpec, task: TaskSpec, harness: HarnessSpec, model: ModelSpec, revision: str, agent_system: AgentSystemSpec | None = None, seed: int | None = None, repetition: int = 0, ) -> RunIdentity: treatment_id = agent_system.system_id if agent_system else harness.harness_id treatment_hash = agent_system.config_hash if agent_system else harness.config_hash 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], seed=experiment.seeds[0] if seed is None else seed, repetition=repetition, repository_sha=task.base_commit, code_revision=revision, ) def _build_task_retrieval( snapshot: GitSnapshot, task: TaskSpec, embedding: EmbeddingSpec, embedding_client: LMStudioEmbeddingClient, cache: SQLiteEmbeddingCache, ) -> TaskRetrieval: chunks = chunk_snapshot( snapshot, task.base_commit, embedding.chunk_lines, embedding.chunk_overlap_lines, embedding.chunk_char_limit, suffixes={"go": (".go",), "python": (".py",)}[task.language], ) symbols = parse_snapshot(snapshot, task.base_commit, task.language) dense, stats = DenseRetriever.build(chunks, embedding, embedding_client, cache) return TaskRetrieval( chunks=chunks, exact=ExactRetriever(chunks), lexical=BM25FuzzyRetriever(chunks), syntax=SyntaxRetriever(symbols), graph=SymbolGraph(symbols), dense=dense, dense_index_stats=asdict(stats), ) def run_live_agent_cell( root: Path, repository: Path, experiment: ExperimentSpec, task: TaskSpec, harness: HarnessSpec, model: ModelSpec, embedding: EmbeddingSpec, retrieval: TaskRetrieval, residency: LMStudioResidencyManager, server: LMStudioServer, tokenizer: QwenTokenCounter, revision: str, agent_system: AgentSystemSpec | None = None, seed: int | None = None, repetition: int = 0, preserve_git_metadata: bool = False, ) -> dict[str, Any]: if agent_system is not None and agent_system.system_id != "A002": raise LiveAgentExperimentError("interactive cell supports only controlled system A002") identity = _identity( experiment, task, harness, model, revision, agent_system=agent_system, seed=seed, repetition=repetition, ) treatment_id = identity.harness_id 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) discovery, resolved = client.resolve() source_suffixes = {"go": (".go",), "python": (".py",)} if task.language not in source_suffixes: raise LiveAgentExperimentError(f"unsupported Study 2 language: {task.language}") tracked_paths = GitSnapshot(repository).tracked_paths( task.base_commit, source_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(), } source_context = ( isolated_git_tree(repository, task.base_commit, task.repository_url) if preserve_git_metadata else isolated_source_tree(repository, task.base_commit) ) resolved_treatment = asdict(agent_system) if agent_system else asdict(harness) with source_context as tree, EventWriter( root / "results", identity, resolved_treatment, resolved_model ) as writer: transitions: list[dict[str, Any]] = [initial_transition.to_dict()] responses: list[dict[str, Any]] = [] protocol_violations: list[str] = [] tool_counts: dict[str, int] = {} tool_call_count = 0 model_elapsed = 0.0 finished_reason = "model_turn_budget" max_test_runs = agent_system.max_test_runs if agent_system else experiment.max_test_runs workspace = AgentWorkspace(tree, tracked_paths, task, max_test_runs) def record_transition(transition: ResidencyTransition) -> None: value = transition.to_dict() transitions.append(value) writer.emit("resource_sample", {"kind": "model_residency_transition", **value}) live_tools: LiveToolHarness | SWEAgentStyleToolHarness if agent_system: live_tools = SWEAgentStyleToolHarness(retrieval, workspace) else: live_tools = LiveToolHarness( harness, retrieval, workspace, residency, model, embedding, record_transition ) task_prompt = f"ISSUE:\n{task.statement}" if not agent_system and harness.control == "oracle_file": task_prompt += "\n\nORACLE FILE LOCATIONS (names only):\n" + "\n".join(task.gold_files) elif not agent_system and harness.control == "no_search": task_prompt += "\n\nThis treatment intentionally provides no repository search tool." messages: list[dict[str, Any]] = [ { "role": "system", "content": live_agent_system(task.language, swe_agent_style=bool(agent_system)), }, {"role": "user", "content": task_prompt}, ] definitions = ( swe_agent_style_tool_definitions(task) if agent_system else tool_definitions(harness, task) ) writer.emit( "run_started", { "confirmatory": True, "blinded": True, "task_config_hash": task.config_hash, "tool_names": [item["function"]["name"] for item in definitions], "budgets": { "model_calls": agent_system.model_calls if agent_system else MAX_MODEL_CALLS, "tool_calls": agent_system.max_tool_calls if agent_system else experiment.max_tool_calls, "test_runs": max_test_runs, "timeout_seconds": experiment.timeout_seconds, }, }, ) cell_started = time.monotonic() model_call_budget = agent_system.model_calls if agent_system else MAX_MODEL_CALLS tool_call_budget = agent_system.max_tool_calls if agent_system else experiment.max_tool_calls for model_turn in range(1, model_call_budget + 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) # Re-resolve after every potential embedding->agent transition. This # makes a wrong variant, context, or reasoning mode a fatal invariant. call_started = time.monotonic() try: discovery, resolved = client.resolve() response = client.chat_completions( resolved.inference_key, messages, tools=definitions, max_tokens=model.max_tokens, seed=identity.seed, ) except LMStudioTransportError as first_error: # A single transparent transport recovery is allowed only because # no valid model response was observed. 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) _, resolved = client.resolve() response = client.chat_completions( resolved.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 >= tool_call_budget: finished_reason = "tool_budget_exhausted" break tool_call_count += 1 call_id = str(call.get("id") or f"tool-{tool_call_count}") try: name, arguments = _parse_tool_arguments(call) tool_counts[name] = tool_counts.get(name, 0) + 1 compact_result, raw_result = live_tools.execute(name, arguments) is_error = False except (ValueError, PatchOutputError, json.JSONDecodeError) as exc: name = str(call.get("function", {}).get("name", "invalid_tool")) tool_counts[name] = tool_counts.get(name, 0) + 1 compact_result = {"error": str(exc)} raw_result = compact_result is_error = True protocol_violations.append(f"{name}: {exc}") messages.append( { "role": "tool", "tool_call_id": call_id, "content": json.dumps(compact_result, sort_keys=True, default=str), } ) event_payload = { "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, } writer.emit("tool_call", event_payload) if (name.startswith("search_") or name == "find_files") 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 == "apply_patch": writer.emit("edit", raw_result) elif name == "run_tests" and not is_error: writer.emit("test_run", raw_result) if live_tools.finished: finished_reason = "finish_tool" break if live_tools.finished or finished_reason == "tool_budget_exhausted": break patch = workspace.final_patch() if patch: validation = validate_generated_patch( root, repository, task, patch, preserve_git_metadata=preserve_git_metadata, ) else: validation = _failure_validation("empty_patch") edited_paths = tuple(sorted(workspace.edited_paths)) localization = retrieval_metrics(edited_paths, task.gold_files) search_metrics = retrieval_metrics(tuple(dict.fromkeys(live_tools.search_paths)), task.gold_files) read_metrics = retrieval_metrics(tuple(dict.fromkeys(live_tools.read_paths)), task.gold_files) usage = _usage_totals(responses) elapsed = time.monotonic() - cell_started switch_seconds = sum( float(item["elapsed_seconds"]) for item in transitions if not item.get("reused") ) final = { "run_id": identity.run_id, "experiment_id": experiment.experiment_id, "task_id": task.task_id, "harness_id": treatment_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": localization, "search_localization_metrics": search_metrics, "read_localization_metrics": read_metrics, "finished_reason": finished_reason, "finish_summary": live_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": switch_seconds, "peak_process_rss_platform_units": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, "dense_index_stats": retrieval.dense_index_stats, "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, "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", "resolved_at_1": validation["resolved_at_1"], "failure_stage": validation["failure_stage"], "finished_reason": finished_reason, }, ) return final def run_live_agent_experiment( root: Path, repository: Path, experiment_id: str = "E07", task_filter: set[str] | None = None, harness_filter: set[str] | None = None, ) -> dict[str, Any]: revision = research_code_revision(root) experiment = load_experiments(root)[experiment_id] if experiment.mode != "live_agent_repair": raise LiveAgentExperimentError("live-agent runner requires mode=live_agent_repair") catalog = load_harnesses(root) task_catalog = load_tasks(root) model = load_models(root)[experiment.model_ids[0]] 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 = [catalog[item] for item in experiment.harness_ids if harness_filter is None or item in harness_filter] if not tasks or not harnesses: raise LiveAgentExperimentError("task or harness filters selected no E07 cells") if any(task.validation_status != "end_to_end_ready" for task in tasks): raise LiveAgentExperimentError("E07 split includes a task without frozen hidden-test validation") server = LMStudioServer(port=1234) server_state = server.ensure_running() residency = LMStudioResidencyManager( model.base_url, model.api_token_env, timeout_seconds=experiment.timeout_seconds ) snapshot = GitSnapshot(repository) tokenizer = QwenTokenCounter() 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]] = [] with SQLiteEmbeddingCache(cache_path, embedding) as cache: for task_index, task in enumerate(tasks): 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 # Cyclic treatment order counterbalances systematic thermal/time drift. offset = task_index % len(harnesses) ordered_harnesses = harnesses[offset:] + harnesses[:offset] task_rows: list[dict[str, Any]] = [] for harness in ordered_harnesses: row = run_live_agent_cell( root, repository, experiment, task, harness, model, embedding, retrieval, residency, server, tokenizer, revision, ) rows.append(row) task_rows.append(row) task_summaries.append( { "task_id": task.task_id, "harness_order": [item.harness_id for item in ordered_harnesses], "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), } ) final_transition = residency.unload_all() report = { "experiment_id": experiment.experiment_id, "code_revision": revision, "server_lifecycle": server_state, "run_count": len(rows), "resolved_count": sum(bool(item["resolved_at_1"]) for item in rows), "task_summaries": task_summaries, "final_residency_transition": final_transition.to_dict(), "rows": rows, } report_dir = root / "results" / "reports" report_dir.mkdir(parents=True, exist_ok=True) report_path = report_dir / f"E07_{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)}