tests.rs - CodeWhale

tests.rs

1	use super::*;
2	use crate::core::events::{Event as EngineEvent, TurnOutcomeStatus};
3	use crate::core::ops::Op;
4	use crate::models::Usage;
5	use crate::runtime_threads::RuntimeEventRecord;
6	use crate::test_support::{EnvVarGuard, lock_test_env};
7	use anyhow::{Context, bail};
8	use futures_util::StreamExt;
9	use std::fs;
10	use std::path::Path;
11	use std::sync::Arc;
12	use tokio::sync::{Mutex, mpsc, oneshot};
13	use tokio::time::sleep;
14	use uuid::Uuid;
15
16	/// Scale a wait budget for shared CI runners.
17	///
18	/// These deadlines are tuned for a developer laptop running one test at a
19	/// time. CI runs the whole workspace suite on a shared runner, where the same
20	/// async progress can legitimately take several times longer. Every budget
21	/// guarded by this helper is a deadline on a poll or a oneshot that resolves
22	/// as soon as the runtime makes progress, so a larger budget does not slow a
23	/// passing run down — it only changes how long a genuinely stuck test waits
24	/// before it fails. Keeping the local value tight preserves fast feedback
25	/// while the tests stop failing for being unlucky about scheduling.
26	fn ci_scaled(base: Duration) -> Duration {
27	if std::env::var_os("CI").is_some() {
28	base * 4
29	} else {
30	base
31	}
32	}
33
34	struct MockExecutor;
35
36	#[cfg(unix)]
37	#[test]
38	fn runtime_session_fallback_retains_non_unicode_explicit_home_boundary() {
39	use std::os::unix::ffi::OsStringExt;
40
41	let _lock = lock_test_env();
42	let tmp = tempfile::tempdir().expect("temporary root");
43	let home = tmp.path().join("home");
44	let explicit = tmp.path().join(std::ffi::OsString::from_vec(
45	b"codewhale-\xff-home".to_vec(),
46	));
47	let _home = EnvVarGuard::set("HOME", &home);
48	let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", &explicit);
49
50	assert_eq!(fallback_sessions_dir(), explicit.join("sessions"));
51	}
52
53	#[test]
54	fn thread_route_credential_error_is_bad_request_not_not_found() {
55	let credential = map_thread_err(anyhow::anyhow!("DeepSeek API key not found"));
56	assert_eq!(credential.status, StatusCode::BAD_REQUEST);
57
58	let missing = map_thread_err(anyhow::anyhow!("thread 'thr_missing' not found"));
59	assert_eq!(missing.status, StatusCode::NOT_FOUND);
60	}
61
62	#[test]
63	fn runtime_tui_settings_reject_legacy_modes_and_do_not_save_env_overlays() -> Result<()> {
64	let _lock = lock_test_env();
65	let tmp = tempfile::tempdir()?;
66	let settings_dir = tmp.path().join(".codewhale");
67	fs::create_dir_all(&settings_dir)?;
68	fs::write(
69	settings_dir.join("settings.toml"),
70	"default_mode = \"plan\"\nlow_motion = false\nfancy_animations = true\nauto_compact = true\n",
71	)?;
72	let _config_path = EnvVarGuard::set(
73	"DEEPSEEK_CONFIG_PATH",
74	settings_dir.join("config.toml").as_os_str(),
75	);
76	let _no_animations = EnvVarGuard::set("NO_ANIMATIONS", "1");
77
78	// yolo remains a permission-migration alias, not a startup mode write.
79	let error = persist_runtime_tui_setting("default_mode", "yolo")
80	.expect_err("yolo must not be saved as a startup mode");
81	assert_eq!(error.status, StatusCode::BAD_REQUEST);
82	assert_eq!(
83	crate::settings::Settings::load_persisted()?.default_mode,
84	"plan",
85	"a rejected write must leave the saved startup mode intact"
86	);
87
88	persist_runtime_tui_setting("default_mode", "operate")
89	.expect("operate is a valid startup mode");
90	assert_eq!(
91	crate::settings::Settings::load_persisted()?.default_mode,
92	"operate"
93	);
94	persist_runtime_tui_setting("default_mode", "agent")
95	.expect("agent should be a valid startup mode");
96	persist_runtime_tui_setting("auto_compact", "false").expect("strict boolean should persist");
97	let saved = crate::settings::Settings::load_persisted()?;
98	assert_eq!(saved.default_mode, "agent");
99	assert!(!saved.auto_compact);
100	assert!(!saved.low_motion, "NO_ANIMATIONS is runtime-only");
101	assert!(saved.fancy_animations, "NO_ANIMATIONS is runtime-only");
102	Ok(())
103	}
104
105	#[async_trait::async_trait]
106	impl crate::task_manager::TaskExecutor for MockExecutor {
107	async fn execute(
108	&self,
109	_task: crate::task_manager::ExecutionTask,
110	events: mpsc::UnboundedSender<crate::task_manager::TaskExecutionEvent>,
111	cancel: tokio_util::sync::CancellationToken,
112	) -> crate::task_manager::TaskExecutionResult {
113	let _ = events.send(crate::task_manager::TaskExecutionEvent::Status {
114	message: "started".to_string(),
115	});
116	sleep(Duration::from_millis(100)).await;
117	if cancel.is_cancelled() {
118	return crate::task_manager::TaskExecutionResult {
119	status: crate::task_manager::TaskStatus::Canceled,
120	result_text: None,
121	error: None,
122	};
123	}
124	crate::task_manager::TaskExecutionResult {
125	status: crate::task_manager::TaskStatus::Completed,
126	result_text: Some("ok".to_string()),
127	error: None,
128	}
129	}
130	}
131
132	fn saved_session_with_blocks(blocks: Vec<crate::models::ContentBlock>) -> SavedSession {
133	SavedSession {
134	schema_version: 1,
135	metadata: SessionMetadata {
136	id: "session-1".to_string(),
137	title: "test session".to_string(),
138	created_at: Utc::now(),
139	updated_at: Utc::now(),
140	message_count: 1,
141	total_tokens: 0,
142	model: "test-model".to_string(),
143	model_provider: "deepseek".to_string(),
144	model_provider_id: None,
145	workspace: PathBuf::from("."),
146	mode: None,
147	cost: Default::default(),
148	parent_session_id: None,
149	forked_from_message_count: None,
150	cumulative_turn_secs: 0,
151	archived: false,
152	},
153	messages: vec![crate::models::Message {
154	role: "assistant".to_string(),
155	content: blocks,
156	}],
157	system_prompt: None,
158	context_references: Vec::new(),
159	artifacts: Vec::new(),
160	work_state: None,
161	last_auto_route: None,
162	}
163	}
164
165	fn run_test_git(workspace: &std::path::Path, args: &[&str]) -> Result<()> {
166	let output = crate::dependencies::Git::output(args, workspace)
167	.with_context(\|\| format!("git {args:?} failed to spawn"))?;
168	if !output.status.success() {
169	bail!(
170	"git {args:?} failed: {}",
171	String::from_utf8_lossy(&output.stderr)
172	);
173	}
174	Ok(())
175	}
176
177	#[test]
178	fn workspace_status_reports_head_and_dirty_counts() -> Result<()> {
179	let tmp = tempfile::tempdir()?;
180	let repo = tmp.path().join("repo");
181	fs::create_dir_all(&repo)?;
182	run_test_git(&repo, &["init", "-b", "main"])?;
183	run_test_git(&repo, &["config", "core.autocrlf", "false"])?;
184	fs::write(repo.join("tracked.txt"), "clean\n")?;
185	run_test_git(&repo, &["add", "tracked.txt"])?;
186	run_test_git(
187	&repo,
188	&[
189	"-c",
190	"user.name=CodeWhale Test",
191	"-c",
192	"user.email=codewhale@example.invalid",
193	"commit",
194	"-m",
195	"init",
196	],
197	)?;
198
199	let clean = collect_workspace_status(&repo);
200	assert!(clean.git_repo);
201	assert_eq!(clean.branch.as_deref(), Some("main"));
202	assert!(clean.head.as_deref().is_some_and(\|head\| !head.is_empty()));
203	assert!(!clean.dirty);
204
205	fs::write(repo.join("tracked.txt"), "dirty\n")?;
206	fs::write(repo.join("untracked.txt"), "new\n")?;
207
208	let dirty = collect_workspace_status(&repo);
209	assert!(dirty.dirty);
210	assert_eq!(dirty.unstaged, 1);
211	assert_eq!(dirty.untracked, 1);
212	Ok(())
213	}
214
215	#[test]
216	fn session_detail_tool_use_preserves_caller_metadata() {
217	let detail = session_to_detail(saved_session_with_blocks(vec![
218	crate::models::ContentBlock::ToolUse {
219	id: "tool-1".to_string(),
220	name: "task_shell_start".to_string(),
221	input: json!({ "cmd": "cargo test" }),
222	caller: Some(crate::models::ToolCaller {
223	caller_type: "subagent".to_string(),
224	tool_id: Some("parent-tool".to_string()),
225	}),
226	},
227	]));
228
229	let block = &detail.messages[0]["content"][0];
230	assert_eq!(block["type"].as_str(), Some("tool_use"));
231	assert_eq!(block["caller"]["type"].as_str(), Some("subagent"));
232	assert_eq!(block["caller"]["tool_id"].as_str(), Some("parent-tool"));
233	}
234
235	#[test]
236	fn session_detail_tool_result_keeps_fallback_content_with_blocks() {
237	let detail = session_to_detail(saved_session_with_blocks(vec![
238	crate::models::ContentBlock::ToolResult {
239	tool_use_id: "tool-1".to_string(),
240	content: "fallback text".to_string(),
241	is_error: Some(false),
242	content_blocks: Some(vec![json!({
243	"type": "text",
244	"text": "structured text"
245	})]),
246	},
247	]));
248
249	let block = &detail.messages[0]["content"][0];
250	assert_eq!(block["type"].as_str(), Some("tool_result"));
251	assert_eq!(block["content"].as_str(), Some("fallback text"));
252	assert_eq!(
253	block["content_blocks"][0]["text"].as_str(),
254	Some("structured text")
255	);
256	assert_eq!(block["is_error"].as_bool(), Some(false));
257	}
258
259	#[test]
260	fn messages_from_thread_detail_batches_tool_results() {
261	let now = Utc::now();
262	let turn_id = "turn_detail".to_string();
263	let thread = ThreadRecord {
264	schema_version: 2,
265	id: "thr_detail".to_string(),
266	created_at: now,
267	updated_at: now,
268	model: DEFAULT_TEXT_MODEL.to_string(),
269	model_provider: None,
270	model_provider_id: None,
271	workspace: PathBuf::from("."),
272	mode: "agent".to_string(),
273	permission_posture: Some("ask".to_string()),
274	allow_shell: false,
275	trust_mode: false,
276	auto_approve: false,
277	latest_turn_id: Some(turn_id.clone()),
278	latest_response_bookmark: None,
279	archived: false,
280	system_prompt: None,
281	task_id: None,
282	title: None,
283	session_id: None,
284	};
285	let turn = TurnRecord {
286	schema_version: 2,
287	id: turn_id.clone(),
288	thread_id: thread.id.clone(),
289	status: RuntimeTurnStatus::Completed,
290	input_summary: "check".to_string(),
291	created_at: now,
292	started_at: Some(now),
293	ended_at: Some(now),
294	duration_ms: Some(0),
295	usage: None,
296	permission_posture: Some("ask".to_string()),
297	effective_provider: None,
298	effective_provider_id: None,
299	effective_billing_surface: None,
300	effective_endpoint_fingerprint: None,
301	effective_billing_mode: None,
302	effective_dispatched_at: None,
303	effective_model: None,
304	routed_usage: Vec::new(),
305	routed_usage_source_ids: Vec::new(),
306	routed_usage_dropped_records: 0,
307	error: None,
308	item_ids: vec![
309	"item_user".to_string(),
310	"item_reasoning".to_string(),
311	"item_tool_use".to_string(),
312	"item_result_one".to_string(),
313	"item_result_two".to_string(),
314	"item_answer".to_string(),
315	],
316	steer_count: 0,
317	};
318	let item = \|id: &str,
319	kind: TurnItemKind,
320	summary: &str,
321	detail: Option<&str>,
322	metadata: Option<Value>\| {
323	crate::runtime_threads::TurnItemRecord {
324	schema_version: 2,
325	id: id.to_string(),
326	turn_id: turn_id.clone(),
327	kind,
328	status: TurnItemLifecycleStatus::Completed,
329	summary: summary.to_string(),
330	detail: detail.map(str::to_string),
331	metadata,
332	artifact_refs: Vec::new(),
333	started_at: Some(now),
334	ended_at: Some(now),
335	}
336	};
337	let detail = ThreadDetail {
338	thread,
339	turns: vec![turn],
340	items: vec![
341	item(
342	"item_user",
343	TurnItemKind::UserMessage,
344	"check",
345	Some("check"),
346	None,
347	),
348	item(
349	"item_reasoning",
350	TurnItemKind::AgentReasoning,
351	"thinking",
352	Some("thinking"),
353	None,
354	),
355	item(
356	"item_tool_use",
357	TurnItemKind::ToolCall,
358	"shell",
359	Some(r#"{"cmd":"pwd"}"#),
360	Some(json!({
361	"tool_use_id": "tool-1",
362	"tool_name": "shell"
363	})),
364	),
365	item(
366	"item_result_one",
367	TurnItemKind::ToolCall,
368	"one",
369	Some("one"),
370	Some(json!({
371	"tool_result_for": "tool-1",
372	"is_error": false,
373	"content_blocks": [{
374	"type": "text",
375	"text": "structured one"
376	}]
377	})),
378	),
379	item(
380	"item_result_two",
381	TurnItemKind::ToolCall,
382	"two",
383	Some("two"),
384	Some(json!({
385	"tool_result_for": "tool-2",
386	"is_error": true
387	})),
388	),
389	item(
390	"item_answer",
391	TurnItemKind::AgentMessage,
392	"done",
393	Some("done"),
394	None,
395	),
396	],
397	latest_seq: 0,
398	pending_approvals: Vec::new(),
399	pending_user_inputs: Vec::new(),
400	pending_dynamic_tool_calls: Vec::new(),
401	};
402
403	let messages = messages_from_thread_detail(&detail);
404	let roles = messages
405	.iter()
406	.map(\|message\| message.role.as_str())
407	.collect::<Vec<_>>();
408	assert_eq!(roles, vec!["user", "assistant", "user", "assistant"]);
409	assert_eq!(messages[2].content.len(), 2);
410	match &messages[2].content[0] {
411	ContentBlock::ToolResult {
412	tool_use_id,
413	content,
414	is_error,
415	content_blocks,
416	} => {
417	assert_eq!(tool_use_id, "tool-1");
418	assert_eq!(content, "one");
419	assert_eq!(*is_error, None);
420	assert_eq!(
421	content_blocks
422	.as_ref()
423	.and_then(\|blocks\| blocks[0].get("text")),
424	Some(&json!("structured one"))
425	);
426	}
427	other => panic!("expected first tool result, got {other:?}"),
428	}
429	match &messages[2].content[1] {
430	ContentBlock::ToolResult {
431	tool_use_id,
432	content,
433	is_error,
434	content_blocks,
435	} => {
436	assert_eq!(tool_use_id, "tool-2");
437	assert_eq!(content, "two");
438	assert_eq!(*is_error, Some(true));
439	assert!(content_blocks.is_none());
440	}
441	other => panic!("expected second tool result, got {other:?}"),
442	}
443	}
444
445	#[test]
446	fn legacy_exact_thread_export_normalizes_provider_kind_and_id() {
447	let now = Utc::now();
448	let detail = ThreadDetail {
449	thread: ThreadRecord {
450	schema_version: 2,
451	id: "thr_legacy_custom".to_string(),
452	created_at: now,
453	updated_at: now,
454	model: "local-model".to_string(),
455	// Pre-additive records overloaded this legacy field with the exact id.
456	model_provider: Some("lm-studio".to_string()),
457	model_provider_id: None,
458	workspace: PathBuf::from("."),
459	mode: "agent".to_string(),
460	permission_posture: None,
461	allow_shell: false,
462	trust_mode: false,
463	auto_approve: false,
464	latest_turn_id: None,
465	latest_response_bookmark: None,
466	archived: false,
467	system_prompt: None,
468	task_id: None,
469	title: None,
470	session_id: None,
471	},
472	turns: Vec::new(),
473	items: Vec::new(),
474	latest_seq: 0,
475	pending_approvals: Vec::new(),
476	pending_user_inputs: Vec::new(),
477	pending_dynamic_tool_calls: Vec::new(),
478	};
479	let config = Config {
480	provider: Some("lm-studio".to_string()),
481	providers: Some(crate::config::ProvidersConfig {
482	custom: std::collections::HashMap::from([(
483	"lm-studio".to_string(),
484	crate::config::ProviderConfig {
485	kind: Some("openai-compatible".to_string()),
486	base_url: Some("http://127.0.0.1:1234/v1".to_string()),
487	model: Some("local-model".to_string()),
488	..Default::default()
489	},
490	)]),
491	..Default::default()
492	}),
493	..Default::default()
494	};
495	let mut session = crate::session_manager::create_saved_session_with_mode(
496	&[],
497	"local-model",
498	std::path::Path::new("."),
499	0,
500	None,
501	Some("agent"),
502	);
503
504	sessions::stamp_session_provider_from_thread(&config, &detail, &mut session.metadata)
505	.expect("normalize legacy exact provider");
506
507	assert_eq!(session.metadata.model_provider, "custom");
508	assert_eq!(
509	session.metadata.model_provider_id.as_deref(),
510	Some("lm-studio")
511	);
512	}
513
514	#[test]
515	fn runtime_auth_generates_token_by_default() {
516	let auth = resolve_runtime_auth(None, None, false);
517	assert!(auth.generated);
518	let token = auth.token.expect("generated token");
519	assert!(token.starts_with("cwrt_"));
520	assert!(token.len() > 32);
521	}
522
523	#[test]
524	fn runtime_auth_status_does_not_render_generated_token() {
525	let auth = ResolvedRuntimeAuth {
526	token: Some("cwrt_super_secret_test_token".to_string()),
527	generated: true,
528	};
529	let rendered = runtime_auth_status_lines(&auth).join("\n");
530
531	assert!(!rendered.contains("cwrt_super_secret_test_token"));
532	assert!(rendered.contains("not printed"));
533	}
534
535	#[test]
536	fn runtime_auth_requires_explicit_insecure_for_no_token() {
537	let auth = resolve_runtime_auth(None, None, true);
538	assert_eq!(
539	auth,
540	ResolvedRuntimeAuth {
541	token: None,
542	generated: false,
543	}
544	);
545	}
546
547	#[test]
548	fn runtime_auth_prefers_cli_token_over_env_token() {
549	let auth = resolve_runtime_auth(
550	Some(" cli-token ".to_string()),
551	Some("env-token".to_string()),
552	false,
553	);
554	assert_eq!(
555	auth,
556	ResolvedRuntimeAuth {
557	token: Some("cli-token".to_string()),
558	generated: false,
559	}
560	);
561	}
562
563	#[test]
564	fn runtime_auth_ignores_blank_configured_tokens() {
565	let auth = resolve_runtime_auth(Some(" ".to_string()), Some("\t".to_string()), false);
566	assert!(auth.generated);
567	assert!(auth.token.is_some());
568	}
569
570	#[test]
571	fn url_query_component_percent_encodes_token() {
572	assert_eq!(
573	url_query_component("abc ABC+/?:=&%"),
574	"abc%20ABC%2B%2F%3F%3A%3D%26%25"
575	);
576	}
577
578	#[test]
579	fn token_from_cookie_header_decodes_percent_encoded_token() {
580	assert_eq!(
581	token_from_cookie_header(Some(
582	"theme=dark; codewhale_runtime_token=abc%20ABC%2B%2F%3F%3A%3D%26%25"
583	)),
584	Some("abc ABC+/?:=&%".to_string())
585	);
586	assert_eq!(
587	token_from_cookie_header(Some("codewhale_runtime_token=bad%ZZ")),
588	None
589	);
590	}
591
592	async fn spawn_test_server_with_root(
593	root: PathBuf,
594	sessions_dir: PathBuf,
595	) -> Result<
596	Option<(
597	SocketAddr,
598	SharedRuntimeThreadManager,
599	tokio::task::JoinHandle<()>,
600	)>,
601	> {
602	spawn_test_server_with_root_and_token(root, sessions_dir, None).await
603	}
604
605	async fn spawn_test_server_with_root_and_token(
606	root: PathBuf,
607	sessions_dir: PathBuf,
608	runtime_token: Option<String>,
609	) -> Result<
610	Option<(
611	SocketAddr,
612	SharedRuntimeThreadManager,
613	tokio::task::JoinHandle<()>,
614	)>,
615	> {
616	spawn_test_server_with_root_token_and_mobile(root, sessions_dir, runtime_token, false).await
617	}
618
619	async fn spawn_test_server_with_root_token_and_mobile(
620	root: PathBuf,
621	sessions_dir: PathBuf,
622	runtime_token: Option<String>,
623	mobile_enabled: bool,
624	) -> Result<
625	Option<(
626	SocketAddr,
627	SharedRuntimeThreadManager,
628	tokio::task::JoinHandle<()>,
629	)>,
630	> {
631	spawn_test_server_with_root_token_mobile_workspace(
632	root,
633	sessions_dir,
634	runtime_token,
635	mobile_enabled,
636	PathBuf::from("."),
637	)
638	.await
639	}
640
641	async fn spawn_test_server_with_root_token_mobile_workspace(
642	root: PathBuf,
643	sessions_dir: PathBuf,
644	runtime_token: Option<String>,
645	mobile_enabled: bool,
646	workspace: PathBuf,
647	) -> Result<
648	Option<(
649	SocketAddr,
650	SharedRuntimeThreadManager,
651	tokio::task::JoinHandle<()>,
652	)>,
653	> {
654	spawn_test_server_with_root_token_mobile_workspace_and_subagents(
655	root,
656	sessions_dir,
657	runtime_token,
658	mobile_enabled,
659	workspace,
660	None,
661	None,
662	)
663	.await
664	}
665
666	#[derive(Default)]
667	struct TestServerOverrides {
668	sub_agent_manager: Option<SharedSubAgentManager>,
669	fleet_codewhale_binary: Option<String>,
670	config_path: Option<PathBuf>,
671	config_profile: Option<String>,
672	web: Option<web::RuntimeWebState>,
673	compat_stream_test_hook: Option<mpsc::UnboundedSender<CompatStreamTestPoint>>,
674	plugin_discovery: Option<Arc<crate::plugins::PluginDiscoveryContext>>,
675	}
676
677	async fn spawn_test_server_with_root_token_mobile_workspace_and_subagents(
678	root: PathBuf,
679	sessions_dir: PathBuf,
680	runtime_token: Option<String>,
681	mobile_enabled: bool,
682	workspace: PathBuf,
683	sub_agent_manager: Option<SharedSubAgentManager>,
684	fleet_codewhale_binary: Option<String>,
685	) -> Result<
686	Option<(
687	SocketAddr,
688	SharedRuntimeThreadManager,
689	tokio::task::JoinHandle<()>,
690	)>,
691	> {
692	spawn_test_server_with_root_token_mobile_workspace_and_overrides(
693	root,
694	sessions_dir,
695	runtime_token,
696	mobile_enabled,
697	workspace,
698	TestServerOverrides {
699	sub_agent_manager,
700	fleet_codewhale_binary,
701	..TestServerOverrides::default()
702	},
703	)
704	.await
705	}
706
707	async fn spawn_test_server_with_root_token_mobile_workspace_and_overrides(
708	root: PathBuf,
709	sessions_dir: PathBuf,
710	runtime_token: Option<String>,
711	mobile_enabled: bool,
712	workspace: PathBuf,
713	overrides: TestServerOverrides,
714	) -> Result<
715	Option<(
716	SocketAddr,
717	SharedRuntimeThreadManager,
718	tokio::task::JoinHandle<()>,
719	)>,
720	> {
721	let _ = rustls::crypto::ring::default_provider().install_default();
722	fs::create_dir_all(&sessions_dir)?;
723	fs::create_dir_all(&workspace)?;
724	let mut config = if let Some(path) = overrides.config_path.clone() {
725	Config::load(Some(path), None)?
726	} else {
727	Config {
728	api_key: Some("runtime-api-test-key".to_string()),
729	base_url: Some("http://127.0.0.1:1/v1".to_string()),
730	..Config::default()
731	}
732	};
733	config.mcp_config_path = Some(root.join("mcp.json").to_string_lossy().to_string());
734
735	config.mcp_config_path = Some(root.join("mcp.json").to_string_lossy().to_string());
736	let manager = TaskManager::start_with_executor(
737	TaskManagerConfig {
738	data_dir: root.join("tasks"),
739	worker_count: 1,
740	default_workspace: workspace.clone(),
741	default_model: DEFAULT_TEXT_MODEL.to_string(),
742	default_mode: "agent".to_string(),
743	allow_shell: false,
744	trust_mode: false,
745	},
746	Arc::new(MockExecutor),
747	)
748	.await?;
749	let runtime_threads: SharedRuntimeThreadManager = Arc::new(RuntimeThreadManager::open(
750	config.clone(),
751	workspace.clone(),
752	RuntimeThreadManagerConfig::from_task_data_dir(root.join("runtime")),
753	)?);
754	runtime_threads.attach_task_manager(manager.clone());
755	let automations = Arc::new(Mutex::new(AutomationManager::open(
756	root.join("automations"),
757	)?));
758	runtime_threads.attach_automation_manager(automations.clone());
759
760	let auth_required = runtime_token.is_some();
761	let sub_agent_manager = overrides
762	.sub_agent_manager
763	.unwrap_or_else(\|\| runtime_api_sub_agent_manager(&workspace, 2));
764	let listener = match TcpListener::bind("127.0.0.1:0").await {
765	Ok(listener) => listener,
766	Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return Ok(None),
767	Err(err) => return Err(err.into()),
768	};
769	let addr = listener.local_addr()?;
770	let state = RuntimeApiState {
771	config: Arc::new(parking_lot::RwLock::new(config)),
772	workspace,
773	plugin_discovery: overrides
774	.plugin_discovery
775	.unwrap_or_else(crate::plugins::PluginDiscoveryContext::capture_pre_dotenv),
776	task_manager: manager,
777	runtime_threads: runtime_threads.clone(),
778	cors_origins: Vec::new(),
779	sessions_dir,
780	config_path: overrides.config_path.clone(),
781	config_profile: overrides.config_profile,
782	mcp_pool: Arc::new(Mutex::new(None)),
783	automations,
784	sub_agent_manager,
785	runtime_token,
786	skill_state: Arc::new(Mutex::new(
787	SkillStateStore::load_from(root.join("skills_state.toml")).unwrap(),
788	)),
789	auth_required,
790	bind_host: "127.0.0.1".to_string(),
791	bind_port: addr.port(),
792	mobile_enabled,
793	web: overrides.web,
794	fleet_codewhale_binary: overrides
795	.fleet_codewhale_binary
796	.unwrap_or_else(configured_codewhale_binary),
797	compat_stream_test_hook: overrides.compat_stream_test_hook,
798	};
799	let app = build_router(state);
800	let handle = tokio::spawn(async move {
801	let _ = axum::serve(
802	listener,
803	app.into_make_service_with_connect_info::<SocketAddr>(),
804	)
805	.await;
806	});
807	Ok(Some((addr, runtime_threads, handle)))
808	}
809
810	async fn spawn_test_server() -> Result<
811	Option<(
812	SocketAddr,
813	SharedRuntimeThreadManager,
814	tokio::task::JoinHandle<()>,
815	)>,
816	> {
817	let root = std::env::temp_dir().join(format!("deepseek-runtime-api-{}", Uuid::new_v4()));
818	let sessions_dir = root.join("sessions");
819	spawn_test_server_with_root(root, sessions_dir).await
820	}
821
822	async fn spawn_test_server_with_config_path(
823	config_path: PathBuf,
824	) -> Result<
825	Option<(
826	SocketAddr,
827	SharedRuntimeThreadManager,
828	tokio::task::JoinHandle<()>,
829	)>,
830	> {
831	let root = std::env::temp_dir().join(format!("codewhale-config-api-{}", Uuid::new_v4()));
832	let sessions_dir = root.join("sessions");
833	let workspace = root.join("workspace");
834	fs::create_dir_all(&root)?;
835	spawn_test_server_with_root_token_mobile_workspace_and_overrides(
836	root,
837	sessions_dir,
838	None,
839	false,
840	workspace,
841	TestServerOverrides {
842	config_path: Some(config_path),
843	..TestServerOverrides::default()
844	},
845	)
846	.await
847	}
848
849	async fn spawn_test_server_with_config_path_and_profile(
850	config_path: PathBuf,
851	config_profile: String,
852	) -> Result<
853	Option<(
854	SocketAddr,
855	SharedRuntimeThreadManager,
856	tokio::task::JoinHandle<()>,
857	)>,
858	> {
859	let root = std::env::temp_dir().join(format!("codewhale-config-api-{}", Uuid::new_v4()));
860	let sessions_dir = root.join("sessions");
861	let workspace = root.join("workspace");
862	fs::create_dir_all(&root)?;
863	spawn_test_server_with_root_token_mobile_workspace_and_overrides(
864	root,
865	sessions_dir,
866	None,
867	false,
868	workspace,
869	TestServerOverrides {
870	config_path: Some(config_path),
871	config_profile: Some(config_profile),
872	..TestServerOverrides::default()
873	},
874	)
875	.await
876	}
877
878	async fn read_first_sse_frame(resp: reqwest::Response) -> Result<String> {
879	let mut stream = resp.bytes_stream();
880	let mut buf = Vec::new();
881	loop {
882	let next = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), stream.next())
883	.await
884	.context("timed out waiting for SSE frame")?
885	.context("SSE stream ended unexpectedly")??;
886	buf.extend_from_slice(&next);
887
888	let text = String::from_utf8_lossy(&buf);
889	if let Some(idx) = text.find("\n\n").or_else(\|\| text.find("\r\n\r\n")) {
890	return Ok(text[..idx].to_string());
891	}
892
893	if buf.len() > 64 * 1024 {
894	bail!("SSE frame exceeded 64KB without delimiter");
895	}
896	}
897	}
898
899	fn take_complete_sse_frame(buffer: &mut Vec<u8>) -> Result<Option<String>> {
900	let text = String::from_utf8_lossy(buffer);
901	let lf = text.find("\n\n").map(\|index\| (index, 2));
902	let crlf = text.find("\r\n\r\n").map(\|index\| (index, 4));
903	let delimiter = match (lf, crlf) {
904	(Some(left), Some(right)) => Some(if left.0 <= right.0 { left } else { right }),
905	(Some(found), None) \| (None, Some(found)) => Some(found),
906	(None, None) => None,
907	};
908	let Some((index, delimiter_len)) = delimiter else {
909	return Ok(None);
910	};
911	let frame = String::from_utf8(buffer[..index].to_vec())?;
912	buffer.drain(..index + delimiter_len);
913	Ok(Some(frame))
914	}
915
916	async fn collect_sse_frames(
917	response: reqwest::Response,
918	frame_tx: mpsc::UnboundedSender<(String, serde_json::Value)>,
919	) -> Result<Vec<(String, serde_json::Value)>> {
920	let mut stream = response.bytes_stream();
921	let mut buffer = Vec::new();
922	let mut frames = Vec::new();
923	while let Some(chunk) = stream.next().await {
924	buffer.extend_from_slice(&chunk?);
925	while let Some(raw) = take_complete_sse_frame(&mut buffer)? {
926	if raw.trim().is_empty() \|\| raw.trim_start().starts_with(':') {
927	continue;
928	}
929	let frame = parse_sse_frame(&raw)?;
930	frame_tx
931	.send(frame.clone())
932	.map_err(\|_\| anyhow::anyhow!("SSE frame observer closed"))?;
933	frames.push(frame);
934	}
935	if buffer.len() > 64 * 1024 {
936	bail!("SSE frame exceeded 64KB without delimiter");
937	}
938	}
939	Ok(frames)
940	}
941
942	#[cfg(unix)]
943	fn write_fake_fleet_binary(root: &Path, marker: &Path) -> Result<PathBuf> {
944	use std::os::unix::fs::PermissionsExt;
945
946	let binary = root.join("fake-codewhale");
947	fs::write(
948	&binary,
949	format!(
950	"#!/bin/sh\ntouch '{}'\nprintf '{{\"type\":\"content\",\"content\":\"restarted through Runtime API\"}}\\n'\nexit 0\n",
951	marker.display()
952	),
953	)?;
954	let mut permissions = fs::metadata(&binary)?.permissions();
955	permissions.set_mode(0o755);
956	fs::set_permissions(&binary, permissions)?;
957	Ok(binary)
958	}
959
960	#[cfg(windows)]
961	fn write_fake_fleet_binary(root: &Path, marker: &Path) -> Result<PathBuf> {
962	// Exercise the same executable/Job Object path as a released Windows
963	// Codewhale binary. A `.cmd` fake introduces an extra `cmd.exe` wrapper
964	// whose lifetime can end before the Fleet host attaches its Job Object,
965	// making the test race a process topology production does not use.
966	let source = root.join("fake-codewhale.rs");
967	let binary = root.join("fake-codewhale.exe");
968	let helper = format!(
969	r##"fn main() {{
970	std::fs::File::create({marker:?}).expect("create Fleet restart marker");
971	println!("{{}}", r#"{{"type":"content","content":"restarted through Runtime API"}}"#);
972	std::thread::sleep(std::time::Duration::from_millis(750));
973	}}
974	"##,
975	marker = marker.to_string_lossy().as_ref(),
976	);
977	fs::write(&source, helper)?;
978	let rustc = std::env::var_os("RUSTC").unwrap_or_else(\|\| "rustc".into());
979	let output = std::process::Command::new(rustc)
980	.arg("--edition=2024")
981	.arg("--crate-name=codewhale_fleet_test_helper")
982	.arg(&source)
983	.arg("-o")
984	.arg(&binary)
985	.output()
986	.context("compile Windows Fleet restart helper")?;
987	if !output.status.success() {
988	bail!(
989	"failed to compile Windows Fleet restart helper: {}",
990	String::from_utf8_lossy(&output.stderr)
991	);
992	}
993	Ok(binary)
994	}
995
996	fn parse_sse_frame(frame: &str) -> Result<(String, serde_json::Value)> {
997	let mut event_name: Option<String> = None;
998	let mut data_lines = Vec::new();
999	for line in frame.lines() {
1000	if let Some(rest) = line.strip_prefix("event:") {
1001	event_name = Some(rest.trim().to_string());
1002	} else if let Some(rest) = line.strip_prefix("data:") {
1003	data_lines.push(rest.trim_start().to_string());
1004	}
1005	}
1006	let event_name = event_name.context("missing SSE event field")?;
1007	let payload = if data_lines.is_empty() {
1008	json!({})
1009	} else {
1010	serde_json::from_str(&data_lines.join("\n"))
1011	.with_context(\|\| format!("invalid SSE data payload: {}", data_lines.join("\n")))?
1012	};
1013	Ok((event_name, payload))
1014	}
1015
1016	async fn wait_for_terminal_turn_status(
1017	client: &reqwest::Client,
1018	addr: SocketAddr,
1019	thread_id: &str,
1020	turn_id: &str,
1021	timeout: Duration,
1022	) -> Result<String> {
1023	let deadline = tokio::time::Instant::now() + ci_scaled(timeout);
1024	loop {
1025	let detail: serde_json::Value = client
1026	.get(format!("http://{addr}/v1/threads/{thread_id}"))
1027	.send()
1028	.await?
1029	.error_for_status()?
1030	.json()
1031	.await?;
1032	let status = detail["turns"]
1033	.as_array()
1034	.and_then(\|turns\| turns.iter().find(\|turn\| turn["id"] == turn_id))
1035	.and_then(\|turn\| turn.get("status"))
1036	.and_then(Value::as_str)
1037	.unwrap_or_default()
1038	.to_string();
1039	if matches!(
1040	status.as_str(),
1041	"completed" \| "failed" \| "interrupted" \| "canceled"
1042	) {
1043	return Ok(status);
1044	}
1045	if tokio::time::Instant::now() >= deadline {
1046	bail!("timed out waiting for terminal turn status for {turn_id}");
1047	}
1048	sleep(Duration::from_millis(25)).await;
1049	}
1050	}
1051
1052	async fn wait_for_in_progress_item(
1053	client: &reqwest::Client,
1054	addr: SocketAddr,
1055	thread_id: &str,
1056	timeout: Duration,
1057	) -> Result<()> {
1058	let deadline = tokio::time::Instant::now() + ci_scaled(timeout);
1059	loop {
1060	let detail: serde_json::Value = client
1061	.get(format!("http://{addr}/v1/threads/{thread_id}"))
1062	.send()
1063	.await?
1064	.error_for_status()?
1065	.json()
1066	.await?;
1067	if detail["items"]
1068	.as_array()
1069	.is_some_and(\|items\| items.iter().any(\|item\| item["status"] == "in_progress"))
1070	{
1071	return Ok(());
1072	}
1073	if tokio::time::Instant::now() >= deadline {
1074	bail!("timed out waiting for in-progress item in thread {thread_id}");
1075	}
1076	sleep(Duration::from_millis(25)).await;
1077	}
1078	}
1079
1080	#[tokio::test]
1081	async fn health_and_tasks_endpoints_work() -> Result<()> {
1082	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
1083	return Ok(());
1084	};
1085	let client = crate::tls::reqwest_client();
1086
1087	let health: serde_json::Value = client
1088	.get(format!("http://{addr}/health"))
1089	.send()
1090	.await?
1091	.error_for_status()?
1092	.json()
1093	.await?;
1094	assert_eq!(health["status"], "ok");
1095	assert_eq!(health["service"], "codewhale-runtime-api");
1096
1097	let created: serde_json::Value = client
1098	.post(format!("http://{addr}/v1/tasks"))
1099	.json(&json!({ "prompt": "hello task" }))
1100	.send()
1101	.await?
1102	.error_for_status()?
1103	.json()
1104	.await?;
1105	let id = created["id"].as_str().expect("task id").to_string();
1106
1107	let listed: serde_json::Value = client
1108	.get(format!("http://{addr}/v1/tasks"))
1109	.send()
1110	.await?
1111	.error_for_status()?
1112	.json()
1113	.await?;
1114	assert!(
1115	listed["tasks"]
1116	.as_array()
1117	.is_some_and(\|tasks\| !tasks.is_empty())
1118	);
1119
1120	let detail: serde_json::Value = client
1121	.get(format!("http://{addr}/v1/tasks/{id}"))
1122	.send()
1123	.await?
1124	.error_for_status()?
1125	.json()
1126	.await?;
1127	assert_eq!(detail["id"], id);
1128
1129	let _cancelled: serde_json::Value = client
1130	.post(format!("http://{addr}/v1/tasks/{id}/cancel"))
1131	.send()
1132	.await?
1133	.error_for_status()?
1134	.json()
1135	.await?;
1136
1137	handle.abort();
1138	Ok(())
1139	}
1140
1141	#[cfg(unix)]
1142	#[tokio::test]
1143	async fn mcp_tools_endpoint_is_passive_until_connect_requested() -> Result<()> {
1144	let root = std::env::temp_dir().join(format!("codewhale-mcp-tools-api-{}", Uuid::new_v4()));
1145	let sessions_dir = root.join("sessions");
1146	fs::create_dir_all(&root)?;
1147	let sentinel = root.join("mcp-spawned");
1148	fs::write(
1149	root.join("mcp.json"),
1150	serde_json::json!({
1151	"servers": {
1152	"sentinel": {
1153	"command": "sh",
1154	"args": [
1155	"-c",
1156	"printf spawned > \"$1\"",
1157	"sh",
1158	sentinel
1159	]
1160	}
1161	}
1162	})
1163	.to_string(),
1164	)?;
1165
1166	let Some((addr, _runtime_threads, handle)) =
1167	spawn_test_server_with_root(root.clone(), sessions_dir).await?
1168	else {
1169	return Ok(());
1170	};
1171	let client = crate::tls::reqwest_client();
1172
1173	let passive: serde_json::Value = client
1174	.get(format!("http://{addr}/v1/apps/mcp/tools"))
1175	.send()
1176	.await?
1177	.error_for_status()?
1178	.json()
1179	.await?;
1180	assert_eq!(passive["tools"].as_array().map(Vec::len), Some(0));
1181	assert!(
1182	!sentinel.exists(),
1183	"passive MCP tool listing must not spawn stdio servers"
1184	);
1185
1186	let _live: serde_json::Value = client
1187	.get(format!("http://{addr}/v1/apps/mcp/tools?connect=true"))
1188	.send()
1189	.await?
1190	.error_for_status()?
1191	.json()
1192	.await?;
1193
1194	for _ in 0..20 {
1195	if sentinel.exists() {
1196	break;
1197	}
1198	tokio::time::sleep(Duration::from_millis(25)).await;
1199	}
1200	assert!(
1201	sentinel.exists(),
1202	"explicit MCP connect should spawn configured stdio servers"
1203	);
1204
1205	handle.abort();
1206	Ok(())
1207	}
1208
1209	#[tokio::test]
1210	async fn runtime_token_guard_protects_v1_routes() -> Result<()> {
1211	let root = std::env::temp_dir().join(format!("deepseek-runtime-api-{}", Uuid::new_v4()));
1212	let sessions_dir = root.join("sessions");
1213	let token = "local-test-token".to_string();
1214	let Some((addr, _runtime_threads, handle)) =
1215	spawn_test_server_with_root_and_token(root, sessions_dir, Some(token.clone())).await?
1216	else {
1217	return Ok(());
1218	};
1219	let client = crate::tls::reqwest_client();
1220
1221	let health = client
1222	.get(format!("http://{addr}/health"))
1223	.send()
1224	.await?
1225	.error_for_status()?;
1226	assert_eq!(health.status(), StatusCode::OK);
1227
1228	let unauthorized = client
1229	.get(format!("http://{addr}/v1/threads/summary"))
1230	.send()
1231	.await?;
1232	assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED);
1233
1234	let bearer = client
1235	.get(format!("http://{addr}/v1/threads/summary"))
1236	.bearer_auth(&token)
1237	.send()
1238	.await?
1239	.error_for_status()?;
1240	assert_eq!(bearer.status(), StatusCode::OK);
1241
1242	let query_token = client
1243	.get(format!("http://{addr}/v1/threads/summary?token={token}"))
1244	.send()
1245	.await?;
1246	assert_eq!(query_token.status(), StatusCode::UNAUTHORIZED);
1247
1248	let cookie_token = client
1249	.get(format!("http://{addr}/v1/threads/summary"))
1250	.header(
1251	header::COOKIE,
1252	format!("codewhale_runtime_token={}", url_query_component(&token)),
1253	)
1254	.send()
1255	.await?
1256	.error_for_status()?;
1257	assert_eq!(cookie_token.status(), StatusCode::OK);
1258
1259	let codewhale_header = client
1260	.get(format!("http://{addr}/v1/threads/summary"))
1261	.header("x-codewhale-runtime-token", &token)
1262	.send()
1263	.await?
1264	.error_for_status()?;
1265	assert_eq!(codewhale_header.status(), StatusCode::OK);
1266
1267	let deepseek_header = client
1268	.get(format!("http://{addr}/v1/threads/summary"))
1269	.header("x-deepseek-runtime-token", &token)
1270	.send()
1271	.await?
1272	.error_for_status()?;
1273	assert_eq!(deepseek_header.status(), StatusCode::OK);
1274
1275	handle.abort();
1276	Ok(())
1277	}
1278
1279	#[tokio::test]
1280	async fn web_bootstrap_sets_strict_cookie_once_and_preserves_v1_auth() -> Result<()> {
1281	let root = std::env::temp_dir().join(format!("codewhale-web-api-{}", Uuid::new_v4()));
1282	let sessions_dir = root.join("sessions");
1283	let workspace = root.join("workspace");
1284	let token = "cwrt_runtime_secret_never_in_browser_storage".to_string();
1285	let (web, nonce) = web::RuntimeWebState::new();
1286	let Some((addr, _runtime_threads, handle)) =
1287	spawn_test_server_with_root_token_mobile_workspace_and_overrides(
1288	root,
1289	sessions_dir,
1290	Some(token.clone()),
1291	false,
1292	workspace,
1293	TestServerOverrides {
1294	web: Some(web),
1295	..TestServerOverrides::default()
1296	},
1297	)
1298	.await?
1299	else {
1300	return Ok(());
1301	};
1302	let client = crate::tls::reqwest_client_builder()
1303	.redirect(reqwest::redirect::Policy::none())
1304	.build()?;
1305
1306	let page = client.get(format!("http://{addr}/")).send().await?;
1307	assert_eq!(page.status(), StatusCode::OK);
1308	assert_eq!(
1309	page.headers()
1310	.get(header::CONTENT_SECURITY_POLICY)
1311	.and_then(\|value\| value.to_str().ok()),
1312	Some(
1313	"default-src 'none'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self'; base-uri 'none'; form-action 'self'; frame-ancestors 'none'; object-src 'none'"
1314	)
1315	);
1316	let page_body = page.text().await?;
1317	assert!(!page_body.contains(&token));
1318	assert!(!page_body.contains(&nonce));
1319
1320	let wrong = client
1321	.get(format!(
1322	"http://{addr}/__codewhale/bootstrap/cwwb_00000000000000000000000000000000"
1323	))
1324	.send()
1325	.await?;
1326	assert_eq!(wrong.status(), StatusCode::UNAUTHORIZED);
1327
1328	let exchange = client
1329	.get(format!("http://{addr}/__codewhale/bootstrap/{nonce}"))
1330	.send()
1331	.await?;
1332	assert_eq!(exchange.status(), StatusCode::SEE_OTHER);
1333	assert_eq!(
1334	exchange
1335	.headers()
1336	.get(header::LOCATION)
1337	.and_then(\|value\| value.to_str().ok()),
1338	Some("/")
1339	);
1340	let set_cookie = exchange
1341	.headers()
1342	.get(header::SET_COOKIE)
1343	.and_then(\|value\| value.to_str().ok())
1344	.context("missing bootstrap Set-Cookie")?
1345	.to_string();
1346	assert!(set_cookie.starts_with("codewhale_web_session=cwws_"));
1347	assert!(set_cookie.ends_with("; HttpOnly; SameSite=Strict; Path=/"));
1348	assert!(!set_cookie.contains(&token));
1349
1350	let unauthorized = client
1351	.get(format!("http://{addr}/v1/threads/summary"))
1352	.send()
1353	.await?;
1354	assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED);
1355
1356	let cookie_pair = set_cookie
1357	.split(';')
1358	.next()
1359	.context("missing web session cookie pair")?;
1360	let authorized = client
1361	.get(format!("http://{addr}/v1/threads/summary"))
1362	.header(header::COOKIE, cookie_pair)
1363	.send()
1364	.await?;
1365	assert_eq!(authorized.status(), StatusCode::OK);
1366
1367	let same_origin_cookie_post = client
1368	.post(format!("http://{addr}/v1/threads"))
1369	.header(header::COOKIE, cookie_pair)
1370	.header(header::ORIGIN, format!("http://{addr}"))
1371	.header("sec-fetch-site", "same-origin")
1372	.json(&json!({}))
1373	.send()
1374	.await?;
1375	assert_eq!(same_origin_cookie_post.status(), StatusCode::CREATED);
1376
1377	let cross_origin_cookie_post = client
1378	.post(format!("http://{addr}/v1/threads"))
1379	.header(header::COOKIE, cookie_pair)
1380	.header(header::ORIGIN, "http://127.0.0.1:3000")
1381	.header("sec-fetch-site", "same-site")
1382	.json(&json!({}))
1383	.send()
1384	.await?;
1385	assert_eq!(cross_origin_cookie_post.status(), StatusCode::UNAUTHORIZED);
1386
1387	let originless_cookie_post = client
1388	.post(format!("http://{addr}/v1/threads"))
1389	.header(header::COOKIE, cookie_pair)
1390	.json(&json!({}))
1391	.send()
1392	.await?;
1393	assert_eq!(originless_cookie_post.status(), StatusCode::UNAUTHORIZED);
1394
1395	let bearer_post = client
1396	.post(format!("http://{addr}/v1/threads"))
1397	.bearer_auth(&token)
1398	.header(header::ORIGIN, "http://127.0.0.1:3000")
1399	.json(&json!({}))
1400	.send()
1401	.await?;
1402	assert_eq!(bearer_post.status(), StatusCode::CREATED);
1403
1404	let reused = client
1405	.get(format!("http://{addr}/__codewhale/bootstrap/{nonce}"))
1406	.send()
1407	.await?;
1408	assert_eq!(reused.status(), StatusCode::UNAUTHORIZED);
1409
1410	let mobile = client.get(format!("http://{addr}/mobile")).send().await?;
1411	assert_eq!(mobile.status(), StatusCode::NOT_FOUND);
1412
1413	handle.abort();
1414	Ok(())
1415	}
1416
1417	#[tokio::test]
1418	async fn web_assets_are_absent_outside_web_mode() -> Result<()> {
1419	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
1420	return Ok(());
1421	};
1422	let client = crate::tls::reqwest_client();
1423	for path in ["/", "/assets/codewhale-web.css", "/assets/codewhale-web.js"] {
1424	let response = client.get(format!("http://{addr}{path}")).send().await?;
1425	assert_eq!(response.status(), StatusCode::NOT_FOUND, "path={path}");
1426	}
1427	handle.abort();
1428	Ok(())
1429	}
1430
1431	#[tokio::test]
1432	async fn thread_summary_includes_workspace_branch_metadata() -> Result<()> {
1433	let tmp = tempfile::tempdir()?;
1434	let root = tmp.path().join("runtime");
1435	let sessions_dir = root.join("sessions");
1436	let repo = tmp.path().join("repo");
1437	fs::create_dir_all(&repo)?;
1438	run_test_git(&repo, &["init", "-b", "feature/agent"])?;
1439	run_test_git(&repo, &["config", "core.autocrlf", "false"])?;
1440	fs::write(repo.join("README.md"), "branch visibility\n")?;
1441	run_test_git(&repo, &["add", "README.md"])?;
1442	run_test_git(
1443	&repo,
1444	&[
1445	"-c",
1446	"user.name=CodeWhale Test",
1447	"-c",
1448	"user.email=codewhale@example.invalid",
1449	"commit",
1450	"-m",
1451	"init",
1452	],
1453	)?;
1454
1455	let non_git = tmp.path().join("non-git");
1456	fs::create_dir_all(&non_git)?;
1457
1458	let Some((addr, _runtime_threads, handle)) =
1459	spawn_test_server_with_root(root, sessions_dir).await?
1460	else {
1461	return Ok(());
1462	};
1463	let client = crate::tls::reqwest_client();
1464
1465	let git_thread: serde_json::Value = client
1466	.post(format!("http://{addr}/v1/threads"))
1467	.json(&json!({
1468	"title": "Git workspace",
1469	"workspace": repo,
1470	}))
1471	.send()
1472	.await?
1473	.error_for_status()?
1474	.json()
1475	.await?;
1476	let git_thread_id = git_thread["id"]
1477	.as_str()
1478	.context("missing git thread id")?
1479	.to_string();
1480	fs::write(
1481	repo.join("dirty.txt"),
1482	"worktree changed after thread spawn\n",
1483	)?;
1484
1485	let plain_thread: serde_json::Value = client
1486	.post(format!("http://{addr}/v1/threads"))
1487	.json(&json!({
1488	"title": "Plain workspace",
1489	"workspace": non_git,
1490	}))
1491	.send()
1492	.await?
1493	.error_for_status()?
1494	.json()
1495	.await?;
1496	let plain_thread_id = plain_thread["id"]
1497	.as_str()
1498	.context("missing plain thread id")?
1499	.to_string();
1500
1501	let summary: serde_json::Value = client
1502	.get(format!("http://{addr}/v1/threads/summary?limit=100"))
1503	.send()
1504	.await?
1505	.error_for_status()?
1506	.json()
1507	.await?;
1508	let summaries = summary.as_array().context("summary should be an array")?;
1509	let git_summary = summaries
1510	.iter()
1511	.find(\|item\| item["id"] == git_thread_id)
1512	.context("missing git workspace summary")?;
1513	assert_eq!(git_summary["branch"], "feature/agent");
1514	assert!(
1515	git_summary["head"]
1516	.as_str()
1517	.is_some_and(\|head\| !head.is_empty())
1518	);
1519	assert_eq!(git_summary["dirty"], true);
1520	assert_eq!(git_summary["workspace"], repo.to_string_lossy().as_ref());
1521
1522	let plain_summary = summaries
1523	.iter()
1524	.find(\|item\| item["id"] == plain_thread_id)
1525	.context("missing plain workspace summary")?;
1526	assert_eq!(plain_summary["branch"], serde_json::Value::Null);
1527	assert_eq!(plain_summary["head"], serde_json::Value::Null);
1528	assert_eq!(plain_summary["dirty"], false);
1529	assert_eq!(
1530	plain_summary["workspace"],
1531	non_git.to_string_lossy().as_ref()
1532	);
1533
1534	handle.abort();
1535	Ok(())
1536	}
1537
1538	#[tokio::test]
1539	async fn workspace_and_automation_endpoints_work() -> Result<()> {
1540	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
1541	return Ok(());
1542	};
1543	let client = crate::tls::reqwest_client();
1544
1545	let workspace: serde_json::Value = client
1546	.get(format!("http://{addr}/v1/workspace/status"))
1547	.send()
1548	.await?
1549	.error_for_status()?
1550	.json()
1551	.await?;
1552	assert!(workspace.get("workspace").is_some());
1553
1554	let created: serde_json::Value = client
1555	.post(format!("http://{addr}/v1/automations"))
1556	.json(&json!({
1557	"name": "Smoke automation",
1558	"prompt": "automation smoke test",
1559	"rrule": "FREQ=HOURLY;INTERVAL=2",
1560	"status": "active"
1561	}))
1562	.send()
1563	.await?
1564	.error_for_status()?
1565	.json()
1566	.await?;
1567	let automation_id = created["id"]
1568	.as_str()
1569	.context("missing automation id")?
1570	.to_string();
1571
1572	let listed: serde_json::Value = client
1573	.get(format!("http://{addr}/v1/automations"))
1574	.send()
1575	.await?
1576	.error_for_status()?
1577	.json()
1578	.await?;
1579	assert!(
1580	listed
1581	.as_array()
1582	.is_some_and(\|items\| items.iter().any(\|item\| item["id"] == automation_id))
1583	);
1584
1585	let run_now: serde_json::Value = client
1586	.post(format!("http://{addr}/v1/automations/{automation_id}/run"))
1587	.send()
1588	.await?
1589	.error_for_status()?
1590	.json()
1591	.await?;
1592	assert_eq!(run_now["automation_id"], automation_id);
1593
1594	let paused: serde_json::Value = client
1595	.post(format!(
1596	"http://{addr}/v1/automations/{automation_id}/pause"
1597	))
1598	.send()
1599	.await?
1600	.error_for_status()?
1601	.json()
1602	.await?;
1603	assert_eq!(paused["status"], "paused");
1604
1605	let resumed: serde_json::Value = client
1606	.post(format!(
1607	"http://{addr}/v1/automations/{automation_id}/resume"
1608	))
1609	.send()
1610	.await?
1611	.error_for_status()?
1612	.json()
1613	.await?;
1614	assert_eq!(resumed["status"], "active");
1615
1616	let updated: serde_json::Value = client
1617	.patch(format!("http://{addr}/v1/automations/{automation_id}"))
1618	.json(&json!({
1619	"name": "Smoke automation edited",
1620	"rrule": "FREQ=WEEKLY;BYDAY=MO,WE;BYHOUR=10;BYMINUTE=15"
1621	}))
1622	.send()
1623	.await?
1624	.error_for_status()?
1625	.json()
1626	.await?;
1627	assert_eq!(updated["name"], "Smoke automation edited");
1628
1629	let runs: serde_json::Value = client
1630	.get(format!(
1631	"http://{addr}/v1/automations/{automation_id}/runs?limit=5"
1632	))
1633	.send()
1634	.await?
1635	.error_for_status()?
1636	.json()
1637	.await?;
1638	assert!(
1639	runs.as_array().is_some_and(\|items\| !items.is_empty()),
1640	"expected at least one run entry"
1641	);
1642
1643	let _deleted: serde_json::Value = client
1644	.delete(format!("http://{addr}/v1/automations/{automation_id}"))
1645	.send()
1646	.await?
1647	.error_for_status()?
1648	.json()
1649	.await?;
1650
1651	let missing_status = client
1652	.get(format!("http://{addr}/v1/automations/{automation_id}"))
1653	.send()
1654	.await?
1655	.status();
1656	assert_eq!(missing_status, StatusCode::NOT_FOUND);
1657
1658	handle.abort();
1659	Ok(())
1660	}
1661
1662	#[tokio::test]
1663	async fn fleet_status_runtime_api_exposes_state_and_actions() -> Result<()> {
1664	let root = std::env::temp_dir().join(format!("codewhale-fleet-api-{}", Uuid::new_v4()));
1665	let workspace = root.join("workspace");
1666	fs::create_dir_all(&workspace)?;
1667	let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, 2);
1668	let manager = FleetManager::open(&workspace)?
1669	.with_sub_agent_manager(sub_agent_manager.clone())
1670	.with_session_model(DEFAULT_TEXT_MODEL);
1671	let task = codewhale_protocol::fleet::FleetTaskSpec {
1672	id: "task-a".to_string(),
1673	name: "Task A".to_string(),
1674	description: None,
1675	objective: Some("Inspect fleet status through Runtime API".to_string()),
1676	instructions: "Stay running for inspection.".to_string(),
1677	worker: Some(codewhale_protocol::fleet::FleetTaskWorkerProfile {
1678	agent_profile: None,
1679	role: Some("reviewer".to_string()),
1680	loadout: None,
1681	model_class: None,
1682	model: None,
1683	tool_profile: Some("read-only".to_string()),
1684	tools: vec!["rg".to_string()],
1685	capabilities: vec!["fleet".to_string()],
1686	}),
1687	workspace: None,
1688	input_files: Vec::new(),
1689	context: Vec::new(),
1690	budget: None,
1691	tags: Vec::new(),
1692	expected_artifacts: vec![FleetArtifactKind::Log],
1693	scorer: None,
1694	retry_policy: None,
1695	alert_policy: None,
1696	timeout_seconds: None,
1697	metadata: std::collections::BTreeMap::new(),
1698	};
1699	let report = manager.create_run(
1700	crate::fleet::task_spec::FleetTaskSpecDocument {
1701	name: Some("api smoke".to_string()),
1702	labels: std::collections::BTreeMap::new(),
1703	security_policy: None,
1704	workers: Vec::new(),
1705	tasks: vec![task],
1706	},
1707	1,
1708	)?;
1709	let restarted_marker = root.join("restarted-worker-ran");
1710	let fake_codewhale = write_fake_fleet_binary(&root, &restarted_marker)?;
1711	let worker_id = report.worker_ids[0].clone();
1712	let sessions_dir = root.join("sessions");
1713	let Some((addr, _runtime_threads, handle)) =
1714	spawn_test_server_with_root_token_mobile_workspace_and_subagents(
1715	root.clone(),
1716	sessions_dir,
1717	None,
1718	false,
1719	workspace,
1720	Some(sub_agent_manager),
1721	Some(fake_codewhale.display().to_string()),
1722	)
1723	.await?
1724	else {
1725	return Ok(());
1726	};
1727	let client = crate::tls::reqwest_client();
1728
1729	let runs: serde_json::Value = client
1730	.get(format!("http://{addr}/v1/fleet/runs"))
1731	.send()
1732	.await?
1733	.error_for_status()?
1734	.json()
1735	.await?;
1736	assert_eq!(runs["status"]["running"], 1);
1737	assert_eq!(runs["runs"][0]["id"], report.run_id.0);
1738
1739	let worker: serde_json::Value = client
1740	.get(format!("http://{addr}/v1/fleet/workers/{worker_id}"))
1741	.send()
1742	.await?
1743	.error_for_status()?
1744	.json()
1745	.await?;
1746	assert_eq!(
1747	worker["objective"],
1748	"Inspect fleet status through Runtime API"
1749	);
1750	assert_eq!(worker["role"], "reviewer");
1751	assert_eq!(worker["host"], "local");
1752	assert_eq!(worker["artifacts"][0]["kind"], "log");
1753	assert_eq!(worker["runtime_state"]["agent_status"], "starting");
1754	assert_eq!(worker["runtime_state"]["steps_taken"], 0);
1755	assert_eq!(worker["runtime_state"]["has_session"], true);
1756
1757	let interrupted: serde_json::Value = client
1758	.post(format!(
1759	"http://{addr}/v1/fleet/workers/{worker_id}/interrupt"
1760	))
1761	.send()
1762	.await?
1763	.error_for_status()?
1764	.json()
1765	.await?;
1766	assert_eq!(interrupted["action"], "interrupt");
1767	assert_eq!(interrupted["worker"]["last_error"], "cancelled by operator");
1768
1769	let restarted: serde_json::Value = client
1770	.post(format!(
1771	"http://{addr}/v1/fleet/workers/{worker_id}/restart"
1772	))
1773	.send()
1774	.await?
1775	.error_for_status()?
1776	.json()
1777	.await?;
1778	assert_eq!(restarted["action"], "restart");
1779	assert_eq!(restarted["execution"], "scheduled");
1780	assert_eq!(restarted["worker"]["status"], "busy");
1781
1782	let terminal_status = tokio::time::timeout(ci_scaled(Duration::from_secs(15)), async {
1783	loop {
1784	let status = manager.run_status(&report.run_id).unwrap();
1785	if status.queued == 0 && status.running == 0 {
1786	break status;
1787	}
1788	tokio::time::sleep(Duration::from_millis(20)).await;
1789	}
1790	})
1791	.await
1792	.context("Runtime API restart never drove the replacement attempt to completion")?;
1793	assert_eq!(
1794	terminal_status.completed, 1,
1795	"replacement attempt did not complete successfully: {terminal_status:?}"
1796	);
1797	assert_eq!(
1798	terminal_status.failed, 0,
1799	"replacement attempt failed: {terminal_status:?}"
1800	);
1801	assert!(
1802	restarted_marker.is_file(),
1803	"Runtime API reported a restart without launching its Fleet worker"
1804	);
1805	let ledger_state = manager.rebuild_state()?;
1806	let restarted_task = ledger_state
1807	.tasks
1808	.values()
1809	.find(\|task\| task.entry.run_id == report.run_id && task.entry.task_id == "task-a")
1810	.context("missing restarted task")?;
1811	assert_eq!(restarted_task.entry.attempts, 2);
1812	assert_eq!(restarted_task.status, FleetTaskLedgerStatus::Completed);
1813	let receipt = ledger_state
1814	.receipts
1815	.values()
1816	.find(\|receipt\| receipt.run_id == report.run_id && receipt.task_id == "task-a")
1817	.context("missing restarted receipt")?;
1818	assert_eq!(receipt.attempt, Some(2));
1819	assert!(receipt.terminal_seq.is_some());
1820
1821	let stopped: serde_json::Value = client
1822	.post(format!(
1823	"http://{addr}/v1/fleet/runs/{}/stop",
1824	report.run_id.0
1825	))
1826	.send()
1827	.await?
1828	.error_for_status()?
1829	.json()
1830	.await?;
1831	assert_eq!(stopped["action"], "stop");
1832	assert_eq!(stopped["stopped"], 0);
1833	assert_eq!(stopped["status"]["completed"], 1);
1834
1835	handle.abort();
1836	Ok(())
1837	}
1838
1839	#[test]
1840	fn fleet_worker_json_includes_runtime_state_projection() {
1841	let inspection = FleetWorkerInspection {
1842	worker_id: "fleet-worker-1".to_string(),
1843	status: FleetWorkerStatus::Busy,
1844	current_run_id: Some(FleetRunId::from("fleet-run-1")),
1845	current_task_id: Some("task-a".to_string()),
1846	objective: Some("Inspect runtime projection".to_string()),
1847	role: Some("reviewer".to_string()),
1848	host: Some("local".to_string()),
1849	latest_heartbeat_at: None,
1850	latest_event: None,
1851	artifacts: Vec::new(),
1852	receipt_summary: None,
1853	last_error: None,
1854	alert_state: None,
1855	runtime_state: Some(FleetWorkerRuntimeProjection {
1856	agent_status: "running".to_string(),
1857	steps_taken: 3,
1858	latest_message: Some("reading files".to_string()),
1859	error: None,
1860	result_summary: None,
1861	has_session: true,
1862	}),
1863	};
1864
1865	let worker = fleet_worker_json(&inspection);
1866
1867	assert_eq!(worker["runtime_state"]["agent_status"], "running");
1868	assert_eq!(worker["runtime_state"]["steps_taken"], 3);
1869	assert_eq!(worker["runtime_state"]["latest_message"], "reading files");
1870	assert_eq!(worker["runtime_state"]["has_session"], true);
1871	}
1872
1873	#[tokio::test]
1874	async fn agent_runs_runtime_api_exposes_persisted_worker_receipts() -> Result<()> {
1875	use crate::tools::subagent::{
1876	AgentRunArtifactRef, AgentRunFollowUpTarget, AgentRunRecommendedAction,
1877	AgentRunTakeoverTarget, AgentRunUsage, AgentRunVerificationSummary, AgentWorkerEvent,
1878	AgentWorkerRecord, AgentWorkerSpec, AgentWorkerStatus, AgentWorkerToolProfile, FleetRole,
1879	};
1880	use crate::worker_profile::{ModelRoute, ToolScope, WorkerRuntimeProfile};
1881	use std::collections::VecDeque;
1882
1883	let root = std::env::temp_dir().join(format!("codewhale-agent-runs-api-{}", Uuid::new_v4()));
1884	let workspace = root.join("workspace");
1885	fs::create_dir_all(workspace.join(".codewhale/state"))?;
1886
1887	let record = AgentWorkerRecord {
1888	spec: AgentWorkerSpec {
1889	worker_id: "agent_receipt".to_string(),
1890	run_id: "run_receipt".to_string(),
1891	parent_run_id: Some("parent_run".to_string()),
1892	session_name: Some("receipt_lane".to_string()),
1893	objective: "Verify run receipt projection".to_string(),
1894	role: Some("verifier".to_string()),
1895	agent_type: FleetRole::Verifier,
1896	model: "deepseek-v4-flash".to_string(),
1897	workspace: workspace.clone(),
1898	git_branch: Some("codex/v0.8.60".to_string()),
1899	context_mode: "fresh".to_string(),
1900	fork_context: false,
1901	tool_profile: AgentWorkerToolProfile::Explicit(vec!["read_file".to_string()]),
1902	runtime_profile: {
1903	let mut profile = WorkerRuntimeProfile::for_role(FleetRole::Verifier);
1904	profile.tools = ToolScope::Explicit(vec!["read_file".to_string()]);
1905	profile.model = ModelRoute::Fixed("deepseek-v4-flash".to_string());
1906	profile.max_spawn_depth =
1907	crate::tools::subagent::DEFAULT_MAX_SPAWN_DEPTH.saturating_sub(1);
1908	profile
1909	},
1910	max_steps: 4,
1911	spawn_depth: 1,
1912	max_spawn_depth: crate::tools::subagent::DEFAULT_MAX_SPAWN_DEPTH,
1913	launch_manifest: None,
1914	},
1915	actor_kind: "subagent".to_string(),
1916	parent_run_id: Some("parent_run".to_string()),
1917	follow_up: AgentRunFollowUpTarget {
1918	tool: "handle_read".to_string(),
1919	agent_id: "agent_receipt".to_string(),
1920	session_name: Some("receipt_lane".to_string()),
1921	accepted_statuses: vec!["running".to_string(), "interrupted_continuable".to_string()],
1922	latest_delivery: None,
1923	},
1924	takeover: AgentRunTakeoverTarget {
1925	kind: "local_subagent_session".to_string(),
1926	supported: true,
1927	agent_id: "agent_receipt".to_string(),
1928	session_name: Some("receipt_lane".to_string()),
1929	instructions: "Use handle_read on the transcript_handle for agent_receipt.".to_string(),
1930	unsupported_reason: None,
1931	},
1932	artifacts: vec![AgentRunArtifactRef {
1933	kind: "transcript".to_string(),
1934	name: "transcript_handle".to_string(),
1935	target: "agent:agent_receipt".to_string(),
1936	description: "Read with handle_read from a live projection.".to_string(),
1937	}],
1938	usage: AgentRunUsage {
1939	status: "unknown".to_string(),
1940	input_tokens: None,
1941	output_tokens: None,
1942	total_tokens: None,
1943	cost_microusd: None,
1944	token_budget: None,
1945	budget_spent_tokens: None,
1946	budget_remaining_tokens: None,
1947	budget_scope: None,
1948	note: "not reported".to_string(),
1949	},
1950	verification: AgentRunVerificationSummary {
1951	status: "self_report_only".to_string(),
1952	summary: "no verified receipt attached".to_string(),
1953	},
1954	recommended_action: AgentRunRecommendedAction {
1955	action: "verify_self_report".to_string(),
1956	tool: Some("handle_read".to_string()),
1957	reason: "Worker agent_receipt completed; verify its self-report.".to_string(),
1958	},
1959	status: AgentWorkerStatus::Completed,
1960	created_at_ms: 1,
1961	updated_at_ms: 2,
1962	started_at_ms: Some(1),
1963	completed_at_ms: Some(2),
1964	latest_message: Some("completed".to_string()),
1965	result_summary: Some("receipt complete".to_string()),
1966	error: None,
1967	steps_taken: 2,
1968	events: VecDeque::from([AgentWorkerEvent {
1969	seq: 1,
1970	worker_id: "agent_receipt".to_string(),
1971	status: AgentWorkerStatus::Completed,
1972	timestamp_ms: 2,
1973	message: Some("completed".to_string()),
1974	step: Some(2),
1975	tool_name: None,
1976	}]),
1977	};
1978	let state_payload = json!({
1979	"schema_version": 1,
1980	"agents": [],
1981	"workers": [record],
1982	});
1983	fs::write(
1984	workspace.join(".codewhale/state/subagents.v1.json"),
1985	serde_json::to_vec_pretty(&state_payload)?,
1986	)?;
1987
1988	let sessions_dir = root.join("sessions");
1989	let Some((addr, _runtime_threads, handle)) =
1990	spawn_test_server_with_root_token_mobile_workspace(
1991	root.clone(),
1992	sessions_dir,
1993	None,
1994	false,
1995	workspace,
1996	)
1997	.await?
1998	else {
1999	return Ok(());
2000	};
2001	let client = crate::tls::reqwest_client();
2002
2003	let runs: serde_json::Value = client
2004	.get(format!("http://{addr}/v1/agent-runs"))
2005	.send()
2006	.await?
2007	.error_for_status()?
2008	.json()
2009	.await?;
2010	assert_eq!(runs["runs"][0]["spec"]["run_id"], "run_receipt");
2011	assert_eq!(runs["runs"][0]["follow_up"]["tool"], "handle_read");
2012	assert_eq!(
2013	runs["runs"][0]["verification"]["status"],
2014	"self_report_only"
2015	);
2016
2017	let run: serde_json::Value = client
2018	.get(format!("http://{addr}/v1/agent-runs/run_receipt"))
2019	.send()
2020	.await?
2021	.error_for_status()?
2022	.json()
2023	.await?;
2024	assert_eq!(run["spec"]["worker_id"], "agent_receipt");
2025	assert_eq!(run["takeover"]["supported"], true);
2026	assert_eq!(run["artifacts"][0]["kind"], "transcript");
2027
2028	let missing = client
2029	.get(format!("http://{addr}/v1/agent-runs/missing"))
2030	.send()
2031	.await?
2032	.status();
2033	assert_eq!(missing, StatusCode::NOT_FOUND);
2034
2035	handle.abort();
2036	Ok(())
2037	}
2038
2039	#[tokio::test]
2040	async fn stream_requires_prompt() -> Result<()> {
2041	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
2042	return Ok(());
2043	};
2044	let client = crate::tls::reqwest_client();
2045
2046	let resp = client
2047	.post(format!("http://{addr}/v1/stream"))
2048	.json(&json!({ "prompt": "" }))
2049	.send()
2050	.await?;
2051	assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
2052	handle.abort();
2053	Ok(())
2054	}
2055
2056	#[tokio::test]
2057	async fn compatibility_stream_closes_losslessly_across_replay_live_handoff() -> Result<()> {
2058	let temp = tempfile::tempdir()?;
2059	let root = temp.path().join("server");
2060	let sessions_dir = root.join("sessions");
2061	let workspace = root.join("workspace");
2062	let (hook_tx, mut hook_rx) = mpsc::unbounded_channel();
2063	let Some((addr, runtime_threads, handle)) =
2064	spawn_test_server_with_root_token_mobile_workspace_and_overrides(
2065	root,
2066	sessions_dir,
2067	None,
2068	false,
2069	workspace,
2070	TestServerOverrides {
2071	compat_stream_test_hook: Some(hook_tx),
2072	..TestServerOverrides::default()
2073	},
2074	)
2075	.await?
2076	else {
2077	return Ok(());
2078	};
2079
2080	let client = crate::tls::reqwest_client();
2081	let stream_client = client.clone();
2082	let stream_task = tokio::spawn(async move {
2083	let response = stream_client
2084	.post(format!("http://{addr}/v1/stream"))
2085	.json(&json!({ "prompt": "cross the replay handoff" }))
2086	.send()
2087	.await?
2088	.error_for_status()?;
2089	let content_type = response
2090	.headers()
2091	.get(reqwest::header::CONTENT_TYPE)
2092	.and_then(\|value\| value.to_str().ok())
2093	.unwrap_or_default()
2094	.to_string();
2095	let body = response.text().await?;
2096	Ok::<_, anyhow::Error>((content_type, body))
2097	});
2098
2099	let created = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), hook_rx.recv())
2100	.await
2101	.context("compatibility stream did not create its thread")?
2102	.context("compatibility stream test hook closed")?;
2103	let (thread_id, resume_created) = match created {
2104	CompatStreamTestPoint::ThreadCreated { thread_id, resume } => (thread_id, resume),
2105	CompatStreamTestPoint::SubscribedBeforeReplay { .. }
2106	\| CompatStreamTestPoint::ReplayLoaded { .. } => {
2107	bail!("compatibility stream loaded replay before its thread was prepared")
2108	}
2109	};
2110
2111	let harness = crate::core::engine::mock_engine_handle();
2112	runtime_threads
2113	.install_test_engine(&thread_id, harness.handle.clone())
2114	.await?;
2115	let mut rx_op = harness.rx_op;
2116	let tx_event = harness.tx_event;
2117	let (release_overlap, wait_for_overlap_release) = oneshot::channel();
2118	let (release_terminal, wait_for_terminal_release) = oneshot::channel();
2119	let engine_task = tokio::spawn(async move {
2120	if !matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
2121	return;
2122	}
2123	let _ = wait_for_overlap_release.await;
2124	let _ = tx_event
2125	.send(EngineEvent::TurnStarted {
2126	turn_id: "mock_compat_handoff".to_string(),
2127	created_at: chrono::Utc::now(),
2128	route: None,
2129	})
2130	.await;
2131	let _ = tx_event
2132	.send(EngineEvent::MessageStarted { index: 0 })
2133	.await;
2134	let _ = tx_event
2135	.send(EngineEvent::MessageDelta {
2136	index: 0,
2137	content: "handoff".to_string(),
2138	})
2139	.await;
2140	let _ = wait_for_terminal_release.await;
2141	let _ = tx_event
2142	.send(EngineEvent::MessageComplete { index: 0 })
2143	.await;
2144	let _ = tx_event
2145	.send(EngineEvent::TurnComplete {
2146	usage: Usage {
2147	input_tokens: 3,
2148	output_tokens: 1,
2149	..Usage::default()
2150	},
2151	status: TurnOutcomeStatus::Completed,
2152	error: None,
2153	tool_catalog: None,
2154	base_url: None,
2155	})
2156	.await;
2157	});
2158	resume_created
2159	.send(())
2160	.map_err(\|_\| anyhow::anyhow!("compatibility stream dropped thread-create handoff"))?;
2161
2162	let subscribed = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), hook_rx.recv())
2163	.await
2164	.context("compatibility stream did not subscribe before replay")?
2165	.context("compatibility stream test hook closed")?;
2166	let (subscribed_thread_id, subscribed_turn_id, resume_subscribed) = match subscribed {
2167	CompatStreamTestPoint::SubscribedBeforeReplay {
2168	thread_id,
2169	turn_id,
2170	resume,
2171	} => (thread_id, turn_id, resume),
2172	CompatStreamTestPoint::ThreadCreated { .. }
2173	\| CompatStreamTestPoint::ReplayLoaded { .. } => {
2174	bail!("compatibility stream did not expose its subscribe-before-replay boundary")
2175	}
2176	};
2177	assert_eq!(subscribed_thread_id, thread_id);
2178
2179	release_overlap
2180	.send(())
2181	.map_err(\|_\| anyhow::anyhow!("mock engine dropped overlap release"))?;
2182	tokio::time::timeout(ci_scaled(Duration::from_secs(2)), async {
2183	loop {
2184	if runtime_threads
2185	.events_since(&thread_id, None)
2186	.is_ok_and(\|events\| {
2187	events.iter().any(\|event\| {
2188	event.turn_id.as_deref() == Some(&subscribed_turn_id)
2189	&& event.event == "item.delta"
2190	})
2191	})
2192	{
2193	break;
2194	}
2195	sleep(Duration::from_millis(10)).await;
2196	}
2197	})
2198	.await
2199	.context("overlap event was not persisted before compatibility replay")?;
2200	resume_subscribed
2201	.send(())
2202	.map_err(\|_\| anyhow::anyhow!("compatibility stream dropped subscribe handoff"))?;
2203
2204	let replay_loaded = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), hook_rx.recv())
2205	.await
2206	.context("compatibility stream did not reach its replay/live handoff")?
2207	.context("compatibility stream test hook closed")?;
2208	let (replay_thread_id, turn_id, resume_replay) = match replay_loaded {
2209	CompatStreamTestPoint::ReplayLoaded {
2210	thread_id,
2211	turn_id,
2212	resume,
2213	} => (thread_id, turn_id, resume),
2214	CompatStreamTestPoint::ThreadCreated { .. }
2215	\| CompatStreamTestPoint::SubscribedBeforeReplay { .. } => {
2216	bail!("compatibility stream created more than one thread")
2217	}
2218	};
2219	assert_eq!(replay_thread_id, thread_id);
2220	assert_eq!(turn_id, subscribed_turn_id);
2221
2222	release_terminal
2223	.send(())
2224	.map_err(\|_\| anyhow::anyhow!("mock engine dropped terminal release"))?;
2225	tokio::time::timeout(ci_scaled(Duration::from_secs(2)), async {
2226	loop {
2227	if runtime_threads
2228	.events_since(&thread_id, None)
2229	.is_ok_and(\|events\| {
2230	events.iter().any(\|event\| {
2231	event.turn_id.as_deref() == Some(&turn_id)
2232	&& event.event == "turn.completed"
2233	})
2234	})
2235	{
2236	break;
2237	}
2238	sleep(Duration::from_millis(10)).await;
2239	}
2240	})
2241	.await
2242	.context("terminal event was not persisted during compatibility handoff")?;
2243	resume_replay
2244	.send(())
2245	.map_err(\|_\| anyhow::anyhow!("compatibility stream dropped replay handoff"))?;
2246
2247	let (content_type, body) = tokio::time::timeout(ci_scaled(Duration::from_secs(3)), stream_task)
2248	.await
2249	.context("compatibility stream hung after its terminal event")?
2250	.context("compatibility stream request task panicked")??;
2251	engine_task.await.context("mock engine task panicked")?;
2252
2253	assert!(content_type.starts_with("text/event-stream"));
2254	assert_eq!(body.matches("event: message.delta").count(), 1, "{body}");
2255	assert_eq!(body.matches("event: turn.completed").count(), 1, "{body}");
2256	assert_eq!(body.matches("event: done").count(), 1, "{body}");
2257
2258	handle.abort();
2259	Ok(())
2260	}
2261
2262	#[tokio::test]
2263	async fn compatibility_stream_exposes_and_resolves_user_input_without_answer_echo() -> Result<()> {
2264	let temp = tempfile::tempdir()?;
2265	let root = temp.path().join("server");
2266	let sessions_dir = root.join("sessions");
2267	let workspace = root.join("workspace");
2268	let (hook_tx, mut hook_rx) = mpsc::unbounded_channel();
2269	let Some((addr, runtime_threads, handle)) =
2270	spawn_test_server_with_root_token_mobile_workspace_and_overrides(
2271	root.clone(),
2272	sessions_dir,
2273	None,
2274	false,
2275	workspace,
2276	TestServerOverrides {
2277	compat_stream_test_hook: Some(hook_tx),
2278	..TestServerOverrides::default()
2279	},
2280	)
2281	.await?
2282	else {
2283	return Ok(());
2284	};
2285
2286	let client = crate::tls::reqwest_client();
2287	let stream_client = client.clone();
2288	let request_task = tokio::spawn(async move {
2289	let response = stream_client
2290	.post(format!("http://{addr}/v1/stream"))
2291	.json(&json!({ "prompt": "ask before continuing" }))
2292	.send()
2293	.await?
2294	.error_for_status()?;
2295	Ok::<_, anyhow::Error>(response)
2296	});
2297
2298	let created = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), hook_rx.recv())
2299	.await
2300	.context("compatibility stream did not create its interaction thread")?
2301	.context("compatibility stream interaction hook closed")?;
2302	let (thread_id, resume_created) = match created {
2303	CompatStreamTestPoint::ThreadCreated { thread_id, resume } => (thread_id, resume),
2304	CompatStreamTestPoint::SubscribedBeforeReplay { .. }
2305	\| CompatStreamTestPoint::ReplayLoaded { .. } => {
2306	bail!("compatibility interaction stream advanced before engine installation")
2307	}
2308	};
2309
2310	let mut harness = crate::core::engine::mock_engine_handle();
2311	runtime_threads
2312	.install_test_engine(&thread_id, harness.handle.clone())
2313	.await?;
2314	let (submission_tx, submission_rx) = oneshot::channel();
2315	let (release_completion, wait_for_completion_release) = oneshot::channel();
2316	let engine_task = tokio::spawn(async move {
2317	if !matches!(harness.rx_op.recv().await, Some(Op::SendMessage { .. })) {
2318	bail!("compatibility interaction engine did not receive a prompt");
2319	}
2320	harness
2321	.tx_event
2322	.send(EngineEvent::TurnStarted {
2323	turn_id: "mock_compat_input".to_string(),
2324	created_at: chrono::Utc::now(),
2325	route: None,
2326	})
2327	.await?;
2328	let request = crate::tools::user_input::UserInputRequest {
2329	questions: vec![crate::tools::user_input::UserInputQuestion {
2330	header: "Continue".to_string(),
2331	id: "choice".to_string(),
2332	question: "Continue the compatibility turn?".to_string(),
2333	options: vec![
2334	crate::tools::user_input::UserInputOption {
2335	label: "Continue".to_string(),
2336	description: "Finish the turn".to_string(),
2337	},
2338	crate::tools::user_input::UserInputOption {
2339	label: "Stop".to_string(),
2340	description: "Cancel the turn".to_string(),
2341	},
2342	],
2343	allow_free_text: false,
2344	multi_select: false,
2345	}],
2346	};
2347	harness
2348	.tx_event
2349	.send(EngineEvent::ToolCallStarted {
2350	id: "input_compat".to_string(),
2351	name: "request_user_input".to_string(),
2352	input: serde_json::to_value(&request)?,
2353	})
2354	.await?;
2355	harness
2356	.tx_event
2357	.send(EngineEvent::UserInputRequired {
2358	id: "input_compat".to_string(),
2359	request,
2360	})
2361	.await?;
2362	let submission = harness.recv_user_input_submission().await;
2363	let tool_result = submission
2364	.as_ref()
2365	.map(\|(_, response)\| crate::tools::spec::ToolResult::json(response))
2366	.transpose()?
2367	.context("compatibility user input was canceled before tool completion")?;
2368	let _ = submission_tx.send(submission);
2369	wait_for_completion_release
2370	.await
2371	.context("compatibility interaction test dropped completion release")?;
2372	harness
2373	.tx_event
2374	.send(EngineEvent::ToolCallComplete {
2375	id: "input_compat".to_string(),
2376	name: "request_user_input".to_string(),
2377	result: Ok(tool_result),
2378	})
2379	.await?;
2380	harness
2381	.tx_event
2382	.send(EngineEvent::MessageStarted { index: 0 })
2383	.await?;
2384	harness
2385	.tx_event
2386	.send(EngineEvent::MessageDelta {
2387	index: 0,
2388	content: "continued".to_string(),
2389	})
2390	.await?;
2391	harness
2392	.tx_event
2393	.send(EngineEvent::MessageComplete { index: 0 })
2394	.await?;
2395	harness
2396	.tx_event
2397	.send(EngineEvent::TurnComplete {
2398	usage: Usage::default(),
2399	status: TurnOutcomeStatus::Completed,
2400	error: None,
2401	tool_catalog: None,
2402	base_url: None,
2403	})
2404	.await?;
2405	Ok::<_, anyhow::Error>(())
2406	});
2407	resume_created
2408	.send(())
2409	.map_err(\|_\| anyhow::anyhow!("compatibility interaction stream dropped create hook"))?;
2410
2411	let subscribed = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), hook_rx.recv())
2412	.await
2413	.context("compatibility interaction stream did not subscribe")?
2414	.context("compatibility stream interaction hook closed")?;
2415	let (subscribed_thread_id, turn_id, resume_subscribed) = match subscribed {
2416	CompatStreamTestPoint::SubscribedBeforeReplay {
2417	thread_id,
2418	turn_id,
2419	resume,
2420	} => (thread_id, turn_id, resume),
2421	CompatStreamTestPoint::ThreadCreated { .. }
2422	\| CompatStreamTestPoint::ReplayLoaded { .. } => {
2423	bail!("compatibility interaction stream missed subscribe-before-replay hook")
2424	}
2425	};
2426	assert_eq!(subscribed_thread_id, thread_id);
2427	resume_subscribed
2428	.send(())
2429	.map_err(\|_\| anyhow::anyhow!("compatibility interaction stream dropped subscribe hook"))?;
2430
2431	let replay_loaded = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), hook_rx.recv())
2432	.await
2433	.context("compatibility interaction stream did not load replay")?
2434	.context("compatibility stream interaction hook closed")?;
2435	let (replay_thread_id, replay_turn_id, resume_replay) = match replay_loaded {
2436	CompatStreamTestPoint::ReplayLoaded {
2437	thread_id,
2438	turn_id,
2439	resume,
2440	} => (thread_id, turn_id, resume),
2441	CompatStreamTestPoint::ThreadCreated { .. }
2442	\| CompatStreamTestPoint::SubscribedBeforeReplay { .. } => {
2443	bail!("compatibility interaction stream missed replay-loaded hook")
2444	}
2445	};
2446	assert_eq!(replay_thread_id, thread_id);
2447	assert_eq!(replay_turn_id, turn_id);
2448	resume_replay
2449	.send(())
2450	.map_err(\|_\| anyhow::anyhow!("compatibility interaction stream dropped replay hook"))?;
2451
2452	let response = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), request_task)
2453	.await
2454	.context("compatibility interaction request did not return SSE headers")?
2455	.context("compatibility interaction request task panicked")??;
2456	let (frame_tx, mut frame_rx) = mpsc::unbounded_channel();
2457	let body_task = tokio::spawn(collect_sse_frames(response, frame_tx));
2458
2459	let required_payload = tokio::time::timeout(ci_scaled(Duration::from_secs(2)), async {
2460	loop {
2461	let (event, payload) = frame_rx
2462	.recv()
2463	.await
2464	.context("compatibility interaction stream ended before user input")?;
2465	if event == "user_input.required" {
2466	break Ok::<_, anyhow::Error>(payload);
2467	}
2468	}
2469	})
2470	.await
2471	.context("compatibility stream did not expose required user input")??;
2472	assert_eq!(required_payload["id"], "input_compat");
2473	assert_eq!(required_payload["input_id"], "input_compat");
2474	assert_eq!(required_payload["thread_id"], thread_id);
2475	assert_eq!(required_payload["turn_id"], turn_id);
2476	assert_eq!(required_payload["status"], "required");
2477	assert_eq!(required_payload["request"]["questions"][0]["id"], "choice");
2478	assert!(required_payload.get("answers").is_none());
2479
2480	const SECRET_ANSWER: &str = "compat-answer-must-not-be-echoed";
2481	let submitted: serde_json::Value = client
2482	.post(format!(
2483	"http://{addr}/v1/user-input/{thread_id}/input_compat"
2484	))
2485	.json(&json!({
2486	"answers": [{
2487	"id": "choice",
2488	"label": "Continue",
2489	"value": SECRET_ANSWER,
2490	}],
2491	}))
2492	.send()
2493	.await?
2494	.error_for_status()?
2495	.json()
2496	.await?;
2497	assert_eq!(submitted["delivered"], true);
2498	let (submitted_id, submitted_response) =
2499	tokio::time::timeout(ci_scaled(Duration::from_secs(2)), submission_rx)
2500	.await
2501	.context("mock engine did not receive compatibility user input")?
2502	.context("mock engine dropped compatibility user input")?
2503	.context("compatibility user input was canceled instead of submitted")?;
2504	assert_eq!(submitted_id, "input_compat");
2505	assert_eq!(submitted_response.answers[0].value, SECRET_ANSWER);
2506	release_completion
2507	.send(())
2508	.map_err(\|_\| anyhow::anyhow!("mock interaction engine dropped completion release"))?;
2509
2510	let frames = tokio::time::timeout(ci_scaled(Duration::from_secs(3)), body_task)
2511	.await
2512	.context("compatibility interaction stream did not terminate")?
2513	.context("compatibility interaction body task panicked")??;
2514	engine_task
2515	.await
2516	.context("compatibility interaction engine task panicked")??;
2517
2518	let answered = frames
2519	.iter()
2520	.find(\|(event, _)\| event == "user_input.answered")
2521	.context("compatibility stream omitted submitted user-input lifecycle")?;
2522	assert_eq!(answered.1["id"], "input_compat");
2523	assert_eq!(answered.1["status"], "submitted");
2524	assert!(answered.1.get("answers").is_none());
2525	assert_eq!(
2526	frames
2527	.iter()
2528	.filter(\|(event, _)\| event == "user_input.required")
2529	.count(),
2530	1
2531	);
2532	assert_eq!(
2533	frames
2534	.iter()
2535	.filter(\|(event, _)\| event == "user_input.answered")
2536	.count(),
2537	1
2538	);
2539	assert!(
2540	!frames
2541	.iter()
2542	.any(\|(event, _)\| event == "user_input.canceled")
2543	);
2544	assert!(frames.iter().any(\|(event, _)\| event == "turn.completed"));
2545	assert_eq!(
2546	frames.iter().filter(\|(event, _)\| event == "done").count(),
2547	1
2548	);
2549	assert!(
2550	!serde_json::to_string(&frames)?.contains(SECRET_ANSWER),
2551	"submitted answer leaked into compatibility SSE"
2552	);
2553	let detail = runtime_threads.get_thread_detail(&thread_id).await?;
2554	let serialized_detail = serde_json::to_string(&detail)?;
2555	assert!(
2556	!serialized_detail.contains(SECRET_ANSWER),
2557	"submitted answer leaked into the thread snapshot"
2558	);
2559	let redacted_item = detail
2560	.items
2561	.iter()
2562	.find(\|item\| {
2563	item.metadata
2564	.as_ref()
2565	.and_then(\|metadata\| metadata.get("tool_name"))
2566	.and_then(Value::as_str)
2567	== Some("request_user_input")
2568	})
2569	.context("request_user_input Runtime receipt was not persisted")?;
2570	assert_eq!(
2571	redacted_item.detail.as_deref(),
2572	Some("User input submitted")
2573	);
2574	assert_eq!(
2575	redacted_item
2576	.metadata
2577	.as_ref()
2578	.and_then(\|metadata\| metadata.get("response_redacted"))
2579	.and_then(Value::as_bool),
2580	Some(true)
2581	);
2582	let durable_events = runtime_threads.events_since(&thread_id, None)?;
2583	assert!(
2584	!serde_json::to_string(&durable_events)?.contains(SECRET_ANSWER),
2585	"submitted answer leaked into the durable Runtime event log"
2586	);
2587	let leaked_file = ignore::WalkBuilder::new(root.join("runtime"))
2588	.hidden(false)
2589	.build()
2590	.filter_map(std::result::Result::ok)
2591	.filter(\|entry\| entry.file_type().is_some_and(\|kind\| kind.is_file()))
2592	.find_map(\|entry\| {
2593	fs::read_to_string(entry.path())
2594	.ok()
2595	.filter(\|contents\| contents.contains(SECRET_ANSWER))
2596	.map(\|_\| entry.path().to_path_buf())
2597	});
2598	assert!(
2599	leaked_file.is_none(),
2600	"submitted answer leaked into Runtime file {}",
2601	leaked_file
2602	.as_deref()
2603	.map(std::path::Path::display)
2604	.map(\|path\| path.to_string())
2605	.unwrap_or_default()
2606	);
2607
2608	handle.abort();
2609	Ok(())
2610	}
2611
2612	#[tokio::test]
2613	async fn thread_endpoints_expose_lifecycle_contract() -> Result<()> {
2614	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
2615	return Ok(());
2616	};
2617	let client = crate::tls::reqwest_client();
2618
2619	let created: serde_json::Value = client
2620	.post(format!("http://{addr}/v1/threads"))
2621	.json(&json!({}))
2622	.send()
2623	.await?
2624	.error_for_status()?
2625	.json()
2626	.await?;
2627	let thread_id = created["id"]
2628	.as_str()
2629	.context("missing thread id")?
2630	.to_string();
2631
2632	let archived: serde_json::Value = client
2633	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
2634	.json(&json!({ "archived": true }))
2635	.send()
2636	.await?
2637	.error_for_status()?
2638	.json()
2639	.await?;
2640	assert_eq!(archived["id"], thread_id);
2641	assert_eq!(archived["archived"], true);
2642
2643	let listed: serde_json::Value = client
2644	.get(format!("http://{addr}/v1/threads"))
2645	.send()
2646	.await?
2647	.error_for_status()?
2648	.json()
2649	.await?;
2650	assert!(
2651	listed
2652	.as_array()
2653	.is_some_and(\|threads\| threads.iter().all(\|t\| t["id"] != thread_id))
2654	);
2655
2656	let listed_all: serde_json::Value = client
2657	.get(format!(
2658	"http://{addr}/v1/threads/summary?include_archived=true&limit=100"
2659	))
2660	.send()
2661	.await?
2662	.error_for_status()?
2663	.json()
2664	.await?;
2665	assert!(
2666	listed_all
2667	.as_array()
2668	.is_some_and(\|threads\| threads.iter().any(\|t\| t["id"] == thread_id))
2669	);
2670
2671	let unarchived: serde_json::Value = client
2672	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
2673	.json(&json!({ "archived": false }))
2674	.send()
2675	.await?
2676	.error_for_status()?
2677	.json()
2678	.await?;
2679	assert_eq!(unarchived["archived"], false);
2680
2681	let invalid_patch = client
2682	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
2683	.json(&json!({}))
2684	.send()
2685	.await?;
2686	assert_eq!(invalid_patch.status(), StatusCode::BAD_REQUEST);
2687
2688	let missing_patch = client
2689	.patch(format!("http://{addr}/v1/threads/thr_missing"))
2690	.json(&json!({ "archived": true }))
2691	.send()
2692	.await?;
2693	assert_eq!(missing_patch.status(), StatusCode::NOT_FOUND);
2694
2695	let detail: serde_json::Value = client
2696	.get(format!("http://{addr}/v1/threads/{thread_id}"))
2697	.send()
2698	.await?
2699	.error_for_status()?
2700	.json()
2701	.await?;
2702	assert_eq!(detail["thread"]["id"], thread_id);
2703
2704	let resumed: serde_json::Value = client
2705	.post(format!("http://{addr}/v1/threads/{thread_id}/resume"))
2706	.send()
2707	.await?
2708	.error_for_status()?
2709	.json()
2710	.await?;
2711	assert_eq!(resumed["id"], thread_id);
2712
2713	let forked: serde_json::Value = client
2714	.post(format!("http://{addr}/v1/threads/{thread_id}/fork"))
2715	.send()
2716	.await?
2717	.error_for_status()?
2718	.json()
2719	.await?;
2720	let forked_id = forked["id"].as_str().context("missing forked id")?;
2721	assert_ne!(forked_id, thread_id);
2722
2723	// Install a mock engine so the turn completes without calling the real API.
2724	// The mock handles both SendMessage and CompactContext ops so the
2725	// compact endpoint tested later also works.
2726	let harness = crate::core::engine::mock_engine_handle();
2727	runtime_threads
2728	.install_test_engine(&thread_id, harness.handle.clone())
2729	.await?;
2730	let mut rx_op = harness.rx_op;
2731	let tx_event = harness.tx_event;
2732	tokio::spawn(async move {
2733	while let Some(op) = rx_op.recv().await {
2734	match op {
2735	Op::SendMessage { .. } => {
2736	let _ = tx_event
2737	.send(EngineEvent::TurnStarted {
2738	turn_id: "mock_lifecycle".to_string(),
2739	created_at: chrono::Utc::now(),
2740	route: None,
2741	})
2742	.await;
2743	let _ = tx_event
2744	.send(EngineEvent::MessageStarted { index: 0 })
2745	.await;
2746	let _ = tx_event
2747	.send(EngineEvent::MessageDelta {
2748	index: 0,
2749	content: "mock reply".to_string(),
2750	})
2751	.await;
2752	let _ = tx_event
2753	.send(EngineEvent::MessageComplete { index: 0 })
2754	.await;
2755	let _ = tx_event
2756	.send(EngineEvent::TurnComplete {
2757	usage: Usage {
2758	input_tokens: 10,
2759	output_tokens: 5,
2760	..Usage::default()
2761	},
2762	status: TurnOutcomeStatus::Completed,
2763	error: None,
2764	tool_catalog: None,
2765	base_url: None,
2766	})
2767	.await;
2768	}
2769	Op::CompactContext { .. } => {
2770	let _ = tx_event
2771	.send(EngineEvent::TurnComplete {
2772	usage: Usage {
2773	input_tokens: 0,
2774	output_tokens: 0,
2775	..Usage::default()
2776	},
2777	status: TurnOutcomeStatus::Completed,
2778	error: None,
2779	tool_catalog: None,
2780	base_url: None,
2781	})
2782	.await;
2783	}
2784	_ => {}
2785	}
2786	}
2787	});
2788
2789	let turn_start: serde_json::Value = client
2790	.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
2791	.json(&json!({ "prompt": "thread endpoint test" }))
2792	.send()
2793	.await?
2794	.error_for_status()?
2795	.json()
2796	.await?;
2797	let turn_id = turn_start["turn"]["id"]
2798	.as_str()
2799	.context("missing turn id")?
2800	.to_string();
2801
2802	let _ =
2803	wait_for_terminal_turn_status(&client, addr, &thread_id, &turn_id, Duration::from_secs(2))
2804	.await?;
2805
2806	let steer_resp = client
2807	.post(format!(
2808	"http://{addr}/v1/threads/{thread_id}/turns/{turn_id}/steer"
2809	))
2810	.json(&json!({ "prompt": "late steer" }))
2811	.send()
2812	.await?;
2813	assert_eq!(steer_resp.status(), StatusCode::CONFLICT);
2814
2815	let interrupt_resp = client
2816	.post(format!(
2817	"http://{addr}/v1/threads/{thread_id}/turns/{turn_id}/interrupt"
2818	))
2819	.send()
2820	.await?;
2821	assert_eq!(interrupt_resp.status(), StatusCode::CONFLICT);
2822
2823	let compact_start: serde_json::Value = client
2824	.post(format!("http://{addr}/v1/threads/{thread_id}/compact"))
2825	.json(&json!({ "reason": "test manual compact" }))
2826	.send()
2827	.await?
2828	.error_for_status()?
2829	.json()
2830	.await?;
2831	assert_eq!(compact_start["thread"]["id"], thread_id);
2832
2833	let events_resp = client
2834	.get(format!(
2835	"http://{addr}/v1/threads/{thread_id}/events?since_seq=0"
2836	))
2837	.send()
2838	.await?
2839	.error_for_status()?;
2840	let content_type = events_resp
2841	.headers()
2842	.get(reqwest::header::CONTENT_TYPE)
2843	.and_then(\|v\| v.to_str().ok())
2844	.unwrap_or_default()
2845	.to_string();
2846	assert!(content_type.starts_with("text/event-stream"));
2847	let chunk_text = read_first_sse_frame(events_resp).await?;
2848	assert!(
2849	chunk_text.contains("event:"),
2850	"expected SSE event chunk, got: {chunk_text}"
2851	);
2852	let (event_name, payload) = parse_sse_frame(&chunk_text)?;
2853	assert_eq!(event_name, "thread.started");
2854	assert!(
2855	event_name.starts_with("item.")
2856	\|\| event_name.starts_with("turn.")
2857	\|\| event_name.starts_with("thread.")
2858	\|\| event_name == "turn.completed"
2859	\|\| event_name == "turn.started"
2860	\|\| event_name == "thread.started",
2861	"unexpected first event name: {event_name}"
2862	);
2863	assert_eq!(payload["event"], payload["kind"]);
2864	assert!(payload.get("turn_id").is_some());
2865	assert!(payload.get("item_id").is_some());
2866	assert!(payload["turn_id"].is_null());
2867	assert!(payload["item_id"].is_null());
2868	assert_eq!(payload["thread_id"], thread_id);
2869	assert!(
2870	payload["schema_version"]
2871	.as_u64()
2872	.is_some_and(\|version\| version >= 1)
2873	);
2874	assert!(payload.get("seq").and_then(Value::as_u64).is_some());
2875	assert!(payload["payload"].is_object() \|\| payload["payload"].is_array());
2876
2877	handle.abort();
2878	Ok(())
2879	}
2880
2881	#[tokio::test]
2882	async fn events_endpoint_respects_since_seq_cursor() -> Result<()> {
2883	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
2884	return Ok(());
2885	};
2886	let client = crate::tls::reqwest_client();
2887
2888	let created: serde_json::Value = client
2889	.post(format!("http://{addr}/v1/threads"))
2890	.json(&json!({}))
2891	.send()
2892	.await?
2893	.error_for_status()?
2894	.json()
2895	.await?;
2896	let thread_id = created["id"]
2897	.as_str()
2898	.context("missing thread id")?
2899	.to_string();
2900
2901	// Install a mock engine so the turn completes without calling the real API.
2902	let harness = crate::core::engine::mock_engine_handle();
2903	runtime_threads
2904	.install_test_engine(&thread_id, harness.handle.clone())
2905	.await?;
2906	let mut rx_op = harness.rx_op;
2907	let tx_event = harness.tx_event;
2908	tokio::spawn(async move {
2909	if !matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
2910	return;
2911	}
2912	let _ = tx_event
2913	.send(EngineEvent::TurnStarted {
2914	turn_id: "mock_cursor".to_string(),
2915	created_at: chrono::Utc::now(),
2916	route: None,
2917	})
2918	.await;
2919	let _ = tx_event
2920	.send(EngineEvent::MessageStarted { index: 0 })
2921	.await;
2922	let _ = tx_event
2923	.send(EngineEvent::MessageComplete { index: 0 })
2924	.await;
2925	let _ = tx_event
2926	.send(EngineEvent::TurnComplete {
2927	usage: Usage {
2928	input_tokens: 5,
2929	output_tokens: 3,
2930	..Usage::default()
2931	},
2932	status: TurnOutcomeStatus::Completed,
2933	error: None,
2934	tool_catalog: None,
2935	base_url: None,
2936	})
2937	.await;
2938	});
2939
2940	let started: serde_json::Value = client
2941	.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
2942	.json(&json!({ "prompt": "cursor replay test" }))
2943	.send()
2944	.await?
2945	.error_for_status()?
2946	.json()
2947	.await?;
2948	let turn_id = started["turn"]["id"]
2949	.as_str()
2950	.context("missing turn id")?
2951	.to_string();
2952
2953	let _ =
2954	wait_for_terminal_turn_status(&client, addr, &thread_id, &turn_id, Duration::from_secs(2))
2955	.await?;
2956
2957	let resp_a = client
2958	.get(format!(
2959	"http://{addr}/v1/threads/{thread_id}/events?since_seq=0"
2960	))
2961	.send()
2962	.await?
2963	.error_for_status()?;
2964	let frame_a = read_first_sse_frame(resp_a).await?;
2965	let (event_a, payload_a) = parse_sse_frame(&frame_a)?;
2966	assert_eq!(event_a, "thread.started");
2967	assert!(payload_a.get("turn_id").is_some());
2968	assert!(payload_a.get("item_id").is_some());
2969	assert!(payload_a["turn_id"].is_null());
2970	assert!(payload_a["item_id"].is_null());
2971	assert!(payload_a.get("schema_version").is_some());
2972	assert_eq!(payload_a["event"], payload_a["kind"]);
2973	assert_eq!(payload_a["thread_id"], thread_id);
2974	let seq_a = payload_a
2975	.get("seq")
2976	.and_then(Value::as_u64)
2977	.context("missing seq in first replay frame")?;
2978
2979	let resp_b = client
2980	.get(format!(
2981	"http://{addr}/v1/threads/{thread_id}/events?since_seq={seq_a}"
2982	))
2983	.send()
2984	.await?
2985	.error_for_status()?;
2986	let frame_b = read_first_sse_frame(resp_b).await?;
2987	let (_event_b, payload_b) = parse_sse_frame(&frame_b)?;
2988	assert!(payload_b.get("schema_version").is_some());
2989	assert_eq!(payload_b["event"], payload_b["kind"]);
2990	assert_eq!(payload_b["thread_id"], thread_id);
2991	let seq_b = payload_b
2992	.get("seq")
2993	.and_then(Value::as_u64)
2994	.context("missing seq in second replay frame")?;
2995	assert!(
2996	seq_b > seq_a,
2997	"expected seq after cursor: {seq_b} <= {seq_a}"
2998	);
2999
3000	handle.abort();
3001	Ok(())
3002	}
3003
3004	#[tokio::test]
3005	async fn event_handoff_replays_and_dedupes_interaction_prompts_without_a_gap() -> Result<()> {
3006	let Some((_addr, runtime_threads, handle)) = spawn_test_server().await? else {
3007	return Ok(());
3008	};
3009	let thread = runtime_threads
3010	.create_thread(CreateThreadRequest::default())
3011	.await?;
3012	let initial_seq = runtime_threads
3013	.events_since(&thread.id, None)?
3014	.last()
3015	.context("thread creation should emit an event")?
3016	.seq;
3017
3018	// Deterministically place approval.required in the old vulnerable window:
3019	// the receiver exists, but durable replay has not been read yet.
3020	let live = runtime_threads.subscribe_events();
3021	let approval = runtime_threads
3022	.emit_event_for_test(
3023	&thread.id,
3024	None,
3025	"approval.required",
3026	json!({
3027	"approval_id": "approval-handoff",
3028	"tool_name": "exec_command",
3029	"description": "Run a local check",
3030	}),
3031	)
3032	.await?;
3033	let backlog = runtime_threads.events_since(&thread.id, Some(initial_seq))?;
3034	let (backlog_tx, backlog_rx) = mpsc::channel(1);
3035	backlog_tx
3036	.send(Ok(backlog))
3037	.await
3038	.map_err(\|_\| anyhow::anyhow!("failed to seed replay backlog"))?;
3039	drop(backlog_tx);
3040
3041	// This request lands after the replay read and is therefore live-only.
3042	let input = runtime_threads
3043	.emit_event_for_test(
3044	&thread.id,
3045	None,
3046	"user_input.required",
3047	json!({
3048	"id": "input-handoff",
3049	"request": {
3050	"questions": [{
3051	"id": "choice",
3052	"question": "Continue?",
3053	"options": [],
3054	}],
3055	},
3056	}),
3057	)
3058	.await?;
3059
3060	let stream = replay_live_thread_events(
3061	runtime_threads.clone(),
3062	thread.id.clone(),
3063	initial_seq,
3064	backlog_rx,
3065	live,
3066	)
3067	.take(2);
3068	let body =
3069	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3070	let rendered = String::from_utf8(body.to_vec())?;
3071	let frames = rendered
3072	.split("\n\n")
3073	.map(str::trim)
3074	.filter(\|frame\| !frame.is_empty())
3075	.map(parse_sse_frame)
3076	.collect::<Result<Vec<_>>>()?;
3077
3078	assert_eq!(frames.len(), 2, "unexpected SSE frames: {rendered}");
3079	assert_eq!(frames[0].0, "approval.required");
3080	assert_eq!(frames[0].1["seq"], approval.seq);
3081	assert_eq!(frames[0].1["previous_seq"], initial_seq);
3082	assert_eq!(frames[1].0, "user_input.required");
3083	assert_eq!(frames[1].1["seq"], input.seq);
3084	assert_eq!(frames[1].1["previous_seq"], approval.seq);
3085	assert_eq!(rendered.matches("approval-handoff").count(), 1);
3086	assert_eq!(rendered.matches("input-handoff").count(), 1);
3087
3088	handle.abort();
3089	Ok(())
3090	}
3091
3092	#[tokio::test]
3093	async fn steer_and_interrupt_endpoints_work_on_active_turn() -> Result<()> {
3094	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
3095	return Ok(());
3096	};
3097	let client = crate::tls::reqwest_client();
3098
3099	let created: serde_json::Value = client
3100	.post(format!("http://{addr}/v1/threads"))
3101	.json(&json!({}))
3102	.send()
3103	.await?
3104	.error_for_status()?
3105	.json()
3106	.await?;
3107	let thread_id = created["id"]
3108	.as_str()
3109	.context("missing thread id")?
3110	.to_string();
3111
3112	let harness = crate::core::engine::mock_engine_handle();
3113	runtime_threads
3114	.install_test_engine(&thread_id, harness.handle.clone())
3115	.await?;
3116	let mut rx_op = harness.rx_op;
3117	let mut rx_steer = harness.rx_steer;
3118	let tx_event = harness.tx_event;
3119	let cancel_token = harness.cancel_token;
3120	tokio::spawn(async move {
3121	if !matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
3122	return;
3123	}
3124	let _ = tx_event
3125	.send(EngineEvent::TurnStarted {
3126	turn_id: "engine_turn_api".to_string(),
3127	created_at: chrono::Utc::now(),
3128	route: None,
3129	})
3130	.await;
3131	let _ = tx_event
3132	.send(EngineEvent::MessageStarted { index: 0 })
3133	.await;
3134	if let Some(steer_text) = rx_steer.recv().await {
3135	let _ = tx_event
3136	.send(EngineEvent::MessageDelta {
3137	index: 0,
3138	content: format!("steer:{steer_text}"),
3139	})
3140	.await;
3141	}
3142	cancel_token.cancelled().await;
3143	sleep(Duration::from_millis(60)).await;
3144	let _ = tx_event
3145	.send(EngineEvent::TurnComplete {
3146	usage: Usage {
3147	input_tokens: 2,
3148	output_tokens: 1,
3149	..Usage::default()
3150	},
3151	status: TurnOutcomeStatus::Completed,
3152	error: None,
3153	tool_catalog: None,
3154	base_url: None,
3155	})
3156	.await;
3157	});
3158
3159	let turn_start: serde_json::Value = client
3160	.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
3161	.json(&json!({ "prompt": "active controls" }))
3162	.send()
3163	.await?
3164	.error_for_status()?
3165	.json()
3166	.await?;
3167	let turn_id = turn_start["turn"]["id"]
3168	.as_str()
3169	.context("missing turn id")?
3170	.to_string();
3171
3172	let steer_resp: serde_json::Value = client
3173	.post(format!(
3174	"http://{addr}/v1/threads/{thread_id}/turns/{turn_id}/steer"
3175	))
3176	.json(&json!({ "prompt": "please steer" }))
3177	.send()
3178	.await?
3179	.error_for_status()?
3180	.json()
3181	.await?;
3182	assert_eq!(steer_resp["id"], turn_id);
3183	assert_eq!(steer_resp["steer_count"], 1);
3184
3185	let interrupt_resp: serde_json::Value = client
3186	.post(format!(
3187	"http://{addr}/v1/threads/{thread_id}/turns/{turn_id}/interrupt"
3188	))
3189	.send()
3190	.await?
3191	.error_for_status()?
3192	.json()
3193	.await?;
3194	assert_eq!(interrupt_resp["id"], turn_id);
3195
3196	let terminal =
3197	wait_for_terminal_turn_status(&client, addr, &thread_id, &turn_id, Duration::from_secs(3))
3198	.await?;
3199	assert_eq!(terminal, "interrupted");
3200
3201	let events = runtime_threads.events_since(&thread_id, None)?;
3202	assert!(events.iter().any(\|ev\| ev.event == "turn.steered"));
3203	assert!(
3204	events
3205	.iter()
3206	.any(\|ev\| ev.event == "turn.interrupt_requested")
3207	);
3208	assert!(events.iter().any(\|ev\| {
3209	ev.event == "turn.completed"
3210	&& ev
3211	.payload
3212	.get("turn")
3213	.and_then(\|turn\| turn.get("status"))
3214	.and_then(Value::as_str)
3215	== Some("interrupted")
3216	}));
3217
3218	handle.abort();
3219	Ok(())
3220	}
3221
3222	#[tokio::test]
3223	async fn stream_compat_mapping_handles_expected_runtime_events() -> Result<()> {
3224	let agent_delta = RuntimeEventRecord {
3225	schema_version: 1,
3226	seq: 1,
3227	timestamp: chrono::Utc::now(),
3228	thread_id: "thr_test".to_string(),
3229	turn_id: Some("turn_test".to_string()),
3230	item_id: Some("item_test".to_string()),
3231	event: "item.delta".to_string(),
3232	payload: json!({
3233	"kind": "agent_message",
3234	"delta": "hello",
3235	}),
3236	};
3237	let mapped = map_compat_stream_event(&agent_delta).context("missing mapped SSE event")?;
3238	let stream = async_stream::stream! {
3239	yield Ok::<_, Infallible>(mapped);
3240	};
3241	let body =
3242	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3243	let text = String::from_utf8_lossy(&body);
3244	assert!(text.contains("event: message.delta"));
3245	assert!(text.contains("\"content\":\"hello\""));
3246
3247	let tool_start = RuntimeEventRecord {
3248	schema_version: 1,
3249	seq: 2,
3250	timestamp: chrono::Utc::now(),
3251	thread_id: "thr_test".to_string(),
3252	turn_id: Some("turn_test".to_string()),
3253	item_id: Some("item_tool".to_string()),
3254	event: "item.started".to_string(),
3255	payload: json!({
3256	"tool": { "id": "tool_1", "name": "exec_shell", "input": { "cmd": "pwd" } }
3257	}),
3258	};
3259	let mapped = map_compat_stream_event(&tool_start).context("missing tool.started event")?;
3260	let stream = async_stream::stream! {
3261	yield Ok::<_, Infallible>(mapped);
3262	};
3263	let body =
3264	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3265	let text = String::from_utf8_lossy(&body);
3266	assert!(text.contains("event: tool.started"));
3267
3268	let tool_done = RuntimeEventRecord {
3269	schema_version: 1,
3270	seq: 3,
3271	timestamp: chrono::Utc::now(),
3272	thread_id: "thr_test".to_string(),
3273	turn_id: Some("turn_test".to_string()),
3274	item_id: Some("item_tool".to_string()),
3275	event: "item.completed".to_string(),
3276	payload: json!({
3277	"item": {
3278	"id": "item_tool",
3279	"kind": "tool_call",
3280	"summary": "ok",
3281	"detail": "done"
3282	}
3283	}),
3284	};
3285	let mapped = map_compat_stream_event(&tool_done).context("missing tool.completed event")?;
3286	let stream = async_stream::stream! {
3287	yield Ok::<_, Infallible>(mapped);
3288	};
3289	let body =
3290	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3291	let text = String::from_utf8_lossy(&body);
3292	assert!(text.contains("event: tool.completed"));
3293	assert!(text.contains("\"success\":true"));
3294
3295	let user_input_required = RuntimeEventRecord {
3296	schema_version: 1,
3297	seq: 4,
3298	timestamp: chrono::Utc::now(),
3299	thread_id: "thr_test".to_string(),
3300	turn_id: Some("turn_test".to_string()),
3301	item_id: None,
3302	event: "user_input.required".to_string(),
3303	payload: json!({
3304	"id": "input_test",
3305	"request": {
3306	"questions": [{
3307	"header": "Continue",
3308	"id": "choice",
3309	"question": "Continue?",
3310	"options": [
3311	{ "label": "Yes", "description": "Continue" },
3312	{ "label": "No", "description": "Stop" }
3313	]
3314	}]
3315	},
3316	"internal_secret": "required-secret",
3317	}),
3318	};
3319	let mapped = map_compat_stream_event(&user_input_required)
3320	.context("missing user_input.required event")?;
3321	let stream = async_stream::stream! {
3322	yield Ok::<_, Infallible>(mapped);
3323	};
3324	let body =
3325	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3326	let text = String::from_utf8_lossy(&body);
3327	assert!(text.contains("event: user_input.required"));
3328	assert!(text.contains("\"input_id\":\"input_test\""));
3329	assert!(text.contains("\"status\":\"required\""));
3330	assert!(!text.contains("required-secret"));
3331
3332	let user_input_answered = RuntimeEventRecord {
3333	schema_version: 1,
3334	seq: 5,
3335	timestamp: chrono::Utc::now(),
3336	thread_id: "thr_test".to_string(),
3337	turn_id: Some("turn_test".to_string()),
3338	item_id: None,
3339	event: "user_input.answered".to_string(),
3340	payload: json!({
3341	"input_id": "input_test",
3342	"answers": [{ "id": "choice", "value": "answer-secret" }],
3343	}),
3344	};
3345	let mapped = map_compat_stream_event(&user_input_answered)
3346	.context("missing user_input.answered event")?;
3347	let stream = async_stream::stream! {
3348	yield Ok::<_, Infallible>(mapped);
3349	};
3350	let body =
3351	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3352	let text = String::from_utf8_lossy(&body);
3353	assert!(text.contains("event: user_input.answered"));
3354	assert!(text.contains("\"status\":\"submitted\""));
3355	assert!(!text.contains("answer-secret"));
3356	assert!(!text.contains("\"answers\""));
3357
3358	let user_input_canceled = RuntimeEventRecord {
3359	schema_version: 1,
3360	seq: 6,
3361	timestamp: chrono::Utc::now(),
3362	thread_id: "thr_test".to_string(),
3363	turn_id: Some("turn_test".to_string()),
3364	item_id: None,
3365	event: "user_input.canceled".to_string(),
3366	payload: json!({ "id": "input_test", "terminal": true }),
3367	};
3368	let mapped = map_compat_stream_event(&user_input_canceled)
3369	.context("missing user_input.canceled event")?;
3370	let stream = async_stream::stream! {
3371	yield Ok::<_, Infallible>(mapped);
3372	};
3373	let body =
3374	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3375	let text = String::from_utf8_lossy(&body);
3376	assert!(text.contains("event: user_input.canceled"));
3377	assert!(text.contains("\"status\":\"canceled\""));
3378	assert!(text.contains("\"terminal\":true"));
3379
3380	let approval_required = RuntimeEventRecord {
3381	schema_version: 1,
3382	seq: 7,
3383	timestamp: chrono::Utc::now(),
3384	thread_id: "thr_test".to_string(),
3385	turn_id: Some("turn_test".to_string()),
3386	item_id: None,
3387	event: "approval.required".to_string(),
3388	payload: json!({
3389	"approval_id": "approval_test",
3390	"tool_name": "exec_command",
3391	"description": "Run tests",
3392	"input": { "token": "approval-secret" },
3393	}),
3394	};
3395	let mapped =
3396	map_compat_stream_event(&approval_required).context("missing approval.required event")?;
3397	let stream = async_stream::stream! {
3398	yield Ok::<_, Infallible>(mapped);
3399	};
3400	let body =
3401	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3402	let text = String::from_utf8_lossy(&body);
3403	assert!(text.contains("event: approval.required"));
3404	assert!(text.contains("\"approval_id\":\"approval_test\""));
3405	assert!(!text.contains("approval-secret"));
3406
3407	let approval_decided = RuntimeEventRecord {
3408	schema_version: 1,
3409	seq: 8,
3410	timestamp: chrono::Utc::now(),
3411	thread_id: "thr_test".to_string(),
3412	turn_id: Some("turn_test".to_string()),
3413	item_id: None,
3414	event: "approval.decided".to_string(),
3415	payload: json!({
3416	"approval_id": "approval_test",
3417	"decision": "allow",
3418	"remember": false,
3419	"internal_secret": "approval-decision-secret",
3420	}),
3421	};
3422	let mapped =
3423	map_compat_stream_event(&approval_decided).context("missing approval.decided event")?;
3424	let stream = async_stream::stream! {
3425	yield Ok::<_, Infallible>(mapped);
3426	};
3427	let body =
3428	axum::body::to_bytes(Sse::new(stream).into_response().into_body(), usize::MAX).await?;
3429	let text = String::from_utf8_lossy(&body);
3430	assert!(text.contains("event: approval.decided"));
3431	assert!(text.contains("\"decision\":\"allow\""));
3432	assert!(!text.contains("approval-decision-secret"));
3433
3434	let unknown = RuntimeEventRecord {
3435	schema_version: 1,
3436	seq: 9,
3437	timestamp: chrono::Utc::now(),
3438	thread_id: "thr_test".to_string(),
3439	turn_id: Some("turn_test".to_string()),
3440	item_id: None,
3441	event: "item.delta".to_string(),
3442	payload: json!({
3443	"kind": "context_compaction",
3444	"delta": "ignored",
3445	}),
3446	};
3447	assert!(map_compat_stream_event(&unknown).is_none());
3448	Ok(())
3449	}
3450
3451	#[tokio::test]
3452	async fn stream_endpoint_remains_backward_compatible() -> Result<()> {
3453	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
3454	return Ok(());
3455	};
3456	let client = crate::tls::reqwest_client();
3457
3458	// Create a thread and install a mock engine so /v1/stream doesn't call the real API.
3459	let created: serde_json::Value = client
3460	.post(format!("http://{addr}/v1/threads"))
3461	.json(&json!({}))
3462	.send()
3463	.await?
3464	.error_for_status()?
3465	.json()
3466	.await?;
3467	let thread_id = created["id"]
3468	.as_str()
3469	.context("missing thread id")?
3470	.to_string();
3471
3472	let harness = crate::core::engine::mock_engine_handle();
3473	runtime_threads
3474	.install_test_engine(&thread_id, harness.handle.clone())
3475	.await?;
3476	let mut rx_op = harness.rx_op;
3477	let tx_event = harness.tx_event;
3478	tokio::spawn(async move {
3479	if !matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
3480	return;
3481	}
3482	let _ = tx_event
3483	.send(EngineEvent::TurnStarted {
3484	turn_id: "mock_stream".to_string(),
3485	created_at: chrono::Utc::now(),
3486	route: None,
3487	})
3488	.await;
3489	let _ = tx_event
3490	.send(EngineEvent::MessageStarted { index: 0 })
3491	.await;
3492	let _ = tx_event
3493	.send(EngineEvent::MessageDelta {
3494	index: 0,
3495	content: "streamed".to_string(),
3496	})
3497	.await;
3498	let _ = tx_event
3499	.send(EngineEvent::MessageComplete { index: 0 })
3500	.await;
3501	let _ = tx_event
3502	.send(EngineEvent::TurnComplete {
3503	usage: Usage {
3504	input_tokens: 4,
3505	output_tokens: 2,
3506	..Usage::default()
3507	},
3508	status: TurnOutcomeStatus::Completed,
3509	error: None,
3510	tool_catalog: None,
3511	base_url: None,
3512	})
3513	.await;
3514	});
3515
3516	// Start the turn and consume events via the SSE endpoint.
3517	let turn_start: serde_json::Value = client
3518	.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
3519	.json(&json!({ "prompt": "compatibility stream" }))
3520	.send()
3521	.await?
3522	.error_for_status()?
3523	.json()
3524	.await?;
3525	let turn_id = turn_start["turn"]["id"]
3526	.as_str()
3527	.context("missing turn id")?
3528	.to_string();
3529
3530	let _ =
3531	wait_for_terminal_turn_status(&client, addr, &thread_id, &turn_id, Duration::from_secs(2))
3532	.await?;
3533
3534	// Verify that the persisted events include the expected turn lifecycle events.
3535	let events = runtime_threads.events_since(&thread_id, None)?;
3536	assert!(
3537	events.iter().any(\|ev\| ev.event == "turn.started"),
3538	"expected turn.started event"
3539	);
3540	assert!(
3541	events.iter().any(\|ev\| ev.event == "turn.completed"),
3542	"expected turn.completed event"
3543	);
3544
3545	// Verify the SSE endpoint returns event-stream content type.
3546	let events_resp = client
3547	.get(format!(
3548	"http://{addr}/v1/threads/{thread_id}/events?since_seq=0"
3549	))
3550	.send()
3551	.await?
3552	.error_for_status()?;
3553	let content_type = events_resp
3554	.headers()
3555	.get(reqwest::header::CONTENT_TYPE)
3556	.and_then(\|v\| v.to_str().ok())
3557	.unwrap_or_default()
3558	.to_string();
3559	assert!(content_type.starts_with("text/event-stream"));
3560
3561	handle.abort();
3562	Ok(())
3563	}
3564
3565	#[tokio::test]
3566	async fn session_get_returns_404_for_missing_id() -> Result<()> {
3567	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
3568	return Ok(());
3569	};
3570	let client = crate::tls::reqwest_client();
3571
3572	let resp = client
3573	.get(format!("http://{addr}/v1/sessions/nonexistent_id"))
3574	.send()
3575	.await?;
3576	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3577
3578	handle.abort();
3579	Ok(())
3580	}
3581
3582	#[tokio::test]
3583	async fn session_endpoints_reject_invalid_id() -> Result<()> {
3584	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
3585	return Ok(());
3586	};
3587	let client = crate::tls::reqwest_client();
3588
3589	let get_resp = client
3590	.get(format!("http://{addr}/v1/sessions/invalid%20id"))
3591	.send()
3592	.await?;
3593	assert_eq!(get_resp.status(), StatusCode::BAD_REQUEST);
3594
3595	let resume_resp = client
3596	.post(format!(
3597	"http://{addr}/v1/sessions/invalid%20id/resume-thread"
3598	))
3599	.json(&json!({}))
3600	.send()
3601	.await?;
3602	assert_eq!(resume_resp.status(), StatusCode::BAD_REQUEST);
3603
3604	let delete_resp = client
3605	.delete(format!("http://{addr}/v1/sessions/invalid%20id"))
3606	.send()
3607	.await?;
3608	assert_eq!(delete_resp.status(), StatusCode::BAD_REQUEST);
3609
3610	handle.abort();
3611	Ok(())
3612	}
3613
3614	#[tokio::test]
3615	async fn session_resume_thread_returns_404_for_missing_session() -> Result<()> {
3616	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
3617	return Ok(());
3618	};
3619	let client = crate::tls::reqwest_client();
3620
3621	let resp = client
3622	.post(format!(
3623	"http://{addr}/v1/sessions/nonexistent_session/resume-thread"
3624	))
3625	.json(&json!({}))
3626	.send()
3627	.await?;
3628	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3629
3630	handle.abort();
3631	Ok(())
3632	}
3633
3634	#[tokio::test]
3635	async fn session_resume_thread_returns_400_when_saved_custom_provider_was_removed() -> Result<()> {
3636	let root = std::env::temp_dir().join(format!(
3637	"codewhale-session-removed-provider-{}",
3638	Uuid::new_v4()
3639	));
3640	let sessions_dir = root.join("sessions");
3641	fs::create_dir_all(&sessions_dir)?;
3642	let session = json!({
3643	"schema_version": 1,
3644	"metadata": {
3645	"id": "sess_removed_custom_provider",
3646	"title": "Removed custom provider",
3647	"created_at": "2025-01-01T00:00:00Z",
3648	"updated_at": "2025-01-01T00:10:00Z",
3649	"message_count": 1,
3650	"total_tokens": 10,
3651	"model": "local-code-model",
3652	"model_provider": "lm-studio",
3653	"workspace": "/tmp/test",
3654	"mode": "agent"
3655	},
3656	"messages": [{
3657	"role": "user",
3658	"content": [{ "type": "text", "text": "Resume me" }]
3659	}],
3660	"system_prompt": null
3661	});
3662	fs::write(
3663	sessions_dir.join("sess_removed_custom_provider.json"),
3664	serde_json::to_string_pretty(&session)?,
3665	)?;
3666
3667	let Some((addr, _runtime_threads, handle)) =
3668	spawn_test_server_with_root(root, sessions_dir).await?
3669	else {
3670	return Ok(());
3671	};
3672	let client = crate::tls::reqwest_client();
3673	let resp = client
3674	.post(format!(
3675	"http://{addr}/v1/sessions/sess_removed_custom_provider/resume-thread"
3676	))
3677	.json(&json!({}))
3678	.send()
3679	.await?;
3680
3681	assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
3682	let body: serde_json::Value = resp.json().await?;
3683	let message = body["error"]["message"].as_str().unwrap_or_default();
3684	assert!(message.contains("[providers.lm-studio]"), "{message}");
3685	assert!(message.contains("will not fall back"), "{message}");
3686
3687	handle.abort();
3688	Ok(())
3689	}
3690
3691	#[tokio::test]
3692	async fn session_resume_thread_creates_thread_from_saved_session() -> Result<()> {
3693	let root = std::env::temp_dir().join(format!("deepseek-session-resume-{}", Uuid::new_v4()));
3694	let sessions_dir = root.join("sessions");
3695	fs::create_dir_all(&sessions_dir)?;
3696	let session = json!({
3697	"schema_version": 1,
3698	"metadata": {
3699	"id": "sess_test_resume",
3700	"title": "Test resume session",
3701	"created_at": "2025-01-01T00:00:00Z",
3702	"updated_at": "2025-01-01T00:10:00Z",
3703	"message_count": 2,
3704	"total_tokens": 100,
3705	"model": "deepseek-v4-pro",
3706	"workspace": "/tmp/test",
3707	"mode": "agent"
3708	},
3709	"messages": [
3710	{
3711	"role": "user",
3712	"content": [{ "type": "text", "text": "Hello, world!" }]
3713	},
3714	{
3715	"role": "assistant",
3716	"content": [{ "type": "text", "text": "Hello! How can I help you?" }]
3717	}
3718	],
3719	"system_prompt": null
3720	});
3721	fs::write(
3722	sessions_dir.join("sess_test_resume.json"),
3723	serde_json::to_string_pretty(&session)?,
3724	)?;
3725
3726	let Some((addr, _runtime_threads, handle)) =
3727	spawn_test_server_with_root(root.clone(), sessions_dir.clone()).await?
3728	else {
3729	return Ok(());
3730	};
3731	let client = crate::tls::reqwest_client();
3732
3733	let resp = client
3734	.post(format!(
3735	"http://{addr}/v1/sessions/sess_test_resume/resume-thread"
3736	))
3737	.json(&json!({ "model": "deepseek-v4-pro" }))
3738	.send()
3739	.await?;
3740	assert_eq!(resp.status(), StatusCode::CREATED);
3741	let resumed: serde_json::Value = resp.json().await?;
3742	assert_eq!(resumed["session_id"], "sess_test_resume");
3743	assert_eq!(resumed["message_count"], 2);
3744
3745	let thread_id = resumed["thread_id"]
3746	.as_str()
3747	.context("missing resumed thread id")?;
3748	let detail: serde_json::Value = client
3749	.get(format!("http://{addr}/v1/threads/{thread_id}"))
3750	.send()
3751	.await?
3752	.error_for_status()?
3753	.json()
3754	.await?;
3755	assert_eq!(detail["thread"]["id"], thread_id);
3756	assert_eq!(detail["thread"]["model_provider"], "deepseek");
3757	assert_eq!(detail["thread"]["workspace"], "/tmp/test");
3758	assert_eq!(detail["turns"].as_array().map_or(0, Vec::len), 1);
3759	assert_eq!(detail["items"].as_array().map_or(0, Vec::len), 2);
3760
3761	handle.abort();
3762	Ok(())
3763	}
3764
3765	#[tokio::test]
3766	async fn session_create_from_completed_thread_saves_messages() -> Result<()> {
3767	let root = std::env::temp_dir().join(format!("deepseek-thread-session-{}", Uuid::new_v4()));
3768	let sessions_dir = root.join("sessions");
3769	let Some((addr, runtime_threads, handle)) =
3770	spawn_test_server_with_root(root.clone(), sessions_dir).await?
3771	else {
3772	return Ok(());
3773	};
3774	let client = crate::tls::reqwest_client();
3775
3776	let created: serde_json::Value = client
3777	.post(format!("http://{addr}/v1/threads"))
3778	.json(&json!({
3779	"model": "deepseek-v4-pro",
3780	"mode": "plan",
3781	"workspace": root.join("workspace")
3782	}))
3783	.send()
3784	.await?
3785	.error_for_status()?
3786	.json()
3787	.await?;
3788	let thread_id = created["id"]
3789	.as_str()
3790	.context("missing thread id")?
3791	.to_string();
3792
3793	let patched: serde_json::Value = client
3794	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
3795	.json(&json!({ "title": "Thread title fallback" }))
3796	.send()
3797	.await?
3798	.error_for_status()?
3799	.json()
3800	.await?;
3801	assert_eq!(patched["title"], "Thread title fallback");
3802
3803	runtime_threads
3804	.seed_thread_from_messages(
3805	&thread_id,
3806	&[
3807	Message {
3808	role: "user".to_string(),
3809	content: vec![ContentBlock::Text {
3810	text: "Please save this runtime thread".to_string(),
3811	cache_control: None,
3812	}],
3813	},
3814	Message {
3815	role: "assistant".to_string(),
3816	content: vec![ContentBlock::Text {
3817	text: "Saved replies should round-trip.".to_string(),
3818	cache_control: None,
3819	}],
3820	},
3821	],
3822	)
3823	.await?;
3824
3825	let resp = client
3826	.post(format!("http://{addr}/v1/sessions"))
3827	.json(&json!({ "thread_id": thread_id }))
3828	.send()
3829	.await?;
3830	assert_eq!(resp.status(), StatusCode::CREATED);
3831	let saved: serde_json::Value = resp.json().await?;
3832	assert_eq!(saved["thread_id"], thread_id);
3833	assert_eq!(saved["message_count"], 2);
3834	assert_eq!(saved["title"], "Thread title fallback");
3835	let saved_session_handle = saved["session_id"]
3836	.as_str()
3837	.context("missing session id")?
3838	.to_string();
3839
3840	let session_manager = crate::session_manager::SessionManager::new(root.join("sessions"))?;
3841	let created_session = session_manager.load_session_by_prefix(&saved_session_handle)?;
3842	assert_eq!(created_session.metadata.title, "Thread title fallback");
3843	assert_eq!(created_session.metadata.model, "deepseek-v4-pro");
3844	assert_eq!(created_session.metadata.mode.as_deref(), Some("plan"));
3845	assert_eq!(created_session.metadata.message_count, 2);
3846	assert_eq!(created_session.messages[0].role, "user");
3847	assert_eq!(created_session.messages[1].role, "assistant");
3848
3849	let mut endpoint_session = crate::session_manager::create_saved_session_with_id_and_mode(
3850	"sess_endpoint_fetch".to_string(),
3851	&created_session.messages,
3852	"deepseek-v4-pro",
3853	&root,
3854	0,
3855	None,
3856	Some("plan"),
3857	);
3858	endpoint_session.metadata.title = "Thread title fallback".to_string();
3859	session_manager.save_session(&endpoint_session)?;
3860
3861	let detail: serde_json::Value = client
3862	.get(format!("http://{addr}/v1/sessions/sess_endpoint_fetch"))
3863	.send()
3864	.await?
3865	.error_for_status()?
3866	.json()
3867	.await?;
3868	assert_eq!(detail["metadata"]["title"], "Thread title fallback");
3869	assert_eq!(detail["metadata"]["model"], "deepseek-v4-pro");
3870	assert_eq!(detail["metadata"]["mode"], "plan");
3871	assert_eq!(detail["metadata"]["message_count"], 2);
3872	assert_eq!(detail["messages"][0]["role"], "user");
3873	assert_eq!(
3874	detail["messages"][0]["content"][0]["text"],
3875	"Please save this runtime thread"
3876	);
3877	assert_eq!(detail["messages"][1]["role"], "assistant");
3878
3879	let manual_title: serde_json::Value = client
3880	.post(format!("http://{addr}/v1/sessions"))
3881	.json(&json!({
3882	"thread_id": thread_id,
3883	"title": "Manual saved title"
3884	}))
3885	.send()
3886	.await?
3887	.error_for_status()?
3888	.json()
3889	.await?;
3890	assert_eq!(manual_title["title"], "Manual saved title");
3891	assert_ne!(manual_title["session_id"], saved_session_handle);
3892
3893	handle.abort();
3894	Ok(())
3895	}
3896
3897	#[tokio::test]
3898	async fn session_create_from_thread_returns_404_for_missing_thread() -> Result<()> {
3899	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
3900	return Ok(());
3901	};
3902	let client = crate::tls::reqwest_client();
3903
3904	let resp = client
3905	.post(format!("http://{addr}/v1/sessions"))
3906	.json(&json!({ "thread_id": "thr_missing" }))
3907	.send()
3908	.await?;
3909	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3910
3911	handle.abort();
3912	Ok(())
3913	}
3914
3915	/// Create a thread over HTTP and seed it with one user/assistant turn.
3916	/// Shared setup for the undo/patch-undo/retry endpoint tests.
3917	async fn create_seeded_thread(
3918	addr: &SocketAddr,
3919	runtime_threads: &SharedRuntimeThreadManager,
3920	root: &FsPath,
3921	user_text: &str,
3922	) -> Result<String> {
3923	let client = crate::tls::reqwest_client();
3924	let created: serde_json::Value = client
3925	.post(format!("http://{addr}/v1/threads"))
3926	.json(&json!({
3927	"model": "deepseek-v4-pro",
3928	"mode": "agent",
3929	"workspace": root.join("workspace")
3930	}))
3931	.send()
3932	.await?
3933	.error_for_status()?
3934	.json()
3935	.await?;
3936	let thread_id = created["id"]
3937	.as_str()
3938	.context("missing thread id")?
3939	.to_string();
3940
3941	runtime_threads
3942	.seed_thread_from_messages(
3943	&thread_id,
3944	&[
3945	Message {
3946	role: "user".to_string(),
3947	content: vec![ContentBlock::Text {
3948	text: user_text.to_string(),
3949	cache_control: None,
3950	}],
3951	},
3952	Message {
3953	role: "assistant".to_string(),
3954	content: vec![ContentBlock::Text {
3955	text: "Done — anything else?".to_string(),
3956	cache_control: None,
3957	}],
3958	},
3959	],
3960	)
3961	.await?;
3962	Ok(thread_id)
3963	}
3964
3965	#[tokio::test]
3966	async fn undo_endpoint_forks_thread_and_returns_original_user_text() -> Result<()> {
3967	let root = std::env::temp_dir().join(format!("deepseek-undo-endpoint-{}", Uuid::new_v4()));
3968	let sessions_dir = root.join("sessions");
3969	let Some((addr, runtime_threads, handle)) =
3970	spawn_test_server_with_root(root.clone(), sessions_dir).await?
3971	else {
3972	return Ok(());
3973	};
3974	let thread_id =
3975	create_seeded_thread(&addr, &runtime_threads, &root, "Please undo this turn").await?;
3976	let client = crate::tls::reqwest_client();
3977
3978	let resp = client
3979	.post(format!("http://{addr}/v1/threads/{thread_id}/undo"))
3980	.json(&json!({}))
3981	.send()
3982	.await?;
3983	assert_eq!(resp.status(), StatusCode::CREATED);
3984	let undone: serde_json::Value = resp.json().await?;
3985	assert_eq!(undone["original_user_text"], "Please undo this turn");
3986	let forked_id = undone["thread"]["id"]
3987	.as_str()
3988	.context("missing forked thread id")?;
3989	assert_ne!(forked_id, thread_id, "undo must fork, not mutate in place");
3990
3991	// The forked thread has the undone turn removed.
3992	let detail: serde_json::Value = client
3993	.get(format!("http://{addr}/v1/threads/{forked_id}"))
3994	.send()
3995	.await?
3996	.error_for_status()?
3997	.json()
3998	.await?;
3999	assert_eq!(detail["turns"].as_array().map_or(usize::MAX, Vec::len), 0);
4000
4001	handle.abort();
4002	Ok(())
4003	}
4004
4005	#[tokio::test]
4006	async fn undo_endpoint_404s_for_missing_thread() -> Result<()> {
4007	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
4008	return Ok(());
4009	};
4010	let client = crate::tls::reqwest_client();
4011	let resp = client
4012	.post(format!("http://{addr}/v1/threads/thr_missing/undo"))
4013	.json(&json!({}))
4014	.send()
4015	.await?;
4016	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
4017	handle.abort();
4018	Ok(())
4019	}
4020
4021	#[tokio::test]
4022	async fn patch_undo_endpoint_forks_and_reports_file_rollback_state() -> Result<()> {
4023	let root =
4024	std::env::temp_dir().join(format!("deepseek-patch-undo-endpoint-{}", Uuid::new_v4()));
4025	let sessions_dir = root.join("sessions");
4026	let Some((addr, runtime_threads, handle)) =
4027	spawn_test_server_with_root(root.clone(), sessions_dir).await?
4028	else {
4029	return Ok(());
4030	};
4031	let thread_id =
4032	create_seeded_thread(&addr, &runtime_threads, &root, "Roll back the patch").await?;
4033	let client = crate::tls::reqwest_client();
4034
4035	let resp = client
4036	.post(format!("http://{addr}/v1/threads/{thread_id}/patch-undo"))
4037	.json(&json!({}))
4038	.send()
4039	.await?;
4040	assert_eq!(resp.status(), StatusCode::CREATED);
4041	let undone: serde_json::Value = resp.json().await?;
4042	// The fresh workspace has no tool/pre-turn snapshots to roll back to,
4043	// so the file-restore step reports failure while the conversation
4044	// undo still forks the thread.
4045	assert_eq!(undone["patch_result"]["files_restored"], false);
4046	assert!(undone["patch_result"]["summary"].is_string());
4047	assert_eq!(undone["original_user_text"], "Roll back the patch");
4048	assert_ne!(undone["thread"]["id"].as_str(), Some(thread_id.as_str()));
4049
4050	handle.abort();
4051	Ok(())
4052	}
4053
4054	#[test]
4055	fn patch_undo_helper_restores_only_the_bound_session() -> Result<()> {
4056	let _lock = lock_test_env();
4057	let root = tempfile::tempdir()?;
4058	let home = root.path().join("home");
4059	fs::create_dir_all(&home)?;
4060	let _home = EnvVarGuard::set("HOME", &home);
4061
4062	let workspace = root.path().join("workspace");
4063	fs::create_dir_all(&workspace)?;
4064	let repo = crate::snapshot::SnapshotRepo::open_or_init(&workspace)?;
4065	let file = workspace.join("a.txt");
4066
4067	fs::write(&file, "legacy")?;
4068	repo.snapshot("pre-turn:legacy")?;
4069	fs::write(&file, "current-before")?;
4070	repo.snapshot_with_session("pre-turn:current", Some("session-current"))?;
4071	fs::write(&file, "foreign-before")?;
4072	repo.snapshot_with_session("pre-turn:foreign", Some("session-foreign"))?;
4073	fs::write(&file, "current-after")?;
4074
4075	let restored = patch_undo_workspace_files(&workspace, Some("session-current"));
4076	assert!(restored.files_restored, "{:?}", restored.summary);
4077	assert_eq!(fs::read_to_string(&file)?, "current-before");
4078
4079	fs::write(&file, "must-stay")?;
4080	let unbound = patch_undo_workspace_files(&workspace, None);
4081	assert!(!unbound.files_restored);
4082	assert_eq!(fs::read_to_string(&file)?, "must-stay");
4083	Ok(())
4084	}
4085
4086	#[tokio::test]
4087	async fn retry_endpoint_reuses_dropped_user_text_to_start_a_turn() -> Result<()> {
4088	let root = std::env::temp_dir().join(format!("deepseek-retry-endpoint-{}", Uuid::new_v4()));
4089	let sessions_dir = root.join("sessions");
4090	let Some((addr, runtime_threads, handle)) =
4091	spawn_test_server_with_root(root.clone(), sessions_dir).await?
4092	else {
4093	return Ok(());
4094	};
4095	let thread_id =
4096	create_seeded_thread(&addr, &runtime_threads, &root, "Retry this request").await?;
4097	let client = crate::tls::reqwest_client();
4098
4099	let resp = client
4100	.post(format!("http://{addr}/v1/threads/{thread_id}/retry"))
4101	.json(&json!({}))
4102	.send()
4103	.await?;
4104	assert_eq!(resp.status(), StatusCode::CREATED);
4105	let retried: serde_json::Value = resp.json().await?;
4106	let forked_id = retried["thread"]["id"]
4107	.as_str()
4108	.context("missing forked thread id")?;
4109	assert_ne!(forked_id, thread_id);
4110	assert_eq!(retried["turn"]["thread_id"], forked_id);
4111
4112	handle.abort();
4113	Ok(())
4114	}
4115
4116	#[test]
4117	fn restore_snapshot_endpoint_helper_restores_workspace_files() -> Result<()> {
4118	let _lock = lock_test_env();
4119	let root = tempfile::tempdir()?;
4120	let home = root.path().join("home");
4121	fs::create_dir_all(&home)?;
4122	let _home = EnvVarGuard::set("HOME", &home);
4123
4124	let workspace = root.path().join("workspace");
4125	fs::create_dir_all(&workspace)?;
4126	let repo = crate::snapshot::SnapshotRepo::open_or_init(&workspace)?;
4127	fs::write(workspace.join("a.txt"), "v1")?;
4128	let snapshot_id = repo.snapshot("pre-turn:1")?;
4129	fs::write(workspace.join("a.txt"), "v2")?;
4130
4131	restore_snapshot_for_workspace(&workspace, snapshot_id.as_str())
4132	.expect("snapshot restore should succeed");
4133	assert_eq!(fs::read_to_string(workspace.join("a.txt"))?, "v1");
4134	Ok(())
4135	}
4136
4137	#[tokio::test]
4138	async fn session_create_from_thread_rejects_active_turn() -> Result<()> {
4139	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
4140	return Ok(());
4141	};
4142	let client = crate::tls::reqwest_client();
4143
4144	let created: serde_json::Value = client
4145	.post(format!("http://{addr}/v1/threads"))
4146	.json(&json!({}))
4147	.send()
4148	.await?
4149	.error_for_status()?
4150	.json()
4151	.await?;
4152	let thread_id = created["id"]
4153	.as_str()
4154	.context("missing thread id")?
4155	.to_string();
4156
4157	let harness = crate::core::engine::mock_engine_handle();
4158	runtime_threads
4159	.install_test_engine(&thread_id, harness.handle.clone())
4160	.await?;
4161	let mut rx_op = harness.rx_op;
4162	let tx_event = harness.tx_event;
4163	let (active_tx, active_rx) = oneshot::channel();
4164	let (finish_tx, finish_rx) = oneshot::channel();
4165	tokio::spawn(async move {
4166	if !matches!(rx_op.recv().await, Some(Op::SendMessage { .. })) {
4167	return;
4168	}
4169	let _ = tx_event
4170	.send(EngineEvent::TurnStarted {
4171	turn_id: "mock_active_session_save".to_string(),
4172	created_at: chrono::Utc::now(),
4173	route: None,
4174	})
4175	.await;
4176	let _ = tx_event
4177	.send(EngineEvent::MessageStarted { index: 0 })
4178	.await;
4179	let _ = active_tx.send(());
4180	let _ = finish_rx.await;
4181	let _ = tx_event
4182	.send(EngineEvent::MessageDelta {
4183	index: 0,
4184	content: "now complete".to_string(),
4185	})
4186	.await;
4187	let _ = tx_event
4188	.send(EngineEvent::MessageComplete { index: 0 })
4189	.await;
4190	let _ = tx_event
4191	.send(EngineEvent::TurnComplete {
4192	usage: Usage {
4193	input_tokens: 2,
4194	output_tokens: 1,
4195	..Usage::default()
4196	},
4197	status: TurnOutcomeStatus::Completed,
4198	error: None,
4199	tool_catalog: None,
4200	base_url: None,
4201	})
4202	.await;
4203	});
4204
4205	let started: serde_json::Value = client
4206	.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
4207	.json(&json!({ "prompt": "save me while active" }))
4208	.send()
4209	.await?
4210	.error_for_status()?
4211	.json()
4212	.await?;
4213	let turn_id = started["turn"]["id"]
4214	.as_str()
4215	.context("missing turn id")?
4216	.to_string();
4217	tokio::time::timeout(ci_scaled(Duration::from_secs(2)), active_rx)
4218	.await
4219	.context("timed out waiting for mock active turn")?
4220	.context("mock active turn sender dropped")?;
4221	wait_for_in_progress_item(&client, addr, &thread_id, Duration::from_secs(2)).await?;
4222
4223	let resp = client
4224	.post(format!("http://{addr}/v1/sessions"))
4225	.json(&json!({ "thread_id": thread_id }))
4226	.send()
4227	.await?;
4228	assert_eq!(resp.status(), StatusCode::CONFLICT);
4229	let body: serde_json::Value = resp.json().await?;
4230	assert!(
4231	body["error"]["message"]
4232	.as_str()
4233	.is_some_and(\|message\| message.contains("queued or active turn"))
4234	);
4235
4236	let _ = finish_tx.send(());
4237	let terminal =
4238	wait_for_terminal_turn_status(&client, addr, &thread_id, &turn_id, Duration::from_secs(2))
4239	.await?;
4240	assert_eq!(terminal, "completed");
4241
4242	handle.abort();
4243	Ok(())
4244	}
4245
4246	#[test]
4247	fn snapshots_endpoint_lists_workspace_snapshots() -> Result<()> {
4248	let _lock = lock_test_env();
4249	let root = tempfile::tempdir()?;
4250	let home = root.path().join("home");
4251	fs::create_dir_all(&home)?;
4252	let _home = EnvVarGuard::set("HOME", &home);
4253
4254	let workspace = root.path().join("workspace");
4255	fs::create_dir_all(&workspace)?;
4256	let repo = crate::snapshot::SnapshotRepo::open_or_init(&workspace)?;
4257	fs::write(workspace.join("a.txt"), "v1")?;
4258	repo.snapshot("pre-turn:1")?;
4259	fs::write(workspace.join("a.txt"), "v2")?;
4260	repo.snapshot("post-turn:1")?;
4261
4262	let snapshots = snapshot_entries_for_workspace(&workspace, SnapshotsQuery { limit: Some(1) })
4263	.expect("snapshot listing should succeed");
4264	assert_eq!(snapshots.len(), 1);
4265	assert_eq!(snapshots[0].label, "post-turn:1");
4266	assert!(snapshots[0].id.len() >= 8);
4267	assert!(snapshots[0].timestamp > 0);
4268
4269	let bad_limit = snapshot_entries_for_workspace(&workspace, SnapshotsQuery { limit: Some(101) })
4270	.expect_err("limit above cap should fail");
4271	assert_eq!(bad_limit.status, StatusCode::BAD_REQUEST);
4272	Ok(())
4273	}
4274
4275	/// Seed a sessions directory and start a server against it.
4276	///
4277	/// The route tests below need to control what is on disk, so they cannot use
4278	/// the `spawn_test_server()` convenience that hides its own temp paths.
4279	async fn spawn_server_with_saved_sessions(
4280	sessions: &[(&str, &str, bool)],
4281	) -> Result<Option<(SocketAddr, PathBuf, tokio::task::JoinHandle<()>)>> {
4282	let root = std::env::temp_dir().join(format!("codewhale-session-routes-{}", Uuid::new_v4()));
4283	let sessions_dir = root.join("sessions");
4284	let workspace = root.join("workspace");
4285	fs::create_dir_all(&workspace)?;
4286	let manager = crate::session_manager::SessionManager::new(sessions_dir.clone())?;
4287	for (id, title, archived) in sessions {
4288	let mut saved = crate::session_manager::create_saved_session_with_id_and_mode(
4289	(*id).to_string(),
4290	&[
4291	crate::models::Message {
4292	role: "user".to_string(),
4293	content: vec![crate::models::ContentBlock::Text {
4294	text: format!("prompt for {title} with token=hunter2"),
4295	cache_control: None,
4296	}],
4297	},
4298	crate::models::Message {
4299	role: "assistant".to_string(),
4300	content: vec![crate::models::ContentBlock::Text {
4301	text: "acknowledged".to_string(),
4302	cache_control: None,
4303	}],
4304	},
4305	],
4306	"deepseek-chat",
4307	&workspace,
4308	12,
4309	None,
4310	Some("agent"),
4311	);
4312	saved.metadata.title = (*title).to_string();
4313	saved.metadata.archived = *archived;
4314	manager.save_session(&saved)?;
4315	}
4316	let Some((addr, _threads, handle)) =
4317	spawn_test_server_with_root(root, sessions_dir.clone()).await?
4318	else {
4319	return Ok(None);
4320	};
4321	Ok(Some((addr, sessions_dir, handle)))
4322	}
4323
4324	/// `/v1/sessions/summary` is routed, projects the shared row shape, and hides
4325	/// archived rows until asked — the dashboard's whole session list depends on
4326	/// all three being true at once.
4327	#[tokio::test]
4328	async fn session_summary_route_projects_rows_and_honours_archive_filters() -> Result<()> {
4329	let Some((addr, _dir, handle)) = spawn_server_with_saved_sessions(&[
4330	("sess-active", "Active work", false),
4331	("sess-putaway", "Put away", true),
4332	])
4333	.await?
4334	else {
4335	return Ok(());
4336	};
4337	let client = crate::tls::reqwest_client();
4338
4339	let active: Vec<serde_json::Value> = client
4340	.get(format!("http://{addr}/v1/sessions/summary"))
4341	.send()
4342	.await?
4343	.error_for_status()?
4344	.json()
4345	.await?;
4346	assert_eq!(
4347	active
4348	.iter()
4349	.map(\|r\| r["id"].as_str().unwrap())
4350	.collect::<Vec<_>>(),
4351	vec!["sess-active"],
4352	"archived sessions must not appear in the default listing"
4353	);
4354	// Field-compatible with /v1/threads/summary — the dashboard renders both
4355	// with one row renderer.
4356	for field in [
4357	"title",
4358	"preview",
4359	"model",
4360	"mode",
4361	"workspace",
4362	"updated_at",
4363	] {
4364	assert!(!active[0][field].is_null(), "summary row missing {field}");
4365	}
4366	assert_eq!(active[0]["preview"], active[0]["title"]);
4367
4368	let archived: Vec<serde_json::Value> = client
4369	.get(format!(
4370	"http://{addr}/v1/sessions/summary?archived_only=true"
4371	))
4372	.send()
4373	.await?
4374	.error_for_status()?
4375	.json()
4376	.await?;
4377	assert_eq!(
4378	archived
4379	.iter()
4380	.map(\|r\| r["id"].as_str().unwrap())
4381	.collect::<Vec<_>>(),
4382	vec!["sess-putaway"]
4383	);
4384
4385	handle.abort();
4386	Ok(())
4387	}
4388
4389	/// `PATCH /v1/sessions/{id}` renames and archives through the one writer, and
4390	/// reports only what actually moved.
4391	#[tokio::test]
4392	async fn session_patch_route_renames_archives_and_reports_real_changes() -> Result<()> {
4393	let Some((addr, sessions_dir, handle)) =
4394	spawn_server_with_saved_sessions(&[("sess-patch", "Before", false)]).await?
4395	else {
4396	return Ok(());
4397	};
4398	let client = crate::tls::reqwest_client();
4399
4400	let patched: serde_json::Value = client
4401	.patch(format!("http://{addr}/v1/sessions/sess-patch"))
4402	.json(&json!({ "title": "After", "archived": true }))
4403	.send()
4404	.await?
4405	.error_for_status()?
4406	.json()
4407	.await?;
4408	assert_eq!(patched["session"]["title"], "After");
4409	assert_eq!(patched["session"]["archived"], true);
4410	assert_eq!(patched["changes"]["title"], "After");
4411	assert_eq!(patched["changes"]["archived"], true);
4412
4413	// Durable, not just echoed back.
4414	let manager = crate::session_manager::SessionManager::new(sessions_dir)?;
4415	let reloaded = manager.load_session("sess-patch")?;
4416	assert_eq!(reloaded.metadata.title, "After");
4417	assert!(reloaded.metadata.archived);
4418
4419	// A re-patch to the same state changes nothing, and says so.
4420	let repeat: serde_json::Value = client
4421	.patch(format!("http://{addr}/v1/sessions/sess-patch"))
4422	.json(&json!({ "archived": true }))
4423	.send()
4424	.await?
4425	.error_for_status()?
4426	.json()
4427	.await?;
4428	assert!(
4429	repeat["changes"]
4430	.as_object()
4431	.expect("changes object")
4432	.is_empty(),
4433	"a no-op patch must report no changes"
4434	);
4435
4436	// An empty body is a client error, not a silent no-op.
4437	let empty = client
4438	.patch(format!("http://{addr}/v1/sessions/sess-patch"))
4439	.json(&json!({}))
4440	.send()
4441	.await?;
4442	assert_eq!(empty.status(), StatusCode::BAD_REQUEST);
4443
4444	// A blank title is rejected with the reason, not accepted.
4445	let blank = client
4446	.patch(format!("http://{addr}/v1/sessions/sess-patch"))
4447	.json(&json!({ "title": " " }))
4448	.send()
4449	.await?;
4450	assert_eq!(blank.status(), StatusCode::BAD_REQUEST);
4451
4452	handle.abort();
4453	Ok(())
4454	}
4455
4456	/// A session the TUI holds open is refused with a typed 409 rather than
4457	/// written behind its back.
4458	#[tokio::test]
4459	async fn session_patch_route_refuses_a_live_session_with_a_conflict() -> Result<()> {
4460	// The live-session claim is process-global by construction (the embedded
4461	// API runs inside the TUI process), so this test must not run alongside
4462	// anything else that claims or clears it.
4463	let _lock = lock_test_env();
4464	let Some((addr, _dir, handle)) =
4465	spawn_server_with_saved_sessions(&[("sess-live", "Held open", false)]).await?
4466	else {
4467	return Ok(());
4468	};
4469	let client = crate::tls::reqwest_client();
4470
4471	crate::session_manager::set_live_session(Some("sess-live"));
4472	let conflict = client
4473	.patch(format!("http://{addr}/v1/sessions/sess-live"))
4474	.json(&json!({ "title": "Renamed from the dashboard" }))
4475	.send()
4476	.await?;
4477	assert_eq!(conflict.status(), StatusCode::CONFLICT);
4478
4479	crate::session_manager::set_live_session(None);
4480	let allowed = client
4481	.patch(format!("http://{addr}/v1/sessions/sess-live"))
4482	.json(&json!({ "title": "Renamed from the dashboard" }))
4483	.send()
4484	.await?;
4485	assert_eq!(allowed.status(), StatusCode::OK);
4486
4487	handle.abort();
4488	Ok(())
4489	}
4490
4491	/// `?peek=true` returns the bounded redacted projection, and the plain route
4492	/// still returns the full detail shape.
4493	#[tokio::test]
4494	async fn session_detail_route_serves_a_bounded_redacted_peek_on_request() -> Result<()> {
4495	let Some((addr, _dir, handle)) =
4496	spawn_server_with_saved_sessions(&[("sess-peek", "Peekable", false)]).await?
4497	else {
4498	return Ok(());
4499	};
4500	let client = crate::tls::reqwest_client();
4501
4502	let peek: serde_json::Value = client
4503	.get(format!(
4504	"http://{addr}/v1/sessions/sess-peek?peek=true&entries=12"
4505	))
4506	.send()
4507	.await?
4508	.error_for_status()?
4509	.json()
4510	.await?;
4511	assert_eq!(peek["session_id"], "sess-peek");
4512	assert_eq!(peek["live"], false);
4513	assert!(peek["entries"].as_array().expect("entries").len() <= 12);
4514	// The seeded prompt carries `token=hunter2`; a peek must not re-emit it.
4515	let body = peek.to_string();
4516	assert!(
4517	!body.contains("hunter2"),
4518	"peek leaked a credential: {body}"
4519	);
4520	// No field a client could read as live turn state.
4521	for forbidden in ["status", "running", "active", "turn"] {
4522	assert!(peek.get(forbidden).is_none(), "peek exposed `{forbidden}`");
4523	}
4524
4525	let detail: serde_json::Value = client
4526	.get(format!("http://{addr}/v1/sessions/sess-peek"))
4527	.send()
4528	.await?
4529	.error_for_status()?
4530	.json()
4531	.await?;
4532	assert_eq!(detail["metadata"]["id"], "sess-peek");
4533	assert!(
4534	detail["messages"].is_array(),
4535	"the plain route keeps returning full detail"
4536	);
4537
4538	handle.abort();
4539	Ok(())
4540	}
4541
4542	#[tokio::test]
4543	async fn session_delete_returns_404_for_missing_id() -> Result<()> {
4544	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
4545	return Ok(());
4546	};
4547	let client = crate::tls::reqwest_client();
4548	let resp = client
4549	.delete(format!("http://{addr}/v1/sessions/nonexistent-id"))
4550	.send()
4551	.await?;
4552	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
4553	handle.abort();
4554	Ok(())
4555	}
4556
4557	/// #561 / whalescale#255 — extra CORS origins from `RuntimeApiOptions`
4558	/// are added on top of the built-in defaults and propagate through to the
4559	/// `Access-Control-Allow-Origin` response header for preflight requests.
4560	/// Built-in defaults must keep working unchanged.
4561	#[tokio::test]
4562	async fn cors_layer_appends_extra_origins_and_keeps_defaults() -> Result<()> {
4563	// The cors_layer fn is the layer factory — exercise it through a
4564	// Router with a single trivial route so we can issue OPTIONS preflights
4565	// and observe the response headers.
4566	let extra = vec!["http://localhost:5173".to_string()];
4567	let layer = cors_layer(&extra);
4568	let router: Router = Router::new()
4569	.route("/probe", get(\|\| async { "ok" }))
4570	.layer(layer);
4571
4572	let listener = match TcpListener::bind("127.0.0.1:0").await {
4573	Ok(listener) => listener,
4574	Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return Ok(()),
4575	Err(err) => return Err(err.into()),
4576	};
4577	let addr = listener.local_addr()?;
4578	let handle = tokio::spawn(async move {
4579	let _ = axum::serve(listener, router).await;
4580	});
4581
4582	let client = crate::tls::reqwest_client();
4583
4584	// The user-supplied origin is allowed.
4585	let resp = client
4586	.request(reqwest::Method::OPTIONS, format!("http://{addr}/probe"))
4587	.header("Origin", "http://localhost:5173")
4588	.header("Access-Control-Request-Method", "GET")
4589	.send()
4590	.await?;
4591	assert_eq!(
4592	resp.headers()
4593	.get("access-control-allow-origin")
4594	.and_then(\|v\| v.to_str().ok()),
4595	Some("http://localhost:5173")
4596	);
4597
4598	// A built-in default origin still works.
4599	let resp = client
4600	.request(reqwest::Method::OPTIONS, format!("http://{addr}/probe"))
4601	.header("Origin", "http://localhost:1420")
4602	.header("Access-Control-Request-Method", "GET")
4603	.send()
4604	.await?;
4605	assert_eq!(
4606	resp.headers()
4607	.get("access-control-allow-origin")
4608	.and_then(\|v\| v.to_str().ok()),
4609	Some("http://localhost:1420")
4610	);
4611
4612	// An origin that's neither configured nor a default is rejected
4613	// (CorsLayer omits the Allow-Origin header on mismatch).
4614	let resp = client
4615	.request(reqwest::Method::OPTIONS, format!("http://{addr}/probe"))
4616	.header("Origin", "http://malicious.example")
4617	.header("Access-Control-Request-Method", "GET")
4618	.send()
4619	.await?;
4620	assert!(
4621	resp.headers().get("access-control-allow-origin").is_none(),
4622	"non-allowed origin must not be echoed back"
4623	);
4624
4625	handle.abort();
4626	Ok(())
4627	}
4628
4629	/// #561 — invalid origins (non-ASCII, etc.) are skipped without aborting
4630	/// the layer build.
4631	#[test]
4632	fn cors_layer_skips_invalid_origins() {
4633	let extras = vec![
4634	"http://valid.example".to_string(),
4635	// Embedded NUL char makes `HeaderValue::from_str` fail.
4636	"http://invalid.example\0".to_string(),
4637	" ".to_string(), // whitespace-only is dropped
4638	];
4639	// Should not panic.
4640	let _ = cors_layer(&extras);
4641	}
4642
4643	/// #562 / whalescale#256 — `PATCH /v1/threads/{id}` accepts the new
4644	/// fields (allow_shell, trust_mode, auto_approve, model, mode, title,
4645	/// system_prompt). Legacy mode aliases remain accepted as one-way inputs and
4646	/// the response returns the canonical product mode. Each field is
4647	/// independently optional; an empty string clears `title` / `system_prompt`
4648	/// back to None.
4649	#[tokio::test]
4650	async fn patch_thread_accepts_extended_field_set() -> Result<()> {
4651	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
4652	return Ok(());
4653	};
4654	let client = crate::tls::reqwest_client();
4655
4656	let created: serde_json::Value = client
4657	.post(format!("http://{addr}/v1/threads"))
4658	.json(&json!({
4659	"model": "deepseek-v4-flash",
4660	"mode": "agent"
4661	}))
4662	.send()
4663	.await?
4664	.error_for_status()?
4665	.json()
4666	.await?;
4667	let thread_id = created["id"]
4668	.as_str()
4669	.context("missing thread id")?
4670	.to_string();
4671
4672	// Patch every new field at once.
4673	let patched: serde_json::Value = client
4674	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
4675	.json(&json!({
4676	"allow_shell": true,
4677	"trust_mode": true,
4678	"auto_approve": true,
4679	"model": "deepseek-v4-pro",
4680	"mode": "yolo",
4681	"title": "Whalescale UI test thread",
4682	"system_prompt": "You are a useful assistant."
4683	}))
4684	.send()
4685	.await?
4686	.error_for_status()?
4687	.json()
4688	.await?;
4689
4690	assert_eq!(patched["allow_shell"], true);
4691	assert_eq!(patched["trust_mode"], true);
4692	assert_eq!(patched["auto_approve"], true);
4693	assert_eq!(patched["model"], "deepseek-v4-pro");
4694	assert_eq!(patched["mode"], "agent");
4695	assert_eq!(patched["permission_posture"], "full_access");
4696	assert_eq!(patched["title"], "Whalescale UI test thread");
4697	assert_eq!(patched["system_prompt"], "You are a useful assistant.");
4698
4699	// Empty string clears title back to None.
4700	let cleared: serde_json::Value = client
4701	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
4702	.json(&json!({ "title": "" }))
4703	.send()
4704	.await?
4705	.error_for_status()?
4706	.json()
4707	.await?;
4708	assert!(
4709	cleared["title"].is_null() \|\| !cleared.as_object().unwrap().contains_key("title"),
4710	"empty title must serialize as None: {cleared:?}"
4711	);
4712
4713	// Empty patch (no fields) is still rejected.
4714	let empty = client
4715	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
4716	.json(&json!({}))
4717	.send()
4718	.await?;
4719	assert_eq!(empty.status(), StatusCode::BAD_REQUEST);
4720
4721	// Empty model is rejected (validation).
4722	let bad_model = client
4723	.patch(format!("http://{addr}/v1/threads/{thread_id}"))
4724	.json(&json!({ "model": " " }))
4725	.send()
4726	.await?;
4727	assert_eq!(bad_model.status(), StatusCode::BAD_REQUEST);
4728
4729	handle.abort();
4730	Ok(())
4731	}
4732
4733	/// #563 / whalescale#260 — `archived_only=true` returns archived-only
4734	/// (no active threads), distinct from `include_archived=true` which
4735	/// returns both.
4736	#[tokio::test]
4737	async fn list_threads_archived_only_filter_matches_only_archived() -> Result<()> {
4738	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
4739	return Ok(());
4740	};
4741	let client = crate::tls::reqwest_client();
4742
4743	// Two threads — keep one active, archive the other.
4744	let active: serde_json::Value = client
4745	.post(format!("http://{addr}/v1/threads"))
4746	.json(&json!({}))
4747	.send()
4748	.await?
4749	.error_for_status()?
4750	.json()
4751	.await?;
4752	let active_id = active["id"].as_str().unwrap().to_string();
4753
4754	let archived: serde_json::Value = client
4755	.post(format!("http://{addr}/v1/threads"))
4756	.json(&json!({}))
4757	.send()
4758	.await?
4759	.error_for_status()?
4760	.json()
4761	.await?;
4762	let archived_id = archived["id"].as_str().unwrap().to_string();
4763
4764	client
4765	.patch(format!("http://{addr}/v1/threads/{archived_id}"))
4766	.json(&json!({ "archived": true }))
4767	.send()
4768	.await?
4769	.error_for_status()?;
4770
4771	// Default (active only) → only the unarchived one.
4772	let active_list: serde_json::Value = client
4773	.get(format!("http://{addr}/v1/threads"))
4774	.send()
4775	.await?
4776	.error_for_status()?
4777	.json()
4778	.await?;
4779	let ids: Vec<&str> = active_list
4780	.as_array()
4781	.unwrap()
4782	.iter()
4783	.filter_map(\|t\| t["id"].as_str())
4784	.collect();
4785	assert!(ids.contains(&active_id.as_str()));
4786	assert!(!ids.contains(&archived_id.as_str()));
4787
4788	// archived_only=true → only the archived one.
4789	let archived_list: serde_json::Value = client
4790	.get(format!("http://{addr}/v1/threads?archived_only=true"))
4791	.send()
4792	.await?
4793	.error_for_status()?
4794	.json()
4795	.await?;
4796	let ids: Vec<&str> = archived_list
4797	.as_array()
4798	.unwrap()
4799	.iter()
4800	.filter_map(\|t\| t["id"].as_str())
4801	.collect();
4802	assert_eq!(ids, vec![archived_id.as_str()]);
4803
4804	// archived_only=true takes precedence over include_archived=true.
4805	let archived_list: serde_json::Value = client
4806	.get(format!(
4807	"http://{addr}/v1/threads?include_archived=true&archived_only=true"
4808	))
4809	.send()
4810	.await?
4811	.error_for_status()?
4812	.json()
4813	.await?;
4814	let ids: Vec<&str> = archived_list
4815	.as_array()
4816	.unwrap()
4817	.iter()
4818	.filter_map(\|t\| t["id"].as_str())
4819	.collect();
4820	assert_eq!(ids, vec![archived_id.as_str()]);
4821
4822	// Same filter works on the summary endpoint.
4823	let summary: serde_json::Value = client
4824	.get(format!(
4825	"http://{addr}/v1/threads/summary?archived_only=true&limit=10"
4826	))
4827	.send()
4828	.await?
4829	.error_for_status()?
4830	.json()
4831	.await?;
4832	let summary_ids: Vec<&str> = summary
4833	.as_array()
4834	.unwrap()
4835	.iter()
4836	.filter_map(\|t\| t["id"].as_str())
4837	.collect();
4838	assert_eq!(summary_ids, vec![archived_id.as_str()]);
4839
4840	handle.abort();
4841	Ok(())
4842	}
4843
4844	/// #564 / whalescale#261 — `GET /v1/usage` aggregates per-turn token +
4845	/// cost data. With no threads the response is well-formed and totals are
4846	/// zero with empty buckets (never a 404).
4847	#[tokio::test]
4848	async fn usage_endpoint_returns_empty_aggregation_for_fresh_store() -> Result<()> {
4849	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
4850	return Ok(());
4851	};
4852	let client = crate::tls::reqwest_client();
4853
4854	let body: serde_json::Value = client
4855	.get(format!("http://{addr}/v1/usage"))
4856	.send()
4857	.await?
4858	.error_for_status()?
4859	.json()
4860	.await?;
4861	assert_eq!(body["group_by"], "day");
4862	assert_eq!(body["totals"]["input_tokens"], 0);
4863	assert_eq!(body["totals"]["output_tokens"], 0);
4864	assert_eq!(body["totals"]["turns"], 0);
4865	assert!(
4866	body["buckets"].as_array().unwrap().is_empty(),
4867	"buckets must be empty when no turns exist: {body}"
4868	);
4869
4870	// group_by query options are validated.
4871	let bad_group = client
4872	.get(format!("http://{addr}/v1/usage?group_by=galaxy"))
4873	.send()
4874	.await?;
4875	assert_eq!(bad_group.status(), StatusCode::BAD_REQUEST);
4876
4877	// Each accepted group_by value succeeds.
4878	for gb in ["day", "model", "provider", "thread"] {
4879	let resp = client
4880	.get(format!("http://{addr}/v1/usage?group_by={gb}"))
4881	.send()
4882	.await?;
4883	assert!(resp.status().is_success(), "group_by={gb} failed: {resp:?}");
4884	}
4885
4886	// Bad ISO-8601 timestamp rejected.
4887	let bad_since = client
4888	.get(format!("http://{addr}/v1/usage?since=not-a-date"))
4889	.send()
4890	.await?;
4891	assert_eq!(bad_since.status(), StatusCode::BAD_REQUEST);
4892
4893	// since > until rejected.
4894	let inverted = client
4895	.get(format!(
4896	"http://{addr}/v1/usage?since=2030-01-02T00:00:00Z&until=2030-01-01T00:00:00Z"
4897	))
4898	.send()
4899	.await?;
4900	assert_eq!(inverted.status(), StatusCode::BAD_REQUEST);
4901
4902	handle.abort();
4903	Ok(())
4904	}
4905
4906	#[tokio::test]
4907	async fn runtime_info_reports_bind_state() -> Result<()> {
4908	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
4909	return Ok(());
4910	};
4911	let client = crate::tls::reqwest_client();
4912	let info: serde_json::Value = client
4913	.get(format!("http://{addr}/v1/runtime/info"))
4914	.send()
4915	.await?
4916	.error_for_status()?
4917	.json()
4918	.await?;
4919	assert_eq!(info["service"], "codewhale-runtime-api");
4920	assert_eq!(info["runtime_api_version"], "1.0");
4921	assert_eq!(info["codewhale_version"], info["version"]);
4922	let commit = info["codewhale_commit"]
4923	.as_str()
4924	.expect("runtime build commit must be a string");
4925	// Since #5245 the commit is env-stamped only: a stamped build (CI /
4926	// release / a `DEEPSEEK_BUILD_SHA=…` dogfood build) reports a full 40-hex
4927	// sha; an unstamped local build honestly reports "unknown" rather than
4928	// reading the checkout. Both are valid provenance — a fabricated sha
4929	// would be the bug.
4930	if commit == "unknown" {
4931	// Unstamped local build — the honest absence.
4932	} else {
4933	assert_eq!(commit.len(), 40, "a stamped build commit is a full sha");
4934	assert!(
4935	commit.bytes().all(\|byte\| byte.is_ascii_hexdigit()),
4936	"runtime build commit must be hexadecimal"
4937	);
4938	}
4939	assert_eq!(info["bind_host"], "127.0.0.1");
4940	assert_eq!(info["auth_required"], false);
4941	assert!(info["version"].is_string());
4942	assert_eq!(info["transports"], json!(["http", "sse"]));
4943	assert_eq!(info["capabilities"]["threads"], true);
4944	assert_eq!(info["capabilities"]["account_session"], true);
4945	assert_eq!(info["capabilities"]["external_tools"], true);
4946	assert_eq!(info["capabilities"]["worker_runtime"], true);
4947	assert_eq!(info["account"]["schema_version"], 1);
4948	assert_eq!(info["account"]["state"], "signed_out");
4949	assert_eq!(info["account"]["api_base"], "https://api.codewhale.net");
4950	assert_eq!(info["account"]["scopes"], json!([]));
4951	assert!(info["account"].get("access_token").is_none());
4952	assert!(info["account"].get("refresh_token").is_none());
4953	assert!(info["account"].get("email").is_none());
4954	assert!(info["experimental"].is_object());
4955
4956	handle.abort();
4957	Ok(())
4958	}
4959
4960	#[test]
4961	fn unauthenticated_runtime_info_redacts_secure_account_identity_without_loading_it() {
4962	use codewhale_secrets::account::{AccountSessionState, RuntimeAccountInfo};
4963
4964	let loaded = std::cell::Cell::new(false);
4965	let info = runtime_account_info_for_request(false, "https://api.codewhale.net", \|\| {
4966	loaded.set(true);
4967	RuntimeAccountInfo {
4968	schema_version: 1,
4969	state: AccountSessionState::Authenticated,
4970	api_base: "https://api.codewhale.net".to_string(),
4971	account_id: Some("acct-private".to_string()),
4972	session_id: Some("session-private".to_string()),
4973	scopes: vec!["identity:read".to_string()],
4974	expires_at: Some("2030-01-01T00:00:00Z".to_string()),
4975	}
4976	});
4977	assert!(
4978	!loaded.get(),
4979	"unauthenticated probes must not read secure storage"
4980	);
4981	assert_eq!(info.state, AccountSessionState::SignedOut);
4982	let json = serde_json::to_string(&info).unwrap();
4983	for private in ["acct-private", "session-private", "identity:read"] {
4984	assert!(!json.contains(private));
4985	}
4986	}
4987
4988	#[test]
4989	fn runtime_account_api_origin_rejects_credentials_paths_and_non_loopback_http() {
4990	assert_eq!(
4991	normalize_runtime_account_api_base("https://api.codewhale.net/"),
4992	Some("https://api.codewhale.net".to_string())
4993	);
4994	assert_eq!(
4995	normalize_runtime_account_api_base("http://127.0.0.1:8787"),
4996	Some("http://127.0.0.1:8787".to_string())
4997	);
4998	for invalid in [
4999	"https://user:secret@example.test",
5000	"https://example.test/account",
5001	"https://example.test?token=secret",
5002	"http://api.codewhale.net",
5003	] {
5004	assert_eq!(
5005	normalize_runtime_account_api_base(invalid),
5006	None,
5007	"{invalid}"
5008	);
5009	}
5010	}
5011
5012	#[tokio::test]
5013	async fn create_thread_accepts_dynamic_tools_and_environments() -> Result<()> {
5014	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
5015	return Ok(());
5016	};
5017	let client = crate::tls::reqwest_client();
5018
5019	let created: serde_json::Value = client
5020	.post(format!("http://{addr}/v1/threads"))
5021	.json(&json!({
5022	"model": "test-model",
5023	"dynamic_tools": [
5024	{
5025	"namespace": "tau_bench",
5026	"name": "get_reservation",
5027	"description": "Look up a reservation.",
5028	"input_schema": { "type": "object" }
5029	}
5030	],
5031	"environments": [
5032	{ "environment_id": "local", "cwd": "/workspace" }
5033	]
5034	}))
5035	.send()
5036	.await?
5037	.error_for_status()?
5038	.json()
5039	.await?;
5040	assert!(created["id"].is_string());
5041
5042	handle.abort();
5043	Ok(())
5044	}
5045
5046	#[tokio::test]
5047	async fn create_thread_normalizes_and_persists_named_permission_posture() -> Result<()> {
5048	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
5049	return Ok(());
5050	};
5051	let client = crate::tls::reqwest_client();
5052
5053	let created: serde_json::Value = client
5054	.post(format!("http://{addr}/v1/threads"))
5055	.json(&json!({
5056	"model": "test-model",
5057	"mode": "operate",
5058	"permission_posture": "auto-review"
5059	}))
5060	.send()
5061	.await?
5062	.error_for_status()?
5063	.json()
5064	.await?;
5065	assert_eq!(created["mode"], "operate");
5066	assert_eq!(created["permission_posture"], "auto_review");
5067	assert_eq!(created["auto_approve"], false);
5068	assert_eq!(created["trust_mode"], false);
5069
5070	let authoritative_ask: serde_json::Value = client
5071	.post(format!("http://{addr}/v1/threads"))
5072	.json(&json!({
5073	"model": "test-model",
5074	"mode": "yolo",
5075	"permission_posture": "ask",
5076	"auto_approve": true
5077	}))
5078	.send()
5079	.await?
5080	.error_for_status()?
5081	.json()
5082	.await?;
5083	assert_eq!(authoritative_ask["mode"], "agent");
5084	assert_eq!(authoritative_ask["permission_posture"], "ask");
5085	assert_eq!(authoritative_ask["auto_approve"], false);
5086
5087	let invalid = client
5088	.post(format!("http://{addr}/v1/threads"))
5089	.json(&json!({
5090	"model": "test-model",
5091	"permission_posture": "owner"
5092	}))
5093	.send()
5094	.await?;
5095	assert_eq!(invalid.status(), StatusCode::BAD_REQUEST);
5096
5097	handle.abort();
5098	Ok(())
5099	}
5100
5101	#[tokio::test]
5102	async fn start_turn_accepts_dynamic_tools_and_environment_id() -> Result<()> {
5103	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
5104	return Ok(());
5105	};
5106	let client = crate::tls::reqwest_client();
5107
5108	let created: serde_json::Value = client
5109	.post(format!("http://{addr}/v1/threads"))
5110	.json(&json!({ "model": "test-model" }))
5111	.send()
5112	.await?
5113	.error_for_status()?
5114	.json()
5115	.await?;
5116	let thread_id = created["id"].as_str().context("missing thread id")?;
5117
5118	let started: serde_json::Value = client
5119	.post(format!("http://{addr}/v1/threads/{thread_id}/turns"))
5120	.json(&json!({
5121	"prompt": "hello",
5122	"dynamic_tools": [
5123	{
5124	"name": "simple_tool",
5125	"description": "A simple tool.",
5126	"input_schema": { "type": "object" }
5127	}
5128	],
5129	"environment_id": "local",
5130	"permission_posture": "auto-review"
5131	}))
5132	.send()
5133	.await?
5134	.error_for_status()?
5135	.json()
5136	.await?;
5137	assert_eq!(started["turn"]["thread_id"], thread_id);
5138	assert_eq!(started["thread"]["permission_posture"], "ask");
5139	assert_eq!(started["turn"]["permission_posture"], "auto_review");
5140
5141	let stored: serde_json::Value = client
5142	.get(format!("http://{addr}/v1/threads/{thread_id}"))
5143	.send()
5144	.await?
5145	.error_for_status()?
5146	.json()
5147	.await?;
5148	assert_eq!(stored["turns"][0]["permission_posture"], "auto_review");
5149
5150	handle.abort();
5151	Ok(())
5152	}
5153
5154	#[tokio::test]
5155	async fn mobile_page_is_available_only_when_enabled() -> Result<()> {
5156	let tmp = tempfile::tempdir()?;
5157	let root = tmp.path().to_path_buf();
5158	let sessions_dir = root.join("sessions");
5159	let Some((addr, _runtime_threads, handle)) = spawn_test_server_with_root_token_and_mobile(
5160	root.clone(),
5161	sessions_dir.clone(),
5162	None,
5163	false,
5164	)
5165	.await?
5166	else {
5167	return Ok(());
5168	};
5169	let client = crate::tls::reqwest_client();
5170	let disabled = client.get(format!("http://{addr}/mobile")).send().await?;
5171	assert_eq!(disabled.status(), StatusCode::NOT_FOUND);
5172	handle.abort();
5173
5174	let Some((addr, _runtime_threads, handle)) =
5175	spawn_test_server_with_root_token_and_mobile(root, sessions_dir, None, true).await?
5176	else {
5177	return Ok(());
5178	};
5179	let enabled = client
5180	.get(format!("http://{addr}/mobile"))
5181	.send()
5182	.await?
5183	.error_for_status()?;
5184	let html = enabled.text().await?;
5185	assert!(html.contains("Codewhale Mobile"));
5186	assert!(html.contains("/v1/approvals/"));
5187	assert!(html.contains("MAX_VISIBLE_EVENTS = 100"));
5188	assert!(html.contains("replay_limit="));
5189
5190	handle.abort();
5191	Ok(())
5192	}
5193
5194	#[tokio::test]
5195	async fn mobile_page_serves_shell_when_auth_enabled() -> Result<()> {
5196	let tmp = tempfile::tempdir()?;
5197	let root = tmp.path().to_path_buf();
5198	let sessions_dir = root.join("sessions");
5199	let token = "abc ABC+/?:=&%".to_string();
5200	let Some((addr, _runtime_threads, handle)) =
5201	spawn_test_server_with_root_token_and_mobile(root, sessions_dir, Some(token.clone()), true)
5202	.await?
5203	else {
5204	return Ok(());
5205	};
5206	let client = crate::tls::reqwest_client();
5207
5208	let shell = client
5209	.get(format!("http://{addr}/mobile"))
5210	.send()
5211	.await?
5212	.error_for_status()?;
5213	let html = shell.text().await?;
5214	assert!(html.contains("Codewhale Mobile"));
5215	assert!(html.contains("TOKEN_COOKIE"));
5216
5217	let bearer = client
5218	.get(format!("http://{addr}/mobile"))
5219	.bearer_auth(&token)
5220	.send()
5221	.await?
5222	.error_for_status()?;
5223	assert!(bearer.text().await?.contains("Codewhale Mobile"));
5224
5225	handle.abort();
5226	Ok(())
5227	}
5228
5229	#[tokio::test]
5230	async fn mobile_insecure_mode_allows_page_and_v1_routes_without_token() -> Result<()> {
5231	let tmp = tempfile::tempdir()?;
5232	let root = tmp.path().to_path_buf();
5233	let sessions_dir = root.join("sessions");
5234	let Some((addr, _runtime_threads, handle)) =
5235	spawn_test_server_with_root_token_and_mobile(root, sessions_dir, None, true).await?
5236	else {
5237	return Ok(());
5238	};
5239	let client = crate::tls::reqwest_client();
5240
5241	let page = client
5242	.get(format!("http://{addr}/mobile"))
5243	.send()
5244	.await?
5245	.error_for_status()?;
5246	assert!(page.text().await?.contains("Codewhale Mobile"));
5247
5248	let summary = client
5249	.get(format!("http://{addr}/v1/threads/summary"))
5250	.send()
5251	.await?
5252	.error_for_status()?;
5253	assert_eq!(summary.status(), StatusCode::OK);
5254
5255	handle.abort();
5256	Ok(())
5257	}
5258
5259	#[tokio::test]
5260	async fn decide_approval_404s_when_nothing_pending() -> Result<()> {
5261	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
5262	return Ok(());
5263	};
5264	let client = crate::tls::reqwest_client();
5265	let resp = client
5266	.post(format!("http://{addr}/v1/approvals/no_such_id"))
5267	.json(&json!({ "decision": "allow" }))
5268	.send()
5269	.await?;
5270	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
5271
5272	handle.abort();
5273	Ok(())
5274	}
5275
5276	#[tokio::test]
5277	async fn submit_user_input_404s_without_entering_engine_mailbox_for_unknown_id() -> Result<()> {
5278	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
5279	return Ok(());
5280	};
5281	let thread = runtime_threads
5282	.create_thread(CreateThreadRequest::default())
5283	.await?;
5284	let mut harness = crate::core::engine::mock_engine_handle();
5285	runtime_threads
5286	.install_test_engine(&thread.id, harness.handle.clone())
5287	.await?;
5288
5289	let response = crate::tls::reqwest_client()
5290	.post(format!(
5291	"http://{addr}/v1/user-input/{}/input-missing",
5292	thread.id
5293	))
5294	.json(&json!({
5295	"answers": [{
5296	"id": "choice",
5297	"label": "Missing",
5298	"value": "must-not-enter-engine-mailbox",
5299	}],
5300	}))
5301	.send()
5302	.await?;
5303	assert_eq!(response.status(), StatusCode::NOT_FOUND);
5304	assert!(
5305	tokio::time::timeout(
5306	Duration::from_millis(25),
5307	harness.recv_user_input_submission()
5308	)
5309	.await
5310	.is_err(),
5311	"unknown user input reached the engine mailbox"
5312	);
5313
5314	handle.abort();
5315	Ok(())
5316	}
5317
5318	#[tokio::test]
5319	async fn events_endpoint_rejects_unbounded_tail_requests() -> Result<()> {
5320	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
5321	return Ok(());
5322	};
5323	let thread = runtime_threads
5324	.create_thread(CreateThreadRequest::default())
5325	.await?;
5326	let response = crate::tls::reqwest_client()
5327	.get(format!(
5328	"http://{addr}/v1/threads/{}/events?replay_limit={}",
5329	thread.id,
5330	MAX_RUNTIME_EVENT_REPLAY_TAIL.saturating_add(1),
5331	))
5332	.send()
5333	.await?;
5334	assert_eq!(response.status(), StatusCode::BAD_REQUEST);
5335
5336	handle.abort();
5337	Ok(())
5338	}
5339
5340	#[tokio::test]
5341	async fn decide_approval_400s_on_bad_decision() -> Result<()> {
5342	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
5343	return Ok(());
5344	};
5345	let client = crate::tls::reqwest_client();
5346	let resp = client
5347	.post(format!("http://{addr}/v1/approvals/whatever"))
5348	.json(&json!({ "decision": "yolo" }))
5349	.send()
5350	.await?;
5351	assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
5352
5353	handle.abort();
5354	Ok(())
5355	}
5356
5357	#[tokio::test]
5358	async fn decide_approval_delivers_to_runtime() -> Result<()> {
5359	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
5360	return Ok(());
5361	};
5362	let client = crate::tls::reqwest_client();
5363	let rx = runtime_threads.register_pending_approval_for_test("ext_id");
5364
5365	let resp = client
5366	.post(format!("http://{addr}/v1/approvals/ext_id"))
5367	.json(&json!({ "decision": "allow", "remember": false }))
5368	.send()
5369	.await?;
5370	assert_eq!(resp.status(), StatusCode::OK);
5371	let body: serde_json::Value = resp.json().await?;
5372	assert_eq!(body["ok"], true);
5373	assert_eq!(body["decision"], "allow");
5374	assert_eq!(body["delivered"], true);
5375
5376	let received = tokio::time::timeout(ci_scaled(Duration::from_secs(1)), rx).await??;
5377	assert_eq!(
5378	received,
5379	ExternalApprovalDecision::Allow { remember: false }
5380	);
5381
5382	handle.abort();
5383	Ok(())
5384	}
5385
5386	#[tokio::test]
5387	async fn dynamic_tool_result_endpoint_delivers_to_runtime() -> Result<()> {
5388	let Some((addr, runtime_threads, handle)) = spawn_test_server().await? else {
5389	return Ok(());
5390	};
5391	let client = crate::tls::reqwest_client();
5392	let thread: serde_json::Value = client
5393	.post(format!("http://{addr}/v1/threads"))
5394	.json(&json!({}))
5395	.send()
5396	.await?
5397	.error_for_status()?
5398	.json()
5399	.await?;
5400	let thread_id = thread["id"].as_str().context("thread id")?;
5401	let rx =
5402	runtime_threads.register_pending_dynamic_tool_for_test(thread_id, "turn_1", "call_1")?;
5403
5404	let wrong_turn = client
5405	.post(format!(
5406	"http://{addr}/v1/threads/{thread_id}/turns/turn_wrong/tool-calls/call_1/result"
5407	))
5408	.json(&json!({ "success": false }))
5409	.send()
5410	.await?;
5411	assert_eq!(wrong_turn.status(), StatusCode::NOT_FOUND);
5412
5413	let resp = client
5414	.post(format!(
5415	"http://{addr}/v1/threads/{thread_id}/turns/turn_1/tool-calls/call_1/result"
5416	))
5417	.json(&json!({
5418	"success": true,
5419	"content": [{ "type": "input_text", "text": "ok" }]
5420	}))
5421	.send()
5422	.await?;
5423	assert_eq!(resp.status(), StatusCode::ACCEPTED);
5424
5425	let received = tokio::time::timeout(ci_scaled(Duration::from_secs(1)), rx).await??;
5426	assert!(received.success);
5427	assert_eq!(received.content.len(), 1);
5428	let resolved = runtime_threads
5429	.events_since(thread_id, None)?
5430	.into_iter()
5431	.filter(\|event\| event.event == "tool_call.resolved")
5432	.collect::<Vec<_>>();
5433	assert_eq!(resolved.len(), 1);
5434	assert_eq!(resolved[0].payload["call_id"], "call_1");
5435	assert!(resolved[0].payload.get("content").is_none());
5436
5437	let duplicate = client
5438	.post(format!(
5439	"http://{addr}/v1/threads/{thread_id}/turns/turn_1/tool-calls/call_1/result"
5440	))
5441	.json(&json!({ "success": true }))
5442	.send()
5443	.await?;
5444	assert_eq!(duplicate.status(), StatusCode::NOT_FOUND);
5445	assert_eq!(
5446	runtime_threads
5447	.events_since(thread_id, None)?
5448	.iter()
5449	.filter(\|event\| event.event == "tool_call.resolved")
5450	.count(),
5451	1
5452	);
5453
5454	handle.abort();
5455	Ok(())
5456	}
5457
5458	#[tokio::test]
5459	async fn skills_endpoint_includes_enabled_field() -> Result<()> {
5460	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
5461	return Ok(());
5462	};
5463	let client = crate::tls::reqwest_client();
5464	let body: serde_json::Value = client
5465	.get(format!("http://{addr}/v1/skills"))
5466	.send()
5467	.await?
5468	.error_for_status()?
5469	.json()
5470	.await?;
5471	if let Some(skills) = body["skills"].as_array() {
5472	for skill in skills {
5473	assert!(skill.get("enabled").is_some());
5474	}
5475	}
5476
5477	handle.abort();
5478	Ok(())
5479	}
5480
5481	#[tokio::test]
5482	async fn skills_endpoint_exposes_safe_plugin_provenance_and_shared_toggle() -> Result<()> {
5483	let tmp = tempfile::tempdir()?;
5484	let root = tmp.path().join("runtime");
5485	let workspace = tmp.path().join("workspace");
5486	let plugin_root = tmp.path().join("plugins/demo");
5487	fs::create_dir_all(plugin_root.join("skills/review"))?;
5488	fs::write(
5489	plugin_root.join("plugin.toml"),
5490	"schema_version = 1\n[plugin]\nname = \"demo\"\nversion = \"1.0.0\"\n[skills]\npath = \"skills\"\n",
5491	)?;
5492	fs::write(
5493	plugin_root.join("skills/review/SKILL.md"),
5494	"---\nname: review\ndescription: reviewed plugin Skill\n---\nbody\n",
5495	)?;
5496	let plugin_config = crate::plugins::discovery::DiscoveryConfig {
5497	workspace: workspace.clone(),
5498	user_plugins_dir: tmp.path().join("plugins"),
5499	workspace_plugins_dir: workspace.join(".codewhale/plugins"),
5500	builtin_plugin_dirs: Vec::new(),
5501	state_path: tmp.path().join("plugin-state/state.json"),
5502	};
5503	let discovery = crate::plugins::PluginDiscoveryContext::from_config_and_environment(
5504	&plugin_config,
5505	crate::plugins::HostEnvironment::default(),
5506	);
5507	let mut plugins = discovery.registry_for_workspace(&workspace);
5508	Arc::make_mut(&mut plugins)
5509	.trust("demo")
5510	.map_err(anyhow::Error::msg)?;
5511	Arc::make_mut(&mut plugins)
5512	.enable("demo")
5513	.map_err(anyhow::Error::msg)?;
5514
5515	let Some((addr, _runtime_threads, handle)) =
5516	spawn_test_server_with_root_token_mobile_workspace_and_overrides(
5517	root.clone(),
5518	root.join("sessions"),
5519	None,
5520	false,
5521	workspace,
5522	TestServerOverrides {
5523	plugin_discovery: Some(discovery),
5524	..TestServerOverrides::default()
5525	},
5526	)
5527	.await?
5528	else {
5529	return Ok(());
5530	};
5531	let client = crate::tls::reqwest_client();
5532	let list = client
5533	.get(format!("http://{addr}/v1/skills"))
5534	.send()
5535	.await?
5536	.error_for_status()?
5537	.json::<serde_json::Value>()
5538	.await?;
5539	let plugin_skill = list["skills"]
5540	.as_array()
5541	.and_then(\|skills\| skills.iter().find(\|skill\| skill["name"] == "demo:review"))
5542	.context("plugin Skill in runtime API catalog")?;
5543	assert_eq!(plugin_skill["enabled"], true);
5544	assert_eq!(plugin_skill["path"], serde_json::Value::Null);
5545	assert_eq!(plugin_skill["source"], "reviewed-plugin-snapshot:demo");
5546	assert!(plugin_skill["plugin_id"].as_str().is_some());
5547	assert!(plugin_skill["plugin_generation"].as_u64().is_some());
5548	assert!(plugin_skill["plugin_content_hash"].as_str().is_some());
5549	assert!(
5550	!plugin_skill
5551	.to_string()
5552	.contains(&plugin_root.display().to_string()),
5553	"runtime API must not expose mutable or staged plugin paths"
5554	);
5555
5556	client
5557	.post(format!("http://{addr}/v1/skills/demo:review"))
5558	.json(&json!({ "enabled": false }))
5559	.send()
5560	.await?
5561	.error_for_status()?;
5562	let after = client
5563	.get(format!("http://{addr}/v1/skills"))
5564	.send()
5565	.await?
5566	.error_for_status()?
5567	.json::<serde_json::Value>()
5568	.await?;
5569	let plugin_skill = after["skills"]
5570	.as_array()
5571	.and_then(\|skills\| skills.iter().find(\|skill\| skill["name"] == "demo:review"))
5572	.context("plugin Skill after toggle")?;
5573	assert_eq!(plugin_skill["enabled"], false);
5574
5575	handle.abort();
5576	Ok(())
5577	}
5578
5579	#[tokio::test]
5580	async fn skill_toggle_endpoint_404s_for_unknown_skill() -> Result<()> {
5581	let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
5582	return Ok(());
5583	};
5584	let client = crate::tls::reqwest_client();
5585	let resp = client
5586	.post(format!("http://{addr}/v1/skills/no-such-skill"))
5587	.json(&json!({ "enabled": false }))
5588	.send()
5589	.await?;
5590	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
5591
5592	handle.abort();
5593	Ok(())
5594	}
5595
5596	#[test]
5597	fn resolve_skills_dir_finds_workspace_local_agents_skills() {
5598	let tmp = tempfile::tempdir().expect("tempdir");
5599	let workspace = tmp.path();
5600	let local_skills = workspace.join(".agents").join("skills");
5601	fs::create_dir_all(&local_skills).expect("create skills dir");
5602
5603	let config = Config::default();
5604	let resolved = resolve_skills_dir(&config, workspace);
5605
5606	let expected = fs::canonicalize(&local_skills).expect("canonical local skills");
5607	assert_eq!(resolved, expected);
5608	}
5609
5610	#[test]
5611	fn resolve_skills_dir_finds_workspace_local_skills_fallback() {
5612	let tmp = tempfile::tempdir().expect("tempdir");
5613	let workspace = tmp.path();
5614	let local_skills = workspace.join("skills");
5615	fs::create_dir_all(&local_skills).expect("create skills dir");
5616
5617	let config = Config::default();
5618	let resolved = resolve_skills_dir(&config, workspace);
5619
5620	let expected = fs::canonicalize(&local_skills).expect("canonical local skills");
5621	assert_eq!(resolved, expected);
5622	}
5623
5624	#[test]
5625	fn resolve_skills_dir_respects_codewhale_only_scan() {
5626	let tmp = tempfile::tempdir().expect("tempdir");
5627	let workspace = tmp.path();
5628	let agents_skills = workspace.join(".agents").join("skills");
5629	let codewhale_skills = workspace.join(".codewhale").join("skills");
5630	fs::create_dir_all(&agents_skills).expect("create agents skills dir");
5631	fs::create_dir_all(&codewhale_skills).expect("create codewhale skills dir");
5632
5633	let config = Config {
5634	skills: Some(crate::config::SkillsConfig {
5635	scan_codewhale_only: Some(true),
5636	..Default::default()
5637	}),
5638	..Default::default()
5639	};
5640	let resolved = resolve_skills_dir(&config, workspace);
5641
5642	let expected = fs::canonicalize(&codewhale_skills).expect("canonical codewhale skills");
5643	assert_eq!(resolved, expected);
5644	}
5645
5646	#[test]
5647	fn resolve_skills_dir_preserves_explicit_dir_in_codewhale_only_scan() {
5648	let tmp = tempfile::tempdir().expect("tempdir");
5649	let workspace = tmp.path().join("workspace");
5650	let codewhale_skills = workspace.join(".codewhale").join("skills");
5651	let configured_skills = tmp.path().join("configured-skills");
5652	fs::create_dir_all(&codewhale_skills).expect("create codewhale skills dir");
5653	fs::create_dir_all(&configured_skills).expect("create configured skills dir");
5654
5655	let config = Config {
5656	skills_dir: Some(configured_skills.to_string_lossy().into_owned()),
5657	skills: Some(crate::config::SkillsConfig {
5658	scan_codewhale_only: Some(true),
5659	..Default::default()
5660	}),
5661	..Default::default()
5662	};
5663	let resolved = resolve_skills_dir(&config, &workspace);
5664
5665	assert_eq!(resolved, configured_skills);
5666	}
5667
5668	#[test]
5669	fn skills_search_directories_includes_custom_skills_dir() {
5670	let tmp = tempfile::tempdir().expect("tempdir");
5671	let workspace = tmp.path().join("workspace");
5672	let custom_skills = tmp.path().join("custom-skills");
5673	fs::create_dir_all(&workspace).expect("create workspace");
5674	fs::create_dir_all(&custom_skills).expect("create custom skills");
5675
5676	let directories = skills_search_directories(
5677	&workspace,
5678	&custom_skills,
5679	crate::skills::SkillDiscoveryMode::Compatible,
5680	);
5681
5682	assert!(
5683	directories.iter().any(\|dir\| dir == &custom_skills),
5684	"custom skills_dir must be reported when discovery searches it"
5685	);
5686	let message = format_skill_search_paths(&directories);
5687	assert!(message.contains("custom-skills"));
5688	}
5689
5690	#[test]
5691	fn skill_entry_is_bundled_requires_configured_bundle_path() {
5692	let tmp = tempfile::tempdir().expect("tempdir");
5693	let bundled_skills_dir = tmp.path().join("bundled-skills");
5694	let bundled_skill_path = bundled_skills_dir.join("delegate").join("SKILL.md");
5695	let override_skill_path = tmp
5696	.path()
5697	.join("workspace")
5698	.join(".agents")
5699	.join("skills")
5700	.join("delegate")
5701	.join("SKILL.md");
5702	fs::create_dir_all(bundled_skill_path.parent().expect("bundled parent"))
5703	.expect("create bundled skill dir");
5704	fs::create_dir_all(override_skill_path.parent().expect("override parent"))
5705	.expect("create override skill dir");
5706	fs::write(
5707	&bundled_skill_path,
5708	"---\nname: delegate\ndescription: bundled\n---\n",
5709	)
5710	.expect("write bundled skill");
5711	fs::write(
5712	&override_skill_path,
5713	"---\nname: delegate\ndescription: override\n---\n",
5714	)
5715	.expect("write override skill");
5716
5717	let bundled_skill = crate::skills::Skill {
5718	name: "delegate".to_string(),
5719	description: String::new(),
5720	localized_descriptions: std::collections::HashMap::new(),
5721	invocation: crate::skills::SkillInvocation::ModelAndUser,
5722	aliases: Vec::new(),
5723	body: String::new(),
5724	path: bundled_skill_path,
5725	source: crate::skills::SkillSource::Native,
5726	};
5727	let override_skill = crate::skills::Skill {
5728	name: "delegate".to_string(),
5729	description: String::new(),
5730	localized_descriptions: std::collections::HashMap::new(),
5731	invocation: crate::skills::SkillInvocation::ModelAndUser,
5732	aliases: Vec::new(),
5733	body: String::new(),
5734	path: override_skill_path,
5735	source: crate::skills::SkillSource::Native,
5736	};
5737
5738	assert!(skill_entry_is_bundled(&bundled_skill, &bundled_skills_dir));
5739	assert!(!skill_entry_is_bundled(
5740	&override_skill,
5741	&bundled_skills_dir
5742	));
5743	}
5744
5745	/// A `skills` symlink that points outside the workspace must NOT be
5746	/// returned as the resolved skills directory. Containment check ensures
5747	/// the canonicalized candidate stays under the canonicalized workspace
5748	/// root, so a malicious or misconfigured symlink can't promote
5749	/// `/etc` (or any other path) into the skills loader.
5750	#[cfg(unix)]
5751	#[test]
5752	fn resolve_skills_dir_rejects_symlink_escaping_workspace() {
5753	let tmp = tempfile::tempdir().expect("tempdir");
5754	let _env_lock = crate::test_support::lock_test_env();
5755	let _home = crate::test_support::EnvVarGuard::set("HOME", tmp.path());
5756	let _userprofile = crate::test_support::EnvVarGuard::set("USERPROFILE", tmp.path());
5757	let workspace_root = tmp.path().join("workspace");
5758	let escape_target = tmp.path().join("escape_target");
5759	fs::create_dir_all(&workspace_root).expect("create workspace");
5760	fs::create_dir_all(&escape_target).expect("create escape target");
5761
5762	let dotagents = workspace_root.join(".agents");
5763	fs::create_dir_all(&dotagents).expect("create .agents");
5764	let bad_link = dotagents.join("skills");
5765	std::os::unix::fs::symlink(&escape_target, &bad_link).expect("symlink");
5766
5767	let config = Config::default();
5768	let resolved = resolve_skills_dir(&config, &workspace_root);
5769
5770	let canon_escape = fs::canonicalize(&escape_target).expect("canon escape");
5771	assert_ne!(
5772	resolved, canon_escape,
5773	"symlink escaping workspace must not be resolved as skills dir"
5774	);
5775	assert_eq!(
5776	resolved,
5777	config.skills_dir(),
5778	"with no valid in-workspace skills dir, resolution should fall back to config"
5779	);
5780	}
5781
5782	#[cfg(unix)]
5783	#[test]
5784	fn resolve_skills_dir_rejects_codewhale_only_symlink_escaping_workspace() {
5785	let tmp = tempfile::tempdir().expect("tempdir");
5786	let _env_lock = crate::test_support::lock_test_env();
5787	let _home = crate::test_support::EnvVarGuard::set("HOME", tmp.path());
5788	let _userprofile = crate::test_support::EnvVarGuard::set("USERPROFILE", tmp.path());
5789	let workspace_root = tmp.path().join("workspace");
5790	let escape_target = tmp.path().join("escape_target");
5791	fs::create_dir_all(&workspace_root).expect("create workspace");
5792	fs::create_dir_all(&escape_target).expect("create escape target");
5793
5794	let dotcodewhale = workspace_root.join(".codewhale");
5795	fs::create_dir_all(&dotcodewhale).expect("create .codewhale");
5796	let bad_link = dotcodewhale.join("skills");
5797	std::os::unix::fs::symlink(&escape_target, &bad_link).expect("symlink");
5798
5799	let config = Config {
5800	skills: Some(crate::config::SkillsConfig {
5801	scan_codewhale_only: Some(true),
5802	..Default::default()
5803	}),
5804	..Default::default()
5805	};
5806	let resolved = resolve_skills_dir(&config, &workspace_root);
5807
5808	let canon_escape = fs::canonicalize(&escape_target).expect("canon escape");
5809	assert_ne!(
5810	resolved, canon_escape,
5811	"CodeWhale-only symlink escaping workspace must not be resolved as skills dir"
5812	);
5813	assert_eq!(
5814	resolved,
5815	config.skills_dir(),
5816	"with no valid in-workspace CodeWhale skills dir, resolution should fall back to config"
5817	);
5818	}
5819
5820	// ---------------------------------------------------------------------------
5821	// /v1/config + /v1/config/reload endpoint tests
5822	// ---------------------------------------------------------------------------
5823
5824	/// Helper: POST to `/v1/config` with the given key/value and return the
5825	/// response status + body JSON.
5826	async fn post_set_config(
5827	client: &reqwest::Client,
5828	addr: &SocketAddr,
5829	key: &str,
5830	value: &str,
5831	persist: bool,
5832	) -> (reqwest::StatusCode, serde_json::Value) {
5833	let resp = client
5834	.post(format!("http://{addr}/v1/config"))
5835	.json(&serde_json::json!({
5836	"key": key,
5837	"value": value,
5838	"persist": persist,
5839	}))
5840	.send()
5841	.await
5842	.expect("POST /v1/config should not fail at transport level");
5843	let status = resp.status();
5844	let body: serde_json::Value = resp
5845	.json()
5846	.await
5847	.unwrap_or_else(\|_\| serde_json::json!({"_error": "non-json response body"}));
5848	(status, body)
5849	}
5850
5851	#[tokio::test]
5852	async fn set_config_rejects_unknown_key_with_bad_request() -> Result<()> {
5853	let root = std::env::temp_dir().join(format!("codewhale-config-unknown-{}", Uuid::new_v4()));
5854	let sessions_dir = root.join("sessions");
5855	let Some((addr, _runtime_threads, handle)) =
5856	spawn_test_server_with_root(root, sessions_dir).await?
5857	else {
5858	return Ok(());
5859	};
5860	let client = crate::tls::reqwest_client();
5861
5862	let (status, body) = post_set_config(&client, &addr, "nonexistent_key", "x", true).await;
5863	assert_eq!(
5864	status,
5865	StatusCode::BAD_REQUEST,
5866	"unknown key should return 400, body: {body}"
5867	);
5868	let message = body["error"]["message"]
5869	.as_str()
5870	.unwrap_or_default()
5871	.to_lowercase();
5872	assert!(
5873	message.contains("unknown config key"),
5874	"error message should mention 'unknown config key', got: {message}"
5875	);
5876
5877	handle.abort();
5878	Ok(())
5879	}
5880
5881	#[tokio::test]
5882	async fn set_config_validates_max_history_input() -> Result<()> {
5883	// Fix #4: invalid max_history input must return 400 instead of silently
5884	// falling back to a default value.
5885	let root = std::env::temp_dir().join(format!("codewhale-config-maxhist-{}", Uuid::new_v4()));
5886	let sessions_dir = root.join("sessions");
5887	let Some((addr, _runtime_threads, handle)) =
5888	spawn_test_server_with_root(root, sessions_dir).await?
5889	else {
5890	return Ok(());
5891	};
5892	let client = crate::tls::reqwest_client();
5893
5894	// Non-integer input must be rejected.
5895	let (status, body) = post_set_config(&client, &addr, "max_history", "not-a-number", true).await;
5896	assert_eq!(
5897	status,
5898	StatusCode::BAD_REQUEST,
5899	"invalid max_history should return 400, body: {body}"
5900	);
5901
5902	// Negative input must also be rejected (parse::<usize> rejects negatives).
5903	let (status, body) = post_set_config(&client, &addr, "max_history", "-5", true).await;
5904	assert_eq!(
5905	status,
5906	StatusCode::BAD_REQUEST,
5907	"negative max_history should return 400, body: {body}"
5908	);
5909
5910	handle.abort();
5911	Ok(())
5912	}
5913
5914	#[tokio::test]
5915	async fn set_config_validates_subagents_enabled_input() -> Result<()> {
5916	// Fix #1: subagents_enabled must validate input and reject non-boolean
5917	// values with a descriptive 400 error.
5918	let root = std::env::temp_dir().join(format!("codewhale-config-subenabled-{}", Uuid::new_v4()));
5919	let sessions_dir = root.join("sessions");
5920	let Some((addr, _runtime_threads, handle)) =
5921	spawn_test_server_with_root(root, sessions_dir).await?
5922	else {
5923	return Ok(());
5924	};
5925	let client = crate::tls::reqwest_client();
5926
5927	let (status, body) = post_set_config(&client, &addr, "subagents_enabled", "maybe", true).await;
5928	assert_eq!(
5929	status,
5930	StatusCode::BAD_REQUEST,
5931	"non-boolean subagents_enabled should return 400, body: {body}"
5932	);
5933	let message = body["error"]["message"]
5934	.as_str()
5935	.unwrap_or_default()
5936	.to_lowercase();
5937	assert!(
5938	message.contains("subagents_enabled"),
5939	"error message should name the key, got: {message}"
5940	);
5941
5942	handle.abort();
5943	Ok(())
5944	}
5945
5946	#[tokio::test]
5947	async fn set_config_validates_subagents_max_depth_input() -> Result<()> {
5948	// Fix #1: subagents_max_depth must validate input and reject non-integer
5949	// values with a descriptive 400 error.
5950	let root = std::env::temp_dir().join(format!("codewhale-config-subdepth-{}", Uuid::new_v4()));
5951	let sessions_dir = root.join("sessions");
5952	let Some((addr, _runtime_threads, handle)) =
5953	spawn_test_server_with_root(root, sessions_dir).await?
5954	else {
5955	return Ok(());
5956	};
5957	let client = crate::tls::reqwest_client();
5958
5959	let (status, body) = post_set_config(&client, &addr, "subagents_max_depth", "deep", true).await;
5960	assert_eq!(
5961	status,
5962	StatusCode::BAD_REQUEST,
5963	"non-integer subagents_max_depth should return 400, body: {body}"
5964	);
5965
5966	handle.abort();
5967	Ok(())
5968	}
5969
5970	#[tokio::test]
5971	async fn set_config_with_config_path_writes_to_specified_file() -> Result<()> {
5972	// Fix #2: when the server is started with --config, set_config must
5973	// persist to that specific file rather than the default discovery path.
5974	let root =
5975	std::env::temp_dir().join(format!("codewhale-config-path-persist-{}", Uuid::new_v4()));
5976	fs::create_dir_all(&root)?;
5977	let config_file = root.join("custom-config.toml");
5978	fs::write(&config_file, "# initial\n")?;
5979
5980	let Some((addr, _runtime_threads, handle)) =
5981	spawn_test_server_with_config_path(config_file.clone()).await?
5982	else {
5983	return Ok(());
5984	};
5985	let client = crate::tls::reqwest_client();
5986
5987	// Persist a subagents_max_depth value above the ceiling to also verify
5988	// clamping (Fix #1).
5989	let over_ceiling = u64::from(codewhale_config::MAX_SPAWN_DEPTH_CEILING) + 10;
5990	let (status, body) = post_set_config(
5991	&client,
5992	&addr,
5993	"subagents_max_depth",
5994	&over_ceiling.to_string(),
5995	true,
5996	)
5997	.await;
5998	assert_eq!(
5999	status,
6000	StatusCode::OK,
6001	"persisting subagents_max_depth should succeed, body: {body}"
6002	);
6003	assert!(
6004	body["persisted"].as_bool().unwrap_or(false),
6005	"response should report persisted=true, body: {body}"
6006	);
6007
6008	// Read the config file and verify the value was clamped and written.
6009	let contents = fs::read_to_string(&config_file)
6010	.with_context(\|\| format!("config file should exist at {}", config_file.display()))?;
6011	assert!(
6012	contents.contains("max_depth"),
6013	"config file should contain max_depth key, got: {contents}"
6014	);
6015	// The value should be clamped to MAX_SPAWN_DEPTH_CEILING.
6016	let expected = format!(
6017	"max_depth = {}",
6018	u64::from(codewhale_config::MAX_SPAWN_DEPTH_CEILING)
6019	);
6020	assert!(
6021	contents.contains(&expected),
6022	"config file should contain clamped value '{expected}', got: {contents}"
6023	);
6024
6025	// Also verify a subagents_enabled persistence writes to the same file.
6026	let (status, body) = post_set_config(&client, &addr, "subagents_enabled", "true", true).await;
6027	assert_eq!(status, StatusCode::OK, "body: {body}");
6028	let contents = fs::read_to_string(&config_file)?;
6029	assert!(
6030	contents.contains("enabled = true"),
6031	"config file should contain enabled = true, got: {contents}"
6032	);
6033
6034	handle.abort();
6035	Ok(())
6036	}
6037
6038	#[tokio::test]
6039	async fn reload_config_endpoint_returns_success() -> Result<()> {
6040	// Basic smoke test that /v1/config/reload returns 200 with a message.
6041	let root = std::env::temp_dir().join(format!("codewhale-config-reload-{}", Uuid::new_v4()));
6042	let sessions_dir = root.join("sessions");
6043	let Some((addr, _runtime_threads, handle)) =
6044	spawn_test_server_with_root(root, sessions_dir).await?
6045	else {
6046	return Ok(());
6047	};
6048	let client = crate::tls::reqwest_client();
6049
6050	let resp = client
6051	.post(format!("http://{addr}/v1/config/reload"))
6052	.send()
6053	.await?;
6054	assert_eq!(resp.status(), StatusCode::OK);
6055	let body: serde_json::Value = resp.json().await?;
6056	let message = body["message"].as_str().unwrap_or_default().to_string();
6057	assert!(
6058	!message.is_empty(),
6059	"reload response should include a non-empty message"
6060	);
6061
6062	handle.abort();
6063	Ok(())
6064	}
6065
6066	/// Helper: GET `/v1/config` and return the parsed response body.
6067	async fn get_config(client: &reqwest::Client, addr: &SocketAddr) -> serde_json::Value {
6068	client
6069	.get(format!("http://{addr}/v1/config"))
6070	.send()
6071	.await
6072	.expect("GET /v1/config should not fail at transport level")
6073	.error_for_status()
6074	.expect("GET /v1/config should return 200")
6075	.json()
6076	.await
6077	.expect("GET /v1/config should return valid JSON")
6078	}
6079
6080	async fn get_providers(client: &reqwest::Client, addr: &SocketAddr) -> serde_json::Value {
6081	client
6082	.get(format!("http://{addr}/v1/providers"))
6083	.send()
6084	.await
6085	.expect("GET /v1/providers should not fail at transport level")
6086	.error_for_status()
6087	.expect("GET /v1/providers should return 200")
6088	.json()
6089	.await
6090	.expect("GET /v1/providers should return valid JSON")
6091	}
6092
6093	async fn get_provider_models(
6094	client: &reqwest::Client,
6095	addr: &SocketAddr,
6096	provider: &str,
6097	) -> serde_json::Value {
6098	client
6099	.get(format!("http://{addr}/v1/providers/{provider}/models"))
6100	.send()
6101	.await
6102	.expect("GET /v1/providers/{id}/models should not fail at transport level")
6103	.error_for_status()
6104	.expect("GET /v1/providers/{id}/models should return 200")
6105	.json()
6106	.await
6107	.expect("GET /v1/providers/{id}/models should return valid JSON")
6108	}
6109
6110	#[tokio::test]
6111	async fn get_config_returns_active_provider_model() -> Result<()> {
6112	let root = std::env::temp_dir().join(format!(
6113	"codewhale-config-active-provider-{}",
6114	Uuid::new_v4()
6115	));
6116	fs::create_dir_all(&root)?;
6117	let config_file = root.join("custom-config.toml");
6118	fs::write(
6119	&config_file,
6120	format!(
6121	"default_text_model = \"deepseek-v4-pro\"\nprovider = \"volcengine\"\n\n[providers.volcengine]\nmodel = \"{}\"\n",
6122	crate::config::DEFAULT_VOLCENGINE_FLASH_MODEL
6123	),
6124	)?;
6125
6126	let Some((addr, _runtime_threads, handle)) =
6127	spawn_test_server_with_config_path(config_file).await?
6128	else {
6129	return Ok(());
6130	};
6131	let client = crate::tls::reqwest_client();
6132
6133	let body = get_config(&client, &addr).await;
6134	assert_eq!(body["provider"].as_str(), Some("volcengine"));
6135	assert_eq!(
6136	body["model"].as_str(),
6137	Some(crate::config::DEFAULT_VOLCENGINE_FLASH_MODEL),
6138	"GET /v1/config should expose the active provider model, not the root DeepSeek default"
6139	);
6140
6141	handle.abort();
6142	Ok(())
6143	}
6144
6145	#[tokio::test]
6146	async fn api_surfaces_only_configured_model_for_custom_provider_route() -> Result<()> {
6147	let root = std::env::temp_dir().join(format!(
6148	"codewhale-config-custom-provider-model-{}",
6149	Uuid::new_v4()
6150	));
6151	fs::create_dir_all(&root)?;
6152	let config_file = root.join("custom-config.toml");
6153	fs::write(
6154	&config_file,
6155	"provider = \"volcengine\"\n\n[providers.volcengine]\nbase_url = \"https://ark.cn-beijing.volces.com/api/plan/v3\"\nmodel = \"glm-5.2\"\napi_key = \"ark-test\"\n",
6156	)?;
6157
6158	let Some((addr, _runtime_threads, handle)) =
6159	spawn_test_server_with_config_path(config_file).await?
6160	else {
6161	return Ok(());
6162	};
6163	let client = crate::tls::reqwest_client();
6164
6165	let config_body = get_config(&client, &addr).await;
6166	assert_eq!(config_body["provider"].as_str(), Some("volcengine"));
6167	assert_eq!(
6168	config_body["model"].as_str(),
6169	Some("glm-5.2"),
6170	"GET /v1/config should preserve the active provider's explicit custom model"
6171	);
6172
6173	let providers = get_providers(&client, &addr).await;
6174	let volcengine = providers["providers"]
6175	.as_array()
6176	.and_then(\|providers\| {
6177	providers
6178	.iter()
6179	.find(\|entry\| entry["id"].as_str() == Some("volcengine"))
6180	})
6181	.expect("volcengine provider entry");
6182	assert_eq!(providers["current"].as_str(), Some("volcengine"));
6183	assert_eq!(
6184	volcengine["default_model"].as_str(),
6185	Some("glm-5.2"),
6186	"GET /v1/providers should mirror the /provider default route when a saved model override exists"
6187	);
6188
6189	let provider_models = get_provider_models(&client, &addr, "volcengine").await;
6190	let model_ids: Vec<_> = provider_models["models"]
6191	.as_array()
6192	.expect("models array")
6193	.iter()
6194	.filter_map(\|entry\| entry["id"].as_str())
6195	.collect();
6196	assert_eq!(
6197	model_ids.first().copied(),
6198	Some("glm-5.2"),
6199	"configured volcengine model should be the only model exposed for a custom provider route"
6200	);
6201	assert_eq!(model_ids, vec!["glm-5.2"]);
6202
6203	handle.abort();
6204	Ok(())
6205	}
6206
6207	#[tokio::test]
6208	async fn api_surfaces_only_active_model_when_runtime_route_passes_ids_through() -> Result<()> {
6209	let root = std::env::temp_dir().join(format!(
6210	"codewhale-config-runtime-pass-through-{}",
6211	Uuid::new_v4()
6212	));
6213	fs::create_dir_all(&root)?;
6214	let config_file = root.join("custom-config.toml");
6215	fs::write(
6216	&config_file,
6217	"provider = \"volcengine\"\nbase_url = \"https://ark.cn-beijing.volces.com/api/plan/v3\"\n\n[providers.volcengine]\nmodel = \"glm-5.2\"\napi_key = \"ark-test\"\n",
6218	)?;
6219
6220	let Some((addr, _runtime_threads, handle)) =
6221	spawn_test_server_with_config_path(config_file).await?
6222	else {
6223	return Ok(());
6224	};
6225	let client = crate::tls::reqwest_client();
6226
6227	let config_body = get_config(&client, &addr).await;
6228	assert_eq!(config_body["provider"].as_str(), Some("volcengine"));
6229	assert_eq!(config_body["model"].as_str(), Some("glm-5.2"));
6230
6231	let providers = get_providers(&client, &addr).await;
6232	let volcengine = providers["providers"]
6233	.as_array()
6234	.and_then(\|providers\| {
6235	providers
6236	.iter()
6237	.find(\|entry\| entry["id"].as_str() == Some("volcengine"))
6238	})
6239	.expect("volcengine provider entry");
6240	assert_eq!(providers["current"].as_str(), Some("volcengine"));
6241	assert_eq!(volcengine["default_model"].as_str(), Some("glm-5.2"));
6242
6243	let provider_models = get_provider_models(&client, &addr, "volcengine").await;
6244	let model_ids: Vec<_> = provider_models["models"]
6245	.as_array()
6246	.expect("models array")
6247	.iter()
6248	.filter_map(\|entry\| entry["id"].as_str())
6249	.collect();
6250	assert_eq!(model_ids, vec!["glm-5.2"]);
6251
6252	handle.abort();
6253	Ok(())
6254	}
6255
6256	#[tokio::test]
6257	async fn reload_config_reads_from_config_path_and_updates_in_memory_state() -> Result<()> {
6258	// Fix #2 + reload behavior: This test proves that reload reads from the
6259	// `--config` path (not default discovery) and actually updates the
6260	// in-memory state visible to GET /v1/config.
6261	//
6262	// If Fix #2 is reverted (reload uses Config::load(None, None) instead of
6263	// state.config_path), the reload will read an empty/default config and
6264	// the persisted value will NOT appear in GET /v1/config → test fails.
6265	let root =
6266	std::env::temp_dir().join(format!("codewhale-config-reload-path-{}", Uuid::new_v4()));
6267	fs::create_dir_all(&root)?;
6268	let config_file = root.join("custom-config.toml");
6269	fs::write(&config_file, "# initial\n")?;
6270
6271	let Some((addr, _runtime_threads, handle)) =
6272	spawn_test_server_with_config_path(config_file.clone()).await?
6273	else {
6274	return Ok(());
6275	};
6276	let client = crate::tls::reqwest_client();
6277
6278	// Step 1: Record initial model value (should be the default, since
6279	// Config::default() has default_text_model = None).
6280	let before = get_config(&client, &addr).await;
6281	let initial_model = before["model"].as_str().unwrap_or_default().to_string();
6282	assert!(
6283	!initial_model.is_empty(),
6284	"initial model should not be empty"
6285	);
6286	// The initial subagents_max_depth should be DEFAULT_SPAWN_DEPTH (3)
6287	// since Config::default() has no subagents config.
6288	let initial_depth = before["subagents_max_depth"]
6289	.as_u64()
6290	.expect("subagents_max_depth should be a number");
6291	assert_eq!(
6292	initial_depth,
6293	u64::from(codewhale_config::DEFAULT_SPAWN_DEPTH),
6294	"initial subagents_max_depth should be DEFAULT_SPAWN_DEPTH"
6295	);
6296
6297	// Step 2: Persist a new model value to the config file.
6298	// set_config must NOT mutate in-memory state (by design — the caller
6299	// must call /v1/config/reload to apply changes).
6300	// Use a valid DeepSeek model ID so Config::validate() doesn't reject
6301	// the reloaded config.
6302	let test_model = "deepseek-v4-flash";
6303	let (status, body) = post_set_config(&client, &addr, "model", test_model, true).await;
6304	assert_eq!(
6305	status,
6306	StatusCode::OK,
6307	"set_config should succeed, body: {body}"
6308	);
6309
6310	// Step 3: Verify in-memory state is NOT mutated by set_config alone.
6311	let after_set = get_config(&client, &addr).await;
6312	assert_eq!(
6313	after_set["model"].as_str().unwrap_or_default(),
6314	initial_model,
6315	"set_config must NOT update in-memory state before reload"
6316	);
6317
6318	// Step 4: Also persist subagents_max_depth = 5 (below ceiling of 8).
6319	let (status, body) = post_set_config(&client, &addr, "subagents_max_depth", "5", true).await;
6320	assert_eq!(status, StatusCode::OK, "body: {body}");
6321
6322	// Step 5: Reload — this must read from config_file (not default discovery).
6323	let reload_resp = client
6324	.post(format!("http://{addr}/v1/config/reload"))
6325	.send()
6326	.await?;
6327	assert_eq!(reload_resp.status(), StatusCode::OK);
6328
6329	// Step 6: Verify in-memory state IS now updated after reload.
6330	let after_reload = get_config(&client, &addr).await;
6331
6332	// Model should reflect the persisted value.
6333	assert_eq!(
6334	after_reload["model"].as_str().unwrap_or_default(),
6335	test_model,
6336	"after reload, model should be the persisted value — \
6337	if this fails, reload is not reading from config_path"
6338	);
6339
6340	// subagents_max_depth should reflect the persisted value (5).
6341	assert_eq!(
6342	after_reload["subagents_max_depth"].as_u64(),
6343	Some(5),
6344	"after reload, subagents_max_depth should be 5"
6345	);
6346
6347	handle.abort();
6348	Ok(())
6349	}
6350
6351	// ---------------------------------------------------------------------------
6352	// POST /v1/providers/{id}/switch endpoint tests
6353	//
6354	// These tests pin down the TUI-parity contract for the GUI's provider
6355	// picker: a bare switch (no model arg) MUST NOT overwrite the user's
6356	// `[providers.<id>].model` config. Regression for the bug where clicking
6357	// volcengine in the picker forced `model = "deepseek-v4-pro"` even when
6358	// the user had configured `model = "glm-2"`.
6359	// ---------------------------------------------------------------------------
6360
6361	/// Helper: POST to `/v1/providers/{id}/switch` and return the response
6362	/// status + body JSON.
6363	async fn post_switch_provider(
6364	client: &reqwest::Client,
6365	addr: &SocketAddr,
6366	provider: &str,
6367	body: &serde_json::Value,
6368	) -> (reqwest::StatusCode, serde_json::Value) {
6369	let resp = client
6370	.post(format!("http://{addr}/v1/providers/{provider}/switch"))
6371	.json(body)
6372	.send()
6373	.await
6374	.expect("POST /v1/providers/{id}/switch should not fail at transport level");
6375	let status = resp.status();
6376	let body: serde_json::Value = resp
6377	.json()
6378	.await
6379	.unwrap_or_else(\|_\| serde_json::json!({"_error": "non-json response body"}));
6380	(status, body)
6381	}
6382
6383	#[tokio::test]
6384	async fn switch_provider_without_model_arg_preserves_user_per_provider_model() -> Result<()> {
6385	// Regression: clicking volcengine in the GUI picker used to send
6386	// `POST /v1/config { key: "model", value: "deepseek-v4-pro" }` (the
6387	// catalog default), clobbering the user's `[providers.volcengine].model
6388	// = "glm-2"`. The new /v1/providers/{id}/switch endpoint MUST NOT
6389	// touch the model key when no model arg is provided — mirroring the
6390	// TUI's `/provider volcengine` (model: None) flow in
6391	// `commands/groups/core/provider.rs` + `tui/ui.rs::switch_provider`.
6392	let root = std::env::temp_dir().join(format!("codewhale-switch-no-model-{}", Uuid::new_v4()));
6393	fs::create_dir_all(&root)?;
6394	let config_file = root.join("custom-config.toml");
6395	fs::write(
6396	&config_file,
6397	r#"provider = "deepseek"
6398	default_text_model = "deepseek-v4-pro"
6399
6400	[providers.volcengine]
6401	api_key = "ark-test"
6402	base_url = "https://ark.cn-beijing.volces.com/api/plan/v3"
6403	model = "glm-2"
6404	"#,
6405	)?;
6406
6407	let Some((addr, _runtime_threads, handle)) =
6408	spawn_test_server_with_config_path(config_file.clone()).await?
6409	else {
6410	return Ok(());
6411	};
6412	let client = crate::tls::reqwest_client();
6413
6414	// Switch to volcengine WITHOUT a model arg — simulates a picker click.
6415	let (status, body) =
6416	post_switch_provider(&client, &addr, "volcengine", &serde_json::json!({})).await;
6417	assert_eq!(
6418	status,
6419	StatusCode::OK,
6420	"switch should succeed, body: {body}"
6421	);
6422
6423	// Response must report the user's configured model, NOT the catalog
6424	// default "deepseek-v4-pro".
6425	assert_eq!(
6426	body["provider"].as_str(),
6427	Some("volcengine"),
6428	"response should echo the switched-to provider"
6429	);
6430	assert_eq!(
6431	body["model"].as_str(),
6432	Some("glm-2"),
6433	"resolved model must be the user's `[providers.volcengine].model`, \
6434	not the catalog default — if this fails the switch endpoint is \
6435	clobbering per-provider config"
6436	);
6437
6438	// The config file on disk must NOT contain a `model = "deepseek-v4-pro"`
6439	// override for volcengine — `glm-2` must be preserved verbatim.
6440	let persisted = fs::read_to_string(&config_file)?;
6441	assert!(
6442	persisted.contains("model = \"glm-2\""),
6443	"user's `[providers.volcengine].model = \"glm-2\"` must be preserved on disk. \
6444	Actual config:\n{persisted}"
6445	);
6446	assert!(
6447	!persisted
6448	.matches("model = \"deepseek-v4-pro\"")
6449	.count()
6450	.ge(&2),
6451	"switch must not add a second `model = \"deepseek-v4-pro\"` line for volcengine. \
6452	Actual config:\n{persisted}"
6453	);
6454
6455	handle.abort();
6456	Ok(())
6457	}
6458
6459	#[tokio::test]
6460	async fn switch_provider_with_explicit_model_arg_persists_model() -> Result<()> {
6461	// When the user explicitly chooses a model (e.g. `/provider volcengine
6462	// glm-2.5` or a model-picker selection), the switch endpoint MUST
6463	// persist that model — mirroring `switch_provider`'s
6464	// `if model_override.is_some()` branch (ui.rs:9400-9405).
6465	let root = std::env::temp_dir().join(format!("codewhale-switch-with-model-{}", Uuid::new_v4()));
6466	fs::create_dir_all(&root)?;
6467	let config_file = root.join("custom-config.toml");
6468	fs::write(
6469	&config_file,
6470	r#"provider = "deepseek"
6471	default_text_model = "deepseek-v4-pro"
6472
6473	[providers.volcengine]
6474	api_key = "ark-test"
6475	base_url = "https://ark.cn-beijing.volces.com/api/plan/v3"
6476	model = "glm-2"
6477	"#,
6478	)?;
6479
6480	let Some((addr, _runtime_threads, handle)) =
6481	spawn_test_server_with_config_path(config_file.clone()).await?
6482	else {
6483	return Ok(());
6484	};
6485	let client = crate::tls::reqwest_client();
6486
6487	// Switch to volcengine WITH an explicit model arg.
6488	let (status, body) = post_switch_provider(
6489	&client,
6490	&addr,
6491	"volcengine",
6492	&serde_json::json!({ "model": "deepseek-v4-flash" }),
6493	)
6494	.await;
6495	assert_eq!(
6496	status,
6497	StatusCode::OK,
6498	"switch with explicit model should succeed, body: {body}"
6499	);
6500
6501	// The persisted config must reflect the explicit override.
6502	let persisted = fs::read_to_string(&config_file)?;
6503	assert!(
6504	persisted.contains("model = \"deepseek-v4-flash\""),
6505	"explicit model arg must be persisted to `[providers.volcengine].model`. \
6506	Actual config:\n{persisted}"
6507	);
6508
6509	handle.abort();
6510	Ok(())
6511	}
6512
6513	#[tokio::test]
6514	async fn switch_provider_rejects_unknown_provider_id() -> Result<()> {
6515	let root = std::env::temp_dir().join(format!("codewhale-switch-unknown-{}", Uuid::new_v4()));
6516	fs::create_dir_all(&root)?;
6517	let config_file = root.join("custom-config.toml");
6518	fs::write(&config_file, "# empty\n")?;
6519
6520	let Some((addr, _runtime_threads, handle)) =
6521	spawn_test_server_with_config_path(config_file.clone()).await?
6522	else {
6523	return Ok(());
6524	};
6525	let client = crate::tls::reqwest_client();
6526
6527	let (status, _body) = post_switch_provider(
6528	&client,
6529	&addr,
6530	"not-a-real-provider",
6531	&serde_json::json!({}),
6532	)
6533	.await;
6534	assert_eq!(
6535	status,
6536	StatusCode::BAD_REQUEST,
6537	"unknown provider id should return 400"
6538	);
6539
6540	handle.abort();
6541	Ok(())
6542	}
6543
6544	#[tokio::test]
6545	async fn switch_provider_rejects_legacy_deepseek_cn_alias() -> Result<()> {
6546	// The legacy `deepseek-cn` alias has no ProviderKind metadata; the
6547	// GUI must use `deepseek` instead. Same guard as list_provider_models.
6548	let root = std::env::temp_dir().join(format!("codewhale-switch-cn-alias-{}", Uuid::new_v4()));
6549	fs::create_dir_all(&root)?;
6550	let config_file = root.join("custom-config.toml");
6551	fs::write(&config_file, "# empty\n")?;
6552
6553	let Some((addr, _runtime_threads, handle)) =
6554	spawn_test_server_with_config_path(config_file.clone()).await?
6555	else {
6556	return Ok(());
6557	};
6558	let client = crate::tls::reqwest_client();
6559
6560	let (status, body) =
6561	post_switch_provider(&client, &addr, "deepseek-cn", &serde_json::json!({})).await;
6562	assert_eq!(
6563	status,
6564	StatusCode::BAD_REQUEST,
6565	"deepseek-cn should be rejected, body: {body}"
6566	);
6567
6568	handle.abort();
6569	Ok(())
6570	}
6571
6572	#[tokio::test]
6573	async fn switch_provider_with_deepseek_and_explicit_model_updates_default_text_model() -> Result<()>
6574	{
6575	// When switching TO a DeepSeek provider with an explicit model, the
6576	// endpoint must persist `default_text_model` (the DeepSeek-specific
6577	// root key) in addition to the provider change, mirroring
6578	// `switch_provider` in ui.rs which pins `default_model` for DeepSeek.
6579	let root = std::env::temp_dir().join(format!(
6580	"codewhale-switch-deepseek-model-{}",
6581	Uuid::new_v4()
6582	));
6583	fs::create_dir_all(&root)?;
6584	let config_file = root.join("custom-config.toml");
6585	fs::write(
6586	&config_file,
6587	r#"provider = "volcengine"
6588	default_text_model = "old-model"
6589
6590	[providers.volcengine]
6591	api_key = "ark-test"
6592	base_url = "https://ark.cn-beijing.volces.com/api/plan/v3"
6593	model = "glm-2"
6594	"#,
6595	)?;
6596
6597	let Some((addr, _runtime_threads, handle)) =
6598	spawn_test_server_with_config_path(config_file.clone()).await?
6599	else {
6600	return Ok(());
6601	};
6602	let client = crate::tls::reqwest_client();
6603
6604	// Switch to deepseek WITH an explicit model override.
6605	let (status, body) = post_switch_provider(
6606	&client,
6607	&addr,
6608	"deepseek",
6609	&serde_json::json!({ "model": "deepseek-v4-pro" }),
6610	)
6611	.await;
6612	assert_eq!(
6613	status,
6614	StatusCode::OK,
6615	"switch to deepseek with model should succeed, body: {body}"
6616	);
6617
6618	// The persisted config must have provider = "deepseek" and
6619	// default_text_model updated to the explicit model.
6620	let persisted = fs::read_to_string(&config_file)?;
6621	assert!(
6622	persisted.contains("provider = \"deepseek\""),
6623	"provider should be persisted as deepseek. Actual config:\n{persisted}"
6624	);
6625	assert!(
6626	persisted.contains("default_text_model = \"deepseek-v4-pro\""),
6627	"DeepSeek explicit model must be persisted as default_text_model. \
6628	Actual config:\n{persisted}"
6629	);
6630
6631	handle.abort();
6632	Ok(())
6633	}
6634
6635	#[tokio::test]
6636	async fn switch_provider_empty_model_string_treated_as_no_override() -> Result<()> {
6637	// An empty string model (`{ "model": "" }`) must be treated the same
6638	// as no model at all — the endpoint should NOT persist a model key,
6639	// matching the TUI's behavior where a blank model arg is ignored.
6640	let root =
6641	std::env::temp_dir().join(format!("codewhale-switch-empty-model-{}", Uuid::new_v4()));
6642	fs::create_dir_all(&root)?;
6643	let config_file = root.join("custom-config.toml");
6644	fs::write(
6645	&config_file,
6646	r#"provider = "deepseek"
6647	default_text_model = "deepseek-v4-pro"
6648
6649	[providers.volcengine]
6650	api_key = "ark-test"
6651	base_url = "https://ark.cn-beijing.volces.com/api/plan/v3"
6652	model = "glm-2"
6653	"#,
6654	)?;
6655
6656	let Some((addr, _runtime_threads, handle)) =
6657	spawn_test_server_with_config_path(config_file.clone()).await?
6658	else {
6659	return Ok(());
6660	};
6661	let client = crate::tls::reqwest_client();
6662
6663	let (status, body) = post_switch_provider(
6664	&client,
6665	&addr,
6666	"volcengine",
6667	&serde_json::json!({ "model": "" }),
6668	)
6669	.await;
6670	assert_eq!(
6671	status,
6672	StatusCode::OK,
6673	"switch with empty model should succeed, body: {body}"
6674	);
6675
6676	// The user's `model = "glm-2"` must NOT be overwritten.
6677	let persisted = fs::read_to_string(&config_file)?;
6678	assert!(
6679	persisted.contains("model = \"glm-2\""),
6680	"user's model must be preserved when empty-string model is sent. \
6681	Actual config:\n{persisted}"
6682	);
6683
6684	handle.abort();
6685	Ok(())
6686	}
6687
6688	#[tokio::test]
6689	async fn switch_provider_persists_provider_key_on_disk() -> Result<()> {
6690	// Verify that the root `provider = "..."` key is correctly written to
6691	// the config file on disk, not just in the response body.
6692	let root =
6693	std::env::temp_dir().join(format!("codewhale-switch-provider-disk-{}", Uuid::new_v4()));
6694	fs::create_dir_all(&root)?;
6695	let config_file = root.join("custom-config.toml");
6696	fs::write(
6697	&config_file,
6698	r#"provider = "deepseek"
6699	default_text_model = "deepseek-v4-pro"
6700
6701	[providers.volcengine]
6702	api_key = "ark-test"
6703	base_url = "https://ark.cn-beijing.volces.com/api/plan/v3"
6704	model = "glm-2"
6705	"#,
6706	)?;
6707
6708	let Some((addr, _runtime_threads, handle)) =
6709	spawn_test_server_with_config_path(config_file.clone()).await?
6710	else {
6711	return Ok(());
6712	};
6713	let client = crate::tls::reqwest_client();
6714
6715	let (status, _body) =
6716	post_switch_provider(&client, &addr, "volcengine", &serde_json::json!({})).await;
6717	assert_eq!(status, StatusCode::OK);
6718
6719	let persisted = fs::read_to_string(&config_file)?;
6720	assert!(
6721	persisted.contains("provider = \"volcengine\""),
6722	"root `provider` key must be updated on disk. Actual config:\n{persisted}"
6723	);
6724
6725	handle.abort();
6726	Ok(())
6727	}
6728
6729	#[tokio::test]
6730	async fn zai_model_update_is_provider_scoped_and_preserves_deepseek_fallback() -> Result<()> {
6731	let root = std::env::temp_dir().join(format!(
6732	"codewhale-config-zai-model-scope-{}",
6733	Uuid::new_v4()
6734	));
6735	fs::create_dir_all(&root)?;
6736	let config_file = root.join("custom-config.toml");
6737	fs::write(
6738	&config_file,
6739	r#"provider = "zai"
6740	default_text_model = "deepseek-v4-pro"
6741
6742	[providers.zai]
6743	api_key = "zai-test-key"
6744	model = "GLM-5.2"
6745	"#,
6746	)?;
6747
6748	let Some((addr, _runtime_threads, handle)) =
6749	spawn_test_server_with_config_path(config_file.clone()).await?
6750	else {
6751	return Ok(());
6752	};
6753	let client = crate::tls::reqwest_client();
6754
6755	let response = client
6756	.post(format!("http://{addr}/v1/config/reload"))
6757	.send()
6758	.await?;
6759	assert_eq!(response.status(), StatusCode::OK);
6760
6761	let before = get_config(&client, &addr).await;
6762	assert_eq!(before["provider"], "zai");
6763	assert_eq!(before["model"], "GLM-5.2");
6764	assert_eq!(before["default_model"], "deepseek-v4-pro");
6765
6766	let (status, body) = post_set_config(&client, &addr, "model", "glm-5-turbo", true).await;
6767	assert_eq!(status, StatusCode::OK, "body: {body}");
6768	assert_eq!(body["value"], "GLM-5-Turbo");
6769
6770	let persisted = fs::read_to_string(&config_file)?;
6771	let persisted: toml::Value = toml::from_str(&persisted)?;
6772	assert_eq!(
6773	persisted["default_text_model"].as_str(),
6774	Some("deepseek-v4-pro"),
6775	"the active Z.ai model must not overwrite the DeepSeek fallback"
6776	);
6777	assert_eq!(
6778	persisted["providers"]["zai"]["model"].as_str(),
6779	Some("GLM-5-Turbo")
6780	);
6781
6782	let (status, body) = post_set_config(&client, &addr, "model", "deepseek-v4-flash", true).await;
6783	assert_eq!(status, StatusCode::BAD_REQUEST, "body: {body}");
6784
6785	let response = client
6786	.post(format!("http://{addr}/v1/config/reload"))
6787	.send()
6788	.await?;
6789	assert_eq!(response.status(), StatusCode::OK);
6790	let after = get_config(&client, &addr).await;
6791	assert_eq!(after["provider"], "zai");
6792	assert_eq!(after["model"], "GLM-5-Turbo");
6793	assert_eq!(after["default_model"], "deepseek-v4-pro");
6794
6795	handle.abort();
6796	Ok(())
6797	}
6798
6799	#[tokio::test]
6800	async fn reload_config_preserves_profile_selected_named_custom_route() -> Result<()> {
6801	let root = std::env::temp_dir().join(format!(
6802	"codewhale-config-reload-profile-{}",
6803	Uuid::new_v4()
6804	));
6805	fs::create_dir_all(&root)?;
6806	let config_file = root.join("custom-config.toml");
6807	fs::write(
6808	&config_file,
6809	r#"provider = "deepseek"
6810	default_text_model = "deepseek-v4-pro"
6811
6812	[profiles.local]
6813	provider = "lm-studio"
6814
6815	[profiles.local.providers.lm-studio]
6816	kind = "openai-compatible"
6817	base_url = "http://127.0.0.1:18190/v1"
6818	model = "profile-local-model"
6819	api_key = "profile-test-key"
6820	"#,
6821	)?;
6822
6823	let Some((addr, _runtime_threads, handle)) =
6824	spawn_test_server_with_config_path_and_profile(config_file, "local".to_string()).await?
6825	else {
6826	return Ok(());
6827	};
6828	let client = crate::tls::reqwest_client();
6829
6830	let response = client
6831	.post(format!("http://{addr}/v1/config/reload"))
6832	.send()
6833	.await?;
6834	assert_eq!(response.status(), StatusCode::OK);
6835
6836	let config = get_config(&client, &addr).await;
6837	assert_eq!(config["provider"], "lm-studio");
6838	assert_eq!(config["model"], "profile-local-model");
6839	assert_eq!(config["base_url"], "http://127.0.0.1:18190/v1");
6840
6841	handle.abort();
6842	Ok(())
6843	}
6844
6845	#[tokio::test]
6846	async fn reload_config_refreshes_mcp_config_path() -> Result<()> {
6847	// Fix #3: After reload, list_mcp_servers should see the new mcp_config_path
6848	// from the reloaded config (not a stale cached value).
6849	//
6850	// This test works by:
6851	// 1. Starting with config_path pointing to custom-config.toml (initially empty)
6852	// 2. Writing mcp_config_path = <new_path> to the config file via set_config
6853	// 3. Reloading
6854	// 4. GET /v1/config and verifying mcp_config_path field changed
6855	//
6856	// If Fix #3 were still needed (stale mcp_config_path field in state),
6857	// this test would fail because the old field wouldn't update. Since we
6858	// removed the stale field and read directly from config, this test also
6859	// validates that architectural decision.
6860	let root =
6861	std::env::temp_dir().join(format!("codewhale-config-mcp-refresh-{}", Uuid::new_v4()));
6862	fs::create_dir_all(&root)?;
6863	let config_file = root.join("custom-config.toml");
6864	fs::write(&config_file, "# initial\n")?;
6865
6866	let Some((addr, _runtime_threads, handle)) =
6867	spawn_test_server_with_config_path(config_file.clone()).await?
6868	else {
6869	return Ok(());
6870	};
6871	let client = crate::tls::reqwest_client();
6872
6873	// Record initial mcp_config_path (set by test helper to root/mcp.json).
6874	let before = get_config(&client, &addr).await;
6875	let initial_mcp_path = before["mcp_config_path"]
6876	.as_str()
6877	.unwrap_or_default()
6878	.to_string();
6879	assert!(
6880	!initial_mcp_path.is_empty(),
6881	"initial mcp_config_path should not be empty"
6882	);
6883
6884	// Persist a new mcp_config_path to the config file.
6885	let new_mcp_path = root.join("custom-mcp.json");
6886	let new_mcp_path_str = new_mcp_path.to_string_lossy().to_string();
6887	let (status, body) =
6888	post_set_config(&client, &addr, "mcp_config_path", &new_mcp_path_str, true).await;
6889	assert_eq!(status, StatusCode::OK, "body: {body}");
6890
6891	// Before reload, GET should still return the old path.
6892	let after_set = get_config(&client, &addr).await;
6893	assert_eq!(
6894	after_set["mcp_config_path"].as_str().unwrap_or_default(),
6895	initial_mcp_path,
6896	"set_config must NOT update in-memory mcp_config_path before reload"
6897	);
6898
6899	// Reload.
6900	let reload_resp = client
6901	.post(format!("http://{addr}/v1/config/reload"))
6902	.send()
6903	.await?;
6904	assert_eq!(reload_resp.status(), StatusCode::OK);
6905
6906	// After reload, GET should return the new path.
6907	let after_reload = get_config(&client, &addr).await;
6908	let reloaded_mcp_path = after_reload["mcp_config_path"]
6909	.as_str()
6910	.unwrap_or_default()
6911	.to_string();
6912	assert_eq!(
6913	reloaded_mcp_path, new_mcp_path_str,
6914	"after reload, mcp_config_path should reflect the persisted value — \
6915	if this fails, the MCP path is stale after reload"
6916	);
6917
6918	handle.abort();
6919	Ok(())
6920	}
6921
6922	#[tokio::test]
6923	async fn set_config_with_persist_false_does_not_write_to_disk() -> Result<()> {
6924	// Verify the persist:false branch: response reports persisted:false and
6925	// the config file on disk is NOT modified. This is the "dry run" path
6926	// the GUI can use to validate input without committing changes.
6927	let root = std::env::temp_dir().join(format!("codewhale-config-nopersist-{}", Uuid::new_v4()));
6928	fs::create_dir_all(&root)?;
6929	let config_file = root.join("custom-config.toml");
6930	let initial_contents = "# initial empty config\n";
6931	fs::write(&config_file, initial_contents)?;
6932
6933	let Some((addr, _runtime_threads, handle)) =
6934	spawn_test_server_with_config_path(config_file.clone()).await?
6935	else {
6936	return Ok(());
6937	};
6938	let client = crate::tls::reqwest_client();
6939
6940	let (status, body) = post_set_config(&client, &addr, "model", "deepseek-v4-flash", false).await;
6941	assert_eq!(
6942	status,
6943	StatusCode::OK,
6944	"persist:false should still return 200, body: {body}"
6945	);
6946	assert_eq!(
6947	body["persisted"].as_bool(),
6948	Some(false),
6949	"persisted should be false when persist:false, body: {body}"
6950	);
6951	assert_eq!(
6952	body["requires_reload"].as_bool(),
6953	Some(false),
6954	"requires_reload should be false when persist:false, body: {body}"
6955	);
6956	assert_eq!(
6957	body["key"].as_str().unwrap_or_default(),
6958	"model",
6959	"key should echo the request key, body: {body}"
6960	);
6961
6962	// The config file on disk must NOT have been modified.
6963	let contents = fs::read_to_string(&config_file)?;
6964	assert_eq!(
6965	contents, initial_contents,
6966	"persist:false must not modify the config file on disk"
6967	);
6968
6969	handle.abort();
6970	Ok(())
6971	}
6972
6973	#[tokio::test]
6974	async fn set_config_subagents_max_depth_below_ceiling_not_clamped() -> Result<()> {
6975	// Verify that values at and below the ceiling pass through unchanged.
6976	// The existing clamping test only verifies over-ceiling clamping; this
6977	// test ensures legitimate values are not accidentally modified.
6978	let root =
6979	std::env::temp_dir().join(format!("codewhale-config-depth-noclamp-{}", Uuid::new_v4()));
6980	fs::create_dir_all(&root)?;
6981	let config_file = root.join("custom-config.toml");
6982	fs::write(&config_file, "# initial\n")?;
6983
6984	let Some((addr, _runtime_threads, handle)) =
6985	spawn_test_server_with_config_path(config_file.clone()).await?
6986	else {
6987	return Ok(());
6988	};
6989	let client = crate::tls::reqwest_client();
6990
6991	// Test a value at the ceiling (should not be clamped).
6992	let ceiling = u64::from(codewhale_config::MAX_SPAWN_DEPTH_CEILING);
6993	let (status, body) = post_set_config(
6994	&client,
6995	&addr,
6996	"subagents_max_depth",
6997	&ceiling.to_string(),
6998	true,
6999	)
7000	.await;
7001	assert_eq!(status, StatusCode::OK, "body: {body}");
7002	let contents = fs::read_to_string(&config_file)?;
7003	let expected = format!("max_depth = {ceiling}");
7004	assert!(
7005	contents.contains(&expected),
7006	"value at ceiling should be written as-is: expected '{expected}', got: {contents}"
7007	);
7008
7009	// Test a value below the ceiling (should not be clamped).
7010	let below = ceiling.saturating_sub(1);
7011	let (status, body) = post_set_config(
7012	&client,
7013	&addr,
7014	"subagents_max_depth",
7015	&below.to_string(),
7016	true,
7017	)
7018	.await;
7019	assert_eq!(status, StatusCode::OK, "body: {body}");
7020	let contents = fs::read_to_string(&config_file)?;
7021	let expected = format!("max_depth = {below}");
7022	assert!(
7023	contents.contains(&expected),
7024	"value below ceiling should be written as-is: expected '{expected}', got: {contents}"
7025	);
7026
7027	handle.abort();
7028	Ok(())
7029	}
7030
7031	#[tokio::test]
7032	async fn set_config_subagents_enabled_false_persists() -> Result<()> {
7033	// Verify that subagents_enabled=false is properly persisted. The
7034	// existing test only verifies the true branch; this covers the false
7035	// branch to ensure both boolean values round-trip correctly.
7036	let root = std::env::temp_dir().join(format!("codewhale-config-subfalse-{}", Uuid::new_v4()));
7037	fs::create_dir_all(&root)?;
7038	let config_file = root.join("custom-config.toml");
7039	fs::write(&config_file, "[subagents]\nenabled = true\n")?;
7040
7041	let Some((addr, _runtime_threads, handle)) =
7042	spawn_test_server_with_config_path(config_file.clone()).await?
7043	else {
7044	return Ok(());
7045	};
7046	let client = crate::tls::reqwest_client();
7047
7048	let (status, body) = post_set_config(&client, &addr, "subagents_enabled", "false", true).await;
7049	assert_eq!(status, StatusCode::OK, "body: {body}");
7050	assert!(
7051	body["persisted"].as_bool().unwrap_or(false),
7052	"should report persisted=true, body: {body}"
7053	);
7054
7055	let contents = fs::read_to_string(&config_file)?;
7056	assert!(
7057	contents.contains("enabled = false"),
7058	"config file should contain 'enabled = false', got: {contents}"
7059	);
7060
7061	handle.abort();
7062	Ok(())
7063	}
7064
7065	#[tokio::test]
7066	async fn reload_config_with_malformed_file_returns_error() -> Result<()> {
7067	// Verify error handling: if the config file contains invalid TOML,
7068	// reload should return 500 instead of crashing or silently succeeding.
7069	// This catches regressions where the map_err is accidentally removed.
7070	let root = std::env::temp_dir().join(format!("codewhale-config-malformed-{}", Uuid::new_v4()));
7071	fs::create_dir_all(&root)?;
7072	let config_file = root.join("custom-config.toml");
7073	fs::write(&config_file, "# initial\n")?;
7074
7075	let Some((addr, _runtime_threads, handle)) =
7076	spawn_test_server_with_config_path(config_file.clone()).await?
7077	else {
7078	return Ok(());
7079	};
7080	let client = crate::tls::reqwest_client();
7081
7082	// Corrupt the config file with invalid TOML.
7083	fs::write(&config_file, "this is = = not valid toml [[[\n")?;
7084
7085	let resp = client
7086	.post(format!("http://{addr}/v1/config/reload"))
7087	.send()
7088	.await?;
7089	assert_eq!(
7090	resp.status(),
7091	StatusCode::INTERNAL_SERVER_ERROR,
7092	"reload with malformed config should return 500"
7093	);
7094
7095	// Verify the error response has a meaningful message.
7096	let body: serde_json::Value = resp.json().await.unwrap_or_default();
7097	let message = body["error"]["message"]
7098	.as_str()
7099	.unwrap_or_default()
7100	.to_lowercase();
7101	assert!(
7102	message.contains("failed to reload config"),
7103	"error message should mention reload failure, got: {message}"
7104	);
7105
7106	handle.abort();
7107	Ok(())
7108	}
7109
7110	#[tokio::test]
7111	async fn set_config_model_follows_persisted_provider_before_reload() -> Result<()> {
7112	let root = std::env::temp_dir().join(format!(
7113	"codewhale-config-provider-model-{}",
7114	Uuid::new_v4()
7115	));
7116	fs::create_dir_all(&root)?;
7117	let config_file = root.join("custom-config.toml");
7118	fs::write(
7119	&config_file,
7120	format!(
7121	"provider = \"deepseek\"\ndefault_text_model = \"deepseek-v4-pro\"\n\n[providers.volcengine]\nmodel = \"{}\"\n",
7122	crate::config::DEFAULT_VOLCENGINE_MODEL
7123	),
7124	)?;
7125
7126	let Some((addr, _runtime_threads, handle)) =
7127	spawn_test_server_with_config_path(config_file.clone()).await?
7128	else {
7129	return Ok(());
7130	};
7131	let client = crate::tls::reqwest_client();
7132
7133	let (status, body) = post_set_config(&client, &addr, "provider", "volcengine", true).await;
7134	assert_eq!(status, StatusCode::OK, "body: {body}");
7135
7136	let target_model = crate::config::DEFAULT_VOLCENGINE_FLASH_MODEL;
7137	let (status, body) = post_set_config(&client, &addr, "model", target_model, true).await;
7138	assert_eq!(status, StatusCode::OK, "body: {body}");
7139
7140	let config_body = fs::read_to_string(&config_file)?;
7141	assert!(
7142	config_body.contains("provider = \"volcengine\""),
7143	"provider should be persisted before reload"
7144	);
7145	// The persisted value is the normalized wire id (lowercase), not the
7146	// display spelling of DEFAULT_VOLCENGINE_FLASH_MODEL.
7147	let expected_wire = crate::config::normalize_model_name_for_provider(
7148	crate::config::ApiProvider::Volcengine,
7149	target_model,
7150	)
7151	.expect("volcengine flash model should normalize");
7152	assert!(
7153	config_body.contains(&format!("model = \"{expected_wire}\"")),
7154	"volcengine model should be written to the provider table as its wire id"
7155	);
7156	assert!(
7157	config_body.contains("default_text_model = \"deepseek-v4-pro\""),
7158	"switching provider model must not overwrite DeepSeek's root default_text_model"
7159	);
7160
7161	let reload_resp = client
7162	.post(format!("http://{addr}/v1/config/reload"))
7163	.send()
7164	.await?;
7165	assert_eq!(reload_resp.status(), StatusCode::OK);
7166
7167	let after_reload = get_config(&client, &addr).await;
7168	assert_eq!(after_reload["provider"].as_str(), Some("volcengine"));
7169	assert_eq!(after_reload["model"].as_str(), Some(expected_wire.as_str()));
7170
7171	handle.abort();
7172	Ok(())
7173	}
7174
7175	#[tokio::test]
7176	async fn reload_config_applies_multiple_persisted_keys() -> Result<()> {
7177	// Verify that multiple set_config calls accumulate on disk and a single
7178	// reload picks up ALL changes. This catches regressions where reload
7179	// only applies the last-written key or where set_config overwrites
7180	// prior keys unexpectedly.
7181	let root = std::env::temp_dir().join(format!("codewhale-config-multi-{}", Uuid::new_v4()));
7182	fs::create_dir_all(&root)?;
7183	let config_file = root.join("custom-config.toml");
7184	fs::write(&config_file, "# initial\n")?;
7185
7186	let Some((addr, _runtime_threads, handle)) =
7187	spawn_test_server_with_config_path(config_file.clone()).await?
7188	else {
7189	return Ok(());
7190	};
7191	let client = crate::tls::reqwest_client();
7192
7193	// Record initial values.
7194	let before = get_config(&client, &addr).await;
7195	let initial_model = before["model"].as_str().unwrap_or_default().to_string();
7196	let initial_depth = before["subagents_max_depth"].as_u64().unwrap_or(0);
7197	let initial_enabled = before["subagents_enabled"].as_bool().unwrap_or(false);
7198
7199	// Persist three different keys.
7200	// Use a valid DeepSeek model ID so Config::validate() doesn't reject
7201	// the reloaded config.
7202	let test_model = "deepseek-v4-pro";
7203	let (status, body) = post_set_config(&client, &addr, "model", test_model, true).await;
7204	assert_eq!(status, StatusCode::OK, "body: {body}");
7205
7206	let (status, body) = post_set_config(&client, &addr, "subagents_max_depth", "4", true).await;
7207	assert_eq!(status, StatusCode::OK, "body: {body}");
7208
7209	// Flip subagents_enabled to the opposite of its initial value.
7210	let target_enabled = !initial_enabled;
7211	let (status, body) = post_set_config(
7212	&client,
7213	&addr,
7214	"subagents_enabled",
7215	&target_enabled.to_string(),
7216	true,
7217	)
7218	.await;
7219	assert_eq!(status, StatusCode::OK, "body: {body}");
7220
7221	// Before reload, in-memory state should be unchanged for all three keys.
7222	let after_set = get_config(&client, &addr).await;
7223	assert_eq!(
7224	after_set["model"].as_str().unwrap_or_default(),
7225	initial_model,
7226	"model should be unchanged before reload"
7227	);
7228	assert_eq!(
7229	after_set["subagents_max_depth"].as_u64(),
7230	Some(initial_depth),
7231	"subagents_max_depth should be unchanged before reload"
7232	);
7233	assert_eq!(
7234	after_set["subagents_enabled"].as_bool(),
7235	Some(initial_enabled),
7236	"subagents_enabled should be unchanged before reload"
7237	);
7238
7239	// Reload.
7240	let reload_resp = client
7241	.post(format!("http://{addr}/v1/config/reload"))
7242	.send()
7243	.await?;
7244	assert_eq!(reload_resp.status(), StatusCode::OK);
7245
7246	// After reload, ALL three keys should reflect their persisted values.
7247	let after_reload = get_config(&client, &addr).await;
7248	assert_eq!(
7249	after_reload["model"].as_str().unwrap_or_default(),
7250	test_model,
7251	"model should be updated after reload"
7252	);
7253	assert_eq!(
7254	after_reload["subagents_max_depth"].as_u64(),
7255	Some(4),
7256	"subagents_max_depth should be 4 after reload"
7257	);
7258	assert_eq!(
7259	after_reload["subagents_enabled"].as_bool(),
7260	Some(target_enabled),
7261	"subagents_enabled should be {} after reload",
7262	target_enabled
7263	);
7264
7265	handle.abort();
7266	Ok(())
7267	}
7268
7269	#[tokio::test]
7270	async fn set_config_response_contains_all_expected_fields() -> Result<()> {
7271	// Verify the SetConfigResponse shape: key, value, message, persisted,
7272	// requires_reload. This catches serialization regressions and ensures
7273	// the GUI client can rely on these fields being present and correct.
7274	let root = std::env::temp_dir().join(format!("codewhale-config-shape-{}", Uuid::new_v4()));
7275	fs::create_dir_all(&root)?;
7276	let config_file = root.join("custom-config.toml");
7277	fs::write(&config_file, "# initial\n")?;
7278
7279	let Some((addr, _runtime_threads, handle)) =
7280	spawn_test_server_with_config_path(config_file.clone()).await?
7281	else {
7282	return Ok(());
7283	};
7284	let client = crate::tls::reqwest_client();
7285
7286	// persist:true → persisted=true, requires_reload=true
7287	let (status, body) = post_set_config(&client, &addr, "model", "deepseek-v4-flash", true).await;
7288	assert_eq!(status, StatusCode::OK, "body: {body}");
7289	assert_eq!(
7290	body["key"].as_str(),
7291	Some("model"),
7292	"key field, body: {body}"
7293	);
7294	assert_eq!(
7295	body["value"].as_str(),
7296	Some("deepseek-v4-flash"),
7297	"value field, body: {body}"
7298	);
7299	assert!(
7300	body["message"].as_str().is_some_and(\|m\| !m.is_empty()),
7301	"message should be non-empty, body: {body}"
7302	);
7303	assert_eq!(
7304	body["persisted"].as_bool(),
7305	Some(true),
7306	"persisted should be true, body: {body}"
7307	);
7308	assert_eq!(
7309	body["requires_reload"].as_bool(),
7310	Some(true),
7311	"requires_reload should be true when persist:true, body: {body}"
7312	);
7313
7314	// persist:false → persisted=false, requires_reload=false
7315	let (status, body) = post_set_config(&client, &addr, "model", "deepseek-v4-pro", false).await;
7316	assert_eq!(status, StatusCode::OK, "body: {body}");
7317	assert_eq!(
7318	body["key"].as_str(),
7319	Some("model"),
7320	"key field, body: {body}"
7321	);
7322	assert_eq!(
7323	body["value"].as_str(),
7324	Some("deepseek-v4-pro"),
7325	"value field, body: {body}"
7326	);
7327	assert_eq!(
7328	body["persisted"].as_bool(),
7329	Some(false),
7330	"persisted should be false, body: {body}"
7331	);
7332	assert_eq!(
7333	body["requires_reload"].as_bool(),
7334	Some(false),
7335	"requires_reload should be false when persist:false, body: {body}"
7336	);
7337
7338	handle.abort();
7339	Ok(())
7340	}
7341
7342	#[tokio::test]
7343	async fn cors_layer_advertises_exact_supported_headers_and_never_an_extra() -> Result<()> {
7344	let layer = cors_layer(&[]);
7345	let router: Router = Router::new()
7346	.route("/probe", get(\|\| async { "ok" }))
7347	.layer(layer);
7348
7349	let listener = match TcpListener::bind("127.0.0.1:0").await {
7350	Ok(listener) => listener,
7351	Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return Ok(()),
7352	Err(err) => return Err(err.into()),
7353	};
7354	let addr = listener.local_addr()?;
7355	let handle = tokio::spawn(async move {
7356	let _ = axum::serve(listener, router).await;
7357	});
7358
7359	let client = crate::tls::reqwest_client();
7360
7361	let allowed = client
7362	.request(reqwest::Method::OPTIONS, format!("http://{addr}/probe"))
7363	.header("Origin", "http://localhost:1420")
7364	.header("Access-Control-Request-Method", "POST")
7365	.header(
7366	"Access-Control-Request-Headers",
7367	"authorization, content-type, accept, x-codewhale-runtime-token, x-deepseek-runtime-token",
7368	)
7369	.send()
7370	.await?;
7371
7372	assert!(allowed.status().is_success());
7373	assert_eq!(
7374	allowed
7375	.headers()
7376	.get("access-control-allow-origin")
7377	.and_then(\|value\| value.to_str().ok()),
7378	Some("http://localhost:1420")
7379	);
7380	let allow_headers = allowed
7381	.headers()
7382	.get("access-control-allow-headers")
7383	.and_then(\|v\| v.to_str().ok())
7384	.unwrap_or("");
7385	let advertised = allow_headers
7386	.split(',')
7387	.map(str::trim)
7388	.map(str::to_ascii_lowercase)
7389	.collect::<std::collections::BTreeSet<_>>();
7390	let expected = [
7391	"accept",
7392	"authorization",
7393	"content-type",
7394	"x-codewhale-runtime-token",
7395	"x-deepseek-runtime-token",
7396	]
7397	.into_iter()
7398	.map(str::to_string)
7399	.collect::<std::collections::BTreeSet<_>>();
7400
7401	assert_eq!(advertised, expected);
7402
7403	let unapproved = client
7404	.request(reqwest::Method::OPTIONS, format!("http://{addr}/probe"))
7405	.header("Origin", "http://localhost:1420")
7406	.header("Access-Control-Request-Method", "POST")
7407	.header(
7408	"Access-Control-Request-Headers",
7409	"authorization, x-malicious-header",
7410	)
7411	.send()
7412	.await?;
7413	let unapproved_headers = unapproved
7414	.headers()
7415	.get("access-control-allow-headers")
7416	.and_then(\|value\| value.to_str().ok())
7417	.unwrap_or_default()
7418	.to_ascii_lowercase();
7419	assert!(
7420	!unapproved_headers.contains("x-malicious-header"),
7421	"an unapproved request header must never be advertised to the browser"
7422	);
7423
7424	handle.abort();
7425	Ok(())
7426	}
7427

7427 lines RUST