Agente Mediador via SSE

Visão geral

O desenho anterior usava um WebSocket com rotas connect/disconnect/init/message/likedislike, mas o "assíncrono" era ilusório: o front travava esperando resposta mesmo com todo o backend rodando atrás de SQS. Essa doc registra o redesenho completo — por que SSE substitui o WS aqui, onde a arquitetura original tinha corridas escondidas, e como cada peça (Redis, SQS FIFO, DynamoDB, ALB, API Gateway) se encaixa.

Diagrama completo

Numeração 1–8 segue a execução real do fluxo de /message. Os passos 3 e 5 acontecem em paralelo — geração/streaming e persistência não dependem um do outro; nenhum bloqueia o outro.

Peça por peça

Auth & token exchange

EventSource nativo do browser não permite setar header Authorization — só query string ou cookie. Isso quebra o padrão de bearer herdado do Azure indo em header, que você já usa em todo o resto do sistema.

Solução: token exchange de curta duração. O front chama POST /stream/token com o bearer real do Azure no header (request normal, sem limitação nenhuma), o backend valida contra o EntraID com o mesmo padrão fail-closed do seu ACL atual, e devolve um JWT de ~60 segundos escopado pra aquele session_id. O front abre GET /stream?session_id=X&token=<curto> com esse token na query string.

@app.post("/stream/token")
async def issue_stream_token(request: Request, authorization: str = Header(None)):
    claims = validate_azure_bearer(authorization)
    now = int(time.time())
    short_token = jwt.encode(
        {
            "session_id": claims["session_id"],
            "client_id": claims["client_id"],
            "operator_id": claims["operator_id"],
            "scope": "stream",
            "iat": now,
            "exp": now + STREAM_TOKEN_TTL_S,  # 60s
        },
        STREAM_SIGNING_KEY,
        algorithm="HS256",
    )
    return {"token": short_token, "expires_in": STREAM_TOKEN_TTL_S}

O limite de 29s do API Gateway

É por isso que o diagrama tem dois caminhos de entrada separados: /stream vai direto no ALB via CloudFront, nunca toca o API Gateway. As rotas síncronas (/message, /warmup, /feedback, /stream/token) respondem em milissegundos e continuam atrás do API Gateway + VPC Link normalmente — o cap de 29s nunca chega perto de ser um problema pra elas.

Fan-out entre tasks do ECS

Sem sticky session no ALB, a task que segura a conexão SSE de um session_id não é necessariamente a task que recebe o POST /message daquele mesmo session_id. Sem um mecanismo de fan-out, o evento gerado pelo agente nunca chega no cliente certo.

Redis Pub/Sub resolve isso desacoplando completamente onde a geração acontece de onde a conexão está aberta: a task que processa a mensagem publica em session:{session_id}:events; qualquer task que esteja segurando aquele SSE está subscrita no canal e repassa pro EventSource do front.

Fluxo de /message

@app.post("/message")
async def post_message(request: Request):
    body = await request.json()
    session_id = body["session_id"]
    prompt = body["prompt"]
    message_id = str(uuid.uuid4())  # gerado na origem — chave de idempotência ponta a ponta

    # dois caminhos em paralelo, nenhum bloqueia o outro
    asyncio.create_task(_generate_and_stream(session_id, message_id, prompt))
    await _enqueue_for_persistence(session_id, message_id, role="user", content=prompt)

    return {"status": "accepted", "message_id": message_id}

Dual-write + idempotência

Dual-write elimina a corrida do TTL do Redis (migração pós-expiração pode perder dado se o job atrasar), mas introduz um problema novo: consistência entre dois writes independentes. Se o worker cair entre gravar no Redis e gravar no DynamoDB, e o SQS reentregar a mensagem, nenhum dos dois pode duplicar.

Resolvido com o mesmo padrão de idempotência que você já usa em produção — SET NX como dedupe key — mais ConditionExpression do lado do Dynamo:

def process_message(body):
    session_id = body["session_id"]
    message_id = body["message_id"]
    dedupe_key = f"processed:{message_id}"

    if not r.set(dedupe_key, "1", nx=True, ex=86400):
        return  # redelivery do SQS — já processado, sai sem duplicar

    r.xadd(f"session:{session_id}:history", {
        "message_id": message_id, "role": body["role"], "content": body["content"],
    })

    try:
        table.put_item(
            Item={
                "PK": f"SESSION#{session_id}",
                "SK": f"MSG#{body['ts']}#{message_id}",
                "role": body["role"], "content": body["content"],
            },
            ConditionExpression="attribute_not_exists(SK)",
        )
    except ClientError as e:
        if e.response["Error"]["Code"] != "ConditionalCheckFailedException":
            raise  # erro real — deixa o SQS re-tentar

Schema do DynamoDB

SK prefixado por timestamp garante ordenação lexicográfica nativa em Query, sem índice extra. Feedback é item próprio (SK: FEEDBACK#{message_id}) referenciando a mensagem original — nunca UpdateItem nela, pra manter Redis (append-only) e Dynamo simétricos e preservar trilha de auditoria de quando o feedback foi dado.

Ordem — por que FIFO e não standard

SQS FIFO com MessageGroupId = session_id garante ordem estrita dentro de uma sessão enquanto ainda permite paralelismo real entre sessões diferentes — cada session_id vira um grupo processado sequencialmente, grupos distintos escalam em paralelo entre os consumers. Só essa fila específica precisa ser FIFO; warmup e outras filas isoladas do sistema continuam standard.

sqs_client.send_message(
    QueueUrl=SQS_QUEUE_URL,
    MessageBody=json.dumps({...}),
    MessageGroupId=session_id,           # ordem garantida dentro da sessão
    MessageDeduplicationId=message_id,   # dedupe nativo do FIFO como camada extra
)

Warmup & a corrida escondida

Ordem correta: abre o SSE primeiro, dispara o warmup depois — nunca em paralelo. Mesmo assim, Pub/Sub é fire-and-forget: se o evento ready for publicado um milissegundo antes da task se inscrever no canal (deploy rolando, scale-out, latência de rede), a mensagem simplesmente deixa de existir pra qualquer um ler.

Warmup não gera XADD nem write no Dynamo — é sinalização de runtime, não conteúdo de conversa. Fica fora do histórico.

Reconexão & heartbeat

• Idle timeout do ALB (60s default) e do CloudFront (origin response timeout, também perto de 30s por padrão): comentário : keep-alive\n\n a cada 15–20s independente de haver evento real — silêncio durante um interrupt() longo não pode significar conexão morta.
• Last-Event-ID real, não decorativo: usa o ID do Redis Stream como id: de cada evento SSE. Em reconexão, o browser manda Last-Event-ID automaticamente — o servidor faz XRANGE a partir dali pra reenviar só o que perdeu, depois volta a assinar o pub/sub ao vivo.

if last_event_id:
    missed = await redis_client.xrange(history_key, min=f"({last_event_id}", max="+")
    for entry_id, fields in missed:
        yield _format_sse(entry_id, fields.get("role", "assistant"), fields)

Deploy sem cortar sessão ativa

ECS rolling deploy manda SIGTERM antes de matar a task antiga. Sem tratamento, qualquer SSE aberto ali morre abrupto — o cliente só percebe pelo timeout de leitura, alguns segundos de silêncio percebidos como travamento.

def handle_sigterm(signum, frame):
    logger.info("SIGTERM recebido, avisando %d conexões ativas", len(active_connections))
    shutdown_event.set()
    # respiro pro ALB parar de rotear novas requests antes de sair de vez
    # alinhar com deregistration_delay do target group

No loop do /stream, quando shutdown_event está setado, emite event: reconnect explícito antes de fechar. O front, ao receber, chama o token exchange de novo e reabre — cai numa task saudável via ALB sem o usuário perceber gap.

Trace propagation através do SQS

# publish
carrier = {}
propagate.inject(carrier)
sqs.send_message(
    ...,
    MessageAttributes={"traceparent": {"DataType": "String", "StringValue": carrier["traceparent"]}},
)

# consumer
ctx = propagate.extract({"traceparent": msg["MessageAttributes"]["traceparent"]["StringValue"]})
with tracer.start_as_current_span("process_message", context=ctx):
    ...

session_id como atributo em todo span — SSE connection, POST handler, mensagem SQS, worker, write no Redis/Dynamo — é o fio que segue uma conversa inteira ponta a ponta no X-Ray mesmo cruzando o boundary assíncrono do SQS.

Tabela de decisões

Handler completo

Código integral do handler — sse_handler.py — entregue junto com esta doc. Pontos de integração marcados com NotImplementedError são onde plugar o client real do LangGraph/Bedrock AgentCore Runtime e a validação do JWT do Azure contra o seu ACL existente.