Harness engineering: quali controlli servono davvero a un agente

Un agente di sviluppo prepara una modifica, esegue un controllo e dichiara di avere finito. Nel frattempo potrebbe aver cambiato ancora i file, consumato il budget disponibile o richiesto un'operazione che nessuno gli aveva autorizzato. Chi deve accorgersene?

L'harness è il software che governa questa esecuzione: riceve le proposte del modello, inoltra le azioni consentite agli strumenti, raccoglie i risultati e decide come proseguire o fermarsi. È il significato usato anche da OpenAI nel descrivere il loop di Codex. Il modello continua a scegliere il prossimo passo; il sistema intorno gli attribuisce possibilità concrete e condizioni da rispettare. Codex as a platform.

Per scegliere i controlli partirei da una domanda precisa: quale errore deve impedire ciascun componente? Un permesso negato, una verifica della versione corrente e un log dettagliato rispondono a esigenze diverse. Metterli tutti sotto l'etichetta “guardrail” rende difficile capire che cosa mantenere e che cosa semplificare.

Dove sta l'harness

Un esempio è un assistente che prepara la configurazione di un servizio e può provarla, ma deve lasciare il rilascio a una persona. Il percorso di una richiesta è questo:

richiesta dell'utente
  → harness: prepara istruzioni, stato e strumenti disponibili
  → modello: propone un'azione
  → gateway degli strumenti: controlla richiesta e autorizzazione
  → ambiente di esecuzione: esegue lo strumento
  → harness: registra l'esito e aggiorna lo stato
  → nuova proposta oppure arresto

Il gateway è il punto del programma attraverso cui passano le richieste agli strumenti. Il modello può chiedere di pubblicare; il gateway deve sapere se quell'azione è disponibile e autorizzata.

Anche l'ambiente ha un compito distinto: isolare file, processi, rete e risorse. La documentazione OpenAI separa il sistema che orchestra l'agente dall'ambiente che esegue il lavoro. Un container può limitare l'accesso al computer, ma non sa da solo se una verifica riguarda l'ultima modifica o se il task soddisfa i requisiti. Sandbox Agents.

Una skill può descrivere una procedura; il gateway continua ad applicare i permessi. Per il contenuto da fornire al modello resta utile il lavoro sul context engineering: qui il problema è che cosa succede quando il modello chiede di agire.

Il task deve avere una condizione di uscita

Il worker di esportazione è un esempio semplice: vogliamo correggerne la configurazione e consegnarla per la review. L'agente può leggere, modificare e verificare; il rilascio rimane una decisione separata. Prima di scegliere il modello, scriverei questo contratto:

La descrizione «sistema la configurazione» lascia troppe decisioni implicite. Anche «esegui i test» è incompleta se l'agente può cambiare la suite che decide il successo. Il contratto assegna al modello il lavoro da svolgere e all'applicazione il criterio con cui accettarlo.

Per una modifica software, la prova può contenere commit o hash dei file verificati, comando, exit code, ambiente e riferimento all'output. Includerei anche la versione della policy e del verificatore. Un report sul commit precedente resta utile nella cronologia; serve una nuova esecuzione per accettare il commit corrente. Le verifiche devono inoltre coprire il comportamento richiesto: questo è il problema affrontato nella guida alle allucinazioni degli agenti.

Le parti che il modello può proporre e quelle che deve possedere il sistema

In un'integrazione scritta da noi separerei le responsabilità così. È una possibile organizzazione dei file, non un formato letto automaticamente dagli agenti:

agent-service/
├── app.py                 # riceve il task e avvia il runtime
├── runtime/
│   ├── loop.py            # budget, transizioni e completamento
│   ├── policy.py          # permessi e approvazioni
│   ├── model_adapter.py   # messaggi del provider ↔ proposte
│   ├── tool_gateway.py    # valida e inoltra le richieste
│   └── verification.py    # controlla risultato e revisione
├── tools/
│   └── export_worker.py   # accesso controllato al servizio
└── workspace/             # soli file modificabili dal lavoro

L'adapter del modello traduce una risposta nel formato dell'applicazione, per esempio {"tool":"write_config","args":{"max_retries":2}}. Non aggiunge l'autorizzazione a scrivere. Non deve neppure trasformare «ho controllato, è corretto» in un esito di verifica.

Il gateway controlla schema, azione, destinatario e permessi del chiamante. L'executor esegue la richiesta autorizzata e restituisce una ricevuta correlata alla chiamata. Il runtime usa quella ricevuta per aggiornare lo stato. La differenza fra proposta e ricevuta deve rimanere anche se entrambe viaggiano come JSON.

Questa separazione ha una conseguenza pratica: un processo che esegue codice generato non dovrebbe poter riscrivere la policy, sostituire il verificatore o leggere le credenziali del controllo. Tenere quei file in una cartella con un nome diverso non li protegge. Servono confini di processo, filesystem, rete e identità applicati dall'infrastruttura. Anthropic descrive proprio la separazione fra sessione, harness e sandbox, con le credenziali fuori dall'ambiente del codice generato. Managed Agents.

Anche l'output di un tool può contenere testo non fidato: una pagina web o un README potrebbero chiedere di eseguire altri comandi. Quel testo entra come dato, non come una nuova policy. Se al gateway affianchiamo una shell con accesso libero agli stessi sistemi, abbiamo aperto un percorso che aggira i controlli degli strumenti.

Due laboratori, due domande

Il primo laboratorio isola i controlli: quale errore passa quando ne togliamo uno? Il secondo aggiunge un loop con feedback, ricevute e retry: chi decide il passo successivo quando qualcosa va storto?

Entrambi usano un proponente programmato, senza chiamate LLM. Possiamo così confrontare gli stessi errori e verificare le risposte del runtime. La capacità di un modello di scegliere bene richiede altre prove: task realistici, più esecuzioni e valutazione del risultato.

Un laboratorio in cui il modello non serve

Il primo laboratorio usa Python e una sequenza fissa di richieste; non servono credenziali.

Il compito è preparare la configurazione di un worker di esportazione. Il timeout deve stare fra 100 e 5.000 millisecondi e il numero di retry fra zero e due. I valori sono interi; campi ulteriori vengono rifiutati.

All'inizio il worker ha otto retry e quindi non rispetta il requisito. Sono disponibili tre strumenti: leggere la configurazione, modificarla e verificarla. La richiesta finish chiede all'harness di concludere. Un quarto handler, publish, cambia soltanto un booleano in memoria: serve a provare il rifiuto di un'azione non autorizzata.

Crea una cartella vuota con questi due file, entrambi scritti da te copiando il contenuto qui sotto:

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

policy.json è un formato di questo esempio, interpretato da lab.py; Claude, Codex e Docker non lo leggono automaticamente.

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

La policy consente solo i tre strumenti, limita a quattro le richieste del proponente e richiede una verifica valida per la configurazione corrente. Anche finish consuma una richiesta. Gli eventi di audit hanno un conteggio separato.

Il file lab.py contiene tutto il programma:

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

L'hash identifica il contenuto della configurazione. Dopo una verifica riuscita, checked conserva quell'identificatore. Se il proponente modifica nuovamente i valori, una richiesta di conclusione trova una revisione diversa e viene respinta. Questo evita di accettare la modifica di oggi usando il controllo passato prima dell'ultima modifica.

Il codice che applica policy e verifiche appartiene al runtime fidato. Il proponente riceve soltanto gli strumenti esposti, senza una shell generica o un accesso in scrittura a quel codice. Se un'integrazione concede anche quei canali, deve applicare il confine lì: una lista di tool nel prompt non basta a proteggerlo.

Esecuzione in Docker e risultati da controllare

Servono Docker locale configurato per eseguire container Linux, una shell compatibile con sh (bash o zsh; su Windows, WSL) e accesso dalla stessa macchina alla directory dei due file. Apri il terminale in quella directory. Non occorre installare pacchetti Python nel progetto. Il primo comando scarica l'immagine ufficiale; il secondo esegue il laboratorio senza accesso alla rete e con i sorgenti in sola lettura.

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

Per usare lo stesso index OCI pubblicato, sostituisci python:3.13-slim nel comando di esecuzione con python@sha256:57b29a7c050c4e679e4bea572e14b3965cf4b5839de2da49d55e49ffc4024f39. Docker seleziona la variante per la piattaforma in uso. La prova qui riportata è stata eseguita con Python 3.13.17 su Linux arm64; su architetture diverse Docker può selezionare immagini diverse dallo stesso index.

Il programma stampa otto righe JSON. Questi sono i passaggi da leggere:

I profili essential e instrumented conservano gli stessi permessi, budget e criterio di conclusione. Il secondo aggiunge un evento con l'hash dopo ogni tool. È un confronto fra esecuzione essenziale e strumentata: quei due eventi possono aiutare una ricostruzione successiva, ma non aggiungono una verifica funzionale.

Gli altri profili tolgono una protezione alla volta per far fallire una proprietà. Sono prove di regressione: mostrano perché mantenere il controllo, non preset da adottare. In particolare, complete con una configurazione errata è l'errore che il caso deve rendere visibile.

Per vedere la traccia completa sostituisci l'ultima parte del comando con python lab.py --trace. Il container viene rimosso all'uscita; il programma non crea file o volumi. Se l'immagine è stata scaricata soltanto per questo esercizio e non serve ad altro, puoi rimuoverla con docker image rm python:3.13-slim.

Dal tool fallito alla prossima decisione

Un'integrazione reale non riceve soltanto risultati positivi. Un argomento può essere errato, un servizio può essere occupato, una risposta può arrivare dopo il timeout. Non tratterei questi eventi con lo stesso «riprova».

Un errore di input è un buon motivo per coinvolgere di nuovo il modello: «args.max_retries deve essere un intero» gli permette di cambiare proposta. Un rifiuto temporaneo prima dell'esecuzione può invece essere gestito dal runtime senza chiedere al modello la stessa decisione. Occorrono un massimo di tentativi e un budget condiviso, altrimenti i retry interni possono aggirare il limite dei turni.

La documentazione OpenAI raccomanda di controllare il lavoro salvato e le azioni già completate prima di riprovare, di correggere gli input invalidi e di limitare i retry. I suoi codici descrivono quel servizio; per i nostri strumenti serve un contratto equivalente, non una classificazione dedotta soltanto dal numero HTTP. Errors and recovery.

Il caso più costoso da sbagliare è il timeout di una scrittura. Il client ha smesso di aspettare; il servizio potrebbe aver accodato l'esportazione. Annullare l'attesa non annulla quell'effetto. Il runtime può cercare l'operazione usando il suo identificatore, ma un elenco incompleto o aggiornato in ritardo non dà una risposta conclusiva. Senza un riscontro autorevole o una garanzia di deduplica del ricevente, lascerei il caso aperto per la riconciliazione. AWS tratta questa distinzione nella sua guida a timeout e retry. Timeouts, retries and backoff.

Un loop completo, con un executor locale

Il secondo laboratorio riduce la configurazione al solo max_retries, fra zero e due, e aggiunge due strumenti locali per simulare l'avvio e la ricerca di un'esportazione. Usa due file separati dal primo esercizio e non legge policy.json. La proposta contiene soltanto tool e args. La ricevuta dell'executor contiene invece identificatore della chiamata, tool, stato, certezza dell'effetto e revisione. Il runtime rifiuta ricevute mancanti o incoerenti; campi come passed aggiunti dal proponente non diventano prove.

Il proponente riceve il feedback del runtime e può rispondere con una nuova azione. Il retry di un errore temporaneo avviene dentro la stessa richiesta logica, con lo stesso identificatore; ogni tentativo consuma il contatore delle chiamate. Una nuova proposta consuma invece il contatore delle proposte. Anche argomenti sbagliati e richieste di conclusione vengono contati. I default del secondo laboratorio sono otto proposte, dieci chiamate e due tentativi per richiesta, con una deadline di dieci secondi. I retry hanno un breve ritardo esponenziale; per un servizio remoto servono anche la policy documentata del servizio e una gestione del carico, per esempio jitter.

L'avvio di un'esportazione richiede prima una verifica valida della configurazione corrente, poi un'approvazione riferita a tool, argomenti e revisione della configurazione. Il runtime la consuma quando ammette quella richiesta: non autorizza una seconda esportazione solo perché ha gli stessi argomenti. Una revisione diversa richiede una nuova decisione.

Il worker simula anche un caso scomodo: registra l'esportazione, poi restituisce un esito sconosciuto. La ricerca successiva restituisce apposta un risultato vuoto non autorevole. La traccia deve conservare l'operazione pendente e rifiutare una nuova scrittura.

Gli oggetti Python qui condividono un processo. È una separazione di responsabilità utile da testare, non una sandbox per eseguire Python ostile. Anche la deadline è controllata fra chiamate e al loro ritorno: il laboratorio non interrompe una chiamata bloccata. In produzione, timeout e cancellazione vanno implementati nell'executor e nei client, mantenendo distinta la sorte dell'effetto remoto.

Crea una seconda cartella vuota con runtime.py e demo.py. Copia entrambi i file completi riportati nell'appendice. Con gli stessi prerequisiti Docker e shell del primo laboratorio, esegui dalla nuova cartella:

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

L'immagine è la stessa del primo esercizio. Il programma stampa cinque righe JSON con eventi, contatori e stato finale:

Nel secondo caso il retry rimane dentro il runtime; nel primo il proponente deve cambiare gli argomenti. Nel quarto, invece, nessuno dei due può concludere che l'operazione sia fallita per il solo fatto di non aver ricevuto conferma.

Il loop mantiene gli eventi in memoria. Per interrompere e riprendere il servizio, occorre salvare task, revisione, contatori, prove e operazioni pendenti fuori dal processo. Un riassunto della conversazione non recupera da solo una scrittura remota: questo passaggio merita una trattazione separata su checkpoint e idempotenza.

Usare un harness esistente o scrivere il proprio

Per lavorare su un repository ordinario partirei dal runtime già disponibile. Policy, istruzioni, strumenti e verifiche del progetto possono bastare; un secondo loop esterno rischia di duplicare approvazioni, retry e gestione delle sessioni.

Quando l'agente deve vivere dentro un'applicazione, userei prima l'interfaccia di integrazione del runtime. OpenAI distingue esecuzione delimitata da script/CI, SDK per avviare o riprendere task e protocollo app-server per integrazioni con eventi e approvazioni. Sono livelli diversi: il controllo che l'applicazione deve possedere cambia con il prodotto. Codex as a platform.

Scriverei un loop personalizzato quando il dominio richiede transizioni che l'integrazione esistente non riesce a rappresentare bene: per esempio operazioni in attesa di riconciliazione, autorizzazioni per specifiche risorse o prove di accettazione prodotte da sistemi separati. La convenienza deve includere il lavoro futuro su errori del provider, sessioni, compatibilità dei tool e diagnostica.

Il confine fra queste scelte è più utile del confronto fra quantità di agenti. Un planner aggiunto a un task mal definito può produrre un piano più lungo senza rendere chiaro che cosa accettare.

Come decidere quali componenti togliere

Nel caso documentato da Anthropic, una semplificazione radicale peggiorò il risultato e rese difficile capire quali parti servissero. Gli autori passarono alla rimozione di un componente alla volta. Il valore di planner e valutatore cambiava con il modello e con la difficoltà del task. Harness design for long-running apps.

Terrei separati due esperimenti. Nel primo controllo il runtime: azione vietata, verifica vecchia, budget finito, ricevuta incoerente. Un caso che smette di essere bloccato è una regressione del contratto. È ciò che mostrano i laboratori.

Nel secondo valuto la qualità del lavoro: tengo modello, task, strumenti e criteri di accettazione comparabili e tolgo, per esempio, il planner oppure una review intermedia. Eseguo più prove, perché il comportamento del modello varia. Conto quanti risultati vengono accettati, quali difetti sfuggono, quante correzioni umane servono e quanto costa arrivare al risultato accettabile, includendo i tentativi falliti.

Anthropic distingue la trascrizione dall'esito effettivo nell'ambiente e separa le valutazioni di capacità dalle regressioni. Una frase convincente nel report non sostituisce lo stato del servizio o il comportamento del software. Demystifying evals.

Toglierei un controllo di coordinamento quando i confronti mostrano che non cattura problemi utili e il suo costo pesa sul lavoro. Se un valutatore trova soltanto difetti già intercettati dai test, proverei a spostarlo sui casi che i test non coprono. Se continua a dichiarare valido il lavoro sbagliato, prima correggerei il criterio di valutazione: aggiungere altri valutatori con lo stesso criterio moltiplica il problema.

I permessi restano quelli richiesti dall'applicazione. Per modificare quel confine serve una decisione sul prodotto e sul rischio ammesso, non un modello che ottiene un punteggio più alto. La mia prima versione sarebbe quindi un loop leggibile, con un responsabile preciso per ogni controllo e una traccia che spieghi perché si è fermato. Planner, memoria aggiuntiva e review entrano quando risolvono una difficoltà osservata.

Appendice: sorgenti del secondo laboratorio

I due file seguenti sono completi, usano soltanto la libreria standard e non generano file o volumi. Il container viene rimosso all'uscita. Se l'immagine Python è stata scaricata solo per gli esercizi e non serve ad altri progetti, puoi rimuoverla con docker image rm python:3.13-slim.

File runtime.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

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