task_manager.rs - CodeWhale

task_manager.rs

1	//! Persistent background task manager for Codewhale agent work.
2	//!
3	//! Tasks are durable across restarts and execute with a bounded worker pool.
4	//! Execution stays DeepSeek-only and now links every task to runtime
5	//! thread/turn records for unified timelines.
6
7	use std::collections::{HashMap, HashSet, VecDeque};
8	use std::fs;
9	use std::path::{Path, PathBuf};
10	use std::sync::Arc;
11	use std::time::Duration;
12	#[cfg(test)]
13	use std::time::Duration as StdDuration;
14
15	use anyhow::{Context, Result, anyhow, bail};
16	use async_trait::async_trait;
17	use chrono::{DateTime, Utc};
18	use serde::{Deserialize, Serialize};
19	use serde_json::{Value, json};
20	use tokio::sync::{Mutex, Notify, mpsc};
21	use tokio::time::sleep;
22	use tokio_util::sync::CancellationToken;
23	use uuid::Uuid;
24
25	use crate::config::{Config, DEFAULT_TEXT_MODEL};
26	use crate::runtime_threads::{
27	CreateThreadRequest, RuntimeThreadManager, RuntimeThreadManagerConfig, RuntimeTurnStatus,
28	SharedRuntimeThreadManager, StartTurnRequest,
29	};
30	use crate::utils::spawn_supervised;
31
32	const DEFAULT_WORKERS: usize = 2;
33	const MAX_WORKERS: usize = 8;
34	const TIMELINE_SUMMARY_LIMIT: usize = 240;
35	const ARTIFACT_THRESHOLD: usize = 1200;
36	// `lifecycle_seq` is an additive, serde-defaulted field. Keep the durable task
37	// schema at v2 so a v0.9.1 rollback can ignore it and still open tasks written
38	// by this build; no existing field changed meaning.
39	const CURRENT_TASK_SCHEMA_VERSION: u32 = 2;
40
41	const fn default_task_schema_version() -> u32 {
42	CURRENT_TASK_SCHEMA_VERSION
43	}
44
45	/// Durable task status.
46	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
47	#[serde(rename_all = "snake_case")]
48	pub enum TaskStatus {
49	Queued,
50	Running,
51	Completed,
52	Failed,
53	Canceled,
54	}
55
56	/// What the manager actually did while handling a cancellation request.
57	///
58	/// This is returned from the same state-lock transaction as the task record,
59	/// so callers never have to infer an outcome from a stale pre-cancel read.
60	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61	pub enum TaskCancelDisposition {
62	Forced,
63	Requested,
64	AlreadyFinished,
65	}
66
67	#[derive(Debug, Clone)]
68	pub struct TaskCancellation {
69	pub task: TaskRecord,
70	pub disposition: TaskCancelDisposition,
71	}
72
73	impl TaskStatus {
74	#[cfg(test)]
75	#[must_use]
76	pub fn is_terminal(self) -> bool {
77	matches!(self, Self::Completed \| Self::Failed \| Self::Canceled)
78	}
79	}
80
81	/// Durable tool-call status within a task timeline.
82	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
83	#[serde(rename_all = "snake_case")]
84	pub enum TaskToolStatus {
85	Running,
86	Success,
87	Failed,
88	Canceled,
89	}
90
91	/// Timeline entry for a task execution.
92	#[derive(Debug, Clone, Serialize, Deserialize)]
93	pub struct TaskTimelineEntry {
94	pub timestamp: DateTime<Utc>,
95	pub kind: String,
96	pub summary: String,
97	#[serde(skip_serializing_if = "Option::is_none")]
98	pub detail_path: Option<PathBuf>,
99	}
100
101	/// Tool call summary for a task.
102	#[derive(Debug, Clone, Serialize, Deserialize)]
103	pub struct TaskToolCallSummary {
104	pub id: String,
105	pub name: String,
106	pub status: TaskToolStatus,
107	pub started_at: DateTime<Utc>,
108	pub ended_at: Option<DateTime<Utc>>,
109	pub duration_ms: Option<u64>,
110	#[serde(skip_serializing_if = "Option::is_none")]
111	pub input_summary: Option<String>,
112	#[serde(skip_serializing_if = "Option::is_none")]
113	pub output_summary: Option<String>,
114	#[serde(skip_serializing_if = "Option::is_none")]
115	pub detail_path: Option<PathBuf>,
116	#[serde(skip_serializing_if = "Option::is_none")]
117	pub patch_ref: Option<PathBuf>,
118	}
119
120	/// Checklist item stored on durable tasks. This is the durable form behind the
121	/// model-visible checklist/todo compatibility tools.
122	#[derive(Debug, Clone, Serialize, Deserialize)]
123	pub struct TaskChecklistItem {
124	pub id: u32,
125	pub content: String,
126	pub status: String,
127	}
128
129	/// Checklist state associated with a task.
130	#[derive(Debug, Clone, Serialize, Deserialize, Default)]
131	pub struct TaskChecklistState {
132	pub items: Vec<TaskChecklistItem>,
133	pub completion_pct: u8,
134	pub in_progress_id: Option<u32>,
135	pub updated_at: Option<DateTime<Utc>>,
136	}
137
138	/// Structured verification evidence attached to a task.
139	#[derive(Debug, Clone, Serialize, Deserialize)]
140	pub struct TaskGateRecord {
141	pub id: String,
142	pub gate: String,
143	pub command: String,
144	pub cwd: PathBuf,
145	pub exit_code: Option<i32>,
146	pub status: String,
147	pub classification: String,
148	pub duration_ms: u64,
149	pub summary: String,
150	#[serde(skip_serializing_if = "Option::is_none")]
151	pub log_path: Option<PathBuf>,
152	pub recorded_at: DateTime<Utc>,
153	}
154
155	/// PR-attempt metadata and artifacts attached to a task.
156	#[derive(Debug, Clone, Serialize, Deserialize)]
157	pub struct TaskAttemptRecord {
158	pub id: String,
159	pub attempt_group_id: String,
160	pub attempt_index: u32,
161	pub attempt_count: u32,
162	#[serde(skip_serializing_if = "Option::is_none")]
163	pub base_ref: Option<String>,
164	#[serde(skip_serializing_if = "Option::is_none")]
165	pub base_sha: Option<String>,
166	#[serde(skip_serializing_if = "Option::is_none")]
167	pub head_ref: Option<String>,
168	#[serde(skip_serializing_if = "Option::is_none")]
169	pub head_sha: Option<String>,
170	pub summary: String,
171	pub changed_files: Vec<String>,
172	#[serde(skip_serializing_if = "Option::is_none")]
173	pub patch_path: Option<PathBuf>,
174	pub verification: Vec<String>,
175	pub selected: bool,
176	pub recorded_at: DateTime<Utc>,
177	}
178
179	/// Durable artifact reference produced by task-aware tools.
180	#[derive(Debug, Clone, Serialize, Deserialize)]
181	pub struct TaskArtifactRef {
182	pub label: String,
183	pub path: PathBuf,
184	pub summary: String,
185	pub created_at: DateTime<Utc>,
186	}
187
188	/// GitHub write/read evidence attached to a task timeline.
189	#[derive(Debug, Clone, Serialize, Deserialize)]
190	pub struct TaskGithubEvent {
191	pub id: String,
192	pub action: String,
193	pub target: String,
194	pub number: u64,
195	pub summary: String,
196	pub url: Option<String>,
197	pub recorded_at: DateTime<Utc>,
198	}
199
200	/// Durable task record.
201	#[derive(Debug, Clone, Serialize, Deserialize)]
202	pub struct TaskRecord {
203	#[serde(default = "default_task_schema_version")]
204	pub schema_version: u32,
205	pub id: String,
206	pub prompt: String,
207	pub model: String,
208	pub workspace: PathBuf,
209	pub mode: String,
210	pub allow_shell: bool,
211	pub trust_mode: bool,
212	#[serde(default = "default_auto_approve")]
213	pub auto_approve: bool,
214	pub status: TaskStatus,
215	pub created_at: DateTime<Utc>,
216	pub started_at: Option<DateTime<Utc>>,
217	pub ended_at: Option<DateTime<Utc>>,
218	pub duration_ms: Option<u64>,
219	#[serde(default, skip_serializing_if = "Option::is_none")]
220	pub hunt_verdict: Option<String>,
221	#[serde(skip_serializing_if = "Option::is_none")]
222	pub result_summary: Option<String>,
223	#[serde(skip_serializing_if = "Option::is_none")]
224	pub result_detail_path: Option<PathBuf>,
225	#[serde(skip_serializing_if = "Option::is_none")]
226	pub error: Option<String>,
227	#[serde(default, skip_serializing_if = "Option::is_none")]
228	pub thread_id: Option<String>,
229	#[serde(default, skip_serializing_if = "Option::is_none")]
230	pub turn_id: Option<String>,
231	#[serde(default, skip_serializing_if = "Option::is_none")]
232	pub owner_session_id: Option<String>,
233	#[serde(default)]
234	pub runtime_event_count: usize,
235	/// Monotonic owner-lifecycle sequence used by Work Graph reconciliation.
236	/// Output/progress events do not advance this counter; only lifecycle
237	/// transitions do, so replay after restart is stable.
238	#[serde(default)]
239	pub lifecycle_seq: u64,
240	#[serde(default)]
241	pub checklist: TaskChecklistState,
242	#[serde(default)]
243	pub gates: Vec<TaskGateRecord>,
244	#[serde(default)]
245	pub attempts: Vec<TaskAttemptRecord>,
246	#[serde(default)]
247	pub artifacts: Vec<TaskArtifactRef>,
248	#[serde(default)]
249	pub github_events: Vec<TaskGithubEvent>,
250	pub tool_calls: Vec<TaskToolCallSummary>,
251	pub timeline: Vec<TaskTimelineEntry>,
252	}
253
254	/// Lightweight task view.
255	#[derive(Debug, Clone, Serialize, Deserialize)]
256	pub struct TaskSummary {
257	pub id: String,
258	pub status: TaskStatus,
259	pub prompt_summary: String,
260	pub model: String,
261	pub mode: String,
262	pub workspace: PathBuf,
263	pub created_at: DateTime<Utc>,
264	pub started_at: Option<DateTime<Utc>>,
265	pub ended_at: Option<DateTime<Utc>>,
266	pub duration_ms: Option<u64>,
267	#[serde(default)]
268	pub lifecycle_seq: u64,
269	#[serde(default, skip_serializing_if = "Option::is_none")]
270	pub hunt_verdict: Option<String>,
271	#[serde(skip_serializing_if = "Option::is_none")]
272	pub error: Option<String>,
273	#[serde(default, skip_serializing_if = "Option::is_none")]
274	pub thread_id: Option<String>,
275	#[serde(default, skip_serializing_if = "Option::is_none")]
276	pub turn_id: Option<String>,
277	#[serde(default, skip_serializing_if = "Option::is_none")]
278	pub owner_session_id: Option<String>,
279	}
280
281	impl From<&TaskRecord> for TaskSummary {
282	fn from(value: &TaskRecord) -> Self {
283	Self {
284	id: value.id.clone(),
285	status: value.status,
286	prompt_summary: summarize_text(&value.prompt, TIMELINE_SUMMARY_LIMIT),
287	model: value.model.clone(),
288	mode: value.mode.clone(),
289	workspace: value.workspace.clone(),
290	created_at: value.created_at,
291	started_at: value.started_at,
292	ended_at: value.ended_at,
293	duration_ms: value.duration_ms,
294	lifecycle_seq: value.lifecycle_seq,
295	hunt_verdict: value.hunt_verdict.clone(),
296	error: value.error.clone(),
297	thread_id: value.thread_id.clone(),
298	turn_id: value.turn_id.clone(),
299	owner_session_id: value.owner_session_id.clone(),
300	}
301	}
302	}
303
304	/// Count totals by status for task dashboards.
305	#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
306	pub struct TaskCounts {
307	pub queued: usize,
308	pub running: usize,
309	pub completed: usize,
310	pub failed: usize,
311	pub canceled: usize,
312	}
313
314	/// Request to enqueue a new task.
315	#[derive(Debug, Clone, Serialize, Deserialize)]
316	pub struct NewTaskRequest {
317	pub prompt: String,
318	pub model: Option<String>,
319	pub workspace: Option<PathBuf>,
320	pub mode: Option<String>,
321	pub allow_shell: Option<bool>,
322	pub trust_mode: Option<bool>,
323	pub auto_approve: Option<bool>,
324	pub owner_session_id: Option<String>,
325	}
326
327	impl NewTaskRequest {
328	#[cfg(test)]
329	#[must_use]
330	pub fn from_prompt(prompt: impl Into<String>) -> Self {
331	Self {
332	prompt: prompt.into(),
333	model: None,
334	workspace: None,
335	mode: None,
336	allow_shell: None,
337	trust_mode: None,
338	auto_approve: Some(true),
339	owner_session_id: None,
340	}
341	}
342	}
343
344	/// Task manager startup options.
345	#[derive(Debug, Clone)]
346	pub struct TaskManagerConfig {
347	pub data_dir: PathBuf,
348	pub worker_count: usize,
349	pub default_workspace: PathBuf,
350	pub default_model: String,
351	pub default_mode: String,
352	pub allow_shell: bool,
353	pub trust_mode: bool,
354	}
355
356	impl TaskManagerConfig {
357	#[must_use]
358	pub fn from_runtime(
359	config: &Config,
360	workspace: PathBuf,
361	default_model: Option<String>,
362	worker_count: Option<usize>,
363	) -> Self {
364	Self {
365	data_dir: default_tasks_dir(),
366	worker_count: worker_count.unwrap_or(DEFAULT_WORKERS),
367	default_workspace: workspace,
368	default_model: default_model.unwrap_or_else(\|\| {
369	config
370	.default_text_model
371	.clone()
372	.unwrap_or_else(\|\| DEFAULT_TEXT_MODEL.to_string())
373	}),
374	default_mode: "agent".to_string(),
375	allow_shell: config.allow_shell(),
376	trust_mode: false,
377	}
378	}
379	}
380
381	#[derive(Debug, Clone)]
382	pub struct ExecutionTask {
383	id: String,
384	prompt: String,
385	model: String,
386	workspace: PathBuf,
387	mode_label: String,
388	allow_shell: bool,
389	trust_mode: bool,
390	auto_approve: bool,
391	}
392
393	/// Event stream produced by an executor while a task runs.
394	#[derive(Debug, Clone)]
395	pub enum TaskExecutionEvent {
396	ThreadLinked {
397	thread_id: String,
398	turn_id: String,
399	},
400	Status {
401	message: String,
402	},
403	MessageDelta {
404	content: String,
405	},
406	ToolStarted {
407	id: String,
408	name: String,
409	input: Value,
410	},
411	ToolProgress {
412	id: String,
413	output: String,
414	},
415	ToolCompleted {
416	id: String,
417	name: String,
418	success: bool,
419	output: String,
420	metadata: Option<Value>,
421	},
422	Error {
423	message: String,
424	},
425	RuntimeEvent {
426	seq: u64,
427	event: String,
428	summary: String,
429	},
430	}
431
432	/// Final executor result.
433	#[derive(Debug, Clone)]
434	pub struct TaskExecutionResult {
435	pub status: TaskStatus,
436	pub result_text: Option<String>,
437	pub error: Option<String>,
438	}
439
440	/// Abstraction for task execution.
441	#[async_trait]
442	pub trait TaskExecutor: Send + Sync {
443	async fn execute(
444	&self,
445	task: ExecutionTask,
446	events: mpsc::UnboundedSender<TaskExecutionEvent>,
447	cancel: CancellationToken,
448	) -> TaskExecutionResult;
449	}
450
451	/// Engine-backed executor (DeepSeek-only).
452	pub struct EngineTaskExecutor {
453	runtime_threads: SharedRuntimeThreadManager,
454	}
455
456	impl EngineTaskExecutor {
457	#[must_use]
458	pub fn new(runtime_threads: SharedRuntimeThreadManager) -> Self {
459	Self { runtime_threads }
460	}
461	}
462
463	#[async_trait]
464	impl TaskExecutor for EngineTaskExecutor {
465	async fn execute(
466	&self,
467	task: ExecutionTask,
468	events: mpsc::UnboundedSender<TaskExecutionEvent>,
469	cancel: CancellationToken,
470	) -> TaskExecutionResult {
471	let thread = match self
472	.runtime_threads
473	.create_thread(CreateThreadRequest {
474	model: Some(task.model.clone()),
475	workspace: Some(task.workspace.clone()),
476	mode: Some(task.mode_label.clone()),
477	allow_shell: Some(task.allow_shell),
478	trust_mode: Some(task.trust_mode),
479	auto_approve: Some(task.auto_approve),
480	archived: false,
481	system_prompt: None,
482	task_id: Some(task.id.clone()),
483	..Default::default()
484	})
485	.await
486	{
487	Ok(thread) => thread,
488	Err(err) => {
489	return TaskExecutionResult {
490	status: TaskStatus::Failed,
491	result_text: None,
492	error: Some(format!("Failed to create runtime thread: {err}")),
493	};
494	}
495	};
496
497	let turn = match self
498	.runtime_threads
499	.start_turn(
500	&thread.id,
501	StartTurnRequest {
502	prompt: task.prompt.clone(),
503	input_summary: Some(summarize_text(&task.prompt, TIMELINE_SUMMARY_LIMIT)),
504	model: Some(task.model.clone()),
505	mode: Some(task.mode_label.clone()),
506	allow_shell: Some(task.allow_shell),
507	trust_mode: Some(task.trust_mode),
508	auto_approve: Some(task.auto_approve),
509	..Default::default()
510	},
511	)
512	.await
513	{
514	Ok(turn) => turn,
515	Err(err) => {
516	return TaskExecutionResult {
517	status: TaskStatus::Failed,
518	result_text: None,
519	error: Some(format!("Failed to start task: {err}")),
520	};
521	}
522	};
523
524	let _ = events.send(TaskExecutionEvent::ThreadLinked {
525	thread_id: thread.id.clone(),
526	turn_id: turn.id.clone(),
527	});
528	let _ = events.send(TaskExecutionEvent::Status {
529	message: format!("Task {} started", task.id),
530	});
531
532	let mut final_text = String::new();
533	let mut seen_seq = 0u64;
534	let mut cancel_requested = false;
535	let mut terminal_status: Option<RuntimeTurnStatus> = None;
536	let mut terminal_error: Option<String> = None;
537
538	loop {
539	if cancel.is_cancelled() && !cancel_requested {
540	cancel_requested = true;
541	let _ = self
542	.runtime_threads
543	.interrupt_turn(&thread.id, &turn.id)
544	.await;
545	let _ = events.send(TaskExecutionEvent::Status {
546	message: "Cancellation requested".to_string(),
547	});
548	}
549
550	let batch = match self
551	.runtime_threads
552	.events_since_async(&thread.id, Some(seen_seq))
553	.await
554	{
555	Ok(batch) => batch,
556	Err(err) => {
557	return TaskExecutionResult {
558	status: TaskStatus::Failed,
559	result_text: if final_text.trim().is_empty() {
560	None
561	} else {
562	Some(final_text)
563	},
564	error: Some(format!("Failed to read runtime events: {err}")),
565	};
566	}
567	};
568
569	for event in batch {
570	seen_seq = seen_seq.max(event.seq);
571	let _ = events.send(TaskExecutionEvent::RuntimeEvent {
572	seq: event.seq,
573	event: event.event.clone(),
574	summary: summarize_text(&event.payload.to_string(), TIMELINE_SUMMARY_LIMIT),
575	});
576
577	match event.event.as_str() {
578	"item.delta" => {
579	let kind = event
580	.payload
581	.get("kind")
582	.and_then(Value::as_str)
583	.unwrap_or_default();
584	if kind == "agent_message" {
585	if let Some(content) =
586	event.payload.get("delta").and_then(Value::as_str)
587	{
588	final_text.push_str(content);
589	let _ = events.send(TaskExecutionEvent::MessageDelta {
590	content: content.to_string(),
591	});
592	}
593	} else if kind == "tool_call" {
594	let output = event
595	.payload
596	.get("delta")
597	.and_then(Value::as_str)
598	.unwrap_or_default()
599	.to_string();
600	let _ = events.send(TaskExecutionEvent::ToolProgress {
601	id: event.item_id.clone().unwrap_or_default(),
602	output,
603	});
604	}
605	}
606	"item.started" => {
607	if let Some(tool) = event.payload.get("tool") {
608	let id = tool
609	.get("id")
610	.and_then(Value::as_str)
611	.unwrap_or_default()
612	.to_string();
613	let name = tool
614	.get("name")
615	.and_then(Value::as_str)
616	.unwrap_or_default()
617	.to_string();
618	let input = tool.get("input").cloned().unwrap_or_else(\|\| json!({}));
619	let _ =
620	events.send(TaskExecutionEvent::ToolStarted { id, name, input });
621	}
622	}
623	"item.completed" \| "item.failed" => {
624	if let Some(item) = event.payload.get("item") {
625	let kind = item.get("kind").and_then(Value::as_str).unwrap_or_default();
626	if kind == "tool_call"
627	\|\| kind == "file_change"
628	\|\| kind == "command_execution"
629	{
630	let id = item
631	.get("id")
632	.and_then(Value::as_str)
633	.unwrap_or_default()
634	.to_string();
635	let name = item
636	.get("summary")
637	.and_then(Value::as_str)
638	.unwrap_or("tool")
639	.split(':')
640	.next()
641	.unwrap_or("tool")
642	.trim()
643	.to_string();
644	let output = item
645	.get("detail")
646	.and_then(Value::as_str)
647	.unwrap_or_default()
648	.to_string();
649	let metadata = item.get("metadata").cloned();
650	let _ = events.send(TaskExecutionEvent::ToolCompleted {
651	id,
652	name,
653	success: event.event == "item.completed",
654	output,
655	metadata,
656	});
657	} else if kind == "status" {
658	let message = item
659	.get("detail")
660	.and_then(Value::as_str)
661	.or_else(\|\| item.get("summary").and_then(Value::as_str))
662	.unwrap_or_default()
663	.to_string();
664	let _ = events.send(TaskExecutionEvent::Status { message });
665	} else if kind == "error" {
666	let message = item
667	.get("detail")
668	.and_then(Value::as_str)
669	.or_else(\|\| item.get("summary").and_then(Value::as_str))
670	.unwrap_or_default()
671	.to_string();
672	let _ = events.send(TaskExecutionEvent::Error { message });
673	}
674	}
675	}
676	"turn.completed" => {
677	if let Some(turn_payload) = event.payload.get("turn") {
678	let status = turn_payload
679	.get("status")
680	.and_then(Value::as_str)
681	.unwrap_or("failed");
682	terminal_status = Some(match status {
683	"completed" => RuntimeTurnStatus::Completed,
684	"interrupted" => RuntimeTurnStatus::Interrupted,
685	"canceled" => RuntimeTurnStatus::Canceled,
686	_ => RuntimeTurnStatus::Failed,
687	});
688	terminal_error = turn_payload
689	.get("error")
690	.and_then(Value::as_str)
691	.map(ToString::to_string);
692	} else {
693	terminal_status = Some(RuntimeTurnStatus::Completed);
694	}
695	}
696	_ => {}
697	}
698	}
699
700	if terminal_status.is_some() {
701	break;
702	}
703
704	sleep(Duration::from_millis(40)).await;
705	}
706
707	match terminal_status.unwrap_or(RuntimeTurnStatus::Failed) {
708	RuntimeTurnStatus::Completed => TaskExecutionResult {
709	status: TaskStatus::Completed,
710	result_text: if final_text.trim().is_empty() {
711	None
712	} else {
713	Some(final_text)
714	},
715	error: None,
716	},
717	RuntimeTurnStatus::Interrupted \| RuntimeTurnStatus::Canceled => TaskExecutionResult {
718	status: TaskStatus::Canceled,
719	result_text: if final_text.trim().is_empty() {
720	None
721	} else {
722	Some(final_text)
723	},
724	error: None,
725	},
726	RuntimeTurnStatus::Queued
727	\| RuntimeTurnStatus::InProgress
728	\| RuntimeTurnStatus::Failed => TaskExecutionResult {
729	status: TaskStatus::Failed,
730	result_text: if final_text.trim().is_empty() {
731	None
732	} else {
733	Some(final_text)
734	},
735	error: terminal_error.or_else(\|\| Some("Task ended unexpectedly".to_string())),
736	},
737	}
738	}
739	}
740
741	/// Thread-safe task manager.
742	pub type SharedTaskManager = Arc<TaskManager>;
743
744	pub struct TaskManager {
745	cfg: TaskManagerConfig,
746	default_workspace: Mutex<PathBuf>,
747	executor: Arc<dyn TaskExecutor>,
748	tasks_dir: PathBuf,
749	artifacts_dir: PathBuf,
750	queue_path: PathBuf,
751	state: Mutex<ManagerState>,
752	notify: Notify,
753	cancel_token: CancellationToken,
754	}
755
756	struct ManagerState {
757	tasks: HashMap<String, TaskRecord>,
758	queue: VecDeque<String>,
759	running_cancel: HashMap<String, CancellationToken>,
760	}
761
762	#[derive(Debug, Serialize, Deserialize, Default)]
763	struct QueueFile {
764	queue: Vec<String>,
765	}
766
767	impl TaskManager {
768	/// Start the manager with the default DeepSeek executor.
769	pub async fn start(
770	cfg: TaskManagerConfig,
771	api_config: Config,
772	plugin_registry: Arc<crate::plugins::PluginRegistry>,
773	) -> Result<SharedTaskManager> {
774	let runtime_threads = Arc::new(RuntimeThreadManager::open_with_plugin_registry(
775	api_config.clone(),
776	cfg.default_workspace.clone(),
777	RuntimeThreadManagerConfig::from_task_data_dir(cfg.data_dir.clone()),
778	plugin_registry,
779	)?);
780	Self::start_with_runtime_manager(cfg, api_config, runtime_threads).await
781	}
782
783	/// Start the manager with an injected runtime thread manager.
784	pub async fn start_with_runtime_manager(
785	cfg: TaskManagerConfig,
786	_api_config: Config,
787	runtime_threads: SharedRuntimeThreadManager,
788	) -> Result<SharedTaskManager> {
789	let executor: Arc<dyn TaskExecutor> =
790	Arc::new(EngineTaskExecutor::new(runtime_threads.clone()));
791	let manager = Self::start_with_executor(cfg, executor).await?;
792	runtime_threads.attach_task_manager(manager.clone());
793	Ok(manager)
794	}
795
796	/// Start the manager with a custom executor (used for tests).
797	pub async fn start_with_executor(
798	cfg: TaskManagerConfig,
799	executor: Arc<dyn TaskExecutor>,
800	) -> Result<SharedTaskManager> {
801	let workers = cfg.worker_count.clamp(1, MAX_WORKERS);
802	let tasks_dir = cfg.data_dir.join("tasks");
803	let artifacts_dir = cfg.data_dir.join("artifacts");
804	let queue_path = cfg.data_dir.join("queue.json");
805	fs::create_dir_all(&tasks_dir)
806	.with_context(\|\| format!("Failed to create tasks dir {}", tasks_dir.display()))?;
807	fs::create_dir_all(&artifacts_dir).with_context(\|\| {
808	format!(
809	"Failed to create task artifacts dir {}",
810	artifacts_dir.display()
811	)
812	})?;
813
814	let LoadedTaskState {
815	tasks,
816	queue,
817	recovered,
818	} = load_state(&tasks_dir, &queue_path)?;
819
820	let cancel_token = CancellationToken::new();
821	let default_workspace = cfg.default_workspace.clone();
822	let manager = Arc::new(Self {
823	cfg,
824	default_workspace: Mutex::new(default_workspace),
825	executor,
826	tasks_dir,
827	artifacts_dir,
828	queue_path,
829	state: Mutex::new(ManagerState {
830	tasks,
831	queue,
832	running_cancel: HashMap::new(),
833	}),
834	notify: Notify::new(),
835	cancel_token: cancel_token.clone(),
836	});
837
838	{
839	// Persist only what boot actually changed: the reconciled queue
840	// and any running->failed recoveries. Rewriting every task record
841	// on every launch was a full-store write storm (#3757).
842	let state = manager.state.lock().await;
843	manager.persist_queue_locked(&state.queue)?;
844	for id in &recovered {
845	if let Some(task) = state.tasks.get(id) {
846	manager.persist_task_locked(task)?;
847	}
848	}
849	}
850
851	for _ in 0..workers {
852	let manager_clone = Arc::clone(&manager);
853	spawn_supervised(
854	"task-manager-worker",
855	std::panic::Location::caller(),
856	async move {
857	manager_clone.worker_loop().await;
858	},
859	);
860	}
861
862	Ok(manager)
863	}
864
865	#[allow(dead_code)] // Public API for external callers (runtime API)
866	pub fn shutdown(&self) {
867	self.cancel_token.cancel();
868	}
869
870	#[allow(dead_code)] // Public API for external callers
871	pub fn is_shutdown(&self) -> bool {
872	self.cancel_token.is_cancelled()
873	}
874
875	pub async fn set_default_workspace(&self, workspace: PathBuf) {
876	let mut default_workspace = self.default_workspace.lock().await;
877	*default_workspace = workspace;
878	}
879
880	pub async fn default_workspace(&self) -> PathBuf {
881	self.default_workspace.lock().await.clone()
882	}
883
884	/// Enqueue a new task.
885	pub async fn add_task(&self, req: NewTaskRequest) -> Result<TaskRecord> {
886	self.add_task_with_id(req, Self::new_task_id()).await
887	}
888
889	/// Allocate the durable owner identity before queue insertion so callers
890	/// can register graph spawn intent first.
891	#[must_use]
892	pub(crate) fn new_task_id() -> String {
893	format!("task_{}", &Uuid::new_v4().simple().to_string()[..16])
894	}
895
896	/// Enqueue using a preallocated id. This is crate-visible only for the
897	/// model tool's register-before-work transaction.
898	pub(crate) async fn add_task_with_id(
899	&self,
900	req: NewTaskRequest,
901	task_id: String,
902	) -> Result<TaskRecord> {
903	let prompt = req.prompt.trim().to_string();
904	if prompt.is_empty() {
905	bail!("Task prompt cannot be empty");
906	}
907	if task_id.len() != 21
908	\|\| !task_id.starts_with("task_")
909	\|\| !task_id[5..].chars().all(\|ch\| ch.is_ascii_hexdigit())
910	{
911	bail!("Invalid preallocated task id: expected task_<16hex>");
912	}
913
914	let task = TaskRecord {
915	schema_version: CURRENT_TASK_SCHEMA_VERSION,
916	// 16 random hex chars (was 8; ~60 bits of entropy once UUIDv4's
917	// fixed version nibble is discounted): task ids live in durable
918	// state that accumulates across restarts, and a collision
919	// overwrites a record while leaving a duplicate queue entry.
920	// `resolve_task_id` matches by prefix, so short references still
921	// work.
922	id: task_id,
923	prompt,
924	model: req.model.unwrap_or_else(\|\| self.cfg.default_model.clone()),
925	workspace: match req.workspace {
926	Some(workspace) => workspace,
927	None => self.default_workspace().await,
928	},
929	mode: req.mode.unwrap_or_else(\|\| self.cfg.default_mode.clone()),
930	allow_shell: req.allow_shell.unwrap_or(self.cfg.allow_shell),
931	trust_mode: req.trust_mode.unwrap_or(self.cfg.trust_mode),
932	// Auto-approval must be opted into explicitly
933	// (GHSA-72w5-pf8h-xfp4).
934	auto_approve: req.auto_approve.unwrap_or(false),
935	status: TaskStatus::Queued,
936	created_at: Utc::now(),
937	started_at: None,
938	ended_at: None,
939	duration_ms: None,
940	hunt_verdict: None,
941	result_summary: None,
942	result_detail_path: None,
943	error: None,
944	thread_id: None,
945	turn_id: None,
946	owner_session_id: req.owner_session_id,
947	runtime_event_count: 0,
948	lifecycle_seq: 1,
949	checklist: TaskChecklistState::default(),
950	gates: Vec::new(),
951	attempts: Vec::new(),
952	artifacts: Vec::new(),
953	github_events: Vec::new(),
954	tool_calls: Vec::new(),
955	timeline: vec![TaskTimelineEntry {
956	timestamp: Utc::now(),
957	kind: "queued".to_string(),
958	summary: "Task queued".to_string(),
959	detail_path: None,
960	}],
961	};
962
963	{
964	let mut state = self.state.lock().await;
965	let task_path = self.tasks_dir.join(format!("{}.json", task.id));
966	// The staged extension is intentionally not `.json`, so startup
967	// replay ignores an interrupted create until the queue write has
968	// succeeded and this file is atomically promoted.
969	let staged_task_path = self.tasks_dir.join(format!(".{}.json.pending", task.id));
970	if state.tasks.contains_key(&task.id) \|\| task_path.exists() \|\| staged_task_path.exists()
971	{
972	bail!("Task id already exists: {}", task.id);
973	}
974	let mut next_queue = state.queue.clone();
975	next_queue.push_back(task.id.clone());
976
977	// Stage the owner record, then persist its queue membership, then
978	// atomically promote it. A crash before promotion leaves either an
979	// ignored staged file or a queue entry with no task (which replay
980	// drops); a crash after promotion leaves the complete runnable
981	// pair. In-memory scheduling is published only after all three.
982	write_json_atomic(&staged_task_path, &task)?;
983	if let Err(err) = self.persist_queue_locked(&next_queue) {
984	if let Err(cleanup_err) = fs::remove_file(&staged_task_path) {
985	tracing::warn!(
986	task_id = %task.id,
987	error = %cleanup_err,
988	"failed to remove ignored staged task after queue write failure"
989	);
990	}
991	return Err(err);
992	}
993	if let Err(promote_err) = fs::rename(&staged_task_path, &task_path) {
994	let rollback_error = self.persist_queue_locked(&state.queue).err();
995	let cleanup_error = fs::remove_file(&staged_task_path).err();
996	let mut message =
997	format!("Failed to promote staged task {}: {promote_err}", task.id);
998	if let Some(rollback_error) = rollback_error {
999	message.push_str(&format!("; queue rollback also failed: {rollback_error:#}"));
1000	}
1001	if let Some(cleanup_error) = cleanup_error {
1002	message.push_str(&format!(
1003	"; ignored staged-file cleanup also failed: {cleanup_error}"
1004	));
1005	}
1006	bail!(message);
1007	}
1008	state.queue = next_queue;
1009	state.tasks.insert(task.id.clone(), task.clone());
1010	}
1011	self.notify.notify_one();
1012	Ok(task)
1013	}
1014
1015	/// List tasks, newest first.
1016	pub async fn list_tasks(&self, limit: Option<usize>) -> Vec<TaskSummary> {
1017	self.list_tasks_scoped(limit, None).await
1018	}
1019
1020	/// List tasks, newest first, optionally scoped to a workspace.
1021	pub async fn list_tasks_scoped(
1022	&self,
1023	limit: Option<usize>,
1024	workspace: Option<&Path>,
1025	) -> Vec<TaskSummary> {
1026	let state = self.state.lock().await;
1027	let mut items = state
1028	.tasks
1029	.values()
1030	.filter(\|record\| {
1031	workspace.is_none_or(\|workspace\| record.workspace.as_path() == workspace)
1032	})
1033	.map(TaskSummary::from)
1034	.collect::<Vec<_>>();
1035	items.sort_by_key(\|i\| std::cmp::Reverse(i.created_at));
1036	if let Some(limit) = limit {
1037	items.truncate(limit);
1038	}
1039	items
1040	}
1041
1042	/// Retrieve a task by full id or prefix.
1043	pub async fn get_task(&self, id_or_prefix: &str) -> Result<TaskRecord> {
1044	let state = self.state.lock().await;
1045	let id = resolve_task_id(&state.tasks, id_or_prefix)?;
1046	state
1047	.tasks
1048	.get(&id)
1049	.cloned()
1050	.ok_or_else(\|\| anyhow!("Task not found: {id_or_prefix}"))
1051	}
1052
1053	/// Cancel a queued or running task by id/prefix.
1054	pub async fn cancel_task(&self, id_or_prefix: &str) -> Result<TaskCancellation> {
1055	let mut state = self.state.lock().await;
1056	let id = resolve_task_id(&state.tasks, id_or_prefix)?;
1057	let now = Utc::now();
1058
1059	let mut cancel_running = false;
1060	let disposition = {
1061	let task = state
1062	.tasks
1063	.get_mut(&id)
1064	.ok_or_else(\|\| anyhow!("Task not found: {id}"))?;
1065	match task.status {
1066	TaskStatus::Queued => {
1067	task.status = TaskStatus::Canceled;
1068	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
1069	task.ended_at = Some(now);
1070	task.duration_ms = Some(0);
1071	task.timeline.push(TaskTimelineEntry {
1072	timestamp: now,
1073	kind: "canceled".to_string(),
1074	summary: "Task canceled before execution".to_string(),
1075	detail_path: None,
1076	});
1077	state.queue.retain(\|queued_id\| queued_id != &id);
1078	TaskCancelDisposition::Forced
1079	}
1080	TaskStatus::Running => {
1081	cancel_running = true;
1082	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
1083	task.timeline.push(TaskTimelineEntry {
1084	timestamp: now,
1085	kind: "cancel_requested".to_string(),
1086	summary: "Cancellation requested".to_string(),
1087	detail_path: None,
1088	});
1089	TaskCancelDisposition::Requested
1090	}
1091	TaskStatus::Completed \| TaskStatus::Failed \| TaskStatus::Canceled => {
1092	TaskCancelDisposition::AlreadyFinished
1093	}
1094	}
1095	};
1096
1097	if cancel_running && let Some(token) = state.running_cancel.get(&id) {
1098	token.cancel();
1099	}
1100
1101	self.persist_all_locked(&state)?;
1102	let task = state
1103	.tasks
1104	.get(&id)
1105	.cloned()
1106	.ok_or_else(\|\| anyhow!("Task not found: {id}"))?;
1107	Ok(TaskCancellation { task, disposition })
1108	}
1109
1110	/// Return aggregate status counters.
1111	pub async fn counts(&self) -> TaskCounts {
1112	let state = self.state.lock().await;
1113	let mut counts = TaskCounts::default();
1114	for task in state.tasks.values() {
1115	match task.status {
1116	TaskStatus::Queued => counts.queued += 1,
1117	TaskStatus::Running => counts.running += 1,
1118	TaskStatus::Completed => counts.completed += 1,
1119	TaskStatus::Failed => counts.failed += 1,
1120	TaskStatus::Canceled => counts.canceled += 1,
1121	}
1122	}
1123	counts
1124	}
1125
1126	/// Root directory for durable task state.
1127	#[must_use]
1128	pub fn data_dir(&self) -> PathBuf {
1129	self.cfg.data_dir.clone()
1130	}
1131
1132	/// Resolve a task artifact reference to an absolute path.
1133	#[must_use]
1134	pub fn artifact_absolute_path(&self, path: &Path) -> PathBuf {
1135	if path.is_absolute() {
1136	path.to_path_buf()
1137	} else {
1138	self.cfg.data_dir.join(path)
1139	}
1140	}
1141
1142	/// Write a durable task artifact and return the persisted path reference.
1143	pub fn write_task_artifact(
1144	&self,
1145	task_id: &str,
1146	label: &str,
1147	content: &str,
1148	) -> Result<PathBuf> {
1149	self.write_artifact(task_id, label, content)
1150	}
1151
1152	/// Apply model-visible tool metadata to a task and persist it.
1153	pub async fn record_tool_metadata(
1154	&self,
1155	id_or_prefix: &str,
1156	metadata: &Value,
1157	) -> Result<TaskRecord> {
1158	let mut state = self.state.lock().await;
1159	let id = resolve_task_id(&state.tasks, id_or_prefix)?;
1160	let updated = {
1161	let task = state
1162	.tasks
1163	.get_mut(&id)
1164	.ok_or_else(\|\| anyhow!("Task not found: {id}"))?;
1165	self.apply_task_update_metadata(task, Some(metadata))?;
1166	task.clone()
1167	};
1168	self.persist_task_locked(&updated)?;
1169	Ok(updated)
1170	}
1171
1172	async fn worker_loop(self: Arc<Self>) {
1173	loop {
1174	if self.cancel_token.is_cancelled() {
1175	tracing::debug!("Worker exiting due to shutdown");
1176	break;
1177	}
1178	let next = {
1179	let mut state = self.state.lock().await;
1180	match state.queue.pop_front() {
1181	None => None,
1182	Some(task_id) => {
1183	if let Some(task) = state.tasks.get_mut(&task_id) {
1184	if task.status != TaskStatus::Queued {
1185	let _ = self.persist_queue_locked(&state.queue);
1186	None
1187	} else {
1188	let now = Utc::now();
1189	task.status = TaskStatus::Running;
1190	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
1191	task.started_at = Some(now);
1192	task.ended_at = None;
1193	task.duration_ms = None;
1194	task.error = None;
1195	task.timeline.push(TaskTimelineEntry {
1196	timestamp: now,
1197	kind: "running".to_string(),
1198	summary: "Task started".to_string(),
1199	detail_path: None,
1200	});
1201
1202	let request = {
1203	ExecutionTask {
1204	id: task.id.clone(),
1205	prompt: task.prompt.clone(),
1206	model: task.model.clone(),
1207	workspace: task.workspace.clone(),
1208	mode_label: task.mode.clone(),
1209	allow_shell: task.allow_shell,
1210	trust_mode: task.trust_mode,
1211	auto_approve: task.auto_approve,
1212	}
1213	};
1214	let cancel = CancellationToken::new();
1215	state.running_cancel.insert(task_id.clone(), cancel.clone());
1216
1217	if let Err(err) = self.persist_all_locked(&state) {
1218	tracing::error!("Failed to persist task start: {err}");
1219	}
1220	Some((task_id, request, cancel))
1221	}
1222	} else {
1223	let _ = self.persist_queue_locked(&state.queue);
1224	None
1225	}
1226	}
1227	}
1228	};
1229
1230	let Some((task_id, request, cancel)) = next else {
1231	tokio::select! {
1232	_ = self.cancel_token.cancelled() => {
1233	tracing::debug!("Worker exiting during wait");
1234	break;
1235	}
1236	_ = self.notify.notified() => {}
1237	}
1238	continue;
1239	};
1240
1241	self.run_task(task_id, request, cancel).await;
1242	}
1243	}
1244
1245	async fn run_task(&self, task_id: String, request: ExecutionTask, cancel: CancellationToken) {
1246	let (event_tx, mut event_rx) = mpsc::unbounded_channel();
1247	let exec_fut = self
1248	.executor
1249	.execute(request.clone(), event_tx, cancel.clone());
1250	tokio::pin!(exec_fut);
1251
1252	let result = loop {
1253	tokio::select! {
1254	maybe_event = event_rx.recv() => {
1255	if let Some(event) = maybe_event
1256	&& let Err(err) = self.apply_execution_event(&task_id, event).await
1257	{
1258	tracing::error!("Failed to apply task event for {task_id}: {err}");
1259	}
1260	}
1261	exec_result = &mut exec_fut => {
1262	break exec_result;
1263	}
1264	}
1265	};
1266
1267	while let Ok(event) = event_rx.try_recv() {
1268	if let Err(err) = self.apply_execution_event(&task_id, event).await {
1269	tracing::error!("Failed to apply trailing task event for {task_id}: {err}");
1270	}
1271	}
1272
1273	if let Err(err) = self
1274	.finish_task(&task_id, result, cancel, &request.mode_label)
1275	.await
1276	{
1277	tracing::error!("Failed to finalize task {task_id}: {err}");
1278	}
1279	}
1280
1281	async fn apply_execution_event(&self, task_id: &str, event: TaskExecutionEvent) -> Result<()> {
1282	let mut state = self.state.lock().await;
1283	let Some(task) = state.tasks.get_mut(task_id) else {
1284	return Ok(());
1285	};
1286
1287	match event {
1288	TaskExecutionEvent::ThreadLinked { thread_id, turn_id } => {
1289	task.thread_id = Some(thread_id.clone());
1290	task.turn_id = Some(turn_id.clone());
1291	task.timeline.push(TaskTimelineEntry {
1292	timestamp: Utc::now(),
1293	kind: "runtime_link".to_string(),
1294	summary: format!("Linked runtime thread {thread_id} turn {turn_id}"),
1295	detail_path: None,
1296	});
1297	}
1298	TaskExecutionEvent::Status { message } => {
1299	task.timeline.push(TaskTimelineEntry {
1300	timestamp: Utc::now(),
1301	kind: "status".to_string(),
1302	summary: summarize_text(&message, TIMELINE_SUMMARY_LIMIT),
1303	detail_path: None,
1304	});
1305	}
1306	TaskExecutionEvent::MessageDelta { content } => {
1307	if !content.trim().is_empty() {
1308	task.timeline.push(TaskTimelineEntry {
1309	timestamp: Utc::now(),
1310	kind: "message".to_string(),
1311	summary: summarize_text(&content, TIMELINE_SUMMARY_LIMIT),
1312	detail_path: None,
1313	});
1314	}
1315	}
1316	TaskExecutionEvent::ToolStarted { id, name, input } => {
1317	let input_summary = summarize_json(&input);
1318	task.tool_calls.push(TaskToolCallSummary {
1319	id: id.clone(),
1320	name: name.clone(),
1321	status: TaskToolStatus::Running,
1322	started_at: Utc::now(),
1323	ended_at: None,
1324	duration_ms: None,
1325	input_summary: input_summary.clone(),
1326	output_summary: None,
1327	detail_path: None,
1328	patch_ref: None,
1329	});
1330	let summary = input_summary
1331	.map(\|s\| format!("{name} started ({s})"))
1332	.unwrap_or_else(\|\| format!("{name} started"));
1333	task.timeline.push(TaskTimelineEntry {
1334	timestamp: Utc::now(),
1335	kind: "tool_started".to_string(),
1336	summary,
1337	detail_path: None,
1338	});
1339	}
1340	TaskExecutionEvent::ToolProgress { id, output } => {
1341	task.timeline.push(TaskTimelineEntry {
1342	timestamp: Utc::now(),
1343	kind: "tool_progress".to_string(),
1344	summary: format!(
1345	"{id}: {}",
1346	summarize_text(&output, TIMELINE_SUMMARY_LIMIT.saturating_sub(8))
1347	),
1348	detail_path: None,
1349	});
1350	}
1351	TaskExecutionEvent::ToolCompleted {
1352	id,
1353	name,
1354	success,
1355	output,
1356	metadata,
1357	} => {
1358	let now = Utc::now();
1359	let detail_path = self.artifact_if_large(task_id, &name, &output)?;
1360	let output_summary = summarize_text(&output, TIMELINE_SUMMARY_LIMIT);
1361	let patch_ref = if name == "apply_patch" {
1362	detail_path.clone()
1363	} else {
1364	None
1365	};
1366
1367	if let Some(call) = task.tool_calls.iter_mut().find(\|call\| call.id == id) {
1368	call.status = if success {
1369	TaskToolStatus::Success
1370	} else {
1371	TaskToolStatus::Failed
1372	};
1373	call.ended_at = Some(now);
1374	call.duration_ms = Some(duration_ms(call.started_at, now));
1375	call.output_summary = Some(output_summary.clone());
1376	call.detail_path = detail_path.clone();
1377	call.patch_ref = patch_ref.clone();
1378
1379	if call.duration_ms.is_none()
1380	&& let Some(duration) = metadata
1381	.as_ref()
1382	.and_then(\|m\| m.get("duration_ms"))
1383	.and_then(Value::as_u64)
1384	{
1385	call.duration_ms = Some(duration);
1386	}
1387	}
1388
1389	let status = if success { "success" } else { "failed" };
1390	task.timeline.push(TaskTimelineEntry {
1391	timestamp: now,
1392	kind: "tool_completed".to_string(),
1393	summary: format!("{name} {status}: {output_summary}"),
1394	detail_path: detail_path.clone(),
1395	});
1396	if let Some(patch_ref) = patch_ref {
1397	task.timeline.push(TaskTimelineEntry {
1398	timestamp: now,
1399	kind: "patch_ref".to_string(),
1400	summary: format!("Patch artifact: {}", patch_ref.display()),
1401	detail_path: Some(patch_ref),
1402	});
1403	}
1404
1405	self.apply_task_update_metadata(task, metadata.as_ref())?;
1406	}
1407	TaskExecutionEvent::Error { message } => {
1408	task.timeline.push(TaskTimelineEntry {
1409	timestamp: Utc::now(),
1410	kind: "error".to_string(),
1411	summary: summarize_text(&message, TIMELINE_SUMMARY_LIMIT),
1412	detail_path: None,
1413	});
1414	}
1415	TaskExecutionEvent::RuntimeEvent {
1416	seq,
1417	event,
1418	summary,
1419	} => {
1420	task.runtime_event_count = task.runtime_event_count.saturating_add(1);
1421	task.timeline.push(TaskTimelineEntry {
1422	timestamp: Utc::now(),
1423	kind: "runtime_event".to_string(),
1424	summary: format!("#{seq} {event}: {summary}"),
1425	detail_path: None,
1426	});
1427	}
1428	}
1429
1430	self.persist_task_locked(task)?;
1431	Ok(())
1432	}
1433
1434	async fn finish_task(
1435	&self,
1436	task_id: &str,
1437	mut result: TaskExecutionResult,
1438	cancel: CancellationToken,
1439	mode_label: &str,
1440	) -> Result<()> {
1441	let mut state = self.state.lock().await;
1442	state.running_cancel.remove(task_id);
1443	let Some(task) = state.tasks.get_mut(task_id) else {
1444	return Ok(());
1445	};
1446
1447	let now = Utc::now();
1448	if cancel.is_cancelled() && result.status == TaskStatus::Completed {
1449	result.status = TaskStatus::Canceled;
1450	result.result_text = None;
1451	result.error = None;
1452	}
1453
1454	task.status = result.status;
1455	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
1456	task.mode = mode_label.to_string();
1457	task.ended_at = Some(now);
1458	task.duration_ms = task.started_at.map(\|start\| duration_ms(start, now));
1459	task.error = result.error.clone();
1460	task.timeline.push(TaskTimelineEntry {
1461	timestamp: now,
1462	kind: "finished".to_string(),
1463	summary: match result.status {
1464	TaskStatus::Completed => "Task completed".to_string(),
1465	TaskStatus::Failed => format!(
1466	"Task failed: {}",
1467	result
1468	.error
1469	.as_deref()
1470	.map(\|e\| summarize_text(e, TIMELINE_SUMMARY_LIMIT))
1471	.unwrap_or_else(\|\| "unknown error".to_string())
1472	),
1473	TaskStatus::Canceled => "Task canceled".to_string(),
1474	TaskStatus::Queued \| TaskStatus::Running => {
1475	format!("Task ended in unexpected state: {mode_label}")
1476	}
1477	},
1478	detail_path: None,
1479	});
1480
1481	if let Some(text) = result.result_text {
1482	let detail_path = self.artifact_if_large(task_id, "result", &text)?;
1483	task.result_summary = Some(summarize_text(&text, TIMELINE_SUMMARY_LIMIT));
1484	task.result_detail_path = detail_path.clone();
1485	if let Some(detail_path) = detail_path {
1486	task.timeline.push(TaskTimelineEntry {
1487	timestamp: now,
1488	kind: "result_ref".to_string(),
1489	summary: format!("Result artifact: {}", detail_path.display()),
1490	detail_path: Some(detail_path),
1491	});
1492	}
1493	} else if result.status == TaskStatus::Completed {
1494	task.result_summary = Some("(no textual output)".to_string());
1495	}
1496
1497	self.persist_all_locked(&state)?;
1498	Ok(())
1499	}
1500
1501	fn artifact_if_large(
1502	&self,
1503	task_id: &str,
1504	label: &str,
1505	content: &str,
1506	) -> Result<Option<PathBuf>> {
1507	if content.len() < ARTIFACT_THRESHOLD {
1508	return Ok(None);
1509	}
1510	self.write_artifact(task_id, label, content).map(Some)
1511	}
1512
1513	fn write_artifact(&self, task_id: &str, label: &str, content: &str) -> Result<PathBuf> {
1514	ensure_safe_storage_id("task id", task_id)?;
1515	let artifact_dir = self.artifacts_dir.join(task_id);
1516	fs::create_dir_all(&artifact_dir)
1517	.with_context(\|\| format!("Failed to create artifact dir {}", artifact_dir.display()))?;
1518	let stamp = Utc::now().format("%Y%m%dT%H%M%S%.3fZ");
1519	let filename = format!("{stamp}_{}.txt", sanitize_filename(label));
1520	let absolute = artifact_dir.join(filename);
1521	fs::write(&absolute, content)
1522	.with_context(\|\| format!("Failed to write artifact {}", absolute.display()))?;
1523	let relative = absolute
1524	.strip_prefix(&self.cfg.data_dir)
1525	.map(PathBuf::from)
1526	.unwrap_or(absolute);
1527	Ok(relative)
1528	}
1529
1530	fn apply_task_update_metadata(
1531	&self,
1532	task: &mut TaskRecord,
1533	metadata: Option<&Value>,
1534	) -> Result<()> {
1535	let Some(updates) = metadata.and_then(\|m\| m.get("task_updates")) else {
1536	return Ok(());
1537	};
1538	let now = Utc::now();
1539
1540	if let Some(value) = updates.get("checklist") {
1541	let mut checklist: TaskChecklistState = serde_json::from_value(value.clone())
1542	.context("Failed to parse checklist task update")?;
1543	checklist.updated_at = checklist.updated_at.or(Some(now));
1544	task.checklist = checklist;
1545	task.timeline.push(TaskTimelineEntry {
1546	timestamp: now,
1547	kind: "checklist".to_string(),
1548	summary: format!(
1549	"Checklist updated: {} item(s), {}% complete",
1550	task.checklist.items.len(),
1551	task.checklist.completion_pct
1552	),
1553	detail_path: None,
1554	});
1555	}
1556
1557	if let Some(value) = updates.get("gate") {
1558	let gate: TaskGateRecord = serde_json::from_value(value.clone())
1559	.context("Failed to parse gate task update")?;
1560	let summary = format!("Gate {} {}: {}", gate.gate, gate.status, gate.summary);
1561	task.gates.retain(\|existing\| existing.id != gate.id);
1562	task.gates.push(gate.clone());
1563	task.timeline.push(TaskTimelineEntry {
1564	timestamp: now,
1565	kind: "gate".to_string(),
1566	summary: summarize_text(&summary, TIMELINE_SUMMARY_LIMIT),
1567	detail_path: gate.log_path,
1568	});
1569	}
1570
1571	if let Some(value) = updates.get("hunt_verdict") {
1572	let raw = value
1573	.as_str()
1574	.ok_or_else(\|\| anyhow!("hunt_verdict task update must be a string"))?;
1575	let verdict = normalize_hunt_verdict(raw)?;
1576	if task.hunt_verdict.as_deref() != Some(verdict) {
1577	task.hunt_verdict = Some(verdict.to_string());
1578	task.timeline.push(TaskTimelineEntry {
1579	timestamp: now,
1580	kind: "hunt_verdict".to_string(),
1581	summary: format!("Hunt verdict updated: {verdict}"),
1582	detail_path: None,
1583	});
1584	}
1585	}
1586
1587	if let Some(value) = updates.get("attempt") {
1588	let attempt: TaskAttemptRecord = serde_json::from_value(value.clone())
1589	.context("Failed to parse attempt task update")?;
1590	task.attempts.retain(\|existing\| existing.id != attempt.id);
1591	task.attempts.push(attempt.clone());
1592	task.timeline.push(TaskTimelineEntry {
1593	timestamp: now,
1594	kind: "pr_attempt".to_string(),
1595	summary: format!(
1596	"Attempt {}/{} recorded for {}",
1597	attempt.attempt_index, attempt.attempt_count, attempt.attempt_group_id
1598	),
1599	detail_path: attempt.patch_path,
1600	});
1601	}
1602
1603	if let Some(value) = updates.get("artifacts")
1604	&& let Some(items) = value.as_array()
1605	{
1606	for item in items {
1607	let artifact: TaskArtifactRef = serde_json::from_value(item.clone())
1608	.context("Failed to parse artifact task update")?;
1609	task.timeline.push(TaskTimelineEntry {
1610	timestamp: now,
1611	kind: "artifact".to_string(),
1612	summary: format!("{}: {}", artifact.label, artifact.summary),
1613	detail_path: Some(artifact.path.clone()),
1614	});
1615	task.artifacts.push(artifact);
1616	}
1617	}
1618
1619	if let Some(value) = updates.get("github_event") {
1620	let event: TaskGithubEvent = serde_json::from_value(value.clone())
1621	.context("Failed to parse GitHub task update")?;
1622	task.timeline.push(TaskTimelineEntry {
1623	timestamp: now,
1624	kind: "github".to_string(),
1625	summary: format!(
1626	"{} {}#{}: {}",
1627	event.action, event.target, event.number, event.summary
1628	),
1629	detail_path: None,
1630	});
1631	task.github_events.push(event);
1632	}
1633
1634	Ok(())
1635	}
1636
1637	fn persist_all_locked(&self, state: &ManagerState) -> Result<()> {
1638	self.persist_queue_locked(&state.queue)?;
1639	for task in state.tasks.values() {
1640	self.persist_task_locked(task)?;
1641	}
1642	Ok(())
1643	}
1644
1645	fn persist_queue_locked(&self, queue: &VecDeque<String>) -> Result<()> {
1646	write_json_atomic(
1647	&self.queue_path,
1648	&QueueFile {
1649	queue: queue.iter().cloned().collect(),
1650	},
1651	)
1652	}
1653
1654	fn persist_task_locked(&self, task: &TaskRecord) -> Result<()> {
1655	let path = self.tasks_dir.join(format!("{}.json", task.id));
1656	write_json_atomic(&path, task)
1657	}
1658	}
1659
1660	fn normalize_hunt_verdict(raw: &str) -> Result<&'static str> {
1661	match raw.trim() {
1662	"hunting" => Ok("hunting"),
1663	"hunted" => Ok("hunted"),
1664	"wounded" => Ok("wounded"),
1665	"escaped" => Ok("escaped"),
1666	other => bail!(
1667	"unsupported hunt_verdict task update '{other}'. Expected one of: hunting, hunted, wounded, escaped"
1668	),
1669	}
1670	}
1671
1672	/// Outcome of loading the persisted task store at boot: the reconciled task
1673	/// map + queue, plus the ids whose status was flipped running->failed by
1674	/// crash recovery (the only records boot needs to re-persist).
1675	struct LoadedTaskState {
1676	tasks: HashMap<String, TaskRecord>,
1677	queue: VecDeque<String>,
1678	recovered: Vec<String>,
1679	}
1680
1681	fn load_state(tasks_dir: &Path, queue_path: &Path) -> Result<LoadedTaskState> {
1682	let mut tasks = HashMap::new();
1683	let mut recovered = Vec::new();
1684	if tasks_dir.exists() {
1685	for entry in fs::read_dir(tasks_dir)
1686	.with_context(\|\| format!("Failed to read tasks dir {}", tasks_dir.display()))?
1687	{
1688	let entry = entry?;
1689	let path = entry.path();
1690	if path.extension().is_none_or(\|ext\| ext != "json") {
1691	continue;
1692	}
1693	let content = fs::read_to_string(&path)
1694	.with_context(\|\| format!("Failed to read task file {}", path.display()))?;
1695	let mut task: TaskRecord = serde_json::from_str(&content)
1696	.with_context(\|\| format!("Failed to parse task file {}", path.display()))?;
1697	if task.schema_version > CURRENT_TASK_SCHEMA_VERSION {
1698	bail!(
1699	"Task schema v{} is newer than supported v{}",
1700	task.schema_version,
1701	CURRENT_TASK_SCHEMA_VERSION
1702	);
1703	}
1704	if task.status == TaskStatus::Running {
1705	let now = Utc::now();
1706	let duration_ms = task.started_at.and_then(\|started\| {
1707	u64::try_from(now.signed_duration_since(started).num_milliseconds()).ok()
1708	});
1709	task.status = TaskStatus::Failed;
1710	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
1711	task.ended_at = Some(now);
1712	task.duration_ms = duration_ms;
1713	task.error = Some(
1714	"Interrupted by process restart; prior process is not attached".to_string(),
1715	);
1716	for tool in &mut task.tool_calls {
1717	if tool.status == TaskToolStatus::Running {
1718	tool.status = TaskToolStatus::Failed;
1719	tool.ended_at = Some(now);
1720	tool.duration_ms = duration_ms.or_else(\|\| {
1721	u64::try_from(
1722	now.signed_duration_since(tool.started_at)
1723	.num_milliseconds(),
1724	)
1725	.ok()
1726	});
1727	}
1728	}
1729	task.timeline.push(TaskTimelineEntry {
1730	timestamp: now,
1731	kind: "recovered".to_string(),
1732	summary: "Interrupted by process restart; prior process is not attached"
1733	.to_string(),
1734	detail_path: None,
1735	});
1736	recovered.push(task.id.clone());
1737	}
1738	tasks.insert(task.id.clone(), task);
1739	}
1740	}
1741
1742	let mut queue = if queue_path.exists() {
1743	let content = fs::read_to_string(queue_path)
1744	.with_context(\|\| format!("Failed to read queue file {}", queue_path.display()))?;
1745	let parsed: QueueFile = serde_json::from_str(&content)
1746	.with_context(\|\| format!("Failed to parse queue file {}", queue_path.display()))?;
1747	VecDeque::from(parsed.queue)
1748	} else {
1749	VecDeque::new()
1750	};
1751
1752	queue.retain(\|id\| {
1753	tasks
1754	.get(id)
1755	.is_some_and(\|task\| task.status == TaskStatus::Queued)
1756	});
1757
1758	let known = queue.iter().cloned().collect::<HashSet<_>>();
1759	let mut missing = tasks
1760	.values()
1761	.filter(\|task\| task.status == TaskStatus::Queued && !known.contains(&task.id))
1762	.map(\|task\| task.id.clone())
1763	.collect::<Vec<_>>();
1764	missing.sort();
1765	for id in missing {
1766	queue.push_back(id);
1767	}
1768
1769	Ok(LoadedTaskState {
1770	tasks,
1771	queue,
1772	recovered,
1773	})
1774	}
1775
1776	fn resolve_task_id(tasks: &HashMap<String, TaskRecord>, id_or_prefix: &str) -> Result<String> {
1777	if tasks.contains_key(id_or_prefix) {
1778	return Ok(id_or_prefix.to_string());
1779	}
1780	let matches = tasks
1781	.keys()
1782	.filter(\|id\| id.starts_with(id_or_prefix))
1783	.cloned()
1784	.collect::<Vec<_>>();
1785	match matches.len() {
1786	0 => bail!("Task not found: {id_or_prefix}"),
1787	1 => Ok(matches[0].clone()),
1788	_ => bail!(
1789	"Ambiguous task prefix '{}': matches {} tasks",
1790	id_or_prefix,
1791	matches.len()
1792	),
1793	}
1794	}
1795
1796	fn summarize_json(value: &Value) -> Option<String> {
1797	let text = serde_json::to_string(value).ok()?;
1798	Some(summarize_text(&text, TIMELINE_SUMMARY_LIMIT))
1799	}
1800
1801	fn summarize_text(text: &str, limit: usize) -> String {
1802	let take = limit.saturating_sub(3);
1803	let mut count = 0;
1804	let mut out = String::new();
1805	for ch in text.chars() {
1806	if count >= take {
1807	out.push_str("...");
1808	return out;
1809	}
1810	if ch.is_control() && ch != '\n' && ch != '\t' {
1811	continue;
1812	}
1813	out.push(ch);
1814	count += 1;
1815	}
1816	out
1817	}
1818
1819	fn ensure_safe_storage_id(kind: &str, value: &str) -> Result<()> {
1820	let mut components = Path::new(value).components();
1821	let Some(component) = components.next() else {
1822	bail!("{kind} must not be empty");
1823	};
1824	if components.next().is_some() \|\| !matches!(component, std::path::Component::Normal(_)) {
1825	bail!("{kind} must be a single path component");
1826	}
1827	Ok(())
1828	}
1829
1830	fn sanitize_filename(input: &str) -> String {
1831	let mut out = String::new();
1832	for ch in input.chars() {
1833	if ch.is_ascii_alphanumeric() \|\| ch == '_' \|\| ch == '-' {
1834	out.push(ch);
1835	} else {
1836	out.push('_');
1837	}
1838	}
1839	if out.is_empty() {
1840	"artifact".to_string()
1841	} else {
1842	out
1843	}
1844	}
1845
1846	fn duration_ms(start: DateTime<Utc>, end: DateTime<Utc>) -> u64 {
1847	let millis = (end - start).num_milliseconds();
1848	if millis.is_negative() {
1849	0
1850	} else {
1851	u64::try_from(millis).unwrap_or(u64::MAX)
1852	}
1853	}
1854
1855	fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> {
1856	if let Some(parent) = path.parent() {
1857	fs::create_dir_all(parent)
1858	.with_context(\|\| format!("Failed to create directory {}", parent.display()))?;
1859	}
1860	let payload = serde_json::to_string_pretty(value)?;
1861	crate::utils::write_atomic(path, payload.as_bytes())
1862	.with_context(\|\| format!("Failed to write {}", path.display()))
1863	}
1864
1865	fn default_auto_approve() -> bool {
1866	true
1867	}
1868
1869	/// Default task manager data location (`~/.codewhale/tasks`, or legacy
1870	/// `~/.deepseek/tasks` when only the legacy directory exists).
1871	#[must_use]
1872	pub fn default_tasks_dir() -> PathBuf {
1873	for var in ["CODEWHALE_TASKS_DIR", "DEEPSEEK_TASKS_DIR"] {
1874	if let Ok(path) = std::env::var(var)
1875	&& !path.trim().is_empty()
1876	{
1877	return PathBuf::from(path);
1878	}
1879	}
1880	if let Some(home) = codewhale_paths::codewhale_home_override().ok().flatten() {
1881	return home.join("tasks");
1882	}
1883	codewhale_paths::user_home()
1884	.map(\|home\| default_tasks_dir_for_home(&home))
1885	.unwrap_or_else(\|\| PathBuf::from(".codewhale").join("tasks"))
1886	}
1887
1888	fn default_tasks_dir_for_home(home: &Path) -> PathBuf {
1889	let primary = home.join(".codewhale").join("tasks");
1890	if primary.is_dir() {
1891	return primary;
1892	}
1893	let legacy = home.join(".deepseek").join("tasks");
1894	if legacy.is_dir() {
1895	return legacy;
1896	}
1897	primary
1898	}
1899
1900	/// Wait for a task to reach a terminal status (tests and API helpers).
1901	#[cfg(test)]
1902	pub async fn wait_for_terminal_state(
1903	manager: &TaskManager,
1904	task_id: &str,
1905	timeout: StdDuration,
1906	) -> Result<TaskRecord> {
1907	let deadline = std::time::Instant::now() + timeout;
1908	loop {
1909	let task = manager.get_task(task_id).await?;
1910	if task.status.is_terminal() {
1911	return Ok(task);
1912	}
1913	if std::time::Instant::now() >= deadline {
1914	bail!("Timed out waiting for task {task_id}");
1915	}
1916	sleep(StdDuration::from_millis(50)).await;
1917	}
1918	}
1919
1920	#[cfg(test)]
1921	mod tests {
1922	use super::*;
1923	use crate::test_support::{EnvVarGuard, lock_test_env};
1924	use std::fs;
1925	use tokio::time::Duration;
1926
1927	struct MockExecutor;
1928
1929	#[async_trait]
1930	impl TaskExecutor for MockExecutor {
1931	async fn execute(
1932	&self,
1933	task: ExecutionTask,
1934	events: mpsc::UnboundedSender<TaskExecutionEvent>,
1935	cancel: CancellationToken,
1936	) -> TaskExecutionResult {
1937	let _ = events.send(TaskExecutionEvent::Status {
1938	message: format!("running {}", task.id),
1939	});
1940	let _ = events.send(TaskExecutionEvent::ThreadLinked {
1941	thread_id: "thr_test".to_string(),
1942	turn_id: "turn_test".to_string(),
1943	});
1944	let _ = events.send(TaskExecutionEvent::ToolStarted {
1945	id: "tool_1".to_string(),
1946	name: "read_file".to_string(),
1947	input: serde_json::json!({ "path": "README.md" }),
1948	});
1949	sleep(Duration::from_millis(50)).await;
1950	if cancel.is_cancelled() {
1951	return TaskExecutionResult {
1952	status: TaskStatus::Canceled,
1953	result_text: None,
1954	error: None,
1955	};
1956	}
1957	let _ = events.send(TaskExecutionEvent::ToolCompleted {
1958	id: "tool_1".to_string(),
1959	name: "read_file".to_string(),
1960	success: true,
1961	output: "read ok".to_string(),
1962	metadata: Some(serde_json::json!({
1963	"duration_ms": 10,
1964	"task_updates": {
1965	"checklist": {
1966	"items": [
1967	{ "id": 1, "content": "read fixture", "status": "in_progress" }
1968	],
1969	"completion_pct": 0,
1970	"in_progress_id": 1,
1971	"updated_at": null
1972	}
1973	}
1974	})),
1975	});
1976	TaskExecutionResult {
1977	status: TaskStatus::Completed,
1978	result_text: Some("done".to_string()),
1979	error: None,
1980	}
1981	}
1982	}
1983
1984	fn test_config(root: PathBuf) -> TaskManagerConfig {
1985	TaskManagerConfig {
1986	data_dir: root,
1987	worker_count: 1,
1988	default_workspace: PathBuf::from("."),
1989	default_model: "deepseek-v4-flash".to_string(),
1990	default_mode: "agent".to_string(),
1991	allow_shell: false,
1992	trust_mode: false,
1993	}
1994	}
1995
1996	#[tokio::test]
1997	async fn persists_and_recovers_task_records() -> Result<()> {
1998	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
1999	let manager =
2000	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2001	.await?;
2002
2003	let task = manager
2004	.add_task(NewTaskRequest {
2005	owner_session_id: Some("session-persist".to_string()),
2006	..NewTaskRequest::from_prompt("test persistence")
2007	})
2008	.await?;
2009	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
2010	assert_eq!(finished.status, TaskStatus::Completed);
2011	assert_eq!(finished.thread_id.as_deref(), Some("thr_test"));
2012	assert_eq!(finished.turn_id.as_deref(), Some("turn_test"));
2013	assert_eq!(finished.checklist.items.len(), 1);
2014	assert_eq!(finished.checklist.in_progress_id, Some(1));
2015	assert!(
2016	finished.lifecycle_seq >= 3,
2017	"queued, running, and terminal owner transitions must advance the sequence"
2018	);
2019
2020	drop(manager);
2021
2022	let recovered =
2023	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2024	.await?;
2025	let loaded = recovered.get_task(&task.id).await?;
2026	assert_eq!(loaded.status, TaskStatus::Completed);
2027	assert_eq!(
2028	loaded.owner_session_id.as_deref(),
2029	Some("session-persist"),
2030	"session ownership should survive persistence and restart"
2031	);
2032	assert!(!loaded.timeline.is_empty());
2033	assert_eq!(loaded.checklist.items[0].content, "read fixture");
2034	Ok(())
2035	}
2036
2037	#[tokio::test]
2038	async fn preallocated_task_ids_are_validated_and_collision_safe() -> Result<()> {
2039	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2040	let manager =
2041	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
2042	let request = NewTaskRequest::from_prompt("preallocated owner identity");
2043
2044	let invalid = manager
2045	.add_task_with_id(request.clone(), "task_short".to_string())
2046	.await
2047	.expect_err("invalid preallocated id");
2048	assert!(invalid.to_string().contains("task_<16hex>"), "{invalid:#}");
2049
2050	let id = "task_0123456789abcdef".to_string();
2051	let created = manager
2052	.add_task_with_id(request.clone(), id.clone())
2053	.await?;
2054	assert_eq!(created.id, id);
2055	assert_eq!(
2056	created.schema_version, 2,
2057	"the additive lifecycle field must remain rollback-readable"
2058	);
2059	assert_eq!(created.lifecycle_seq, 1);
2060	let collision = manager
2061	.add_task_with_id(request, id)
2062	.await
2063	.expect_err("task id collision");
2064	assert!(
2065	collision.to_string().contains("already exists"),
2066	"{collision:#}"
2067	);
2068	assert_eq!(manager.list_tasks(None).await.len(), 1);
2069	Ok(())
2070	}
2071
2072	#[tokio::test]
2073	async fn failed_queue_write_leaves_no_replayable_task_record() -> Result<()> {
2074	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2075	let manager =
2076	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2077	.await?;
2078	std::fs::remove_file(root.join("queue.json"))?;
2079	std::fs::create_dir(root.join("queue.json"))?;
2080
2081	let id = "task_fedcba9876543210".to_string();
2082	let error = manager
2083	.add_task_with_id(
2084	NewTaskRequest::from_prompt("must not resurrect"),
2085	id.clone(),
2086	)
2087	.await
2088	.expect_err("queue path directory must reject the atomic queue write");
2089	assert!(error.to_string().contains("queue.json"), "{error:#}");
2090	assert!(manager.list_tasks(None).await.is_empty());
2091	assert!(!root.join("tasks").join(format!("{id}.json")).exists());
2092	assert!(
2093	!root
2094	.join("tasks")
2095	.join(format!(".{id}.json.pending"))
2096	.exists(),
2097	"a failed queue write may leave no replayable or staged task record"
2098	);
2099	Ok(())
2100	}
2101
2102	#[tokio::test]
2103	async fn list_tasks_scopes_results_to_workspace_before_limit() -> Result<()> {
2104	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2105	let manager =
2106	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
2107
2108	manager
2109	.add_task(NewTaskRequest {
2110	prompt: "task in workspace a".to_string(),
2111	workspace: Some(PathBuf::from("/tmp/workspace-a")),
2112	..NewTaskRequest::from_prompt("task in workspace a")
2113	})
2114	.await?;
2115	manager
2116	.add_task(NewTaskRequest {
2117	prompt: "task in workspace b".to_string(),
2118	workspace: Some(PathBuf::from("/tmp/workspace-b")),
2119	..NewTaskRequest::from_prompt("task in workspace b")
2120	})
2121	.await?;
2122
2123	let scoped = manager
2124	.list_tasks_scoped(Some(1), Some(Path::new("/tmp/workspace-a")))
2125	.await;
2126	assert_eq!(scoped.len(), 1);
2127	assert_eq!(scoped[0].workspace, PathBuf::from("/tmp/workspace-a"));
2128	Ok(())
2129	}
2130
2131	#[tokio::test]
2132	async fn boot_does_not_rewrite_non_recovered_task_files() -> Result<()> {
2133	// #3757 boot-persist narrowing: TaskManager::start must persist only
2134	// the reconciled queue and the running->failed recoveries — a
2135	// completed task's file must be byte-identical across a restart.
2136	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2137	let manager =
2138	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2139	.await?;
2140	let task = manager
2141	.add_task(NewTaskRequest::from_prompt("finish then persist"))
2142	.await?;
2143	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
2144	assert_eq!(finished.status, TaskStatus::Completed);
2145	drop(manager);
2146
2147	let task_file = root.join("tasks").join(format!("{}.json", task.id));
2148	let before = fs::read(&task_file)?;
2149
2150	let recovered =
2151	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2152	.await?;
2153	// Give start() a beat to run its (narrowed) boot persist.
2154	sleep(Duration::from_millis(50)).await;
2155	drop(recovered);
2156
2157	let after = fs::read(&task_file)?;
2158	assert_eq!(
2159	before, after,
2160	"a completed task file must not be rewritten on boot"
2161	);
2162	Ok(())
2163	}
2164
2165	#[test]
2166	fn running_tasks_are_not_requeued_after_restart() -> Result<()> {
2167	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2168	let tasks_dir = root.join("tasks");
2169	fs::create_dir_all(&tasks_dir)?;
2170	let queue_path = root.join("queue.json");
2171	let task_id = "task_stale_running".to_string();
2172	let started_at = Utc::now() - chrono::Duration::seconds(30);
2173	let task = TaskRecord {
2174	schema_version: CURRENT_TASK_SCHEMA_VERSION,
2175	id: task_id.clone(),
2176	prompt: "long-running shell work".to_string(),
2177	model: "deepseek-v4-flash".to_string(),
2178	workspace: PathBuf::from("."),
2179	mode: "agent".to_string(),
2180	allow_shell: true,
2181	trust_mode: false,
2182	auto_approve: false,
2183	status: TaskStatus::Running,
2184	created_at: started_at,
2185	started_at: Some(started_at),
2186	ended_at: None,
2187	duration_ms: None,
2188	hunt_verdict: None,
2189	result_summary: None,
2190	result_detail_path: None,
2191	error: None,
2192	thread_id: Some("thr_stale".to_string()),
2193	turn_id: Some("turn_stale".to_string()),
2194	owner_session_id: Some("session-old".to_string()),
2195	runtime_event_count: 0,
2196	lifecycle_seq: 2,
2197	checklist: TaskChecklistState::default(),
2198	gates: Vec::new(),
2199	attempts: Vec::new(),
2200	artifacts: Vec::new(),
2201	github_events: Vec::new(),
2202	tool_calls: vec![TaskToolCallSummary {
2203	id: "tool_shell".to_string(),
2204	name: "task_shell_start".to_string(),
2205	status: TaskToolStatus::Running,
2206	started_at,
2207	ended_at: None,
2208	duration_ms: None,
2209	input_summary: Some("shell: sleep 999".to_string()),
2210	output_summary: None,
2211	detail_path: None,
2212	patch_ref: None,
2213	}],
2214	timeline: vec![TaskTimelineEntry {
2215	timestamp: started_at,
2216	kind: "running".to_string(),
2217	summary: "Task started".to_string(),
2218	detail_path: None,
2219	}],
2220	};
2221	fs::write(
2222	tasks_dir.join(format!("{task_id}.json")),
2223	serde_json::to_string_pretty(&task)?,
2224	)?;
2225	fs::write(
2226	&queue_path,
2227	serde_json::to_string_pretty(&QueueFile {
2228	queue: vec![task_id.clone()],
2229	})?,
2230	)?;
2231
2232	let loaded = load_state(&tasks_dir, &queue_path)?;
2233	let queue = loaded.queue;
2234	let recovered = loaded.tasks.get(&task_id).expect("task loaded");
2235
2236	assert!(queue.is_empty(), "stale running task must not be requeued");
2237	assert_eq!(recovered.status, TaskStatus::Failed);
2238	assert!(
2239	recovered
2240	.error
2241	.as_deref()
2242	.is_some_and(\|err\| err.contains("prior process is not attached")),
2243	"recovered task should explain stale process ownership: {recovered:?}"
2244	);
2245	assert!(recovered.ended_at.is_some());
2246	assert!(recovered.duration_ms.is_some());
2247	assert_eq!(recovered.tool_calls[0].status, TaskToolStatus::Failed);
2248	assert!(recovered.tool_calls[0].ended_at.is_some());
2249	assert!(
2250	recovered
2251	.timeline
2252	.iter()
2253	.any(\|entry\| entry.kind == "recovered"
2254	&& entry.summary.contains("prior process is not attached")),
2255	"recovery timeline should explain why the task is terminal: {:?}",
2256	recovered.timeline
2257	);
2258	Ok(())
2259	}
2260
2261	#[tokio::test]
2262	async fn default_workspace_updates_for_future_tasks() -> Result<()> {
2263	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2264	let new_workspace =
2265	std::env::temp_dir().join(format!("deepseek-workspace-{}", Uuid::new_v4()));
2266	let manager =
2267	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
2268
2269	manager.set_default_workspace(new_workspace.clone()).await;
2270	let task = manager
2271	.add_task(NewTaskRequest::from_prompt("test workspace default"))
2272	.await?;
2273
2274	assert_eq!(manager.default_workspace().await, new_workspace);
2275	assert_eq!(task.workspace, new_workspace);
2276	Ok(())
2277	}
2278
2279	#[tokio::test]
2280	async fn record_tool_metadata_updates_explicit_task() -> Result<()> {
2281	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2282	let manager =
2283	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
2284
2285	let task = manager
2286	.add_task(NewTaskRequest::from_prompt("test metadata"))
2287	.await?;
2288	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
2289	let updated = manager
2290	.record_tool_metadata(
2291	&finished.id,
2292	&serde_json::json!({
2293	"task_updates": {
2294	"gate": {
2295	"id": "gate_test",
2296	"gate": "test",
2297	"command": "cargo test -p codewhale-tui --lib",
2298	"cwd": ".",
2299	"exit_code": 0,
2300	"status": "passed",
2301	"classification": "passed",
2302	"duration_ms": 1,
2303	"summary": "ok",
2304	"log_path": null,
2305	"recorded_at": Utc::now()
2306	}
2307	}
2308	}),
2309	)
2310	.await?;
2311
2312	assert_eq!(updated.gates.len(), 1);
2313	assert_eq!(updated.gates[0].classification, "passed");
2314	Ok(())
2315	}
2316
2317	#[tokio::test]
2318	async fn record_tool_metadata_updates_hunt_verdict_summary() -> Result<()> {
2319	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2320	let manager =
2321	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
2322
2323	let task = manager
2324	.add_task(NewTaskRequest::from_prompt("test verdict metadata"))
2325	.await?;
2326	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
2327	let updated = manager
2328	.record_tool_metadata(
2329	&finished.id,
2330	&serde_json::json!({
2331	"task_updates": {
2332	"hunt_verdict": "wounded"
2333	}
2334	}),
2335	)
2336	.await?;
2337
2338	assert_eq!(updated.hunt_verdict.as_deref(), Some("wounded"));
2339	let summaries = manager.list_tasks(Some(10)).await;
2340	let summary = summaries
2341	.iter()
2342	.find(\|summary\| summary.id == updated.id)
2343	.expect("updated task summary");
2344	assert_eq!(summary.hunt_verdict.as_deref(), Some("wounded"));
2345	Ok(())
2346	}
2347
2348	#[tokio::test]
2349	async fn write_task_artifact_rejects_traversal_task_id() -> Result<()> {
2350	let temp = tempfile::tempdir()?;
2351	let root = temp.path().join("tasks-root");
2352	let escaped = temp.path().join("escape");
2353	let manager =
2354	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2355	.await?;
2356
2357	let err = manager
2358	.write_task_artifact("../escape", "result", "artifact body")
2359	.expect_err("traversal task ids must be rejected");
2360
2361	assert!(err.to_string().contains("single path component"));
2362	assert!(!escaped.exists(), "artifact write escaped the task root");
2363	Ok(())
2364	}
2365
2366	#[tokio::test]
2367	async fn cancel_running_task_marks_canceled() -> Result<()> {
2368	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2369	let manager =
2370	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
2371
2372	let task = manager
2373	.add_task(NewTaskRequest::from_prompt("test cancellation"))
2374	.await?;
2375
2376	sleep(Duration::from_millis(10)).await;
2377	let cancellation = manager.cancel_task(&task.id).await?;
2378	assert_eq!(cancellation.disposition, TaskCancelDisposition::Requested);
2379	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
2380	assert_eq!(finished.status, TaskStatus::Canceled);
2381	Ok(())
2382	}
2383
2384	#[tokio::test]
2385	async fn cancel_finished_task_returns_atomic_already_finished_outcome() -> Result<()> {
2386	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2387	let manager =
2388	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
2389	let task = manager
2390	.add_task(NewTaskRequest::from_prompt("finish before cancellation"))
2391	.await?;
2392	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
2393	assert_eq!(finished.status, TaskStatus::Completed);
2394
2395	let cancellation = manager.cancel_task(&task.id).await?;
2396
2397	assert_eq!(
2398	cancellation.disposition,
2399	TaskCancelDisposition::AlreadyFinished
2400	);
2401	assert_eq!(cancellation.task.status, TaskStatus::Completed);
2402	Ok(())
2403	}
2404
2405	// GHSA-72w5-pf8h-xfp4 — regression: omitted optional fields must not
2406	// silently elevate the spawned task's privileges.
2407	#[tokio::test]
2408	async fn add_task_without_optional_fields_does_not_grant_shell_or_auto_approve() -> Result<()> {
2409	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2410	let manager =
2411	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2412	.await?;
2413
2414	let req = NewTaskRequest {
2415	prompt: "fix TODOs and write a README".to_string(),
2416	model: None,
2417	workspace: None,
2418	mode: None,
2419	allow_shell: None,
2420	trust_mode: None,
2421	auto_approve: None,
2422	owner_session_id: None,
2423	};
2424	let task = manager.add_task(req).await?;
2425
2426	assert!(
2427	!task.allow_shell,
2428	"model-omitted allow_shell must default to false (no silent shell grant)"
2429	);
2430	assert!(
2431	!task.auto_approve,
2432	"model-omitted auto_approve must default to false (no silent auto-approval)"
2433	);
2434	assert!(
2435	!task.trust_mode,
2436	"model-omitted trust_mode must default to false"
2437	);
2438	Ok(())
2439	}
2440
2441	#[tokio::test]
2442	async fn rejects_newer_task_schema_on_recovery() -> Result<()> {
2443	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
2444	let manager =
2445	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
2446	.await?;
2447
2448	let task = manager
2449	.add_task(NewTaskRequest::from_prompt("test schema gate"))
2450	.await?;
2451	let _ = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
2452	drop(manager);
2453
2454	let task_path = root.join("tasks").join(format!("{}.json", task.id));
2455	let mut value: serde_json::Value = serde_json::from_str(&fs::read_to_string(&task_path)?)?;
2456	value["schema_version"] = serde_json::json!(999);
2457	fs::write(&task_path, serde_json::to_string_pretty(&value)?)?;
2458
2459	match TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await {
2460	Ok(_) => panic!("manager should reject newer task schema"),
2461	Err(err) => assert!(err.to_string().contains("newer than supported")),
2462	}
2463	Ok(())
2464	}
2465
2466	#[test]
2467	fn default_tasks_dir_falls_back_to_legacy_deepseek_tasks() {
2468	let temp_home = tempfile::tempdir().unwrap();
2469	let home = temp_home.path();
2470	let legacy_tasks = home.join(".deepseek").join("tasks");
2471	std::fs::create_dir_all(&legacy_tasks).unwrap();
2472
2473	assert_eq!(default_tasks_dir_for_home(home), legacy_tasks);
2474	}
2475
2476	#[test]
2477	fn default_tasks_dir_prefers_existing_codewhale_tasks() {
2478	let temp_home = tempfile::tempdir().unwrap();
2479	let home = temp_home.path();
2480	let primary_tasks = home.join(".codewhale").join("tasks");
2481	let legacy_tasks = home.join(".deepseek").join("tasks");
2482	std::fs::create_dir_all(&primary_tasks).unwrap();
2483	std::fs::create_dir_all(&legacy_tasks).unwrap();
2484
2485	assert_eq!(default_tasks_dir_for_home(home), primary_tasks);
2486	}
2487
2488	#[test]
2489	fn default_tasks_dir_falls_back_to_legacy_when_primary_is_file() {
2490	let temp_home = tempfile::tempdir().unwrap();
2491	let home = temp_home.path();
2492	let primary_tasks = home.join(".codewhale").join("tasks");
2493	let legacy_tasks = home.join(".deepseek").join("tasks");
2494	std::fs::create_dir_all(primary_tasks.parent().unwrap()).unwrap();
2495	std::fs::write(&primary_tasks, "not a directory").unwrap();
2496	std::fs::create_dir_all(&legacy_tasks).unwrap();
2497
2498	assert_eq!(default_tasks_dir_for_home(home), legacy_tasks);
2499	}
2500
2501	#[test]
2502	fn default_tasks_dir_ignores_legacy_file_for_new_installs() {
2503	let temp_home = tempfile::tempdir().unwrap();
2504	let home = temp_home.path();
2505	let primary_tasks = home.join(".codewhale").join("tasks");
2506	let legacy_tasks = home.join(".deepseek").join("tasks");
2507	std::fs::create_dir_all(legacy_tasks.parent().unwrap()).unwrap();
2508	std::fs::write(&legacy_tasks, "not a directory").unwrap();
2509
2510	assert_eq!(default_tasks_dir_for_home(home), primary_tasks);
2511	}
2512
2513	#[test]
2514	fn default_tasks_dir_uses_codewhale_tasks_for_new_installs() {
2515	let temp_home = tempfile::tempdir().unwrap();
2516	let home = temp_home.path();
2517
2518	assert_eq!(
2519	default_tasks_dir_for_home(home),
2520	home.join(".codewhale").join("tasks")
2521	);
2522	}
2523
2524	#[test]
2525	fn task_and_runtime_roots_honor_explicit_codewhale_home() {
2526	let _lock = lock_test_env();
2527	let temp_root = tempfile::tempdir().unwrap();
2528	let ambient_home = temp_root.path().join("ambient-home");
2529	let explicit_home = temp_root.path().join("explicit-home");
2530	std::fs::create_dir_all(ambient_home.join(".deepseek").join("tasks")).unwrap();
2531	let _home = EnvVarGuard::set("HOME", &ambient_home);
2532	let _userprofile = EnvVarGuard::set("USERPROFILE", &ambient_home);
2533	let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", &explicit_home);
2534	let _tasks_override = EnvVarGuard::remove("CODEWHALE_TASKS_DIR");
2535	let _legacy_tasks_override = EnvVarGuard::remove("DEEPSEEK_TASKS_DIR");
2536	let _runtime_override = EnvVarGuard::remove("CODEWHALE_RUNTIME_DIR");
2537	let _legacy_runtime_override = EnvVarGuard::remove("DEEPSEEK_RUNTIME_DIR");
2538
2539	let task_root = default_tasks_dir();
2540	let task_manager =
2541	TaskManagerConfig::from_runtime(&Config::default(), PathBuf::from("."), None, None);
2542	let runtime = RuntimeThreadManagerConfig::from_task_data_dir(task_manager.data_dir.clone());
2543
2544	assert_eq!(task_root, explicit_home.join("tasks"));
2545	assert_eq!(task_manager.data_dir, task_root);
2546	assert_eq!(runtime.task_data_dir, task_root);
2547	assert_eq!(
2548	runtime.data_dir,
2549	explicit_home.join("tasks").join("runtime")
2550	);
2551	}
2552
2553	#[test]
2554	fn whitespace_codewhale_home_keeps_ambient_legacy_task_and_runtime_fallbacks() {
2555	let _lock = lock_test_env();
2556	let temp_root = tempfile::tempdir().unwrap();
2557	let ambient_home = temp_root.path().join("ambient-home");
2558	let legacy_tasks = ambient_home.join(".deepseek").join("tasks");
2559	std::fs::create_dir_all(&legacy_tasks).unwrap();
2560	let _home = EnvVarGuard::set("HOME", &ambient_home);
2561	let _userprofile = EnvVarGuard::set("USERPROFILE", &ambient_home);
2562	let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", " \t ");
2563	let _tasks_override = EnvVarGuard::remove("CODEWHALE_TASKS_DIR");
2564	let _legacy_tasks_override = EnvVarGuard::remove("DEEPSEEK_TASKS_DIR");
2565	let _runtime_override = EnvVarGuard::remove("CODEWHALE_RUNTIME_DIR");
2566	let _legacy_runtime_override = EnvVarGuard::remove("DEEPSEEK_RUNTIME_DIR");
2567
2568	let task_root = default_tasks_dir();
2569	let task_manager =
2570	TaskManagerConfig::from_runtime(&Config::default(), PathBuf::from("."), None, None);
2571	let runtime = RuntimeThreadManagerConfig::from_task_data_dir(task_manager.data_dir.clone());
2572
2573	assert_eq!(task_root, legacy_tasks);
2574	assert_eq!(task_manager.data_dir, task_root);
2575	assert_eq!(runtime.task_data_dir, task_root);
2576	assert_eq!(runtime.data_dir, task_root.join("runtime"));
2577	}
2578
2579	#[cfg(unix)]
2580	#[test]
2581	fn non_unicode_codewhale_home_is_preserved_by_task_and_runtime_roots() {
2582	use std::os::unix::ffi::OsStringExt;
2583
2584	let _lock = lock_test_env();
2585	let temp_root = tempfile::tempdir().unwrap();
2586	let explicit_home = temp_root.path().join(std::ffi::OsString::from_vec(
2587	b"codewhale-\xff-home".to_vec(),
2588	));
2589	let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", &explicit_home);
2590	let _tasks_override = EnvVarGuard::remove("CODEWHALE_TASKS_DIR");
2591	let _legacy_tasks_override = EnvVarGuard::remove("DEEPSEEK_TASKS_DIR");
2592	let _runtime_override = EnvVarGuard::remove("CODEWHALE_RUNTIME_DIR");
2593	let _legacy_runtime_override = EnvVarGuard::remove("DEEPSEEK_RUNTIME_DIR");
2594
2595	let task_root = default_tasks_dir();
2596	let task_manager =
2597	TaskManagerConfig::from_runtime(&Config::default(), PathBuf::from("."), None, None);
2598	let runtime = RuntimeThreadManagerConfig::from_task_data_dir(task_manager.data_dir.clone());
2599
2600	assert_eq!(task_root, explicit_home.join("tasks"));
2601	assert_eq!(task_manager.data_dir, task_root);
2602	assert_eq!(runtime.task_data_dir, task_root);
2603	assert_eq!(
2604	runtime.data_dir,
2605	explicit_home.join("tasks").join("runtime")
2606	);
2607	}
2608	}
2609

2609 lines RUST