handshake_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;
18pub(crate) mod 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    // 20: Custom workflow titles and read-only reopened history.
270    "ALTER TABLE workflows ADD COLUMN custom_name TEXT;
271     ALTER TABLE workflows ADD COLUMN archived INTEGER NOT NULL DEFAULT 0 CHECK(archived IN (0,1))"
272];
273
274/// How long a write waits for another connection, such as a second Handshake process, to release the
275/// database.
276const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
277
278/// A recent folder as stored, without the file system checks the folder module adds.
279#[derive(Debug, Eq, PartialEq)]
280pub(crate) struct StoredFolder {
281    pub(crate) path: PathBuf,
282    /// Whether the folder was open when Handshake last quit.
283    pub(crate) is_open: bool,
284}
285
286/// The SQLite database that holds Handshake's persistent state.
287#[derive(Debug)]
288pub(crate) struct Store {
289    connection: Connection,
290    payloads: trace_payloads::TracePayloads,
291    pub(crate) workflow_directory: std::path::PathBuf,
292    _workflow_temp: Option<tempfile::TempDir>,
293}
294
295fn workflow_file_directory(config_path: Option<&Path>) -> PathBuf {
296    config_path.and_then(Path::parent).map_or_else(
297        || std::env::temp_dir().join("handshake/workflows"),
298        |directory| directory.join("workflows"),
299    )
300}
301
302impl Store {
303    /// Opens the database at `path`, creating the file and its directory if needed, and applies
304    /// pending migrations.
305    pub(crate) fn open(path: &Path) -> Result<Self, StoreError> {
306        if let Some(directory) = path.parent() {
307            std::fs::create_dir_all(directory).map_err(|source| StoreError::CreateDirectory {
308                path: directory.to_owned(),
309                source,
310            })?;
311        }
312        let connection = Connection::open(path).map_err(|source| StoreError::Open {
313            path: path.to_owned(),
314            source,
315        })?;
316        Self::with_connection(connection)
317    }
318
319    #[cfg(test)]
320    pub(crate) fn execute_test_sql(&self, sql: &str) {
321        self.connection
322            .execute_batch(sql)
323            .expect("test SQL should execute");
324    }
325
326    #[cfg(test)]
327    pub(crate) fn open_in_memory() -> Result<Self, StoreError> {
328        Self::with_connection(Connection::open_in_memory()?)
329    }
330
331    fn with_connection(mut connection: Connection) -> Result<Self, StoreError> {
332        connection.busy_timeout(BUSY_TIMEOUT)?;
333        connection.pragma_update(None, "journal_mode", "WAL")?;
334        connection.pragma_update(None, "foreign_keys", true)?;
335        migrate(&mut connection)?;
336        let payloads = trace_payloads::TracePayloads::new(connection.path())?;
337        let temporary = if connection.path().is_none() || connection.path() == Some("") {
338            Some(
339                tempfile::tempdir().map_err(|source| StoreError::CreateDirectory {
340                    path: std::env::temp_dir(),
341                    source,
342                })?,
343            )
344        } else {
345            None
346        };
347        let workflow_directory = if let Some(temporary) = &temporary {
348            temporary.path().join("workflows")
349        } else if cfg!(test) {
350            std::path::Path::new(connection.path().expect("disk store"))
351                .parent()
352                .unwrap_or_else(|| std::path::Path::new("."))
353                .join("workflows")
354        } else {
355            workflow_file_directory(crate::config::Config::user_path().as_deref())
356        };
357        Ok(Self {
358            connection,
359            payloads,
360            workflow_directory,
361            _workflow_temp: temporary,
362        })
363    }
364
365    /// Verify every expected domain column before a copied database becomes authoritative.
366    /// A valid SQLite file and a current `user_version` alone do not establish a usable app schema.
367    pub(crate) fn verify_schema(&self) -> Result<(), StoreError> {
368        let mut expected = Connection::open_in_memory()?;
369        migrate(&mut expected)?;
370        let mut tables = expected.prepare(
371            "SELECT name FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%'",
372        )?;
373        for table in tables.query_map([], |row| row.get::<_, String>(0))? {
374            let table = table?;
375            // Identifiers come solely from our embedded schema, never caller input.
376            let mut columns = expected.prepare(&format!("PRAGMA table_info(\"{table}\")"))?;
377            let columns = columns
378                .query_map([], |row| row.get::<_, String>(1))?
379                .map(|column| column.map(|name| format!("\"{table}\".\"{name}\"")))
380                .collect::<rusqlite::Result<Vec<_>>>()?;
381            self.connection.prepare(&format!(
382                "SELECT {} FROM \"{table}\" LIMIT 0",
383                columns.join(",")
384            ))?;
385        }
386        let broken = self
387            .connection
388            .prepare("PRAGMA foreign_key_check")?
389            .exists([])?;
390        if broken {
391            return Err(rusqlite::Error::InvalidQuery.into());
392        }
393        Ok(())
394    }
395
396    /// Returns the recent folders, most recently opened first.
397    pub(crate) fn recent_folders(&self) -> Result<Vec<StoredFolder>, StoreError> {
398        query_recent_folders(&self.connection)
399    }
400
401    pub(crate) fn data_version(&self) -> Result<i64, StoreError> {
402        Ok(self
403            .connection
404            .pragma_query_value(None, "data_version", |row| row.get(0))?)
405    }
406
407    pub(crate) fn open_folder_paths(&self) -> Result<Vec<PathBuf>, StoreError> {
408        let mut statement = self
409            .connection
410            .prepare("SELECT path FROM open_folders ORDER BY last_opened_at DESC")?;
411        Ok(statement
412            .query_map([], |row| Ok(PathBuf::from(row.get::<_, String>(0)?)))?
413            .collect::<Result<_, _>>()?)
414    }
415
416    /// Records `path` as the open folder and the most recently opened one, then forgets all but the
417    /// `limit` most recent folders. Returns the recent folders after the change.
418    ///
419    /// The recorded time is at least one millisecond after every stored time, so the new folder
420    /// sorts first even if the clock went backward or two folders open in the same millisecond.
421    pub(crate) fn record_folder_opened(
422        &mut self,
423        path: &str,
424        opened_at_millis: i64,
425        limit: i64,
426    ) -> Result<Vec<StoredFolder>, StoreError> {
427        self.record_window_folder_opened(path, opened_at_millis, limit, None, true)
428    }
429
430    pub(crate) fn record_window_folder_opened(
431        &mut self,
432        path: &str,
433        opened_at_millis: i64,
434        limit: i64,
435        previous: Option<&Path>,
436        exclusive: bool,
437    ) -> Result<Vec<StoredFolder>, StoreError> {
438        let transaction = self.connection.transaction()?;
439        if exclusive {
440            transaction.execute("DELETE FROM open_folders", [])?;
441            transaction.execute(
442                "UPDATE recent_folders SET is_open = 0 WHERE is_open = 1",
443                [],
444            )?;
445        } else if let Some(previous) = previous {
446            transaction.execute(
447                "DELETE FROM open_folders WHERE path = ?1",
448                [previous.to_string_lossy()],
449            )?;
450            transaction.execute(
451                "UPDATE recent_folders SET is_open = 0 WHERE path = ?1",
452                [previous.to_string_lossy()],
453            )?;
454        }
455        transaction.execute(
456            "INSERT INTO recent_folders (path, last_opened_at, is_open)
457             VALUES (?1, MAX(?2, (SELECT COALESCE(MAX(last_opened_at), 0) + 1 FROM
458                (SELECT last_opened_at FROM recent_folders UNION ALL SELECT last_opened_at FROM open_folders))), 1)
459             ON CONFLICT (path) DO UPDATE
460             SET last_opened_at = excluded.last_opened_at, is_open = 1",
461            params![path, opened_at_millis],
462        )?;
463        transaction.execute(
464            "INSERT INTO open_folders (path, last_opened_at)
465            SELECT path, last_opened_at FROM recent_folders WHERE path = ?1
466            ON CONFLICT(path) DO UPDATE SET last_opened_at = excluded.last_opened_at",
467            [path],
468        )?;
469        transaction.execute(
470            "DELETE FROM recent_folders WHERE path NOT IN
471             (SELECT path FROM recent_folders ORDER BY last_opened_at DESC LIMIT ?1)",
472            [limit],
473        )?;
474        let folders = query_recent_folders(&transaction)?;
475        transaction.commit()?;
476        Ok(folders)
477    }
478
479    /// Records that no folder is open, so the next launch shows the start page. Returns the recent
480    /// folders after the change.
481    pub(crate) fn record_folder_closed(&mut self) -> Result<Vec<StoredFolder>, StoreError> {
482        let transaction = self.connection.transaction()?;
483        transaction.execute("DELETE FROM open_folders", [])?;
484        transaction.execute(
485            "UPDATE recent_folders SET is_open = 0 WHERE is_open = 1",
486            [],
487        )?;
488        let folders = query_recent_folders(&transaction)?;
489        transaction.commit()?;
490        Ok(folders)
491    }
492
493    pub(crate) fn record_window_folder_closed(
494        &mut self,
495        path: Option<&Path>,
496    ) -> Result<Vec<StoredFolder>, StoreError> {
497        let transaction = self.connection.transaction()?;
498        if let Some(path) = path {
499            transaction.execute(
500                "DELETE FROM open_folders WHERE path = ?1",
501                [path.to_string_lossy()],
502            )?;
503            transaction.execute(
504                "UPDATE recent_folders SET is_open = 0 WHERE path = ?1",
505                [path.to_string_lossy()],
506            )?;
507        }
508        let folders = query_recent_folders(&transaction)?;
509        transaction.commit()?;
510        Ok(folders)
511    }
512
513    /// Forgets the recent folder at `path`. Returns the recent folders after the change.
514    pub(crate) fn remove_recent_folder(
515        &mut self,
516        path: &str,
517    ) -> Result<Vec<StoredFolder>, StoreError> {
518        self.remove_recent_folder_preserving_windows(path, false)
519    }
520
521    pub(crate) fn remove_recent_folder_preserving_windows(
522        &mut self,
523        path: &str,
524        preserve_open: bool,
525    ) -> Result<Vec<StoredFolder>, StoreError> {
526        let transaction = self.connection.transaction()?;
527        if !preserve_open {
528            transaction.execute("DELETE FROM open_folders WHERE path = ?1", [path])?;
529        }
530        transaction.execute("DELETE FROM recent_folders WHERE path = ?1", [path])?;
531        let folders = query_recent_folders(&transaction)?;
532        transaction.commit()?;
533        Ok(folders)
534    }
535}
536
537fn query_recent_folders(connection: &Connection) -> Result<Vec<StoredFolder>, StoreError> {
538    let mut statement = connection
539        .prepare("SELECT path, is_open FROM recent_folders ORDER BY last_opened_at DESC")?;
540    let folders = statement
541        .query_map([], |row| {
542            Ok(StoredFolder {
543                path: PathBuf::from(row.get::<_, String>(0)?),
544                is_open: row.get(1)?,
545            })
546        })?
547        .collect::<Result<_, _>>()?;
548    Ok(folders)
549}
550
551/// Applies pending migrations in one transaction that takes the write lock before reading the
552/// schema version, so a second Handshake process opening the database waits instead of migrating it
553/// again.
554fn migrate(connection: &mut Connection) -> Result<(), StoreError> {
555    let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
556    let version: i64 = transaction.pragma_query_value(None, "user_version", |row| row.get(0))?;
557    let applied = usize::try_from(version)
558        .ok()
559        .filter(|applied| *applied <= MIGRATIONS.len())
560        .ok_or(StoreError::UnsupportedSchemaVersion { version })?;
561
562    for (version, migration) in (1_i64..).zip(MIGRATIONS).skip(applied) {
563        transaction
564            .execute_batch(migration)
565            .map_err(|source| StoreError::Migrate { version, source })?;
566        transaction.pragma_update(None, "user_version", version)?;
567        info!(version, "applied database migration");
568    }
569    transaction.commit()?;
570    Ok(())
571}
572
573fn sql_integer(value: u64) -> Result<i64, StoreError> {
574    i64::try_from(value).map_err(|_| StoreError::InvalidIdentifier)
575}
576
577fn unsigned_column(row: &rusqlite::Row<'_>, index: usize) -> rusqlite::Result<u64> {
578    let value: i64 = row.get(index)?;
579    u64::try_from(value).map_err(|_| rusqlite::Error::IntegralValueOutOfRange(index, value))
580}
581
582#[derive(Debug, Error)]
583pub enum StoreError {
584    #[error("couldn't read or store full trace details")]
585    TracePayload(#[source] std::io::Error),
586    #[error("failed to resolve data directory {}", path.display())]
587    ResolveDirectory {
588        path: PathBuf,
589        source: std::io::Error,
590    },
591    #[error("failed to encode workflow run state")]
592    RunEncoding(#[from] serde_json::Error),
593    #[error("database returned an invalid identifier")]
594    InvalidIdentifier,
595    #[error("failed to create the data directory {}", path.display())]
596    CreateDirectory {
597        path: PathBuf,
598        #[source]
599        source: std::io::Error,
600    },
601    #[error("failed to apply database migration {version}")]
602    Migrate {
603        version: i64,
604        #[source]
605        source: rusqlite::Error,
606    },
607    #[error("failed to open the database {}", path.display())]
608    Open {
609        path: PathBuf,
610        #[source]
611        source: rusqlite::Error,
612    },
613    #[error(transparent)]
614    Sqlite(#[from] rusqlite::Error),
615    #[error("database schema version {version} is not one this version of Handshake supports")]
616    UnsupportedSchemaVersion { version: i64 },
617}
618
619#[cfg(test)]
620mod tests {
621    use super::*;
622    use crate::workflow::SessionId;
623
624    fn schema_version(connection: &Connection) -> i64 {
625        connection
626            .pragma_query_value(None, "user_version", |row| row.get(0))
627            .expect("the schema version should be readable")
628    }
629
630    #[test]
631    fn workflows_follow_the_selected_config_directory() {
632        for directory in [
633            "/checkout/out/runtime/debug/config",
634            "/home/user/.config/handshake",
635            "/custom/handshake",
636        ] {
637            assert_eq!(
638                workflow_file_directory(Some(&Path::new(directory).join("config.toml"))),
639                Path::new(directory).join("workflows")
640            );
641        }
642    }
643
644    #[test]
645    fn open_creates_the_directory_and_migrates_once() {
646        let directory = tempfile::tempdir().expect("a temporary directory should be available");
647        let path = directory.path().join("nested").join("twine.db");
648
649        let store = Store::open(&path).expect("the store should open");
650        assert_eq!(
651            schema_version(&store.connection),
652            i64::try_from(MIGRATIONS.len()).expect("the migration count fits in i64")
653        );
654        drop(store);
655
656        let reopened = Store::open(&path).expect("the migrated store should reopen");
657        assert_eq!(
658            schema_version(&reopened.connection),
659            i64::try_from(MIGRATIONS.len()).expect("the migration count fits in i64")
660        );
661    }
662
663    fn legacy_workflow_history() -> Connection {
664        let connection = Connection::open_in_memory().unwrap();
665        connection
666            .pragma_update(None, "foreign_keys", true)
667            .unwrap();
668        for migration in &MIGRATIONS[..11] {
669            connection.execute_batch(migration).unwrap();
670        }
671        connection.pragma_update(None, "user_version", 11).unwrap();
672        connection.execute_batch(
673            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Session', 123);
674             INSERT INTO workflows (id, session_id, name, kind, harness, lifecycle_status, closed_at)
675                 VALUES (1, 1, 'Claude Code', 'single_agent', 'claude_code', 'cancelled', 456),
676                        (2, 1, 'Agents', 'agents', NULL, NULL, NULL),
677                        (99, 1, 'Deleted', 'draft', NULL, NULL, NULL);
678             DELETE FROM workflows WHERE id = 99;
679             INSERT INTO agents (id, workflow_id, role) VALUES (7, 2, 'Worker');
680             INSERT INTO workflow_runs (workflow_id, state, trace_sequence) VALUES (2, 'kept run state', 3);
681             INSERT INTO trace_lanes (id, workflow_id, lane_key, name, is_agent, harness)
682                 VALUES (8, 2, 'worker', 'Worker', 1, 'pi');
683             INSERT INTO trace_spans (id, lane_id, title, started_at, status, harness_turn_id)
684                 VALUES (9, 8, 'Task', 123, 'running', 'turn-1');
685             INSERT INTO trace_events (id, workflow_id, span_id, timestamp, kind, message)
686                 VALUES (10, 2, 9, 123, 'processStarted', 'Started');"
687        ).unwrap();
688        connection
689    }
690
691    #[test]
692    fn new_harness_migration_preserves_workflows_and_dependent_history() {
693        let connection = legacy_workflow_history();
694        let mut store = Store::with_connection(connection).unwrap();
695        let preserved: (String, String, i64) = store
696            .connection
697            .query_row(
698                "SELECT harness_id, lifecycle_status, closed_at FROM workflows WHERE id = 1",
699                [],
700                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
701            )
702            .unwrap();
703        assert_eq!(preserved, ("claude_code".into(), "cancelled".into(), 456));
704        let run: (String, i64) = store
705            .connection
706            .query_row(
707                "SELECT state, trace_sequence FROM workflow_runs WHERE workflow_id = 2",
708                [],
709                |row| Ok((row.get(0)?, row.get(1)?)),
710            )
711            .unwrap();
712        assert_eq!(run, ("kept run state".into(), 3));
713        for (table, id) in [
714            ("agents", 7),
715            ("trace_lanes", 8),
716            ("trace_spans", 9),
717            ("trace_events", 10),
718        ] {
719            let stored: i64 = store
720                .connection
721                .query_row(&format!("SELECT id FROM {table}"), [], |row| row.get(0))
722                .unwrap();
723            assert_eq!(stored, id);
724        }
725        assert!(
726            store
727                .connection
728                .prepare("PRAGMA foreign_key_check")
729                .unwrap()
730                .query([])
731                .unwrap()
732                .next()
733                .unwrap()
734                .is_none()
735        );
736        for harness in [
737            crate::HarnessId::Antigravity,
738            crate::HarnessId::Omp,
739            crate::HarnessId::Opencode,
740        ] {
741            store
742                .update_workflow(
743                    crate::WorkflowId(1),
744                    harness.definition().name,
745                    crate::WorkflowKind::SingleAgent,
746                    Some(harness),
747                )
748                .unwrap();
749            let stored: String = store
750                .connection
751                .query_row("SELECT harness_id FROM workflows WHERE id = 1", [], |row| {
752                    row.get(0)
753                })
754                .unwrap();
755            assert_eq!(stored, super::workflows::harness_name(harness));
756        }
757        let (id, _) = store
758            .create_workflow(SessionId(1), "New", crate::WorkflowKind::Draft, &[])
759            .unwrap();
760        assert!(id.0 > 99, "Deleted workflow IDs must not be reused");
761        store
762            .connection
763            .execute("DELETE FROM workflows WHERE id = 2", [])
764            .unwrap();
765        for table in [
766            "agents",
767            "workflow_runs",
768            "trace_lanes",
769            "trace_spans",
770            "trace_events",
771        ] {
772            let remaining: i64 = store
773                .connection
774                .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| {
775                    row.get(0)
776                })
777                .unwrap();
778            assert_eq!(
779                remaining, 0,
780                "Deleting a workflow still cascades to {table}"
781            );
782        }
783    }
784
785    #[test]
786    fn harness_migrations_preserve_version_twelve_terminal_ownership() {
787        let connection = legacy_workflow_history();
788        connection.execute_batch(MIGRATIONS[11]).unwrap();
789        connection.pragma_update(None, "user_version", 12).unwrap();
790        connection
791            .execute_batch(
792                "INSERT INTO workflow_terminals (terminal_id, workflow_id, agent_id)
793                 VALUES (42, 1, NULL), (43, 2, 7)",
794            )
795            .unwrap();
796
797        let store = Store::with_connection(connection).unwrap();
798        let harness: String = store
799            .connection
800            .query_row("SELECT harness_id FROM workflows WHERE id = 1", [], |row| {
801                row.get(0)
802            })
803            .unwrap();
804        assert_eq!(harness, "claude_code");
805        for (workflow, terminal, agent) in [(1, 42, None), (2, 43, Some(crate::AgentId(7)))] {
806            let history = store.terminal_history(crate::WorkflowId(workflow)).unwrap();
807            assert_eq!(history.len(), 1);
808            assert_eq!(history[0].terminal_id.value(), terminal);
809            assert_eq!(history[0].agent_id, agent);
810        }
811        store
812            .connection
813            .execute("DELETE FROM workflows WHERE id = 2", [])
814            .unwrap();
815        assert_eq!(store.terminal_history(crate::WorkflowId(2)).unwrap(), []);
816        assert_eq!(
817            store.terminal_history(crate::WorkflowId(1)).unwrap().len(),
818            1
819        );
820        assert!(
821            store
822                .connection
823                .prepare("PRAGMA foreign_key_check")
824                .unwrap()
825                .query([])
826                .unwrap()
827                .next()
828                .unwrap()
829                .is_none()
830        );
831    }
832
833    #[test]
834    fn extensible_harness_migration_preserves_version_thirteen_choices_and_history() {
835        let connection = legacy_workflow_history();
836        for migration in &MIGRATIONS[11..13] {
837            connection.execute_batch(migration).unwrap();
838        }
839        connection.pragma_update(None, "user_version", 13).unwrap();
840        connection
841            .execute("UPDATE workflows SET harness_id = 'omp' WHERE id = 1", [])
842            .unwrap();
843        let mut store = Store::with_connection(connection).unwrap();
844        let preserved: (String, String, i64) = store
845            .connection
846            .query_row(
847                "SELECT harness_id, lifecycle_status, closed_at FROM workflows WHERE id = 1",
848                [],
849                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
850            )
851            .unwrap();
852        assert_eq!(preserved, ("omp".into(), "cancelled".into(), 456));
853        store
854            .update_workflow(
855                crate::WorkflowId(1),
856                "OpenCode",
857                crate::WorkflowKind::SingleAgent,
858                Some(crate::HarnessId::Opencode),
859            )
860            .unwrap();
861        let harness: String = store
862            .connection
863            .query_row("SELECT harness_id FROM workflows WHERE id = 1", [], |row| {
864                row.get(0)
865            })
866            .unwrap();
867        assert_eq!(harness, "opencode");
868        for (table, id) in [
869            ("agents", 7),
870            ("trace_lanes", 8),
871            ("trace_spans", 9),
872            ("trace_events", 10),
873        ] {
874            let stored: i64 = store
875                .connection
876                .query_row(&format!("SELECT id FROM {table}"), [], |row| row.get(0))
877                .unwrap();
878            assert_eq!(stored, id);
879        }
880        let (id, _) = store
881            .create_workflow(SessionId(1), "New", crate::WorkflowKind::Draft, &[])
882            .unwrap();
883        assert!(id.0 > 99);
884        assert!(
885            store
886                .connection
887                .prepare("PRAGMA foreign_key_check")
888                .unwrap()
889                .query([])
890                .unwrap()
891                .next()
892                .unwrap()
893                .is_none()
894        );
895    }
896
897    #[test]
898    fn migration_from_version_one_preserves_folders_and_adds_session_storage() {
899        let mut connection = Connection::open_in_memory().unwrap();
900        connection.execute_batch(MIGRATIONS[0]).unwrap();
901        connection
902            .execute("INSERT INTO recent_folders VALUES ('/folder', 1, 1)", [])
903            .unwrap();
904        connection.pragma_update(None, "user_version", 1).unwrap();
905        migrate(&mut connection).unwrap();
906        let mut store = Store::with_connection(connection).unwrap();
907        assert_eq!(
908            store.recent_folders().unwrap(),
909            [StoredFolder {
910                path: "/folder".into(),
911                is_open: true
912            }]
913        );
914        let id = store
915            .create_session(Path::new("/folder"), "Session", 123)
916            .unwrap();
917        store
918            .create_workflow(id, "Terminal", crate::WorkflowKind::Terminal, &[])
919            .unwrap();
920        assert_eq!(
921            store.selected_session(Path::new("/folder")).unwrap(),
922            Some(id)
923        );
924        assert_eq!(store.workflows(Path::new("/folder")).unwrap().len(), 1);
925    }
926
927    #[test]
928    fn migration_from_version_two_preserves_sessions_and_adds_workflow_types() {
929        let mut connection = Connection::open_in_memory().unwrap();
930        connection.execute_batch(MIGRATIONS[0]).unwrap();
931        connection.execute_batch(MIGRATIONS[1]).unwrap();
932        connection
933            .execute(
934                "INSERT INTO sessions (folder, name, started_at) VALUES ('/folder', 'Kept', 1)",
935                [],
936            )
937            .unwrap();
938        connection.pragma_update(None, "user_version", 2).unwrap();
939        migrate(&mut connection).unwrap();
940        let mut store = Store::with_connection(connection).unwrap();
941        assert_eq!(
942            store.sessions(Path::new("/folder")).unwrap()[0].name,
943            "Kept"
944        );
945        let type_id = store.create_workflow_type("definition", 1).unwrap();
946        assert_eq!(
947            store
948                .add_workflow_type_version(type_id, "edited", 2)
949                .unwrap(),
950            Some(2)
951        );
952        assert_eq!(
953            store
954                .add_workflow_type_version(type_id + 1, "missing", 3)
955                .unwrap(),
956            None
957        );
958    }
959
960    #[test]
961    fn migration_from_version_two_keeps_workflows_and_stores_a_harness() {
962        let mut connection = Connection::open_in_memory().unwrap();
963        connection
964            .pragma_update(None, "foreign_keys", true)
965            .unwrap();
966        connection.execute_batch(MIGRATIONS[0]).unwrap();
967        connection.execute_batch(MIGRATIONS[1]).unwrap();
968        connection
969            .execute_batch(
970                "INSERT INTO sessions VALUES (1, '/folder', 'Session', 1);
971                 INSERT INTO workflows VALUES (1, 1, 'Terminal', 'terminal');
972                 INSERT INTO workflows VALUES (2, 1, 'Terminal', 'terminal');
973                 INSERT INTO workflows VALUES (3, 1, 'Terminal', 'terminal');
974                 DELETE FROM workflows WHERE id IN (2, 3)",
975            )
976            .unwrap();
977        connection.pragma_update(None, "user_version", 2).unwrap();
978        migrate(&mut connection).unwrap();
979        let mut store = Store::with_connection(connection).unwrap();
980
981        // Workflow 3 was deleted before the migration, and its ID must not come back.
982        let (id, _) = store
983            .create_workflow(
984                SessionId(1),
985                "New workflow",
986                crate::WorkflowKind::Draft,
987                &[],
988            )
989            .unwrap();
990        assert!(
991            id.0 > 3,
992            "deleted workflow IDs must not be reused, got {id:?}"
993        );
994        store
995            .update_workflow(
996                id,
997                "Single agent",
998                crate::WorkflowKind::SingleAgent,
999                Some(crate::HarnessId::ClaudeCode),
1000            )
1001            .unwrap();
1002
1003        let workflows = store.workflows(Path::new("/folder")).unwrap();
1004        assert_eq!(workflows.len(), 2);
1005        assert_eq!(workflows[0].kind, crate::WorkflowKind::Terminal);
1006        assert_eq!(workflows[0].harness, None);
1007        assert_eq!(workflows[1].kind, crate::WorkflowKind::SingleAgent);
1008        assert_eq!(workflows[1].harness, Some(crate::HarnessId::ClaudeCode));
1009    }
1010
1011    #[test]
1012    fn migration_to_agents_keeps_workflows_and_never_reuses_their_ids() {
1013        let connection = Connection::open_in_memory().unwrap();
1014        connection
1015            .pragma_update(None, "foreign_keys", true)
1016            .unwrap();
1017        for migration in &MIGRATIONS[..4] {
1018            connection.execute_batch(migration).unwrap();
1019        }
1020        connection.pragma_update(None, "user_version", 4).unwrap();
1021        connection
1022            .execute_batch(
1023                "INSERT INTO sessions (folder, name, started_at) VALUES ('/folder', 'Session', 1);
1024                 INSERT INTO workflows (session_id, name, kind)
1025                     VALUES (1, 'New workflow', 'draft'), (1, 'Terminal', 'terminal'),
1026                         (1, 'Closed', 'terminal');
1027                 DELETE FROM workflows WHERE id = 3;",
1028            )
1029            .unwrap();
1030        let mut store = Store::with_connection(connection).unwrap();
1031        let folder = Path::new("/folder");
1032        assert_eq!(
1033            store
1034                .workflows(folder)
1035                .unwrap()
1036                .iter()
1037                .map(|workflow| (
1038                    workflow.workflow_id.0,
1039                    workflow.name.as_str(),
1040                    workflow.kind
1041                ))
1042                .collect::<Vec<_>>(),
1043            [
1044                (1, "New workflow", crate::WorkflowKind::Draft),
1045                (2, "Terminal", crate::WorkflowKind::Terminal)
1046            ]
1047        );
1048
1049        let (workflow_id, agent_ids) = store
1050            .create_workflow(
1051                crate::SessionId(1),
1052                "Agents",
1053                crate::WorkflowKind::Agents,
1054                &["Implementer", "Reviewer"],
1055            )
1056            .unwrap();
1057        assert_eq!(
1058            workflow_id.0, 4,
1059            "a closed workflow's ID must not be reused"
1060        );
1061        let workflows = store.workflows(folder).unwrap();
1062        let agents = &workflows.last().unwrap().agents;
1063        assert_eq!(
1064            agents
1065                .iter()
1066                .map(|agent| (agent.agent_id, agent.role.as_str()))
1067                .collect::<Vec<_>>(),
1068            [(agent_ids[0], "Implementer"), (agent_ids[1], "Reviewer")]
1069        );
1070        assert!(
1071            workflows[..2]
1072                .iter()
1073                .all(|workflow| workflow.agents.is_empty())
1074        );
1075
1076        store
1077            .connection
1078            .execute(
1079                "DELETE FROM workflows WHERE id = ?1",
1080                [sql_integer(workflow_id.0).unwrap()],
1081            )
1082            .unwrap();
1083        let remaining: i64 = store
1084            .connection
1085            .query_row("SELECT COUNT(*) FROM agents", [], |row| row.get(0))
1086            .unwrap();
1087        assert_eq!(remaining, 0, "deleting a workflow deletes its agents");
1088        assert!(
1089            store
1090                .connection
1091                .execute(
1092                    "INSERT INTO workflows (session_id, name, kind) VALUES (1, 'Other', 'other')",
1093                    [],
1094                )
1095                .is_err()
1096        );
1097    }
1098
1099    #[test]
1100    fn migration_from_version_three_preserves_types_sessions_and_workflows() {
1101        let connection = Connection::open_in_memory().unwrap();
1102        for migration in &MIGRATIONS[..3] {
1103            connection.execute_batch(migration).unwrap();
1104        }
1105        connection.pragma_update(None, "user_version", 3).unwrap();
1106        connection.execute_batch(
1107            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Kept session', 123);
1108             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Kept workflow', 'terminal');
1109             INSERT INTO folder_selection (folder, session_id) VALUES ('/folder', 1);
1110             INSERT INTO workflow_types (id) VALUES (1);
1111             INSERT INTO workflow_type_versions (type_id, version, definition, created_at)
1112             VALUES (1, 1, 'kept definition', 123);"
1113        ).unwrap();
1114        let mut store = Store::with_connection(connection).unwrap();
1115        assert_eq!(
1116            usize::try_from(schema_version(&store.connection)).unwrap(),
1117            MIGRATIONS.len()
1118        );
1119        assert_eq!(
1120            store.sessions(Path::new("/folder")).unwrap()[0].name,
1121            "Kept session"
1122        );
1123        assert_eq!(
1124            store.selected_session(Path::new("/folder")).unwrap(),
1125            Some(crate::SessionId(1))
1126        );
1127        let workflows = store.workflows(Path::new("/folder")).unwrap();
1128        assert_eq!(workflows.len(), 1);
1129        assert_eq!(workflows[0].name, "Kept workflow");
1130        assert_eq!(
1131            store.workflow_type_version(1, 1).unwrap().as_deref(),
1132            Some("kept definition")
1133        );
1134        assert_eq!(
1135            store
1136                .add_workflow_type_version(1, "new definition", 124)
1137                .unwrap(),
1138            Some(2)
1139        );
1140        let summary = store.trace_summary(crate::WorkflowId(1)).unwrap();
1141        assert_eq!(summary.span_count, 0);
1142        let span = store
1143            .start_trace_span(&NewTraceSpan {
1144                workflow_id: crate::WorkflowId(1),
1145                lane_key: "terminal",
1146                lane_name: "Terminal",
1147                is_agent: false,
1148                role: None,
1149                harness: None,
1150                title: "Shell",
1151                started_at: 124,
1152                anchor: None,
1153            })
1154            .unwrap();
1155        assert_eq!(
1156            store
1157                .trace_summary(crate::WorkflowId(1))
1158                .unwrap()
1159                .span_count,
1160            1
1161        );
1162        assert_eq!(store.trace_events(span, None, 10).unwrap().events.len(), 1);
1163        store.close_workflow(crate::WorkflowId(1), 125).unwrap();
1164        assert!(store.workflows(Path::new("/folder")).unwrap().is_empty());
1165        assert_eq!(
1166            store
1167                .workflow_trace(crate::WorkflowId(1), None, 10)
1168                .unwrap()
1169                .spans
1170                .len(),
1171            1
1172        );
1173    }
1174
1175    #[test]
1176    fn migration_from_version_four_keeps_agent_harness_status_and_adds_traces() {
1177        let connection = Connection::open_in_memory().unwrap();
1178        for migration in &MIGRATIONS[..4] {
1179            connection.execute_batch(migration).unwrap();
1180        }
1181        connection.pragma_update(None, "user_version", 4).unwrap();
1182        connection.execute_batch(
1183            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Session', 123);
1184             INSERT INTO workflows (id, session_id, name, kind, harness, agent_status)
1185             VALUES (1, 1, 'Claude Code', 'single_agent', 'claude_code', 'cancelled');"
1186        ).unwrap();
1187        let mut store = Store::with_connection(connection).unwrap();
1188        assert_eq!(
1189            usize::try_from(schema_version(&store.connection)).unwrap(),
1190            MIGRATIONS.len()
1191        );
1192        let workflows = store.workflows(Path::new("/folder")).unwrap();
1193        assert_eq!(workflows.len(), 1);
1194        assert_eq!(workflows[0].kind, crate::WorkflowKind::SingleAgent);
1195        assert_eq!(workflows[0].harness, Some(crate::HarnessId::ClaudeCode));
1196        assert_eq!(
1197            workflows[0].agent_status,
1198            Some(crate::WorkflowStatus::Cancelled)
1199        );
1200        let span = store
1201            .start_trace_span(&NewTraceSpan {
1202                workflow_id: workflows[0].workflow_id,
1203                lane_key: "agent",
1204                lane_name: "Agent",
1205                is_agent: true,
1206                role: Some("agent"),
1207                harness: Some("Claude Code"),
1208                title: "Claude Code",
1209                started_at: 124,
1210                anchor: None,
1211            })
1212            .unwrap();
1213        assert_eq!(store.trace_events(span, None, 10).unwrap().events.len(), 1);
1214        store.close_workflow(workflows[0].workflow_id, 125).unwrap();
1215        assert!(store.workflows(Path::new("/folder")).unwrap().is_empty());
1216        assert_eq!(
1217            store
1218                .workflow_trace(workflows[0].workflow_id, None, 10)
1219                .unwrap()
1220                .spans
1221                .len(),
1222            1
1223        );
1224    }
1225
1226    #[test]
1227    fn migration_from_version_five_preserves_agent_identity_and_role_order() {
1228        let connection = Connection::open_in_memory().unwrap();
1229        for migration in &MIGRATIONS[..5] {
1230            connection.execute_batch(migration).unwrap();
1231        }
1232        connection.pragma_update(None, "user_version", 5).unwrap();
1233        connection.execute_batch(
1234            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Session', 123);
1235             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Agents', 'agents');
1236             INSERT INTO agents (id, workflow_id, role) VALUES (7, 1, 'Reviewer'), (8, 1, 'Implementer');"
1237        ).unwrap();
1238        let store = Store::with_connection(connection).unwrap();
1239        let workflows = store.workflows(Path::new("/folder")).unwrap();
1240        assert_eq!(workflows[0].kind, crate::WorkflowKind::Agents);
1241        assert_eq!(
1242            workflows[0]
1243                .agents
1244                .iter()
1245                .map(|agent| (agent.agent_id.0, agent.role.as_str()))
1246                .collect::<Vec<_>>(),
1247            [(7, "Reviewer"), (8, "Implementer")]
1248        );
1249        assert_eq!(
1250            store
1251                .trace_summary(workflows[0].workflow_id)
1252                .unwrap()
1253                .span_count,
1254            0
1255        );
1256        assert_eq!(
1257            usize::try_from(schema_version(&store.connection)).unwrap(),
1258            MIGRATIONS.len()
1259        );
1260    }
1261
1262    #[test]
1263    fn migration_from_version_six_keeps_trace_ids_and_accepts_workflow_events() {
1264        let connection = Connection::open_in_memory().unwrap();
1265        for migration in &MIGRATIONS[..6] {
1266            connection.execute_batch(migration).unwrap();
1267        }
1268        connection.pragma_update(None, "user_version", 6).unwrap();
1269        connection.execute_batch(
1270            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Work', 123);
1271             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Shell', 'terminal');
1272             INSERT INTO trace_events (id, workflow_id, timestamp, kind, message)
1273             VALUES (80, 1, 123, 'processStarted', 'Earlier event');
1274             DELETE FROM trace_events WHERE id = 80;"
1275        ).unwrap();
1276        let store = Store::with_connection(connection).unwrap();
1277        store
1278            .connection
1279            .execute(
1280                "INSERT INTO trace_events (workflow_id, timestamp, kind, message)
1281             VALUES (1, 124, 'workflowEvent', 'Stage started')",
1282                [],
1283            )
1284            .unwrap();
1285        assert_eq!(store.connection.last_insert_rowid(), 81);
1286        assert_eq!(store.workflows(Path::new("/folder")).unwrap().len(), 1);
1287    }
1288
1289    #[test]
1290    fn migration_preserves_idle_shell_and_agent_terminal_ownership() {
1291        let connection = Connection::open_in_memory().unwrap();
1292        for migration in &MIGRATIONS[..11] {
1293            connection.execute_batch(migration).unwrap();
1294        }
1295        connection.pragma_update(None, "user_version", 11).unwrap();
1296        connection.execute_batch(
1297            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Work', 1);
1298             INSERT INTO workflows (id, session_id, name, kind) VALUES
1299                (1, 1, 'Shell', 'terminal'), (2, 1, 'Agent', 'single_agent'), (3, 1, 'Run', 'agents');
1300             INSERT INTO trace_lanes (id, workflow_id, lane_key, name, is_agent) VALUES
1301                (1, 2, 'terminal', 'Terminal', 0), (2, 2, 'agent', 'Agent', 1), (3, 3, 'agent:8', 'Worker', 1);
1302             INSERT INTO trace_spans (lane_id, title, started_at, status, terminal_id) VALUES
1303                (1, 'Shell', 1, 'stopped', 20), (2, 'Agent', 2, 'exited', 21), (3, 'Worker', 2, 'exited', 31);
1304             INSERT INTO trace_events (workflow_id, timestamp, kind, message, terminal_id, byte_offset)
1305                VALUES (1, 1, 'processStarted', 'Idle shell', 10, 0);"
1306        ).unwrap();
1307        let store = Store::with_connection(connection).unwrap();
1308        let history = |id| store.terminal_history(crate::WorkflowId(id)).unwrap();
1309        assert_eq!(history(1)[0].terminal_id.value(), 10);
1310        assert_eq!(history(2).len(), 1);
1311        assert_eq!(history(2)[0].terminal_id.value(), 21);
1312        assert_eq!(history(3)[0].agent_id, Some(crate::AgentId(8)));
1313    }
1314
1315    #[test]
1316    fn migration_from_version_nine_preserves_process_spans_and_terminal_anchors() {
1317        let connection = Connection::open_in_memory().unwrap();
1318        for migration in &MIGRATIONS[..9] {
1319            connection.execute_batch(migration).unwrap();
1320        }
1321        connection.pragma_update(None, "user_version", 9).unwrap();
1322        connection.execute_batch(
1323            "INSERT INTO sessions (id, folder, name, started_at) VALUES (1, '/folder', 'Work', 100);
1324             INSERT INTO workflows (id, session_id, name, kind) VALUES (1, 1, 'Adversarial', 'agents');
1325             INSERT INTO trace_lanes (id, workflow_id, lane_key, name, is_agent, role, harness)
1326             VALUES (1, 1, 'agent:1', 'Implementer', 1, 'implementer', 'pi');
1327             INSERT INTO trace_spans (id, lane_id, title, started_at, ended_at, status, terminal_id, run_generation)
1328             VALUES (1, 1, 'Implement', 100, 200, 'exited', 99, 1);
1329             INSERT INTO trace_events (id, workflow_id, span_id, timestamp, kind, message, terminal_id, byte_offset, boundary_sizes)
1330             VALUES (1, 1, 1, 100, 'processStarted', 'Process started.', 99, 0, X''),
1331                    (2, 1, 1, 200, 'processExited', 'Process exited with code 0.', 99, 42, X'');"
1332        ).unwrap();
1333        let store = Store::with_connection(connection).unwrap();
1334        let page = store
1335            .workflow_trace(crate::WorkflowId(1), None, 10)
1336            .unwrap();
1337        let span = &page.spans[0];
1338        assert_eq!(span.title, "Implement");
1339        assert_eq!((span.started_at, span.ended_at), (100, Some(200)));
1340        assert_eq!(span.status, crate::TraceSpanStatus::Exited);
1341        let events = store.trace_events(span.span_id, None, 10).unwrap().events;
1342        assert_eq!(events.len(), 2);
1343        assert_eq!(events[1].kind, crate::TraceEventKind::ProcessExited);
1344        assert_eq!(
1345            events[1].anchor,
1346            Some(crate::TraceAnchor {
1347                terminal_id: crate::TerminalId::from_value(99),
1348                byte_offset: 42,
1349                boundary_sizes: Some(vec![]),
1350            })
1351        );
1352    }
1353
1354    #[test]
1355    fn a_newer_schema_is_rejected() {
1356        let directory = tempfile::tempdir().expect("a temporary directory should be available");
1357        let path = directory.path().join("twine.db");
1358        let connection = Connection::open(&path).expect("the database should open");
1359        connection
1360            .pragma_update(None, "user_version", 99)
1361            .expect("the schema version should be writable");
1362        drop(connection);
1363
1364        assert!(matches!(
1365            Store::open(&path),
1366            Err(StoreError::UnsupportedSchemaVersion { version: 99 })
1367        ));
1368    }
1369
1370    #[test]
1371    fn recent_folders_are_ordered_capped_and_track_the_open_folder() {
1372        let mut store = Store::open_in_memory().expect("the store should open");
1373        store
1374            .record_folder_opened("/a", 5_000, 2)
1375            .expect("a should be recorded");
1376        // An earlier clock reading still sorts after the folder opened before it.
1377        store
1378            .record_folder_opened("/b", 1_000, 2)
1379            .expect("b should be recorded");
1380        store
1381            .record_folder_opened("/c", 1_000, 2)
1382            .expect("c should be recorded");
1383
1384        assert_eq!(
1385            store.recent_folders().expect("folders should load"),
1386            [
1387                StoredFolder {
1388                    path: PathBuf::from("/c"),
1389                    is_open: true,
1390                },
1391                StoredFolder {
1392                    path: PathBuf::from("/b"),
1393                    is_open: false,
1394                },
1395            ]
1396        );
1397
1398        store
1399            .record_folder_opened("/b", 1_000, 2)
1400            .expect("b should be recorded again");
1401        store
1402            .record_folder_closed()
1403            .expect("closing should be recorded");
1404        let remaining = store
1405            .remove_recent_folder("/c")
1406            .expect("c should be removed");
1407        assert_eq!(
1408            remaining,
1409            [StoredFolder {
1410                path: PathBuf::from("/b"),
1411                is_open: false,
1412            }]
1413        );
1414        assert_eq!(
1415            store.recent_folders().expect("folders should load"),
1416            remaining
1417        );
1418    }
1419}
handshake_core/store.rs

handshake_core/
store.rs
