src/agent_harness/study2_experiment.py

"""Prospective E08 multi-repository live-agent experiment.

Component harness cells reuse the audited E07 engine with generalized inputs.
A001 is a three-stage Agentless-style controlled adaptation; A002 uses the
same engine with a controlled SWE-agent-style tool interface.
"""

from __future__ import annotations

from dataclasses import asdict, replace
from hashlib import sha256
import json
from pathlib import Path
import resource
import time
from typing import Any, Sequence

from .live_agent_experiment import (
    AgentWorkspace,
    TaskRetrieval,
    _build_task_retrieval,
    _failure_validation,
    _usage_totals,
    run_live_agent_cell,
)
from .lm_studio import LMStudioClient, LMStudioTransportError
from .lm_studio_embeddings import LMStudioEmbeddingClient
from .lm_studio_management import LMStudioResidencyManager, LMStudioServer
from .pilot import research_code_revision, retrieval_metrics
from .repair_experiment import (
    PatchOutputError,
    extract_unified_diff,
    isolated_git_tree,
    validate_generated_patch,
)
from .repository import GitSnapshot
from .retrieval import SQLiteEmbeddingCache
from .specs import (
    AgentSystemSpec,
    EmbeddingSpec,
    ExperimentSpec,
    ModelSpec,
    RepositorySpec,
    TaskSpec,
    load_agent_systems,
    load_embeddings,
    load_experiments,
    load_harnesses,
    load_models,
    load_repositories,
    load_task_split,
    load_tasks,
)
from .telemetry import EventWriter, RunIdentity, load_completed_or_archive_incomplete
from .tokenization import QwenTokenCounter


AGENTLESS_INDEX_TOKENS = 42_000
AGENTLESS_SOURCE_TOKENS = 48_000
TOKENIZER_PATHS = {
    "M002": Path.home()
    / ".lmstudio/models/lmstudio-community/Qwen3.6-35B-A3B-MLX-4bit/tokenizer.json",
    "M003": Path.home()
    / ".lmstudio/models/mlx-community/gpt-oss-20b-MXFP4-Q8/tokenizer.json",
    "M004": Path.home()
    / ".lmstudio/models/lmstudio-community/Qwen3-Coder-30B-A3B-Instruct-MLX-4bit/tokenizer.json",
}


class Study2ExperimentError(RuntimeError):
    """Raised when E08 cannot preserve its preregistered protocol."""


class _RuntimeLease:
    """Guarantee exclusive-model cleanup on success and on infrastructure errors."""

    def __init__(
        self,
        server: LMStudioServer,
        residency: LMStudioResidencyManager,
        stop_server: bool,
    ):
        self.server = server
        self.residency = residency
        self.stop_server = stop_server
        self.server_state: dict[str, Any] | None = None
        self.final_transition: dict[str, Any] | None = None
        self.stop_state: dict[str, Any] | None = None
        self.cleanup_errors: list[str] = []

    def __enter__(self) -> dict[str, Any]:
        self.server_state = self.server.ensure_running()
        return self.server_state

    def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
        try:
            self.final_transition = self.residency.unload_all().to_dict()
        except Exception as cleanup_error:
            self.cleanup_errors.append(f"unload_all: {cleanup_error}")
        if self.stop_server:
            try:
                status = self.server.status()
                if status["running"]:
                    self.stop_state = self.server.stop()
                else:
                    self.stop_state = {
                        "action": "already_stopped",
                        "status": status,
                    }
            except Exception as cleanup_error:
                self.cleanup_errors.append(f"server_stop: {cleanup_error}")
        if exc_type is None and self.cleanup_errors:
            raise Study2ExperimentError(
                "runtime cleanup failed: " + "; ".join(self.cleanup_errors)
            )


def tokenizer_for(model: ModelSpec) -> QwenTokenCounter:
    try:
        path = TOKENIZER_PATHS[model.model_id]
    except KeyError as exc:
        raise Study2ExperimentError(f"no frozen tokenizer for {model.model_id}") from exc
    if not path.exists():
        raise Study2ExperimentError(f"frozen tokenizer is unavailable: {path}")
    return QwenTokenCounter(path)


def _identity(
    experiment: ExperimentSpec,
    task: TaskSpec,
    system: AgentSystemSpec,
    model: ModelSpec,
    revision: str,
    seed: int,
    repetition: int,
) -> RunIdentity:
    return RunIdentity(
        experiment_id=experiment.experiment_id,
        task_id=task.task_id,
        harness_id=system.system_id,
        harness_hash=system.config_hash,
        model_id=model.model_id,
        model_key=model.expected_inference_key,
        model_config_hash=model.config_hash,
        context_budget=experiment.context_budgets[0],
        seed=seed,
        repetition=repetition,
        repository_sha=task.base_commit,
        code_revision=revision,
    )


def _assistant_content(response: dict[str, Any]) -> str:
    try:
        content = response["choices"][0]["message"]["content"]
    except (KeyError, IndexError, TypeError) as exc:
        raise Study2ExperimentError("agentless stage returned no assistant message") from exc
    if not isinstance(content, str):
        raise Study2ExperimentError("agentless stage assistant content is not text")
    return content


def _json_object(content: str) -> dict[str, Any]:
    stripped = content.strip()
    fence = chr(96) * 3
    if stripped.startswith(fence):
        stripped = stripped.strip(chr(96))
        if "\n" in stripped:
            stripped = stripped.split("\n", 1)[1]
    start, end = stripped.find("{"), stripped.rfind("}")
    if start < 0 or end < start:
        raise ValueError("response contains no JSON object")
    value = json.loads(stripped[start : end + 1])
    if not isinstance(value, dict):
        raise ValueError("response JSON is not an object")
    return value


def _select_files(content: str, tracked_paths: Sequence[str]) -> tuple[str, ...]:
    allowed = set(tracked_paths)
    selected: list[str] = []
    try:
        value = _json_object(content)
        raw = value.get("files", [])
        if isinstance(raw, list):
            selected.extend(str(item) for item in raw if str(item) in allowed)
    except (ValueError, json.JSONDecodeError):
        pass
    if not selected:
        selected.extend(path for path in tracked_paths if path in content)
    return tuple(dict.fromkeys(selected))[:10]


def _pack_blocks(
    tokenizer: QwenTokenCounter,
    blocks: Sequence[tuple[str, str]],
    budget: int,
) -> tuple[str, tuple[str, ...], int]:
    selected: list[str] = []
    paths: list[str] = []
    used = 0
    for path, block in blocks:
        count = tokenizer.count(block)
        if count > budget and not selected:
            block = block[: budget * 3]
            count = tokenizer.count(block)
        if used + count > budget:
            continue
        selected.append(block)
        paths.append(path)
        used += count
    return "\n".join(selected), tuple(paths), used


def _repository_index(
    retrieval: TaskRetrieval,
    tracked_paths: Sequence[str],
    tokenizer: QwenTokenCounter,
) -> tuple[str, int]:
    blocks: list[tuple[str, str]] = []
    for path in sorted(tracked_paths):
        symbols = retrieval.graph.by_path.get(path, ())
        names = ", ".join(item.name for item in symbols[:30])
        line = path if not names else f"{path} :: {names}"
        blocks.append((path, line + "\n"))
    text, _, tokens = _pack_blocks(tokenizer, blocks, AGENTLESS_INDEX_TOKENS)
    return text, tokens


def _source_context(
    tree: Path,
    paths: Sequence[str],
    tokenizer: QwenTokenCounter,
) -> tuple[str, tuple[str, ...], int]:
    blocks: list[tuple[str, str]] = []
    for path in paths:
        target = tree / path
        if target.is_file():
            blocks.append(
                (
                    path,
                    f"\n--- FILE: {path} ---\n"
                    + target.read_text(encoding="utf-8", errors="replace")
                    + "\n",
                )
            )
    return _pack_blocks(tokenizer, blocks, AGENTLESS_SOURCE_TOKENS)


def run_agentless_cell(
    root: Path,
    repository: Path,
    experiment: ExperimentSpec,
    task: TaskSpec,
    system: AgentSystemSpec,
    model: ModelSpec,
    embedding: EmbeddingSpec,
    retrieval: TaskRetrieval,
    residency: LMStudioResidencyManager,
    server: LMStudioServer,
    tokenizer: QwenTokenCounter,
    revision: str,
    seed: int = 0,
    repetition: int = 0,
) -> dict[str, Any]:
    if system.system_id != "A001":
        raise Study2ExperimentError("agentless cell requires A001")
    identity = _identity(experiment, task, system, model, revision, seed, repetition)
    completed = load_completed_or_archive_incomplete(root / "results", identity)
    if completed is not None:
        return completed

    initial_transition = residency.ensure_exclusive(
        model.expected_inference_key, model.context_length
    )
    client = LMStudioClient(model, timeout_seconds=experiment.timeout_seconds)
    _, resolved = client.resolve()
    suffixes = {"go": (".go",), "python": (".py",)}
    tracked_paths = GitSnapshot(repository).tracked_paths(
        task.base_commit, suffixes[task.language]
    )
    resolved_model = {
        "agent_model": asdict(model),
        "embedding_model": asdict(embedding),
        "agent_runtime": resolved.to_dict(),
        "tokenizer_path": str(tokenizer.path),
        "tokenizer_sha256": tokenizer.sha256,
        "initial_residency_transition": initial_transition.to_dict(),
    }

    with isolated_git_tree(repository, task.base_commit, task.repository_url) as tree, EventWriter(
        root / "results", identity, asdict(system), resolved_model
    ) as writer:
        workspace = AgentWorkspace(tree, tracked_paths, task, system.max_test_runs)
        transitions: list[dict[str, Any]] = [initial_transition.to_dict()]
        responses: list[dict[str, Any]] = []
        stage_records: list[dict[str, Any]] = []
        protocol_violations: list[str] = []
        model_elapsed = 0.0
        selected_paths: tuple[str, ...] = ()
        patch = ""
        finished_reason = "completed_three_stages"
        cell_started = time.monotonic()

        def call_stage(name: str, prompt: str, max_tokens: int) -> str:
            nonlocal model_elapsed
            transition = residency.ensure_exclusive(
                model.expected_inference_key, model.context_length
            )
            transitions.append(transition.to_dict())
            writer.emit(
                "resource_sample",
                {"kind": "model_residency_transition", **transition.to_dict()},
            )
            messages = [
                {
                    "role": "system",
                    "content": (
                        "You are one stage in a controlled Agentless-style coding "
                        "pipeline. Follow the requested output schema exactly. Hidden "
                        "tests and gold changes are unavailable."
                    ),
                },
                {"role": "user", "content": prompt},
            ]
            started = time.monotonic()
            try:
                _, active = client.resolve()
                response = client.chat_completions(
                    active.inference_key,
                    messages,
                    max_tokens=max_tokens,
                    seed=seed,
                )
            except LMStudioTransportError as first_error:
                recovery = server.ensure_running()
                writer.emit(
                    "resource_sample",
                    {"kind": "server_recovery", "error": str(first_error), "recovery": recovery},
                )
                transition = residency.ensure_exclusive(
                    model.expected_inference_key, model.context_length
                )
                transitions.append(transition.to_dict())
                _, active = client.resolve()
                response = client.chat_completions(
                    active.inference_key,
                    messages,
                    max_tokens=max_tokens,
                    seed=seed,
                )
            elapsed = time.monotonic() - started
            model_elapsed += elapsed
            responses.append(response)
            content = _assistant_content(response)
            record = {
                "stage": name,
                "prompt": prompt,
                "response": content,
                "elapsed_seconds": elapsed,
                "usage": response.get("usage", {}),
                "prompt_tokens_local": tokenizer.count(prompt),
            }
            stage_records.append(record)
            turn = len(stage_records)
            writer.write_artifact(
                f"model_response_{turn:02d}.json",
                json.dumps(response, indent=2, sort_keys=True) + "\n",
            )
            writer.emit(
                "model_call",
                {
                    "turn": turn,
                    "stage": name,
                    "elapsed_seconds": elapsed,
                    "usage": response.get("usage", {}),
                    "input_conversation_tokens": record["prompt_tokens_local"],
                },
            )
            return content

        writer.emit(
            "run_started",
            {
                "confirmatory": True,
                "blinded": True,
                "task_config_hash": task.config_hash,
                "stages": ["file_localization", "line_localization", "repair"],
                "budgets": {
                    "model_calls": system.model_calls,
                    "tool_calls": system.max_tool_calls,
                    "test_runs": system.max_test_runs,
                    "timeout_seconds": experiment.timeout_seconds,
                },
            },
        )
        index_tokens = 0
        source_tokens = 0
        try:
            index, index_tokens = _repository_index(retrieval, tracked_paths, tokenizer)
            stage1 = call_stage(
                "file_localization",
                f"""ISSUE:
{task.statement}

REPOSITORY FILE/SYMBOL INDEX:
{index}

Select at most 10 likely production files. Return only JSON:
{{"files": ["path/from/index"], "rationale": "brief"}}""",
                max_tokens=2_048,
            )
            selected_paths = _select_files(stage1, tracked_paths)
            if not selected_paths:
                protocol_violations.append("A001 file localization selected no valid paths")
                finished_reason = "file_localization_failure"
                raise ValueError("no valid localized files")
            source, included_paths, source_tokens = _source_context(
                tree, selected_paths, tokenizer
            )
            selected_paths = included_paths
            stage2 = call_stage(
                "line_localization",
                f"""ISSUE:
{task.statement}

CANDIDATE SOURCE:
{source}

Identify the exact functions or line regions that require change. Return only JSON:
{{"locations": [{{"path": "...", "symbol_or_lines": "...", "reason": "..."}}]}}""",
                max_tokens=2_048,
            )
            stage3 = call_stage(
                "repair",
                f"""ISSUE:
{task.statement}

LOCALIZATION:
{stage2}

CANDIDATE SOURCE:
{source}

Return only a standard unified diff. Modify production files only, do not add tests,
and make the smallest correct change.""",
                max_tokens=model.max_tokens,
            )
            patch = extract_unified_diff(
                {"choices": [{"message": {"content": stage3}}]}
            )
            apply_result = workspace.apply_patch(patch)
            writer.emit("edit", apply_result)
            if not apply_result["accepted"]:
                finished_reason = "patch_apply_failure"
            elif system.max_test_runs:
                command = task.pass_to_pass_tests[0]
                public_test = workspace.run_tests(command)
                writer.emit("test_run", public_test)
        except (ValueError, PatchOutputError, json.JSONDecodeError) as exc:
            protocol_violations.append(f"A001 protocol: {exc}")
            if not patch:
                finished_reason = "protocol_failure"

        final_patch = workspace.final_patch()
        if final_patch:
            validation = validate_generated_patch(
                root,
                repository,
                task,
                final_patch,
                preserve_git_metadata=True,
            )
        else:
            validation = _failure_validation("empty_patch")
        edited_paths = tuple(sorted(workspace.edited_paths))
        usage = _usage_totals(responses)
        elapsed = time.monotonic() - cell_started
        final = {
            "run_id": identity.run_id,
            "experiment_id": experiment.experiment_id,
            "task_id": task.task_id,
            "harness_id": system.system_id,
            "resolved_at_1": validation["resolved_at_1"],
            "failure_stage": validation["failure_stage"],
            "patch_applied": bool(
                validation.get("model_patch_apply")
                and validation["model_patch_apply"].get("returncode") == 0
            ),
            "fail_to_pass": validation["fail_to_pass"],
            "pass_to_pass": validation["pass_to_pass"],
            "modified_files": edited_paths,
            "localization_metrics": retrieval_metrics(edited_paths, task.gold_files),
            "search_localization_metrics": retrieval_metrics(
                selected_paths, task.gold_files
            ),
            "read_localization_metrics": retrieval_metrics(
                selected_paths, task.gold_files
            ),
            "finished_reason": finished_reason,
            "finish_summary": "",
            "protocol_violations": protocol_violations,
            "model_calls": len(responses),
            "tool_calls": 0,
            "tool_counts": {},
            "test_runs": len(workspace.test_runs),
            "usage": usage,
            "elapsed_seconds": elapsed,
            "model_elapsed_seconds": model_elapsed,
            "model_switch_count": sum(
                not item.get("reused") for item in transitions
            ),
            "model_switch_seconds": sum(
                float(item["elapsed_seconds"])
                for item in transitions
                if not item.get("reused")
            ),
            "peak_process_rss_platform_units": resource.getrusage(
                resource.RUSAGE_SELF
            ).ru_maxrss,
            "dense_index_stats": retrieval.dense_index_stats,
            "agentless_index_tokens": index_tokens,
            "agentless_source_tokens": source_tokens,
            "trajectory_sha256": sha256(
                json.dumps(stage_records, sort_keys=True).encode()
            ).hexdigest(),
            "patch_sha256": sha256(final_patch.encode()).hexdigest()
            if final_patch
            else None,
            "residency_transitions": transitions,
            "test_results": validation["tests"],
        }
        writer.write_artifact(
            "messages.json", json.dumps(stage_records, indent=2, sort_keys=True) + "\n"
        )
        writer.write_artifact("model.patch", final_patch)
        writer.write_artifact(
            "validation.json", json.dumps(validation, indent=2, sort_keys=True) + "\n"
        )
        writer.write_artifact(
            "final_metrics.json", json.dumps(final, indent=2, sort_keys=True) + "\n"
        )
        writer.emit(
            "run_finished",
            {
                "status": "completed",
                "resolved_at_1": validation["resolved_at_1"],
                "failure_stage": validation["failure_stage"],
                "finished_reason": finished_reason,
            },
        )
        return final


def _repository_for_task(
    repositories: dict[str, RepositorySpec],
    task: TaskSpec,
) -> RepositorySpec:
    matches = [
        repository
        for repository in repositories.values()
        if repository.repository_url == task.repository_url
    ]
    if len(matches) != 1:
        raise Study2ExperimentError(
            f"{task.task_id} has no unique repository registry entry"
        )
    return matches[0]


def _write_progress(
    root: Path,
    experiment: ExperimentSpec,
    revision: str,
    rows: Sequence[dict[str, Any]],
    task_summaries: Sequence[dict[str, Any]],
) -> Path:
    path = root / "results" / "reports" / f"{experiment.experiment_id}_progress.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "experiment_id": experiment.experiment_id,
                "code_revision": revision,
                "completed_cells": len(rows),
                "resolved_cells": sum(bool(item["resolved_at_1"]) for item in rows),
                "task_summaries": task_summaries,
                "rows": rows,
            },
            indent=2,
            sort_keys=True,
        )
        + "\n",
        encoding="utf-8",
    )
    return path


def run_study2_experiment(
    root: Path,
    experiment_id: str = "E08",
    task_filter: set[str] | None = None,
    treatment_filter: set[str] | None = None,
    model_filter: set[str] | None = None,
    stop_server_when_complete: bool = True,
) -> dict[str, Any]:
    revision = research_code_revision(root)
    experiment = load_experiments(root)[experiment_id]
    if experiment.mode != "study2_live_agent":
        raise Study2ExperimentError("Study 2 runner requires mode=study2_live_agent")
    harness_catalog = load_harnesses(root)
    system_catalog = load_agent_systems(root)
    model_catalog = load_models(root)
    repository_catalog = load_repositories(root)
    task_catalog = load_tasks(root)
    embedding = load_embeddings(root)[experiment.embedding_id]
    split = load_task_split(
        root / "tasks" / "splits" / f"{experiment.task_split}.txt"
    )
    tasks = [
        task_catalog[item]
        for item in split
        if task_filter is None or item in task_filter
    ]
    harnesses = [
        harness_catalog[item]
        for item in experiment.harness_ids
        if treatment_filter is None or item in treatment_filter
    ]
    systems = [
        system_catalog[item]
        for item in experiment.agent_system_ids
        if treatment_filter is None or item in treatment_filter
    ]
    models = [
        model_catalog[item]
        for item in experiment.model_ids
        if model_filter is None or item in model_filter
    ]
    if not tasks or not models or not (harnesses or systems):
        raise Study2ExperimentError("Study 2 filters selected an empty execution block")
    if any(task.validation_status != "end_to_end_ready" for task in tasks):
        raise Study2ExperimentError("Study 2 includes a task without hidden-test validation")

    server = LMStudioServer(port=1234)
    residency = LMStudioResidencyManager(
        models[0].base_url,
        models[0].api_token_env,
        timeout_seconds=experiment.timeout_seconds,
    )
    embedding_client = LMStudioEmbeddingClient(
        embedding, timeout_seconds=experiment.timeout_seconds
    )
    cache_path = root / "indexes" / "embeddings" / f"{embedding.config_hash}.sqlite3"
    treatments: list[tuple[str, Any]] = [
        *((item.harness_id, item) for item in harnesses),
        *((item.system_id, item) for item in systems),
    ]
    rows: list[dict[str, Any]] = []
    task_summaries: list[dict[str, Any]] = []
    runtime = _RuntimeLease(server, residency, stop_server_when_complete)
    with runtime as server_state, SQLiteEmbeddingCache(cache_path, embedding) as cache:
        for task_index, task in enumerate(tasks):
            repository_spec = _repository_for_task(repository_catalog, task)
            repository = (root / repository_spec.local_path).resolve()
            snapshot = GitSnapshot(repository)
            snapshot.verify_commit(task.base_commit)
            index_transition = residency.ensure_exclusive(
                embedding.model_key, embedding.loaded_context_length
            )
            embedding_client.resolve()
            index_started = time.monotonic()
            retrieval = _build_task_retrieval(
                snapshot, task, embedding, embedding_client, cache
            )
            index_elapsed = time.monotonic() - index_started
            model_order = models if task_index % 2 == 0 else list(reversed(models))
            offset = task_index % len(treatments)
            treatment_order = treatments[offset:] + treatments[:offset]
            task_rows: list[dict[str, Any]] = []
            for model in model_order:
                tokenizer = tokenizer_for(model)
                for treatment_id, treatment in treatment_order:
                    if treatment_id.startswith("H"):
                        row = run_live_agent_cell(
                            root,
                            repository,
                            experiment,
                            task,
                            treatment,
                            model,
                            embedding,
                            retrieval,
                            residency,
                            server,
                            tokenizer,
                            revision,
                            seed=experiment.seeds[0],
                            repetition=0,
                            preserve_git_metadata=True,
                        )
                    elif treatment_id == "A001":
                        row = run_agentless_cell(
                            root,
                            repository,
                            experiment,
                            task,
                            treatment,
                            model,
                            embedding,
                            retrieval,
                            residency,
                            server,
                            tokenizer,
                            revision,
                            seed=experiment.seeds[0],
                            repetition=0,
                        )
                    elif treatment_id == "A002":
                        row = run_live_agent_cell(
                            root,
                            repository,
                            experiment,
                            task,
                            harness_catalog["H000"],
                            model,
                            embedding,
                            retrieval,
                            residency,
                            server,
                            tokenizer,
                            revision,
                            agent_system=treatment,
                            seed=experiment.seeds[0],
                            repetition=0,
                            preserve_git_metadata=True,
                        )
                    else:
                        raise Study2ExperimentError(
                            f"unsupported Study 2 treatment: {treatment_id}"
                        )
                    rows.append(row)
                    task_rows.append(row)
                    _write_progress(root, experiment, revision, rows, task_summaries)
            summary = {
                "task_id": task.task_id,
                "repository_id": repository_spec.repository_id,
                "language": task.language,
                "model_order": [item.model_id for item in model_order],
                "treatment_order": [item[0] for item in treatment_order],
                "embedding_index_transition": index_transition.to_dict(),
                "index_elapsed_seconds": index_elapsed,
                "dense_index_stats": retrieval.dense_index_stats,
                "cells": len(task_rows),
                "resolved": sum(bool(item["resolved_at_1"]) for item in task_rows),
            }
            task_summaries.append(summary)
            _write_progress(root, experiment, revision, rows, task_summaries)

    report = {
        "experiment_id": experiment.experiment_id,
        "code_revision": revision,
        "server_lifecycle": server_state,
        "server_stop": runtime.stop_state,
        "run_count": len(rows),
        "resolved_count": sum(bool(item["resolved_at_1"]) for item in rows),
        "task_summaries": task_summaries,
        "final_residency_transition": runtime.final_transition,
        "cleanup_errors": runtime.cleanup_errors,
        "rows": rows,
    }
    report_dir = root / "results" / "reports"
    report_dir.mkdir(parents=True, exist_ok=True)
    report_path = report_dir / f"E08_{revision[:12]}_{int(time.time())}.json"
    report_path.write_text(
        json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )
    return {**report, "report_path": str(report_path)}


def run_study2_reliability(
    root: Path,
    manifest_path: Path | None = None,
    stop_server_when_complete: bool = True,
) -> dict[str, Any]:
    """Run the 24 frozen non-oracle cells under the stochastic sensitivity profile."""

    revision = research_code_revision(root)
    experiment = load_experiments(root)["E08"]
    path = manifest_path or root / "configs" / "reliability" / "E08_repeat_cells.json"
    value = json.loads(path.read_text(encoding="utf-8"))
    cells = value.get("cells", [])
    seeds = tuple(int(item) for item in value.get("seeds", []))
    temperature = float(value.get("temperature", -1.0))
    top_p = float(value.get("top_p", -1.0))
    if len(cells) != 24 or seeds != (0, 1, 2):
        raise Study2ExperimentError(
            "reliability manifest must freeze 24 cells and seeds 0,1,2"
        )
    if temperature != 0.2 or top_p != 1.0:
        raise Study2ExperimentError(
            "reliability manifest must freeze temperature=0.2 and top_p=1.0"
        )

    harness_catalog = load_harnesses(root)
    system_catalog = load_agent_systems(root)
    model_catalog = load_models(root)
    repository_catalog = load_repositories(root)
    task_catalog = load_tasks(root)
    embedding = load_embeddings(root)[experiment.embedding_id]
    server = LMStudioServer(port=1234)
    residency = LMStudioResidencyManager(
        model_catalog["M002"].base_url,
        model_catalog["M002"].api_token_env,
        timeout_seconds=experiment.timeout_seconds,
    )
    embedding_client = LMStudioEmbeddingClient(
        embedding, timeout_seconds=experiment.timeout_seconds
    )
    cache_path = root / "indexes" / "embeddings" / f"{embedding.config_hash}.sqlite3"
    rows: list[dict[str, Any]] = []
    grouped: dict[str, list[dict[str, str]]] = {}
    for raw in cells:
        if not isinstance(raw, dict):
            raise Study2ExperimentError("reliability cell must be an object")
        cell = {key: str(raw[key]) for key in ("task_id", "treatment_id", "model_id")}
        grouped.setdefault(cell["task_id"], []).append(cell)

    runtime = _RuntimeLease(server, residency, stop_server_when_complete)
    with runtime as server_state, SQLiteEmbeddingCache(cache_path, embedding) as cache:
        for task_id, task_cells in grouped.items():
            task = task_catalog[task_id]
            repository_spec = _repository_for_task(repository_catalog, task)
            repository = (root / repository_spec.local_path).resolve()
            snapshot = GitSnapshot(repository)
            index_transition = residency.ensure_exclusive(
                embedding.model_key, embedding.loaded_context_length
            )
            embedding_client.resolve()
            retrieval = _build_task_retrieval(
                snapshot, task, embedding, embedding_client, cache
            )
            for cell in task_cells:
                treatment_id = cell["treatment_id"]
                base_model = model_catalog[cell["model_id"]]
                model = replace(base_model, temperature=temperature, top_p=top_p)
                tokenizer = tokenizer_for(model)
                for seed in seeds:
                    if treatment_id.startswith("H"):
                        row = run_live_agent_cell(
                            root,
                            repository,
                            experiment,
                            task,
                            harness_catalog[treatment_id],
                            model,
                            embedding,
                            retrieval,
                            residency,
                            server,
                            tokenizer,
                            revision,
                            seed=seed,
                            repetition=1,
                            preserve_git_metadata=True,
                        )
                    elif treatment_id == "A001":
                        row = run_agentless_cell(
                            root,
                            repository,
                            experiment,
                            task,
                            system_catalog[treatment_id],
                            model,
                            embedding,
                            retrieval,
                            residency,
                            server,
                            tokenizer,
                            revision,
                            seed=seed,
                            repetition=1,
                        )
                    elif treatment_id == "A002":
                        row = run_live_agent_cell(
                            root,
                            repository,
                            experiment,
                            task,
                            harness_catalog["H000"],
                            model,
                            embedding,
                            retrieval,
                            residency,
                            server,
                            tokenizer,
                            revision,
                            agent_system=system_catalog[treatment_id],
                            seed=seed,
                            repetition=1,
                            preserve_git_metadata=True,
                        )
                    else:
                        raise Study2ExperimentError(
                            f"unknown reliability treatment: {treatment_id}"
                        )
                    rows.append(row)
                    progress = root / "results" / "reports" / "E08_reliability_progress.json"
                    progress.parent.mkdir(parents=True, exist_ok=True)
                    progress.write_text(
                        json.dumps(
                            {
                                "schema_version": 1,
                                "code_revision": revision,
                                "completed_cells": len(rows),
                                "planned_cells": 72,
                                "generation_profile": {
                                    "temperature": temperature,
                                    "top_p": top_p,
                                    "seeds": seeds,
                                },
                                "rows": rows,
                                "last_index_transition": index_transition.to_dict(),
                            },
                            indent=2,
                            sort_keys=True,
                        )
                        + "\n",
                        encoding="utf-8",
                    )

    report = {
        "experiment_id": "E08_reliability",
        "code_revision": revision,
        "server_lifecycle": server_state,
        "server_stop": runtime.stop_state,
        "run_count": len(rows),
        "resolved_count": sum(bool(item["resolved_at_1"]) for item in rows),
        "generation_profile": {
            "temperature": temperature,
            "top_p": top_p,
            "seeds": seeds,
        },
        "final_residency_transition": runtime.final_transition,
        "cleanup_errors": runtime.cleanup_errors,
        "rows": rows,
    }
    report_path = (
        root
        / "results"
        / "reports"
        / f"E08_reliability_{revision[:12]}_{int(time.time())}.json"
    )
    report_path.write_text(
        json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )
    return {**report, "report_path": str(report_path)}