Struct FrankenStorage 

pub struct FrankenStorage { /* private fields */ }

Primary frankensqlite-backed storage backend.

Implementations

impl FrankenStorage

pub fn open(path: &Path) -> Result<Self>

Open a frankensqlite connection, run migrations, and apply config.

This initializes canonical schema state only. Derived fallback search structures like the in-database fts_messages table are repaired separately so ordinary opens never block on heavyweight maintenance.

pub fn open_writer(path: &Path) -> Result<Self>

Open a writer connection that skips migration (assumes DB already migrated).

Used by the BEGIN CONCURRENT parallel writer pool: each writer needs its own connection with config applied, but migrations have already been run by the primary connection.

pub fn open_readonly(path: &Path) -> Result<Self>

Open in read-only mode using frankensqlite compat flags.

pub fn open_readonly_with_doctor_lock_timeout( path: &Path, timeout: Duration, ) -> Result<Self>

Open in read-only mode with an explicit doctor mutation-lock timeout.

This is primarily useful for probes that need to prove a reader would not enter the archive while cass doctor --fix owns the repair lock.

pub fn close(self) -> Result<()>

pub fn close_without_checkpoint(self) -> Result<()>

pub fn close_best_effort_in_place(&mut self)

pub fn close_without_checkpoint_in_place(&mut self) -> Result<()>

pub fn raw(&self) -> &FrankenConnection

Access the raw frankensqlite connection.

pub fn into_raw(self) -> FrankenConnection

Consume the storage wrapper and return the underlying frankensqlite connection after migrations/repair have already been applied.

pub fn apply_config(&self) -> Result<()>

Apply connection PRAGMAs for parity with SqliteStorage’s apply_pragmas().

Frankensqlite supports all PRAGMAs cass uses (journal_mode, synchronous, cache_size, foreign_keys, busy_timeout). Its default journal_mode is already WAL and default synchronous is NORMAL, matching cass’s requirements.

pub fn run_migrations(&self) -> Result<()>

Run all schema migrations, handling transition from meta table versioning.

The existing SqliteStorage tracks schema version in a meta table entry. The new MigrationRunner uses a _schema_migrations table. This method:

1. Transitions existing databases from meta table → _schema_migrations
2. Runs pending migrations via MigrationRunner
3. Syncs meta.schema_version for backward compatibility

Fresh vs existing databases

Fresh databases use a single combined migration (MIGRATION_FRESH_SCHEMA) that creates the complete V13 schema directly. This avoids the incremental V5 migration which uses DROP TABLE — an operation that triggers a known frankensqlite autoindex limitation.

Existing databases (transitioned from SqliteStorage) are typically at V13 or newer already; additive post-V13 migrations are applied normally.

pub fn schema_version(&self) -> Result<i64>

Return the current schema version from _schema_migrations.

pub fn database_path(&self) -> Result<PathBuf>

Resolve the database file path for this connection.

pub fn open_or_rebuild(path: &Path) -> Result<Self, MigrationError>

Open database with migration, backing up if schema is incompatible.

impl FrankenStorage

pub fn ensure_agent(&self, agent: &Agent) -> Result<i64>

Ensure an agent exists in the database, returning its ID.

pub fn ensure_workspace( &self, path: &Path, display_name: Option<&str>, ) -> Result<i64>

Ensure a workspace exists in the database, returning its ID.

pub fn now_millis() -> i64

Get current time as milliseconds since epoch.

pub fn day_id_from_millis(timestamp_ms: i64) -> i64

Convert a millisecond timestamp to a day ID (days since 2020-01-01).

pub fn hour_id_from_millis(timestamp_ms: i64) -> i64

Convert a millisecond timestamp to an hour ID (hours since 2020-01-01 00:00 UTC).

pub fn millis_from_day_id(day_id: i64) -> i64

Convert a day ID back to milliseconds (start of day).

pub fn millis_from_hour_id(hour_id: i64) -> i64

Convert an hour ID back to milliseconds (start of hour).

pub fn get_last_scan_ts(&self) -> Result<Option<i64>>

Get the timestamp of the last successful scan.

pub fn set_last_scan_ts(&self, ts: i64) -> Result<()>

Set the timestamp of the last successful scan (milliseconds since epoch).

pub fn get_last_indexed_at(&self) -> Result<Option<i64>>

Get the timestamp of the last successful index completion.

pub fn set_last_indexed_at(&self, ts: i64) -> Result<()>

Set the timestamp of the last successful index completion (milliseconds since epoch).

pub fn list_agents(&self) -> Result<Vec<Agent>>

List all registered agents.

pub fn total_conversation_count(&self) -> Result<usize>

Count all archived conversations.

pub fn total_message_count(&self) -> Result<usize>

Count all archived messages.

pub fn purge_agent_archive_data( &self, agent_slug: &str, ) -> Result<AgentArchivePurgeResult>

Remove all archived conversations/messages for one agent slug.

This only affects cass’s local archive database. Source session files on disk are untouched.

pub fn list_workspaces(&self) -> Result<Vec<Workspace>>

List all registered workspaces.

pub fn list_conversations( &self, limit: i64, offset: i64, ) -> Result<Vec<Conversation>>

List conversations with pagination.

pub fn build_lexical_rebuild_lookups( &self, ) -> Result<(HashMap<i64, String>, HashMap<i64, PathBuf>)>

Build lookup maps for agents and workspaces to avoid JOINs in paged conversation queries. Both tables are tiny (tens of rows) so this is effectively free.

pub fn list_conversation_footprints_for_lexical_rebuild( &self, ) -> Result<Vec<LexicalRebuildConversationFootprintRow>>

List per-conversation message footprints in primary-key order.

This deliberately avoids rebuild-path JOINs. Instead we merge ordered single-table reads over conversations and the narrow conversation_tail_state cache in Rust, then use last_message_idx + 1 as a planning estimate.

The planner only needs a sizing heuristic; exact message and byte accounting is performed later by the rebuild packet pipeline as it reads message content for indexing. Rows missing both tail-cache sources fall back to MAX(messages.idx) + 1, which preserves legacy upgraded databases without treating populated conversations as empty.

pub fn lexical_rebuild_has_tail_footprint_metadata(&self) -> Result<bool>

pub fn list_conversation_ids_for_lexical_rebuild(&self) -> Result<Vec<i64>>

List conversation ids in the stable order used by lexical rebuilds.

pub fn list_conversations_for_lexical_rebuild_by_offset( &self, limit: i64, offset: i64, agent_slugs: &HashMap<i64, String>, workspace_paths: &HashMap<i64, PathBuf>, ) -> Result<Vec<LexicalRebuildConversationRow>>

Legacy OFFSET-based traversal for one-time checkpoint migration only.

New code must use list_conversations_for_lexical_rebuild_after_id for keyset pagination.

pub fn list_conversations_for_lexical_rebuild_after_id( &self, limit: i64, after_conversation_id: i64, agent_slugs: &HashMap<i64, String>, workspace_paths: &HashMap<i64, PathBuf>, ) -> Result<Vec<LexicalRebuildConversationRow>>

List lexical rebuild conversations strictly after the given primary key.

Keyset pagination keeps later rebuild pages as cheap as earlier ones, avoiding the ever-growing OFFSET scan cost during large rebuilds.

pub fn list_conversations_for_lexical_rebuild_after_id_through_id( &self, limit: i64, after_conversation_id: i64, through_conversation_id: i64, agent_slugs: &HashMap<i64, String>, workspace_paths: &HashMap<i64, PathBuf>, ) -> Result<Vec<LexicalRebuildConversationRow>>

List lexical rebuild conversations inside an (after_id, through_id] primary-key window.

This lets the rebuild producer respect planned shard boundaries without falling back to client-side trimming or multi-table joins.

pub fn fetch_messages(&self, conversation_id: i64) -> Result<Vec<Message>>

Fetch messages for a conversation.

pub fn fetch_messages_for_lexical_rebuild( &self, conversation_id: i64, ) -> Result<Vec<Message>>

Fetch messages for lexical index rebuilds without deserializing extra metadata.

Tantivy only needs message text and core envelope fields, so avoiding extra_json here prevents rebuilds from rehydrating enormous historical payloads that are irrelevant to lexical search.

pub fn fetch_messages_for_lexical_rebuild_batch( &self, conversation_ids: &[i64], max_messages: Option<usize>, max_content_bytes: Option<usize>, ) -> Result<HashMap<i64, Vec<Message>>>

Fetch messages for multiple conversations during lexical rebuilds.

This preserves the lightweight lexical-rebuild projection while avoiding one round-trip per conversation when rebuilding large canonical indexes.

pub fn stream_messages_for_lexical_rebuild_between_conversation_ids<F>( &self, start_conversation_id: i64, end_conversation_id: i64, f: F, ) -> Result<()>

where F: FnMut(LexicalRebuildMessageRow) -> Result<()>,

Stream lexical rebuild message rows in (conversation_id, idx) order without materializing the full result set.

pub fn stream_grouped_messages_for_lexical_rebuild_between_conversation_ids<F>( &self, start_conversation_id: i64, end_conversation_id: i64, f: F, ) -> Result<()>

where F: FnMut(i64, LexicalRebuildGroupedMessageRows, i64) -> Result<()>,

Stream grouped lexical rebuild message rows in (conversation_id, idx) order by reusing the canonical per-message stream and coalescing rows per conversation.

pub fn stream_grouped_messages_for_lexical_rebuild_from_conversation_id<F>( &self, start_conversation_id: i64, f: F, ) -> Result<()>

where F: FnMut(i64, LexicalRebuildGroupedMessageRows, i64) -> Result<()>,

Stream grouped lexical rebuild message rows from a starting conversation id to the end of the table.

pub fn stream_messages_for_lexical_rebuild_from_conversation_id<F>( &self, start_conversation_id: i64, f: F, ) -> Result<()>

where F: FnMut(LexicalRebuildMessageRow) -> Result<()>,

Stream lexical rebuild message rows from a starting conversation id to the end of the table.

pub fn get_source(&self, id: &str) -> Result<Option<Source>>

Get a source by ID.

pub fn list_sources(&self) -> Result<Vec<Source>>

List all sources.

pub fn get_source_ids(&self) -> Result<Vec<String>>

Get IDs of all non-local sources.

pub fn upsert_source(&self, source: &Source) -> Result<()>

Create or update a source.

pub fn salvage_historical_databases( &self, canonical_db_path: &Path, ) -> Result<HistoricalSalvageOutcome>

pub fn delete_source(&self, id: &str, _cascade: bool) -> Result<bool>

Delete a source by ID. Returns true if a row was deleted.

pub fn insert_conversation_tree( &self, agent_id: i64, workspace_id: Option<i64>, conv: &Conversation, ) -> Result<InsertOutcome>

Insert a conversation tree (conversation + messages + snippets + FTS).

pub fn rebuild_fts(&self) -> Result<()>

Rebuild the FTS5 index from scratch (chunked to avoid OOM on large databases, #110).

pub fn fetch_messages_for_embedding(&self) -> Result<Vec<MessageForEmbedding>>

Fetch all messages for embedding generation.

pub fn get_last_embedded_message_id(&self) -> Result<Option<i64>>

Get the watermark for incremental semantic embedding.

pub fn set_last_embedded_message_id(&self, id: i64) -> Result<()>

Set the watermark for incremental semantic embedding.

pub fn get_embedding_jobs(&self, db_path: &str) -> Result<Vec<EmbeddingJobRow>>

Get embedding jobs for a database path.

pub fn upsert_embedding_job( &self, db_path: &str, model_id: &str, total_docs: i64, ) -> Result<i64>

Create or update an embedding job.

pub fn start_embedding_job(&self, job_id: i64) -> Result<()>

Mark an embedding job as started.

pub fn complete_embedding_job(&self, job_id: i64) -> Result<()>

Mark an embedding job as completed.

pub fn fail_embedding_job(&self, job_id: i64, error: &str) -> Result<()>

Mark an embedding job as failed.

pub fn cancel_embedding_jobs( &self, db_path: &str, model_id: Option<&str>, ) -> Result<usize>

Cancel embedding jobs for a database path.

pub fn update_job_progress( &self, job_id: i64, completed_docs: i64, ) -> Result<()>

Update embedding job progress.

pub fn count_sessions_in_range( &self, start_ts_ms: Option<i64>, end_ts_ms: Option<i64>, agent_slug: Option<&str>, source_id: Option<&str>, ) -> Result<(i64, bool)>

Get session count for a date range using materialized stats. Returns (count, is_from_cache) where is_from_cache is true if from daily_stats.

Falls back to COUNT(*) query when daily_stats table is empty or stale.

pub fn get_daily_histogram( &self, start_ts_ms: i64, end_ts_ms: i64, agent_slug: Option<&str>, source_id: Option<&str>, ) -> Result<Vec<DailyCount>>

Get daily histogram data for a date range.

pub fn daily_stats_health(&self) -> Result<DailyStatsHealth>

Check health of daily stats table.

pub fn insert_conversations_batched( &self, conversations: &[(i64, Option<i64>, &Conversation)], ) -> Result<Vec<InsertOutcome>>

Batch insert multiple conversations with full analytics (token usage, message metrics, rollups). Frankensqlite equivalent of SqliteStorage::insert_conversations_batched.

impl FrankenStorage

pub fn rebuild_token_daily_stats(&self) -> Result<usize>

Rebuild token_daily_stats from the token_usage ledger.

pub fn rebuild_analytics(&self) -> Result<AnalyticsRebuildResult>

Rebuild analytics tables (message_metrics + rollups) from existing messages in the database. Does NOT re-parse raw agent session files.

pub fn rebuild_daily_stats(&self) -> Result<DailyStatsRebuildResult>

Rebuild all daily stats from scratch.

Trait Implementations

impl IndexingCacheStorage for FrankenStorage

fn ensure_indexing_agent(&self, agent: &Agent) -> Result<i64>

fn ensure_indexing_workspace( &self, path: &Path, display_name: Option<&str>, ) -> Result<i64>