lib.rs - CodeWhale

lib.rs

1	//! Persistent state management for conversation threads, messages, and jobs.
2	//!
3	//! The [`StateStore`] is the primary entry point, backed by a SQLite database and an
4	//! append-only JSONL session index file. It provides CRUD operations for:
5	//!
6	//! - Threads — conversation metadata, archival, and session indexing.
7	//! - Messages — append-only message storage with tree-structured branching.
8	//! - Checkpoints — named state snapshots for restoring conversation progress.
9	//! - Jobs — background task tracking with status and progress.
10	//! - Dynamic tools — per-thread tool registrations.
11
12	use std::collections::HashMap;
13	use std::fs::{self, OpenOptions};
14	use std::io::{BufRead, BufReader, Write};
15	use std::path::{Path, PathBuf};
16	use std::sync::{Arc, LazyLock, Mutex, MutexGuard};
17
18	/// Serializes all `session_index.jsonl` read/append/compact/rename operations so
19	/// concurrent `StateStore` clones cannot interleave an append with a compaction
20	/// rename and silently drop entries.
21	static SESSION_INDEX_LOCK: LazyLock<Mutex<()>> = LazyLock::new(\|\| Mutex::new(()));
22
23	use anyhow::{Context, Result};
24	use chrono::Utc;
25	use codewhale_paths::{CODEWHALE_APP_DIR, LEGACY_APP_DIR, codewhale_home_override};
26	use rusqlite::{Connection, OptionalExtension, params};
27	use serde::{Deserialize, Serialize};
28	use serde_json::Value;
29
30	// Re-export protocol's ThreadStatus so callers in the state crate and
31	// external consumers (e.g. core) can reference a single canonical definition.
32	pub use codewhale_protocol::ThreadStatus;
33
34	/// Indicates how a session was initiated.
35	///
36	/// Serialized as lowercase snake_case strings.
37	#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
38	#[serde(rename_all = "snake_case")]
39	pub enum SessionSource {
40	/// Started by a user interacting with the CLI.
41	Interactive,
42	/// Resumed from a previously persisted session.
43	Resume,
44	/// Created by forking an existing conversation at a specific message.
45	Fork,
46	/// Initiated programmatically via the API.
47	Api,
48	/// Source is unknown or unspecified.
49	Unknown,
50	}
51
52	/// Metadata for a persisted conversation thread.
53	///
54	/// Each thread represents a single conversation session and stores its
55	/// configuration, git context, and current status.
56	#[derive(Debug, Clone, Serialize, Deserialize)]
57	pub struct ThreadMetadata {
58	/// Unique identifier for this thread.
59	pub id: String,
60	/// Optional filesystem path to the rollout (JSONL transcript) file.
61	pub rollout_path: Option<PathBuf>,
62	/// Short preview or summary of the thread content.
63	pub preview: String,
64	/// Whether this thread is ephemeral (not persisted long-term).
65	pub ephemeral: bool,
66	/// Identifier of the model provider used for this thread (e.g. `"openai"`).
67	pub model_provider: String,
68	/// Unix timestamp (seconds) when the thread was created.
69	pub created_at: i64,
70	/// Unix timestamp (seconds) of the most recent update to the thread.
71	pub updated_at: i64,
72	/// Current lifecycle status of the thread.
73	pub status: ThreadStatus,
74	/// Optional filesystem path associated with the thread working context.
75	pub path: Option<PathBuf>,
76	/// Working directory that was active when the thread was created.
77	pub cwd: PathBuf,
78	/// Version of the CLI that created this thread.
79	pub cli_version: String,
80	/// How this session was initiated.
81	pub source: SessionSource,
82	/// User-assigned display name for the thread.
83	pub name: Option<String>,
84	/// Serialized sandbox policy applied to this thread, if any.
85	pub sandbox_policy: Option<String>,
86	/// Approval mode configured for tool calls in this thread.
87	pub approval_mode: Option<String>,
88	/// Whether the thread has been archived.
89	pub archived: bool,
90	/// Unix timestamp (seconds) when the thread was archived, or `None` if not archived.
91	pub archived_at: Option<i64>,
92	/// Git commit SHA of the working tree when the thread was created.
93	pub git_sha: Option<String>,
94	/// Git branch checked out when the thread was created.
95	pub git_branch: Option<String>,
96	/// URL of the git remote origin, if available.
97	pub git_origin_url: Option<String>,
98	/// Memory mode configured for this thread (e.g. `"local"`, `"remote"`).
99	pub memory_mode: Option<String>,
100	/// ID of the current leaf message in the conversation tree.
101	pub current_leaf_id: Option<i64>,
102	}
103
104	/// A dynamically registered tool associated with a thread.
105	#[derive(Debug, Clone, Serialize, Deserialize)]
106	pub struct DynamicToolRecord {
107	/// Ordinal position of this tool in the thread tool list.
108	pub position: i64,
109	/// Unique name identifying the tool.
110	pub name: String,
111	/// Human-readable description of what the tool does.
112	pub description: Option<String>,
113	/// JSON Schema describing the tool input parameters.
114	pub input_schema: Value,
115	}
116
117	/// A single message entry in a conversation thread.
118	///
119	/// Messages form a tree structure via [`parent_entry_id`](Self::parent_entry_id),
120	/// enabling conversation branching and forking.
121	#[derive(Debug, Clone, Serialize, Deserialize)]
122	pub struct MessageRecord {
123	/// Auto-incremented unique identifier for this message.
124	pub id: i64,
125	/// ID of the thread this message belongs to.
126	pub thread_id: String,
127	/// Role of the message sender (e.g. `"user"`, `"assistant"`, `"system"`).
128	pub role: String,
129	/// Text content of the message.
130	pub content: String,
131	/// Optional structured item payload (tool calls, tool results, etc.).
132	pub item: Option<Value>,
133	/// Unix timestamp (seconds) when the message was created.
134	pub created_at: i64,
135	/// ID of the parent message, forming a tree structure. `None` for root messages.
136	pub parent_entry_id: Option<i64>,
137	}
138
139	/// A named checkpoint capturing the state of a thread at a point in time.
140	#[derive(Debug, Clone, Serialize, Deserialize)]
141	pub struct CheckpointRecord {
142	/// ID of the thread this checkpoint belongs to.
143	pub thread_id: String,
144	/// Unique identifier for this checkpoint within its thread.
145	pub checkpoint_id: String,
146	/// Serialized state snapshot stored as a JSON value.
147	pub state: Value,
148	/// Unix timestamp (seconds) when the checkpoint was created or last updated.
149	pub created_at: i64,
150	}
151
152	/// Status of a background job.
153	///
154	/// Serialized as lowercase snake_case strings.
155	#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
156	#[serde(rename_all = "snake_case")]
157	pub enum JobStateStatus {
158	/// Job is waiting to be executed.
159	Queued,
160	/// Job is currently executing.
161	Running,
162	/// Job has been temporarily paused.
163	Paused,
164	/// Job has finished successfully.
165	Completed,
166	/// Job has failed with an error.
167	Failed,
168	/// Job was cancelled before completion.
169	Cancelled,
170	}
171
172	/// Persisted state of a background job.
173	#[derive(Debug, Clone, Serialize, Deserialize)]
174	pub struct JobStateRecord {
175	/// Unique identifier for the job.
176	pub id: String,
177	/// Human-readable name describing the job.
178	pub name: String,
179	/// Current lifecycle status of the job.
180	pub status: JobStateStatus,
181	/// Completion progress as a percentage (0--100), if available.
182	pub progress: Option<u8>,
183	/// Optional detail message providing additional status information.
184	pub detail: Option<String>,
185	/// Unix timestamp (seconds) when the job was created.
186	pub created_at: i64,
187	/// Unix timestamp (seconds) of the most recent status update.
188	pub updated_at: i64,
189	}
190
191	/// Persisted lifecycle status for a thread goal.
192	#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
193	#[serde(rename_all = "snake_case")]
194	pub enum ThreadGoalStatus {
195	/// Goal is active and should continue receiving work.
196	Active,
197	/// Goal is paused by the user.
198	Paused,
199	/// Goal is blocked and cannot make meaningful progress.
200	Blocked,
201	/// Goal stopped because account/service usage limits were reached.
202	UsageLimited,
203	/// Goal stopped because its explicit token budget was reached.
204	BudgetLimited,
205	/// Goal has been completed.
206	Complete,
207	}
208
209	/// Persisted goal state attached to a thread.
210	#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
211	pub struct ThreadGoalRecord {
212	/// Thread this goal belongs to.
213	pub thread_id: String,
214	/// Stable identifier for this goal revision.
215	pub goal_id: String,
216	/// User-visible objective.
217	pub objective: String,
218	/// Current lifecycle status.
219	pub status: ThreadGoalStatus,
220	/// Optional token budget requested by the user.
221	pub token_budget: Option<i64>,
222	/// Tokens consumed while pursuing the goal.
223	pub tokens_used: i64,
224	/// Elapsed wall-clock work time in seconds.
225	pub time_used_seconds: i64,
226	/// Durable continuation passes dispatched for this objective.
227	pub continuation_count: i64,
228	/// Unix timestamp (seconds) when the goal was created.
229	pub created_at: i64,
230	/// Unix timestamp (seconds) when the goal was last updated.
231	pub updated_at: i64,
232	#[serde(default, skip_serializing_if = "Option::is_none")]
233	pub last_gap_fingerprint: Option<String>,
234	#[serde(default)]
235	pub repeated_gap_count: u32,
236	#[serde(default, skip_serializing_if = "Option::is_none")]
237	pub last_gap_pass: Option<u32>,
238	#[serde(default, skip_serializing_if = "Option::is_none")]
239	pub pause_reason: Option<codewhale_protocol::GoalPauseReason>,
240	}
241
242	/// Filters for listing conversation threads.
243	#[derive(Debug, Clone)]
244	pub struct ThreadListFilters {
245	/// Whether to include archived threads in the results.
246	pub include_archived: bool,
247	/// Maximum number of threads to return. Defaults to 50.
248	pub limit: Option<usize>,
249	}
250
251	impl Default for ThreadListFilters {
252	fn default() -> Self {
253	Self {
254	include_archived: false,
255	limit: Some(50),
256	}
257	}
258	}
259
260	#[derive(Debug, Clone, Serialize, Deserialize)]
261	struct SessionIndexEntry {
262	thread_id: String,
263	thread_name: Option<String>,
264	updated_at: i64,
265	rollout_path: Option<PathBuf>,
266	}
267
268	/// Rewrite the session index once the append-only log grows large enough that
269	/// full-file scans become costly. Lookups already dedupe by thread id, so
270	/// compaction keeps only the latest entry per thread.
271	fn session_index_compact_line_threshold() -> usize {
272	if cfg!(test) { 5 } else { 5_000 }
273	}
274
275	/// Persistent storage for conversation threads, messages, checkpoints, and jobs.
276	///
277	/// Backed by a SQLite database and an append-only JSONL session index file.
278	/// The database schema is automatically initialized and migrated on [`open`](Self::open).
279	#[derive(Debug, Clone)]
280	pub struct StateStore {
281	db_path: PathBuf,
282	session_index_path: PathBuf,
283	// Single long-lived connection shared by all clones. SQLite pragmas are
284	// per-connection, so opening once in `open` and applying them there keeps
285	// every operation consistent without re-opening the database per call.
286	conn: Arc<Mutex<Connection>>,
287	}
288
289	impl StateStore {
290	/// Open (or create) a state store at the given database path.
291	///
292	/// If `path` is `None`, the default location (`~/.codewhale/state.db`, with
293	/// `~/.deepseek/state.db` as a legacy fallback) is used.
294	/// The database schema is created automatically if it does not exist.
295	pub fn open(path: Option<PathBuf>) -> Result<Self> {
296	let db_path = path.unwrap_or_else(default_state_db_path);
297	let session_index_path = db_path
298	.parent()
299	.unwrap_or_else(\|\| Path::new("."))
300	.join("session_index.jsonl");
301	if let Some(parent) = db_path.parent() {
302	fs::create_dir_all(parent).with_context(\|\| {
303	format!("failed to create state directory {}", parent.display())
304	})?;
305	}
306	let conn = Connection::open(&db_path)
307	.with_context(\|\| format!("failed to open state db {}", db_path.display()))?;
308	Self::configure_connection(&conn, &db_path)?;
309	Self::init_schema(&conn)?;
310	Ok(Self {
311	db_path,
312	session_index_path,
313	conn: Arc::new(Mutex::new(conn)),
314	})
315	}
316
317	/// Apply connection-level SQLite settings that must hold for every open.
318	///
319	/// Enables WAL so readers and writers from concurrent CodeWhale processes
320	/// do not block each other as aggressively as the default rollback journal,
321	/// and sets a multi-second busy timeout so a second process retries on
322	/// `SQLITE_BUSY` instead of failing immediately (issue #4734).
323	fn configure_connection(conn: &Connection, db_path: &Path) -> Result<()> {
324	// Install our wait policy before touching database-level settings or
325	// schema. Connection::open may currently provide a dependency default,
326	// but StateStore must not rely on that incidental behavior.
327	conn.busy_timeout(std::time::Duration::from_secs(5))
328	.with_context(\|\| format!("failed to set busy_timeout for {}", db_path.display()))?;
329	conn.pragma_update(None, "foreign_keys", "ON")
330	.with_context(\|\| format!("failed to enable foreign keys for {}", db_path.display()))?;
331
332	// WAL persists in the database header, so established stores need no
333	// write-like journal transition on every process start. Fresh or
334	// explicitly downgraded stores still transition once, and we verify
335	// SQLite accepted the requested mode instead of silently retaining the
336	// previous one (for example on an unsupported VFS).
337	let journal_mode: String = conn
338	.pragma_query_value(None, "journal_mode", \|row\| row.get(0))
339	.with_context(\|\| format!("failed to read journal mode for {}", db_path.display()))?;
340	if !journal_mode.eq_ignore_ascii_case("wal") {
341	let configured_mode: String = conn
342	.pragma_update_and_check(None, "journal_mode", "WAL", \|row\| row.get(0))
343	.with_context(\|\| format!("failed to enable WAL for {}", db_path.display()))?;
344	if !configured_mode.eq_ignore_ascii_case("wal") {
345	anyhow::bail!(
346	"failed to enable WAL for {}: SQLite retained journal mode {configured_mode}",
347	db_path.display()
348	);
349	}
350	}
351	Ok(())
352	}
353
354	/// Returns the filesystem path of the underlying SQLite database.
355	pub fn db_path(&self) -> &Path {
356	&self.db_path
357	}
358
359	fn conn(&self) -> Result<MutexGuard<'_, Connection>> {
360	// Poisoning means a panic mid-operation; any open transaction was
361	// rolled back when it dropped, but surface the condition rather than
362	// silently continuing on a connection whose state we can't vouch for.
363	self.conn
364	.lock()
365	.map_err(\|_\| anyhow::anyhow!("state db connection mutex poisoned"))
366	}
367
368	fn init_schema(conn: &Connection) -> Result<()> {
369	let mut user_version: u32 = conn.query_row("PRAGMA user_version;", [], \|row\| row.get(0))?;
370	if user_version == 0 {
371	// Guard each ALTER: a database restored with a v0 header (or
372	// stamped by a racing process that crashed before setting
373	// user_version) can already carry these columns, and an
374	// unguarded ADD COLUMN aborts the whole open with
375	// "duplicate column name".
376	let add_parent_entry_id = if column_exists(conn, "messages", "parent_entry_id")? {
377	""
378	} else {
379	"ALTER TABLE messages ADD COLUMN parent_entry_id INTEGER NULL;"
380	};
381	let add_current_leaf_id = if column_exists(conn, "threads", "current_leaf_id")? {
382	""
383	} else {
384	"ALTER TABLE threads ADD COLUMN current_leaf_id INTEGER NULL;"
385	};
386	conn.execute_batch(&format!(
387	r#"
388	BEGIN;
389	CREATE TABLE IF NOT EXISTS threads (
390	id TEXT PRIMARY KEY,
391	rollout_path TEXT,
392	preview TEXT NOT NULL,
393	ephemeral INTEGER NOT NULL,
394	model_provider TEXT NOT NULL,
395	created_at INTEGER NOT NULL,
396	updated_at INTEGER NOT NULL,
397	status TEXT NOT NULL,
398	path TEXT,
399	cwd TEXT NOT NULL,
400	cli_version TEXT NOT NULL,
401	source TEXT NOT NULL,
402	title TEXT,
403	sandbox_policy TEXT,
404	approval_mode TEXT,
405	archived INTEGER NOT NULL DEFAULT 0,
406	archived_at INTEGER,
407	git_sha TEXT,
408	git_branch TEXT,
409	git_origin_url TEXT,
410	memory_mode TEXT
411	);
412	CREATE INDEX IF NOT EXISTS idx_threads_updated_at ON threads(updated_at DESC);
413	CREATE INDEX IF NOT EXISTS idx_threads_archived_at ON threads(archived_at DESC);
414	CREATE INDEX IF NOT EXISTS idx_threads_archived_updated ON threads(archived, updated_at DESC);
415
416	CREATE TABLE IF NOT EXISTS thread_dynamic_tools (
417	thread_id TEXT NOT NULL,
418	position INTEGER NOT NULL,
419	name TEXT NOT NULL,
420	description TEXT,
421	input_schema TEXT NOT NULL,
422	PRIMARY KEY (thread_id, position),
423	FOREIGN KEY(thread_id) REFERENCES threads(id) ON DELETE CASCADE
424	);
425
426	CREATE TABLE IF NOT EXISTS messages (
427	id INTEGER PRIMARY KEY AUTOINCREMENT,
428	thread_id TEXT NOT NULL,
429	role TEXT NOT NULL,
430	content TEXT NOT NULL,
431	item_json TEXT,
432	created_at INTEGER NOT NULL,
433	FOREIGN KEY(thread_id) REFERENCES threads(id) ON DELETE CASCADE
434	);
435	CREATE INDEX IF NOT EXISTS idx_messages_thread_created_at ON messages(thread_id, created_at ASC);
436
437	CREATE TABLE IF NOT EXISTS checkpoints (
438	thread_id TEXT NOT NULL,
439	checkpoint_id TEXT NOT NULL,
440	state_json TEXT NOT NULL,
441	created_at INTEGER NOT NULL,
442	PRIMARY KEY(thread_id, checkpoint_id),
443	FOREIGN KEY(thread_id) REFERENCES threads(id) ON DELETE CASCADE
444	);
445	CREATE INDEX IF NOT EXISTS idx_checkpoints_thread_created_at ON checkpoints(thread_id, created_at DESC);
446
447	CREATE TABLE IF NOT EXISTS jobs (
448	id TEXT PRIMARY KEY,
449	name TEXT NOT NULL,
450	status TEXT NOT NULL,
451	progress INTEGER,
452	detail TEXT,
453	created_at INTEGER NOT NULL,
454	updated_at INTEGER NOT NULL
455	);
456	CREATE INDEX IF NOT EXISTS idx_jobs_updated_at ON jobs(updated_at DESC);
457
458	-- Add parent_entry_id column, and set to last message before current message
459	{add_parent_entry_id}
460	UPDATE messages
461	SET parent_entry_id = (
462	SELECT m2.id
463	FROM messages m2
464	WHERE m2.thread_id = messages.thread_id
465	AND (
466	m2.created_at < messages.created_at
467	OR (
468	m2.created_at = messages.created_at
469	AND m2.id < messages.id
470	)
471	)
472	ORDER BY m2.created_at DESC, m2.id DESC
473	LIMIT 1
474	);
475	CREATE INDEX IF NOT EXISTS idx_messages_parent_entry_id ON messages(parent_entry_id);
476
477	-- Add current_leaf_id column, and set to last message in thread
478	{add_current_leaf_id}
479	UPDATE threads
480	SET current_leaf_id = (
481	SELECT m.id
482	FROM messages m
483	WHERE m.thread_id = threads.id
484	ORDER BY m.id DESC
485	LIMIT 1
486	);
487
488	PRAGMA user_version = 1;
489	COMMIT;
490	"#
491	))
492	.context("failed to initialize thread schema")?;
493	user_version = 1;
494	}
495	if user_version < 2 {
496	conn.execute_batch(
497	r#"
498	BEGIN;
499	CREATE TABLE IF NOT EXISTS workflow_runs (
500	id TEXT PRIMARY KEY,
501	workflow_id TEXT NOT NULL,
502	goal TEXT NOT NULL,
503	status TEXT NOT NULL,
504	input_hash TEXT,
505	started_at INTEGER NOT NULL,
506	completed_at INTEGER,
507	metadata_json TEXT NOT NULL DEFAULT '{}'
508	);
509	CREATE INDEX IF NOT EXISTS idx_workflow_runs_status_started_at
510	ON workflow_runs(status, started_at DESC);
511	CREATE INDEX IF NOT EXISTS idx_workflow_runs_workflow_started_at
512	ON workflow_runs(workflow_id, started_at DESC);
513
514	CREATE TABLE IF NOT EXISTS branch_runs (
515	id TEXT PRIMARY KEY,
516	workflow_run_id TEXT NOT NULL,
517	branch_id TEXT NOT NULL,
518	node_id TEXT NOT NULL,
519	status TEXT NOT NULL,
520	started_at INTEGER NOT NULL,
521	completed_at INTEGER,
522	result_json TEXT NOT NULL DEFAULT '{}',
523	FOREIGN KEY(workflow_run_id) REFERENCES workflow_runs(id) ON DELETE CASCADE
524	);
525	CREATE INDEX IF NOT EXISTS idx_branch_runs_workflow_run_id
526	ON branch_runs(workflow_run_id);
527	CREATE INDEX IF NOT EXISTS idx_branch_runs_branch_id
528	ON branch_runs(branch_id);
529
530	CREATE TABLE IF NOT EXISTS leaf_runs (
531	id TEXT PRIMARY KEY,
532	workflow_run_id TEXT NOT NULL,
533	branch_run_id TEXT,
534	leaf_id TEXT NOT NULL,
535	task_id TEXT NOT NULL,
536	input_hash TEXT,
537	status TEXT NOT NULL,
538	output_json TEXT NOT NULL DEFAULT '{}',
539	artifacts_json TEXT NOT NULL DEFAULT '[]',
540	started_at INTEGER NOT NULL,
541	completed_at INTEGER,
542	FOREIGN KEY(workflow_run_id) REFERENCES workflow_runs(id) ON DELETE CASCADE,
543	FOREIGN KEY(branch_run_id) REFERENCES branch_runs(id) ON DELETE SET NULL
544	);
545	CREATE INDEX IF NOT EXISTS idx_leaf_runs_workflow_run_id
546	ON leaf_runs(workflow_run_id);
547	CREATE INDEX IF NOT EXISTS idx_leaf_runs_replay_lookup
548	ON leaf_runs(workflow_run_id, leaf_id, input_hash);
549
550	CREATE TABLE IF NOT EXISTS control_node_runs (
551	id TEXT PRIMARY KEY,
552	workflow_run_id TEXT NOT NULL,
553	node_id TEXT NOT NULL,
554	kind TEXT NOT NULL,
555	status TEXT NOT NULL,
556	selected_children_json TEXT NOT NULL DEFAULT '[]',
557	result_json TEXT NOT NULL DEFAULT '{}',
558	started_at INTEGER NOT NULL,
559	completed_at INTEGER,
560	FOREIGN KEY(workflow_run_id) REFERENCES workflow_runs(id) ON DELETE CASCADE
561	);
562	CREATE INDEX IF NOT EXISTS idx_control_node_runs_workflow_run_id
563	ON control_node_runs(workflow_run_id);
564	CREATE INDEX IF NOT EXISTS idx_control_node_runs_node_id
565	ON control_node_runs(node_id);
566
567	CREATE TABLE IF NOT EXISTS teacher_candidates (
568	id TEXT PRIMARY KEY,
569	workflow_run_id TEXT NOT NULL,
570	control_node_run_id TEXT NOT NULL,
571	candidate_id TEXT NOT NULL,
572	branch_run_id TEXT,
573	score REAL,
574	passed INTEGER,
575	rationale_json TEXT NOT NULL DEFAULT '{}',
576	created_at INTEGER NOT NULL,
577	FOREIGN KEY(workflow_run_id) REFERENCES workflow_runs(id) ON DELETE CASCADE,
578	FOREIGN KEY(control_node_run_id) REFERENCES control_node_runs(id) ON DELETE CASCADE,
579	FOREIGN KEY(branch_run_id) REFERENCES branch_runs(id) ON DELETE SET NULL
580	);
581	CREATE INDEX IF NOT EXISTS idx_teacher_candidates_workflow_run_id
582	ON teacher_candidates(workflow_run_id);
583	CREATE INDEX IF NOT EXISTS idx_teacher_candidates_control_node_run_id
584	ON teacher_candidates(control_node_run_id);
585
586	PRAGMA user_version = 2;
587	COMMIT;
588	"#,
589	)
590	.context("failed to initialize workflow trace schema")?;
591	user_version = 2;
592	}
593	if user_version < 3 {
594	conn.execute_batch(
595	r#"
596	BEGIN;
597	CREATE TABLE IF NOT EXISTS thread_goals (
598	thread_id TEXT PRIMARY KEY NOT NULL,
599	goal_id TEXT NOT NULL,
600	objective TEXT NOT NULL,
601	status TEXT NOT NULL CHECK(status IN (
602	'active',
603	'paused',
604	'blocked',
605	'usage_limited',
606	'budget_limited',
607	'complete'
608	)),
609	token_budget INTEGER,
610	tokens_used INTEGER NOT NULL DEFAULT 0,
611	time_used_seconds INTEGER NOT NULL DEFAULT 0,
612	created_at INTEGER NOT NULL,
613	updated_at INTEGER NOT NULL,
614	FOREIGN KEY(thread_id) REFERENCES threads(id) ON DELETE CASCADE
615	);
616
617	PRAGMA user_version = 3;
618	COMMIT;
619	"#,
620	)
621	.context("failed to initialize thread goal schema")?;
622	user_version = 3;
623	}
624	if user_version < 4 {
625	// Same restore/race guard as the v0 block: the column may
626	// already exist even though the header predates version 4.
627	let add_continuation_count = if column_exists(
628	conn,
629	"thread_goals",
630	"continuation_count",
631	)? {
632	""
633	} else {
634	"ALTER TABLE thread_goals\n ADD COLUMN continuation_count INTEGER NOT NULL DEFAULT 0;"
635	};
636	conn.execute_batch(&format!(
637	r#"
638	BEGIN;
639	{add_continuation_count}
640
641	PRAGMA user_version = 4;
642	COMMIT;
643	"#
644	))
645	.context("failed to initialize thread goal continuation schema")?;
646	user_version = 4;
647	}
648	if user_version < 5 {
649	let mut additions = String::new();
650	for (column, definition) in [
651	("last_gap_fingerprint", "TEXT"),
652	("repeated_gap_count", "INTEGER NOT NULL DEFAULT 0"),
653	("last_gap_pass", "INTEGER"),
654	("pause_reason", "TEXT"),
655	] {
656	if !column_exists(conn, "thread_goals", column)? {
657	additions.push_str(&format!(
658	"ALTER TABLE thread_goals ADD COLUMN {column} {definition};\n"
659	));
660	}
661	}
662	conn.execute_batch(&format!(
663	"BEGIN; {additions} PRAGMA user_version = 5; COMMIT;"
664	))
665	.context("failed to initialize durable goal stall schema")?;
666	}
667	Ok(())
668	}
669
670	/// Insert or update thread metadata.
671	///
672	/// This does not update `current_leaf_id`; use [`append_message`](Self::append_message)
673	/// or [`set_current_leaf_id`](Self::set_current_leaf_id) for that.
674	pub fn upsert_thread(&self, thread: &ThreadMetadata) -> Result<()> {
675	let conn = self.conn()?;
676	conn.execute(
677	r#"
678	INSERT INTO threads (
679	id, rollout_path, preview, ephemeral, model_provider, created_at, updated_at, status, path, cwd,
680	cli_version, source, title, sandbox_policy, approval_mode, archived, archived_at,
681	git_sha, git_branch, git_origin_url, memory_mode
682	) VALUES (
683	?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10,
684	?11, ?12, ?13, ?14, ?15, ?16, ?17,
685	?18, ?19, ?20, ?21
686	)
687	ON CONFLICT(id) DO UPDATE SET
688	rollout_path=excluded.rollout_path,
689	preview=excluded.preview,
690	ephemeral=excluded.ephemeral,
691	model_provider=excluded.model_provider,
692	created_at=excluded.created_at,
693	updated_at=excluded.updated_at,
694	status=excluded.status,
695	path=excluded.path,
696	cwd=excluded.cwd,
697	cli_version=excluded.cli_version,
698	source=excluded.source,
699	title=excluded.title,
700	sandbox_policy=excluded.sandbox_policy,
701	approval_mode=excluded.approval_mode,
702	archived=excluded.archived,
703	archived_at=excluded.archived_at,
704	git_sha=excluded.git_sha,
705	git_branch=excluded.git_branch,
706	git_origin_url=excluded.git_origin_url,
707	memory_mode=excluded.memory_mode
708	"#,
709	params![
710	thread.id,
711	path_to_opt_string(thread.rollout_path.as_deref()),
712	thread.preview,
713	bool_to_i64(thread.ephemeral),
714	thread.model_provider,
715	thread.created_at,
716	thread.updated_at,
717	thread_status_to_str(&thread.status),
718	path_to_opt_string(thread.path.as_deref()),
719	thread.cwd.display().to_string(),
720	thread.cli_version,
721	session_source_to_str(&thread.source),
722	thread.name,
723	thread.sandbox_policy,
724	thread.approval_mode,
725	bool_to_i64(thread.archived),
726	thread.archived_at,
727	thread.git_sha,
728	thread.git_branch,
729	thread.git_origin_url,
730	thread.memory_mode,
731	],
732	)
733	.context("failed to upsert thread metadata")?;
734
735	self.append_thread_name(
736	&thread.id,
737	thread.name.clone(),
738	thread.updated_at,
739	thread.rollout_path.clone(),
740	)?;
741	Ok(())
742	}
743
744	/// Retrieve a single thread by its ID.
745	///
746	/// Returns `None` if no thread with the given ID exists.
747	pub fn get_thread(&self, id: &str) -> Result<Option<ThreadMetadata>> {
748	let conn = self.conn()?;
749	conn.query_row(
750	r#"
751	SELECT id, rollout_path, preview, ephemeral, model_provider, created_at, updated_at, status, path, cwd,
752	cli_version, source, title, sandbox_policy, approval_mode, archived, archived_at,
753	git_sha, git_branch, git_origin_url, memory_mode, current_leaf_id
754	FROM threads
755	WHERE id = ?1
756	"#,
757	params![id],
758	row_to_thread,
759	)
760	.optional()
761	.context("failed to read thread")
762	}
763
764	/// List threads ordered by most recently updated.
765	///
766	/// Use [`ThreadListFilters`] to control whether archived threads are included
767	/// and the maximum number of results returned.
768	pub fn list_threads(&self, filters: ThreadListFilters) -> Result<Vec<ThreadMetadata>> {
769	let conn = self.conn()?;
770	let sql = if filters.include_archived {
771	"SELECT id, rollout_path, preview, ephemeral, model_provider, created_at, updated_at, status, path, cwd, cli_version, source, title, sandbox_policy, approval_mode, archived, archived_at, git_sha, git_branch, git_origin_url, memory_mode, current_leaf_id FROM threads ORDER BY updated_at DESC LIMIT ?1"
772	} else {
773	"SELECT id, rollout_path, preview, ephemeral, model_provider, created_at, updated_at, status, path, cwd, cli_version, source, title, sandbox_policy, approval_mode, archived, archived_at, git_sha, git_branch, git_origin_url, memory_mode, current_leaf_id FROM threads WHERE archived = 0 ORDER BY updated_at DESC LIMIT ?1"
774	};
775
776	let mut stmt = conn.prepare(sql).context("failed to prepare list query")?;
777	let limit = i64::try_from(filters.limit.unwrap_or(50)).unwrap_or(50);
778	let mut rows = stmt
779	.query(params![limit])
780	.context("failed to query threads")?;
781	let mut out = Vec::new();
782	while let Some(row) = rows.next().context("failed to iterate thread rows")? {
783	out.push(row_to_thread(row)?);
784	}
785	Ok(out)
786	}
787
788	/// Archive a thread, setting its status to [`ThreadStatus::Archived`] and
789	/// recording the current timestamp.
790	pub fn mark_archived(&self, id: &str) -> Result<()> {
791	let conn = self.conn()?;
792	conn.execute(
793	"UPDATE threads SET archived = 1, archived_at = ?2, status = ?3 WHERE id = ?1",
794	params![
795	id,
796	Utc::now().timestamp(),
797	thread_status_to_str(&ThreadStatus::Archived)
798	],
799	)
800	.context("failed to archive thread")?;
801	Ok(())
802	}
803
804	/// Unarchive a thread, removing the archived flag and clearing `archived_at`.
805	pub fn mark_unarchived(&self, id: &str) -> Result<()> {
806	let conn = self.conn()?;
807	conn.execute(
808	"UPDATE threads SET archived = 0, archived_at = NULL, status = CASE WHEN status = ?2 THEN ?3 ELSE status END WHERE id = ?1",
809	params![
810	id,
811	thread_status_to_str(&ThreadStatus::Archived),
812	thread_status_to_str(&ThreadStatus::Idle),
813	],
814	)
815	.context("failed to unarchive thread")?;
816	Ok(())
817	}
818
819	/// Permanently delete a thread and all of its associated data
820	/// (messages, checkpoints, dynamic tools) via cascading foreign keys.
821	pub fn delete_thread(&self, id: &str) -> Result<()> {
822	let conn = self.conn()?;
823	conn.execute("DELETE FROM threads WHERE id = ?1", params![id])
824	.context("failed to delete thread")?;
825	Ok(())
826	}
827
828	/// Insert or replace the persisted goal for a thread.
829	pub fn upsert_thread_goal(&self, goal: &ThreadGoalRecord) -> Result<()> {
830	codewhale_protocol::validate_goal_stall_state(
831	goal.last_gap_fingerprint.as_deref(),
832	goal.repeated_gap_count,
833	goal.last_gap_pass,
834	u32::try_from(goal.continuation_count.max(0)).unwrap_or(u32::MAX),
835	)
836	.map_err(anyhow::Error::msg)?;
837	let pause_reason = goal
838	.pause_reason
839	.map(\|reason\| serde_json::to_string(&reason))
840	.transpose()?;
841	let conn = self.conn()?;
842	let exists: Option<i64> = conn
843	.query_row(
844	"SELECT 1 FROM threads WHERE id = ?1",
845	params![goal.thread_id],
846	\|row\| row.get(0),
847	)
848	.optional()
849	.context("failed to verify thread before saving goal")?;
850	if exists.is_none() {
851	anyhow::bail!("thread {} not found", goal.thread_id);
852	}
853
854	conn.execute(
855	r#"
856	INSERT INTO thread_goals (
857	thread_id, goal_id, objective, status, token_budget, tokens_used,
858	time_used_seconds, continuation_count, created_at, updated_at,
859	last_gap_fingerprint, repeated_gap_count, last_gap_pass, pause_reason
860	) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)
861	ON CONFLICT(thread_id) DO UPDATE SET
862	goal_id=excluded.goal_id,
863	objective=excluded.objective,
864	status=excluded.status,
865	token_budget=excluded.token_budget,
866	tokens_used=excluded.tokens_used,
867	time_used_seconds=excluded.time_used_seconds,
868	continuation_count=excluded.continuation_count,
869	created_at=excluded.created_at,
870	updated_at=excluded.updated_at,
871	last_gap_fingerprint=excluded.last_gap_fingerprint,
872	repeated_gap_count=excluded.repeated_gap_count,
873	last_gap_pass=excluded.last_gap_pass,
874	pause_reason=excluded.pause_reason
875	"#,
876	params![
877	goal.thread_id,
878	goal.goal_id,
879	goal.objective,
880	thread_goal_status_to_str(&goal.status),
881	goal.token_budget,
882	goal.tokens_used,
883	goal.time_used_seconds,
884	goal.continuation_count,
885	goal.created_at,
886	goal.updated_at,
887	goal.last_gap_fingerprint,
888	goal.repeated_gap_count,
889	goal.last_gap_pass,
890	pause_reason,
891	],
892	)
893	.context("failed to upsert thread goal")?;
894	Ok(())
895	}
896
897	/// Accrue additional token and wall-clock usage onto a thread's persisted goal.
898	///
899	/// This is the durable, additive accounting path for the persistent goal loop: it
900	/// increments `tokens_used` and `time_used_seconds` in a single atomic SQL `UPDATE`
901	/// (`col = col + ?`) so concurrent accruals do not race a read-modify-write. The
902	/// goal's `updated_at` is advanced to the larger of its current value and `now`,
903	/// keeping the timestamp monotonic even if a stale `now` is supplied.
904	///
905	/// `token_delta` and `time_delta_seconds` are added on the database side; callers
906	/// should pass non-negative deltas (negative values are accepted and will decrement,
907	/// which is intentionally left to the caller's discretion).
908	///
909	/// Returns the updated [`ThreadGoalRecord`], or `Ok(None)` if the thread has no
910	/// persisted goal. Unlike [`upsert_thread_goal`](Self::upsert_thread_goal) this never
911	/// creates a goal row; it only accumulates onto an existing one.
912	pub fn record_thread_goal_usage(
913	&self,
914	thread_id: &str,
915	token_delta: i64,
916	time_delta_seconds: i64,
917	now: i64,
918	) -> Result<Option<ThreadGoalRecord>> {
919	let conn = self.conn()?;
920	let changed = conn
921	.execute(
922	r#"
923	UPDATE thread_goals
924	SET tokens_used = tokens_used + ?2,
925	time_used_seconds = time_used_seconds + ?3,
926	updated_at = MAX(updated_at, ?4)
927	WHERE thread_id = ?1
928	"#,
929	params![thread_id, token_delta, time_delta_seconds, now],
930	)
931	.context("failed to record thread goal usage")?;
932	if changed == 0 {
933	return Ok(None);
934	}
935	Self::read_thread_goal(&conn, thread_id)
936	}
937
938	/// Increment the durable cross-turn continuation counter for a thread goal.
939	///
940	/// The older TUI continuation guard is scoped to one engine turn. This
941	/// counter is intentionally persisted so a resumed goal loop can feed
942	/// `goal_loop::decide_continuation` with the true cross-turn count.
943	pub fn record_thread_goal_continuation(
944	&self,
945	thread_id: &str,
946	now: i64,
947	) -> Result<Option<ThreadGoalRecord>> {
948	let conn = self.conn()?;
949	let changed = conn
950	.execute(
951	r#"
952	UPDATE thread_goals
953	SET continuation_count = continuation_count + 1,
954	updated_at = MAX(updated_at, ?2)
955	WHERE thread_id = ?1
956	"#,
957	params![thread_id, now],
958	)
959	.context("failed to record thread goal continuation")?;
960	if changed == 0 {
961	return Ok(None);
962	}
963	Self::read_thread_goal(&conn, thread_id)
964	}
965
966	/// Retrieve the persisted goal for a thread.
967	pub fn get_thread_goal(&self, thread_id: &str) -> Result<Option<ThreadGoalRecord>> {
968	let conn = self.conn()?;
969	Self::read_thread_goal(&conn, thread_id)
970	}
971
972	/// Read a goal on an already-held connection. The `record_*` mutators call
973	/// this instead of [`Self::get_thread_goal`], which would re-lock the
974	/// connection mutex and self-deadlock.
975	fn read_thread_goal(conn: &Connection, thread_id: &str) -> Result<Option<ThreadGoalRecord>> {
976	conn.query_row(
977	r#"
978	SELECT thread_id, goal_id, objective, status, token_budget, tokens_used,
979	time_used_seconds, continuation_count, created_at, updated_at,
980	last_gap_fingerprint, repeated_gap_count, last_gap_pass, pause_reason
981	FROM thread_goals
982	WHERE thread_id = ?1
983	"#,
984	params![thread_id],
985	row_to_thread_goal,
986	)
987	.optional()
988	.context("failed to read thread goal")
989	}
990
991	/// Delete the persisted goal for a thread.
992	pub fn delete_thread_goal(&self, thread_id: &str) -> Result<bool> {
993	let conn = self.conn()?;
994	let changed = conn
995	.execute(
996	"DELETE FROM thread_goals WHERE thread_id = ?1",
997	params![thread_id],
998	)
999	.context("failed to delete thread goal")?;
1000	Ok(changed > 0)
1001	}
1002
1003	/// List all leaf messages in a thread.
1004	///
1005	/// A leaf message is one that has no other message referencing it as a parent.
1006	/// In a branching conversation tree, there may be multiple leaf messages.
1007	pub fn list_leaf_messages(&self, thread_id: &str) -> Result<Vec<MessageRecord>> {
1008	let conn = self.conn()?;
1009	let mut stmt = conn
1010	.prepare(
1011	r#"
1012	SELECT m1.id, m1.thread_id, m1.role, m1.content, m1.item_json, m1.created_at, m1.parent_entry_id
1013	FROM messages m1
1014	LEFT JOIN messages m2 ON m1.id = m2.parent_entry_id
1015	WHERE m1.thread_id = ?1 AND m2.id IS NULL
1016	"#,
1017	)
1018	.context("failed to prepare message listing query")?;
1019	let mut rows = stmt
1020	.query(params![thread_id])
1021	.with_context(\|\| format!("failed to list leaf messages for thread {thread_id}"))?;
1022	let mut out = Vec::new();
1023	while let Some(row) = rows.next().context("failed to iterate message rows")? {
1024	let item_json: Option<String> = row.get(4).context("failed to read item json")?;
1025	let item = item_json
1026	.as_deref()
1027	.map(serde_json::from_str)
1028	.transpose()
1029	.with_context(\|\| {
1030	format!("failed to parse message item json in thread {thread_id}")
1031	})?;
1032	out.push(MessageRecord {
1033	id: row.get(0).context("failed to read message id")?,
1034	thread_id: row.get(1).context("failed to read message thread id")?,
1035	role: row.get(2).context("failed to read message role")?,
1036	content: row.get(3).context("failed to read message content")?,
1037	item,
1038	created_at: row.get(5).context("failed to read message timestamp")?,
1039	parent_entry_id: row.get(6).context("failed to read parent entry id")?,
1040	});
1041	}
1042	Ok(out)
1043	}
1044
1045	/// Update the current leaf message pointer for a thread.
1046	///
1047	/// This controls which branch of the conversation tree is considered active
1048	/// when listing messages via [`list_messages`](Self::list_messages).
1049	pub fn set_current_leaf_id(&self, thread_id: &str, current_leaf_id: &str) -> Result<()> {
1050	let conn = self.conn()?;
1051	conn.execute(
1052	"UPDATE threads SET current_leaf_id = ?1 WHERE id = ?2",
1053	params![current_leaf_id, thread_id],
1054	)
1055	.context("failed to update thread current leaf id")?;
1056	Ok(())
1057	}
1058
1059	/// Replace the dynamic tools for a thread.
1060	///
1061	/// All existing dynamic tools for the thread are deleted and replaced with the
1062	/// provided list. The operation is performed within a transaction.
1063	pub fn persist_dynamic_tools(
1064	&self,
1065	thread_id: &str,
1066	tools: &[DynamicToolRecord],
1067	) -> Result<()> {
1068	let mut conn = self.conn()?;
1069	let tx = conn
1070	.transaction()
1071	.context("failed to begin dynamic tools transaction")?;
1072	tx.execute(
1073	"DELETE FROM thread_dynamic_tools WHERE thread_id = ?1",
1074	params![thread_id],
1075	)
1076	.context("failed to clear dynamic tools")?;
1077	for tool in tools {
1078	tx.execute(
1079	"INSERT INTO thread_dynamic_tools(thread_id, position, name, description, input_schema) VALUES (?1, ?2, ?3, ?4, ?5)",
1080	params![
1081	thread_id,
1082	tool.position,
1083	tool.name,
1084	tool.description,
1085	tool.input_schema.to_string()
1086	],
1087	)
1088	.with_context(\|\| format!("failed to persist dynamic tool {}", tool.name))?;
1089	}
1090	tx.commit().context("failed to commit dynamic tools")?;
1091	Ok(())
1092	}
1093
1094	/// Retrieve all dynamic tools registered for a thread, ordered by position.
1095	pub fn get_dynamic_tools(&self, thread_id: &str) -> Result<Vec<DynamicToolRecord>> {
1096	let conn = self.conn()?;
1097	let mut stmt = conn
1098	.prepare(
1099	"SELECT position, name, description, input_schema FROM thread_dynamic_tools WHERE thread_id = ?1 ORDER BY position ASC",
1100	)
1101	.context("failed to prepare get dynamic tools query")?;
1102	let mut rows = stmt
1103	.query(params![thread_id])
1104	.context("failed to query dynamic tools")?;
1105	let mut out = Vec::new();
1106	while let Some(row) = rows.next().context("failed to iterate dynamic tools")? {
1107	let input_schema_raw: String =
1108	row.get(3).context("failed to read tool input schema")?;
1109	let input_schema: Value =
1110	serde_json::from_str(&input_schema_raw).with_context(\|\| {
1111	format!("failed to parse input schema for dynamic tool in thread {thread_id}")
1112	})?;
1113	out.push(DynamicToolRecord {
1114	position: row.get(0).context("failed to read tool position")?,
1115	name: row.get(1).context("failed to read tool name")?,
1116	description: row.get(2).context("failed to read tool description")?,
1117	input_schema,
1118	});
1119	}
1120	Ok(out)
1121	}
1122
1123	/// Append a new message to a thread.
1124	///
1125	/// The message is linked to the thread's current leaf as its parent, and the
1126	/// thread's `current_leaf_id` is updated to the new message. Returns the ID
1127	/// of the newly created message.
1128	pub fn append_message(
1129	&self,
1130	thread_id: &str,
1131	role: &str,
1132	content: &str,
1133	item: Option<Value>,
1134	) -> Result<i64> {
1135	let mut conn = self.conn()?;
1136	let created_at = Utc::now().timestamp();
1137	let item_json = item
1138	.as_ref()
1139	.map(serde_json::to_string)
1140	.transpose()
1141	.context("failed to serialize message item payload")?;
1142
1143	let tx = conn
1144	.transaction()
1145	.context("failed to begin append message transaction")?;
1146
1147	let current_leaf_id: Option<i64> = tx
1148	.query_row(
1149	"SELECT current_leaf_id FROM threads WHERE id = ?1",
1150	params![thread_id],
1151	\|row\| row.get(0),
1152	)
1153	.with_context(\|\| {
1154	format!("failed to query thread current leaf id for thread {thread_id}")
1155	})?;
1156
1157	let next_leaf_id: i64 = tx.query_row(
1158	r#"
1159	INSERT INTO messages(thread_id, role, content, item_json, created_at, parent_entry_id)
1160	SELECT ?1, ?2, ?3, ?4, ?5, ?6
1161	RETURNING id
1162	"#, params![thread_id, role, content, item_json, created_at, current_leaf_id], \|row\| row.get(0)
1163	).with_context(\|\| format!("failed to append message for thread {thread_id}"))?;
1164
1165	tx.execute(
1166	r#"
1167	UPDATE threads
1168	SET current_leaf_id = ?1
1169	WHERE id = ?2;
1170	"#,
1171	params![next_leaf_id, thread_id],
1172	)
1173	.with_context(\|\| {
1174	format!("failed to update thread current leaf id for thread {thread_id}")
1175	})?;
1176
1177	tx.commit()
1178	.context("failed to commit append message transaction")?;
1179
1180	Ok(next_leaf_id)
1181	}
1182
1183	/// List messages in the current conversation branch, walking backwards from
1184	/// the thread's `current_leaf_id`.
1185	///
1186	/// Messages are returned in chronological order (oldest first). The `limit`
1187	/// parameter caps how many ancestor messages are traversed; it defaults to 500.
1188	pub fn list_messages(
1189	&self,
1190	thread_id: &str,
1191	limit: Option<usize>,
1192	) -> Result<Vec<MessageRecord>> {
1193	let conn = self.conn()?;
1194	let limit = i64::try_from(limit.unwrap_or(500)).unwrap_or(500);
1195	let mut stmt = conn
1196	.prepare(
1197	r#"
1198	WITH RECURSIVE
1199	leaf_id AS (
1200	SELECT current_leaf_id FROM threads WHERE id = ?1
1201	),
1202	ancestors AS (
1203	SELECT id, thread_id, role, content, item_json, created_at, parent_entry_id, 0 AS depth
1204	FROM messages
1205	WHERE id = (SELECT current_leaf_id FROM leaf_id)
1206
1207	UNION ALL
1208
1209	SELECT m.id, m.thread_id, m.role, m.content, m.item_json, m.created_at, m.parent_entry_id, a.depth + 1
1210	FROM messages m
1211	JOIN ancestors a ON m.id = a.parent_entry_id
1212	WHERE a.depth < ?2
1213	)
1214	SELECT id, thread_id, role, content, item_json, created_at, parent_entry_id FROM ancestors
1215	ORDER BY depth DESC
1216	"#
1217	)
1218	.context("failed to prepare message listing query")?;
1219	let mut rows = stmt
1220	.query(params![thread_id, limit - 1])
1221	.with_context(\|\| format!("failed to list messages for thread {thread_id}"))?;
1222	let mut out = Vec::new();
1223	while let Some(row) = rows.next().context("failed to iterate message rows")? {
1224	let item_json: Option<String> = row.get(4).context("failed to read item json")?;
1225	let item = item_json
1226	.as_deref()
1227	.map(serde_json::from_str)
1228	.transpose()
1229	.with_context(\|\| {
1230	format!("failed to parse message item json in thread {thread_id}")
1231	})?;
1232	out.push(MessageRecord {
1233	id: row.get(0).context("failed to read message id")?,
1234	thread_id: row.get(1).context("failed to read message thread id")?,
1235	role: row.get(2).context("failed to read message role")?,
1236	content: row.get(3).context("failed to read message content")?,
1237	item,
1238	created_at: row.get(5).context("failed to read message timestamp")?,
1239	parent_entry_id: row.get(6).context("failed to read parent entry id")?,
1240	});
1241	}
1242	Ok(out)
1243	}
1244
1245	/// Fork the conversation at a specific message.
1246	///
1247	/// Creates a new message whose parent is `message_id` and updates the thread's
1248	/// `current_leaf_id` to the new message. Returns the ID of the new message.
1249	/// This enables branching conversations from any point in the history.
1250	pub fn fork_at_message(
1251	&self,
1252	message_id: &str,
1253	role: &str,
1254	content: &str,
1255	item: Option<Value>,
1256	) -> Result<i64> {
1257	let mut conn = self.conn()?;
1258	let created_at = Utc::now().timestamp();
1259	let item_json = item
1260	.as_ref()
1261	.map(serde_json::to_string)
1262	.transpose()
1263	.context("failed to serialize message item payload")?;
1264
1265	let tx = conn
1266	.transaction()
1267	.context("failed to begin fork message transaction")?;
1268
1269	let thread_id: String = tx
1270	.query_row(
1271	"SELECT thread_id FROM messages WHERE id = ?1",
1272	params![message_id],
1273	\|row\| row.get(0),
1274	)
1275	.with_context(\|\| format!("failed to query thread id for message {message_id}"))?;
1276
1277	let next_leaf_id: i64 = tx.query_row(
1278	r#"
1279	INSERT INTO messages(thread_id, role, content, item_json, created_at, parent_entry_id)
1280	SELECT ?1, ?2, ?3, ?4, ?5, ?6
1281	RETURNING id
1282	"#, params![thread_id, role, content, item_json, created_at, message_id], \|row\| row.get(0)
1283	).with_context(\|\| format!("failed to fork at message for thread {thread_id:?}"))?;
1284
1285	tx.execute(
1286	r#"
1287	UPDATE threads
1288	SET current_leaf_id = ?1
1289	WHERE id = ?2;
1290	"#,
1291	params![next_leaf_id, thread_id],
1292	)
1293	.with_context(\|\| {
1294	format!("failed to update thread current leaf id for thread {thread_id:?}")
1295	})?;
1296
1297	tx.commit()
1298	.context("failed to commit fork message transaction")?;
1299
1300	Ok(next_leaf_id)
1301	}
1302
1303	/// Delete all messages belonging to a thread and reset its `current_leaf_id`.
1304	///
1305	/// Returns the number of messages deleted.
1306	pub fn clear_messages(&self, thread_id: &str) -> Result<usize> {
1307	let mut conn = self.conn()?;
1308	let tx = conn
1309	.transaction()
1310	.context("failed to begin clear messages transaction")?;
1311
1312	tx.execute(
1313	r#"
1314	UPDATE threads
1315	SET current_leaf_id = NULL
1316	WHERE id = ?1;
1317	"#,
1318	params![thread_id],
1319	)
1320	.with_context(\|\| format!("failed to clear messages for thread {thread_id}"))?;
1321	let result = tx
1322	.execute(
1323	r#"
1324	DELETE FROM messages WHERE thread_id = ?1
1325	"#,
1326	params![thread_id],
1327	)
1328	.with_context(\|\| format!("failed to clear messages for thread {thread_id}"))?;
1329	tx.commit()
1330	.context("failed to commit clear messages transaction")?;
1331
1332	Ok(result)
1333	}
1334
1335	/// Save (or update) a named checkpoint for a thread.
1336	///
1337	/// If a checkpoint with the same `thread_id` and `checkpoint_id` already exists,
1338	/// its state and timestamp are overwritten.
1339	pub fn save_checkpoint(
1340	&self,
1341	thread_id: &str,
1342	checkpoint_id: &str,
1343	state: &Value,
1344	) -> Result<()> {
1345	let conn = self.conn()?;
1346	let state_json =
1347	serde_json::to_string(state).context("failed to encode checkpoint state")?;
1348	conn.execute(
1349	r#"
1350	INSERT INTO checkpoints(thread_id, checkpoint_id, state_json, created_at)
1351	VALUES (?1, ?2, ?3, ?4)
1352	ON CONFLICT(thread_id, checkpoint_id) DO UPDATE SET
1353	state_json = excluded.state_json,
1354	created_at = excluded.created_at
1355	"#,
1356	params![thread_id, checkpoint_id, state_json, Utc::now().timestamp()],
1357	)
1358	.with_context(\|\| {
1359	format!("failed to save checkpoint {checkpoint_id} for thread {thread_id}")
1360	})?;
1361	Ok(())
1362	}
1363
1364	/// Load a checkpoint for a thread.
1365	///
1366	/// If `checkpoint_id` is provided, loads that specific checkpoint. Otherwise,
1367	/// loads the most recently created checkpoint for the thread. Returns `None`
1368	/// if no matching checkpoint exists.
1369	pub fn load_checkpoint(
1370	&self,
1371	thread_id: &str,
1372	checkpoint_id: Option<&str>,
1373	) -> Result<Option<CheckpointRecord>> {
1374	let conn = self.conn()?;
1375	if let Some(checkpoint_id) = checkpoint_id {
1376	let row = conn
1377	.query_row(
1378	"SELECT thread_id, checkpoint_id, state_json, created_at FROM checkpoints WHERE thread_id = ?1 AND checkpoint_id = ?2",
1379	params![thread_id, checkpoint_id],
1380	\|row\| {
1381	Ok((
1382	row.get::<_, String>(0)?,
1383	row.get::<_, String>(1)?,
1384	row.get::<_, String>(2)?,
1385	row.get::<_, i64>(3)?,
1386	))
1387	},
1388	)
1389	.optional()
1390	.with_context(\|\| {
1391	format!("failed to load checkpoint {checkpoint_id} for thread {thread_id}")
1392	})?;
1393	if let Some((thread_id, checkpoint_id, state_json, created_at)) = row {
1394	let state = parse_checkpoint_state(&state_json)?;
1395	return Ok(Some(CheckpointRecord {
1396	thread_id,
1397	checkpoint_id,
1398	state,
1399	created_at,
1400	}));
1401	}
1402	return Ok(None);
1403	}
1404
1405	let row = conn
1406	.query_row(
1407	"SELECT thread_id, checkpoint_id, state_json, created_at FROM checkpoints WHERE thread_id = ?1 ORDER BY created_at DESC LIMIT 1",
1408	params![thread_id],
1409	\|row\| {
1410	Ok((
1411	row.get::<_, String>(0)?,
1412	row.get::<_, String>(1)?,
1413	row.get::<_, String>(2)?,
1414	row.get::<_, i64>(3)?,
1415	))
1416	},
1417	)
1418	.optional()
1419	.with_context(\|\| format!("failed to load latest checkpoint for thread {thread_id}"))?;
1420	if let Some((thread_id, checkpoint_id, state_json, created_at)) = row {
1421	let state = parse_checkpoint_state(&state_json)?;
1422	return Ok(Some(CheckpointRecord {
1423	thread_id,
1424	checkpoint_id,
1425	state,
1426	created_at,
1427	}));
1428	}
1429	Ok(None)
1430	}
1431
1432	/// List checkpoints for a thread, ordered by creation time (newest first).
1433	///
1434	/// The `limit` parameter caps the number of results and defaults to 100.
1435	pub fn list_checkpoints(
1436	&self,
1437	thread_id: &str,
1438	limit: Option<usize>,
1439	) -> Result<Vec<CheckpointRecord>> {
1440	let conn = self.conn()?;
1441	let limit = i64::try_from(limit.unwrap_or(100)).unwrap_or(100);
1442	let mut stmt = conn
1443	.prepare(
1444	"SELECT thread_id, checkpoint_id, state_json, created_at FROM checkpoints WHERE thread_id = ?1 ORDER BY created_at DESC LIMIT ?2",
1445	)
1446	.context("failed to prepare checkpoint list query")?;
1447	let mut rows = stmt
1448	.query(params![thread_id, limit])
1449	.with_context(\|\| format!("failed to list checkpoints for thread {thread_id}"))?;
1450
1451	let mut out = Vec::new();
1452	while let Some(row) = rows.next().context("failed to iterate checkpoint rows")? {
1453	let state_json: String = row.get(2).context("failed to read checkpoint state json")?;
1454	let state = parse_checkpoint_state(&state_json)?;
1455	out.push(CheckpointRecord {
1456	thread_id: row.get(0).context("failed to read checkpoint thread id")?,
1457	checkpoint_id: row.get(1).context("failed to read checkpoint id")?,
1458	state,
1459	created_at: row.get(3).context("failed to read checkpoint timestamp")?,
1460	});
1461	}
1462	Ok(out)
1463	}
1464
1465	/// Delete a specific checkpoint from a thread.
1466	pub fn delete_checkpoint(&self, thread_id: &str, checkpoint_id: &str) -> Result<()> {
1467	let conn = self.conn()?;
1468	conn.execute(
1469	"DELETE FROM checkpoints WHERE thread_id = ?1 AND checkpoint_id = ?2",
1470	params![thread_id, checkpoint_id],
1471	)
1472	.with_context(\|\| {
1473	format!("failed to delete checkpoint {checkpoint_id} for thread {thread_id}")
1474	})?;
1475	Ok(())
1476	}
1477
1478	/// Insert or update a background job record.
1479	pub fn upsert_job(&self, job: &JobStateRecord) -> Result<()> {
1480	let conn = self.conn()?;
1481	conn.execute(
1482	r#"
1483	INSERT INTO jobs(id, name, status, progress, detail, created_at, updated_at)
1484	VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
1485	ON CONFLICT(id) DO UPDATE SET
1486	name = excluded.name,
1487	status = excluded.status,
1488	progress = excluded.progress,
1489	detail = excluded.detail,
1490	created_at = excluded.created_at,
1491	updated_at = excluded.updated_at
1492	"#,
1493	params![
1494	job.id,
1495	job.name,
1496	job_state_status_to_str(&job.status),
1497	job.progress.map(i64::from),
1498	job.detail,
1499	job.created_at,
1500	job.updated_at
1501	],
1502	)
1503	.with_context(\|\| format!("failed to upsert job {}", job.id))?;
1504	Ok(())
1505	}
1506
1507	/// List jobs ordered by most recently updated.
1508	///
1509	/// The `limit` parameter caps the number of results and defaults to 100.
1510	pub fn list_jobs(&self, limit: Option<usize>) -> Result<Vec<JobStateRecord>> {
1511	let conn = self.conn()?;
1512	let limit = i64::try_from(limit.unwrap_or(100)).unwrap_or(100);
1513	let mut stmt = conn
1514	.prepare(
1515	"SELECT id, name, status, progress, detail, created_at, updated_at FROM jobs ORDER BY updated_at DESC LIMIT ?1",
1516	)
1517	.context("failed to prepare job list query")?;
1518	let mut rows = stmt
1519	.query(params![limit])
1520	.context("failed to query persisted jobs")?;
1521	let mut out = Vec::new();
1522	while let Some(row) = rows.next().context("failed to iterate persisted jobs")? {
1523	let status_raw: String = row.get(2).context("failed to read job status")?;
1524	let progress: Option<i64> = row.get(3).context("failed to read job progress")?;
1525	out.push(JobStateRecord {
1526	id: row.get(0).context("failed to read job id")?,
1527	name: row.get(1).context("failed to read job name")?,
1528	status: job_state_status_from_str(&status_raw),
1529	progress: progress.and_then(\|v\| u8::try_from(v).ok()),
1530	detail: row.get(4).context("failed to read job detail")?,
1531	created_at: row.get(5).context("failed to read job created_at")?,
1532	updated_at: row.get(6).context("failed to read job updated_at")?,
1533	});
1534	}
1535	Ok(out)
1536	}
1537
1538	/// Look up the rollout file path for a thread by its ID.
1539	pub fn find_rollout_path_by_id(&self, id: &str) -> Result<Option<PathBuf>> {
1540	let conn = self.conn()?;
1541	conn.query_row(
1542	"SELECT rollout_path FROM threads WHERE id = ?1",
1543	params![id],
1544	\|row\| row.get::<_, Option<String>>(0),
1545	)
1546	.optional()
1547	.context("failed to lookup rollout path")
1548	.map(\|opt\| opt.flatten().map(PathBuf::from))
1549	}
1550
1551	/// Append an entry to the JSONL session index file.
1552	///
1553	/// The session index is an append-only log that maps thread IDs to their names,
1554	/// update timestamps, and rollout paths. It is used for fast name-based lookups
1555	/// without opening the SQLite database.
1556	pub fn append_thread_name(
1557	&self,
1558	thread_id: &str,
1559	thread_name: Option<String>,
1560	updated_at: i64,
1561	rollout_path: Option<PathBuf>,
1562	) -> Result<()> {
1563	// Hold the index lock for the entire append + compaction so a concurrent
1564	// `StateStore` clone cannot rename the index while we are appending.
1565	let _guard = SESSION_INDEX_LOCK.lock().unwrap();
1566	if let Some(parent) = self.session_index_path.parent() {
1567	fs::create_dir_all(parent).with_context(\|\| {
1568	format!(
1569	"failed to create session index directory {}",
1570	parent.display()
1571	)
1572	})?;
1573	}
1574	let entry = SessionIndexEntry {
1575	thread_id: thread_id.to_string(),
1576	thread_name,
1577	updated_at,
1578	rollout_path,
1579	};
1580	let encoded =
1581	serde_json::to_string(&entry).context("failed to serialize session index entry")?;
1582	// Append and compaction share one lock. Compaction rewrites the file
1583	// from a snapshot and renames over it, so an append landing between
1584	// that snapshot and the rename would be discarded — silently, since
1585	// the append already returned success to its caller.
1586	self.with_session_index_lock(\|\| {
1587	let mut file = OpenOptions::new()
1588	.create(true)
1589	.append(true)
1590	.open(&self.session_index_path)
1591	.with_context(\|\| {
1592	format!(
1593	"failed to open session index {}",
1594	self.session_index_path.display()
1595	)
1596	})?;
1597	writeln!(file, "{encoded}").context("failed to append session index entry")?;
1598	// Durability: without this a crash mid-write can leave a torn
1599	// final line. Reads tolerate one (see `session_index_map`), but
1600	// not losing the entry beats recovering from having lost it.
1601	file.sync_data()
1602	.context("failed to flush session index entry")?;
1603	drop(file);
1604	self.compact_session_index_locked()
1605	})
1606	}
1607
1608	/// Run `operation` holding the exclusive session-index lock.
1609	///
1610	/// The lock is an adjacent `.lock` file rather than the index itself, so
1611	/// compaction's rename cannot pull the lock out from under a waiter. This
1612	/// mirrors the discipline `codewhale-config` uses for `config.toml`.
1613	fn with_session_index_lock<T>(&self, operation: impl FnOnce() -> Result<T>) -> Result<T> {
1614	if let Some(parent) = self.session_index_path.parent() {
1615	fs::create_dir_all(parent).with_context(\|\| {
1616	format!(
1617	"failed to create session index directory {}",
1618	parent.display()
1619	)
1620	})?;
1621	}
1622	let lock_path = self.session_index_path.with_extension("jsonl.lock");
1623	let lock_file = OpenOptions::new()
1624	.create(true)
1625	.read(true)
1626	.write(true)
1627	// The file is only a lock handle; its contents are never read and
1628	// truncating it would race other holders for no benefit.
1629	.truncate(false)
1630	.open(&lock_path)
1631	.with_context(\|\| {
1632	format!("failed to open session index lock {}", lock_path.display())
1633	})?;
1634	#[cfg(unix)]
1635	{
1636	use std::os::unix::fs::PermissionsExt as _;
1637	lock_file
1638	.set_permissions(fs::Permissions::from_mode(0o600))
1639	.with_context(\|\| {
1640	format!(
1641	"failed to secure session index lock {}",
1642	lock_path.display()
1643	)
1644	})?;
1645	}
1646	let mut lock = fd_lock::RwLock::new(lock_file);
1647	let _guard = lock
1648	.write()
1649	.with_context(\|\| format!("failed to lock session index {}", lock_path.display()))?;
1650	operation()
1651	}
1652
1653	/// Find the display name for a thread by its ID, using the session index.
1654	///
1655	/// Returns `None` if the thread is not in the index or has no name.
1656	pub fn find_thread_name_by_id(&self, thread_id: &str) -> Result<Option<String>> {
1657	let map = self.session_index_map()?;
1658	Ok(map
1659	.get(thread_id)
1660	.and_then(\|entry\| entry.thread_name.clone()))
1661	}
1662
1663	/// Look up display names for multiple thread IDs at once.
1664	///
1665	/// Returns a map from thread ID to its name (which may be `None`).
1666	pub fn find_thread_names_by_ids(
1667	&self,
1668	ids: &[String],
1669	) -> Result<HashMap<String, Option<String>>> {
1670	let map = self.session_index_map()?;
1671	let mut out = HashMap::new();
1672	for id in ids {
1673	let name = map.get(id).and_then(\|entry\| entry.thread_name.clone());
1674	out.insert(id.clone(), name);
1675	}
1676	Ok(out)
1677	}
1678
1679	/// Find the rollout path for a thread by its display name (case-insensitive).
1680	///
1681	/// If multiple threads share the same name, the most recently updated one is returned.
1682	/// Returns `None` if no matching thread is found.
1683	pub fn find_thread_path_by_name_str(&self, name: &str) -> Result<Option<PathBuf>> {
1684	let map = self.session_index_map()?;
1685	let matched = map
1686	.values()
1687	.filter(\|entry\| {
1688	entry
1689	.thread_name
1690	.as_deref()
1691	.is_some_and(\|n\| n.eq_ignore_ascii_case(name))
1692	})
1693	.max_by_key(\|entry\| entry.updated_at);
1694	Ok(matched.and_then(\|entry\| entry.rollout_path.clone()))
1695	}
1696
1697	/// Compact the session index. The caller must already hold the lock from
1698	/// [`Self::with_session_index_lock`]: this reads a snapshot and renames a
1699	/// rewritten file over the live one, and an append interleaved between
1700	/// those two steps is lost.
1701	fn compact_session_index_locked(&self) -> Result<()> {
1702	if !self.session_index_path.exists() {
1703	return Ok(());
1704	}
1705	let line_count = BufReader::new(
1706	OpenOptions::new()
1707	.read(true)
1708	.open(&self.session_index_path)
1709	.with_context(\|\| {
1710	format!(
1711	"failed to read session index {}",
1712	self.session_index_path.display()
1713	)
1714	})?,
1715	)
1716	.lines()
1717	.filter(\|line\| {
1718	line.as_ref()
1719	.map(\|value\| !value.trim().is_empty())
1720	.unwrap_or(false)
1721	})
1722	.count();
1723	if line_count <= session_index_compact_line_threshold() {
1724	return Ok(());
1725	}
1726
1727	let latest = self.session_index_map()?;
1728	let compact_path = self.session_index_path.with_extension("jsonl.compact");
1729	{
1730	let mut file = OpenOptions::new()
1731	.create(true)
1732	.write(true)
1733	.truncate(true)
1734	.open(&compact_path)
1735	.with_context(\|\| {
1736	format!(
1737	"failed to open compact session index {}",
1738	compact_path.display()
1739	)
1740	})?;
1741	for entry in latest.values() {
1742	let encoded = serde_json::to_string(entry)
1743	.context("failed to serialize compact session index entry")?;
1744	writeln!(file, "{encoded}")
1745	.context("failed to write compact session index entry")?;
1746	}
1747	}
1748	// The snapshot is written but the live file is still the old one:
1749	// this is the window an unsynchronized appender would write into and
1750	// lose. Tests widen it deliberately to prove the lock closes it.
1751	#[cfg(test)]
1752	tests::compaction_midpoint(&self.session_index_path);
1753	fs::rename(&compact_path, &self.session_index_path).with_context(\|\| {
1754	format!(
1755	"failed to replace session index {}",
1756	self.session_index_path.display()
1757	)
1758	})?;
1759	Ok(())
1760	}
1761
1762	#[cfg(test)]
1763	fn session_index_line_count(&self) -> Result<usize> {
1764	if !self.session_index_path.exists() {
1765	return Ok(0);
1766	}
1767	Ok(BufReader::new(
1768	OpenOptions::new()
1769	.read(true)
1770	.open(&self.session_index_path)
1771	.with_context(\|\| {
1772	format!(
1773	"failed to read session index {}",
1774	self.session_index_path.display()
1775	)
1776	})?,
1777	)
1778	.lines()
1779	.filter(\|line\| {
1780	line.as_ref()
1781	.map(\|value\| !value.trim().is_empty())
1782	.unwrap_or(false)
1783	})
1784	.count())
1785	}
1786
1787	fn session_index_map(&self) -> Result<HashMap<String, SessionIndexEntry>> {
1788	if !self.session_index_path.exists() {
1789	return Ok(HashMap::new());
1790	}
1791	let file = OpenOptions::new()
1792	.read(true)
1793	.open(&self.session_index_path)
1794	.with_context(\|\| {
1795	format!(
1796	"failed to read session index {}",
1797	self.session_index_path.display()
1798	)
1799	})?;
1800	let reader = BufReader::new(file);
1801	let mut latest = HashMap::<String, SessionIndexEntry>::new();
1802	for line in reader.lines() {
1803	let line = line.context("failed to read session index line")?;
1804	if line.trim().is_empty() {
1805	continue;
1806	}
1807	// Skip a line we can't parse instead of failing the whole read.
1808	// An append that was interrupted mid-write leaves a torn final
1809	// line; aborting here broke every thread-name lookup, and because
1810	// compaction reads through this same function, the index could
1811	// never repair itself either — the file stayed broken until
1812	// someone deleted it by hand.
1813	match serde_json::from_str::<SessionIndexEntry>(&line) {
1814	Ok(parsed) => {
1815	latest.insert(parsed.thread_id.clone(), parsed);
1816	}
1817	Err(err) => {
1818	tracing::warn!(
1819	"skipping unparseable session index entry in {}: {err}",
1820	self.session_index_path.display()
1821	);
1822	}
1823	}
1824	}
1825	Ok(latest)
1826	}
1827	}
1828
1829	/// Resolve the default SQLite state path without opening or creating it.
1830	///
1831	/// An explicit `CODEWHALE_HOME` always yields `<override>/state.db` and blocks
1832	/// ambient legacy fallback. Without an override, an existing legacy database
1833	/// remains readable until it is migrated.
1834	#[must_use]
1835	pub fn default_state_db_path() -> PathBuf {
1836	// $CODEWHALE_HOME is a hard override of the base data directory
1837	// (docs/CONFIGURATION.md): when set, the state DB lives under it and we do
1838	// NOT fall back to the legacy ~/.deepseek path — silent fallback would
1839	// defeat the isolation the override promises (CI, containers, multi-project,
1840	// test harnesses). Legacy ~/.deepseek migration only applies to the default
1841	// home location.
1842	if let Some(overridden) = codewhale_home_override().ok().flatten() {
1843	return overridden.join("state.db");
1844	}
1845	let home = codewhale_paths::user_home().unwrap_or_else(\|\| PathBuf::from("."));
1846	// Prefer the CodeWhale directory, falling back to legacy DeepSeek path
1847	// so existing installs don't lose their session history.
1848	let primary = home.join(CODEWHALE_APP_DIR).join("state.db");
1849	if primary.exists() \|\| !home.join(LEGACY_APP_DIR).join("state.db").exists() {
1850	primary
1851	} else {
1852	home.join(LEGACY_APP_DIR).join("state.db")
1853	}
1854	}
1855
1856	fn bool_to_i64(value: bool) -> i64 {
1857	if value { 1 } else { 0 }
1858	}
1859
1860	/// Whether `table` currently has a column named `column`.
1861	///
1862	/// Used to guard `ALTER TABLE ... ADD COLUMN` migrations so they are
1863	/// idempotent. Both identifiers are compile-time literals at every call
1864	/// site, never user input. A missing table reports `false`, matching the
1865	/// fresh-database case where the migration must still run.
1866	fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
1867	let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
1868	let names = stmt.query_map([], \|row\| row.get::<_, String>(1))?;
1869	for name in names {
1870	if name? == column {
1871	return Ok(true);
1872	}
1873	}
1874	Ok(false)
1875	}
1876
1877	fn i64_to_bool(value: i64) -> bool {
1878	value != 0
1879	}
1880
1881	fn thread_status_to_str(status: &ThreadStatus) -> &'static str {
1882	match status {
1883	ThreadStatus::Running => "running",
1884	ThreadStatus::Idle => "idle",
1885	ThreadStatus::Completed => "completed",
1886	ThreadStatus::Failed => "failed",
1887	ThreadStatus::Paused => "paused",
1888	ThreadStatus::Archived => "archived",
1889	}
1890	}
1891
1892	fn thread_status_from_str(value: &str) -> ThreadStatus {
1893	match value {
1894	"running" => ThreadStatus::Running,
1895	"idle" => ThreadStatus::Idle,
1896	"completed" => ThreadStatus::Completed,
1897	"failed" => ThreadStatus::Failed,
1898	"paused" => ThreadStatus::Paused,
1899	"archived" => ThreadStatus::Archived,
1900	_ => ThreadStatus::Idle,
1901	}
1902	}
1903
1904	fn session_source_to_str(source: &SessionSource) -> &'static str {
1905	match source {
1906	SessionSource::Interactive => "interactive",
1907	SessionSource::Resume => "resume",
1908	SessionSource::Fork => "fork",
1909	SessionSource::Api => "api",
1910	SessionSource::Unknown => "unknown",
1911	}
1912	}
1913
1914	fn session_source_from_str(value: &str) -> SessionSource {
1915	match value {
1916	"interactive" => SessionSource::Interactive,
1917	"resume" => SessionSource::Resume,
1918	"fork" => SessionSource::Fork,
1919	"api" => SessionSource::Api,
1920	_ => SessionSource::Unknown,
1921	}
1922	}
1923
1924	fn path_to_opt_string(path: Option<&Path>) -> Option<String> {
1925	path.map(\|p\| p.display().to_string())
1926	}
1927
1928	fn parse_checkpoint_state(state_json: &str) -> Result<Value> {
1929	serde_json::from_str(state_json).context("failed to parse checkpoint state json")
1930	}
1931
1932	fn job_state_status_to_str(status: &JobStateStatus) -> &'static str {
1933	match status {
1934	JobStateStatus::Queued => "queued",
1935	JobStateStatus::Running => "running",
1936	JobStateStatus::Paused => "paused",
1937	JobStateStatus::Completed => "completed",
1938	JobStateStatus::Failed => "failed",
1939	JobStateStatus::Cancelled => "cancelled",
1940	}
1941	}
1942
1943	fn job_state_status_from_str(value: &str) -> JobStateStatus {
1944	match value {
1945	"queued" => JobStateStatus::Queued,
1946	"running" => JobStateStatus::Running,
1947	"paused" => JobStateStatus::Paused,
1948	"completed" => JobStateStatus::Completed,
1949	"failed" => JobStateStatus::Failed,
1950	"cancelled" => JobStateStatus::Cancelled,
1951	_ => JobStateStatus::Queued,
1952	}
1953	}
1954
1955	fn thread_goal_status_to_str(status: &ThreadGoalStatus) -> &'static str {
1956	match status {
1957	ThreadGoalStatus::Active => "active",
1958	ThreadGoalStatus::Paused => "paused",
1959	ThreadGoalStatus::Blocked => "blocked",
1960	ThreadGoalStatus::UsageLimited => "usage_limited",
1961	ThreadGoalStatus::BudgetLimited => "budget_limited",
1962	ThreadGoalStatus::Complete => "complete",
1963	}
1964	}
1965
1966	fn thread_goal_status_from_str(value: &str) -> ThreadGoalStatus {
1967	match value {
1968	"active" => ThreadGoalStatus::Active,
1969	"paused" => ThreadGoalStatus::Paused,
1970	"blocked" => ThreadGoalStatus::Blocked,
1971	"usage_limited" => ThreadGoalStatus::UsageLimited,
1972	"budget_limited" => ThreadGoalStatus::BudgetLimited,
1973	"complete" => ThreadGoalStatus::Complete,
1974	// Fail closed: an unknown or corrupted persisted value must never
1975	// resurrect a self-driving goal. The user can inspect and explicitly
1976	// resume a paused goal after repairing or replacing the record.
1977	_ => ThreadGoalStatus::Paused,
1978	}
1979	}
1980
1981	fn row_to_thread(row: &rusqlite::Row<'_>) -> rusqlite::Result<ThreadMetadata> {
1982	let status_raw: String = row.get(7)?;
1983	let source_raw: String = row.get(11)?;
1984	let rollout_path: Option<String> = row.get(1)?;
1985	let path: Option<String> = row.get(8)?;
1986	Ok(ThreadMetadata {
1987	id: row.get(0)?,
1988	rollout_path: rollout_path.map(PathBuf::from),
1989	preview: row.get(2)?,
1990	ephemeral: i64_to_bool(row.get(3)?),
1991	model_provider: row.get(4)?,
1992	created_at: row.get(5)?,
1993	updated_at: row.get(6)?,
1994	status: thread_status_from_str(&status_raw),
1995	path: path.map(PathBuf::from),
1996	cwd: PathBuf::from(row.get::<_, String>(9)?),
1997	cli_version: row.get(10)?,
1998	source: session_source_from_str(&source_raw),
1999	name: row.get(12)?,
2000	sandbox_policy: row.get(13)?,
2001	approval_mode: row.get(14)?,
2002	archived: i64_to_bool(row.get(15)?),
2003	archived_at: row.get(16)?,
2004	git_sha: row.get(17)?,
2005	git_branch: row.get(18)?,
2006	git_origin_url: row.get(19)?,
2007	memory_mode: row.get(20)?,
2008	current_leaf_id: row.get(21)?,
2009	})
2010	}
2011
2012	fn row_to_thread_goal(row: &rusqlite::Row<'_>) -> rusqlite::Result<ThreadGoalRecord> {
2013	let status_raw: String = row.get(3)?;
2014	let pause_reason: Option<String> = row.get(13)?;
2015	let mut goal = ThreadGoalRecord {
2016	thread_id: row.get(0)?,
2017	goal_id: row.get(1)?,
2018	objective: row.get(2)?,
2019	status: thread_goal_status_from_str(&status_raw),
2020	token_budget: row.get(4)?,
2021	tokens_used: row.get(5)?,
2022	time_used_seconds: row.get(6)?,
2023	continuation_count: row.get(7)?,
2024	created_at: row.get(8)?,
2025	updated_at: row.get(9)?,
2026	last_gap_fingerprint: row.get(10)?,
2027	repeated_gap_count: row.get(11)?,
2028	last_gap_pass: row.get(12)?,
2029	pause_reason: pause_reason
2030	.map(\|value\| serde_json::from_str(&value))
2031	.transpose()
2032	.map_err(\|error\| {
2033	rusqlite::Error::FromSqlConversionFailure(
2034	13,
2035	rusqlite::types::Type::Text,
2036	Box::new(error),
2037	)
2038	})?,
2039	};
2040	codewhale_protocol::validate_goal_stall_state(
2041	goal.last_gap_fingerprint.as_deref(),
2042	goal.repeated_gap_count,
2043	goal.last_gap_pass,
2044	u32::try_from(goal.continuation_count.max(0)).unwrap_or(u32::MAX),
2045	)
2046	.map_err(\|error\| {
2047	rusqlite::Error::FromSqlConversionFailure(
2048	10,
2049	rusqlite::types::Type::Text,
2050	Box::new(std::io::Error::new(std::io::ErrorKind::InvalidData, error)),
2051	)
2052	})?;
2053	goal.normalize_restored_stall_state();
2054	Ok(goal)
2055	}
2056
2057	impl ThreadGoalRecord {
2058	/// Same restore rule as `codewhale_protocol::ThreadGoal`: an Active
2059	/// record with an exhausted stall window is corrupt and reads as paused.
2060	fn normalize_restored_stall_state(&mut self) {
2061	if matches!(self.status, ThreadGoalStatus::Active)
2062	&& self.repeated_gap_count >= codewhale_protocol::MAX_REPEATED_GAP_COUNT
2063	{
2064	self.status = ThreadGoalStatus::Paused;
2065	self.pause_reason = Some(codewhale_protocol::GoalPauseReason::NoProgress);
2066	}
2067	}
2068	}
2069
2070	#[cfg(test)]
2071	mod tests {
2072	use super::*;
2073	use serde_json::json;
2074	use std::sync::{Arc, Barrier, Mutex, mpsc};
2075	use std::thread;
2076	use std::time::{Duration, SystemTime, UNIX_EPOCH};
2077
2078	fn temp_state_dir(name: &str) -> PathBuf {
2079	let suffix = SystemTime::now()
2080	.duration_since(UNIX_EPOCH)
2081	.expect("system time")
2082	.as_nanos();
2083	let dir = std::env::temp_dir().join(format!(
2084	"codewhale-state-{name}-{}-{suffix}",
2085	std::process::id()
2086	));
2087	fs::create_dir_all(&dir).expect("create temp state dir");
2088	dir
2089	}
2090
2091	fn temp_state_store(name: &str) -> StateStore {
2092	let dir = temp_state_dir(name);
2093	StateStore::open(Some(dir.join("state.db"))).expect("open state store")
2094	}
2095
2096	fn test_thread(id: &str) -> ThreadMetadata {
2097	ThreadMetadata {
2098	id: id.to_string(),
2099	rollout_path: None,
2100	preview: "test thread".to_string(),
2101	ephemeral: false,
2102	model_provider: "deepseek".to_string(),
2103	created_at: 10,
2104	updated_at: 10,
2105	status: ThreadStatus::Running,
2106	path: None,
2107	cwd: PathBuf::from("/tmp/codewhale"),
2108	cli_version: "0.0.0-test".to_string(),
2109	source: SessionSource::Interactive,
2110	name: None,
2111	sandbox_policy: None,
2112	approval_mode: None,
2113	archived: false,
2114	archived_at: None,
2115	git_sha: None,
2116	git_branch: None,
2117	git_origin_url: None,
2118	memory_mode: None,
2119	current_leaf_id: None,
2120	}
2121	}
2122
2123	fn test_goal(thread_id: &str, objective: &str) -> ThreadGoalRecord {
2124	ThreadGoalRecord {
2125	thread_id: thread_id.to_string(),
2126	goal_id: "goal-1".to_string(),
2127	objective: objective.to_string(),
2128	status: ThreadGoalStatus::Active,
2129	token_budget: Some(123),
2130	tokens_used: 7,
2131	time_used_seconds: 11,
2132	continuation_count: 0,
2133	last_gap_fingerprint: None,
2134	repeated_gap_count: 0,
2135	last_gap_pass: None,
2136	pause_reason: None,
2137	created_at: 100,
2138	updated_at: 101,
2139	}
2140	}
2141
2142	#[test]
2143	fn unknown_persisted_goal_status_fails_closed() {
2144	assert_eq!(
2145	thread_goal_status_from_str("future_or_corrupt_status"),
2146	ThreadGoalStatus::Paused
2147	);
2148	}
2149
2150	#[test]
2151	fn thread_goal_crud_round_trips_and_replaces() {
2152	let store = temp_state_store("thread-goal-crud");
2153	store
2154	.upsert_thread(&test_thread("thread-1"))
2155	.expect("upsert thread");
2156
2157	let goal = test_goal("thread-1", "Ship v0.8.59");
2158	store.upsert_thread_goal(&goal).expect("upsert goal");
2159	assert_eq!(
2160	store
2161	.get_thread_goal("thread-1")
2162	.expect("read goal")
2163	.as_ref(),
2164	Some(&goal)
2165	);
2166
2167	let mut replacement = test_goal("thread-1", "Ship v0.8.59 safely");
2168	replacement.goal_id = "goal-2".to_string();
2169	replacement.status = ThreadGoalStatus::BudgetLimited;
2170	replacement.token_budget = None;
2171	replacement.updated_at = 202;
2172	store
2173	.upsert_thread_goal(&replacement)
2174	.expect("replace goal");
2175	assert_eq!(
2176	store.get_thread_goal("thread-1").expect("read replacement"),
2177	Some(replacement)
2178	);
2179
2180	assert!(store.delete_thread_goal("thread-1").expect("delete goal"));
2181	assert!(
2182	store
2183	.get_thread_goal("thread-1")
2184	.expect("read empty")
2185	.is_none()
2186	);
2187	assert!(!store.delete_thread_goal("thread-1").expect("delete empty"));
2188	}
2189
2190	#[test]
2191	fn thread_goal_requires_existing_thread() {
2192	let store = temp_state_store("thread-goal-missing-thread");
2193	let err = store
2194	.upsert_thread_goal(&test_goal("missing-thread", "nope"))
2195	.expect_err("goal without a thread should fail");
2196	assert!(err.to_string().contains("thread missing-thread not found"));
2197	}
2198
2199	#[test]
2200	fn delete_thread_cascades_child_rows() {
2201	let store = temp_state_store("thread-delete-cascade");
2202	store
2203	.upsert_thread(&test_thread("thread-1"))
2204	.expect("upsert thread");
2205	store
2206	.append_message("thread-1", "user", "hello", None)
2207	.expect("append message");
2208	store
2209	.save_checkpoint("thread-1", "checkpoint-1", &serde_json::json!({"ok": true}))
2210	.expect("save checkpoint");
2211	store
2212	.persist_dynamic_tools(
2213	"thread-1",
2214	&[DynamicToolRecord {
2215	position: 0,
2216	name: "test_tool".to_string(),
2217	description: Some("test".to_string()),
2218	input_schema: serde_json::json!({"type": "object"}),
2219	}],
2220	)
2221	.expect("persist dynamic tools");
2222	store
2223	.upsert_thread_goal(&test_goal("thread-1", "Ship v0.8.67"))
2224	.expect("upsert goal");
2225
2226	store.delete_thread("thread-1").expect("delete thread");
2227
2228	let conn = store.conn().expect("conn");
2229	for table in [
2230	"messages",
2231	"checkpoints",
2232	"thread_dynamic_tools",
2233	"thread_goals",
2234	] {
2235	let sql = format!("SELECT COUNT(*) FROM {table} WHERE thread_id = ?1");
2236	let count: i64 = conn
2237	.query_row(&sql, params!["thread-1"], \|row\| row.get(0))
2238	.expect("count child rows");
2239	assert_eq!(count, 0, "{table} row survived thread deletion");
2240	}
2241	}
2242
2243	#[test]
2244	fn state_store_reuses_one_connection_across_operations_and_clones() {
2245	let store = temp_state_store("conn-reuse");
2246	{
2247	let conn = store.conn().expect("conn");
2248	conn.execute_batch("CREATE TEMP TABLE conn_reuse_probe(id INTEGER);")
2249	.expect("create temp table");
2250	}
2251	// TEMP tables are visible only on the connection that created them, so
2252	// seeing the probe again — through a clone, after real operations ran —
2253	// proves the store holds one long-lived connection instead of
2254	// reopening the database (and reapplying pragmas) per call.
2255	let clone = store.clone();
2256	clone
2257	.upsert_thread(&test_thread("thread-conn-reuse"))
2258	.expect("upsert thread");
2259	let conn = clone.conn().expect("conn");
2260	let probe_count: i64 = conn
2261	.query_row(
2262	"SELECT COUNT(*) FROM sqlite_temp_master WHERE name = 'conn_reuse_probe'",
2263	[],
2264	\|row\| row.get(0),
2265	)
2266	.expect("query temp master");
2267	assert_eq!(
2268	probe_count, 1,
2269	"temp table not visible: a fresh connection was opened"
2270	);
2271	// The pragma applied once at open still governs the shared connection.
2272	let foreign_keys: i64 = conn
2273	.query_row("PRAGMA foreign_keys;", [], \|row\| row.get(0))
2274	.expect("read foreign_keys pragma");
2275	assert_eq!(foreign_keys, 1);
2276	let journal_mode: String = conn
2277	.query_row("PRAGMA journal_mode;", [], \|row\| row.get(0))
2278	.expect("read journal_mode pragma");
2279	assert_eq!(
2280	journal_mode.to_ascii_lowercase(),
2281	"wal",
2282	"open should enable WAL for multi-process readers/writers"
2283	);
2284	}
2285
2286	#[test]
2287	fn connection_setup_waits_for_database_lock_before_enabling_wal() {
2288	let dir = temp_state_dir("locked-open");
2289	let db_path = dir.join("state.db");
2290
2291	let candidate = Connection::open(&db_path).expect("open candidate connection");
2292	// Do not let rusqlite's current default mask StateStore's own setup
2293	// contract: configure_connection must install the wait policy before
2294	// it performs any operation that can need a database lock.
2295	candidate
2296	.busy_timeout(Duration::ZERO)
2297	.expect("disable dependency default timeout");
2298	let blocker = Connection::open(&db_path).expect("open blocking connection");
2299	let (locked_tx, locked_rx) = mpsc::sync_channel(0);
2300	let blocker_thread = thread::spawn(move \|\| {
2301	blocker
2302	.execute_batch("BEGIN EXCLUSIVE;")
2303	.expect("acquire exclusive database lock");
2304	locked_tx.send(()).expect("announce database lock");
2305	thread::sleep(Duration::from_millis(200));
2306	blocker
2307	.execute_batch("COMMIT;")
2308	.expect("release exclusive database lock");
2309	});
2310
2311	locked_rx.recv().expect("wait for database lock");
2312	StateStore::configure_connection(&candidate, &db_path)
2313	.expect("connection setup should wait for the brief database lock");
2314	blocker_thread.join().expect("blocking thread panicked");
2315
2316	let journal_mode: String = candidate
2317	.query_row("PRAGMA journal_mode;", [], \|row\| row.get(0))
2318	.expect("read journal_mode");
2319	assert_eq!(journal_mode.to_ascii_lowercase(), "wal");
2320
2321	drop(candidate);
2322	let _ = fs::remove_dir_all(dir);
2323	}
2324
2325	/// A second process must wait for a brief active writer instead of
2326	/// surfacing SQLITE_BUSY (#4734).
2327	///
2328	/// The lock handoff is explicit: unlike a race between many autocommit
2329	/// writes, this proves the busy timeout while keeping the contention
2330	/// duration below its documented five-second bound on every platform.
2331	#[test]
2332	fn second_connection_waits_for_active_writer() {
2333	let dir = temp_state_dir("concurrent-write");
2334	let db_path = dir.join("state.db");
2335
2336	let store_a = StateStore::open(Some(db_path.clone())).expect("open store a");
2337	let store_b = StateStore::open(Some(db_path.clone())).expect("open store b");
2338	let (locked_tx, locked_rx) = mpsc::sync_channel(0);
2339	let (release_tx, release_rx) = mpsc::sync_channel(0);
2340
2341	let writer_a = thread::spawn(move \|\| {
2342	let conn = store_a.conn().expect("connection a");
2343	conn.execute_batch(
2344	r#"
2345	BEGIN IMMEDIATE;
2346	INSERT INTO jobs(id, name, status, created_at, updated_at)
2347	VALUES ('job-a', 'writer-a', 'running', 0, 0);
2348	"#,
2349	)
2350	.expect("writer a should acquire the database write lock");
2351	locked_tx.send(()).expect("announce active writer");
2352	release_rx.recv().expect("wait to release active writer");
2353	conn.execute_batch("COMMIT;")
2354	.expect("writer a should commit");
2355	});
2356
2357	locked_rx.recv().expect("wait for active writer");
2358	let (attempting_tx, attempting_rx) = mpsc::sync_channel(0);
2359	let writer_b = thread::spawn(move \|\| {
2360	attempting_tx.send(()).expect("announce second write");
2361	store_b.upsert_job(&JobStateRecord {
2362	id: "job-b".to_string(),
2363	name: "writer-b".to_string(),
2364	status: JobStateStatus::Running,
2365	progress: None,
2366	detail: Some("waited for writer a".to_string()),
2367	created_at: 1,
2368	updated_at: 1,
2369	})
2370	});
2371
2372	attempting_rx.recv().expect("wait for second write attempt");
2373	thread::sleep(Duration::from_millis(100));
2374	assert!(
2375	!writer_b.is_finished(),
2376	"second writer should still be waiting while the first holds the lock"
2377	);
2378	release_tx.send(()).expect("release active writer");
2379	writer_a.join().expect("writer a panicked");
2380	writer_b
2381	.join()
2382	.expect("writer b panicked")
2383	.expect("writer b should succeed after the lock is released");
2384
2385	let store = StateStore::open(Some(db_path)).expect("reopen for verify");
2386	let listed = store.list_jobs(Some(2)).expect("list jobs");
2387	assert_eq!(listed.len(), 2, "both writers should persist their jobs");
2388
2389	let _ = fs::remove_dir_all(dir);
2390	}
2391
2392	#[test]
2393	fn migration_runs_cleanly_when_schema_predates_user_version_header() {
2394	// Simulate a restore (or a racing process that crashed before
2395	// stamping user_version): the on-disk schema is fully migrated but
2396	// the header still says 0. The v0 block used to re-run unconditional
2397	// ADD COLUMN statements and abort the open with
2398	// "duplicate column name".
2399	let dir = temp_state_dir("migration-v0-idempotent");
2400	let db_path = dir.join("state.db");
2401	drop(StateStore::open(Some(db_path.clone())).expect("initial open"));
2402	{
2403	let conn = Connection::open(&db_path).expect("raw connection");
2404	conn.pragma_update(None, "user_version", 0)
2405	.expect("reset user_version");
2406	}
2407
2408	let store = StateStore::open(Some(db_path.clone())).expect("reopen with v0 header");
2409	store
2410	.upsert_thread(&test_thread("thread-migrated"))
2411	.expect("write after guarded migration");
2412
2413	// Reopening again (now stamped at the current version) still works.
2414	drop(store);
2415	let store = StateStore::open(Some(db_path)).expect("third open");
2416	let persisted = store
2417	.get_thread("thread-migrated")
2418	.expect("read after reopen");
2419	assert!(persisted.is_some());
2420
2421	let _ = fs::remove_dir_all(dir);
2422	}
2423
2424	#[test]
2425	fn record_thread_goal_usage_accumulates_tokens_and_time() {
2426	let store = temp_state_store("thread-goal-usage");
2427	store
2428	.upsert_thread(&test_thread("thread-1"))
2429	.expect("upsert thread");
2430
2431	// Mirror the runtime, which creates goals with zeroed accounting.
2432	let mut goal = test_goal("thread-1", "Ship the persistent goal loop");
2433	goal.tokens_used = 0;
2434	goal.time_used_seconds = 0;
2435	goal.updated_at = 100;
2436	store.upsert_thread_goal(&goal).expect("upsert goal");
2437
2438	// First accrual lands the deltas and advances updated_at.
2439	let after_first = store
2440	.record_thread_goal_usage("thread-1", 250, 12, 150)
2441	.expect("record usage")
2442	.expect("goal exists");
2443	assert_eq!(after_first.tokens_used, 250);
2444	assert_eq!(after_first.time_used_seconds, 12);
2445	assert_eq!(after_first.updated_at, 150);
2446	// Identity fields are preserved across accrual.
2447	assert_eq!(after_first.goal_id, goal.goal_id);
2448	assert_eq!(after_first.objective, goal.objective);
2449	assert_eq!(after_first.status, goal.status);
2450	assert_eq!(after_first.token_budget, goal.token_budget);
2451	assert_eq!(after_first.created_at, goal.created_at);
2452	assert_eq!(after_first.continuation_count, 0);
2453
2454	// Second accrual adds on top of the first (additive, not replacing).
2455	let after_second = store
2456	.record_thread_goal_usage("thread-1", 75, 8, 200)
2457	.expect("record usage")
2458	.expect("goal exists");
2459	assert_eq!(after_second.tokens_used, 325);
2460	assert_eq!(after_second.time_used_seconds, 20);
2461	assert_eq!(after_second.updated_at, 200);
2462
2463	// A stale `now` must not move updated_at backwards.
2464	let after_stale = store
2465	.record_thread_goal_usage("thread-1", 5, 1, 1)
2466	.expect("record usage")
2467	.expect("goal exists");
2468	assert_eq!(after_stale.tokens_used, 330);
2469	assert_eq!(after_stale.time_used_seconds, 21);
2470	assert_eq!(after_stale.updated_at, 200);
2471
2472	// Read back through the normal getter to confirm durability.
2473	let persisted = store
2474	.get_thread_goal("thread-1")
2475	.expect("read goal")
2476	.expect("goal exists");
2477	assert_eq!(persisted.tokens_used, 330);
2478	assert_eq!(persisted.time_used_seconds, 21);
2479	}
2480
2481	#[test]
2482	fn record_thread_goal_usage_returns_none_without_goal() {
2483	let store = temp_state_store("thread-goal-usage-missing");
2484	store
2485	.upsert_thread(&test_thread("thread-1"))
2486	.expect("upsert thread");
2487	// Thread exists but has no goal row yet: accrual is a no-op, not an error,
2488	// and must not create a goal.
2489	let result = store
2490	.record_thread_goal_usage("thread-1", 100, 5, 999)
2491	.expect("record usage on goalless thread");
2492	assert!(result.is_none());
2493	assert!(
2494	store
2495	.get_thread_goal("thread-1")
2496	.expect("read goal")
2497	.is_none()
2498	);
2499	}
2500
2501	#[test]
2502	fn record_thread_goal_continuation_accumulates_durably() {
2503	let store = temp_state_store("thread-goal-continuation");
2504	store
2505	.upsert_thread(&test_thread("thread-1"))
2506	.expect("upsert thread");
2507
2508	let mut goal = test_goal("thread-1", "Keep working across turns");
2509	goal.updated_at = 100;
2510	store.upsert_thread_goal(&goal).expect("upsert goal");
2511
2512	let after_first = store
2513	.record_thread_goal_continuation("thread-1", 120)
2514	.expect("record continuation")
2515	.expect("goal exists");
2516	assert_eq!(after_first.continuation_count, 1);
2517	assert_eq!(after_first.tokens_used, goal.tokens_used);
2518	assert_eq!(after_first.time_used_seconds, goal.time_used_seconds);
2519	assert_eq!(after_first.updated_at, 120);
2520
2521	let after_second = store
2522	.record_thread_goal_continuation("thread-1", 110)
2523	.expect("record second continuation")
2524	.expect("goal exists");
2525	assert_eq!(after_second.continuation_count, 2);
2526	assert_eq!(after_second.updated_at, 120);
2527
2528	let persisted = store
2529	.get_thread_goal("thread-1")
2530	.expect("read goal")
2531	.expect("goal exists");
2532	assert_eq!(persisted.continuation_count, 2);
2533	}
2534
2535	// ── $CODEWHALE_HOME override tests ──────────────────────────────
2536	//
2537	// These touch a process-global env var, so they serialize against each
2538	// other (and restore the prior value) to stay hermetic under parallel test
2539	// runs — the same concern AGENTS.md flags for config_command_allow_shell_*.
2540
2541	static CODEWHALE_HOME_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
2542
2543	struct CodeWhaleHomeGuard {
2544	prior: Option<std::ffi::OsString>,
2545	}
2546	impl CodeWhaleHomeGuard {
2547	fn set(value: &str) -> Self {
2548	let prior = std::env::var_os("CODEWHALE_HOME");
2549	// SAFETY: serialised by CODEWHALE_HOME_TEST_LOCK.
2550	unsafe { std::env::set_var("CODEWHALE_HOME", value) };
2551	Self { prior }
2552	}
2553	fn remove() -> Self {
2554	let prior = std::env::var_os("CODEWHALE_HOME");
2555	// SAFETY: serialised by CODEWHALE_HOME_TEST_LOCK.
2556	unsafe { std::env::remove_var("CODEWHALE_HOME") };
2557	Self { prior }
2558	}
2559	}
2560	impl Drop for CodeWhaleHomeGuard {
2561	fn drop(&mut self) {
2562	// SAFETY: serialised by CODEWHALE_HOME_TEST_LOCK.
2563	unsafe {
2564	match &self.prior {
2565	Some(value) => std::env::set_var("CODEWHALE_HOME", value),
2566	None => std::env::remove_var("CODEWHALE_HOME"),
2567	}
2568	}
2569	}
2570	}
2571
2572	#[test]
2573	fn codewhale_home_override_returns_the_env_value_verbatim() {
2574	let _lock = CODEWHALE_HOME_TEST_LOCK.lock().unwrap();
2575	let override_path = std::env::temp_dir().join("cw-isolated-state");
2576	let _g = CodeWhaleHomeGuard::set(override_path.to_str().unwrap());
2577	// The env var IS the home dir — no ".codewhale" appended. This matches
2578	// codewhale_home() in config ($CODEWHALE_HOME=/x means home is /x).
2579	assert_eq!(
2580	codewhale_home_override().unwrap().as_deref(),
2581	Some(override_path.as_path())
2582	);
2583	}
2584
2585	#[test]
2586	fn codewhale_home_override_none_when_unset() {
2587	let _lock = CODEWHALE_HOME_TEST_LOCK.lock().unwrap();
2588	let _g = CodeWhaleHomeGuard::remove();
2589	assert!(codewhale_home_override().unwrap().is_none());
2590	}
2591
2592	#[test]
2593	fn codewhale_home_override_none_when_whitespace_only() {
2594	let _lock = CODEWHALE_HOME_TEST_LOCK.lock().unwrap();
2595	let _g = CodeWhaleHomeGuard::set(" ");
2596	assert!(
2597	codewhale_home_override().unwrap().is_none(),
2598	"whitespace-only CODEWHALE_HOME must not establish isolation"
2599	);
2600	}
2601
2602	#[test]
2603	fn default_state_db_path_uses_codewhale_home_when_set() {
2604	let _lock = CODEWHALE_HOME_TEST_LOCK.lock().unwrap();
2605	let dir = std::env::temp_dir().join(format!(
2606	"cw-home-state-{}-{}",
2607	std::process::id(),
2608	std::time::SystemTime::now()
2609	.duration_since(std::time::UNIX_EPOCH)
2610	.unwrap()
2611	.as_nanos()
2612	));
2613	let _g = CodeWhaleHomeGuard::set(dir.to_str().unwrap());
2614	// Hard override: the DB is <CODEWHALE_HOME>/state.db, NOT
2615	// <CODEWHALE_HOME>/.codewhale/state.db, and the legacy ~/.deepseek
2616	// fallback is bypassed entirely.
2617	assert_eq!(default_state_db_path(), dir.join("state.db"));
2618	}
2619
2620	#[test]
2621	fn load_checkpoint_propagates_invalid_state_json() {
2622	let store = temp_state_store("checkpoint-parse-error");
2623	store
2624	.upsert_thread(&test_thread("thread-1"))
2625	.expect("upsert thread");
2626	store
2627	.save_checkpoint("thread-1", "broken", &json!({"ok": true}))
2628	.expect("save checkpoint");
2629
2630	{
2631	let conn = store.conn().expect("conn");
2632	conn.execute(
2633	"UPDATE checkpoints SET state_json = ?1 WHERE thread_id = ?2 AND checkpoint_id = ?3",
2634	params!["not-json", "thread-1", "broken"],
2635	)
2636	.expect("corrupt checkpoint");
2637	}
2638
2639	let err = store
2640	.load_checkpoint("thread-1", Some("broken"))
2641	.expect_err("invalid checkpoint json should fail");
2642	assert!(
2643	err.to_string()
2644	.contains("failed to parse checkpoint state json")
2645	);
2646	}
2647
2648	#[test]
2649	fn session_index_compacts_after_threshold() {
2650	let store = temp_state_store("session-index-compact");
2651	for idx in 0..6 {
2652	store
2653	.append_thread_name("thread-1", Some(format!("name-{idx}")), idx, None)
2654	.expect("append session index entry");
2655	}
2656
2657	let line_count = store
2658	.session_index_line_count()
2659	.expect("count session index lines");
2660	assert_eq!(line_count, 1);
2661
2662	let name = store
2663	.find_thread_name_by_id("thread-1")
2664	.expect("lookup thread name");
2665	assert_eq!(name.as_deref(), Some("name-5"));
2666	}
2667
2668	#[test]
2669	fn session_index_read_skips_a_torn_line() {
2670	// #4735: a crash mid-append leaves a truncated final line. Failing the
2671	// whole read broke every thread-name lookup at once, and compaction
2672	// reads through the same path, so the index could not repair itself.
2673	let store = temp_state_store("session-index-torn");
2674	store
2675	.append_thread_name("thread-1", Some("first".to_string()), 1, None)
2676	.expect("append first entry");
2677
2678	{
2679	let mut file = OpenOptions::new()
2680	.append(true)
2681	.open(&store.session_index_path)
2682	.expect("open session index");
2683	// A write cut off mid-JSON, exactly as a crash would leave it.
2684	writeln!(file, "{{\"thread_id\":\"thread-2\",\"thread_na").expect("write torn line");
2685	}
2686
2687	store
2688	.append_thread_name("thread-3", Some("third".to_string()), 3, None)
2689	.expect("append third entry");
2690
2691	assert_eq!(
2692	store
2693	.find_thread_name_by_id("thread-1")
2694	.expect("lookup thread-1")
2695	.as_deref(),
2696	Some("first"),
2697	);
2698	assert_eq!(
2699	store
2700	.find_thread_name_by_id("thread-3")
2701	.expect("lookup thread-3")
2702	.as_deref(),
2703	Some("third"),
2704	);
2705	}
2706
2707	/// Hook fired by compaction between writing the snapshot and renaming it
2708	/// over the live index — the window a concurrent append can be lost in.
2709	/// Only the store whose path a test registered is affected, so tests
2710	/// running in parallel don't disturb each other.
2711	type MidpointHook = Box<dyn Fn() + Send + Sync>;
2712	static COMPACTION_MIDPOINT: Mutex<Option<(PathBuf, MidpointHook)>> = Mutex::new(None);
2713
2714	/// Fires at most once: the racing append compacts too, and a hook that
2715	/// fired twice would re-enter the test's one-shot handshake.
2716	pub(super) fn compaction_midpoint(index_path: &Path) {
2717	let mut hook = COMPACTION_MIDPOINT.lock().expect("midpoint hook lock");
2718	let registered_for_this_store = match hook.as_ref() {
2719	Some((registered, _)) => registered == index_path,
2720	None => return,
2721	};
2722	if !registered_for_this_store {
2723	return;
2724	}
2725	let (_, callback) = hook.take().expect("presence checked above");
2726	drop(hook);
2727	callback();
2728	}
2729
2730	#[test]
2731	fn session_index_compaction_does_not_drop_a_concurrent_append() {
2732	// #4736: compaction snapshots the file, rewrites it, and renames over
2733	// the live one. An append landing between those two steps used to
2734	// vanish — silently, since it had already returned success to its
2735	// caller. The shared lock serializes the two.
2736	//
2737	// The race is real but narrow, so the test drives it deterministically:
2738	// a hook at the compaction midpoint releases the appender and then
2739	// waits. Without the lock the appender writes into the doomed file and
2740	// the rename discards it; with the lock it blocks until compaction
2741	// finishes, and its entry survives.
2742	let store = Arc::new(temp_state_store("session-index-race"));
2743	let threshold = session_index_compact_line_threshold();
2744
2745	// One line short of the threshold, so the next append compacts.
2746	for idx in 0..threshold {
2747	store
2748	.append_thread_name(
2749	&format!("thread-{idx}"),
2750	Some(format!("name-{idx}")),
2751	1,
2752	None,
2753	)
2754	.expect("append filler entry");
2755	}
2756
2757	let appender_released = Arc::new(Barrier::new(2));
2758	{
2759	let released = Arc::clone(&appender_released);
2760	*COMPACTION_MIDPOINT.lock().expect("midpoint hook lock") = Some((
2761	store.session_index_path.clone(),
2762	Box::new(move \|\| {
2763	released.wait();
2764	// Give the appender time to complete its write into the
2765	// window. Under the fix it is blocked on the lock instead.
2766	thread::sleep(Duration::from_millis(300));
2767	}),
2768	));
2769	}
2770
2771	let appender = {
2772	let store = Arc::clone(&store);
2773	let released = Arc::clone(&appender_released);
2774	thread::spawn(move \|\| {
2775	released.wait();
2776	store
2777	.append_thread_name("racer", Some("racer-name".to_string()), 2, None)
2778	.expect("append racing entry");
2779	})
2780	};
2781
2782	store
2783	.append_thread_name("trigger", Some("trigger-name".to_string()), 1, None)
2784	.expect("append entry that triggers compaction");
2785	appender.join().expect("appender thread");
2786	*COMPACTION_MIDPOINT.lock().expect("midpoint hook lock") = None;
2787
2788	assert_eq!(
2789	store
2790	.find_thread_name_by_id("racer")
2791	.expect("lookup racer")
2792	.as_deref(),
2793	Some("racer-name"),
2794	"append was dropped by a concurrent compaction",
2795	);
2796	assert_eq!(
2797	store
2798	.find_thread_name_by_id("trigger")
2799	.expect("lookup trigger")
2800	.as_deref(),
2801	Some("trigger-name"),
2802	);
2803	}
2804	}
2805

2805 lines RUST