Struct Store 

pub struct Store { /* private fields */ }

Persistent memory store backed by SQLite.

Implementations

impl Store

pub async fn upsert_phase_checkpoint( &self, run_id: &str, topology_name: &str, phase_name: &str, sender_id: &str, project: &str, status: &str, output: Option<&str>, error_message: Option<&str>, attempt: i64, ) -> Result<(), MemoryError>

Create or update a checkpoint for one phase within a run.

Uses INSERT OR REPLACE keyed on (run_id, phase_name), so calling this multiple times as the phase progresses is safe and idempotent.

pub async fn get_phase_checkpoint( &self, run_id: &str, phase_name: &str, ) -> Result<Option<PhaseCheckpoint>, MemoryError>

Fetch the checkpoint for a specific phase within a run.

pub async fn get_run_checkpoints( &self, run_id: &str, ) -> Result<Vec<PhaseCheckpoint>, MemoryError>

Fetch all phase checkpoints for a run, ordered by creation time.

Use this to inspect which phases have already completed when resuming a failed run.

pub async fn clear_run_checkpoints( &self, run_id: &str, ) -> Result<(), MemoryError>

Delete all checkpoints for a run.

Call this after a pipeline run completes successfully to reclaim space, or before re-running a pipeline from scratch.

impl Store

pub async fn build_context( &self, channel: &str, incoming: &Request, base_system_prompt: &str, needs: &ContextNeeds, active_project: Option<&str>, summarizer: Option<&dyn Summarizer>, ) -> Result<Context, MemoryError>

Build a conversation context from memory for the provider.

The channel parameter identifies the communication channel since Request is channel-agnostic.

When needs.compact is CompactionStrategy::Summarize and a summarizer is provided, overflow messages (those beyond max_context_messages) are summarized and prepended to the system prompt instead of being silently dropped.

impl Store

pub async fn find_idle_conversations( &self, ) -> Result<Vec<(String, String, String, String)>, MemoryError>

Find active conversations that have been idle beyond the timeout.

pub async fn find_all_active_conversations( &self, ) -> Result<Vec<(String, String, String, String)>, MemoryError>

Find all active conversations (for shutdown).

pub async fn get_conversation_messages( &self, conversation_id: &str, ) -> Result<Vec<(String, String)>, MemoryError>

Get all messages for a conversation (for summarization).

pub async fn close_conversation( &self, conversation_id: &str, summary: &str, ) -> Result<(), MemoryError>

Close a conversation with a summary.

pub async fn close_current_conversation( &self, channel: &str, sender_id: &str, project: &str, ) -> Result<bool, MemoryError>

Close the current active conversation for a sender + project.

pub async fn get_recent_summaries( &self, channel: &str, sender_id: &str, limit: i64, ) -> Result<Vec<(String, String)>, MemoryError>

Get recent closed conversation summaries for a sender.

pub async fn get_all_recent_summaries( &self, limit: i64, ) -> Result<Vec<(String, String)>, MemoryError>

Get recent conversation summaries across all users.

pub async fn get_history( &self, channel: &str, sender_id: &str, limit: i64, ) -> Result<Vec<(String, String)>, MemoryError>

Get conversation history (summaries with timestamps) for a sender.

pub async fn get_memory_stats( &self, sender_id: &str, ) -> Result<(i64, i64, i64), MemoryError>

Get memory statistics for a sender.

impl Store

pub async fn store_fact( &self, sender_id: &str, key: &str, value: &str, ) -> Result<(), MemoryError>

Store a fact (upsert by sender_id + key). If the row was previously soft-deleted, re-storing clears deleted_at so the value is visible again to default-filtered reads.

pub async fn get_fact( &self, sender_id: &str, key: &str, ) -> Result<Option<String>, MemoryError>

Get a single fact by sender and key. Returns None if the row is soft-deleted.

pub async fn delete_fact( &self, sender_id: &str, key: &str, ) -> Result<bool, MemoryError>

Hard-delete a single fact by sender and key. Returns true if a row was deleted. Emergency cleanup only — not exposed on the MemoryStore trait. Default consumer paths should call Self::soft_delete_fact.

pub async fn soft_delete_fact( &self, sender_id: &str, key: &str, ) -> Result<bool, MemoryError>

Soft-delete a single fact by setting its deleted_at timestamp. Returns true if a row transitioned from active to deleted.

pub async fn get_facts( &self, sender_id: &str, ) -> Result<Vec<(String, String)>, MemoryError>

Get all active (not soft-deleted) facts for a sender.

pub async fn delete_facts( &self, sender_id: &str, key: Option<&str>, ) -> Result<u64, MemoryError>

Hard-delete facts for a sender — all if key is None, specific fact if key is Some. Emergency cleanup only — not exposed on the MemoryStore trait. Default consumer paths should call Self::soft_delete_facts.

pub async fn soft_delete_facts( &self, sender_id: &str, key: Option<&str>, ) -> Result<u64, MemoryError>

Soft-delete facts for a sender — every active fact if key is None, only the matching active fact if key is Some. Returns the count of rows that transitioned from active to deleted.

pub async fn list_soft_deleted_facts( &self, sender_id: &str, ) -> Result<Vec<(String, String, String)>, MemoryError>

Read soft-deleted facts for a sender (debug / recovery helper). Returns (key, value, deleted_at) rows ordered by deletion time descending.

pub async fn get_all_facts(&self) -> Result<Vec<(String, String)>, MemoryError>

Get all active facts across all users (excluding the welcomed marker key).

pub async fn get_all_facts_by_key( &self, key: &str, ) -> Result<Vec<(String, String)>, MemoryError>

Get all active (sender_id, value) pairs for a given fact key across all users.

pub async fn is_new_user(&self, sender_id: &str) -> Result<bool, MemoryError>

Check if a sender has never been welcomed (no active welcomed fact).

pub async fn resolve_sender_id( &self, sender_id: &str, ) -> Result<String, MemoryError>

Resolve a sender_id to its canonical form via the user_aliases table.

pub async fn create_alias( &self, alias_id: &str, canonical_id: &str, ) -> Result<(), MemoryError>

Create an alias mapping: alias_id → canonical_id.

pub async fn find_canonical_user( &self, exclude_sender_id: &str, ) -> Result<Option<String>, MemoryError>

Find an existing welcomed user different from sender_id. Skips soft-deleted welcomed markers.

pub async fn store_limitation( &self, title: &str, description: &str, proposed_plan: &str, ) -> Result<bool, MemoryError>

Store a limitation (deduplicates by title, case-insensitive).

pub async fn get_open_limitations( &self, ) -> Result<Vec<(String, String, String)>, MemoryError>

Get all open limitations: (title, description, proposed_plan).

impl Store

pub async fn store_exchange( &self, channel: &str, incoming: &Request, response: &Response, project: &str, ) -> Result<(), MemoryError>

Store a user message and assistant response.

The channel parameter identifies the communication channel (e.g. “api”, “slack”) since Request is channel-agnostic.

pub async fn search_messages( &self, query: &str, exclude_conversation_id: &str, sender_id: &str, limit: i64, ) -> Result<Vec<(String, String, String)>, MemoryError>

Search past messages across all conversations using FTS5 full-text search.

impl Store

pub async fn store_outcome( &self, sender_id: &str, domain: &str, score: i32, lesson: &str, source: &str, project: &str, ) -> Result<(), MemoryError>

Store a raw outcome from a REWARD marker.

pub async fn get_recent_outcomes( &self, sender_id: &str, limit: i64, project: Option<&str>, ) -> Result<Vec<(i32, String, String, String)>, MemoryError>

Get recent outcomes for a sender.

When project is Some, returns only outcomes for that project. When project is None, returns all outcomes.

pub async fn get_all_recent_outcomes( &self, hours: i64, limit: i64, project: Option<&str>, ) -> Result<Vec<(i32, String, String, String)>, MemoryError>

Get recent outcomes across all users.

pub async fn store_lesson( &self, sender_id: &str, domain: &str, rule: &str, project: &str, ) -> Result<(), MemoryError>

Store a distilled lesson with content-based deduplication.

Multiple lessons can exist per (sender_id, domain, project). If the exact same rule text already exists, its occurrences counter is bumped instead of creating a duplicate. After insertion, a cap of 10 lessons per (sender_id, domain, project) is enforced — oldest are pruned.

pub async fn get_lessons( &self, sender_id: &str, project: Option<&str>, ) -> Result<Vec<(String, String, String)>, MemoryError>

Get lessons for a sender.

When project is Some, returns project-specific lessons first, then general. When project is None, returns general lessons only (project = ‘’).

pub async fn get_all_lessons( &self, project: Option<&str>, ) -> Result<Vec<(String, String, String)>, MemoryError>

Get all lessons across all users.

impl Store

pub async fn store_session( &self, channel: &str, sender_id: &str, project: &str, session_id: &str, ) -> Result<(), MemoryError>

Upsert a CLI session for a (channel, sender_id, project) tuple.

pub async fn get_session( &self, channel: &str, sender_id: &str, project: &str, ) -> Result<Option<String>, MemoryError>

Look up the CLI session_id for a (channel, sender_id, project) tuple.

pub async fn clear_session( &self, channel: &str, sender_id: &str, project: &str, ) -> Result<(), MemoryError>

Delete the CLI session for a specific (channel, sender_id, project).

pub async fn clear_all_sessions_for_sender( &self, sender_id: &str, ) -> Result<(), MemoryError>

Delete all CLI sessions for a sender.

impl Store

pub async fn create_task( &self, channel: &str, sender_id: &str, reply_target: &str, description: &str, due_at: &str, repeat: Option<&str>, task_type: &str, project: &str, ) -> Result<String, MemoryError>

Create a scheduled task. Deduplicates on two levels:

1. Exact match: same sender + description + normalized due_at.
2. Fuzzy match: same sender + similar description + due_at within 30 min.

pub async fn get_due_tasks(&self) -> Result<Vec<DueTask>, MemoryError>

Get tasks that are due for delivery.

pub async fn complete_task( &self, id: &str, repeat: Option<&str>, ) -> Result<(), MemoryError>

Complete a task: one-shot tasks become ‘delivered’, recurring tasks advance due_at.

pub async fn fail_task( &self, id: &str, error: &str, max_retries: u32, ) -> Result<bool, MemoryError>

Fail an action task: increment retry count and either reschedule or permanently fail.

Returns true if the task will be retried, false if permanently failed.

pub async fn get_tasks_for_sender( &self, sender_id: &str, ) -> Result<Vec<(String, String, String, Option<String>, String, String)>, MemoryError>

Get pending tasks for a sender.

pub async fn cancel_task( &self, id_prefix: &str, sender_id: &str, ) -> Result<bool, MemoryError>

Cancel a task by ID prefix (must match sender).

pub async fn update_task( &self, id_prefix: &str, sender_id: &str, description: Option<&str>, due_at: Option<&str>, repeat: Option<&str>, ) -> Result<bool, MemoryError>

Update fields of a pending task by ID prefix (must match sender).

pub async fn defer_task( &self, id: &str, new_due_at: &str, ) -> Result<(), MemoryError>

Defer a pending task to a new due_at time (by exact ID).

impl Store

pub async fn record_usage( &self, sender_id: &str, session_id: &str, tokens: u64, model: &str, ) -> Result<(), MemoryError>

Record token usage for a completed API request, total tokens only.

Thin wrapper over Store::record_usage_full for callers that do not have a per-dimension breakdown. Cost is estimated using known per-model pricing; unrecognized models record cost as 0.0.

pub async fn record_usage_full( &self, sender_id: &str, session_id: &str, tokens: u64, model: &str, breakdown: UsageBreakdown, ) -> Result<(), MemoryError>

Record token usage with a per-dimension breakdown.

tokens is the authoritative total used for cost estimation and summary aggregation. The breakdown columns are stored verbatim and surface in UsageSummary for cost telemetry (e.g. cache hit ratio).

pub async fn get_session_usage( &self, session_id: &str, ) -> Result<UsageSummary, MemoryError>

Get aggregated token usage for a session.

pub async fn get_total_usage(&self) -> Result<UsageSummary, MemoryError>

Get aggregated token usage across all sessions in the store.

Useful for project-wide cost reporting (e.g. the kx /cost command) when callers do not maintain a stable session id.

impl Store

pub async fn new(config: &MemoryConfig) -> Result<Self, MemoryError>

Create a new store, running migrations on first use.

pub fn pool(&self) -> &SqlitePool

Get a reference to the underlying connection pool.

impl Store

pub async fn db_size(&self) -> Result<u64, MemoryError>

Get the database file size in bytes.

Trait Implementations

impl Clone for Store

fn clone(&self) -> Store

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl MemoryStore for Store

fn close_current_conversation<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, channel: &'life1 str, sender_id: &'life2 str, project: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<bool, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Mark the active conversation for (channel, sender_id, project) as closed. Returns true if a row transitioned from active to closed.

fn get_memory_stats<'life0, 'life1, 'async_trait>( &'life0 self, sender_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(i64, i64, i64), MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Aggregate counters: (conversation_count, message_count, fact_count).

fn db_size<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<u64, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

On-disk byte size of the SQLite database file.

fn get_total_usage<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<UsageSummary, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Aggregate token usage across all sessions.

fn get_history<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, channel: &'life1 str, sender_id: &'life2 str, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, String)>, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Recent closed-conversation summaries for a given channel + sender, newest first, capped at limit.

fn search_messages<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, query: &'life1 str, exclude_conversation_id: &'life2 str, sender_id: &'life3 str, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, String, String)>, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

FTS5 full-text search over user messages, excluding the live conversation. Returns up to limit (channel, role, content) rows.

fn store_fact<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, sender_id: &'life1 str, key: &'life2 str, value: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<(), MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Upsert a fact for (sender_id, key). If the row was previously soft-deleted, this clears deleted_at so the value is visible again to default-filtered reads.

fn get_fact<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, sender_id: &'life1 str, key: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Read a single active fact by (sender_id, key). Returns None if the row is soft-deleted, missing, or never existed.

fn get_facts<'life0, 'life1, 'async_trait>( &'life0 self, sender_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, String)>, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Active (not soft-deleted) facts for sender_id.

fn soft_delete_fact<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, sender_id: &'life1 str, key: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<bool, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Soft-delete a single fact by setting its deleted_at timestamp. Returns true if a row transitioned from active to deleted; false if the row was already deleted, missing, or never existed.

fn soft_delete_facts<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, sender_id: &'life1 str, key: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<u64, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Soft-delete multiple facts. With Some(key), soft-deletes that specific key. With None, soft-deletes every active fact for the sender. Returns the count of rows that transitioned from active to deleted.

fn list_soft_deleted_facts<'life0, 'life1, 'async_trait>( &'life0 self, sender_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, String, String)>, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read soft-deleted facts (debug / recovery helper). Returns (key, value, deleted_at) rows for sender_id.

fn create_task<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'life6, 'life7, 'life8, 'async_trait>( &'life0 self, channel: &'life1 str, sender_id: &'life2 str, reply_target: &'life3 str, description: &'life4 str, due_at: &'life5 str, repeat: Option<&'life6 str>, task_type: &'life7 str, project: &'life8 str, ) -> Pin<Box<dyn Future<Output = Result<String, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, 'life6: 'async_trait, 'life7: 'async_trait, 'life8: 'async_trait,

Insert a new scheduled task. Returns the new task id.

fn get_tasks_for_sender<'life0, 'life1, 'async_trait>( &'life0 self, sender_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, String, String, Option<String>, String, String)>, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Pending tasks for sender_id as raw (id, description, due_at, repeat, task_type, project) rows, ordered by due_at ascending.

fn complete_task<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id: &'life1 str, repeat: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<(), MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Mark a task as completed. With Some("daily") / Some("weekly") / etc., reschedules the next occurrence; with None or Some("once"), the task transitions to a terminal status.

fn fail_task<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id: &'life1 str, error: &'life2 str, max_retries: u32, ) -> Pin<Box<dyn Future<Output = Result<bool, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Record a task failure. Increments retry counter; transitions to a terminal failed status when retries exhaust. Returns true if the task transitioned to a terminal state.

fn cancel_task<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id_prefix: &'life1 str, sender_id: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<bool, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Cancel a pending task whose id starts with id_prefix, scoped to sender_id. Returns true if a row was cancelled.

fn get_due_tasks<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<DueTask>, MemoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

All pending tasks whose due_at is in the past.