Redis & DynamoDB
Memory Architecture

session:{session_id}              → Hash    TTL: 2h sliding
session:{session_id}:user         → String  TTL: 2h sliding   valor: user_id
user:{user_id}:active_session     → String  TTL: 2h sliding   valor: session_id

{
  "session_id":  "sess_abc123",
  "user_id":     "usr_xyz789",
  "started_at":  "2025-01-15T10:00:00Z",
  "last_active": "2025-01-15T10:45:00Z",
  "status":      "active",
  "metadata":    "{\"channel\": \"web\", \"ip\": \"...\"}"
}

thread:{thread_id}:state          → Hash    TTL: 30min sliding
thread:{thread_id}:messages       → List    TTL: 30min sliding
thread:{thread_id}:checkpoint     → String  TTL: 30min sliding  JSON do LangGraph state
session:{session_id}:current_thread → String TTL: 2h            valor: thread_id

{
  "thread_id":   "thrd_def456",
  "session_id":  "sess_abc123",
  "user_id":     "usr_xyz789",
  "created_at":  "2025-01-15T10:05:00Z",
  "last_active": "2025-01-15T10:40:00Z",
  "status":      "active",
  "agent_route": "billing_agent",
  "turn_count":  "7"
}

mem:pref:{user_id}                → Hash    sem TTL  (invalidação por write-through)
mem:ltm:{user_id}:summaries       → List    sem TTL  (LPUSH + LTRIM — top 10)
mem:wm:{thread_id}                → Hash    TTL: 30min (mesma vida da thread)

{
  "language":        "pt-BR",
  "response_style":  "concise",
  "segment":         "fintech",
  "plan":            "enterprise",
  "last_updated":    "2025-01-15T10:00:00Z",
  "inferred_count":  "12"
}

{
  "session_id": "sess_prev001",
  "summary":    "Usuário perguntou sobre integração via API REST. Resolvido com sucesso.",
  "topics":     ["api", "integration", "billing"],
  "sentiment":  "positive",
  "resolved":   true,
  "created_at": "2025-01-14T15:30:00Z"
}

{
  "PK":          "USER#usr_xyz789",
  "SK":          "SESSION#sess_abc123#MSG#2025-01-15T10:05:00.000Z",
  "message_id":  "msg_001",
  "thread_id":   "thrd_def456",
  "role":        "user",
  "content":     "Preciso entender minha fatura",
  "agent_name":  null,
  "token_count": 8,
  "ttl":         1744924800
}

{
  "PK":         "USER#usr_xyz789",
  "SK":         "SUMMARY#sess_abc123",
  "session_id": "sess_abc123",
  "summary":    "Usuário perguntou sobre integração via API REST. Resolvido.",
  "topics":     ["api", "integration"],
  "sentiment":  "positive",
  "resolved":   true,
  "turn_count": 7,
  "started_at": "2025-01-15T10:00:00Z",
  "ended_at":   "2025-01-15T10:50:00Z",
  "ttl":        1744924800
}

import redis.asyncio as aioredis

class RedisClients:
    """Singleton para os 3 clusters Redis."""

    def __init__(self, config: dict):
        self.sessions = aioredis.Redis(
            host=config["sessions"]["host"],
            port=config["sessions"]["port"],
            decode_responses=True,
            ssl=True,
        )
        self.threads = aioredis.Redis(
            host=config["threads"]["host"],
            port=config["threads"]["port"],
            decode_responses=True,
            ssl=True,
        )
        self.memory = aioredis.Redis(
            host=config["memory"]["host"],
            port=config["memory"]["port"],
            decode_responses=True,
            ssl=True,
        )

    async def close(self):
        await self.sessions.aclose()
        await self.threads.aclose()
        await self.memory.aclose()

import json
from datetime import datetime, timezone
from typing import Optional
from infrastructure.redis_clients import RedisClients

SESSION_TTL = 7200   # 2h
THREAD_TTL  = 1800   # 30min

class SessionStore:
    def __init__(self, redis: RedisClients):
        self.r = redis.sessions

    async def create_session(
        self, session_id: str, user_id: str, metadata: dict = {}
    ) -> dict:
        now = datetime.now(timezone.utc).isoformat()
        session = {
            "session_id":  session_id,
            "user_id":     user_id,
            "started_at":  now,
            "last_active": now,
            "status":      "active",
            "metadata":    json.dumps(metadata),
        }
        pipe = self.r.pipeline()
        pipe.hset(f"session:{session_id}", mapping=session)
        pipe.expire(f"session:{session_id}", SESSION_TTL)
        pipe.set(f"session:{session_id}:user", user_id, ex=SESSION_TTL)
        pipe.set(f"user:{user_id}:active_session", session_id, ex=SESSION_TTL)
        await pipe.execute()
        return session

    async def get_session(self, session_id: str) -> Optional[dict]:
        session = await self.r.hgetall(f"session:{session_id}")
        if not session:
            return None
        # Sliding TTL — renova nas 3 chaves ao mesmo tempo
        pipe = self.r.pipeline()
        pipe.expire(f"session:{session_id}", SESSION_TTL)
        pipe.expire(f"session:{session_id}:user", SESSION_TTL)
        user_id = session.get("user_id")
        if user_id:
            pipe.expire(f"user:{user_id}:active_session", SESSION_TTL)
        await pipe.execute()
        return session

    async def get_active_session_for_user(self, user_id: str) -> Optional[str]:
        return await self.r.get(f"user:{user_id}:active_session")

    async def touch_session(self, session_id: str) -> None:
        now = datetime.now(timezone.utc).isoformat()
        await self.r.hset(f"session:{session_id}", "last_active", now)
        await self.r.expire(f"session:{session_id}", SESSION_TTL)

    async def close_session(self, session_id: str) -> None:
        user_id = await self.r.get(f"session:{session_id}:user")
        pipe = self.r.pipeline()
        pipe.hset(f"session:{session_id}", "status", "closed")
        pipe.expire(f"session:{session_id}", 300)  # mantém 5min para cleanup
        if user_id:
            pipe.delete(f"user:{user_id}:active_session")
        await pipe.execute()

import json
from typing import AsyncIterator
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
    BaseCheckpointSaver, Checkpoint, CheckpointMetadata, CheckpointTuple
)
from infrastructure.redis_clients import RedisClients

THREAD_TTL = 1800  # 30min

class RedisCheckpointer(BaseCheckpointSaver):
    """
    Checkpointer LangGraph usando redis-threads.
    Sliding TTL renovado a cada aput() — garante que o TTL
    reflita inatividade real, não tempo desde a criação.
    """

    def __init__(self, redis: RedisClients):
        super().__init__()
        self.r = redis.threads

    def _checkpoint_key(self, thread_id: str) -> str:
        return f"thread:{thread_id}:checkpoint"

    def _meta_key(self, thread_id: str) -> str:
        return f"thread:{thread_id}:state"

    # interface sync — não utilizada
    def get_tuple(self, config): raise NotImplementedError("Use async")
    def list(self, config, **kw): raise NotImplementedError("Use async")
    def put(self, config, checkpoint, metadata): raise NotImplementedError("Use async")

    async def aget_tuple(self, config: RunnableConfig):
        thread_id = config["configurable"]["thread_id"]
        raw = await self.r.get(self._checkpoint_key(thread_id))
        if not raw:
            return None
        # Sliding TTL ao ler
        await self.r.expire(self._checkpoint_key(thread_id), THREAD_TTL)
        await self.r.expire(self._meta_key(thread_id), THREAD_TTL)
        data = json.loads(raw)
        return CheckpointTuple(
            config=config,
            checkpoint=data["checkpoint"],
            metadata=data["metadata"],
            parent_config=data.get("parent_config"),
        )

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        thread_id = config["configurable"]["thread_id"]
        import datetime as dt
        data = json.dumps({
            "checkpoint":    checkpoint,
            "metadata":      metadata,
            "parent_config": config,
        })
        pipe = self.r.pipeline()
        pipe.set(self._checkpoint_key(thread_id), data, ex=THREAD_TTL)
        pipe.hset(self._meta_key(thread_id), mapping={
            "last_active": dt.datetime.now(dt.timezone.utc).isoformat(),
            "turn_count":  str(metadata.get("step", 0)),
        })
        pipe.expire(self._meta_key(thread_id), THREAD_TTL)
        await pipe.execute()
        return config

    async def alist(self, config: RunnableConfig, **kw) -> AsyncIterator:
        result = await self.aget_tuple(config)
        if result:
            yield result

import json
from typing import Optional
from infrastructure.redis_clients import RedisClients

WM_TTL            = 1800  # Working memory: 30min
MAX_LTM_SUMMARIES = 10    # Top 10 sessões no cache

class MemoryStore:
    def __init__(self, redis: RedisClients):
        self.r = redis.memory

    # ── UserPreferences ────────────────────────────────────────────────
    async def get_preferences(self, user_id: str) -> Optional[dict]:
        """None = cache miss → Supervisor faz fallback ao DynamoDB."""
        raw = await self.r.hgetall(f"mem:pref:{user_id}")
        return raw if raw else None

    async def set_preferences(self, user_id: str, prefs: dict) -> None:
        """Chamado após escrita no DynamoDB (write-through). Sem TTL."""
        await self.r.hset(f"mem:pref:{user_id}", mapping=prefs)

    # ── LTM Summaries ──────────────────────────────────────────────────
    async def get_ltm_summaries(self, user_id: str, n: int = 5) -> list[dict]:
        raw_list = await self.r.lrange(f"mem:ltm:{user_id}:summaries", 0, n - 1)
        return [json.loads(s) for s in raw_list]

    async def push_ltm_summary(self, user_id: str, summary: dict) -> None:
        key = f"mem:ltm:{user_id}:summaries"
        pipe = self.r.pipeline()
        pipe.lpush(key, json.dumps(summary))          # mais recente na frente
        pipe.ltrim(key, 0, MAX_LTM_SUMMARIES - 1)    # mantém top 10
        await pipe.execute()

    # ── Working Memory ─────────────────────────────────────────────────
    async def get_working_memory(self, thread_id: str) -> Optional[dict]:
        raw = await self.r.hgetall(f"mem:wm:{thread_id}")
        return raw if raw else None

    async def set_working_memory(self, thread_id: str, data: dict) -> None:
        pipe = self.r.pipeline()
        pipe.hset(f"mem:wm:{thread_id}", mapping=data)
        pipe.expire(f"mem:wm:{thread_id}", WM_TTL)
        await pipe.execute()

    async def update_working_memory(self, thread_id: str, updates: dict) -> None:
        pipe = self.r.pipeline()
        pipe.hset(f"mem:wm:{thread_id}", mapping=updates)
        pipe.expire(f"mem:wm:{thread_id}", WM_TTL)  # renova TTL
        await pipe.execute()

import boto3
from datetime import datetime, timezone, timedelta
from typing import Optional
from boto3.dynamodb.conditions import Key

def _ttl_epoch(days: int = 90) -> int:
    return int((datetime.now(timezone.utc) + timedelta(days=days)).timestamp())

class DynamoRepository:
    def __init__(self, region: str = "us-east-1"):
        ddb = boto3.resource("dynamodb", region_name=region)
        self.history  = ddb.Table("ConversationHistory")
        self.prefs    = ddb.Table("UserPreferences")
        self.sessions = ddb.Table("SessionIndex")

    # ── ConversationHistory ────────────────────────────────────────────
    async def save_message(
        self, user_id: str, session_id: str, thread_id: str,
        message_id: str, role: str, content: str,
        agent_name: Optional[str] = None, token_count: int = 0,
    ) -> None:
        ts = datetime.now(timezone.utc).isoformat()
        self.history.put_item(Item={
            "PK": f"USER#{user_id}",
            "SK": f"SESSION#{session_id}#MSG#{ts}",
            "message_id": message_id, "thread_id": thread_id,
            "role": role, "content": content,
            "agent_name": agent_name or "supervisor",
            "token_count": token_count, "ttl": _ttl_epoch(),
        })

    async def save_session_summary(
        self, user_id: str, session_id: str, summary: str,
        topics: list, sentiment: str, resolved: bool,
        turn_count: int, started_at: str, ended_at: str,
    ) -> None:
        self.history.put_item(Item={
            "PK": f"USER#{user_id}", "SK": f"SUMMARY#{session_id}",
            "session_id": session_id, "summary": summary,
            "topics": topics, "sentiment": sentiment,
            "resolved": resolved, "turn_count": turn_count,
            "started_at": started_at, "ended_at": ended_at,
            "ttl": _ttl_epoch(),
        })

    async def get_session_messages(self, user_id: str, session_id: str) -> list:
        resp = self.history.query(
            KeyConditionExpression=
                Key("PK").eq(f"USER#{user_id}") &
                Key("SK").begins_with(f"SESSION#{session_id}#MSG#"),
        )
        return sorted(resp["Items"], key=lambda x: x["SK"])

    async def get_recent_summaries(self, user_id: str, limit: int = 5) -> list:
        resp = self.history.query(
            IndexName="SessionSummaryIndex",
            KeyConditionExpression=
                Key("PK").eq(f"USER#{user_id}") &
                Key("SK").begins_with("SUMMARY#"),
            ScanIndexForward=False,
            Limit=limit,
        )
        return resp["Items"]

    # ── UserPreferences ────────────────────────────────────────────────
    async def get_preferences(self, user_id: str) -> Optional[dict]:
        resp = self.prefs.get_item(
            Key={"PK": f"USER#{user_id}", "SK": "PREF#PROFILE"}
        )
        return resp.get("Item")

    async def save_preferences(self, user_id: str, prefs: dict) -> None:
        now = datetime.now(timezone.utc).isoformat()
        self.prefs.put_item(Item={
            "PK": f"USER#{user_id}", "SK": "PREF#PROFILE",
            **prefs, "last_updated": now,
        })

    async def save_preference_fact(
        self, user_id: str, fact_key: str, fact_value: str,
        confidence: float, source_session: str,
    ) -> None:
        self.prefs.put_item(Item={
            "PK": f"USER#{user_id}", "SK": f"PREF#FACT#{fact_key}",
            "fact_key": fact_key, "fact_value": fact_value,
            "confidence": str(confidence),
            "inferred_at": datetime.now(timezone.utc).isoformat(),
            "source_session": source_session,
        })

    # ── SessionIndex ───────────────────────────────────────────────────
    async def register_session(
        self, user_id: str, session_id: str, status: str = "active"
    ) -> None:
        now = datetime.now(timezone.utc).isoformat()
        self.sessions.put_item(Item={
            "PK": f"USER#{user_id}",
            "SK": f"SESSION#{now}#{session_id}",
            "session_id": session_id, "status": status,
            "thread_count": 0, "ttl": _ttl_epoch(),
        })

import json
from typing import TypedDict, Optional
from memory.session_store     import SessionStore
from memory.memory_store      import MemoryStore
from memory.dynamo_repository import DynamoRepository

class EnrichedContext(TypedDict):
    user_id:          str
    session_id:       str
    thread_id:        str
    preferences:      dict
    ltm_summaries:    list[dict]
    working_memory:   dict
    session_metadata: dict

class ContextEnricher:
    def __init__(self, sessions: SessionStore, memory: MemoryStore, dynamo: DynamoRepository):
        self.sessions = sessions
        self.memory   = memory
        self.dynamo   = dynamo

    async def enrich(self, session_id: str, thread_id: str, user_id: str) -> EnrichedContext:
        # 1. Session metadata
        session = await self.sessions.get_session(session_id) or {}

        # 2. UserPreferences — Redis primeiro, fallback DynamoDB (warm-up automático)
        preferences = await self.memory.get_preferences(user_id)
        if not preferences:
            preferences = await self.dynamo.get_preferences(user_id) or {}
            if preferences:
                await self.memory.set_preferences(user_id, preferences)

        # 3. LTM Summaries — Redis primeiro, fallback DynamoDB
        ltm_summaries = await self.memory.get_ltm_summaries(user_id, n=5)
        if not ltm_summaries:
            ltm_summaries = await self.dynamo.get_recent_summaries(user_id, limit=5)

        # 4. Working memory da thread ativa
        working_memory = await self.memory.get_working_memory(thread_id) or {}

        return EnrichedContext(
            user_id=user_id, session_id=session_id, thread_id=thread_id,
            preferences=preferences, ltm_summaries=ltm_summaries,
            working_memory=working_memory, session_metadata=session,
        )

    def build_context_prompt(self, ctx: EnrichedContext) -> str:
        """Serializa contexto como texto para o system prompt."""
        parts = []
        if ctx["preferences"]:
            parts.append(
                f"## User Profile\n{json.dumps(ctx['preferences'], indent=2, ensure_ascii=False)}"
            )
        if ctx["ltm_summaries"]:
            lines = "\n".join(
                f"- [{s.get('created_at','')[:10]}] {s.get('summary','')}"
                for s in ctx["ltm_summaries"]
            )
            parts.append(f"## Recent Session History\n{lines}")
        if ctx["working_memory"]:
            parts.append(
                f"## Current Working Memory\n{json.dumps(ctx['working_memory'], indent=2, ensure_ascii=False)}"
            )
        return "\n\n".join(parts)

from typing import TypedDict, Annotated, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from memory.langgraph_checkpointer import RedisCheckpointer
from supervisor.context_enricher   import ContextEnricher
from supervisor.thread_archiver    import ThreadArchiver

class AgentState(TypedDict):
    messages:         Annotated[list[BaseMessage], add_messages]
    session_id:       str
    thread_id:        str
    user_id:          str
    enriched_context: dict
    agent_route:      Optional[str]
    resolved:         bool

# ── Nó: enriquecimento ───────────────────────────────────────────────
async def enrich_context_node(state: AgentState, enricher: ContextEnricher) -> AgentState:
    ctx = await enricher.enrich(
        session_id=state["session_id"],
        thread_id=state["thread_id"],
        user_id=state["user_id"],
    )
    context_prompt = enricher.build_context_prompt(ctx)
    system_msg = SystemMessage(content=f"""Você é o Supervisor de um sistema multi-agente.

{context_prompt}

Roteie a solicitação para o agente correto.
Agentes: billing_agent, technical_agent, onboarding_agent.""")
    return {
        **state,
        "enriched_context": dict(ctx),
        "messages": [system_msg] + [m for m in state["messages"]
                                    if not isinstance(m, SystemMessage)],
    }

# ── Nó: supervisor LLM ───────────────────────────────────────────────
async def supervisor_node(state: AgentState) -> AgentState:
    llm = ChatAnthropic(model="claude-sonnet-4-20250514")
    response = await llm.ainvoke(state["messages"])
    return {**state, "messages": [response]}

# ── Nó: roteamento MCP ───────────────────────────────────────────────
async def route_to_agent_node(state: AgentState) -> AgentState:
    mcp_payload = {
        "thread_id": state["thread_id"],
        "user_id":   state["user_id"],
        "context":   state["enriched_context"],
        "messages":  [m.dict() for m in state["messages"][-10:]],
    }
    # agent_response = await mcp_client.call(state["agent_route"], mcp_payload)
    return state

# ── Nó: verificar encerramento ────────────────────────────────────────
async def check_resolution_node(state: AgentState, archiver: ThreadArchiver) -> AgentState:
    last = state["messages"][-1] if state["messages"] else None
    resolved = False
    if last and isinstance(last, AIMessage):
        signals = ["posso ajudar com mais algo", "há algo mais", "problema resolvido"]
        resolved = any(s in last.content.lower() for s in signals)
    if resolved:
        await archiver.archive_thread(
            thread_id=state["thread_id"],
            session_id=state["session_id"],
            user_id=state["user_id"],
            messages=state["messages"],
        )
    return {**state, "resolved": resolved}

def should_end(state: AgentState) -> str:
    return END if state.get("resolved") else "supervisor"

def build_supervisor_graph(
    enricher: ContextEnricher,
    archiver: ThreadArchiver,
    checkpointer: RedisCheckpointer,
) -> StateGraph:
    graph = StateGraph(AgentState)
    graph.add_node("enrich_context",    lambda s: enrich_context_node(s, enricher))
    graph.add_node("supervisor",        supervisor_node)
    graph.add_node("route_agent",       route_to_agent_node)
    graph.add_node("check_resolution",  lambda s: check_resolution_node(s, archiver))
    graph.set_entry_point("enrich_context")
    graph.add_edge("enrich_context",    "supervisor")
    graph.add_edge("supervisor",        "route_agent")
    graph.add_edge("route_agent",       "check_resolution")
    graph.add_conditional_edges("check_resolution", should_end)
    return graph.compile(checkpointer=checkpointer)

import json
from datetime import datetime, timezone
from langchain_core.messages import BaseMessage
from langchain_anthropic import ChatAnthropic
from memory.memory_store      import MemoryStore
from memory.dynamo_repository import DynamoRepository

class ThreadArchiver:
    def __init__(self, memory: MemoryStore, dynamo: DynamoRepository):
        self.memory = memory
        self.dynamo = dynamo
        self.llm    = ChatAnthropic(model="claude-haiku-4-5-20251001")

    async def archive_thread(
        self, thread_id: str, session_id: str, user_id: str, messages: list[BaseMessage]
    ) -> None:
        ended_at = datetime.now(timezone.utc).isoformat()

        # 1. Gerar summary via LLM
        conv = "\n".join(
            f"[{m.__class__.__name__}]: {m.content}"
            for m in messages if hasattr(m, "content")
        )
        resp = await self.llm.ainvoke([{"role": "user", "content": f"""Analise esta conversa e retorne um JSON com:
- summary: resumo em 1-2 frases do que foi discutido e resolvido
- topics: lista de tópicos principais (array de strings)
- sentiment: "positive" | "neutral" | "negative"
- resolved: true | false
- inferred_preferences: objeto com preferências inferidas (ex: language, response_style)
- facts: objeto com fatos estruturados (ex: plan, segment, last_issue)

Conversa:
{conv}

Retorne SOMENTE o JSON, sem texto adicional."""}])

        try:
            a = json.loads(resp.content)
        except json.JSONDecodeError:
            a = {"summary": "Conversa encerrada.", "topics": [],
                 "sentiment": "neutral", "resolved": True,
                 "inferred_preferences": {}, "facts": {}}

        # 2. Salvar no DynamoDB
        await self.dynamo.save_session_summary(
            user_id=user_id, session_id=session_id,
            summary=a["summary"], topics=a.get("topics", []),
            sentiment=a.get("sentiment", "neutral"), resolved=a.get("resolved", True),
            turn_count=len(messages),
            started_at=ended_at, ended_at=ended_at,
        )

        # 3. Atualizar LTM no Redis
        await self.memory.push_ltm_summary(user_id, {
            "session_id": session_id, "summary": a["summary"],
            "topics": a.get("topics", []), "sentiment": a.get("sentiment", "neutral"),
            "resolved": a.get("resolved", True), "created_at": ended_at,
        })

        # 4. Write-through UserPreferences (DynamoDB → Redis)
        if a.get("inferred_preferences"):
            existing = await self.dynamo.get_preferences(user_id) or {}
            merged   = {**existing, **a["inferred_preferences"]}
            merged["inferred_count"] = str(int(existing.get("inferred_count", 0)) + 1)
            await self.dynamo.save_preferences(user_id, merged)
            await self.memory.set_preferences(user_id, merged)

        # 5. Fatos estruturados
        for key, val in a.get("facts", {}).items():
            await self.dynamo.save_preference_fact(
                user_id=user_id, fact_key=key, fact_value=str(val),
                confidence=0.85, source_session=session_id,
            )

import asyncio, uuid
from infrastructure.redis_clients  import RedisClients
from memory.session_store           import SessionStore
from memory.memory_store            import MemoryStore
from memory.dynamo_repository       import DynamoRepository
from memory.langgraph_checkpointer  import RedisCheckpointer
from supervisor.context_enricher    import ContextEnricher
from supervisor.thread_archiver     import ThreadArchiver
from supervisor.graph               import build_supervisor_graph
from langchain_core.messages        import HumanMessage

REDIS_CONFIG = {
    "sessions": {"host": "redis-sessions.xxx.cache.amazonaws.com", "port": 6379},
    "threads":  {"host": "redis-threads.xxx.cache.amazonaws.com",  "port": 6379},
    "memory":   {"host": "redis-memory.xxx.cache.amazonaws.com",   "port": 6379},
}

async def handle_message(user_id: str, message: str, session_id: str = None):
    redis        = RedisClients(REDIS_CONFIG)
    sessions     = SessionStore(redis)
    memory       = MemoryStore(redis)
    dynamo       = DynamoRepository()
    checkpointer = RedisCheckpointer(redis)
    enricher     = ContextEnricher(sessions, memory, dynamo)
    archiver     = ThreadArchiver(memory, dynamo)
    graph        = build_supervisor_graph(enricher, archiver, checkpointer)

    if not session_id:
        session_id = f"sess_{uuid.uuid4().hex[:12]}"
        await sessions.create_session(session_id, user_id, {"channel": "api"})
        await dynamo.register_session(user_id, session_id)
    else:
        await sessions.touch_session(session_id)

    thread_id = f"thrd_{uuid.uuid4().hex[:12]}"  # nova thread por conversa

    config = {"configurable": {"thread_id": thread_id}}
    state  = {
        "messages":         [HumanMessage(content=message)],
        "session_id":       session_id,
        "thread_id":        thread_id,
        "user_id":          user_id,
        "enriched_context": {},
        "agent_route":      None,
        "resolved":         False,
    }
    result = await graph.ainvoke(state, config=config)
    await redis.close()
    return result

if __name__ == "__main__":
    asyncio.run(handle_message(
        user_id="usr_xyz789",
        message="Preciso entender minha fatura do mês passado",
    ))