src/agent_harness/live_agent_experiment.py

"""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)}