runtime_threads.rs - DeepSeek-TUI-2026

返回 DeepSeek-TUI-2026

runtime_threads.rs

1	//! Durable thread/turn/item runtime for the HTTP API and background tasks.
2	//!
3	//! This module keeps DeepSeek-only execution while exposing Codex-like lifecycle
4	//! semantics (threads, turns, items, interrupt/steer, and replayable events).
5
6	use std::collections::{HashMap, HashSet, VecDeque};
7	use std::fs::{self, File, OpenOptions};
8	use std::io::{BufRead, BufReader, Write};
9	use std::path::{Path, PathBuf};
10	use std::sync::{Arc, Mutex as StdMutex};
11
12	use anyhow::{Context, Result, anyhow, bail};
13	use chrono::{DateTime, Utc};
14	use serde::{Deserialize, Serialize};
15	use serde_json::{Value, json};
16	use tokio::sync::{Mutex, broadcast};
17	use tokio_util::sync::CancellationToken;
18	use uuid::Uuid;
19
20	use crate::compaction::CompactionConfig;
21	use crate::config::{Config, DEFAULT_TEXT_MODEL, MAX_SUBAGENTS};
22	use crate::core::coherence::CoherenceState;
23	use crate::core::engine::{EngineConfig, EngineHandle, spawn_engine};
24	use crate::core::events::{Event as EngineEvent, TurnOutcomeStatus};
25	use crate::core::ops::Op;
26	use crate::models::{ContentBlock, Message, SystemPrompt, Usage, compaction_threshold_for_model};
27	use crate::tools::plan::new_shared_plan_state;
28	use crate::tools::subagent::SubAgentStatus;
29	use crate::tools::todo::new_shared_todo_list;
30	use crate::tui::app::AppMode;
31
32	const EVENT_CHANNEL_CAPACITY: usize = 1024;
33	const MAX_ACTIVE_THREADS_DEFAULT: usize = 8;
34	const SUMMARY_LIMIT: usize = 280;
35	/// Bumped to 2 for v0.6.6 — see issue #124. The persisted thread/turn/item
36	/// records didn't change shape, but the live engine semantics did: cycle
37	/// boundaries advance the `Session.cycle_count` and produce archived JSONL
38	/// files at `~/.deepseek/sessions/<id>/cycles/<n>.jsonl`. A v1 reader on a
39	/// session written by v2 wouldn't know about the cycle archive directory and
40	/// might misinterpret message counts; bumping is the safe choice.
41	const CURRENT_RUNTIME_SCHEMA_VERSION: u32 = 2;
42	const RUNTIME_RESTART_REASON: &str = "Interrupted by process restart";
43
44	const fn default_runtime_schema_version() -> u32 {
45	CURRENT_RUNTIME_SCHEMA_VERSION
46	}
47
48	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
49	#[serde(rename_all = "snake_case")]
50	pub enum RuntimeTurnStatus {
51	Queued,
52	InProgress,
53	Completed,
54	Failed,
55	Interrupted,
56	Canceled,
57	}
58
59	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
60	#[serde(rename_all = "snake_case")]
61	pub enum TurnItemKind {
62	UserMessage,
63	AgentMessage,
64	ToolCall,
65	FileChange,
66	CommandExecution,
67	ContextCompaction,
68	Status,
69	Error,
70	}
71
72	#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
73	#[serde(rename_all = "snake_case")]
74	pub enum TurnItemLifecycleStatus {
75	Queued,
76	InProgress,
77	Completed,
78	Failed,
79	Interrupted,
80	Canceled,
81	}
82
83	#[derive(Debug, Clone, Serialize, Deserialize)]
84	pub struct ThreadRecord {
85	#[serde(default = "default_runtime_schema_version")]
86	pub schema_version: u32,
87	pub id: String,
88	pub created_at: DateTime<Utc>,
89	pub updated_at: DateTime<Utc>,
90	pub model: String,
91	pub workspace: PathBuf,
92	pub mode: String,
93	pub allow_shell: bool,
94	pub trust_mode: bool,
95	pub auto_approve: bool,
96	#[serde(skip_serializing_if = "Option::is_none")]
97	pub latest_turn_id: Option<String>,
98	#[serde(skip_serializing_if = "Option::is_none")]
99	pub latest_response_bookmark: Option<String>,
100	#[serde(default)]
101	pub archived: bool,
102	#[serde(default, skip_serializing_if = "Option::is_none")]
103	pub system_prompt: Option<String>,
104	#[serde(default, skip_serializing_if = "Option::is_none")]
105	pub task_id: Option<String>,
106	/// User-set title for the thread. When `None`, consumers fall back to a
107	/// derived title (typically the latest turn's input summary). Added in
108	/// v0.8.10 (#562); old runtime records simply have no `title` and behave
109	/// as before. Schema version is not bumped because this field is purely
110	/// additive metadata — older readers ignore it without misinterpretation.
111	#[serde(default, skip_serializing_if = "Option::is_none")]
112	pub title: Option<String>,
113	#[serde(default)]
114	pub coherence_state: CoherenceState,
115	}
116
117	#[derive(Debug, Clone, Serialize, Deserialize)]
118	pub struct TurnRecord {
119	#[serde(default = "default_runtime_schema_version")]
120	pub schema_version: u32,
121	pub id: String,
122	pub thread_id: String,
123	pub status: RuntimeTurnStatus,
124	pub input_summary: String,
125	pub created_at: DateTime<Utc>,
126	#[serde(skip_serializing_if = "Option::is_none")]
127	pub started_at: Option<DateTime<Utc>>,
128	#[serde(skip_serializing_if = "Option::is_none")]
129	pub ended_at: Option<DateTime<Utc>>,
130	#[serde(skip_serializing_if = "Option::is_none")]
131	pub duration_ms: Option<u64>,
132	#[serde(skip_serializing_if = "Option::is_none")]
133	pub usage: Option<Usage>,
134	#[serde(skip_serializing_if = "Option::is_none")]
135	pub error: Option<String>,
136	#[serde(default)]
137	pub item_ids: Vec<String>,
138	#[serde(default)]
139	pub steer_count: usize,
140	}
141
142	#[derive(Debug, Clone, Serialize, Deserialize)]
143	pub struct TurnItemRecord {
144	#[serde(default = "default_runtime_schema_version")]
145	pub schema_version: u32,
146	pub id: String,
147	pub turn_id: String,
148	pub kind: TurnItemKind,
149	pub status: TurnItemLifecycleStatus,
150	pub summary: String,
151	#[serde(skip_serializing_if = "Option::is_none")]
152	pub detail: Option<String>,
153	#[serde(default, skip_serializing_if = "Option::is_none")]
154	pub metadata: Option<Value>,
155	#[serde(default)]
156	pub artifact_refs: Vec<PathBuf>,
157	#[serde(skip_serializing_if = "Option::is_none")]
158	pub started_at: Option<DateTime<Utc>>,
159	#[serde(skip_serializing_if = "Option::is_none")]
160	pub ended_at: Option<DateTime<Utc>>,
161	}
162
163	#[derive(Debug, Clone, Serialize, Deserialize)]
164	pub struct RuntimeEventRecord {
165	#[serde(default = "default_runtime_schema_version")]
166	pub schema_version: u32,
167	pub seq: u64,
168	pub timestamp: DateTime<Utc>,
169	pub thread_id: String,
170	#[serde(skip_serializing_if = "Option::is_none")]
171	pub turn_id: Option<String>,
172	#[serde(skip_serializing_if = "Option::is_none")]
173	pub item_id: Option<String>,
174	pub event: String,
175	pub payload: Value,
176	}
177
178	#[derive(Debug, Clone, Serialize, Deserialize)]
179	pub struct RuntimeStoreState {
180	#[serde(default = "default_runtime_schema_version")]
181	schema_version: u32,
182	next_seq: u64,
183	}
184
185	impl Default for RuntimeStoreState {
186	fn default() -> Self {
187	Self {
188	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
189	next_seq: 1,
190	}
191	}
192	}
193
194	#[derive(Debug, Clone)]
195	pub struct RuntimeThreadStore {
196	threads_dir: PathBuf,
197	turns_dir: PathBuf,
198	items_dir: PathBuf,
199	events_dir: PathBuf,
200	state_path: PathBuf,
201	state: Arc<Mutex<RuntimeStoreState>>,
202	}
203
204	impl RuntimeThreadStore {
205	pub fn open(root: PathBuf) -> Result<Self> {
206	let threads_dir = root.join("threads");
207	let turns_dir = root.join("turns");
208	let items_dir = root.join("items");
209	let events_dir = root.join("events");
210	fs::create_dir_all(&threads_dir)
211	.with_context(\|\| format!("Failed to create {}", threads_dir.display()))?;
212	fs::create_dir_all(&turns_dir)
213	.with_context(\|\| format!("Failed to create {}", turns_dir.display()))?;
214	fs::create_dir_all(&items_dir)
215	.with_context(\|\| format!("Failed to create {}", items_dir.display()))?;
216	fs::create_dir_all(&events_dir)
217	.with_context(\|\| format!("Failed to create {}", events_dir.display()))?;
218
219	let state_path = root.join("state.json");
220	let state = if state_path.exists() {
221	let raw = fs::read_to_string(&state_path)
222	.with_context(\|\| format!("Failed to read {}", state_path.display()))?;
223	serde_json::from_str::<RuntimeStoreState>(&raw)
224	.with_context(\|\| format!("Failed to parse {}", state_path.display()))?
225	} else {
226	let default = RuntimeStoreState::default();
227	write_json_atomic(&state_path, &default)?;
228	default
229	};
230
231	Ok(Self {
232	threads_dir,
233	turns_dir,
234	items_dir,
235	events_dir,
236	state_path,
237	state: Arc::new(Mutex::new(state)),
238	})
239	}
240
241	fn thread_path(&self, thread_id: &str) -> PathBuf {
242	self.threads_dir.join(format!("{thread_id}.json"))
243	}
244
245	fn turn_path(&self, turn_id: &str) -> PathBuf {
246	self.turns_dir.join(format!("{turn_id}.json"))
247	}
248
249	fn item_path(&self, item_id: &str) -> PathBuf {
250	self.items_dir.join(format!("{item_id}.json"))
251	}
252
253	fn events_path(&self, thread_id: &str) -> PathBuf {
254	self.events_dir.join(format!("{thread_id}.jsonl"))
255	}
256
257	pub fn save_thread(&self, thread: &ThreadRecord) -> Result<()> {
258	write_json_atomic(&self.thread_path(&thread.id), thread)
259	}
260
261	pub fn save_turn(&self, turn: &TurnRecord) -> Result<()> {
262	write_json_atomic(&self.turn_path(&turn.id), turn)
263	}
264
265	pub fn save_item(&self, item: &TurnItemRecord) -> Result<()> {
266	write_json_atomic(&self.item_path(&item.id), item)
267	}
268
269	pub fn load_thread(&self, thread_id: &str) -> Result<ThreadRecord> {
270	let path = self.thread_path(thread_id);
271	let raw = fs::read_to_string(&path)
272	.with_context(\|\| format!("Failed to read thread {}", path.display()))?;
273	let record: ThreadRecord = serde_json::from_str(&raw)
274	.with_context(\|\| format!("Failed to parse thread {}", path.display()))?;
275	if record.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
276	bail!(
277	"Thread schema v{} is newer than supported v{}",
278	record.schema_version,
279	CURRENT_RUNTIME_SCHEMA_VERSION
280	);
281	}
282	Ok(record)
283	}
284
285	pub fn load_turn(&self, turn_id: &str) -> Result<TurnRecord> {
286	let path = self.turn_path(turn_id);
287	let raw = fs::read_to_string(&path)
288	.with_context(\|\| format!("Failed to read turn {}", path.display()))?;
289	let record: TurnRecord = serde_json::from_str(&raw)
290	.with_context(\|\| format!("Failed to parse turn {}", path.display()))?;
291	if record.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
292	bail!(
293	"Turn schema v{} is newer than supported v{}",
294	record.schema_version,
295	CURRENT_RUNTIME_SCHEMA_VERSION
296	);
297	}
298	Ok(record)
299	}
300
301	pub fn load_item(&self, item_id: &str) -> Result<TurnItemRecord> {
302	let path = self.item_path(item_id);
303	let raw = fs::read_to_string(&path)
304	.with_context(\|\| format!("Failed to read item {}", path.display()))?;
305	let record: TurnItemRecord = serde_json::from_str(&raw)
306	.with_context(\|\| format!("Failed to parse item {}", path.display()))?;
307	if record.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
308	bail!(
309	"Item schema v{} is newer than supported v{}",
310	record.schema_version,
311	CURRENT_RUNTIME_SCHEMA_VERSION
312	);
313	}
314	Ok(record)
315	}
316
317	pub fn list_threads(&self) -> Result<Vec<ThreadRecord>> {
318	let mut out = Vec::new();
319	for entry in fs::read_dir(&self.threads_dir)
320	.with_context(\|\| format!("Failed to read {}", self.threads_dir.display()))?
321	{
322	let entry = entry?;
323	let path = entry.path();
324	if path.extension().is_none_or(\|ext\| ext != "json") {
325	continue;
326	}
327	let raw = fs::read_to_string(&path)
328	.with_context(\|\| format!("Failed to read {}", path.display()))?;
329	let thread: ThreadRecord = serde_json::from_str(&raw)
330	.with_context(\|\| format!("Failed to parse {}", path.display()))?;
331	if thread.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
332	bail!(
333	"Thread schema v{} is newer than supported v{}",
334	thread.schema_version,
335	CURRENT_RUNTIME_SCHEMA_VERSION
336	);
337	}
338	out.push(thread);
339	}
340	out.sort_by_key(\|t\| std::cmp::Reverse(t.updated_at));
341	Ok(out)
342	}
343
344	pub fn list_turns_for_thread(&self, thread_id: &str) -> Result<Vec<TurnRecord>> {
345	let mut out = Vec::new();
346	for entry in fs::read_dir(&self.turns_dir)
347	.with_context(\|\| format!("Failed to read {}", self.turns_dir.display()))?
348	{
349	let entry = entry?;
350	let path = entry.path();
351	if path.extension().is_none_or(\|ext\| ext != "json") {
352	continue;
353	}
354	let raw = fs::read_to_string(&path)
355	.with_context(\|\| format!("Failed to read {}", path.display()))?;
356	let turn: TurnRecord = serde_json::from_str(&raw)
357	.with_context(\|\| format!("Failed to parse {}", path.display()))?;
358	if turn.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
359	bail!(
360	"Turn schema v{} is newer than supported v{}",
361	turn.schema_version,
362	CURRENT_RUNTIME_SCHEMA_VERSION
363	);
364	}
365	if turn.thread_id == thread_id {
366	out.push(turn);
367	}
368	}
369	out.sort_by_key(\|a\| a.created_at);
370	Ok(out)
371	}
372
373	pub fn list_items_for_turn(&self, turn_id: &str) -> Result<Vec<TurnItemRecord>> {
374	let mut out = Vec::new();
375	for entry in fs::read_dir(&self.items_dir)
376	.with_context(\|\| format!("Failed to read {}", self.items_dir.display()))?
377	{
378	let entry = entry?;
379	let path = entry.path();
380	if path.extension().is_none_or(\|ext\| ext != "json") {
381	continue;
382	}
383	let raw = fs::read_to_string(&path)
384	.with_context(\|\| format!("Failed to read {}", path.display()))?;
385	let item: TurnItemRecord = serde_json::from_str(&raw)
386	.with_context(\|\| format!("Failed to parse {}", path.display()))?;
387	if item.schema_version > CURRENT_RUNTIME_SCHEMA_VERSION {
388	bail!(
389	"Item schema v{} is newer than supported v{}",
390	item.schema_version,
391	CURRENT_RUNTIME_SCHEMA_VERSION
392	);
393	}
394	if item.turn_id == turn_id {
395	out.push(item);
396	}
397	}
398	out.sort_by(\|a, b\| {
399	let left = a.started_at.unwrap_or_else(Utc::now);
400	let right = b.started_at.unwrap_or_else(Utc::now);
401	left.cmp(&right)
402	});
403	Ok(out)
404	}
405
406	pub async fn append_event(
407	&self,
408	thread_id: &str,
409	turn_id: Option<&str>,
410	item_id: Option<&str>,
411	event: impl Into<String>,
412	payload: Value,
413	) -> Result<RuntimeEventRecord> {
414	let mut state = self.state.lock().await;
415	let seq = state.next_seq;
416	state.next_seq = state.next_seq.saturating_add(1);
417	write_json_atomic(&self.state_path, &*state)?;
418	drop(state);
419
420	let record = RuntimeEventRecord {
421	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
422	seq,
423	timestamp: Utc::now(),
424	thread_id: thread_id.to_string(),
425	turn_id: turn_id.map(ToString::to_string),
426	item_id: item_id.map(ToString::to_string),
427	event: event.into(),
428	payload,
429	};
430
431	let path = self.events_path(thread_id);
432	let mut file = OpenOptions::new()
433	.create(true)
434	.append(true)
435	.open(&path)
436	.with_context(\|\| format!("Failed to open {}", path.display()))?;
437	let line = serde_json::to_string(&record)?;
438	writeln!(file, "{line}").with_context(\|\| format!("Failed to append {}", path.display()))?;
439	file.flush()
440	.with_context(\|\| format!("Failed to flush {}", path.display()))?;
441	file.sync_all()
442	.with_context(\|\| format!("Failed to fsync {}", path.display()))?;
443	Ok(record)
444	}
445
446	pub fn events_since(
447	&self,
448	thread_id: &str,
449	since_seq: Option<u64>,
450	) -> Result<Vec<RuntimeEventRecord>> {
451	let path = self.events_path(thread_id);
452	if !path.exists() {
453	return Ok(Vec::new());
454	}
455	let file =
456	File::open(&path).with_context(\|\| format!("Failed to open {}", path.display()))?;
457	let reader = BufReader::new(file);
458	let mut out = Vec::new();
459	for line in reader.lines() {
460	let line = line?;
461	if line.trim().is_empty() {
462	continue;
463	}
464	let event: RuntimeEventRecord = serde_json::from_str(&line)
465	.with_context(\|\| format!("Failed to parse event line in {}", path.display()))?;
466	if let Some(since) = since_seq
467	&& event.seq <= since
468	{
469	continue;
470	}
471	out.push(event);
472	}
473	Ok(out)
474	}
475
476	pub async fn current_seq(&self) -> u64 {
477	let state = self.state.lock().await;
478	state.next_seq.saturating_sub(1)
479	}
480	}
481
482	#[derive(Debug, Clone)]
483	pub struct RuntimeThreadManagerConfig {
484	pub data_dir: PathBuf,
485	pub task_data_dir: PathBuf,
486	pub max_active_threads: usize,
487	}
488
489	impl RuntimeThreadManagerConfig {
490	#[must_use]
491	pub fn from_task_data_dir(task_data_dir: PathBuf) -> Self {
492	let data_dir = if let Ok(override_dir) = std::env::var("DEEPSEEK_RUNTIME_DIR") {
493	if override_dir.trim().is_empty() {
494	task_data_dir.join("runtime")
495	} else {
496	PathBuf::from(override_dir)
497	}
498	} else {
499	task_data_dir.join("runtime")
500	};
501	Self {
502	data_dir,
503	task_data_dir,
504	max_active_threads: MAX_ACTIVE_THREADS_DEFAULT,
505	}
506	}
507	}
508
509	/// Visibility filter for `list_threads`. Default is `ActiveOnly`. The runtime
510	/// API exposes this as the combination of `include_archived` and
511	/// `archived_only` query params (see `runtime_api.rs`); whalescale#260 / #563.
512	#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
513	pub enum ThreadListFilter {
514	/// Only `archived = false` threads. The original default.
515	#[default]
516	ActiveOnly,
517	/// Active and archived threads, sorted as the store returns them.
518	IncludeArchived,
519	/// Only `archived = true` threads.
520	ArchivedOnly,
521	}
522
523	#[derive(Debug, Clone, Serialize, Deserialize)]
524	pub struct CreateThreadRequest {
525	pub model: Option<String>,
526	pub workspace: Option<PathBuf>,
527	pub mode: Option<String>,
528	pub allow_shell: Option<bool>,
529	pub trust_mode: Option<bool>,
530	pub auto_approve: Option<bool>,
531	#[serde(default)]
532	pub archived: bool,
533	#[serde(default)]
534	pub system_prompt: Option<String>,
535	#[serde(default)]
536	pub task_id: Option<String>,
537	}
538
539	/// Mutable fields accepted by `PATCH /v1/threads/{id}`.
540	///
541	/// Each field is optional — missing means "no change". Extended in v0.8.10
542	/// (#562, whalescale#256) so the UI can flip persistent thread state without
543	/// having to recreate a thread or pass per-turn overrides on every send.
544	#[derive(Debug, Clone, Serialize, Deserialize, Default)]
545	pub struct UpdateThreadRequest {
546	pub archived: Option<bool>,
547	pub allow_shell: Option<bool>,
548	pub trust_mode: Option<bool>,
549	pub auto_approve: Option<bool>,
550	pub model: Option<String>,
551	pub mode: Option<String>,
552	pub title: Option<String>,
553	pub system_prompt: Option<String>,
554	}
555
556	#[derive(Debug, Clone, Serialize, Deserialize)]
557	pub struct StartTurnRequest {
558	pub prompt: String,
559	#[serde(default)]
560	pub input_summary: Option<String>,
561	pub model: Option<String>,
562	pub mode: Option<String>,
563	pub allow_shell: Option<bool>,
564	pub trust_mode: Option<bool>,
565	pub auto_approve: Option<bool>,
566	}
567
568	#[derive(Debug, Clone, Serialize, Deserialize)]
569	pub struct SteerTurnRequest {
570	pub prompt: String,
571	}
572
573	#[derive(Debug, Clone, Serialize, Deserialize, Default)]
574	pub struct CompactThreadRequest {
575	#[serde(default)]
576	pub reason: Option<String>,
577	}
578
579	#[derive(Debug, Clone, Serialize, Deserialize)]
580	pub struct ThreadDetail {
581	pub thread: ThreadRecord,
582	pub turns: Vec<TurnRecord>,
583	pub items: Vec<TurnItemRecord>,
584	pub latest_seq: u64,
585	}
586
587	/// Aggregation key for `aggregate_usage`. Whalescale#261 / #564.
588	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
589	pub enum UsageGroupBy {
590	Day,
591	Model,
592	Provider,
593	Thread,
594	}
595
596	#[derive(Debug, Clone, Default, Serialize)]
597	pub struct UsageTotals {
598	pub input_tokens: u64,
599	pub output_tokens: u64,
600	pub cached_tokens: u64,
601	pub reasoning_tokens: u64,
602	pub cost_usd: f64,
603	pub turns: u64,
604	}
605
606	#[derive(Debug, Clone, Default, Serialize)]
607	pub struct UsageBucket {
608	pub key: String,
609	pub input_tokens: u64,
610	pub output_tokens: u64,
611	pub cached_tokens: u64,
612	pub reasoning_tokens: u64,
613	pub cost_usd: f64,
614	pub turns: u64,
615	}
616
617	#[derive(Debug, Clone, Serialize)]
618	pub struct UsageAggregation {
619	pub since: Option<DateTime<Utc>>,
620	pub until: Option<DateTime<Utc>>,
621	pub group_by: String,
622	pub totals: UsageTotals,
623	pub buckets: Vec<UsageBucket>,
624	}
625
626	/// Best-effort provider classification from a model name. Used as a grouping
627	/// key for `/v1/usage?group_by=provider`. Cost-tracking already runs the
628	/// model→pricing→cost path; this only labels the bucket.
629	fn provider_label_for_model(model: &str) -> &'static str {
630	if model.starts_with("deepseek-ai/") {
631	"nvidia-nim"
632	} else if model.starts_with("deepseek-") {
633	"deepseek"
634	} else if model.starts_with("openai/") \|\| model.starts_with("anthropic/") {
635	"openrouter"
636	} else {
637	"unknown"
638	}
639	}
640
641	#[derive(Debug, Clone)]
642	struct ActiveTurnState {
643	turn_id: String,
644	interrupt_requested: bool,
645	auto_approve: bool,
646	trust_mode: bool,
647	}
648
649	#[derive(Clone)]
650	struct ActiveThreadState {
651	engine: EngineHandle,
652	active_turn: Option<ActiveTurnState>,
653	}
654
655	#[derive(Default)]
656	struct ActiveThreads {
657	engines: HashMap<String, ActiveThreadState>,
658	lru: VecDeque<String>,
659	}
660
661	pub type SharedRuntimeThreadManager = Arc<RuntimeThreadManager>;
662
663	/// Manages active engine threads, lifecycle, and event persistence.
664	///
665	/// # Lock ordering invariant
666	///
667	/// Two `Mutex`es exist across this module:
668	/// - `RuntimeThreadStore::state` — protects the monotonic event sequence counter.
669	/// - `RuntimeThreadManager::active` — protects the set of loaded engine handles.
670	///
671	/// No code path holds both locks simultaneously. The `state` lock is only
672	/// acquired inside `RuntimeThreadStore::append_event` (where it is explicitly
673	/// dropped before any I/O) and `current_seq`. All `emit_event` calls (which
674	/// call `append_event`) happen after `active` has been released. If you add
675	/// new code that touches both, always acquire `state` before `active` to
676	/// preserve a consistent ordering.
677	#[derive(Clone)]
678	pub struct RuntimeThreadManager {
679	config: Config,
680	workspace: PathBuf,
681	store: RuntimeThreadStore,
682	active: Arc<Mutex<ActiveThreads>>,
683	event_tx: broadcast::Sender<RuntimeEventRecord>,
684	manager_cfg: RuntimeThreadManagerConfig,
685	cancel_token: CancellationToken,
686	task_manager: Arc<StdMutex<Option<crate::task_manager::SharedTaskManager>>>,
687	automations: Arc<StdMutex<Option<crate::automation_manager::SharedAutomationManager>>>,
688	}
689
690	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
691	enum RuntimeApprovalDecision {
692	ApproveTool,
693	DenyTool,
694	RetryWithFullAccess,
695	}
696
697	impl RuntimeThreadManager {
698	pub fn open(
699	config: Config,
700	workspace: PathBuf,
701	manager_cfg: RuntimeThreadManagerConfig,
702	) -> Result<Self> {
703	let store = RuntimeThreadStore::open(manager_cfg.data_dir.clone())?;
704	let (event_tx, _event_rx) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
705	let manager = Self {
706	config,
707	workspace,
708	store,
709	active: Arc::new(Mutex::new(ActiveThreads::default())),
710	event_tx,
711	manager_cfg,
712	cancel_token: CancellationToken::new(),
713	task_manager: Arc::new(StdMutex::new(None)),
714	automations: Arc::new(StdMutex::new(None)),
715	};
716	manager.recover_interrupted_state()?;
717	Ok(manager)
718	}
719
720	/// Attach the durable task manager so model-visible task tools work inside
721	/// runtime thread turns as well as interactive TUI turns.
722	pub fn attach_task_manager(&self, task_manager: crate::task_manager::SharedTaskManager) {
723	if let Ok(mut slot) = self.task_manager.lock() {
724	*slot = Some(task_manager);
725	}
726	}
727
728	/// Attach the automation manager for model-visible scheduling tools.
729	pub fn attach_automation_manager(
730	&self,
731	automations: crate::automation_manager::SharedAutomationManager,
732	) {
733	if let Ok(mut slot) = self.automations.lock() {
734	*slot = Some(automations);
735	}
736	}
737
738	#[allow(dead_code)] // Public API for external callers (runtime API, task manager)
739	pub fn shutdown(&self) {
740	self.cancel_token.cancel();
741	}
742
743	#[allow(dead_code)] // Public API for external callers
744	pub fn is_shutdown(&self) -> bool {
745	self.cancel_token.is_cancelled()
746	}
747
748	#[must_use]
749	pub fn subscribe_events(&self) -> broadcast::Receiver<RuntimeEventRecord> {
750	self.event_tx.subscribe()
751	}
752
753	async fn emit_event(
754	&self,
755	thread_id: &str,
756	turn_id: Option<&str>,
757	item_id: Option<&str>,
758	event: impl Into<String>,
759	payload: Value,
760	) -> Result<RuntimeEventRecord> {
761	let record = self
762	.store
763	.append_event(thread_id, turn_id, item_id, event, payload)
764	.await?;
765	if let Err(e) = self.event_tx.send(record.clone()) {
766	tracing::debug!(
767	"Runtime event broadcast failed (no receivers or channel full): {}",
768	e
769	);
770	}
771	Ok(record)
772	}
773
774	pub async fn create_thread(&self, req: CreateThreadRequest) -> Result<ThreadRecord> {
775	let now = Utc::now();
776	let model = req
777	.model
778	.filter(\|m\| !m.trim().is_empty())
779	.or_else(\|\| self.config.default_text_model.clone())
780	.unwrap_or_else(\|\| DEFAULT_TEXT_MODEL.to_string());
781	let workspace = req.workspace.unwrap_or_else(\|\| self.workspace.clone());
782	let mode = req
783	.mode
784	.filter(\|m\| !m.trim().is_empty())
785	.unwrap_or_else(\|\| "agent".to_string());
786	let allow_shell = req.allow_shell.unwrap_or_else(\|\| self.config.allow_shell());
787	let trust_mode = req.trust_mode.unwrap_or(false);
788	let auto_approve = req.auto_approve.unwrap_or(false);
789
790	let thread = ThreadRecord {
791	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
792	id: format!("thr_{}", &Uuid::new_v4().to_string()[..8]),
793	created_at: now,
794	updated_at: now,
795	model,
796	workspace,
797	mode,
798	allow_shell,
799	trust_mode,
800	auto_approve,
801	latest_turn_id: None,
802	latest_response_bookmark: None,
803	archived: req.archived,
804	system_prompt: req.system_prompt,
805	task_id: req.task_id,
806	title: None,
807	coherence_state: CoherenceState::default(),
808	};
809	self.store.save_thread(&thread)?;
810	self.emit_event(
811	&thread.id,
812	None,
813	None,
814	"thread.started",
815	json!({ "thread": thread }),
816	)
817	.await?;
818	Ok(thread)
819	}
820
821	pub async fn list_threads(
822	&self,
823	filter: ThreadListFilter,
824	limit: Option<usize>,
825	) -> Result<Vec<ThreadRecord>> {
826	let mut threads = self.store.list_threads()?;
827	match filter {
828	ThreadListFilter::ActiveOnly => threads.retain(\|t\| !t.archived),
829	ThreadListFilter::ArchivedOnly => threads.retain(\|t\| t.archived),
830	ThreadListFilter::IncludeArchived => {}
831	}
832	if let Some(limit) = limit {
833	threads.truncate(limit);
834	}
835	Ok(threads)
836	}
837
838	/// Aggregate token + cost usage across all threads/turns inside the time
839	/// range `[since, until]`. Each turn's cost is computed via
840	/// `pricing::calculate_turn_cost_from_usage` using the thread's model
841	/// (turns inherit it). Whalescale#261 / #564.
842	///
843	/// Buckets are sorted by ascending key for deterministic output. Empty
844	/// ranges produce empty `buckets` (never an error).
845	pub async fn aggregate_usage(
846	&self,
847	since: Option<DateTime<Utc>>,
848	until: Option<DateTime<Utc>>,
849	group_by: UsageGroupBy,
850	) -> Result<UsageAggregation> {
851	use std::collections::BTreeMap;
852
853	let mut buckets: BTreeMap<String, UsageBucket> = BTreeMap::new();
854	let mut totals = UsageTotals::default();
855
856	for thread in self.store.list_threads()? {
857	let turns = self.store.list_turns_for_thread(&thread.id)?;
858	for turn in turns {
859	if let Some(s) = since
860	&& turn.created_at < s
861	{
862	continue;
863	}
864	if let Some(u) = until
865	&& turn.created_at > u
866	{
867	continue;
868	}
869	let Some(usage) = turn.usage.as_ref() else {
870	continue;
871	};
872	let cached = usage.prompt_cache_hit_tokens.unwrap_or(0) as u64;
873	let reasoning = usage.reasoning_tokens.unwrap_or(0) as u64;
874	let input = usage.input_tokens as u64;
875	let output = usage.output_tokens as u64;
876	let cost = crate::pricing::calculate_turn_cost_from_usage(&thread.model, usage)
877	.unwrap_or(0.0);
878
879	totals.input_tokens += input;
880	totals.output_tokens += output;
881	totals.cached_tokens += cached;
882	totals.reasoning_tokens += reasoning;
883	totals.cost_usd += cost;
884	totals.turns += 1;
885
886	let key = match group_by {
887	UsageGroupBy::Day => turn.created_at.format("%Y-%m-%d").to_string(),
888	UsageGroupBy::Model => thread.model.clone(),
889	UsageGroupBy::Provider => provider_label_for_model(&thread.model).to_string(),
890	UsageGroupBy::Thread => thread.id.clone(),
891	};
892	let bucket = buckets.entry(key.clone()).or_insert_with(\|\| UsageBucket {
893	key,
894	..UsageBucket::default()
895	});
896	bucket.input_tokens += input;
897	bucket.output_tokens += output;
898	bucket.cached_tokens += cached;
899	bucket.reasoning_tokens += reasoning;
900	bucket.cost_usd += cost;
901	bucket.turns += 1;
902	}
903	}
904
905	let group_by_str = match group_by {
906	UsageGroupBy::Day => "day",
907	UsageGroupBy::Model => "model",
908	UsageGroupBy::Provider => "provider",
909	UsageGroupBy::Thread => "thread",
910	}
911	.to_string();
912
913	Ok(UsageAggregation {
914	since,
915	until,
916	group_by: group_by_str,
917	totals,
918	buckets: buckets.into_values().collect(),
919	})
920	}
921
922	pub async fn get_thread(&self, id: &str) -> Result<ThreadRecord> {
923	self.store
924	.load_thread(id)
925	.with_context(\|\| format!("Thread not found: {id}"))
926	}
927
928	pub async fn update_thread(&self, id: &str, req: UpdateThreadRequest) -> Result<ThreadRecord> {
929	if req.archived.is_none()
930	&& req.allow_shell.is_none()
931	&& req.trust_mode.is_none()
932	&& req.auto_approve.is_none()
933	&& req.model.is_none()
934	&& req.mode.is_none()
935	&& req.title.is_none()
936	&& req.system_prompt.is_none()
937	{
938	bail!("At least one thread field is required");
939	}
940
941	if let Some(model) = req.model.as_ref()
942	&& model.trim().is_empty()
943	{
944	bail!("model must not be empty");
945	}
946	if let Some(mode) = req.mode.as_ref()
947	&& mode.trim().is_empty()
948	{
949	bail!("mode must not be empty");
950	}
951
952	let mut thread = self.get_thread(id).await?;
953	let mut changes = serde_json::Map::new();
954
955	if let Some(archived) = req.archived
956	&& thread.archived != archived
957	{
958	thread.archived = archived;
959	changes.insert("archived".to_string(), json!(archived));
960	}
961	if let Some(allow_shell) = req.allow_shell
962	&& thread.allow_shell != allow_shell
963	{
964	thread.allow_shell = allow_shell;
965	changes.insert("allow_shell".to_string(), json!(allow_shell));
966	}
967	if let Some(trust_mode) = req.trust_mode
968	&& thread.trust_mode != trust_mode
969	{
970	thread.trust_mode = trust_mode;
971	changes.insert("trust_mode".to_string(), json!(trust_mode));
972	}
973	if let Some(auto_approve) = req.auto_approve
974	&& thread.auto_approve != auto_approve
975	{
976	thread.auto_approve = auto_approve;
977	changes.insert("auto_approve".to_string(), json!(auto_approve));
978	}
979	if let Some(model) = req.model
980	&& thread.model != model
981	{
982	thread.model = model.clone();
983	changes.insert("model".to_string(), json!(model));
984	}
985	if let Some(mode) = req.mode
986	&& thread.mode != mode
987	{
988	thread.mode = mode.clone();
989	changes.insert("mode".to_string(), json!(mode));
990	}
991	if let Some(title) = req.title {
992	// Empty string clears a previously-set title and reverts to derived.
993	let new_title = if title.trim().is_empty() {
994	None
995	} else {
996	Some(title)
997	};
998	if thread.title != new_title {
999	thread.title = new_title.clone();
1000	changes.insert("title".to_string(), json!(new_title));
1001	}
1002	}
1003	if let Some(system_prompt) = req.system_prompt {
1004	let new_sys = if system_prompt.trim().is_empty() {
1005	None
1006	} else {
1007	Some(system_prompt)
1008	};
1009	if thread.system_prompt != new_sys {
1010	thread.system_prompt = new_sys.clone();
1011	changes.insert("system_prompt".to_string(), json!(new_sys));
1012	}
1013	}
1014
1015	if !changes.is_empty() {
1016	thread.updated_at = Utc::now();
1017	self.store.save_thread(&thread)?;
1018	self.emit_event(
1019	&thread.id,
1020	None,
1021	None,
1022	"thread.updated",
1023	json!({
1024	"thread": thread.clone(),
1025	"changes": Value::Object(changes),
1026	}),
1027	)
1028	.await?;
1029	}
1030
1031	Ok(thread)
1032	}
1033
1034	pub async fn get_thread_detail(&self, id: &str) -> Result<ThreadDetail> {
1035	let thread = self.get_thread(id).await?;
1036	let turns = self.store.list_turns_for_thread(id)?;
1037	let mut items = Vec::new();
1038	for turn in &turns {
1039	items.extend(self.store.list_items_for_turn(&turn.id)?);
1040	}
1041	let latest_seq = self.store.current_seq().await;
1042	Ok(ThreadDetail {
1043	thread,
1044	turns,
1045	items,
1046	latest_seq,
1047	})
1048	}
1049
1050	pub async fn resume_thread(&self, id: &str) -> Result<ThreadRecord> {
1051	let thread = self.get_thread(id).await?;
1052	self.ensure_engine_loaded(&thread).await?;
1053	Ok(thread)
1054	}
1055
1056	/// Resume a thread and recover the sub-agent rebind hints needed to
1057	/// reconstruct in-transcript cards (issue #128). Drains the persisted
1058	/// `agent.*` event stream and collapses it into the latest known
1059	/// status per `agent_id` — the UI consumes this to seed empty
1060	/// `DelegateCard` / `FanoutCard` placeholders so subsequent live
1061	/// mailbox envelopes mutate them in place.
1062	#[allow(dead_code)] // exposed for the runtime API resume flow; consumed by #128 follow-up.
1063	pub async fn resume_thread_with_agent_rebind(
1064	&self,
1065	id: &str,
1066	) -> Result<(ThreadRecord, Vec<AgentRebindHint>)> {
1067	let thread = self.resume_thread(id).await?;
1068	let events = self.store.events_since(&thread.id, None)?;
1069	let hints = collect_agent_rebind_hints(&events);
1070	Ok((thread, hints))
1071	}
1072
1073	pub async fn fork_thread(&self, id: &str) -> Result<ThreadRecord> {
1074	let source = self.get_thread(id).await?;
1075	let mut forked = source.clone();
1076	let now = Utc::now();
1077	forked.id = format!("thr_{}", &Uuid::new_v4().to_string()[..8]);
1078	forked.created_at = now;
1079	forked.updated_at = now;
1080	forked.latest_turn_id = None;
1081	forked.archived = false;
1082	self.store.save_thread(&forked)?;
1083
1084	let source_turns = self.store.list_turns_for_thread(&source.id)?;
1085	for source_turn in source_turns {
1086	let mut cloned_turn = source_turn.clone();
1087	cloned_turn.id = format!("turn_{}", &Uuid::new_v4().to_string()[..8]);
1088	cloned_turn.thread_id = forked.id.clone();
1089	cloned_turn.item_ids.clear();
1090	self.store.save_turn(&cloned_turn)?;
1091
1092	let items = self.store.list_items_for_turn(&source_turn.id)?;
1093	for item in items {
1094	let mut cloned_item = item.clone();
1095	cloned_item.id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
1096	cloned_item.turn_id = cloned_turn.id.clone();
1097	self.store.save_item(&cloned_item)?;
1098	cloned_turn.item_ids.push(cloned_item.id.clone());
1099	}
1100	self.store.save_turn(&cloned_turn)?;
1101	forked.latest_turn_id = Some(cloned_turn.id.clone());
1102	forked.updated_at = now;
1103	self.store.save_thread(&forked)?;
1104	}
1105
1106	self.emit_event(
1107	&forked.id,
1108	None,
1109	None,
1110	"thread.forked",
1111	json!({
1112	"thread": forked,
1113	"source_thread_id": source.id,
1114	}),
1115	)
1116	.await?;
1117	Ok(forked)
1118	}
1119
1120	/// Fork a thread, dropping every turn from the Nth-from-tail user
1121	/// message onward (issue #133 — Esc-Esc backtrack).
1122	///
1123	/// `depth_from_tail` selects which user turn to roll back to:
1124	///
1125	/// - `0` — drop the most recent turn (the freshest user message and
1126	/// everything after it)
1127	/// - `1` — drop the two most recent turns (rewind one further)
1128	/// - …and so on
1129	///
1130	/// Returns a tuple of `(forked_thread, original_user_text)` where the
1131	/// second element is the `detail` of the first `UserMessage` item in
1132	/// the first dropped turn — i.e. the input the user typed to start
1133	/// that turn — so the caller can pre-populate the composer with it.
1134	/// `None` when no detail was recorded (defensive — every persisted
1135	/// `UserMessage` since v0.6 carries a detail string).
1136	///
1137	/// Counts user turns by iterating `list_turns_for_thread` (sorted
1138	/// oldest → newest) backwards. A turn is counted as a "user turn"
1139	/// when at least one of its items has `kind ==
1140	/// TurnItemKind::UserMessage`. Steered turns (which append additional
1141	/// `UserMessage` items) still count as one turn — backtrack rewinds
1142	/// at the turn boundary, not at the steer boundary.
1143	///
1144	/// Errors:
1145	/// - `depth_from_tail` exceeds the number of user turns
1146	/// - source thread not found
1147	#[allow(dead_code)] // exposed for the runtime/HTTP fork-on-backtrack path; the in-TUI Esc-Esc flow trims `App` state directly. Issue #133.
1148	pub async fn fork_at_user_message(
1149	&self,
1150	id: &str,
1151	depth_from_tail: usize,
1152	) -> Result<(ThreadRecord, Option<String>)> {
1153	let source = self.get_thread(id).await?;
1154	let source_turns = self.store.list_turns_for_thread(&source.id)?;
1155
1156	// Walk turns from newest to oldest. For each turn, ask: does it
1157	// contain a UserMessage item? If yes, it counts toward the depth.
1158	let mut user_turn_indices: Vec<usize> = Vec::new();
1159	for (idx, turn) in source_turns.iter().enumerate().rev() {
1160	let items = self.store.list_items_for_turn(&turn.id)?;
1161	if items
1162	.iter()
1163	.any(\|item\| item.kind == TurnItemKind::UserMessage)
1164	{
1165	user_turn_indices.push(idx);
1166	}
1167	}
1168	if depth_from_tail >= user_turn_indices.len() {
1169	bail!(
1170	"fork_at_user_message: depth {} exceeds {} user turn(s)",
1171	depth_from_tail,
1172	user_turn_indices.len()
1173	);
1174	}
1175	// `user_turn_indices` is newest-first because we iterated in
1176	// reverse, so the Nth element is exactly the Nth-from-tail user
1177	// turn in the original chronological list.
1178	let target_turn_idx = user_turn_indices[depth_from_tail];
1179	let target_turn_id = source_turns[target_turn_idx].id.clone();
1180
1181	// Pull the original user-message text out of the dropped turn so
1182	// the caller can drop it back into the composer.
1183	let target_items = self.store.list_items_for_turn(&target_turn_id)?;
1184	let original_user_text = target_items
1185	.iter()
1186	.find(\|item\| item.kind == TurnItemKind::UserMessage)
1187	.and_then(\|item\| item.detail.clone());
1188
1189	// Copy turns strictly before `target_turn_idx` into a new thread.
1190	// Mirrors `fork_thread` but stops at the cutoff instead of copying
1191	// every turn. Kept structurally close so future parity reviews
1192	// can spot drift between the two paths.
1193	let mut forked = source.clone();
1194	let now = Utc::now();
1195	forked.id = format!("thr_{}", &Uuid::new_v4().to_string()[..8]);
1196	forked.created_at = now;
1197	forked.updated_at = now;
1198	forked.latest_turn_id = None;
1199	forked.archived = false;
1200	self.store.save_thread(&forked)?;
1201
1202	for source_turn in source_turns.iter().take(target_turn_idx) {
1203	let mut cloned_turn = source_turn.clone();
1204	cloned_turn.id = format!("turn_{}", &Uuid::new_v4().to_string()[..8]);
1205	cloned_turn.thread_id = forked.id.clone();
1206	cloned_turn.item_ids.clear();
1207	self.store.save_turn(&cloned_turn)?;
1208
1209	let items = self.store.list_items_for_turn(&source_turn.id)?;
1210	for item in items {
1211	let mut cloned_item = item.clone();
1212	cloned_item.id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
1213	cloned_item.turn_id = cloned_turn.id.clone();
1214	self.store.save_item(&cloned_item)?;
1215	cloned_turn.item_ids.push(cloned_item.id.clone());
1216	}
1217	self.store.save_turn(&cloned_turn)?;
1218	forked.latest_turn_id = Some(cloned_turn.id.clone());
1219	forked.updated_at = now;
1220	self.store.save_thread(&forked)?;
1221	}
1222
1223	self.emit_event(
1224	&forked.id,
1225	None,
1226	None,
1227	"thread.forked",
1228	json!({
1229	"thread": forked,
1230	"source_thread_id": source.id,
1231	"backtrack_depth_from_tail": depth_from_tail,
1232	"dropped_turn_id": target_turn_id,
1233	}),
1234	)
1235	.await?;
1236	Ok((forked, original_user_text))
1237	}
1238
1239	/// Seed a thread with messages from a saved session so subsequent turns
1240	/// continue with the prior conversation context.
1241	pub async fn seed_thread_from_messages(
1242	&self,
1243	thread_id: &str,
1244	messages: &[Message],
1245	) -> Result<()> {
1246	let mut thread = self.get_thread(thread_id).await?;
1247	let now = Utc::now();
1248
1249	let mut user_buf: Vec<String> = Vec::new();
1250	let mut pending_pairs: Vec<(String, Option<String>)> = Vec::new();
1251
1252	for msg in messages {
1253	let text = msg
1254	.content
1255	.iter()
1256	.filter_map(\|block\| match block {
1257	ContentBlock::Text { text, .. } => Some(text.as_str()),
1258	_ => None,
1259	})
1260	.collect::<Vec<_>>()
1261	.join("\n");
1262	if text.trim().is_empty() {
1263	continue;
1264	}
1265	if msg.role == "user" {
1266	user_buf.push(text);
1267	} else if msg.role == "assistant" {
1268	let user_text = if user_buf.is_empty() {
1269	String::new()
1270	} else {
1271	std::mem::take(&mut user_buf).join("\n")
1272	};
1273	pending_pairs.push((user_text, Some(text)));
1274	}
1275	}
1276	if !user_buf.is_empty() {
1277	let user_text = std::mem::take(&mut user_buf).join("\n");
1278	pending_pairs.push((user_text, None));
1279	}
1280
1281	for (user_text, assistant_text) in pending_pairs {
1282	let turn_id = format!("turn_{}", &Uuid::new_v4().to_string()[..8]);
1283	let summary = crate::utils::truncate_with_ellipsis(&user_text, SUMMARY_LIMIT, "...");
1284	let mut item_ids = Vec::new();
1285
1286	if !user_text.is_empty() {
1287	let item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
1288	self.store.save_item(&TurnItemRecord {
1289	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1290	id: item_id.clone(),
1291	turn_id: turn_id.clone(),
1292	kind: TurnItemKind::UserMessage,
1293	status: TurnItemLifecycleStatus::Completed,
1294	summary: summary.clone(),
1295	detail: Some(user_text),
1296	metadata: None,
1297	artifact_refs: Vec::new(),
1298	started_at: Some(now),
1299	ended_at: Some(now),
1300	})?;
1301	item_ids.push(item_id);
1302	}
1303
1304	if let Some(assistant_text) = assistant_text {
1305	let asst_summary = if assistant_text.len() > SUMMARY_LIMIT {
1306	format!("{}...", &assistant_text[..SUMMARY_LIMIT.saturating_sub(3)])
1307	} else {
1308	assistant_text.clone()
1309	};
1310	let item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
1311	self.store.save_item(&TurnItemRecord {
1312	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1313	id: item_id.clone(),
1314	turn_id: turn_id.clone(),
1315	kind: TurnItemKind::AgentMessage,
1316	status: TurnItemLifecycleStatus::Completed,
1317	summary: asst_summary,
1318	detail: Some(assistant_text),
1319	metadata: None,
1320	artifact_refs: Vec::new(),
1321	started_at: Some(now),
1322	ended_at: Some(now),
1323	})?;
1324	item_ids.push(item_id);
1325	}
1326
1327	self.store.save_turn(&TurnRecord {
1328	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1329	id: turn_id.clone(),
1330	thread_id: thread_id.to_string(),
1331	status: RuntimeTurnStatus::Completed,
1332	input_summary: summary,
1333	created_at: now,
1334	started_at: Some(now),
1335	ended_at: Some(now),
1336	duration_ms: Some(0),
1337	usage: None,
1338	error: None,
1339	item_ids,
1340	steer_count: 0,
1341	})?;
1342
1343	thread.latest_turn_id = Some(turn_id);
1344	thread.updated_at = now;
1345	}
1346
1347	self.store.save_thread(&thread)?;
1348	self.emit_event(
1349	thread_id,
1350	None,
1351	None,
1352	"thread.updated",
1353	json!({ "thread": thread, "reason": "session_resume" }),
1354	)
1355	.await?;
1356	Ok(())
1357	}
1358
1359	pub async fn start_turn(&self, thread_id: &str, req: StartTurnRequest) -> Result<TurnRecord> {
1360	let prompt = req.prompt.trim().to_string();
1361	if prompt.is_empty() {
1362	bail!("prompt is required");
1363	}
1364
1365	let mut thread = self.get_thread(thread_id).await?;
1366	let engine = self.ensure_engine_loaded(&thread).await?;
1367
1368	{
1369	let active = self.active.lock().await;
1370	if let Some(active_thread) = active.engines.get(thread_id)
1371	&& active_thread.active_turn.is_some()
1372	{
1373	bail!("Thread already has an active turn");
1374	}
1375	}
1376
1377	let now = Utc::now();
1378	let turn_id = format!("turn_{}", &Uuid::new_v4().to_string()[..8]);
1379	let mut turn = TurnRecord {
1380	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1381	id: turn_id.clone(),
1382	thread_id: thread_id.to_string(),
1383	status: RuntimeTurnStatus::InProgress,
1384	input_summary: req
1385	.input_summary
1386	.unwrap_or_else(\|\| summarize_text(&prompt, SUMMARY_LIMIT)),
1387	created_at: now,
1388	started_at: Some(now),
1389	ended_at: None,
1390	duration_ms: None,
1391	usage: None,
1392	error: None,
1393	item_ids: Vec::new(),
1394	steer_count: 0,
1395	};
1396
1397	let user_item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
1398	let user_item = TurnItemRecord {
1399	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1400	id: user_item_id.clone(),
1401	turn_id: turn_id.clone(),
1402	kind: TurnItemKind::UserMessage,
1403	status: TurnItemLifecycleStatus::Completed,
1404	summary: summarize_text(&prompt, SUMMARY_LIMIT),
1405	detail: Some(prompt.clone()),
1406	metadata: None,
1407	artifact_refs: Vec::new(),
1408	started_at: Some(now),
1409	ended_at: Some(now),
1410	};
1411
1412	turn.item_ids.push(user_item_id.clone());
1413	self.store.save_item(&user_item)?;
1414	self.store.save_turn(&turn)?;
1415
1416	thread.latest_turn_id = Some(turn_id.clone());
1417	thread.updated_at = now;
1418	self.store.save_thread(&thread)?;
1419
1420	self.emit_event(
1421	thread_id,
1422	Some(&turn_id),
1423	None,
1424	"turn.started",
1425	json!({ "turn": turn.clone() }),
1426	)
1427	.await?;
1428	self.emit_event(
1429	thread_id,
1430	Some(&turn_id),
1431	Some(&user_item_id),
1432	"item.started",
1433	json!({ "item": user_item.clone() }),
1434	)
1435	.await?;
1436	self.emit_event(
1437	thread_id,
1438	Some(&turn_id),
1439	Some(&user_item_id),
1440	"item.completed",
1441	json!({ "item": user_item }),
1442	)
1443	.await?;
1444
1445	{
1446	let mut active = self.active.lock().await;
1447	let Some(state) = active.engines.get_mut(thread_id) else {
1448	bail!("Thread engine not loaded");
1449	};
1450	state.active_turn = Some(ActiveTurnState {
1451	turn_id: turn_id.clone(),
1452	interrupt_requested: false,
1453	auto_approve: req.auto_approve.unwrap_or(thread.auto_approve),
1454	trust_mode: req.trust_mode.unwrap_or(thread.trust_mode),
1455	});
1456	touch_lru(&mut active.lru, thread_id);
1457	}
1458
1459	let mode = parse_mode(req.mode.as_deref().unwrap_or(&thread.mode));
1460	let requested_model = req.model.unwrap_or_else(\|\| thread.model.clone());
1461	let auto_model = requested_model.trim().eq_ignore_ascii_case("auto");
1462	let (model, reasoning_effort) = if auto_model {
1463	let selection = crate::commands::resolve_auto_route_with_flash(
1464	&self.config,
1465	&prompt,
1466	"",
1467	"auto",
1468	"auto",
1469	)
1470	.await;
1471	(
1472	selection.model,
1473	selection
1474	.reasoning_effort
1475	.map(\|effort\| effort.as_setting().to_string()),
1476	)
1477	} else {
1478	(requested_model, None)
1479	};
1480	let allow_shell = req.allow_shell.unwrap_or(thread.allow_shell);
1481	let trust_mode = req.trust_mode.unwrap_or(thread.trust_mode);
1482	let auto_approve = req.auto_approve.unwrap_or(thread.auto_approve);
1483
1484	engine
1485	.send(Op::SendMessage {
1486	content: prompt,
1487	mode,
1488	model: model.clone(),
1489	goal_objective: None,
1490	reasoning_effort,
1491	reasoning_effort_auto: auto_model,
1492	auto_model,
1493	allow_shell,
1494	trust_mode,
1495	auto_approve,
1496	approval_mode: if auto_approve {
1497	crate::tui::approval::ApprovalMode::Auto
1498	} else {
1499	crate::tui::approval::ApprovalMode::Suggest
1500	},
1501	})
1502	.await
1503	.map_err(\|e\| anyhow!("Failed to start turn: {e}"))?;
1504
1505	let manager = Arc::new(self.clone());
1506	let thread_id_owned = thread_id.to_string();
1507	let turn_id_owned = turn_id.clone();
1508	let engine_clone = engine.clone();
1509	let cancel_token = self.cancel_token.clone();
1510	tokio::spawn(async move {
1511	if cancel_token.is_cancelled() {
1512	tracing::debug!("Skipping turn monitor: shutdown requested");
1513	return;
1514	}
1515	use futures_util::FutureExt;
1516	let result = std::panic::AssertUnwindSafe(manager.monitor_turn(
1517	thread_id_owned,
1518	turn_id_owned,
1519	engine_clone,
1520	))
1521	.catch_unwind()
1522	.await;
1523	match result {
1524	Ok(res) => {
1525	if let Err(err) = res {
1526	tracing::error!("Failed to monitor turn: {err}");
1527	}
1528	}
1529	Err(panic_err) => {
1530	if let Some(msg) = panic_err.downcast_ref::<&str>() {
1531	tracing::error!("Turn monitor panicked: {}", msg);
1532	} else if let Some(msg) = panic_err.downcast_ref::<String>() {
1533	tracing::error!("Turn monitor panicked: {}", msg);
1534	} else {
1535	tracing::error!("Turn monitor panicked with unknown error");
1536	}
1537	}
1538	}
1539	});
1540
1541	Ok(turn)
1542	}
1543
1544	pub async fn interrupt_turn(&self, thread_id: &str, turn_id: &str) -> Result<TurnRecord> {
1545	{
1546	let mut active = self.active.lock().await;
1547	let Some(active_thread) = active.engines.get_mut(thread_id) else {
1548	bail!("Thread is not loaded");
1549	};
1550	let Some(active_turn) = active_thread.active_turn.as_mut() else {
1551	bail!("No active turn on thread {thread_id}");
1552	};
1553	if active_turn.turn_id != turn_id {
1554	bail!("Turn {turn_id} is not active on thread {thread_id}");
1555	}
1556	active_turn.interrupt_requested = true;
1557	active_thread.engine.cancel();
1558	touch_lru(&mut active.lru, thread_id);
1559	}
1560
1561	self.emit_event(
1562	thread_id,
1563	Some(turn_id),
1564	None,
1565	"turn.interrupt_requested",
1566	json!({ "thread_id": thread_id, "turn_id": turn_id }),
1567	)
1568	.await?;
1569
1570	self.store.load_turn(turn_id)
1571	}
1572
1573	pub async fn steer_turn(
1574	&self,
1575	thread_id: &str,
1576	turn_id: &str,
1577	req: SteerTurnRequest,
1578	) -> Result<TurnRecord> {
1579	let prompt = req.prompt.trim().to_string();
1580	if prompt.is_empty() {
1581	bail!("prompt is required");
1582	}
1583
1584	let engine = {
1585	let mut active = self.active.lock().await;
1586	let engine = {
1587	let Some(active_thread) = active.engines.get_mut(thread_id) else {
1588	bail!("Thread is not loaded");
1589	};
1590	let Some(active_turn) = active_thread.active_turn.as_mut() else {
1591	bail!("No active turn on thread {thread_id}");
1592	};
1593	if active_turn.turn_id != turn_id {
1594	bail!("Turn {turn_id} is not active on thread {thread_id}");
1595	}
1596	active_thread.engine.clone()
1597	};
1598	touch_lru(&mut active.lru, thread_id);
1599	engine
1600	};
1601
1602	engine
1603	.steer(prompt.clone())
1604	.await
1605	.map_err(\|e\| anyhow!("Failed to steer turn: {e}"))?;
1606
1607	let now = Utc::now();
1608	let mut turn = self.store.load_turn(turn_id)?;
1609	turn.steer_count = turn.steer_count.saturating_add(1);
1610	self.store.save_turn(&turn)?;
1611
1612	let item = TurnItemRecord {
1613	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1614	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
1615	turn_id: turn_id.to_string(),
1616	kind: TurnItemKind::UserMessage,
1617	status: TurnItemLifecycleStatus::Completed,
1618	summary: summarize_text(&prompt, SUMMARY_LIMIT),
1619	detail: Some(prompt.clone()),
1620	metadata: None,
1621	artifact_refs: Vec::new(),
1622	started_at: Some(now),
1623	ended_at: Some(now),
1624	};
1625	turn.item_ids.push(item.id.clone());
1626	self.store.save_item(&item)?;
1627	self.store.save_turn(&turn)?;
1628
1629	self.emit_event(
1630	thread_id,
1631	Some(turn_id),
1632	Some(&item.id),
1633	"turn.steered",
1634	json!({
1635	"thread_id": thread_id,
1636	"turn_id": turn_id,
1637	"input": prompt,
1638	}),
1639	)
1640	.await?;
1641	self.emit_event(
1642	thread_id,
1643	Some(turn_id),
1644	Some(&item.id),
1645	"item.completed",
1646	json!({ "item": item }),
1647	)
1648	.await?;
1649
1650	Ok(turn)
1651	}
1652
1653	pub async fn compact_thread(
1654	&self,
1655	thread_id: &str,
1656	req: CompactThreadRequest,
1657	) -> Result<TurnRecord> {
1658	let mut thread = self.get_thread(thread_id).await?;
1659	let engine = self.ensure_engine_loaded(&thread).await?;
1660
1661	{
1662	let active = self.active.lock().await;
1663	if let Some(active_thread) = active.engines.get(thread_id)
1664	&& active_thread.active_turn.is_some()
1665	{
1666	bail!("Thread already has an active turn");
1667	}
1668	}
1669
1670	let now = Utc::now();
1671	let turn_id = format!("turn_{}", &Uuid::new_v4().to_string()[..8]);
1672	let turn = TurnRecord {
1673	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1674	id: turn_id.clone(),
1675	thread_id: thread_id.to_string(),
1676	status: RuntimeTurnStatus::InProgress,
1677	input_summary: req
1678	.reason
1679	.as_deref()
1680	.map(\|s\| summarize_text(s, SUMMARY_LIMIT))
1681	.unwrap_or_else(\|\| "Manual context compaction".to_string()),
1682	created_at: now,
1683	started_at: Some(now),
1684	ended_at: None,
1685	duration_ms: None,
1686	usage: None,
1687	error: None,
1688	item_ids: Vec::new(),
1689	steer_count: 0,
1690	};
1691	self.store.save_turn(&turn)?;
1692
1693	thread.latest_turn_id = Some(turn_id.clone());
1694	thread.updated_at = now;
1695	self.store.save_thread(&thread)?;
1696
1697	{
1698	let mut active = self.active.lock().await;
1699	let Some(state) = active.engines.get_mut(thread_id) else {
1700	bail!("Thread engine not loaded");
1701	};
1702	state.active_turn = Some(ActiveTurnState {
1703	turn_id: turn_id.clone(),
1704	interrupt_requested: false,
1705	auto_approve: thread.auto_approve,
1706	trust_mode: thread.trust_mode,
1707	});
1708	touch_lru(&mut active.lru, thread_id);
1709	}
1710
1711	self.emit_event(
1712	thread_id,
1713	Some(&turn_id),
1714	None,
1715	"turn.started",
1716	json!({ "turn": turn.clone(), "manual_compaction": true }),
1717	)
1718	.await?;
1719
1720	engine
1721	.send(Op::CompactContext)
1722	.await
1723	.map_err(\|e\| anyhow!("Failed to trigger compaction: {e}"))?;
1724
1725	let manager = Arc::new(self.clone());
1726	let thread_id_owned = thread_id.to_string();
1727	let turn_id_owned = turn_id.clone();
1728	let engine_clone = engine.clone();
1729	let cancel_token = self.cancel_token.clone();
1730	tokio::spawn(async move {
1731	if cancel_token.is_cancelled() {
1732	tracing::debug!("Skipping compaction monitor: shutdown requested");
1733	return;
1734	}
1735	use futures_util::FutureExt;
1736	let result = std::panic::AssertUnwindSafe(manager.monitor_turn(
1737	thread_id_owned,
1738	turn_id_owned,
1739	engine_clone,
1740	))
1741	.catch_unwind()
1742	.await;
1743	match result {
1744	Ok(res) => {
1745	if let Err(err) = res {
1746	tracing::error!("Failed to monitor compaction turn: {err}");
1747	}
1748	}
1749	Err(panic_err) => {
1750	if let Some(msg) = panic_err.downcast_ref::<&str>() {
1751	tracing::error!("Compaction monitor panicked: {}", msg);
1752	} else if let Some(msg) = panic_err.downcast_ref::<String>() {
1753	tracing::error!("Compaction monitor panicked: {}", msg);
1754	} else {
1755	tracing::error!("Compaction monitor panicked with unknown error");
1756	}
1757	}
1758	}
1759	});
1760
1761	Ok(turn)
1762	}
1763
1764	pub fn events_since(
1765	&self,
1766	thread_id: &str,
1767	since_seq: Option<u64>,
1768	) -> Result<Vec<RuntimeEventRecord>> {
1769	self.store.events_since(thread_id, since_seq)
1770	}
1771
1772	async fn ensure_engine_loaded(&self, thread: &ThreadRecord) -> Result<EngineHandle> {
1773	{
1774	let mut active = self.active.lock().await;
1775	if let Some(engine) = active
1776	.engines
1777	.get(thread.id.as_str())
1778	.map(\|state\| state.engine.clone())
1779	{
1780	touch_lru(&mut active.lru, &thread.id);
1781	return Ok(engine);
1782	}
1783	}
1784
1785	// Compaction defaults to disabled in v0.6.6 — the cycle architecture
1786	// (issue #124) handles long-context resets. Threads keep the
1787	// legacy summarizer wired off unless an operator opts in via config.
1788	let compaction = CompactionConfig {
1789	enabled: false,
1790	model: thread.model.clone(),
1791	token_threshold: compaction_threshold_for_model(&thread.model),
1792	..Default::default()
1793	};
1794	let network_policy = self.config.network.clone().map(\|toml_cfg\| {
1795	crate::network_policy::NetworkPolicyDecider::with_default_audit(toml_cfg.into_runtime())
1796	});
1797	let lsp_config = self
1798	.config
1799	.lsp
1800	.clone()
1801	.map(crate::config::LspConfigToml::into_runtime);
1802	let engine_cfg = EngineConfig {
1803	model: thread.model.clone(),
1804	workspace: thread.workspace.clone(),
1805	allow_shell: thread.allow_shell,
1806	trust_mode: thread.trust_mode,
1807	notes_path: self.config.notes_path(),
1808	mcp_config_path: self.config.mcp_config_path(),
1809	skills_dir: self.config.skills_dir(),
1810	instructions: self.config.instructions_paths(),
1811	max_steps: 100,
1812	max_subagents: self.config.max_subagents().clamp(1, MAX_SUBAGENTS),
1813	features: self.config.features(),
1814	compaction,
1815	cycle: crate::cycle_manager::CycleConfig::default(),
1816	capacity: crate::core::capacity::CapacityControllerConfig::from_app_config(
1817	&self.config,
1818	),
1819	todos: new_shared_todo_list(),
1820	plan_state: new_shared_plan_state(),
1821	max_spawn_depth: crate::tools::subagent::DEFAULT_MAX_SPAWN_DEPTH,
1822	network_policy,
1823	snapshots_enabled: self.config.snapshots_config().enabled,
1824	lsp_config,
1825	runtime_services: crate::tools::spec::RuntimeToolServices {
1826	task_manager: self.task_manager.lock().ok().and_then(\|slot\| slot.clone()),
1827	automations: self.automations.lock().ok().and_then(\|slot\| slot.clone()),
1828	task_data_dir: Some(self.manager_cfg.task_data_dir.clone()),
1829	active_task_id: thread.task_id.clone(),
1830	active_thread_id: Some(thread.id.clone()),
1831	shell_manager: None,
1832	hook_executor: None,
1833	},
1834	subagent_model_overrides: self.config.subagent_model_overrides(),
1835	memory_enabled: self.config.memory_enabled(),
1836	memory_path: self.config.memory_path(),
1837	strict_tool_mode: self.config.strict_tool_mode.unwrap_or(false),
1838	goal_objective: None,
1839	locale_tag: crate::localization::resolve_locale(
1840	&crate::settings::Settings::load().unwrap_or_default().locale,
1841	)
1842	.tag()
1843	.to_string(),
1844	workshop: self.config.workshop.clone(),
1845	};
1846
1847	let engine = spawn_engine(engine_cfg, &self.config);
1848
1849	let turns = self.store.list_turns_for_thread(&thread.id)?;
1850	let session_messages = self.reconstruct_messages_from_turns(&turns)?;
1851	let sys_prompt = thread
1852	.system_prompt
1853	.as_ref()
1854	.map(\|s\| SystemPrompt::Text(s.clone()));
1855	if !session_messages.is_empty() \|\| sys_prompt.is_some() {
1856	engine
1857	.send(Op::SyncSession {
1858	messages: session_messages,
1859	system_prompt: sys_prompt,
1860	model: thread.model.clone(),
1861	workspace: thread.workspace.clone(),
1862	})
1863	.await
1864	.map_err(\|e\| anyhow!("Failed to sync thread session: {e}"))?;
1865	}
1866
1867	let mut active = self.active.lock().await;
1868	let evicted = enforce_lru_capacity(&mut active, self.manager_cfg.max_active_threads);
1869	active.engines.insert(
1870	thread.id.clone(),
1871	ActiveThreadState {
1872	engine: engine.clone(),
1873	active_turn: None,
1874	},
1875	);
1876	touch_lru(&mut active.lru, &thread.id);
1877	drop(active);
1878	for handle in evicted {
1879	let _ = handle.send(Op::Shutdown).await;
1880	}
1881	Ok(engine)
1882	}
1883
1884	fn reconstruct_messages_from_turns(&self, turns: &[TurnRecord]) -> Result<Vec<Message>> {
1885	let mut messages = Vec::new();
1886	for turn in turns {
1887	let items = self.store.list_items_for_turn(&turn.id)?;
1888	for item in items {
1889	match item.kind {
1890	TurnItemKind::UserMessage => {
1891	let text = item.detail.unwrap_or(item.summary);
1892	messages.push(Message {
1893	role: "user".to_string(),
1894	content: vec![ContentBlock::Text {
1895	text,
1896	cache_control: None,
1897	}],
1898	});
1899	}
1900	TurnItemKind::AgentMessage => {
1901	let text = item.detail.unwrap_or(item.summary);
1902	messages.push(Message {
1903	role: "assistant".to_string(),
1904	content: vec![ContentBlock::Text {
1905	text,
1906	cache_control: None,
1907	}],
1908	});
1909	}
1910	_ => {}
1911	}
1912	}
1913	}
1914	Ok(messages)
1915	}
1916
1917	async fn monitor_turn(
1918	&self,
1919	thread_id: String,
1920	turn_id: String,
1921	engine: EngineHandle,
1922	) -> Result<()> {
1923	let mut current_message_item: Option<(String, String)> = None;
1924	let mut tool_items: HashMap<String, String> = HashMap::new();
1925	let mut compaction_items: HashMap<String, String> = HashMap::new();
1926	let mut turn_usage: Option<Usage> = None;
1927	let mut turn_status = RuntimeTurnStatus::Completed;
1928	let mut turn_error: Option<String> = None;
1929
1930	loop {
1931	let event = {
1932	let mut rx = engine.rx_event.write().await;
1933	rx.recv().await
1934	};
1935	let Some(event) = event else {
1936	if self
1937	.is_interrupt_requested(&thread_id, &turn_id)
1938	.await
1939	.unwrap_or(false)
1940	{
1941	turn_status = RuntimeTurnStatus::Interrupted;
1942	}
1943	break;
1944	};
1945
1946	match event {
1947	EngineEvent::TurnStarted { .. } => {
1948	self.emit_event(
1949	&thread_id,
1950	Some(&turn_id),
1951	None,
1952	"turn.lifecycle",
1953	json!({ "status": "in_progress" }),
1954	)
1955	.await?;
1956	}
1957	EngineEvent::MessageStarted { .. } => {
1958	let item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
1959	let item = TurnItemRecord {
1960	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
1961	id: item_id.clone(),
1962	turn_id: turn_id.clone(),
1963	kind: TurnItemKind::AgentMessage,
1964	status: TurnItemLifecycleStatus::InProgress,
1965	summary: String::new(),
1966	detail: Some(String::new()),
1967	metadata: None,
1968	artifact_refs: Vec::new(),
1969	started_at: Some(Utc::now()),
1970	ended_at: None,
1971	};
1972	self.store.save_item(&item)?;
1973	self.attach_item_to_turn(&turn_id, &item.id)?;
1974	self.emit_event(
1975	&thread_id,
1976	Some(&turn_id),
1977	Some(&item_id),
1978	"item.started",
1979	json!({ "item": item }),
1980	)
1981	.await?;
1982	current_message_item = Some((item_id, String::new()));
1983	}
1984	EngineEvent::MessageDelta { content, .. } => {
1985	if let Some((item_id, text)) = current_message_item.as_mut() {
1986	text.push_str(&content);
1987	self.emit_event(
1988	&thread_id,
1989	Some(&turn_id),
1990	Some(item_id),
1991	"item.delta",
1992	json!({ "delta": content, "kind": "agent_message" }),
1993	)
1994	.await?;
1995	}
1996	}
1997	EngineEvent::MessageComplete { .. } => {
1998	if let Some((item_id, text)) = current_message_item.take() {
1999	let mut item = self.store.load_item(&item_id)?;
2000	item.status = TurnItemLifecycleStatus::Completed;
2001	item.summary = summarize_text(&text, SUMMARY_LIMIT);
2002	item.detail = Some(text);
2003	item.ended_at = Some(Utc::now());
2004	self.store.save_item(&item)?;
2005	self.emit_event(
2006	&thread_id,
2007	Some(&turn_id),
2008	Some(&item_id),
2009	"item.completed",
2010	json!({ "item": item }),
2011	)
2012	.await?;
2013	}
2014	}
2015	EngineEvent::ToolCallStarted { id, name, input } => {
2016	let item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
2017	tool_items.insert(id.clone(), item_id.clone());
2018	let kind = tool_kind_for_name(&name);
2019	let summary = summarize_text(&format!("{name} started"), SUMMARY_LIMIT);
2020	let item = TurnItemRecord {
2021	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2022	id: item_id.clone(),
2023	turn_id: turn_id.clone(),
2024	kind,
2025	status: TurnItemLifecycleStatus::InProgress,
2026	summary,
2027	detail: Some(serde_json::to_string(&input).unwrap_or_default()),
2028	metadata: None,
2029	artifact_refs: Vec::new(),
2030	started_at: Some(Utc::now()),
2031	ended_at: None,
2032	};
2033	self.store.save_item(&item)?;
2034	self.attach_item_to_turn(&turn_id, &item.id)?;
2035	self.emit_event(
2036	&thread_id,
2037	Some(&turn_id),
2038	Some(&item_id),
2039	"item.started",
2040	json!({ "item": item, "tool": { "id": id, "name": name, "input": input } }),
2041	)
2042	.await?;
2043	}
2044	EngineEvent::ToolCallProgress { id, output } => {
2045	if let Some(item_id) = tool_items.get(&id) {
2046	self.emit_event(
2047	&thread_id,
2048	Some(&turn_id),
2049	Some(item_id),
2050	"item.delta",
2051	json!({ "delta": output, "kind": "tool_call" }),
2052	)
2053	.await?;
2054	}
2055	}
2056	EngineEvent::ToolCallComplete { id, name, result } => {
2057	if let Some(item_id) = tool_items.remove(&id) {
2058	let mut item = self.store.load_item(&item_id)?;
2059	let now = Utc::now();
2060	item.ended_at = Some(now);
2061	match result {
2062	Ok(output) => {
2063	item.status = if output.success {
2064	TurnItemLifecycleStatus::Completed
2065	} else {
2066	TurnItemLifecycleStatus::Failed
2067	};
2068	item.summary = summarize_text(
2069	&format!("{name}: {}", output.content),
2070	SUMMARY_LIMIT,
2071	);
2072	item.detail = Some(output.content.clone());
2073	item.metadata = output.metadata.clone();
2074	}
2075	Err(err) => {
2076	item.status = TurnItemLifecycleStatus::Failed;
2077	item.summary =
2078	summarize_text(&format!("{name} failed: {err}"), SUMMARY_LIMIT);
2079	item.detail = Some(err.to_string());
2080	}
2081	}
2082	self.store.save_item(&item)?;
2083	self.emit_event(
2084	&thread_id,
2085	Some(&turn_id),
2086	Some(&item_id),
2087	if item.status == TurnItemLifecycleStatus::Completed {
2088	"item.completed"
2089	} else {
2090	"item.failed"
2091	},
2092	json!({ "item": item }),
2093	)
2094	.await?;
2095	}
2096	}
2097	EngineEvent::CompactionStarted { id, auto, message } => {
2098	let item_id = format!("item_{}", &Uuid::new_v4().to_string()[..8]);
2099	compaction_items.insert(id.clone(), item_id.clone());
2100	let item = TurnItemRecord {
2101	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2102	id: item_id.clone(),
2103	turn_id: turn_id.clone(),
2104	kind: TurnItemKind::ContextCompaction,
2105	status: TurnItemLifecycleStatus::InProgress,
2106	summary: summarize_text(&message, SUMMARY_LIMIT),
2107	detail: Some(message.clone()),
2108	metadata: None,
2109	artifact_refs: Vec::new(),
2110	started_at: Some(Utc::now()),
2111	ended_at: None,
2112	};
2113	self.store.save_item(&item)?;
2114	self.attach_item_to_turn(&turn_id, &item.id)?;
2115	self.emit_event(
2116	&thread_id,
2117	Some(&turn_id),
2118	Some(&item_id),
2119	"item.started",
2120	json!({ "item": item, "auto": auto }),
2121	)
2122	.await?;
2123	}
2124	EngineEvent::CompactionCompleted {
2125	id,
2126	auto,
2127	message,
2128	messages_before,
2129	messages_after,
2130	} => {
2131	if let Some(item_id) = compaction_items.remove(&id) {
2132	let mut item = self.store.load_item(&item_id)?;
2133	item.status = TurnItemLifecycleStatus::Completed;
2134	item.summary = summarize_text(&message, SUMMARY_LIMIT);
2135	item.detail = Some(message);
2136	item.ended_at = Some(Utc::now());
2137	self.store.save_item(&item)?;
2138	self.emit_event(
2139	&thread_id,
2140	Some(&turn_id),
2141	Some(&item_id),
2142	"item.completed",
2143	json!({
2144	"item": item,
2145	"auto": auto,
2146	"messages_before": messages_before,
2147	"messages_after": messages_after,
2148	}),
2149	)
2150	.await?;
2151	}
2152	}
2153	EngineEvent::CompactionFailed { id, auto, message } => {
2154	if let Some(item_id) = compaction_items.remove(&id) {
2155	let mut item = self.store.load_item(&item_id)?;
2156	item.status = TurnItemLifecycleStatus::Failed;
2157	item.summary = summarize_text(&message, SUMMARY_LIMIT);
2158	item.detail = Some(message);
2159	item.ended_at = Some(Utc::now());
2160	self.store.save_item(&item)?;
2161	self.emit_event(
2162	&thread_id,
2163	Some(&turn_id),
2164	Some(&item_id),
2165	"item.failed",
2166	json!({ "item": item, "auto": auto }),
2167	)
2168	.await?;
2169	}
2170	}
2171	EngineEvent::CycleAdvanced { from, to, briefing } => {
2172	// Surface the cycle boundary in the runtime event timeline so
2173	// background-task subscribers and replay see it. The actual
2174	// archive write is the engine's responsibility (see
2175	// `cycle_manager::archive_cycle`); this event is informational.
2176	self.emit_event(
2177	&thread_id,
2178	Some(&turn_id),
2179	None,
2180	"cycle.advanced",
2181	json!({
2182	"from": from,
2183	"to": to,
2184	"briefing_tokens": briefing.token_estimate,
2185	"cycle": briefing.cycle,
2186	"timestamp": briefing.timestamp,
2187	}),
2188	)
2189	.await?;
2190	}
2191	EngineEvent::CoherenceState {
2192	state,
2193	label,
2194	description,
2195	reason,
2196	} => {
2197	let mut thread = self.store.load_thread(&thread_id)?;
2198	thread.coherence_state = state;
2199	thread.updated_at = Utc::now();
2200	self.store.save_thread(&thread)?;
2201	self.emit_event(
2202	&thread_id,
2203	Some(&turn_id),
2204	None,
2205	"coherence.state",
2206	json!({
2207	"state": state,
2208	"label": label,
2209	"description": description,
2210	"reason": reason,
2211	"thread": thread,
2212	}),
2213	)
2214	.await?;
2215	}
2216	EngineEvent::CapacityDecision {
2217	risk_band,
2218	action,
2219	reason,
2220	..
2221	} => {
2222	let message = format!(
2223	"Capacity decision: risk={risk_band} action={action} reason={reason}"
2224	);
2225	let item = TurnItemRecord {
2226	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2227	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2228	turn_id: turn_id.clone(),
2229	kind: TurnItemKind::Status,
2230	status: TurnItemLifecycleStatus::Completed,
2231	summary: summarize_text(&message, SUMMARY_LIMIT),
2232	detail: Some(message),
2233	metadata: None,
2234	artifact_refs: Vec::new(),
2235	started_at: Some(Utc::now()),
2236	ended_at: Some(Utc::now()),
2237	};
2238	self.store.save_item(&item)?;
2239	self.attach_item_to_turn(&turn_id, &item.id)?;
2240	self.emit_event(
2241	&thread_id,
2242	Some(&turn_id),
2243	Some(&item.id),
2244	"item.completed",
2245	json!({ "item": item }),
2246	)
2247	.await?;
2248	}
2249	EngineEvent::CapacityIntervention {
2250	action,
2251	before_prompt_tokens,
2252	after_prompt_tokens,
2253	replay_outcome,
2254	replan_performed,
2255	..
2256	} => {
2257	let message = format!(
2258	"Capacity intervention: {action} (~{before_prompt_tokens} -> ~{after_prompt_tokens}) replay={:?} replan={replan_performed}",
2259	replay_outcome
2260	);
2261	let item = TurnItemRecord {
2262	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2263	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2264	turn_id: turn_id.clone(),
2265	kind: TurnItemKind::Status,
2266	status: TurnItemLifecycleStatus::Completed,
2267	summary: summarize_text(&message, SUMMARY_LIMIT),
2268	detail: Some(message),
2269	metadata: None,
2270	artifact_refs: Vec::new(),
2271	started_at: Some(Utc::now()),
2272	ended_at: Some(Utc::now()),
2273	};
2274	self.store.save_item(&item)?;
2275	self.attach_item_to_turn(&turn_id, &item.id)?;
2276	self.emit_event(
2277	&thread_id,
2278	Some(&turn_id),
2279	Some(&item.id),
2280	"item.completed",
2281	json!({ "item": item }),
2282	)
2283	.await?;
2284	}
2285	EngineEvent::CapacityMemoryPersistFailed { action, error, .. } => {
2286	let message =
2287	format!("Capacity memory persist failed: action={action} error={error}");
2288	let item = TurnItemRecord {
2289	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2290	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2291	turn_id: turn_id.clone(),
2292	kind: TurnItemKind::Status,
2293	status: TurnItemLifecycleStatus::Failed,
2294	summary: summarize_text(&message, SUMMARY_LIMIT),
2295	detail: Some(message),
2296	metadata: None,
2297	artifact_refs: Vec::new(),
2298	started_at: Some(Utc::now()),
2299	ended_at: Some(Utc::now()),
2300	};
2301	self.store.save_item(&item)?;
2302	self.attach_item_to_turn(&turn_id, &item.id)?;
2303	self.emit_event(
2304	&thread_id,
2305	Some(&turn_id),
2306	Some(&item.id),
2307	"item.failed",
2308	json!({ "item": item }),
2309	)
2310	.await?;
2311	}
2312	EngineEvent::AgentSpawned { id, prompt } => {
2313	let message = format!(
2314	"Sub-agent {id} spawned: {}",
2315	summarize_text(&prompt, SUMMARY_LIMIT)
2316	);
2317	let item = TurnItemRecord {
2318	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2319	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2320	turn_id: turn_id.clone(),
2321	kind: TurnItemKind::Status,
2322	status: TurnItemLifecycleStatus::Completed,
2323	summary: summarize_text(&message, SUMMARY_LIMIT),
2324	detail: Some(message),
2325	metadata: None,
2326	artifact_refs: Vec::new(),
2327	started_at: Some(Utc::now()),
2328	ended_at: Some(Utc::now()),
2329	};
2330	self.store.save_item(&item)?;
2331	self.attach_item_to_turn(&turn_id, &item.id)?;
2332	self.emit_event(
2333	&thread_id,
2334	Some(&turn_id),
2335	Some(&item.id),
2336	"agent.spawned",
2337	json!({ "item": item, "agent_id": id }),
2338	)
2339	.await?;
2340	}
2341	EngineEvent::AgentProgress { id, status } => {
2342	let message = format!("Sub-agent {id}: {status}");
2343	let item = TurnItemRecord {
2344	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2345	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2346	turn_id: turn_id.clone(),
2347	kind: TurnItemKind::Status,
2348	status: TurnItemLifecycleStatus::Completed,
2349	summary: summarize_text(&message, SUMMARY_LIMIT),
2350	detail: Some(message),
2351	metadata: None,
2352	artifact_refs: Vec::new(),
2353	started_at: Some(Utc::now()),
2354	ended_at: Some(Utc::now()),
2355	};
2356	self.store.save_item(&item)?;
2357	self.attach_item_to_turn(&turn_id, &item.id)?;
2358	self.emit_event(
2359	&thread_id,
2360	Some(&turn_id),
2361	Some(&item.id),
2362	"agent.progress",
2363	json!({ "item": item, "agent_id": id }),
2364	)
2365	.await?;
2366	}
2367	EngineEvent::AgentComplete { id, result } => {
2368	let message = format!(
2369	"Sub-agent {id} completed: {}",
2370	summarize_text(&result, SUMMARY_LIMIT)
2371	);
2372	let item = TurnItemRecord {
2373	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2374	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2375	turn_id: turn_id.clone(),
2376	kind: TurnItemKind::Status,
2377	status: TurnItemLifecycleStatus::Completed,
2378	summary: summarize_text(&message, SUMMARY_LIMIT),
2379	detail: Some(message),
2380	metadata: None,
2381	artifact_refs: Vec::new(),
2382	started_at: Some(Utc::now()),
2383	ended_at: Some(Utc::now()),
2384	};
2385	self.store.save_item(&item)?;
2386	self.attach_item_to_turn(&turn_id, &item.id)?;
2387	self.emit_event(
2388	&thread_id,
2389	Some(&turn_id),
2390	Some(&item.id),
2391	"agent.completed",
2392	json!({ "item": item, "agent_id": id }),
2393	)
2394	.await?;
2395	}
2396	EngineEvent::AgentList { agents } => {
2397	let running = agents
2398	.iter()
2399	.filter(\|agent\| matches!(agent.status, SubAgentStatus::Running))
2400	.count();
2401	let interrupted = agents
2402	.iter()
2403	.filter(\|agent\| matches!(agent.status, SubAgentStatus::Interrupted(_)))
2404	.count();
2405	let completed = agents
2406	.iter()
2407	.filter(\|agent\| matches!(agent.status, SubAgentStatus::Completed))
2408	.count();
2409	let message = format!(
2410	"Sub-agent list refreshed: {} total ({running} running, {interrupted} interrupted, {completed} completed)",
2411	agents.len()
2412	);
2413	let item = TurnItemRecord {
2414	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2415	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2416	turn_id: turn_id.clone(),
2417	kind: TurnItemKind::Status,
2418	status: TurnItemLifecycleStatus::Completed,
2419	summary: summarize_text(&message, SUMMARY_LIMIT),
2420	detail: Some(message),
2421	metadata: None,
2422	artifact_refs: Vec::new(),
2423	started_at: Some(Utc::now()),
2424	ended_at: Some(Utc::now()),
2425	};
2426	self.store.save_item(&item)?;
2427	self.attach_item_to_turn(&turn_id, &item.id)?;
2428	self.emit_event(
2429	&thread_id,
2430	Some(&turn_id),
2431	Some(&item.id),
2432	"agent.list",
2433	json!({ "item": item, "agents": agents }),
2434	)
2435	.await?;
2436	}
2437	EngineEvent::ApprovalRequired {
2438	id,
2439	tool_name,
2440	description,
2441	..
2442	} => {
2443	self.emit_event(
2444	&thread_id,
2445	Some(&turn_id),
2446	None,
2447	"approval.required",
2448	json!({
2449	"id": id,
2450	"tool_name": tool_name,
2451	"description": description,
2452	}),
2453	)
2454	.await?;
2455
2456	let (auto_approve, trust_mode) = self
2457	.active_turn_flags(&thread_id, &turn_id)
2458	.await
2459	.unwrap_or((false, false));
2460	match Self::approval_decision(auto_approve, trust_mode, false) {
2461	RuntimeApprovalDecision::ApproveTool => {
2462	let _ = engine.approve_tool_call(id).await;
2463	}
2464	RuntimeApprovalDecision::DenyTool
2465	\| RuntimeApprovalDecision::RetryWithFullAccess => {
2466	let _ = engine.deny_tool_call(id).await;
2467	}
2468	}
2469	}
2470	EngineEvent::ElevationRequired {
2471	tool_id,
2472	tool_name,
2473	denial_reason,
2474	..
2475	} => {
2476	self.emit_event(
2477	&thread_id,
2478	Some(&turn_id),
2479	None,
2480	"sandbox.denied",
2481	json!({
2482	"tool_id": tool_id,
2483	"tool_name": tool_name,
2484	"reason": denial_reason,
2485	}),
2486	)
2487	.await?;
2488	let (auto_approve, trust_mode) = self
2489	.active_turn_flags(&thread_id, &turn_id)
2490	.await
2491	.unwrap_or((false, false));
2492	match Self::approval_decision(auto_approve, trust_mode, true) {
2493	RuntimeApprovalDecision::RetryWithFullAccess => {
2494	let _ = engine
2495	.retry_tool_with_policy(
2496	tool_id,
2497	crate::sandbox::SandboxPolicy::DangerFullAccess,
2498	)
2499	.await;
2500	}
2501	RuntimeApprovalDecision::ApproveTool
2502	\| RuntimeApprovalDecision::DenyTool => {
2503	let _ = engine.deny_tool_call(tool_id).await;
2504	}
2505	}
2506	}
2507	EngineEvent::Status { message } => {
2508	let item = TurnItemRecord {
2509	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2510	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2511	turn_id: turn_id.clone(),
2512	kind: TurnItemKind::Status,
2513	status: TurnItemLifecycleStatus::Completed,
2514	summary: summarize_text(&message, SUMMARY_LIMIT),
2515	detail: Some(message.clone()),
2516	metadata: None,
2517	artifact_refs: Vec::new(),
2518	started_at: Some(Utc::now()),
2519	ended_at: Some(Utc::now()),
2520	};
2521	self.store.save_item(&item)?;
2522	self.attach_item_to_turn(&turn_id, &item.id)?;
2523	self.emit_event(
2524	&thread_id,
2525	Some(&turn_id),
2526	Some(&item.id),
2527	"item.completed",
2528	json!({ "item": item }),
2529	)
2530	.await?;
2531	}
2532	EngineEvent::Error { envelope, .. } => {
2533	turn_status = RuntimeTurnStatus::Failed;
2534	turn_error = Some(envelope.message.clone());
2535	let message = envelope.message.clone();
2536	let item = TurnItemRecord {
2537	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2538	id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
2539	turn_id: turn_id.clone(),
2540	kind: TurnItemKind::Error,
2541	status: TurnItemLifecycleStatus::Failed,
2542	summary: summarize_text(&message, SUMMARY_LIMIT),
2543	detail: Some(message),
2544	metadata: None,
2545	artifact_refs: Vec::new(),
2546	started_at: Some(Utc::now()),
2547	ended_at: Some(Utc::now()),
2548	};
2549	self.store.save_item(&item)?;
2550	self.attach_item_to_turn(&turn_id, &item.id)?;
2551	self.emit_event(
2552	&thread_id,
2553	Some(&turn_id),
2554	Some(&item.id),
2555	"item.failed",
2556	json!({ "item": item }),
2557	)
2558	.await?;
2559	}
2560	EngineEvent::TurnComplete {
2561	usage,
2562	status,
2563	error,
2564	} => {
2565	turn_usage = Some(usage);
2566	turn_status = match status {
2567	TurnOutcomeStatus::Completed => RuntimeTurnStatus::Completed,
2568	TurnOutcomeStatus::Interrupted => RuntimeTurnStatus::Interrupted,
2569	TurnOutcomeStatus::Failed => RuntimeTurnStatus::Failed,
2570	};
2571	if let Some(err) = error {
2572	turn_error = Some(err);
2573	}
2574	break;
2575	}
2576	_ => {}
2577	}
2578	}
2579
2580	if self
2581	.is_interrupt_requested(&thread_id, &turn_id)
2582	.await
2583	.unwrap_or(false)
2584	{
2585	turn_status = RuntimeTurnStatus::Interrupted;
2586	}
2587
2588	if let Some((item_id, text)) = current_message_item.take() {
2589	let mut item = self.store.load_item(&item_id)?;
2590	if turn_status == RuntimeTurnStatus::Interrupted {
2591	item.status = TurnItemLifecycleStatus::Interrupted;
2592	} else {
2593	item.status = TurnItemLifecycleStatus::Completed;
2594	}
2595	item.summary = summarize_text(&text, SUMMARY_LIMIT);
2596	item.detail = Some(text);
2597	item.ended_at = Some(Utc::now());
2598	self.store.save_item(&item)?;
2599	self.emit_event(
2600	&thread_id,
2601	Some(&turn_id),
2602	Some(&item_id),
2603	if item.status == TurnItemLifecycleStatus::Interrupted {
2604	"item.interrupted"
2605	} else {
2606	"item.completed"
2607	},
2608	json!({ "item": item }),
2609	)
2610	.await?;
2611	}
2612
2613	let ended_at = Utc::now();
2614	let mut turn = self.store.load_turn(&turn_id)?;
2615	turn.status = turn_status;
2616	turn.ended_at = Some(ended_at);
2617	turn.duration_ms = turn.started_at.map(\|start\| duration_ms(start, ended_at));
2618	turn.usage = turn_usage;
2619	turn.error = turn_error;
2620	self.store.save_turn(&turn)?;
2621
2622	let mut thread = self.get_thread(&thread_id).await?;
2623	thread.latest_turn_id = Some(turn_id.clone());
2624	thread.updated_at = Utc::now();
2625	self.store.save_thread(&thread)?;
2626
2627	self.emit_event(
2628	&thread_id,
2629	Some(&turn_id),
2630	None,
2631	"turn.completed",
2632	json!({ "turn": turn.clone() }),
2633	)
2634	.await?;
2635
2636	{
2637	let mut active = self.active.lock().await;
2638	if let Some(state) = active.engines.get_mut(&thread_id)
2639	&& state
2640	.active_turn
2641	.as_ref()
2642	.is_some_and(\|t\| t.turn_id == turn_id)
2643	{
2644	state.active_turn = None;
2645	}
2646	touch_lru(&mut active.lru, &thread_id);
2647	}
2648
2649	Ok(())
2650	}
2651
2652	fn attach_item_to_turn(&self, turn_id: &str, item_id: &str) -> Result<()> {
2653	let mut turn = self.store.load_turn(turn_id)?;
2654	if !turn.item_ids.iter().any(\|id\| id == item_id) {
2655	turn.item_ids.push(item_id.to_string());
2656	self.store.save_turn(&turn)?;
2657	}
2658	Ok(())
2659	}
2660
2661	async fn is_interrupt_requested(&self, thread_id: &str, turn_id: &str) -> Result<bool> {
2662	let active = self.active.lock().await;
2663	let Some(state) = active.engines.get(thread_id) else {
2664	return Ok(false);
2665	};
2666	let Some(turn) = state.active_turn.as_ref() else {
2667	return Ok(false);
2668	};
2669	Ok(turn.turn_id == turn_id && turn.interrupt_requested)
2670	}
2671
2672	async fn active_turn_flags(&self, thread_id: &str, turn_id: &str) -> Option<(bool, bool)> {
2673	let active = self.active.lock().await;
2674	let state = active.engines.get(thread_id)?;
2675	let turn = state.active_turn.as_ref()?;
2676	if turn.turn_id != turn_id {
2677	return None;
2678	}
2679	Some((turn.auto_approve, turn.trust_mode))
2680	}
2681
2682	fn approval_decision(
2683	auto_approve: bool,
2684	trust_mode: bool,
2685	requires_full_access: bool,
2686	) -> RuntimeApprovalDecision {
2687	if !auto_approve {
2688	return RuntimeApprovalDecision::DenyTool;
2689	}
2690	if requires_full_access {
2691	if trust_mode {
2692	RuntimeApprovalDecision::RetryWithFullAccess
2693	} else {
2694	RuntimeApprovalDecision::DenyTool
2695	}
2696	} else {
2697	RuntimeApprovalDecision::ApproveTool
2698	}
2699	}
2700
2701	fn recover_interrupted_state(&self) -> Result<()> {
2702	let now = Utc::now();
2703	for mut thread in self.store.list_threads()? {
2704	let mut thread_changed = false;
2705	for mut turn in self.store.list_turns_for_thread(&thread.id)? {
2706	if !matches!(
2707	turn.status,
2708	RuntimeTurnStatus::Queued \| RuntimeTurnStatus::InProgress
2709	) {
2710	continue;
2711	}
2712
2713	turn.status = RuntimeTurnStatus::Interrupted;
2714	turn.error = Some(RUNTIME_RESTART_REASON.to_string());
2715	turn.ended_at = Some(now);
2716	if let Some(started_at) = turn.started_at {
2717	let elapsed = now.signed_duration_since(started_at);
2718	turn.duration_ms = Some(elapsed.num_milliseconds().max(0) as u64);
2719	}
2720	self.store.save_turn(&turn)?;
2721
2722	for item_id in &turn.item_ids {
2723	let mut item = self.store.load_item(item_id)?;
2724	if matches!(
2725	item.status,
2726	TurnItemLifecycleStatus::Queued \| TurnItemLifecycleStatus::InProgress
2727	) {
2728	item.status = TurnItemLifecycleStatus::Interrupted;
2729	item.ended_at = Some(now);
2730	self.store.save_item(&item)?;
2731	}
2732	}
2733
2734	thread.updated_at = now;
2735	thread_changed = true;
2736	}
2737
2738	if thread_changed {
2739	self.store.save_thread(&thread)?;
2740	}
2741	}
2742
2743	Ok(())
2744	}
2745
2746	#[cfg(test)]
2747	pub(crate) async fn install_test_engine(
2748	&self,
2749	thread_id: &str,
2750	engine: EngineHandle,
2751	) -> Result<()> {
2752	let _ = self.get_thread(thread_id).await?;
2753	let mut active = self.active.lock().await;
2754	active.engines.insert(
2755	thread_id.to_string(),
2756	ActiveThreadState {
2757	engine,
2758	active_turn: None,
2759	},
2760	);
2761	touch_lru(&mut active.lru, thread_id);
2762	Ok(())
2763	}
2764	}
2765
2766	fn touch_lru(lru: &mut VecDeque<String>, thread_id: &str) {
2767	if let Some(idx) = lru.iter().position(\|id\| id == thread_id) {
2768	lru.remove(idx);
2769	}
2770	lru.push_back(thread_id.to_string());
2771	}
2772
2773	fn enforce_lru_capacity(
2774	active: &mut ActiveThreads,
2775	max_active_threads: usize,
2776	) -> Vec<EngineHandle> {
2777	let mut evicted = Vec::new();
2778	if max_active_threads == 0 \|\| active.engines.len() < max_active_threads {
2779	return evicted;
2780	}
2781	let protected = active
2782	.engines
2783	.iter()
2784	.filter_map(\|(thread_id, state)\| {
2785	if state.active_turn.is_some() {
2786	Some(thread_id.clone())
2787	} else {
2788	None
2789	}
2790	})
2791	.collect::<HashSet<_>>();
2792
2793	let scan_limit = active.lru.len();
2794	for _ in 0..scan_limit {
2795	let Some(candidate) = active.lru.pop_front() else {
2796	break;
2797	};
2798	if protected.contains(&candidate) {
2799	active.lru.push_back(candidate);
2800	continue;
2801	}
2802	if let Some(state) = active.engines.remove(&candidate) {
2803	evicted.push(state.engine);
2804	}
2805	break;
2806	}
2807	evicted
2808	}
2809
2810	fn parse_mode(mode: &str) -> AppMode {
2811	match mode.trim().to_ascii_lowercase().as_str() {
2812	"plan" => AppMode::Plan,
2813	"yolo" => AppMode::Yolo,
2814	_ => AppMode::Agent,
2815	}
2816	}
2817
2818	fn tool_kind_for_name(name: &str) -> TurnItemKind {
2819	let lower = name.to_ascii_lowercase();
2820	if lower == "exec_shell" \|\| lower == "exec_shell_wait" \|\| lower == "exec_shell_interact" {
2821	return TurnItemKind::CommandExecution;
2822	}
2823	if lower.contains("patch") \|\| lower.contains("write") \|\| lower.contains("edit") {
2824	return TurnItemKind::FileChange;
2825	}
2826	TurnItemKind::ToolCall
2827	}
2828
2829	/// One sub-agent rebind hint extracted from a thread's persisted event
2830	/// timeline (issue #128). When the TUI resumes a session that was
2831	/// mid-fanout, the in-transcript card stack is empty — these hints let the
2832	/// UI know which agent_ids were live (or recently terminal) so it can
2833	/// reconstruct the matching `DelegateCard` / `FanoutCard` placeholders
2834	/// before fresh mailbox envelopes arrive on a re-attached engine.
2835	///
2836	/// The helper is the testable contract here — actual TUI wire-up to the
2837	/// resume flow is a follow-up; the runtime API consumer (`runtime_api.rs`)
2838	/// can already call `resume_thread_with_agent_rebind` to drive it.
2839	#[derive(Debug, Clone, PartialEq, Eq)]
2840	#[allow(dead_code)] // consumed by #128 follow-up TUI resume wiring; tested here.
2841	pub struct AgentRebindHint {
2842	pub agent_id: String,
2843	pub status: AgentRebindStatus,
2844	}
2845
2846	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2847	#[allow(dead_code)]
2848	pub enum AgentRebindStatus {
2849	Spawned,
2850	InProgress,
2851	Completed,
2852	}
2853
2854	/// Collapse a chronologically ordered slice of `RuntimeEventRecord` into
2855	/// the latest known status per `agent_id`. Drops entries that aren't in
2856	/// the `agent.*` family. Cards built from these hints are immediately
2857	/// open to mutation by subsequent live mailbox envelopes (each envelope's
2858	/// `agent_id` matches one already in the rebind map).
2859	#[must_use]
2860	#[allow(dead_code)]
2861	pub fn collect_agent_rebind_hints(events: &[RuntimeEventRecord]) -> Vec<AgentRebindHint> {
2862	use std::collections::BTreeMap;
2863	let mut latest: BTreeMap<String, AgentRebindStatus> = BTreeMap::new();
2864	for event in events {
2865	let id = match event.payload.get("agent_id").and_then(\|v\| v.as_str()) {
2866	Some(id) => id.to_string(),
2867	None => continue,
2868	};
2869	let next_status = match event.event.as_str() {
2870	"agent.spawned" => Some(AgentRebindStatus::Spawned),
2871	"agent.progress" => Some(AgentRebindStatus::InProgress),
2872	"agent.completed" => Some(AgentRebindStatus::Completed),
2873	_ => None,
2874	};
2875	if let Some(status) = next_status {
2876	// Don't downgrade Completed → InProgress on out-of-order events.
2877	let entry = latest.entry(id).or_insert(status);
2878	if !matches!(*entry, AgentRebindStatus::Completed) {
2879	*entry = status;
2880	}
2881	}
2882	}
2883	latest
2884	.into_iter()
2885	.map(\|(agent_id, status)\| AgentRebindHint { agent_id, status })
2886	.collect()
2887	}
2888
2889	pub fn summarize_text(text: &str, limit: usize) -> String {
2890	let take = limit.saturating_sub(3);
2891	let mut count = 0;
2892	let mut out = String::new();
2893	for ch in text.chars() {
2894	if count >= take {
2895	out.push_str("...");
2896	return out;
2897	}
2898	if ch.is_control() && ch != '\n' && ch != '\t' {
2899	continue;
2900	}
2901	out.push(ch);
2902	count += 1;
2903	}
2904	out
2905	}
2906
2907	fn duration_ms(start: DateTime<Utc>, end: DateTime<Utc>) -> u64 {
2908	let millis = (end - start).num_milliseconds();
2909	if millis.is_negative() {
2910	0
2911	} else {
2912	u64::try_from(millis).unwrap_or(u64::MAX)
2913	}
2914	}
2915
2916	fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> {
2917	if let Some(parent) = path.parent() {
2918	fs::create_dir_all(parent)
2919	.with_context(\|\| format!("Failed to create directory {}", parent.display()))?;
2920	}
2921	let payload = serde_json::to_string_pretty(value)?;
2922	crate::utils::write_atomic(path, payload.as_bytes())
2923	.with_context(\|\| format!("Failed to write {}", path.display()))
2924	}
2925
2926	#[cfg(test)]
2927	mod tests {
2928	use super::*;
2929	use crate::core::engine::{MockApprovalEvent, mock_engine_handle};
2930	use crate::core::events::{Event as EngineEvent, TurnOutcomeStatus};
2931	use std::time::{Duration, Instant};
2932	use tokio::sync::oneshot;
2933	use tokio::time::sleep;
2934	use uuid::Uuid;
2935
2936	fn test_runtime_dir() -> PathBuf {
2937	std::env::temp_dir().join(format!("deepseek-runtime-threads-{}", Uuid::new_v4()))
2938	}
2939
2940	fn test_manager_config(data_dir: PathBuf) -> RuntimeThreadManagerConfig {
2941	RuntimeThreadManagerConfig {
2942	task_data_dir: data_dir.clone(),
2943	data_dir,
2944	max_active_threads: 4,
2945	}
2946	}
2947
2948	fn test_manager(data_dir: PathBuf) -> Result<RuntimeThreadManager> {
2949	RuntimeThreadManager::open(
2950	Config::default(),
2951	PathBuf::from("."),
2952	test_manager_config(data_dir),
2953	)
2954	}
2955
2956	fn sample_thread(thread_id: &str) -> ThreadRecord {
2957	let now = Utc::now();
2958	ThreadRecord {
2959	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2960	id: thread_id.to_string(),
2961	created_at: now,
2962	updated_at: now,
2963	model: DEFAULT_TEXT_MODEL.to_string(),
2964	workspace: PathBuf::from("."),
2965	mode: AppMode::Agent.as_setting().to_string(),
2966	allow_shell: false,
2967	trust_mode: false,
2968	auto_approve: false,
2969	latest_turn_id: None,
2970	latest_response_bookmark: None,
2971	archived: false,
2972	system_prompt: None,
2973	task_id: None,
2974	title: None,
2975	coherence_state: CoherenceState::default(),
2976	}
2977	}
2978
2979	fn sample_turn(thread_id: &str, turn_id: &str, status: RuntimeTurnStatus) -> TurnRecord {
2980	let now = Utc::now();
2981	TurnRecord {
2982	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
2983	id: turn_id.to_string(),
2984	thread_id: thread_id.to_string(),
2985	status,
2986	input_summary: "sample".to_string(),
2987	created_at: now,
2988	started_at: Some(now),
2989	ended_at: None,
2990	duration_ms: None,
2991	usage: None,
2992	error: None,
2993	item_ids: Vec::new(),
2994	steer_count: 0,
2995	}
2996	}
2997
2998	fn sample_item(
2999	turn_id: &str,
3000	item_id: &str,
3001	status: TurnItemLifecycleStatus,
3002	) -> TurnItemRecord {
3003	TurnItemRecord {
3004	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
3005	id: item_id.to_string(),
3006	turn_id: turn_id.to_string(),
3007	kind: TurnItemKind::Status,
3008	status,
3009	summary: "sample item".to_string(),
3010	detail: None,
3011	metadata: None,
3012	artifact_refs: Vec::new(),
3013	started_at: Some(Utc::now()),
3014	ended_at: None,
3015	}
3016	}
3017
3018	async fn install_mock_engine(
3019	manager: &RuntimeThreadManager,
3020	thread_id: &str,
3021	) -> crate::core::engine::MockEngineHandle {
3022	let harness = mock_engine_handle();
3023	let mut active = manager.active.lock().await;
3024	active.engines.insert(
3025	thread_id.to_string(),
3026	ActiveThreadState {
3027	engine: harness.handle.clone(),
3028	active_turn: None,
3029	},
3030	);
3031	touch_lru(&mut active.lru, thread_id);
3032	harness
3033	}
3034
3035	async fn wait_for_terminal_turn(
3036	manager: &RuntimeThreadManager,
3037	turn_id: &str,
3038	timeout: Duration,
3039	) -> Result<TurnRecord> {
3040	let deadline = Instant::now() + timeout;
3041	loop {
3042	let turn = manager.store.load_turn(turn_id)?;
3043	if matches!(
3044	turn.status,
3045	RuntimeTurnStatus::Completed
3046	\| RuntimeTurnStatus::Failed
3047	\| RuntimeTurnStatus::Interrupted
3048	\| RuntimeTurnStatus::Canceled
3049	) {
3050	return Ok(turn);
3051	}
3052	if Instant::now() >= deadline {
3053	bail!("Timed out waiting for turn {turn_id}");
3054	}
3055	sleep(Duration::from_millis(20)).await;
3056	}
3057	}
3058
3059	#[test]
3060	fn store_load_thread_rejects_newer_schema_version() {
3061	let dir = test_runtime_dir();
3062	let store = RuntimeThreadStore::open(dir.clone()).expect("open store");
3063
3064	// Construct a thread record persisted with a future schema version.
3065	let mut thread = sample_thread("thr_future");
3066	thread.schema_version = CURRENT_RUNTIME_SCHEMA_VERSION + 1;
3067
3068	// Bypass save_thread (which would respect our local schema_version)
3069	// by writing the JSON directly so we can simulate a future writer.
3070	let path = store.threads_dir.join(format!("{}.json", thread.id));
3071	std::fs::create_dir_all(path.parent().unwrap()).expect("mkdirs");
3072	let payload = serde_json::to_string(&thread).expect("serialize thread");
3073	std::fs::write(&path, payload).expect("write thread");
3074
3075	let err = store
3076	.load_thread(&thread.id)
3077	.expect_err("load_thread must reject newer schema");
3078	let msg = format!("{err:#}");
3079	assert!(msg.contains("newer than supported"), "got: {msg}");
3080
3081	// Cleanup so we don't leak across tests.
3082	let _ = std::fs::remove_dir_all(dir);
3083	}
3084
3085	#[test]
3086	fn current_runtime_schema_version_is_two_on_v066() {
3087	// Locks the bump in (issue #124). Bump deliberately when persisted
3088	// shape changes.
3089	assert_eq!(CURRENT_RUNTIME_SCHEMA_VERSION, 2);
3090	}
3091
3092	#[test]
3093	fn store_load_turn_rejects_newer_schema_version() {
3094	let dir = test_runtime_dir();
3095	let store = RuntimeThreadStore::open(dir.clone()).expect("open store");
3096
3097	let mut turn = sample_turn("thr_t", "trn_future", RuntimeTurnStatus::InProgress);
3098	turn.schema_version = CURRENT_RUNTIME_SCHEMA_VERSION + 1;
3099
3100	let path = store.turns_dir.join(format!("{}.json", turn.id));
3101	std::fs::create_dir_all(path.parent().unwrap()).expect("mkdirs");
3102	std::fs::write(&path, serde_json::to_string(&turn).expect("serialize turn"))
3103	.expect("write turn");
3104
3105	let err = store
3106	.load_turn(&turn.id)
3107	.expect_err("load_turn must reject newer schema");
3108	assert!(
3109	format!("{err:#}").contains("newer than supported"),
3110	"got: {err:#}"
3111	);
3112
3113	let _ = std::fs::remove_dir_all(dir);
3114	}
3115
3116	#[test]
3117	fn store_load_item_rejects_newer_schema_version() {
3118	let dir = test_runtime_dir();
3119	let store = RuntimeThreadStore::open(dir.clone()).expect("open store");
3120
3121	let mut item = sample_item("trn_t", "itm_future", TurnItemLifecycleStatus::InProgress);
3122	item.schema_version = CURRENT_RUNTIME_SCHEMA_VERSION + 1;
3123
3124	let path = store.items_dir.join(format!("{}.json", item.id));
3125	std::fs::create_dir_all(path.parent().unwrap()).expect("mkdirs");
3126	std::fs::write(&path, serde_json::to_string(&item).expect("serialize item"))
3127	.expect("write item");
3128
3129	let err = store
3130	.load_item(&item.id)
3131	.expect_err("load_item must reject newer schema");
3132	assert!(
3133	format!("{err:#}").contains("newer than supported"),
3134	"got: {err:#}"
3135	);
3136
3137	let _ = std::fs::remove_dir_all(dir);
3138	}
3139
3140	#[test]
3141	fn enforce_lru_capacity_does_not_loop_when_all_threads_are_active() {
3142	let mut active = ActiveThreads::default();
3143	let harness_a = mock_engine_handle();
3144	let harness_b = mock_engine_handle();
3145
3146	active.engines.insert(
3147	"thr_a".to_string(),
3148	ActiveThreadState {
3149	engine: harness_a.handle,
3150	active_turn: Some(ActiveTurnState {
3151	turn_id: "turn_a".to_string(),
3152	interrupt_requested: false,
3153	auto_approve: true,
3154	trust_mode: false,
3155	}),
3156	},
3157	);
3158	active.engines.insert(
3159	"thr_b".to_string(),
3160	ActiveThreadState {
3161	engine: harness_b.handle,
3162	active_turn: Some(ActiveTurnState {
3163	turn_id: "turn_b".to_string(),
3164	interrupt_requested: false,
3165	auto_approve: true,
3166	trust_mode: false,
3167	}),
3168	},
3169	);
3170	active.lru.push_back("thr_a".to_string());
3171	active.lru.push_back("thr_b".to_string());
3172
3173	let evicted = enforce_lru_capacity(&mut active, 2);
3174	assert!(evicted.is_empty(), "no idle threads should be evicted");
3175	assert_eq!(active.engines.len(), 2);
3176	assert_eq!(active.lru.len(), 2);
3177	}
3178
3179	#[test]
3180	fn approval_decision_matches_auto_approve_and_trust_mode() {
3181	assert!(matches!(
3182	RuntimeThreadManager::approval_decision(false, false, false),
3183	RuntimeApprovalDecision::DenyTool
3184	));
3185	assert!(matches!(
3186	RuntimeThreadManager::approval_decision(true, false, false),
3187	RuntimeApprovalDecision::ApproveTool
3188	));
3189	assert!(matches!(
3190	RuntimeThreadManager::approval_decision(true, false, true),
3191	RuntimeApprovalDecision::DenyTool
3192	));
3193	assert!(matches!(
3194	RuntimeThreadManager::approval_decision(true, true, true),
3195	RuntimeApprovalDecision::RetryWithFullAccess
3196	));
3197	}
3198
3199	#[test]
3200	fn open_recovers_queued_and_in_progress_turns() -> Result<()> {
3201	let runtime_dir = test_runtime_dir();
3202	let store = RuntimeThreadStore::open(runtime_dir.clone())?;
3203	let thread = sample_thread("thr_recover");
3204	store.save_thread(&thread)?;
3205
3206	let mut queued_turn = sample_turn(&thread.id, "turn_queued", RuntimeTurnStatus::Queued);
3207	let mut in_progress_turn =
3208	sample_turn(&thread.id, "turn_running", RuntimeTurnStatus::InProgress);
3209	let completed_turn = sample_turn(&thread.id, "turn_done", RuntimeTurnStatus::Completed);
3210
3211	let queued_item = sample_item(
3212	&queued_turn.id,
3213	"item_queued",
3214	TurnItemLifecycleStatus::Queued,
3215	);
3216	let in_progress_item = sample_item(
3217	&in_progress_turn.id,
3218	"item_running",
3219	TurnItemLifecycleStatus::InProgress,
3220	);
3221	let completed_item = sample_item(
3222	&completed_turn.id,
3223	"item_done",
3224	TurnItemLifecycleStatus::Completed,
3225	);
3226
3227	queued_turn.item_ids = vec![queued_item.id.clone()];
3228	in_progress_turn.item_ids = vec![in_progress_item.id.clone()];
3229
3230	store.save_item(&queued_item)?;
3231	store.save_item(&in_progress_item)?;
3232	store.save_item(&completed_item)?;
3233	store.save_turn(&queued_turn)?;
3234	store.save_turn(&in_progress_turn)?;
3235	store.save_turn(&completed_turn)?;
3236
3237	let manager = test_manager(runtime_dir)?;
3238
3239	let queued_turn = manager.store.load_turn(&queued_turn.id)?;
3240	assert_eq!(queued_turn.status, RuntimeTurnStatus::Interrupted);
3241	assert_eq!(queued_turn.error.as_deref(), Some(RUNTIME_RESTART_REASON));
3242	assert!(queued_turn.ended_at.is_some());
3243	assert!(queued_turn.duration_ms.is_some());
3244
3245	let in_progress_turn = manager.store.load_turn(&in_progress_turn.id)?;
3246	assert_eq!(in_progress_turn.status, RuntimeTurnStatus::Interrupted);
3247	assert_eq!(
3248	in_progress_turn.error.as_deref(),
3249	Some(RUNTIME_RESTART_REASON)
3250	);
3251	assert!(in_progress_turn.ended_at.is_some());
3252	assert!(in_progress_turn.duration_ms.is_some());
3253
3254	let completed_turn = manager.store.load_turn(&completed_turn.id)?;
3255	assert_eq!(completed_turn.status, RuntimeTurnStatus::Completed);
3256	assert!(completed_turn.error.is_none());
3257
3258	let queued_item = manager.store.load_item("item_queued")?;
3259	assert_eq!(queued_item.status, TurnItemLifecycleStatus::Interrupted);
3260	assert!(queued_item.ended_at.is_some());
3261
3262	let in_progress_item = manager.store.load_item("item_running")?;
3263	assert_eq!(
3264	in_progress_item.status,
3265	TurnItemLifecycleStatus::Interrupted
3266	);
3267	assert!(in_progress_item.ended_at.is_some());
3268
3269	let completed_item = manager.store.load_item("item_done")?;
3270	assert_eq!(completed_item.status, TurnItemLifecycleStatus::Completed);
3271
3272	Ok(())
3273	}
3274
3275	#[tokio::test]
3276	async fn thread_lifecycle_persists_across_restart() -> Result<()> {
3277	let runtime_dir = test_runtime_dir();
3278	let manager = test_manager(runtime_dir.clone())?;
3279	let thread = manager
3280	.create_thread(CreateThreadRequest {
3281	model: None,
3282	workspace: None,
3283	mode: None,
3284	allow_shell: None,
3285	trust_mode: None,
3286	auto_approve: None,
3287	archived: false,
3288	system_prompt: None,
3289	task_id: None,
3290	})
3291	.await?;
3292
3293	let harness = install_mock_engine(&manager, &thread.id).await;
3294	let mut rx_op = harness.rx_op;
3295	let tx_event = harness.tx_event;
3296	tokio::spawn(async move {
3297	if matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
3298	let _ = tx_event
3299	.send(EngineEvent::TurnStarted {
3300	turn_id: "engine_turn_1".to_string(),
3301	})
3302	.await;
3303	let _ = tx_event
3304	.send(EngineEvent::MessageStarted { index: 0 })
3305	.await;
3306	let _ = tx_event
3307	.send(EngineEvent::MessageDelta {
3308	index: 0,
3309	content: "mock response".to_string(),
3310	})
3311	.await;
3312	let _ = tx_event
3313	.send(EngineEvent::MessageComplete { index: 0 })
3314	.await;
3315	let _ = tx_event
3316	.send(EngineEvent::CoherenceState {
3317	state: CoherenceState::GettingCrowded,
3318	label: "getting crowded".to_string(),
3319	description: "The session is approaching context pressure.".to_string(),
3320	reason: "test capacity signal".to_string(),
3321	})
3322	.await;
3323	let _ = tx_event
3324	.send(EngineEvent::TurnComplete {
3325	usage: Usage {
3326	input_tokens: 10,
3327	output_tokens: 12,
3328	..Usage::default()
3329	},
3330	status: TurnOutcomeStatus::Completed,
3331	error: None,
3332	})
3333	.await;
3334	}
3335	});
3336
3337	let turn = manager
3338	.start_turn(
3339	&thread.id,
3340	StartTurnRequest {
3341	prompt: "first prompt".to_string(),
3342	input_summary: None,
3343	model: None,
3344	mode: None,
3345	allow_shell: None,
3346	trust_mode: None,
3347	auto_approve: None,
3348	},
3349	)
3350	.await?;
3351	let completed = wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(2)).await?;
3352	assert_eq!(completed.status, RuntimeTurnStatus::Completed);
3353
3354	drop(manager);
3355
3356	let reopened = test_manager(runtime_dir)?;
3357	let detail = reopened.get_thread_detail(&thread.id).await?;
3358	assert_eq!(detail.thread.id, thread.id);
3359	assert_eq!(
3360	detail.thread.coherence_state,
3361	CoherenceState::GettingCrowded
3362	);
3363	assert_eq!(detail.turns.len(), 1);
3364	assert!(detail.latest_seq >= 1);
3365	assert!(!detail.items.is_empty());
3366	let events = reopened.events_since(&thread.id, None)?;
3367	assert!(
3368	events.iter().any(\|ev\| ev.event == "turn.completed"),
3369	"expected turn.completed event after restart"
3370	);
3371	assert!(
3372	events.iter().any(\|ev\| ev.event == "coherence.state"
3373	&& ev.payload.get("state").and_then(serde_json::Value::as_str)
3374	== Some("getting_crowded")),
3375	"expected machine-readable coherence event after restart"
3376	);
3377	Ok(())
3378	}
3379
3380	#[tokio::test]
3381	async fn create_thread_defaults_auto_approve_to_false() -> Result<()> {
3382	let manager = test_manager(test_runtime_dir())?;
3383	let thread = manager
3384	.create_thread(CreateThreadRequest {
3385	model: None,
3386	workspace: None,
3387	mode: None,
3388	allow_shell: None,
3389	trust_mode: None,
3390	auto_approve: None,
3391	archived: false,
3392	system_prompt: None,
3393	task_id: None,
3394	})
3395	.await?;
3396
3397	assert!(!thread.auto_approve);
3398	assert_eq!(thread.coherence_state, CoherenceState::Healthy);
3399	Ok(())
3400	}
3401
3402	#[tokio::test]
3403	async fn start_turn_passes_effective_auto_approve_to_engine() -> Result<()> {
3404	let manager = test_manager(test_runtime_dir())?;
3405	let thread = manager
3406	.create_thread(CreateThreadRequest {
3407	model: None,
3408	workspace: None,
3409	mode: None,
3410	allow_shell: None,
3411	trust_mode: None,
3412	auto_approve: Some(false),
3413	archived: false,
3414	system_prompt: None,
3415	task_id: None,
3416	})
3417	.await?;
3418
3419	let harness = install_mock_engine(&manager, &thread.id).await;
3420	let mut rx_op = harness.rx_op;
3421
3422	let _turn = manager
3423	.start_turn(
3424	&thread.id,
3425	StartTurnRequest {
3426	prompt: "override approval".to_string(),
3427	input_summary: None,
3428	model: None,
3429	mode: None,
3430	allow_shell: None,
3431	trust_mode: None,
3432	auto_approve: Some(true),
3433	},
3434	)
3435	.await?;
3436
3437	match rx_op.recv().await {
3438	Some(Op::SendMessage { auto_approve, .. }) => assert!(auto_approve),
3439	other => panic!("expected SendMessage op, got {other:?}"),
3440	}
3441
3442	Ok(())
3443	}
3444
3445	#[tokio::test]
3446	async fn start_turn_can_override_thread_auto_approve_to_false() -> Result<()> {
3447	let manager = test_manager(test_runtime_dir())?;
3448	let thread = manager
3449	.create_thread(CreateThreadRequest {
3450	model: None,
3451	workspace: None,
3452	mode: None,
3453	allow_shell: None,
3454	trust_mode: None,
3455	auto_approve: Some(true),
3456	archived: false,
3457	system_prompt: None,
3458	task_id: None,
3459	})
3460	.await?;
3461
3462	let harness = install_mock_engine(&manager, &thread.id).await;
3463	let mut rx_op = harness.rx_op;
3464
3465	let _turn = manager
3466	.start_turn(
3467	&thread.id,
3468	StartTurnRequest {
3469	prompt: "disable approval".to_string(),
3470	input_summary: None,
3471	model: None,
3472	mode: None,
3473	allow_shell: None,
3474	trust_mode: None,
3475	auto_approve: Some(false),
3476	},
3477	)
3478	.await?;
3479
3480	match rx_op.recv().await {
3481	Some(Op::SendMessage { auto_approve, .. }) => assert!(!auto_approve),
3482	other => panic!("expected SendMessage op, got {other:?}"),
3483	}
3484
3485	Ok(())
3486	}
3487
3488	#[tokio::test]
3489	async fn compact_thread_preserves_thread_auto_approve_policy() -> Result<()> {
3490	let manager = test_manager(test_runtime_dir())?;
3491	let thread = manager
3492	.create_thread(CreateThreadRequest {
3493	model: None,
3494	workspace: None,
3495	mode: None,
3496	allow_shell: None,
3497	trust_mode: None,
3498	auto_approve: Some(false),
3499	archived: false,
3500	system_prompt: None,
3501	task_id: None,
3502	})
3503	.await?;
3504
3505	let harness = install_mock_engine(&manager, &thread.id).await;
3506	let mut rx_op = harness.rx_op;
3507
3508	let turn = manager
3509	.compact_thread(&thread.id, CompactThreadRequest::default())
3510	.await?;
3511
3512	assert!(matches!(rx_op.recv().await, Some(Op::CompactContext)));
3513	assert_eq!(
3514	manager.active_turn_flags(&thread.id, &turn.id).await,
3515	Some((false, false))
3516	);
3517
3518	Ok(())
3519	}
3520
3521	#[tokio::test]
3522	async fn compact_thread_with_real_engine_reaches_terminal_status() -> Result<()> {
3523	let manager = test_manager(test_runtime_dir())?;
3524	let thread = manager
3525	.create_thread(CreateThreadRequest {
3526	model: None,
3527	workspace: None,
3528	mode: None,
3529	allow_shell: None,
3530	trust_mode: None,
3531	auto_approve: None,
3532	archived: false,
3533	system_prompt: None,
3534	task_id: None,
3535	})
3536	.await?;
3537
3538	let turn = manager
3539	.compact_thread(&thread.id, CompactThreadRequest::default())
3540	.await?;
3541	let terminal = wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(2)).await?;
3542
3543	assert!(matches!(
3544	terminal.status,
3545	RuntimeTurnStatus::Completed \| RuntimeTurnStatus::Failed
3546	));
3547	assert!(
3548	terminal.ended_at.is_some(),
3549	"manual compaction should reach a terminal turn state"
3550	);
3551	assert_eq!(manager.active_turn_flags(&thread.id, &turn.id).await, None);
3552
3553	let expected_status = match terminal.status {
3554	RuntimeTurnStatus::Completed => "completed",
3555	RuntimeTurnStatus::Failed => "failed",
3556	other => panic!("unexpected non-terminal compaction status: {other:?}"),
3557	};
3558	let events = manager.events_since(&thread.id, None)?;
3559	assert!(events.iter().any(\|ev\| {
3560	ev.event == "turn.completed"
3561	&& ev
3562	.payload
3563	.get("turn")
3564	.and_then(\|turn\| turn.get("status"))
3565	.and_then(Value::as_str)
3566	== Some(expected_status)
3567	}));
3568	Ok(())
3569	}
3570
3571	#[tokio::test]
3572	async fn multi_turn_continuity_same_thread() -> Result<()> {
3573	let manager = test_manager(test_runtime_dir())?;
3574	let thread = manager
3575	.create_thread(CreateThreadRequest {
3576	model: None,
3577	workspace: None,
3578	mode: None,
3579	allow_shell: None,
3580	trust_mode: None,
3581	auto_approve: None,
3582	archived: false,
3583	system_prompt: None,
3584	task_id: None,
3585	})
3586	.await?;
3587
3588	let harness = install_mock_engine(&manager, &thread.id).await;
3589	let mut rx_op = harness.rx_op;
3590	let tx_event = harness.tx_event;
3591	tokio::spawn(async move {
3592	let mut turn_index = 0u8;
3593	while let Some(op) = rx_op.recv().await {
3594	if !matches!(op, Op::SendMessage { .. }) {
3595	continue;
3596	}
3597	turn_index = turn_index.saturating_add(1);
3598	let _ = tx_event
3599	.send(EngineEvent::TurnStarted {
3600	turn_id: format!("engine_turn_{turn_index}"),
3601	})
3602	.await;
3603	let _ = tx_event
3604	.send(EngineEvent::MessageStarted { index: 0 })
3605	.await;
3606	let _ = tx_event
3607	.send(EngineEvent::MessageDelta {
3608	index: 0,
3609	content: format!("reply {turn_index}"),
3610	})
3611	.await;
3612	let _ = tx_event
3613	.send(EngineEvent::MessageComplete { index: 0 })
3614	.await;
3615	let _ = tx_event
3616	.send(EngineEvent::TurnComplete {
3617	usage: Usage {
3618	input_tokens: 5,
3619	output_tokens: 5,
3620	..Usage::default()
3621	},
3622	status: TurnOutcomeStatus::Completed,
3623	error: None,
3624	})
3625	.await;
3626	if turn_index >= 2 {
3627	break;
3628	}
3629	}
3630	});
3631
3632	let turn_1 = manager
3633	.start_turn(
3634	&thread.id,
3635	StartTurnRequest {
3636	prompt: "first".to_string(),
3637	input_summary: None,
3638	model: None,
3639	mode: None,
3640	allow_shell: None,
3641	trust_mode: None,
3642	auto_approve: None,
3643	},
3644	)
3645	.await?;
3646	let turn_1 = wait_for_terminal_turn(&manager, &turn_1.id, Duration::from_secs(2)).await?;
3647	assert_eq!(turn_1.status, RuntimeTurnStatus::Completed);
3648
3649	let turn_2 = manager
3650	.start_turn(
3651	&thread.id,
3652	StartTurnRequest {
3653	prompt: "second".to_string(),
3654	input_summary: None,
3655	model: None,
3656	mode: None,
3657	allow_shell: None,
3658	trust_mode: None,
3659	auto_approve: None,
3660	},
3661	)
3662	.await?;
3663	let turn_2 = wait_for_terminal_turn(&manager, &turn_2.id, Duration::from_secs(2)).await?;
3664	assert_eq!(turn_2.status, RuntimeTurnStatus::Completed);
3665
3666	let detail = manager.get_thread_detail(&thread.id).await?;
3667	assert_eq!(
3668	detail.thread.latest_turn_id.as_deref(),
3669	Some(turn_2.id.as_str())
3670	);
3671	assert_eq!(detail.turns.len(), 2);
3672	assert!(detail.items.iter().any(\|item\| {
3673	item.kind == TurnItemKind::UserMessage && item.detail.as_deref() == Some("first")
3674	}));
3675	assert!(detail.items.iter().any(\|item\| {
3676	item.kind == TurnItemKind::UserMessage && item.detail.as_deref() == Some("second")
3677	}));
3678
3679	let events = manager.events_since(&thread.id, None)?;
3680	let started = events
3681	.iter()
3682	.filter(\|ev\| ev.event == "turn.started")
3683	.count();
3684	let completed = events
3685	.iter()
3686	.filter(\|ev\| ev.event == "turn.completed")
3687	.count();
3688	assert_eq!(started, 2);
3689	assert_eq!(completed, 2);
3690	Ok(())
3691	}
3692
3693	#[tokio::test]
3694	async fn interrupt_turn_marks_interrupted_after_cleanup() -> Result<()> {
3695	let manager = test_manager(test_runtime_dir())?;
3696	let thread = manager
3697	.create_thread(CreateThreadRequest {
3698	model: None,
3699	workspace: None,
3700	mode: None,
3701	allow_shell: None,
3702	trust_mode: None,
3703	auto_approve: None,
3704	archived: false,
3705	system_prompt: None,
3706	task_id: None,
3707	})
3708	.await?;
3709
3710	let harness = install_mock_engine(&manager, &thread.id).await;
3711	let mut rx_op = harness.rx_op;
3712	let tx_event = harness.tx_event;
3713	let cancel_token = harness.cancel_token;
3714	let cleanup_delay = Duration::from_millis(140);
3715	tokio::spawn(async move {
3716	if matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
3717	let _ = tx_event
3718	.send(EngineEvent::TurnStarted {
3719	turn_id: "engine_turn_interrupt".to_string(),
3720	})
3721	.await;
3722	let _ = tx_event
3723	.send(EngineEvent::MessageStarted { index: 0 })
3724	.await;
3725	let _ = tx_event
3726	.send(EngineEvent::MessageDelta {
3727	index: 0,
3728	content: "partial".to_string(),
3729	})
3730	.await;
3731	cancel_token.cancelled().await;
3732	sleep(cleanup_delay).await;
3733	}
3734	});
3735
3736	let turn = manager
3737	.start_turn(
3738	&thread.id,
3739	StartTurnRequest {
3740	prompt: "interrupt me".to_string(),
3741	input_summary: None,
3742	model: None,
3743	mode: None,
3744	allow_shell: None,
3745	trust_mode: None,
3746	auto_approve: None,
3747	},
3748	)
3749	.await?;
3750
3751	sleep(Duration::from_millis(20)).await;
3752	let interrupted_at = Instant::now();
3753	let interrupt_result = manager.interrupt_turn(&thread.id, &turn.id).await?;
3754	assert_eq!(interrupt_result.status, RuntimeTurnStatus::InProgress);
3755
3756	let final_turn = wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(3)).await?;
3757	assert_eq!(final_turn.status, RuntimeTurnStatus::Interrupted);
3758	assert!(
3759	interrupted_at.elapsed() >= cleanup_delay,
3760	"turn transitioned before cleanup finished"
3761	);
3762
3763	let events = manager.events_since(&thread.id, None)?;
3764	let interrupt_seq = events
3765	.iter()
3766	.find(\|ev\| ev.event == "turn.interrupt_requested")
3767	.map(\|ev\| ev.seq)
3768	.context("missing turn.interrupt_requested event")?;
3769	let completed = events
3770	.iter()
3771	.find(\|ev\| ev.event == "turn.completed")
3772	.context("missing turn.completed event")?;
3773	assert!(completed.seq > interrupt_seq);
3774	assert_eq!(
3775	completed
3776	.payload
3777	.get("turn")
3778	.and_then(\|turn\| turn.get("status"))
3779	.and_then(Value::as_str),
3780	Some("interrupted")
3781	);
3782	Ok(())
3783	}
3784
3785	#[tokio::test]
3786	async fn approval_required_with_stale_active_turn_is_denied() -> Result<()> {
3787	let manager = test_manager(test_runtime_dir())?;
3788	let thread = manager
3789	.create_thread(CreateThreadRequest {
3790	model: None,
3791	workspace: None,
3792	mode: None,
3793	allow_shell: None,
3794	trust_mode: None,
3795	auto_approve: Some(true),
3796	archived: false,
3797	system_prompt: None,
3798	task_id: None,
3799	})
3800	.await?;
3801
3802	let mut harness = install_mock_engine(&manager, &thread.id).await;
3803	let turn = manager
3804	.start_turn(
3805	&thread.id,
3806	StartTurnRequest {
3807	prompt: "needs approval".to_string(),
3808	input_summary: None,
3809	model: None,
3810	mode: None,
3811	allow_shell: None,
3812	trust_mode: None,
3813	auto_approve: Some(true),
3814	},
3815	)
3816	.await?;
3817
3818	assert!(matches!(
3819	harness.rx_op.recv().await,
3820	Some(Op::SendMessage { .. })
3821	));
3822	{
3823	let mut active = manager.active.lock().await;
3824	let state = active
3825	.engines
3826	.get_mut(&thread.id)
3827	.context("missing active thread state")?;
3828	state.active_turn = None;
3829	}
3830
3831	harness
3832	.tx_event
3833	.send(EngineEvent::ApprovalRequired {
3834	approval_key: "test_key".to_string(),
3835	id: "tool_stale".to_string(),
3836	tool_name: "exec_command".to_string(),
3837	description: "stale approval".to_string(),
3838	})
3839	.await?;
3840
3841	assert_eq!(
3842	harness.recv_approval_event().await,
3843	Some(MockApprovalEvent::Denied {
3844	id: "tool_stale".to_string(),
3845	})
3846	);
3847
3848	harness
3849	.tx_event
3850	.send(EngineEvent::TurnComplete {
3851	usage: Usage {
3852	input_tokens: 0,
3853	output_tokens: 0,
3854	..Usage::default()
3855	},
3856	status: TurnOutcomeStatus::Completed,
3857	error: None,
3858	})
3859	.await?;
3860
3861	let terminal = wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(2)).await?;
3862	assert_eq!(terminal.status, RuntimeTurnStatus::Completed);
3863	Ok(())
3864	}
3865
3866	#[tokio::test]
3867	async fn elevation_required_with_stale_active_turn_is_denied() -> Result<()> {
3868	let manager = test_manager(test_runtime_dir())?;
3869	let thread = manager
3870	.create_thread(CreateThreadRequest {
3871	model: None,
3872	workspace: None,
3873	mode: None,
3874	allow_shell: None,
3875	trust_mode: Some(true),
3876	auto_approve: Some(true),
3877	archived: false,
3878	system_prompt: None,
3879	task_id: None,
3880	})
3881	.await?;
3882
3883	let mut harness = install_mock_engine(&manager, &thread.id).await;
3884	let turn = manager
3885	.start_turn(
3886	&thread.id,
3887	StartTurnRequest {
3888	prompt: "needs elevation".to_string(),
3889	input_summary: None,
3890	model: None,
3891	mode: None,
3892	allow_shell: None,
3893	trust_mode: Some(true),
3894	auto_approve: Some(true),
3895	},
3896	)
3897	.await?;
3898
3899	assert!(matches!(
3900	harness.rx_op.recv().await,
3901	Some(Op::SendMessage { .. })
3902	));
3903	{
3904	let mut active = manager.active.lock().await;
3905	let state = active
3906	.engines
3907	.get_mut(&thread.id)
3908	.context("missing active thread state")?;
3909	state.active_turn = None;
3910	}
3911
3912	harness
3913	.tx_event
3914	.send(EngineEvent::ElevationRequired {
3915	tool_id: "tool_stale_elevated".to_string(),
3916	tool_name: "exec_command".to_string(),
3917	command: None,
3918	denial_reason: "sandbox denied".to_string(),
3919	blocked_network: false,
3920	blocked_write: false,
3921	})
3922	.await?;
3923
3924	assert_eq!(
3925	harness.recv_approval_event().await,
3926	Some(MockApprovalEvent::Denied {
3927	id: "tool_stale_elevated".to_string(),
3928	})
3929	);
3930
3931	harness
3932	.tx_event
3933	.send(EngineEvent::TurnComplete {
3934	usage: Usage {
3935	input_tokens: 0,
3936	output_tokens: 0,
3937	..Usage::default()
3938	},
3939	status: TurnOutcomeStatus::Completed,
3940	error: None,
3941	})
3942	.await?;
3943
3944	let terminal = wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(2)).await?;
3945	assert_eq!(terminal.status, RuntimeTurnStatus::Completed);
3946	Ok(())
3947	}
3948
3949	#[tokio::test]
3950	async fn steer_turn_on_active_turn_records_item_and_event() -> Result<()> {
3951	let manager = test_manager(test_runtime_dir())?;
3952	let thread = manager
3953	.create_thread(CreateThreadRequest {
3954	model: None,
3955	workspace: None,
3956	mode: None,
3957	allow_shell: None,
3958	trust_mode: None,
3959	auto_approve: None,
3960	archived: false,
3961	system_prompt: None,
3962	task_id: None,
3963	})
3964	.await?;
3965
3966	let harness = install_mock_engine(&manager, &thread.id).await;
3967	let mut rx_op = harness.rx_op;
3968	let mut rx_steer = harness.rx_steer;
3969	let tx_event = harness.tx_event;
3970	let (steer_seen_tx, steer_seen_rx) = oneshot::channel::<String>();
3971	tokio::spawn(async move {
3972	if matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
3973	let _ = tx_event
3974	.send(EngineEvent::TurnStarted {
3975	turn_id: "engine_turn_steer".to_string(),
3976	})
3977	.await;
3978	if let Some(steer) = rx_steer.recv().await {
3979	let _ = steer_seen_tx.send(steer);
3980	}
3981	let _ = tx_event
3982	.send(EngineEvent::MessageStarted { index: 0 })
3983	.await;
3984	let _ = tx_event
3985	.send(EngineEvent::MessageDelta {
3986	index: 0,
3987	content: "steered response".to_string(),
3988	})
3989	.await;
3990	let _ = tx_event
3991	.send(EngineEvent::MessageComplete { index: 0 })
3992	.await;
3993	let _ = tx_event
3994	.send(EngineEvent::TurnComplete {
3995	usage: Usage {
3996	input_tokens: 8,
3997	output_tokens: 9,
3998	..Usage::default()
3999	},
4000	status: TurnOutcomeStatus::Completed,
4001	error: None,
4002	})
4003	.await;
4004	}
4005	});
4006
4007	let turn = manager
4008	.start_turn(
4009	&thread.id,
4010	StartTurnRequest {
4011	prompt: "initial".to_string(),
4012	input_summary: None,
4013	model: None,
4014	mode: None,
4015	allow_shell: None,
4016	trust_mode: None,
4017	auto_approve: None,
4018	},
4019	)
4020	.await?;
4021
4022	let steer_text = "add bullet list".to_string();
4023	let steered_turn = manager
4024	.steer_turn(
4025	&thread.id,
4026	&turn.id,
4027	SteerTurnRequest {
4028	prompt: steer_text.clone(),
4029	},
4030	)
4031	.await?;
4032	assert_eq!(steered_turn.steer_count, 1);
4033	let observed_steer = steer_seen_rx
4034	.await
4035	.context("driver did not receive steer")?;
4036	assert_eq!(observed_steer, steer_text);
4037
4038	let final_turn = wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(2)).await?;
4039	assert_eq!(final_turn.status, RuntimeTurnStatus::Completed);
4040	assert_eq!(final_turn.steer_count, 1);
4041
4042	let events = manager.events_since(&thread.id, None)?;
4043	assert!(events.iter().any(\|ev\| ev.event == "turn.steered"));
4044	assert!(events.iter().any(\|ev\| {
4045	ev.event == "item.completed"
4046	&& ev
4047	.payload
4048	.get("item")
4049	.and_then(\|item\| item.get("detail"))
4050	.and_then(Value::as_str)
4051	== Some("add bullet list")
4052	}));
4053	Ok(())
4054	}
4055
4056	#[tokio::test]
4057	async fn compaction_lifecycle_emits_item_events_with_compaction_counts() -> Result<()> {
4058	let manager = test_manager(test_runtime_dir())?;
4059	let thread = manager
4060	.create_thread(CreateThreadRequest {
4061	model: None,
4062	workspace: None,
4063	mode: None,
4064	allow_shell: None,
4065	trust_mode: None,
4066	auto_approve: None,
4067	archived: false,
4068	system_prompt: None,
4069	task_id: None,
4070	})
4071	.await?;
4072
4073	let harness = install_mock_engine(&manager, &thread.id).await;
4074	let mut rx_op = harness.rx_op;
4075	let tx_event = harness.tx_event;
4076	tokio::spawn(async move {
4077	let mut op_count = 0usize;
4078	while let Some(op) = rx_op.recv().await {
4079	match op {
4080	Op::SendMessage { .. } => {
4081	op_count = op_count.saturating_add(1);
4082	let _ = tx_event
4083	.send(EngineEvent::TurnStarted {
4084	turn_id: "engine_turn_auto".to_string(),
4085	})
4086	.await;
4087	let _ = tx_event
4088	.send(EngineEvent::CompactionStarted {
4089	id: "auto_compact_1".to_string(),
4090	auto: true,
4091	message: "auto compact begin".to_string(),
4092	})
4093	.await;
4094	let _ = tx_event
4095	.send(EngineEvent::CompactionCompleted {
4096	id: "auto_compact_1".to_string(),
4097	auto: true,
4098	message: "auto compact done".to_string(),
4099	messages_before: Some(7),
4100	messages_after: Some(3),
4101	})
4102	.await;
4103	let _ = tx_event
4104	.send(EngineEvent::TurnComplete {
4105	usage: Usage {
4106	input_tokens: 3,
4107	output_tokens: 3,
4108	..Usage::default()
4109	},
4110	status: TurnOutcomeStatus::Completed,
4111	error: None,
4112	})
4113	.await;
4114	}
4115	Op::CompactContext => {
4116	op_count = op_count.saturating_add(1);
4117	let _ = tx_event
4118	.send(EngineEvent::CompactionStarted {
4119	id: "manual_compact_1".to_string(),
4120	auto: false,
4121	message: "manual compact begin".to_string(),
4122	})
4123	.await;
4124	let _ = tx_event
4125	.send(EngineEvent::CompactionCompleted {
4126	id: "manual_compact_1".to_string(),
4127	auto: false,
4128	message: "manual compact done".to_string(),
4129	messages_before: Some(5),
4130	messages_after: Some(2),
4131	})
4132	.await;
4133	let _ = tx_event
4134	.send(EngineEvent::TurnComplete {
4135	usage: Usage {
4136	input_tokens: 1,
4137	output_tokens: 1,
4138	..Usage::default()
4139	},
4140	status: TurnOutcomeStatus::Completed,
4141	error: None,
4142	})
4143	.await;
4144	}
4145	_ => {}
4146	}
4147	if op_count >= 2 {
4148	break;
4149	}
4150	}
4151	});
4152
4153	let auto_turn = manager
4154	.start_turn(
4155	&thread.id,
4156	StartTurnRequest {
4157	prompt: "trigger auto".to_string(),
4158	input_summary: None,
4159	model: None,
4160	mode: None,
4161	allow_shell: None,
4162	trust_mode: None,
4163	auto_approve: None,
4164	},
4165	)
4166	.await?;
4167	let auto_turn =
4168	wait_for_terminal_turn(&manager, &auto_turn.id, Duration::from_secs(2)).await?;
4169	assert_eq!(auto_turn.status, RuntimeTurnStatus::Completed);
4170
4171	let manual_turn = manager
4172	.compact_thread(
4173	&thread.id,
4174	CompactThreadRequest {
4175	reason: Some("manual request".to_string()),
4176	},
4177	)
4178	.await?;
4179	let manual_turn =
4180	wait_for_terminal_turn(&manager, &manual_turn.id, Duration::from_secs(2)).await?;
4181	assert_eq!(manual_turn.status, RuntimeTurnStatus::Completed);
4182
4183	let events = manager.events_since(&thread.id, None)?;
4184	assert!(events.iter().any(\|ev\| {
4185	ev.event == "item.started"
4186	&& ev
4187	.payload
4188	.get("item")
4189	.and_then(\|item\| item.get("kind"))
4190	.and_then(Value::as_str)
4191	== Some("context_compaction")
4192	&& ev.payload.get("auto").and_then(Value::as_bool) == Some(true)
4193	}));
4194	assert!(events.iter().any(\|ev\| {
4195	ev.event == "item.completed"
4196	&& ev
4197	.payload
4198	.get("item")
4199	.and_then(\|item\| item.get("kind"))
4200	.and_then(Value::as_str)
4201	== Some("context_compaction")
4202	&& ev.payload.get("auto").and_then(Value::as_bool) == Some(true)
4203	&& ev.payload.get("messages_before").and_then(Value::as_u64) == Some(7)
4204	&& ev.payload.get("messages_after").and_then(Value::as_u64) == Some(3)
4205	}));
4206	assert!(events.iter().any(\|ev\| {
4207	ev.event == "item.completed"
4208	&& ev
4209	.payload
4210	.get("item")
4211	.and_then(\|item\| item.get("kind"))
4212	.and_then(Value::as_str)
4213	== Some("context_compaction")
4214	&& ev.payload.get("auto").and_then(Value::as_bool) == Some(false)
4215	&& ev.payload.get("messages_before").and_then(Value::as_u64) == Some(5)
4216	&& ev.payload.get("messages_after").and_then(Value::as_u64) == Some(2)
4217	}));
4218	Ok(())
4219	}
4220
4221	#[test]
4222	fn summarize_text_truncates() {
4223	let out = summarize_text("abcdefghijklmnopqrstuvwxyz", 10);
4224	assert_eq!(out, "abcdefg...");
4225	}
4226
4227	#[test]
4228	fn approval_decision_requires_auto_approve_and_trust_for_full_access() {
4229	assert_eq!(
4230	RuntimeThreadManager::approval_decision(false, false, false),
4231	RuntimeApprovalDecision::DenyTool
4232	);
4233	assert_eq!(
4234	RuntimeThreadManager::approval_decision(true, false, false),
4235	RuntimeApprovalDecision::ApproveTool
4236	);
4237	assert_eq!(
4238	RuntimeThreadManager::approval_decision(true, false, true),
4239	RuntimeApprovalDecision::DenyTool
4240	);
4241	assert_eq!(
4242	RuntimeThreadManager::approval_decision(true, true, true),
4243	RuntimeApprovalDecision::RetryWithFullAccess
4244	);
4245	}
4246
4247	#[test]
4248	fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
4249	let data_dir = test_runtime_dir();
4250	let manager = test_manager(data_dir.clone())?;
4251	let started_at = Utc::now() - chrono::Duration::seconds(5);
4252	let created_at = started_at - chrono::Duration::seconds(1);
4253
4254	let thread = ThreadRecord {
4255	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4256	id: "thr_restart".to_string(),
4257	created_at,
4258	updated_at: created_at,
4259	model: DEFAULT_TEXT_MODEL.to_string(),
4260	workspace: PathBuf::from("."),
4261	mode: "agent".to_string(),
4262	allow_shell: false,
4263	trust_mode: false,
4264	auto_approve: false,
4265	latest_turn_id: Some("turn_in_progress".to_string()),
4266	latest_response_bookmark: None,
4267	archived: false,
4268	system_prompt: None,
4269	task_id: None,
4270	title: None,
4271	coherence_state: CoherenceState::default(),
4272	};
4273	manager.store.save_thread(&thread)?;
4274
4275	let completed_item = TurnItemRecord {
4276	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4277	id: "item_completed".to_string(),
4278	turn_id: "turn_in_progress".to_string(),
4279	kind: TurnItemKind::Status,
4280	status: TurnItemLifecycleStatus::Completed,
4281	summary: "done".to_string(),
4282	detail: None,
4283	metadata: None,
4284	artifact_refs: Vec::new(),
4285	started_at: Some(started_at),
4286	ended_at: Some(started_at + chrono::Duration::seconds(1)),
4287	};
4288	let in_progress_item = TurnItemRecord {
4289	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4290	id: "item_in_progress".to_string(),
4291	turn_id: "turn_in_progress".to_string(),
4292	kind: TurnItemKind::ToolCall,
4293	status: TurnItemLifecycleStatus::InProgress,
4294	summary: "running".to_string(),
4295	detail: None,
4296	metadata: None,
4297	artifact_refs: Vec::new(),
4298	started_at: Some(started_at),
4299	ended_at: None,
4300	};
4301	let queued_item = TurnItemRecord {
4302	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4303	id: "item_queued".to_string(),
4304	turn_id: "turn_queued".to_string(),
4305	kind: TurnItemKind::ToolCall,
4306	status: TurnItemLifecycleStatus::Queued,
4307	summary: "queued".to_string(),
4308	detail: None,
4309	metadata: None,
4310	artifact_refs: Vec::new(),
4311	started_at: None,
4312	ended_at: None,
4313	};
4314	manager.store.save_item(&completed_item)?;
4315	manager.store.save_item(&in_progress_item)?;
4316	manager.store.save_item(&queued_item)?;
4317
4318	manager.store.save_turn(&TurnRecord {
4319	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4320	id: "turn_in_progress".to_string(),
4321	thread_id: thread.id.clone(),
4322	status: RuntimeTurnStatus::InProgress,
4323	input_summary: "hello".to_string(),
4324	created_at,
4325	started_at: Some(started_at),
4326	ended_at: None,
4327	duration_ms: None,
4328	usage: None,
4329	error: None,
4330	item_ids: vec![completed_item.id.clone(), in_progress_item.id.clone()],
4331	steer_count: 0,
4332	})?;
4333	manager.store.save_turn(&TurnRecord {
4334	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4335	id: "turn_queued".to_string(),
4336	thread_id: thread.id.clone(),
4337	status: RuntimeTurnStatus::Queued,
4338	input_summary: "later".to_string(),
4339	created_at,
4340	started_at: None,
4341	ended_at: None,
4342	duration_ms: None,
4343	usage: None,
4344	error: None,
4345	item_ids: vec![queued_item.id.clone()],
4346	steer_count: 0,
4347	})?;
4348	drop(manager);
4349
4350	let recovered = test_manager(data_dir)?;
4351
4352	let recovered_thread = recovered.store.load_thread(&thread.id)?;
4353	assert!(recovered_thread.updated_at >= thread.updated_at);
4354
4355	let recovered_in_progress_turn = recovered.store.load_turn("turn_in_progress")?;
4356	assert_eq!(
4357	recovered_in_progress_turn.status,
4358	RuntimeTurnStatus::Interrupted
4359	);
4360	assert_eq!(
4361	recovered_in_progress_turn.error.as_deref(),
4362	Some(RUNTIME_RESTART_REASON)
4363	);
4364	assert!(recovered_in_progress_turn.ended_at.is_some());
4365	assert!(
4366	recovered_in_progress_turn
4367	.duration_ms
4368	.is_some_and(\|duration\| duration >= 5_000)
4369	);
4370
4371	let recovered_queued_turn = recovered.store.load_turn("turn_queued")?;
4372	assert_eq!(recovered_queued_turn.status, RuntimeTurnStatus::Interrupted);
4373	assert_eq!(
4374	recovered_queued_turn.error.as_deref(),
4375	Some(RUNTIME_RESTART_REASON)
4376	);
4377	assert!(recovered_queued_turn.ended_at.is_some());
4378	assert_eq!(recovered_queued_turn.duration_ms, None);
4379
4380	assert_eq!(
4381	recovered.store.load_item(&completed_item.id)?.status,
4382	TurnItemLifecycleStatus::Completed
4383	);
4384	let recovered_in_progress_item = recovered.store.load_item(&in_progress_item.id)?;
4385	assert_eq!(
4386	recovered_in_progress_item.status,
4387	TurnItemLifecycleStatus::Interrupted
4388	);
4389	assert!(recovered_in_progress_item.ended_at.is_some());
4390
4391	let recovered_queued_item = recovered.store.load_item(&queued_item.id)?;
4392	assert_eq!(
4393	recovered_queued_item.status,
4394	TurnItemLifecycleStatus::Interrupted
4395	);
4396	assert!(recovered_queued_item.ended_at.is_some());
4397
4398	Ok(())
4399	}
4400
4401	#[test]
4402	fn parse_mode_defaults_to_agent() {
4403	assert_eq!(parse_mode("unknown"), AppMode::Agent);
4404	assert_eq!(parse_mode("plan"), AppMode::Plan);
4405	}
4406
4407	fn rebind_event(event: &str, agent_id: &str, seq: u64) -> RuntimeEventRecord {
4408	RuntimeEventRecord {
4409	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4410	seq,
4411	timestamp: Utc::now(),
4412	thread_id: "thr_test".to_string(),
4413	turn_id: Some("turn_test".to_string()),
4414	item_id: None,
4415	event: event.to_string(),
4416	payload: json!({ "agent_id": agent_id }),
4417	}
4418	}
4419
4420	#[test]
4421	fn collect_agent_rebind_hints_resumes_a_mid_fanout_session() {
4422	// Mirror what runtime_threads persists during a real fanout: three
4423	// workers spawned, two finished, one still running when the session
4424	// was killed. The TUI re-attach must rebuild placeholders for the
4425	// running worker AND the two completed workers (the fanout card
4426	// tracks all of them so the dot-grid stays accurate post-resume).
4427	let events = vec![
4428	rebind_event("agent.spawned", "agent_a", 1),
4429	rebind_event("agent.spawned", "agent_b", 2),
4430	rebind_event("agent.spawned", "agent_c", 3),
4431	rebind_event("agent.progress", "agent_a", 4),
4432	rebind_event("agent.completed", "agent_a", 5),
4433	rebind_event("agent.progress", "agent_b", 6),
4434	rebind_event("agent.completed", "agent_b", 7),
4435	rebind_event("agent.progress", "agent_c", 8),
4436	];
4437	let hints = collect_agent_rebind_hints(&events);
4438	assert_eq!(hints.len(), 3, "every fanout worker must be rebound");
4439	let by_id: std::collections::BTreeMap<&str, AgentRebindStatus> = hints
4440	.iter()
4441	.map(\|h\| (h.agent_id.as_str(), h.status))
4442	.collect();
4443	assert_eq!(by_id.get("agent_a"), Some(&AgentRebindStatus::Completed));
4444	assert_eq!(by_id.get("agent_b"), Some(&AgentRebindStatus::Completed));
4445	assert_eq!(
4446	by_id.get("agent_c"),
4447	Some(&AgentRebindStatus::InProgress),
4448	"in-flight worker must rebind in InProgress, not downgrade"
4449	);
4450	}
4451
4452	#[test]
4453	fn collect_agent_rebind_hints_ignores_unrelated_events() {
4454	// Status / tool events should not produce phantom hints — only the
4455	// agent.* family carries the contract we re-bind from.
4456	let events = vec![
4457	RuntimeEventRecord {
4458	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4459	seq: 1,
4460	timestamp: Utc::now(),
4461	thread_id: "thr".to_string(),
4462	turn_id: None,
4463	item_id: None,
4464	event: "tool.completed".to_string(),
4465	payload: json!({"name": "read_file"}),
4466	},
4467	rebind_event("agent.spawned", "agent_x", 2),
4468	RuntimeEventRecord {
4469	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4470	seq: 3,
4471	timestamp: Utc::now(),
4472	thread_id: "thr".to_string(),
4473	turn_id: None,
4474	item_id: None,
4475	event: "compaction.completed".to_string(),
4476	payload: json!({"messages_after": 12}),
4477	},
4478	];
4479	let hints = collect_agent_rebind_hints(&events);
4480	assert_eq!(hints.len(), 1);
4481	assert_eq!(hints[0].agent_id, "agent_x");
4482	}
4483
4484	#[test]
4485	fn collect_agent_rebind_hints_does_not_downgrade_completed_to_in_progress() {
4486	// Out-of-order replay: a stale `agent.progress` arriving after the
4487	// completed event must NOT clobber the terminal status. This matters
4488	// when an event log is concatenated from interrupted segments.
4489	let events = vec![
4490	rebind_event("agent.spawned", "agent_y", 1),
4491	rebind_event("agent.completed", "agent_y", 2),
4492	rebind_event("agent.progress", "agent_y", 3),
4493	];
4494	let hints = collect_agent_rebind_hints(&events);
4495	assert_eq!(hints.len(), 1);
4496	assert_eq!(hints[0].status, AgentRebindStatus::Completed);
4497	}
4498
4499	/// Helper for the `fork_at_user_message` tests: write a sequence of
4500	/// (user, assistant) turns under the given thread id. Each turn gets
4501	/// one UserMessage item carrying `user_text` in `detail` plus one
4502	/// AgentMessage item. Turn `created_at` is monotonically increasing
4503	/// so the chronological sort in `list_turns_for_thread` is stable.
4504	fn seed_turns_with_user_messages(
4505	manager: &RuntimeThreadManager,
4506	thread_id: &str,
4507	user_texts: &[&str],
4508	) -> Result<Vec<String>> {
4509	let mut turn_ids = Vec::new();
4510	let base = Utc::now();
4511	for (offset, text) in user_texts.iter().enumerate() {
4512	let created_at = base + chrono::Duration::milliseconds(offset as i64);
4513	let turn_id = format!("turn_test_{offset}");
4514	let user_item_id = format!("item_user_{offset}");
4515	let asst_item_id = format!("item_asst_{offset}");
4516	manager.store.save_item(&TurnItemRecord {
4517	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4518	id: user_item_id.clone(),
4519	turn_id: turn_id.clone(),
4520	kind: TurnItemKind::UserMessage,
4521	status: TurnItemLifecycleStatus::Completed,
4522	summary: (*text).to_string(),
4523	detail: Some((*text).to_string()),
4524	metadata: None,
4525	artifact_refs: Vec::new(),
4526	started_at: Some(created_at),
4527	ended_at: Some(created_at),
4528	})?;
4529	manager.store.save_item(&TurnItemRecord {
4530	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4531	id: asst_item_id.clone(),
4532	turn_id: turn_id.clone(),
4533	kind: TurnItemKind::AgentMessage,
4534	status: TurnItemLifecycleStatus::Completed,
4535	summary: format!("reply {offset}"),
4536	detail: Some(format!("reply {offset}")),
4537	metadata: None,
4538	artifact_refs: Vec::new(),
4539	started_at: Some(created_at),
4540	ended_at: Some(created_at),
4541	})?;
4542	manager.store.save_turn(&TurnRecord {
4543	schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
4544	id: turn_id.clone(),
4545	thread_id: thread_id.to_string(),
4546	status: RuntimeTurnStatus::Completed,
4547	input_summary: (*text).to_string(),
4548	created_at,
4549	started_at: Some(created_at),
4550	ended_at: Some(created_at),
4551	duration_ms: Some(0),
4552	usage: None,
4553	error: None,
4554	item_ids: vec![user_item_id, asst_item_id],
4555	steer_count: 0,
4556	})?;
4557	turn_ids.push(turn_id);
4558	}
4559	Ok(turn_ids)
4560	}
4561
4562	#[tokio::test]
4563	async fn fork_at_user_message_drops_tail_and_returns_user_text() -> Result<()> {
4564	// Seed three completed user/assistant turns. Backtracking with
4565	// depth=0 should drop only the most recent turn ("third") and
4566	// hand back its original text so the caller can refill the
4567	// composer.
4568	let manager = test_manager(test_runtime_dir())?;
4569	let thread = manager
4570	.create_thread(CreateThreadRequest {
4571	model: None,
4572	workspace: None,
4573	mode: None,
4574	allow_shell: None,
4575	trust_mode: None,
4576	auto_approve: None,
4577	archived: false,
4578	system_prompt: None,
4579	task_id: None,
4580	})
4581	.await?;
4582	seed_turns_with_user_messages(&manager, &thread.id, &["first", "second", "third"])?;
4583
4584	let (forked, original_text) = manager.fork_at_user_message(&thread.id, 0).await?;
4585	assert_eq!(original_text.as_deref(), Some("third"));
4586	assert_ne!(forked.id, thread.id);
4587
4588	let forked_turns = manager.store.list_turns_for_thread(&forked.id)?;
4589	assert_eq!(
4590	forked_turns.len(),
4591	2,
4592	"depth=0 should drop the most recent turn"
4593	);
4594	let summaries: Vec<&str> = forked_turns
4595	.iter()
4596	.map(\|t\| t.input_summary.as_str())
4597	.collect();
4598	assert_eq!(summaries, vec!["first", "second"]);
4599	Ok(())
4600	}
4601
4602	#[tokio::test]
4603	async fn fork_at_user_message_depth_one_drops_two_turns() -> Result<()> {
4604	let manager = test_manager(test_runtime_dir())?;
4605	let thread = manager
4606	.create_thread(CreateThreadRequest {
4607	model: None,
4608	workspace: None,
4609	mode: None,
4610	allow_shell: None,
4611	trust_mode: None,
4612	auto_approve: None,
4613	archived: false,
4614	system_prompt: None,
4615	task_id: None,
4616	})
4617	.await?;
4618	seed_turns_with_user_messages(&manager, &thread.id, &["a", "b", "c", "d"])?;
4619
4620	let (forked, original_text) = manager.fork_at_user_message(&thread.id, 1).await?;
4621	assert_eq!(original_text.as_deref(), Some("c"));
4622	let forked_turns = manager.store.list_turns_for_thread(&forked.id)?;
4623	let summaries: Vec<&str> = forked_turns
4624	.iter()
4625	.map(\|t\| t.input_summary.as_str())
4626	.collect();
4627	assert_eq!(summaries, vec!["a", "b"]);
4628	Ok(())
4629	}
4630
4631	#[tokio::test]
4632	async fn fork_at_user_message_out_of_range_errors() -> Result<()> {
4633	let manager = test_manager(test_runtime_dir())?;
4634	let thread = manager
4635	.create_thread(CreateThreadRequest {
4636	model: None,
4637	workspace: None,
4638	mode: None,
4639	allow_shell: None,
4640	trust_mode: None,
4641	auto_approve: None,
4642	archived: false,
4643	system_prompt: None,
4644	task_id: None,
4645	})
4646	.await?;
4647	seed_turns_with_user_messages(&manager, &thread.id, &["only"])?;
4648
4649	let err = manager.fork_at_user_message(&thread.id, 5).await.err();
4650	assert!(err.is_some(), "depth past the end should bail out");
4651	Ok(())
4652	}
4653
4654	#[tokio::test]
4655	async fn fork_at_user_message_does_not_mutate_source() -> Result<()> {
4656	// The source thread must be untouched: turns still present, items
4657	// still present, latest_turn_id still pointing at the original
4658	// tail. Backtrack creates a sibling, never edits in place.
4659	let manager = test_manager(test_runtime_dir())?;
4660	let thread = manager
4661	.create_thread(CreateThreadRequest {
4662	model: None,
4663	workspace: None,
4664	mode: None,
4665	allow_shell: None,
4666	trust_mode: None,
4667	auto_approve: None,
4668	archived: false,
4669	system_prompt: None,
4670	task_id: None,
4671	})
4672	.await?;
4673	let turn_ids = seed_turns_with_user_messages(&manager, &thread.id, &["x", "y", "z"])?;
4674
4675	let _ = manager.fork_at_user_message(&thread.id, 0).await?;
4676
4677	let source_turns = manager.store.list_turns_for_thread(&thread.id)?;
4678	assert_eq!(
4679	source_turns.len(),
4680	3,
4681	"source thread must still hold every turn after fork"
4682	);
4683	for tid in &turn_ids {
4684	assert!(
4685	manager.store.load_turn(tid).is_ok(),
4686	"turn {tid} must remain on disk"
4687	);
4688	}
4689	Ok(())
4690	}
4691	}
4692

4692 lines RUST