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

2764 lines RUST