Capstone: Proving the Harness with Offline Evidence

The capstone integrates the mechanisms built across the course: an async agent loop, typed tools, context and session state, safety policy, hooks, sub-agent boundaries, and an MCP bridge. Integration is not a victory lap. It is the point at which a local guarantee can fail because another layer ignored it.

The chapter therefore makes one modest claim falsifiable: the offline harness satisfies a stated set of contracts on deterministic fixtures, and each contract can be made to fail by removing its corresponding control. No model call, MCP server, subprocess, or secret is required. The evidence is narrower than a production benchmark, but it is repeatable and inspectable. See the sub-agent patterns and MCP bridge chapters for the two external-authority boundaries being integrated here.

Evaluation contract

The capstone artifact is not just the final assistant string. A passing run must leave observable evidence for four questions:

• Task: did the scripted agent complete the requested repository change?
• Authority: did tools, sub-agents, and external schemas retain their boundaries?
• Control: did safety, context, and loop controls catch the planted failures?
• Evidence: can a judge score those claims from events, files, and structured observations?

The offline runner below uses the real package classes wherever a boundary exists. Fakes appear only at the model and MCP transport edges, where deterministic inputs are more informative than a flaky live service.

Workspace and registry

Use .tmp/ rather than the operating system temporary directory so the fixture’s lifecycle is visible to the repository workflow and remains gitignored. The registry is the same default registry used by Session, including the delegation tool introduced in Chapter 10.

import asyncio
import shutil
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from agent_harness import (
    Agent,
    ApprovalManager,
    ApprovalPolicy,
    Config,
    ContextManager,
    LoopDetector,
    MCPToolAdapter,
    ModelConfig,
    Session,
    SubAgentParams,
    SubAgentTool,
    TokenUsage,
    ToolInvocation,
    ToolRegistry,
    create_default_registry,
)
from agent_harness.events import (
    AgentEventType,
    StreamEvent,
    StreamEventType,
    TextDelta,
    ToolCall,
)
from agent_harness.mcp.bridge import mcp_tool_to_openai_schema
from agent_harness.tools.files import ReadFileTool
from agent_harness.tools.shell import ShellTool


fixture_root = Path(".tmp") / "agent-harness-capstone-fixture"
if fixture_root.exists():
    shutil.rmtree(fixture_root)
fixture_root.mkdir(parents=True)
(fixture_root / "README.md").write_text("offline fixture\n", encoding="utf-8")

config = Config(
    cwd=fixture_root,
    approval=ApprovalPolicy.YOLO,
    max_turns=3,
)
registry = create_default_registry(config)
print(
    {
        "workspace": str(fixture_root),
        "tool_count": len(registry.get_tools()),
        "tools": [tool.name for tool in registry.get_tools()],
    }
)
assert "run_sub_agent" in {tool.name for tool in registry.get_tools()}
assert "write_file" in {tool.name for tool in registry.get_tools()}

{'workspace': '.tmp/agent-harness-capstone-fixture', 'tool_count': 9, 'tools': ['read_file', 'write_file', 'edit', 'shell', 'list_dir', 'grep', 'glob', 'memory', 'run_sub_agent']}

The fixture is now attached to a normal Config and ToolRegistry. YOLO is explicit here because the current Agent loop dispatches through the registry but does not itself call an approval callback. The safety policy is still evaluated separately below; naming that limitation prevents the integration demo from overstating what the current loop enforces.

Scripted agent run

The fake client emits the same StreamEvent objects that LLMClient produces. On turn one it requests write_file; on turn two it returns a final answer. The agent, session, registry dispatch, file tool, event stream, and stored usage are real. Only the model decision is fixed.

class ScriptedClient:
    def __init__(self, turns: list[list[StreamEvent]]):
        self.turns = turns
        self.calls = 0

    async def chat_completion(self, messages, tools=None, stream=True):
        events = self.turns[self.calls]
        self.calls += 1
        for event in events:
            yield event


scripted = ScriptedClient(
    [
        [
            StreamEvent(
                type=StreamEventType.TEXT_DELTA,
                text_delta=TextDelta("I will create the report. "),
            ),
            StreamEvent(
                type=StreamEventType.TOOL_CALL_COMPLETE,
                tool_call=ToolCall(
                    call_id="write-1",
                    name="write_file",
                    arguments={"path": "report.txt", "content": "verified offline\n"},
                ),
            ),
            StreamEvent(
                type=StreamEventType.MESSAGE_COMPLETE,
                usage=TokenUsage(prompt_tokens=42, completion_tokens=12, total_tokens=54),
            ),
        ],
        [
            StreamEvent(
                type=StreamEventType.TEXT_DELTA,
                text_delta=TextDelta("Report created and verified."),
            ),
            StreamEvent(
                type=StreamEventType.MESSAGE_COMPLETE,
                usage=TokenUsage(prompt_tokens=58, completion_tokens=7, total_tokens=65),
            ),
        ],
    ]
)

session = Session(config, client=scripted, registry=registry)
agent = Agent(config, session=session)


async def collect_events():
    return [event async for event in agent.run("Create report.txt with the fixture result.")]


events = await collect_events()
event_types = [event.type.value for event in events]
print(event_types)
print("final text:", [event.data["content"] for event in events if event.type == AgentEventType.TEXT_COMPLETE])
assert scripted.calls == 2
assert (fixture_root / "report.txt").read_text(encoding="utf-8") == "verified offline\n"
assert AgentEventType.TOOL_CALL_COMPLETE.value in event_types
assert event_types[-1] == AgentEventType.AGENT_END.value

['agent_start', 'text_delta', 'text_complete', 'tool_call_start', 'tool_call_complete', 'text_delta', 'text_complete', 'agent_end']
final text: ['I will create the report. ', 'Report created and verified.']

The successful artifact is stronger than a printed “done”: the file content, two model calls, a tool-complete event, and a terminal agent event all agree. The session accumulated 54 + 65 = 119 total tokens across turns. The event stream is an observability surface that later evaluation code can score without re-running the model.

Journal replay

The session keeps an append-only journal of user messages, assistant tool calls, tool results, turns, and usage. Replay provides a second route to the same state. This is useful in evaluation because a stored trajectory can be inspected or re-judged without contacting the model again.

print("journal:", [(entry["type"], sorted(entry)) for entry in session.journal])
replayed = Session.replay(
    session.journal,
    config,
    client=scripted,
    registry=registry,
)
print(
    {
        "messages": len(replayed.messages),
        "turn_count": replayed.turn_count,
        "total_tokens": replayed.total_usage.total_tokens,
    }
)
assert replayed.messages == session.messages
assert replayed.turn_count == session.turn_count == 2
assert replayed.total_usage == session.total_usage
assert replayed.journal == session.journal

journal: [('user_message', ['content', 'type']), ('turn', ['count', 'type']), ('usage', ['type', 'usage']), ('assistant_message', ['content', 'tool_calls', 'type']), ('tool_result', ['content', 'is_error', 'tool_call_id', 'type']), ('turn', ['count', 'type']), ('usage', ['type', 'usage']), ('assistant_message', ['content', 'tool_calls', 'type'])]
{'messages': 5, 'turn_count': 2, 'total_tokens': 119}

Replay establishes state equivalence, not semantic correctness. A journal can faithfully preserve a bad tool call. That is why the suite below scores safety, authority boundaries, and bounded termination in addition to task completion.

Held-out fixtures

A suite is useful only when its cases are fixed before an ablation is run. These seven cases cover the integrated loop and one boundary from each hardening layer. They are small enough to understand individually, but they exercise files, events, schemas, async dispatch, safety classification, hook validation, pruning, and loop detection through the project APIs.

@dataclass(frozen=True)
class EvalCase:
    case_id: str
    family: str
    description: str
    dimensions: tuple[str, ...]


@dataclass(frozen=True)
class FixtureObservation:
    case_id: str
    checks: dict[str, bool]
    details: dict[str, object]


SUITE_CASES = (
    EvalCase(
        "agent-loop",
        "integration",
        "complete a scripted write and expose its lifecycle events",
        ("task_success", "observability"),
    ),
    EvalCase(
        "path-boundary",
        "safety",
        "reject a file path outside the workspace and classify a dangerous shell command",
        ("safety",),
    ),
    EvalCase(
        "hook-contract",
        "extension",
        "reject an ambiguous lifecycle hook before external code can run",
        ("hooks",),
    ),
    EvalCase(
        "delegation-boundary",
        "authority",
        "keep an investigator read-scoped and bounded",
        ("delegation",),
    ),
    EvalCase(
        "mcp-bridge",
        "protocol",
        "translate and invoke a namespaced external tool offline",
        ("mcp",),
    ),
    EvalCase(
        "context-budget",
        "control",
        "prune old context while retaining the system prompt and tail",
        ("context",),
    ),
    EvalCase(
        "loop-stop",
        "control",
        "detect a repeated action before it consumes unbounded turns",
        ("bounded_termination",),
    ),
)

print([(case.case_id, case.dimensions) for case in SUITE_CASES])
assert len(SUITE_CASES) == 7
assert len({case.case_id for case in SUITE_CASES}) == len(SUITE_CASES)

[('agent-loop', ('task_success', 'observability')), ('path-boundary', ('safety',)), ('hook-contract', ('hooks',)), ('delegation-boundary', ('delegation',)), ('mcp-bridge', ('mcp',)), ('context-budget', ('context',)), ('loop-stop', ('bounded_termination',))]

Each case names the dimension it owns. This prevents a single aggregate number from hiding which mechanism failed. The fixture suite is held out from the model because it does not ask a model to discover the expected answer; it asks the harness to preserve explicit invariants.

Boundary fixtures

The runner below is longer than a unit test because it records details alongside booleans. A failed row should explain what was observed, not merely lower a score. The variant argument removes one named control for ablation; the full path always uses the actual project mechanism.

eval_root = Path(".tmp") / "agent-harness-eval-fixtures"
if eval_root.exists():
    shutil.rmtree(eval_root)
eval_root.mkdir(parents=True)


def fresh_workspace(name: str) -> Path:
    workspace = eval_root / name
    if workspace.exists():
        shutil.rmtree(workspace)
    workspace.mkdir(parents=True)
    (workspace / "README.md").write_text("fixture\n", encoding="utf-8")
    return workspace


async def run_agent_fixture() -> FixtureObservation:
    workspace = fresh_workspace("agent-loop")
    fixture_config = Config(
        cwd=workspace,
        approval=ApprovalPolicy.YOLO,
        max_turns=3,
    )
    fixture_registry = create_default_registry(fixture_config)
    fixture_client = ScriptedClient(
        [
            [
                StreamEvent(
                    type=StreamEventType.TOOL_CALL_COMPLETE,
                    tool_call=ToolCall(
                        call_id="fixture-write",
                        name="write_file",
                        arguments={"path": "result.txt", "content": "pass\n"},
                    ),
                ),
                StreamEvent(
                    type=StreamEventType.MESSAGE_COMPLETE,
                    usage=TokenUsage(total_tokens=10),
                ),
            ],
            [
                StreamEvent(
                    type=StreamEventType.TEXT_DELTA,
                    text_delta=TextDelta("fixture complete"),
                ),
                StreamEvent(type=StreamEventType.MESSAGE_COMPLETE),
            ],
        ]
    )
    fixture_session = Session(
        fixture_config,
        client=fixture_client,
        registry=fixture_registry,
    )
    fixture_agent = Agent(fixture_config, session=fixture_session)

    async def collect_fixture_events():
        return [event async for event in fixture_agent.run("Create result.txt.")]

    events = await collect_fixture_events()
    event_types = [event.type.value for event in events]
    tool_results = [
        event.data for event in events if event.type == AgentEventType.TOOL_CALL_COMPLETE
    ]
    checks = {
        "task_success": (
            (workspace / "result.txt").read_text(encoding="utf-8") == "pass\n"
            and bool(tool_results)
            and tool_results[0]["success"]
        ),
        "observability": (
            event_types[0] == AgentEventType.AGENT_START.value
            and event_types[-1] == AgentEventType.AGENT_END.value
            and "tool_result" in {entry["type"] for entry in fixture_session.journal}
        ),
    }
    return FixtureObservation(
        "agent-loop",
        checks,
        {"event_types": event_types, "journal_entries": len(fixture_session.journal)},
    )


async def run_safety_fixture(variant: str) -> FixtureObservation:
    workspace = fresh_workspace("path-boundary")
    (workspace.parent / "outside.txt").write_text("outside fixture\n", encoding="utf-8")
    fixture_config = Config(cwd=workspace, approval=ApprovalPolicy.ON_REQUEST)
    fixture_registry = ToolRegistry(fixture_config)
    fixture_registry.register(ReadFileTool(fixture_config))
    escaped = await fixture_registry.invoke(
        "read_file", {"path": "../outside.txt"}, workspace
    )
    approval = ApprovalManager(fixture_config)
    dangerous_tool = ShellTool(fixture_config)
    approval_required = (
        approval.needs_approval(dangerous_tool, {"command": "rm -rf build"})
        if variant != "no_safety_gate"
        else False
    )
    return FixtureObservation(
        "path-boundary",
        {"safety": not escaped.success and approval_required},
        {"path_error": escaped.error, "approval_required": approval_required},
    )


def run_delegation_fixture(variant: str) -> FixtureObservation:
    fixture_config = Config(cwd=Path.cwd(), max_turns=12)
    delegation_tool = SubAgentTool(fixture_config)
    params = SubAgentParams(
        task="List the files relevant to the issue and cite each path.",
        role="codebase_investigator",
        max_turns=5,
    )
    role_prompt, role_tools = delegation_tool._resolve_role(params)
    child_config = delegation_tool._build_child_config(params, role_prompt, role_tools)
    effective_tools = None if variant == "no_delegation_boundary" else child_config.allowed_tools
    bounded_and_read_only = (
        effective_tools is not None
        and set(effective_tools).isdisjoint({"write_file", "edit"})
        and child_config.max_turns == params.max_turns
    )
    return FixtureObservation(
        "delegation-boundary",
        {"delegation": bounded_and_read_only},
        {"effective_tools": effective_tools, "max_turns": child_config.max_turns},
    )

Remaining fixtures

The first three runners cover the integrated loop, permissions, and delegation. The remaining runners exercise hook configuration, MCP schema namespacing and dispatch, context pruning, and loop detection. Each runner accepts the same variant string, so removing one control changes only the observation owned by that control.

from types import SimpleNamespace

from agent_harness import HookConfig, HookTrigger


def run_hook_fixture(variant: str) -> FixtureObservation:
    rejected_ambiguous_hook = False
    if variant != "no_hook_validation":
        try:
            HookConfig(
                name="ambiguous",
                trigger=HookTrigger.BEFORE_TOOL,
                command="true",
                script="true",
            )
        except ValueError:
            rejected_ambiguous_hook = True
    return FixtureObservation(
        "hook-contract",
        {"hooks": rejected_ambiguous_hook},
        {"ambiguous_config_rejected": rejected_ambiguous_hook},
    )


class FakeMCPClient:
    async def call_tool(self, name: str, arguments: dict[str, Any]):
        content = [SimpleNamespace(text=f"{name}:{arguments['text']}")]
        return SimpleNamespace(is_error=False, content=content, data=None)


async def run_mcp_fixture(variant: str) -> FixtureObservation:
    workspace = fresh_workspace("mcp-bridge")
    fixture_config = Config(cwd=workspace)
    raw_tool = {
        "name": "echo",
        "description": "Echo one string without changing external state.",
        "inputSchema": {
            "type": "object",
            "properties": {"text": {"type": "string"}},
            "required": ["text"],
        },
    }
    translated = (
        mcp_tool_to_openai_schema("fixture", raw_tool)
        if variant != "no_mcp_namespace"
        else {
            "name": raw_tool["name"],
            "description": raw_tool["description"],
            "parameters": raw_tool["inputSchema"],
        }
    )
    adapter = MCPToolAdapter(
        server_name="fixture",
        mcp_tool_name="echo",
        description=raw_tool["description"],
        input_schema=raw_tool["inputSchema"],
        client=FakeMCPClient(),
        config=fixture_config,
    )
    fixture_registry = ToolRegistry(fixture_config)
    fixture_registry.register(adapter)
    result = await fixture_registry.invoke(
        adapter.name, {"text": "hello"}, workspace
    )
    namespaced = translated["name"] == adapter.name == "fixture__echo"
    return FixtureObservation(
        "mcp-bridge",
        {"mcp": result.success and result.output == "echo:hello" and namespaced},
        {
            "translated_name": translated["name"],
            "registered_name": adapter.name,
            "output": result.output,
        },
    )


def run_context_fixture(variant: str) -> FixtureObservation:
    workspace = fresh_workspace("context-budget")
    fixture_config = Config(
        cwd=workspace,
        model=ModelConfig(name="offline-fixture", context_window=220),
    )
    manager = ContextManager(
        fixture_config,
        compaction_threshold=0.6,
        pruning_threshold=0.8,
        keep_last=2,
    )
    messages = [
        {"role": "system", "content": "Never modify tests."},
        {"role": "tool", "content": "x" * 500, "tool_call_id": "old-call"},
        {"role": "assistant", "content": "y" * 500},
        {"role": "user", "content": "Keep this recent request."},
        {"role": "assistant", "content": "Keep this recent response."},
    ]
    managed = (
        manager.prune_messages(messages)
        if variant != "no_context_pruning"
        else list(messages)
    )
    protected = managed[0] == messages[0] and managed[-2:] == messages[-2:]
    old_context_removed = len(managed) < len(messages)
    return FixtureObservation(
        "context-budget",
        {"context": protected and old_context_removed},
        {
            "before_messages": len(messages),
            "after_messages": len(managed),
            "protected_regions_preserved": protected,
        },
    )


def run_loop_fixture(variant: str) -> FixtureObservation:
    detector = LoopDetector(max_repeats=3, window_size=6)
    for _ in range(3):
        detector.record("read_file", {"path": "missing.py"})
    detected = detector.is_looping() if variant != "no_loop_detector" else False
    return FixtureObservation(
        "loop-stop",
        {"bounded_termination": detected},
        {"detected": detected, "diagnostic": detector.get_loop_message()},
    )


full_boundary_observations = [
    run_hook_fixture("full"),
    await run_mcp_fixture("full"),
    run_context_fixture("full"),
    run_loop_fixture("full"),
]
print(
    {
        observation.case_id: observation.checks
        for observation in full_boundary_observations
    }
)
assert all(
    all(observation.checks.values())
    for observation in full_boundary_observations
)

{'hook-contract': {'hooks': True}, 'mcp-bridge': {'mcp': True}, 'context-budget': {'context': True}, 'loop-stop': {'bounded_termination': True}}

The MCP check requires both a successful call and a namespaced schema, so transport success cannot hide a collision-prone registration. The context check separately asserts preservation of the system message and recent tail. The hook case stays offline by testing HookConfig validation rather than launching user code; Chapter 9 already established the subprocess timeout contract.

These are narrow mechanism checks. They do not claim that a namespaced MCP description is free of prompt injection, that mechanical pruning preserves every fact, or that rejecting an ambiguous hook proves a future veto protocol correct.

Ablation matrix

Convert each fixture observation into the project evaluator’s EvalTask, Trajectory, and JudgeResult types. The conversion deliberately keeps task success and contract compliance separate: a failed boundary produces a failed trajectory even when the Python runner itself completed. Every variant writes its trajectories under a distinct directory, then the Scorecard aggregates the same fixed task suite.

from IPython.display import Markdown, display

from agent_harness import EvalTask, Judge, Scorecard, TaskSuite, Trajectory, TrajectoryStore


CASE_BY_ID = {case.case_id: case for case in SUITE_CASES}
VARIANTS = (
    "full",
    "no_safety_gate",
    "no_hook_validation",
    "no_delegation_boundary",
    "no_mcp_namespace",
    "no_context_pruning",
    "no_loop_detector",
)


async def run_variant(variant: str) -> list[FixtureObservation]:
    return [
        await run_agent_fixture(),
        await run_safety_fixture(variant),
        run_hook_fixture(variant),
        run_delegation_fixture(variant),
        await run_mcp_fixture(variant),
        run_context_fixture(variant),
        run_loop_fixture(variant),
    ]


evaluation_tasks = [
    EvalTask(
        id=case.case_id,
        prompt=case.description,
        expected_text="PASS",
        category=case.family,
        metadata={"dimensions": list(case.dimensions)},
    )
    for case in SUITE_CASES
]
task_suite = TaskSuite(
    "agent-harness-offline-v1",
    evaluation_tasks,
    metadata={"fixture_only": True, "model_calls": 0},
)
artifact_root = Path(".tmp") / "agent-harness-capstone-artifacts"
artifact_root.mkdir(parents=True, exist_ok=True)
task_suite.save(artifact_root / "suite.json")

scorecards = {}
observations_by_variant = {}
for variant in VARIANTS:
    observations = await run_variant(variant)
    observations_by_variant[variant] = observations
    store = TrajectoryStore(artifact_root / "trajectories" / variant)
    results = []
    for observation in observations:
        case = CASE_BY_ID[observation.case_id]
        passed = all(observation.checks.values())
        trajectory = Trajectory(
            observation.case_id,
            events=[
                {
                    "type": "fixture_observation",
                    "data": {
                        "checks": observation.checks,
                        "details": observation.details,
                    },
                }
            ],
            response="PASS" if passed else "FAIL",
            status="completed" if passed else "failed",
            metadata={
                "success": passed,
                "variant": variant,
                "category": case.family,
            },
        )
        store.save(trajectory)
        task = task_suite[observation.case_id]
        results.append(Judge().evaluate(task, trajectory))
    scorecard = Scorecard(
        results,
        suite=task_suite.name,
        variant=variant,
        metadata={"offline": True},
    )
    scorecard.save(artifact_root / f"scorecard-{variant}.json")
    scorecards[variant] = scorecard

comparison = Scorecard.compare(scorecards)
table_lines = [
    "| Variant | Passed | Pass rate | Mean score |",
    "| --- | ---: | ---: | ---: |",
]
for variant in VARIANTS:
    metrics = comparison[variant]
    table_lines.append(
        f"| `{variant}` | {metrics['passed_tasks']}/{metrics['total_tasks']} "
        f"| {metrics['pass_rate']:.3f} | {metrics['mean_score']:.3f} |"
    )
display(Markdown("\n".join(table_lines)))

assert comparison["full"]["passed_tasks"] == len(SUITE_CASES)
for variant in VARIANTS[1:]:
    failed = [
        observation.case_id
        for observation in observations_by_variant[variant]
        if not all(observation.checks.values())
    ]
    assert len(failed) == 1, (variant, failed)

print("artifacts:", artifact_root)

Variant	Passed	Pass rate	Mean score
full	7/7	1.000	1.000
no_safety_gate	6/7	0.857	0.857
no_hook_validation	6/7	0.857	0.857
no_delegation_boundary	6/7	0.857	0.857
no_mcp_namespace	6/7	0.857	0.857
no_context_pruning	6/7	0.857	0.857
no_loop_detector	6/7	0.857	0.857

artifacts: .tmp/agent-harness-capstone-artifacts

The full harness passes all seven fixtures. Every one-control ablation fails exactly one owned case, while the other six remain green. This is the required causal sanity check: the evaluator can observe a control disappearing, and the failure is attributed to the matching dimension instead of reducing an unexplained aggregate score.

The score difference is intentionally coarse. With two equally weighted judge criteria, an ablated case contributes zero while every unaffected case contributes one. The useful evidence is the per-case failure and stored observation, not the third decimal place in the mean.

Reproducibility manifest

A result without an implementation identifier is not replayable. This offline run has no provider dashboard or decoding parameters, so record those fields explicitly as not applicable. Hash the harness source and effective configuration, version the evaluator protocol, and point to the suite and scorecards written above.

import hashlib
import json
import platform


source_files = sorted(
    Path("projects/agent-harness/src/agent_harness").rglob("*.py")
)
source_digest = hashlib.sha256()
for source_file in source_files:
    relative_source = source_file.resolve().relative_to(Path.cwd().resolve())
    source_digest.update(str(relative_source).encode())
    source_digest.update(source_file.read_bytes())

config_payload = json.dumps(
    config.model_dump(mode="json"),
    sort_keys=True,
    separators=(",", ":"),
)
manifest = {
    "suite": task_suite.name,
    "evaluator_version": "offline-contracts-v1",
    "model": "scripted/offline-v1",
    "decoding_parameters": None,
    "seeds": [0],
    "provider_cost_reconciliation": "not applicable: zero provider calls",
    "python": platform.python_version(),
    "harness_source_sha256": source_digest.hexdigest(),
    "config_sha256": hashlib.sha256(config_payload.encode()).hexdigest(),
    "variants": list(VARIANTS),
    "suite_path": "suite.json",
    "scorecard_paths": [f"scorecard-{variant}.json" for variant in VARIANTS],
}
manifest_path = artifact_root / "manifest.json"
manifest_path.write_text(
    json.dumps(manifest, indent=2, sort_keys=True) + "\n",
    encoding="utf-8",
)
print(json.dumps(manifest, indent=2, sort_keys=True))
assert len(manifest["harness_source_sha256"]) == 64
assert manifest_path.is_file()

{
  "config_sha256": "23e4ffbfd57420aab7c5955f30507960ca89da3664f411e7b65a28dc87aadf48",
  "decoding_parameters": null,
  "evaluator_version": "offline-contracts-v1",
  "harness_source_sha256": "5faba62fc88e080f945b507ded7f24d49c355c0cdbb30128a59cc886b6bd2224",
  "model": "scripted/offline-v1",
  "provider_cost_reconciliation": "not applicable: zero provider calls",
  "python": "3.14.6",
  "scorecard_paths": [
    "scorecard-full.json",
    "scorecard-no_safety_gate.json",
    "scorecard-no_hook_validation.json",
    "scorecard-no_delegation_boundary.json",
    "scorecard-no_mcp_namespace.json",
    "scorecard-no_context_pruning.json",
    "scorecard-no_loop_detector.json"
  ],
  "seeds": [
    0
  ],
  "suite": "agent-harness-offline-v1",
  "suite_path": "suite.json",
  "variants": [
    "full",
    "no_safety_gate",
    "no_hook_validation",
    "no_delegation_boundary",
    "no_mcp_namespace",
    "no_context_pruning",
    "no_loop_detector"
  ]
}

Capstone results and limits

The offline result proves that the current package composes on deterministic fixtures: the agent loop can perform a real file write and replay its journal; path and approval checks reject the planted hazards; hook configuration rejects an ambiguous contract; delegation preserves a read-only tool boundary and turn budget; MCP translation preserves a namespace through dispatch; context pruning protects the system message and recent tail; and the loop detector catches a repeated action. The persisted suite, trajectories, scorecards, and manifest make each observation inspectable without contacting a model.

It does not prove the broader production claim from the course plan. A held-out repository benchmark still needs development and held-out tasks with no shared files, at least three seeds per configuration, fixed live-model decoding parameters, wall-clock and provider cost reconciliation, calibrated LLM-judge prompts, sandbox escape attempts, and failure exemplars from trajectories longer than 50 calls. Those requirements cannot be manufactured by an offline notebook.

Treat this chapter as the deterministic base layer for that study. Clone a repository fixture per attempt, freeze suite.json before the first held-out run, store one trajectory per seed and variant, and keep the offline matrix as a preflight test. If an external evaluator cannot reproduce these seven rows first, a larger benchmark score is not trustworthy.