Caderno de campo: da spec à frontier

Objetivo e hipótese

Você receberá uma pequena spec: “um supervisor autorizado aprova uma OS pendente, a decisão é auditada e sincronizada com o ERP sem duplicação”. O exercício cria um plano, testa sua estrutura, isola a incerteza de retry e transfere a conclusão para o próximo contexto.

Hipótese: se cada ticket possuir comportamento demonstrável e prova local, blockers formarem um grafo acíclico, a migração preservar compatibilidade e o protótipo tiver uma pergunta e parada explícitas, então a frontier conterá trabalho que uma sessão nova consegue executar sem adivinhar. Um handoff referenciado deverá permitir retomar sem copiar o histórico.

Duração: 120 a 150 minutos.
Pré-requisitos: Python 3.11 ou superior.
Dependências: somente biblioteca padrão.
Dados e rede: fixtures sintéticas, sem rede, segredo, ERP ou banco reais.
Condição de parada: o plano ruim é rejeitado; o bom revela a frontier esperada; o protótipo distingue duas políticas de retry; o handoff passa validação.

O que será provado

O laboratório separa quatro afirmações:

1. o grafo é estruturalmente executável;
2. os tickets de implementação são verticais segundo um contrato didático;
3. reutilizar a mesma chave evita duplicação na simulação de timeout após commit;
4. o handoff contém estado vivo e aponta para uma evidência existente.

Ele não prova que um ERP real honra idempotência, que a arquitetura está pronta para produção ou que toda fatia vertical precisa conter exatamente três camadas. Esses limites aparecem no relatório para impedir que um resultado didático vire garantia operacional.

Arquivo executável

Crie uma pasta vazia e salve o código abaixo como workflow_lab.py.

from __future__ import annotations

import json
import sys
from pathlib import Path


GOOD_TICKETS = [
    {
        "id": "P-01", "kind": "prototype", "blocked_by": [],
        "question": "O ERP deduplica retry com o mesmo eventId após timeout?",
        "metric": "quantidade de aprovações remotas", "threshold": 1,
        "stop": "parar após comparar chave estável e chave nova",
    },
    {
        "id": "T-01", "kind": "implementation", "blocked_by": [],
        "demo": "supervisor visualiza uma OS pendente individual",
        "layers": ["api", "ui", "test"],
    },
    {
        "id": "T-02", "kind": "implementation", "blocked_by": ["P-01", "T-01"],
        "demo": "supervisor aprova OS 742; operador comum recebe 403",
        "layers": ["domain", "persistence", "api", "ui", "audit", "test"],
    },
    {
        "id": "M-EXP", "kind": "migration", "phase": "expand", "blocked_by": ["T-02"],
        "demo": "status e status_code coexistem sem quebrar o conector",
        "layers": ["persistence", "api", "contract-test"],
    },
    {
        "id": "M-MIG", "kind": "migration", "phase": "migrate", "blocked_by": ["M-EXP"],
        "demo": "backfill idempotente e consumidores usam status_code",
        "layers": ["persistence", "consumer", "test"],
    },
    {
        "id": "M-CON", "kind": "migration", "phase": "contract", "blocked_by": ["M-MIG"],
        "demo": "status antigo removido após zero leitores observados",
        "layers": ["persistence", "observability", "test"],
    },
]

BAD_TICKETS = [
    {
        "id": "H-DB", "kind": "implementation", "blocked_by": [],
        "demo": "", "layers": ["persistence"],
    },
    {
        "id": "H-API", "kind": "implementation", "blocked_by": ["H-DB"],
        "demo": "", "layers": ["api"],
    },
    {
        "id": "M-EXP", "kind": "migration", "phase": "expand", "blocked_by": ["H-API"],
        "demo": "adicionar status_code", "layers": ["persistence", "api", "test"],
    },
    {
        "id": "M-CON", "kind": "migration", "phase": "contract", "blocked_by": ["M-EXP"],
        "demo": "remover status imediatamente", "layers": ["persistence", "api", "test"],
    },
]


def find_cycle(tickets: list[dict]) -> bool:
    graph = {ticket["id"]: ticket.get("blocked_by", []) for ticket in tickets}
    visiting: set[str] = set()
    visited: set[str] = set()

    def visit(node: str) -> bool:
        if node in visiting:
            return True
        if node in visited:
            return False
        visiting.add(node)
        if any(dep in graph and visit(dep) for dep in graph[node]):
            return True
        visiting.remove(node)
        visited.add(node)
        return False

    return any(visit(node) for node in graph)


def validate_tickets(tickets: list[dict]) -> list[str]:
    findings: list[str] = []
    ids = [ticket["id"] for ticket in tickets]
    known = set(ids)
    if len(ids) != len(known):
        findings.append("duplicate-id")
    for ticket in tickets:
        for blocker in ticket.get("blocked_by", []):
            if blocker not in known:
                findings.append(f"{ticket['id']}:missing-blocker:{blocker}")
        if ticket["kind"] == "implementation":
            if not ticket.get("demo"):
                findings.append(f"{ticket['id']}:no-demo")
            layers = set(ticket.get("layers", []))
            if len(layers) < 3 or "test" not in layers:
                findings.append(f"{ticket['id']}:horizontal-or-unverified")
        if ticket["kind"] == "prototype":
            required = {"question", "metric", "threshold", "stop"}
            missing = sorted(required - set(ticket))
            if missing:
                findings.append(f"{ticket['id']}:prototype-missing:{','.join(missing)}")
    if find_cycle(tickets):
        findings.append("dependency-cycle")

    by_phase = {t.get("phase"): t for t in tickets if t["kind"] == "migration"}
    if "contract" in by_phase:
        migrate = by_phase.get("migrate")
        if migrate is None or migrate["id"] not in by_phase["contract"].get("blocked_by", []):
            findings.append("contract-before-migrate")
    return findings


def frontier(tickets: list[dict], completed: set[str]) -> list[str]:
    return sorted(
        ticket["id"] for ticket in tickets
        if ticket["id"] not in completed
        and set(ticket.get("blocked_by", [])) <= completed
    )


class FakeERP:
    def __init__(self) -> None:
        self.approvals: dict[str, str] = {}

    def approve(self, event_id: str, work_order_id: str, fail_after_commit: bool) -> None:
        self.approvals.setdefault(event_id, work_order_id)
        if fail_after_commit:
            raise TimeoutError("timeout sintético depois do commit")


def retry_experiment(stable_key: bool) -> int:
    erp = FakeERP()
    first_key = "approval:742:v3"
    try:
        erp.approve(first_key, "742", fail_after_commit=True)
    except TimeoutError:
        retry_key = first_key if stable_key else "approval:742:v3:retry-2"
        erp.approve(retry_key, "742", fail_after_commit=False)
    return len(erp.approvals)


def run_prototype(output_dir: Path) -> dict:
    report = {
        "question": "O ERP deduplica retry com o mesmo eventId após timeout?",
        "fixture": "FakeERP em memória; timeout sintético após commit",
        "stable_key_remote_approvals": retry_experiment(stable_key=True),
        "new_key_remote_approvals": retry_experiment(stable_key=False),
        "threshold": 1,
        "decision": "reutilizar eventId em todas as tentativas da mesma aprovação",
        "stop_reason": "as duas políticas produziram resultados discriminantes",
        "limitations": [
            "não prova comportamento do ERP real",
            "não mede latência, concorrência ou indisponibilidade prolongada",
            "código experimental não entra na implementação de produção",
        ],
    }
    output_dir.mkdir(exist_ok=True)
    (output_dir / "prototype-report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    return report


def create_handoff(output_dir: Path) -> dict:
    handoff = {
        "objective": "implementar T-02 com retry idempotente do ERP",
        "in_flight": "P-01 concluído; T-01 continua na frontier",
        "decisions": ["reutilizar approval:{workOrderId}:{version} como eventId"],
        "evidence": ["lab-output/prototype-report.json"],
        "risks": ["o ERP real ainda precisa de teste de contrato homologado"],
        "next_step": "incorporar a decisão no aceite de T-02",
        "stop_reason": "pergunta de P-01 respondida; implementação fora deste contexto",
    }
    (output_dir / "handoff.json").write_text(
        json.dumps(handoff, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    return handoff


def validate_handoff(handoff: dict, root: Path) -> list[str]:
    required = {"objective", "in_flight", "decisions", "evidence", "risks", "next_step", "stop_reason"}
    findings = [f"handoff-missing:{key}" for key in sorted(required - set(handoff))]
    for reference in handoff.get("evidence", []):
        if not (root / reference).exists():
            findings.append(f"handoff-broken-reference:{reference}")
    serialized = json.dumps(handoff).lower()
    if any(marker in serialized for marker in ("api_key", "password", "bearer ")):
        findings.append("handoff-possible-secret")
    return findings


def main() -> int:
    mode = sys.argv[1] if len(sys.argv) > 1 else "all"
    if mode == "bad-plan":
        findings = validate_tickets(BAD_TICKETS)
        print(json.dumps({"accepted": not findings, "findings": findings}, indent=2))
        return 2 if findings else 1
    if mode != "all":
        print("uso: python workflow_lab.py [bad-plan|all]", file=sys.stderr)
        return 64

    plan_findings = validate_tickets(GOOD_TICKETS)
    output_dir = Path("lab-output")
    prototype = run_prototype(output_dir)
    handoff = create_handoff(output_dir)
    handoff_findings = validate_handoff(handoff, Path("."))
    report = {
        "plan_findings": plan_findings,
        "initial_frontier": frontier(GOOD_TICKETS, completed=set()),
        "frontier_after_p01_t01": frontier(GOOD_TICKETS, completed={"P-01", "T-01"}),
        "prototype": prototype,
        "handoff_findings": handoff_findings,
        "accepted": (
            not plan_findings
            and frontier(GOOD_TICKETS, set()) == ["P-01", "T-01"]
            and prototype["stable_key_remote_approvals"] == 1
            and prototype["new_key_remote_approvals"] == 2
            and not handoff_findings
        ),
    }
    (output_dir / "lab-report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0 if report["accepted"] else 1


if __name__ == "__main__":
    raise SystemExit(main())

Etapa 1 — prove que o validador rejeita o plano ruim

No PowerShell, execute:

python workflow_lab.py bad-plan
if ($LASTEXITCODE -ne 2) { throw "o plano horizontal deveria ser rejeitado" }

O resultado deve conter H-DB:no-demo, H-DB:horizontal-or-unverified, falhas equivalentes para H-API e contract-before-migrate. O exit code 2 é evidência negativa intencional, não defeito do script.

Observe a diferença entre regra didática e lei arquitetural. O validador exige três “camadas” e test para tickets de implementação porque a fixture foi desenhada para capturar cortes obviamente horizontais. Num serviço sem UI, o contrato real deve aceitar outro conjunto. Automatizar uma heurística sem declarar seu domínio cria burocracia, não qualidade.

Etapa 2 — calcule a frontier e execute o plano bom

python workflow_lab.py all
if ($LASTEXITCODE -ne 0) { throw "o fluxo válido deveria passar" }

Confirme no output:

{
  "plan_findings": [],
  "initial_frontier": ["P-01", "T-01"],
  "frontier_after_p01_t01": ["T-02"],
  "accepted": true
}

P-01 e T-01 são independentes no grafo. Depois de concluídos, T-02 entra na frontier. Isso não autoriza paralelismo ilimitado: antes de despachar agentes, verifique worktrees, arquivos compartilhados, ambiente de teste e decisões concorrentes.

O plano também mostra a exceção de migração. M-CON depende de M-MIG, que depende de M-EXP. Remova M-MIG ou ligue contract diretamente a expand e o validador deve bloquear a remoção prematura.

Etapa 3 — leia o experimento, não apenas o verde

Abra lab-output/prototype-report.json. Na primeira execução, o simulador grava a aprovação e lança timeout antes de o cliente receber confirmação. Com chave estável, o retry usa o mesmo identificador e o dicionário conserva uma operação. Com chave nova, cria duas.

O experimento é discriminante porque as duas políticas produzem outcomes diferentes sob a mesma falha. Sua condição de parada foi alcançada. Não acrescente UI, autenticação ou PostgreSQL “para ficar mais real”. O relatório declara que FakeERP não prova o contrato do fornecedor; o próximo passo correto é transformar a conclusão em hipótese de teste no sandbox homologado.

Esta simulação não é a implementação de produção. O verdadeiro controle exigirá chave durável, persistência de tentativa, contrato com o ERP, observabilidade e recuperação. Copiar a classe FakeERP para o serviço seria confundir evidência experimental com componente operacional.

Etapa 4 — audite o handoff

Abra lab-output/handoff.json. Ele não contém chat, código ou relatório duplicado. O campo evidence aponta para o artefato que sustenta a decisão. validate_handoff confirma que a referência existe e procura marcadores simples de segredo. Essa busca não é scanner completo; serve para mostrar que redaction precisa de controle próprio.

Teste uma falha: renomeie temporariamente prototype-report.json e execute all novamente somente depois de comentar run_prototype. O handoff deve acusar referência quebrada. Restaure o arquivo e o código ao final. Um handoff perfeito em prosa, mas com pointer inexistente, não é retomável.

Casos negativos para praticar

Faça uma mutação por vez, execute e desfaça:

1. adicione P-01 como blocker de si mesmo; espere dependency-cycle;
2. troque o blocker de T-02 por T-99; espere missing-blocker;
3. remova test das layers de T-01; espere horizontal-or-unverified;
4. em retry_experiment, sempre gere chave nova; o gate final deve falhar;
5. remova limitations do relatório e explique por que o script ainda passa: essa é uma lacuna que poderia virar nova regra;
6. coloque uma senha fictícia no handoff com o marcador password; espere bloqueio.

Registre mutação, resultado esperado, observado e conclusão. Um teste só merece confiança quando também falha sob uma alteração que deveria detectar.

Evidências, limpeza e segurança

Entregue workflow_lab.py, lab-output/lab-report.json, prototype-report.json, handoff.json e sua tabela de mutações. Outra pessoa deve executar os dois comandos sem instalar pacote nem conhecer a conversa.

Os valores são didáticos. Não substitua fixtures por dados produtivos. Um protótipo que toca ERP real pode gerar ordem duplicada, custo e incidente. Use sandbox autorizado, contas técnicas de menor privilégio e identificadores sintéticos. Não inclua segredos em branch experimental ou handoff.

Para limpar apenas a saída do laboratório, a partir da pasta criada:

Remove-Item -Recurse -Force -LiteralPath './lab-output'

Não execute a limpeza fora dessa pasta nem use caminho calculado sem conferência.

Critérios de aceite

O laboratório está concluído quando:

• bad-plan retorna 2 e explica cortes horizontais e contract prematuro;
• all retorna 0, com frontier inicial P-01 e T-01;
• após concluir ambos, somente T-02 entra na frontier;
• o protótipo produz uma aprovação com chave estável e duas com chave nova;
• relatório contém pergunta, fixture, threshold, decisão, parada e limitações;
• handoff aponta para evidência existente e nomeia o próximo passo;
• pelo menos três mutações negativas falham pelo motivo esperado;
• você consegue explicar o que o laboratório não prova.

Solução comentada

find_cycle percorre o grafo com conjuntos visiting e visited; reencontrar um nó ainda em visita revela ciclo. frontier seleciona tickets abertos cujos blockers são subconjunto dos concluídos. validate_tickets aplica invariantes estruturais e uma heurística pedagógica de verticalidade. O protótipo usa setdefault para representar deduplicação por event ID e injeta a falha depois do efeito, o momento ambíguo que exige idempotência. O handoff transfere a decisão por um pointer verificável.

Em um sistema real, substitua dicionários por ambientes resetáveis, contratos do tracker e fixtures versionadas. A skill to-tickets inspira tracer bullets e frontier; prototype inspira a pergunta e a separação da branch; handoff inspira compaction por referências. Nenhum desses nomes é comando universal do Python, Codex ou Claude Code.

Recuperação ativa

1. Por que dois tickets na frontier ainda podem não ser seguros para execução paralela?
2. Que observação diferencia um protótipo discriminante de uma demonstração decorativa?
3. Por que M-CON deve depender de M-MIG, e não apenas de M-EXP?
4. O que faz T-02 ser uma fatia vertical?
5. Qual parte do protótipo alimenta a produção: código, decisão ou ambos?
6. Por que um handoff deve referenciar o relatório em vez de copiá-lo?