Storage Layer¶
Storage is abstracted behind a StoragePort protocol. The agent and all tools talk to the protocol — never to a specific adapter directly.
Architecture¶
graph TD
AG[Agent tools] --> SP[StoragePort protocol\napp/memory/ports.py]
BG[Background loops] --> SP
DASH[Dashboard router] --> SP
MINI[Mini App router] --> SP
SP --> PG[PostgresAdapter\napp/memory/adapters/postgres.py]
PG --> DB2[(PostgreSQL + pgvector\nprod, dev and tests)]
There is one adapter. The env var is validated rather than dispatched on — anything but postgres is a hard config error, and a leftover sqlite:// URL fails at startup instead of silently falling back:
Schema¶
The table/column listing below is hand-maintained reference. At runtime the agent never relies on it:
describe_schema()introspects the live schema frominformation_schemaand injects it into the system prompt as a "## Live database schema" section, so the raw-SQL tools (db_query/db_execute) can never drift from the real database. It is cached for the process lifetime and fails safe to an empty section.
interactions¶
Every conversation exchange, from any channel.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
user_message |
TEXT | What you sent |
agent_response |
TEXT | What Kwasi replied |
tools_used |
TEXT | JSON array of tool call records |
channel |
TEXT | "telegram", "cli", or "whatsapp" |
user_id |
TEXT | Telegram user ID (nullable) |
created_at |
TIMESTAMP | UTC |
embedding |
halfvec(3072) | 3072-float embedding from Gemini gemini-embedding-001 (16-bit). NULL until embedded. Used by semantic_search. |
notes¶
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
title |
TEXT | |
content |
TEXT | |
created_at |
TIMESTAMP | UTC |
embedding |
halfvec(3072) | 3072-float embedding from Gemini gemini-embedding-001 (16-bit). NULL until embedded. Used by semantic_search. |
tasks¶
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
title |
TEXT | |
status |
TEXT | "todo" or "done" |
due_date |
TEXT | ISO date string (YYYY-MM-DD), nullable |
notified |
BOOLEAN | true once a due-date notification has been sent |
todo_task_id |
TEXT | Nullable. Links this row to its Microsoft To Do counterpart (Spec 013). NULL = not mirrored |
last_synced |
TIMESTAMP | Nullable. The To Do item's lastModifiedDateTime as of the last push/reconcile — the echo-guard watermark. See Background Loops → To Do Sync Loop |
created_at |
TIMESTAMP | UTC |
reminders¶
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
user_id |
TEXT | Telegram user ID or WhatsApp phone number |
chat_id |
TEXT | Delivery target: Telegram chat ID or WhatsApp phone number |
message |
TEXT | The reminder text |
remind_at |
TIMESTAMP | UTC — when to fire |
status |
TEXT | "pending", "sent", or "cancelled" |
channel |
TEXT | "telegram" or "whatsapp" — delivery channel |
created_at |
TIMESTAMP | UTC |
context¶
Dual-purpose: it stores the long-term user profile produced by the Reflection Engine and doubles as a general-purpose system key-value store — runtime config overrides, deployment markers, loop heartbeats, and auth caches all live here under system:* keys. Reads are exact-key only (get_context(user_id) / save_context(...)), so consumers fetch a known fixed set of keys. The one prefix-scoped operation is delete_old_context(prefix, days), which exists for the per-message keys below — the only family that grows without bound.
| Column | Type | Notes |
|---|---|---|
user_id |
TEXT | Primary key. Holds the profile or a system key (see below). |
content |
TEXT | Markdown profile ≤550 words (for "global"); ISO date string, JSON, or iso\|status\|detail for system keys; "" = cleared (read back as absent). |
last_updated |
TIMESTAMP | UTC |
System keys (user_id prefixes):
| Key | Written by | Content |
|---|---|---|
global |
Reflection Engine | The user profile (markdown) |
system:briefing |
morning-briefing loop | Today's date — dedups the daily send across restarts |
system:stale_check |
reflection | This week's Monday — weekly stale-facts nudge cooldown |
system:meeting_prep:<event_id> |
meeting-prep loop | Per-event prep-sent marker |
system:meeting_followup:gmail:<message_id> |
meeting-followup loop | Per-email processed marker |
system:config:<field> |
set_runtime_config |
A runtime config override, e.g. system:config:model_name → anthropic:claude-sonnet-4-6. Re-applied at boot by load_overrides. |
system:pending_self_redeploy |
railway_set_env / railway_redeploy executors |
JSON marker {chat_id, kind, change, expected, triggered_at, prev_deploy_id}. Consumed on next boot by confirm_pending_self_redeploy to post a ✅/⚠️. |
system:<loop>:heartbeat |
background loops | Loop health signal iso\|status\|detail (e.g. 2026-06-09T…\|ok\|3_processed). diagnose_self reads these to surface silent failures. |
system:outlook_msal_cache |
Outlook MSAL | Serialized MSAL token cache (opaque) |
system:oauth:<provider>:<account> |
credential_broker |
Cached Google access token + rotated refresh token. A stale row here once shadowed a freshly re-authed env var and kept an account broken; the broker now self-heals |
system:msgtrace:<chat_id>:<message_id> |
every sent message | JSON {trace_id, genre, sent_at, reaction} — maps a Telegram message back to the trace that produced it so a reaction minutes later can still be scored. The only per-message key, hence the 30-day prune |
system:pr_review:<owner>/<repo>#<n> |
PR-review pipeline | Per-PR dedup marker, 18h TTL so an updated PR can be re-reviewed next day |
system:podcast_note:<task_id> |
sync_podcast_insights |
Keys the generated note to the Graph task ID, so a re-run updates in place and a retitle does not duplicate |
user_facts¶
Durable key-value facts about the user, treated as ground truth. Written by the agent (remember_fact) or auto-extracted by nightly reflection. Fetched at runtime via recall_facts() — not injected into the system prompt. Mostly permanent, but decay_user_facts() removes a narrow class of rapidly-stale auto-facts (date-stamped snapshots, completed-task records, external-system shadows) once per nightly reflection.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
key |
TEXT | UNIQUE. snake_case identifier, e.g. home_address, partner_name |
value |
TEXT | The actual fact value |
category |
TEXT | location, personal, preference, work, health, or general |
source |
TEXT | "agent" (proactive/explicit) or "reflection" (auto-extracted nightly) |
created_at |
TIMESTAMP | UTC — set on first insert, never changed |
updated_at |
TIMESTAMP | UTC — refreshed on every upsert |
The unique index on key makes save_user_fact an upsert — with one exception: a row whose source is "user" cannot be overwritten by a reflection or agent write. The ON CONFLICT clause carries WHERE user_facts.source <> 'user' OR EXCLUDED.source = 'user', so a fact you corrected in the Mini App survives the nightly reflection re-extracting the same key. Delete the row to hand it back to extraction. See Memory → Permanent User Facts.
news_topics¶
Explicitly followed news topics.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
topic |
TEXT | Unique, stored lowercase |
created_at |
TIMESTAMP | UTC |
seen_news¶
Rolling 7-day dedup store for fetched news stories.
| Column | Type | Notes |
|---|---|---|
url |
TEXT | Primary key |
title |
TEXT | Story headline |
seen_at |
TIMESTAMP | UTC — entries older than 7 days are purged on next get_news call |
scheduled_tasks¶
User-defined recurring jobs evaluated by _user_scheduled_tasks_loop every 60 seconds.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
name |
TEXT | Human-readable label |
prompt |
TEXT | What to run the agent with when the task fires |
cron_expression |
TEXT | Standard 5-field cron, e.g. 0 9 * * 1 = Monday 9am |
timezone |
TEXT | IANA timezone (defaults to USER_TIMEZONE) |
enabled |
BOOLEAN | true to run, false to pause |
created_by |
TEXT | "user" or "system" |
last_run |
TIMESTAMP | UTC — last time this task successfully fired |
created_at |
TIMESTAMP | UTC |
audit_log¶
Every tool call the agent makes, with sanitised arguments and truncated output.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
timestamp |
TIMESTAMP | UTC |
tool_name |
TEXT | Name of the tool called |
arguments |
TEXT | JSON — keys matching key, token, secret, password are redacted to "***" |
result_summary |
TEXT | First 200 characters of tool output |
channel |
TEXT | "telegram", "whatsapp", or "cli" |
user_id |
TEXT | Nullable |
read_later¶
Articles saved by the user via the Content Curator skill.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key (prefix match supported in delete_read_later) |
url |
TEXT | Original URL |
title |
TEXT | Article title |
summary |
TEXT | Auto-generated at save time via LLM summarisation |
note |
TEXT | Optional user annotation |
tags |
TEXT | Comma-separated keywords extracted from title + summary at save time (frequency-ranked, stop-word filtered, up to 10). Used by find_relevant_read_later() to surface contextually relevant articles. |
added_at |
TIMESTAMP | UTC |
read_at |
TIMESTAMP | Set when marked read via mark_read_later_read (nullable); list_read_later(include_read=False) filters on it |
embedding |
halfvec(3072) | 3072-float embedding from Gemini gemini-embedding-001 (16-bit). NULL until embedded. Used by semantic_search. |
alert_rules¶
Proactive alert rules evaluated every 5 minutes by _alert_loop. Two system defaults seeded on first run.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
name |
TEXT | Human-readable label |
trigger_type |
TEXT | task_due_today, task_overdue (phase 2: meeting_soon, email_arrived) |
conditions |
TEXT | JSON object of trigger-specific params (e.g. {"fire_hour": 11}) |
cooldown_hours |
INTEGER | Minimum hours between consecutive firings — prevents spam |
enabled |
BOOLEAN | true to evaluate, false to pause |
created_by |
TEXT | "user" or "system" |
last_fired |
TIMESTAMP | UTC — last time this rule triggered a message |
created_at |
TIMESTAMP | UTC |
agent_learnings¶
Behavioral corrections extracted by the nightly Reflection Engine. Promoted from "candidate" to "active" when seen in ≥2 reflection cycles.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
rule |
TEXT | UNIQUE (normalized lowercase). Short imperative sentence ≤200 chars, e.g. "always confirm before deleting any item" |
category |
TEXT | response_format, tool_usage, decision_making, tone, or scope |
recurrence_count |
INTEGER | Incremented each time the same rule is re-extracted — triggers promotion at ≥2 |
status |
TEXT | "candidate" (new, not yet confirmed) or "active" (shown in system prompt) |
first_seen |
TIMESTAMP | UTC |
last_seen |
TIMESTAMP | UTC |
pending_actions¶
Holds consequential actions awaiting user confirmation via the Telegram inline approval flow. Created by approval_gate() when the agent calls a gated tool; resolved by the user tapping Confirm, Cancel, or Edit.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
tool_name |
TEXT | e.g. "create_task", "send_email_wrapper", "__plan__", "__plan_resume__" |
action_type |
TEXT | "task", "reminder", "note", "journal", "email", "calendar", "todo", "slack", "code", "delegation", "database", "config", "deployment", "plan" (Spec 008 — multi-step plan awaiting Confirm), or "plan_resume" (Spec 008 — partial-failure resume) |
payload |
TEXT | JSON-serialized kwargs to pass to the executor on Confirm. For "plan" this is the serialised ExecutionPlan; for "plan_resume" it is the serialised PlanResumePayload (plan + remaining_steps + scratchpad + failed_step + failure_reason). |
preview_text |
TEXT | Human-readable description shown to the user |
chat_id |
TEXT | Telegram chat ID that triggered the action |
message_id |
INTEGER | Telegram message ID of the preview message (set after send, used by expiry loop to edit) |
status |
TEXT | "pending", "confirmed", "cancelled", "expired", or "failed" |
notified |
BOOLEAN | true once the user has seen an expiry notice |
edit_count |
INTEGER | Number of times the user has used the Edit button on this action |
original_user_message |
TEXT | The user's original message, preserved for the Edit re-run prompt |
trace_id |
TEXT | Nullable. OTEL trace ID captured at gate time so the eventual Confirm/Cancel/Edit decision can attach a user_approval / user_edit score to the originating trace asynchronously (Spec 009 — Langfuse observability). |
created_at |
TIMESTAMP | UTC |
expires_at |
TIMESTAMP | UTC — created_at + 30 min for a chat request; + 24 h for one proposed by a background loop or /message (#144), which is read whenever the owner next looks. Gate rejects execution after this point |
Indexed on (chat_id, status) and (expires_at, status) for efficient per-chat and expiry lookups.
pending_intentions¶
Soft personal commitments extracted from conversation by the nightly Reflection Engine. Follow-ups delivered proactively.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
text |
TEXT | The intention text, e.g. "call doctor about knee" |
mentioned_at |
TIMESTAMP | UTC — when the intention was first expressed |
follow_up_days |
INTEGER | Days until first follow-up (default 3) |
follow_up_at |
TIMESTAMP | UTC — next scheduled follow-up delivery |
status |
TEXT | pending, snoozed, resolved, dismissed |
last_follow_up_at |
TIMESTAMP | UTC — when the last follow-up was sent (nullable) |
created_at |
TIMESTAMP | UTC |
journal_entries¶
Private journal entries saved by the agent via save_journal_entry. Included in the nightly Reflection Engine context so they inform the long-term narrative profile and intention extraction. A weekly digest fires via _journal_digest_loop.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
content |
TEXT | Full text of the entry |
title |
TEXT | ≤10-word title capturing the main theme, generated by the model |
source |
TEXT | "voice" (default) or "text" — how the entry was created |
created_at |
TIMESTAMP | UTC |
Indexed on created_at for efficient time-range lookups. Entries are never embedded (too personal/short for vector search) — retrieved by date range only.
health_samples¶
Wearable / Health Connect samples ingested by the Android bridge (Spec 010). Single normalised, append-only table covering all 11 metric types with a flexible JSON value payload — schema is unchanged when adding a new metric type.
| Column | Type | Notes |
|---|---|---|
id |
BIGSERIAL | Primary key |
metric_type |
TEXT | One of: steps, sleep_session, heart_rate, hrv_rmssd, spo2, resting_hr, respiratory_rate, exercise, body_fat, weight, blood_pressure |
start_time |
TIMESTAMP WITH TIME ZONE | UTC |
end_time |
TIMESTAMP WITH TIME ZONE | UTC, nullable — null for instantaneous samples (HR, HRV, SpO2, RHR, …); set for sessions (sleep, exercise) and intervals (steps) |
value |
JSONB | Metric-specific payload — e.g. {"bpm": 62}, {"rmssd_ms": 38}, {"score": 78, "stages": [...]}, {"count": 4231} |
source_device |
TEXT | e.g. Galaxy Watch 5 Pro |
source_app |
TEXT | e.g. com.sec.android.app.shealth |
created_at |
TIMESTAMP WITH TIME ZONE | UTC, when the row landed in the server |
Indexes:
- UNIQUE (metric_type, start_time, source_device) — idempotency key; duplicate POSTs from the bridge are dropped silently
- (metric_type, start_time DESC) — for time-range reads by the agent's health tools
Writes never embed (high volume, low semantic value). Reads are always time-bounded — get_health_samples(metric_type, since, until=None, limit=5000).
app_usage_daily¶
Per-app foreground stats for a single local day, upserted from the Bridge's daily UsageStatsManager poll. Retained indefinitely — it is aggregate, not content.
| Column | Type | Notes |
|---|---|---|
id |
BIGSERIAL | Primary key |
date |
DATE | Local calendar date |
package |
TEXT | App package name |
display_name |
TEXT | Human-readable app name |
foreground_ms |
BIGINT | Foreground time for this app on this date |
launches |
INTEGER | Launch count |
first_used / last_used |
TIMESTAMP | Nullable |
screen_unlocks |
INTEGER | Day-level, denormalised onto every row for the date |
total_foreground_ms |
BIGINT | Day-level, denormalised onto every row for the date |
timezone |
TEXT | IANA tz the local date was computed in |
device |
TEXT | Source device id |
created_at |
TIMESTAMP | Ingest time |
Unique index on (date, package, device) — a re-sent day overwrites rather than duplicating.
Why the day-level columns are denormalised
screen_unlocks and total_foreground_ms describe the day, not the app, and are identical across every row for that date. Copying them onto each row keeps usage in one table with no summary side-table; reads take MAX(screen_unlocks) for a date.
capability_gaps¶
Logged when Kwasi declines a request. Classified by whether the underlying capability exists in the deployed source — see Architecture → Self-Improvement Loop.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
user_message |
TEXT | The original user message |
agent_response |
TEXT | Kwasi's decline response |
classification |
TEXT | "gap" (no source matches → real capability hole), "available" (source matches found → routing/prompt failure), or "unknown" |
suspected_capability |
TEXT | Short noun phrase extracted by the mini-model classifier, e.g. "linkedin scraping" |
classifier_reason |
TEXT | Why this classification — combines grep-result summary + mini-model rationale |
grep_evidence |
TEXT | Truncated grep_source output that informed the classification (nullable) |
status |
TEXT | "open", "resolved", or "dismissed" |
created_at |
TIMESTAMP | UTC |
resolved_at |
TIMESTAMP | UTC, nullable |
Indexes: (status, created_at DESC) for the digest, (classification) for filtering by class.
delegations¶
Each row tracks one delegate_to_coding_agent or delegate_web_task invocation — see Architecture → Task Delegation.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
task_description |
TEXT | The task sent to the sandbox |
status |
TEXT | "queued", "running", "completed", "failed", "cancelled" |
backend |
TEXT | "opencode" (default), "claude_code", or "web_task" |
chat_id |
TEXT | Telegram chat to deliver result to (nullable) |
session_id |
TEXT | Backend session id (for future --resume; nullable) |
started_at / completed_at |
TIMESTAMP | UTC (nullable) |
result_summary |
TEXT | Final answer (default '') |
files_summary |
TEXT | Comma-separated generated-file paths (default '') |
duration_seconds |
INT | Wall-clock (nullable) |
input_tokens / output_tokens |
INT | Stream-json telemetry (Claude Code path only; nullable) |
estimated_cost_usd |
REAL/NUMERIC | Soft-cap enforced for Claude Code; null otherwise |
error |
TEXT | Failure summary (nullable) |
credential |
TEXT | web_task only: name of a vault credential used for login — never the secret (nullable) |
created_at |
TIMESTAMP | UTC |
trail |
JSONB | web_task only: one row per browser step — action, URL, the model's goal, any error. Nullable. |
Indexes: (status, created_at DESC) for listing + recovering orphans on container restart.
Full-text index — idx_notes_fts, a GIN index on to_tsvector('english', title || ' ' || content), backs StoragePort.keyword_search. The expression must match PostgresAdapter._FTS_DOCUMENT exactly or the planner drops to a sequential scan silently, visible only as latency.¶
vault_credentials¶
Encrypted website logins for delegate_web_task(credential=...) — see Deployment → Credential Vault. The *_enc columns are Fernet ciphertext (master key from VAULT_MASTER_KEY, env only); domains is plaintext (not a secret — needed for allowed_domains scoping). db_query refuses this table.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
service |
TEXT | Unique lookup name, e.g. navette |
domains |
TEXT | Comma-separated hostnames (plaintext) |
username_enc / password_enc |
TEXT | Fernet ciphertext |
totp_seed_enc |
TEXT | Fernet ciphertext, nullable |
storage_state_enc |
TEXT | Fernet ciphertext of a captured browser session (cookies + localStorage), nullable |
session_captured_at |
TIMESTAMPTZ | When a human captured that session, nullable |
created_at / updated_at |
TIMESTAMP | UTC |
Data Models¶
@dataclass
class SemanticSearchResult:
source: str # "notes" | "interactions" | "read_later"
id: str
title: str # note title / "Conversation YYYY-MM-DD" / article title
snippet: str # first 200 chars of relevant content
similarity: float # cosine similarity 0.0–1.0
created_at: datetime | None # used for recency weighting in context injection
StoragePort Protocol¶
Defined in app/memory/ports.py (the single source of truth — read it directly rather than relying on a copy here). Every method is async. The protocol groups into:
- Interactions —
save_interaction,get_interactions,get_interactions_by_user,get_interactions_since,get_interactions_by_ids,search_interactions - Context / facts / learnings / intentions —
get_context/save_context/delete_old_context; user-fact CRUD +decay_user_facts;save_agent_learning/get_agent_learnings/delete_agent_learning; intention CRUD +get_due_intentions/count_followups_today - Notes / tasks / reminders / read-later / journal / news — standard CRUD (+
get_tasks_due/mark_task_notified,mark_read_later_read,get_due_reminders) - Scheduled tasks / alert rules / audit log — CRUD +
get_request_log/delete_old_audit_entries - Pending actions (approval gate), Delegations (
save/get/update/count_delegations_today/list_running_delegations), Vault (save_vault_entry/get_vault_entry/list_vault_services/list_vault_meta) - Self-improvement — capability-gap + proposed-skill CRUD
- Health samples —
save_health_samples(idempotent),get_health_samples,get_health_freshness - Search —
semantic_search,hybrid_search(keyword + dense, RRF-fused) - Raw SQL / schema —
raw_sql_query(read-only),raw_sql_execute,raw_sql_dryrun,describe_schema close
Representative signatures
class StoragePort(Protocol):
async def save_interaction(self, interaction: Interaction) -> None: ...
async def get_interactions(self, limit: int = 50) -> list[Interaction]: ...
async def get_interactions_by_user(self, user_id: str, limit: int = 20) -> list[Interaction]: ...
async def get_interactions_since(self, since: datetime) -> list[Interaction]: ...
async def search_interactions(self, query: str, limit: int = 10) -> list[Interaction]: ...
# Context
async def get_context(self, user_id: str) -> UserContext | None: ...
async def save_context(self, context: UserContext) -> None: ...
async def delete_old_context(self, prefix: str, days: int) -> int: ... # prefix-scoped TTL
# Notes
async def save_note(self, note: Note) -> None: ...
async def get_notes(self) -> list[Note]: ...
async def search_notes(self, query: str) -> list[Note]: ...
async def update_note(self, note_id: str, title: str, content: str) -> bool: ...
async def delete_note(self, note_id: str) -> bool: ...
# Tasks
async def create_task(self, task: Task) -> None: ...
async def list_tasks(self, status: str | None = None) -> list[Task]: ...
async def search_tasks(self, query: str) -> list[Task]: ...
async def get_tasks_due(self) -> list[Task]: ...
async def mark_task_notified(self, task_id: str) -> None: ...
async def update_task(self, task_id: str, title: str | None, due_date: str | None) -> bool: ...
async def complete_task(self, task_id: str) -> bool: ...
async def delete_task(self, task_id: str) -> bool: ...
# Reminders
async def create_reminder(self, reminder: Reminder) -> None: ...
async def get_due_reminders(self) -> list[Reminder]: ...
async def list_reminders(self, user_id: str | None = None) -> list[Reminder]: ...
async def update_reminder_status(self, reminder_id: str, status: str) -> None: ...
async def cancel_reminder(self, reminder_id: str) -> bool: ...
# Scheduled tasks
async def create_scheduled_task(self, task: ScheduledTask) -> None: ...
async def list_scheduled_tasks(self) -> list[ScheduledTask]: ...
async def update_scheduled_task(self, task_id: str, **kwargs) -> bool: ...
async def delete_scheduled_task(self, task_id: str) -> bool: ...
async def update_scheduled_task_last_run(self, task_id: str, last_run: datetime) -> None: ...
# Audit log
async def log_audit_entry(self, entry: AuditEntry) -> None: ...
async def get_audit_log(self, limit: int = 50) -> list[AuditEntry]: ...
# News
async def add_news_topic(self, topic: NewsTopic) -> None: ...
async def remove_news_topic(self, topic: str) -> bool: ...
async def list_news_topics(self) -> list[NewsTopic]: ...
async def add_seen_story(self, story: SeenStory) -> None: ...
async def get_seen_urls(self) -> set[str]: ...
async def cleanup_seen_stories(self, days: int = 7) -> None: ...
# Read Later
async def save_read_later(self, item: ReadLaterItem) -> None: ...
async def list_read_later(self) -> list[ReadLaterItem]: ...
async def delete_read_later(self, item_id: str) -> None: ...
# User Facts
async def save_user_fact(self, fact: UserFact) -> None: ... # upsert on key
async def get_user_fact(self, key: str) -> UserFact | None: ...
async def get_all_user_facts(self) -> list[UserFact]: ... # ordered by category, key
async def search_user_facts(self, query: str) -> list[UserFact]: ...
async def delete_user_fact(self, key: str) -> bool: ...
# Alert Rules
async def create_alert_rule(self, rule: AlertRule) -> None: ...
async def list_alert_rules(self, enabled_only: bool = False) -> list[AlertRule]: ...
async def update_alert_rule(self, rule_id: str, **kwargs) -> bool: ...
async def delete_alert_rule(self, rule_id: str) -> bool: ...
async def update_alert_rule_last_fired(self, rule_id: str, fired_at: datetime) -> None: ...
# Pending Intentions
async def create_intention(self, intention: PendingIntention) -> None: ...
async def list_intentions(self, status: str | None = None) -> list[PendingIntention]: ...
async def get_due_intentions(self) -> list[PendingIntention]: ...
async def update_intention(self, intention_id: str, **kwargs) -> bool: ...
async def count_followups_today(self) -> int: ...
# Agent Learnings
async def save_agent_learning(self, learning: AgentLearning) -> None: ... # upsert on rule
async def get_agent_learnings(self, status: str | None = None) -> list[AgentLearning]: ...
async def delete_agent_learning(self, learning_id: str) -> bool: ...
# Pending Actions (approval flow)
async def save_pending_action(self, action: PendingAction) -> None: ...
async def get_pending_action(self, action_id: str) -> PendingAction | None: ...
async def update_pending_action(self, action_id: str, **kwargs) -> bool: ...
# accepts status/message_id/notified/edit_count/payload/preview_text;
# anything else raises — it used to drop unknown keys silently, which
# turned a missing field into an update that reported success and
# changed nothing
async def list_expired_pending_actions(self) -> list[PendingAction]: ... # pending rows past expires_at
async def get_pending_actions_for_chat(self, chat_id: str, status: str = "pending") -> list[PendingAction]: ...
async def delete_old_pending_actions(self, days: int = 7) -> int: ... # prune confirmed/cancelled/expired rows
async def delete_old_audit_entries(self, days: int = 90) -> int: ... # prune old audit_log rows
# Journal Entries
async def save_journal_entry(self, entry: JournalEntry) -> None: ...
async def list_journal_entries(self, days: int | None = None) -> list[JournalEntry]: ...
# Health Samples (Spec 010 — Wearable / Health Connect ingest)
async def save_health_samples(self, samples: list[dict]) -> tuple[int, int]: ...
# Idempotent on (metric_type, start_time, source_device); returns (accepted, duplicates).
async def get_health_samples(
self,
metric_type: str,
since: datetime,
until: datetime | None = None,
limit: int = 5000,
) -> list[dict]: ...
# Semantic Search
async def semantic_search(
self,
embedding: list[float],
sources: list[str] | None = None, # default: all three
limit: int = 5,
) -> list[SemanticSearchResult]: ...
async def close(self) -> None: ...
Adapter Characteristics¶
A SQLite adapter existed alongside this one until the Phase 1.1 cleanup. It was
removed because it duplicated ~2,800 lines of the same StoragePort and forced
every schema change to be written twice, in two type systems — which is exactly
where parity bugs lived. Postgres is now the only backend, in production and in
tests.
| Feature | PostgreSQL |
|---|---|
| Case-insensitive search | ILIKE |
| Timestamps | Native TIMESTAMP WITH TIME ZONE |
| Connection | asyncpg connection pool (1–10), with SSL fallback |
| Schema init | CREATE TABLE IF NOT EXISTS on first pool use |
| Migrations | ALTER TABLE ... ADD COLUMN IF NOT EXISTS — idempotent by construction |
| Boolean | Native BOOLEAN |
| Vector search | pgvector <=> cosine operator with HNSW index |
| Embedding storage | Native halfvec(3072) (16-bit; requires pgvector ≥0.7) |
Semantic Search¶
The adapter implements semantic_search(), which finds content by meaning rather than keywords. The query text is embedded via embed_text() in app/tools/embedding.py (Gemini gemini-embedding-001, 3072 dimensions) before calling this method.
Uses the pgvector extension with the <=> cosine distance operator and an HNSW index (halfvec_cosine_ops) on each table. Each source independently returns its best limit candidates; results are merged and re-ranked before returning the top limit overall. Embeddings are stored as halfvec(3072) (16-bit floats) — full-precision vector(3072) exceeds pgvector's 2000-dim HNSW limit and would force a sequential scan.
Graceful degradation: If GOOGLE_API_KEY is not set, embed_text() returns None and no embedding is stored or queried — keyword search continues to work normally. If the pgvector extension is unavailable, the CREATE EXTENSION and ADD COLUMN statements fail silently and semantic_search() returns an empty list.
Backfilling: Existing rows without embeddings can be embedded in bulk via POST /embed-backfill (protected by X-Reflection-Secret). Rate-limited to ~10 rows/second.
Schema Migrations¶
The adapter does not use a migration framework. New columns are added via ALTER TABLE statements run on every startup. New tables are created with CREATE TABLE IF NOT EXISTS. Columns use ADD COLUMN IF NOT EXISTS, so re-running a migration is a no-op by construction. No rollback support.
Current migrations applied at startup:
tasks.notified— due-date notification trackingaudit_log.duration_ms,audit_log.tokens_used— request timing and token trackingread_later.tags— keyword tags for relevance matching;read_later.read_at— read-state filteringreminders.channel— delivery channel (telegram/whatsapp)pending_actions.trace_id— Langfuse trace id for delayed approval scoringdelegations.backend,delegations.credential— backend label + vault-credential name (web_task)notes.embedding,interactions.embedding,read_later.embedding— vector embeddings for semantic search (halfvec(3072)via the pgvector ≥0.7 extension; legacyvector(3072)columns auto-migrate in-place on first pool init)health_samples,vault_credentials,capability_gaps,app_usage_daily— created on first connecttasks.todo_task_id/tasks.last_synced— Microsoft To Do sync linkage (Spec 013), added as additive migrations
feed_cards¶
Spec 017. Proactive content Kwasi wants Lawrence to see without interrupting him, shown at the top of the Mini App's Today tab. Each status change keeps its own timestamp, which is how the feed measures its own engagement per kind (StoragePort.feed_engagement, scripts/feed_report.sql).
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
kind |
TEXT | Open-ended: briefing, meeting_prep, task_state, weekly_recap, digest kinds; later goal |
source |
TEXT | The loop or label that produced it, e.g. alert: Overdue tasks |
dedup_key |
TEXT | Nullable. One live card per thing: a partial unique index on dedup_key WHERE status IN ('new','seen') makes recurring state update one card instead of adding one per check. A handled (dismissed/acted) card holds while title and body are unchanged — the next check creates nothing — and a new card appears only when the state changes (found in the first live test: dismissing brought an identical card back five minutes later) |
title / body |
TEXT | One line collapsed; Markdown expanded |
action_id |
TEXT | Nullable. A pending_actions.id when the card proposes something |
actions |
TEXT (JSON list) | Things the card lets you do in place, e.g. {"type": "complete_task", "task_id": …, "label": …}. Default [] |
status |
TEXT | new → seen → acted | dismissed, or new/seen → expired. Forward-only; a replayed transition is a no-op |
created_at |
TIMESTAMPTZ | |
updated_at |
TIMESTAMPTZ | Moves only when an upsert changes title or body; the feed sorts by it |
expires_at |
TIMESTAMPTZ | Default created_at + 7 days; time-bound kinds earlier |
seen_at / acted_at / dismissed_at |
TIMESTAMPTZ | One per transition. Acting or dismissing also sets seen_at |
Retention: unhandled cards past expires_at become expired every 5 minutes; rows older than 60 days are deleted by the hourly cleanup in _approval_expiry_loop.
miniapp_opens¶
Spec 017. One row per opening of the Mini App and per tab switch, posted by the page itself — the first record of whether the Mini App is used at all.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
opened_at |
TIMESTAMPTZ | |
tab |
TEXT | One of today, approvals, memory, jobs, status, inbox (the page's VIEWS); anything else is refused |
Retention: 90 days.
triage_items¶
Spec 018. One email in one account, sorted into a group for the Mini App's Inbox tab. Written by the sort loop, which only reads mail and records it: archive, Trash and replies happen only on his tap.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
account |
TEXT | work_gmail · outlook · personal_gmail. Unique with message_id |
message_id |
TEXT | The provider's id |
thread_id |
TEXT | Gmail threadId / Outlook conversationId |
from_name / from_addr |
TEXT | |
subject / snippet |
TEXT | The snippet is the provider's own preview (~200 chars); the body is never stored |
received_at |
TIMESTAMPTZ | |
grp |
TEXT | reply · act · fyi · noise |
reason |
TEXT | One line, ≤ 120 chars, rendered escaped |
placed_by |
TEXT | rule:<name> · model · user. His move wins: a re-sort never changes grp/reason once this is user (the same rule as user-corrected facts, #108) |
unsubscribe |
TEXT | The List-Unsubscribe value, if any (Spec 018 US5) |
open_url |
TEXT | The email in its own app (Gmail web link with authuser, Outlook webLink) |
status |
TEXT | open → done | deleted | replied (his taps), once; or open → gone when a complete read shows the email left the inbox outside Kwasi (archived in Gmail itself). A re-sort never touches it, so a handled email never reopens |
acted_at |
TIMESTAMPTZ | When it was handled |
created_at |
TIMESTAMPTZ |
Indexes: unique (account, message_id); (status, received_at DESC). Open items list the two main accounts (work_gmail, outlook) first, then newest first.
| snoozed_until | TIMESTAMPTZ | Hidden from the Inbox and its counts until then (his Snooze); back afterwards, even if older than the window. Never written by the sort's upsert |
Retention: rows older than 60 days are deleted hourly.
sender_prefs¶
Spec 018. The group he moved a sender to, per account; the first sorting rule reads it, so a correction applies to that sender's next email.
| Column | Type | Notes |
|---|---|---|
account / from_addr |
TEXT | Primary key together |
grp |
TEXT | The group he chose |
updated_at |
TIMESTAMPTZ |
Retention: kept; a few rows per sender he has ever corrected.
inbox_daily_unread¶
Spec 018 SC-003. Unread and in-inbox counts per account, written once a day by the sort loop, so the unread trend can be read without asking the mail services for history they don't keep.
| Column | Type | Notes |
|---|---|---|
day / account |
DATE / TEXT | Primary key together; a second write the same day is ignored |
unread / in_inbox |
INTEGER |
Retention: kept (one row per account per day).
cleanup_runs¶
Spec 018 US4 (Clean up by sender). One row per Clean up session he starts from the Inbox tab: a read-only scan of mail older than the cut-off in every inbox, then archives of the addresses he ticks.
| Column | Type | Notes |
|---|---|---|
id |
TEXT | Primary key |
status |
TEXT | scanning · ready · archiving · stopping · done · stopped · failed; a row left active with no task running (a deploy) is settled on the next read |
cutoff_days |
INTEGER | 7 / 30 / 90 |
scanned |
INTEGER | emails the scan offered |
excluded / accounts |
TEXT (JSON) | what was left out and why; each account's scan result |
groups |
TEXT (JSON) | sender groups (by domain; by address for personal providers), computed when the scan ends and again after each archive |
picked |
TEXT (JSON) | the addresses he ticked |
archived / failed |
INTEGER | progress, updated per chunk |
error |
TEXT | |
scanned_at / created_at / updated_at |
TIMESTAMPTZ | a scan older than an hour cannot be archived from |
Retention: 7 days, with their items (hourly cleanup).
cleanup_items¶
One row per email a Clean up scan found. Metadata only, never a body.
| Column | Type | Notes |
|---|---|---|
run_id / account / message_id |
TEXT | Primary key together; run_id cascades from cleanup_runs |
thread_id |
TEXT | Gmail archives by conversation, as Done does |
from_addr / from_name / subject |
TEXT | |
received_at |
TIMESTAMPTZ | |
archived_at |
TIMESTAMPTZ | set per chunk as the mail service agrees; NULL = still in the inbox |
Retention: with its run.
trip_watches¶
Spec 019. A booked trip (found in his booking emails) watched on its day.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
kind / identifier / service_date |
TEXT / TEXT / DATE | Unique together: train numbers repeat daily |
origin / destination / departure_time |
TEXT | His stops and booked time, from the booking |
told |
TEXT (JSON) | What he was last told ({"delay", "cancelled"}); the watch compares against this, never the previous reading |
status |
TEXT | watching → done after arrival |
created_at |
TIMESTAMPTZ |
Retention: rows (and their checks) older than 30 days deleted hourly.
trip_watch_checks¶
Spec 019 FR-010. Every check the watch made, sent or not, so "why did (or didn't) it message me" has an answer.
| Column | Type | Notes |
|---|---|---|
id |
TEXT (UUID) | Primary key |
watch_id |
TEXT | trip_watches.id |
checked_at |
TIMESTAMPTZ | |
reading |
TEXT (JSON) | The status read, or the error |
sent / reason |
BOOLEAN / TEXT | Whether a message went, and why or why not |