Caderno de campo: runtime recuperável

Objetivo e hipótese

Você construirá um simulador sem modelo externo, segredo ou acesso à Internet. Essa escolha é intencional: o laboratório testa o harness, não a qualidade de um LLM. Uma função determinística representa a proposta que um modelo poderia produzir; policy e executor continuam independentes.

Hipótese: se a identidade, o alvo e a rede forem validados fora da proposta; se o efeito receber uma chave idempotente; e se o estado for persistido antes e depois da execução, então uma queda após o efeito poderá ser recuperada sem duplicação e toda interrupção terá causa observável.

Duração estimada: 120 a 150 minutos.
Ambiente: Python 3.11 ou superior, somente biblioteca padrão.
Segredos: nenhum. Os tenants, URLs e ordens são fictícios.

Cenário e baseline

O runtime deve preparar um relatório para a OS 742 do tenant shopping-a. O efeito simulado é uma gravação em lab-output/effects.json. A baseline ingênua faria apenas isto:

proposta → executar → imprimir “concluído”

Ela falha se o processo cair após gravar e antes de responder: na retomada, a mesma proposta grava outra vez. Também aceita que o texto da proposta escolha outro tenant ou uma URL externa.

O candidato adiciona fronteiras explícitas:

Arquivos do laboratório

Crie uma pasta vazia e, dentro dela, o arquivo lab_harness.py com o código a seguir. lab-output/ será criado durante a execução. O exemplo é executável e deliberadamente pequeno; arquivos JSON representam armazenamento durável e um serviço externo apenas para fins didáticos.

from __future__ import annotations

import json
import shutil
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any

OUT = Path("lab-output")
CHECKPOINT = OUT / "checkpoint.json"
EFFECTS = OUT / "effects.json"
TRACE = OUT / "trace.jsonl"


def read_json(path: Path, default: Any) -> Any:
    return json.loads(path.read_text(encoding="utf-8")) if path.exists() else default


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
    temporary.replace(path)  # troca atômica no mesmo filesystem


def emit(event: str, **fields: Any) -> None:
    OUT.mkdir(exist_ok=True)
    safe = {key: value for key, value in fields.items() if key not in {"token", "secret"}}
    with TRACE.open("a", encoding="utf-8") as stream:
        stream.write(json.dumps({"event": event, **safe}, ensure_ascii=False) + "\n")


@dataclass
class State:
    version: int = 1
    run_id: str = "run-os-742"
    phase: str = "new"
    step: int = 0
    budget_left: int = 3
    idempotency_key: str = "run-os-742:write-report:v1"
    stop_reason: str | None = None
    recovered: bool = False


def load_state() -> State:
    return State(**read_json(CHECKPOINT, asdict(State())))


def save_state(state: State) -> None:
    write_json(CHECKPOINT, asdict(state))
    emit("checkpoint", phase=state.phase, step=state.step, stop_reason=state.stop_reason)


def propose(mode: str) -> dict[str, str]:
    proposal = {
        "tool": "write_report",
        "tenant": "shopping-a",
        "order_id": "742",
        "url": "https://api.internal.example/orders/742/report",
    }
    if mode == "deny-tenant":
        proposal["tenant"] = "shopping-b"
    if mode == "deny-egress":
        proposal["url"] = "https://malicious.example/payload"
    return proposal


def policy(proposal: dict[str, str]) -> tuple[bool, str]:
    if proposal["tool"] not in {"write_report"}:
        return False, "tool_not_allowed"
    if proposal["tenant"] != "shopping-a":
        return False, "tenant_not_allowed"
    if not proposal["url"].startswith("https://api.internal.example/"):
        return False, "egress_not_allowed"
    return True, "allowed"


def execute(proposal: dict[str, str], key: str, crash: bool) -> dict[str, Any]:
    effects = read_json(EFFECTS, {})
    if key in effects:
        emit("effect_reused", idempotency_key=key)
        return {**effects[key], "reused": True}
    effect = {"tenant": proposal["tenant"], "order_id": proposal["order_id"], "status": "written"}
    effects[key] = effect
    write_json(EFFECTS, effects)
    emit("effect_committed", idempotency_key=key)
    if crash:
        raise RuntimeError("injected_crash_after_effect")
    return {**effect, "reused": False}


def run(mode: str) -> int:
    state = load_state()
    state.recovered = state.phase == "prepared"
    proposal = propose(mode)
    allowed, reason = policy(proposal)
    emit("policy_decision", allowed=allowed, reason=reason, tool=proposal["tool"])
    if not allowed:
        state.phase, state.stop_reason = "stopped", "policy_denied:" + reason
        save_state(state)
        print(state.stop_reason)
        return 2

    if state.budget_left <= 0:
        state.phase, state.stop_reason = "stopped", "budget_exhausted"
        save_state(state)
        print(state.stop_reason)
        return 3

    state.phase, state.step = "prepared", 1
    state.budget_left -= 1
    save_state(state)  # persiste intenção e chave antes do efeito
    result = execute(proposal, state.idempotency_key, crash=mode == "crash")
    state.phase, state.step = "effect_observed", 2
    save_state(state)

    effects = read_json(EFFECTS, {})
    verified = effects.get(state.idempotency_key, {}).get("status") == "written"
    state.phase = "stopped"
    state.stop_reason = "accepted" if verified else "final_invalid"
    save_state(state)
    emit("outcome", verified=verified, recovered=state.recovered, reused=result["reused"])
    print(json.dumps({"stop_reason": state.stop_reason, "recovered": state.recovered,
                      "effect_reused": result["reused"]}, sort_keys=True))
    return 0 if verified else 4


if __name__ == "__main__":
    selected = sys.argv[1] if len(sys.argv) > 1 else "happy"
    if selected == "clean":
        shutil.rmtree(OUT, ignore_errors=True)
        print("clean")
        raise SystemExit(0)
    raise SystemExit(run(selected))

Execução passo a passo

1. Caminho nominal

Execute em ambiente limpo:

python lab_harness.py clean
python lab_harness.py happy

Resultado esperado: JSON com stop_reason igual a accepted, recovered falso e effect_reused falso. Confirme que existe exatamente uma chave em lab-output/effects.json e que o checkpoint terminou em phase: stopped.

Não aceite apenas o texto do terminal. A evidência é o conjunto coerente de checkpoint, efeito e trace.

2. Tenant proibido

python lab_harness.py clean
python lab_harness.py deny-tenant

O processo retorna código diferente de zero e imprime policy_denied:tenant_not_allowed. effects.json não deve existir. O trace deve conter uma decisão de policy negada. Isso prova política fora do prompt: mesmo que a função propose peça outro tenant, o executor não recebe a ação.

3. Egress proibido

python lab_harness.py clean
python lab_harness.py deny-egress

O resultado é policy_denied:egress_not_allowed, sem efeito. Observe que o laboratório não tenta abrir a URL; a allowlist valida antes da capacidade de rede. Em produção, complemente com bloqueio real no ambiente, validação de DNS/redirects e credencial incapaz de acessar destinos arbitrários. A string no código ensina o princípio, mas não substitui firewall ou política de egress.

4. Falha depois do efeito

python lab_harness.py clean
python lab_harness.py crash
python lab_harness.py happy

A primeira execução falha deliberadamente depois de gravar effects.json e antes de salvar effect_observed. A segunda carrega phase: prepared, reutiliza a mesma idempotency key, encontra o efeito e termina com:

{"effect_reused": true, "recovered": true, "stop_reason": "accepted"}

Confirme novamente que existe apenas uma chave de efeito. Esse é o teste central do runtime recuperável.

Casos negativos adicionais

Sem alterar a solução definitiva, faça cópias temporárias e provoque:

1. remova a verificação de tenant: deny-tenant passa a criar efeito; o controle falhou;
2. gere uma idempotency key aleatória a cada retomada: crash seguido de happy cria duas intenções;
3. remova o save_state anterior ao executor: a retomada perde fase e chave;
4. registre um campo secret diretamente: o teste de privacidade deve rejeitar o trace;
5. configure budget_left: 0: a execução deve parar como budget_exhausted, sem efeito.

Não execute URLs nem credenciais reais. Os casos são simulações locais; restaurar o arquivo original faz parte da limpeza.

Evidência e relatório

Entregue:

• lab_harness.py e versão do Python;
• checkpoint.json, effects.json e trace.jsonl de cada cenário em pastas separadas;
• tabela com comando, exit code, stop reason, policy decision, quantidade de efeitos e recuperação;
• explicação da janela efeito–checkpoint e de como a chave idempotente a contém;
• limitação explícita: arquivos locais não oferecem transações distribuídas nem sandbox de rede real.

Uma tabela mínima:

Cenário	Stop reason	Efeitos	Violação bloqueada	Recuperado
nominal	accepted	1	não aplicável	não
outro tenant	policy_denied:tenant_not_allowed	0	sim	não
egress externo	policy_denied:egress_not_allowed	0	sim	não
queda + retomada	accepted	1	duplicação evitada	sim

Antes de aceitar a tabela, confronte cada célula com o artefato correspondente. O stop reason vem do evento final do trace; a quantidade de efeitos vem de effects.json; a fase e a chave vêm de checkpoint.json. Se a tabela disser “um efeito”, mas houver duas chaves no arquivo, o arquivo vence. Se o trace disser “negado” depois de um efeito, a política foi aplicada tarde demais. Essa conferência ensina uma regra operacional: resumo é navegação, não fonte de verdade. A evidência precisa ser consultada no sistema que realmente registrou estado ou consequência.

Critérios de aceite

O laboratório passa quando outra pessoa, em pasta limpa, reproduz os quatro cenários; policy impede executor e efeito nos casos proibidos; a queda ocorre após o efeito e a retomada não duplica; checkpoint contém objetivo operacional suficiente sem segredo; todo término normal ou bloqueado possui stop reason; trace distingue policy, checkpoint, efeito e outcome; e o estudante explica por que a allowlist na aplicação deve ser reforçada pelo ambiente.

Ele falha se “concluído” for a única evidência, se effects.json tiver duas intenções após recuperação, se a URL proibida alcançar o executor, se o checkpoint depender da conversa ou se os logs contiverem credenciais.

Troubleshooting

Sintoma	Diagnóstico	Correção
python não encontrado	runtime fora do PATH	use o executável Python 3.11+ instalado no ambiente
recuperação aparece falsa	checkpoint foi limpo entre crash e happy	limpe somente antes do par de execuções
dois efeitos	chave mudou ou executor não a consultou	derive a chave da intenção estável e consulte antes de gravar
efeito em caso negado	policy ocorreu depois do executor	mova resolução e policy para antes de qualquer capacidade
JSON corrompido	escrita foi interrompida no arquivo final	mantenha temporário + replace ou use armazenamento transacional

Limpeza e solução comentada

Ao terminar:

python lab_harness.py clean

A solução mantém quatro fronteiras. propose é conteúdo não confiável; policy toma a decisão determinística; execute é o único componente com efeito e reconhece a idempotency key; State registra a intenção antes do efeito e o outcome depois. O trace não decide nada, mas permite explicar o que ocorreu. Em um sistema real, substitua JSON por armazenamento transacional, use sandbox e egress no nível de infraestrutura, obtenha identidade do sistema autenticado e gere traces com redaction antes da exportação.

Recuperação ativa

1. Qual arquivo comprova o efeito e qual comprova a fase do runtime?
2. Por que a primeira execução do cenário crash não deve ser repetida com uma chave nova?
3. Que diferença existe entre validar uma URL em código e restringir egress no ambiente?
4. Qual stop reason indica bloqueio de política e qual indica sucesso verificado?
5. Por que uma resposta final do modelo não substituiria a consulta a effects.json?