Architecture

Module layout changed (2026-06 cleanup)

Behavior is unchanged, but the code was decomposed: agent tools → app/agent_tools/<domain>.py, background loops → app/background.py, HTTP routes → app/routers/api.py, Telegram helpers/media/commands split out of bot.py, prompt text → app/prompt_text.py, AgentDeps/build_deps → app/deps.py. Original modules re-export their public symbols. See the banner on the home page and CLAUDE.md → Recent Changes.

System Context

Zoom-out view of the runtime topology — who talks to Kwasi, where Kwasi lives, and every external service it depends on. Useful for "if X goes down, what breaks?" and "where does my data leave the container?"

%%{init: {"flowchart": {"defaultRenderer": "elk", "nodeSpacing": 25, "rankSpacing": 45, "wrappingWidth": 180, "elk": {"mergeEdges": true, "nodePlacementStrategy": "NETWORK_SIMPLEX"}}}}%%
graph TB
    subgraph Edge["Users and Devices"]
        direction LR
        U["Lawrence\nTelegram client\n+ Mini App (in-chat web page)"]
        PHONE["Android phone (bridge-android/)\nHealth Connect\n+ app usage · HTTP Shortcuts"]
        WATCH[Samsung Watch /\nwearable]
        DASHU[Dashboard browser]
    end

    subgraph Railway["Railway (Docker container)"]
        KWASI["Kwasi\nFastAPI + Telegram bot\n+ background loops"]
        PG[("PostgreSQL\n+ pgvector\nRailway managed service")]
        KWASI <--> PG
    end

    subgraph AI["AI / LLM Providers"]
        direction LR
        GEM["Google Gemini\nLLM · Vision · STT · embeddings"]
        ANT["Anthropic Claude\noptional primary\n+ delegation backend"]
        OAI["OpenAI\noptional primary"]
        E2B["E2B Cloud\nsandboxed Python\n+ delegation VMs"]
    end

    subgraph Obs["Observability (Cloud)"]
        direction LR
        LF["Logfire Cloud\ninfra · FastAPI · loops"]
        LFC["Langfuse Cloud\nLLM generations · prompts · scores"]
        WV["W&B Weave\noptional · WEAVE_ENABLED"]
    end

    subgraph Personal["Google / Microsoft APIs"]
        direction LR
        GMAIL["Gmail API\npersonal + work OAuth2"]
        GCAL["Google Calendar API"]
        GDRIVE["Google Drive API\npersonal + work OAuth2"]
        MSG["Microsoft Graph\nOutlook mail · calendar · To Do\nMSAL"]
        NOTION["Notion API\nconnected Notes database"]
    end

    subgraph Work["Messaging / Productivity APIs"]
        direction LR
        TG[Telegram BotAPI\nlong-polling]
        SLACK[Slack Web API]
        JIRA[Jira REST API]
        GH[GitHub API\nPyGitHub]
    end

    subgraph Utility["Utility APIs"]
        direction LR
        TAV[Tavily\nweb search]
        WX[WeatherAPI]
        GMAP[Google Maps\ntransit + places]
        YV[YouVersion API\nverse of the day]
    end

    U <--> TG
    TG <--> KWASI
    PHONE -->|"X-Health-Secret\nPOST /health/ingest"| KWASI
    PHONE -->|"X-Phone-Secret\nPOST /phone/usage/ingest"| KWASI
    PHONE -->|"X-API-Token\nPOST /message"| KWASI
    WATCH --> PHONE
    DASHU -->|X-Dashboard-Secret| KWASI

    KWASI --> GEM
    KWASI -.-> ANT
    KWASI -.-> OAI
    KWASI --> E2B

    KWASI -->|OTEL spans| LF
    KWASI -->|OTEL spans + scores| LFC
    KWASI -.->|OTEL spans| WV

    KWASI --> GMAIL
    KWASI --> GCAL
    KWASI --> GDRIVE
    KWASI --> MSG
    KWASI -.-> NOTION
    KWASI -.-> YV

    KWASI --> SLACK
    KWASI --> JIRA
    KWASI --> GH

    KWASI --> TAV
    KWASI --> WX
    KWASI --> GMAP

Reading the diagram:

• Solid arrows are always-on dependencies. Dotted arrows are optional/feature-flagged (Anthropic, OpenAI, YouVersion verse-of-the-day).
• There is no external scheduler. Every periodic job is an in-process asyncio loop started by the FastAPI lifespan — railway.json declares no cron, and nothing but a human with curl has ever called POST /reflect. See Background Loops → Reflection Loop.
• Railway boundary = one container + one Postgres service. Everything else is somebody else's cloud. The container is stateless aside from in-flight asyncio tasks; all durable state lives in Postgres.
• Three trust boundaries cross the Railway edge:
• Inbound from users/devices, gated by token headers: X-Health-Secret (Android health bridge), X-API-Token (Android HTTP Shortcuts → POST /message), X-Phone-Secret (Android phone bridge → /phone/usage/ingest), X-Dashboard-Secret (dashboard browser), X-Reflection-Secret (manual/ops → POST /reflect, POST /embed-backfill). Telegram is gated by ALLOWED_TELEGRAM_USER_IDS at handler level.
• Outbound to AI providers, gated by API keys in env. Gemini is the only one that's load-bearing on the default config; the others are alternative primary models or alternative delegation backends.
• Outbound to personal/work APIs, gated by per-service OAuth refresh tokens (Google, Microsoft) or PATs (GitHub, Jira, Slack).
• What breaks if X is down:
• Gemini down → no LLM responses, no embeddings, no Vision/STT. Hard failure.
• Postgres down → no history, no facts, no approval gate. Hard failure.
• Telegram BotAPI down → can't receive or reply. Background loops still run (and queue notifications that will fail to send).
• Logfire or Langfuse down → observability gap, app keeps working (OTEL exporter retries in background).
• E2B down → execute_python and delegate_to_coding_agent fail; everything else fine.
• Any single personal/work API down (Gmail, Slack, etc.) → that domain agent's tools return errors; routing falls through to the full agent which retries or summarises the failure to the user.

---

System Overview

%%{init: {"flowchart": {"defaultRenderer": "elk", "nodeSpacing": 30, "rankSpacing": 50, "wrappingWidth": 180}}}%%
graph TB
    subgraph Interfaces
        TG[Telegram\npython-telegram-bot polling]
        WA[WhatsApp\nwebhook]
        CLI[CLI\nlocal REPL]
        DASH[Dashboard\n/dashboard/]
        MINI["Mini App\n/miniapp/ — in-chat\ntoday · memory · approvals · status"]
    end

    subgraph Planning["Planning Layer (app/planning/) — Spec 008"]
        CC[classify_complexity\nregex pre-filter]
        GP[generate_plan\nplanner agent]
        EP[execute_plan\nstep loop · scratchpad threading]
    end

    subgraph Routing["Intent Routing (app/tools/router.py + app/routing/agents.py)"]
        CI[classify_intent\nkeyword → semantic top-K → context_hint]
        SA[select_agent\ndispatcher]
        DA["Domain Agents (13)\nemail · calendar · memory · github · news\nslack · jira · drive · meetings\ndiagnostics · health · utility · database"]
        CA[Composed Agent\nambiguous / cross-domain\nlru_cache]
        FA2[Full Agent\nfallback]
    end

    subgraph Core["Core (FastAPI + asyncio)"]
        FAST[FastAPI\napp/main.py]
        AG[Pydantic AI Agent\napp/agent.py]
        SP[System Prompt\nbuild_system_prompt]
    end

    subgraph Tools["Native Tools"]
        T1[search_web\nTavily]
        T2[get_weather\nWeatherAPI]
        T3[get_datetime]
        T4[summarize_url]
        T5[transcribe_audio\nGemini STT]
        T6[analyze_image\nGemini Vision]
        T7[browse_web\nbrowser_submit_form\nPlaywright]
        T8[github_*\nPyGitHub]
    end

    subgraph Skills["Skills (file-drop plugins)\napp/skills/"]
        S1[content_curator\nread_later.py]
        S2[travel_briefing]
        S3[cv]
        S4[deep_research\nresearch.py]
        S5[meeting_notes\nget_meeting_insights · list_recent_meetings\ncheck_meeting_notes]
        S6[sync_podcast_insights\nTo Do 'Podcasts' list → Knowledge Notes]
    end

    subgraph MCPTools["MCP Tools"]
        M1[Gmail personal + work\ngmail_* tools]
        M2[Outlook Email\noutlook_* tools]
        M3[Google Calendar\ngoogle_* tools]
        M4[Outlook Calendar\noutlook_calendar_* tools]
        M5[Microsoft To Do\ntodo_* tools]
        M6[Google Drive\ndrive_* tools]
        M7[Logfire\nlogfire_* tools]
    end

    subgraph Ext["External Note Stores"]
        NOTION["Notion Notes DB\nsave_to_notion · search_notion\nlist_recent_notion · delete_notion_note\nsource of truth — no local mirror"]
    end

    subgraph Memory["Memory Tools (agent tools over storage)"]
        N[Notes\nsave · get · search · edit · delete · semantic]
        TK[Tasks\ncreate · list · search · complete · edit · delete]
        RM[Reminders\nset · list · cancel]
        SC[Scheduled Tasks\ncreate · list · update · delete]
        HX[History\nsearch_history · semantic_search]
        UF[User Facts\nremember_fact · recall_facts · forget_fact]
        JR[Journal\nsave_journal_entry · get_journal_digest]
        IN2[Intentions + Learnings\nlist/resolve/snooze · dismiss_learning]
        AR[Alert Rules\ncreate · list · update · delete]
        PH[Phone Awareness\nget_app_usage_summary\nget_attention_snapshot]
    end

    subgraph Storage["Storage (StoragePort)"]
        PG[("PostgreSQL + pgvector\nprod, dev and tests\nthe only implementation")]
    end

    subgraph HealthIngest["Device Ingest (push model — no server-side collection loop)"]
        WATCH[Samsung Watch / wearable]
        HC[Google Health Connect\non-device]
        NOTIF["UsageStatsManager"]
        BR["bridge-android/\nKotlin · WorkManager\nwhitelist + on-device redaction"]
        IN["POST /health/ingest\nX-Health-Secret · idempotent"]
        IN2P["POST /phone/usage/ingest\nX-Phone-Secret"]
        WATCH --> HC
        HC --> BR
        NOTIF --> BR
        BR --> IN
        BR --> IN2P
        IN --> Storage
        IN2P --> Storage
    end

    subgraph BG["Background Loops"]
        RL[Reminder Loop\nevery 5 min]
        BL[Briefing Loop\ndaily at BRIEFING_TIME]
        ST[Scheduled Tasks Loop\nevery 60 sec]
        RF[Reflection Loop\ndaily at 2 AM UTC]
        ER[Evening Recap Loop\ndaily local evening]
        WR[Weekly Recap Loop\nFriday local]
        WP[Weekly Prep Loop\nSunday local]
        RLD[Read-Later Digest\nSaturday morning]
        JD[Journal Digest\nSunday evening]
        EI[Email Intelligence\ndaily at EMAIL_INTEL_TIME]
        AL[Alert + Intentions Loop\nevery 5 min]
        MP[Meeting Prep Loop\nevery 5 min]
        AE["Approval Expiry Loop\nevery 5 min\n+ hourly TTL pruning"]
        PRL[PR Review Loop\ndaily at PR_REVIEW_TIME]
        MFU[Meeting Follow-up Loop\nevery MEETING_FOLLOWUP_POLL_MINUTES]
        TSL[To Do Sync Loop\nevery TODO_SYNC_INTERVAL_SECONDS]
        PDL[Phone Digest Loop\ndaily at PHONE_DIGEST_TIME]
    end

    subgraph Models["LLM Models"]
        MR["run_with_resilience\nretry 429/5xx → fall back to mini"]
        LLM[Primary LLM\nMODEL_NAME\nGemini / Claude / GPT-4]
        MINI[Mini LLM\nMINI_MODEL_NAME\nskill synthesis]
        REFL[Reflection LLM\nREFLECTION_MODEL_NAME\nnightly reflection]
    end

    subgraph SelfImprove["Capability-gap detection"]
        RS[read_source_file · grep_source\napp/tools/source.py\non every domain agent]
        GD[detect_and_save_capability_gap\npost-turn fire-and-forget\nregex → mini-classify → grep-verify]
        GD --> GAPS[(capability_gaps\nweekly digest)]
    end

    TG --> CI
    CI --> SA
    SA --> CC
    CC -- complex --> GP
    GP -- 2+ steps --> EP
    CC -- simple --> AG
    GP -- 0–1 steps --> AG
    EP -- per step --> CI
    WA --> FAST
    CLI --> AG
    CI --> SA
    SA --> DA
    SA --> CA
    SA --> FA2
    DA --> MR
    CA --> MR
    FA2 --> MR
    MR --> LLM
    MR -.->|primary exhausted| MINI
    AG --> RS
    AG -- post-turn fire-and-forget --> GD
    GD --> Storage
    PR --> LLM
    AC --> Skills
    FAST --> FAST
    DASH --> Storage
    AG --> SP
    SP --> Storage
    AG --> LLM
    AG --> Tools
    AG --> Skills
    AG --> MCPTools
    AG --> Ext
    AG --> Memory
    Memory --> Storage
    BG --> Storage
    BG --> TG
    Skills --> MINI
    BG --> REFL

---

Component Breakdown

FastAPI (app/main.py)

The web process entry point. Responsibilities:

• Mounts the WhatsApp webhook router, dashboard router, and health router (/health/ingest, /health/backfill, /health/status)
• Exposes GET /health and POST /reflect
• Exposes POST /embed-backfill — bulk-embeds existing rows without embeddings; protected by X-Reflection-Secret; rate-limited to ~10 rows/sec
• Exposes POST /health/ingest — idempotent batch ingest for wearable samples from the Android bridge; protected by X-Health-Secret (Spec 010)
• Exposes POST /health/backfill — one-shot upload of a Samsung Health "Download Personal Data" CSV/JSON archive to seed historical samples; protected by X-Health-Secret
• Exposes GET /health/status — per-metric ingest freshness summary for monitoring the bridge from outside the device. Returns {now, latest_ingest, lag_minutes, metrics: [...]} with one entry per known metric type (zeros when never ingested) plus per-metric ingest_lag_minutes. Protected by X-Health-Secret so polling it doesn't expose data shape (Spec 010)
• Exposes POST /message — external message endpoint for Android HTTP Shortcuts and other clients. Accepts text and/or images (multipart/form-data or JSON with base64). Protected by X-API-Token header (API_TOKEN env var). Processes input through the full agent pipeline (intent routing, read-later injection, message history). Delivers response to Telegram (BRIEFING_CHAT_ID) and returns it in the JSON body. Conversation history is unified with Telegram — both interfaces use BRIEFING_CHAT_ID as the user ID so context is shared across surfaces.
• Manages the FastAPI lifespan context — starts and stops the Telegram bot and all background asyncio tasks
• Exports API keys to os.environ so LLM SDKs can find them

Agent (app/agent.py)

The brain. Built with Pydantic AI. Responsibilities:

• Defines AgentDeps — the dependency container passed to every tool and the system prompt
• build_deps() — single factory function used by all interfaces to create AgentDeps
• build_system_prompt() — async function registered with @agent.instructions (not @agent.system_prompt — see System Prompt Architecture), assembles the full prompt from the static instruction sections + dynamic storage-fetched sections (behavioral learnings, narrative profile, live DB schema, and — if the vault is enabled — stored web logins). Permanent facts are not injected; the prompt tells the model to call recall_facts(). See System Prompt Architecture below.
• build_briefing_system_prompt() — extends build_system_prompt by appending all briefing structural templates (_MORNING_BRIEFING, _EVENING_RECAP, _WEEKLY_RECAP, _WEEKLY_PREP, _MEETING_PREP) at the system level. Used exclusively by briefing_agent.
• Registers all native tools via @agent.tool
• Calls load_skills(agent) at startup to auto-register skills from app/skills/

Multi-Model Architecture

Three model roles are configured independently via env vars. Each is optimised for a different cost/capability trade-off:

Role	Env var	Default	Used for
Primary	MODEL_NAME	google-gla:gemini-2.5-flash	Main conversational agent and all 13 domain agents — every interactive user message
Mini	MINI_MODEL_NAME	google-gla:gemini-3-1-flash-lite-preview	Skill synthesis: CV extraction, deep research synthesis, read-later summarisation, travel briefing, post-conversation fact extraction, PR review structuring, capability-gap classification
Reflection	REFLECTION_MODEL_NAME	(empty — falls back to MODEL_NAME)	Nightly reflection run only — set to a larger/more capable model for deeper analysis without affecting interactive latency

settings.effective_reflection_model resolves the fallback: if REFLECTION_MODEL_NAME is empty, it returns MODEL_NAME.

Model binding is per-request, not at agent construction. The Agent instances are built with no model (Agent(deps_type=AgentDeps, retries=3)); every call passes model=ctx.deps.settings.model_name at invocation time. Because settings and deps.settings are the same object, mutating deps.settings.model_name in memory changes the model on the next message with no restart — this is the mechanism that makes set_runtime_config("model_name", …) instant (see Self-Management Subsystem). (Limitation: a few skill-synthesis "mini" agents bind mini_model_name at import time, so a runtime change to the mini model only affects call sites that read deps.settings.mini_model_name dynamically until the next restart.)

Why separate models? The mini model handles high-frequency, low-stakes synthesis tasks (run after every message) where cost and latency matter most. The primary model handles user-facing responses where quality matters. The reflection model is a deliberate override slot — if you want a frontier model for nightly analysis without paying for it on every chat message, set only REFLECTION_MODEL_NAME.

Model Resilience (app/model_resilience.py)

A retry + model-fallback wrapper around Agent.run, wired into the Telegram call site.

Production telemetry (Langfuse, Jun–Aug 2026) showed 86 429 quota exceeded and 11 503 overloaded responses from the primary Gemini model. Nothing in the codebase retried them: the single agent call site caught Exception and replied "I'm having trouble thinking right now", so one transient quota blip cost the user their whole turn.

run_with_resilience(agent, prompt, *, deps, model, fallback_model, **run_kwargs) adds the missing layer:

1. Retry the primary model on transient failures — status 429, 500, 502, 503, 504 — up to MAX_PRIMARY_ATTEMPTS (3) with (1s, 4s) backoff. Three attempts spans ~5s, which comfortably outlasts the per-minute quota windows that produce most 429s.
2. Fall back to the mini model once the primary is exhausted. A smaller answer beats no answer, and the mini model draws on a separate quota bucket.
3. Re-raise the primary error if everything fails, so the caller's existing error handling and Langfuse agent_error scoring still work. The fallback's own error is logged, not surfaced — the primary failure is the one worth diagnosing.

run_kwargs is forwarded verbatim (message_history, usage, usage_limits, …), so it is a drop-in replacement for the direct call.

Deliberately not retried: 4xx other than 429 (bad request, auth). Those are deterministic — retrying only burns latency and quota.

Notion Notes (app/tools/notion.py) — Spec 015

Two-way capture into a single connected Notion "Notes" database. Notion is the source of truth — no local mirror, no sync loop, no background reconciliation; reads query the API live. That is the deliberate contrast with To Do sync, where the local table is the brain and needs reconciling.

Implemented as a raw-REST client (like the Graph client) to avoid a new dependency; the sync functions are wrapped in asyncio.to_thread by the agent-tools layer and raise on API failure so callers degrade gracefully. Note bodies are chunked under Notion's 2,000-char rich-text cap.

Least privilege by construction — the integration can only reach pages explicitly connected to it, never the rest of the user's Notion. Requires NOTION_API_KEY + NOTION_NOTES_DATABASE_ID; every tool degrades to an offer to save a normal Kwasi note when unset. See Tools → Notion Notes.

Telegram Interface (app/interfaces/telegram/bot.py)

The primary user-facing interface. Handles:

• Text messages — a "…" placeholder is sent immediately and edited once with the complete answer. Not streamed: run_stream returns the first model response and treats its text as final even when tool calls follow, so on Gemini a "Let me check…" preamble became the visible answer while the real post-tool synthesis was dropped. The run goes through run_with_resilience.
• Voice/audio — downloads from Telegram, transcribes via Gemini STT, runs agent, replies with a voice note (TTS) for responses ≤500 words
• Photos — downloads highest resolution, runs Gemini Vision, runs agent
• Documents — PDFs via Gemini Vision, text files decoded directly
• Slash commands — /start, /help, /notes, /tasks, /reminders (bypass the agent, hit storage directly)
• Allowlist — ALLOWED_TELEGRAM_USER_IDS checked on every handler; blocks all if unset
• Audit logging — _write_audit_entries() writes sanitised tool call records to audit_log after each interaction
• Inline approval flow — after the run completes, scans tool call outputs for sentinel tokens ([APPROVAL_PENDING:<uuid>]); edits the sent message to add Confirm/Cancel/Edit inline keyboard buttons. CallbackQueryHandler handles button taps — Confirm executes via ACTION_REGISTRY, Cancel marks cancelled, Edit prompts for a correction and re-runs the agent. Edit state is tracked in context.user_data.
• Post-conversation background tasks — after audit logging, fires asyncio.create_task() calls for: (1) extract_facts_from_exchange() to immediately extract explicitly stated user facts; (2) detect_and_save_capability_gap() to log routing/capability gaps for the self-improvement loop. Both are fire-and-forget. (The earlier 30-min summarise_session was removed 2026-05-09 — topic summaries now come only from the nightly reflection.) See Memory & Reflection.

Browser Tools (app/tools/browser.py)

Playwright-based headless Chromium. Features:

• Chrome user agent spoofing to reduce bot detection
• Asset blocking (images, fonts, media) for faster loads
• Semantic content extraction — targets <main>, <article>, [role="main"] before falling back to <body>
• Headline extraction from h1/h2/h3 in the content zone
• URL security: blocks file://, localhost, and private IPs; optional BROWSER_ALLOWED_DOMAINS allowlist
• Graceful degradation if Playwright is not installed

GitHub Tools (app/tools/github.py)

PyGitHub-based integration. All API calls are synchronous and wrapped with asyncio.to_thread(). Gracefully returns a clear message if GITHUB_TOKEN is not configured or PyGitHub is not installed.

Embedding (app/tools/embedding.py)

embed_text(text, api_key) → list[float] | None — calls the Gemini gemini-embedding-001 REST API directly via httpx (3072 dimensions). Called internally by both storage adapters after each write to notes, interactions, and read_later. Returns None when GOOGLE_API_KEY is not set or the API call fails — never raises. Falls back to gemini-embedding-2-preview if the primary model returns 404.

Slack Tools (app/tools/slack.py)

slack-sdk WebClient integration. All API calls are synchronous and wrapped with asyncio.to_thread(). Username and channel name resolution are cached within each tool call. Gracefully returns a clear message if SLACK_BOT_TOKEN is not configured or slack-sdk is not installed. Requires the bot to be invited to any channel it reads or posts in.

Intent Router (app/tools/router.py + app/routing/agents.py)

Keyword-based intent classification and domain agent dispatch. Runs on every text message (Telegram and WhatsApp) before the agent is invoked.

Taxonomy (TOOL_CATEGORIES) — maps thirteen domain names to their canonical tool lists: utility, memory, email, calendar, github, news, slack, jira, drive, meetings, diagnostics, health, database. TOOL_TO_CATEGORY is the reverse lookup used in tests.

classify_intent(message, context_hint=None) — three-stage classifier:

1. Keyword match (microseconds, zero LLM cost) — checks the message against per-domain keyword lists. Returns matched categories with "utility" always added when any domain hits. keyword_hits per category is logged to the Langfuse span.
2. Semantic fallback with confidence bands — when keywords find nothing, embed the message against pre-computed per-domain anchor embeddings (cached lazily). Returns top-K based on cosine score:
3. ≥ 0.65 (high confidence) → single best domain
4. 0.45 – 0.65 (ambiguous) → top-K within 0.10 of best, capped at 3 → composed_agent so the LLM picks
5. < 0.45 → empty (no confident match) Each routing decision logs semantic_top3: [{domain, score}] so misroutes are debuggable from traces.
6. Context-hint inheritance — when both keyword and semantic miss, an optional context_hint: set[str] (single non-utility domain from the previous turn) is inherited so short follow-ups like "anything else?" continue in the same domain agent.

The three-band semantic stage replaced an earlier binary ≥ 0.60 cutoff that caused misroute hallucinations: borderline queries scored just below threshold and fell through to the wrong agent, leading the model to claim "I don't have a tool" for capabilities that existed.

select_agent(categories) — dispatches to the right agent: - Empty set → full_agent (conversational fallback / no confident match) - {"utility"} only → utility_agent - One non-utility domain (e.g. {"email", "utility"}) → email_agent - Multiple non-utility domains → _get_composed_agent(frozenset(categories)) — a dynamically built agent whose tool set is the union of all matched domains' tools, cached with lru_cache(maxsize=32). With the three-band semantic stage above, this path is used not just for explicit cross-domain requests but also for ambiguous single-domain queries where multiple anchors scored within 0.10 of each other — the LLM picks across the candidates rather than the router guessing wrong.

Domain agents — 13 interactive Agent instances + 1 briefing_agent, all built once at import time in app/routing/agents.py. Each carries only its relevant tool subset, reducing the tool list the LLM sees. They share build_system_prompt and the active_categories focus hint injected into AgentDeps per request. Several universal meta-toolsets are registered on every domain agent via _UTILITY_FNS — they are meta, not domain-scoped, because the model often reaches for them from non-matching contexts:

• Source introspection — read_source_file, grep_source ("do I have a tool for X?").
• Raw DB ops — db_query (read-only), db_execute (approval-gated). database / diagnostics exist as focused-attention modes for the same tools.
• Runtime config + self-management — set_runtime_config / get/list/clear_runtime_config, railway_set_env / railway_redeploy (approval-gated, action_type config / deployment), and read-only railway_deployment_status / railway_deploy_logs / diagnose_self. See Self-Management Subsystem.
• LLM trace introspection — langfuse_recent_traces / langfuse_trace_details (Logfire's read tools live on diagnostics_agent).
• Self-improvement — get_capability_gap_digest, - Conversation recall — search_history, semantic_search.
• Phone Awareness — get_app_usage_summary, get_attention_snapshot.
• Notion reads — search_notion, list_recent_notion (writes stay memory-scoped).
• Facts + intentions — remember_fact / recall_facts / forget_fact; list / resolve / dismiss / snooze_intention.
• Everyday utility + delegation — web search, weather, datetime, URL/browser, maps/transit, execute_python, delegate_to_coding_agent, delegate_web_task, get_verse_of_the_day.

So "switch your model", "redeploy", or "any errors lately?" route to the right universal tool from any domain rather than being declined.

The rule this list encodes, and how it keeps getting learned

A tool belongs in _UTILITY_FNS when the request for it doesn't arrive framed in its own domain. Production kept proving the point the hard way: db_query was database-only until non-DB-framed requests ("count my interactions") failed; in August, prod logged ToolRetryError: Unknown tool name: 'search_history' on the github agent and 'list_recent_notion' on the health agent — "what did we say about X" is a follow-up to any domain, not a memory-framed request.

_build_domain_agent now deduplicates its tool lists, because a tool present in both a domain list and _UTILITY_FNS raised into the swallow-and-warn path and silently produced the very missing tool the promotion was meant to prevent.

The jira_agent handles Jira queries (tickets, sprints, projects) via the six jira_* tools. Requires JIRA_BASE_URL, JIRA_EMAIL, and JIRA_API_TOKEN.

The drive_agent handles Google Drive queries (search, read, list recent, transcripts) via the five Drive MCP tools. Supports personal (GOOGLE_DRIVE_REFRESH_TOKEN) and work (GOOGLE_DRIVE_WORK_REFRESH_TOKEN) accounts.

The meetings_agent handles meeting intelligence queries via get_meeting_insights and list_recent_meetings skills (from app/skills/meeting_notes.py), plus Drive MCP tools and calendar MCP tools for context. Intent keywords include "recap the meeting", "notes from the", "what came out of", and "standup notes".

The diagnostics_agent handles self-diagnosis queries (errors, performance, loop health) via the three Logfire MCP tools. Its intent keywords use specific multi-word phrases (e.g. "your errors", "did the briefing run") to avoid false positives on common words like "error".

The health_agent (Spec 010) handles wearable / Samsung Watch queries via four read-only tools (get_recent_health, get_sleep_summary, get_hrv_trend, get_health_snapshot). Routed by intent keywords like "how did i sleep", "my hrv", "resting heart rate", "step count". Data is ingested separately by the Android bridge — the agent only reads.

The dba_agent (database category) is structurally identical to utility_agent — its value is the focused system-prompt mode (_DB_OPS_RULES, plus a live "## Live database schema" section introspected via describe_schema() rather than a hand-maintained appendix) for SQL-flavored requests. The two raw-DB tools themselves are universal on every domain agent.

Planning Layer (app/planning/) — Spec 008

A multi-step decomposition pass that runs on Telegram messages before any agent run, so requests like "check my email, then post a summary in #standup, then add a follow-up task" execute as a sequenced plan instead of one ambiguous tool dance.

Where the gate actually sits in handle_message

It runs after history retrieval, intent routing, and context injection — not before them. A message that turns out to be a plan has already paid for that work; a message that isn't reaches the agent with everything in place. Each step is then routed independently by classify_intent + select_agent, so routing happens twice for a planned request: once on the original message (discarded when a plan is produced) and once per step.

• planner.py — classify_complexity(text) is a regex pre-filter (≥30 chars + connectives like "and then", "after that", "also send") that gates the LLM call. generate_plan(text, deps) runs a tool-less _planner_agent: Agent[None, ExecutionPlan]. Returns None when needs_planning=False or len(steps) < 2 — the message then falls through to the normal single-step flow.
• executor.py — format_plan_preview(plan) renders the plan as a Telegram approval message. execute_plan(plan, deps, send_progress) runs steps sequentially; each step is independently routed via classify_intent + select_agent, with truncated outputs of prior steps threaded forward as a scratchpad so later steps can build on earlier results. A live ▶/✓ progress checklist is edited into the same message after each step.
• Resume on failure — when a step throws, execute_plan saves a PendingAction(action_type="plan_resume") with a PlanResumePayload(plan, remaining_steps, scratchpad, failed_step, failure_reason) and stops. Confirm restarts from remaining_steps; Cancel aborts cleanly.

The plan itself is always behind the standard inline approval keyboard, even when none of the constituent tools would normally be gated. Full sequence: Multi-step plan flow.

Observability — Logfire + Langfuse + W&B Weave (app/observability.py) — Spec 009

Up to three observability stacks share a single OpenTelemetry tracer provider and run side-by-side without duplicating instrumentation:

graph LR
    PA[Pydantic AI agent.run] --> OTEL[OTEL TracerProvider]
    FAST[FastAPI requests] --> OTEL
    OTEL --> LF[Logfire processor\ninfra · FastAPI routes · loop spans]
    OTEL --> LFC[Langfuse processor\nLLM generations · sessions · scores]
    OTEL --> WB[Weave OTLP exporter\nagent + tool spans · optional]
    LF --> LFCloud[(Logfire Cloud)]
    LFC --> LFCloud2[(Langfuse Cloud)]
    WB --> WBCloud[(W&B Weave Cloud)]
    LF -. self-diagnosis MCP .-> AG[Agent — diagnostics_agent]

init_observability(settings, app) is called once from create_app() before any agent runs. It first installs an SDK TracerProvider as the OTEL global (_ensure_sdk_tracer_provider, so processors aren't silently swallowed by the default proxy provider), then configures Logfire (logfire.configure() + instrument_fastapi) when LOGFIRE_TOKEN is set, and calls Agent.instrument_all() from Pydantic AI when Langfuse is enabled so spans land on the global provider where all processors observe them. (When only Logfire is enabled, logfire.instrument_pydantic_ai() is used instead — it routes through Logfire's internal tracer, which Langfuse would not see.)

W&B Weave (optional, off by default). When WEAVE_ENABLED=true (plus WEAVE_API_KEY + WEAVE_PROJECT as <entity>/<project>), _init_weave() does two things: calls weave.init() so the Weave SDK auto-patches supported LLM clients, and attaches an OTLP BatchSpanProcessor to the shared provider that ships Pydantic AI spans to https://trace.wandb.ai/otel/v1/traces. So agent + tool traces land in the Weave UI alongside Logfire and Langfuse — all three observe the same OTEL spans. Weave init fails soft (a warning, never a boot failure) if the key/project are missing.

Per-turn trace grouping. langfuse_root_span(name, session_id, user_id, …) is an async context manager wrapping the whole turn so every agent.run() inside it nests under one Langfuse trace instead of producing one trace per call.

Fire-and-forget nesting. spawn_with_otel_context(coro) is an asyncio.create_task() variant that propagates the current OTEL span context, so post-conversation background work (fact extraction, capability-gap detection, delegation orchestration, the self-redeploy marker) nests under the originating turn trace instead of appearing as an orphan. Genuinely independent work (scheduled loops) keeps bare create_task().

Stage-level spans inside a turn. Pydantic AI's Agent.instrument_all() covers LLM generations + tool calls automatically, but the preprocessing and routing stages would otherwise be invisible. The following functions are decorated with Langfuse @observe so they appear as nested children of telegram.turn:

• preprocess.stt — app.tools.voice.transcribe_audio (audio bytes → transcript)
• preprocess.vision — app.tools.vision.analyze_image (image bytes → description)
• embed_text — app.tools.embedding.embed_text (sanitized: only chars/dims/model captured, never the vector or api_key)
• routing.classify_intent — app.tools.router.classify_intent (output: {categories, method} where method is keyword | semantic | context_hint | none)
• context_injection — app.utils.message_utils.inject_context (output: per-layer counts + 1500-char preview of what was prepended to the user message)
• message_history_retrieval — app.utils.message_utils.fetch_message_history (metadata: {mode: chronological|semantic, recent_count, semantic_count, semantic_enabled}; mode=error indicates the semantic path failed and chronological was used)

update_current_span(input=, output=, metadata=) in observability.py is the helper they use to attach sanitized payloads; it degrades to a no-op when Langfuse is disabled.

Asynchronous trace scoring. When a Confirm/Cancel/Edit happens minutes after the original turn closed, score_trace(action.trace_id, name, value) attaches a quality score to the original trace. PendingAction.trace_id is captured at gate time via OTEL context (_current_otel_trace_id() in app/approval.py) precisely so this delayed score can land. Three score names are emitted today: user_approval (Confirm 1.0 / Cancel 0.0), user_edit (Edit tap), and agent_error (in-flight exception, written via score_current_trace).

Langfuse-managed Prompts (app/prompts.py + prompts.lock.json) — Spec 009

Nine system prompts can be tuned in the Langfuse UI without redeploying — get_prompt(name, fallback) returns the production-labeled Langfuse version when reachable, otherwise the code constant. prompts.lock.json (repo root) pins each constant's sha256, and check_drift() runs at startup to warn when code has been edited without being pushed. The sync workflow lives in scripts/sync_prompts.py; full prompt list and CLI usage in Tools → Langfuse-managed Prompts.

Briefing Agent (app/routing/agents.py → briefing_agent)

A dedicated agent used exclusively by the scheduled briefing loops (morning, evening, weekly). It is never invoked for interactive user messages.

Tool set: memory (tasks, notes, reminders) + utility (weather, datetime, search, URL, maps) + all email MCP tools + all calendar MCP tools + Microsoft To Do (todo_*) MCP tools.

Why separate? The interactive agents are designed around one domain. The morning briefing legitimately needs email + calendar + tasks + weather simultaneously — a cross-domain request that would normally fall back to the full 45-tool full_agent. briefing_agent gives it a pre-composed, focused tool set without the overhead of GitHub, news, or browser tools that are irrelevant for briefings.

Skill Registry (app/skills/)

File-drop plugin architecture. Any .py file placed in app/skills/ with a function decorated @skill is auto-discovered and registered as an agent tool on startup via load_skills(agent). Modules starting with _ are ignored. Loading is idempotent — calling load_skills() twice registers nothing new.

Built-in skills: - read_later.py — Content Curator (save_to_read_later, list_read_later, delete_read_later, get_read_later_digest) - travel_briefing.py — get_travel_briefing — on-demand travel summary (weather + maps/transit + LLM synthesis) - cv.py — store_cv, get_cv — parse and persist CV as structured user facts for job application workflows - research.py — deep_research — multi-step web research: generates sub-questions, fetches and synthesises sources in parallel, saves result as a "Research: <topic>" note - meeting_notes.py — get_meeting_insights, list_recent_meetings — extract decisions/actions/key points from Drive transcripts via mini-model; save as notes. Source-agnostic: _SOURCE_REGISTRY maps source name to fetcher, allowing Teams or Notion to be wired in later.

Capability-Gap Detection (app/tools/source.py + app/memory/capability_gaps.py)

Kwasi inspects its own deployed code and logs the requests it declines, so real capability holes surface instead of being lost in conversation.

flowchart LR
    subgraph turn[Each Telegram turn]
        A[agent response] --> B[regex prefilter]
        B -->|negation match| C[mini-model classify]
        C --> D[grep_source verify capability]
        D -->|no source matches| E[gap]
        D -->|matches found| F[available]
    end
    E & F -->|fire-and-forget| G[(capability_gaps)]
    G --> H[weekly digest → build it yourself]

The classification matters: available means the capability exists and the model failed to find it — a routing or prompt problem — while gap is a genuine hole. Only the second is worth building for.

Removed 2026-08-22. A propose_skill / activate_proposed_skill flow used to sit on the end of this pipeline: the model drafted a complete .py file, a static AST validator checked it, and an approval card wrote it to app/skills/ and hot-reloaded it onto every agent. In five months it produced six proposals, five rejected, and the one activated skill (sync_podcast_insights) had to be rewritten from scratch in August after fabricating content. Adding a tool is a small PR with tests, review and CI — all three of which that path bypassed. The detection half stays; the drafting half is gone.

System Prompt Architecture

build_system_prompt() assembles the prompt from static and live parts, in that order — static first, because Gemini's implicit cache keys on a stable prefix and any per-turn variation early in the prompt costs the discount for the whole run. That is also why the datetime line goes in the user turn, not here.

Live sections: the reflection profile, active behavioural learnings (ranked by recurrence_count, capped at 10), the introspected database schema, enrolled vault service names, and tool guidance narrowed to active_categories.

It is registered as instructions, not system_prompt. A dynamic @agent.system_prompt is silently dropped whenever a non-empty message_history is passed — and since build_message_history reconstructs history without a SystemPromptPart, the entire prompt vanished on every follow-up turn until PR #72. tests/test_system_prompt_survives_history.py pins it structurally.

Self-management subsystem (app/tools/config_registry.py)

Kwasi changes its own configuration and redeploys itself from Telegram.

CONFIG_REGISTRY is the single source of truth: a list of ConfigField entries from which both allowlists are derived — ALLOWED_CONFIG_KEYS (runtime overrides) and RAILWAY_ENV_ALLOWLIST (environment writes). Adding a self-manageable setting is one line, and the two lists cannot drift apart. An independent deny-substring gate (TOKEN/SECRET/KEY/PASSWORD/URL/…) blocks secret-shaped names regardless of the registry.

Two control planes: set_runtime_config (instant, in-memory, persisted to system:config:<field> and re-applied at boot) and railway_set_env (bakes into the real environment, needs a redeploy). Both are approval-gated. Values are validated by the field's registry validator before the write, so a bad MODEL_NAME cannot bake in and crash-loop the deploy.

After a self-redeploy, app/self_redeploy.py leaves a marker that the next boot consumes to post a ✅/⚠️ — so a failed build is reported rather than silently leaving the previous version running.

Task Delegation (app/delegate.py)

For work no existing tool covers, Kwasi spawns a coding agent in an ephemeral E2B sandbox. Three backends behind two tools: delegate_to_coding_agent (OpenCode by default — Gemini-backed and cheap; Claude Code for harder agentic work) and a dedicated delegate_web_task (browser-use + Gemini, for DOM-driven scraping and multi-step form-fills).

The browser backend is a separate tool rather than a backend parameter, because the model routes on tool names: a static page is browse_web, a live DOM is delegate_web_task, iterative code is delegate_to_coding_agent.

Approval-gated, with a daily rate limit, a wall-clock cap, and a soft cost cap on the metered path. Runs are asynchronous — the executor returns a "working on it" sentinel and the result arrives later as its own Telegram message. Container restarts are recovered at boot: orphaned running rows are marked failed and the user is told, rather than the delegation vanishing.

Google Credential Broker (app/interfaces/mcp/credential_broker.py)

Caches Google access tokens and persists rotated refresh tokens to storage under system:oauth:<provider>:<account>, preferring the stored token over the env var.

That preference caused a two-time outage: a persisted token that later went invalid permanently shadowed a freshly re-authed GMAIL_*_REFRESH_TOKEN, so updating the environment did nothing and the account stayed broken with invalid_grant. The broker now self-heals — on a RefreshError, if the env refresh token differs from the stored one it retries with it and adopts it. If the env token is itself bad it still raises.

Mini App (app/routers/miniapp.py + app/miniapp.py)

A web page served inside the Telegram chat — today, memory, approvals, status. Authenticated on Telegram-signed initData rather than DASHBOARD_SECRET, and scoped to ALLOWED_TELEGRAM_USER_IDS. Approvals resolve through the same app.approval.confirm_action the inline keyboard uses, so the two surfaces cannot diverge. Full detail in Telegram Mini App.

Key Design Decisions

Why Pydantic AI?

Pydantic AI handles the agent loop (tool calls → LLM → tool calls → ...), type-safe tool definitions, and model switching. The Agent class abstracts over different LLM providers — switching from Gemini to Claude is a single env var change.

Why polling instead of webhook for Telegram?

Long-polling (updater.start_polling) is simpler to deploy on Railway — no public URL needed for the bot itself, no TLS certificate management. The WhatsApp webhook still uses a proper HTTP endpoint since the Meta platform requires it.

Why asyncio tasks for background loops?

Avoids the complexity of a separate worker process, Celery, or a job scheduler. Since the app is single-user and the loops are low-frequency (5 min / daily / 60 sec), asyncio tasks inside the FastAPI lifespan are sufficient and operationally simple.

Why MCP for email/calendar?

MCP (Model Context Protocol via fastmcp) provides a clean abstraction for external tool servers. Gmail and Outlook have separate auth flows (OAuth2 with Google vs. MSAL with Microsoft) — MCP isolates this complexity from the agent.

Why native PyGitHub instead of the official GitHub MCP server?

The existing "MCP" tools in this project are local Python wrappers, not external server processes. PyGitHub follows the same pattern as other tools (weather, search, news) — in-process, async-wrapped, consistent output formatting. The official GitHub MCP server is a Go binary designed for IDE integrations, not embedded service use.

Why a file-drop skill registry?

Allows extending the agent's capabilities without touching agent.py. New skills can be developed, tested, and dropped in independently. The registry is idempotent and fail-safe — a broken skill file is skipped with a logged error.

Why keyword-based intent routing instead of LLM-based routing?

The routing step runs on every message, so it must have near-zero latency and zero cost. Keyword matching is deterministic, testable, and trivially debuggable — if it misroutes a message, you can see exactly why by reading the keyword list. An LLM-based router would add a full extra inference step per message, significantly increasing latency and cost for a single-user assistant. The keyword approach covers ~95% of real messages correctly; the remaining 5% fall through to the full agent, which is always the safe fallback.

Why a sentinel token instead of pausing the agent mid-run for approval?

The agent runs to completion — it always produces a text response. Pausing mid-run would require a resumable coroutine, persistent agent state, and a way to re-inject the approval result back into an in-progress LLM call. The sentinel pattern avoids all of this: the agent sees "[APPROVAL_PENDING:uuid]" as a normal tool result, writes a response like "your action is pending approval", and exits. handle_message post-processes the result synchronously. On Confirm, a simple direct executor call (no LLM re-run) performs the action. This keeps the agent loop stateless and the approval flow outside the LLM's critical path.

Why a separate app/approval.py module?

approval.py imports only from app/memory/ports — it has no knowledge of agent.py, bot.py, or client.py. This breaks the potential circular import chain: agent.py imports approval.py, bot.py imports approval.py, and client.py imports approval.py — but none of these files import each other. If the gate lived in agent.py or bot.py, it would create a cycle.

Why domain agents instead of a single agent with all tools?

Reducing the tool list the LLM sees for each request reduces token usage per call and improves model focus — a model with 8 tools performs more reliably than one with 50. Domain agents are built once at import time (zero per-request cost), share the same system prompt function, and fall back transparently to the full agent for anything ambiguous. The full agent is always the safety net.

Why both Logfire and Langfuse?

Different jobs. Logfire owns infrastructure and request-shape tracing — FastAPI routes, background-loop ticks, exceptions, latency by file path. The diagnostics_agent uses it via Logfire's MCP read tool to answer "did the briefing run?" or "which tool is slowest?". Langfuse owns the LLM layer — per-generation token usage and cost, prompt versions, session/user grouping, trace-level quality scores tied to user actions (Confirm/Cancel/Edit). W&B Weave is an optional third backend (WEAVE_ENABLED) for teams already on the W&B ecosystem who want agent traces and evaluations in Weave. All processors hang off the same OTEL tracer provider in app/observability.py, so Pydantic AI is instrumented once and spans flow to every enabled backend without duplication. Any one can be disabled (skip its env vars) and the others keep working.

Why a push model for health data instead of a server-side polling loop?

Neither Samsung Health nor Google Health Connect exposes a cloud / server-side REST API — both store data on-device, encrypted, behind explicit user consent. There is no OAuth Kwasi could call from FastAPI to pull samples. Two options remained: (a) integrate a paid third-party aggregator (Terra/Rook/Thryve, ~$400/mo minimum) which itself still requires an Android piece, or (b) write a tiny sideloaded Android bridge (bridge-android/) that reads Health Connect locally and POSTs to /health/ingest. The bridge is one Kotlin Gradle project, ~600 LOC, and never goes through the Play Store — so we sidestep Google's sensitive-permissions review entirely. Per-metric watermarks in DataStore prevent re-sends and gaps across reboots; the server is idempotent on (metric_type, start_time, source_device) so retries are safe. Result: zero ongoing cost, full data ownership, and the rest of Kwasi's read path is identical to any other agent tool.

Why semantic search instead of upgrading keyword search?

Keyword (LIKE / ILIKE) search is fast and good enough for exact recall. Semantic search complements it — it finds content by meaning when the user's words don't match the stored text (e.g. "anything about burnout" finding a note titled "work-life balance thoughts"). Rather than replacing keyword search, the agent uses both: keyword first (zero latency, no API call), semantic as a fallback or in parallel via find_everything. Embeddings are stored at write time so queries are instant. The embed_text call at write time costs one Gemini API call per saved record — negligible for a personal assistant.