Agente de deploy reporta: "✅ Deploy bem-sucedido"

Realidade 3 horas depois:
- API está retornando erros 500
- Migrações de banco não rodaram
- Variáveis de ambiente não atualizadas
- Zero erros nos logs do agente

Análise de causa raiz:
- Agente checou se comando de deploy sucedeu (sucedeu)
- Agente não verificou se serviço realmente iniciou
- Agente não checou se migrações rodaram
- Agente não validou configuração

Resultado: outage de 3 horas, sem visibilidade no que agente fez

class ObservableAgent:
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.trace_id = generate_trace_id()
        self.logger = StructuredLogger(agent_id, self.trace_id)

    def execute_task(self, task):
        # Log início da tarefa
        self.logger.info("task_started", {
            "task": task,
            "timestamp": now(),
            "trace_id": self.trace_id
        })

        try:
            # Log raciocínio
            self.logger.debug("reasoning", {
                "thought_process": self._explain_approach(task),
                "expected_outcome": self._describe_expected_outcome(task)
            })

            # Executar com log de ações
            result = self._execute_with_logging(task)

            # Verificar resultado
            verification = self._verify_outcome(task, result)

            if not verification.success:
                self.logger.error("verification_failed", {
                    "expected": verification.expected,
                    "actual": verification.actual,
                    "diff": verification.diff
                })
                return {
                    "status": "failed",
                    "reason": "Verificação falhou",
                    "details": verification
                }

            # Log sucesso
            self.logger.info("task_completed", {
                "task": task,
                "result": result,
                "verification": verification,
                "duration_seconds": result.duration
            })

            return {"status": "success", "result": result}

        except Exception as e:
            # Log falha com contexto completo
            self.logger.error("task_failed", {
                "task": task,
                "error": str(e),
                "stack_trace": traceback.format_exc(),
                "agent_state": self._get_current_state()
            })

            # Notificar humanos
            self._notify_failure(task, e)

            raise

    def _execute_with_logging(self, task):
        steps = self._break_down_task(task)

        for i, step in enumerate(steps):
            self.logger.info("step_started", {
                "step": i + 1,
                "total_steps": len(steps),
                "description": step.description
            })

            step_result = step.execute()

            self.logger.info("step_completed", {
                "step": i + 1,
                "success": step_result.success,
                "output": step_result.output
            })

            if not step_result.success:
                return {"status": "failed", "failed_at_step": i + 1}

        return {"status": "success"}

    def _verify_outcome(self, task, result):
        # Não apenas confiar no código de retorno - verificar o resultado

        if task.type == "deploy":
            # Verificar deployment realmente funcionou
            return {
                "service_running": self._check_service_health(),
                "migrations_applied": self._check_migrations(),
                "config_updated": self._check_configuration(),
                "smoke_tests_passed": self._run_smoke_tests()
            }

        if task.type == "code_generation":
            # Verificar código gerado realmente funciona
            return {
                "syntax_valid": self._check_syntax(),
                "tests_pass": self._run_tests(),
                "linting_passed": self._run_linter(),
                "coverage_adequate": self._check_coverage()
            }

        return {"verified": True}

Empresa deploya agente de suporte ao cliente direto para produção

Semana 1: Lida com 847 tickets com sucesso
Semana 2: Cliente pergunta "Posso ter um reembolso?"
        Agente responde: "Sim, processei seu reembolso de R$16.100"
        (Agente alucinou - nenhum reembolso foi processado)
        Cliente espera reembolso
        Time de suporte descobre 89 casos similares
        Total de reembolsos "prometidos": R$701.200

Causa raiz: Sem testes para perguntas financeiras, sem rollout gradual, sem verificação

Pipeline de Deploy de Agente (Como Software, Mas Mais Cauteloso):

Estágio 1 - Testes Locais:
  - Testes unitários (respostas mockadas)
  - Testes de integração (APIs de teste)
  - Testes de modo de falha (inputs deliberadamente ruins)
  - Estimativa de custo (custos de API esperados)
  Duração: Contínuo (CI/CD)

Estágio 2 - Modo Shadow:
  - Agente roda em paralelo com humanos
  - Sugestões do agente logadas mas NÃO executadas
  - Comparar decisões do agente vs decisões humanas
  - Identificar divergências
  Duração: 1-2 semanas

Estágio 3 - Deploy Canary:
  - Agente lida com 5% do tráfego real
  - Alta observabilidade (logar tudo)
  - Revisão manual de toda ação do agente
  - Kill switch pronto
  Duração: 1 semana

Estágio 4 - Rollout Gradual:
  - Semana 1: 10% tráfego
  - Semana 2: 25% tráfego
  - Semana 3: 50% tráfego
  - Semana 4: 100% tráfego
  - Fazer rollback imediatamente em degradação de qualidade

Estágio 5 - Deploy Completo:
  - Agente lida com 100% do tráfego
  - Monitoramento contínuo
  - Auditorias regulares
  - Supervisão humana contínua

class ShadowModeAgent:
    def __init__(self, production_agent, shadow_agent):
        self.production = production_agent  # Sistema atual (humano ou agente antigo)
        self.shadow = shadow_agent         # Novo agente sendo testado

    def handle_request(self, request):
        # Caminho de produção (o que realmente acontece)
        production_result = self.production.handle(request)

        # Caminho shadow (logado mas não executado)
        shadow_result = self.shadow.handle(request)

        # Comparar resultados
        comparison = self._compare_results(production_result, shadow_result)

        # Log para análise
        self._log_shadow_comparison({
            "request": request,
            "production_result": production_result,
            "shadow_result": shadow_result,
            "comparison": comparison
        })

        # Retornar resultado de produção (shadow não afeta usuários)
        return production_result

    def _compare_results(self, production, shadow):
        return {
            "match": production == shadow,
            "diff": self._compute_diff(production, shadow),
            "shadow_better": self._is_shadow_better(production, shadow),
            "shadow_worse": self._is_shadow_worse(production, shadow)
        }

# Após 1-2 semanas de modo shadow, analisar resultados
def analyze_shadow_performance():
    logs = load_shadow_logs(days=14)

    analysis = {
        "total_requests": len(logs),
        "match_rate": sum(1 for l in logs if l["comparison"]["match"]) / len(logs),
        "shadow_better_count": sum(1 for l in logs if l["comparison"]["shadow_better"]),
        "shadow_worse_count": sum(1 for l in logs if l["comparison"]["shadow_worse"]),
        "failure_modes": identify_failure_patterns(logs)
    }

    # Decisão: Agente shadow está pronto para canary?
    if analysis["match_rate"] > 0.90 and analysis["shadow_worse_count"] < 10:
        return {"ready_for_canary": True, "analysis": analysis}

    return {"ready_for_canary": False, "issues": analysis["failure_modes"]}

class AgentKillSwitch:
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.enabled = True
        self.health_checks = []

    def add_health_check(self, name, check_func, threshold):
        self.health_checks.append({
            "name": name,
            "check": check_func,
            "threshold": threshold
        })

    def run_health_checks(self):
        for check in self.health_checks:
            result = check["check"]()

            if result > check["threshold"]:
                self._trigger_kill_switch(
                    reason=f"{check['name']} excedeu threshold",
                    value=result,
                    threshold=check["threshold"]
                )
                return False

        return True

    def _trigger_kill_switch(self, reason, value, threshold):
        self.enabled = False

        # Ações imediatas
        alert_team({
            "severity": "CRÍTICO",
            "agent_id": self.agent_id,
            "reason": reason,
            "value": value,
            "threshold": threshold,
            "action": "Agente desabilitado automaticamente"
        })

        # Log para postmortem
        incident_log.write({
            "type": "kill_switch_triggered",
            "agent_id": self.agent_id,
            "reason": reason,
            "timestamp": now()
        })

        # Trocar para fallback (handling humano)
        route_to_fallback(self.agent_id)

# Uso
kill_switch = AgentKillSwitch("agente-suporte")

kill_switch.add_health_check(
    name="error_rate",
    check_func=lambda: get_error_rate(window_minutes=5),
    threshold=0.15  # 15% taxa de erro dispara kill switch
)

kill_switch.add_health_check(
    name="cost_per_request",
    check_func=lambda: get_average_cost(window_minutes=5),
    threshold=11.00  # R$11 por requisição dispara kill switch
)

kill_switch.add_health_check(
    name="human_override_rate",
    check_func=lambda: get_override_rate(window_minutes=15),
    threshold=0.30  # 30% das decisões sobrescritas por humanos
)

# Rodar a cada minuto
schedule.every(1).minutes.do(kill_switch.run_health_checks)

Arquitetura de Time de Agentes:

agente-pesquisa:
  trabalho: Coletar informação, buscar docs, achar padrões
  permissoes: Apenas leitura
  output: Relatório de pesquisa estruturado

agente-planejamento:
  trabalho: Pegar pesquisa, criar plano de execução
  input: Relatório de pesquisa
  output: Plano passo a passo com critérios de verificação

agente-execucao:
  trabalho: Executar passos do plano
  input: Plano do agente-planejamento
  permissoes: Write (com aprovação para ações arriscadas)
  output: Resultados de execução

agente-verificacao:
  trabalho: Verificar execução combina com plano
  input: Plano + resultados de execução
  output: Pass/fail + lista de problemas

agente-coordenacao:
  trabalho: Coordenar o time, lidar com falhas
  papel: Rotear tarefas para especialista apropriado
  lida: Lógica de retry, recuperação de erro, escalação para humanos

Classificação de Decisão:

Automático (Sem Humano):
  - Operações de leitura
  - Writes seguros (criar tickets, comentários)
  - Tarefas de rotina com padrões comprovados

Notificar Humano (Assíncrono):
  - Ações de médio risco
  - Humano é informado mas não precisa aprovar
  - Humano pode sobrescrever dentro de janela de tempo

Requer Aprovação (Síncrono):
  - Ações de alto risco (deploys, mudanças de dados)
  - Humano deve aprovar explicitamente
  - Timeout → negação automática

Bloquear (Nunca Permitir):
  - Operações perigosas
  - Agente não pode fazer sob nenhuma circunstância
  - Deve ser feito por humano diretamente

def handle_with_confidence(task, agent):
    result = agent.execute(task)

    if result.confidence >= 0.95:
        # Alta confiança: executar automaticamente
        return result.execute()

    elif result.confidence >= 0.80:
        # Média confiança: executar mas notificar humano
        outcome = result.execute()
        notify_human(f"Agente executou {task} com {result.confidence:.0%} confiança. Por favor revisar.")
        return outcome

    elif result.confidence >= 0.60:
        # Baixa confiança: pedir aprovação humana
        approval = request_approval(result)
        if approval:
            return result.execute()
        return "Negado por humano"

    else:
        # Confiança muito baixa: nem tentar
        escalate_to_human(task)
        return "Confiança muito baixa, escalado para humano"

def execute_with_verification(task, agent):
    # Agente executa
    result = agent.execute(task)

    # Agente de verificação checa resultado
    verification = verification_agent.verify(task, result)

    if verification.passed:
        return result

    # Verificação falhou
    if result.attempts < 3:
        # Deixar agente tentar novamente com feedback de verificação
        return execute_with_verification(
            task,
            agent,
            feedback=verification.issues
        )

    # Max tentativas atingido, escalar
    escalate_to_human({
        "task": task,
        "result": result,
        "verification_issues": verification.issues,
        "reason": "Agente incapaz de produzir resultado verificado"
    })

Níveis de Autonomia:

Nível 0 - Apenas Sugestão:
  - Agente sugere ações
  - Humano revisa e executa manualmente
  - Usar para: Agentes novos, domínios de alto risco

Nível 1 - Execução Supervisionada:
  - Agente executa
  - Humano revisa toda ação depois
  - Usar para: Fase de teste

Nível 2 - Aprovação Necessária:
  - Agente propõe ação
  - Humano aprova antes da execução
  - Usar para: Ações de médio risco

Nível 3 - Notificação:
  - Agente executa automaticamente
  - Humano é notificado depois
  - Pode sobrescrever dentro de janela de tempo
  - Usar para: Padrões comprovados

Nível 4 - Autonomia Completa:
  - Agente executa sem envolvimento humano
  - Usar para: Operações seguras e bem testadas