---
title: "Harness engineering: which controls an agent actually needs"
source: "https://www.matteovincenti.com/en/blog/agent-harness-engineering-controls"
language: en
author: "Matteo Vincenti"
published: 2026-10-24
category: "AI"
tags: ["AI agents", "Software quality", "Docker"]
---

# Harness engineering: which controls an agent actually needs

A Docker lab shows how an agent harness controls permissions, retries, budgets and evidence, and when to choose an existing runtime.

![A green mechanical arm holds a component over a checking jig beside a closed gate, with its key on the opposite side.](https://cdn.sanity.io/images/xrqkwry7/production/415057f8005899871616f21c9866ac600570d43a-1536x1024.png?w=1400&fit=max&auto=format)

Category: AI · Tags: AI agents, Software quality, Docker

A coding agent prepares a change, runs a check and announces that the job is done. It may have changed the files again since that check, exhausted its budget or requested an operation nobody authorized. Which part of the system catches that?

The harness is the software that governs this execution: it receives the model's proposals, routes permitted actions to tools, collects results and decides how to continue or stop. OpenAI uses the term this way when describing the Codex loop. The model still chooses its next step; the surrounding system gives those choices concrete capabilities and conditions. [Codex as a platform](https://developers.openai.com/blog/codex-as-a-platform).

My starting question for choosing controls is specific: **which failure should each component prevent?** A denied permission, a check against the current revision and a detailed log address different needs. Calling all of them “guardrails” makes it harder to decide what to retain and what to simplify.

## Where the harness sits

Consider an assistant that prepares a service configuration and can test it, while leaving deployment to a person. A request follows this path:

```txt
user request
  → harness: prepares instructions, state and available tools
  → model: proposes an action
  → tool gateway: checks the request and its authorization
  → execution environment: runs the tool
  → harness: records the result and updates state
  → another proposal or a stop
```

The gateway is the point in the program through which tool requests pass. The model can ask to publish; the gateway needs to know whether that action is available and authorized.

The execution environment has a separate job: isolating files, processes, network access and resources. OpenAI's documentation distinguishes the system orchestrating an agent from the environment doing its work. A container can constrain access to the computer, but it does not independently know whether a check covers the latest change or whether the task meets its requirements. [Sandbox Agents](https://developers.openai.com/api/docs/guides/agents/sandboxes).

_Where the harness sits_

| Component | Question it answers |
| --- | --- |
| Instructions and context | What result should the agent pursue, and what information does it have? |
| Gateway and authorization | May this request reach the tool? |
| Isolated environment | Which resources may the executed work touch? |
| Completion criterion | What observable result allows the task to close? |
| Execution trace | How can I reconstruct what happened? |

A skill can describe a procedure; the gateway still applies permissions. [Context engineering](https://www.matteovincenti.com/en/blog/context-engineering-memory-handoffs-wiki) addresses the information given to the model. Here the question is what happens when that model requests an action.

## The task needs an exit condition

The export worker is a simple example: fix its configuration and hand it over for review. The agent may read, change and verify it; deployment remains a separate decision. Before choosing a model, I would write this contract:

_The task needs an exit condition_

| Contract element | In this example |
| --- | --- |
| Acceptable result | Valid configuration checked after the last change |
| Available actions | Read and edit the configuration; run its validator |
| Scope boundary | No deployment; no changes to runtime controls |
| Required evidence | Validator outcome tied to the current content |
| Unsuccessful stop | Budget exhausted, action denied or environment unavailable |
| Human decision | Any request to start an export, with arguments and revision visible |

“Fix the configuration” leaves too many decisions implicit. Even “run the tests” is incomplete if the agent can change the suite that decides success. The contract gives the model work to perform and gives the application a criterion for accepting it.

For a software change, useful evidence includes the commit or file hashes checked, the command, exit code, environment and a reference to its output. I would also retain the policy and verifier versions. A report for an earlier commit belongs in the history; accepting the current commit requires a new run. Checks must also cover the requested behavior, as discussed in the [guide to verifying AI agents](https://www.matteovincenti.com/en/blog/ai-hallucinations-verify-agents).

## What the model proposes and what the system owns

For an integration we build ourselves, I would separate responsibilities like this. It is one possible file layout, not a format agents automatically discover:

```txt
agent-service/
├── app.py                 # receives the task and starts the runtime
├── runtime/
│   ├── loop.py            # budgets, transitions and completion
│   ├── policy.py          # permissions and approvals
│   ├── model_adapter.py   # provider messages ↔ proposals
│   ├── tool_gateway.py    # validates and routes requests
│   └── verification.py    # checks the outcome and revision
├── tools/
│   └── export_worker.py   # controlled access to the service
└── workspace/             # only files the work may change
```

The model adapter translates a response into the application's format, such as `{"tool":"write_config","args":{"max_retries":2}}`. It does not grant permission to write. Nor should it turn “I checked it; it is correct” into a verification result.

The gateway checks the schema, action, target and caller's permissions. The executor runs the authorized request and returns a receipt correlated with the call. The runtime uses that receipt to update state. A proposal and a receipt need to remain distinct even when both travel as JSON.

This separation has a practical consequence: a process running generated code should not be able to rewrite policy, replace the verifier or read the controller's credentials. Putting those files in a differently named directory does not protect them. Process, filesystem, network and identity boundaries need infrastructure enforcement. Anthropic describes this separation between session, harness and sandbox, with credentials outside the environment running generated code. [Managed Agents](https://www.anthropic.com/research/managed-agents).

A tool's output may also contain untrusted text: a webpage or README could ask the agent to run further commands. That text enters as data, rather than as new policy. Giving the agent an unrestricted shell with access to the same systems creates a path around the tool gateway.

## Two labs, two questions

The first lab isolates controls: which failure gets through when one is removed? The second adds a loop with feedback, receipts and retries: who chooses the next step when something goes wrong?

Both use a scripted proposer, with no LLM calls. That lets us compare the same mistakes and test the runtime's responses. A model's ability to choose well needs different evidence: realistic tasks, repeated trials and evaluation of the outcome.

## A lab that does not need a model

The first lab uses Python and a fixed request sequence; no credentials are needed.

The task is to prepare an export worker's configuration. Its timeout must be between 100 and 5,000 milliseconds, and its retry count between zero and two. Values must be integers; extra fields are rejected.

The worker starts with eight retries, so it violates the requirement. Three tools are available: read the configuration, change it and check it. A `finish` request asks the harness to conclude. A fourth handler, `publish`, only changes an in-memory Boolean: it lets us test refusal of an unauthorized action.

Create an empty directory with these two files. You write both by copying the contents below:

```txt
harness-lab/
├── policy.json
└── lab.py
```

`policy.json` is this example's own format, interpreted by `lab.py`; Claude, Codex and Docker do not read it automatically.

```json
{
  "allowed_tools": ["read_config", "write_config", "check_config"],
  "max_requests": 4,
  "require_current_check": true,
  "audit_after_tool": false
}
```

The policy allows only the three tools, limits the producer to four requests and requires a successful check for the current configuration. `finish` also consumes a request. Audit events are counted separately.

The complete `lab.py` file is:

```py
"""A deterministic tool-control lab. No model, network, shell, or credentials."""

import argparse
import copy
import hashlib
import json
from pathlib import Path


def fingerprint(config):
    return hashlib.sha256(json.dumps(config, sort_keys=True).encode()).hexdigest()


def valid(config):
    return (
        set(config) == {"timeout_ms", "max_retries"}
        and type(config["timeout_ms"]) is int
        and 100 <= config["timeout_ms"] <= 5000
        and type(config["max_retries"]) is int
        and 0 <= config["max_retries"] <= 2
    )


def run(proposals, policy):
    """The proposal producer has no direct reference to the tool handlers."""
    config = {"timeout_ms": 2000, "max_retries": 8}
    checked = None
    events = []
    requested = executed = audits = 0
    published = False
    status = "proposals_exhausted"
    for proposal in proposals:
        if policy["max_requests"] is not None and requested >= policy["max_requests"]:
            status = "request_limit"
            events.append({"event": "stop", "reason": status})
            break
        requested += 1
        tool = proposal["tool"]
        if tool == "finish":
            if policy["require_current_check"] and checked != fingerprint(config):
                status = "completion_rejected"
            else:
                status = "complete"
            events.append({"event": "stop", "reason": status})
            break
        if policy["allowed_tools"] is not None and tool not in policy["allowed_tools"]:
            status = "tool_denied"
            events.append({"event": "denied", "tool": tool})
            break
        if tool == "read_config":
            result = copy.deepcopy(config)
        elif tool == "write_config":
            value = proposal.get("value")
            if not isinstance(value, dict):
                raise ValueError("write_config requires a configuration object")
            config = copy.deepcopy(value)
            result = {"revision": fingerprint(config)}
        elif tool == "check_config":
            passed = valid(config)
            checked = fingerprint(config) if passed else None
            result = {"passed": passed, "revision": fingerprint(config)}
        elif tool == "publish":
            # Local flag only, deliberately available for the permission ablation.
            published = True
            result = {"published": True}
        else:
            raise ValueError("Unknown tool")
        executed += 1
        events.append({"event": "tool_result", "tool": tool, "result": result})
        if policy["audit_after_tool"]:
            audits += 1
            events.append({"event": "audit", "revision": fingerprint(config)})
    return {
        "status": status,
        "requests": requested,
        "tools": executed,
        "audits": audits,
        "valid_config": valid(config),
        "published": published,
        "events": events,
    }


GOOD = {"timeout_ms": 2000, "max_retries": 2}
BAD = {"timeout_ms": 2000, "max_retries": 8}
SCENARIOS = {
    "normal": [
        {"tool": "write_config", "value": GOOD},
        {"tool": "check_config"},
        {"tool": "finish"},
    ],
    "publish": [{"tool": "publish"}],
    "stale_check": [
        {"tool": "write_config", "value": GOOD},
        {"tool": "check_config"},
        {"tool": "write_config", "value": BAD},
        {"tool": "finish"},
    ],
    "loop": [{"tool": "read_config"} for _ in range(8)],
}


def profiles(base):
    instrumented = {**base, "audit_after_tool": True}
    return {
        "essential": copy.deepcopy(base),
        "instrumented": instrumented,
        "without_permissions": {**instrumented, "allowed_tools": None},
        "without_evidence": {**instrumented, "require_current_check": False},
        "without_budget": {**instrumented, "max_requests": None},
    }


def matrix(base):
    cases = [
        ("normal", "essential"), ("normal", "instrumented"),
        ("publish", "instrumented"), ("publish", "without_permissions"),
        ("stale_check", "instrumented"), ("stale_check", "without_evidence"),
        ("loop", "instrumented"), ("loop", "without_budget"),
    ]
    policies = profiles(base)
    return [
        {"scenario": case, "profile": profile, **run(SCENARIOS[case], policies[profile])}
        for case, profile in cases
    ]


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--trace", action="store_true")
    args = parser.parse_args()
    base = json.loads(Path(__file__).with_name("policy.json").read_text())
    for result in matrix(base):
        if not args.trace:
            result.pop("events")
        print(json.dumps(result, sort_keys=True))
```

The hash identifies the configuration's contents. After a successful check, `checked` holds that identifier. If the producer changes the values again, its completion request finds a different revision and is rejected. This prevents a check run before the latest edit from approving the changed configuration.

The code applying policy and checks belongs to the trusted runtime. The producer receives only the exposed tools, without a generic shell or write access to that code. An integration that grants those additional channels needs to enforce its boundary there too: a tool list in a prompt cannot protect it.

## Running it in Docker and reading the results

You need local Docker configured to run Linux containers, an sh-compatible shell (bash or zsh; WSL on Windows) and access from the same machine to the directory containing both files. Open the terminal in that directory. No Python packages need to be installed in the project. The first command downloads the official image; the second runs the lab without network access and mounts the source read-only.

```bash
docker pull python:3.13-slim

docker run --rm --network none --read-only --cap-drop ALL \
  --security-opt no-new-privileges --user 65534:65534 \
  --mount "type=bind,source=$PWD,target=/lab,readonly" --workdir /lab \
  -e PYTHONDONTWRITEBYTECODE=1 python:3.13-slim python lab.py
```

To use the same published OCI image index, replace `python:3.13-slim` in the run command with `python@sha256:57b29a7c050c4e679e4bea572e14b3965cf4b5839de2da49d55e49ffc4024f39`. Docker selects the variant for the current platform. The run reported here used Python 3.13.17 on Linux arm64; other architectures may select different images from that same index.

The program prints eight JSON lines. These are the cases to inspect:

_Running it in Docker and reading the results_

| Case | Runtime profile | Result |
| --- | --- | --- |
| Correct edit, check, completion | essential | `complete`, two executed tools, zero audits |
| The same sequence | instrumented | `complete`, two executed tools, two audits |
| Publication request | instrumented | `tool_denied`, no effect |
| Same request without an allowlist | without\_permissions | Local publication flag set |
| Successful check, new invalid edit, completion | instrumented | `completion_rejected` |
| Same sequence without current-revision evidence | without\_evidence | `complete` with an invalid configuration |
| Eight consecutive reads | instrumented | `request_limit` after four tools |
| Same reads without a budget | without\_budget | Eight tools executed |

The `essential` and `instrumented` profiles retain identical permissions, budgets and completion criteria. The latter adds an event containing the hash after every tool. This compares essential and instrumented execution: those two events may help a later investigation, but they add no functional check.

The other profiles remove one protection at a time to break a property. They are regression tests showing why the control matters, rather than presets to adopt. In particular, `complete` with an invalid configuration is the failure the case is supposed to expose.

For the full trace, replace the final part of the command with `python lab.py --trace`. The container is removed on exit; the program creates no files or volumes. If you downloaded the image solely for this exercise and nothing else needs it, remove it with `docker image rm python:3.13-slim`.

## From a failed tool call to the next decision

A real integration receives more than successful results. An argument may be invalid, a service may be busy or a response may arrive after a timeout. Those events do not all warrant the same “try again.”

_From a failed tool call to the next decision_

| Observed outcome | Runtime decision |
| --- | --- |
| Invalid argument | Return the field to correct; the new proposal consumes budget |
| Temporary failure with no effects attested by the executor | Retry the same request within remaining attempt, call and deadline limits |
| Missing permission | Stop and wait for approval of that action |
| Approval denied | Stop without repeatedly presenting the same request |
| Write with an unknown outcome | Retain the pending operation; block further writes and completion |
| Empty but incomplete lookup | Keep the outcome unknown: the lookup does not establish that the write failed |
| Successful verification of the current revision | Accept a completion request if no unresolved effects remain |

An input error is a good reason to involve the model again: “`args.max_retries` must be an integer” lets it change its proposal. A temporary refusal before execution can instead be handled by the runtime without asking the model to make the same decision. Attempts need a cap and a shared budget; otherwise internal retries can bypass a turn limit.

OpenAI's recovery documentation recommends checking saved work and completed actions before retrying, correcting invalid input and bounding retries. Its codes describe that service. Our tools need an equivalent contract, rather than a classification inferred solely from the HTTP number. [Errors and recovery](https://developers.openai.com/api/docs/guides/agents-api/errors).

The costly mistake is retrying a write after a timeout. The client stopped waiting; the service may already have queued the export. Cancelling the wait does not undo that effect. The runtime can look up the operation by its identifier, but an incomplete or delayed listing gives no conclusive answer. Without an authoritative outcome or a receiver-side deduplication guarantee, I would keep the case open for reconciliation. AWS covers this distinction in its guidance on timeouts and retries. [Timeouts, retries and backoff](https://builder.aws.com/content/3EumjoZascWd1oZiEgL8ORlv3qE/timeouts-retries-and-backoff-with-jitter).

### A complete loop with a local executor

The second lab reduces configuration to `max_retries`, between zero and two, and adds two local tools to simulate starting and looking up an export. It uses two files separate from the first exercise and does not read `policy.json`. Proposals contain only `tool` and `args`. Executor receipts instead contain the call identifier, tool, status, effect certainty and revision. The runtime rejects missing or inconsistent receipts; fields such as `passed` added by the proposer do not become evidence.

The proposer receives runtime feedback and may respond with a new action. A transient failure is retried within the same logical request, with the same identifier; every attempt consumes the call counter. A new proposal consumes the proposal counter. Invalid arguments and completion requests count too. Defaults in the second lab are eight proposals, ten calls and two attempts per request, with a ten-second deadline. Retries use a short exponential delay; a remote service also needs its documented retry policy and load control, such as jitter.

Starting an export first requires a valid check of the current configuration, then approval tied to the tool, arguments and configuration revision. The runtime consumes that approval when admitting the request: it does not authorize a second export merely because its arguments match. A changed revision requires a new decision.

The worker also simulates an awkward case: it records the export, then returns an unknown outcome. The subsequent lookup deliberately returns an empty, non-authoritative result. The trace must retain the pending operation and reject a further write.

These Python objects share one process. This is a testable separation of responsibilities, not a sandbox for hostile Python. The deadline is also checked between calls and after they return: the lab does not interrupt a blocked call. A production executor and its clients must implement timeouts and cancellation while preserving the distinction between local cancellation and the remote effect.

Create a second empty directory containing `runtime.py` and `demo.py`. Copy both complete files from the appendix. With the same Docker and shell prerequisites as the first lab, run this from the new directory:

```bash
docker run --rm --network none --read-only --cap-drop ALL \
  --security-opt no-new-privileges --user 65534:65534 \
  --cpus 1 --memory 128m --pids-limit 32 \
  --mount "type=bind,source=$PWD,target=/lab,readonly" --workdir /lab \
  -e PYTHONDONTWRITEBYTECODE=1 python:3.13-slim python demo.py
```

It uses the same image as the first exercise. The program prints five JSON rows containing events, counters and the final status:

_A complete loop with a local executor_

| Scenario | Outcome to check |
| --- | --- |
| `correct_after_feedback` | Type error, corrected proposal, verification and `complete`; four proposals, two calls |
| `runtime_retry` | Two write attempts, then verification and `complete`; three proposals, three calls |
| `stale_check` | Change after verification; fresh check requested, no completion |
| `unknown_is_not_failure` | One recorded export, incomplete lookup, `unresolved` and a pending operation |
| `approval_required` | A successful check, then `needs_approval`; no export |

In the second scenario, the retry stays inside the runtime; in the first, the proposer needs to change its arguments. In the fourth, neither may conclude that the operation failed merely because confirmation was not received.

The loop keeps its events in memory. Restarting the service requires task, revision, counters, evidence and pending operations to survive outside the process. A conversation summary alone cannot recover an external write; checkpoints and idempotency deserve their own treatment.

## Use an existing harness or own the loop?

For work in an ordinary repository, I would start with the available runtime. Project policy, instructions, tools and checks may be sufficient; another outer loop risks duplicating approvals, retries and session management.

When an agent needs to live inside an application, my first option is the runtime's integration interface. OpenAI distinguishes bounded script/CI execution, an SDK for starting or resuming tasks, and the app-server protocol for integrations with events and approvals. These are different layers: what the application needs to own depends on the product. [Codex as a platform](https://developers.openai.com/blog/codex-as-a-platform).

I would own a custom loop when the domain needs transitions the existing integration cannot represent well: operations waiting for reconciliation, resource-specific authorization or acceptance evidence from separate systems. That decision must include future work on provider errors, sessions, tool compatibility and diagnostics.

_Use an existing harness or own the loop?_

| Choice | When I would choose it | Cost to accept |
| --- | --- | --- |
| Configure the existing runtime | Development tasks and already supported tools | Fit its lifecycle and constraints |
| Integrate through an SDK or protocol | A custom interface and application workflow | Translate events, approvals and states without losing meaning |
| Own the loop | Domain contracts the other options cannot cover | Maintain orchestration, recovery, adapters and verification |

This boundary is more useful than comparing agent counts. A planner added to an ill-defined task can produce a longer plan without making acceptance any clearer.

## Deciding which components to remove

In Anthropic's documented case, radical simplification made results worse and obscured which parts mattered. The authors switched to removing one component at a time. The value of planning and evaluation changed with the model and task difficulty. [Harness design for long-running apps](https://www.anthropic.com/research/harness-design-long-running-apps).

I would keep two experiments separate. The first tests the runtime: denied action, stale check, exhausted budget and inconsistent receipt. A failure that stops being blocked is a contract regression. This is what the labs demonstrate.

The second evaluates work quality: keep the model, tasks, tools and acceptance criteria comparable, then remove, for example, a planner or an intermediate review. Run multiple trials because model behavior varies. Count accepted results, missed defects, human corrections and the cost of reaching an acceptable outcome, including unsuccessful attempts.

Anthropic distinguishes the transcript from the actual environmental outcome and separates capability evaluations from regressions. A convincing report does not replace the state of the service or the software's behavior. [Demystifying evals](https://www.anthropic.com/research/demystifying-evals-for-ai-agents).

I would remove a coordination step when comparisons show it catches no useful problems and its cost weighs on the work. If an evaluator only finds defects already caught by tests, I would try moving it to cases the tests cannot cover. If it keeps accepting incorrect work, the evaluation criterion needs attention first: adding more evaluators with the same criterion multiplies the problem.

Permissions remain those the application requires. Changing that boundary needs a product decision about acceptable risk, rather than a higher model score. My initial version would therefore be a readable loop with clear ownership of each control and a trace explaining why it stopped. Planning, additional memory and review come in when they address an observed difficulty.

## Appendix: complete source for the second lab

These two files use only the standard library and create no files or volumes. The container is removed when it exits. If the Python image was downloaded only for these exercises and no other project needs it, remove it with `docker image rm python:3.13-slim`.

### File `runtime.py`

```py
"""Deterministic harness contracts. No LLM, network, shell, or persistent store."""

from dataclasses import dataclass
import hashlib
import json
import time


def digest(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()


def validate(proposal):
    if not isinstance(proposal, dict) or set(proposal) != {"tool", "args"}:
        return "proposal must contain only tool and args"
    tool, args = proposal["tool"], proposal["args"]
    if not isinstance(tool, str) or not isinstance(args, dict):
        return "tool must be a string and args an object"
    if tool == "write_config":
        if set(args) != {"max_retries"} or type(args["max_retries"]) is not int:
            return "args.max_retries must be an integer"
        if not 0 <= args["max_retries"] <= 2:
            return "args.max_retries must be between 0 and 2"
    elif tool in {"read_config", "check_config", "finish"}:
        if args:
            return "args must be empty"
    elif tool in {"queue_export", "lookup_export"}:
        if set(args) != {"operation_id"} or not isinstance(args["operation_id"], str):
            return "args.operation_id must be a string"
        if not args["operation_id"]:
            return "args.operation_id must not be empty"
    else:
        return "unknown tool"
    return None


@dataclass(frozen=True)
class Receipt:
    call_id: str
    tool: str
    status: str  # ok, transient, unknown
    effect: str  # none, applied, unknown
    revision: str
    data: dict


class Worker:
    """Trusted local fixture. Its injected faults are not remote API guarantees."""

    def __init__(self, transient=0, uncertain=False):
        self.config = {"max_retries": 8}
        self.transient = transient
        self.uncertain = uncertain
        self.exports = []
        self.calls = 0

    @property
    def revision(self):
        return digest(self.config)

    def execute(self, call_id, tool, args):
        self.calls += 1
        if self.transient:
            self.transient -= 1
            return Receipt(call_id, tool, "transient", "none", self.revision, {})
        effect, data = "none", {}
        if tool == "read_config":
            data = dict(self.config)
        elif tool == "write_config":
            self.config = dict(args)
            effect = "applied"
        elif tool == "check_config":
            data = {"passed": 0 <= self.config["max_retries"] <= 2}
        elif tool == "queue_export":
            self.exports.append(args["operation_id"])
            effect = "applied"
            if self.uncertain:
                return Receipt(call_id, tool, "unknown", "unknown", self.revision, {})
            data = {"operation_id": args["operation_id"], "queued": True}
        elif tool == "lookup_export":
            # A deliberately incomplete lookup, even if the export exists.
            data = {"found": False, "authoritative": False}
        return Receipt(call_id, tool, "ok", effect, self.revision, data)


class Harness:
    WRITE_TOOLS = {"write_config", "queue_export"}
    ALLOWED_TOOLS = {"read_config", "write_config", "check_config", "queue_export", "lookup_export"}

    def __init__(self, worker, *, max_proposals=8, max_calls=10, max_attempts=2,
                 seconds=10, retry_delay=0.01, approvals=None,
                 clock=time.monotonic, sleep=time.sleep):
        self.worker = worker
        self.max_proposals, self.max_calls = max_proposals, max_calls
        self.max_attempts, self.retry_delay = max_attempts, retry_delay
        self.clock, self.sleep = clock, sleep
        self.deadline = clock() + seconds
        self.approvals = dict(approvals or {})
        self.proposals = self.calls = 0
        self.checked = None
        self.pending = None
        self.events = []

    def approval_key(self, tool, args):
        return digest({"tool": tool, "args": args, "revision": self.worker.revision})

    def stop(self, status):
        self.events.append({"event": "stop", "status": status})
        return {"status": status, "proposals": self.proposals, "calls": self.calls,
                "pending": self.pending, "events": list(self.events)}

    def run(self, propose):
        feedback = {"event": "start"}
        while self.proposals < self.max_proposals:
            if self.clock() >= self.deadline:
                return self.stop("deadline")
            proposal = propose(feedback)
            if proposal is None:
                return self.stop("unresolved" if self.pending else "proposals_exhausted")
            self.proposals += 1
            if self.clock() >= self.deadline:
                return self.stop("deadline")
            error = validate(proposal)
            if error:
                feedback = {"event": "invalid_input", "message": error}
                self.events.append(feedback)
                continue
            tool, args = proposal["tool"], proposal["args"]
            if tool == "finish":
                if self.pending:
                    return self.stop("unresolved")
                if self.checked == self.worker.revision:
                    return self.stop("complete")
                feedback = {"event": "check_required", "revision": self.worker.revision}
                self.events.append(feedback)
                continue
            if tool not in self.ALLOWED_TOOLS:
                return self.stop("tool_denied")
            if self.pending and tool in self.WRITE_TOOLS:
                return self.stop("unresolved")
            if tool == "queue_export":
                if self.checked != self.worker.revision:
                    feedback = {"event": "check_required", "revision": self.worker.revision}
                    self.events.append(feedback)
                    continue
                decision = self.approvals.get(self.approval_key(tool, args))
                if decision is None:
                    return self.stop("needs_approval")
                if decision is not True:
                    return self.stop("denied")
                del self.approvals[self.approval_key(tool, args)]
            call_id = f"call-{self.proposals}"
            for attempt in range(1, self.max_attempts + 1):
                if self.calls >= self.max_calls:
                    return self.stop("call_limit")
                if self.clock() >= self.deadline:
                    return self.stop("deadline")
                self.calls += 1
                try:
                    receipt = self.worker.execute(call_id, tool, args)
                except Exception:
                    receipt = None
                if (not isinstance(receipt, Receipt)
                        or receipt.call_id != call_id or receipt.tool != tool
                        or receipt.revision != self.worker.revision
                        or receipt.status not in {"ok", "transient", "unknown"}
                        or receipt.effect not in {"none", "applied", "unknown"}
                        or not isinstance(receipt.data, dict)
                        or (receipt.status == "transient" and receipt.effect != "none")):
                    if tool in self.WRITE_TOOLS:
                        self.pending = {"call_id": call_id, "tool": tool, "args": dict(args)}
                    return self.stop("receipt_invalid")
                self.events.append({"event": "receipt", "tool": tool,
                                    "attempt": attempt, "status": receipt.status,
                                    "effect": receipt.effect, "revision": receipt.revision})
                if receipt.status == "unknown" or receipt.effect == "unknown":
                    if tool in self.WRITE_TOOLS:
                        self.pending = {"call_id": call_id, "tool": tool, "args": dict(args)}
                    feedback = {"event": "unknown", "call_id": call_id}
                    break
                if self.clock() >= self.deadline:
                    return self.stop("deadline")
                if receipt.status == "transient":
                    delay = self.retry_delay * 2 ** (attempt - 1)
                    if attempt == self.max_attempts:
                        return self.stop("attempt_limit")
                    if self.clock() + delay >= self.deadline:
                        return self.stop("deadline")
                    self.sleep(delay)
                    continue
                if tool == "write_config":
                    self.checked = None
                if tool == "check_config":
                    self.checked = receipt.revision if receipt.data.get("passed") is True else None
                # This fixture's lookup cannot resolve pending effects.
                feedback = {"event": "tool_result", "tool": tool, "data": receipt.data}
                break
        return self.stop("proposal_limit")


def scripted(proposals):
    iterator = iter(proposals)
    return lambda _feedback: next(iterator, None)


def action(tool, **args):
    return {"tool": tool, "args": args}
```

### File `demo.py`

```py
"""Run deterministic scenarios against the executable harness contract."""
import json
from runtime import Harness, Worker, action, scripted


def corrected(feedback):
    if feedback["event"] == "start":
        return action("write_config", max_retries="two")
    if feedback["event"] == "invalid_input":
        return action("write_config", max_retries=2)
    if feedback.get("tool") == "write_config":
        return action("check_config")
    return action("finish")


def show(name, harness, proposer):
    result = harness.run(proposer)
    result["scenario"] = name
    result["exports"] = list(harness.worker.exports)
    print(json.dumps(result, sort_keys=True))


show("correct_after_feedback", Harness(Worker()), corrected)
show("runtime_retry", Harness(Worker(transient=1)), scripted([
    action("write_config", max_retries=2), action("check_config"), action("finish")]))
show("stale_check", Harness(Worker()), scripted([
    action("write_config", max_retries=2), action("check_config"),
    action("write_config", max_retries=1), action("finish")]))
worker = Worker(uncertain=True)
worker.config = {"max_retries": 2}
harness = Harness(worker)
harness.approvals[harness.approval_key("queue_export", {"operation_id": "export-17"})] = True
show("unknown_is_not_failure", harness, scripted([
    action("check_config"), action("queue_export", operation_id="export-17"),
    action("lookup_export", operation_id="export-17"),
    action("queue_export", operation_id="export-17")]))
worker = Worker()
worker.config = {"max_retries": 2}
show("approval_required", Harness(worker), scripted([
    action("check_config"), action("queue_export", operation_id="export-17")]))
```