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 uses the shared runtime provider route and links every task to
5	//! runtime 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	#[cfg(test)]
12	use std::time::Duration as StdDuration;
13	use std::time::{Duration, Instant};
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;
26	use crate::runtime_threads::{
27	CreateThreadRequest, RUNTIME_STORE_FAILURE_EVENT, RuntimeEventRecord, RuntimeProcessOwnerLock,
28	RuntimeThreadManager, RuntimeThreadManagerConfig, RuntimeTurnStatus,
29	SharedRuntimeThreadManager, StartTurnRequest,
30	};
31	use crate::utils::spawn_supervised;
32
33	const DEFAULT_WORKERS: usize = 2;
34	const MAX_WORKERS: usize = 8;
35	const TIMELINE_SUMMARY_LIMIT: usize = 240;
36	const TIMELINE_ENTRY_LIMIT: usize = 256;
37	const TIMELINE_HEAD_KEEP: usize = 8;
38	const ARTIFACT_THRESHOLD: usize = 1200;
39	const TASK_EVENT_CHANNEL_CAPACITY: usize = 256;
40	const EVENT_CURSOR_BATCH: usize = 256;
41	const EVENT_CATCHUP_POLL: Duration = Duration::from_millis(200);
42	// v4 binds execution to a trusted Runtime scope. Older executors must not
43	// ignore its eligibility or generation fence.
44	const CURRENT_TASK_SCHEMA_VERSION: u32 = 4;
45	const STORE_REFRESH_INTERVAL: Duration = Duration::from_millis(200);
46
47	const fn default_task_schema_version() -> u32 {
48	CURRENT_TASK_SCHEMA_VERSION
49	}
50
51	/// Durable task status.
52	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
53	#[serde(rename_all = "snake_case")]
54	pub enum TaskStatus {
55	Queued,
56	Running,
57	Completed,
58	Failed,
59	Canceled,
60	}
61
62	/// What the manager actually did while handling a cancellation request.
63	///
64	/// This is returned from the same state-lock transaction as the task record,
65	/// so callers never have to infer an outcome from a stale pre-cancel read.
66	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
67	pub enum TaskCancelDisposition {
68	Forced,
69	Requested,
70	AlreadyFinished,
71	}
72
73	#[derive(Debug, Clone)]
74	pub struct TaskCancellation {
75	pub task: TaskRecord,
76	pub disposition: TaskCancelDisposition,
77	}
78
79	impl TaskStatus {
80	#[cfg(test)]
81	#[must_use]
82	pub fn is_terminal(self) -> bool {
83	matches!(self, Self::Completed \| Self::Failed \| Self::Canceled)
84	}
85	}
86
87	/// Why a durable task left the running state. Stored on the task record so
88	/// receipts and status views can show a forced timeout separately from a
89	/// cooperative cancel.
90	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91	pub enum TaskTerminalReason {
92	Completed,
93	Canceled,
94	CancelTimeout,
95	Shutdown,
96	WallTimeout,
97	IdleTimeout,
98	Failed,
99	}
100
101	impl TaskTerminalReason {
102	#[must_use]
103	pub const fn as_str(self) -> &'static str {
104	match self {
105	Self::Completed => "completed",
106	Self::Canceled => "canceled",
107	Self::CancelTimeout => "cancel_timeout",
108	Self::Shutdown => "shutdown",
109	Self::WallTimeout => "wall_timeout",
110	Self::IdleTimeout => "idle_timeout",
111	Self::Failed => "failed",
112	}
113	}
114
115	#[must_use]
116	pub const fn task_status(self) -> TaskStatus {
117	match self {
118	Self::Completed => TaskStatus::Completed,
119	Self::Canceled \| Self::CancelTimeout \| Self::Shutdown => TaskStatus::Canceled,
120	Self::WallTimeout \| Self::IdleTimeout \| Self::Failed => TaskStatus::Failed,
121	}
122	}
123
124	#[must_use]
125	pub fn receipt_message(self) -> String {
126	match self {
127	Self::Completed => "Task completed".to_string(),
128	Self::Canceled => "Task canceled".to_string(),
129	Self::CancelTimeout => {
130	"Task did not terminalize after cancellation; worker released".to_string()
131	}
132	Self::Shutdown => "Task canceled because the task manager shut down".to_string(),
133	Self::WallTimeout => {
134	"Task exceeded its wall-time deadline without completing".to_string()
135	}
136	Self::IdleTimeout => {
137	"Task made no model or tool progress before the idle deadline".to_string()
138	}
139	Self::Failed => "Task ended unexpectedly".to_string(),
140	}
141	}
142
143	fn after_grace(self) -> Self {
144	match self {
145	Self::Canceled => Self::CancelTimeout,
146	other => other,
147	}
148	}
149	}
150
151	/// Durable tool-call status within a task timeline.
152	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
153	#[serde(rename_all = "snake_case")]
154	pub enum TaskToolStatus {
155	Running,
156	Success,
157	Failed,
158	Canceled,
159	}
160
161	/// Timeline entry for a task execution.
162	#[derive(Debug, Clone, Serialize, Deserialize)]
163	pub struct TaskTimelineEntry {
164	pub timestamp: DateTime<Utc>,
165	pub kind: String,
166	pub summary: String,
167	#[serde(skip_serializing_if = "Option::is_none")]
168	pub detail_path: Option<PathBuf>,
169	}
170
171	/// Tool call summary for a task.
172	#[derive(Debug, Clone, Serialize, Deserialize)]
173	pub struct TaskToolCallSummary {
174	pub id: String,
175	pub name: String,
176	pub status: TaskToolStatus,
177	pub started_at: DateTime<Utc>,
178	pub ended_at: Option<DateTime<Utc>>,
179	pub duration_ms: Option<u64>,
180	#[serde(skip_serializing_if = "Option::is_none")]
181	pub input_summary: Option<String>,
182	#[serde(skip_serializing_if = "Option::is_none")]
183	pub output_summary: Option<String>,
184	#[serde(skip_serializing_if = "Option::is_none")]
185	pub detail_path: Option<PathBuf>,
186	#[serde(skip_serializing_if = "Option::is_none")]
187	pub patch_ref: Option<PathBuf>,
188	}
189
190	/// Checklist item stored on durable tasks. This is the durable form behind the
191	/// model-visible checklist/todo compatibility tools.
192	#[derive(Debug, Clone, Serialize, Deserialize)]
193	pub struct TaskChecklistItem {
194	pub id: u32,
195	pub content: String,
196	pub status: String,
197	}
198
199	/// Checklist state associated with a task.
200	#[derive(Debug, Clone, Serialize, Deserialize, Default)]
201	pub struct TaskChecklistState {
202	pub items: Vec<TaskChecklistItem>,
203	pub completion_pct: u8,
204	pub in_progress_id: Option<u32>,
205	pub updated_at: Option<DateTime<Utc>>,
206	}
207
208	/// Structured verification evidence attached to a task.
209	#[derive(Debug, Clone, Serialize, Deserialize)]
210	pub struct TaskGateRecord {
211	pub id: String,
212	pub gate: String,
213	pub command: String,
214	pub cwd: PathBuf,
215	pub exit_code: Option<i32>,
216	pub status: String,
217	pub classification: String,
218	pub duration_ms: u64,
219	pub summary: String,
220	#[serde(skip_serializing_if = "Option::is_none")]
221	pub log_path: Option<PathBuf>,
222	pub recorded_at: DateTime<Utc>,
223	}
224
225	/// PR-attempt metadata and artifacts attached to a task.
226	#[derive(Debug, Clone, Serialize, Deserialize)]
227	pub struct TaskAttemptRecord {
228	pub id: String,
229	pub attempt_group_id: String,
230	pub attempt_index: u32,
231	pub attempt_count: u32,
232	#[serde(skip_serializing_if = "Option::is_none")]
233	pub base_ref: Option<String>,
234	#[serde(skip_serializing_if = "Option::is_none")]
235	pub base_sha: Option<String>,
236	#[serde(skip_serializing_if = "Option::is_none")]
237	pub head_ref: Option<String>,
238	#[serde(skip_serializing_if = "Option::is_none")]
239	pub head_sha: Option<String>,
240	pub summary: String,
241	pub changed_files: Vec<String>,
242	#[serde(skip_serializing_if = "Option::is_none")]
243	pub patch_path: Option<PathBuf>,
244	pub verification: Vec<String>,
245	pub selected: bool,
246	pub recorded_at: DateTime<Utc>,
247	}
248
249	/// Durable artifact reference produced by task-aware tools.
250	#[derive(Debug, Clone, Serialize, Deserialize)]
251	pub struct TaskArtifactRef {
252	pub label: String,
253	pub path: PathBuf,
254	pub summary: String,
255	pub created_at: DateTime<Utc>,
256	}
257
258	/// GitHub write/read evidence attached to a task timeline.
259	#[derive(Debug, Clone, Serialize, Deserialize)]
260	pub struct TaskGithubEvent {
261	pub id: String,
262	pub action: String,
263	pub target: String,
264	pub number: u64,
265	pub summary: String,
266	pub url: Option<String>,
267	pub recorded_at: DateTime<Utc>,
268	}
269
270	/// Durable task record.
271	#[derive(Debug, Clone, Serialize, Deserialize)]
272	pub struct TaskRecord {
273	#[serde(default = "default_task_schema_version")]
274	pub schema_version: u32,
275	pub id: String,
276	pub prompt: String,
277	#[serde(default, skip_serializing_if = "Option::is_none")]
278	pub name: Option<String>,
279	pub model: String,
280	#[serde(default, skip_serializing_if = "Option::is_none")]
281	pub model_provider: Option<String>,
282	#[serde(default, skip_serializing_if = "Option::is_none")]
283	pub model_provider_id: Option<String>,
284	pub workspace: PathBuf,
285	pub mode: String,
286	pub allow_shell: bool,
287	pub trust_mode: bool,
288	#[serde(default = "default_auto_approve")]
289	pub auto_approve: bool,
290	pub status: TaskStatus,
291	pub created_at: DateTime<Utc>,
292	pub started_at: Option<DateTime<Utc>>,
293	pub ended_at: Option<DateTime<Utc>>,
294	pub duration_ms: Option<u64>,
295	#[serde(skip_serializing_if = "Option::is_none")]
296	pub result_summary: Option<String>,
297	#[serde(skip_serializing_if = "Option::is_none")]
298	pub result_detail_path: Option<PathBuf>,
299	#[serde(skip_serializing_if = "Option::is_none")]
300	pub error: Option<String>,
301	#[serde(default, skip_serializing_if = "Option::is_none")]
302	pub terminal_reason: Option<String>,
303	#[serde(default, skip_serializing_if = "Option::is_none")]
304	pub thread_id: Option<String>,
305	#[serde(default, skip_serializing_if = "Option::is_none")]
306	pub turn_id: Option<String>,
307	#[serde(default, skip_serializing_if = "Option::is_none")]
308	pub owner_session_id: Option<String>,
309	/// Trusted execution provenance, distinct from model-visible ownership.
310	#[serde(default, skip_serializing_if = "Option::is_none")]
311	pub execution_scope: Option<String>,
312	#[serde(default, skip_serializing_if = "Option::is_none")]
313	pub execution_generation: Option<String>,
314	/// Durable cancellation acknowledged by the actual execution owner.
315	#[serde(default)]
316	pub cancel_requested_seq: u64,
317	#[serde(default)]
318	pub runtime_event_count: usize,
319	/// Monotonic owner-lifecycle sequence used by Work Graph reconciliation.
320	/// Output/progress events do not advance this counter; only lifecycle
321	/// transitions do, so replay after restart is stable.
322	#[serde(default)]
323	pub lifecycle_seq: u64,
324	#[serde(default)]
325	pub checklist: TaskChecklistState,
326	#[serde(default)]
327	pub gates: Vec<TaskGateRecord>,
328	#[serde(default)]
329	pub attempts: Vec<TaskAttemptRecord>,
330	#[serde(default)]
331	pub artifacts: Vec<TaskArtifactRef>,
332	#[serde(default)]
333	pub github_events: Vec<TaskGithubEvent>,
334	pub tool_calls: Vec<TaskToolCallSummary>,
335	pub timeline: Vec<TaskTimelineEntry>,
336	}
337
338	/// Lightweight task view.
339	#[derive(Debug, Clone, Serialize, Deserialize)]
340	pub struct TaskSummary {
341	pub id: String,
342	pub status: TaskStatus,
343	pub prompt_summary: String,
344	#[serde(default, skip_serializing_if = "Option::is_none")]
345	pub name: Option<String>,
346	pub model: String,
347	#[serde(default, skip_serializing_if = "Option::is_none")]
348	pub model_provider: Option<String>,
349	#[serde(default, skip_serializing_if = "Option::is_none")]
350	pub model_provider_id: Option<String>,
351	pub mode: String,
352	pub workspace: PathBuf,
353	pub created_at: DateTime<Utc>,
354	pub started_at: Option<DateTime<Utc>>,
355	pub ended_at: Option<DateTime<Utc>>,
356	pub duration_ms: Option<u64>,
357	#[serde(default)]
358	pub lifecycle_seq: u64,
359	#[serde(skip_serializing_if = "Option::is_none")]
360	pub error: Option<String>,
361	#[serde(default, skip_serializing_if = "Option::is_none")]
362	pub terminal_reason: Option<String>,
363	#[serde(default, skip_serializing_if = "Option::is_none")]
364	pub thread_id: Option<String>,
365	#[serde(default, skip_serializing_if = "Option::is_none")]
366	pub turn_id: Option<String>,
367	#[serde(default, skip_serializing_if = "Option::is_none")]
368	pub owner_session_id: Option<String>,
369	pub execution_binding_known: bool,
370	}
371
372	impl From<&TaskRecord> for TaskSummary {
373	fn from(value: &TaskRecord) -> Self {
374	Self {
375	id: value.id.clone(),
376	status: value.status,
377	prompt_summary: summarize_text(&value.prompt, TIMELINE_SUMMARY_LIMIT),
378	name: value.name.clone(),
379	model: value.model.clone(),
380	model_provider: value.model_provider.clone(),
381	model_provider_id: value.model_provider_id.clone(),
382	mode: value.mode.clone(),
383	workspace: value.workspace.clone(),
384	created_at: value.created_at,
385	started_at: value.started_at,
386	ended_at: value.ended_at,
387	duration_ms: value.duration_ms,
388	lifecycle_seq: value.lifecycle_seq,
389	error: value.error.clone(),
390	terminal_reason: value.terminal_reason.clone(),
391	thread_id: value.thread_id.clone(),
392	turn_id: value.turn_id.clone(),
393	owner_session_id: value.owner_session_id.clone(),
394	execution_binding_known: value.execution_scope.is_some()
395	&& (value.status != TaskStatus::Running \|\| value.execution_generation.is_some()),
396	}
397	}
398	}
399
400	/// Count totals by status for task dashboards.
401	#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
402	pub struct TaskCounts {
403	pub queued: usize,
404	pub running: usize,
405	pub completed: usize,
406	pub failed: usize,
407	pub canceled: usize,
408	}
409
410	/// Request to enqueue a new task.
411	#[derive(Debug, Clone, Serialize, Deserialize)]
412	pub struct NewTaskRequest {
413	pub prompt: String,
414	/// Caller-given run name, stored as-given. Absent names stay absent —
415	/// titles derived from the prompt are a presentation concern.
416	#[serde(default)]
417	pub name: Option<String>,
418	pub model: Option<String>,
419	#[serde(default)]
420	pub model_provider: Option<String>,
421	#[serde(default)]
422	pub model_provider_id: Option<String>,
423	pub workspace: Option<PathBuf>,
424	pub mode: Option<String>,
425	pub allow_shell: Option<bool>,
426	pub trust_mode: Option<bool>,
427	pub auto_approve: Option<bool>,
428	pub owner_session_id: Option<String>,
429	}
430
431	impl NewTaskRequest {
432	/// Preserve values already resolved into a staged or accepted task.
433	pub(crate) fn from_task(task: &TaskRecord) -> Self {
434	Self {
435	prompt: task.prompt.clone(),
436	name: task.name.clone(),
437	model: Some(task.model.clone()),
438	model_provider: task.model_provider.clone(),
439	model_provider_id: task.model_provider_id.clone(),
440	workspace: Some(task.workspace.clone()),
441	mode: Some(task.mode.clone()),
442	allow_shell: Some(task.allow_shell),
443	trust_mode: Some(task.trust_mode),
444	auto_approve: Some(task.auto_approve),
445	owner_session_id: task.owner_session_id.clone(),
446	}
447	}
448
449	#[cfg(test)]
450	#[must_use]
451	pub fn from_prompt(prompt: impl Into<String>) -> Self {
452	Self {
453	prompt: prompt.into(),
454	name: None,
455	model: None,
456	model_provider: None,
457	model_provider_id: None,
458	workspace: None,
459	mode: None,
460	allow_shell: None,
461	trust_mode: None,
462	auto_approve: Some(true),
463	owner_session_id: None,
464	}
465	}
466	}
467
468	/// Task manager startup options.
469	#[derive(Debug, Clone)]
470	pub struct TaskManagerConfig {
471	pub data_dir: PathBuf,
472	pub worker_count: usize,
473	pub default_workspace: PathBuf,
474	pub default_model: String,
475	pub default_mode: String,
476	pub allow_shell: bool,
477	pub trust_mode: bool,
478	pub execution_limits: TaskExecutionLimits,
479	}
480
481	/// Deadlines and persistence cadence for one durable execution.
482	#[derive(Debug, Clone, Copy)]
483	pub struct TaskExecutionLimits {
484	pub wall_time: Duration,
485	pub idle_progress: Duration,
486	pub cancel_grace: Duration,
487	pub persist_debounce: Duration,
488	}
489
490	impl Default for TaskExecutionLimits {
491	fn default() -> Self {
492	Self {
493	wall_time: Duration::from_secs(30 * 60),
494	idle_progress: Duration::from_secs(2 * 60),
495	cancel_grace: Duration::from_secs(5),
496	persist_debounce: Duration::from_millis(250),
497	}
498	}
499	}
500
501	#[cfg(test)]
502	impl TaskExecutionLimits {
503	fn short_for_tests() -> Self {
504	Self {
505	wall_time: Duration::from_millis(400),
506	idle_progress: Duration::from_millis(150),
507	cancel_grace: Duration::from_millis(50),
508	persist_debounce: Duration::from_millis(10),
509	}
510	}
511	}
512
513	/// Pure watchdog for cancel grace, wall time, and idle progress.
514	struct ExecutionGuard {
515	started_at: Instant,
516	last_progress_at: Instant,
517	interrupt_at: Option<Instant>,
518	interrupt_reason: Option<TaskTerminalReason>,
519	limits: TaskExecutionLimits,
520	}
521
522	#[derive(Debug)]
523	enum GuardAction {
524	Run { wait: Duration },
525	Interrupt { reason: TaskTerminalReason },
526	Terminalize { reason: TaskTerminalReason },
527	}
528
529	impl ExecutionGuard {
530	fn new(limits: TaskExecutionLimits, now: Instant) -> Self {
531	Self {
532	started_at: now,
533	last_progress_at: now,
534	interrupt_at: None,
535	interrupt_reason: None,
536	limits,
537	}
538	}
539
540	fn note_progress(&mut self, now: Instant) {
541	self.last_progress_at = now;
542	}
543
544	fn note_interrupt(&mut self, now: Instant, reason: TaskTerminalReason) {
545	if self.interrupt_at.is_none() {
546	self.interrupt_at = Some(now);
547	self.interrupt_reason = Some(reason);
548	}
549	}
550
551	fn evaluate(&self, now: Instant, cancel: bool, shutdown: bool) -> GuardAction {
552	if let Some(interrupt_at) = self.interrupt_at {
553	let elapsed = now.saturating_duration_since(interrupt_at);
554	if elapsed >= self.limits.cancel_grace {
555	let reason = self
556	.interrupt_reason
557	.unwrap_or(TaskTerminalReason::CancelTimeout)
558	.after_grace();
559	return GuardAction::Terminalize { reason };
560	}
561	return GuardAction::Run {
562	wait: self.limits.cancel_grace.saturating_sub(elapsed),
563	};
564	}
565
566	let wall_elapsed = now.saturating_duration_since(self.started_at);
567	let idle_elapsed = now.saturating_duration_since(self.last_progress_at);
568	// A limit whose deadline does not fit in `Instant` can never fire.
569	let wall_deadline = self.started_at.checked_add(self.limits.wall_time);
570	let idle_deadline = self.last_progress_at.checked_add(self.limits.idle_progress);
571	let pending = if shutdown {
572	Some(TaskTerminalReason::Shutdown)
573	} else if cancel {
574	Some(TaskTerminalReason::Canceled)
575	} else {
576	// Attribute the timeout to the limit that was crossed first, not
577	// to the one this tick happens to check first. When the watchdog
578	// is starved past both deadlines (a >=250 ms scheduler stall on a
579	// loaded CI runner is enough with the test budgets), the idle
580	// limit that expired earlier is still the truthful reason; a tie
581	// keeps the wall limit's precedence (issue #5898).
582	match (wall_deadline, idle_deadline) {
583	(Some(wall), Some(idle)) if now >= wall && wall <= idle => {
584	Some(TaskTerminalReason::WallTimeout)
585	}
586	(_, Some(idle)) if now >= idle => Some(TaskTerminalReason::IdleTimeout),
587	(Some(wall), _) if now >= wall => Some(TaskTerminalReason::WallTimeout),
588	_ => None,
589	}
590	};
591	if let Some(reason) = pending {
592	return GuardAction::Interrupt { reason };
593	}
594
595	let wait = self
596	.limits
597	.wall_time
598	.saturating_sub(wall_elapsed)
599	.min(self.limits.idle_progress.saturating_sub(idle_elapsed))
600	.min(EVENT_CATCHUP_POLL);
601	GuardAction::Run {
602	wait: wait.max(Duration::from_millis(1)),
603	}
604	}
605
606	fn preserve_timeout_reason(&self, result: TaskExecutionResult) -> TaskExecutionResult {
607	if result.terminal_reason != TaskTerminalReason::Canceled {
608	return result;
609	}
610	match self.interrupt_reason {
611	Some(reason @ (TaskTerminalReason::WallTimeout \| TaskTerminalReason::IdleTimeout)) => {
612	TaskExecutionResult::from_reason(reason, result.result_text)
613	}
614	_ => result,
615	}
616	}
617	}
618
619	impl TaskManagerConfig {
620	#[must_use]
621	pub fn from_runtime(
622	config: &Config,
623	workspace: PathBuf,
624	default_model: Option<String>,
625	worker_count: Option<usize>,
626	) -> Self {
627	Self {
628	data_dir: default_tasks_dir(),
629	worker_count: worker_count.unwrap_or(DEFAULT_WORKERS),
630	default_workspace: workspace,
631	default_model: default_model.unwrap_or_else(\|\| config.default_model()),
632	default_mode: "agent".to_string(),
633	allow_shell: config.allow_shell(),
634	trust_mode: false,
635	execution_limits: TaskExecutionLimits::default(),
636	}
637	}
638	}
639
640	#[derive(Debug, Clone)]
641	pub struct ExecutionTask {
642	id: String,
643	prompt: String,
644	model: String,
645	model_provider: Option<String>,
646	model_provider_id: Option<String>,
647	workspace: PathBuf,
648	mode_label: String,
649	allow_shell: bool,
650	trust_mode: bool,
651	auto_approve: bool,
652	}
653
654	impl From<&TaskRecord> for ExecutionTask {
655	fn from(task: &TaskRecord) -> Self {
656	Self {
657	id: task.id.clone(),
658	prompt: task.prompt.clone(),
659	model: task.model.clone(),
660	model_provider: task.model_provider.clone(),
661	model_provider_id: task.model_provider_id.clone(),
662	workspace: task.workspace.clone(),
663	mode_label: task.mode.clone(),
664	allow_shell: task.allow_shell,
665	trust_mode: task.trust_mode,
666	auto_approve: task.auto_approve,
667	}
668	}
669	}
670
671	impl ExecutionTask {
672	pub(crate) fn thread_request(&self) -> CreateThreadRequest {
673	CreateThreadRequest {
674	model: Some(self.model.clone()),
675	model_provider: self.model_provider.clone(),
676	model_provider_id: self.model_provider_id.clone(),
677	workspace: Some(self.workspace.clone()),
678	mode: Some(self.mode_label.clone()),
679	allow_shell: Some(self.allow_shell),
680	trust_mode: Some(self.trust_mode),
681	auto_approve: Some(self.auto_approve),
682	task_id: Some(self.id.clone()),
683	..Default::default()
684	}
685	}
686	}
687
688	/// Event stream produced by an executor while a task runs.
689	#[derive(Debug, Clone)]
690	pub enum TaskExecutionEvent {
691	ThreadLinked {
692	thread_id: String,
693	turn_id: String,
694	},
695	Status {
696	message: String,
697	},
698	MessageDelta {
699	content: String,
700	},
701	ToolStarted {
702	id: String,
703	name: String,
704	input: Value,
705	},
706	ToolProgress {
707	id: String,
708	output: String,
709	},
710	ToolCompleted {
711	id: String,
712	name: String,
713	success: bool,
714	output: String,
715	metadata: Option<Value>,
716	},
717	Error {
718	message: String,
719	},
720	RuntimeEvent {
721	seq: u64,
722	event: String,
723	summary: String,
724	},
725	}
726
727	/// Final executor result.
728	#[derive(Debug, Clone)]
729	pub struct TaskExecutionResult {
730	pub status: TaskStatus,
731	pub result_text: Option<String>,
732	pub error: Option<String>,
733	pub terminal_reason: TaskTerminalReason,
734	}
735
736	impl TaskExecutionResult {
737	fn failed(error: impl Into<String>) -> Self {
738	Self {
739	status: TaskStatus::Failed,
740	result_text: None,
741	error: Some(error.into()),
742	terminal_reason: TaskTerminalReason::Failed,
743	}
744	}
745
746	fn from_reason(reason: TaskTerminalReason, result_text: Option<String>) -> Self {
747	let error = match reason {
748	TaskTerminalReason::Completed \| TaskTerminalReason::Canceled => None,
749	_ => Some(reason.receipt_message()),
750	};
751	Self {
752	status: reason.task_status(),
753	result_text,
754	error,
755	terminal_reason: reason,
756	}
757	}
758	}
759
760	/// Abstraction for task execution.
761	#[async_trait]
762	pub trait TaskExecutor: Send + Sync {
763	async fn execute(
764	&self,
765	task: ExecutionTask,
766	events: mpsc::Sender<TaskExecutionEvent>,
767	cancel: CancellationToken,
768	) -> TaskExecutionResult;
769	}
770
771	/// Executor backed by the shared runtime and its canonical provider resolver.
772	pub struct EngineTaskExecutor {
773	runtime_threads: SharedRuntimeThreadManager,
774	limits: TaskExecutionLimits,
775	}
776
777	impl EngineTaskExecutor {
778	#[must_use]
779	pub fn new(runtime_threads: SharedRuntimeThreadManager, limits: TaskExecutionLimits) -> Self {
780	Self {
781	runtime_threads,
782	limits,
783	}
784	}
785	}
786
787	#[async_trait]
788	impl TaskExecutor for EngineTaskExecutor {
789	async fn execute(
790	&self,
791	task: ExecutionTask,
792	events: mpsc::Sender<TaskExecutionEvent>,
793	cancel: CancellationToken,
794	) -> TaskExecutionResult {
795	if cancel.is_cancelled() {
796	return TaskExecutionResult::from_reason(TaskTerminalReason::Canceled, None);
797	}
798	let thread = match self
799	.runtime_threads
800	.create_thread(task.thread_request())
801	.await
802	{
803	Ok(thread) => thread,
804	Err(err) => {
805	return TaskExecutionResult::failed(format!(
806	"Failed to create runtime thread: {err}"
807	));
808	}
809	};
810
811	if cancel.is_cancelled() {
812	return TaskExecutionResult::from_reason(TaskTerminalReason::Canceled, None);
813	}
814	let turn = match self
815	.runtime_threads
816	.start_turn(
817	&thread.id,
818	StartTurnRequest {
819	prompt: task.prompt.clone(),
820	input_summary: Some(summarize_text(&task.prompt, TIMELINE_SUMMARY_LIMIT)),
821	model: Some(task.model.clone()),
822	mode: Some(task.mode_label.clone()),
823	allow_shell: Some(task.allow_shell),
824	trust_mode: Some(task.trust_mode),
825	auto_approve: Some(task.auto_approve),
826	..Default::default()
827	},
828	)
829	.await
830	{
831	Ok(turn) => turn,
832	Err(err) => {
833	return TaskExecutionResult::failed(format!("Failed to start task: {err}"));
834	}
835	};
836
837	emit_task_event(
838	&events,
839	TaskExecutionEvent::ThreadLinked {
840	thread_id: thread.id.clone(),
841	turn_id: turn.id.clone(),
842	},
843	)
844	.await;
845	emit_task_event(
846	&events,
847	TaskExecutionEvent::Status {
848	message: format!("Task {} started", task.id),
849	},
850	)
851	.await;
852
853	drive_engine_turn(
854	self.runtime_threads.as_ref(),
855	&thread.id,
856	&turn.id,
857	events,
858	cancel,
859	self.limits,
860	)
861	.await
862	}
863	}
864
865	async fn drive_engine_turn(
866	runtime_threads: &RuntimeThreadManager,
867	thread_id: &str,
868	turn_id: &str,
869	events: mpsc::Sender<TaskExecutionEvent>,
870	cancel: CancellationToken,
871	limits: TaskExecutionLimits,
872	) -> TaskExecutionResult {
873	let mut subscription = runtime_threads.subscribe_events();
874	let mut guard = ExecutionGuard::new(limits, Instant::now());
875	let mut final_text = RuntimeTaskOutput::default();
876	let mut cursor = 0u64;
877	let mut terminal_status: Option<RuntimeTurnStatus> = None;
878	let mut terminal_error: Option<String> = None;
879	// Approval requests this turn is waiting on, each with the deadline the
880	// runtime bridge will resolve it by (#6118).
881	let mut pending_approvals: HashMap<String, Instant> = HashMap::new();
882
883	loop {
884	let batch = match runtime_threads
885	.events_from_offset_async(thread_id, cursor, Some(EVENT_CURSOR_BATCH))
886	.await
887	{
888	Ok((batch, next_cursor)) => {
889	cursor = next_cursor;
890	batch
891	}
892	Err(err) => {
893	return TaskExecutionResult {
894	status: TaskStatus::Failed,
895	result_text: final_text.into_result(true),
896	error: Some(format!("Failed to read runtime events: {err}")),
897	terminal_reason: TaskTerminalReason::Failed,
898	};
899	}
900	};
901
902	let more_pending = batch.len() >= EVENT_CURSOR_BATCH;
903	for event in batch {
904	if event.thread_id != thread_id {
905	continue;
906	}
907	if event
908	.turn_id
909	.as_deref()
910	.is_some_and(\|event_turn\| event_turn != turn_id)
911	{
912	continue;
913	}
914	match event.event.as_str() {
915	// An approval parks the turn on an external decision until
916	// the runtime bridge answers or its own window closes; note
917	// that deadline so the idle watchdog stays off it (#6118).
918	"approval.required" => {
919	let approval_id = event
920	.payload
921	.get("approval_id")
922	.and_then(Value::as_str)
923	.map(ToString::to_string)
924	.unwrap_or_else(\|\| format!("approval-{}", event.seq));
925	let until = match runtime_threads.approval_decision_timeout() {
926	Some(wait) => Instant::now() + wait,
927	// `0` waits indefinitely by configuration; the wall
928	// deadline still bounds the run.
929	None => Instant::now() + limits.wall_time,
930	};
931	pending_approvals.insert(approval_id, until);
932	}
933	"approval.decided" => {
934	if let Some(approval_id) =
935	event.payload.get("approval_id").and_then(Value::as_str)
936	{
937	pending_approvals.remove(approval_id);
938	}
939	}
940	_ => {}
941	}
942	if runtime_event_is_progress(&event) {
943	guard.note_progress(Instant::now());
944	}
945	if let Some((status, error)) =
946	ingest_runtime_event(&event, &mut final_text, &events).await
947	{
948	// The decision window closed on a pending approval: the
949	// runtime already denied the tool, and an unattended run has
950	// no operator to answer, so stop the turn instead of letting
951	// it keep burning under a failure nobody sees (#6118).
952	if event.event.as_str() == "approval.timeout" {
953	let _ = runtime_threads.interrupt_turn(thread_id, turn_id).await;
954	}
955	terminal_status = Some(status);
956	terminal_error = error;
957	}
958	}
959
960	if terminal_status.is_some() {
961	break;
962	}
963
964	// While an approval is pending the turn is deliberately waiting on an
965	// external decision, not drifting: keep the idle deadline from firing
966	// so the bridge's own window can resolve and record it. Entries expire
967	// with their window, so a decision that never arrives cannot suspend
968	// the watchdog forever (#6118).
969	let now = Instant::now();
970	pending_approvals.retain(\|_, until\| now < *until);
971	if !pending_approvals.is_empty() {
972	guard.note_progress(now);
973	}
974
975	match guard.evaluate(now, cancel.is_cancelled(), false) {
976	GuardAction::Interrupt { reason } => {
977	let _ = runtime_threads.interrupt_turn(thread_id, turn_id).await;
978	emit_task_event(
979	&events,
980	TaskExecutionEvent::Status {
981	message: reason.receipt_message(),
982	},
983	)
984	.await;
985	guard.note_interrupt(Instant::now(), reason);
986	}
987	GuardAction::Terminalize { reason } => {
988	return TaskExecutionResult::from_reason(reason, final_text.into_result(true));
989	}
990	GuardAction::Run { wait } => {
991	if more_pending {
992	continue;
993	}
994	tokio::select! {
995	_ = cancel.cancelled(), if !cancel.is_cancelled() => {}
996	_ = subscription.recv() => {}
997	_ = sleep(wait) => {}
998	}
999	}
1000	}
1001	}
1002
1003	let result = match terminal_status.unwrap_or(RuntimeTurnStatus::Failed) {
1004	RuntimeTurnStatus::Completed => TaskExecutionResult {
1005	status: TaskStatus::Completed,
1006	result_text: final_text.into_result(false),
1007	error: None,
1008	terminal_reason: TaskTerminalReason::Completed,
1009	},
1010	RuntimeTurnStatus::Interrupted \| RuntimeTurnStatus::Canceled => TaskExecutionResult {
1011	status: TaskStatus::Canceled,
1012	result_text: final_text.into_result(true),
1013	error: None,
1014	terminal_reason: TaskTerminalReason::Canceled,
1015	},
1016	RuntimeTurnStatus::Queued \| RuntimeTurnStatus::InProgress \| RuntimeTurnStatus::Failed => {
1017	TaskExecutionResult {
1018	status: TaskStatus::Failed,
1019	result_text: final_text.into_result(true),
1020	error: terminal_error
1021	.or_else(\|\| Some(TaskTerminalReason::Failed.receipt_message())),
1022	terminal_reason: TaskTerminalReason::Failed,
1023	}
1024	}
1025	};
1026	guard.preserve_timeout_reason(result)
1027	}
1028
1029	async fn emit_task_event(events: &mpsc::Sender<TaskExecutionEvent>, event: TaskExecutionEvent) {
1030	let _ = events.send(event).await;
1031	}
1032
1033	fn optional_nonzero_text(text: String) -> Option<String> {
1034	if text.trim().is_empty() {
1035	None
1036	} else {
1037	Some(text)
1038	}
1039	}
1040
1041	/// A task result is the last message, not concatenated progress commentary.
1042	/// Only interrupted/failed execution may return a still-streaming message.
1043	#[derive(Default)]
1044	struct RuntimeTaskOutput {
1045	text: String,
1046	completed: bool,
1047	}
1048
1049	impl RuntimeTaskOutput {
1050	fn into_result(self, allow_partial: bool) -> Option<String> {
1051	(self.completed \|\| allow_partial)
1052	.then(\|\| optional_nonzero_text(self.text))
1053	.flatten()
1054	}
1055	}
1056
1057	fn append_message_delta(result_text: &mut String, event: &TaskExecutionEvent) {
1058	if let TaskExecutionEvent::MessageDelta { content } = event {
1059	result_text.push_str(content);
1060	}
1061	}
1062
1063	fn runtime_event_is_progress(event: &RuntimeEventRecord) -> bool {
1064	matches!(
1065	event.event.as_str(),
1066	"item.delta" \| "item.started" \| "item.completed" \| "item.failed" \| "turn.completed"
1067	)
1068	}
1069
1070	async fn ingest_runtime_event(
1071	event: &RuntimeEventRecord,
1072	final_text: &mut RuntimeTaskOutput,
1073	events: &mpsc::Sender<TaskExecutionEvent>,
1074	) -> Option<(RuntimeTurnStatus, Option<String>)> {
1075	emit_task_event(
1076	events,
1077	TaskExecutionEvent::RuntimeEvent {
1078	seq: event.seq,
1079	event: event.event.clone(),
1080	summary: summarize_text(&event.payload.to_string(), TIMELINE_SUMMARY_LIMIT),
1081	},
1082	)
1083	.await;
1084
1085	match event.event.as_str() {
1086	"item.delta" => {
1087	let kind = event
1088	.payload
1089	.get("kind")
1090	.and_then(Value::as_str)
1091	.unwrap_or_default();
1092	if kind == "agent_message" {
1093	if let Some(content) = event.payload.get("delta").and_then(Value::as_str) {
1094	final_text.text.push_str(content);
1095	final_text.completed = false;
1096	emit_task_event(
1097	events,
1098	TaskExecutionEvent::MessageDelta {
1099	content: content.to_string(),
1100	},
1101	)
1102	.await;
1103	}
1104	} else if kind == "tool_call" {
1105	let output = event
1106	.payload
1107	.get("delta")
1108	.and_then(Value::as_str)
1109	.unwrap_or_default()
1110	.to_string();
1111	emit_task_event(
1112	events,
1113	TaskExecutionEvent::ToolProgress {
1114	id: event.item_id.clone().unwrap_or_default(),
1115	output,
1116	},
1117	)
1118	.await;
1119	}
1120	None
1121	}
1122	"item.started" => {
1123	if event.payload.pointer("/item/kind").and_then(Value::as_str) == Some("agent_message")
1124	{
1125	*final_text = RuntimeTaskOutput::default();
1126	}
1127	if let Some(tool) = event.payload.get("tool") {
1128	let id = tool
1129	.get("id")
1130	.and_then(Value::as_str)
1131	.unwrap_or_default()
1132	.to_string();
1133	let name = tool
1134	.get("name")
1135	.and_then(Value::as_str)
1136	.unwrap_or_default()
1137	.to_string();
1138	let input = tool.get("input").cloned().unwrap_or_else(\|\| json!({}));
1139	emit_task_event(events, TaskExecutionEvent::ToolStarted { id, name, input }).await;
1140	}
1141	None
1142	}
1143	"item.completed" \| "item.failed" => {
1144	if let Some(item) = event.payload.get("item") {
1145	let kind = item.get("kind").and_then(Value::as_str).unwrap_or_default();
1146	if kind == "tool_call" \|\| kind == "file_change" \|\| kind == "command_execution" {
1147	let metadata = item.get("metadata");
1148	// Starts carry the provider call ID; item.id is Runtime's
1149	// separate receipt ID. Runtime preserves the call identity
1150	// in terminal metadata, including errors and redacted input.
1151	let id = metadata
1152	.and_then(\|meta\| {
1153	meta.get("tool_result_for")
1154	.or_else(\|\| meta.get("tool_use_id"))
1155	.or_else(\|\| meta.get("tool_call_id"))
1156	})
1157	.and_then(Value::as_str)
1158	.filter(\|id\| !id.is_empty())
1159	.or_else(\|\| item.get("id").and_then(Value::as_str))
1160	.unwrap_or_default()
1161	.to_string();
1162	let name = metadata
1163	.and_then(\|meta\| meta.get("tool_name"))
1164	.and_then(Value::as_str)
1165	.filter(\|name\| !name.is_empty())
1166	.unwrap_or_else(\|\| {
1167	item.get("summary")
1168	.and_then(Value::as_str)
1169	.unwrap_or("tool")
1170	.split(':')
1171	.next()
1172	.unwrap_or("tool")
1173	.trim()
1174	})
1175	.to_string();
1176	let output = item
1177	.get("detail")
1178	.and_then(Value::as_str)
1179	.unwrap_or_default()
1180	.to_string();
1181	emit_task_event(
1182	events,
1183	TaskExecutionEvent::ToolCompleted {
1184	id,
1185	name,
1186	success: event.event == "item.completed",
1187	output,
1188	metadata: metadata.cloned(),
1189	},
1190	)
1191	.await;
1192	} else if kind == "agent_message" {
1193	// The completed item is authoritative even when catch-up
1194	// did not receive its deltas. Replacing avoids duplication.
1195	final_text.text = item
1196	.get("detail")
1197	.and_then(Value::as_str)
1198	.or_else(\|\| item.get("summary").and_then(Value::as_str))
1199	.unwrap_or_default()
1200	.to_string();
1201	final_text.completed = event.event == "item.completed";
1202	} else if kind == "status" {
1203	let message = item
1204	.get("detail")
1205	.and_then(Value::as_str)
1206	.or_else(\|\| item.get("summary").and_then(Value::as_str))
1207	.unwrap_or_default()
1208	.to_string();
1209	emit_task_event(events, TaskExecutionEvent::Status { message }).await;
1210	} else if kind == "error" {
1211	let message = item
1212	.get("detail")
1213	.and_then(Value::as_str)
1214	.or_else(\|\| item.get("summary").and_then(Value::as_str))
1215	.unwrap_or_default()
1216	.to_string();
1217	emit_task_event(events, TaskExecutionEvent::Error { message }).await;
1218	}
1219	}
1220	None
1221	}
1222	"turn.completed" => {
1223	if let Some(turn_payload) = event.payload.get("turn") {
1224	let status = turn_payload
1225	.get("status")
1226	.and_then(Value::as_str)
1227	.unwrap_or("failed");
1228	let terminal_status = match status {
1229	"completed" => RuntimeTurnStatus::Completed,
1230	"interrupted" => RuntimeTurnStatus::Interrupted,
1231	"canceled" => RuntimeTurnStatus::Canceled,
1232	_ => RuntimeTurnStatus::Failed,
1233	};
1234	let terminal_error = turn_payload
1235	.get("error")
1236	.and_then(Value::as_str)
1237	.map(ToString::to_string);
1238	Some((terminal_status, terminal_error))
1239	} else {
1240	Some((RuntimeTurnStatus::Completed, None))
1241	}
1242	}
1243	RUNTIME_STORE_FAILURE_EVENT => {
1244	// The runtime's own store failed. The notice names the file and
1245	// the next action; `terminal` means no `turn.completed` can
1246	// follow, so the driver stops waiting instead of idling out (#5931).
1247	let message = event
1248	.payload
1249	.get("message")
1250	.and_then(Value::as_str)
1251	.unwrap_or("Session runtime store failure")
1252	.to_string();
1253	emit_task_event(
1254	events,
1255	TaskExecutionEvent::Error {
1256	message: message.clone(),
1257	},
1258	)
1259	.await;
1260	event
1261	.payload
1262	.get("terminal")
1263	.and_then(Value::as_bool)
1264	.unwrap_or(false)
1265	.then_some((RuntimeTurnStatus::Failed, Some(message)))
1266	}
1267
1268	"approval.timeout" => Some((
1269	RuntimeTurnStatus::Failed,
1270	Some(
1271	"Tool approval was not answered within the decision window; the runtime denied the tool and the run stopped."
1272	.to_string(),
1273	),
1274	)),
1275	_ => None,
1276	}
1277	}
1278
1279	/// Thread-safe task manager.
1280	pub type SharedTaskManager = Arc<TaskManager>;
1281
1282	pub(crate) struct TaskManagerShutdownGuard(std::sync::Weak<TaskManager>);
1283
1284	impl Drop for TaskManagerShutdownGuard {
1285	fn drop(&mut self) {
1286	if let Some(manager) = self.0.upgrade() {
1287	manager.shutdown();
1288	if let Ok(runtime) = tokio::runtime::Handle::try_current() {
1289	runtime.spawn(async move {
1290	if let Err(error) = manager.shutdown_and_wait().await {
1291	tracing::error!(%error, "Task service shutdown failed");
1292	}
1293	});
1294	}
1295	}
1296	}
1297	}
1298
1299	/// A process generation stays live as long as either the manager or actual
1300	/// Runtime work retains it. The Runtime's existing scope lock excludes another
1301	/// executor for that scope; this generation lock proves a particular owner died.
1302	#[derive(Debug)]
1303	pub(crate) struct TaskExecutionLease {
1304	scope: String,
1305	generation: String,
1306	_scope_owner: Arc<RuntimeProcessOwnerLock>,
1307	_generation_owner: RuntimeProcessOwnerLock,
1308	}
1309
1310	impl TaskExecutionLease {
1311	fn new(
1312	root: &Path,
1313	scope: String,
1314	scope_owner: Arc<RuntimeProcessOwnerLock>,
1315	) -> Result<Arc<Self>> {
1316	validate_execution_id(&scope, 64)?;
1317	let generation = Uuid::new_v4().simple().to_string();
1318	let path = execution_lease_path(root, &scope, &generation)?;
1319	let owner = RuntimeProcessOwnerLock::try_acquire_file(&path, true)?
1320	.context("Task execution generation is already owned")?;
1321	Ok(Arc::new(Self {
1322	scope,
1323	generation,
1324	_scope_owner: scope_owner,
1325	_generation_owner: owner,
1326	}))
1327	}
1328	}
1329
1330	fn validate_execution_id(value: &str, length: usize) -> Result<()> {
1331	if value.len() != length \|\| !value.bytes().all(\|byte\| byte.is_ascii_hexdigit()) {
1332	bail!("Invalid task execution identity");
1333	}
1334	Ok(())
1335	}
1336
1337	fn execution_lease_path(root: &Path, scope: &str, generation: &str) -> Result<PathBuf> {
1338	validate_execution_id(scope, 64)?;
1339	validate_execution_id(generation, 32)?;
1340	Ok(root
1341	.join("execution-owners")
1342	.join(format!("{scope}.{generation}.lock")))
1343	}
1344
1345	#[cfg(test)]
1346	pub(crate) fn test_execution_scope(name: &str) -> String {
1347	use sha2::{Digest, Sha256};
1348	Sha256::digest(name.as_bytes())
1349	.iter()
1350	.map(\|byte\| format!("{byte:02x}"))
1351	.collect()
1352	}
1353
1354	pub struct TaskManager {
1355	cfg: TaskManagerConfig,
1356	default_workspace: Mutex<PathBuf>,
1357	executor: Arc<dyn TaskExecutor>,
1358	/// The runtime thread store this manager drives, when it owns one.
1359	runtime_threads: Option<SharedRuntimeThreadManager>,
1360	tasks_dir: PathBuf,
1361	artifacts_dir: PathBuf,
1362	queue_path: PathBuf,
1363	state: Mutex<ManagerState>,
1364	notify: Notify,
1365	cancel_token: CancellationToken,
1366	execution_lease: Arc<TaskExecutionLease>,
1367	workers: Mutex<Vec<tokio::task::JoinHandle<()>>>,
1368	shutdown_drain: Mutex<()>,
1369	}
1370
1371	struct ManagerState {
1372	tasks: HashMap<String, TaskRecord>,
1373	queue: VecDeque<String>,
1374	running_cancel: HashMap<String, CancellationToken>,
1375	/// Uncommitted typed deltas, reapplied to a fresh record before persistence.
1376	pending_events: HashMap<String, Vec<TaskExecutionEvent>>,
1377	}
1378
1379	#[derive(Debug, Serialize, Deserialize, Default)]
1380	struct QueueFile {
1381	queue: Vec<String>,
1382	}
1383
1384	impl TaskManager {
1385	/// Start the manager with the default DeepSeek executor.
1386	///
1387	/// Interactive callers pass an initial session id to isolate new hosts,
1388	/// or the saved store binding to retain the same authority across resume.
1389	pub async fn start(
1390	cfg: TaskManagerConfig,
1391	api_config: Config,
1392	plugin_registry: Arc<crate::plugins::PluginRegistry>,
1393	session_id: &str,
1394	binding: Option<&crate::runtime_threads::RuntimeStoreBinding>,
1395	) -> Result<SharedTaskManager> {
1396	let runtime_threads = Arc::new(RuntimeThreadManager::open_for_session(
1397	api_config.clone(),
1398	cfg.default_workspace.clone(),
1399	RuntimeThreadManagerConfig::for_session(cfg.data_dir.clone(), session_id),
1400	plugin_registry,
1401	binding,
1402	)?);
1403	Self::start_with_runtime_manager(cfg, api_config, runtime_threads).await
1404	}
1405
1406	/// Start the manager with an injected runtime thread manager.
1407	pub async fn start_with_runtime_manager(
1408	cfg: TaskManagerConfig,
1409	_api_config: Config,
1410	runtime_threads: SharedRuntimeThreadManager,
1411	) -> Result<SharedTaskManager> {
1412	let executor: Arc<dyn TaskExecutor> = Arc::new(EngineTaskExecutor::new(
1413	runtime_threads.clone(),
1414	cfg.execution_limits,
1415	));
1416	let identity = runtime_threads.task_execution_identity();
1417	let manager = Self::start_with_executor_and_runtime(
1418	cfg,
1419	executor,
1420	Some(runtime_threads.clone()),
1421	identity,
1422	)
1423	.await?;
1424	runtime_threads.attach_task_manager(manager.clone());
1425	Ok(manager)
1426	}
1427
1428	/// Start the manager with a custom executor (used for tests).
1429	#[cfg(test)]
1430	pub async fn start_with_executor(
1431	cfg: TaskManagerConfig,
1432	executor: Arc<dyn TaskExecutor>,
1433	) -> Result<SharedTaskManager> {
1434	Self::start_with_executor_in_scope(cfg, executor, "test").await
1435	}
1436
1437	#[cfg(test)]
1438	pub(crate) async fn start_with_executor_in_scope(
1439	cfg: TaskManagerConfig,
1440	executor: Arc<dyn TaskExecutor>,
1441	scope: &str,
1442	) -> Result<SharedTaskManager> {
1443	let scope = test_execution_scope(scope);
1444	let path = cfg
1445	.data_dir
1446	.join("execution-owners")
1447	.join(format!("test-{scope}.lock"));
1448	let owner = RuntimeProcessOwnerLock::try_acquire_file(&path, true)?
1449	.context("Mock Runtime execution scope is already owned")?;
1450	Self::start_with_executor_and_runtime(cfg, executor, None, (scope, Arc::new(owner))).await
1451	}
1452
1453	async fn start_with_executor_and_runtime(
1454	cfg: TaskManagerConfig,
1455	executor: Arc<dyn TaskExecutor>,
1456	runtime_threads: Option<SharedRuntimeThreadManager>,
1457	identity: (String, Arc<RuntimeProcessOwnerLock>),
1458	) -> Result<SharedTaskManager> {
1459	let workers = cfg.worker_count.clamp(1, MAX_WORKERS);
1460	let tasks_dir = cfg.data_dir.join("tasks");
1461	let artifacts_dir = cfg.data_dir.join("artifacts");
1462	let queue_path = cfg.data_dir.join("queue.json");
1463	tokio::fs::create_dir_all(&tasks_dir)
1464	.await
1465	.with_context(\|\| format!("Failed to create tasks dir {}", tasks_dir.display()))?;
1466	tokio::fs::create_dir_all(&artifacts_dir)
1467	.await
1468	.with_context(\|\| {
1469	format!(
1470	"Failed to create task artifacts dir {}",
1471	artifacts_dir.display()
1472	)
1473	})?;
1474
1475	let execution_lease = TaskExecutionLease::new(&cfg.data_dir, identity.0, identity.1)?;
1476	let cancel_token = CancellationToken::new();
1477	let default_workspace = cfg.default_workspace.clone();
1478	let manager = Arc::new(Self {
1479	cfg,
1480	default_workspace: Mutex::new(default_workspace),
1481	executor,
1482	runtime_threads,
1483	tasks_dir,
1484	artifacts_dir,
1485	queue_path,
1486	state: Mutex::new(ManagerState {
1487	tasks: HashMap::new(),
1488	queue: VecDeque::new(),
1489	running_cancel: HashMap::new(),
1490	pending_events: HashMap::new(),
1491	}),
1492	notify: Notify::new(),
1493	cancel_token: cancel_token.clone(),
1494	execution_lease: execution_lease.clone(),
1495	workers: Mutex::new(Vec::new()),
1496	shutdown_drain: Mutex::new(()),
1497	});
1498
1499	{
1500	let mut state = manager.state.lock().await;
1501	let _transaction = manager.lock_store().await?;
1502	manager.refresh_locked(&mut state)?;
1503	manager.recover_dead_executions_locked(&mut state)?;
1504	manager.persist_queue_locked(&state.queue)?;
1505	}
1506	if let Some(runtime) = &manager.runtime_threads {
1507	runtime.retain_task_execution_lease(execution_lease)?;
1508	}
1509
1510	for _ in 0..workers {
1511	let manager_clone = Arc::clone(&manager);
1512	let worker = spawn_supervised(
1513	"task-manager-worker",
1514	std::panic::Location::caller(),
1515	async move {
1516	manager_clone.worker_loop().await;
1517	},
1518	);
1519	manager.workers.lock().await.push(worker);
1520	}
1521
1522	Ok(manager)
1523	}
1524
1525	/// Request shutdown. Ownership remains retained through actual execution.
1526	pub fn shutdown(&self) {
1527	self.cancel_token.cancel();
1528	}
1529
1530	pub(crate) fn shutdown_guard(self: &Arc<Self>) -> TaskManagerShutdownGuard {
1531	TaskManagerShutdownGuard(Arc::downgrade(self))
1532	}
1533
1534	pub(crate) async fn shutdown_and_wait(&self) -> Result<()> {
1535	self.shutdown();
1536	let _drain = self.shutdown_drain.lock().await;
1537	if let Some(runtime) = &self.runtime_threads {
1538	runtime.close_execution_admission().await;
1539	}
1540	// Waiting through the admission fence settles requests that passed
1541	// their admission check before shutdown was requested.
1542	{
1543	let _transaction = self.lock_store().await?;
1544	}
1545	let mut failure = None;
1546	{
1547	let mut workers = self.workers.lock().await;
1548	while let Some(worker) = workers.last_mut() {
1549	// Await by reference: canceling this caller leaves the join in
1550	// the manager, so a later drain still waits for actual exit.
1551	let result = worker.await;
1552	workers.pop();
1553	if let Err(error) = result {
1554	failure = Some(anyhow!("Task worker shutdown failed: {error}"));
1555	}
1556	}
1557	}
1558	if let Some(runtime) = &self.runtime_threads {
1559	runtime.shutdown_and_wait().await?;
1560	}
1561	failure.map_or(Ok(()), Err)
1562	}
1563
1564	pub(crate) fn execution_scope(&self) -> &str {
1565	&self.execution_lease.scope
1566	}
1567
1568	pub(crate) fn session_store_binding(
1569	&self,
1570	) -> Option<crate::runtime_threads::RuntimeStoreBinding> {
1571	self.runtime_threads
1572	.as_ref()
1573	.map(\|runtime\| runtime.session_store_binding())
1574	}
1575
1576	pub async fn set_default_workspace(&self, workspace: PathBuf) {
1577	let mut default_workspace = self.default_workspace.lock().await;
1578	*default_workspace = workspace;
1579	}
1580
1581	pub async fn default_workspace(&self) -> PathBuf {
1582	self.default_workspace.lock().await.clone()
1583	}
1584
1585	/// Enqueue a new task.
1586	pub async fn add_task(&self, req: NewTaskRequest) -> Result<TaskRecord> {
1587	self.add_task_with_id(req, Self::new_task_id()).await
1588	}
1589
1590	/// Allocate the durable owner identity before queue insertion so callers
1591	/// can register graph spawn intent first.
1592	#[must_use]
1593	pub(crate) fn new_task_id() -> String {
1594	format!("task_{}", &Uuid::new_v4().simple().to_string()[..16])
1595	}
1596
1597	/// Read the exact durable task binding without adopting another process's
1598	/// queue. Used by the automation dispatcher while it owns the store claim.
1599	pub(crate) fn read_bound_task(&self, task_id: &str) -> Result<Option<TaskRecord>> {
1600	if task_id.is_empty()
1601	\|\| !task_id
1602	.bytes()
1603	.all(\|byte\| byte.is_ascii_alphanumeric() \|\| matches!(byte, b'_' \| b'-'))
1604	{
1605	bail!("Invalid durable task id");
1606	}
1607	read_bound_task_file(&self.tasks_dir.join(format!("{task_id}.json")), task_id)
1608	}
1609
1610	/// Recover a persisted automation admission under its cross-process
1611	/// dispatch lock. A promoted task is accepted work, including failed or
1612	/// interrupted work; returning it must never enqueue it again.
1613	pub(crate) async fn recover_task_admission(
1614	&self,
1615	request: NewTaskRequest,
1616	task_id: String,
1617	) -> Result<TaskRecord> {
1618	validate_preallocated_task_id(&task_id)?;
1619	if let Some(task) = self.read_bound_task(&task_id)? {
1620	validate_bound_task_request(&task, &request)?;
1621	return Ok(task);
1622	}
1623	let staged_path = self.tasks_dir.join(format!(".{task_id}.json.pending"));
1624	let admission = if let Some(staged) = read_bound_task_file(&staged_path, &task_id)? {
1625	validate_bound_task_request(&staged, &request)?;
1626	if staged.status != TaskStatus::Queued
1627	\|\| staged.started_at.is_some()
1628	\|\| staged.thread_id.is_some()
1629	\|\| staged.turn_id.is_some()
1630	{
1631	bail!("Unpromoted task stage contains execution evidence; refusing to replay it");
1632	}
1633	// The original resolved settings remain durable in this stage
1634	// until promotion, including if recovery itself is interrupted.
1635	self.admit_task_record(staged, true).await
1636	} else {
1637	self.add_task_with_id(request.clone(), task_id.clone())
1638	.await
1639	};
1640	match admission {
1641	Ok(task) => Ok(task),
1642	Err(error) => {
1643	// Preserve a task promoted before a torn response; do not
1644	// replace its identity or convert accepted work into a retry.
1645	if let Some(task) = self.read_bound_task(&task_id)? {
1646	validate_bound_task_request(&task, &request)?;
1647	Ok(task)
1648	} else {
1649	Err(error)
1650	}
1651	}
1652	}
1653	}
1654
1655	/// Enqueue using a preallocated id. This is crate-visible only for the
1656	/// model tool's register-before-work transaction.
1657	pub(crate) async fn add_task_with_id(
1658	&self,
1659	req: NewTaskRequest,
1660	task_id: String,
1661	) -> Result<TaskRecord> {
1662	let prompt = req.prompt.trim().to_string();
1663	if prompt.is_empty() {
1664	bail!("Task prompt cannot be empty");
1665	}
1666	if (req.model_provider.is_some() \|\| req.model_provider_id.is_some())
1667	&& req
1668	.model
1669	.as_deref()
1670	.is_none_or(\|model\| model.trim().is_empty())
1671	{
1672	bail!("A pinned task provider requires an explicit model");
1673	}
1674	validate_preallocated_task_id(&task_id)?;
1675
1676	let task = TaskRecord {
1677	schema_version: CURRENT_TASK_SCHEMA_VERSION,
1678	// 16 random hex chars (was 8; ~60 bits of entropy once UUIDv4's
1679	// fixed version nibble is discounted): task ids live in durable
1680	// state that accumulates across restarts, and a collision
1681	// overwrites a record while leaving a duplicate queue entry.
1682	// `resolve_task_id` matches by prefix, so short references still
1683	// work.
1684	id: task_id,
1685	prompt,
1686	name: req
1687	.name
1688	.map(\|name\| name.trim().to_string())
1689	.filter(\|name\| !name.is_empty()),
1690	model: req.model.unwrap_or_else(\|\| self.cfg.default_model.clone()),
1691	model_provider: req.model_provider,
1692	model_provider_id: req.model_provider_id,
1693	workspace: match req.workspace {
1694	Some(workspace) => workspace,
1695	None => self.default_workspace().await,
1696	},
1697	mode: req.mode.unwrap_or_else(\|\| self.cfg.default_mode.clone()),
1698	allow_shell: req.allow_shell.unwrap_or(self.cfg.allow_shell),
1699	trust_mode: req.trust_mode.unwrap_or(self.cfg.trust_mode),
1700	// Auto-approval must be opted into explicitly
1701	// (GHSA-72w5-pf8h-xfp4).
1702	auto_approve: req.auto_approve.unwrap_or(false),
1703	status: TaskStatus::Queued,
1704	created_at: Utc::now(),
1705	started_at: None,
1706	ended_at: None,
1707	duration_ms: None,
1708	result_summary: None,
1709	result_detail_path: None,
1710	error: None,
1711	terminal_reason: None,
1712	thread_id: None,
1713	turn_id: None,
1714	owner_session_id: req.owner_session_id,
1715	execution_scope: Some(self.execution_scope().to_string()),
1716	execution_generation: None,
1717	cancel_requested_seq: 0,
1718	runtime_event_count: 0,
1719	lifecycle_seq: 1,
1720	checklist: TaskChecklistState::default(),
1721	gates: Vec::new(),
1722	attempts: Vec::new(),
1723	artifacts: Vec::new(),
1724	github_events: Vec::new(),
1725	tool_calls: Vec::new(),
1726	timeline: vec![TaskTimelineEntry {
1727	timestamp: Utc::now(),
1728	kind: "queued".to_string(),
1729	summary: "Task queued".to_string(),
1730	detail_path: None,
1731	}],
1732	};
1733
1734	self.admit_task_record(task, false).await
1735	}
1736
1737	async fn admit_task_record(&self, task: TaskRecord, recover_stage: bool) -> Result<TaskRecord> {
1738	{
1739	let mut state = self.state.lock().await;
1740	let _transaction = self.lock_store().await?;
1741	self.refresh_locked(&mut state)?;
1742	if self.cancel_token.is_cancelled() {
1743	bail!("Task manager is shutting down; admission is closed");
1744	}
1745	if task.execution_scope.as_deref() != Some(self.execution_scope()) {
1746	bail!(
1747	"Task execution scope is unverified or belongs to another Runtime; refusing adoption"
1748	);
1749	}
1750	let task_path = self.tasks_dir.join(format!("{}.json", task.id));
1751	// The staged extension is intentionally not `.json`, so startup
1752	// replay ignores an interrupted create until the queue write has
1753	// succeeded and this file is atomically promoted.
1754	let staged_task_path = self.tasks_dir.join(format!(".{}.json.pending", task.id));
1755	if recover_stage {
1756	if let Some(accepted) = self.read_bound_task(&task.id)? {
1757	validate_bound_task_request(&accepted, &NewTaskRequest::from_task(&task))?;
1758	return Ok(accepted);
1759	}
1760	let current = read_bound_task_file(&staged_task_path, &task.id)?
1761	.context("Unaccepted task stage disappeared during recovery")?;
1762	if serde_json::to_value(&current)? != serde_json::to_value(&task)? {
1763	bail!("Unaccepted task stage changed during recovery");
1764	}
1765	}
1766	if state.tasks.contains_key(&task.id)
1767	\|\| task_path.exists()
1768	\|\| (!recover_stage && staged_task_path.exists())
1769	{
1770	bail!("Task id already exists: {}", task.id);
1771	}
1772	let mut next_queue = state.queue.clone();
1773	if !next_queue.contains(&task.id) {
1774	next_queue.push_back(task.id.clone());
1775	}
1776
1777	// Stage the owner record, then persist its queue membership, then
1778	// atomically promote it. A crash before promotion leaves either an
1779	// ignored staged file or a queue entry with no task (which replay
1780	// drops); a crash after promotion leaves the complete runnable
1781	// pair. In-memory scheduling is published only after all three.
1782	if !recover_stage {
1783	write_json_atomic(&staged_task_path, &task)?;
1784	}
1785	if let Err(err) = self.persist_queue_locked(&next_queue) {
1786	if !recover_stage
1787	&& let Err(cleanup_err) = tokio::fs::remove_file(&staged_task_path).await
1788	{
1789	tracing::warn!(
1790	task_id = %task.id,
1791	error = %cleanup_err,
1792	"failed to remove ignored staged task after queue write failure"
1793	);
1794	}
1795	return Err(err);
1796	}
1797	if let Err(promote_err) = tokio::fs::rename(&staged_task_path, &task_path).await {
1798	let rollback_error = self.persist_queue_locked(&state.queue).err();
1799	let cleanup_error = if recover_stage {
1800	None
1801	} else {
1802	tokio::fs::remove_file(&staged_task_path).await.err()
1803	};
1804	let mut message =
1805	format!("Failed to promote staged task {}: {promote_err}", task.id);
1806	if let Some(rollback_error) = rollback_error {
1807	message.push_str(&format!("; queue rollback also failed: {rollback_error:#}"));
1808	}
1809	if let Some(cleanup_error) = cleanup_error {
1810	message.push_str(&format!(
1811	"; ignored staged-file cleanup also failed: {cleanup_error}"
1812	));
1813	}
1814	bail!(message);
1815	}
1816	state.queue = next_queue;
1817	state.tasks.insert(task.id.clone(), task.clone());
1818	}
1819	self.notify.notify_one();
1820	Ok(task)
1821	}
1822
1823	/// List tasks, newest first.
1824	pub async fn list_tasks(&self, limit: Option<usize>) -> Result<Vec<TaskSummary>> {
1825	self.list_tasks_scoped(limit, None).await
1826	}
1827
1828	/// List tasks, newest first, optionally scoped to a workspace.
1829	pub async fn list_tasks_scoped(
1830	&self,
1831	limit: Option<usize>,
1832	workspace: Option<&Path>,
1833	) -> Result<Vec<TaskSummary>> {
1834	self.list_tasks_visible_to(limit, workspace, None).await
1835	}
1836
1837	/// List tasks owned by a session, newest first, optionally scoped to a workspace.
1838	///
1839	/// Ownerless legacy records fail closed and are not model-visible.
1840	pub async fn list_tasks_for_owner(
1841	&self,
1842	limit: Option<usize>,
1843	workspace: Option<&Path>,
1844	owner_session_id: &str,
1845	) -> Result<Vec<TaskSummary>> {
1846	self.list_tasks_visible_to(limit, workspace, Some(owner_session_id))
1847	.await
1848	}
1849
1850	async fn list_tasks_visible_to(
1851	&self,
1852	limit: Option<usize>,
1853	workspace: Option<&Path>,
1854	owner_session_id: Option<&str>,
1855	) -> Result<Vec<TaskSummary>> {
1856	let mut state = self.state.lock().await;
1857	let _transaction = self.lock_store().await?;
1858	self.refresh_locked(&mut state)?;
1859	let mut items = state
1860	.tasks
1861	.values()
1862	.filter(\|record\| {
1863	workspace.is_none_or(\|workspace\| record.workspace.as_path() == workspace)
1864	&& owner_session_id.is_none_or(\|owner_session_id\| {
1865	record.owner_session_id.as_deref() == Some(owner_session_id)
1866	})
1867	})
1868	.map(TaskSummary::from)
1869	.collect::<Vec<_>>();
1870	items.sort_by_key(\|i\| std::cmp::Reverse(i.created_at));
1871	if let Some(limit) = limit {
1872	items.truncate(limit);
1873	}
1874	Ok(items)
1875	}
1876
1877	/// Retrieve a task by full id or prefix.
1878	pub async fn get_task(&self, id_or_prefix: &str) -> Result<TaskRecord> {
1879	self.get_task_visible_to(id_or_prefix, None).await
1880	}
1881
1882	/// Retrieve a session-owned task by full id or prefix.
1883	///
1884	/// Ownership is applied before id resolution so foreign records cannot
1885	/// disclose their existence through exact matches or prefix ambiguity.
1886	pub async fn get_task_for_owner(
1887	&self,
1888	id_or_prefix: &str,
1889	owner_session_id: &str,
1890	) -> Result<TaskRecord> {
1891	self.get_task_visible_to(id_or_prefix, Some(owner_session_id))
1892	.await
1893	}
1894
1895	/// Retrieve a task the interactive operator can inspect by full id or prefix.
1896	///
1897	/// Scheduled automations have no session owner because they can outlive the
1898	/// session that configured them. They remain operator-visible only while
1899	/// bound to this manager's verified Runtime execution scope. Session-owned
1900	/// tasks keep their existing isolation, and unscoped legacy records remain
1901	/// hidden.
1902	pub(crate) async fn get_task_for_interactive_session(
1903	&self,
1904	id_or_prefix: &str,
1905	owner_session_id: &str,
1906	) -> Result<TaskRecord> {
1907	let mut state = self.state.lock().await;
1908	let _transaction = self.lock_store().await?;
1909	self.refresh_locked(&mut state)?;
1910	let id = resolve_task_id_visible_to_operator(
1911	&state.tasks,
1912	id_or_prefix,
1913	owner_session_id,
1914	self.execution_scope(),
1915	)?;
1916	state
1917	.tasks
1918	.get(&id)
1919	.cloned()
1920	.ok_or_else(\|\| anyhow!("Task not found: {id_or_prefix}"))
1921	}
1922
1923	/// Retrieve the exact owned task stamped onto a trusted runtime thread.
1924	///
1925	/// The runtime thread supplies a full durable id rather than model input.
1926	/// An ownerless task needs the current trusted execution scope. Legacy
1927	/// ownerless tasks without that provenance still fail closed.
1928	pub(crate) async fn get_task_for_active_runtime(&self, task_id: &str) -> Result<TaskRecord> {
1929	let mut state = self.state.lock().await;
1930	let _transaction = self.lock_store().await?;
1931	self.refresh_locked(&mut state)?;
1932	state
1933	.tasks
1934	.get(task_id)
1935	.filter(\|task\| match task.execution_scope.as_deref() {
1936	Some(scope) => scope == self.execution_scope(),
1937	None => task.owner_session_id.is_some(),
1938	})
1939	.cloned()
1940	.ok_or_else(\|\| anyhow!("Task not found: {task_id}"))
1941	}
1942
1943	async fn get_task_visible_to(
1944	&self,
1945	id_or_prefix: &str,
1946	owner_session_id: Option<&str>,
1947	) -> Result<TaskRecord> {
1948	let mut state = self.state.lock().await;
1949	let _transaction = self.lock_store().await?;
1950	self.refresh_locked(&mut state)?;
1951	let id = resolve_task_id_visible_to(&state.tasks, id_or_prefix, owner_session_id)?;
1952	state
1953	.tasks
1954	.get(&id)
1955	.cloned()
1956	.ok_or_else(\|\| anyhow!("Task not found: {id_or_prefix}"))
1957	}
1958
1959	/// Cancel a queued or running task by id/prefix.
1960	pub async fn cancel_task(&self, id_or_prefix: &str) -> Result<TaskCancellation> {
1961	self.cancel_task_visible_to(id_or_prefix, None, None).await
1962	}
1963
1964	/// Cancel a queued or running task owned by the given session.
1965	///
1966	/// Ownerless legacy and foreign records fail closed.
1967	pub async fn cancel_task_for_owner(
1968	&self,
1969	id_or_prefix: &str,
1970	owner_session_id: &str,
1971	) -> Result<TaskCancellation> {
1972	self.cancel_task_visible_to(id_or_prefix, Some(owner_session_id), None)
1973	.await
1974	}
1975
1976	/// Cancel a task visible to the interactive operator.
1977	///
1978	/// This is the cancellation counterpart to
1979	/// [`Self::get_task_for_interactive_session`]. It exists for human TUI
1980	/// actions, including the automation view's cancel button; model and child
1981	/// task APIs retain session-only visibility.
1982	pub(crate) async fn cancel_task_for_interactive_session(
1983	&self,
1984	id_or_prefix: &str,
1985	owner_session_id: &str,
1986	) -> Result<TaskCancellation> {
1987	self.cancel_task_visible_to(
1988	id_or_prefix,
1989	Some(owner_session_id),
1990	Some(self.execution_scope()),
1991	)
1992	.await
1993	}
1994
1995	/// Cancel the exact owned task stamped onto a trusted runtime thread.
1996	pub(crate) async fn cancel_task_for_active_runtime(
1997	&self,
1998	task_id: &str,
1999	) -> Result<TaskCancellation> {
2000	self.get_task_for_active_runtime(task_id).await?;
2001	self.cancel_task(task_id).await
2002	}
2003
2004	async fn cancel_task_visible_to(
2005	&self,
2006	id_or_prefix: &str,
2007	owner_session_id: Option<&str>,
2008	operator_execution_scope: Option<&str>,
2009	) -> Result<TaskCancellation> {
2010	let mut state = self.state.lock().await;
2011	let _transaction = self.lock_store().await?;
2012	self.refresh_locked(&mut state)?;
2013	let id = match (owner_session_id, operator_execution_scope) {
2014	(Some(owner_session_id), Some(execution_scope)) => resolve_task_id_visible_to_operator(
2015	&state.tasks,
2016	id_or_prefix,
2017	owner_session_id,
2018	execution_scope,
2019	)?,
2020	_ => resolve_task_id_visible_to(&state.tasks, id_or_prefix, owner_session_id)?,
2021	};
2022	let now = Utc::now();
2023
2024	let mut cancel_running = false;
2025	let disposition = {
2026	let task = state
2027	.tasks
2028	.get_mut(&id)
2029	.ok_or_else(\|\| anyhow!("Task not found: {id}"))?;
2030	match task.status {
2031	TaskStatus::Queued => {
2032	task.status = TaskStatus::Canceled;
2033	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
2034	task.ended_at = Some(now);
2035	task.duration_ms = Some(0);
2036	push_timeline_entry(
2037	task,
2038	TaskTimelineEntry {
2039	timestamp: now,
2040	kind: "canceled".to_string(),
2041	summary: "Task canceled before execution".to_string(),
2042	detail_path: None,
2043	},
2044	);
2045	state.queue.retain(\|queued_id\| queued_id != &id);
2046	TaskCancelDisposition::Forced
2047	}
2048	TaskStatus::Running => {
2049	cancel_running = true;
2050	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
2051	task.cancel_requested_seq = task.lifecycle_seq;
2052	push_timeline_entry(
2053	task,
2054	TaskTimelineEntry {
2055	timestamp: now,
2056	kind: "cancel_requested".to_string(),
2057	summary: "Cancellation requested".to_string(),
2058	detail_path: None,
2059	},
2060	);
2061	TaskCancelDisposition::Requested
2062	}
2063	TaskStatus::Completed \| TaskStatus::Failed \| TaskStatus::Canceled => {
2064	TaskCancelDisposition::AlreadyFinished
2065	}
2066	}
2067	};
2068
2069	if cancel_running && let Some(token) = state.running_cancel.get(&id) {
2070	token.cancel();
2071	}
2072
2073	self.persist_changed_task_locked(&mut state, &id)?;
2074	self.persist_queue_locked(&state.queue)?;
2075	let task = state
2076	.tasks
2077	.get(&id)
2078	.cloned()
2079	.ok_or_else(\|\| anyhow!("Task not found: {id}"))?;
2080	Ok(TaskCancellation { task, disposition })
2081	}
2082
2083	/// Return aggregate status counters.
2084	pub async fn counts(&self) -> Result<TaskCounts> {
2085	let mut state = self.state.lock().await;
2086	let _transaction = self.lock_store().await?;
2087	self.refresh_locked(&mut state)?;
2088	let mut counts = TaskCounts::default();
2089	for task in state.tasks.values() {
2090	match task.status {
2091	TaskStatus::Queued => counts.queued += 1,
2092	TaskStatus::Running => counts.running += 1,
2093	TaskStatus::Completed => counts.completed += 1,
2094	TaskStatus::Failed => counts.failed += 1,
2095	TaskStatus::Canceled => counts.canceled += 1,
2096	}
2097	}
2098	Ok(counts)
2099	}
2100
2101	/// Root directory for durable task state.
2102	#[must_use]
2103	pub fn data_dir(&self) -> PathBuf {
2104	self.cfg.data_dir.clone()
2105	}
2106
2107	/// Live events from the runtime thread store this manager drives, when it
2108	/// owns one. The TUI taps `runtime.store_failure` here (#5931).
2109	#[must_use]
2110	pub fn subscribe_runtime_events(
2111	&self,
2112	) -> Option<tokio::sync::broadcast::Receiver<RuntimeEventRecord>> {
2113	self.runtime_threads
2114	.as_ref()
2115	.map(\|runtime\| runtime.subscribe_events())
2116	}
2117
2118	/// Resolve a task artifact reference to an absolute path.
2119	#[must_use]
2120	pub fn artifact_absolute_path(&self, path: &Path) -> PathBuf {
2121	if path.is_absolute() {
2122	path.to_path_buf()
2123	} else {
2124	self.cfg.data_dir.join(path)
2125	}
2126	}
2127
2128	/// Write a durable task artifact and return the persisted path reference.
2129	pub fn write_task_artifact(
2130	&self,
2131	task_id: &str,
2132	label: &str,
2133	content: &str,
2134	) -> Result<PathBuf> {
2135	self.write_artifact(task_id, label, content)
2136	}
2137
2138	/// Apply model-visible tool metadata to a task and persist it.
2139	pub async fn record_tool_metadata(
2140	&self,
2141	id_or_prefix: &str,
2142	metadata: &Value,
2143	) -> Result<TaskRecord> {
2144	let mut state = self.state.lock().await;
2145	let _transaction = self.lock_store().await?;
2146	self.refresh_locked(&mut state)?;
2147	let id = resolve_task_id(&state.tasks, id_or_prefix)?;
2148	let updated = {
2149	let task = state
2150	.tasks
2151	.get_mut(&id)
2152	.ok_or_else(\|\| anyhow!("Task not found: {id}"))?;
2153	self.apply_task_update_metadata(task, Some(metadata))?;
2154	task.clone()
2155	};
2156	self.persist_changed_task_locked(&mut state, &id)?;
2157	Ok(updated)
2158	}
2159
2160	async fn claim_next_task(&self) -> Result<Option<(String, ExecutionTask, CancellationToken)>> {
2161	let mut state = self.state.lock().await;
2162	let _transaction = self.lock_store().await?;
2163	self.refresh_locked(&mut state)?;
2164	if self.cancel_token.is_cancelled() {
2165	return Ok(None);
2166	}
2167	let Some(id) = state
2168	.queue
2169	.iter()
2170	.find(\|id\| {
2171	state.tasks.get(*id).is_some_and(\|task\| {
2172	task.status == TaskStatus::Queued
2173	&& task.execution_scope.as_deref() == Some(self.execution_scope())
2174	})
2175	})
2176	.cloned()
2177	else {
2178	return Ok(None);
2179	};
2180	state.queue.retain(\|queued\| queued != &id);
2181	let task = state
2182	.tasks
2183	.get_mut(&id)
2184	.context("Claimed task is missing")?;
2185	let now = Utc::now();
2186	task.status = TaskStatus::Running;
2187	task.execution_generation = Some(self.execution_lease.generation.clone());
2188	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
2189	task.started_at = Some(now);
2190	task.ended_at = None;
2191	task.duration_ms = None;
2192	task.error = None;
2193	push_timeline_entry(
2194	task,
2195	TaskTimelineEntry {
2196	timestamp: now,
2197	kind: "running".into(),
2198	summary: "Task started".into(),
2199	detail_path: None,
2200	},
2201	);
2202	let request = ExecutionTask::from(&*task);
2203	// Removing queue membership first is recoverable from the still-Queued
2204	// record. Executor polling requires BOTH durable writes to succeed.
2205	self.persist_queue_locked(&state.queue)?;
2206	self.persist_changed_task_locked(&mut state, &id)?;
2207	let cancel = CancellationToken::new();
2208	state.running_cancel.insert(id.clone(), cancel.clone());
2209	Ok(Some((id, request, cancel)))
2210	}
2211
2212	async fn worker_loop(self: Arc<Self>) {
2213	loop {
2214	if self.cancel_token.is_cancelled() {
2215	break;
2216	}
2217	match self.claim_next_task().await {
2218	Ok(Some((id, request, cancel))) => {
2219	self.run_task(id, request, cancel).await;
2220	continue;
2221	}
2222	Ok(None) => {}
2223	Err(error) => {
2224	tracing::error!(%error, "Task claim unavailable; executor was not polled")
2225	}
2226	}
2227	tokio::select! {
2228	_ = self.cancel_token.cancelled() => break,
2229	_ = self.notify.notified() => {},
2230	_ = sleep(STORE_REFRESH_INTERVAL) => {},
2231	}
2232	}
2233	}
2234
2235	fn observe_task_cancellation(&self, task_id: &str, cancel: &CancellationToken) -> Result<()> {
2236	let task = self
2237	.read_bound_task(task_id)?
2238	.context("Running task is missing")?;
2239	self.require_execution_owner(&task)?;
2240	if task.cancel_requested_seq > 0 {
2241	cancel.cancel();
2242	}
2243	Ok(())
2244	}
2245
2246	async fn run_task(&self, task_id: String, request: ExecutionTask, cancel: CancellationToken) {
2247	let (event_tx, mut event_rx) = mpsc::channel(TASK_EVENT_CHANNEL_CAPACITY);
2248	let exec_fut = self
2249	.executor
2250	.execute(request.clone(), event_tx, cancel.clone());
2251	tokio::pin!(exec_fut);
2252
2253	let mut guard = ExecutionGuard::new(self.cfg.execution_limits, Instant::now());
2254	let mut dirty = false;
2255	let mut accumulated_result_text = String::new();
2256	let persist_debounce = self.cfg.execution_limits.persist_debounce;
2257
2258	let mut next_store_poll = Instant::now();
2259	let mut execution_started = false;
2260	let mut blocked_event: Option<TaskExecutionEvent> = None;
2261	let mut next_event_retry = Instant::now();
2262	let (mut result, manager_terminalized) = loop {
2263	if Instant::now() >= next_store_poll {
2264	if let Err(error) = self.observe_task_cancellation(&task_id, &cancel) {
2265	tracing::error!(%error, "Task ownership unavailable; requesting Runtime cancellation");
2266	cancel.cancel();
2267	}
2268	next_store_poll = Instant::now() + STORE_REFRESH_INTERVAL;
2269	}
2270	if !execution_started && (cancel.is_cancelled() \|\| self.cancel_token.is_cancelled()) {
2271	let reason = if self.cancel_token.is_cancelled() {
2272	TaskTerminalReason::Shutdown
2273	} else {
2274	TaskTerminalReason::Canceled
2275	};
2276	break (TaskExecutionResult::from_reason(reason, None), true);
2277	}
2278	if Instant::now() >= next_event_retry {
2279	if let Some(event) = blocked_event.take()
2280	&& self
2281	.process_execution_event(
2282	&task_id,
2283	event.clone(),
2284	&mut guard,
2285	&mut accumulated_result_text,
2286	&mut dirty,
2287	)
2288	.await
2289	.is_err()
2290	{
2291	blocked_event = Some(event);
2292	cancel.cancel();
2293	}
2294	next_event_retry = Instant::now() + STORE_REFRESH_INTERVAL;
2295	}
2296	let mut action = guard.evaluate(
2297	Instant::now(),
2298	cancel.is_cancelled(),
2299	self.cancel_token.is_cancelled(),
2300	);
2301	if matches!(
2302	action,
2303	GuardAction::Interrupt {
2304	reason: TaskTerminalReason::IdleTimeout
2305	}
2306	) {
2307	// Progress already accepted by the executor must win an idle
2308	// deadline race. Drain only the events queued at this instant
2309	// so a producer cannot keep the watchdog from re-evaluating
2310	// wall time, shutdown, or explicit cancellation indefinitely.
2311	let queued = event_rx.len();
2312	for _ in 0..queued {
2313	if blocked_event.is_some() {
2314	break;
2315	}
2316	let Ok(event) = event_rx.try_recv() else {
2317	break;
2318	};
2319	if self
2320	.process_execution_event(
2321	&task_id,
2322	event.clone(),
2323	&mut guard,
2324	&mut accumulated_result_text,
2325	&mut dirty,
2326	)
2327	.await
2328	.is_err()
2329	{
2330	blocked_event = Some(event);
2331	cancel.cancel();
2332	}
2333	}
2334	action = guard.evaluate(
2335	Instant::now(),
2336	cancel.is_cancelled(),
2337	self.cancel_token.is_cancelled(),
2338	);
2339	}
2340	match action {
2341	GuardAction::Interrupt { reason } => {
2342	cancel.cancel();
2343	guard.note_interrupt(Instant::now(), reason);
2344	continue;
2345	}
2346	GuardAction::Terminalize { reason } => {
2347	break (TaskExecutionResult::from_reason(reason, None), true);
2348	}
2349	GuardAction::Run { wait } => {
2350	execution_started = true;
2351	tokio::select! {
2352	biased;
2353	exec_result = &mut exec_fut => {
2354	break (guard.preserve_timeout_reason(exec_result), false);
2355	}
2356	maybe_event = event_rx.recv(), if blocked_event.is_none() => {
2357	if let Some(event) = maybe_event
2358	&& self.process_execution_event(
2359	&task_id, event.clone(), &mut guard,
2360	&mut accumulated_result_text, &mut dirty,
2361	).await.is_err() {
2362	blocked_event = Some(event);
2363	cancel.cancel();
2364	}
2365	}
2366	_ = self.cancel_token.cancelled(), if !self.cancel_token.is_cancelled() => {
2367	cancel.cancel();
2368	}
2369	_ = sleep(persist_debounce), if dirty => {
2370	match self.flush_task(&task_id).await {
2371	Ok(()) => dirty = false,
2372	Err(err) => {
2373	tracing::error!("Failed to debounce-persist task {task_id}: {err}");
2374	cancel.cancel();
2375	}
2376	}
2377	}
2378	_ = sleep(wait.min(STORE_REFRESH_INTERVAL)) => {}
2379	}
2380	}
2381	}
2382	};
2383
2384	// Stop accepting producer events while one event cannot be retained.
2385	// The pending delta set and channel remain bounded during disk failure;
2386	// keep the execution lease until accepted events and the terminal receipt
2387	// have actually been persisted. A storage failure is not completion.
2388	event_rx.close();
2389	loop {
2390	let event = blocked_event.take().or_else(\|\| event_rx.try_recv().ok());
2391	let Some(event) = event else {
2392	break;
2393	};
2394	while self
2395	.process_execution_event(
2396	&task_id,
2397	event.clone(),
2398	&mut guard,
2399	&mut accumulated_result_text,
2400	&mut dirty,
2401	)
2402	.await
2403	.is_err()
2404	{
2405	sleep(STORE_REFRESH_INTERVAL).await;
2406	}
2407	}
2408	if manager_terminalized {
2409	result.result_text = optional_nonzero_text(accumulated_result_text);
2410	}
2411	loop {
2412	match self
2413	.finish_task(
2414	&task_id,
2415	result.clone(),
2416	cancel.clone(),
2417	&request.mode_label,
2418	)
2419	.await
2420	{
2421	Ok(()) => break,
2422	Err(err) => {
2423	tracing::error!(
2424	"Task {task_id} terminal receipt is pending storage recovery: {err}"
2425	);
2426	sleep(STORE_REFRESH_INTERVAL).await;
2427	}
2428	}
2429	}
2430	}
2431
2432	async fn process_execution_event(
2433	&self,
2434	task_id: &str,
2435	event: TaskExecutionEvent,
2436	guard: &mut ExecutionGuard,
2437	accumulated_result_text: &mut String,
2438	dirty: &mut bool,
2439	) -> Result<()> {
2440	match self.apply_execution_event(task_id, event.clone()).await {
2441	Ok(outcome) => {
2442	if execution_event_is_progress(&event) {
2443	guard.note_progress(Instant::now());
2444	}
2445	append_message_delta(accumulated_result_text, &event);
2446	*dirty = !outcome.persisted;
2447	Ok(())
2448	}
2449	Err(err) => {
2450	tracing::error!("Task {task_id} event is waiting for storage recovery: {err}");
2451	Err(err)
2452	}
2453	}
2454	}
2455
2456	async fn apply_execution_event(
2457	&self,
2458	task_id: &str,
2459	event: TaskExecutionEvent,
2460	) -> Result<EventApplyOutcome> {
2461	let urgent = execution_event_persist_urgent(&event);
2462	let mut state = self.state.lock().await;
2463	let _transaction = self.lock_store().await?;
2464	self.refresh_locked(&mut state)?;
2465	if state
2466	.pending_events
2467	.get(task_id)
2468	.is_some_and(\|events\| events.len() >= TASK_EVENT_CHANNEL_CAPACITY)
2469	{
2470	self.persist_changed_task_locked(&mut state, task_id)?;
2471	}
2472	let task = state
2473	.tasks
2474	.get_mut(task_id)
2475	.context("Event task is missing")?;
2476	self.require_execution_owner(task)?;
2477	self.apply_event_to_task(task, event.clone())?;
2478	let pending = state.pending_events.entry(task_id.to_string()).or_default();
2479	pending.push(event);
2480	let persist_now = urgent \|\| pending.len() >= TASK_EVENT_CHANNEL_CAPACITY;
2481	let persisted = if persist_now {
2482	match self.persist_changed_task_locked(&mut state, task_id) {
2483	Ok(()) => true,
2484	Err(error) => {
2485	// This event is already retained in the bounded delta set.
2486	// Return acceptance so the caller must not append it again.
2487	if let Some(cancel) = state.running_cancel.get(task_id) {
2488	cancel.cancel();
2489	}
2490	tracing::error!(%error, "Task event retained pending storage recovery; cancellation requested");
2491	false
2492	}
2493	}
2494	} else {
2495	false
2496	};
2497	Ok(EventApplyOutcome { persisted })
2498	}
2499
2500	fn apply_event_to_task(&self, task: &mut TaskRecord, event: TaskExecutionEvent) -> Result<()> {
2501	let task_id = task.id.clone();
2502	match event {
2503	TaskExecutionEvent::ThreadLinked { thread_id, turn_id } => {
2504	task.thread_id = Some(thread_id.clone());
2505	task.turn_id = Some(turn_id.clone());
2506	push_timeline_entry(
2507	task,
2508	TaskTimelineEntry {
2509	timestamp: Utc::now(),
2510	kind: "runtime_link".to_string(),
2511	summary: format!("Linked runtime thread {thread_id} turn {turn_id}"),
2512	detail_path: None,
2513	},
2514	);
2515	}
2516	TaskExecutionEvent::Status { message } => {
2517	push_timeline_entry(
2518	task,
2519	TaskTimelineEntry {
2520	timestamp: Utc::now(),
2521	kind: "status".to_string(),
2522	summary: summarize_text(&message, TIMELINE_SUMMARY_LIMIT),
2523	detail_path: None,
2524	},
2525	);
2526	}
2527	TaskExecutionEvent::MessageDelta { content } => {
2528	if !content.trim().is_empty() {
2529	push_timeline_entry(
2530	task,
2531	TaskTimelineEntry {
2532	timestamp: Utc::now(),
2533	kind: "message".to_string(),
2534	summary: summarize_text(&content, TIMELINE_SUMMARY_LIMIT),
2535	detail_path: None,
2536	},
2537	);
2538	}
2539	}
2540	TaskExecutionEvent::ToolStarted { id, name, input } => {
2541	let input_summary = summarize_json(&input);
2542	task.tool_calls.push(TaskToolCallSummary {
2543	id: id.clone(),
2544	name: name.clone(),
2545	status: TaskToolStatus::Running,
2546	started_at: Utc::now(),
2547	ended_at: None,
2548	duration_ms: None,
2549	input_summary: input_summary.clone(),
2550	output_summary: None,
2551	detail_path: None,
2552	patch_ref: None,
2553	});
2554	let summary = input_summary
2555	.map(\|s\| format!("{name} started ({s})"))
2556	.unwrap_or_else(\|\| format!("{name} started"));
2557	push_timeline_entry(
2558	task,
2559	TaskTimelineEntry {
2560	timestamp: Utc::now(),
2561	kind: "tool_started".to_string(),
2562	summary,
2563	detail_path: None,
2564	},
2565	);
2566	}
2567	TaskExecutionEvent::ToolProgress { id, output } => {
2568	push_timeline_entry(
2569	task,
2570	TaskTimelineEntry {
2571	timestamp: Utc::now(),
2572	kind: "tool_progress".to_string(),
2573	summary: format!(
2574	"{id}: {}",
2575	summarize_text(&output, TIMELINE_SUMMARY_LIMIT.saturating_sub(8))
2576	),
2577	detail_path: None,
2578	},
2579	);
2580	}
2581	TaskExecutionEvent::ToolCompleted {
2582	id,
2583	name,
2584	success,
2585	output,
2586	metadata,
2587	} => {
2588	let now = Utc::now();
2589	let detail_path = self.artifact_if_large(&task_id, &name, &output)?;
2590	let output_summary = summarize_text(&output, TIMELINE_SUMMARY_LIMIT);
2591	let patch_ref = if name == "apply_patch" {
2592	detail_path.clone()
2593	} else {
2594	None
2595	};
2596
2597	if let Some(call) = task.tool_calls.iter_mut().find(\|call\| call.id == id) {
2598	call.status = if success {
2599	TaskToolStatus::Success
2600	} else {
2601	TaskToolStatus::Failed
2602	};
2603	call.ended_at = Some(now);
2604	call.duration_ms = Some(duration_ms(call.started_at, now));
2605	call.output_summary = Some(output_summary.clone());
2606	call.detail_path = detail_path.clone();
2607	call.patch_ref = patch_ref.clone();
2608
2609	if call.duration_ms.is_none()
2610	&& let Some(duration) = metadata
2611	.as_ref()
2612	.and_then(\|m\| m.get("duration_ms"))
2613	.and_then(Value::as_u64)
2614	{
2615	call.duration_ms = Some(duration);
2616	}
2617	}
2618
2619	let status = if success { "success" } else { "failed" };
2620	push_timeline_entry(
2621	task,
2622	TaskTimelineEntry {
2623	timestamp: now,
2624	kind: "tool_completed".to_string(),
2625	summary: format!("{name} {status}: {output_summary}"),
2626	detail_path: detail_path.clone(),
2627	},
2628	);
2629	if let Some(patch_ref) = patch_ref {
2630	push_timeline_entry(
2631	task,
2632	TaskTimelineEntry {
2633	timestamp: now,
2634	kind: "patch_ref".to_string(),
2635	summary: format!("Patch artifact: {}", patch_ref.display()),
2636	detail_path: Some(patch_ref),
2637	},
2638	);
2639	}
2640
2641	self.apply_task_update_metadata(task, metadata.as_ref())?;
2642	}
2643	TaskExecutionEvent::Error { message } => {
2644	push_timeline_entry(
2645	task,
2646	TaskTimelineEntry {
2647	timestamp: Utc::now(),
2648	kind: "error".to_string(),
2649	summary: summarize_text(&message, TIMELINE_SUMMARY_LIMIT),
2650	detail_path: None,
2651	},
2652	);
2653	}
2654	TaskExecutionEvent::RuntimeEvent {
2655	seq,
2656	event,
2657	summary,
2658	} => {
2659	task.runtime_event_count = task.runtime_event_count.saturating_add(1);
2660	push_timeline_entry(
2661	task,
2662	TaskTimelineEntry {
2663	timestamp: Utc::now(),
2664	kind: "runtime_event".to_string(),
2665	summary: format!("#{seq} {event}: {summary}"),
2666	detail_path: None,
2667	},
2668	);
2669	}
2670	}
2671
2672	Ok(())
2673	}
2674
2675	async fn flush_task(&self, task_id: &str) -> Result<()> {
2676	let mut state = self.state.lock().await;
2677	let _transaction = self.lock_store().await?;
2678	self.refresh_locked(&mut state)?;
2679	let task = state
2680	.tasks
2681	.get(task_id)
2682	.context("Flushed task is missing")?;
2683	self.require_execution_owner(task)?;
2684	self.persist_changed_task_locked(&mut state, task_id)
2685	}
2686
2687	async fn finish_task(
2688	&self,
2689	task_id: &str,
2690	mut result: TaskExecutionResult,
2691	cancel: CancellationToken,
2692	mode_label: &str,
2693	) -> Result<()> {
2694	let mut state = self.state.lock().await;
2695	let _transaction = self.lock_store().await?;
2696	self.refresh_locked(&mut state)?;
2697	state.running_cancel.remove(task_id);
2698	let task = state
2699	.tasks
2700	.get_mut(task_id)
2701	.context("Finished task is missing")?;
2702	self.require_execution_owner(task)?;
2703
2704	let now = Utc::now();
2705	if (cancel.is_cancelled() \|\| task.cancel_requested_seq > 0)
2706	&& result.status == TaskStatus::Completed
2707	{
2708	result.status = TaskStatus::Canceled;
2709	result.result_text = None;
2710	result.error = None;
2711	result.terminal_reason = TaskTerminalReason::Canceled;
2712	}
2713	if self.cancel_token.is_cancelled()
2714	&& result.status != TaskStatus::Completed
2715	&& matches!(
2716	result.terminal_reason,
2717	TaskTerminalReason::Canceled \| TaskTerminalReason::Failed
2718	)
2719	{
2720	result.status = TaskStatus::Canceled;
2721	result.terminal_reason = TaskTerminalReason::Shutdown;
2722	result.error = Some(TaskTerminalReason::Shutdown.receipt_message());
2723	}
2724
2725	task.status = result.status;
2726	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
2727	task.mode = mode_label.to_string();
2728	task.ended_at = Some(now);
2729	task.duration_ms = task.started_at.map(\|start\| duration_ms(start, now));
2730	task.error = result.error.clone();
2731	task.terminal_reason = Some(result.terminal_reason.as_str().to_string());
2732	let finished_summary = if matches!(result.status, TaskStatus::Queued \| TaskStatus::Running)
2733	{
2734	format!("Task ended in unexpected state: {mode_label}")
2735	} else {
2736	match result.terminal_reason {
2737	TaskTerminalReason::Completed
2738	\| TaskTerminalReason::Canceled
2739	\| TaskTerminalReason::Shutdown => result.terminal_reason.receipt_message(),
2740	TaskTerminalReason::Failed
2741	\| TaskTerminalReason::WallTimeout
2742	\| TaskTerminalReason::IdleTimeout
2743	\| TaskTerminalReason::CancelTimeout => format!(
2744	"{}: {}",
2745	result.terminal_reason.as_str(),
2746	result
2747	.error
2748	.as_deref()
2749	.map(\|e\| summarize_text(e, TIMELINE_SUMMARY_LIMIT))
2750	.unwrap_or_else(\|\| result.terminal_reason.receipt_message())
2751	),
2752	}
2753	};
2754	push_timeline_entry(
2755	task,
2756	TaskTimelineEntry {
2757	timestamp: now,
2758	kind: "finished".to_string(),
2759	summary: finished_summary,
2760	detail_path: None,
2761	},
2762	);
2763
2764	if let Some(text) = result.result_text {
2765	let detail_path = self.artifact_if_large(task_id, "result", &text)?;
2766	task.result_summary = Some(summarize_text(&text, TIMELINE_SUMMARY_LIMIT));
2767	task.result_detail_path = detail_path.clone();
2768	if let Some(detail_path) = detail_path {
2769	push_timeline_entry(
2770	task,
2771	TaskTimelineEntry {
2772	timestamp: now,
2773	kind: "result_ref".to_string(),
2774	summary: format!("Result artifact: {}", detail_path.display()),
2775	detail_path: Some(detail_path),
2776	},
2777	);
2778	}
2779	} else if result.status == TaskStatus::Completed {
2780	task.result_summary = Some("(no textual output)".to_string());
2781	}
2782
2783	self.persist_changed_task_locked(&mut state, task_id)?;
2784	Ok(())
2785	}
2786
2787	fn artifact_if_large(
2788	&self,
2789	task_id: &str,
2790	label: &str,
2791	content: &str,
2792	) -> Result<Option<PathBuf>> {
2793	if content.len() < ARTIFACT_THRESHOLD {
2794	return Ok(None);
2795	}
2796	self.write_artifact(task_id, label, content).map(Some)
2797	}
2798
2799	fn write_artifact(&self, task_id: &str, label: &str, content: &str) -> Result<PathBuf> {
2800	ensure_safe_storage_id("task id", task_id)?;
2801	let artifact_dir = self.artifacts_dir.join(task_id);
2802	fs::create_dir_all(&artifact_dir)
2803	.with_context(\|\| format!("Failed to create artifact dir {}", artifact_dir.display()))?;
2804	let stamp = Utc::now().format("%Y%m%dT%H%M%S%.3fZ");
2805	let filename = format!("{stamp}_{}.txt", sanitize_filename(label));
2806	let absolute = artifact_dir.join(filename);
2807	fs::write(&absolute, content)
2808	.with_context(\|\| format!("Failed to write artifact {}", absolute.display()))?;
2809	let relative = absolute
2810	.strip_prefix(&self.cfg.data_dir)
2811	.map(PathBuf::from)
2812	.unwrap_or(absolute);
2813	Ok(relative)
2814	}
2815
2816	fn apply_task_update_metadata(
2817	&self,
2818	task: &mut TaskRecord,
2819	metadata: Option<&Value>,
2820	) -> Result<()> {
2821	let Some(updates) = metadata.and_then(\|m\| m.get("task_updates")) else {
2822	return Ok(());
2823	};
2824	let now = Utc::now();
2825
2826	if let Some(value) = updates.get("checklist") {
2827	let mut checklist: TaskChecklistState = serde_json::from_value(value.clone())
2828	.context("Failed to parse checklist task update")?;
2829	checklist.updated_at = checklist.updated_at.or(Some(now));
2830	task.checklist = checklist;
2831	push_timeline_entry(
2832	task,
2833	TaskTimelineEntry {
2834	timestamp: now,
2835	kind: "checklist".to_string(),
2836	summary: format!(
2837	"Checklist updated: {} item(s), {}% complete",
2838	task.checklist.items.len(),
2839	task.checklist.completion_pct
2840	),
2841	detail_path: None,
2842	},
2843	);
2844	}
2845
2846	if let Some(value) = updates.get("gate") {
2847	let gate: TaskGateRecord = serde_json::from_value(value.clone())
2848	.context("Failed to parse gate task update")?;
2849	let summary = format!("Gate {} {}: {}", gate.gate, gate.status, gate.summary);
2850	task.gates.retain(\|existing\| existing.id != gate.id);
2851	task.gates.push(gate.clone());
2852	push_timeline_entry(
2853	task,
2854	TaskTimelineEntry {
2855	timestamp: now,
2856	kind: "gate".to_string(),
2857	summary: summarize_text(&summary, TIMELINE_SUMMARY_LIMIT),
2858	detail_path: gate.log_path,
2859	},
2860	);
2861	}
2862
2863	if let Some(value) = updates.get("attempt") {
2864	let attempt: TaskAttemptRecord = serde_json::from_value(value.clone())
2865	.context("Failed to parse attempt task update")?;
2866	task.attempts.retain(\|existing\| existing.id != attempt.id);
2867	task.attempts.push(attempt.clone());
2868	push_timeline_entry(
2869	task,
2870	TaskTimelineEntry {
2871	timestamp: now,
2872	kind: "pr_attempt".to_string(),
2873	summary: format!(
2874	"Attempt {}/{} recorded for {}",
2875	attempt.attempt_index, attempt.attempt_count, attempt.attempt_group_id
2876	),
2877	detail_path: attempt.patch_path,
2878	},
2879	);
2880	}
2881
2882	if let Some(value) = updates.get("artifacts")
2883	&& let Some(items) = value.as_array()
2884	{
2885	for item in items {
2886	let artifact: TaskArtifactRef = serde_json::from_value(item.clone())
2887	.context("Failed to parse artifact task update")?;
2888	push_timeline_entry(
2889	task,
2890	TaskTimelineEntry {
2891	timestamp: now,
2892	kind: "artifact".to_string(),
2893	summary: format!("{}: {}", artifact.label, artifact.summary),
2894	detail_path: Some(artifact.path.clone()),
2895	},
2896	);
2897	task.artifacts.push(artifact);
2898	}
2899	}
2900
2901	if let Some(value) = updates.get("github_event") {
2902	let event: TaskGithubEvent = serde_json::from_value(value.clone())
2903	.context("Failed to parse GitHub task update")?;
2904	push_timeline_entry(
2905	task,
2906	TaskTimelineEntry {
2907	timestamp: now,
2908	kind: "github".to_string(),
2909	summary: format!(
2910	"{} {}#{}: {}",
2911	event.action, event.target, event.number, event.summary
2912	),
2913	detail_path: None,
2914	},
2915	);
2916	task.github_events.push(event);
2917	}
2918
2919	Ok(())
2920	}
2921
2922	/// Acquire the cross-process task-store lock.
2923	///
2924	/// Polls with exponential backoff (5ms → 50ms) instead of a flat 5ms
2925	/// interval: under contention the old shape woke ~200×/s for up to its
2926	/// whole five-second deadline (#6211 R7c). The deadline and the busy
2927	/// error are unchanged. What this does not do: it does not add the
2928	/// in-process mutex the issue also suggested — in-process contenders
2929	/// just back off against the same file lock.
2930	async fn lock_store(&self) -> Result<RuntimeProcessOwnerLock> {
2931	let path = self.cfg.data_dir.join("task-store.lock");
2932	let deadline = Instant::now() + Duration::from_secs(5);
2933	let mut wait = Duration::from_millis(5);
2934	loop {
2935	if let Some(owner) = RuntimeProcessOwnerLock::try_acquire_file(&path, true)? {
2936	return Ok(owner);
2937	}
2938	if Instant::now() >= deadline {
2939	bail!("Task store is busy; state is unavailable");
2940	}
2941	sleep(wait).await;
2942	wait = (wait * 2).min(Duration::from_millis(50));
2943	}
2944	}
2945
2946	fn refresh_locked(&self, state: &mut ManagerState) -> Result<()> {
2947	let loaded = load_state(&self.tasks_dir, &self.queue_path)?;
2948	state.tasks = loaded.tasks;
2949	state.queue = loaded.queue;
2950	for (id, events) in &state.pending_events {
2951	let task = state
2952	.tasks
2953	.get_mut(id)
2954	.context("Pending task disappeared")?;
2955	self.require_execution_owner(task)?;
2956	for event in events {
2957	self.apply_event_to_task(task, event.clone())?;
2958	}
2959	}
2960	Ok(())
2961	}
2962
2963	fn require_execution_owner(&self, task: &TaskRecord) -> Result<()> {
2964	if task.execution_scope.as_deref() != Some(self.execution_scope())
2965	\|\| task.execution_generation.as_deref() != Some(&self.execution_lease.generation)
2966	\|\| task.status != TaskStatus::Running
2967	{
2968	bail!("Task execution ownership changed; refusing a stale write");
2969	}
2970	Ok(())
2971	}
2972
2973	fn persist_changed_task_locked(&self, state: &mut ManagerState, id: &str) -> Result<()> {
2974	let task = state.tasks.get(id).context("Changed task is missing")?;
2975	self.persist_task_locked(task)?;
2976	state.pending_events.remove(id);
2977	Ok(())
2978	}
2979
2980	fn recover_dead_executions_locked(&self, state: &mut ManagerState) -> Result<()> {
2981	for task in state.tasks.values_mut() {
2982	// Unknown legacy ownership is preserved, never guessed from the
2983	// visibility owner, model spelling, or current process defaults.
2984	if task.status != TaskStatus::Running {
2985	continue;
2986	}
2987	let (Some(scope), Some(generation)) =
2988	(&task.execution_scope, &task.execution_generation)
2989	else {
2990	continue;
2991	};
2992	let path = execution_lease_path(&self.cfg.data_dir, scope, generation)?;
2993	let Some(_dead_owner) = RuntimeProcessOwnerLock::try_acquire_file(&path, false)? else {
2994	continue;
2995	};
2996	let now = Utc::now();
2997	let duration_ms = task.started_at.and_then(\|started\| {
2998	u64::try_from(now.signed_duration_since(started).num_milliseconds()).ok()
2999	});
3000	task.status = TaskStatus::Failed;
3001	task.lifecycle_seq = task.lifecycle_seq.saturating_add(1);
3002	task.ended_at = Some(now);
3003	task.duration_ms = duration_ms;
3004	task.terminal_reason = Some(TaskTerminalReason::Failed.as_str().to_string());
3005	task.error =
3006	Some("Interrupted by process restart; prior process is not attached".to_string());
3007	for tool in &mut task.tool_calls {
3008	if tool.status == TaskToolStatus::Running {
3009	tool.status = TaskToolStatus::Failed;
3010	tool.ended_at = Some(now);
3011	tool.duration_ms = duration_ms.or_else(\|\| {
3012	u64::try_from(
3013	now.signed_duration_since(tool.started_at)
3014	.num_milliseconds(),
3015	)
3016	.ok()
3017	});
3018	}
3019	}
3020	push_timeline_entry(
3021	task,
3022	TaskTimelineEntry {
3023	timestamp: now,
3024	kind: "recovered".to_string(),
3025	summary: "Interrupted by process restart; prior process is not attached"
3026	.to_string(),
3027	detail_path: None,
3028	},
3029	);
3030
3031	self.persist_task_locked(task)?;
3032	}
3033	Ok(())
3034	}
3035
3036	fn persist_queue_locked(&self, queue: &VecDeque<String>) -> Result<()> {
3037	write_json_atomic(
3038	&self.queue_path,
3039	&QueueFile {
3040	queue: queue.iter().cloned().collect(),
3041	},
3042	)
3043	}
3044
3045	fn persist_task_locked(&self, task: &TaskRecord) -> Result<()> {
3046	let path = self.tasks_dir.join(format!("{}.json", task.id));
3047	write_json_atomic(&path, task)
3048	}
3049	}
3050
3051	fn validate_preallocated_task_id(task_id: &str) -> Result<()> {
3052	if task_id.len() != 21
3053	\|\| !task_id.starts_with("task_")
3054	\|\| !task_id[5..].chars().all(\|ch\| ch.is_ascii_hexdigit())
3055	{
3056	bail!("Invalid preallocated task id: expected task_<16hex>");
3057	}
3058	Ok(())
3059	}
3060
3061	fn read_bound_task_file(path: &Path, task_id: &str) -> Result<Option<TaskRecord>> {
3062	let bytes = match fs::read(path) {
3063	Ok(bytes) => bytes,
3064	Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
3065	Err(error) => return Err(error).context("read bound task"),
3066	};
3067	let task: TaskRecord = serde_json::from_slice(&bytes).context("decode bound task")?;
3068	if task.id != task_id \|\| task.schema_version > CURRENT_TASK_SCHEMA_VERSION {
3069	bail!("Bound task identity or schema does not match its durable admission");
3070	}
3071	Ok(Some(task))
3072	}
3073
3074	pub(crate) fn validate_bound_task_request(
3075	task: &TaskRecord,
3076	request: &NewTaskRequest,
3077	) -> Result<()> {
3078	if task.prompt != request.prompt.trim()
3079	\|\| task.owner_session_id != request.owner_session_id
3080	\|\| task.model_provider != request.model_provider
3081	\|\| task.model_provider_id != request.model_provider_id
3082	\|\| request
3083	.model
3084	.as_ref()
3085	.is_some_and(\|value\| value != &task.model)
3086	\|\| request
3087	.workspace
3088	.as_ref()
3089	.is_some_and(\|value\| value != &task.workspace)
3090	\|\| request
3091	.mode
3092	.as_ref()
3093	.is_some_and(\|value\| value != &task.mode)
3094	\|\| request
3095	.allow_shell
3096	.is_some_and(\|value\| value != task.allow_shell)
3097	\|\| request
3098	.trust_mode
3099	.is_some_and(\|value\| value != task.trust_mode)
3100	\|\| task.auto_approve != request.auto_approve.unwrap_or(false)
3101	{
3102	bail!("Task admission replay does not match the bound request");
3103	}
3104	Ok(())
3105	}
3106
3107	/// A read-only inventory and reconstructed queue. Execution recovery is a
3108	/// separate transaction requiring proof that a particular generation died.
3109	struct LoadedTaskState {
3110	tasks: HashMap<String, TaskRecord>,
3111	queue: VecDeque<String>,
3112	}
3113
3114	fn load_state(tasks_dir: &Path, queue_path: &Path) -> Result<LoadedTaskState> {
3115	let mut tasks = HashMap::new();
3116	if tasks_dir.exists() {
3117	for entry in fs::read_dir(tasks_dir)
3118	.with_context(\|\| format!("Failed to read tasks dir {}", tasks_dir.display()))?
3119	{
3120	let entry = entry?;
3121	let path = entry.path();
3122	if path.extension().is_none_or(\|ext\| ext != "json") {
3123	continue;
3124	}
3125	let content = fs::read_to_string(&path)
3126	.with_context(\|\| format!("Failed to read task file {}", path.display()))?;
3127	let task: TaskRecord = serde_json::from_str(&content)
3128	.with_context(\|\| format!("Failed to parse task file {}", path.display()))?;
3129	if task.schema_version > CURRENT_TASK_SCHEMA_VERSION {
3130	bail!(
3131	"Task schema v{} is newer than supported v{}",
3132	task.schema_version,
3133	CURRENT_TASK_SCHEMA_VERSION
3134	);
3135	}
3136	ensure_safe_storage_id("task id", &task.id)?;
3137	if path.file_stem().and_then(\|stem\| stem.to_str()) != Some(task.id.as_str()) {
3138	bail!("Task record identity differs from its path");
3139	}
3140	if let Some(scope) = &task.execution_scope {
3141	validate_execution_id(scope, 64)?;
3142	}
3143	if let Some(generation) = &task.execution_generation {
3144	validate_execution_id(generation, 32)?;
3145	if task.execution_scope.is_none() {
3146	bail!("Task generation has no execution scope");
3147	}
3148	}
3149	tasks.insert(task.id.clone(), task);
3150	}
3151	}
3152
3153	let mut queue = if queue_path.exists() {
3154	let content = fs::read_to_string(queue_path)
3155	.with_context(\|\| format!("Failed to read queue file {}", queue_path.display()))?;
3156	let parsed: QueueFile = serde_json::from_str(&content)
3157	.with_context(\|\| format!("Failed to parse queue file {}", queue_path.display()))?;
3158	VecDeque::from(parsed.queue)
3159	} else {
3160	VecDeque::new()
3161	};
3162
3163	queue.retain(\|id\| {
3164	tasks
3165	.get(id)
3166	.is_some_and(\|task\| task.status == TaskStatus::Queued)
3167	});
3168
3169	let known = queue.iter().cloned().collect::<HashSet<_>>();
3170	let mut missing = tasks
3171	.values()
3172	.filter(\|task\| task.status == TaskStatus::Queued && !known.contains(&task.id))
3173	.map(\|task\| task.id.clone())
3174	.collect::<Vec<_>>();
3175	missing.sort();
3176	for id in missing {
3177	queue.push_back(id);
3178	}
3179
3180	Ok(LoadedTaskState { tasks, queue })
3181	}
3182
3183	struct EventApplyOutcome {
3184	persisted: bool,
3185	}
3186
3187	fn execution_event_is_progress(event: &TaskExecutionEvent) -> bool {
3188	matches!(
3189	event,
3190	TaskExecutionEvent::MessageDelta { .. }
3191	\| TaskExecutionEvent::ToolStarted { .. }
3192	\| TaskExecutionEvent::ToolProgress { .. }
3193	\| TaskExecutionEvent::ToolCompleted { .. }
3194	)
3195	}
3196
3197	fn execution_event_persist_urgent(event: &TaskExecutionEvent) -> bool {
3198	!matches!(
3199	event,
3200	TaskExecutionEvent::MessageDelta { .. }
3201	\| TaskExecutionEvent::ToolProgress { .. }
3202	\| TaskExecutionEvent::RuntimeEvent { .. }
3203	)
3204	}
3205
3206	fn timeline_kinds_coalesce(left: &str, right: &str) -> bool {
3207	matches!(
3208	(left, right),
3209	("message", "message")
3210	\| ("tool_progress", "tool_progress")
3211	\| ("runtime_event", "runtime_event")
3212	)
3213	}
3214
3215	fn push_timeline_entry(task: &mut TaskRecord, entry: TaskTimelineEntry) {
3216	if let Some(last) = task.timeline.last_mut()
3217	&& timeline_kinds_coalesce(last.kind.as_str(), entry.kind.as_str())
3218	{
3219	last.timestamp = entry.timestamp;
3220	last.summary = entry.summary;
3221	last.detail_path = entry.detail_path;
3222	return;
3223	}
3224	task.timeline.push(entry);
3225	trim_task_timeline(&mut task.timeline);
3226	}
3227
3228	fn trim_task_timeline(entries: &mut Vec<TaskTimelineEntry>) {
3229	if entries.len() <= TIMELINE_ENTRY_LIMIT {
3230	return;
3231	}
3232	let overflow = entries.len() - TIMELINE_ENTRY_LIMIT;
3233	let start = TIMELINE_HEAD_KEEP.min(entries.len().saturating_sub(overflow + 1));
3234	let end = start + overflow;
3235	if start >= end \|\| end > entries.len() {
3236	entries.truncate(TIMELINE_ENTRY_LIMIT);
3237	return;
3238	}
3239	entries.drain(start..end);
3240	let omitted = TaskTimelineEntry {
3241	timestamp: Utc::now(),
3242	kind: "omitted".to_string(),
3243	summary: format!("{overflow} earlier events omitted to bound storage"),
3244	detail_path: None,
3245	};
3246	if entries
3247	.get(start)
3248	.is_none_or(\|entry\| entry.kind != "omitted")
3249	{
3250	entries.insert(start, omitted);
3251	}
3252	if entries.len() > TIMELINE_ENTRY_LIMIT {
3253	let extra = entries.len() - TIMELINE_ENTRY_LIMIT;
3254	let drop_at = (start + 1).min(entries.len().saturating_sub(1));
3255	let drop_end = (drop_at + extra).min(entries.len());
3256	if drop_at < drop_end {
3257	entries.drain(drop_at..drop_end);
3258	} else {
3259	entries.truncate(TIMELINE_ENTRY_LIMIT);
3260	}
3261	}
3262	}
3263
3264	fn resolve_task_id_visible_to(
3265	tasks: &HashMap<String, TaskRecord>,
3266	id_or_prefix: &str,
3267	owner_session_id: Option<&str>,
3268	) -> Result<String> {
3269	let visible = \|record: &TaskRecord\| {
3270	owner_session_id.is_none_or(\|owner_session_id\| {
3271	record.owner_session_id.as_deref() == Some(owner_session_id)
3272	})
3273	};
3274	if tasks.get(id_or_prefix).is_some_and(visible) {
3275	return Ok(id_or_prefix.to_string());
3276	}
3277	let matches = tasks
3278	.iter()
3279	.filter(\|(id, record)\| id.starts_with(id_or_prefix) && visible(record))
3280	.map(\|(id, _)\| id)
3281	.cloned()
3282	.collect::<Vec<_>>();
3283	match matches.len() {
3284	0 => bail!("Task not found: {id_or_prefix}"),
3285	1 => Ok(matches[0].clone()),
3286	_ => bail!(
3287	"Ambiguous task prefix '{}': matches {} tasks",
3288	id_or_prefix,
3289	matches.len()
3290	),
3291	}
3292	}
3293
3294	fn resolve_task_id_visible_to_operator(
3295	tasks: &HashMap<String, TaskRecord>,
3296	id_or_prefix: &str,
3297	owner_session_id: &str,
3298	execution_scope: &str,
3299	) -> Result<String> {
3300	let visible = \|record: &TaskRecord\| {
3301	record.owner_session_id.as_deref() == Some(owner_session_id)
3302	\|\| record.owner_session_id.is_none()
3303	&& !execution_scope.is_empty()
3304	&& record.execution_scope.as_deref() == Some(execution_scope)
3305	};
3306	if tasks.get(id_or_prefix).is_some_and(visible) {
3307	return Ok(id_or_prefix.to_string());
3308	}
3309	let matches = tasks
3310	.iter()
3311	.filter(\|(id, record)\| id.starts_with(id_or_prefix) && visible(record))
3312	.map(\|(id, _)\| id)
3313	.cloned()
3314	.collect::<Vec<_>>();
3315	match matches.len() {
3316	0 => bail!("Task not found: {id_or_prefix}"),
3317	1 => Ok(matches[0].clone()),
3318	_ => bail!(
3319	"Ambiguous task prefix '{}': matches {} tasks",
3320	id_or_prefix,
3321	matches.len()
3322	),
3323	}
3324	}
3325
3326	fn resolve_task_id(tasks: &HashMap<String, TaskRecord>, id_or_prefix: &str) -> Result<String> {
3327	resolve_task_id_visible_to(tasks, id_or_prefix, None)
3328	}
3329
3330	fn summarize_json(value: &Value) -> Option<String> {
3331	let text = serde_json::to_string(value).ok()?;
3332	Some(summarize_text(&text, TIMELINE_SUMMARY_LIMIT))
3333	}
3334
3335	fn summarize_text(text: &str, limit: usize) -> String {
3336	let take = limit.saturating_sub(3);
3337	let mut count = 0;
3338	let mut out = String::new();
3339	for ch in text.chars() {
3340	if count >= take {
3341	out.push_str("...");
3342	return out;
3343	}
3344	if ch.is_control() && ch != '\n' && ch != '\t' {
3345	continue;
3346	}
3347	out.push(ch);
3348	count += 1;
3349	}
3350	out
3351	}
3352
3353	fn ensure_safe_storage_id(kind: &str, value: &str) -> Result<()> {
3354	let mut components = Path::new(value).components();
3355	let Some(component) = components.next() else {
3356	bail!("{kind} must not be empty");
3357	};
3358	if components.next().is_some() \|\| !matches!(component, std::path::Component::Normal(_)) {
3359	bail!("{kind} must be a single path component");
3360	}
3361	Ok(())
3362	}
3363
3364	fn sanitize_filename(input: &str) -> String {
3365	let mut out = String::new();
3366	for ch in input.chars() {
3367	if ch.is_ascii_alphanumeric() \|\| ch == '_' \|\| ch == '-' {
3368	out.push(ch);
3369	} else {
3370	out.push('_');
3371	}
3372	}
3373	if out.is_empty() {
3374	"artifact".to_string()
3375	} else {
3376	out
3377	}
3378	}
3379
3380	fn duration_ms(start: DateTime<Utc>, end: DateTime<Utc>) -> u64 {
3381	let millis = (end - start).num_milliseconds();
3382	if millis.is_negative() {
3383	0
3384	} else {
3385	u64::try_from(millis).unwrap_or(u64::MAX)
3386	}
3387	}
3388
3389	fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> {
3390	if let Some(parent) = path.parent() {
3391	fs::create_dir_all(parent)
3392	.with_context(\|\| format!("Failed to create directory {}", parent.display()))?;
3393	}
3394	let payload = serde_json::to_string_pretty(value)?;
3395	crate::utils::write_atomic(path, payload.as_bytes())
3396	.with_context(\|\| format!("Failed to write {}", path.display()))
3397	}
3398
3399	fn default_auto_approve() -> bool {
3400	true
3401	}
3402
3403	/// Default task manager data location (`~/.codewhale/tasks`, or legacy
3404	/// `~/.deepseek/tasks` when only the legacy directory exists).
3405	#[must_use]
3406	pub fn default_tasks_dir() -> PathBuf {
3407	for var in ["CODEWHALE_TASKS_DIR", "DEEPSEEK_TASKS_DIR"] {
3408	if let Ok(path) = std::env::var(var)
3409	&& !path.trim().is_empty()
3410	{
3411	return PathBuf::from(path);
3412	}
3413	}
3414	if let Some(home) = codewhale_paths::codewhale_home_override().ok().flatten() {
3415	return home.join("tasks");
3416	}
3417	codewhale_paths::user_home()
3418	.map(\|home\| default_tasks_dir_for_home(&home))
3419	.unwrap_or_else(\|\| PathBuf::from(".codewhale").join("tasks"))
3420	}
3421
3422	fn default_tasks_dir_for_home(home: &Path) -> PathBuf {
3423	let primary = home.join(".codewhale").join("tasks");
3424	if primary.is_dir() {
3425	return primary;
3426	}
3427	let legacy = home.join(".deepseek").join("tasks");
3428	if legacy.is_dir() {
3429	return legacy;
3430	}
3431	primary
3432	}
3433
3434	/// Wait for a task to reach a terminal status (tests and API helpers).
3435	#[cfg(test)]
3436	pub async fn wait_for_terminal_state(
3437	manager: &TaskManager,
3438	task_id: &str,
3439	timeout: StdDuration,
3440	) -> Result<TaskRecord> {
3441	let deadline = std::time::Instant::now() + timeout;
3442	loop {
3443	let task = manager.get_task(task_id).await?;
3444	if task.status.is_terminal() {
3445	return Ok(task);
3446	}
3447	if std::time::Instant::now() >= deadline {
3448	bail!("Timed out waiting for task {task_id}");
3449	}
3450	sleep(StdDuration::from_millis(50)).await;
3451	}
3452	}
3453
3454	#[cfg(test)]
3455	mod tests {
3456	use super::*;
3457	use crate::test_support::{EnvVarGuard, lock_test_env};
3458	use std::fs;
3459	use std::sync::atomic::{AtomicUsize, Ordering};
3460	use tokio::time::Duration;
3461
3462	struct MockExecutor;
3463
3464	fn provider_default_model_cases() -> Vec<(&'static str, Config, &'static str)> {
3465	let deepseek = Config {
3466	provider: Some("deepseek".to_string()),
3467	default_text_model: Some("deepseek-v4-flash".to_string()),
3468	..Config::default()
3469	};
3470
3471	let zai = Config {
3472	provider: Some("zai".to_string()),
3473	// Exercise provider-aware rejection of a stale DeepSeek root default.
3474	default_text_model: Some(crate::config::DEFAULT_TEXT_MODEL.to_string()),
3475	..Config::default()
3476	};
3477
3478	let mut custom_providers = crate::config::ProvidersConfig::default();
3479	custom_providers.custom.insert(
3480	"acme".to_string(),
3481	crate::config::ProviderConfig {
3482	base_url: Some("http://127.0.0.1:1/v1".to_string()),
3483	model: Some("acme-coder".to_string()),
3484	kind: Some("openai-compatible".to_string()),
3485	..crate::config::ProviderConfig::default()
3486	},
3487	);
3488	let custom = Config {
3489	provider: Some("acme".to_string()),
3490	providers: Some(custom_providers),
3491	..Config::default()
3492	};
3493
3494	vec![
3495	("deepseek", deepseek, "deepseek-v4-flash"),
3496	("zai", zai, crate::config::DEFAULT_ZAI_MODEL),
3497	("custom", custom, "acme-coder"),
3498	]
3499	}
3500
3501	#[test]
3502	fn task_manager_config_uses_the_active_provider_default() {
3503	for (label, config, expected) in provider_default_model_cases() {
3504	let task_config =
3505	TaskManagerConfig::from_runtime(&config, PathBuf::from("."), None, Some(1));
3506	assert_eq!(
3507	task_config.default_model, expected,
3508	"{label} durable task default"
3509	);
3510	}
3511	}
3512
3513	#[async_trait]
3514	impl TaskExecutor for MockExecutor {
3515	async fn execute(
3516	&self,
3517	task: ExecutionTask,
3518	events: mpsc::Sender<TaskExecutionEvent>,
3519	cancel: CancellationToken,
3520	) -> TaskExecutionResult {
3521	let _ = events
3522	.send(TaskExecutionEvent::Status {
3523	message: format!("running {}", task.id),
3524	})
3525	.await;
3526	let _ = events
3527	.send(TaskExecutionEvent::ThreadLinked {
3528	thread_id: "thr_test".to_string(),
3529	turn_id: "turn_test".to_string(),
3530	})
3531	.await;
3532	let _ = events
3533	.send(TaskExecutionEvent::ToolStarted {
3534	id: "tool_1".to_string(),
3535	name: "read_file".to_string(),
3536	input: serde_json::json!({ "path": "README.md" }),
3537	})
3538	.await;
3539	sleep(Duration::from_millis(50)).await;
3540	if cancel.is_cancelled() {
3541	return TaskExecutionResult {
3542	status: TaskStatus::Canceled,
3543	result_text: None,
3544	error: None,
3545	terminal_reason: TaskTerminalReason::Canceled,
3546	};
3547	}
3548	let _ = events
3549	.send(TaskExecutionEvent::ToolCompleted {
3550	id: "tool_1".to_string(),
3551	name: "read_file".to_string(),
3552	success: true,
3553	output: "read ok".to_string(),
3554	metadata: Some(serde_json::json!({
3555	"duration_ms": 10,
3556	"task_updates": {
3557	"checklist": {
3558	"items": [
3559	{ "id": 1, "content": "read fixture", "status": "in_progress" }
3560	],
3561	"completion_pct": 0,
3562	"in_progress_id": 1,
3563	"updated_at": null
3564	}
3565	}
3566	})),
3567	})
3568	.await;
3569	TaskExecutionResult {
3570	status: TaskStatus::Completed,
3571	result_text: Some("done".to_string()),
3572	error: None,
3573	terminal_reason: TaskTerminalReason::Completed,
3574	}
3575	}
3576	}
3577
3578	fn test_config(root: PathBuf) -> TaskManagerConfig {
3579	TaskManagerConfig {
3580	data_dir: root,
3581	worker_count: 1,
3582	default_workspace: PathBuf::from("."),
3583	default_model: "deepseek-v4-flash".to_string(),
3584	default_mode: "agent".to_string(),
3585	allow_shell: false,
3586	trust_mode: false,
3587	execution_limits: TaskExecutionLimits::default(),
3588	}
3589	}
3590
3591	fn short_test_config(root: PathBuf) -> TaskManagerConfig {
3592	TaskManagerConfig {
3593	execution_limits: TaskExecutionLimits::short_for_tests(),
3594	..test_config(root)
3595	}
3596	}
3597
3598	fn wall_timeout_test_config(root: PathBuf) -> TaskManagerConfig {
3599	let mut config = short_test_config(root);
3600	config.execution_limits.idle_progress = config
3601	.execution_limits
3602	.wall_time
3603	.saturating_add(config.execution_limits.cancel_grace);
3604	config
3605	}
3606
3607	#[tokio::test]
3608	async fn persists_and_recovers_task_records() -> Result<()> {
3609	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
3610	let manager =
3611	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
3612	.await?;
3613
3614	let task = manager
3615	.add_task(NewTaskRequest {
3616	owner_session_id: Some("session-persist".to_string()),
3617	..NewTaskRequest::from_prompt("test persistence")
3618	})
3619	.await?;
3620	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
3621	assert_eq!(finished.status, TaskStatus::Completed);
3622	assert_eq!(finished.thread_id.as_deref(), Some("thr_test"));
3623	assert_eq!(finished.turn_id.as_deref(), Some("turn_test"));
3624	assert_eq!(finished.checklist.items.len(), 1);
3625	assert_eq!(finished.checklist.in_progress_id, Some(1));
3626	assert!(
3627	finished.lifecycle_seq >= 3,
3628	"queued, running, and terminal owner transitions must advance the sequence"
3629	);
3630
3631	manager.shutdown_and_wait().await?;
3632	drop(manager);
3633
3634	let recovered =
3635	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
3636	.await?;
3637	let loaded = recovered.get_task(&task.id).await?;
3638	assert_eq!(loaded.status, TaskStatus::Completed);
3639	assert_eq!(
3640	loaded.owner_session_id.as_deref(),
3641	Some("session-persist"),
3642	"session ownership should survive persistence and restart"
3643	);
3644	assert!(!loaded.timeline.is_empty());
3645	assert_eq!(loaded.checklist.items[0].content, "read fixture");
3646	Ok(())
3647	}
3648
3649	struct AdmissionCountingExecutor(Arc<AtomicUsize>);
3650
3651	#[async_trait]
3652	impl TaskExecutor for AdmissionCountingExecutor {
3653	async fn execute(
3654	&self,
3655	_task: ExecutionTask,
3656	_events: mpsc::Sender<TaskExecutionEvent>,
3657	_cancel: CancellationToken,
3658	) -> TaskExecutionResult {
3659	self.0.fetch_add(1, Ordering::SeqCst);
3660	TaskExecutionResult {
3661	status: TaskStatus::Completed,
3662	result_text: Some("admission fixture completed".into()),
3663	error: None,
3664	terminal_reason: TaskTerminalReason::Completed,
3665	}
3666	}
3667	}
3668
3669	#[tokio::test]
3670	async fn interrupted_task_stage_preserves_resolved_request_and_executes_once() -> Result<()> {
3671	for queue_was_written in [false, true] {
3672	let root = tempfile::tempdir()?;
3673	let tasks_dir = root.path().join("tasks");
3674	fs::create_dir_all(&tasks_dir)?;
3675	let mut staged = sample_task_record();
3676	staged.status = TaskStatus::Queued;
3677	staged.started_at = None;
3678	staged.model = "staged-model".into();
3679	staged.workspace = root.path().join("staged-workspace");
3680	fs::create_dir(&staged.workspace)?;
3681	let mut request = NewTaskRequest::from_task(&staged);
3682	request.model = None;
3683	request.workspace = None;
3684	request.mode = None;
3685	request.allow_shell = None;
3686	request.trust_mode = None;
3687	let staged_path = tasks_dir.join(format!(".{}.json.pending", staged.id));
3688	write_json_atomic(&staged_path, &staged)?;
3689	if queue_was_written {
3690	write_json_atomic(
3691	&root.path().join("queue.json"),
3692	&QueueFile {
3693	queue: vec![staged.id.clone()],
3694	},
3695	)?;
3696	}
3697	let executions = Arc::new(AtomicUsize::new(0));
3698	let mut config = test_config(root.path().to_path_buf());
3699	config.default_model = "new-default-model".into();
3700	config.default_workspace = root.path().join("new-workspace");
3701	config.default_mode = "plan".into();
3702	config.allow_shell = true;
3703	config.trust_mode = true;
3704	let manager = TaskManager::start_with_executor(
3705	config,
3706	Arc::new(AdmissionCountingExecutor(executions.clone())),
3707	)
3708	.await?;
3709	// Interrupt recovery while its admission is waiting for the queue
3710	// lock. The resolved intent must remain durable for another retry.
3711	let stage_before = fs::read(&staged_path)?;
3712	let queue_guard = manager.state.lock().await;
3713	let mut recovery =
3714	Box::pin(manager.recover_task_admission(request.clone(), staged.id.clone()));
3715	assert!(
3716	tokio::time::timeout(Duration::from_millis(25), &mut recovery)
3717	.await
3718	.is_err()
3719	);
3720	assert_eq!(
3721	fs::read(&staged_path)?,
3722	stage_before,
3723	"interrupted recovery must preserve its resolved staged intent"
3724	);
3725	assert!(manager.read_bound_task(&staged.id)?.is_none());
3726	assert_eq!(executions.load(Ordering::SeqCst), 0);
3727	drop(recovery);
3728	drop(queue_guard);
3729	let admitted = manager
3730	.recover_task_admission(request.clone(), staged.id.clone())
3731	.await?;
3732	assert_eq!(admitted.id, staged.id);
3733	assert_eq!(admitted.model, staged.model);
3734	assert_eq!(admitted.workspace, staged.workspace);
3735	assert_eq!(admitted.mode, staged.mode);
3736	assert_eq!(admitted.allow_shell, staged.allow_shell);
3737	assert_eq!(admitted.trust_mode, staged.trust_mode);
3738	assert!(
3739	!staged_path.exists(),
3740	"unaccepted stage recovered through TaskManager"
3741	);
3742	let completed =
3743	wait_for_terminal_state(&manager, &admitted.id, Duration::from_secs(5)).await?;
3744	assert_eq!(completed.status, TaskStatus::Completed);
3745	assert!(
3746	completed
3747	.result_summary
3748	.as_deref()
3749	.unwrap_or_default()
3750	.contains("admission fixture completed")
3751	);
3752	assert_eq!(executions.load(Ordering::SeqCst), 1);
3753	let replay = manager
3754	.recover_task_admission(request.clone(), staged.id.clone())
3755	.await?;
3756	assert_eq!(replay.status, TaskStatus::Completed);
3757	assert_eq!(replay.id, staged.id);
3758	let canonical = tasks_dir.join(format!("{}.json", staged.id));
3759	let before = fs::read(&canonical)?;
3760	let mut mismatched = request;
3761	mismatched.prompt = "a different operation".into();
3762	let error = manager
3763	.recover_task_admission(mismatched, staged.id)
3764	.await
3765	.expect_err("mismatched replay must be rejected");
3766	assert!(error.to_string().contains("does not match"));
3767	assert_eq!(
3768	fs::read(canonical)?,
3769	before,
3770	"replay cannot rewrite accepted work"
3771	);
3772	assert_eq!(executions.load(Ordering::SeqCst), 1);
3773	manager.shutdown();
3774	}
3775	Ok(())
3776	}
3777
3778	#[tokio::test]
3779	async fn accepted_task_interrupted_by_restart_is_reconciled_without_execution() -> Result<()> {
3780	let root = tempfile::tempdir()?;
3781	let tasks_dir = root.path().join("tasks");
3782	fs::create_dir_all(&tasks_dir)?;
3783	let mut accepted = sample_task_record();
3784	accepted.execution_generation = Some(Uuid::new_v4().simple().to_string());
3785	let lease_path = execution_lease_path(
3786	root.path(),
3787	accepted.execution_scope.as_deref().unwrap(),
3788	accepted.execution_generation.as_deref().unwrap(),
3789	)?;
3790	drop(
3791	RuntimeProcessOwnerLock::try_acquire_file(&lease_path, true)?
3792	.context("fixture generation")?,
3793	);
3794	let request = NewTaskRequest::from_task(&accepted);
3795	write_json_atomic(&tasks_dir.join(format!("{}.json", accepted.id)), &accepted)?;
3796	write_json_atomic(
3797	&root.path().join("queue.json"),
3798	&QueueFile {
3799	queue: vec![accepted.id.clone()],
3800	},
3801	)?;
3802	let executions = Arc::new(AtomicUsize::new(0));
3803	let manager = TaskManager::start_with_executor(
3804	test_config(root.path().to_path_buf()),
3805	Arc::new(AdmissionCountingExecutor(executions.clone())),
3806	)
3807	.await?;
3808	let recovered = manager
3809	.recover_task_admission(request, accepted.id.clone())
3810	.await?;
3811	assert_eq!(recovered.id, accepted.id);
3812	assert_eq!(recovered.status, TaskStatus::Failed);
3813	assert!(
3814	recovered
3815	.error
3816	.as_deref()
3817	.unwrap_or_default()
3818	.contains("Interrupted by process restart")
3819	);
3820	assert_eq!(manager.list_tasks(None).await?.len(), 1);
3821	assert_eq!(
3822	executions.load(Ordering::SeqCst),
3823	0,
3824	"accepted work cannot be replayed after restart"
3825	);
3826	manager.shutdown();
3827	Ok(())
3828	}
3829
3830	#[tokio::test]
3831	async fn preallocated_task_ids_are_validated_and_collision_safe() -> Result<()> {
3832	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
3833	let manager =
3834	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
3835	let request = NewTaskRequest::from_prompt("preallocated owner identity");
3836
3837	let invalid = manager
3838	.add_task_with_id(request.clone(), "task_short".to_string())
3839	.await
3840	.expect_err("invalid preallocated id");
3841	assert!(invalid.to_string().contains("task_<16hex>"), "{invalid:#}");
3842
3843	let id = "task_0123456789abcdef".to_string();
3844	let created = manager
3845	.add_task_with_id(request.clone(), id.clone())
3846	.await?;
3847	assert_eq!(created.id, id);
3848	assert_eq!(
3849	created.schema_version, CURRENT_TASK_SCHEMA_VERSION,
3850	"execution provenance requires readers that preserve the binding"
3851	);
3852	assert_eq!(created.lifecycle_seq, 1);
3853	let collision = manager
3854	.add_task_with_id(request, id)
3855	.await
3856	.expect_err("task id collision");
3857	assert!(
3858	collision.to_string().contains("already exists"),
3859	"{collision:#}"
3860	);
3861	assert_eq!(manager.list_tasks(None).await?.len(), 1);
3862	Ok(())
3863	}
3864
3865	#[tokio::test]
3866	async fn failed_queue_write_leaves_no_replayable_task_record() -> Result<()> {
3867	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
3868	let manager =
3869	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
3870	.await?;
3871	std::fs::remove_file(root.join("queue.json"))?;
3872	std::fs::create_dir(root.join("queue.json"))?;
3873
3874	let id = "task_fedcba9876543210".to_string();
3875	let error = manager
3876	.add_task_with_id(
3877	NewTaskRequest::from_prompt("must not resurrect"),
3878	id.clone(),
3879	)
3880	.await
3881	.expect_err("queue path directory must reject the atomic queue write");
3882	assert!(error.to_string().contains("queue.json"), "{error:#}");
3883	assert!(
3884	manager.list_tasks(None).await.is_err(),
3885	"unavailable storage cannot be reported as empty"
3886	);
3887	assert!(!root.join("tasks").join(format!("{id}.json")).exists());
3888	assert!(
3889	!root
3890	.join("tasks")
3891	.join(format!(".{id}.json.pending"))
3892	.exists(),
3893	"a failed queue write may leave no replayable or staged task record"
3894	);
3895	Ok(())
3896	}
3897
3898	#[tokio::test]
3899	async fn list_tasks_scopes_results_to_workspace_before_limit() -> Result<()> {
3900	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
3901	let manager =
3902	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
3903
3904	manager
3905	.add_task(NewTaskRequest {
3906	prompt: "task in workspace a".to_string(),
3907	workspace: Some(PathBuf::from("/tmp/workspace-a")),
3908	..NewTaskRequest::from_prompt("task in workspace a")
3909	})
3910	.await?;
3911	manager
3912	.add_task(NewTaskRequest {
3913	prompt: "task in workspace b".to_string(),
3914	workspace: Some(PathBuf::from("/tmp/workspace-b")),
3915	..NewTaskRequest::from_prompt("task in workspace b")
3916	})
3917	.await?;
3918
3919	let scoped = manager
3920	.list_tasks_scoped(Some(1), Some(Path::new("/tmp/workspace-a")))
3921	.await?;
3922	assert_eq!(scoped.len(), 1);
3923	assert_eq!(scoped[0].workspace, PathBuf::from("/tmp/workspace-a"));
3924	Ok(())
3925	}
3926
3927	#[tokio::test]
3928	async fn task_controls_are_session_owned_and_legacy_records_fail_closed() -> Result<()> {
3929	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
3930	let manager =
3931	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
3932
3933	let mut session_a = sample_task_record();
3934	session_a.id = "task_dead000000000001".to_string();
3935	session_a.owner_session_id = Some("session-a".to_string());
3936	session_a.status = TaskStatus::Completed;
3937	session_a.created_at = Utc::now() - chrono::Duration::seconds(3);
3938
3939	let mut session_b = sample_task_record();
3940	session_b.id = "task_dead000000000002".to_string();
3941	session_b.owner_session_id = Some("session-b".to_string());
3942	session_b.status = TaskStatus::Completed;
3943	session_b.created_at = Utc::now() - chrono::Duration::seconds(2);
3944
3945	let mut session_b_newest = sample_task_record();
3946	session_b_newest.id = "task_beef000000000002".to_string();
3947	session_b_newest.owner_session_id = Some("session-b".to_string());
3948	session_b_newest.status = TaskStatus::Completed;
3949	session_b_newest.created_at = Utc::now();
3950
3951	let mut legacy = sample_task_record();
3952	legacy.id = "task_dead000000000003".to_string();
3953	legacy.owner_session_id = None;
3954	legacy.execution_scope = None;
3955	legacy.status = TaskStatus::Completed;
3956	legacy.created_at = Utc::now() - chrono::Duration::seconds(1);
3957
3958	{
3959	let mut state = manager.state.lock().await;
3960	for record in [
3961	session_a.clone(),
3962	session_b.clone(),
3963	session_b_newest.clone(),
3964	legacy.clone(),
3965	] {
3966	manager.persist_task_locked(&record)?;
3967	state.tasks.insert(record.id.clone(), record);
3968	}
3969	}
3970
3971	let session_b_list = manager
3972	.list_tasks_for_owner(Some(1), None, "session-b")
3973	.await?;
3974	assert_eq!(session_b_list.len(), 1);
3975	assert_eq!(session_b_list[0].id, session_b_newest.id);
3976
3977	let session_b_prefix = manager.get_task_for_owner("task_dead", "session-b").await?;
3978	assert_eq!(session_b_prefix.id, session_b.id);
3979	assert!(
3980	manager
3981	.get_task_for_owner(&session_a.id, "session-b")
3982	.await
3983	.unwrap_err()
3984	.to_string()
3985	.contains("Task not found")
3986	);
3987	assert!(
3988	manager
3989	.get_task_for_owner(&legacy.id, "session-b")
3990	.await
3991	.unwrap_err()
3992	.to_string()
3993	.contains("Task not found")
3994	);
3995	assert!(
3996	manager
3997	.get_task_for_active_runtime(&legacy.id)
3998	.await
3999	.unwrap_err()
4000	.to_string()
4001	.contains("Task not found"),
4002	"legacy ownerless active tasks must fail closed"
4003	);
4004	assert_eq!(
4005	manager.get_task_for_active_runtime(&session_a.id).await?.id,
4006	session_a.id
4007	);
4008
4009	assert!(
4010	manager
4011	.cancel_task_for_owner(&session_a.id, "session-b")
4012	.await
4013	.unwrap_err()
4014	.to_string()
4015	.contains("Task not found")
4016	);
4017	assert_eq!(
4018	manager.get_task(&session_a.id).await?.status,
4019	TaskStatus::Completed
4020	);
4021	assert!(
4022	manager
4023	.cancel_task_for_owner(&legacy.id, "session-b")
4024	.await
4025	.unwrap_err()
4026	.to_string()
4027	.contains("Task not found")
4028	);
4029
4030	let own = manager
4031	.cancel_task_for_owner(&session_b.id, "session-b")
4032	.await?;
4033	assert_eq!(own.disposition, TaskCancelDisposition::AlreadyFinished);
4034	let active_own = manager
4035	.cancel_task_for_active_runtime(&session_a.id)
4036	.await?;
4037	assert_eq!(
4038	active_own.disposition,
4039	TaskCancelDisposition::AlreadyFinished
4040	);
4041	assert_eq!(
4042	manager
4043	.get_task_for_owner(&session_a.id, "session-a")
4044	.await?
4045	.id,
4046	session_a.id,
4047	"switching A to B and back must restore A's controls"
4048	);
4049	Ok(())
4050	}
4051
4052	#[tokio::test]
4053	async fn interactive_task_controls_include_only_owned_or_same_scope_records() -> Result<()> {
4054	let root = tempfile::tempdir()?;
4055	let manager = TaskManager::start_with_executor(
4056	test_config(root.path().to_path_buf()),
4057	Arc::new(MockExecutor),
4058	)
4059	.await?;
4060	// Exercise persisted controls without a worker racing to execute the
4061	// queued fixture records. The manager retains its verified scope.
4062	manager.shutdown_and_wait().await?;
4063	let mut scheduled = sample_task_record();
4064	scheduled.id = "task_dead000000000001".to_string();
4065	scheduled.owner_session_id = None;
4066	scheduled.execution_scope = Some(manager.execution_scope().to_string());
4067	scheduled.status = TaskStatus::Queued;
4068
4069	let mut owned = scheduled.clone();
4070	owned.id = "task_beef000000000001".to_string();
4071	owned.owner_session_id = Some("session-a".to_string());
4072	owned.status = TaskStatus::Completed;
4073
4074	let mut foreign_scope = scheduled.clone();
4075	foreign_scope.id = "task_dead000000000002".to_string();
4076	foreign_scope.execution_scope = Some(test_execution_scope("other"));
4077	let mut other_session = scheduled.clone();
4078	other_session.id = "task_dead000000000003".to_string();
4079	other_session.owner_session_id = Some("session-b".to_string());
4080	let mut legacy = scheduled.clone();
4081	legacy.id = "task_dead000000000004".to_string();
4082	legacy.execution_scope = None;
4083	let hidden = [foreign_scope, other_session, legacy];
4084	{
4085	let mut state = manager.state.lock().await;
4086	for record in std::iter::once(&scheduled)
4087	.chain(std::iter::once(&owned))
4088	.chain(hidden.iter())
4089	{
4090	manager.persist_task_locked(record)?;
4091	state.tasks.insert(record.id.clone(), record.clone());
4092	}
4093	}
4094
4095	for record in [&scheduled, &owned] {
4096	assert_eq!(
4097	manager
4098	.get_task_for_interactive_session(&record.id, "session-a")
4099	.await?
4100	.id,
4101	record.id
4102	);
4103	}
4104	// Hidden records sharing this prefix must not make it ambiguous.
4105	assert_eq!(
4106	manager
4107	.get_task_for_interactive_session("task_dead", "session-a")
4108	.await?
4109	.id,
4110	scheduled.id
4111	);
4112	let ambiguous = manager
4113	.get_task_for_interactive_session("task_", "session-a")
4114	.await
4115	.unwrap_err();
4116	assert!(ambiguous.to_string().contains("matches 2 tasks"));
4117	for record in &hidden {
4118	assert!(
4119	manager
4120	.get_task_for_interactive_session(&record.id, "session-a")
4121	.await
4122	.unwrap_err()
4123	.to_string()
4124	.contains("Task not found")
4125	);
4126	assert!(
4127	manager
4128	.cancel_task_for_interactive_session(&record.id, "session-a")
4129	.await
4130	.unwrap_err()
4131	.to_string()
4132	.contains("Task not found")
4133	);
4134	assert_eq!(
4135	manager.get_task(&record.id).await?.status,
4136	TaskStatus::Queued
4137	);
4138	}
4139	// Model and child-session APIs do not inherit the human-only access.
4140	assert!(
4141	manager
4142	.get_task_for_owner(&scheduled.id, "session-a")
4143	.await
4144	.is_err()
4145	);
4146	assert!(
4147	manager
4148	.cancel_task_for_owner(&scheduled.id, "session-a")
4149	.await
4150	.is_err()
4151	);
4152	let canceled = manager
4153	.cancel_task_for_interactive_session("task_dead", "session-a")
4154	.await?;
4155	assert_eq!(canceled.task.id, scheduled.id);
4156	assert_eq!(canceled.task.status, TaskStatus::Canceled);
4157	assert_eq!(
4158	manager.get_task(&scheduled.id).await?.status,
4159	TaskStatus::Canceled
4160	);
4161	manager.shutdown_and_wait().await?;
4162	Ok(())
4163	}
4164
4165	#[test]
4166	fn interactive_task_controls_reject_an_empty_manager_scope() {
4167	let mut record = sample_task_record();
4168	record.owner_session_id = None;
4169	record.execution_scope = Some(String::new());
4170	let id = record.id.clone();
4171	let tasks = HashMap::from([(id.clone(), record)]);
4172	assert!(resolve_task_id_visible_to_operator(&tasks, &id, "session-a", "").is_err());
4173	}
4174
4175	#[tokio::test]
4176	async fn boot_does_not_rewrite_non_recovered_task_files() -> Result<()> {
4177	// #3757 boot-persist narrowing: TaskManager::start must persist only
4178	// the reconciled queue and the running->failed recoveries — a
4179	// completed task's file must be byte-identical across a restart.
4180	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4181	let manager =
4182	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
4183	.await?;
4184	let task = manager
4185	.add_task(NewTaskRequest::from_prompt("finish then persist"))
4186	.await?;
4187	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
4188	assert_eq!(finished.status, TaskStatus::Completed);
4189	manager.shutdown_and_wait().await?;
4190	drop(manager);
4191
4192	let task_file = root.join("tasks").join(format!("{}.json", task.id));
4193	let before = fs::read(&task_file)?;
4194
4195	let recovered =
4196	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
4197	.await?;
4198	// Give start() a beat to run its (narrowed) boot persist.
4199	sleep(Duration::from_millis(50)).await;
4200	drop(recovered);
4201
4202	let after = fs::read(&task_file)?;
4203	assert_eq!(
4204	before, after,
4205	"a completed task file must not be rewritten on boot"
4206	);
4207	Ok(())
4208	}
4209
4210	#[test]
4211	fn legacy_running_tasks_are_preserved_without_assuming_owner_death() -> Result<()> {
4212	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4213	let tasks_dir = root.join("tasks");
4214	fs::create_dir_all(&tasks_dir)?;
4215	let queue_path = root.join("queue.json");
4216	let task_id = "task_stale_running".to_string();
4217	let started_at = Utc::now() - chrono::Duration::seconds(30);
4218	let task = TaskRecord {
4219	schema_version: CURRENT_TASK_SCHEMA_VERSION,
4220	id: task_id.clone(),
4221	prompt: "long-running shell work".to_string(),
4222	name: None,
4223	model: "deepseek-v4-flash".to_string(),
4224	model_provider: None,
4225	model_provider_id: None,
4226	workspace: PathBuf::from("."),
4227	mode: "agent".to_string(),
4228	allow_shell: true,
4229	trust_mode: false,
4230	auto_approve: false,
4231	status: TaskStatus::Running,
4232	created_at: started_at,
4233	started_at: Some(started_at),
4234	ended_at: None,
4235	duration_ms: None,
4236	result_summary: None,
4237	result_detail_path: None,
4238	error: None,
4239	terminal_reason: None,
4240	thread_id: Some("thr_stale".to_string()),
4241	turn_id: Some("turn_stale".to_string()),
4242	owner_session_id: Some("session-old".to_string()),
4243	execution_scope: None,
4244	execution_generation: None,
4245	cancel_requested_seq: 0,
4246	runtime_event_count: 0,
4247	lifecycle_seq: 2,
4248	checklist: TaskChecklistState::default(),
4249	gates: Vec::new(),
4250	attempts: Vec::new(),
4251	artifacts: Vec::new(),
4252	github_events: Vec::new(),
4253	tool_calls: vec![TaskToolCallSummary {
4254	id: "tool_shell".to_string(),
4255	name: "task_shell_start".to_string(),
4256	status: TaskToolStatus::Running,
4257	started_at,
4258	ended_at: None,
4259	duration_ms: None,
4260	input_summary: Some("shell: sleep 999".to_string()),
4261	output_summary: None,
4262	detail_path: None,
4263	patch_ref: None,
4264	}],
4265	timeline: vec![TaskTimelineEntry {
4266	timestamp: started_at,
4267	kind: "running".to_string(),
4268	summary: "Task started".to_string(),
4269	detail_path: None,
4270	}],
4271	};
4272	fs::write(
4273	tasks_dir.join(format!("{task_id}.json")),
4274	serde_json::to_string_pretty(&task)?,
4275	)?;
4276	fs::write(
4277	&queue_path,
4278	serde_json::to_string_pretty(&QueueFile {
4279	queue: vec![task_id.clone()],
4280	})?,
4281	)?;
4282
4283	let loaded = load_state(&tasks_dir, &queue_path)?;
4284	let queue = loaded.queue;
4285	let recovered = loaded.tasks.get(&task_id).expect("task loaded");
4286
4287	assert!(queue.is_empty(), "stale running task must not be requeued");
4288	assert_eq!(recovered.status, TaskStatus::Running);
4289	assert!(recovered.ended_at.is_none());
4290	assert!(recovered.error.is_none());
4291	assert_eq!(recovered.tool_calls[0].status, TaskToolStatus::Running);
4292	assert!(!TaskSummary::from(recovered).execution_binding_known);
4293	assert_eq!(
4294	fs::read(tasks_dir.join(format!("{task_id}.json")))?,
4295	serde_json::to_string_pretty(&task)?.as_bytes()
4296	);
4297	Ok(())
4298	}
4299
4300	#[tokio::test]
4301	async fn default_workspace_updates_for_future_tasks() -> Result<()> {
4302	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4303	let new_workspace =
4304	std::env::temp_dir().join(format!("deepseek-workspace-{}", Uuid::new_v4()));
4305	let manager =
4306	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
4307
4308	manager.set_default_workspace(new_workspace.clone()).await;
4309	let task = manager
4310	.add_task(NewTaskRequest::from_prompt("test workspace default"))
4311	.await?;
4312
4313	assert_eq!(manager.default_workspace().await, new_workspace);
4314	assert_eq!(task.workspace, new_workspace);
4315	Ok(())
4316	}
4317
4318	#[tokio::test]
4319	async fn record_tool_metadata_updates_explicit_task() -> Result<()> {
4320	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4321	let manager =
4322	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
4323
4324	let task = manager
4325	.add_task(NewTaskRequest::from_prompt("test metadata"))
4326	.await?;
4327	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
4328	let updated = manager
4329	.record_tool_metadata(
4330	&finished.id,
4331	&serde_json::json!({
4332	"task_updates": {
4333	"gate": {
4334	"id": "gate_test",
4335	"gate": "test",
4336	"command": "cargo test -p codewhale-tui --lib",
4337	"cwd": ".",
4338	"exit_code": 0,
4339	"status": "passed",
4340	"classification": "passed",
4341	"duration_ms": 1,
4342	"summary": "ok",
4343	"log_path": null,
4344	"recorded_at": Utc::now()
4345	}
4346	}
4347	}),
4348	)
4349	.await?;
4350
4351	assert_eq!(updated.gates.len(), 1);
4352	assert_eq!(updated.gates[0].classification, "passed");
4353	Ok(())
4354	}
4355
4356	#[tokio::test]
4357	async fn write_task_artifact_rejects_traversal_task_id() -> Result<()> {
4358	let temp = tempfile::tempdir()?;
4359	let root = temp.path().join("tasks-root");
4360	let escaped = temp.path().join("escape");
4361	let manager =
4362	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
4363	.await?;
4364
4365	let err = manager
4366	.write_task_artifact("../escape", "result", "artifact body")
4367	.expect_err("traversal task ids must be rejected");
4368
4369	assert!(err.to_string().contains("single path component"));
4370	assert!(!escaped.exists(), "artifact write escaped the task root");
4371	Ok(())
4372	}
4373
4374	#[tokio::test]
4375	async fn cancel_running_task_marks_canceled() -> Result<()> {
4376	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4377	let manager =
4378	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
4379
4380	let task = manager
4381	.add_task(NewTaskRequest::from_prompt("test cancellation"))
4382	.await?;
4383
4384	sleep(Duration::from_millis(10)).await;
4385	let cancellation = manager.cancel_task(&task.id).await?;
4386	assert_eq!(cancellation.disposition, TaskCancelDisposition::Requested);
4387	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
4388	assert_eq!(finished.status, TaskStatus::Canceled);
4389	Ok(())
4390	}
4391
4392	#[tokio::test]
4393	async fn cancel_finished_task_returns_atomic_already_finished_outcome() -> Result<()> {
4394	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4395	let manager =
4396	TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await?;
4397	let task = manager
4398	.add_task(NewTaskRequest::from_prompt("finish before cancellation"))
4399	.await?;
4400	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
4401	assert_eq!(finished.status, TaskStatus::Completed);
4402
4403	let cancellation = manager.cancel_task(&task.id).await?;
4404
4405	assert_eq!(
4406	cancellation.disposition,
4407	TaskCancelDisposition::AlreadyFinished
4408	);
4409	assert_eq!(cancellation.task.status, TaskStatus::Completed);
4410	Ok(())
4411	}
4412
4413	// GHSA-72w5-pf8h-xfp4 — regression: omitted optional fields must not
4414	// silently elevate the spawned task's privileges.
4415	#[tokio::test]
4416	async fn add_task_without_optional_fields_does_not_grant_shell_or_auto_approve() -> Result<()> {
4417	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4418	let manager =
4419	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
4420	.await?;
4421
4422	let req = NewTaskRequest {
4423	prompt: "fix TODOs and write a README".to_string(),
4424	name: None,
4425	model: None,
4426	model_provider: None,
4427	model_provider_id: None,
4428	workspace: None,
4429	mode: None,
4430	allow_shell: None,
4431	trust_mode: None,
4432	auto_approve: None,
4433	owner_session_id: None,
4434	};
4435	let task = manager.add_task(req).await?;
4436
4437	assert!(
4438	!task.allow_shell,
4439	"model-omitted allow_shell must default to false (no silent shell grant)"
4440	);
4441	assert!(
4442	!task.auto_approve,
4443	"model-omitted auto_approve must default to false (no silent auto-approval)"
4444	);
4445	assert!(
4446	!task.trust_mode,
4447	"model-omitted trust_mode must default to false"
4448	);
4449	Ok(())
4450	}
4451
4452	#[tokio::test]
4453	async fn rejects_newer_task_schema_on_recovery() -> Result<()> {
4454	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
4455	let manager =
4456	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(MockExecutor))
4457	.await?;
4458
4459	let task = manager
4460	.add_task(NewTaskRequest::from_prompt("test schema gate"))
4461	.await?;
4462	let _ = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
4463	manager.shutdown_and_wait().await?;
4464	drop(manager);
4465
4466	let task_path = root.join("tasks").join(format!("{}.json", task.id));
4467	let mut value: serde_json::Value = serde_json::from_str(&fs::read_to_string(&task_path)?)?;
4468	value["schema_version"] = serde_json::json!(999);
4469	fs::write(&task_path, serde_json::to_string_pretty(&value)?)?;
4470
4471	match TaskManager::start_with_executor(test_config(root), Arc::new(MockExecutor)).await {
4472	Ok(_) => panic!("manager should reject newer task schema"),
4473	Err(err) => assert!(err.to_string().contains("newer than supported")),
4474	}
4475	Ok(())
4476	}
4477
4478	#[test]
4479	fn default_tasks_dir_falls_back_to_legacy_deepseek_tasks() {
4480	let temp_home = tempfile::tempdir().unwrap();
4481	let home = temp_home.path();
4482	let legacy_tasks = home.join(".deepseek").join("tasks");
4483	std::fs::create_dir_all(&legacy_tasks).unwrap();
4484
4485	assert_eq!(default_tasks_dir_for_home(home), legacy_tasks);
4486	}
4487
4488	#[test]
4489	fn default_tasks_dir_prefers_existing_codewhale_tasks() {
4490	let temp_home = tempfile::tempdir().unwrap();
4491	let home = temp_home.path();
4492	let primary_tasks = home.join(".codewhale").join("tasks");
4493	let legacy_tasks = home.join(".deepseek").join("tasks");
4494	std::fs::create_dir_all(&primary_tasks).unwrap();
4495	std::fs::create_dir_all(&legacy_tasks).unwrap();
4496
4497	assert_eq!(default_tasks_dir_for_home(home), primary_tasks);
4498	}
4499
4500	#[test]
4501	fn default_tasks_dir_falls_back_to_legacy_when_primary_is_file() {
4502	let temp_home = tempfile::tempdir().unwrap();
4503	let home = temp_home.path();
4504	let primary_tasks = home.join(".codewhale").join("tasks");
4505	let legacy_tasks = home.join(".deepseek").join("tasks");
4506	std::fs::create_dir_all(primary_tasks.parent().unwrap()).unwrap();
4507	std::fs::write(&primary_tasks, "not a directory").unwrap();
4508	std::fs::create_dir_all(&legacy_tasks).unwrap();
4509
4510	assert_eq!(default_tasks_dir_for_home(home), legacy_tasks);
4511	}
4512
4513	#[test]
4514	fn default_tasks_dir_ignores_legacy_file_for_new_installs() {
4515	let temp_home = tempfile::tempdir().unwrap();
4516	let home = temp_home.path();
4517	let primary_tasks = home.join(".codewhale").join("tasks");
4518	let legacy_tasks = home.join(".deepseek").join("tasks");
4519	std::fs::create_dir_all(legacy_tasks.parent().unwrap()).unwrap();
4520	std::fs::write(&legacy_tasks, "not a directory").unwrap();
4521
4522	assert_eq!(default_tasks_dir_for_home(home), primary_tasks);
4523	}
4524
4525	#[test]
4526	fn default_tasks_dir_uses_codewhale_tasks_for_new_installs() {
4527	let temp_home = tempfile::tempdir().unwrap();
4528	let home = temp_home.path();
4529
4530	assert_eq!(
4531	default_tasks_dir_for_home(home),
4532	home.join(".codewhale").join("tasks")
4533	);
4534	}
4535
4536	#[test]
4537	fn task_and_runtime_roots_honor_explicit_codewhale_home() {
4538	let _lock = lock_test_env();
4539	let temp_root = tempfile::tempdir().unwrap();
4540	let ambient_home = temp_root.path().join("ambient-home");
4541	let explicit_home = temp_root.path().join("explicit-home");
4542	std::fs::create_dir_all(ambient_home.join(".deepseek").join("tasks")).unwrap();
4543	let _home = EnvVarGuard::set("HOME", &ambient_home);
4544	let _userprofile = EnvVarGuard::set("USERPROFILE", &ambient_home);
4545	let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", &explicit_home);
4546	let _tasks_override = EnvVarGuard::remove("CODEWHALE_TASKS_DIR");
4547	let _legacy_tasks_override = EnvVarGuard::remove("DEEPSEEK_TASKS_DIR");
4548	let _runtime_override = EnvVarGuard::remove("CODEWHALE_RUNTIME_DIR");
4549	let _legacy_runtime_override = EnvVarGuard::remove("DEEPSEEK_RUNTIME_DIR");
4550
4551	let task_root = default_tasks_dir();
4552	let task_manager =
4553	TaskManagerConfig::from_runtime(&Config::default(), PathBuf::from("."), None, None);
4554	let runtime = RuntimeThreadManagerConfig::from_task_data_dir(task_manager.data_dir.clone());
4555
4556	assert_eq!(task_root, explicit_home.join("tasks"));
4557	assert_eq!(task_manager.data_dir, task_root);
4558	assert_eq!(runtime.task_data_dir, task_root);
4559	assert_eq!(
4560	runtime.data_dir,
4561	explicit_home.join("tasks").join("runtime")
4562	);
4563	}
4564
4565	#[test]
4566	fn whitespace_codewhale_home_keeps_ambient_legacy_task_and_runtime_fallbacks() {
4567	let _lock = lock_test_env();
4568	let temp_root = tempfile::tempdir().unwrap();
4569	let ambient_home = temp_root.path().join("ambient-home");
4570	let legacy_tasks = ambient_home.join(".deepseek").join("tasks");
4571	std::fs::create_dir_all(&legacy_tasks).unwrap();
4572	let _home = EnvVarGuard::set("HOME", &ambient_home);
4573	let _userprofile = EnvVarGuard::set("USERPROFILE", &ambient_home);
4574	let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", " \t ");
4575	let _tasks_override = EnvVarGuard::remove("CODEWHALE_TASKS_DIR");
4576	let _legacy_tasks_override = EnvVarGuard::remove("DEEPSEEK_TASKS_DIR");
4577	let _runtime_override = EnvVarGuard::remove("CODEWHALE_RUNTIME_DIR");
4578	let _legacy_runtime_override = EnvVarGuard::remove("DEEPSEEK_RUNTIME_DIR");
4579
4580	let task_root = default_tasks_dir();
4581	let task_manager =
4582	TaskManagerConfig::from_runtime(&Config::default(), PathBuf::from("."), None, None);
4583	let runtime = RuntimeThreadManagerConfig::from_task_data_dir(task_manager.data_dir.clone());
4584
4585	assert_eq!(task_root, legacy_tasks);
4586	assert_eq!(task_manager.data_dir, task_root);
4587	assert_eq!(runtime.task_data_dir, task_root);
4588	assert_eq!(runtime.data_dir, task_root.join("runtime"));
4589	}
4590
4591	#[cfg(unix)]
4592	#[test]
4593	fn non_unicode_codewhale_home_is_preserved_by_task_and_runtime_roots() {
4594	use std::os::unix::ffi::OsStringExt;
4595
4596	let _lock = lock_test_env();
4597	let temp_root = tempfile::tempdir().unwrap();
4598	let explicit_home = temp_root.path().join(std::ffi::OsString::from_vec(
4599	b"codewhale-\xff-home".to_vec(),
4600	));
4601	let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", &explicit_home);
4602	let _tasks_override = EnvVarGuard::remove("CODEWHALE_TASKS_DIR");
4603	let _legacy_tasks_override = EnvVarGuard::remove("DEEPSEEK_TASKS_DIR");
4604	let _runtime_override = EnvVarGuard::remove("CODEWHALE_RUNTIME_DIR");
4605	let _legacy_runtime_override = EnvVarGuard::remove("DEEPSEEK_RUNTIME_DIR");
4606
4607	let task_root = default_tasks_dir();
4608	let task_manager =
4609	TaskManagerConfig::from_runtime(&Config::default(), PathBuf::from("."), None, None);
4610	let runtime = RuntimeThreadManagerConfig::from_task_data_dir(task_manager.data_dir.clone());
4611
4612	assert_eq!(task_root, explicit_home.join("tasks"));
4613	assert_eq!(task_manager.data_dir, task_root);
4614	assert_eq!(runtime.task_data_dir, task_root);
4615	assert_eq!(
4616	runtime.data_dir,
4617	explicit_home.join("tasks").join("runtime")
4618	);
4619	}
4620
4621	struct DeafHangExecutor;
4622
4623	#[async_trait]
4624	impl TaskExecutor for DeafHangExecutor {
4625	async fn execute(
4626	&self,
4627	_task: ExecutionTask,
4628	_events: mpsc::Sender<TaskExecutionEvent>,
4629	_cancel: CancellationToken,
4630	) -> TaskExecutionResult {
4631	std::future::pending().await
4632	}
4633	}
4634
4635	struct PartialThenHangExecutor;
4636
4637	#[async_trait]
4638	impl TaskExecutor for PartialThenHangExecutor {
4639	async fn execute(
4640	&self,
4641	_task: ExecutionTask,
4642	events: mpsc::Sender<TaskExecutionEvent>,
4643	_cancel: CancellationToken,
4644	) -> TaskExecutionResult {
4645	let _ = events
4646	.send(TaskExecutionEvent::MessageDelta {
4647	content: "partial ".to_string(),
4648	})
4649	.await;
4650	let _ = events
4651	.send(TaskExecutionEvent::MessageDelta {
4652	content: "result".to_string(),
4653	})
4654	.await;
4655	std::future::pending().await
4656	}
4657	}
4658
4659	struct PollCountingHangExecutor {
4660	polls: Arc<AtomicUsize>,
4661	}
4662
4663	#[async_trait]
4664	impl TaskExecutor for PollCountingHangExecutor {
4665	async fn execute(
4666	&self,
4667	_task: ExecutionTask,
4668	_events: mpsc::Sender<TaskExecutionEvent>,
4669	_cancel: CancellationToken,
4670	) -> TaskExecutionResult {
4671	std::future::poll_fn(\|_\| {
4672	self.polls.fetch_add(1, Ordering::Relaxed);
4673	std::task::Poll::Pending
4674	})
4675	.await
4676	}
4677	}
4678
4679	struct HeartbeatExecutor;
4680
4681	#[async_trait]
4682	impl TaskExecutor for HeartbeatExecutor {
4683	async fn execute(
4684	&self,
4685	_task: ExecutionTask,
4686	events: mpsc::Sender<TaskExecutionEvent>,
4687	_cancel: CancellationToken,
4688	) -> TaskExecutionResult {
4689	loop {
4690	let _ = events
4691	.send(TaskExecutionEvent::Status {
4692	message: "heartbeat".to_string(),
4693	})
4694	.await;
4695	sleep(Duration::from_millis(10)).await;
4696	}
4697	}
4698	}
4699
4700	struct ProgressHeartbeatExecutor;
4701
4702	#[async_trait]
4703	impl TaskExecutor for ProgressHeartbeatExecutor {
4704	async fn execute(
4705	&self,
4706	_task: ExecutionTask,
4707	events: mpsc::Sender<TaskExecutionEvent>,
4708	_cancel: CancellationToken,
4709	) -> TaskExecutionResult {
4710	loop {
4711	let _ = events
4712	.send(TaskExecutionEvent::MessageDelta {
4713	content: "working".to_string(),
4714	})
4715	.await;
4716	sleep(Duration::from_millis(10)).await;
4717	}
4718	}
4719	}
4720
4721	struct CooperativeIdleCancelExecutor;
4722
4723	#[async_trait]
4724	impl TaskExecutor for CooperativeIdleCancelExecutor {
4725	async fn execute(
4726	&self,
4727	_task: ExecutionTask,
4728	_events: mpsc::Sender<TaskExecutionEvent>,
4729	cancel: CancellationToken,
4730	) -> TaskExecutionResult {
4731	cancel.cancelled().await;
4732	TaskExecutionResult::from_reason(TaskTerminalReason::Canceled, None)
4733	}
4734	}
4735
4736	struct CooperativeProgressCancelExecutor;
4737
4738	#[async_trait]
4739	impl TaskExecutor for CooperativeProgressCancelExecutor {
4740	async fn execute(
4741	&self,
4742	_task: ExecutionTask,
4743	events: mpsc::Sender<TaskExecutionEvent>,
4744	cancel: CancellationToken,
4745	) -> TaskExecutionResult {
4746	loop {
4747	tokio::select! {
4748	_ = cancel.cancelled() => {
4749	return TaskExecutionResult::from_reason(
4750	TaskTerminalReason::Canceled,
4751	None,
4752	);
4753	}
4754	_ = sleep(Duration::from_millis(10)) => {
4755	let _ = events
4756	.send(TaskExecutionEvent::MessageDelta {
4757	content: "working".to_string(),
4758	})
4759	.await;
4760	}
4761	}
4762	}
4763	}
4764	}
4765
4766	struct PromptRouterExecutor;
4767
4768	#[async_trait]
4769	impl TaskExecutor for PromptRouterExecutor {
4770	async fn execute(
4771	&self,
4772	task: ExecutionTask,
4773	events: mpsc::Sender<TaskExecutionEvent>,
4774	_cancel: CancellationToken,
4775	) -> TaskExecutionResult {
4776	if task.prompt.starts_with("hang ") {
4777	std::future::pending().await
4778	} else {
4779	// The follow-up task must complete without a single await
4780	// point: `run_task` polls the executor future before its
4781	// guard can observe the (test-shortened) idle/wall budgets,
4782	// so an await-free future always finishes first and an
4783	// interrupt can never be recorded against it. The previous
4784	// MockExecutor delegation (`send(...).await` x4 plus a 50 ms
4785	// sleep) left windows where CI scheduler/storage stalls of
4786	// >=150 ms tripped the idle watchdog mid-flight; the executor
4787	// then observed the cancellation and returned `Canceled`,
4788	// which `preserve_timeout_reason` rewrote into the timeout
4789	// reason -> `Failed` (issue #5898). `try_send` keeps the
4790	// released worker's event pipeline exercised without
4791	// suspending this future.
4792	let _ = events.try_send(TaskExecutionEvent::Status {
4793	message: format!("running after forced release {}", task.id),
4794	});
4795	TaskExecutionResult {
4796	status: TaskStatus::Completed,
4797	result_text: Some("done after hang".to_string()),
4798	error: None,
4799	terminal_reason: TaskTerminalReason::Completed,
4800	}
4801	}
4802	}
4803	}
4804
4805	struct FloodExecutor;
4806
4807	#[async_trait]
4808	impl TaskExecutor for FloodExecutor {
4809	async fn execute(
4810	&self,
4811	_task: ExecutionTask,
4812	events: mpsc::Sender<TaskExecutionEvent>,
4813	_cancel: CancellationToken,
4814	) -> TaskExecutionResult {
4815	for i in 0..400 {
4816	// Mirror the runtime path: each raw event is followed by its
4817	// derived message delta. Alternating the two non-urgent stream
4818	// kinds prevents timeline coalescing without turning this
4819	// storage-bound test into hundreds of synchronous fsyncs.
4820	let _ = events
4821	.send(TaskExecutionEvent::RuntimeEvent {
4822	seq: i,
4823	event: "item.delta".to_string(),
4824	summary: format!("tick {i}"),
4825	})
4826	.await;
4827	let _ = events
4828	.send(TaskExecutionEvent::MessageDelta {
4829	content: format!("chunk {i}"),
4830	})
4831	.await;
4832	}
4833	TaskExecutionResult {
4834	status: TaskStatus::Completed,
4835	result_text: Some("flooded".to_string()),
4836	error: None,
4837	terminal_reason: TaskTerminalReason::Completed,
4838	}
4839	}
4840	}
4841
4842	struct CompleteAfterCancelExecutor;
4843
4844	#[async_trait]
4845	impl TaskExecutor for CompleteAfterCancelExecutor {
4846	async fn execute(
4847	&self,
4848	_task: ExecutionTask,
4849	_events: mpsc::Sender<TaskExecutionEvent>,
4850	cancel: CancellationToken,
4851	) -> TaskExecutionResult {
4852	cancel.cancelled().await;
4853	TaskExecutionResult {
4854	status: TaskStatus::Completed,
4855	result_text: Some("late complete".to_string()),
4856	error: None,
4857	terminal_reason: TaskTerminalReason::Completed,
4858	}
4859	}
4860	}
4861
4862	fn sample_task_record() -> TaskRecord {
4863	TaskRecord {
4864	schema_version: CURRENT_TASK_SCHEMA_VERSION,
4865	id: "task_0123456789abcdef".to_string(),
4866	prompt: "bound timeline".to_string(),
4867	name: None,
4868	model: "deepseek-v4-flash".to_string(),
4869	model_provider: None,
4870	model_provider_id: None,
4871	workspace: PathBuf::from("."),
4872	mode: "agent".to_string(),
4873	allow_shell: false,
4874	trust_mode: false,
4875	auto_approve: false,
4876	status: TaskStatus::Running,
4877	created_at: Utc::now(),
4878	started_at: Some(Utc::now()),
4879	ended_at: None,
4880	duration_ms: None,
4881	result_summary: None,
4882	result_detail_path: None,
4883	error: None,
4884	terminal_reason: None,
4885	thread_id: None,
4886	turn_id: None,
4887	owner_session_id: None,
4888	execution_scope: Some(test_execution_scope("test")),
4889	execution_generation: None,
4890	cancel_requested_seq: 0,
4891	runtime_event_count: 0,
4892	lifecycle_seq: 2,
4893	checklist: TaskChecklistState::default(),
4894	gates: Vec::new(),
4895	attempts: Vec::new(),
4896	artifacts: Vec::new(),
4897	github_events: Vec::new(),
4898	tool_calls: Vec::new(),
4899	timeline: Vec::new(),
4900	}
4901	}
4902
4903	#[test]
4904	fn execution_guard_idle_does_not_reset_without_progress() {
4905	let start = Instant::now();
4906	let limits = TaskExecutionLimits::short_for_tests();
4907	let guard = ExecutionGuard::new(limits, start);
4908	match guard.evaluate(start + limits.idle_progress, false, false) {
4909	GuardAction::Interrupt { reason } => {
4910	assert_eq!(reason, TaskTerminalReason::IdleTimeout);
4911	}
4912	other => panic!("expected idle interrupt, got {other:?}"),
4913	}
4914	}
4915
4916	#[test]
4917	fn execution_guard_reports_the_limit_that_expired_first_when_both_elapsed() {
4918	let start = Instant::now();
4919	let limits = TaskExecutionLimits::short_for_tests();
4920	let guard = ExecutionGuard::new(limits, start);
4921	// A starved watchdog that first ticks after both budgets ran out must
4922	// still report the idle limit, which expired first.
4923	match guard.evaluate(
4924	start + limits.wall_time + limits.idle_progress,
4925	false,
4926	false,
4927	) {
4928	GuardAction::Interrupt { reason } => {
4929	assert_eq!(reason, TaskTerminalReason::IdleTimeout);
4930	}
4931	other => panic!("expected idle interrupt, got {other:?}"),
4932	}
4933
4934	// Late progress pushes the idle deadline past the wall deadline, so
4935	// the same starved tick reports the wall limit instead.
4936	let mut guard = ExecutionGuard::new(limits, start);
4937	guard.note_progress(start + limits.wall_time - Duration::from_millis(1));
4938	match guard.evaluate(
4939	start + limits.wall_time + limits.idle_progress,
4940	false,
4941	false,
4942	) {
4943	GuardAction::Interrupt { reason } => {
4944	assert_eq!(reason, TaskTerminalReason::WallTimeout);
4945	}
4946	other => panic!("expected wall interrupt, got {other:?}"),
4947	}
4948	}
4949
4950	#[test]
4951	fn execution_guard_progress_refreshes_idle_until_wall_timeout() {
4952	let start = Instant::now();
4953	let limits = TaskExecutionLimits::short_for_tests();
4954	let mut guard = ExecutionGuard::new(limits, start);
4955	let progressed = start + (limits.idle_progress / 2);
4956	guard.note_progress(progressed);
4957	match guard.evaluate(progressed + (limits.idle_progress / 2), false, false) {
4958	GuardAction::Run { .. } => {}
4959	other => panic!("progress should keep idle from firing, got {other:?}"),
4960	}
4961	// Progress keeps arriving, so the idle deadline never expires before
4962	// the wall deadline does.
4963	guard.note_progress(start + limits.wall_time - (limits.idle_progress / 2));
4964	match guard.evaluate(start + limits.wall_time, false, false) {
4965	GuardAction::Interrupt { reason } => {
4966	assert_eq!(reason, TaskTerminalReason::WallTimeout);
4967	}
4968	other => panic!("expected wall interrupt, got {other:?}"),
4969	}
4970	}
4971
4972	#[test]
4973	fn execution_guard_cancel_grace_terminalizes_stuck_work() {
4974	let start = Instant::now();
4975	let limits = TaskExecutionLimits::short_for_tests();
4976	let mut guard = ExecutionGuard::new(limits, start);
4977	match guard.evaluate(start, true, false) {
4978	GuardAction::Interrupt { reason } => {
4979	assert_eq!(reason, TaskTerminalReason::Canceled);
4980	guard.note_interrupt(start, reason);
4981	}
4982	other => panic!("expected cancel interrupt, got {other:?}"),
4983	}
4984	match guard.evaluate(start + limits.cancel_grace, true, false) {
4985	GuardAction::Terminalize { reason } => {
4986	assert_eq!(reason, TaskTerminalReason::CancelTimeout);
4987	}
4988	other => panic!("expected cancel timeout, got {other:?}"),
4989	}
4990	}
4991
4992	#[test]
4993	fn consecutive_message_deltas_coalesce_on_the_timeline() {
4994	let mut task = sample_task_record();
4995	for i in 0..50 {
4996	push_timeline_entry(
4997	&mut task,
4998	TaskTimelineEntry {
4999	timestamp: Utc::now(),
5000	kind: "message".to_string(),
5001	summary: format!("chunk {i}"),
5002	detail_path: None,
5003	},
5004	);
5005	}
5006	assert_eq!(
5007	task.timeline
5008	.iter()
5009	.filter(\|entry\| entry.kind == "message")
5010	.count(),
5011	1
5012	);
5013	assert_eq!(
5014	task.timeline.last().map(\|e\| e.summary.as_str()),
5015	Some("chunk 49")
5016	);
5017	}
5018
5019	#[test]
5020	fn timeline_trim_bounds_growth_and_keeps_a_head() {
5021	let mut task = sample_task_record();
5022	for i in 0..400 {
5023	push_timeline_entry(
5024	&mut task,
5025	TaskTimelineEntry {
5026	timestamp: Utc::now(),
5027	kind: "status".to_string(),
5028	summary: format!("tick {i}"),
5029	detail_path: None,
5030	},
5031	);
5032	}
5033	assert!(task.timeline.len() <= TIMELINE_ENTRY_LIMIT);
5034	assert_eq!(task.timeline[0].summary, "tick 0");
5035	assert!(
5036	task.timeline.iter().any(\|entry\| entry.kind == "omitted"),
5037	"bounded timeline should record omitted history: {:?}",
5038	task.timeline
5039	.iter()
5040	.map(\|e\| e.kind.as_str())
5041	.collect::<Vec<_>>()
5042	);
5043	}
5044
5045	#[tokio::test]
5046	async fn never_terminalizing_execution_fails_with_idle_timeout() -> Result<()> {
5047	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5048	let manager =
5049	TaskManager::start_with_executor(short_test_config(root), Arc::new(DeafHangExecutor))
5050	.await?;
5051	let task = manager
5052	.add_task(NewTaskRequest::from_prompt("never finish"))
5053	.await?;
5054	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5055	assert_eq!(finished.status, TaskStatus::Failed);
5056	assert_eq!(finished.terminal_reason.as_deref(), Some("idle_timeout"));
5057	assert!(
5058	finished
5059	.error
5060	.as_deref()
5061	.is_some_and(\|err\| err.contains("idle")),
5062	"idle timeout must be visible on the receipt: {finished:?}"
5063	);
5064	Ok(())
5065	}
5066
5067	#[tokio::test]
5068	async fn forced_timeout_keeps_all_partial_message_output() -> Result<()> {
5069	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5070	let manager = TaskManager::start_with_executor(
5071	short_test_config(root),
5072	Arc::new(PartialThenHangExecutor),
5073	)
5074	.await?;
5075	let task = manager
5076	.add_task(NewTaskRequest::from_prompt("retain partial result"))
5077	.await?;
5078	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5079
5080	assert_eq!(finished.status, TaskStatus::Failed);
5081	assert_eq!(finished.terminal_reason.as_deref(), Some("idle_timeout"));
5082	assert_eq!(finished.result_summary.as_deref(), Some("partial result"));
5083	Ok(())
5084	}
5085
5086	#[tokio::test]
5087	async fn cooperative_cancel_after_idle_timeout_keeps_timeout_reason() -> Result<()> {
5088	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5089	let manager = TaskManager::start_with_executor(
5090	short_test_config(root),
5091	Arc::new(CooperativeIdleCancelExecutor),
5092	)
5093	.await?;
5094	let task = manager
5095	.add_task(NewTaskRequest::from_prompt("cooperative idle timeout"))
5096	.await?;
5097	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5098	assert_eq!(finished.status, TaskStatus::Failed);
5099	assert_eq!(finished.terminal_reason.as_deref(), Some("idle_timeout"));
5100	assert!(
5101	finished
5102	.error
5103	.as_deref()
5104	.is_some_and(\|error\| error.contains("idle")),
5105	"cooperative cancellation must retain the timeout receipt: {finished:?}"
5106	);
5107	Ok(())
5108	}
5109
5110	#[tokio::test]
5111	async fn heartbeat_status_does_not_refresh_idle_timeout() -> Result<()> {
5112	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5113	let manager =
5114	TaskManager::start_with_executor(short_test_config(root), Arc::new(HeartbeatExecutor))
5115	.await?;
5116	let task = manager
5117	.add_task(NewTaskRequest::from_prompt("heartbeat only"))
5118	.await?;
5119	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5120	assert_eq!(finished.status, TaskStatus::Failed);
5121	assert_eq!(finished.terminal_reason.as_deref(), Some("idle_timeout"));
5122	Ok(())
5123	}
5124
5125	#[tokio::test]
5126	async fn active_progress_keeps_idle_alive_until_wall_timeout() -> Result<()> {
5127	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5128	let manager = TaskManager::start_with_executor(
5129	short_test_config(root),
5130	Arc::new(ProgressHeartbeatExecutor),
5131	)
5132	.await?;
5133	let task = manager
5134	.add_task(NewTaskRequest::from_prompt("genuine progress"))
5135	.await?;
5136	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5137	assert_eq!(finished.status, TaskStatus::Failed);
5138	assert_eq!(finished.terminal_reason.as_deref(), Some("wall_timeout"));
5139	Ok(())
5140	}
5141
5142	#[tokio::test]
5143	async fn cooperative_cancel_after_wall_timeout_keeps_timeout_reason() -> Result<()> {
5144	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5145	let manager = TaskManager::start_with_executor(
5146	wall_timeout_test_config(root),
5147	Arc::new(CooperativeProgressCancelExecutor),
5148	)
5149	.await?;
5150	let task = manager
5151	.add_task(NewTaskRequest::from_prompt("cooperative wall timeout"))
5152	.await?;
5153	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5154	assert_eq!(finished.status, TaskStatus::Failed);
5155	assert_eq!(finished.terminal_reason.as_deref(), Some("wall_timeout"));
5156	assert!(
5157	finished
5158	.error
5159	.as_deref()
5160	.is_some_and(\|error\| error.contains("wall-time")),
5161	"cooperative cancellation must retain the timeout receipt: {finished:?}"
5162	);
5163	Ok(())
5164	}
5165
5166	#[tokio::test]
5167	async fn shutdown_terminalizes_a_stuck_running_task() -> Result<()> {
5168	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5169	let manager =
5170	TaskManager::start_with_executor(short_test_config(root), Arc::new(DeafHangExecutor))
5171	.await?;
5172	let task = manager
5173	.add_task(NewTaskRequest::from_prompt("stuck during shutdown"))
5174	.await?;
5175	sleep(Duration::from_millis(5)).await;
5176	manager.shutdown();
5177	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5178	assert_eq!(finished.status, TaskStatus::Canceled);
5179	assert_eq!(finished.terminal_reason.as_deref(), Some("shutdown"));
5180	Ok(())
5181	}
5182
5183	#[tokio::test]
5184	async fn shutdown_cancel_signal_does_not_spin_during_grace() -> Result<()> {
5185	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5186	let polls = Arc::new(AtomicUsize::new(0));
5187	let manager = TaskManager::start_with_executor(
5188	short_test_config(root),
5189	Arc::new(PollCountingHangExecutor {
5190	polls: Arc::clone(&polls),
5191	}),
5192	)
5193	.await?;
5194	let task = manager
5195	.add_task(NewTaskRequest::from_prompt("stuck during shutdown"))
5196	.await?;
5197
5198	let deadline = std::time::Instant::now() + Duration::from_secs(5);
5199	while manager.get_task(&task.id).await?.status != TaskStatus::Running {
5200	if std::time::Instant::now() >= deadline {
5201	bail!("task never started running");
5202	}
5203	sleep(Duration::from_millis(5)).await;
5204	}
5205
5206	manager.shutdown();
5207	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5208	assert_eq!(finished.terminal_reason.as_deref(), Some("shutdown"));
5209	assert!(
5210	polls.load(Ordering::Relaxed) <= 10,
5211	"already-canceled shutdown signal repeatedly repolled the executor during grace: {} polls",
5212	polls.load(Ordering::Relaxed)
5213	);
5214	Ok(())
5215	}
5216
5217	#[tokio::test]
5218	async fn forced_idle_timeout_releases_the_worker_for_later_tasks() -> Result<()> {
5219	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5220	let manager = TaskManager::start_with_executor(
5221	short_test_config(root),
5222	Arc::new(PromptRouterExecutor),
5223	)
5224	.await?;
5225	let stuck = manager
5226	.add_task(NewTaskRequest::from_prompt("hang until idle timeout"))
5227	.await?;
5228	let finished =
5229	wait_for_terminal_state(&manager, &stuck.id, Duration::from_secs(10)).await?;
5230	assert_eq!(
5231	finished.terminal_reason.as_deref(),
5232	Some("idle_timeout"),
5233	"stuck task terminal record: {finished:?}"
5234	);
5235
5236	let next = manager
5237	.add_task(NewTaskRequest::from_prompt("run after hang"))
5238	.await?;
5239	let completed =
5240	wait_for_terminal_state(&manager, &next.id, Duration::from_secs(10)).await?;
5241	assert_eq!(
5242	completed.status,
5243	TaskStatus::Completed,
5244	"follow-up task terminal record: {completed:?}"
5245	);
5246	assert_eq!(
5247	completed.terminal_reason.as_deref(),
5248	Some("completed"),
5249	"follow-up task terminal record: {completed:?}"
5250	);
5251	Ok(())
5252	}
5253
5254	#[tokio::test]
5255	async fn cancel_then_completed_result_is_recorded_as_canceled() -> Result<()> {
5256	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5257	let manager = TaskManager::start_with_executor(
5258	test_config(root),
5259	Arc::new(CompleteAfterCancelExecutor),
5260	)
5261	.await?;
5262	let task = manager
5263	.add_task(NewTaskRequest::from_prompt("race complete after cancel"))
5264	.await?;
5265	let deadline = std::time::Instant::now() + Duration::from_secs(5);
5266	loop {
5267	let current = manager.get_task(&task.id).await?;
5268	if current.status == TaskStatus::Running {
5269	break;
5270	}
5271	if std::time::Instant::now() >= deadline {
5272	bail!("task never started running");
5273	}
5274	sleep(Duration::from_millis(5)).await;
5275	}
5276	sleep(Duration::from_millis(5)).await;
5277	let cancellation = manager.cancel_task(&task.id).await?;
5278	assert_eq!(cancellation.disposition, TaskCancelDisposition::Requested);
5279	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5280	assert_eq!(finished.status, TaskStatus::Canceled);
5281	assert_eq!(finished.terminal_reason.as_deref(), Some("canceled"));
5282	Ok(())
5283	}
5284
5285	#[tokio::test]
5286	async fn long_stream_timeline_is_bounded() -> Result<()> {
5287	let root = std::env::temp_dir().join(format!("deepseek-task-test-{}", Uuid::new_v4()));
5288	let manager =
5289	TaskManager::start_with_executor(test_config(root.clone()), Arc::new(FloodExecutor))
5290	.await?;
5291	let task = manager
5292	.add_task(NewTaskRequest::from_prompt("flood the timeline"))
5293	.await?;
5294	let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5295	assert_eq!(finished.status, TaskStatus::Completed);
5296	assert_eq!(finished.runtime_event_count, 400);
5297	assert!(
5298	finished.timeline.len() <= TIMELINE_ENTRY_LIMIT,
5299	"timeline grew to {}",
5300	finished.timeline.len()
5301	);
5302	assert!(
5303	finished
5304	.timeline
5305	.iter()
5306	.any(\|entry\| entry.kind == "omitted"),
5307	"long streams should drop older timeline entries: {:?}",
5308	finished
5309	.timeline
5310	.iter()
5311	.map(\|e\| e.kind.as_str())
5312	.collect::<Vec<_>>()
5313	);
5314
5315	let persisted_path = root.join("tasks").join(format!("{}.json", task.id));
5316	let persisted: TaskRecord = serde_json::from_slice(&fs::read(&persisted_path)?)?;
5317	assert_eq!(persisted.status, TaskStatus::Completed);
5318	assert_eq!(persisted.runtime_event_count, 400);
5319	assert!(persisted.timeline.len() <= TIMELINE_ENTRY_LIMIT);
5320	assert!(
5321	persisted
5322	.timeline
5323	.iter()
5324	.any(\|entry\| entry.kind == "omitted")
5325	);
5326	Ok(())
5327	}
5328
5329	async fn test_runtime_manager() -> Result<RuntimeThreadManager> {
5330	let root = tempfile::tempdir()?.keep();
5331	RuntimeThreadManager::open(
5332	Config::default(),
5333	PathBuf::from("."),
5334	RuntimeThreadManagerConfig::from_task_data_dir(root),
5335	)
5336	}
5337
5338	async fn drain_task_events(mut rx: mpsc::Receiver<TaskExecutionEvent>) {
5339	while rx.recv().await.is_some() {}
5340	}
5341
5342	struct RuntimeProjectionExecutor(Vec<(&'static str, Value)>);
5343
5344	#[async_trait]
5345	impl TaskExecutor for RuntimeProjectionExecutor {
5346	async fn execute(
5347	&self,
5348	_task: ExecutionTask,
5349	events: mpsc::Sender<TaskExecutionEvent>,
5350	cancel: CancellationToken,
5351	) -> TaskExecutionResult {
5352	let runtime = test_runtime_manager().await.expect("fixture runtime");
5353	let thread = runtime
5354	.create_thread(CreateThreadRequest::default())
5355	.await
5356	.expect("fixture thread");
5357	for (event, payload) in &self.0 {
5358	runtime
5359	.emit_event_for_test(
5360	&thread.id,
5361	Some("turn_projection"),
5362	event,
5363	payload.clone(),
5364	)
5365	.await
5366	.expect("persist fixture runtime event");
5367	}
5368	drive_engine_turn(
5369	&runtime,
5370	&thread.id,
5371	"turn_projection",
5372	events,
5373	cancel,
5374	TaskExecutionLimits::default(),
5375	)
5376	.await
5377	}
5378	}
5379
5380	async fn project_runtime_task(events: Vec<(&'static str, Value)>) -> Result<TaskRecord> {
5381	let root = tempfile::tempdir()?;
5382	let manager = TaskManager::start_with_executor(
5383	test_config(root.path().to_path_buf()),
5384	Arc::new(RuntimeProjectionExecutor(events)),
5385	)
5386	.await?;
5387	let task = manager
5388	.add_task(NewTaskRequest::from_prompt("runtime projection fixture"))
5389	.await?;
5390	wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
5391	manager.shutdown_and_wait().await?;
5392	// Assert the durable task projection, not just an adapter event.
5393	let path = root.path().join("tasks").join(format!("{}.json", task.id));
5394	Ok(serde_json::from_slice(&fs::read(path)?)?)
5395	}
5396
5397	#[tokio::test]
5398	async fn runtime_task_projection_preserves_provider_ids_and_terminal_tool_statuses()
5399	-> Result<()> {
5400	let mut events = Vec::new();
5401	for (item, provider) in [
5402	("item_a", "call_a"),
5403	("item_b", "call_b"),
5404	("item_c", "call_c"),
5405	] {
5406	events.push((
5407	"item.started",
5408	json!({
5409	"item": { "id": item, "kind": "tool_call" },
5410	"tool": { "id": provider, "name": "read", "input": {} }
5411	}),
5412	));
5413	}
5414	// Parallel same-name calls finish out of order. Success preserves
5415	// tool_result_for; errors retain tool_use_id; redaction uses tool_call_id.
5416	for (item, provider, identity_key, terminal) in [
5417	("item_c", "call_c", "tool_use_id", "item.failed"),
5418	("item_b", "call_b", "tool_call_id", "item.completed"),
5419	("item_a", "call_a", "tool_result_for", "item.completed"),
5420	] {
5421	events.push((
5422	terminal,
5423	json!({ "item": {
5424	"id": item, "kind": "tool_call", "summary": "redacted receipt",
5425	"detail": format!("result for {provider}"),
5426	"metadata": { identity_key: provider, "tool_name": "read" }
5427	}}),
5428	));
5429	}
5430	// Old event shapes with no metadata retain their existing identity.
5431	events.extend([
5432	("item.started", json!({ "tool": { "id": "legacy", "name": "read", "input": {} } })),
5433	("item.completed", json!({ "item": { "id": "legacy", "kind": "tool_call", "summary": "read: ok", "detail": "ok" } })),
5434	("turn.completed", json!({ "turn": { "status": "completed" } })),
5435	]);
5436	let task = project_runtime_task(events).await?;
5437	assert_eq!(task.status, TaskStatus::Completed);
5438	assert_eq!(task.tool_calls.len(), 4);
5439	for (call, expected_id, expected_status) in [
5440	(&task.tool_calls[0], "call_a", TaskToolStatus::Success),
5441	(&task.tool_calls[1], "call_b", TaskToolStatus::Success),
5442	(&task.tool_calls[2], "call_c", TaskToolStatus::Failed),
5443	(&task.tool_calls[3], "legacy", TaskToolStatus::Success),
5444	] {
5445	assert_eq!(call.id, expected_id);
5446	assert_eq!(call.status, expected_status);
5447	assert!(call.ended_at.is_some());
5448	assert!(call.duration_ms.is_some());
5449	}
5450	assert_eq!(
5451	task.tool_calls[0].output_summary.as_deref(),
5452	Some("result for call_a")
5453	);
5454	assert_eq!(
5455	task.tool_calls[2].output_summary.as_deref(),
5456	Some("result for call_c")
5457	);
5458	Ok(())
5459	}
5460
5461	#[tokio::test]
5462	async fn runtime_task_projection_uses_last_completed_message_without_delta_duplication()
5463	-> Result<()> {
5464	let commentary = "Checking fixture state. ".repeat(20);
5465	let task = project_runtime_task(vec![
5466	("item.started", json!({ "item": { "id": "commentary", "kind": "agent_message" } })),
5467	("item.delta", json!({ "kind": "agent_message", "delta": commentary })),
5468	("item.completed", json!({ "item": { "id": "commentary", "kind": "agent_message", "detail": commentary } })),
5469	("item.started", json!({ "item": { "id": "final", "kind": "agent_message" } })),
5470	("item.delta", json!({ "kind": "agent_message", "delta": "NOTHING_" })),
5471	("item.completed", json!({ "item": { "id": "final", "kind": "agent_message", "detail": "NOTHING_TO_REPORT" } })),
5472	("turn.completed", json!({ "turn": { "status": "completed" } })),
5473	]).await?;
5474	assert_eq!(task.result_summary.as_deref(), Some("NOTHING_TO_REPORT"));
5475	assert!(task.result_detail_path.is_none());
5476	assert!(
5477	task.timeline.iter().any(\|entry\| entry.kind == "message"
5478	&& entry.summary.starts_with("Checking fixture state."))
5479	);
5480	Ok(())
5481	}
5482
5483	#[tokio::test]
5484	async fn runtime_task_projection_preserves_partial_output_only_for_unfinished_results()
5485	-> Result<()> {
5486	for (status, expected) in [
5487	("interrupted", "partial final"),
5488	("failed", "partial final"),
5489	("completed", "(no textual output)"),
5490	] {
5491	let task = project_runtime_task(vec![
5492	(
5493	"item.completed",
5494	json!({ "item": { "kind": "agent_message", "detail": "earlier commentary" } }),
5495	),
5496	(
5497	"item.started",
5498	json!({ "item": { "kind": "agent_message" } }),
5499	),
5500	(
5501	"item.delta",
5502	json!({ "kind": "agent_message", "delta": "partial final" }),
5503	),
5504	("turn.completed", json!({ "turn": { "status": status } })),
5505	])
5506	.await?;
5507	assert_eq!(task.result_summary.as_deref(), Some(expected), "{status}");
5508	}
5509	Ok(())
5510	}
5511
5512	#[tokio::test]
5513	async fn engine_turn_without_terminal_event_idle_times_out() -> Result<()> {
5514	let runtime = test_runtime_manager().await?;
5515	let thread = runtime
5516	.create_thread(CreateThreadRequest::default())
5517	.await?;
5518	let (tx, rx) = mpsc::channel(64);
5519	tokio::spawn(drain_task_events(rx));
5520	let result = drive_engine_turn(
5521	&runtime,
5522	&thread.id,
5523	"turn_missing",
5524	tx,
5525	CancellationToken::new(),
5526	TaskExecutionLimits::short_for_tests(),
5527	)
5528	.await;
5529	assert_eq!(result.status, TaskStatus::Failed);
5530	assert_eq!(result.terminal_reason, TaskTerminalReason::IdleTimeout);
5531	Ok(())
5532	}
5533
5534	#[tokio::test]
5535	async fn engine_turn_keeps_idle_timeout_when_runtime_interrupts_during_grace() -> Result<()> {
5536	let runtime = Arc::new(test_runtime_manager().await?);
5537	let thread = runtime
5538	.create_thread(CreateThreadRequest::default())
5539	.await?;
5540	let (tx, mut rx) = mpsc::channel(64);
5541	let runtime_for_drive = Arc::clone(&runtime);
5542	let thread_id = thread.id.clone();
5543	let drive = tokio::spawn(async move {
5544	drive_engine_turn(
5545	runtime_for_drive.as_ref(),
5546	&thread_id,
5547	"turn_timeout",
5548	tx,
5549	CancellationToken::new(),
5550	TaskExecutionLimits {
5551	wall_time: Duration::from_secs(2),
5552	idle_progress: Duration::from_millis(80),
5553	cancel_grace: Duration::from_millis(500),
5554	persist_debounce: Duration::from_millis(10),
5555	},
5556	)
5557	.await
5558	});
5559
5560	tokio::time::timeout(Duration::from_secs(2), async {
5561	loop {
5562	match rx.recv().await {
5563	Some(TaskExecutionEvent::Status { message })
5564	if message.contains("idle deadline") =>
5565	{
5566	break;
5567	}
5568	Some(_) => {}
5569	None => panic!("engine task event stream closed before timeout interrupt"),
5570	}
5571	}
5572	})
5573	.await
5574	.context("engine did not request the idle-timeout interrupt")?;
5575
5576	runtime
5577	.emit_event_for_test(
5578	&thread.id,
5579	Some("turn_timeout"),
5580	"turn.completed",
5581	json!({ "turn": { "status": "interrupted" } }),
5582	)
5583	.await?;
5584	let result = drive.await?;
5585	assert_eq!(result.status, TaskStatus::Failed);
5586	assert_eq!(result.terminal_reason, TaskTerminalReason::IdleTimeout);
5587	Ok(())
5588	}
5589
5590	#[tokio::test]
5591	async fn engine_turn_uses_cursor_catchup_for_completed_event() -> Result<()> {
5592	let runtime = test_runtime_manager().await?;
5593	let thread = runtime
5594	.create_thread(CreateThreadRequest::default())
5595	.await?;
5596	runtime
5597	.emit_event_for_test(
5598	&thread.id,
5599	Some("turn_done"),
5600	"turn.completed",
5601	json!({ "turn": { "status": "completed" } }),
5602	)
5603	.await?;
5604	let (tx, rx) = mpsc::channel(64);
5605	tokio::spawn(drain_task_events(rx));
5606	let result = drive_engine_turn(
5607	&runtime,
5608	&thread.id,
5609	"turn_done",
5610	tx,
5611	CancellationToken::new(),
5612	TaskExecutionLimits::short_for_tests(),
5613	)
5614	.await;
5615	assert_eq!(result.status, TaskStatus::Completed);
5616	assert_eq!(result.terminal_reason, TaskTerminalReason::Completed);
5617	Ok(())
5618	}
5619
5620	#[tokio::test]
5621	async fn pending_approval_suspends_idle_and_timeout_denial_settles_failed() -> Result<()> {
5622	// #6118: a run that needs a tool approval must not die as a silent
5623	// idle-timeout cancel; the pending approval suspends the idle
5624	// watchdog, and the bridge's own deadline denial then settles the
5625	// run Failed with the reason recorded.
5626	let runtime = Arc::new(test_runtime_manager().await?);
5627	let thread = runtime
5628	.create_thread(CreateThreadRequest::default())
5629	.await?;
5630	let thread_id = thread.id.clone();
5631	runtime
5632	.emit_event_for_test(
5633	&thread.id,
5634	Some("turn_approval"),
5635	"approval.required",
5636	json!({
5637	"approval_id": "approval_fixture_1",
5638	"tool_call_id": "call_fixture_1",
5639	"tool_name": "shell",
5640	}),
5641	)
5642	.await?;
5643	let (tx, mut rx) = mpsc::channel(64);
5644	let runtime_for_drive = Arc::clone(&runtime);
5645	let drive = tokio::spawn(async move {
5646	drive_engine_turn(
5647	runtime_for_drive.as_ref(),
5648	&thread_id,
5649	"turn_approval",
5650	tx,
5651	CancellationToken::new(),
5652	TaskExecutionLimits {
5653	wall_time: Duration::from_secs(5),
5654	idle_progress: Duration::from_millis(120),
5655	cancel_grace: Duration::from_millis(200),
5656	persist_debounce: Duration::from_millis(10),
5657	},
5658	)
5659	.await
5660	});
5661
5662	// Well past the idle window, the pending approval must keep the run
5663	// alive; the old behavior killed it here with no receipt.
5664	tokio::time::sleep(Duration::from_millis(400)).await;
5665	assert!(
5666	!drive.is_finished(),
5667	"a pending approval must suspend the idle watchdog (#6118)"
5668	);
5669	while let Ok(event) = rx.try_recv() {
5670	if let TaskExecutionEvent::Status { message } = event {
5671	assert!(
5672	!message.contains("idle deadline"),
5673	"no idle interrupt may fire while an approval is pending: {message}"
5674	);
5675	}
5676	}
5677
5678	// The decision window closes: the run settles Failed with the reason.
5679	runtime
5680	.emit_event_for_test(
5681	&thread.id,
5682	Some("turn_approval"),
5683	"approval.timeout",
5684	json!({ "approval_id": "approval_fixture_1", "tool_call_id": "call_fixture_1" }),
5685	)
5686	.await?;
5687	let result = tokio::time::timeout(Duration::from_secs(2), drive)
5688	.await
5689	.context("the decision-window denial must settle the run promptly")??;
5690	assert_eq!(result.status, TaskStatus::Failed);
5691	assert_eq!(result.terminal_reason, TaskTerminalReason::Failed);
5692	assert!(
5693	result
5694	.error
5695	.as_deref()
5696	.is_some_and(\|error\| error.contains("Tool approval was not answered")),
5697	"the run must record why it stopped, got {:?}",
5698	result.error
5699	);
5700	Ok(())
5701	}
5702
5703	#[tokio::test]
5704	async fn resolved_approval_restores_the_idle_watchdog() -> Result<()> {
5705	// #6118 counter-check: the suspension ends with the decision, so a
5706	// run that then stops making progress is idle-killed exactly as
5707	// before.
5708	let runtime = test_runtime_manager().await?;
5709	let thread = runtime
5710	.create_thread(CreateThreadRequest::default())
5711	.await?;
5712	runtime
5713	.emit_event_for_test(
5714	&thread.id,
5715	Some("turn_approval_resolved"),
5716	"approval.required",
5717	json!({
5718	"approval_id": "approval_fixture_2",
5719	"tool_call_id": "call_fixture_2",
5720	"tool_name": "shell",
5721	}),
5722	)
5723	.await?;
5724	runtime
5725	.emit_event_for_test(
5726	&thread.id,
5727	Some("turn_approval_resolved"),
5728	"approval.decided",
5729	json!({
5730	"approval_id": "approval_fixture_2",
5731	"tool_call_id": "call_fixture_2",
5732	"decision": "allow",
5733	}),
5734	)
5735	.await?;
5736	let (tx, rx) = mpsc::channel(64);
5737	tokio::spawn(drain_task_events(rx));
5738	let result = drive_engine_turn(
5739	&runtime,
5740	&thread.id,
5741	"turn_approval_resolved",
5742	tx,
5743	CancellationToken::new(),
5744	TaskExecutionLimits::short_for_tests(),
5745	)
5746	.await;
5747	assert_eq!(result.status, TaskStatus::Failed);
5748	assert_eq!(result.terminal_reason, TaskTerminalReason::IdleTimeout);
5749	Ok(())
5750	}
5751
5752	#[tokio::test]
5753	async fn engine_turn_prefers_runtime_terminal_over_cancel_grace() -> Result<()> {
5754	let runtime = test_runtime_manager().await?;
5755	let thread = runtime
5756	.create_thread(CreateThreadRequest::default())
5757	.await?;
5758	let cancel = CancellationToken::new();
5759	cancel.cancel();
5760	runtime
5761	.emit_event_for_test(
5762	&thread.id,
5763	Some("turn_done"),
5764	"turn.completed",
5765	json!({ "turn": { "status": "interrupted" } }),
5766	)
5767	.await?;
5768	let (tx, rx) = mpsc::channel(64);
5769	tokio::spawn(drain_task_events(rx));
5770	let result = drive_engine_turn(
5771	&runtime,
5772	&thread.id,
5773	"turn_done",
5774	tx,
5775	cancel,
5776	TaskExecutionLimits::short_for_tests(),
5777	)
5778	.await;
5779	assert_eq!(result.status, TaskStatus::Canceled);
5780	assert_eq!(result.terminal_reason, TaskTerminalReason::Canceled);
5781	Ok(())
5782	}
5783
5784	#[tokio::test]
5785	async fn runtime_store_failure_event_reaches_the_task_timeline() -> Result<()> {
5786	// #5931: the runtime's own store fault lands in the task timeline,
5787	// and a terminal one stops the driver instead of idling it out.
5788	let (tx, mut rx) = mpsc::channel(8);
5789	let mut final_text = RuntimeTaskOutput::default();
5790	let path = "/tmp/runtime/turns/turn_store.json";
5791	let event = RuntimeEventRecord {
5792	schema_version: 1,
5793	seq: 7,
5794	timestamp: Utc::now(),
5795	thread_id: "thr_store".to_string(),
5796	turn_id: Some("turn_store".to_string()),
5797	item_id: None,
5798	event: RUNTIME_STORE_FAILURE_EVENT.to_string(),
5799	payload: json!({
5800	"operation": "read",
5801	"record_kind": "turn",
5802	"record_id": "turn_store",
5803	"path": path,
5804	"error": format!("Failed to read turn {path}: No such file"),
5805	"reason": "No such file",
5806	"next_action": format!("Move {path} aside (or delete it) and retry."),
5807	"message": format!(
5808	"Session runtime store: turn turn_store at {path} could not be read: No such file. Move {path} aside (or delete it) and retry."
5809	),
5810	}),
5811	};
5812
5813	assert!(
5814	ingest_runtime_event(&event, &mut final_text, &tx)
5815	.await
5816	.is_none(),
5817	"a non-terminal store fault leaves the driver waiting"
5818	);
5819	let mut saw_error = false;
5820	while let Ok(received) = rx.try_recv() {
5821	if let TaskExecutionEvent::Error { message } = received {
5822	assert!(message.contains(path), "{message}");
5823	saw_error = true;
5824	}
5825	}
5826	assert!(saw_error, "store fault missing from the task timeline");
5827
5828	let mut terminal = event.clone();
5829	terminal.payload["terminal"] = json!(true);
5830	let (status, error) = ingest_runtime_event(&terminal, &mut final_text, &tx)
5831	.await
5832	.expect("a terminal store fault ends the turn");
5833	assert_eq!(status, RuntimeTurnStatus::Failed);
5834	assert!(error.is_some_and(\|message\| message.contains(path)));
5835	Ok(())
5836	}
5837	}
5838
5839	#[cfg(test)]
5840	mod ownership_tests;
5841

5841 lines RUST