twine_core/
store.rs

1use std::path::{Path, PathBuf};
2use std::time::Duration;
3
4use rusqlite::{Connection, TransactionBehavior, params};
5use thiserror::Error;
6use tracing::info;
7
8mod activities;
9mod harness_history;
10mod harness_sessions;
11mod harness_steps;
12mod recovery;
13mod run_traces;
14mod trace_payloads;
15mod trace_storage;
16mod traces;
17mod workflow_types;
18mod workflows;
19pub(crate) use traces::{NewTraceSpan, TraceEnding};
20
21pub(crate) use harness_steps::HarnessStepContext;
22pub(crate) use workflow_types::StoredWorkflowType;
23
24/// Schema migrations in the order they apply. `PRAGMA user_version` counts the ones already applied,
25/// so append new migrations and never change or reorder one that has shipped.
26const MIGRATIONS: &[&str] = &[
27    // 1: Recent folders, and the folder that was open when Twine last quit.
28    "CREATE TABLE recent_folders (
29        path TEXT PRIMARY KEY NOT NULL,
30        last_opened_at INTEGER NOT NULL,
31        is_open INTEGER NOT NULL DEFAULT 0 CHECK (is_open IN (0, 1))
32    ) STRICT",
33    // 2: Sessions and workflow tabs outlive terminal processes and the recent-folder list.
34    "CREATE TABLE sessions (
35        id INTEGER PRIMARY KEY AUTOINCREMENT,
36        folder TEXT NOT NULL,
37        name TEXT NOT NULL,
38        started_at INTEGER NOT NULL
39    ) STRICT;
40    CREATE INDEX sessions_folder ON sessions(folder);
41    CREATE TABLE workflows (
42        id INTEGER PRIMARY KEY AUTOINCREMENT,
43        session_id INTEGER NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
44        name TEXT NOT NULL,
45        kind TEXT NOT NULL CHECK (kind IN ('draft', 'terminal'))
46    ) STRICT;
47    CREATE INDEX workflows_session ON workflows(session_id);
48    CREATE TABLE folder_selection (
49        folder TEXT PRIMARY KEY NOT NULL,
50        session_id INTEGER REFERENCES sessions(id) ON DELETE SET NULL
51    ) STRICT",
52    // 3: User-made workflow types. Versions are only ever added, so running workflows keep theirs.
53    "CREATE TABLE workflow_types (
54        id INTEGER PRIMARY KEY AUTOINCREMENT
55    ) STRICT;
56    CREATE TABLE workflow_type_versions (
57        type_id INTEGER NOT NULL REFERENCES workflow_types(id) ON DELETE CASCADE,
58        version INTEGER NOT NULL CHECK (version > 0),
59        definition TEXT NOT NULL,
60        created_at INTEGER NOT NULL,
61        PRIMARY KEY (type_id, version)
62    ) STRICT",
63    // 4: Single-agent workflows record their harness and how the agent last ended.
64    "CREATE TABLE workflows_new (
65        id INTEGER PRIMARY KEY AUTOINCREMENT,
66        session_id INTEGER NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
67        name TEXT NOT NULL,
68        kind TEXT NOT NULL CHECK (kind IN ('draft', 'terminal', 'single_agent')),
69        harness TEXT CHECK (harness IN ('codex', 'claude_code', 'pi')),
70        agent_status TEXT CHECK (agent_status IN ('running', 'exited', 'failed', 'cancelled'))
71    ) STRICT;
72    INSERT INTO workflows_new (id, session_id, name, kind)
73        SELECT id, session_id, name, kind FROM workflows;
74    UPDATE sqlite_sequence SET seq = MAX(seq, COALESCE(
75        (SELECT seq FROM sqlite_sequence WHERE name = 'workflows'), 0))
76        WHERE name = 'workflows_new';
77    DROP TABLE workflows;
78    ALTER TABLE workflows_new RENAME TO workflows;
79    CREATE INDEX workflows_session ON workflows(session_id)",
80    // 5: Agents workflows and their agents' roles. SQLite can't change a CHECK constraint, so the
81    // workflows table is rebuilt, keeping its IDs and its AUTOINCREMENT high-water mark so closed
82    // workflows' IDs are never reused.
83    "CREATE TABLE workflows_v5 (
84        id INTEGER PRIMARY KEY AUTOINCREMENT,
85        session_id INTEGER NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
86        name TEXT NOT NULL,
87        kind TEXT NOT NULL CHECK (kind IN ('draft', 'terminal', 'single_agent', 'agents')),
88        harness TEXT CHECK (harness IN ('codex', 'claude_code', 'pi')),
89        agent_status TEXT CHECK (agent_status IN ('running', 'exited', 'failed', 'cancelled'))
90    ) STRICT;
91    INSERT INTO workflows_v5 (id, session_id, name, kind, harness, agent_status)
92        SELECT id, session_id, name, kind, harness, agent_status FROM workflows;
93    DELETE FROM sqlite_sequence WHERE name = 'workflows_v5';
94    INSERT INTO sqlite_sequence (name, seq)
95        SELECT 'workflows_v5', seq FROM sqlite_sequence WHERE name = 'workflows';
96    DROP TABLE workflows;
97    ALTER TABLE workflows_v5 RENAME TO workflows;
98    CREATE INDEX workflows_session ON workflows(session_id);
99    CREATE TABLE agents (
100        id INTEGER PRIMARY KEY AUTOINCREMENT,
101        workflow_id INTEGER NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
102        role TEXT NOT NULL
103    ) STRICT;
104    CREATE INDEX agents_workflow ON agents(workflow_id)",
105    // 6: Trace history, including closed tabs.
106    "ALTER TABLE workflows ADD COLUMN closed_at INTEGER;
107    CREATE TABLE trace_lanes (
108        id INTEGER PRIMARY KEY AUTOINCREMENT,
109        workflow_id INTEGER NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
110        lane_key TEXT NOT NULL,
111        name TEXT NOT NULL,
112        is_agent INTEGER NOT NULL CHECK (is_agent IN (0, 1)),
113        role TEXT,
114        harness TEXT,
115        UNIQUE (workflow_id, lane_key)
116    ) STRICT;
117    CREATE TABLE trace_spans (
118        id INTEGER PRIMARY KEY AUTOINCREMENT,
119        lane_id INTEGER NOT NULL REFERENCES trace_lanes(id) ON DELETE CASCADE,
120        title TEXT NOT NULL,
121        started_at INTEGER NOT NULL CHECK (started_at >= 0),
122        ended_at INTEGER CHECK (ended_at >= started_at),
123        status TEXT NOT NULL CHECK (status IN ('running', 'exited', 'failed', 'stopped')),
124        terminal_id INTEGER
125    ) STRICT;
126    CREATE INDEX trace_spans_lane ON trace_spans(lane_id, id);
127    CREATE INDEX trace_spans_terminal ON trace_spans(terminal_id);
128    CREATE TABLE trace_events (
129        id INTEGER PRIMARY KEY AUTOINCREMENT,
130        workflow_id INTEGER NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
131        span_id INTEGER REFERENCES trace_spans(id) ON DELETE CASCADE,
132        timestamp INTEGER NOT NULL CHECK (timestamp >= 0),
133        kind TEXT NOT NULL CHECK (kind IN ('processStarted', 'processExited', 'processFailed', 'processStopped')),
134        message TEXT NOT NULL,
135        terminal_id INTEGER,
136        byte_offset INTEGER,
137        CHECK ((terminal_id IS NULL) = (byte_offset IS NULL)),
138        CHECK (terminal_id IS NULL OR terminal_id > 0),
139        CHECK (byte_offset IS NULL OR byte_offset >= 0)
140    ) STRICT;
141    CREATE INDEX trace_events_workflow ON trace_events(workflow_id, id);
142    CREATE INDEX trace_events_span ON trace_events(span_id, id)",
143    // 7: Pinned workflow executions and durable events for each stage invocation.
144    "CREATE TABLE workflow_runs (
145        workflow_id INTEGER PRIMARY KEY REFERENCES workflows(id) ON DELETE CASCADE,
146        state TEXT NOT NULL,
147        trace_sequence INTEGER NOT NULL DEFAULT 0
148    ) STRICT;
149    ALTER TABLE trace_spans ADD COLUMN run_generation INTEGER;
150    CREATE TABLE trace_events_v7 (
151        id INTEGER PRIMARY KEY AUTOINCREMENT,
152        workflow_id INTEGER NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
153        span_id INTEGER REFERENCES trace_spans(id) ON DELETE CASCADE,
154        timestamp INTEGER NOT NULL CHECK (timestamp >= 0),
155        kind TEXT NOT NULL CHECK (kind IN ('processStarted', 'processExited', 'processFailed', 'processStopped', 'workflowEvent')),
156        message TEXT NOT NULL,
157        terminal_id INTEGER,
158        byte_offset INTEGER,
159        CHECK ((terminal_id IS NULL) = (byte_offset IS NULL)),
160        CHECK (terminal_id IS NULL OR terminal_id > 0),
161        CHECK (byte_offset IS NULL OR byte_offset >= 0)
162    ) STRICT;
163    INSERT INTO trace_events_v7 SELECT * FROM trace_events;
164    DELETE FROM sqlite_sequence WHERE name = 'trace_events_v7';
165    INSERT INTO sqlite_sequence (name, seq)
166        SELECT 'trace_events_v7', seq FROM sqlite_sequence WHERE name = 'trace_events';
167    DROP TABLE trace_events;
168    ALTER TABLE trace_events_v7 RENAME TO trace_events;
169    CREATE INDEX trace_events_workflow ON trace_events(workflow_id, id);
170    CREATE INDEX trace_events_span ON trace_events(span_id, id)",
171    // 8: Ordered terminal sizes at an anchored trace event's byte boundary.
172    "ALTER TABLE trace_events ADD COLUMN boundary_sizes BLOB CHECK
173        (boundary_sizes IS NULL OR (length(boundary_sizes) <= 2048 AND length(boundary_sizes) % 8 = 0))",
174    // 9: Single-agent lifecycle includes interruption without rebuilding history's parent table.
175    "ALTER TABLE workflows ADD COLUMN lifecycle_status TEXT CHECK
176        (lifecycle_status IN ('running', 'exited', 'failed', 'cancelled', 'interrupted'));
177    UPDATE workflows SET lifecycle_status = agent_status WHERE kind = 'single_agent'",
178    // 10: Assignment spans end on explicit completion, independently of process lifetimes.
179    // Existing spans retain their recorded lifecycle semantics.
180    "ALTER TABLE trace_spans ADD COLUMN work_span INTEGER NOT NULL DEFAULT 0 CHECK (work_span IN (0, 1))",
181    // 11: Correlate asynchronous harness events with the prompt that owns them.
182    "ALTER TABLE trace_spans ADD COLUMN harness_turn_id TEXT;
183     CREATE UNIQUE INDEX trace_harness_turn ON trace_spans(terminal_id, harness_turn_id)
184     WHERE harness_turn_id IS NOT NULL",
185    // 12: Keep terminal ownership even when an idle process has no Activity span.
186    "CREATE TABLE workflow_terminals (
187        terminal_id INTEGER PRIMARY KEY CHECK (terminal_id > 0),
188        workflow_id INTEGER NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
189        agent_id INTEGER
190    ) STRICT;
191    CREATE INDEX workflow_terminals_workflow ON workflow_terminals(workflow_id, terminal_id);
192    INSERT OR IGNORE INTO workflow_terminals (terminal_id, workflow_id, agent_id)
193        SELECT s.terminal_id, l.workflow_id,
194            CASE WHEN l.lane_key LIKE 'agent:%' THEN CAST(SUBSTR(l.lane_key, 7) AS INTEGER) END
195        FROM trace_spans s JOIN trace_lanes l ON l.id = s.lane_id
196        JOIN workflows w ON w.id = l.workflow_id
197        WHERE s.terminal_id > 0 AND (w.kind IN ('draft', 'terminal') OR l.is_agent = 1);
198    INSERT OR IGNORE INTO workflow_terminals (terminal_id, workflow_id)
199        SELECT e.terminal_id, e.workflow_id FROM trace_events e
200        JOIN workflows w ON w.id = e.workflow_id
201        WHERE e.terminal_id > 0 AND w.kind IN ('draft', 'terminal')",
202    // 13: Extend harness choices without rebuilding history's parent table. The legacy harness
203    // column's CHECK constraint remains; new reads and writes use harness_id instead.
204    "ALTER TABLE workflows ADD COLUMN harness_id TEXT CHECK
205        (harness_id IN ('codex', 'claude_code', 'pi', 'antigravity', 'omp'));
206     UPDATE workflows SET harness_id = harness",
207    // 14: Keep the checked column for existing databases and use an extensible harness ID.
208    // Rust validates harness choices; retaining the parent table preserves dependent history.
209    "ALTER TABLE workflows RENAME COLUMN harness_id TO legacy_harness_id;
210     ALTER TABLE workflows ADD COLUMN harness_id TEXT;
211     UPDATE workflows SET harness_id = legacy_harness_id",
212    // 15: Exact harness sessions, tied to terminal ownership rather than a shared folder.
213    "ALTER TABLE workflow_terminals ADD COLUMN harness_session TEXT",
214    // 16: Open windows outlive removal from recents and the recent-folder limit.
215    "CREATE TABLE open_folders (
216        path TEXT PRIMARY KEY NOT NULL,
217        last_opened_at INTEGER NOT NULL
218    ) STRICT;
219    INSERT INTO open_folders SELECT path, last_opened_at FROM recent_folders WHERE is_open = 1",
220    // 17: Native call identities and optional endpoints survive restart and pagination.
221    "CREATE TABLE trace_activities (
222        id INTEGER PRIMARY KEY AUTOINCREMENT,
223        span_id INTEGER NOT NULL REFERENCES trace_spans(id) ON DELETE CASCADE,
224        source_id TEXT NOT NULL,
225        parent_source_id TEXT,
226        kind TEXT NOT NULL CHECK (kind IN ('tool', 'subagent')),
227        title TEXT NOT NULL,
228        started_at INTEGER,
229        ended_at INTEGER,
230        status TEXT NOT NULL CHECK (status IN ('running', 'completed', 'failed', 'interrupted')),
231        input TEXT NOT NULL,
232        output TEXT NOT NULL,
233        anchor TEXT,
234        UNIQUE(span_id, source_id)
235    ) STRICT;
236    CREATE INDEX trace_activities_parent ON trace_activities(span_id, parent_source_id)",
237    // 18: Model calls and observer notes, full detail references, and native metadata.
238    "ALTER TABLE trace_activities RENAME TO trace_activities_v17;
239     DROP INDEX trace_activities_parent;
240     CREATE TABLE trace_activities (
241        id INTEGER PRIMARY KEY AUTOINCREMENT,
242        span_id INTEGER NOT NULL REFERENCES trace_spans(id) ON DELETE CASCADE,
243        source_id TEXT NOT NULL, parent_source_id TEXT,
244        kind TEXT NOT NULL CHECK (kind IN ('tool', 'subagent', 'model', 'note')),
245        title TEXT NOT NULL, started_at INTEGER, ended_at INTEGER,
246        status TEXT NOT NULL CHECK (status IN ('running', 'completed', 'failed', 'interrupted')),
247        input TEXT NOT NULL, output TEXT NOT NULL, anchor TEXT,
248        input_key TEXT, output_key TEXT,
249        input_bytes INTEGER, output_bytes INTEGER,
250        metadata TEXT NOT NULL DEFAULT '{}',
251        UNIQUE(span_id, source_id)
252     ) STRICT;
253     INSERT INTO trace_activities (id,span_id,source_id,parent_source_id,kind,title,started_at,ended_at,status,input,output,anchor)
254        SELECT id,span_id,source_id,parent_source_id,kind,title,started_at,ended_at,status,input,output,anchor FROM trace_activities_v17;
255     INSERT INTO sqlite_sequence(name,seq) SELECT 'trace_activities',seq FROM sqlite_sequence WHERE name='trace_activities_v17'
256        AND NOT EXISTS(SELECT 1 FROM sqlite_sequence WHERE name='trace_activities');
257     UPDATE sqlite_sequence SET seq=MAX(seq,COALESCE((SELECT seq FROM sqlite_sequence WHERE name='trace_activities_v17'),0))
258        WHERE name='trace_activities';
259     DROP TABLE trace_activities_v17;
260     CREATE INDEX trace_activities_parent ON trace_activities(span_id, parent_source_id)",
261    // 19: Independent full-detail retention with explicit pinning and cached storage statistics.
262    "ALTER TABLE trace_spans ADD COLUMN pinned_details INTEGER NOT NULL DEFAULT 0 CHECK(pinned_details IN (0,1));
263     CREATE TABLE trace_storage_preferences (
264       id INTEGER PRIMARY KEY CHECK(id=1),budget_bytes INTEGER NOT NULL,retention_days INTEGER NOT NULL,
265       payload_bytes INTEGER NOT NULL DEFAULT 0,payload_files INTEGER NOT NULL DEFAULT 0,updated_at INTEGER NOT NULL DEFAULT 0,
266       clear_generation INTEGER NOT NULL DEFAULT 0,completed_clear_generation INTEGER NOT NULL DEFAULT 0
267     ) STRICT;
268     INSERT INTO trace_storage_preferences(id,budget_bytes,retention_days) VALUES(1,536870912,90)"
269];
270
271/// How long a write waits for another connection, such as a second Twine process, to release the
272/// database.
273const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
274
275/// A recent folder as stored, without the file system checks the folder module adds.
276#[derive(Debug, Eq, PartialEq)]
277pub(crate) struct StoredFolder {
278    pub(crate) path: PathBuf,
279    /// Whether the folder was open when Twine last quit.
280    pub(crate) is_open: bool,
281}
282
283/// The SQLite database that holds Twine's persistent state.
284#[derive(Debug)]
285pub(crate) struct Store {
286    connection: Connection,
287    payloads: trace_payloads::TracePayloads,
288    pub(crate) workflow_directory: std::path::PathBuf,
289    _workflow_temp: Option<tempfile::TempDir>,
290}
291
292fn workflow_file_directory(config_path: Option<&Path>) -> PathBuf {
293    config_path.and_then(Path::parent).map_or_else(
294        || std::env::temp_dir().join("twine/workflows"),
295        |directory| directory.join("workflows"),
296    )
297}
298
299impl Store {
300    /// Opens the database at `path`, creating the file and its directory if needed, and applies
301    /// pending migrations.
302    pub(crate) fn open(path: &Path) -> Result<Self, StoreError> {
303        if let Some(directory) = path.parent() {
304            std::fs::create_dir_all(directory).map_err(|source| StoreError::CreateDirectory {
305                path: directory.to_owned(),
306                source,
307            })?;
308        }
309        let connection = Connection::open(path).map_err(|source| StoreError::Open {
310            path: path.to_owned(),
311            source,
312        })?;
313        Self::with_connection(connection)
314    }
315
316    #[cfg(test)]
317    pub(crate) fn execute_test_sql(&self, sql: &str) {
318        self.connection
319            .execute_batch(sql)
320            .expect("test SQL should execute");
321    }
322
323    #[cfg(test)]
324    pub(crate) fn open_in_memory() -> Result<Self, StoreError> {
325        Self::with_connection(Connection::open_in_memory()?)
326    }
327
328    fn with_connection(mut connection: Connection) -> Result<Self, StoreError> {
329        connection.busy_timeout(BUSY_TIMEOUT)?;
330        connection.pragma_update(None, "journal_mode", "WAL")?;
331        connection.pragma_update(None, "foreign_keys", true)?;
332        migrate(&mut connection)?;
333        let payloads = trace_payloads::TracePayloads::new(connection.path())?;
334        let temporary = if connection.path().is_none() || connection.path() == Some("") {
335            Some(
336                tempfile::tempdir().map_err(|source| StoreError::CreateDirectory {
337                    path: std::env::temp_dir(),
338                    source,
339                })?,
340            )
341        } else {
342            None
343        };
344        let workflow_directory = if let Some(temporary) = &temporary {
345            temporary.path().join("workflows")
346        } else if cfg!(test) {
347            std::path::Path::new(connection.path().expect("disk store"))
348                .parent()
349                .unwrap_or_else(|| std::path::Path::new("."))
350                .join("workflows")
351        } else {
352            workflow_file_directory(crate::config::Config::user_path().as_deref())
353        };
354        Ok(Self {
355            connection,
356            payloads,
357            workflow_directory,
358            _workflow_temp: temporary,
359        })
360    }
361
362    /// Returns the recent folders, most recently opened first.
363    pub(crate) fn recent_folders(&self) -> Result<Vec<StoredFolder>, StoreError> {
364        query_recent_folders(&self.connection)
365    }
366
367    pub(crate) fn data_version(&self) -> Result<i64, StoreError> {
368        Ok(self
369            .connection
370            .pragma_query_value(None, "data_version", |row| row.get(0))?)
371    }
372
373    pub(crate) fn open_folder_paths(&self) -> Result<Vec<PathBuf>, StoreError> {
374        let mut statement = self
375            .connection
376            .prepare("SELECT path FROM open_folders ORDER BY last_opened_at DESC")?;
377        Ok(statement
378            .query_map([], |row| Ok(PathBuf::from(row.get::<_, String>(0)?)))?
379            .collect::<Result<_, _>>()?)
380    }
381
382    /// Records `path` as the open folder and the most recently opened one, then forgets all but the
383    /// `limit` most recent folders. Returns the recent folders after the change.
384    ///
385    /// The recorded time is at least one millisecond after every stored time, so the new folder
386    /// sorts first even if the clock went backward or two folders open in the same millisecond.
387    pub(crate) fn record_folder_opened(
388        &mut self,
389        path: &str,
390        opened_at_millis: i64,
391        limit: i64,
392    ) -> Result<Vec<StoredFolder>, StoreError> {
393        self.record_window_folder_opened(path, opened_at_millis, limit, None, true)
394    }
395
396    pub(crate) fn record_window_folder_opened(
397        &mut self,
398        path: &str,
399        opened_at_millis: i64,
400        limit: i64,
401        previous: Option<&Path>,
402        exclusive: bool,
403    ) -> Result<Vec<StoredFolder>, StoreError> {
404        let transaction = self.connection.transaction()?;
405        if exclusive {
406            transaction.execute("DELETE FROM open_folders", [])?;
407            transaction.execute(
408                "UPDATE recent_folders SET is_open = 0 WHERE is_open = 1",
409                [],
410            )?;
411        } else if let Some(previous) = previous {
412            transaction.execute(
413                "DELETE FROM open_folders WHERE path = ?1",
414                [previous.to_string_lossy()],
415            )?;
416            transaction.execute(
417                "UPDATE recent_folders SET is_open = 0 WHERE path = ?1",
418                [previous.to_string_lossy()],
419            )?;
420        }
421        transaction.execute(
422            "INSERT INTO recent_folders (path, last_opened_at, is_open)
423             VALUES (?1, MAX(?2, (SELECT COALESCE(MAX(last_opened_at), 0) + 1 FROM
424                (SELECT last_opened_at FROM recent_folders UNION ALL SELECT last_opened_at FROM open_folders))), 1)
425             ON CONFLICT (path) DO UPDATE
426             SET last_opened_at = excluded.last_opened_at, is_open = 1",
427            params![path, opened_at_millis],
428        )?;
429        transaction.execute(
430            "INSERT INTO open_folders (path, last_opened_at)
431            SELECT path, last_opened_at FROM recent_folders WHERE path = ?1
432            ON CONFLICT(path) DO UPDATE SET last_opened_at = excluded.last_opened_at",
433            [path],
434        )?;
435        transaction.execute(
436            "DELETE FROM recent_folders WHERE path NOT IN
437             (SELECT path FROM recent_folders ORDER BY last_opened_at DESC LIMIT ?1)",
438            [limit],
439        )?;
440        let folders = query_recent_folders(&transaction)?;
441        transaction.commit()?;
442        Ok(folders)
443    }
444
445    /// Records that no folder is open, so the next launch shows the start page. Returns the recent
446    /// folders after the change.
447    pub(crate) fn record_folder_closed(&mut self) -> Result<Vec<StoredFolder>, StoreError> {
448        let transaction = self.connection.transaction()?;
449        transaction.execute("DELETE FROM open_folders", [])?;
450        transaction.execute(
451            "UPDATE recent_folders SET is_open = 0 WHERE is_open = 1",
452            [],
453        )?;
454        let folders = query_recent_folders(&transaction)?;
455        transaction.commit()?;
456        Ok(folders)
457    }
458
459    pub(crate) fn record_window_folder_closed(
460        &mut self,
461        path: Option<&Path>,
462    ) -> Result<Vec<StoredFolder>, StoreError> {
463        let transaction = self.connection.transaction()?;
464        if let Some(path) = path {
465            transaction.execute(
466                "DELETE FROM open_folders WHERE path = ?1",
467                [path.to_string_lossy()],
468            )?;
469            transaction.execute(
470                "UPDATE recent_folders SET is_open = 0 WHERE path = ?1",
471                [path.to_string_lossy()],
472            )?;
473        }
474        let folders = query_recent_folders(&transaction)?;
475        transaction.commit()?;
476        Ok(folders)
477    }
478
479    /// Forgets the recent folder at `path`. Returns the recent folders after the change.
480    pub(crate) fn remove_recent_folder(
481        &mut self,
482        path: &str,
483    ) -> Result<Vec<StoredFolder>, StoreError> {
484        self.remove_recent_folder_preserving_windows(path, false)
485    }
486
487    pub(crate) fn remove_recent_folder_preserving_windows(
488        &mut self,
489        path: &str,
490        preserve_open: bool,
491    ) -> Result<Vec<StoredFolder>, StoreError> {
492        let transaction = self.connection.transaction()?;
493        if !preserve_open {
494            transaction.execute("DELETE FROM open_folders WHERE path = ?1", [path])?;
495        }
496        transaction.execute("DELETE FROM recent_folders WHERE path = ?1", [path])?;
497        let folders = query_recent_folders(&transaction)?;
498        transaction.commit()?;
499        Ok(folders)
500    }
501}
502
503fn query_recent_folders(connection: &Connection) -> Result<Vec<StoredFolder>, StoreError> {
504    let mut statement = connection
505        .prepare("SELECT path, is_open FROM recent_folders ORDER BY last_opened_at DESC")?;
506    let folders = statement
507        .query_map([], |row| {
508            Ok(StoredFolder {
509                path: PathBuf::from(row.get::<_, String>(0)?),
510                is_open: row.get(1)?,
511            })
512        })?
513        .collect::<Result<_, _>>()?;
514    Ok(folders)
515}
516
517/// Applies pending migrations in one transaction that takes the write lock before reading the
518/// schema version, so a second Twine process opening the database waits instead of migrating it
519/// again.
520fn migrate(connection: &mut Connection) -> Result<(), StoreError> {
521    let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
522    let version: i64 = transaction.pragma_query_value(None, "user_version", |row| row.get(0))?;
523    let applied = usize::try_from(version)
524        .ok()
525        .filter(|applied| *applied <= MIGRATIONS.len())
526        .ok_or(StoreError::UnsupportedSchemaVersion { version })?;
527
528    for (version, migration) in (1_i64..).zip(MIGRATIONS).skip(applied) {
529        transaction
530            .execute_batch(migration)
531            .map_err(|source| StoreError::Migrate { version, source })?;
532        transaction.pragma_update(None, "user_version", version)?;
533        info!(version, "applied database migration");
534    }
535    transaction.commit()?;
536    Ok(())
537}
538
539fn sql_integer(value: u64) -> Result<i64, StoreError> {
540    i64::try_from(value).map_err(|_| StoreError::InvalidIdentifier)
541}
542
543fn unsigned_column(row: &rusqlite::Row<'_>, index: usize) -> rusqlite::Result<u64> {
544    let value: i64 = row.get(index)?;
545    u64::try_from(value).map_err(|_| rusqlite::Error::IntegralValueOutOfRange(index, value))
546}
547
548#[derive(Debug, Error)]
549pub enum StoreError {
550    #[error("couldn't read or store full trace details")]
551    TracePayload(#[source] std::io::Error),
552    #[error("failed to resolve data directory {}", path.display())]
553    ResolveDirectory {
554        path: PathBuf,
555        source: std::io::Error,
556    },
557    #[error("failed to encode workflow run state")]
558    RunEncoding(#[from] serde_json::Error),
559    #[error("database returned an invalid identifier")]
560    InvalidIdentifier,
561    #[error("failed to create the data directory {}", path.display())]
562    CreateDirectory {
563        path: PathBuf,
564        #[source]
565        source: std::io::Error,
566    },
567    #[error("failed to apply database migration {version}")]
568    Migrate {
569        version: i64,
570        #[source]
571        source: rusqlite::Error,
572    },
573    #[error("failed to open the database {}", path.display())]
574    Open {
575        path: PathBuf,
576        #[source]
577        source: rusqlite::Error,
578    },
579    #[error(transparent)]
580    Sqlite(#[from] rusqlite::Error),
581    #[error("database schema version {version} is not one this version of Twine supports")]
582    UnsupportedSchemaVersion { version: i64 },
583}
584
585#[cfg(test)]
586mod tests {
587    use super::*;
588    use crate::workflow::SessionId;
589
590    fn schema_version(connection: &Connection) -> i64 {
591        connection
592            .pragma_query_value(None, "user_version", |row| row.get(0))
593            .expect("the schema version should be readable")
594    }
595
596    #[test]
597    fn workflows_follow_the_selected_config_directory() {
598        for directory in [
599            "/checkout/out/runtime/debug/config",
600            "/home/user/.config/twine",
601            "/custom/twine",
602        ] {
603            assert_eq!(
604                workflow_file_directory(Some(&Path::new(directory).join("config.toml"))),
605                Path::new(directory).join("workflows")
606            );
607        }
608    }
609
610    #[test]
611    fn open_creates_the_directory_and_migrates_once() {
612        let directory = tempfile::tempdir().expect("a temporary directory should be available");
613        let path = directory.path().join("nested").join("twine.db");
614
615        let store = Store::open(&path).expect("the store should open");
616        assert_eq!(
617            schema_version(&store.connection),
618            i64::try_from(MIGRATIONS.len()).expect("the migration count fits in i64")
619        );
620        drop(store);
621
622        let reopened = Store::open(&path).expect("the migrated store should reopen");
623        assert_eq!(
624            schema_version(&reopened.connection),
625            i64::try_from(MIGRATIONS.len()).expect("the migration count fits in i64")
626        );
627    }
628
629    fn legacy_workflow_history() -> Connection {
630        let connection = Connection::open_in_memory().unwrap();
631        connection
632            .pragma_update(None, "foreign_keys", true)
633            .unwrap();
634        for migration in &MIGRATIONS[..11] {
635            connection.execute_batch(migration).unwrap();
636        }
637        connection.pragma_update(None, "user_version", 11).unwrap();
638        connection.execute_batch(
639            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Session', 123);
640             INSERT INTO workflows (id, session_id, name, kind, harness, lifecycle_status, closed_at)
641                 VALUES (1, 1, 'Claude Code', 'single_agent', 'claude_code', 'cancelled', 456),
642                        (2, 1, 'Agents', 'agents', NULL, NULL, NULL),
643                        (99, 1, 'Deleted', 'draft', NULL, NULL, NULL);
644             DELETE FROM workflows WHERE id = 99;
645             INSERT INTO agents (id, workflow_id, role) VALUES (7, 2, 'Worker');
646             INSERT INTO workflow_runs (workflow_id, state, trace_sequence) VALUES (2, 'kept run state', 3);
647             INSERT INTO trace_lanes (id, workflow_id, lane_key, name, is_agent, harness)
648                 VALUES (8, 2, 'worker', 'Worker', 1, 'pi');
649             INSERT INTO trace_spans (id, lane_id, title, started_at, status, harness_turn_id)
650                 VALUES (9, 8, 'Task', 123, 'running', 'turn-1');
651             INSERT INTO trace_events (id, workflow_id, span_id, timestamp, kind, message)
652                 VALUES (10, 2, 9, 123, 'processStarted', 'Started');"
653        ).unwrap();
654        connection
655    }
656
657    #[test]
658    fn new_harness_migration_preserves_workflows_and_dependent_history() {
659        let connection = legacy_workflow_history();
660        let mut store = Store::with_connection(connection).unwrap();
661        let preserved: (String, String, i64) = store
662            .connection
663            .query_row(
664                "SELECT harness_id, lifecycle_status, closed_at FROM workflows WHERE id = 1",
665                [],
666                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
667            )
668            .unwrap();
669        assert_eq!(preserved, ("claude_code".into(), "cancelled".into(), 456));
670        let run: (String, i64) = store
671            .connection
672            .query_row(
673                "SELECT state, trace_sequence FROM workflow_runs WHERE workflow_id = 2",
674                [],
675                |row| Ok((row.get(0)?, row.get(1)?)),
676            )
677            .unwrap();
678        assert_eq!(run, ("kept run state".into(), 3));
679        for (table, id) in [
680            ("agents", 7),
681            ("trace_lanes", 8),
682            ("trace_spans", 9),
683            ("trace_events", 10),
684        ] {
685            let stored: i64 = store
686                .connection
687                .query_row(&format!("SELECT id FROM {table}"), [], |row| row.get(0))
688                .unwrap();
689            assert_eq!(stored, id);
690        }
691        assert!(
692            store
693                .connection
694                .prepare("PRAGMA foreign_key_check")
695                .unwrap()
696                .query([])
697                .unwrap()
698                .next()
699                .unwrap()
700                .is_none()
701        );
702        for harness in [
703            crate::HarnessId::Antigravity,
704            crate::HarnessId::Omp,
705            crate::HarnessId::Opencode,
706        ] {
707            store
708                .update_workflow(
709                    crate::WorkflowId(1),
710                    harness.definition().name,
711                    crate::WorkflowKind::SingleAgent,
712                    Some(harness),
713                )
714                .unwrap();
715            let stored: String = store
716                .connection
717                .query_row("SELECT harness_id FROM workflows WHERE id = 1", [], |row| {
718                    row.get(0)
719                })
720                .unwrap();
721            assert_eq!(stored, super::workflows::harness_name(harness));
722        }
723        let (id, _) = store
724            .create_workflow(SessionId(1), "New", crate::WorkflowKind::Draft, &[])
725            .unwrap();
726        assert!(id.0 > 99, "Deleted workflow IDs must not be reused");
727        store
728            .connection
729            .execute("DELETE FROM workflows WHERE id = 2", [])
730            .unwrap();
731        for table in [
732            "agents",
733            "workflow_runs",
734            "trace_lanes",
735            "trace_spans",
736            "trace_events",
737        ] {
738            let remaining: i64 = store
739                .connection
740                .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| {
741                    row.get(0)
742                })
743                .unwrap();
744            assert_eq!(
745                remaining, 0,
746                "Deleting a workflow still cascades to {table}"
747            );
748        }
749    }
750
751    #[test]
752    fn harness_migrations_preserve_version_twelve_terminal_ownership() {
753        let connection = legacy_workflow_history();
754        connection.execute_batch(MIGRATIONS[11]).unwrap();
755        connection.pragma_update(None, "user_version", 12).unwrap();
756        connection
757            .execute_batch(
758                "INSERT INTO workflow_terminals (terminal_id, workflow_id, agent_id)
759                 VALUES (42, 1, NULL), (43, 2, 7)",
760            )
761            .unwrap();
762
763        let store = Store::with_connection(connection).unwrap();
764        let harness: String = store
765            .connection
766            .query_row("SELECT harness_id FROM workflows WHERE id = 1", [], |row| {
767                row.get(0)
768            })
769            .unwrap();
770        assert_eq!(harness, "claude_code");
771        for (workflow, terminal, agent) in [(1, 42, None), (2, 43, Some(crate::AgentId(7)))] {
772            let history = store.terminal_history(crate::WorkflowId(workflow)).unwrap();
773            assert_eq!(history.len(), 1);
774            assert_eq!(history[0].terminal_id.value(), terminal);
775            assert_eq!(history[0].agent_id, agent);
776        }
777        store
778            .connection
779            .execute("DELETE FROM workflows WHERE id = 2", [])
780            .unwrap();
781        assert_eq!(store.terminal_history(crate::WorkflowId(2)).unwrap(), []);
782        assert_eq!(
783            store.terminal_history(crate::WorkflowId(1)).unwrap().len(),
784            1
785        );
786        assert!(
787            store
788                .connection
789                .prepare("PRAGMA foreign_key_check")
790                .unwrap()
791                .query([])
792                .unwrap()
793                .next()
794                .unwrap()
795                .is_none()
796        );
797    }
798
799    #[test]
800    fn extensible_harness_migration_preserves_version_thirteen_choices_and_history() {
801        let connection = legacy_workflow_history();
802        for migration in &MIGRATIONS[11..13] {
803            connection.execute_batch(migration).unwrap();
804        }
805        connection.pragma_update(None, "user_version", 13).unwrap();
806        connection
807            .execute("UPDATE workflows SET harness_id = 'omp' WHERE id = 1", [])
808            .unwrap();
809        let mut store = Store::with_connection(connection).unwrap();
810        let preserved: (String, String, i64) = store
811            .connection
812            .query_row(
813                "SELECT harness_id, lifecycle_status, closed_at FROM workflows WHERE id = 1",
814                [],
815                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
816            )
817            .unwrap();
818        assert_eq!(preserved, ("omp".into(), "cancelled".into(), 456));
819        store
820            .update_workflow(
821                crate::WorkflowId(1),
822                "OpenCode",
823                crate::WorkflowKind::SingleAgent,
824                Some(crate::HarnessId::Opencode),
825            )
826            .unwrap();
827        let harness: String = store
828            .connection
829            .query_row("SELECT harness_id FROM workflows WHERE id = 1", [], |row| {
830                row.get(0)
831            })
832            .unwrap();
833        assert_eq!(harness, "opencode");
834        for (table, id) in [
835            ("agents", 7),
836            ("trace_lanes", 8),
837            ("trace_spans", 9),
838            ("trace_events", 10),
839        ] {
840            let stored: i64 = store
841                .connection
842                .query_row(&format!("SELECT id FROM {table}"), [], |row| row.get(0))
843                .unwrap();
844            assert_eq!(stored, id);
845        }
846        let (id, _) = store
847            .create_workflow(SessionId(1), "New", crate::WorkflowKind::Draft, &[])
848            .unwrap();
849        assert!(id.0 > 99);
850        assert!(
851            store
852                .connection
853                .prepare("PRAGMA foreign_key_check")
854                .unwrap()
855                .query([])
856                .unwrap()
857                .next()
858                .unwrap()
859                .is_none()
860        );
861    }
862
863    #[test]
864    fn migration_from_version_one_preserves_folders_and_adds_session_storage() {
865        let mut connection = Connection::open_in_memory().unwrap();
866        connection.execute_batch(MIGRATIONS[0]).unwrap();
867        connection
868            .execute("INSERT INTO recent_folders VALUES ('/folder', 1, 1)", [])
869            .unwrap();
870        connection.pragma_update(None, "user_version", 1).unwrap();
871        migrate(&mut connection).unwrap();
872        let mut store = Store::with_connection(connection).unwrap();
873        assert_eq!(
874            store.recent_folders().unwrap(),
875            [StoredFolder {
876                path: "/folder".into(),
877                is_open: true
878            }]
879        );
880        let id = store
881            .create_session(Path::new("/folder"), "Session", 123)
882            .unwrap();
883        store
884            .create_workflow(id, "Terminal", crate::WorkflowKind::Terminal, &[])
885            .unwrap();
886        assert_eq!(
887            store.selected_session(Path::new("/folder")).unwrap(),
888            Some(id)
889        );
890        assert_eq!(store.workflows(Path::new("/folder")).unwrap().len(), 1);
891    }
892
893    #[test]
894    fn migration_from_version_two_preserves_sessions_and_adds_workflow_types() {
895        let mut connection = Connection::open_in_memory().unwrap();
896        connection.execute_batch(MIGRATIONS[0]).unwrap();
897        connection.execute_batch(MIGRATIONS[1]).unwrap();
898        connection
899            .execute(
900                "INSERT INTO sessions (folder, name, started_at) VALUES ('/folder', 'Kept', 1)",
901                [],
902            )
903            .unwrap();
904        connection.pragma_update(None, "user_version", 2).unwrap();
905        migrate(&mut connection).unwrap();
906        let mut store = Store::with_connection(connection).unwrap();
907        assert_eq!(
908            store.sessions(Path::new("/folder")).unwrap()[0].name,
909            "Kept"
910        );
911        let type_id = store.create_workflow_type("definition", 1).unwrap();
912        assert_eq!(
913            store
914                .add_workflow_type_version(type_id, "edited", 2)
915                .unwrap(),
916            Some(2)
917        );
918        assert_eq!(
919            store
920                .add_workflow_type_version(type_id + 1, "missing", 3)
921                .unwrap(),
922            None
923        );
924    }
925
926    #[test]
927    fn migration_from_version_two_keeps_workflows_and_stores_a_harness() {
928        let mut connection = Connection::open_in_memory().unwrap();
929        connection
930            .pragma_update(None, "foreign_keys", true)
931            .unwrap();
932        connection.execute_batch(MIGRATIONS[0]).unwrap();
933        connection.execute_batch(MIGRATIONS[1]).unwrap();
934        connection
935            .execute_batch(
936                "INSERT INTO sessions VALUES (1, '/folder', 'Session', 1);
937                 INSERT INTO workflows VALUES (1, 1, 'Terminal', 'terminal');
938                 INSERT INTO workflows VALUES (2, 1, 'Terminal', 'terminal');
939                 INSERT INTO workflows VALUES (3, 1, 'Terminal', 'terminal');
940                 DELETE FROM workflows WHERE id IN (2, 3)",
941            )
942            .unwrap();
943        connection.pragma_update(None, "user_version", 2).unwrap();
944        migrate(&mut connection).unwrap();
945        let mut store = Store::with_connection(connection).unwrap();
946
947        // Workflow 3 was deleted before the migration, and its ID must not come back.
948        let (id, _) = store
949            .create_workflow(
950                SessionId(1),
951                "New workflow",
952                crate::WorkflowKind::Draft,
953                &[],
954            )
955            .unwrap();
956        assert!(
957            id.0 > 3,
958            "deleted workflow IDs must not be reused, got {id:?}"
959        );
960        store
961            .update_workflow(
962                id,
963                "Single agent",
964                crate::WorkflowKind::SingleAgent,
965                Some(crate::HarnessId::ClaudeCode),
966            )
967            .unwrap();
968
969        let workflows = store.workflows(Path::new("/folder")).unwrap();
970        assert_eq!(workflows.len(), 2);
971        assert_eq!(workflows[0].kind, crate::WorkflowKind::Terminal);
972        assert_eq!(workflows[0].harness, None);
973        assert_eq!(workflows[1].kind, crate::WorkflowKind::SingleAgent);
974        assert_eq!(workflows[1].harness, Some(crate::HarnessId::ClaudeCode));
975    }
976
977    #[test]
978    fn migration_to_agents_keeps_workflows_and_never_reuses_their_ids() {
979        let connection = Connection::open_in_memory().unwrap();
980        connection
981            .pragma_update(None, "foreign_keys", true)
982            .unwrap();
983        for migration in &MIGRATIONS[..4] {
984            connection.execute_batch(migration).unwrap();
985        }
986        connection.pragma_update(None, "user_version", 4).unwrap();
987        connection
988            .execute_batch(
989                "INSERT INTO sessions (folder, name, started_at) VALUES ('/folder', 'Session', 1);
990                 INSERT INTO workflows (session_id, name, kind)
991                     VALUES (1, 'New workflow', 'draft'), (1, 'Terminal', 'terminal'),
992                         (1, 'Closed', 'terminal');
993                 DELETE FROM workflows WHERE id = 3;",
994            )
995            .unwrap();
996        let mut store = Store::with_connection(connection).unwrap();
997        let folder = Path::new("/folder");
998        assert_eq!(
999            store
1000                .workflows(folder)
1001                .unwrap()
1002                .iter()
1003                .map(|workflow| (
1004                    workflow.workflow_id.0,
1005                    workflow.name.as_str(),
1006                    workflow.kind
1007                ))
1008                .collect::<Vec<_>>(),
1009            [
1010                (1, "New workflow", crate::WorkflowKind::Draft),
1011                (2, "Terminal", crate::WorkflowKind::Terminal)
1012            ]
1013        );
1014
1015        let (workflow_id, agent_ids) = store
1016            .create_workflow(
1017                crate::SessionId(1),
1018                "Agents",
1019                crate::WorkflowKind::Agents,
1020                &["Implementer", "Reviewer"],
1021            )
1022            .unwrap();
1023        assert_eq!(
1024            workflow_id.0, 4,
1025            "a closed workflow's ID must not be reused"
1026        );
1027        let workflows = store.workflows(folder).unwrap();
1028        let agents = &workflows.last().unwrap().agents;
1029        assert_eq!(
1030            agents
1031                .iter()
1032                .map(|agent| (agent.agent_id, agent.role.as_str()))
1033                .collect::<Vec<_>>(),
1034            [(agent_ids[0], "Implementer"), (agent_ids[1], "Reviewer")]
1035        );
1036        assert!(
1037            workflows[..2]
1038                .iter()
1039                .all(|workflow| workflow.agents.is_empty())
1040        );
1041
1042        store
1043            .connection
1044            .execute(
1045                "DELETE FROM workflows WHERE id = ?1",
1046                [sql_integer(workflow_id.0).unwrap()],
1047            )
1048            .unwrap();
1049        let remaining: i64 = store
1050            .connection
1051            .query_row("SELECT COUNT(*) FROM agents", [], |row| row.get(0))
1052            .unwrap();
1053        assert_eq!(remaining, 0, "deleting a workflow deletes its agents");
1054        assert!(
1055            store
1056                .connection
1057                .execute(
1058                    "INSERT INTO workflows (session_id, name, kind) VALUES (1, 'Other', 'other')",
1059                    [],
1060                )
1061                .is_err()
1062        );
1063    }
1064
1065    #[test]
1066    fn migration_from_version_three_preserves_types_sessions_and_workflows() {
1067        let connection = Connection::open_in_memory().unwrap();
1068        for migration in &MIGRATIONS[..3] {
1069            connection.execute_batch(migration).unwrap();
1070        }
1071        connection.pragma_update(None, "user_version", 3).unwrap();
1072        connection.execute_batch(
1073            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Kept session', 123);
1074             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Kept workflow', 'terminal');
1075             INSERT INTO folder_selection (folder, session_id) VALUES ('/folder', 1);
1076             INSERT INTO workflow_types (id) VALUES (1);
1077             INSERT INTO workflow_type_versions (type_id, version, definition, created_at)
1078             VALUES (1, 1, 'kept definition', 123);"
1079        ).unwrap();
1080        let mut store = Store::with_connection(connection).unwrap();
1081        assert_eq!(
1082            usize::try_from(schema_version(&store.connection)).unwrap(),
1083            MIGRATIONS.len()
1084        );
1085        assert_eq!(
1086            store.sessions(Path::new("/folder")).unwrap()[0].name,
1087            "Kept session"
1088        );
1089        assert_eq!(
1090            store.selected_session(Path::new("/folder")).unwrap(),
1091            Some(crate::SessionId(1))
1092        );
1093        let workflows = store.workflows(Path::new("/folder")).unwrap();
1094        assert_eq!(workflows.len(), 1);
1095        assert_eq!(workflows[0].name, "Kept workflow");
1096        assert_eq!(
1097            store.workflow_type_version(1, 1).unwrap().as_deref(),
1098            Some("kept definition")
1099        );
1100        assert_eq!(
1101            store
1102                .add_workflow_type_version(1, "new definition", 124)
1103                .unwrap(),
1104            Some(2)
1105        );
1106        let summary = store.trace_summary(crate::WorkflowId(1)).unwrap();
1107        assert_eq!(summary.span_count, 0);
1108        let span = store
1109            .start_trace_span(&NewTraceSpan {
1110                workflow_id: crate::WorkflowId(1),
1111                lane_key: "terminal",
1112                lane_name: "Terminal",
1113                is_agent: false,
1114                role: None,
1115                harness: None,
1116                title: "Shell",
1117                started_at: 124,
1118                anchor: None,
1119            })
1120            .unwrap();
1121        assert_eq!(
1122            store
1123                .trace_summary(crate::WorkflowId(1))
1124                .unwrap()
1125                .span_count,
1126            1
1127        );
1128        assert_eq!(store.trace_events(span, None, 10).unwrap().events.len(), 1);
1129        store.close_workflow(crate::WorkflowId(1), 125).unwrap();
1130        assert!(store.workflows(Path::new("/folder")).unwrap().is_empty());
1131        assert_eq!(
1132            store
1133                .workflow_trace(crate::WorkflowId(1), None, 10)
1134                .unwrap()
1135                .spans
1136                .len(),
1137            1
1138        );
1139    }
1140
1141    #[test]
1142    fn migration_from_version_four_keeps_agent_harness_status_and_adds_traces() {
1143        let connection = Connection::open_in_memory().unwrap();
1144        for migration in &MIGRATIONS[..4] {
1145            connection.execute_batch(migration).unwrap();
1146        }
1147        connection.pragma_update(None, "user_version", 4).unwrap();
1148        connection.execute_batch(
1149            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Session', 123);
1150             INSERT INTO workflows (id, session_id, name, kind, harness, agent_status)
1151             VALUES (1, 1, 'Claude Code', 'single_agent', 'claude_code', 'cancelled');"
1152        ).unwrap();
1153        let mut store = Store::with_connection(connection).unwrap();
1154        assert_eq!(
1155            usize::try_from(schema_version(&store.connection)).unwrap(),
1156            MIGRATIONS.len()
1157        );
1158        let workflows = store.workflows(Path::new("/folder")).unwrap();
1159        assert_eq!(workflows.len(), 1);
1160        assert_eq!(workflows[0].kind, crate::WorkflowKind::SingleAgent);
1161        assert_eq!(workflows[0].harness, Some(crate::HarnessId::ClaudeCode));
1162        assert_eq!(
1163            workflows[0].agent_status,
1164            Some(crate::WorkflowStatus::Cancelled)
1165        );
1166        let span = store
1167            .start_trace_span(&NewTraceSpan {
1168                workflow_id: workflows[0].workflow_id,
1169                lane_key: "agent",
1170                lane_name: "Agent",
1171                is_agent: true,
1172                role: Some("agent"),
1173                harness: Some("Claude Code"),
1174                title: "Claude Code",
1175                started_at: 124,
1176                anchor: None,
1177            })
1178            .unwrap();
1179        assert_eq!(store.trace_events(span, None, 10).unwrap().events.len(), 1);
1180        store.close_workflow(workflows[0].workflow_id, 125).unwrap();
1181        assert!(store.workflows(Path::new("/folder")).unwrap().is_empty());
1182        assert_eq!(
1183            store
1184                .workflow_trace(workflows[0].workflow_id, None, 10)
1185                .unwrap()
1186                .spans
1187                .len(),
1188            1
1189        );
1190    }
1191
1192    #[test]
1193    fn migration_from_version_five_preserves_agent_identity_and_role_order() {
1194        let connection = Connection::open_in_memory().unwrap();
1195        for migration in &MIGRATIONS[..5] {
1196            connection.execute_batch(migration).unwrap();
1197        }
1198        connection.pragma_update(None, "user_version", 5).unwrap();
1199        connection.execute_batch(
1200            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Session', 123);
1201             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Agents', 'agents');
1202             INSERT INTO agents (id, workflow_id, role) VALUES (7, 1, 'Reviewer'), (8, 1, 'Implementer');"
1203        ).unwrap();
1204        let store = Store::with_connection(connection).unwrap();
1205        let workflows = store.workflows(Path::new("/folder")).unwrap();
1206        assert_eq!(workflows[0].kind, crate::WorkflowKind::Agents);
1207        assert_eq!(
1208            workflows[0]
1209                .agents
1210                .iter()
1211                .map(|agent| (agent.agent_id.0, agent.role.as_str()))
1212                .collect::<Vec<_>>(),
1213            [(7, "Reviewer"), (8, "Implementer")]
1214        );
1215        assert_eq!(
1216            store
1217                .trace_summary(workflows[0].workflow_id)
1218                .unwrap()
1219                .span_count,
1220            0
1221        );
1222        assert_eq!(
1223            usize::try_from(schema_version(&store.connection)).unwrap(),
1224            MIGRATIONS.len()
1225        );
1226    }
1227
1228    #[test]
1229    fn migration_from_version_six_keeps_trace_ids_and_accepts_workflow_events() {
1230        let connection = Connection::open_in_memory().unwrap();
1231        for migration in &MIGRATIONS[..6] {
1232            connection.execute_batch(migration).unwrap();
1233        }
1234        connection.pragma_update(None, "user_version", 6).unwrap();
1235        connection.execute_batch(
1236            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Work', 123);
1237             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Shell', 'terminal');
1238             INSERT INTO trace_events (id, workflow_id, timestamp, kind, message)
1239             VALUES (80, 1, 123, 'processStarted', 'Earlier event');
1240             DELETE FROM trace_events WHERE id = 80;"
1241        ).unwrap();
1242        let store = Store::with_connection(connection).unwrap();
1243        store
1244            .connection
1245            .execute(
1246                "INSERT INTO trace_events (workflow_id, timestamp, kind, message)
1247             VALUES (1, 124, 'workflowEvent', 'Stage started')",
1248                [],
1249            )
1250            .unwrap();
1251        assert_eq!(store.connection.last_insert_rowid(), 81);
1252        assert_eq!(store.workflows(Path::new("/folder")).unwrap().len(), 1);
1253    }
1254
1255    #[test]
1256    fn migration_preserves_idle_shell_and_agent_terminal_ownership() {
1257        let connection = Connection::open_in_memory().unwrap();
1258        for migration in &MIGRATIONS[..11] {
1259            connection.execute_batch(migration).unwrap();
1260        }
1261        connection.pragma_update(None, "user_version", 11).unwrap();
1262        connection.execute_batch(
1263            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Work', 1);
1264             INSERT INTO workflows (id, session_id, name, kind) VALUES
1265                (1, 1, 'Shell', 'terminal'), (2, 1, 'Agent', 'single_agent'), (3, 1, 'Run', 'agents');
1266             INSERT INTO trace_lanes (id, workflow_id, lane_key, name, is_agent) VALUES
1267                (1, 2, 'terminal', 'Terminal', 0), (2, 2, 'agent', 'Agent', 1), (3, 3, 'agent:8', 'Worker', 1);
1268             INSERT INTO trace_spans (lane_id, title, started_at, status, terminal_id) VALUES
1269                (1, 'Shell', 1, 'stopped', 20), (2, 'Agent', 2, 'exited', 21), (3, 'Worker', 2, 'exited', 31);
1270             INSERT INTO trace_events (workflow_id, timestamp, kind, message, terminal_id, byte_offset)
1271                VALUES (1, 1, 'processStarted', 'Idle shell', 10, 0);"
1272        ).unwrap();
1273        let store = Store::with_connection(connection).unwrap();
1274        let history = |id| store.terminal_history(crate::WorkflowId(id)).unwrap();
1275        assert_eq!(history(1)[0].terminal_id.value(), 10);
1276        assert_eq!(history(2).len(), 1);
1277        assert_eq!(history(2)[0].terminal_id.value(), 21);
1278        assert_eq!(history(3)[0].agent_id, Some(crate::AgentId(8)));
1279    }
1280
1281    #[test]
1282    fn migration_from_version_nine_preserves_process_spans_and_terminal_anchors() {
1283        let connection = Connection::open_in_memory().unwrap();
1284        for migration in &MIGRATIONS[..9] {
1285            connection.execute_batch(migration).unwrap();
1286        }
1287        connection.pragma_update(None, "user_version", 9).unwrap();
1288        connection.execute_batch(
1289            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Work', 100);
1290             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Adversarial', 'agents');
1291             INSERT INTO trace_lanes (id, workflow_id, lane_key, name, is_agent, role, harness)
1292             VALUES (1, 1, 'agent:1', 'Implementer', 1, 'implementer', 'pi');
1293             INSERT INTO trace_spans (id, lane_id, title, started_at, ended_at, status, terminal_id, run_generation)
1294             VALUES (1, 1, 'Implement', 100, 200, 'exited', 99, 1);
1295             INSERT INTO trace_events (id, workflow_id, span_id, timestamp, kind, message, terminal_id, byte_offset, boundary_sizes)
1296             VALUES (1, 1, 1, 100, 'processStarted', 'Process started.', 99, 0, X''),
1297                    (2, 1, 1, 200, 'processExited', 'Process exited with code 0.', 99, 42, X'');"
1298        ).unwrap();
1299        let store = Store::with_connection(connection).unwrap();
1300        let page = store
1301            .workflow_trace(crate::WorkflowId(1), None, 10)
1302            .unwrap();
1303        let span = &page.spans[0];
1304        assert_eq!(span.title, "Implement");
1305        assert_eq!((span.started_at, span.ended_at), (100, Some(200)));
1306        assert_eq!(span.status, crate::TraceSpanStatus::Exited);
1307        let events = store.trace_events(span.span_id, None, 10).unwrap().events;
1308        assert_eq!(events.len(), 2);
1309        assert_eq!(events[1].kind, crate::TraceEventKind::ProcessExited);
1310        assert_eq!(
1311            events[1].anchor,
1312            Some(crate::TraceAnchor {
1313                terminal_id: crate::TerminalId::from_value(99),
1314                byte_offset: 42,
1315                boundary_sizes: Some(vec![]),
1316            })
1317        );
1318    }
1319
1320    #[test]
1321    fn a_newer_schema_is_rejected() {
1322        let directory = tempfile::tempdir().expect("a temporary directory should be available");
1323        let path = directory.path().join("twine.db");
1324        let connection = Connection::open(&path).expect("the database should open");
1325        connection
1326            .pragma_update(None, "user_version", 99)
1327            .expect("the schema version should be writable");
1328        drop(connection);
1329
1330        assert!(matches!(
1331            Store::open(&path),
1332            Err(StoreError::UnsupportedSchemaVersion { version: 99 })
1333        ));
1334    }
1335
1336    #[test]
1337    fn recent_folders_are_ordered_capped_and_track_the_open_folder() {
1338        let mut store = Store::open_in_memory().expect("the store should open");
1339        store
1340            .record_folder_opened("/a", 5_000, 2)
1341            .expect("a should be recorded");
1342        // An earlier clock reading still sorts after the folder opened before it.
1343        store
1344            .record_folder_opened("/b", 1_000, 2)
1345            .expect("b should be recorded");
1346        store
1347            .record_folder_opened("/c", 1_000, 2)
1348            .expect("c should be recorded");
1349
1350        assert_eq!(
1351            store.recent_folders().expect("folders should load"),
1352            [
1353                StoredFolder {
1354                    path: PathBuf::from("/c"),
1355                    is_open: true,
1356                },
1357                StoredFolder {
1358                    path: PathBuf::from("/b"),
1359                    is_open: false,
1360                },
1361            ]
1362        );
1363
1364        store
1365            .record_folder_opened("/b", 1_000, 2)
1366            .expect("b should be recorded again");
1367        store
1368            .record_folder_closed()
1369            .expect("closing should be recorded");
1370        let remaining = store
1371            .remove_recent_folder("/c")
1372            .expect("c should be removed");
1373        assert_eq!(
1374            remaining,
1375            [StoredFolder {
1376                path: PathBuf::from("/b"),
1377                is_open: false,
1378            }]
1379        );
1380        assert_eq!(
1381            store.recent_folders().expect("folders should load"),
1382            remaining
1383        );
1384    }
1385}
twine_core/store.rs

twine_core/
store.rs
