Message router.

The single entry point

Every webhook handler in app/routers/webhook.py (Telegram, WhatsApp Cloud via 360Dialog, Instagram) normalises its channel-specific payload into a common shape and calls into msg_router.handle_message. The PWA's /api/agent/chatendpoint does the same for browser traffic. This single-entry-point discipline is what makes multi-channel support cheap: add a new channel, add a new webhook, and the rest of the stack doesn't care.

End-to-end flow

1. Normalise inbound payload. Channel-specific fields (Telegram update.message, WhatsApp messages[0], Instagram messaging[0].message) become a common (user_id, platform, text, image_url) tuple.
2. Resolve or create the user. The router looks up the user by platform + platform_user_id, creates one on first contact, and fires a background task to register a consumer agent on ClawPulse via agent_registry.
3. Classify intent. If there is an image attached, the image classifier decides whether it looks like a food photo (→ calorie scan), a product (→ dupe search), or general context. Text-only messages skip this branch.
4. Load session history. Session storereturns the last 12 messages from Redis, encrypted with Fernet. Platform-scoped keys ensure a user who chats on Telegram and WhatsApp doesn't leak context between the two.
5. Build the soul prompt. app/core/soul_loader.py assembles the system prompt: base persona + user memory chunks + weather + transport preferences + 35 intent patterns + any venue overrides.
6. Call the LLM gateway. LLM gateway dispatches the Anthropic Messages call with the 16 tool schemas attached and enforces the per-user token budget.
7. Run tool calls if any. When the model returns a tool_use block, tool_executor.run executes them with a 5 second timeout and feeds the results back into the conversation.
8. Persist and send. The final assistant message is appended to the session store and queued for outbound delivery through the channel sender.

Image handling

Image URLs come in from the channel webhook as either a direct link (Telegram gives you a file_path) or a media ID that must be resolved to a download URL (WhatsApp and Instagram). The router delegates resolution to channel-specific adapters, then runs the bytes through app/core/image_utils.py which applies an SSRF check, size limit, and base64 encoding before handing the image to the classifier or the vision-capable LLM call.

Error handling

Every failure mode is logged with the Sentry breadcrumb for the current request. User-visible errors are translated into a friendly reply in the user's language via the session's language hint. Silent failures are an anti-pattern: if the router catches an exception it must either reply to the user or escalate to the ErrorLog table with enough context to reproduce the failure.

async def handle_message(
    *,
    user_id: uuid.UUID,
    platform: str,
    text: str | None,
    image_url: str | None,
    db: AsyncSession,
) -> str:
    """Main entry point for every inbound message."""
    history = await session_store.get_history(str(user_id), platform=platform)

    if image_url:
        intent = await image_classifier.classify(image_url)
    else:
        intent = None

    prompt = await soul_loader.build(user_id=user_id, db=db, intent=intent)
    response = await llm_gateway.chat(
        user_id=user_id,
        messages=[*history, LLMMessage(role="user", content=text or "")],
        system=prompt,
        tools=TOOL_SCHEMAS,
    )

    if response.tool_calls:
        tool_results = await tool_executor.run(response.tool_calls, user_id, db)
        # Feed results back for a follow-up turn
        response = await llm_gateway.chat(
            user_id=user_id,
            messages=[
                *history,
                LLMMessage(role="user", content=text or ""),
                response.as_assistant_message(),
                *tool_results,
            ],
            system=prompt,
        )

    await session_store.append(
        str(user_id),
        [LLMMessage(role="user", content=text or ""), response.as_assistant_message()],
        platform=platform,
    )
    return response.text

Testing

The router is covered by the wave-1 / wave-4 hardening suite under tests/test_wave1_wave4_hardening.py. Use the fixture at the top of that file to construct an in-memory DB session and a fake Redis so tests stay fast and deterministic.