lib.rs - CodeWhale

lib.rs

1	pub mod context_reference;
2	pub mod fragments;
3	pub mod ids;
4	pub mod journal;
5	pub mod prefix_cache;
6	pub mod request;
7	pub mod role;
8	pub mod session;
9	pub mod tool_parser;
10
11	pub use context_reference::{
12	ContextReference, ContextReferenceKind, ContextReferenceSource, MediaAttachmentReference,
13	media_attachment_references,
14	};
15
16	use std::collections::{HashMap, HashSet};
17	use std::path::{Path, PathBuf};
18	use std::sync::Arc;
19
20	use std::time::Duration;
21
22	use anyhow::{Result, anyhow};
23	use codewhale_config::{ConfigToml, ProviderKind};
24	use codewhale_execpolicy::{
25	AskForApproval, ExecApprovalRequirement, ExecPolicyContext, ExecPolicyDecision,
26	ExecPolicyEngine,
27	};
28	use codewhale_hooks::{HookDispatcher, HookEvent};
29	use codewhale_mcp::{
30	McpManager, McpStartupCompleteEvent, McpStartupStatus as McpManagerStartupStatus,
31	};
32	use codewhale_protocol::{
33	AppResponse, EventFrame, ExecApprovalRequestEvent, ResponseChannel, ReviewDecision, Status,
34	Thread, ThreadForkParams, ThreadGoal, ThreadGoalClearParams, ThreadGoalGetParams,
35	ThreadGoalProgressParams, ThreadGoalSetParams, ThreadGoalStatus, ThreadListParams,
36	ThreadReadParams, ThreadRequest, ThreadResponse, ThreadResumeParams, ThreadSetNameParams,
37	ThreadStatus, ToolPayload, UserInputRequestEvent,
38	};
39	use codewhale_state::{
40	JobStateRecord, JobStateStatus, SessionSource, StateStore, ThreadGoalRecord,
41	ThreadGoalStatus as PersistedThreadGoalStatus, ThreadListFilters, ThreadMetadata,
42	ThreadStatus as PersistedThreadStatus,
43	};
44	use codewhale_tools::{ToolCall, ToolRegistry};
45	use serde_json::{Value, json};
46	use tokio::time;
47	use uuid::Uuid;
48
49	/// Per-tool dispatch budget for the headless runtime. Matches the generous
50	/// subagent default so long-running tools are not cut off prematurely.
51	fn tool_dispatch_timeout() -> Duration {
52	if cfg!(test) {
53	Duration::from_millis(50)
54	} else {
55	Duration::from_secs(300)
56	}
57	}
58
59	/// How a new thread's conversation history is initialized.
60	#[derive(Debug, Clone)]
61	pub enum InitialHistory {
62	/// Start with an empty conversation.
63	New,
64	/// Forked from an existing thread with the given history items.
65	Forked(Vec<Value>),
66	/// Resumed from a persisted thread with its full history.
67	Resumed {
68	conversation_id: String,
69	history: Vec<Value>,
70	rollout_path: PathBuf,
71	},
72	}
73
74	/// Result of spawning or resuming a thread.
75	#[derive(Debug, Clone)]
76	pub struct NewThread {
77	/// The thread metadata.
78	pub thread: Thread,
79	/// Resolved model identifier.
80	pub model: String,
81	/// Provider that serves the model.
82	pub model_provider: String,
83	/// Working directory for the thread.
84	pub cwd: PathBuf,
85	/// Approval policy override, if any.
86	pub approval_policy: Option<String>,
87	/// Sandbox mode override, if any.
88	pub sandbox: Option<String>,
89	}
90
91	/// Status of a background job.
92	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93	pub enum JobStatus {
94	/// Waiting to be picked up.
95	Queued,
96	/// Currently executing.
97	Running,
98	/// Temporarily paused.
99	Paused,
100	/// Finished successfully.
101	Completed,
102	/// Finished with an error.
103	Failed,
104	/// Cancelled by the user.
105	Cancelled,
106	}
107
108	impl Status for JobStatus {
109	fn is_terminal(&self) -> bool {
110	matches!(self, Self::Completed \| Self::Failed \| Self::Cancelled)
111	}
112	fn is_active(&self) -> bool {
113	matches!(self, Self::Queued \| Self::Running)
114	}
115	fn is_paused(&self) -> bool {
116	matches!(self, Self::Paused)
117	}
118	}
119
120	const JOB_DETAIL_SCHEMA_VERSION: u8 = 1;
121	const DEFAULT_JOB_MAX_ATTEMPTS: u32 = 3;
122	const DEFAULT_JOB_BACKOFF_BASE_MS: u64 = 500;
123	const MAX_JOB_HISTORY_ENTRIES: usize = 64;
124
125	/// Retry state for a job that failed and may be retried.
126	#[derive(Debug, Clone)]
127	pub struct JobRetryMetadata {
128	/// Current attempt number (0 = not yet retried).
129	pub attempt: u32,
130	/// Maximum number of retry attempts before giving up.
131	pub max_attempts: u32,
132	/// Base delay in milliseconds for exponential backoff.
133	pub backoff_base_ms: u64,
134	/// Computed delay in milliseconds until the next retry.
135	pub next_backoff_ms: u64,
136	/// Timestamp when the next retry should be attempted.
137	pub next_retry_at: Option<i64>,
138	}
139
140	impl Default for JobRetryMetadata {
141	fn default() -> Self {
142	Self {
143	attempt: 0,
144	max_attempts: DEFAULT_JOB_MAX_ATTEMPTS,
145	backoff_base_ms: DEFAULT_JOB_BACKOFF_BASE_MS,
146	next_backoff_ms: 0,
147	next_retry_at: None,
148	}
149	}
150	}
151
152	/// A single entry in a job's history log.
153	#[derive(Debug, Clone)]
154	pub struct JobHistoryEntry {
155	/// Timestamp when this entry was recorded.
156	pub at: i64,
157	/// Phase name (e.g., "created", "running", "failed").
158	pub phase: String,
159	/// Job status at this point in time.
160	pub status: JobStatus,
161	/// Progress percentage at this point, if available.
162	pub progress: Option<u8>,
163	/// Human-readable detail message.
164	pub detail: Option<String>,
165	/// Retry state snapshot at this point.
166	pub retry: JobRetryMetadata,
167	}
168
169	#[derive(Debug, Clone)]
170	struct PersistedJobDetail {
171	pub status: JobStatus,
172	pub detail: Option<String>,
173	pub retry: JobRetryMetadata,
174	pub history: Vec<JobHistoryEntry>,
175	}
176
177	/// A complete job record with all metadata and history.
178	#[derive(Debug, Clone)]
179	pub struct JobRecord {
180	/// Unique job identifier.
181	pub id: String,
182	/// Human-readable job name.
183	pub name: String,
184	/// Current job status.
185	pub status: JobStatus,
186	/// Current progress percentage (0-100).
187	pub progress: Option<u8>,
188	/// Human-readable detail about the current state.
189	pub detail: Option<String>,
190	/// Retry state for failed jobs.
191	pub retry: JobRetryMetadata,
192	/// Chronological history of state transitions.
193	pub history: Vec<JobHistoryEntry>,
194	/// Timestamp when the job was created.
195	pub created_at: i64,
196	/// Timestamp of the last state change.
197	pub updated_at: i64,
198	}
199
200	/// Map a durable [`JobRecord`] to the dependency-neutral run read model.
201	///
202	/// Pure projection of the record as persisted: unknown budgets stay unset and
203	/// nothing is fabricated. `updated_at` (epoch seconds) provides the terminal
204	/// timestamp because the job manager records no separate end time. The
205	/// free-form job detail is intentionally omitted because this owner does not
206	/// classify it as safe for a cross-surface read model.
207	#[must_use]
208	pub fn job_record_to_agent_run(
209	record: &JobRecord,
210	) -> codewhale_protocol::agent_run::AgentRunSnapshot {
211	use codewhale_protocol::agent_run::{
212	AgentRunSnapshot, BudgetSummary, RunSource, RunState, TerminalOutcome, TerminalSummary,
213	};
214
215	let (state, terminal) = match record.status {
216	JobStatus::Queued => (RunState::Queued, None),
217	JobStatus::Running => (RunState::Running, None),
218	JobStatus::Paused => (RunState::Paused, None),
219	JobStatus::Completed \| JobStatus::Failed \| JobStatus::Cancelled => {
220	let outcome = match record.status {
221	JobStatus::Completed => TerminalOutcome::Completed,
222	JobStatus::Failed => TerminalOutcome::Failed,
223	_ => TerminalOutcome::Cancelled,
224	};
225	(
226	RunState::Terminal,
227	Some(TerminalSummary {
228	outcome,
229	ended_at_ms: record.updated_at.checked_mul(1000),
230	detail: None,
231	}),
232	)
233	}
234	};
235
236	AgentRunSnapshot {
237	run_id: record.id.clone(),
238	parent: None,
239	source: RunSource::CoreJob,
240	state,
241	budget: BudgetSummary::default(),
242	terminal,
243	refs: Vec::new(),
244	}
245	}
246
247	/// Manages background jobs with retry logic and persistence.
248	#[derive(Debug, Default)]
249	pub struct JobManager {
250	jobs: HashMap<String, JobRecord>,
251	}
252
253	impl JobManager {
254	fn now_ts() -> i64 {
255	chrono::Utc::now().timestamp()
256	}
257
258	fn deterministic_backoff_ms(retry: &JobRetryMetadata) -> u64 {
259	if retry.attempt == 0 {
260	return 0;
261	}
262	let exponent = retry.attempt.saturating_sub(1).min(20);
263	let multiplier = 1u64.checked_shl(exponent).unwrap_or(u64::MAX);
264	retry.backoff_base_ms.saturating_mul(multiplier)
265	}
266
267	fn clear_retry_schedule(retry: &mut JobRetryMetadata) {
268	retry.next_backoff_ms = 0;
269	retry.next_retry_at = None;
270	}
271
272	fn push_history(job: &mut JobRecord, phase: &str) {
273	job.history.push(JobHistoryEntry {
274	at: job.updated_at,
275	phase: phase.to_string(),
276	status: job.status,
277	progress: job.progress,
278	detail: job.detail.clone(),
279	retry: job.retry.clone(),
280	});
281	if job.history.len() > MAX_JOB_HISTORY_ENTRIES {
282	let to_drain = job.history.len() - MAX_JOB_HISTORY_ENTRIES;
283	job.history.drain(0..to_drain);
284	}
285	}
286
287	fn parse_persisted_detail(raw: Option<&str>) -> Option<PersistedJobDetail> {
288	let raw = raw?;
289	let parsed: Value = serde_json::from_str(raw).ok()?;
290	let status = parsed
291	.get("status")
292	.and_then(Value::as_str)
293	.and_then(job_status_from_str)?;
294	let detail = parsed.get("detail").and_then(json_optional_string);
295	let retry = parse_retry_metadata(parsed.get("retry"));
296	let history = parsed
297	.get("history")
298	.and_then(Value::as_array)
299	.map(\|items\| {
300	items
301	.iter()
302	.filter_map(parse_history_entry)
303	.collect::<Vec<_>>()
304	})
305	.unwrap_or_default();
306	Some(PersistedJobDetail {
307	status,
308	detail,
309	retry,
310	history,
311	})
312	}
313
314	fn encode_persisted_detail(job: &JobRecord) -> Result<Option<String>> {
315	let encoded = json!({
316	"schema_version": JOB_DETAIL_SCHEMA_VERSION,
317	"status": job_status_to_str(job.status),
318	"detail": job.detail.clone(),
319	"retry": job_retry_to_value(&job.retry),
320	"history": job.history.iter().map(job_history_to_value).collect::<Vec<_>>()
321	})
322	.to_string();
323	Ok(Some(encoded))
324	}
325
326	/// Enqueues a new job and returns its record.
327	pub fn enqueue(&mut self, name: impl Into<String>) -> JobRecord {
328	let now = Self::now_ts();
329	let id = format!("job-{}", Uuid::new_v4());
330	let mut job = JobRecord {
331	id: id.clone(),
332	name: name.into(),
333	status: JobStatus::Queued,
334	progress: Some(0),
335	detail: None,
336	retry: JobRetryMetadata::default(),
337	history: Vec::new(),
338	created_at: now,
339	updated_at: now,
340	};
341	Self::push_history(&mut job, "created");
342	self.jobs.insert(id, job.clone());
343	job
344	}
345
346	/// Transitions a job to running and clears its retry schedule.
347	pub fn set_running(&mut self, id: &str) {
348	if let Some(job) = self.jobs.get_mut(id) {
349	job.status = JobStatus::Running;
350	Self::clear_retry_schedule(&mut job.retry);
351	job.updated_at = Self::now_ts();
352	Self::push_history(job, "running");
353	}
354	}
355
356	/// Updates a job's progress (clamped to 100) and optional detail message.
357	pub fn update_progress(&mut self, id: &str, progress: u8, detail: Option<String>) {
358	if let Some(job) = self.jobs.get_mut(id) {
359	job.progress = Some(progress.min(100));
360	job.detail = detail;
361	job.updated_at = Self::now_ts();
362	Self::push_history(job, "progress_updated");
363	}
364	}
365
366	/// Marks a job as completed with 100% progress and clears its retry schedule.
367	pub fn complete(&mut self, id: &str) {
368	if let Some(job) = self.jobs.get_mut(id) {
369	job.status = JobStatus::Completed;
370	job.progress = Some(100);
371	Self::clear_retry_schedule(&mut job.retry);
372	job.updated_at = Self::now_ts();
373	Self::push_history(job, "completed");
374	}
375	}
376
377	/// Marks a job as failed and schedules a retry if attempts remain.
378	pub fn fail(&mut self, id: &str, detail: impl Into<String>) {
379	if let Some(job) = self.jobs.get_mut(id) {
380	let now = Self::now_ts();
381	job.status = JobStatus::Failed;
382	job.detail = Some(detail.into());
383	if job.retry.attempt < job.retry.max_attempts {
384	job.retry.attempt += 1;
385	job.retry.next_backoff_ms = Self::deterministic_backoff_ms(&job.retry);
386	let delay_secs = ((job.retry.next_backoff_ms.saturating_add(999)) / 1000)
387	.min(i64::MAX as u64) as i64;
388	job.retry.next_retry_at = Some(now.saturating_add(delay_secs));
389	} else {
390	Self::clear_retry_schedule(&mut job.retry);
391	}
392	job.updated_at = now;
393	Self::push_history(job, "failed");
394	}
395	}
396
397	/// Cancels a job and clears any pending retry schedule.
398	pub fn cancel(&mut self, id: &str) {
399	if let Some(job) = self.jobs.get_mut(id) {
400	job.status = JobStatus::Cancelled;
401	Self::clear_retry_schedule(&mut job.retry);
402	job.updated_at = Self::now_ts();
403	Self::push_history(job, "cancelled");
404	}
405	}
406
407	/// Pauses a job, optionally updating its detail message.
408	pub fn pause(&mut self, id: &str, detail: Option<String>) {
409	if let Some(job) = self.jobs.get_mut(id) {
410	job.status = JobStatus::Paused;
411	if detail.is_some() {
412	job.detail = detail;
413	}
414	job.updated_at = Self::now_ts();
415	Self::push_history(job, "paused");
416	}
417	}
418
419	/// Resumes a paused or failed job back to running status.
420	pub fn resume(&mut self, id: &str, detail: Option<String>) {
421	if let Some(job) = self.jobs.get_mut(id) {
422	job.status = JobStatus::Running;
423	if detail.is_some() {
424	job.detail = detail;
425	}
426	Self::clear_retry_schedule(&mut job.retry);
427	job.updated_at = Self::now_ts();
428	Self::push_history(job, "resumed");
429	}
430	}
431
432	/// Returns all jobs sorted by most recently updated first.
433	pub fn list(&self) -> Vec<JobRecord> {
434	let mut out = self.jobs.values().cloned().collect::<Vec<_>>();
435	out.sort_by_key(\|job\| std::cmp::Reverse(job.updated_at));
436	out
437	}
438
439	/// Returns the history entries for a job, or an empty vec if not found.
440	pub fn history(&self, id: &str) -> Vec<JobHistoryEntry> {
441	self.jobs
442	.get(id)
443	.map(\|job\| job.history.clone())
444	.unwrap_or_default()
445	}
446
447	/// Resets queued or running jobs back to queued on application resume.
448	pub fn resume_pending(&mut self) -> Vec<JobRecord> {
449	let mut resumed = Vec::new();
450	for job in self.jobs.values_mut() {
451	if matches!(job.status, JobStatus::Queued \| JobStatus::Running) {
452	job.status = JobStatus::Queued;
453	job.updated_at = Self::now_ts();
454	Self::push_history(job, "queued_after_resume");
455	resumed.push(job.clone());
456	}
457	}
458	resumed
459	}
460
461	/// Loads jobs from the state store, deserializing extended detail when available.
462	pub fn load_from_store(&mut self, store: &StateStore) -> Result<()> {
463	let persisted = store.list_jobs(Some(500))?;
464	for job in persisted {
465	let fallback_status = job_state_status_to_runtime(job.status);
466	let parsed = Self::parse_persisted_detail(job.detail.as_deref());
467	let (status, detail, retry, history) = if let Some(detail_state) = parsed {
468	(
469	detail_state.status,
470	detail_state.detail,
471	detail_state.retry,
472	detail_state.history,
473	)
474	} else {
475	(
476	fallback_status,
477	job.detail,
478	JobRetryMetadata::default(),
479	Vec::new(),
480	)
481	};
482	self.jobs.insert(
483	job.id.clone(),
484	JobRecord {
485	id: job.id,
486	name: job.name,
487	status,
488	progress: job.progress,
489	detail,
490	retry,
491	history,
492	created_at: job.created_at,
493	updated_at: job.updated_at,
494	},
495	);
496	}
497	Ok(())
498	}
499
500	/// Persists a single job's current state to the state store.
501	pub fn persist_job(&self, store: &StateStore, id: &str) -> Result<()> {
502	let Some(job) = self.jobs.get(id) else {
503	return Ok(());
504	};
505	let encoded_detail = Self::encode_persisted_detail(job)?;
506	store.upsert_job(&JobStateRecord {
507	id: job.id.clone(),
508	name: job.name.clone(),
509	status: runtime_status_to_job_state(job.status),
510	progress: job.progress,
511	detail: encoded_detail,
512	created_at: job.created_at,
513	updated_at: job.updated_at,
514	})
515	}
516
517	/// Persists all in-memory jobs to the state store.
518	pub fn persist_all(&self, store: &StateStore) -> Result<()> {
519	for id in self.jobs.keys() {
520	self.persist_job(store, id)?;
521	}
522	Ok(())
523	}
524	}
525
526	/// Manages thread lifecycle: spawn, resume, fork, archive, and persistence.
527	pub struct ThreadManager {
528	store: StateStore,
529	running_threads: HashMap<String, Thread>,
530	cli_version: String,
531	}
532
533	impl ThreadManager {
534	/// Creates a new `ThreadManager` backed by the given state store.
535	pub fn new(store: StateStore) -> Self {
536	Self {
537	store,
538	running_threads: HashMap::new(),
539	cli_version: env!("CARGO_PKG_VERSION").to_string(),
540	}
541	}
542
543	/// Returns a reference to the underlying state store.
544	pub fn state_store(&self) -> &StateStore {
545	&self.store
546	}
547
548	/// Spawns a new thread with the given initial history and persists it.
549	pub fn spawn_thread_with_history(
550	&mut self,
551	model_provider: String,
552	cwd: PathBuf,
553	initial_history: InitialHistory,
554	persist_extended_history: bool,
555	) -> Result<NewThread> {
556	let id = format!("thread-{}", Uuid::new_v4());
557	let now = chrono::Utc::now().timestamp();
558	let preview = preview_from_initial_history(&initial_history);
559	let source = match initial_history {
560	InitialHistory::New => SessionSource::Interactive,
561	InitialHistory::Forked(_) => SessionSource::Fork,
562	InitialHistory::Resumed { .. } => SessionSource::Resume,
563	};
564	let thread = Thread {
565	id: id.clone(),
566	preview,
567	ephemeral: !persist_extended_history,
568	model_provider: model_provider.clone(),
569	created_at: now,
570	updated_at: now,
571	status: ThreadStatus::Running,
572	path: None,
573	cwd: cwd.clone(),
574	cli_version: self.cli_version.clone(),
575	source: match source {
576	SessionSource::Interactive => codewhale_protocol::SessionSource::Interactive,
577	SessionSource::Resume => codewhale_protocol::SessionSource::Resume,
578	SessionSource::Fork => codewhale_protocol::SessionSource::Fork,
579	SessionSource::Api => codewhale_protocol::SessionSource::Api,
580	SessionSource::Unknown => codewhale_protocol::SessionSource::Unknown,
581	},
582	name: None,
583	};
584	self.persist_thread(&thread, None)?;
585	match &initial_history {
586	InitialHistory::Forked(items) => {
587	for item in items {
588	self.store.append_message(
589	&thread.id,
590	"history",
591	&item.to_string(),
592	Some(item.clone()),
593	)?;
594	}
595	}
596	InitialHistory::Resumed { history, .. } => {
597	for item in history {
598	self.store.append_message(
599	&thread.id,
600	"history",
601	&item.to_string(),
602	Some(item.clone()),
603	)?;
604	}
605	}
606	InitialHistory::New => {}
607	}
608	self.running_threads
609	.insert(thread.id.clone(), thread.clone());
610	Ok(NewThread {
611	thread,
612	model: "auto".to_string(),
613	model_provider,
614	cwd,
615	approval_policy: None,
616	sandbox: None,
617	})
618	}
619
620	/// Resumes an existing thread, returning `None` if not found.
621	pub fn resume_thread_with_history(
622	&mut self,
623	params: &ThreadResumeParams,
624	fallback_cwd: &Path,
625	model_provider: String,
626	) -> Result<Option<NewThread>> {
627	if params.history.is_none()
628	&& let Some(thread) = self.running_threads.get(&params.thread_id).cloned()
629	{
630	return Ok(Some(NewThread {
631	model: params.model.clone().unwrap_or_else(\|\| "auto".to_string()),
632	model_provider: params.model_provider.clone().unwrap_or(model_provider),
633	cwd: params.cwd.clone().unwrap_or_else(\|\| thread.cwd.clone()),
634	approval_policy: params.approval_policy.clone(),
635	sandbox: params.sandbox.clone(),
636	thread,
637	}));
638	}
639
640	let persisted = self.store.get_thread(&params.thread_id)?;
641	let Some(metadata) = persisted else {
642	return Ok(None);
643	};
644	let mut thread = to_protocol_thread(metadata);
645	thread.status = ThreadStatus::Running;
646	thread.updated_at = chrono::Utc::now().timestamp();
647	thread.cwd = params
648	.cwd
649	.clone()
650	.unwrap_or_else(\|\| fallback_cwd.to_path_buf());
651	self.persist_thread(&thread, None)?;
652	self.running_threads
653	.insert(thread.id.clone(), thread.clone());
654	if let Some(history) = params.history.as_ref() {
655	// A read→resume flow hands back items that are already on the
656	// persisted chain; appending them again would double the
657	// conversation on every resume, compounding. Dedup by content
658	// fingerprint (the item's JSON, matching what append_message
659	// stores as content) against the persisted chain and against
660	// items already appended in this loop.
661	let mut seen: HashSet<String> = self
662	.store
663	.list_messages(&thread.id, None)?
664	.into_iter()
665	.map(\|message\| {
666	message
667	.item
668	.as_ref()
669	.map_or(message.content.clone(), \|item\| item.to_string())
670	})
671	.collect();
672	for item in history {
673	let fingerprint = item.to_string();
674	if !seen.insert(fingerprint.clone()) {
675	continue;
676	}
677	self.store.append_message(
678	&thread.id,
679	"history",
680	&fingerprint,
681	Some(item.clone()),
682	)?;
683	}
684	}
685
686	Ok(Some(NewThread {
687	model: params.model.clone().unwrap_or_else(\|\| "auto".to_string()),
688	model_provider: params.model_provider.clone().unwrap_or(model_provider),
689	cwd: thread.cwd.clone(),
690	approval_policy: params.approval_policy.clone(),
691	sandbox: params.sandbox.clone(),
692	thread,
693	}))
694	}
695
696	/// Forks an existing thread into a new one, inheriting the parent's provider.
697	pub fn fork_thread(
698	&mut self,
699	params: &ThreadForkParams,
700	fallback_cwd: &Path,
701	) -> Result<Option<NewThread>> {
702	let parent = self.store.get_thread(&params.thread_id)?;
703	let Some(parent) = parent else {
704	return Ok(None);
705	};
706	let parent_thread = to_protocol_thread(parent);
707	let new = self.spawn_thread_with_history(
708	params
709	.model_provider
710	.clone()
711	.unwrap_or_else(\|\| parent_thread.model_provider.clone()),
712	params
713	.cwd
714	.clone()
715	.unwrap_or_else(\|\| fallback_cwd.to_path_buf()),
716	InitialHistory::Forked(vec![json!({
717	"type": "fork",
718	"from_thread_id": parent_thread.id
719	})]),
720	params.persist_extended_history,
721	)?;
722	Ok(Some(new))
723	}
724
725	/// Lists threads matching the given filter parameters.
726	pub fn list_threads(&self, params: &ThreadListParams) -> Result<Vec<Thread>> {
727	let list = self.store.list_threads(ThreadListFilters {
728	include_archived: params.include_archived,
729	limit: params.limit,
730	})?;
731	Ok(list.into_iter().map(to_protocol_thread).collect())
732	}
733
734	/// Reads a single thread by id, or `None` if not found.
735	pub fn read_thread(&self, params: &ThreadReadParams) -> Result<Option<Thread>> {
736	Ok(self
737	.store
738	.get_thread(&params.thread_id)?
739	.map(to_protocol_thread))
740	}
741
742	/// Sets the display name for a thread, returning the updated thread or `None`.
743	pub fn set_thread_name(&mut self, params: &ThreadSetNameParams) -> Result<Option<Thread>> {
744	let Some(mut metadata) = self.store.get_thread(&params.thread_id)? else {
745	return Ok(None);
746	};
747	metadata.name = Some(params.name.clone());
748	metadata.updated_at = chrono::Utc::now().timestamp();
749	self.store.upsert_thread(&metadata)?;
750	let updated = to_protocol_thread(metadata);
751	self.running_threads
752	.insert(updated.id.clone(), updated.clone());
753	Ok(Some(updated))
754	}
755
756	/// Sets or replaces the persisted goal for a thread.
757	pub fn set_thread_goal(&mut self, params: &ThreadGoalSetParams) -> Result<Option<ThreadGoal>> {
758	if self.store.get_thread(&params.thread_id)?.is_none() {
759	return Ok(None);
760	}
761	let now = chrono::Utc::now().timestamp();
762	let goal = ThreadGoalRecord {
763	thread_id: params.thread_id.clone(),
764	goal_id: format!("goal-{}", Uuid::new_v4()),
765	objective: params.objective.clone(),
766	status: PersistedThreadGoalStatus::Active,
767	token_budget: params.token_budget,
768	tokens_used: 0,
769	time_used_seconds: 0,
770	continuation_count: 0,
771	last_gap_fingerprint: None,
772	repeated_gap_count: 0,
773	last_gap_pass: None,
774	pause_reason: None,
775	created_at: now,
776	updated_at: now,
777	};
778	self.store.upsert_thread_goal(&goal)?;
779	Ok(Some(to_protocol_goal(goal)))
780	}
781
782	/// Reads the persisted goal for a thread.
783	pub fn get_thread_goal(&self, params: &ThreadGoalGetParams) -> Result<Option<ThreadGoal>> {
784	Ok(self
785	.store
786	.get_thread_goal(&params.thread_id)?
787	.map(to_protocol_goal))
788	}
789
790	/// Accrues durable per-goal usage and/or a continuation pass for a thread.
791	pub fn record_thread_goal_progress(
792	&mut self,
793	params: &ThreadGoalProgressParams,
794	) -> Result<Option<ThreadGoal>> {
795	if self.store.get_thread(&params.thread_id)?.is_none() {
796	return Ok(None);
797	}
798
799	let now = chrono::Utc::now().timestamp();
800	let mut goal = if params.token_delta != 0 \|\| params.time_delta_seconds != 0 {
801	self.store.record_thread_goal_usage(
802	&params.thread_id,
803	params.token_delta,
804	params.time_delta_seconds,
805	now,
806	)?
807	} else {
808	self.store.get_thread_goal(&params.thread_id)?
809	};
810
811	if params.record_continuation {
812	goal = self
813	.store
814	.record_thread_goal_continuation(&params.thread_id, now)?;
815	}
816
817	Ok(goal.map(to_protocol_goal))
818	}
819
820	/// Clears the persisted goal for a thread, returning whether one existed.
821	pub fn clear_thread_goal(&mut self, params: &ThreadGoalClearParams) -> Result<bool> {
822	self.store.delete_thread_goal(&params.thread_id)
823	}
824
825	/// Archives a thread so it no longer appears in default listings.
826	pub fn archive_thread(&mut self, thread_id: &str) -> Result<()> {
827	self.store.mark_archived(thread_id)?;
828	if let Some(thread) = self.running_threads.get_mut(thread_id) {
829	thread.status = ThreadStatus::Archived;
830	}
831	Ok(())
832	}
833
834	/// Restores an archived thread to active status.
835	pub fn unarchive_thread(&mut self, thread_id: &str) -> Result<()> {
836	self.store.mark_unarchived(thread_id)?;
837	if let Some(metadata) = self.store.get_thread(thread_id)? {
838	let thread = to_protocol_thread(metadata);
839	if let Some(cached) = self.running_threads.get_mut(thread_id) {
840	*cached = thread;
841	}
842	}
843	Ok(())
844	}
845
846	/// Records a user message in a thread and updates its preview and timestamp.
847	pub fn touch_message(&mut self, thread_id: &str, input: &str) -> Result<()> {
848	let Some(mut metadata) = self.store.get_thread(thread_id)? else {
849	return Ok(());
850	};
851	metadata.updated_at = chrono::Utc::now().timestamp();
852	metadata.preview = truncate_preview(input);
853	metadata.status = PersistedThreadStatus::Running;
854	self.store.upsert_thread(&metadata)?;
855	if let Some(thread) = self.running_threads.get_mut(thread_id) {
856	thread.updated_at = metadata.updated_at;
857	thread.preview = metadata.preview;
858	thread.status = ThreadStatus::Running;
859	}
860	let message_id = self.store.append_message(thread_id, "user", input, None)?;
861	self.store.save_checkpoint(
862	thread_id,
863	"latest",
864	&json!({
865	"reason": "thread_message",
866	"message_id": message_id,
867	"role": "user",
868	"preview": truncate_preview(input),
869	"updated_at": metadata.updated_at
870	}),
871	)?;
872	Ok(())
873	}
874
875	fn persist_thread(&self, thread: &Thread, rollout_path: Option<PathBuf>) -> Result<()> {
876	// This update payload carries no per-thread policy, so preserve any
877	// policy already stored for the thread rather than erasing it with
878	// NULLs on every persist/resume.
879	let existing = self.store.get_thread(&thread.id)?;
880	self.store.upsert_thread(&ThreadMetadata {
881	id: thread.id.clone(),
882	rollout_path,
883	preview: thread.preview.clone(),
884	ephemeral: thread.ephemeral,
885	model_provider: thread.model_provider.clone(),
886	created_at: thread.created_at,
887	updated_at: thread.updated_at,
888	status: to_persisted_status(&thread.status),
889	path: thread.path.clone(),
890	cwd: thread.cwd.clone(),
891	cli_version: thread.cli_version.clone(),
892	source: to_persisted_source(&thread.source),
893	name: thread.name.clone(),
894	sandbox_policy: existing
895	.as_ref()
896	.and_then(\|metadata\| metadata.sandbox_policy.clone()),
897	approval_mode: existing
898	.as_ref()
899	.and_then(\|metadata\| metadata.approval_mode.clone()),
900	archived: matches!(thread.status, ThreadStatus::Archived),
901	archived_at: None,
902	git_sha: None,
903	git_branch: None,
904	git_origin_url: None,
905	memory_mode: None,
906	current_leaf_id: None,
907	})
908	}
909	}
910
911	/// Top-level runtime combining config, threads, tools, MCP, and hooks.
912	pub struct Runtime {
913	/// Resolved application configuration.
914	pub config: ConfigToml,
915	/// Manages conversation thread lifecycle.
916	pub thread_manager: ThreadManager,
917	/// Registry of callable tools.
918	pub tool_registry: Arc<ToolRegistry>,
919	/// Manager for MCP server connections.
920	pub mcp_manager: Arc<McpManager>,
921	/// Engine for evaluating execution policy decisions.
922	pub exec_policy: ExecPolicyEngine,
923	/// Dispatcher for lifecycle hooks.
924	pub hooks: HookDispatcher,
925	/// Manager for background job lifecycle.
926	pub jobs: JobManager,
927	}
928
929	impl Runtime {
930	/// Constructs a new `Runtime`, loading existing jobs from the state store.
931	pub fn new(
932	config: ConfigToml,
933	state: StateStore,
934	tool_registry: Arc<ToolRegistry>,
935	mcp_manager: Arc<McpManager>,
936	exec_policy: ExecPolicyEngine,
937	hooks: HookDispatcher,
938	) -> Self {
939	let mut jobs = JobManager::default();
940	if let Err(e) = jobs.load_from_store(&state) {
941	tracing::warn!("Failed to load job store, starting with empty job list: {e}");
942	}
943	Self {
944	config,
945	thread_manager: ThreadManager::new(state),
946	tool_registry,
947	mcp_manager,
948	exec_policy,
949	hooks,
950	jobs,
951	}
952	}
953
954	/// Update the live configuration in-place so the next turn picks up
955	/// changes without a restart. Called by the app-server after
956	/// `ConfigSet` or `ConfigUnset`.
957	///
958	/// Only `config.toml` is touched by those operations, so the sibling
959	/// `permissions.toml` (and therefore `exec_policy`) is left unchanged.
960	///
961	/// Fields that the TUI caches on its `App` struct (`api_provider`,
962	/// `reasoning_effort`, `mcp_config_path`, `skills_dir`, …) are read
963	/// live from `self.config` here via `resolve_runtime_options`, so they
964	/// take effect on the next prompt turn without any extra plumbing.
965	pub fn update_config(&mut self, config: ConfigToml) {
966	self.config = config;
967	}
968
969	/// Reload the live configuration and the exec policy from a
970	/// freshly-loaded `ConfigStore`. Used by the app-server's
971	/// `ConfigReload` request, which re-reads both `config.toml` and the
972	/// sibling `permissions.toml` from disk.
973	///
974	/// Unlike `update_config`, this also refreshes `self.exec_policy` so
975	/// externally edited permission rules take effect without a restart.
976	///
977	/// Mirrors the TUI `reload_runtime_config` codepath for everything
978	/// that is reachable from the headless `Runtime`. The TUI-only caches
979	/// (`last_effective_reasoning_effort`, `model_compaction_budget`,
980	/// `ui_locale`, …) do not exist on `Runtime` and need no work here.
981	///
982	/// Not refreshed by this call:
983	/// * `mcp_manager` — MCP server connections are loaded once at
984	/// startup from `mcp_config_path`. Changing `mcp_config_path` or the
985	/// referenced `mcp.json` still requires a headless-runtime restart;
986	/// the TUI owns a separate explicit `/mcp reload` operation.
987	/// * `tool_registry` — built once at startup.
988	pub fn reload_config_and_policy(&mut self, config: ConfigToml, exec_policy: ExecPolicyEngine) {
989	self.config = config;
990	self.exec_policy = exec_policy;
991	}
992
993	fn persisted_thread_data(&self, thread_id: &str) -> Result<Value> {
994	let history = self
995	.thread_manager
996	.state_store()
997	.list_messages(thread_id, Some(500))?
998	.into_iter()
999	.map(\|message\| {
1000	json!({
1001	"id": message.id,
1002	"role": message.role,
1003	"content": message.content,
1004	"item": message.item,
1005	"created_at": message.created_at
1006	})
1007	})
1008	.collect::<Vec<_>>();
1009
1010	let checkpoint = self
1011	.thread_manager
1012	.state_store()
1013	.load_checkpoint(thread_id, None)?
1014	.map(\|record\| {
1015	json!({
1016	"checkpoint_id": record.checkpoint_id,
1017	"state": record.state,
1018	"created_at": record.created_at
1019	})
1020	});
1021
1022	let goal = self
1023	.thread_manager
1024	.state_store()
1025	.get_thread_goal(thread_id)?
1026	.map(to_protocol_goal);
1027
1028	Ok(json!({
1029	"history": history,
1030	"checkpoint": checkpoint,
1031	"goal": goal
1032	}))
1033	}
1034
1035	/// Dispatches a thread request (create, start, resume, fork, list, read, etc.).
1036	pub async fn handle_thread(&mut self, req: ThreadRequest) -> Result<ThreadResponse> {
1037	match req {
1038	ThreadRequest::Create { .. } => {
1039	let cwd = std::env::current_dir().unwrap_or_else(\|_\| PathBuf::from("."));
1040	let new = self.thread_manager.spawn_thread_with_history(
1041	"deepseek".to_string(),
1042	cwd,
1043	InitialHistory::New,
1044	false,
1045	)?;
1046	let mut response = thread_response_from_new("created", new);
1047	response.data = self.persisted_thread_data(&response.thread_id)?;
1048	Ok(response)
1049	}
1050	ThreadRequest::Start(params) => {
1051	let cwd = params.cwd.clone().unwrap_or_else(\|\| {
1052	std::env::current_dir().unwrap_or_else(\|_\| PathBuf::from("."))
1053	});
1054	let new = self.thread_manager.spawn_thread_with_history(
1055	params
1056	.model_provider
1057	.clone()
1058	.unwrap_or_else(\|\| "deepseek".to_string()),
1059	cwd,
1060	InitialHistory::New,
1061	params.persist_extended_history,
1062	)?;
1063	let mut response = thread_response_from_new("started", new);
1064	response.data = self.persisted_thread_data(&response.thread_id)?;
1065	Ok(response)
1066	}
1067	ThreadRequest::Resume(params) => {
1068	let fallback_cwd = std::env::current_dir().unwrap_or_else(\|_\| PathBuf::from("."));
1069	if let Some(new) = self.thread_manager.resume_thread_with_history(
1070	&params,
1071	&fallback_cwd,
1072	"deepseek".to_string(),
1073	)? {
1074	let mut response = thread_response_from_new("resumed", new);
1075	response.data = self.persisted_thread_data(&response.thread_id)?;
1076	Ok(response)
1077	} else {
1078	Ok(ThreadResponse {
1079	thread_id: params.thread_id,
1080	status: "missing".to_string(),
1081	thread: None,
1082	threads: Vec::new(),
1083	goal: None,
1084	model: None,
1085	model_provider: None,
1086	cwd: None,
1087	approval_policy: params.approval_policy,
1088	sandbox: params.sandbox,
1089	events: Vec::new(),
1090	data: json!({"error":"thread not found"}),
1091	})
1092	}
1093	}
1094	ThreadRequest::Fork(params) => {
1095	let cwd = std::env::current_dir().unwrap_or_else(\|_\| PathBuf::from("."));
1096	if let Some(new) = self.thread_manager.fork_thread(&params, &cwd)? {
1097	let mut response = thread_response_from_new("forked", new);
1098	response.data = self.persisted_thread_data(&response.thread_id)?;
1099	Ok(response)
1100	} else {
1101	Ok(ThreadResponse {
1102	thread_id: params.thread_id,
1103	status: "missing".to_string(),
1104	thread: None,
1105	threads: Vec::new(),
1106	goal: None,
1107	model: None,
1108	model_provider: None,
1109	cwd: None,
1110	approval_policy: params.approval_policy,
1111	sandbox: params.sandbox,
1112	events: Vec::new(),
1113	data: json!({"error":"thread not found"}),
1114	})
1115	}
1116	}
1117	ThreadRequest::List(params) => Ok(ThreadResponse {
1118	thread_id: "list".to_string(),
1119	status: "ok".to_string(),
1120	thread: None,
1121	threads: self.thread_manager.list_threads(&params)?,
1122	goal: None,
1123	model: None,
1124	model_provider: None,
1125	cwd: None,
1126	approval_policy: None,
1127	sandbox: None,
1128	events: Vec::new(),
1129	data: json!({}),
1130	}),
1131	ThreadRequest::Read(params) => {
1132	let id = params.thread_id.clone();
1133	let data = self.persisted_thread_data(&id)?;
1134	Ok(ThreadResponse {
1135	thread_id: id,
1136	status: "ok".to_string(),
1137	thread: self.thread_manager.read_thread(&params)?,
1138	threads: Vec::new(),
1139	goal: self.thread_manager.get_thread_goal(&ThreadGoalGetParams {
1140	thread_id: params.thread_id,
1141	})?,
1142	model: None,
1143	model_provider: None,
1144	cwd: None,
1145	approval_policy: None,
1146	sandbox: None,
1147	events: Vec::new(),
1148	data,
1149	})
1150	}
1151	ThreadRequest::SetName(params) => Ok(ThreadResponse {
1152	thread_id: params.thread_id.clone(),
1153	status: "ok".to_string(),
1154	thread: self.thread_manager.set_thread_name(&params)?,
1155	threads: Vec::new(),
1156	goal: None,
1157	model: None,
1158	model_provider: None,
1159	cwd: None,
1160	approval_policy: None,
1161	sandbox: None,
1162	events: Vec::new(),
1163	data: json!({}),
1164	}),
1165	ThreadRequest::GoalSet(params) => {
1166	let thread_id = params.thread_id.clone();
1167	if let Some(goal) = self.thread_manager.set_thread_goal(&params)? {
1168	Ok(ThreadResponse {
1169	thread_id,
1170	status: "ok".to_string(),
1171	thread: None,
1172	threads: Vec::new(),
1173	goal: Some(goal.clone()),
1174	model: None,
1175	model_provider: None,
1176	cwd: None,
1177	approval_policy: None,
1178	sandbox: None,
1179	events: vec![EventFrame::ThreadGoalUpdated { goal: goal.clone() }],
1180	data: json!({ "goal": goal }),
1181	})
1182	} else {
1183	Ok(ThreadResponse {
1184	thread_id,
1185	status: "missing".to_string(),
1186	thread: None,
1187	threads: Vec::new(),
1188	goal: None,
1189	model: None,
1190	model_provider: None,
1191	cwd: None,
1192	approval_policy: None,
1193	sandbox: None,
1194	events: Vec::new(),
1195	data: json!({"error":"thread not found"}),
1196	})
1197	}
1198	}
1199	ThreadRequest::GoalGet(params) => {
1200	let goal = self.thread_manager.get_thread_goal(&params)?;
1201	Ok(ThreadResponse {
1202	thread_id: params.thread_id,
1203	status: "ok".to_string(),
1204	thread: None,
1205	threads: Vec::new(),
1206	goal: goal.clone(),
1207	model: None,
1208	model_provider: None,
1209	cwd: None,
1210	approval_policy: None,
1211	sandbox: None,
1212	events: Vec::new(),
1213	data: json!({ "goal": goal }),
1214	})
1215	}
1216	ThreadRequest::GoalClear(params) => {
1217	let thread_id = params.thread_id.clone();
1218	let cleared = self.thread_manager.clear_thread_goal(&params)?;
1219	Ok(ThreadResponse {
1220	thread_id: thread_id.clone(),
1221	status: if cleared { "cleared" } else { "empty" }.to_string(),
1222	thread: None,
1223	threads: Vec::new(),
1224	goal: None,
1225	model: None,
1226	model_provider: None,
1227	cwd: None,
1228	approval_policy: None,
1229	sandbox: None,
1230	events: if cleared {
1231	vec![EventFrame::ThreadGoalCleared { thread_id }]
1232	} else {
1233	Vec::new()
1234	},
1235	data: json!({ "cleared": cleared }),
1236	})
1237	}
1238	ThreadRequest::GoalRecordProgress(params) => {
1239	let thread_id = params.thread_id.clone();
1240	if let Some(goal) = self.thread_manager.record_thread_goal_progress(&params)? {
1241	Ok(ThreadResponse {
1242	thread_id,
1243	status: "ok".to_string(),
1244	thread: None,
1245	threads: Vec::new(),
1246	goal: Some(goal.clone()),
1247	model: None,
1248	model_provider: None,
1249	cwd: None,
1250	approval_policy: None,
1251	sandbox: None,
1252	events: vec![EventFrame::ThreadGoalUpdated { goal: goal.clone() }],
1253	data: json!({ "goal": goal }),
1254	})
1255	} else {
1256	Ok(ThreadResponse {
1257	thread_id,
1258	status: "missing".to_string(),
1259	thread: None,
1260	threads: Vec::new(),
1261	goal: None,
1262	model: None,
1263	model_provider: None,
1264	cwd: None,
1265	approval_policy: None,
1266	sandbox: None,
1267	events: Vec::new(),
1268	data: json!({"error":"thread or goal not found"}),
1269	})
1270	}
1271	}
1272	ThreadRequest::Archive { thread_id } => {
1273	self.thread_manager.archive_thread(&thread_id)?;
1274	Ok(ThreadResponse {
1275	thread_id,
1276	status: "archived".to_string(),
1277	thread: None,
1278	threads: Vec::new(),
1279	goal: None,
1280	model: None,
1281	model_provider: None,
1282	cwd: None,
1283	approval_policy: None,
1284	sandbox: None,
1285	events: Vec::new(),
1286	data: json!({}),
1287	})
1288	}
1289	ThreadRequest::Unarchive { thread_id } => {
1290	self.thread_manager.unarchive_thread(&thread_id)?;
1291	Ok(ThreadResponse {
1292	thread_id,
1293	status: "unarchived".to_string(),
1294	thread: None,
1295	threads: Vec::new(),
1296	goal: None,
1297	model: None,
1298	model_provider: None,
1299	cwd: None,
1300	approval_policy: None,
1301	sandbox: None,
1302	events: Vec::new(),
1303	data: json!({}),
1304	})
1305	}
1306	// A thread message is a turn, and this type is not the turn
1307	// engine — it owns thread bookkeeping and persistence only. The
1308	// app-server routes messages through its runtime bridge
1309	// (`POST /v1/threads/{id}/turns` on the runtime API) and never
1310	// reaches this arm. Returning an error rather than a canned
1311	// "accepted" keeps any other caller from mistaking bookkeeping
1312	// for execution.
1313	ThreadRequest::Message { thread_id, .. } => Err(anyhow!(
1314	"thread message for {thread_id} cannot be executed here: \
1315	Runtime::handle_thread does not run turns. Send it through the \
1316	app-server runtime bridge (POST /v1/threads/{{id}}/turns)."
1317	)),
1318	}
1319	}
1320
1321	/// Evaluates execution policy and dispatches a tool call.
1322	pub async fn invoke_tool(
1323	&self,
1324	call: ToolCall,
1325	approval_mode: AskForApproval,
1326	cwd: &Path,
1327	) -> Result<Value> {
1328	let fallback_cwd = cwd.display().to_string();
1329	let (command, policy_cwd, execution_kind) = call.execution_subject(&fallback_cwd);
1330	let policy_tool = match &call.payload {
1331	ToolPayload::LocalShell { .. } => "exec_shell",
1332	_ => call.name.as_str(),
1333	};
1334	let policy_path = permission_path_for_call(&call);
1335	let decision = self.exec_policy.check(ExecPolicyContext {
1336	command: &command,
1337	cwd: &policy_cwd,
1338	tool: Some(policy_tool),
1339	path: policy_path.as_deref(),
1340	ask_for_approval: approval_mode,
1341	sandbox_mode: None,
1342	})?;
1343	let precheck = policy_precheck_payload(&decision, &command, &policy_cwd, execution_kind);
1344	let response_id = format!("tool-{}", Uuid::new_v4());
1345	let call_id = call
1346	.raw_tool_call_id
1347	.clone()
1348	.unwrap_or_else(\|\| format!("tool-call-{}", Uuid::new_v4()));
1349	self.hooks
1350	.emit(HookEvent::ToolLifecycle {
1351	response_id: response_id.clone(),
1352	tool_name: call.name.clone(),
1353	phase: "precheck".to_string(),
1354	payload: precheck.clone(),
1355	})
1356	.await;
1357
1358	if !decision.allow {
1359	let reason = decision.reason().to_string();
1360	let approval_id = format!("approval-{}", Uuid::new_v4());
1361	let error_frame = EventFrame::Error {
1362	response_id: response_id.clone(),
1363	message: reason.clone(),
1364	};
1365	self.hooks
1366	.emit(HookEvent::ApprovalLifecycle {
1367	approval_id,
1368	phase: "denied".to_string(),
1369	reason: Some(reason.clone()),
1370	})
1371	.await;
1372	self.hooks
1373	.emit(HookEvent::GenericEventFrame {
1374	frame: Box::new(error_frame.clone()),
1375	})
1376	.await;
1377	return Ok(json!({
1378	"ok": false,
1379	"status": "denied",
1380	"execution_kind": execution_kind,
1381	"response_id": response_id,
1382	"precheck": precheck,
1383	"error": reason,
1384	"events": [event_frame_payload(&error_frame)],
1385	}));
1386	}
1387
1388	if decision.requires_approval {
1389	let approval_id = format!("approval-{}", Uuid::new_v4());
1390	let reason = decision.reason().to_string();
1391	let maybe_approval_frame = approval_request_frame(
1392	&decision.requirement,
1393	decision.matched_rule.as_deref(),
1394	call_id,
1395	approval_id.clone(),
1396	response_id.clone(),
1397	command.clone(),
1398	policy_cwd.clone(),
1399	);
1400	self.hooks
1401	.emit(HookEvent::ApprovalLifecycle {
1402	approval_id: approval_id.clone(),
1403	phase: "requested".to_string(),
1404	reason: Some(reason.clone()),
1405	})
1406	.await;
1407	let mut events = Vec::new();
1408	if let Some(frame) = maybe_approval_frame {
1409	self.hooks
1410	.emit(HookEvent::GenericEventFrame {
1411	frame: Box::new(frame.clone()),
1412	})
1413	.await;
1414	events.push(event_frame_payload(&frame));
1415	}
1416	return Ok(json!({
1417	"ok": false,
1418	"status": "approval_required",
1419	"execution_kind": execution_kind,
1420	"response_id": response_id,
1421	"approval_id": approval_id,
1422	"precheck": precheck,
1423	"error": reason,
1424	"events": events,
1425	}));
1426	}
1427
1428	// Headless `request_user_input`: mirror the approval fire-and-return
1429	// branch (issue #3102). The TUI intercepts this tool by name before
1430	// dispatch and blocks on a reply channel; the headless runtime instead
1431	// emits a typed `UserInputRequest` frame and returns a
1432	// `user_input_required` status so the client can render the question.
1433	// It does NOT block — consistent with the headless approval model,
1434	// which has no resume channel either.
1435	//
1436	// The reply goes to the runtime API
1437	// (`POST /v1/user-input/{thread_id}/{request_id}`), which owns the
1438	// pending request and can resume the turn. The app-server control
1439	// transport cannot: it executes only `thread/interrupt` mid-turn, so
1440	// an answer sent over it would queue behind the very turn that is
1441	// waiting for it. `AppRequest::SubmitUserInput` therefore refuses
1442	// explicitly instead of pretending to have delivered the answer.
1443	if call.name == REQUEST_USER_INPUT_TOOL_NAME {
1444	let request_id = format!("user-input-{}", Uuid::new_v4());
1445	let arguments = match &call.payload {
1446	ToolPayload::Function { arguments } => arguments.as_str(),
1447	// Custom/Mcp/LocalShell can't carry a user_input payload; fall
1448	// through to the generic dispatch error below.
1449	_ => "",
1450	};
1451	let maybe_frame = user_input_request_frame(
1452	call_id.clone(),
1453	response_id.clone(),
1454	request_id.clone(),
1455	arguments,
1456	);
1457	let mut events = Vec::new();
1458	if let Some(frame) = maybe_frame {
1459	self.hooks
1460	.emit(HookEvent::GenericEventFrame {
1461	frame: Box::new(frame.clone()),
1462	})
1463	.await;
1464	events.push(event_frame_payload(&frame));
1465	}
1466	return Ok(json!({
1467	"ok": false,
1468	"status": "user_input_required",
1469	"execution_kind": execution_kind,
1470	"response_id": response_id,
1471	"request_id": request_id,
1472	"precheck": precheck,
1473	"events": events,
1474	}));
1475	}
1476
1477	let start_frame = EventFrame::ToolCallStart {
1478	response_id: response_id.clone(),
1479	tool_name: call.name.clone(),
1480	arguments: tool_payload_value(&call.payload),
1481	};
1482	self.hooks
1483	.emit(HookEvent::GenericEventFrame {
1484	frame: Box::new(start_frame.clone()),
1485	})
1486	.await;
1487	self.hooks
1488	.emit(HookEvent::ToolLifecycle {
1489	response_id: response_id.clone(),
1490	tool_name: call.name.clone(),
1491	phase: "dispatching".to_string(),
1492	payload: json!({
1493	"call_id": call_id,
1494	"execution_kind": execution_kind
1495	}),
1496	})
1497	.await;
1498
1499	match time::timeout(
1500	tool_dispatch_timeout(),
1501	self.tool_registry.dispatch(call.clone(), true),
1502	)
1503	.await
1504	{
1505	Ok(Ok(tool_output)) => {
1506	let success = tool_output.success();
1507	let status = if success { "completed" } else { "failed" };
1508	let result_frame = EventFrame::ToolCallResult {
1509	response_id: response_id.clone(),
1510	tool_name: call.name.clone(),
1511	output: tool_output_value(&tool_output),
1512	};
1513	self.hooks
1514	.emit(HookEvent::GenericEventFrame {
1515	frame: Box::new(result_frame.clone()),
1516	})
1517	.await;
1518	self.hooks
1519	.emit(HookEvent::ToolLifecycle {
1520	response_id: response_id.clone(),
1521	tool_name: call.name,
1522	phase: status.to_string(),
1523	payload: json!({ "ok": success }),
1524	})
1525	.await;
1526	Ok(json!({
1527	"ok": success,
1528	"status": status,
1529	"execution_kind": execution_kind,
1530	"response_id": response_id,
1531	"precheck": precheck,
1532	"output": tool_output,
1533	"events": [
1534	event_frame_payload(&start_frame),
1535	event_frame_payload(&result_frame)
1536	]
1537	}))
1538	}
1539	Ok(Err(err)) => {
1540	let message = format!("{err:?}");
1541	let error_frame = EventFrame::Error {
1542	response_id: response_id.clone(),
1543	message: message.clone(),
1544	};
1545	self.hooks
1546	.emit(HookEvent::GenericEventFrame {
1547	frame: Box::new(error_frame.clone()),
1548	})
1549	.await;
1550	self.hooks
1551	.emit(HookEvent::ToolLifecycle {
1552	response_id: response_id.clone(),
1553	tool_name: call.name,
1554	phase: "failed".to_string(),
1555	payload: json!({ "error": message.clone() }),
1556	})
1557	.await;
1558	Ok(json!({
1559	"ok": false,
1560	"status": "failed",
1561	"execution_kind": execution_kind,
1562	"response_id": response_id,
1563	"precheck": precheck,
1564	"error": message,
1565	"events": [
1566	event_frame_payload(&start_frame),
1567	event_frame_payload(&error_frame)
1568	]
1569	}))
1570	}
1571	Err(_elapsed) => {
1572	let seconds = tool_dispatch_timeout().as_secs().max(1);
1573	let message = format!("Tool '{}' timed out after {seconds}s", call.name);
1574	let error_frame = EventFrame::Error {
1575	response_id: response_id.clone(),
1576	message: message.clone(),
1577	};
1578	self.hooks
1579	.emit(HookEvent::GenericEventFrame {
1580	frame: Box::new(error_frame.clone()),
1581	})
1582	.await;
1583	self.hooks
1584	.emit(HookEvent::ToolLifecycle {
1585	response_id: response_id.clone(),
1586	tool_name: call.name,
1587	phase: "failed".to_string(),
1588	payload: json!({ "error": message.clone(), "timeout": true }),
1589	})
1590	.await;
1591	Ok(json!({
1592	"ok": false,
1593	"status": "timeout",
1594	"execution_kind": execution_kind,
1595	"response_id": response_id,
1596	"precheck": precheck,
1597	"error": message,
1598	"events": [
1599	event_frame_payload(&start_frame),
1600	event_frame_payload(&error_frame)
1601	]
1602	}))
1603	}
1604	}
1605	}
1606
1607	/// Starts all configured MCP servers and emits startup events via hooks.
1608	pub async fn mcp_startup(&self) -> McpStartupCompleteEvent {
1609	let mut updates = Vec::new();
1610	let summary = self.mcp_manager.start_all(\|update\| {
1611	updates.push(update);
1612	});
1613	for update in updates {
1614	let status = match update.status {
1615	McpManagerStartupStatus::Starting => codewhale_protocol::McpStartupStatus::Starting,
1616	McpManagerStartupStatus::Ready => codewhale_protocol::McpStartupStatus::Ready,
1617	McpManagerStartupStatus::Failed { error } => {
1618	codewhale_protocol::McpStartupStatus::Failed { error }
1619	}
1620	McpManagerStartupStatus::Cancelled => {
1621	codewhale_protocol::McpStartupStatus::Cancelled
1622	}
1623	};
1624	self.hooks
1625	.emit(HookEvent::GenericEventFrame {
1626	frame: Box::new(EventFrame::McpStartupUpdate {
1627	update: codewhale_protocol::McpStartupUpdateEvent {
1628	server_name: update.server_name,
1629	status,
1630	},
1631	}),
1632	})
1633	.await;
1634	}
1635	self.hooks
1636	.emit(HookEvent::GenericEventFrame {
1637	frame: Box::new(EventFrame::McpStartupComplete {
1638	summary: codewhale_protocol::McpStartupCompleteEvent {
1639	ready: summary.ready.clone(),
1640	failed: summary
1641	.failed
1642	.iter()
1643	.map(\|f\| codewhale_protocol::McpStartupFailure {
1644	server_name: f.server_name.clone(),
1645	error: f.error.clone(),
1646	})
1647	.collect(),
1648	cancelled: summary.cancelled.clone(),
1649	},
1650	}),
1651	})
1652	.await;
1653	summary
1654	}
1655
1656	/// Returns the current application status including all jobs and their history.
1657	pub fn app_status(&self) -> AppResponse {
1658	let jobs = self.jobs.list();
1659	let events = jobs
1660	.iter()
1661	.flat_map(\|job\| {
1662	job.history.iter().map(\|entry\| EventFrame::ResponseDelta {
1663	response_id: job.id.clone(),
1664	delta: json!({
1665	"kind": "job_transition",
1666	"job_id": job.id.clone(),
1667	"phase": entry.phase.clone(),
1668	"status": job_status_to_str(entry.status),
1669	"progress": entry.progress,
1670	"detail": entry.detail.clone(),
1671	"retry": job_retry_to_value(&entry.retry),
1672	"at": entry.at
1673	})
1674	.to_string(),
1675	channel: ResponseChannel::Text,
1676	})
1677	})
1678	.collect::<Vec<_>>();
1679	AppResponse {
1680	ok: true,
1681	data: json!({
1682	"jobs": jobs.into_iter().map(\|job\| {
1683	json!({
1684	"id": job.id,
1685	"name": job.name,
1686	"status": job_status_to_str(job.status),
1687	"progress": job.progress,
1688	"detail": job.detail,
1689	"retry": job_retry_to_value(&job.retry),
1690	"history": job.history.iter().map(job_history_to_value).collect::<Vec<_>>()
1691	})
1692	}).collect::<Vec<_>>()
1693	}),
1694	events,
1695	}
1696	}
1697
1698	/// Returns the default model provider from the resolved configuration.
1699	pub fn provider_default(&self) -> ProviderKind {
1700	self.config.provider
1701	}
1702
1703	/// Saves a named checkpoint for a thread.
1704	pub fn save_thread_checkpoint(
1705	&self,
1706	thread_id: &str,
1707	checkpoint_id: &str,
1708	state: &Value,
1709	) -> Result<()> {
1710	self.thread_manager
1711	.state_store()
1712	.save_checkpoint(thread_id, checkpoint_id, state)
1713	}
1714
1715	/// Loads a checkpoint for a thread. Pass `None` for the latest.
1716	pub fn load_thread_checkpoint(
1717	&self,
1718	thread_id: &str,
1719	checkpoint_id: Option<&str>,
1720	) -> Result<Option<Value>> {
1721	Ok(self
1722	.thread_manager
1723	.state_store()
1724	.load_checkpoint(thread_id, checkpoint_id)?
1725	.map(\|checkpoint\| checkpoint.state))
1726	}
1727	}
1728
1729	fn thread_response_from_new(status: &str, new: NewThread) -> ThreadResponse {
1730	ThreadResponse {
1731	thread_id: new.thread.id.clone(),
1732	status: status.to_string(),
1733	thread: Some(new.thread),
1734	threads: Vec::new(),
1735	goal: None,
1736	model: Some(new.model),
1737	model_provider: Some(new.model_provider),
1738	cwd: Some(new.cwd),
1739	approval_policy: new.approval_policy,
1740	sandbox: new.sandbox,
1741	events: Vec::new(),
1742	data: json!({}),
1743	}
1744	}
1745
1746	fn preview_from_initial_history(initial_history: &InitialHistory) -> String {
1747	match initial_history {
1748	InitialHistory::New => "New conversation".to_string(),
1749	InitialHistory::Forked(items) => truncate_preview(
1750	&items
1751	.first()
1752	.map(Value::to_string)
1753	.unwrap_or_else(\|\| "Forked conversation".to_string()),
1754	),
1755	InitialHistory::Resumed { history, .. } => truncate_preview(
1756	&history
1757	.first()
1758	.map(Value::to_string)
1759	.unwrap_or_else(\|\| "Resumed conversation".to_string()),
1760	),
1761	}
1762	}
1763
1764	fn permission_path_for_call(call: &ToolCall) -> Option<String> {
1765	match &call.payload {
1766	ToolPayload::Function { arguments } => serde_json::from_str::<Value>(arguments)
1767	.ok()
1768	.and_then(\|value\| {
1769	value
1770	.get("path")
1771	.and_then(Value::as_str)
1772	.map(str::to_string)
1773	}),
1774	ToolPayload::Mcp { raw_arguments, .. } => raw_arguments
1775	.get("path")
1776	.and_then(Value::as_str)
1777	.map(str::to_string),
1778	ToolPayload::Custom { .. } \| ToolPayload::LocalShell { .. } => None,
1779	}
1780	}
1781
1782	fn truncate_preview(value: &str) -> String {
1783	value.chars().take(120).collect()
1784	}
1785
1786	fn to_protocol_thread(thread: ThreadMetadata) -> Thread {
1787	Thread {
1788	id: thread.id,
1789	preview: thread.preview,
1790	ephemeral: thread.ephemeral,
1791	model_provider: thread.model_provider,
1792	created_at: thread.created_at,
1793	updated_at: thread.updated_at,
1794	status: match thread.status {
1795	PersistedThreadStatus::Running => ThreadStatus::Running,
1796	PersistedThreadStatus::Idle => ThreadStatus::Idle,
1797	PersistedThreadStatus::Completed => ThreadStatus::Completed,
1798	PersistedThreadStatus::Failed => ThreadStatus::Failed,
1799	PersistedThreadStatus::Paused => ThreadStatus::Paused,
1800	PersistedThreadStatus::Archived => ThreadStatus::Archived,
1801	},
1802	path: thread.path,
1803	cwd: thread.cwd,
1804	cli_version: thread.cli_version,
1805	source: match thread.source {
1806	SessionSource::Interactive => codewhale_protocol::SessionSource::Interactive,
1807	SessionSource::Resume => codewhale_protocol::SessionSource::Resume,
1808	SessionSource::Fork => codewhale_protocol::SessionSource::Fork,
1809	SessionSource::Api => codewhale_protocol::SessionSource::Api,
1810	SessionSource::Unknown => codewhale_protocol::SessionSource::Unknown,
1811	},
1812	name: thread.name,
1813	}
1814	}
1815
1816	fn to_protocol_goal(goal: ThreadGoalRecord) -> ThreadGoal {
1817	ThreadGoal {
1818	thread_id: goal.thread_id,
1819	goal_id: goal.goal_id,
1820	objective: goal.objective,
1821	status: to_protocol_goal_status(goal.status),
1822	token_budget: goal.token_budget,
1823	tokens_used: goal.tokens_used,
1824	time_used_seconds: goal.time_used_seconds,
1825	continuation_count: goal.continuation_count,
1826	last_gap_fingerprint: goal.last_gap_fingerprint,
1827	repeated_gap_count: goal.repeated_gap_count,
1828	last_gap_pass: goal.last_gap_pass,
1829	pause_reason: goal.pause_reason,
1830	created_at: goal.created_at,
1831	updated_at: goal.updated_at,
1832	}
1833	}
1834
1835	fn to_protocol_goal_status(status: PersistedThreadGoalStatus) -> ThreadGoalStatus {
1836	match status {
1837	PersistedThreadGoalStatus::Active => ThreadGoalStatus::Active,
1838	PersistedThreadGoalStatus::Paused => ThreadGoalStatus::Paused,
1839	PersistedThreadGoalStatus::Blocked => ThreadGoalStatus::Blocked,
1840	PersistedThreadGoalStatus::UsageLimited => ThreadGoalStatus::UsageLimited,
1841	PersistedThreadGoalStatus::BudgetLimited => ThreadGoalStatus::BudgetLimited,
1842	PersistedThreadGoalStatus::Complete => ThreadGoalStatus::Complete,
1843	}
1844	}
1845
1846	fn to_persisted_status(status: &ThreadStatus) -> PersistedThreadStatus {
1847	match status {
1848	ThreadStatus::Running => PersistedThreadStatus::Running,
1849	ThreadStatus::Idle => PersistedThreadStatus::Idle,
1850	ThreadStatus::Completed => PersistedThreadStatus::Completed,
1851	ThreadStatus::Failed => PersistedThreadStatus::Failed,
1852	ThreadStatus::Paused => PersistedThreadStatus::Paused,
1853	ThreadStatus::Archived => PersistedThreadStatus::Archived,
1854	}
1855	}
1856
1857	fn to_persisted_source(source: &codewhale_protocol::SessionSource) -> SessionSource {
1858	match source {
1859	codewhale_protocol::SessionSource::Interactive => SessionSource::Interactive,
1860	codewhale_protocol::SessionSource::Resume => SessionSource::Resume,
1861	codewhale_protocol::SessionSource::Fork => SessionSource::Fork,
1862	codewhale_protocol::SessionSource::Api => SessionSource::Api,
1863	codewhale_protocol::SessionSource::Unknown => SessionSource::Unknown,
1864	}
1865	}
1866
1867	fn approval_request_frame(
1868	requirement: &ExecApprovalRequirement,
1869	matched_rule: Option<&str>,
1870	call_id: String,
1871	approval_id: String,
1872	turn_id: String,
1873	command: String,
1874	cwd: String,
1875	) -> Option<EventFrame> {
1876	let ExecApprovalRequirement::NeedsApproval {
1877	reason,
1878	proposed_execpolicy_amendment,
1879	proposed_network_policy_amendments,
1880	} = requirement
1881	else {
1882	return None;
1883	};
1884
1885	let mut available_decisions = vec![
1886	ReviewDecision::Approved,
1887	ReviewDecision::ApprovedForSession,
1888	ReviewDecision::Denied,
1889	ReviewDecision::Abort,
1890	];
1891	if proposed_execpolicy_amendment
1892	.as_ref()
1893	.is_some_and(\|amendment\| !amendment.prefixes.is_empty())
1894	{
1895	available_decisions.push(ReviewDecision::ApprovedExecpolicyAmendment);
1896	}
1897	available_decisions.extend(proposed_network_policy_amendments.iter().cloned().map(
1898	\|amendment\| ReviewDecision::NetworkPolicyAmendment {
1899	host: amendment.host,
1900	action: amendment.action,
1901	},
1902	));
1903
1904	Some(EventFrame::ExecApprovalRequest {
1905	request: ExecApprovalRequestEvent {
1906	call_id,
1907	approval_id,
1908	turn_id,
1909	command,
1910	cwd,
1911	reason: reason.clone(),
1912	matched_rule: matched_rule.map(\|rule\| rule.to_string().into_boxed_str()),
1913	network_approval_context: None,
1914	proposed_execpolicy_amendment: proposed_execpolicy_amendment
1915	.as_ref()
1916	.map(\|amendment\| amendment.prefixes.clone())
1917	.unwrap_or_default(),
1918	proposed_network_policy_amendments: proposed_network_policy_amendments.clone(),
1919	additional_permissions: Vec::new(),
1920	available_decisions,
1921	},
1922	})
1923	}
1924
1925	/// Build an [`EventFrame::UserInputRequest`] for a headless
1926	/// `request_user_input` tool call, mirroring [`approval_request_frame`].
1927	///
1928	/// `arguments` is the raw JSON arguments string the model supplied to the
1929	/// `request_user_input` tool (a `ToolPayload::Function` body). On parse
1930	/// failure we return `None` so the caller falls through to the generic tool
1931	/// error path rather than silently dropping the request.
1932	fn user_input_request_frame(
1933	call_id: String,
1934	turn_id: String,
1935	request_id: String,
1936	arguments: &str,
1937	) -> Option<EventFrame> {
1938	let parsed: Value = serde_json::from_str(arguments).ok()?;
1939	// Extract the `questions` array and lift it into the headless event
1940	// shape. We tolerate missing `allow_free_text`/`multi_select` (default
1941	// false) and extra fields, matching the lenient TUI `from_value` path.
1942	let questions = parsed.get("questions").cloned().filter(Value::is_array)?;
1943	let request = UserInputRequestEvent {
1944	call_id,
1945	turn_id,
1946	request_id,
1947	questions: serde_json::from_value(questions).ok()?,
1948	};
1949	Some(EventFrame::UserInputRequest { request })
1950	}
1951
1952	fn approval_requirement_payload(requirement: &ExecApprovalRequirement) -> Value {
1953	match requirement {
1954	ExecApprovalRequirement::Skip {
1955	bypass_sandbox,
1956	proposed_execpolicy_amendment,
1957	} => json!({
1958	"type": "skip",
1959	"bypass_sandbox": bypass_sandbox,
1960	"reason": requirement.reason(),
1961	"proposed_execpolicy_amendment": proposed_execpolicy_amendment
1962	.as_ref()
1963	.map(\|amendment\| amendment.prefixes.clone())
1964	.unwrap_or_default()
1965	}),
1966	ExecApprovalRequirement::NeedsApproval {
1967	reason,
1968	proposed_execpolicy_amendment,
1969	proposed_network_policy_amendments,
1970	} => json!({
1971	"type": "needs_approval",
1972	"reason": reason,
1973	"proposed_execpolicy_amendment": proposed_execpolicy_amendment
1974	.as_ref()
1975	.map(\|amendment\| amendment.prefixes.clone())
1976	.unwrap_or_default(),
1977	"proposed_network_policy_amendments": proposed_network_policy_amendments
1978	}),
1979	ExecApprovalRequirement::Forbidden { reason } => json!({
1980	"type": "forbidden",
1981	"reason": reason
1982	}),
1983	}
1984	}
1985
1986	fn policy_precheck_payload(
1987	decision: &ExecPolicyDecision,
1988	command: &str,
1989	cwd: &str,
1990	execution_kind: &str,
1991	) -> Value {
1992	json!({
1993	"execution_kind": execution_kind,
1994	"command": command,
1995	"cwd": cwd,
1996	"allow": decision.allow,
1997	"requires_approval": decision.requires_approval,
1998	"matched_rule": decision.matched_rule.clone(),
1999	"phase": decision.requirement.phase(),
2000	"reason": decision.reason(),
2001	"requirement": approval_requirement_payload(&decision.requirement)
2002	})
2003	}
2004
2005	fn tool_payload_value(payload: &ToolPayload) -> Value {
2006	serde_json::to_value(payload).unwrap_or_else(
2007	\|_\| json!({"type":"serialization_error","message":"tool payload unavailable"}),
2008	)
2009	}
2010
2011	fn tool_output_value(output: &codewhale_protocol::ToolOutput) -> Value {
2012	serde_json::to_value(output).unwrap_or_else(
2013	\|_\| json!({"type":"serialization_error","message":"tool output unavailable"}),
2014	)
2015	}
2016
2017	fn event_frame_payload(frame: &EventFrame) -> Value {
2018	serde_json::to_value(frame)
2019	.unwrap_or_else(\|_\| json!({"event":"error","message":"failed to encode event frame"}))
2020	}
2021
2022	/// Tool name that triggers the headless clarification-question flow.
2023	///
2024	/// Mirrors the TUI's `REQUEST_USER_INPUT_NAME`
2025	/// (`crates/tui/src/core/engine/tool_catalog.rs`); duplicated here rather than
2026	/// depended on across crates so `core` stays free of `tui` imports.
2027	const REQUEST_USER_INPUT_TOOL_NAME: &str = "request_user_input";
2028
2029	fn json_optional_string(value: &Value) -> Option<String> {
2030	if value.is_null() {
2031	None
2032	} else {
2033	value.as_str().map(ToString::to_string)
2034	}
2035	}
2036
2037	fn parse_retry_metadata(value: Option<&Value>) -> JobRetryMetadata {
2038	let Some(value) = value else {
2039	return JobRetryMetadata::default();
2040	};
2041	JobRetryMetadata {
2042	attempt: value
2043	.get("attempt")
2044	.and_then(Value::as_u64)
2045	.unwrap_or(0)
2046	.min(u32::MAX as u64) as u32,
2047	max_attempts: value
2048	.get("max_attempts")
2049	.and_then(Value::as_u64)
2050	.unwrap_or(DEFAULT_JOB_MAX_ATTEMPTS as u64)
2051	.min(u32::MAX as u64) as u32,
2052	backoff_base_ms: value
2053	.get("backoff_base_ms")
2054	.and_then(Value::as_u64)
2055	.unwrap_or(DEFAULT_JOB_BACKOFF_BASE_MS),
2056	next_backoff_ms: value
2057	.get("next_backoff_ms")
2058	.and_then(Value::as_u64)
2059	.unwrap_or(0),
2060	next_retry_at: value.get("next_retry_at").and_then(Value::as_i64),
2061	}
2062	}
2063
2064	fn parse_history_entry(value: &Value) -> Option<JobHistoryEntry> {
2065	let status = value
2066	.get("status")
2067	.and_then(Value::as_str)
2068	.and_then(job_status_from_str)?;
2069	Some(JobHistoryEntry {
2070	at: value.get("at").and_then(Value::as_i64).unwrap_or(0),
2071	phase: value
2072	.get("phase")
2073	.and_then(Value::as_str)
2074	.unwrap_or("unknown")
2075	.to_string(),
2076	status,
2077	progress: value
2078	.get("progress")
2079	.and_then(Value::as_u64)
2080	.map(\|v\| v.min(u8::MAX as u64) as u8),
2081	detail: value.get("detail").and_then(json_optional_string),
2082	retry: parse_retry_metadata(value.get("retry")),
2083	})
2084	}
2085
2086	fn job_status_to_str(status: JobStatus) -> &'static str {
2087	match status {
2088	JobStatus::Queued => "queued",
2089	JobStatus::Running => "running",
2090	JobStatus::Paused => "paused",
2091	JobStatus::Completed => "completed",
2092	JobStatus::Failed => "failed",
2093	JobStatus::Cancelled => "cancelled",
2094	}
2095	}
2096
2097	fn job_status_from_str(value: &str) -> Option<JobStatus> {
2098	match value {
2099	"queued" => Some(JobStatus::Queued),
2100	"running" => Some(JobStatus::Running),
2101	"paused" => Some(JobStatus::Paused),
2102	"completed" => Some(JobStatus::Completed),
2103	"failed" => Some(JobStatus::Failed),
2104	"cancelled" => Some(JobStatus::Cancelled),
2105	_ => None,
2106	}
2107	}
2108
2109	fn job_retry_to_value(retry: &JobRetryMetadata) -> Value {
2110	json!({
2111	"attempt": retry.attempt,
2112	"max_attempts": retry.max_attempts,
2113	"backoff_base_ms": retry.backoff_base_ms,
2114	"next_backoff_ms": retry.next_backoff_ms,
2115	"next_retry_at": retry.next_retry_at
2116	})
2117	}
2118
2119	fn job_history_to_value(entry: &JobHistoryEntry) -> Value {
2120	json!({
2121	"at": entry.at,
2122	"phase": entry.phase.clone(),
2123	"status": job_status_to_str(entry.status),
2124	"progress": entry.progress,
2125	"detail": entry.detail.clone(),
2126	"retry": job_retry_to_value(&entry.retry)
2127	})
2128	}
2129
2130	fn runtime_status_to_job_state(status: JobStatus) -> JobStateStatus {
2131	match status {
2132	JobStatus::Queued => JobStateStatus::Queued,
2133	JobStatus::Running => JobStateStatus::Running,
2134	JobStatus::Paused => JobStateStatus::Paused,
2135	JobStatus::Completed => JobStateStatus::Completed,
2136	JobStatus::Failed => JobStateStatus::Failed,
2137	JobStatus::Cancelled => JobStateStatus::Cancelled,
2138	}
2139	}
2140
2141	fn job_state_status_to_runtime(status: JobStateStatus) -> JobStatus {
2142	match status {
2143	JobStateStatus::Queued => JobStatus::Queued,
2144	JobStateStatus::Running => JobStatus::Running,
2145	JobStateStatus::Paused => JobStatus::Paused,
2146	JobStateStatus::Completed => JobStatus::Completed,
2147	JobStateStatus::Failed => JobStatus::Failed,
2148	JobStateStatus::Cancelled => JobStatus::Cancelled,
2149	}
2150	}
2151
2152	#[cfg(test)]
2153	mod tests {
2154	use super::*;
2155	use codewhale_protocol::ThreadResumeParams;
2156	use codewhale_tools::ToolCallSource;
2157
2158	fn temp_core_state(name: &str) -> StateStore {
2159	let dir =
2160	std::env::temp_dir().join(format!("codewhale-core-{name}-{}", Uuid::new_v4().simple()));
2161	std::fs::create_dir_all(&dir).expect("create temp state dir");
2162	StateStore::open(Some(dir.join("state.db"))).expect("open state store")
2163	}
2164
2165	fn test_thread_metadata(id: &str) -> ThreadMetadata {
2166	ThreadMetadata {
2167	id: id.to_string(),
2168	rollout_path: None,
2169	preview: "test thread".to_string(),
2170	ephemeral: false,
2171	model_provider: "deepseek".to_string(),
2172	created_at: 10,
2173	updated_at: 10,
2174	status: PersistedThreadStatus::Running,
2175	path: None,
2176	cwd: PathBuf::from("/tmp/codewhale"),
2177	cli_version: "0.0.0-test".to_string(),
2178	source: SessionSource::Interactive,
2179	name: None,
2180	sandbox_policy: None,
2181	approval_mode: None,
2182	archived: false,
2183	archived_at: None,
2184	git_sha: None,
2185	git_branch: None,
2186	git_origin_url: None,
2187	memory_mode: None,
2188	current_leaf_id: None,
2189	}
2190	}
2191
2192	// ── JobManager: lifecycle ──────────────────────────────────────────
2193
2194	#[test]
2195	fn permission_path_for_call_extracts_function_path_argument() {
2196	let call = ToolCall {
2197	name: "read_file".to_string(),
2198	payload: ToolPayload::Function {
2199	arguments: json!({ "path": "README.md" }).to_string(),
2200	},
2201	source: ToolCallSource::Direct,
2202	raw_tool_call_id: None,
2203	};
2204
2205	assert_eq!(
2206	permission_path_for_call(&call).as_deref(),
2207	Some("README.md")
2208	);
2209	}
2210
2211	#[test]
2212	fn permission_path_for_call_extracts_mcp_path_argument() {
2213	let call = ToolCall {
2214	name: "mcp_fs_read".to_string(),
2215	payload: ToolPayload::Mcp {
2216	server: "fs".to_string(),
2217	tool: "read".to_string(),
2218	raw_arguments: json!({ "path": "secrets/token.txt" }),
2219	raw_tool_call_id: None,
2220	},
2221	source: ToolCallSource::Direct,
2222	raw_tool_call_id: None,
2223	};
2224
2225	assert_eq!(
2226	permission_path_for_call(&call).as_deref(),
2227	Some("secrets/token.txt")
2228	);
2229	}
2230
2231	#[test]
2232	fn permission_path_for_call_ignores_shell_payload() {
2233	let call = ToolCall {
2234	name: "exec_shell".to_string(),
2235	payload: ToolPayload::LocalShell {
2236	params: codewhale_protocol::LocalShellParams {
2237	command: "cargo test".to_string(),
2238	cwd: None,
2239	timeout_ms: None,
2240	},
2241	},
2242	source: ToolCallSource::Direct,
2243	raw_tool_call_id: None,
2244	};
2245
2246	assert_eq!(permission_path_for_call(&call), None);
2247	}
2248
2249	#[test]
2250	fn thread_goal_progress_accumulates_durable_accounting() {
2251	let store = temp_core_state("thread-goal-progress");
2252	store
2253	.upsert_thread(&test_thread_metadata("thread-1"))
2254	.expect("upsert thread");
2255	let mut manager = ThreadManager::new(store);
2256	manager
2257	.set_thread_goal(&ThreadGoalSetParams {
2258	thread_id: "thread-1".to_string(),
2259	objective: "Carry the goal across turns".to_string(),
2260	token_budget: Some(2_000),
2261	})
2262	.expect("set goal")
2263	.expect("goal exists");
2264
2265	let updated = manager
2266	.record_thread_goal_progress(&ThreadGoalProgressParams {
2267	thread_id: "thread-1".to_string(),
2268	token_delta: 750,
2269	time_delta_seconds: 12,
2270	record_continuation: true,
2271	})
2272	.expect("record progress")
2273	.expect("goal exists");
2274
2275	assert_eq!(updated.tokens_used, 750);
2276	assert_eq!(updated.time_used_seconds, 12);
2277	assert_eq!(updated.continuation_count, 1);
2278
2279	let persisted = manager
2280	.get_thread_goal(&ThreadGoalGetParams {
2281	thread_id: "thread-1".to_string(),
2282	})
2283	.expect("read goal")
2284	.expect("goal exists");
2285	assert_eq!(persisted.tokens_used, 750);
2286	assert_eq!(persisted.time_used_seconds, 12);
2287	assert_eq!(persisted.continuation_count, 1);
2288	}
2289
2290	#[test]
2291	fn approval_request_frame_includes_matched_rule() {
2292	let requirement = ExecApprovalRequirement::NeedsApproval {
2293	reason: "Typed ask rule 'tool=exec_shell command=cargo test' requires approval."
2294	.to_string(),
2295	proposed_execpolicy_amendment: None,
2296	proposed_network_policy_amendments: Vec::new(),
2297	};
2298
2299	let frame = approval_request_frame(
2300	&requirement,
2301	Some("tool=exec_shell command=cargo test"),
2302	"call-1".to_string(),
2303	"approval-1".to_string(),
2304	"turn-1".to_string(),
2305	"cargo test --workspace".to_string(),
2306	"/repo".to_string(),
2307	)
2308	.expect("approval frame");
2309
2310	let EventFrame::ExecApprovalRequest { request } = frame else {
2311	panic!("expected exec approval request frame");
2312	};
2313	assert_eq!(
2314	request.matched_rule.as_deref(),
2315	Some("tool=exec_shell command=cargo test")
2316	);
2317	assert_eq!(request.reason, requirement.reason());
2318	}
2319
2320	#[test]
2321	fn user_input_request_frame_lifts_questions_from_arguments() {
2322	// issue #3102: the headless frame constructor must parse the model's
2323	// `request_user_input` arguments and lift the questions into the
2324	// UserInputRequestEvent, defaulting the boolean flags when omitted.
2325	let arguments = r#"{"questions":[{"header":"Scope","id":"scope","question":"Which?","options":[{"label":"A","description":"a"},{"label":"B","description":"b"}],"allow_free_text":true}]}"#;
2326	let frame = user_input_request_frame(
2327	"call-1".to_string(),
2328	"turn-1".to_string(),
2329	"ui-1".to_string(),
2330	arguments,
2331	)
2332	.expect("user input frame");
2333
2334	let EventFrame::UserInputRequest { request } = frame else {
2335	panic!("expected user_input_request frame");
2336	};
2337	assert_eq!(request.call_id, "call-1");
2338	assert_eq!(request.turn_id, "turn-1");
2339	assert_eq!(request.request_id, "ui-1");
2340	assert_eq!(request.questions.len(), 1);
2341	assert_eq!(request.questions[0].id, "scope");
2342	assert!(request.questions[0].allow_free_text);
2343	// multi_select omitted in the payload → defaults to false.
2344	assert!(!request.questions[0].multi_select);
2345	assert_eq!(request.questions[0].options.len(), 2);
2346	}
2347
2348	#[test]
2349	fn user_input_request_frame_returns_none_on_invalid_arguments() {
2350	// On parse failure the constructor returns None so invoke_tool falls
2351	// through to the generic tool error path instead of silently dropping.
2352	let frame = user_input_request_frame(
2353	"call-1".to_string(),
2354	"turn-1".to_string(),
2355	"ui-1".to_string(),
2356	"not json",
2357	);
2358	assert!(frame.is_none());
2359
2360	// Valid JSON but missing the questions array is also rejected.
2361	let frame = user_input_request_frame(
2362	"call-1".to_string(),
2363	"turn-1".to_string(),
2364	"ui-1".to_string(),
2365	r#"{"foo":"bar"}"#,
2366	);
2367	assert!(frame.is_none());
2368	}
2369
2370	#[test]
2371	fn enqueue_creates_queued_job_with_zero_progress() {
2372	let mut jm = JobManager::default();
2373	let job = jm.enqueue("build");
2374	assert_eq!(job.name, "build");
2375	assert_eq!(job.status, JobStatus::Queued);
2376	assert_eq!(job.progress, Some(0));
2377	assert!(job.detail.is_none());
2378	assert_eq!(job.history.len(), 1);
2379	assert_eq!(job.history[0].phase, "created");
2380	}
2381
2382	#[test]
2383	fn set_running_transitions_from_queued() {
2384	let mut jm = JobManager::default();
2385	let job = jm.enqueue("deploy");
2386	let id = job.id.clone();
2387	jm.set_running(&id);
2388	let jobs = jm.list();
2389	let updated = jobs.iter().find(\|j\| j.id == id).unwrap();
2390	assert_eq!(updated.status, JobStatus::Running);
2391	assert_eq!(updated.history.last().unwrap().phase, "running");
2392	}
2393
2394	#[test]
2395	fn update_progress_clamps_to_100() {
2396	let mut jm = JobManager::default();
2397	let job = jm.enqueue("task");
2398	let id = job.id.clone();
2399	jm.update_progress(&id, 150, Some("over".to_string()));
2400	let jobs = jm.list();
2401	let updated = jobs.iter().find(\|j\| j.id == id).unwrap();
2402	assert_eq!(updated.progress, Some(100));
2403	}
2404
2405	#[test]
2406	fn complete_sets_progress_to_100() {
2407	let mut jm = JobManager::default();
2408	let job = jm.enqueue("task");
2409	let id = job.id.clone();
2410	jm.set_running(&id);
2411	jm.complete(&id);
2412	let jobs = jm.list();
2413	let updated = jobs.iter().find(\|j\| j.id == id).unwrap();
2414	assert_eq!(updated.status, JobStatus::Completed);
2415	assert_eq!(updated.progress, Some(100));
2416	}
2417
2418	#[test]
2419	fn fail_increments_attempt_and_sets_backoff() {
2420	let mut jm = JobManager::default();
2421	let job = jm.enqueue("fragile");
2422	let id = job.id.clone();
2423	jm.set_running(&id);
2424	jm.fail(&id, "crashed");
2425	let jobs = jm.list();
2426	let updated = jobs.iter().find(\|j\| j.id == id).unwrap();
2427	assert_eq!(updated.status, JobStatus::Failed);
2428	assert_eq!(updated.retry.attempt, 1);
2429	assert!(updated.retry.next_backoff_ms > 0);
2430	assert!(updated.retry.next_retry_at.is_some());
2431	assert_eq!(updated.detail.as_deref(), Some("crashed"));
2432	}
2433
2434	#[test]
2435	fn fail_clears_retry_after_max_attempts() {
2436	let mut jm = JobManager::default();
2437	let job = jm.enqueue("fragile");
2438	let id = job.id.clone();
2439	for _ in 0..=DEFAULT_JOB_MAX_ATTEMPTS {
2440	jm.set_running(&id);
2441	jm.fail(&id, "boom");
2442	}
2443	let jobs = jm.list();
2444	let updated = jobs.iter().find(\|j\| j.id == id).unwrap();
2445	assert_eq!(updated.retry.attempt, DEFAULT_JOB_MAX_ATTEMPTS);
2446	assert_eq!(updated.retry.next_backoff_ms, 0);
2447	assert!(updated.retry.next_retry_at.is_none());
2448	}
2449
2450	#[test]
2451	fn cancel_sets_status_and_clears_retry() {
2452	let mut jm = JobManager::default();
2453	let job = jm.enqueue("task");
2454	let id = job.id.clone();
2455	jm.cancel(&id);
2456	let jobs = jm.list();
2457	let updated = jobs.iter().find(\|j\| j.id == id).unwrap();
2458	assert_eq!(updated.status, JobStatus::Cancelled);
2459	assert_eq!(updated.retry.next_backoff_ms, 0);
2460	}
2461
2462	#[test]
2463	fn pause_and_resume_round_trip() {
2464	let mut jm = JobManager::default();
2465	let job = jm.enqueue("task");
2466	let id = job.id.clone();
2467	jm.set_running(&id);
2468	jm.pause(&id, Some("waiting".to_string()));
2469	let jobs = jm.list();
2470	let paused = jobs.iter().find(\|j\| j.id == id).unwrap();
2471	assert_eq!(paused.status, JobStatus::Paused);
2472	assert_eq!(paused.detail.as_deref(), Some("waiting"));
2473
2474	jm.resume(&id, None);
2475	let jobs = jm.list();
2476	let resumed = jobs.iter().find(\|j\| j.id == id).unwrap();
2477	assert_eq!(resumed.status, JobStatus::Running);
2478	assert_eq!(resumed.history.last().unwrap().phase, "resumed");
2479	}
2480
2481	#[test]
2482	fn list_returns_jobs_sorted_by_updated_at_desc() {
2483	let mut jm = JobManager::default();
2484	jm.enqueue("first");
2485	jm.enqueue("second");
2486	jm.enqueue("third");
2487	let jobs = jm.list();
2488	assert_eq!(jobs.len(), 3);
2489	for window in jobs.windows(2) {
2490	assert!(window[0].updated_at >= window[1].updated_at);
2491	}
2492	}
2493
2494	#[test]
2495	fn history_returns_entries_for_existing_job() {
2496	let mut jm = JobManager::default();
2497	let job = jm.enqueue("task");
2498	let id = job.id.clone();
2499	jm.set_running(&id);
2500	jm.complete(&id);
2501	let history = jm.history(&id);
2502	assert_eq!(history.len(), 3); // created, running, completed
2503	assert_eq!(history[0].phase, "created");
2504	assert_eq!(history[1].phase, "running");
2505	assert_eq!(history[2].phase, "completed");
2506	}
2507
2508	#[test]
2509	fn history_returns_empty_for_unknown_job() {
2510	let jm = JobManager::default();
2511	assert!(jm.history("nonexistent").is_empty());
2512	}
2513
2514	#[test]
2515	fn resume_pending_requeues_running_and_queued() {
2516	let mut jm = JobManager::default();
2517	let _j1 = jm.enqueue("queued_task");
2518	let j2 = jm.enqueue("running_task");
2519	let j3 = jm.enqueue("completed_task");
2520	let id2 = j2.id.clone();
2521	let id3 = j3.id.clone();
2522	jm.set_running(&id2);
2523	jm.set_running(&id3);
2524	jm.complete(&id3);
2525
2526	let resumed = jm.resume_pending();
2527	assert_eq!(resumed.len(), 2);
2528	for job in &resumed {
2529	assert_eq!(job.status, JobStatus::Queued);
2530	}
2531	}
2532
2533	// ── JobManager: backoff ────────────────────────────────────────────
2534
2535	#[test]
2536	fn deterministic_backoff_zero_on_first_attempt() {
2537	let retry = JobRetryMetadata {
2538	attempt: 0,
2539	..Default::default()
2540	};
2541	assert_eq!(JobManager::deterministic_backoff_ms(&retry), 0);
2542	}
2543
2544	#[test]
2545	fn deterministic_backoff_exponential_growth() {
2546	let base = DEFAULT_JOB_BACKOFF_BASE_MS;
2547	for attempt in 1..=5 {
2548	let retry = JobRetryMetadata {
2549	attempt,
2550	backoff_base_ms: base,
2551	..Default::default()
2552	};
2553	let expected = base * 2u64.pow(attempt.saturating_sub(1).min(20));
2554	assert_eq!(
2555	JobManager::deterministic_backoff_ms(&retry),
2556	expected,
2557	"attempt {attempt}"
2558	);
2559	}
2560	}
2561
2562	#[test]
2563	fn deterministic_backoff_saturates_at_high_exponent() {
2564	let retry = JobRetryMetadata {
2565	attempt: 63,
2566	backoff_base_ms: 1000,
2567	..Default::default()
2568	};
2569	// Should not panic; result saturates
2570	let _ = JobManager::deterministic_backoff_ms(&retry);
2571	}
2572
2573	// ── JobManager: history truncation ─────────────────────────────────
2574
2575	#[test]
2576	fn push_history_truncates_beyond_max() {
2577	let mut jm = JobManager::default();
2578	let job = jm.enqueue("task");
2579	let id = job.id.clone();
2580	// Generate more history entries than the limit
2581	for i in 0..(MAX_JOB_HISTORY_ENTRIES + 20) {
2582	jm.update_progress(&id, (i % 100) as u8, Some(format!("step {i}")));
2583	}
2584	let history = jm.history(&id);
2585	assert_eq!(history.len(), MAX_JOB_HISTORY_ENTRIES);
2586	}
2587
2588	// ── JobManager: persistence encoding/parsing ───────────────────────
2589
2590	#[test]
2591	fn encode_and_parse_persisted_detail_round_trip() {
2592	let mut jm = JobManager::default();
2593	let job = jm.enqueue("task");
2594	let id = job.id.clone();
2595	jm.set_running(&id);
2596	jm.fail(&id, "oops");
2597	let job = jm.list().into_iter().find(\|j\| j.id == id).unwrap();
2598
2599	let encoded = JobManager::encode_persisted_detail(&job).unwrap().unwrap();
2600	let parsed = JobManager::parse_persisted_detail(Some(&encoded)).unwrap();
2601
2602	assert_eq!(parsed.status, job.status);
2603	assert_eq!(parsed.detail, job.detail);
2604	assert_eq!(parsed.retry.attempt, job.retry.attempt);
2605	assert_eq!(parsed.history.len(), job.history.len());
2606	}
2607
2608	#[test]
2609	fn parse_persisted_detail_returns_none_for_none_input() {
2610	assert!(JobManager::parse_persisted_detail(None).is_none());
2611	}
2612
2613	#[test]
2614	fn parse_persisted_detail_returns_none_for_invalid_json() {
2615	assert!(JobManager::parse_persisted_detail(Some("not json")).is_none());
2616	}
2617
2618	// ── Helper functions ───────────────────────────────────────────────
2619
2620	#[test]
2621	fn job_status_round_trip_str() {
2622	let statuses = [
2623	JobStatus::Queued,
2624	JobStatus::Running,
2625	JobStatus::Paused,
2626	JobStatus::Completed,
2627	JobStatus::Failed,
2628	JobStatus::Cancelled,
2629	];
2630	for status in &statuses {
2631	let s = job_status_to_str(*status);
2632	let parsed = job_status_from_str(s);
2633	assert_eq!(parsed, Some(*status), "round-trip failed for {s:?}");
2634	}
2635	}
2636
2637	#[test]
2638	fn job_status_from_str_returns_none_for_unknown() {
2639	assert_eq!(job_status_from_str("unknown"), None);
2640	assert_eq!(job_status_from_str(""), None);
2641	}
2642
2643	#[test]
2644	fn truncate_preview_limits_to_120_chars() {
2645	let long = "a".repeat(200);
2646	let truncated = truncate_preview(&long);
2647	assert_eq!(truncated.len(), 120);
2648	}
2649
2650	#[test]
2651	fn truncate_preview_preserves_short_strings() {
2652	let short = "hello";
2653	assert_eq!(truncate_preview(short), "hello");
2654	}
2655
2656	#[test]
2657	fn runtime_status_to_job_state_maps_correctly() {
2658	assert_eq!(
2659	runtime_status_to_job_state(JobStatus::Queued),
2660	JobStateStatus::Queued
2661	);
2662	assert_eq!(
2663	runtime_status_to_job_state(JobStatus::Running),
2664	JobStateStatus::Running
2665	);
2666	assert_eq!(
2667	runtime_status_to_job_state(JobStatus::Paused),
2668	JobStateStatus::Paused
2669	);
2670	assert_eq!(
2671	runtime_status_to_job_state(JobStatus::Completed),
2672	JobStateStatus::Completed
2673	);
2674	assert_eq!(
2675	runtime_status_to_job_state(JobStatus::Failed),
2676	JobStateStatus::Failed
2677	);
2678	assert_eq!(
2679	runtime_status_to_job_state(JobStatus::Cancelled),
2680	JobStateStatus::Cancelled
2681	);
2682	}
2683
2684	#[test]
2685	fn job_state_status_to_runtime_maps_correctly() {
2686	assert_eq!(
2687	job_state_status_to_runtime(JobStateStatus::Queued),
2688	JobStatus::Queued
2689	);
2690	assert_eq!(
2691	job_state_status_to_runtime(JobStateStatus::Running),
2692	JobStatus::Running
2693	);
2694	assert_eq!(
2695	job_state_status_to_runtime(JobStateStatus::Paused),
2696	JobStatus::Paused
2697	);
2698	assert_eq!(
2699	job_state_status_to_runtime(JobStateStatus::Completed),
2700	JobStatus::Completed
2701	);
2702	assert_eq!(
2703	job_state_status_to_runtime(JobStateStatus::Failed),
2704	JobStatus::Failed
2705	);
2706	assert_eq!(
2707	job_state_status_to_runtime(JobStateStatus::Cancelled),
2708	JobStatus::Cancelled
2709	);
2710	}
2711
2712	#[test]
2713	fn preview_from_initial_history_new() {
2714	let preview = preview_from_initial_history(&InitialHistory::New);
2715	assert_eq!(preview, "New conversation");
2716	}
2717
2718	#[test]
2719	fn preview_from_initial_history_forked() {
2720	let preview = preview_from_initial_history(&InitialHistory::Forked(vec![json!("hello")]));
2721	assert!(preview.contains("hello"));
2722	}
2723
2724	#[test]
2725	fn preview_from_initial_history_resumed() {
2726	let preview = preview_from_initial_history(&InitialHistory::Resumed {
2727	conversation_id: "test".to_string(),
2728	history: vec![json!("world")],
2729	rollout_path: PathBuf::from("/tmp/test"),
2730	});
2731	assert!(preview.contains("world"));
2732	}
2733
2734	#[test]
2735	fn json_optional_string_handles_null() {
2736	assert!(json_optional_string(&Value::Null).is_none());
2737	}
2738
2739	#[test]
2740	fn json_optional_string_handles_string() {
2741	assert_eq!(
2742	json_optional_string(&Value::String("hello".to_string())),
2743	Some("hello".to_string())
2744	);
2745	}
2746
2747	#[test]
2748	fn json_optional_string_handles_non_string() {
2749	assert!(json_optional_string(&json!(42)).is_none());
2750	}
2751
2752	#[test]
2753	fn parse_retry_metadata_returns_default_for_none() {
2754	let retry = parse_retry_metadata(None);
2755	assert_eq!(retry.attempt, 0);
2756	assert_eq!(retry.max_attempts, DEFAULT_JOB_MAX_ATTEMPTS);
2757	assert_eq!(retry.backoff_base_ms, DEFAULT_JOB_BACKOFF_BASE_MS);
2758	}
2759
2760	#[test]
2761	fn parse_retry_metadata_parses_fields() {
2762	let value = json!({
2763	"attempt": 2,
2764	"max_attempts": 5,
2765	"backoff_base_ms": 1000,
2766	"next_backoff_ms": 2000,
2767	"next_retry_at": 1234567890i64
2768	});
2769	let retry = parse_retry_metadata(Some(&value));
2770	assert_eq!(retry.attempt, 2);
2771	assert_eq!(retry.max_attempts, 5);
2772	assert_eq!(retry.backoff_base_ms, 1000);
2773	assert_eq!(retry.next_backoff_ms, 2000);
2774	assert_eq!(retry.next_retry_at, Some(1234567890));
2775	}
2776
2777	#[test]
2778	fn parse_history_entry_returns_none_without_status() {
2779	let value = json!({"at": 1, "phase": "test"});
2780	assert!(parse_history_entry(&value).is_none());
2781	}
2782
2783	#[test]
2784	fn parse_history_entry_parses_valid_entry() {
2785	let value = json!({
2786	"at": 100,
2787	"phase": "running",
2788	"status": "running",
2789	"progress": 50,
2790	"detail": "working",
2791	"retry": {"attempt": 0, "max_attempts": 3, "backoff_base_ms": 500}
2792	});
2793	let entry = parse_history_entry(&value).unwrap();
2794	assert_eq!(entry.at, 100);
2795	assert_eq!(entry.phase, "running");
2796	assert_eq!(entry.status, JobStatus::Running);
2797	assert_eq!(entry.progress, Some(50));
2798	assert_eq!(entry.detail.as_deref(), Some("working"));
2799	}
2800
2801	#[test]
2802	fn paused_job_persists_as_paused_not_running() {
2803	let store = temp_core_state("paused-persist");
2804	let mut jm = JobManager::default();
2805	let job = jm.enqueue("task");
2806	let id = job.id.clone();
2807	jm.set_running(&id);
2808	jm.pause(&id, Some("waiting".to_string()));
2809	jm.persist_job(&store, &id).expect("persist paused job");
2810
2811	let persisted = store.list_jobs(Some(10)).expect("list jobs");
2812	let record = persisted.iter().find(\|job\| job.id == id).unwrap();
2813	assert_eq!(record.status, JobStateStatus::Paused);
2814
2815	let mut reloaded = JobManager::default();
2816	reloaded.load_from_store(&store).expect("reload jobs");
2817	let jobs = reloaded.list();
2818	let reloaded_job = jobs.iter().find(\|job\| job.id == id).unwrap();
2819	assert_eq!(reloaded_job.status, JobStatus::Paused);
2820	}
2821
2822	// ── O1: JobRecord → AgentRunSnapshot adapter ────────────────────────
2823
2824	fn sample_job_record(status: JobStatus, detail: Option<&str>) -> JobRecord {
2825	JobRecord {
2826	id: "job-o1-1".to_string(),
2827	name: "sample".to_string(),
2828	status,
2829	progress: None,
2830	detail: detail.map(str::to_string),
2831	retry: JobRetryMetadata {
2832	attempt: 0,
2833	max_attempts: DEFAULT_JOB_MAX_ATTEMPTS,
2834	backoff_base_ms: DEFAULT_JOB_BACKOFF_BASE_MS,
2835	next_backoff_ms: 0,
2836	next_retry_at: None,
2837	},
2838	history: Vec::new(),
2839	created_at: 1_700_000_000,
2840	updated_at: 1_700_000_042,
2841	}
2842	}
2843
2844	#[test]
2845	fn job_record_to_agent_run_maps_non_terminal_states() {
2846	use codewhale_protocol::agent_run::RunState;
2847
2848	for (status, expected) in [
2849	(JobStatus::Queued, RunState::Queued),
2850	(JobStatus::Running, RunState::Running),
2851	(JobStatus::Paused, RunState::Paused),
2852	] {
2853	let snapshot = job_record_to_agent_run(&sample_job_record(status, None));
2854	assert!(snapshot.is_coherent());
2855	assert_eq!(snapshot.run_id, "job-o1-1");
2856	assert_eq!(snapshot.parent, None);
2857	assert_eq!(
2858	snapshot.source,
2859	codewhale_protocol::agent_run::RunSource::CoreJob
2860	);
2861	assert_eq!(snapshot.state, expected);
2862	assert!(snapshot.terminal.is_none());
2863	assert!(snapshot.refs.is_empty());
2864	assert_eq!(
2865	snapshot.budget,
2866	codewhale_protocol::agent_run::BudgetSummary::default()
2867	);
2868	}
2869	}
2870
2871	#[test]
2872	fn job_record_to_agent_run_maps_terminal_states_without_fabricating_fields() {
2873	use codewhale_protocol::agent_run::{RunState, TerminalOutcome};
2874
2875	let cases = [
2876	(
2877	JobStatus::Completed,
2878	TerminalOutcome::Completed,
2879	Some("done"),
2880	),
2881	(JobStatus::Failed, TerminalOutcome::Failed, Some("boom")),
2882	(JobStatus::Cancelled, TerminalOutcome::Cancelled, None),
2883	];
2884
2885	for (status, outcome, detail) in cases {
2886	let snapshot = job_record_to_agent_run(&sample_job_record(status, detail));
2887	assert!(snapshot.is_coherent());
2888	assert_eq!(snapshot.state, RunState::Terminal);
2889	let terminal = snapshot.terminal.expect("terminal summary");
2890	assert_eq!(terminal.outcome, outcome);
2891	assert_eq!(terminal.ended_at_ms, Some(1_700_000_042_000));
2892	assert_eq!(terminal.detail, None);
2893	assert_eq!(
2894	snapshot.budget,
2895	codewhale_protocol::agent_run::BudgetSummary::default()
2896	);
2897	assert!(snapshot.refs.is_empty());
2898	assert_eq!(snapshot.parent, None);
2899	}
2900	}
2901
2902	#[test]
2903	fn job_record_to_agent_run_does_not_export_unclassified_detail() {
2904	let record = sample_job_record(JobStatus::Failed, Some("owner-private diagnostic"));
2905	let snapshot = job_record_to_agent_run(&record);
2906	let terminal = snapshot.terminal.as_ref().expect("terminal summary");
2907	assert_eq!(terminal.detail, None);
2908	let serialized = serde_json::to_string(&snapshot).expect("serialize snapshot");
2909	assert!(!serialized.contains("owner-private diagnostic"));
2910	}
2911
2912	#[test]
2913	fn job_record_to_agent_run_omits_ended_at_on_updated_at_overflow() {
2914	let mut record = sample_job_record(JobStatus::Completed, Some("ok"));
2915	record.updated_at = i64::MAX;
2916	let snapshot = job_record_to_agent_run(&record);
2917	assert!(snapshot.is_coherent());
2918	let terminal = snapshot.terminal.expect("terminal summary");
2919	assert_eq!(terminal.ended_at_ms, None);
2920	}
2921
2922	#[test]
2923	fn unarchive_thread_updates_running_threads_cache() {
2924	let store = temp_core_state("unarchive-cache");
2925	let mut manager = ThreadManager::new(store);
2926	let spawned = manager
2927	.spawn_thread_with_history(
2928	"deepseek".to_string(),
2929	PathBuf::from("/tmp/codewhale"),
2930	InitialHistory::New,
2931	true,
2932	)
2933	.expect("spawn thread");
2934	let thread_id = spawned.thread.id.clone();
2935	let resume_params = ThreadResumeParams {
2936	thread_id: thread_id.clone(),
2937	history: None,
2938	path: None,
2939	model: None,
2940	model_provider: None,
2941	cwd: None,
2942	approval_policy: None,
2943	sandbox: None,
2944	config: None,
2945	base_instructions: None,
2946	developer_instructions: None,
2947	personality: None,
2948	persist_extended_history: false,
2949	};
2950
2951	manager.archive_thread(&thread_id).expect("archive thread");
2952	let archived = manager
2953	.resume_thread_with_history(
2954	&resume_params,
2955	Path::new("/tmp/codewhale"),
2956	"deepseek".to_string(),
2957	)
2958	.expect("resume archived thread")
2959	.expect("thread in cache");
2960	assert_eq!(archived.thread.status, ThreadStatus::Archived);
2961
2962	manager
2963	.unarchive_thread(&thread_id)
2964	.expect("unarchive thread");
2965	let restored = manager
2966	.resume_thread_with_history(
2967	&resume_params,
2968	Path::new("/tmp/codewhale"),
2969	"deepseek".to_string(),
2970	)
2971	.expect("resume unarchived thread")
2972	.expect("thread in cache");
2973	assert_eq!(restored.thread.status, ThreadStatus::Idle);
2974	}
2975
2976	#[test]
2977	fn resume_with_history_does_not_reappend_persisted_messages() {
2978	// A read→resume flow hands the thread's own history back to
2979	// `thread/resume`; appending it verbatim doubled the conversation on
2980	// every resume, compounding.
2981	let store = temp_core_state("resume-history-dedup");
2982	let mut manager = ThreadManager::new(store);
2983	let history = vec![
2984	json!({"type": "user_message", "message": "hello"}),
2985	json!({"type": "assistant_message", "message": "hi there"}),
2986	];
2987	let spawned = manager
2988	.spawn_thread_with_history(
2989	"deepseek".to_string(),
2990	PathBuf::from("/tmp/codewhale"),
2991	InitialHistory::Forked(history.clone()),
2992	true,
2993	)
2994	.expect("spawn thread");
2995	let thread_id = spawned.thread.id.clone();
2996	let message_count = \|manager: &ThreadManager\| {
2997	manager
2998	.state_store()
2999	.list_messages(&thread_id, None)
3000	.expect("list messages")
3001	.len()
3002	};
3003	assert_eq!(message_count(&manager), 2);
3004
3005	let resume_params = ThreadResumeParams {
3006	thread_id: thread_id.clone(),
3007	history: Some(history.clone()),
3008	path: None,
3009	model: None,
3010	model_provider: None,
3011	cwd: None,
3012	approval_policy: None,
3013	sandbox: None,
3014	config: None,
3015	base_instructions: None,
3016	developer_instructions: None,
3017	personality: None,
3018	persist_extended_history: false,
3019	};
3020
3021	// Resuming twice with the same history must be idempotent.
3022	for _ in 0..2 {
3023	manager
3024	.resume_thread_with_history(
3025	&resume_params,
3026	Path::new("/tmp/codewhale"),
3027	"deepseek".to_string(),
3028	)
3029	.expect("resume thread")
3030	.expect("thread found");
3031	}
3032	assert_eq!(
3033	message_count(&manager),
3034	2,
3035	"resume re-appended messages already on the persisted chain"
3036	);
3037
3038	// A genuinely new history item is still appended, exactly once.
3039	let mut extended = history.clone();
3040	extended.push(json!({"type": "user_message", "message": "something new"}));
3041	let resume_params = ThreadResumeParams {
3042	history: Some(extended),
3043	..resume_params
3044	};
3045	manager
3046	.resume_thread_with_history(
3047	&resume_params,
3048	Path::new("/tmp/codewhale"),
3049	"deepseek".to_string(),
3050	)
3051	.expect("resume thread")
3052	.expect("thread found");
3053	assert_eq!(message_count(&manager), 3);
3054	}
3055
3056	#[test]
3057	fn persist_thread_preserves_stored_policy() {
3058	// persist_thread's update payload carries no per-thread policy;
3059	// writing NULLs unconditionally erased any policy stored earlier
3060	// (e.g. on every resume).
3061	let store = temp_core_state("persist-policy");
3062	let mut metadata = test_thread_metadata("thread-policy");
3063	metadata.sandbox_policy = Some("workspace-write".to_string());
3064	metadata.approval_mode = Some("on-request".to_string());
3065	store.upsert_thread(&metadata).expect("seed thread");
3066
3067	// A fresh manager has an empty running-thread cache, so resume goes
3068	// through the persisted path, which calls persist_thread.
3069	let mut manager = ThreadManager::new(store);
3070	let resume_params = ThreadResumeParams {
3071	thread_id: "thread-policy".to_string(),
3072	history: None,
3073	path: None,
3074	model: None,
3075	model_provider: None,
3076	cwd: None,
3077	approval_policy: None,
3078	sandbox: None,
3079	config: None,
3080	base_instructions: None,
3081	developer_instructions: None,
3082	personality: None,
3083	persist_extended_history: false,
3084	};
3085	manager
3086	.resume_thread_with_history(
3087	&resume_params,
3088	Path::new("/tmp/codewhale"),
3089	"deepseek".to_string(),
3090	)
3091	.expect("resume thread")
3092	.expect("thread found");
3093
3094	let persisted = manager
3095	.state_store()
3096	.get_thread("thread-policy")
3097	.expect("read thread")
3098	.expect("thread persisted");
3099	assert_eq!(persisted.sandbox_policy.as_deref(), Some("workspace-write"));
3100	assert_eq!(persisted.approval_mode.as_deref(), Some("on-request"));
3101	}
3102
3103	#[tokio::test]
3104	async fn invoke_tool_returns_timeout_status_for_slow_tools() {
3105	use async_trait::async_trait;
3106	use codewhale_config::ConfigToml;
3107	use codewhale_execpolicy::{AskForApproval, ExecPolicyEngine};
3108	use codewhale_hooks::HookDispatcher;
3109	use codewhale_mcp::McpManager;
3110	use codewhale_protocol::{ToolKind, ToolOutput, ToolPayload};
3111	use codewhale_tools::{FunctionCallError, ToolDescriptor, ToolHandler, ToolInvocation};
3112
3113	struct SlowTool;
3114	#[async_trait]
3115	impl ToolHandler for SlowTool {
3116	fn kind(&self) -> ToolKind {
3117	ToolKind::Function
3118	}
3119
3120	async fn handle(
3121	&self,
3122	_invocation: ToolInvocation,
3123	) -> std::result::Result<ToolOutput, FunctionCallError> {
3124	time::sleep(Duration::from_millis(200)).await;
3125	Ok(ToolOutput::Function {
3126	body: Some(json!("late")),
3127	success: true,
3128	})
3129	}
3130	}
3131
3132	let mut registry = ToolRegistry::default();
3133	registry
3134	.register(
3135	ToolDescriptor {
3136	name: "slow_tool".to_string(),
3137	input_schema: json!({"type":"object"}),
3138	output_schema: json!({"type":"object"}),
3139	supports_parallel_tool_calls: true,
3140	timeout_ms: None,
3141	},
3142	Arc::new(SlowTool),
3143	)
3144	.expect("register slow tool");
3145
3146	let runtime = Runtime::new(
3147	ConfigToml::default(),
3148	temp_core_state("invoke-tool-timeout"),
3149	Arc::new(registry),
3150	Arc::new(McpManager::default()),
3151	ExecPolicyEngine::new(vec![], vec![]),
3152	HookDispatcher::default(),
3153	);
3154
3155	let result = runtime
3156	.invoke_tool(
3157	ToolCall {
3158	name: "slow_tool".to_string(),
3159	payload: ToolPayload::Function {
3160	arguments: "{}".to_string(),
3161	},
3162	source: ToolCallSource::Direct,
3163	raw_tool_call_id: None,
3164	},
3165	AskForApproval::Never,
3166	Path::new("/tmp/codewhale"),
3167	)
3168	.await
3169	.expect("invoke tool");
3170
3171	assert_eq!(result["status"], "timeout");
3172	assert_eq!(result["ok"], false);
3173	}
3174
3175	#[tokio::test]
3176	async fn thread_message_is_refused_rather_than_faked() {
3177	// This arm used to record the user message, emit canned
3178	// ResponseStart/ResponseDelta("queued")/ResponseEnd frames and report
3179	// status "accepted" — with no worker, no model, and nothing queued.
3180	// `Runtime` owns thread bookkeeping, not the turn engine, so the only
3181	// honest answer here is a refusal that names where turns actually run.
3182	let mut runtime = Runtime::new(
3183	ConfigToml::default(),
3184	temp_core_state("message-refused"),
3185	Arc::new(ToolRegistry::default()),
3186	Arc::new(McpManager::default()),
3187	ExecPolicyEngine::new(vec![], vec![]),
3188	HookDispatcher::default(),
3189	);
3190	let spawned = runtime
3191	.thread_manager
3192	.spawn_thread_with_history(
3193	"deepseek".to_string(),
3194	PathBuf::from("/tmp/codewhale"),
3195	InitialHistory::New,
3196	true,
3197	)
3198	.expect("spawn thread");
3199	let thread_id = spawned.thread.id.clone();
3200
3201	let err = runtime
3202	.handle_thread(ThreadRequest::Message {
3203	images: Vec::new(),
3204	max_output_tokens: None,
3205	thread_id: thread_id.clone(),
3206	input: "run this".to_string(),
3207	})
3208	.await
3209	.expect_err("handle_thread must not pretend to execute a turn");
3210	assert!(
3211	err.to_string().contains("/v1/threads/{id}/turns"),
3212	"the refusal must name the surface that does run turns: {err}"
3213	);
3214
3215	// Nothing was written to history on the way out.
3216	let history = runtime
3217	.thread_manager
3218	.store
3219	.list_messages(&thread_id, None)
3220	.expect("list messages");
3221	assert!(
3222	history.is_empty(),
3223	"a refused message must leave no history rows: {history:?}"
3224	);
3225	}
3226	}
3227

3227 lines RUST