Guia de Debug - unified_bot.py

🐛 Métodos de Debug para SSE Endpoints

1. Debug via Logging (Mais Simples)

Configurar Nível de Log

# .env
LOG_LEVEL=DEBUG  # Altere de INFO para DEBUG

Adicionar Logs Estratégicos nos Endpoints SSE

# unified_bot.py - No endpoint /sse/chat
logger.debug(f"📋 Request headers: {dict(request.headers)}")
logger.debug(f"📋 Request body: {await request.json()}")
logger.debug(f"🔐 JWT token extraído: {jwt_token[:20]}..." if jwt_token else "Sem JWT")

# unified_bot.py - No endpoint /sse/stream/{stream_id}
logger.debug(f"🍪 Cookies recebidos: {request.cookies}")
logger.debug(f"🔍 Query params: {dict(request.query_params)}")
logger.debug(f"📡 Stream {stream_id} - eventos: {len(list(stream_manager.get_stream(stream_id).events))}")

Rodar com Debug Ativado

# Local
LOG_LEVEL=DEBUG python unified_bot.py

# Com uvicorn
LOG_LEVEL=DEBUG uvicorn unified_bot:app --reload --log-level debug

---

2. Debug com VS Code (Recomendado)

Criar .vscode/launch.json

{
  "version": "0.2.0",
  "configurations": [
    {
      "name": "Python: Debug unified_bot.py",
      "type": "debugpy",
      "request": "launch",
      "module": "uvicorn",
      "args": [
        "unified_bot:app",
        "--reload",
        "--host", "0.0.0.0",
        "--port", "8080"
      ],
      "jinja": true,
      "justMyCode": false,
      "env": {
        "LOG_LEVEL": "DEBUG"
      },
      "console": "integratedTerminal"
    },
    {
      "name": "Python: Debug SSE Only",
      "type": "debugpy",
      "request": "launch",
      "module": "uvicorn",
      "args": [
        "unified_bot:app",
        "--reload",
        "--host", "0.0.0.0",
        "--port", "8080",
        "--log-level", "debug"
      ],
      "jinja": true,
      "justMyCode": false,
      "env": {
        "LOG_LEVEL": "DEBUG",
        "SSE_ENABLED": "true"
      },
      "console": "integratedTerminal",
      "preLaunchTask": "Activate venv"
    }
  ]
}

Como Usar

1. Abra unified_bot.py no VS Code
2. Coloque breakpoints (clique na margem esquerda) nos endpoints SSE:
3. Linha do @app.post("/sse/chat")
4. Linha do @app.get("/sse/stream/{stream_id}")
5. Dentro do event_generator()
6. Aperte F5 ou vá em Run > Start Debugging
7. Faça requisição para http://localhost:8080/sse/chat
8. Debug pausará nos breakpoints

Breakpoints Estratégicos

# unified_bot.py

@app.post("/sse/chat")
async def sse_chat_start(request: Request, response: Response):
    # 🔴 BREAKPOINT 1: Entrada da requisição
    adapter = adapter_factory.get_adapter("sse")

    # 🔴 BREAKPOINT 2: Parse da mensagem
    incoming_message = await adapter.parse_webhook_request(request)

    # 🔴 BREAKPOINT 3: JWT do metadata
    jwt_token = incoming_message.metadata.get("jwt_token")

    # 🔴 BREAKPOINT 4: Criação do stream
    stream_id = adapter.create_stream(incoming_message)

    # 🔴 BREAKPOINT 5: Cookie configurado
    if jwt_token:
        response.set_cookie(...)

@app.get("/sse/stream/{stream_id}")
async def sse_stream(...):
    # 🔴 BREAKPOINT 6: Extração de JWT
    jwt_token = request.cookies.get("jwt_token")

    # 🔴 BREAKPOINT 7: Validação JWT
    if jwt_token:
        user_context = await validate_jwt(jwt_token)

    # 🔴 BREAKPOINT 8: Stream encontrado
    stream = stream_manager.get_stream(stream_id)

    # 🔴 BREAKPOINT 9: Generator de eventos
    async def event_generator():
        async for event in stream_manager.subscribe_to_stream(...):
            # 🔴 BREAKPOINT 10: Cada evento SSE
            yield sse_message

---

3. Debug Remoto com debugpy (Docker/Cloud Run)

Adicionar debugpy ao código

# unified_bot.py - No topo do arquivo

import os

# Debug remoto
if os.getenv("ENABLE_DEBUGPY", "false").lower() == "true":
    import debugpy
    debugpy.listen(("0.0.0.0", 5678))
    print("🐛 Debugpy rodando na porta 5678. Aguardando conexão...")
    # debugpy.wait_for_client()  # Descomente para bloquear até conectar

Configurar .env

ENABLE_DEBUGPY=true

VS Code launch.json para Remote

{
  "name": "Python: Remote Attach",
  "type": "debugpy",
  "request": "attach",
  "connect": {
    "host": "localhost",
    "port": 5678
  },
  "pathMappings": [
    {
      "localRoot": "${workspaceFolder}",
      "remoteRoot": "/app"
    }
  ]
}

Docker Compose com Debug

# docker-compose.yml
services:
  unified-bot:
    ports:
      - "8080:8080"
      - "5678:5678"  # Porta de debug
    environment:
      - ENABLE_DEBUGPY=true

---

4. Debug de SSE com cURL e HTTPie

Testar POST /sse/chat

# Com cURL
curl -X POST http://localhost:8080/sse/chat \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "test_user",
    "message": "Olá!",
    "jwt_token": "eyJ..."
  }' \
  -v  # Verbose para ver headers

# Com HTTPie (mais legível)
http POST localhost:8080/sse/chat \
  user_id=test_user \
  message="Olá!" \
  jwt_token="eyJ..." \
  --verbose

Testar GET /sse/stream com curl

# Sem auth
curl -N http://localhost:8080/sse/stream/sse_abc123

# Com JWT via query
curl -N "http://localhost:8080/sse/stream/sse_abc123?token=eyJ..."

# Com cookie
curl -N http://localhost:8080/sse/stream/sse_abc123 \
  -H "Cookie: jwt_token=eyJ..."

Com EventSource via Browser Console

// Abra DevTools > Console
const es = new EventSource('http://localhost:8080/sse/stream/sse_abc123');

es.addEventListener('stream_start', (e) => {
  console.log('🚀 Stream Start:', JSON.parse(e.data));
});

es.addEventListener('text_chunk', (e) => {
  console.log('📝 Text:', JSON.parse(e.data));
});

es.addEventListener('stream_complete', (e) => {
  console.log('✅ Complete:', JSON.parse(e.data));
  es.close();
});

es.onerror = (e) => {
  console.error('❌ Error:', e);
};

---

5. Debug de Streaming com Python

# test_sse_debug.py
import asyncio
import httpx
import json

async def test_sse_stream():
    async with httpx.AsyncClient() as client:
        # 1. POST para criar stream
        print("📤 Criando stream...")
        response = await client.post(
            "http://localhost:8080/sse/chat",
            json={
                "user_id": "debug_user",
                "message": "Teste de debug",
                "jwt_token": "eyJ..."
            }
        )

        data = response.json()
        stream_id = data["stream_id"]
        print(f"✅ Stream criado: {stream_id}")

        # Pegar cookie se presente
        cookies = response.cookies
        print(f"🍪 Cookies: {cookies}")

        # 2. GET para conectar ao stream
        print(f"\n📡 Conectando ao stream {stream_id}...")

        async with client.stream(
            "GET",
            f"http://localhost:8080/sse/stream/{stream_id}",
            cookies=cookies,
            timeout=60.0
        ) as stream_response:
            print(f"Status: {stream_response.status_code}")
            print(f"Headers: {stream_response.headers}")

            # 3. Ler eventos SSE
            async for line in stream_response.aiter_lines():
                if line.startswith("event:"):
                    event_type = line.split(":", 1)[1].strip()
                    print(f"\n🎯 Event: {event_type}")
                elif line.startswith("data:"):
                    data = line.split(":", 1)[1].strip()
                    event_data = json.loads(data)
                    print(f"   Data: {event_data}")
                elif line.startswith("id:"):
                    event_id = line.split(":", 1)[1].strip()
                    print(f"   ID: {event_id}")

if __name__ == "__main__":
    asyncio.run(test_sse_stream())

Executar:

python test_sse_debug.py

---

6. Monitoramento em Produção (Cloud Run)

Ver Logs em Tempo Real

# Cloud Run
gcloud run services logs tail ifriend-agent-unified \
  --region=us-central1 \
  --project=ifriend-platform

# Filtrar apenas SSE
gcloud run services logs tail ifriend-agent-unified \
  --region=us-central1 \
  --project=ifriend-platform \
  | grep -i "sse\|stream"

Structured Logging para Debug

# unified_bot.py
import json
from datetime import datetime

def log_sse_debug(event_type: str, data: dict):
    """Log estruturado para Cloud Logging"""
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "severity": "DEBUG",
        "component": "sse",
        "event_type": event_type,
        **data
    }
    print(json.dumps(log_entry))

# Uso nos endpoints
log_sse_debug("stream_created", {
    "stream_id": stream_id,
    "user_id": incoming_message.user_id,
    "has_jwt": bool(jwt_token)
})

---

7. Ferramentas de Debug Úteis

Install

pip install debugpy httpx ipdb

IPython Debugger (ipdb)

# unified_bot.py - Qualquer lugar
import ipdb; ipdb.set_trace()  # Breakpoint interativo

Rich para Logs Bonitos

# unified_bot.py
from rich import print as rprint
from rich.console import Console

console = Console()

# Logs formatados
console.log(f"[green]✅ Stream criado:[/green] {stream_id}")
console.log("[red]❌ Erro:[/red]", exc_info=True)

# Print de objetos
rprint(incoming_message)

---

🎯 Checklist de Debug SSE

Antes de Debugar

• [ ] LOG_LEVEL=DEBUG no .env
• [ ] Servidor rodando com --reload
• [ ] DevTools aberto (Network + Console)
• [ ] Breakpoints colocados

Durante Debug

• [ ] Verificar headers da requisição
• [ ] Verificar cookies enviados/recebidos
• [ ] Verificar JWT extraído e validado
• [ ] Verificar stream_id retornado
• [ ] Verificar eventos SSE sendo gerados

Problemas Comuns

Sintoma	Causa	Debug
Stream não encontrado	stream_id incorreto	Verificar response do POST
Cookie não enviado	SameSite/Secure	DevTools > Application > Cookies
JWT inválido	Token expirado	Breakpoint na validação
Eventos não chegam	Generator travado	Breakpoint no event_generator()
CORS erro	Origem não permitida	Ver logs "CORS origens"

---

🛡️ Loop Guard — Diagnosticando Bloqueios Anti-Loop

O LoopGuard (messaging/loop_guard.py) protege contra loops bot-to-bot com 4 camadas. Logs aparecem com prefixo [LoopGuard].

Identificando bloqueios nos logs

[LoopGuard] DUPLICATE platform=whatsapp user=5511... message_id=xyz
[LoopGuard] CIRCUIT_OPEN platform=whatsapp user=5511... reason=rate_limit burst=6/5
[LoopGuard] CIRCUIT_OPEN platform=whatsapp user=5511... reason=repetition count=3
[LoopGuard] TURN_LIMIT platform=whatsapp user=5511... session=... turns=201

Ajustando thresholds via env vars

# Aumentar tolerância (ex: humano digitando rápido)
LOOP_GUARD_MAX_MESSAGES=10        # default: 5
LOOP_GUARD_WINDOW_SECONDS=15      # default: 10

# Reduzir cooldown para testes
LOOP_GUARD_COOLDOWN_SECONDS=5     # default: 30

# Aumentar limite de turnos
LOOP_GUARD_MAX_TURNS_PER_SESSION=500  # default: 200

# Tolerância a repetições
LOOP_GUARD_REPEAT_THRESHOLD=5     # default: 3
LOOP_GUARD_REPEAT_WINDOW_SECONDS=120  # default: 60

Rodando testes

python -m pytest tests/test_loop_guard.py -v

---

Última atualização: 15/02/2026