surfaces/.planning/codebase/ARCHITECTURE.md

Architecture

Analysis Date: 2026-04-01

Pattern Overview

Overall: Hexagonal / Ports-and-Adapters

Key Characteristics:

• A platform-neutral core/ defines all business logic and unified event types
• Adapters (adapter/telegram/, adapter/matrix/) translate platform-specific events into core types and back
• The AI platform SDK is hidden behind a PlatformClient Protocol; the current implementation (sdk/mock.py) is swappable without touching core or adapters
• All state is stored through a StateStore Protocol, with InMemoryStore for tests and SQLiteStore for production

Layers

Protocol Layer:

• Purpose: Defines every data structure crossing layer boundaries
• Location: core/protocol.py
• Contains: IncomingMessage, IncomingCommand, IncomingCallback, OutgoingMessage, OutgoingUI, OutgoingNotification, OutgoingTyping, ChatContext, AuthFlow, SettingsAction, type aliases IncomingEvent and OutgoingEvent
• Depends on: Python stdlib only
• Used by: All other layers

Core / Business Logic Layer:

• Purpose: Handles all domain logic independent of any platform
• Location: core/
• Contains:
  • core/handler.py — EventDispatcher: routes IncomingEvent to registered handler functions; returns list[OutgoingEvent]
  • core/handlers/ — one module per event category (start, message, chat, settings, callback)
  • core/store.py — StateStore Protocol + InMemoryStore + SQLiteStore
  • core/chat.py — ChatManager: creates/renames/archives chat workspaces (C1/C2/C3); persists via StateStore
  • core/auth.py — AuthManager: tracks auth flow state (pending → confirmed); persists via StateStore
  • core/settings.py — SettingsManager: fetches/caches user settings from SDK; invalidates on write
• Depends on: core/protocol.py, sdk/interface.py (Protocol only), core/store.py
• Used by: Adapters

SDK / Platform Layer:

• Purpose: Wraps the external Lambda AI platform; isolated behind a Protocol
• Location: sdk/
• Contains:
  • sdk/interface.py — PlatformClient Protocol: get_or_create_user, send_message, stream_message, get_settings, update_settings; also WebhookReceiver Protocol, Pydantic models (User, MessageResponse, MessageChunk, UserSettings, AgentEvent)
  • sdk/mock.py — MockPlatformClient: full in-memory implementation with simulated latency; supports both sync (send_message) and streaming (stream_message, currently returns single chunk); includes webhook simulation via simulate_agent_event()
• Depends on: sdk/interface.py
• Used by: core/ managers, adapters during bot startup

Adapter Layer:

• Purpose: Translates platform-native events into IncomingEvent and OutgoingEvent back to platform-native calls
• Location: adapter/matrix/, adapter/telegram/ (in .worktrees/telegram/)
• Contains per adapter: bot.py (entry point + send logic), converter.py (native event → protocol), handlers/ (adapter-specific handler overrides registered on top of core handlers), optional store.py / room_router.py / reactions.py for adapter state
• Depends on: core/, sdk/, platform SDK (aiogram or matrix-nio)
• Used by: __main__ / asyncio.run(main())

Data Flow

Incoming Message (Matrix example):

1. matrix-nio fires RoomMessageText callback → MatrixBot.on_room_message() in adapter/matrix/bot.py
2. resolve_chat_id() in adapter/matrix/room_router.py maps room_id → logical chat_id (e.g. C1), persisted in StateStore
3. from_room_event() in adapter/matrix/converter.py converts the nio event to IncomingMessage or IncomingCommand
4. EventDispatcher.dispatch(incoming) in core/handler.py selects the handler by routing key (command name, callback action, or "*" for messages)
5. Handler (e.g. core/handlers/message.py:handle_message) calls platform.send_message() on MockPlatformClient, receives MessageResponse
6. Handler returns list[OutgoingEvent] (e.g. [OutgoingTyping(..., False), OutgoingMessage(...)])
7. MatrixBot._send_all() iterates the list; send_outgoing() converts each to a client.room_send() / client.room_typing() call

Incoming Reaction (Matrix):

1. ReactionEvent callback → MatrixBot.on_reaction()
2. from_reaction() maps emoji key to IncomingCallback with action="confirm", "cancel", or "toggle_skill"
3. Dispatch → core/handlers/callback.py

Command Routing:

The EventDispatcher uses a routing key per event type:

• IncomingCommand → event.command (e.g. "start", "new", "settings")
• IncomingCallback → event.action (e.g. "confirm", "toggle_skill")
• IncomingMessage → "*" (catch-all), or event.attachments[0].type if attachments present

Adapters call register_all(dispatcher) first (core handlers), then register_matrix_handlers(dispatcher, ...) to override or add platform-specific variants (e.g. !new creates a real Matrix room via the nio client).

State Management:

• All persistent state goes through StateStore (key-value, async interface)
• Key namespaces: chat:{user_id}:{chat_id}, auth:{user_id}, settings:{user_id}, matrix_room:{room_id}, matrix_user:{matrix_user_id}, matrix_state:{room_id}, matrix_skills_msg:{room_id}
• Production uses SQLiteStore (row-per-key, JSON-serialised values); tests use InMemoryStore

Key Abstractions

EventDispatcher (core/handler.py):

• Purpose: Single dispatch table for all event types; decouples handler logic from transport
• Pattern: Registry (map of event_type → {key → HandlerFn}); wildcard "*" as fallback
• Handler signature: async def handler(event, chat_mgr, auth_mgr, settings_mgr, platform) → list[OutgoingEvent]

StateStore Protocol (core/store.py):

• Purpose: Pluggable persistence behind a minimal get/set/delete/keys interface
• Implementations: InMemoryStore (tests/dev), SQLiteStore (production)
• Key pattern: "{namespace}:{discriminator}"

PlatformClient Protocol (sdk/interface.py):

• Purpose: Contracts the entire surface of the Lambda AI SDK
• Current implementation: MockPlatformClient in sdk/mock.py
• Swap path: Replace sdk/mock.py with a real SDK client; no changes needed elsewhere

Converter functions (adapter/matrix/converter.py):

• Purpose: One-way transformation from platform-native event to IncomingEvent
• Always produce canonical protocol types; adapters never pass raw library objects to core

Entry Points

Matrix Bot:

• Location: adapter/matrix/bot.py:main()
• Run: python -m adapter.matrix.bot
• Startup sequence: load .env → build AsyncClient → build_runtime() → register callbacks → client.sync_forever()

Telegram Bot:

• Location: .worktrees/telegram/adapter/telegram/bot.py (feature branch, not merged to main yet)
• Run: python -m adapter.telegram.bot

Error Handling

Strategy: Errors propagate up to the adapter's event callback. The adapter logs and drops the event; the bot keeps running.

Patterns:

• EventDispatcher.dispatch() returns [] (empty list) when no handler is found and logs a warning
• AuthManager and ChatManager raise ValueError for not-found entities; callers are responsible for catching
• MockPlatformClient raises PlatformError (defined in sdk/interface.py) on unexpected states

Cross-Cutting Concerns

Logging: structlog throughout; all managers and the dispatcher use structlog.get_logger(__name__) Validation: Pydantic models in sdk/interface.py for SDK responses; plain dataclasses in core/protocol.py for internal events Authentication: AuthManager.is_authenticated() is checked in handle_message before forwarding to platform; unauthenticated users receive a prompt to run !start / /start

---

Architecture analysis: 2026-04-01