orch/store/
mod.rs

1//! Unified task store — SQLite as single source of truth.
2//!
3//! `TaskStore` replaces the legacy `internal_tasks` table and GitHub labels as the
4//! authoritative task state. External backends (GitHub, Linear) become sync adapters
5//! that mirror state changes outward.
6//!
7//! Uses sqlx for async SQLite access with file-based migrations.
8//!
9//! All task state, metrics, KV, and rate limits flow through this module.
10//!
11//! # Safety Warning: Never Use `SELECT *`
12//!
13//! Using `SELECT *` with `sqlx-sqlite`'s prepared statement cache can cause **OOB
14//! (out-of-bounds) panics** when the schema changes (e.g., new migration adds a column).
15//!
16//! The issue occurs because:
17//! 1. SQLite caches prepared statements with column metadata at prepare time
18//! 2. When a migration adds a new column, `sqlite3_column_count()` returns the updated count
19//! 3. The cached statement's column metadata vector was built with the old count
20//! 4. Accessing column index N when only N columns exist (0..N-1) causes a panic
21//!
22//! **Solution:** Always use explicit column lists via constants like `TASK_COLS`,
23//! `CONTROL_MESSAGE_COLS`, `JOB_STATE_COLS`, etc. See `tasks.rs` and `control.rs`
24//! for examples. CI will fail if `SELECT *` is detected in store files.
25
26use anyhow::Context;
27use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqlitePoolOptions};
28use std::path::Path;
29
30mod control;
31mod helpers;
32mod jobs;
33mod kv;
34mod metrics;
35mod pricing;
36mod tasks;
37
38#[allow(unused_imports)]
39pub use control::{parse_since_duration, ChatCostSummary, ControlMessage, MemoryEntry};
40#[allow(unused_imports)]
41pub use helpers::{
42    get_cost_estimate, get_cost_estimate_result, get_recent_memory, get_recent_memory_result,
43    get_task_field, get_task_field_direct, get_token_usage, get_token_usage_result,
44    opt_store_get_task, opt_store_get_task_by_id, resolve_store_id, review_session_expected,
45    set_review_session_expected, store_increment, store_increment_by_id, store_log_activity,
46    store_reset_counters, store_reset_failure_counters, store_set, store_set_by_id,
47    store_set_result, store_set_result_by_id, store_touch_updated_at, store_touch_updated_at_by_id,
48    StoreResult,
49};
50#[allow(unused_imports)]
51pub use jobs::JobState;
52#[allow(unused_imports)]
53pub use metrics::{
54    AgentStat, CostByGroup, CostPeriod, CostSummary, ErrorStat, HighReviewCycleTask,
55    InsertTaskMetric, MetricsSummary, SlowTaskInfo,
56};
57#[allow(unused_imports)]
58pub use pricing::{pricing_for_model, CostEstimate, ModelPricing, TokenUsage};
59#[allow(unused_imports)]
60pub use tasks::{
61    CompleteRun, NewTask, RunTokenUsage, StartRun, StoreRoute, Task, TaskActivity, TaskRun,
62    TaskStatus, UpsertExternal,
63};
64
65/// Default database path: `~/.orch/orch.db`
66pub async fn default_db_path() -> anyhow::Result<std::path::PathBuf> {
67    crate::home::db_path().await
68}
69
70/// The unified task store backed by SQLite (via sqlx).
71#[derive(Clone)]
72pub struct TaskStore {
73    pool: SqlitePool,
74}
75
76impl TaskStore {
77    /// Open (or create) the store at the given database path.
78    ///
79    /// Runs file-based migrations from the `migrations/` directory.
80    pub async fn open(db_path: &Path) -> anyhow::Result<Self> {
81        let options = SqliteConnectOptions::new()
82            .filename(db_path)
83            .journal_mode(SqliteJournalMode::Wal)
84            .create_if_missing(true)
85            .busy_timeout(std::time::Duration::from_secs(5))
86            .pragma("foreign_keys", "ON");
87
88        let pool = SqlitePoolOptions::new()
89            .max_connections(5)
90            .connect_with(options)
91            .await
92            .with_context(|| format!("opening task store: {}", db_path.display()))?;
93
94        let store = Self { pool };
95        store.migrate().await?;
96        Ok(store)
97    }
98
99    /// Open a store with a single connection.
100    #[allow(dead_code)]
101    ///
102    /// Uses `max_connections(1)` to avoid WAL visibility issues where a write
103    /// on one pooled connection isn't visible to a read on another. Used by
104    /// integration tests that need read-your-own-writes consistency.
105    pub async fn open_single(db_path: &Path) -> anyhow::Result<Self> {
106        let options = SqliteConnectOptions::new()
107            .filename(db_path)
108            .journal_mode(SqliteJournalMode::Wal)
109            .create_if_missing(true)
110            .busy_timeout(std::time::Duration::from_secs(5))
111            .pragma("foreign_keys", "ON");
112
113        let pool = SqlitePoolOptions::new()
114            .max_connections(1)
115            .connect_with(options)
116            .await
117            .with_context(|| format!("opening task store (single): {}", db_path.display()))?;
118
119        let store = Self { pool };
120        store.migrate().await?;
121        Ok(store)
122    }
123
124    /// Open an in-memory store (for testing).
125    #[cfg(test)]
126    pub async fn open_memory() -> anyhow::Result<Self> {
127        let options = SqliteConnectOptions::new()
128            .filename(":memory:")
129            .pragma("foreign_keys", "ON");
130        let pool = SqlitePoolOptions::new()
131            .max_connections(1)
132            .connect_with(options)
133            .await?;
134
135        let store = Self { pool };
136        store.migrate().await?;
137        Ok(store)
138    }
139
140    /// Run migrations.
141    async fn migrate(&self) -> anyhow::Result<()> {
142        sqlx::migrate!("./migrations")
143            .run(&self.pool)
144            .await
145            .context("running task store migrations")?;
146        Ok(())
147    }
148
149    /// Get a reference to the underlying pool (for advanced queries).
150    #[allow(dead_code)]
151    pub fn pool(&self) -> &SqlitePool {
152        &self.pool
153    }
154}
155
156#[cfg(test)]
157mod tests;
orch/store/mod.rs

orch/store/
mod.rs
