lib.rs - CodeWhale

lib.rs

1	use std::collections::{HashMap, VecDeque};
2	use std::net::SocketAddr;
3	use std::path::{Path, PathBuf};
4	use std::process::{Child, Command, Stdio};
5	use std::sync::Arc;
6	use std::time::{Duration, Instant};
7
8	use anyhow::{Context, Result, anyhow, bail};
9	use axum::extract::{DefaultBodyLimit, Request, State};
10	use axum::http::{HeaderValue, Method, StatusCode, header};
11	use axum::middleware::{self, Next};
12	use axum::response::{IntoResponse, Response};
13	use axum::routing::{get, post};
14	use axum::{Json, Router};
15	use codewhale_agent::ModelRegistry;
16	use codewhale_config::{CliRuntimeOverrides, ConfigStore};
17	use codewhale_core::Runtime;
18	use codewhale_hooks::{HookDispatcher, JsonlHookSink, StdoutHookSink, UnixSocketHookSink};
19	use codewhale_mcp::McpManager;
20	use codewhale_protocol::{
21	AppRequest, AppResponse, PromptRequest, PromptResponse, ThreadGoalClearParams,
22	ThreadGoalGetParams, ThreadGoalSetParams, ThreadRequest, ThreadResponse, UserInputAnswerEvent,
23	};
24	use codewhale_state::StateStore;
25	use codewhale_tools::{ToolCall, ToolRegistry};
26	use serde::de::DeserializeOwned;
27	use serde::{Deserialize, Serialize};
28	use serde_json::{Value, json};
29	use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader};
30	use tokio::sync::{Mutex, RwLock};
31	use tower_http::cors::CorsLayer;
32	use uuid::Uuid;
33
34	/// Answers submitted for a pending `request_user_input` clarification.
35	///
36	/// The headless runtime emits [`codewhale_protocol::EventFrame::UserInputRequest`]
37	/// fire-and-return (it has no resume channel, mirroring headless approval).
38	/// Clients POST answers back via [`AppRequest::SubmitUserInput`]; we record
39	/// them here keyed by `request_id` so a driver can retrieve and feed them into
40	/// the next turn as structured context. True in-flight resume would require an
41	/// awaiter in `invoke_tool` and is left as a follow-up.
42	type PendingUserInputAnswers = Vec<UserInputAnswerEvent>;
43
44	mod chat_completions;
45
46	/// Legacy DeepSeek-era naming kept for external compatibility.
47	///
48	/// CodeWhale began life as a DeepSeek CLI; existing health probes, SDK
49	/// harnesses, and on-disk layouts still key off these names. Every remaining
50	/// legacy reference in this crate routes through this shim so a future
51	/// coordinated migration touches exactly one place (repo policy: preserve
52	/// legacy migration care).
53	mod legacy_deepseek_compat {
54	use std::path::PathBuf;
55
56	/// Service name advertised by the HTTP and stdio health probes.
57	pub(crate) const SERVICE_NAME: &str = "deepseek-app-server";
58
59	/// Fallback hook-event log location used when no config path is
60	/// provided (legacy `.deepseek/` dot-directory layout).
61	pub(crate) fn default_events_log_path() -> PathBuf {
62	PathBuf::from(".deepseek/events.jsonl")
63	}
64	}
65
66	/// Upper bound on JSON request bodies accepted by the HTTP app-server.
67	const MAX_HTTP_BODY_BYTES: usize = 16 * 1024 * 1024;
68	const MAX_SSE_FRAME_BYTES: usize = 16 * 1024 * 1024;
69
70	const DEFAULT_CORS_ORIGINS: &[&str] = &[
71	"http://localhost",
72	"http://localhost:1420",
73	"http://localhost:3000",
74	"http://localhost:5173",
75	"http://127.0.0.1",
76	"http://127.0.0.1:1420",
77	"tauri://localhost",
78	];
79
80	#[derive(Clone)]
81	pub struct AppServerOptions {
82	pub listen: SocketAddr,
83	pub config_path: Option<PathBuf>,
84	pub auth_token: Option<String>,
85	pub insecure_no_auth: bool,
86	pub cors_origins: Vec<String>,
87	}
88
89	impl std::fmt::Debug for AppServerOptions {
90	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91	f.debug_struct("AppServerOptions")
92	.field("listen", &self.listen)
93	.field("config_path", &self.config_path)
94	.field(
95	"auth_token",
96	&self.auth_token.as_ref().map(\|_\| "<redacted>"),
97	)
98	.field("insecure_no_auth", &self.insecure_no_auth)
99	.field("cors_origins", &self.cors_origins)
100	.finish()
101	}
102	}
103
104	/// Cached stdio→runtime bridge handle.
105	///
106	/// The outer [`AppState::stdio_bridge`] mutex guards only the cache slot;
107	/// this inner mutex serializes traffic on one bridge (single child process
108	/// plus per-thread seq bookkeeping requires ordered access).
109	type SharedRuntimeBridge = Arc<Mutex<RuntimeBridge>>;
110
111	#[derive(Clone)]
112	struct AppState {
113	config_path: Option<PathBuf>,
114	config: Arc<RwLock<codewhale_config::ConfigToml>>,
115	/// Read/write split mirrors [`Runtime`]'s own receivers: `&self`
116	/// operations (tool calls, status, MCP startup) share a read guard and
117	/// run concurrently; `&mut self` turns (prompt/thread) and config pushes
118	/// take the write guard because the runtime genuinely requires
119	/// exclusivity there.
120	runtime: Arc<RwLock<Runtime>>,
121	registry: ModelRegistry,
122	auth_token: Option<String>,
123	stdio_bridge: Arc<Mutex<Option<SharedRuntimeBridge>>>,
124	stdio_thread_hints: Arc<Mutex<HashMap<String, RuntimeThreadHint>>>,
125	/// Answers submitted via `AppRequest::SubmitUserInput`, keyed by
126	/// `request_id`. A driver polls this to resolve clarification questions
127	/// raised by the model during a headless run.
128	pending_user_input: Arc<Mutex<std::collections::HashMap<String, PendingUserInputAnswers>>>,
129	/// Turns currently streaming over stdio, keyed by stdio thread id.
130	///
131	/// Deliberately kept outside the bridge mutex: a streaming turn holds
132	/// that mutex for its entire duration, so anything reachable only through
133	/// it cannot be used to stop the turn. This holds its own copy of what an
134	/// interrupt needs, so a cancel never waits on the turn it is cancelling.
135	in_flight_turns: Arc<Mutex<HashMap<String, InFlightTurn>>>,
136	}
137
138	/// Everything needed to interrupt a running turn without the bridge lock.
139	#[derive(Debug, Clone)]
140	struct InFlightTurn {
141	base_url: String,
142	auth_token: Option<String>,
143	/// Thread id as the runtime knows it, not the stdio-facing id.
144	runtime_thread_id: String,
145	turn_id: String,
146	}
147
148	type TurnRegistry = Arc<Mutex<HashMap<String, InFlightTurn>>>;
149
150	#[derive(Debug, Clone, Serialize, Deserialize)]
151	struct ToolCallRequest {
152	call: ToolCall,
153	#[serde(default)]
154	cwd: Option<PathBuf>,
155	}
156
157	#[derive(Debug, Deserialize)]
158	struct JsonRpcRequest {
159	#[serde(default)]
160	jsonrpc: Option<String>,
161	#[serde(default)]
162	id: Option<Value>,
163	method: String,
164	#[serde(default)]
165	params: Value,
166	}
167
168	#[derive(Debug)]
169	struct JsonRpcError {
170	code: i64,
171	message: String,
172	data: Option<Value>,
173	}
174
175	#[derive(Debug)]
176	struct StdioDispatchResult {
177	result: Value,
178	should_exit: bool,
179	}
180
181	#[derive(Debug)]
182	struct RuntimeBridge {
183	base_url: String,
184	client: reqwest::Client,
185	auth_token: Option<String>,
186	child: Option<Child>,
187	thread_map: HashMap<String, String>,
188	last_seq_by_thread: HashMap<String, u64>,
189	}
190
191	#[derive(Debug, Clone, Default)]
192	struct RuntimeThreadHint {
193	model: Option<String>,
194	workspace: Option<PathBuf>,
195	}
196
197	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
198	enum TurnTerminalStatus {
199	Completed,
200	Failed,
201	Interrupted,
202	Canceled,
203	}
204
205	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
206	enum AppTransport {
207	Http,
208	Stdio,
209	}
210
211	#[derive(Debug, Deserialize)]
212	struct ConfigGetParams {
213	key: String,
214	}
215
216	#[derive(Debug, Deserialize)]
217	struct ConfigSetParams {
218	key: String,
219	value: String,
220	}
221
222	#[derive(Debug, Deserialize)]
223	struct ThreadIdParams {
224	thread_id: String,
225	}
226
227	#[derive(Debug, Deserialize)]
228	struct ThreadMessageParams {
229	thread_id: String,
230	input: String,
231	}
232
233	#[derive(Debug, Deserialize)]
234	struct ThreadInterruptParams {
235	thread_id: String,
236	}
237
238	pub async fn run(options: AppServerOptions) -> Result<()> {
239	let auth_token = resolve_auth_token(&options)?;
240	let state = build_state(options.config_path.clone(), auth_token)?;
241	let app = app_router(state, &options.cors_origins);
242
243	let listener = tokio::net::TcpListener::bind(options.listen).await?;
244	axum::serve(listener, app)
245	.with_graceful_shutdown(shutdown_signal())
246	.await?;
247	Ok(())
248	}
249
250	async fn shutdown_signal() {
251	let ctrl_c = async {
252	let _ = tokio::signal::ctrl_c().await;
253	};
254
255	#[cfg(unix)]
256	let terminate = async {
257	match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
258	Ok(mut signal) => {
259	signal.recv().await;
260	}
261	Err(_) => std::future::pending::<()>().await,
262	}
263	};
264
265	#[cfg(not(unix))]
266	let terminate = std::future::pending::<()>();
267
268	tokio::select! {
269	_ = ctrl_c => {}
270	_ = terminate => {}
271	}
272	}
273
274	fn app_router(state: AppState, cors_origins: &[String]) -> Router {
275	let protected_routes = Router::new()
276	.route("/thread", post(thread_handler))
277	.route("/app", post(app_handler))
278	.route("/prompt", post(prompt_handler))
279	.route("/tool", post(tool_handler))
280	.route("/jobs", get(jobs_handler))
281	.route("/mcp/startup", post(mcp_startup_handler))
282	.route(
283	"/v1/chat/completions",
284	post(chat_completions::chat_completions_handler),
285	)
286	.route_layer(middleware::from_fn_with_state(
287	state.clone(),
288	require_app_server_token,
289	));
290
291	Router::new()
292	.route("/healthz", get(healthz))
293	.merge(protected_routes)
294	.layer(DefaultBodyLimit::max(MAX_HTTP_BODY_BYTES))
295	.layer(cors_layer(cors_origins))
296	.with_state(state)
297	}
298
299	pub async fn run_stdio(config_path: Option<PathBuf>) -> Result<()> {
300	let state = build_state_with_transport(config_path, None, AppTransport::Stdio)?;
301	let reader = BufReader::new(tokio::io::stdin()).lines();
302	let writer = tokio::io::BufWriter::new(tokio::io::stdout());
303	run_stdio_loop(&state, reader, writer).await
304	}
305
306	/// The stdio JSON-RPC loop, generic over its transport so it can be driven by
307	/// a duplex pipe in tests rather than the process's real stdin/stdout.
308	async fn run_stdio_loop<R, W>(
309	state: &AppState,
310	mut reader: tokio::io::Lines<R>,
311	mut writer: W,
312	) -> Result<()>
313	where
314	R: AsyncBufRead + Unpin,
315	W: AsyncWrite + Unpin,
316	{
317	// Work that arrived while a turn was streaming. The turn owns the writer
318	// for its whole duration, so these wait for it rather than interleaving
319	// into the middle of a response.
320	let mut pending: VecDeque<PendingStdioWork> = VecDeque::new();
321	let mut stdin_open = true;
322
323	loop {
324	let request = match pending.pop_front() {
325	Some(PendingStdioWork::Response(response)) => {
326	write_stdio_line(&mut writer, &response).await?;
327	continue;
328	}
329	Some(PendingStdioWork::Request(request)) => request,
330	None => {
331	if !stdin_open {
332	break;
333	}
334	let Some(line) = reader.next_line().await? else {
335	break;
336	};
337	match parse_stdio_line(&line) {
338	ParsedStdioLine::Blank => continue,
339	ParsedStdioLine::Rejected(response) => {
340	write_stdio_line(&mut writer, &response).await?;
341	continue;
342	}
343	ParsedStdioLine::Request(request) => request,
344	}
345	}
346	};
347
348	let id = request.id.clone();
349	let dispatched = if request.method == "thread/message" {
350	// A turn can run for minutes. Keep reading stdin while it streams
351	// so an interrupt (or a shutdown) can actually reach it — with a
352	// plain `await` here, nothing could be read until it finished.
353	let dispatch = dispatch_stdio_request_with_writer(
354	state,
355	&mut writer,
356	&request.method,
357	request.params,
358	);
359	tokio::pin!(dispatch);
360	loop {
361	tokio::select! {
362	outcome = &mut dispatch => break outcome,
363	line = reader.next_line(), if stdin_open => {
364	match line? {
365	None => stdin_open = false,
366	Some(line) => {
367	handle_line_during_turn(state, &line, &mut pending).await;
368	}
369	}
370	}
371	}
372	}
373	} else {
374	dispatch_stdio_request_with_writer(state, &mut writer, &request.method, request.params)
375	.await
376	};
377
378	match dispatched {
379	Ok(dispatch) => {
380	write_stdio_line(&mut writer, &jsonrpc_result(id, dispatch.result)).await?;
381	if dispatch.should_exit {
382	break;
383	}
384	}
385	Err(err) => {
386	write_stdio_line(&mut writer, &jsonrpc_error(id, err)).await?;
387	}
388	}
389	}
390
391	Ok(())
392	}
393
394	/// Work deferred until a streaming turn releases the writer.
395	enum PendingStdioWork {
396	/// Already answered (an interrupt acted immediately); just needs writing.
397	Response(Value),
398	/// Not started yet; runs normally once the turn is done.
399	Request(JsonRpcRequest),
400	}
401
402	enum ParsedStdioLine {
403	Blank,
404	Request(JsonRpcRequest),
405	Rejected(Value),
406	}
407
408	fn parse_stdio_line(line: &str) -> ParsedStdioLine {
409	if line.trim().is_empty() {
410	return ParsedStdioLine::Blank;
411	}
412	let request: JsonRpcRequest = match serde_json::from_str(line) {
413	Ok(value) => value,
414	Err(err) => {
415	return ParsedStdioLine::Rejected(jsonrpc_error(
416	None,
417	JsonRpcError::parse_error(format!("invalid json: {err}")),
418	));
419	}
420	};
421	if request
422	.jsonrpc
423	.as_deref()
424	.is_some_and(\|version\| version != "2.0")
425	{
426	return ParsedStdioLine::Rejected(jsonrpc_error(
427	request.id,
428	JsonRpcError::invalid_request("jsonrpc version must be 2.0"),
429	));
430	}
431	ParsedStdioLine::Request(request)
432	}
433
434	/// Triage a request that arrived mid-turn.
435	///
436	/// Cancellation is the whole point of reading here, so `thread/interrupt`
437	/// runs immediately and only its reply waits for the writer. `shutdown` also
438	/// interrupts immediately — otherwise it would block on the bridge mutex the
439	/// turn is holding — and then queues so the turn can unwind first. Everything
440	/// else simply queues: it was never urgent, and running it now would race the
441	/// turn for the writer.
442	async fn handle_line_during_turn(
443	state: &AppState,
444	line: &str,
445	pending: &mut VecDeque<PendingStdioWork>,
446	) {
447	let request = match parse_stdio_line(line) {
448	ParsedStdioLine::Blank => return,
449	ParsedStdioLine::Rejected(response) => {
450	pending.push_back(PendingStdioWork::Response(response));
451	return;
452	}
453	ParsedStdioLine::Request(request) => request,
454	};
455
456	match request.method.as_str() {
457	"thread/interrupt" => {
458	let id = request.id.clone();
459	let response = match parse_params::<ThreadInterruptParams>(params_or_object(
460	request.params.clone(),
461	)) {
462	Ok(parsed) => match interrupt_stdio_turn(state, &parsed.thread_id).await {
463	Ok(interrupted) => jsonrpc_result(
464	id,
465	json!({ "thread_id": parsed.thread_id, "interrupted": interrupted }),
466	),
467	Err(err) => jsonrpc_error(id, err),
468	},
469	Err(err) => jsonrpc_error(id, err),
470	};
471	pending.push_back(PendingStdioWork::Response(response));
472	}
473	"shutdown" => {
474	let live: Vec<String> = state.in_flight_turns.lock().await.keys().cloned().collect();
475	for thread_id in live {
476	let _ = interrupt_stdio_turn(state, &thread_id).await;
477	}
478	pending.push_back(PendingStdioWork::Request(request));
479	}
480	_ => pending.push_back(PendingStdioWork::Request(request)),
481	}
482	}
483
484	async fn write_stdio_line<W: AsyncWrite + Unpin>(writer: &mut W, response: &Value) -> Result<()> {
485	writer.write_all(&serde_json::to_vec(response)?).await?;
486	writer.write_all(b"\n").await?;
487	writer.flush().await?;
488	Ok(())
489	}
490
491	async fn healthz() -> Json<Value> {
492	Json(json!({
493	"status": "ok",
494	"protocol": "v2",
495	"service": legacy_deepseek_compat::SERVICE_NAME
496	}))
497	}
498
499	async fn thread_handler(
500	State(state): State<AppState>,
501	Json(req): Json<ThreadRequest>,
502	) -> (StatusCode, Json<ThreadResponse>) {
503	let mut runtime = state.runtime.write().await;
504	match runtime.handle_thread(req).await {
505	Ok(res) => (StatusCode::OK, Json(res)),
506	Err(err) => (
507	StatusCode::INTERNAL_SERVER_ERROR,
508	Json(ThreadResponse {
509	thread_id: "error".to_string(),
510	status: format!("error:{err}"),
511	thread: None,
512	threads: Vec::new(),
513	goal: None,
514	model: None,
515	model_provider: None,
516	cwd: None,
517	approval_policy: None,
518	sandbox: None,
519	events: Vec::new(),
520	data: json!({}),
521	}),
522	),
523	}
524	}
525
526	async fn prompt_handler(
527	State(state): State<AppState>,
528	Json(req): Json<PromptRequest>,
529	) -> (StatusCode, Json<PromptResponse>) {
530	let mut runtime = state.runtime.write().await;
531	let overrides = CliRuntimeOverrides::default();
532	match runtime.handle_prompt(req, &overrides).await {
533	Ok(res) => (StatusCode::OK, Json(res)),
534	Err(err) => (
535	StatusCode::INTERNAL_SERVER_ERROR,
536	Json(PromptResponse {
537	output: err.to_string(),
538	model: "unknown".to_string(),
539	events: Vec::new(),
540	}),
541	),
542	}
543	}
544
545	async fn tool_handler(
546	State(state): State<AppState>,
547	Json(req): Json<ToolCallRequest>,
548	) -> (StatusCode, Json<Value>) {
549	let cwd = req
550	.cwd
551	.unwrap_or_else(\|\| std::env::current_dir().unwrap_or_else(\|_\| PathBuf::from(".")));
552	// Resolve approval policy from config instead of hardcoding.
553	let approval_mode = {
554	let cfg = state.config.read().await;
555	cfg.approval_policy
556	.as_deref()
557	.and_then(\|p\| match p.trim().to_ascii_lowercase().as_str() {
558	"auto" \| "yolo" => Some(codewhale_execpolicy::AskForApproval::UnlessTrusted),
559	"never" \| "deny" => Some(codewhale_execpolicy::AskForApproval::Never),
560	_ => None,
561	})
562	.unwrap_or(codewhale_execpolicy::AskForApproval::OnRequest)
563	};
564	// `invoke_tool` takes `&self`, so long-running tool executions share a
565	// read guard: they run concurrently with each other and with status
566	// reads instead of serializing every request behind one Mutex.
567	let runtime = state.runtime.read().await;
568	match runtime.invoke_tool(req.call, approval_mode, &cwd).await {
569	Ok(value) => (StatusCode::OK, Json(value)),
570	Err(err) => (
571	StatusCode::INTERNAL_SERVER_ERROR,
572	Json(json!({ "ok": false, "error": err.to_string() })),
573	),
574	}
575	}
576
577	async fn jobs_handler(State(state): State<AppState>) -> Json<AppResponse> {
578	let runtime = state.runtime.read().await;
579	Json(runtime.app_status())
580	}
581
582	async fn mcp_startup_handler(State(state): State<AppState>) -> Json<Value> {
583	let runtime = state.runtime.read().await;
584	let summary = runtime.mcp_startup().await;
585	Json(json!({
586	"ok": true,
587	"summary": summary
588	}))
589	}
590
591	async fn app_handler(
592	State(state): State<AppState>,
593	Json(req): Json<AppRequest>,
594	) -> (StatusCode, Json<AppResponse>) {
595	let response = process_app_request(&state, req, AppTransport::Http).await;
596	(app_response_status(&response), Json(response))
597	}
598
599	fn app_response_status(response: &AppResponse) -> StatusCode {
600	if response.ok {
601	return StatusCode::OK;
602	}
603	if response.data.get("request_id").is_some() {
604	StatusCode::CONFLICT
605	} else if response
606	.data
607	.get("error")
608	.and_then(Value::as_str)
609	.is_some_and(\|err\| err.contains("failed to load config"))
610	{
611	StatusCode::INTERNAL_SERVER_ERROR
612	} else {
613	StatusCode::BAD_REQUEST
614	}
615	}
616
617	fn build_state(config_path: Option<PathBuf>, auth_token: Option<String>) -> Result<AppState> {
618	build_state_with_transport(config_path, auth_token, AppTransport::Http)
619	}
620
621	fn build_state_with_transport(
622	config_path: Option<PathBuf>,
623	auth_token: Option<String>,
624	transport: AppTransport,
625	) -> Result<AppState> {
626	let has_explicit_config_path = config_path.is_some();
627	let store = ConfigStore::load(config_path)?;
628	let config_path = has_explicit_config_path.then(\|\| store.path().to_path_buf());
629	let config = store.config.clone();
630	let exec_policy = store.exec_policy_engine();
631	let registry = ModelRegistry::default();
632
633	let state_db_path = config_path
634	.as_ref()
635	.and_then(\|p\| p.parent().map(\|parent\| parent.join("state.db")));
636	let state_store = StateStore::open(state_db_path)?;
637
638	let mut hooks = HookDispatcher::default();
639	// Stdio carries JSON-RPC on stdout: printing raw hook events there
640	// corrupts the protocol stream (#5165). HTTP mode keeps the stdout
641	// sink for local development visibility.
642	if transport == AppTransport::Http {
643	hooks.add_sink(Arc::new(StdoutHookSink));
644	}
645	let hook_log_path = config_path
646	.as_ref()
647	.and_then(\|p\| p.parent().map(\|parent\| parent.join("events.jsonl")))
648	.unwrap_or_else(legacy_deepseek_compat::default_events_log_path);
649	hooks.add_sink(Arc::new(JsonlHookSink::new(hook_log_path)));
650
651	if let Some(socket_path) = config
652	.hook_sinks
653	.as_ref()
654	.and_then(\|sinks\| sinks.unix_socket_path.as_ref())
655	.filter(\|path\| !path.as_os_str().is_empty())
656	{
657	hooks.add_sink(Arc::new(UnixSocketHookSink::new(socket_path.clone())));
658	}
659
660	let runtime = Runtime::new(
661	config.clone(),
662	registry.clone(),
663	state_store,
664	Arc::new(ToolRegistry::default()),
665	Arc::new(McpManager::default()),
666	exec_policy,
667	hooks,
668	);
669
670	Ok(AppState {
671	config_path,
672	config: Arc::new(RwLock::new(config)),
673	runtime: Arc::new(RwLock::new(runtime)),
674	registry,
675	auth_token,
676	stdio_bridge: Arc::new(Mutex::new(None)),
677	stdio_thread_hints: Arc::new(Mutex::new(HashMap::new())),
678	pending_user_input: Arc::new(Mutex::new(std::collections::HashMap::new())),
679	in_flight_turns: Arc::new(Mutex::new(HashMap::new())),
680	})
681	}
682
683	fn resolve_auth_token(options: &AppServerOptions) -> Result<Option<String>> {
684	let configured = options.auth_token.as_ref().map(\|token\| token.trim());
685	if let Some(token) = configured
686	&& token.is_empty()
687	{
688	bail!("app-server auth token cannot be empty");
689	}
690	let has_explicit_token = configured.is_some();
691
692	if options.insecure_no_auth {
693	if !options.listen.ip().is_loopback() {
694	bail!("refusing unauthenticated app-server bind on non-loopback address");
695	}
696	eprintln!("warning: app-server HTTP auth disabled by --insecure-no-auth");
697	return Ok(None);
698	}
699
700	if !has_explicit_token && !options.listen.ip().is_loopback() {
701	bail!(
702	"refusing non-loopback app-server bind without explicit auth token; pass --auth-token or set CODEWHALE_APP_SERVER_TOKEN"
703	);
704	}
705
706	let token = configured
707	.map(str::to_string)
708	.unwrap_or_else(\|\| format!("cwapp_{}", Uuid::new_v4().simple()));
709	for line in app_server_auth_status_lines(has_explicit_token) {
710	eprintln!("{line}");
711	}
712	Ok(Some(token))
713	}
714
715	fn app_server_auth_status_lines(has_explicit_token: bool) -> Vec<&'static str> {
716	if has_explicit_token {
717	return vec!["app-server auth: bearer token required for HTTP routes."];
718	}
719	vec![
720	"app-server auth: generated bearer token for this process (not printed).",
721	" Pass --auth-token or set CODEWHALE_APP_SERVER_TOKEN when another client needs to connect.",
722	]
723	}
724
725	fn cors_layer(extra_origins: &[String]) -> CorsLayer {
726	let mut origins: Vec<HeaderValue> = DEFAULT_CORS_ORIGINS
727	.iter()
728	.filter_map(\|origin\| HeaderValue::from_str(origin).ok())
729	.collect();
730	for raw in extra_origins {
731	let trimmed = raw.trim();
732	if trimmed.is_empty() {
733	continue;
734	}
735	match HeaderValue::from_str(trimmed) {
736	Ok(value) if !origins.contains(&value) => origins.push(value),
737	Ok(_) => {}
738	Err(err) => {
739	eprintln!("warning: ignoring invalid app-server CORS origin `{trimmed}`: {err}")
740	}
741	}
742	}
743
744	CorsLayer::new()
745	.allow_origin(origins)
746	.allow_methods([Method::GET, Method::POST, Method::OPTIONS])
747	.allow_headers([header::AUTHORIZATION, header::CONTENT_TYPE])
748	}
749
750	async fn require_app_server_token(
751	State(state): State<AppState>,
752	req: Request,
753	next: Next,
754	) -> Response {
755	let Some(expected) = state.auth_token.as_deref() else {
756	return next.run(req).await;
757	};
758	let authorized = req
759	.headers()
760	.get(header::AUTHORIZATION)
761	.and_then(\|value\| value.to_str().ok())
762	.and_then(\|raw\| raw.strip_prefix("Bearer "))
763	.is_some_and(\|token\| constant_time_eq(token.as_bytes(), expected.as_bytes()));
764
765	if authorized {
766	next.run(req).await
767	} else {
768	(
769	StatusCode::UNAUTHORIZED,
770	Json(json!({
771	"error": {
772	"message": "app-server bearer token required",
773	"status": StatusCode::UNAUTHORIZED.as_u16(),
774	}
775	})),
776	)
777	.into_response()
778	}
779	}
780
781	/// Compares the full length of both inputs regardless of where they first
782	/// differ, so auth failures don't leak the matching prefix length via timing.
783	fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
784	let mut diff = a.len() ^ b.len();
785	for i in 0..a.len().max(b.len()) {
786	let x = a.get(i).copied().unwrap_or(0);
787	let y = b.get(i).copied().unwrap_or(0);
788	diff \|= usize::from(x ^ y);
789	}
790	diff == 0
791	}
792
793	fn params_or_object(params: Value) -> Value {
794	if params.is_null() { json!({}) } else { params }
795	}
796
797	fn parse_params<T: DeserializeOwned>(params: Value) -> std::result::Result<T, JsonRpcError> {
798	serde_json::from_value(params).map_err(\|err\| JsonRpcError::invalid_params(err.to_string()))
799	}
800
801	fn jsonrpc_result(id: Option<Value>, result: Value) -> Value {
802	json!({
803	"jsonrpc": "2.0",
804	"id": id.unwrap_or(Value::Null),
805	"result": result
806	})
807	}
808
809	fn jsonrpc_error(id: Option<Value>, err: JsonRpcError) -> Value {
810	json!({
811	"jsonrpc": "2.0",
812	"id": id.unwrap_or(Value::Null),
813	"error": {
814	"code": err.code,
815	"message": err.message,
816	"data": err.data
817	}
818	})
819	}
820
821	impl JsonRpcError {
822	fn parse_error(message: impl Into<String>) -> Self {
823	Self {
824	code: -32700,
825	message: message.into(),
826	data: None,
827	}
828	}
829
830	fn invalid_request(message: impl Into<String>) -> Self {
831	Self {
832	code: -32600,
833	message: message.into(),
834	data: None,
835	}
836	}
837
838	fn method_not_found(method: &str) -> Self {
839	Self {
840	code: -32601,
841	message: format!("unsupported method: {method}"),
842	data: None,
843	}
844	}
845
846	fn invalid_params(message: impl Into<String>) -> Self {
847	Self {
848	code: -32602,
849	message: message.into(),
850	data: None,
851	}
852	}
853
854	/// Server error (-32000..-32099): the named thread does not exist.
855	fn thread_not_found(thread_id: &str) -> Self {
856	Self {
857	code: -32004,
858	message: format!("thread not found: {thread_id}"),
859	data: Some(json!({
860	"error": "thread_not_found",
861	"thread_id": thread_id,
862	})),
863	}
864	}
865
866	fn internal(message: impl Into<String>) -> Self {
867	Self {
868	code: -32603,
869	message: message.into(),
870	data: None,
871	}
872	}
873	}
874
875	async fn handle_thread_request(
876	state: &AppState,
877	req: ThreadRequest,
878	) -> std::result::Result<ThreadResponse, JsonRpcError> {
879	let mut runtime = state.runtime.write().await;
880	runtime
881	.handle_thread(req)
882	.await
883	.map_err(\|err\| JsonRpcError::internal(err.to_string()))
884	}
885
886	async fn handle_prompt_request(
887	state: &AppState,
888	req: PromptRequest,
889	) -> std::result::Result<PromptResponse, JsonRpcError> {
890	let mut runtime = state.runtime.write().await;
891	runtime
892	.handle_prompt(req, &CliRuntimeOverrides::default())
893	.await
894	.map_err(\|err\| JsonRpcError::internal(err.to_string()))
895	}
896
897	async fn handle_stdio_thread_message<W: AsyncWrite + Unpin>(
898	state: &AppState,
899	writer: &mut W,
900	parsed: ThreadMessageParams,
901	) -> std::result::Result<Value, JsonRpcError> {
902	let hint = {
903	let hints = state.stdio_thread_hints.lock().await;
904	hints.get(&parsed.thread_id).cloned()
905	};
906	let bridge = acquire_stdio_bridge(state).await?;
907	// The inner bridge lock is held for the whole turn: one child process
908	// serves all threads and per-thread seq tracking requires ordered
909	// access. The cache slot itself stays unlocked, so config updates and
910	// bridge invalidation are never queued behind a streaming turn.
911	let mut bridge = bridge.lock().await;
912	let runtime_thread_id = bridge
913	.ensure_runtime_thread(&parsed.thread_id, hint)
914	.await
915	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?;
916	let mut result = bridge
917	.message_thread(
918	&runtime_thread_id,
919	&parsed.input,
920	writer,
921	Some((state.in_flight_turns.clone(), parsed.thread_id.clone())),
922	)
923	.await
924	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?;
925	if let Some(object) = result.as_object_mut() {
926	object.insert("thread_id".to_string(), Value::String(parsed.thread_id));
927	}
928	Ok(result)
929	}
930
931	/// Resuming or forking a thread the runtime reports as `missing` must fail
932	/// with a named not-found error. Recording the null model/workspace of that
933	/// response as a stdio hint would clobber any previously cached hint for
934	/// the same thread id (#5171).
935	fn ensure_thread_found(response: &ThreadResponse) -> std::result::Result<(), JsonRpcError> {
936	if response.status == "missing" {
937	return Err(JsonRpcError::thread_not_found(&response.thread_id));
938	}
939	Ok(())
940	}
941
942	async fn record_stdio_thread_hint(state: &AppState, response: &ThreadResponse) {
943	let mut hints = state.stdio_thread_hints.lock().await;
944	hints.insert(
945	response.thread_id.clone(),
946	RuntimeThreadHint {
947	model: response.model.clone(),
948	workspace: response.cwd.clone(),
949	},
950	);
951	}
952
953	/// Fetch the cached stdio→runtime bridge, spawning one on first use.
954	///
955	/// The cache-slot lock is held only for the lookup/insert — never across
956	/// the child spawn or any request traffic — so [`invalidate_stdio_bridge`]
957	/// and other slot users are never blocked behind a slow bridge operation.
958	async fn acquire_stdio_bridge(
959	state: &AppState,
960	) -> std::result::Result<SharedRuntimeBridge, JsonRpcError> {
961	if let Some(bridge) = state.stdio_bridge.lock().await.as_ref() {
962	return Ok(bridge.clone());
963	}
964	let bridge = Arc::new(Mutex::new(
965	RuntimeBridge::start(state.config_path.as_deref())
966	.await
967	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
968	));
969	let mut slot = state.stdio_bridge.lock().await;
970	// Prefer a bridge cached by a concurrent caller while we were spawning;
971	// dropping our unused one kills the extra child via `Drop`.
972	Ok(slot.get_or_insert_with(\|\| bridge.clone()).clone())
973	}
974
975	/// Ask the runtime to interrupt a turn that is streaming right now.
976	///
977	/// Everything this needs was copied out of the bridge when the turn started,
978	/// so it never touches the bridge mutex the turn is holding. Returns whether
979	/// a live turn was found for `thread_id`.
980	async fn interrupt_stdio_turn(
981	state: &AppState,
982	thread_id: &str,
983	) -> std::result::Result<bool, JsonRpcError> {
984	let Some(turn) = state.in_flight_turns.lock().await.get(thread_id).cloned() else {
985	return Ok(false);
986	};
987	let mut request = codewhale_release::platform_http_client_builder()
988	.timeout(Duration::from_secs(10))
989	.build()
990	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?
991	.post(format!(
992	"{}/v1/threads/{}/turns/{}/interrupt",
993	turn.base_url, turn.runtime_thread_id, turn.turn_id
994	));
995	if let Some(token) = turn.auth_token.as_deref() {
996	request = request.bearer_auth(token);
997	}
998	request
999	.send()
1000	.await
1001	.and_then(reqwest::Response::error_for_status)
1002	.map_err(\|err\| JsonRpcError::internal(format!("interrupt failed: {err}")))?;
1003	Ok(true)
1004	}
1005
1006	/// Drop the cached runtime bridge so the next stdio thread message spawns a
1007	/// fresh child that re-reads the persisted config. An in-flight message
1008	/// keeps its own [`SharedRuntimeBridge`] clone and finishes against the old
1009	/// child, which is killed when the last clone drops.
1010	async fn invalidate_stdio_bridge(state: &AppState) {
1011	let mut bridge = state.stdio_bridge.lock().await;
1012	*bridge = None;
1013	}
1014
1015	impl RuntimeBridge {
1016	async fn start(config_path: Option<&Path>) -> Result<Self> {
1017	install_rustls_crypto_provider();
1018	let port = reserve_runtime_port()?;
1019	let auth_token = format!("cwrt_{}", Uuid::new_v4().simple());
1020	let child = Self::runtime_command(config_path, port, &auth_token)?
1021	.spawn()
1022	.context("failed to start runtime API bridge")?;
1023	let mut bridge = Self {
1024	base_url: format!("http://127.0.0.1:{port}"),
1025	client: codewhale_release::platform_http_client_builder()
1026	.build()
1027	.context("failed to build runtime API client")?,
1028	auth_token: Some(auth_token),
1029	child: Some(child),
1030	thread_map: HashMap::new(),
1031	last_seq_by_thread: HashMap::new(),
1032	};
1033	bridge.wait_until_ready().await?;
1034	Ok(bridge)
1035	}
1036
1037	fn runtime_command(config_path: Option<&Path>, port: u16, auth_token: &str) -> Result<Command> {
1038	let current_exe = std::env::current_exe().ok();
1039	let mut command = if let Some(path) = current_exe {
1040	Command::new(path)
1041	} else {
1042	Command::new("codewhale")
1043	};
1044	// Pass the runtime auth token out-of-band via env (not argv) so local
1045	// `ps` cannot read credential material from the child command line.
1046	// The TUI/runtime server already accepts CODEWHALE_RUNTIME_TOKEN /
1047	// DEEPSEEK_RUNTIME_TOKEN when --auth-token is absent.
1048	command
1049	.arg("app-server")
1050	.arg("--http")
1051	.arg("--host")
1052	.arg("127.0.0.1")
1053	.arg("--port")
1054	.arg(port.to_string())
1055	.env("CODEWHALE_RUNTIME_TOKEN", auth_token)
1056	.env("DEEPSEEK_RUNTIME_TOKEN", auth_token)
1057	.stdin(Stdio::null())
1058	.stdout(Stdio::null())
1059	.stderr(Stdio::null());
1060	if let Some(config_path) = config_path {
1061	command.arg("--config").arg(config_path);
1062	}
1063	Ok(command)
1064	}
1065
1066	async fn wait_until_ready(&mut self) -> Result<()> {
1067	let deadline = Instant::now() + Duration::from_secs(15);
1068	loop {
1069	if let Some(child) = self.child.as_mut()
1070	&& let Some(status) = child.try_wait()?
1071	{
1072	return Err(anyhow!(
1073	"runtime API bridge exited before becoming ready (status {status})"
1074	));
1075	}
1076
1077	match self
1078	.client
1079	.get(format!("{}/health", self.base_url))
1080	.send()
1081	.await
1082	{
1083	Ok(response) if response.status().is_success() => return Ok(()),
1084	_ if Instant::now() >= deadline => {
1085	bail!(
1086	"timed out waiting for runtime API bridge at {}/health",
1087	self.base_url
1088	)
1089	}
1090	_ => tokio::time::sleep(Duration::from_millis(50)).await,
1091	}
1092	}
1093	}
1094
1095	fn authed(&self, builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
1096	match self.auth_token.as_deref() {
1097	Some(token) => builder.bearer_auth(token),
1098	None => builder,
1099	}
1100	}
1101
1102	async fn request_json(&self, builder: reqwest::RequestBuilder) -> Result<Value> {
1103	let response = builder.send().await?;
1104	let status = response.status();
1105	let body = response.text().await?;
1106	if !status.is_success() {
1107	let detail = body.trim();
1108	if detail.is_empty() {
1109	bail!("runtime API returned {status}");
1110	}
1111	bail!("runtime API returned {status}: {detail}");
1112	}
1113	serde_json::from_str(&body).with_context(\|\| format!("invalid runtime API json: {body}"))
1114	}
1115
1116	async fn ensure_runtime_thread(
1117	&mut self,
1118	stdio_thread_id: &str,
1119	hint: Option<RuntimeThreadHint>,
1120	) -> Result<String> {
1121	if let Some(runtime_thread_id) = self.thread_map.get(stdio_thread_id) {
1122	return Ok(runtime_thread_id.clone());
1123	}
1124	let hint = hint.unwrap_or_default();
1125	let runtime_thread_id = self
1126	.create_runtime_thread(hint.model, hint.workspace)
1127	.await?;
1128	self.thread_map
1129	.insert(stdio_thread_id.to_string(), runtime_thread_id.clone());
1130	Ok(runtime_thread_id)
1131	}
1132
1133	async fn create_runtime_thread(
1134	&mut self,
1135	model: Option<String>,
1136	workspace: Option<PathBuf>,
1137	) -> Result<String> {
1138	let record = self
1139	.request_json(
1140	self.authed(self.client.post(format!("{}/v1/threads", self.base_url)))
1141	.json(&json!({
1142	"model": model,
1143	"workspace": workspace,
1144	"mode": "agent",
1145	"archived": false,
1146	})),
1147	)
1148	.await?;
1149	let thread_id = extract_runtime_thread_id(&record)?.to_string();
1150	self.last_seq_by_thread
1151	.entry(thread_id.clone())
1152	.or_insert(0);
1153	Ok(thread_id)
1154	}
1155
1156	/// Run one turn to completion, streaming its events to `writer`.
1157	///
1158	/// `registration` is `Some` on the stdio path: it publishes the live turn
1159	/// so an `thread/interrupt` arriving mid-stream can reach the runtime
1160	/// without waiting on the bridge mutex this call holds.
1161	async fn message_thread<W: AsyncWrite + Unpin>(
1162	&mut self,
1163	thread_id: &str,
1164	input: &str,
1165	writer: &mut W,
1166	registration: Option<(TurnRegistry, String)>,
1167	) -> Result<Value> {
1168	let turn = self
1169	.request_json(
1170	self.authed(
1171	self.client
1172	.post(format!("{}/v1/threads/{thread_id}/turns", self.base_url)),
1173	)
1174	.json(&json!({ "prompt": input })),
1175	)
1176	.await?;
1177	let turn_id = turn
1178	.pointer("/turn/id")
1179	.and_then(Value::as_str)
1180	.ok_or_else(\|\| anyhow!("runtime API turn response missing turn.id"))?
1181	.to_string();
1182	let response_id = format!("{thread_id}:{turn_id}");
1183
1184	emit_stdio_event(
1185	writer,
1186	json!({
1187	"type": "response_start",
1188	"response_id": response_id,
1189	}),
1190	)
1191	.await?;
1192
1193	// Publish the turn only for the streaming window, and take it back
1194	// before any `?` below: a turn that has already finished must never
1195	// look cancellable.
1196	if let Some((registry, key)) = registration.as_ref() {
1197	registry.lock().await.insert(
1198	key.clone(),
1199	InFlightTurn {
1200	base_url: self.base_url.clone(),
1201	auth_token: self.auth_token.clone(),
1202	runtime_thread_id: thread_id.to_string(),
1203	turn_id: turn_id.clone(),
1204	},
1205	);
1206	}
1207
1208	let since_seq = self.last_seq_by_thread.get(thread_id).copied().unwrap_or(0);
1209	let stream_result = self
1210	.stream_turn_events(thread_id, &turn_id, &response_id, writer, since_seq)
1211	.await;
1212
1213	if let Some((registry, key)) = registration.as_ref() {
1214	registry.lock().await.remove(key);
1215	}
1216
1217	let _ = emit_stdio_event(
1218	writer,
1219	json!({
1220	"type": "response_end",
1221	"response_id": response_id,
1222	}),
1223	)
1224	.await;
1225
1226	let (last_seq, status, error) = stream_result?;
1227	self.last_seq_by_thread
1228	.insert(thread_id.to_string(), last_seq);
1229
1230	match status {
1231	TurnTerminalStatus::Completed => Ok(json!({
1232	"thread_id": thread_id,
1233	"status": "accepted",
1234	"thread": Value::Null,
1235	"threads": [],
1236	"model": Value::Null,
1237	"model_provider": Value::Null,
1238	"cwd": Value::Null,
1239	"approval_policy": Value::Null,
1240	"sandbox": Value::Null,
1241	"events": [],
1242	"data": { "turn_id": turn_id },
1243	})),
1244	TurnTerminalStatus::Failed => Err(anyhow!(
1245	"{}",
1246	error.unwrap_or_else(\|\| "turn failed".to_string())
1247	)),
1248	TurnTerminalStatus::Interrupted => Err(anyhow!(
1249	"{}",
1250	error.unwrap_or_else(\|\| "turn interrupted".to_string())
1251	)),
1252	TurnTerminalStatus::Canceled => Err(anyhow!(
1253	"{}",
1254	error.unwrap_or_else(\|\| "turn canceled".to_string())
1255	)),
1256	}
1257	}
1258
1259	async fn stream_turn_events<W: AsyncWrite + Unpin>(
1260	&self,
1261	thread_id: &str,
1262	turn_id: &str,
1263	response_id: &str,
1264	writer: &mut W,
1265	since_seq: u64,
1266	) -> Result<(u64, TurnTerminalStatus, Option<String>)> {
1267	let mut response = self
1268	.authed(self.client.get(format!(
1269	"{}/v1/threads/{thread_id}/events?since_seq={since_seq}",
1270	self.base_url
1271	)))
1272	.send()
1273	.await?
1274	.error_for_status()?;
1275
1276	let mut buffer = Vec::new();
1277	let mut last_seq = since_seq;
1278
1279	while let Some(chunk) = response.chunk().await? {
1280	buffer.extend_from_slice(&chunk);
1281	if buffer.len() > MAX_SSE_FRAME_BYTES {
1282	bail!(
1283	"runtime SSE frame exceeded {MAX_SSE_FRAME_BYTES} bytes without a frame delimiter"
1284	);
1285	}
1286	while let Some(frame_bytes) = take_sse_frame(&mut buffer) {
1287	let Some((event_name, frame_data)) = parse_sse_frame(&frame_bytes) else {
1288	continue;
1289	};
1290	let envelope: Value = serde_json::from_str(&frame_data)
1291	.with_context(\|\| format!("invalid SSE json for {event_name}: {frame_data}"))?;
1292	if let Some(seq) = envelope.get("seq").and_then(Value::as_u64) {
1293	last_seq = last_seq.max(seq);
1294	}
1295	if envelope.get("turn_id").and_then(Value::as_str) != Some(turn_id) {
1296	continue;
1297	}
1298	let payload = envelope.get("payload").cloned().unwrap_or(Value::Null);
1299	match event_name.as_str() {
1300	"item.delta" => {
1301	let kind = payload
1302	.get("kind")
1303	.and_then(Value::as_str)
1304	.unwrap_or_default();
1305	if kind == "agent_message"
1306	&& let Some(delta) = payload.get("delta").and_then(Value::as_str)
1307	&& !delta.is_empty()
1308	{
1309	emit_stdio_event(
1310	writer,
1311	json!({
1312	"type": "response_delta",
1313	"response_id": response_id,
1314	"delta": delta,
1315	}),
1316	)
1317	.await?;
1318	}
1319	}
1320	"turn.completed" => {
1321	let status = turn_terminal_status(&payload);
1322	let error = payload
1323	.pointer("/turn/error")
1324	.and_then(Value::as_str)
1325	.map(str::to_string);
1326	return Ok((last_seq, status, error));
1327	}
1328	_ => {}
1329	}
1330	}
1331	}
1332
1333	bail!("runtime event stream ended before turn.completed")
1334	}
1335
1336	#[cfg(test)]
1337	fn from_base_url_for_test(base_url: String) -> Self {
1338	install_rustls_crypto_provider();
1339	Self {
1340	base_url,
1341	client: codewhale_release::platform_http_client_builder()
1342	.timeout(Duration::from_secs(5))
1343	.build()
1344	.expect("build reqwest test client"),
1345	auth_token: None,
1346	child: None,
1347	thread_map: HashMap::new(),
1348	last_seq_by_thread: HashMap::new(),
1349	}
1350	}
1351	}
1352
1353	impl RuntimeBridge {
1354	/// Kills the managed runtime child and reaps it on a detached thread so
1355	/// neither an explicit shutdown nor Drop blocks a Tokio runtime thread.
1356	fn shutdown_child(&mut self) {
1357	if let Some(mut child) = self.child.take() {
1358	let _ = child.kill();
1359	std::thread::spawn(move \|\| {
1360	let _ = child.wait();
1361	});
1362	}
1363	}
1364	}
1365
1366	impl Drop for RuntimeBridge {
1367	fn drop(&mut self) {
1368	self.shutdown_child();
1369	}
1370	}
1371
1372	fn reserve_runtime_port() -> Result<u16> {
1373	let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
1374	Ok(listener.local_addr()?.port())
1375	}
1376
1377	fn install_rustls_crypto_provider() {
1378	let _ = rustls::crypto::ring::default_provider().install_default();
1379	}
1380
1381	fn extract_runtime_thread_id(record: &Value) -> Result<&str> {
1382	record
1383	.get("id")
1384	.and_then(Value::as_str)
1385	.ok_or_else(\|\| anyhow!("runtime API thread response missing id"))
1386	}
1387
1388	fn turn_terminal_status(payload: &Value) -> TurnTerminalStatus {
1389	match payload
1390	.pointer("/turn/status")
1391	.and_then(Value::as_str)
1392	.unwrap_or("completed")
1393	.to_ascii_lowercase()
1394	.as_str()
1395	{
1396	"failed" => TurnTerminalStatus::Failed,
1397	"interrupted" => TurnTerminalStatus::Interrupted,
1398	"canceled" \| "cancelled" => TurnTerminalStatus::Canceled,
1399	_ => TurnTerminalStatus::Completed,
1400	}
1401	}
1402
1403	async fn emit_stdio_event<W: AsyncWrite + Unpin>(writer: &mut W, event: Value) -> Result<()> {
1404	writer.write_all(&serde_json::to_vec(&event)?).await?;
1405	writer.write_all(b"\n").await?;
1406	writer.flush().await?;
1407	Ok(())
1408	}
1409
1410	fn take_sse_frame(buffer: &mut Vec<u8>) -> Option<Vec<u8>> {
1411	if let Some(pos) = buffer.windows(4).position(\|window\| window == b"\r\n\r\n") {
1412	return Some(buffer.drain(..pos + 4).collect());
1413	}
1414	buffer
1415	.windows(2)
1416	.position(\|window\| window == b"\n\n")
1417	.map(\|pos\| buffer.drain(..pos + 2).collect())
1418	}
1419
1420	fn parse_sse_frame(frame_bytes: &[u8]) -> Option<(String, String)> {
1421	let text = String::from_utf8(frame_bytes.to_vec()).ok()?;
1422	let mut event_name = None;
1423	let mut data_lines = Vec::new();
1424	for raw_line in text.lines() {
1425	let line = raw_line.trim_end_matches('\r');
1426	if let Some(value) = line.strip_prefix("event:") {
1427	event_name = Some(value.trim().to_string());
1428	} else if let Some(value) = line.strip_prefix("data:") {
1429	data_lines.push(value.trim_start().to_string());
1430	}
1431	}
1432	match (event_name, data_lines.is_empty()) {
1433	(Some(event), false) => Some((event, data_lines.join("\n"))),
1434	_ => None,
1435	}
1436	}
1437
1438	#[cfg(test)]
1439	async fn dispatch_stdio_request(
1440	state: &AppState,
1441	method: &str,
1442	params: Value,
1443	) -> std::result::Result<StdioDispatchResult, JsonRpcError> {
1444	let mut sink = tokio::io::sink();
1445	dispatch_stdio_request_with_writer(state, &mut sink, method, params).await
1446	}
1447
1448	async fn dispatch_stdio_app_request(
1449	state: &AppState,
1450	request: AppRequest,
1451	) -> std::result::Result<StdioDispatchResult, JsonRpcError> {
1452	let response = Box::pin(process_app_request(state, request, AppTransport::Stdio)).await;
1453	Ok(StdioDispatchResult {
1454	result: serde_json::to_value(response)
1455	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1456	should_exit: false,
1457	})
1458	}
1459
1460	async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
1461	state: &AppState,
1462	writer: &mut W,
1463	method: &str,
1464	params: Value,
1465	) -> std::result::Result<StdioDispatchResult, JsonRpcError> {
1466	let outcome = match method {
1467	"healthz" \| "app/healthz" => StdioDispatchResult {
1468	result: json!({
1469	"status": "ok",
1470	"service": legacy_deepseek_compat::SERVICE_NAME,
1471	"transport": "stdio"
1472	}),
1473	should_exit: false,
1474	},
1475	"capabilities" => StdioDispatchResult {
1476	result: json!({
1477	"transport": "stdio",
1478	"families": ["thread/", "app/", "prompt/*"],
1479	"methods": [
1480	"healthz",
1481	"thread/capabilities",
1482	"thread/request",
1483	"thread/create",
1484	"thread/start",
1485	"thread/resume",
1486	"thread/fork",
1487	"thread/list",
1488	"thread/read",
1489	"thread/set_name",
1490	"thread/goal/set",
1491	"thread/goal/get",
1492	"thread/goal/clear",
1493	"thread/archive",
1494	"thread/unarchive",
1495	"thread/message",
1496	"thread/interrupt",
1497	"app/capabilities",
1498	"app/request",
1499	"app/config/get",
1500	"app/config/set",
1501	"app/config/unset",
1502	"app/config/list",
1503	"app/config/reload",
1504	"app/models",
1505	"app/thread_loaded_list",
1506	"prompt/capabilities",
1507	"prompt/request",
1508	"prompt/run",
1509	"shutdown"
1510	]
1511	}),
1512	should_exit: false,
1513	},
1514	"thread/capabilities" => StdioDispatchResult {
1515	result: json!({
1516	"methods": [
1517	"thread/request",
1518	"thread/create",
1519	"thread/start",
1520	"thread/resume",
1521	"thread/fork",
1522	"thread/list",
1523	"thread/read",
1524	"thread/set_name",
1525	"thread/goal/set",
1526	"thread/goal/get",
1527	"thread/goal/clear",
1528	"thread/archive",
1529	"thread/unarchive",
1530	"thread/message",
1531	"thread/interrupt"
1532	]
1533	}),
1534	should_exit: false,
1535	},
1536	"thread/request" => {
1537	let request: ThreadRequest = parse_params(params)?;
1538	if let ThreadRequest::Message { thread_id, input } = request {
1539	let response = handle_stdio_thread_message(
1540	state,
1541	writer,
1542	ThreadMessageParams { thread_id, input },
1543	)
1544	.await?;
1545	return Ok(StdioDispatchResult {
1546	result: response,
1547	should_exit: false,
1548	});
1549	}
1550	let should_record_hint = matches!(
1551	&request,
1552	ThreadRequest::Create { .. }
1553	\| ThreadRequest::Start(_)
1554	\| ThreadRequest::Resume(_)
1555	\| ThreadRequest::Fork(_)
1556	);
1557	let response = handle_thread_request(state, request).await?;
1558	if should_record_hint {
1559	record_stdio_thread_hint(state, &response).await;
1560	}
1561	StdioDispatchResult {
1562	result: serde_json::to_value(response)
1563	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1564	should_exit: false,
1565	}
1566	}
1567	"thread/create" => {
1568	#[derive(Debug, Deserialize)]
1569	struct CreateParams {
1570	#[serde(default)]
1571	metadata: Value,
1572	}
1573	let parsed: CreateParams = parse_params(params_or_object(params))?;
1574	let response = handle_thread_request(
1575	state,
1576	ThreadRequest::Create {
1577	metadata: parsed.metadata,
1578	},
1579	)
1580	.await?;
1581	record_stdio_thread_hint(state, &response).await;
1582	StdioDispatchResult {
1583	result: serde_json::to_value(response)
1584	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1585	should_exit: false,
1586	}
1587	}
1588	"thread/start" => {
1589	let request = ThreadRequest::Start(parse_params(params_or_object(params))?);
1590	let response = handle_thread_request(state, request).await?;
1591	record_stdio_thread_hint(state, &response).await;
1592	StdioDispatchResult {
1593	result: serde_json::to_value(response)
1594	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1595	should_exit: false,
1596	}
1597	}
1598	"thread/resume" => {
1599	let request = ThreadRequest::Resume(parse_params(params_or_object(params))?);
1600	let response = handle_thread_request(state, request).await?;
1601	ensure_thread_found(&response)?;
1602	record_stdio_thread_hint(state, &response).await;
1603	StdioDispatchResult {
1604	result: serde_json::to_value(response)
1605	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1606	should_exit: false,
1607	}
1608	}
1609	"thread/fork" => {
1610	let request = ThreadRequest::Fork(parse_params(params_or_object(params))?);
1611	let response = handle_thread_request(state, request).await?;
1612	ensure_thread_found(&response)?;
1613	record_stdio_thread_hint(state, &response).await;
1614	StdioDispatchResult {
1615	result: serde_json::to_value(response)
1616	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1617	should_exit: false,
1618	}
1619	}
1620	"thread/list" => {
1621	let request = ThreadRequest::List(parse_params(params_or_object(params))?);
1622	let response = handle_thread_request(state, request).await?;
1623	StdioDispatchResult {
1624	result: serde_json::to_value(response)
1625	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1626	should_exit: false,
1627	}
1628	}
1629	"thread/read" => {
1630	let request = ThreadRequest::Read(parse_params(params_or_object(params))?);
1631	let response = handle_thread_request(state, request).await?;
1632	StdioDispatchResult {
1633	result: serde_json::to_value(response)
1634	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1635	should_exit: false,
1636	}
1637	}
1638	"thread/set_name" \| "thread/set-name" => {
1639	let request = ThreadRequest::SetName(parse_params(params_or_object(params))?);
1640	let response = handle_thread_request(state, request).await?;
1641	StdioDispatchResult {
1642	result: serde_json::to_value(response)
1643	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1644	should_exit: false,
1645	}
1646	}
1647	"thread/goal/set" \| "thread/goal_set" \| "thread/goal-set" => {
1648	let request = ThreadRequest::GoalSet(parse_params::<ThreadGoalSetParams>(
1649	params_or_object(params),
1650	)?);
1651	let response = handle_thread_request(state, request).await?;
1652	StdioDispatchResult {
1653	result: serde_json::to_value(response)
1654	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1655	should_exit: false,
1656	}
1657	}
1658	"thread/goal/get" \| "thread/goal_get" \| "thread/goal-get" => {
1659	let request = ThreadRequest::GoalGet(parse_params::<ThreadGoalGetParams>(
1660	params_or_object(params),
1661	)?);
1662	let response = handle_thread_request(state, request).await?;
1663	StdioDispatchResult {
1664	result: serde_json::to_value(response)
1665	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1666	should_exit: false,
1667	}
1668	}
1669	"thread/goal/clear" \| "thread/goal_clear" \| "thread/goal-clear" => {
1670	let request = ThreadRequest::GoalClear(parse_params::<ThreadGoalClearParams>(
1671	params_or_object(params),
1672	)?);
1673	let response = handle_thread_request(state, request).await?;
1674	StdioDispatchResult {
1675	result: serde_json::to_value(response)
1676	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1677	should_exit: false,
1678	}
1679	}
1680	"thread/archive" => {
1681	let parsed: ThreadIdParams = parse_params(params_or_object(params))?;
1682	let response = handle_thread_request(
1683	state,
1684	ThreadRequest::Archive {
1685	thread_id: parsed.thread_id,
1686	},
1687	)
1688	.await?;
1689	StdioDispatchResult {
1690	result: serde_json::to_value(response)
1691	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1692	should_exit: false,
1693	}
1694	}
1695	"thread/unarchive" => {
1696	let parsed: ThreadIdParams = parse_params(params_or_object(params))?;
1697	let response = handle_thread_request(
1698	state,
1699	ThreadRequest::Unarchive {
1700	thread_id: parsed.thread_id,
1701	},
1702	)
1703	.await?;
1704	StdioDispatchResult {
1705	result: serde_json::to_value(response)
1706	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1707	should_exit: false,
1708	}
1709	}
1710	"thread/message" => {
1711	let parsed: ThreadMessageParams = parse_params(params_or_object(params))?;
1712	let response = handle_stdio_thread_message(state, writer, parsed).await?;
1713	StdioDispatchResult {
1714	result: response,
1715	should_exit: false,
1716	}
1717	}
1718	"app/capabilities" => dispatch_stdio_app_request(state, AppRequest::Capabilities).await?,
1719	"app/request" => {
1720	let request: AppRequest = parse_params(params)?;
1721	dispatch_stdio_app_request(state, request).await?
1722	}
1723	"app/config/get" => {
1724	let parsed: ConfigGetParams = parse_params(params_or_object(params))?;
1725	dispatch_stdio_app_request(state, AppRequest::ConfigGet { key: parsed.key }).await?
1726	}
1727	"app/config/set" => {
1728	let parsed: ConfigSetParams = parse_params(params_or_object(params))?;
1729	dispatch_stdio_app_request(
1730	state,
1731	AppRequest::ConfigSet {
1732	key: parsed.key,
1733	value: parsed.value,
1734	},
1735	)
1736	.await?
1737	}
1738	"app/config/unset" => {
1739	let parsed: ConfigGetParams = parse_params(params_or_object(params))?;
1740	dispatch_stdio_app_request(state, AppRequest::ConfigUnset { key: parsed.key }).await?
1741	}
1742	"app/config/list" => dispatch_stdio_app_request(state, AppRequest::ConfigList).await?,
1743	"app/config/reload" => dispatch_stdio_app_request(state, AppRequest::ConfigReload).await?,
1744	"app/models" => dispatch_stdio_app_request(state, AppRequest::Models).await?,
1745	"app/thread_loaded_list" \| "app/thread-loaded-list" => {
1746	dispatch_stdio_app_request(state, AppRequest::ThreadLoadedList).await?
1747	}
1748	"prompt/capabilities" => StdioDispatchResult {
1749	result: json!({
1750	"methods": ["prompt/request", "prompt/run"]
1751	}),
1752	should_exit: false,
1753	},
1754	"prompt/request" \| "prompt/run" => {
1755	let request: PromptRequest = parse_params(params)?;
1756	let response = handle_prompt_request(state, request).await?;
1757	StdioDispatchResult {
1758	result: serde_json::to_value(response)
1759	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
1760	should_exit: false,
1761	}
1762	}
1763	"thread/interrupt" => {
1764	let parsed: ThreadInterruptParams = parse_params(params_or_object(params))?;
1765	let interrupted = interrupt_stdio_turn(state, &parsed.thread_id).await?;
1766	StdioDispatchResult {
1767	result: json!({
1768	"thread_id": parsed.thread_id,
1769	"interrupted": interrupted,
1770	}),
1771	should_exit: false,
1772	}
1773	}
1774	"shutdown" => {
1775	// A turn streaming right now holds the bridge mutex, so taking it
1776	// to kill the child would block until that turn ends — the exact
1777	// deadlock that made shutdown useless against a runaway turn.
1778	// Interrupt live turns first; they release the mutex promptly.
1779	let live: Vec<String> = state.in_flight_turns.lock().await.keys().cloned().collect();
1780	for thread_id in live {
1781	let _ = interrupt_stdio_turn(state, &thread_id).await;
1782	}
1783	if let Some(bridge) = state.stdio_bridge.lock().await.take() {
1784	bridge.lock().await.shutdown_child();
1785	}
1786	StdioDispatchResult {
1787	result: json!({"ok": true, "status": "stopped"}),
1788	should_exit: true,
1789	}
1790	}
1791	_ => return Err(JsonRpcError::method_not_found(method)),
1792	};
1793	Ok(outcome)
1794	}
1795
1796	async fn process_app_request(
1797	state: &AppState,
1798	req: AppRequest,
1799	_transport: AppTransport,
1800	) -> AppResponse {
1801	match req {
1802	AppRequest::Capabilities => AppResponse {
1803	ok: true,
1804	data: json!({
1805	"routes": ["/thread", "/app", "/prompt", "/tool", "/jobs", "/mcp/startup"],
1806	"config": ["get", "set", "unset", "list", "reload"],
1807	"events": ["response_start", "response_delta", "response_end", "tool_call_start", "tool_call_result", "mcp_startup_update", "mcp_startup_complete"],
1808	"transport": "stdio+http",
1809	"config_path": state.config_path.as_ref().map(\|p\| p.display().to_string()),
1810	}),
1811	events: Vec::new(),
1812	},
1813	AppRequest::ConfigGet { key } => {
1814	let cfg = state.config.read().await;
1815	let value = cfg.get_display_value(&key);
1816	AppResponse {
1817	ok: true,
1818	data: json!({ "key": key, "value": value }),
1819	events: Vec::new(),
1820	}
1821	}
1822	AppRequest::ConfigSet { key, value } => {
1823	let (result, snapshot) = {
1824	let mut cfg = state.config.write().await;
1825	let result = cfg.set_value(&key, &value);
1826	(result, cfg.clone())
1827	};
1828	let ok = result.is_ok();
1829	let message = result.err().map(\|e\| e.to_string());
1830	// Only propagate a mutation that actually happened. `set_value`
1831	// leaves the config untouched on an unknown key or invalid value,
1832	// so this is a no-op from the caller's point of view — but
1833	// `apply_config_update` invalidates the cached stdio bridge
1834	// regardless, and dropping the last reference kills the running
1835	// child runtime along with its thread map. A single typo'd key
1836	// would orphan every in-flight thread on that bridge.
1837	if ok {
1838	apply_config_update(state, snapshot, None, true).await;
1839	}
1840	AppResponse {
1841	ok,
1842	data: json!({ "key": key, "value": value, "error": message }),
1843	events: Vec::new(),
1844	}
1845	}
1846	AppRequest::ConfigUnset { key } => {
1847	let (result, snapshot) = {
1848	let mut cfg = state.config.write().await;
1849	let result = cfg.unset_value(&key);
1850	(result, cfg.clone())
1851	};
1852	let ok = result.is_ok();
1853	let message = result.err().map(\|e\| e.to_string());
1854	// See ConfigSet: a failed unset changed nothing and must not tear
1855	// down the runtime bridge.
1856	if ok {
1857	apply_config_update(state, snapshot, None, true).await;
1858	}
1859	AppResponse {
1860	ok,
1861	data: json!({ "key": key, "error": message }),
1862	events: Vec::new(),
1863	}
1864	}
1865	AppRequest::ConfigList => {
1866	let cfg = state.config.read().await;
1867	AppResponse {
1868	ok: true,
1869	data: json!({ "values": cfg.list_values() }),
1870	events: Vec::new(),
1871	}
1872	}
1873	AppRequest::ConfigReload => {
1874	// Re-read both `config.toml` and the sibling `permissions.toml`
1875	// from disk (the headless equivalent of the TUI
1876	// `reload_runtime_config` codepath) and push the fresh
1877	// snapshots into `state.config` and the live `Runtime`.
1878	//
1879	// `ConfigStore::load` resolves the same default config path
1880	// that `build_state` used at startup when `config_path` is
1881	// `None`, so a `None` here reloads from the same on-disk file
1882	// the server booted from.
1883	let store = match ConfigStore::load(state.config_path.clone()) {
1884	Ok(store) => store,
1885	Err(e) => {
1886	return AppResponse {
1887	ok: false,
1888	data: json!({ "error": format!("failed to load config: {e}") }),
1889	events: Vec::new(),
1890	};
1891	}
1892	};
1893	let new_config = store.config.clone();
1894	let new_exec_policy = store.exec_policy_engine();
1895
1896	// Disk is already the source of truth here, so nothing to
1897	// persist; the exec policy rides along so the runtime picks up
1898	// external `permissions.toml` edits too.
1899	apply_config_update(state, new_config, Some(new_exec_policy), false).await;
1900
1901	AppResponse {
1902	ok: true,
1903	data: json!({ "reloaded": true }),
1904	events: Vec::new(),
1905	}
1906	}
1907	AppRequest::Models => AppResponse {
1908	ok: true,
1909	data: json!({ "models": state.registry.list() }),
1910	events: Vec::new(),
1911	},
1912	AppRequest::ThreadLoadedList => {
1913	let mut runtime = state.runtime.write().await;
1914	let response = runtime
1915	.handle_thread(codewhale_protocol::ThreadRequest::List(
1916	codewhale_protocol::ThreadListParams {
1917	include_archived: false,
1918	limit: Some(50),
1919	},
1920	))
1921	.await;
1922	match response {
1923	Ok(thread_resp) => AppResponse {
1924	ok: true,
1925	data: json!({ "threads": thread_resp.threads }),
1926	events: thread_resp.events,
1927	},
1928	Err(err) => AppResponse {
1929	ok: false,
1930	data: json!({ "error": err.to_string() }),
1931	events: Vec::new(),
1932	},
1933	}
1934	}
1935	AppRequest::SubmitUserInput {
1936	request_id,
1937	answers,
1938	} => {
1939	// Record the user's answers against the pending clarification
1940	// request so a driver can retrieve them. The headless runtime does
1941	// not block on `request_user_input` (fire-and-return, like
1942	// approval), so there is no in-flight turn to resume here — the
1943	// caller is expected to feed these answers into the next turn.
1944	let mut pending = state.pending_user_input.lock().await;
1945	if pending.contains_key(&request_id) {
1946	return AppResponse {
1947	ok: false,
1948	data: json!({
1949	"error": "request_id already resolved",
1950	"request_id": request_id,
1951	}),
1952	events: Vec::new(),
1953	};
1954	}
1955	pending.insert(request_id.clone(), answers);
1956	AppResponse {
1957	ok: true,
1958	data: json!({ "request_id": request_id, "resolved": true }),
1959	events: Vec::new(),
1960	}
1961	}
1962	}
1963	}
1964
1965	/// Propagate a new config snapshot to every place that must observe it:
1966	/// optionally persist it to disk, install it in the shared `state.config`,
1967	/// push it into the live [`Runtime`], and invalidate the cached stdio
1968	/// bridge so the next stdio request spawns a fresh child that reads the
1969	/// new on-disk config. Shared by `ConfigSet` / `ConfigUnset` / `ConfigReload`.
1970	///
1971	/// `exec_policy` is `Some` only on the reload path, which re-reads
1972	/// `permissions.toml` from disk; set/unset intentionally leave the live
1973	/// exec policy alone (use `ConfigReload` to pick up external permission
1974	/// edits). `persist` is false on the reload path because disk is already
1975	/// the source of truth there.
1976	async fn apply_config_update(
1977	state: &AppState,
1978	snapshot: codewhale_config::ConfigToml,
1979	exec_policy: Option<codewhale_execpolicy::ExecPolicyEngine>,
1980	persist: bool,
1981	) {
1982	if persist && let Err(e) = persist_config(state, snapshot.clone()).await {
1983	tracing::error!("Failed to persist config update: {e}");
1984	}
1985	{
1986	let mut cfg = state.config.write().await;
1987	*cfg = snapshot.clone();
1988	}
1989	// Sync into the live Runtime so the next turn picks up the change
1990	// without a restart. MCP server connections are NOT refreshed here —
1991	// see `Runtime::reload_config_and_policy` for the headless boundary;
1992	// the TUI's explicit `/mcp reload` operation is a separate path.
1993	{
1994	let mut runtime = state.runtime.write().await;
1995	match exec_policy {
1996	Some(policy) => runtime.reload_config_and_policy(snapshot, policy),
1997	None => runtime.update_config(snapshot),
1998	}
1999	}
2000	invalidate_stdio_bridge(state).await;
2001	}
2002
2003	async fn persist_config(state: &AppState, config: codewhale_config::ConfigToml) -> Result<()> {
2004	if state.config_path.is_none() {
2005	return Ok(());
2006	}
2007	let mut store = ConfigStore::load(state.config_path.clone())?;
2008	store.config = config;
2009	store.save()
2010	}
2011
2012	#[cfg(test)]
2013	mod tests {
2014	use super::*;
2015	use axum::body::{Body, to_bytes};
2016	use axum::extract::{Path as AxumPath, Query};
2017	use axum::http::header;
2018	use codewhale_protocol::AppRequest;
2019	use std::collections::HashMap;
2020	use std::fs;
2021	use tokio::io::AsyncReadExt;
2022	use tower::ServiceExt;
2023
2024	fn app_with_config(auth_token: Option<&str>) -> (Router, tempfile::TempDir) {
2025	let tmp = tempfile::tempdir().expect("tempdir");
2026	let config_path = tmp.path().join("config.toml");
2027	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2028	let state = build_state(
2029	Some(config_path),
2030	auth_token.map(std::string::ToString::to_string),
2031	)
2032	.expect("state");
2033	(app_router(state, &[]), tmp)
2034	}
2035
2036	#[test]
2037	fn build_state_keeps_resolved_explicit_config_path() {
2038	let tmp = tempfile::tempdir().expect("tempdir");
2039	let config_dir = tmp.path().join("config-dir");
2040	fs::create_dir_all(&config_dir).expect("config dir");
2041	let config_path = config_dir.join("config.toml");
2042	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2043
2044	let state = build_state(Some(config_path.clone()), None).expect("state");
2045
2046	assert_eq!(
2047	state.config_path.as_deref(),
2048	Some(
2049	config_path
2050	.canonicalize()
2051	.expect("canonical config")
2052	.as_path()
2053	)
2054	);
2055	}
2056
2057	#[tokio::test]
2058	async fn stdio_transport_never_registers_the_stdout_hook_sink() {
2059	let tmp = tempfile::tempdir().expect("tempdir");
2060	let config_path = tmp.path().join("config.toml");
2061	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2062
2063	let http_state =
2064	build_state_with_transport(Some(config_path.clone()), None, AppTransport::Http)
2065	.expect("http state");
2066	let stdio_state = build_state_with_transport(Some(config_path), None, AppTransport::Stdio)
2067	.expect("stdio state");
2068
2069	let http_sinks = http_state.runtime.read().await.hooks.sink_count();
2070	let stdio_sinks = stdio_state.runtime.read().await.hooks.sink_count();
2071	assert_eq!(
2072	http_sinks,
2073	stdio_sinks + 1,
2074	"HTTP mode keeps StdoutHookSink + JsonlHookSink; stdio must drop the stdout sink (#5165)"
2075	);
2076	}
2077
2078	async fn response_body_json(response: Response) -> Value {
2079	let bytes = to_bytes(response.into_body(), usize::MAX)
2080	.await
2081	.expect("body bytes");
2082	serde_json::from_slice(&bytes).expect("json response")
2083	}
2084
2085	#[tokio::test]
2086	async fn http_app_routes_require_bearer_token_when_auth_enabled() {
2087	let (app, _tmp) = app_with_config(Some("test-token"));
2088	let response = app
2089	.oneshot(
2090	Request::builder()
2091	.method(Method::POST)
2092	.uri("/app")
2093	.header(header::CONTENT_TYPE, "application/json")
2094	.body(Body::from(
2095	serde_json::to_vec(&AppRequest::ConfigGet {
2096	key: "api_key".to_string(),
2097	})
2098	.expect("request json"),
2099	))
2100	.expect("request"),
2101	)
2102	.await
2103	.expect("response");
2104
2105	assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
2106	}
2107
2108	#[tokio::test]
2109	async fn http_config_get_redacts_sensitive_values_after_auth() {
2110	let (app, _tmp) = app_with_config(Some("test-token"));
2111	let response = app
2112	.oneshot(
2113	Request::builder()
2114	.method(Method::POST)
2115	.uri("/app")
2116	.header(header::AUTHORIZATION, "Bearer test-token")
2117	.header(header::CONTENT_TYPE, "application/json")
2118	.body(Body::from(
2119	serde_json::to_vec(&AppRequest::ConfigGet {
2120	key: "api_key".to_string(),
2121	})
2122	.expect("request json"),
2123	))
2124	.expect("request"),
2125	)
2126	.await
2127	.expect("response");
2128
2129	assert_eq!(response.status(), StatusCode::OK);
2130	let body = response_body_json(response).await;
2131	assert_eq!(body["data"]["value"], "sk-d***cret");
2132	}
2133
2134	#[tokio::test]
2135	async fn cors_does_not_allow_arbitrary_origins() {
2136	let (app, _tmp) = app_with_config(Some("test-token"));
2137	let response = app
2138	.oneshot(
2139	Request::builder()
2140	.method(Method::GET)
2141	.uri("/healthz")
2142	.header(header::ORIGIN, "https://attacker.example")
2143	.body(Body::empty())
2144	.expect("request"),
2145	)
2146	.await
2147	.expect("response");
2148
2149	assert_eq!(response.status(), StatusCode::OK);
2150	assert!(
2151	response
2152	.headers()
2153	.get(header::ACCESS_CONTROL_ALLOW_ORIGIN)
2154	.is_none()
2155	);
2156	}
2157
2158	#[tokio::test]
2159	async fn build_state_loads_permissions_into_runtime_policy() {
2160	let tmp = tempfile::tempdir().expect("tempdir");
2161	let config_path = tmp.path().join("config.toml");
2162	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2163	fs::write(
2164	tmp.path().join("permissions.toml"),
2165	r#"
2166	[[rules]]
2167	tool = "exec_shell"
2168	command = "cargo test"
2169	"#,
2170	)
2171	.expect("write permissions");
2172
2173	let state = build_state(Some(config_path), None).expect("state");
2174	let runtime = state.runtime.read().await;
2175	let decision = runtime
2176	.exec_policy
2177	.check(codewhale_execpolicy::ExecPolicyContext {
2178	command: "cargo test --workspace",
2179	cwd: "/workspace",
2180	tool: Some("exec_shell"),
2181	path: None,
2182	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
2183	sandbox_mode: Some("workspace-write"),
2184	})
2185	.expect("policy check");
2186
2187	assert!(decision.allow);
2188	assert!(decision.requires_approval);
2189	assert_eq!(
2190	decision.matched_rule.as_deref(),
2191	Some("tool=exec_shell command=cargo test")
2192	);
2193	}
2194
2195	#[tokio::test]
2196	async fn config_reload_refreshes_runtime_config_and_exec_policy_from_disk() {
2197	let tmp = tempfile::tempdir().expect("tempdir");
2198	let config_path = tmp.path().join("config.toml");
2199	fs::write(
2200	&config_path,
2201	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
2202	)
2203	.expect("write config");
2204	// No permissions.toml at startup → exec_policy starts empty.
2205	let state = build_state(Some(config_path.clone()), None).expect("state");
2206
2207	// Sanity: initial runtime sees the on-disk model and has no rule.
2208	{
2209	let runtime = state.runtime.read().await;
2210	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-chat"));
2211	let decision = runtime
2212	.exec_policy
2213	.check(codewhale_execpolicy::ExecPolicyContext {
2214	command: "cargo test",
2215	cwd: "/workspace",
2216	tool: Some("exec_shell"),
2217	path: None,
2218	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
2219	sandbox_mode: Some("workspace-write"),
2220	})
2221	.expect("policy check");
2222	assert!(decision.matched_rule.is_none());
2223	}
2224
2225	// Edit both files on disk: new model + a permission rule.
2226	fs::write(
2227	&config_path,
2228	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-reasoner\"\n",
2229	)
2230	.expect("rewrite config");
2231	fs::write(
2232	tmp.path().join("permissions.toml"),
2233	r#"
2234	[[rules]]
2235	tool = "exec_shell"
2236	command = "cargo test"
2237	"#,
2238	)
2239	.expect("write permissions");
2240
2241	// ConfigReload must re-read both files and push them into the
2242	// live Runtime without a restart.
2243	let response =
2244	process_app_request(&state, AppRequest::ConfigReload, AppTransport::Stdio).await;
2245	assert!(response.ok, "reload should succeed");
2246	assert_eq!(response.data["reloaded"], true);
2247
2248	// The shared config lock reflects the new model.
2249	{
2250	let cfg = state.config.read().await;
2251	assert_eq!(cfg.model.as_deref(), Some("deepseek-reasoner"));
2252	}
2253	// The live Runtime reflects both the new model and the new rule.
2254	{
2255	let runtime = state.runtime.read().await;
2256	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-reasoner"));
2257	let decision = runtime
2258	.exec_policy
2259	.check(codewhale_execpolicy::ExecPolicyContext {
2260	command: "cargo test --workspace",
2261	cwd: "/workspace",
2262	tool: Some("exec_shell"),
2263	path: None,
2264	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
2265	sandbox_mode: Some("workspace-write"),
2266	})
2267	.expect("policy check");
2268	assert!(decision.allow);
2269	assert!(decision.requires_approval);
2270	assert_eq!(
2271	decision.matched_rule.as_deref(),
2272	Some("tool=exec_shell command=cargo test")
2273	);
2274	}
2275	}
2276
2277	#[tokio::test]
2278	async fn config_set_propagates_to_runtime_config_without_touching_exec_policy() {
2279	let tmp = tempfile::tempdir().expect("tempdir");
2280	let config_path = tmp.path().join("config.toml");
2281	fs::write(
2282	&config_path,
2283	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
2284	)
2285	.expect("write config");
2286	let state = build_state(Some(config_path.clone()), None).expect("state");
2287
2288	// Set a new model via the API. Only config.toml is touched; no
2289	// permissions.toml exists, so exec_policy must stay empty.
2290	let response = process_app_request(
2291	&state,
2292	AppRequest::ConfigSet {
2293	key: "model".to_string(),
2294	value: "deepseek-reasoner".to_string(),
2295	},
2296	AppTransport::Stdio,
2297	)
2298	.await;
2299	assert!(response.ok, "set should succeed");
2300
2301	// Live runtime sees the new model.
2302	{
2303	let runtime = state.runtime.read().await;
2304	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-reasoner"));
2305	// exec_policy was empty at startup and must remain empty.
2306	let decision = runtime
2307	.exec_policy
2308	.check(codewhale_execpolicy::ExecPolicyContext {
2309	command: "cargo test",
2310	cwd: "/workspace",
2311	tool: Some("exec_shell"),
2312	path: None,
2313	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
2314	sandbox_mode: Some("workspace-write"),
2315	})
2316	.expect("policy check");
2317	assert!(decision.matched_rule.is_none());
2318	}
2319	// The on-disk file was persisted.
2320	let persisted = fs::read_to_string(&config_path).expect("read config");
2321	assert!(persisted.contains("deepseek-reasoner"));
2322	}
2323
2324	/// A bridge stand-in with no child process: this test only cares about
2325	/// whether the cache slot survives, not about talking to a runtime.
2326	fn sentinel_bridge() -> SharedRuntimeBridge {
2327	Arc::new(Mutex::new(RuntimeBridge {
2328	base_url: "http://127.0.0.1:0".to_string(),
2329	client: reqwest::Client::new(),
2330	auth_token: None,
2331	child: None,
2332	thread_map: HashMap::from([("stdio-1".to_string(), "runtime-1".to_string())]),
2333	last_seq_by_thread: HashMap::new(),
2334	}))
2335	}
2336
2337	#[tokio::test]
2338	async fn failed_config_set_keeps_the_stdio_bridge() {
2339	// #4737: `set_value` rejects an invalid value before assigning, so the
2340	// request is a no-op — but `apply_config_update` ran anyway and
2341	// invalidated the cached bridge, dropping the child runtime along with
2342	// its thread map. A single bad value orphaned every in-flight stdio
2343	// thread, behind a response that correctly reported `ok: false`.
2344	//
2345	// Only `set_value` is exercised: an unknown key lands in `extras` and
2346	// succeeds, and `unset_value` has no failing input today, so its
2347	// identical guard has nothing to assert against.
2348	let tmp = tempfile::tempdir().expect("tempdir");
2349	let config_path = tmp.path().join("config.toml");
2350	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
2351	let state = build_state(Some(config_path.clone()), None).expect("state");
2352	*state.stdio_bridge.lock().await = Some(sentinel_bridge());
2353
2354	let response = process_app_request(
2355	&state,
2356	AppRequest::ConfigSet {
2357	key: "telemetry".to_string(),
2358	value: "not-a-bool".to_string(),
2359	},
2360	AppTransport::Stdio,
2361	)
2362	.await;
2363	assert!(!response.ok, "invalid value must fail: {response:?}");
2364
2365	let slot = state.stdio_bridge.lock().await;
2366	let kept = slot
2367	.as_ref()
2368	.expect("bridge must survive a failed config/set");
2369	assert_eq!(
2370	kept.lock()
2371	.await
2372	.thread_map
2373	.get("stdio-1")
2374	.map(String::as_str),
2375	Some("runtime-1"),
2376	"the live thread map must be intact",
2377	);
2378	}
2379
2380	#[tokio::test]
2381	async fn successful_config_set_still_invalidates_the_stdio_bridge() {
2382	// The other half of #4737: a mutation that did happen must still
2383	// rebuild the bridge, or the runtime keeps serving the old config.
2384	let tmp = tempfile::tempdir().expect("tempdir");
2385	let config_path = tmp.path().join("config.toml");
2386	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
2387	let state = build_state(Some(config_path.clone()), None).expect("state");
2388	*state.stdio_bridge.lock().await = Some(sentinel_bridge());
2389
2390	let response = process_app_request(
2391	&state,
2392	AppRequest::ConfigSet {
2393	key: "model".to_string(),
2394	value: "deepseek-reasoner".to_string(),
2395	},
2396	AppTransport::Stdio,
2397	)
2398	.await;
2399	assert!(response.ok, "valid set should succeed: {response:?}");
2400	assert!(
2401	state.stdio_bridge.lock().await.is_none(),
2402	"a successful config change must invalidate the cached bridge",
2403	);
2404	}
2405
2406	#[tokio::test]
2407	async fn config_unset_propagates_to_runtime_config() {
2408	let tmp = tempfile::tempdir().expect("tempdir");
2409	let config_path = tmp.path().join("config.toml");
2410	fs::write(
2411	&config_path,
2412	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
2413	)
2414	.expect("write config");
2415	let state = build_state(Some(config_path.clone()), None).expect("state");
2416
2417	// Sanity: runtime starts with the on-disk model.
2418	{
2419	let runtime = state.runtime.read().await;
2420	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-chat"));
2421	}
2422
2423	// Unset the model via the API. This walks a separate code path
2424	// from ConfigSet (unset_value + update_config), so it needs its
2425	// own regression coverage.
2426	let response = process_app_request(
2427	&state,
2428	AppRequest::ConfigUnset {
2429	key: "model".to_string(),
2430	},
2431	AppTransport::Stdio,
2432	)
2433	.await;
2434	assert!(response.ok, "unset should succeed");
2435
2436	// Live runtime sees the cleared model.
2437	{
2438	let runtime = state.runtime.read().await;
2439	assert!(runtime.config.model.is_none());
2440	}
2441	// Shared config lock agrees.
2442	{
2443	let cfg = state.config.read().await;
2444	assert!(cfg.model.is_none());
2445	}
2446	// The on-disk file no longer carries the model value.
2447	let persisted = fs::read_to_string(&config_path).expect("read config");
2448	assert!(!persisted.contains("deepseek-chat"));
2449	}
2450
2451	#[tokio::test]
2452	async fn config_reload_returns_error_when_disk_config_is_invalid() {
2453	let tmp = tempfile::tempdir().expect("tempdir");
2454	let config_path = tmp.path().join("config.toml");
2455	fs::write(
2456	&config_path,
2457	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
2458	)
2459	.expect("write config");
2460	let state = build_state(Some(config_path.clone()), None).expect("state");
2461
2462	// Corrupt the on-disk config so ConfigStore::load fails to parse.
2463	fs::write(&config_path, "api_key = \"unterminated\n").expect("corrupt config");
2464
2465	let response =
2466	process_app_request(&state, AppRequest::ConfigReload, AppTransport::Stdio).await;
2467	assert!(!response.ok, "reload of corrupt config must fail");
2468	let err = response.data["error"]
2469	.as_str()
2470	.expect("error message present")
2471	.to_string();
2472	assert!(
2473	err.contains("failed to load config"),
2474	"error should mention load failure, got: {err}"
2475	);
2476
2477	// Live state is untouched: the early-return on load error must
2478	// not have clobbered runtime.config or state.config.
2479	{
2480	let runtime = state.runtime.read().await;
2481	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-chat"));
2482	}
2483	{
2484	let cfg = state.config.read().await;
2485	assert_eq!(cfg.model.as_deref(), Some("deepseek-chat"));
2486	}
2487	}
2488
2489	async fn seed_test_bridge(state: &AppState) -> SharedRuntimeBridge {
2490	let bridge = Arc::new(Mutex::new(RuntimeBridge::from_base_url_for_test(
2491	"http://127.0.0.1:9".to_string(),
2492	)));
2493	*state.stdio_bridge.lock().await = Some(bridge.clone());
2494	bridge
2495	}
2496
2497	#[tokio::test]
2498	async fn config_set_invalidates_cached_stdio_bridge() {
2499	let tmp = tempfile::tempdir().expect("tempdir");
2500	let config_path = tmp.path().join("config.toml");
2501	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
2502	let state = build_state(Some(config_path), None).expect("state");
2503	seed_test_bridge(&state).await;
2504
2505	let response = process_app_request(
2506	&state,
2507	AppRequest::ConfigSet {
2508	key: "model".to_string(),
2509	value: "deepseek-reasoner".to_string(),
2510	},
2511	AppTransport::Stdio,
2512	)
2513	.await;
2514	assert!(response.ok, "set should succeed");
2515
2516	// The cached bridge child must be dropped so the next stdio request
2517	// spawns a fresh runtime that reads the persisted config.
2518	assert!(state.stdio_bridge.lock().await.is_none());
2519	}
2520
2521	#[tokio::test]
2522	async fn config_reload_invalidates_cached_stdio_bridge() {
2523	let tmp = tempfile::tempdir().expect("tempdir");
2524	let config_path = tmp.path().join("config.toml");
2525	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
2526	let state = build_state(Some(config_path), None).expect("state");
2527	seed_test_bridge(&state).await;
2528
2529	let response =
2530	process_app_request(&state, AppRequest::ConfigReload, AppTransport::Stdio).await;
2531	assert!(response.ok, "reload should succeed");
2532
2533	assert!(state.stdio_bridge.lock().await.is_none());
2534	}
2535
2536	#[tokio::test]
2537	async fn stdio_bridge_invalidation_not_blocked_by_in_flight_turn() {
2538	let (state, _tmp) = capability_test_state();
2539	let bridge = seed_test_bridge(&state).await;
2540
2541	// Simulate a long streaming turn holding the inner bridge lock.
2542	let _in_flight = bridge.lock().await;
2543
2544	// Invalidation only touches the cache slot, so it must complete
2545	// without waiting for the in-flight turn to release the bridge.
2546	tokio::time::timeout(Duration::from_secs(1), invalidate_stdio_bridge(&state))
2547	.await
2548	.expect("invalidation must not wait on bridge traffic");
2549	assert!(state.stdio_bridge.lock().await.is_none());
2550	}
2551
2552	#[tokio::test]
2553	async fn runtime_read_paths_run_concurrently() {
2554	// Tool/status/mcp handlers take read guards; two must coexist so a
2555	// long-running tool call cannot serialize unrelated requests. With
2556	// the old `Mutex<Runtime>` this pattern would deadlock.
2557	let (state, _tmp) = capability_test_state();
2558	let first = state.runtime.read().await;
2559	let second = state.runtime.read().await;
2560	assert!(first.app_status().ok);
2561	assert!(second.app_status().ok);
2562	}
2563
2564	#[tokio::test]
2565	async fn health_probes_advertise_legacy_deepseek_service_name() {
2566	// External probes still key off the DeepSeek-era service name; both
2567	// transports must serve it from the single compat shim.
2568	let (app, _tmp) = app_with_config(None);
2569	let response = app
2570	.oneshot(
2571	Request::builder()
2572	.method(Method::GET)
2573	.uri("/healthz")
2574	.body(Body::empty())
2575	.expect("request"),
2576	)
2577	.await
2578	.expect("response");
2579	let body = response_body_json(response).await;
2580	assert_eq!(body["service"], legacy_deepseek_compat::SERVICE_NAME);
2581	assert_eq!(body["service"], "deepseek-app-server");
2582
2583	let (state, _tmp) = capability_test_state();
2584	let stdio = dispatch_stdio_request(&state, "healthz", json!({}))
2585	.await
2586	.expect("stdio healthz");
2587	assert_eq!(
2588	stdio.result["service"],
2589	legacy_deepseek_compat::SERVICE_NAME
2590	);
2591	}
2592
2593	#[test]
2594	fn non_loopback_bind_without_auth_fails_fast() {
2595	let options = AppServerOptions {
2596	listen: "0.0.0.0:8787".parse().expect("socket addr"),
2597	config_path: None,
2598	auth_token: None,
2599	insecure_no_auth: false,
2600	cors_origins: Vec::new(),
2601	};
2602
2603	let err =
2604	resolve_auth_token(&options).expect_err("non-loopback generated auth should fail");
2605	assert!(err.to_string().contains("without explicit auth token"));
2606	}
2607
2608	#[tokio::test]
2609	async fn stdio_transport_redacts_config_get_secrets() {
2610	let tmp = tempfile::tempdir().expect("tempdir");
2611	let config_path = tmp.path().join("config.toml");
2612	fs::write(&config_path, "").expect("write config");
2613	let state = build_state(Some(config_path), None).expect("state");
2614	{
2615	let mut cfg = state.config.write().await;
2616	cfg.api_key = Some("sk-deepseek-secret".to_string());
2617	}
2618
2619	let response = process_app_request(
2620	&state,
2621	AppRequest::ConfigGet {
2622	key: "api_key".to_string(),
2623	},
2624	AppTransport::Stdio,
2625	)
2626	.await;
2627
2628	assert_eq!(response.data["value"], "sk-d***cret");
2629	}
2630
2631	#[tokio::test]
2632	async fn stdio_thread_goal_methods_round_trip_persisted_goal() {
2633	let tmp = tempfile::tempdir().expect("tempdir");
2634	let config_path = tmp.path().join("config.toml");
2635	fs::write(&config_path, "").expect("write config");
2636	let state = build_state(Some(config_path), None).expect("state");
2637
2638	let capabilities = dispatch_stdio_request(&state, "thread/capabilities", json!({}))
2639	.await
2640	.expect("thread capabilities");
2641	assert!(
2642	capabilities.result["methods"]
2643	.as_array()
2644	.expect("methods")
2645	.iter()
2646	.any(\|method\| method == "thread/goal/set")
2647	);
2648
2649	let started = dispatch_stdio_request(&state, "thread/start", json!({}))
2650	.await
2651	.expect("start thread");
2652	let thread_id = started.result["thread_id"]
2653	.as_str()
2654	.expect("thread id")
2655	.to_string();
2656
2657	let set = dispatch_stdio_request(
2658	&state,
2659	"thread/goal/set",
2660	json!({
2661	"thread_id": thread_id,
2662	"objective": "Release 0.8.59",
2663	"token_budget": 59000
2664	}),
2665	)
2666	.await
2667	.expect("set goal");
2668	assert_eq!(set.result["status"], "ok");
2669	assert_eq!(set.result["goal"]["objective"], "Release 0.8.59");
2670	assert_eq!(set.result["goal"]["status"], "active");
2671
2672	let got = dispatch_stdio_request(
2673	&state,
2674	"thread/goal/get",
2675	json!({
2676	"thread_id": thread_id
2677	}),
2678	)
2679	.await
2680	.expect("get goal");
2681	assert_eq!(got.result["goal"]["token_budget"], 59000);
2682
2683	let cleared = dispatch_stdio_request(
2684	&state,
2685	"thread/goal/clear",
2686	json!({
2687	"thread_id": thread_id
2688	}),
2689	)
2690	.await
2691	.expect("clear goal");
2692	assert_eq!(cleared.result["status"], "cleared");
2693	assert_eq!(cleared.result["data"]["cleared"], true);
2694	}
2695
2696	#[tokio::test]
2697	async fn stdio_resume_of_missing_thread_fails_without_clobbering_the_hint() {
2698	let tmp = tempfile::tempdir().expect("tempdir");
2699	let config_path = tmp.path().join("config.toml");
2700	fs::write(&config_path, "").expect("write config");
2701	let state = build_state(Some(config_path), None).expect("state");
2702
2703	// A cached hint for a thread the runtime no longer knows: the exact
2704	// clobber scenario from #5171.
2705	let workspace = tmp.path().join("ws");
2706	{
2707	let mut hints = state.stdio_thread_hints.lock().await;
2708	hints.insert(
2709	"ghost-thread".to_string(),
2710	RuntimeThreadHint {
2711	model: Some("deepseek-v4-pro".to_string()),
2712	workspace: Some(workspace.clone()),
2713	},
2714	);
2715	}
2716
2717	let err = dispatch_stdio_request(
2718	&state,
2719	"thread/resume",
2720	json!({ "thread_id": "ghost-thread" }),
2721	)
2722	.await
2723	.expect_err("resuming a missing thread must fail with a named not-found error");
2724	assert_eq!(err.code, -32004);
2725	assert!(err.message.contains("ghost-thread"), "{}", err.message);
2726
2727	let fork_err = dispatch_stdio_request(
2728	&state,
2729	"thread/fork",
2730	json!({ "thread_id": "ghost-thread" }),
2731	)
2732	.await
2733	.expect_err("forking a missing thread must fail with a named not-found error");
2734	assert_eq!(fork_err.code, -32004);
2735
2736	let hints = state.stdio_thread_hints.lock().await;
2737	let hint = hints.get("ghost-thread").expect("cached hint survives");
2738	assert_eq!(hint.model.as_deref(), Some("deepseek-v4-pro"));
2739	assert_eq!(hint.workspace.as_deref(), Some(workspace.as_path()));
2740	}
2741
2742	fn sse_frame(event: &str, payload: Value) -> String {
2743	format!("event: {event}\ndata: {payload}\n\n")
2744	}
2745	/// A runtime whose turn never ends on its own — only an interrupt stops
2746	/// it. That is the shape of the runaway turn this protects against.
2747	async fn spawn_uninterruptible_until_asked_runtime() -> (
2748	String,
2749	Arc<tokio::sync::Notify>,
2750	tokio::task::JoinHandle<()>,
2751	) {
2752	use axum::body::Body;
2753	use axum::extract::Path as AxumPath;
2754
2755	let interrupted = Arc::new(tokio::sync::Notify::new());
2756
2757	async fn create_turn(AxumPath(_thread_id): AxumPath<String>) -> Json<Value> {
2758	Json(json!({ "turn": { "id": "turn_runaway" } }))
2759	}
2760	async fn create_thread() -> Json<Value> {
2761	Json(json!({ "id": "thr_runaway" }))
2762	}
2763	async fn interrupt(
2764	State(notify): State<Arc<tokio::sync::Notify>>,
2765	AxumPath((_thread_id, _turn_id)): AxumPath<(String, String)>,
2766	) -> Json<Value> {
2767	notify.notify_waiters();
2768	Json(json!({ "ok": true }))
2769	}
2770	async fn thread_events(
2771	State(notify): State<Arc<tokio::sync::Notify>>,
2772	AxumPath(_thread_id): AxumPath<String>,
2773	) -> ([(header::HeaderName, &'static str); 1], Body) {
2774	// Hold the event response open until something interrupts the
2775	// turn. Nothing else can end it, which is the point.
2776	notify.notified().await;
2777	let body = [
2778	sse_frame(
2779	"item.delta",
2780	json!({
2781	"seq": 1,
2782	"turn_id": "turn_runaway",
2783	"payload": { "kind": "agent_message", "delta": "thinking" }
2784	}),
2785	),
2786	sse_frame(
2787	"turn.completed",
2788	json!({
2789	"seq": 2,
2790	"turn_id": "turn_runaway",
2791	"payload": { "turn": { "status": "interrupted" } }
2792	}),
2793	),
2794	]
2795	.concat();
2796	(
2797	[(header::CONTENT_TYPE, "text/event-stream")],
2798	Body::from(body),
2799	)
2800	}
2801
2802	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2803	.await
2804	.expect("bind test listener");
2805	let addr = listener.local_addr().expect("listener addr");
2806	let app = Router::new()
2807	.route("/v1/threads", post(create_thread))
2808	.route("/v1/threads/{thread_id}/turns", post(create_turn))
2809	.route(
2810	"/v1/threads/{thread_id}/turns/{turn_id}/interrupt",
2811	post(interrupt),
2812	)
2813	.route("/v1/threads/{thread_id}/events", get(thread_events))
2814	.with_state(interrupted.clone());
2815	let server = tokio::spawn(async move {
2816	axum::serve(listener, app)
2817	.await
2818	.expect("serve test runtime");
2819	});
2820	(format!("http://{addr}"), interrupted, server)
2821	}
2822
2823	#[tokio::test]
2824	async fn interrupt_stops_a_turn_that_would_otherwise_stream_forever() {
2825	let (base_url, _notify, server) = spawn_uninterruptible_until_asked_runtime().await;
2826	let (state, _tmp) = capability_test_state();
2827	*state.stdio_bridge.lock().await = Some(Arc::new(Mutex::new(
2828	RuntimeBridge::from_base_url_for_test(base_url),
2829	)));
2830
2831	let (client, server_side) = tokio::io::duplex(16 * 1024);
2832	let (client_reader, mut client_writer) = tokio::io::split(client);
2833
2834	let loop_state = state.clone();
2835	let loop_handle = tokio::spawn(async move {
2836	let (rx, tx) = tokio::io::split(server_side);
2837	run_stdio_loop(&loop_state, BufReader::new(rx).lines(), tx).await
2838	});
2839
2840	// Start the runaway turn.
2841	client_writer
2842	.write_all(
2843	b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"thread/message\",\
2844	\"params\":{\"thread_id\":\"thr_a\",\"input\":\"go\"}}\n",
2845	)
2846	.await
2847	.expect("send thread/message");
2848
2849	// Wait until the turn is genuinely in flight before cancelling, so the
2850	// test exercises mid-stream cancellation rather than a race.
2851	tokio::time::timeout(Duration::from_secs(10), async {
2852	loop {
2853	if state.in_flight_turns.lock().await.contains_key("thr_a") {
2854	return;
2855	}
2856	tokio::time::sleep(Duration::from_millis(10)).await;
2857	}
2858	})
2859	.await
2860	.expect("turn should register itself as in flight");
2861
2862	// The read loop must accept this while the turn holds the bridge.
2863	client_writer
2864	.write_all(
2865	b"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"thread/interrupt\",\
2866	\"params\":{\"thread_id\":\"thr_a\"}}\n",
2867	)
2868	.await
2869	.expect("send thread/interrupt");
2870	client_writer
2871	.write_all(b"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"shutdown\"}\n")
2872	.await
2873	.expect("send shutdown");
2874
2875	let finished = tokio::time::timeout(Duration::from_secs(20), loop_handle)
2876	.await
2877	.expect("the loop must exit rather than hang on the runaway turn");
2878	finished.expect("join loop").expect("loop result");
2879
2880	let mut output = String::new();
2881	let mut lines = BufReader::new(client_reader);
2882	lines
2883	.read_to_string(&mut output)
2884	.await
2885	.expect("read stdio output");
2886
2887	let responses: Vec<Value> = output
2888	.lines()
2889	.filter_map(\|line\| serde_json::from_str::<Value>(line).ok())
2890	.collect();
2891	let by_id = \|id: u64\| {
2892	responses
2893	.iter()
2894	.find(\|value\| value["id"] == json!(id))
2895	.unwrap_or_else(\|\| panic!("no response for id {id} in {output}"))
2896	.clone()
2897	};
2898
2899	// The turn ended as interrupted rather than running to completion.
2900	assert!(
2901	by_id(1)["error"].is_object(),
2902	"the interrupted turn should report an error, got: {}",
2903	by_id(1)
2904	);
2905	assert_eq!(by_id(2)["result"]["interrupted"], json!(true));
2906	assert_eq!(by_id(3)["result"]["status"], json!("stopped"));
2907
2908	server.abort();
2909	let _ = server.await;
2910	}
2911
2912	#[tokio::test]
2913	async fn interrupting_an_idle_thread_is_not_an_error() {
2914	let (state, _tmp) = capability_test_state();
2915	let response = dispatch_stdio_request(
2916	&state,
2917	"thread/interrupt",
2918	json!({ "thread_id": "thr_nothing_running" }),
2919	)
2920	.await
2921	.expect("interrupt dispatch");
2922	assert_eq!(response.result["interrupted"], json!(false));
2923	}
2924
2925	#[tokio::test]
2926	async fn stdio_runtime_bridge_streams_response_delta_events() {
2927	async fn create_turn(AxumPath(thread_id): AxumPath<String>) -> Json<Value> {
2928	Json(json!({
2929	"thread": { "id": thread_id },
2930	"turn": { "id": "turn_test" },
2931	}))
2932	}
2933
2934	async fn thread_events(
2935	AxumPath(thread_id): AxumPath<String>,
2936	Query(query): Query<HashMap<String, String>>,
2937	) -> ([(header::HeaderName, &'static str); 1], String) {
2938	assert_eq!(thread_id, "thr_test");
2939	assert_eq!(query.get("since_seq").map(String::as_str), Some("0"));
2940
2941	let body = [
2942	sse_frame(
2943	"item.delta",
2944	json!({
2945	"seq": 1,
2946	"turn_id": "turn_test",
2947	"payload": {
2948	"kind": "agent_message",
2949	"delta": "hello"
2950	}
2951	}),
2952	),
2953	sse_frame(
2954	"turn.completed",
2955	json!({
2956	"seq": 2,
2957	"turn_id": "turn_test",
2958	"payload": {
2959	"turn": {
2960	"status": "completed"
2961	}
2962	}
2963	}),
2964	),
2965	]
2966	.concat();
2967
2968	([(header::CONTENT_TYPE, "text/event-stream")], body)
2969	}
2970
2971	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2972	.await
2973	.expect("bind test listener");
2974	let addr = listener.local_addr().expect("listener addr");
2975	let app = Router::new()
2976	.route("/v1/threads/{thread_id}/turns", post(create_turn))
2977	.route("/v1/threads/{thread_id}/events", get(thread_events));
2978
2979	let server = tokio::spawn(async move {
2980	axum::serve(listener, app)
2981	.await
2982	.expect("serve test runtime");
2983	});
2984
2985	let mut bridge = RuntimeBridge::from_base_url_for_test(format!("http://{addr}"));
2986	let (mut reader, mut writer) = tokio::io::duplex(4096);
2987
2988	let result = bridge
2989	.message_thread("thr_test", "hello", &mut writer, None)
2990	.await
2991	.expect("message_thread should succeed");
2992	drop(writer);
2993
2994	let mut stdout = Vec::new();
2995	reader
2996	.read_to_end(&mut stdout)
2997	.await
2998	.expect("read stdio output");
2999	server.abort();
3000	let _ = server.await;
3001
3002	let lines: Vec<Value> = String::from_utf8(stdout)
3003	.expect("utf8 output")
3004	.lines()
3005	.map(\|line\| serde_json::from_str(line).expect("json line"))
3006	.collect();
3007
3008	assert_eq!(
3009	result.get("status").and_then(Value::as_str),
3010	Some("accepted")
3011	);
3012	assert_eq!(
3013	result.pointer("/data/turn_id").and_then(Value::as_str),
3014	Some("turn_test")
3015	);
3016	assert_eq!(bridge.last_seq_by_thread.get("thr_test"), Some(&2));
3017
3018	let event_types: Vec<&str> = lines
3019	.iter()
3020	.map(\|line\| {
3021	line.get("type")
3022	.and_then(Value::as_str)
3023	.expect("event type")
3024	})
3025	.collect();
3026	assert_eq!(
3027	event_types,
3028	vec!["response_start", "response_delta", "response_end"]
3029	);
3030	assert_eq!(lines[1]["delta"], "hello");
3031	}
3032
3033	#[tokio::test]
3034	async fn stdio_runtime_bridge_applies_thread_start_hints() {
3035	async fn create_thread(Json(body): Json<Value>) -> Json<Value> {
3036	assert_eq!(body["model"], "deepseek-v4");
3037	assert_eq!(body["workspace"], "/tmp/codewhale-stdio");
3038	Json(json!({
3039	"id": "thr_runtime",
3040	"model": body["model"].clone(),
3041	"workspace": body["workspace"].clone(),
3042	}))
3043	}
3044
3045	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
3046	.await
3047	.expect("bind test listener");
3048	let addr = listener.local_addr().expect("listener addr");
3049	let app = Router::new().route("/v1/threads", post(create_thread));
3050
3051	let server = tokio::spawn(async move {
3052	axum::serve(listener, app)
3053	.await
3054	.expect("serve test runtime");
3055	});
3056
3057	let mut bridge = RuntimeBridge::from_base_url_for_test(format!("http://{addr}"));
3058	let runtime_id = bridge
3059	.ensure_runtime_thread(
3060	"legacy_thread",
3061	Some(RuntimeThreadHint {
3062	model: Some("deepseek-v4".to_string()),
3063	workspace: Some(PathBuf::from("/tmp/codewhale-stdio")),
3064	}),
3065	)
3066	.await
3067	.expect("runtime thread");
3068	server.abort();
3069	let _ = server.await;
3070
3071	assert_eq!(runtime_id, "thr_runtime");
3072	assert_eq!(
3073	bridge.thread_map.get("legacy_thread").map(String::as_str),
3074	Some("thr_runtime")
3075	);
3076	}
3077
3078	// ── capability drift guard ─────────────────────────────────────────
3079	//
3080	// The stdio `capabilities` method is the benchmark/SDK contract: external
3081	// harnesses probe it (without spending model tokens) to learn what the
3082	// app-server can do. Pin the advertised method set so any change forces a
3083	// deliberate update here, in the dispatcher, and in docs/RUNTIME_API.md.
3084
3085	/// Methods advertised by the top-level `capabilities` probe, in order.
3086	const EXPECTED_CAPABILITY_METHODS: &[&str] = &[
3087	"healthz",
3088	"thread/capabilities",
3089	"thread/request",
3090	"thread/create",
3091	"thread/start",
3092	"thread/resume",
3093	"thread/fork",
3094	"thread/list",
3095	"thread/read",
3096	"thread/set_name",
3097	"thread/goal/set",
3098	"thread/goal/get",
3099	"thread/goal/clear",
3100	"thread/archive",
3101	"thread/unarchive",
3102	"thread/message",
3103	"thread/interrupt",
3104	"app/capabilities",
3105	"app/request",
3106	"app/config/get",
3107	"app/config/set",
3108	"app/config/unset",
3109	"app/config/list",
3110	"app/config/reload",
3111	"app/models",
3112	"app/thread_loaded_list",
3113	"prompt/capabilities",
3114	"prompt/request",
3115	"prompt/run",
3116	"shutdown",
3117	];
3118
3119	fn capability_test_state() -> (AppState, tempfile::TempDir) {
3120	let tmp = tempfile::tempdir().expect("tempdir");
3121	let config_path = tmp.path().join("config.toml");
3122	fs::write(&config_path, "").expect("write config");
3123	let state = build_state(Some(config_path), None).expect("state");
3124	(state, tmp)
3125	}
3126
3127	#[tokio::test]
3128	async fn capabilities_method_set_is_stable() {
3129	let (state, _tmp) = capability_test_state();
3130	let caps = dispatch_stdio_request(&state, "capabilities", json!({}))
3131	.await
3132	.expect("capabilities dispatch");
3133	let methods: Vec<String> = caps.result["methods"]
3134	.as_array()
3135	.expect("methods array")
3136	.iter()
3137	.map(\|m\| m.as_str().expect("method string").to_string())
3138	.collect();
3139	assert_eq!(
3140	methods, EXPECTED_CAPABILITY_METHODS,
3141	"app-server stdio capability set drifted; update the dispatcher, this \
3142	snapshot, and docs/RUNTIME_API.md together"
3143	);
3144	}
3145
3146	#[tokio::test]
3147	async fn every_advertised_capability_is_dispatchable() {
3148	let (state, _tmp) = capability_test_state();
3149	// Empty params: methods may fail validation (-32602), but none may report
3150	// method-not-found (-32601). Required fields (e.g. PromptRequest.prompt)
3151	// make the prompt routes fail at parse time, so no model tokens are spent.
3152	for method in EXPECTED_CAPABILITY_METHODS {
3153	if let Err(err) = dispatch_stdio_request(&state, method, json!({})).await {
3154	assert_ne!(
3155	err.code,
3156	JsonRpcError::method_not_found(method).code,
3157	"advertised capability `{method}` is not dispatchable"
3158	);
3159	}
3160	}
3161	}
3162
3163	// ── resolve_auth_token ─────────────────────────────────────────────
3164
3165	#[test]
3166	fn auth_token_empty_string_fails() {
3167	let options = AppServerOptions {
3168	listen: "127.0.0.1:0".parse().expect("addr"),
3169	config_path: None,
3170	auth_token: Some(" ".to_string()),
3171	insecure_no_auth: false,
3172	cors_origins: Vec::new(),
3173	};
3174	let err = resolve_auth_token(&options).expect_err("empty token should fail");
3175	assert!(err.to_string().contains("cannot be empty"));
3176	}
3177
3178	#[test]
3179	fn auth_token_generated_when_none_provided() {
3180	let options = AppServerOptions {
3181	listen: "127.0.0.1:0".parse().expect("addr"),
3182	config_path: None,
3183	auth_token: None,
3184	insecure_no_auth: false,
3185	cors_origins: Vec::new(),
3186	};
3187	let token = resolve_auth_token(&options).unwrap();
3188	assert!(token.is_some());
3189	assert!(token.unwrap().starts_with("cwapp_"));
3190	}
3191
3192	#[test]
3193	fn runtime_bridge_command_keeps_auth_token_out_of_argv() {
3194	// FR001-C001: runtime auth token must not appear on the child argv
3195	// (visible via local `ps`); pass it via env instead.
3196	let token = "cwrt_unit_test_secret_token_not_for_argv";
3197	let cmd = RuntimeBridge::runtime_command(None, 18787, token).expect("command");
3198	let argv: Vec<String> = cmd
3199	.get_args()
3200	.map(\|a\| a.to_string_lossy().into_owned())
3201	.collect();
3202	assert!(
3203	!argv
3204	.iter()
3205	.any(\|a\| a.contains(token) \|\| a == "--auth-token"),
3206	"auth token must not be present in child argv: {argv:?}"
3207	);
3208	let envs: Vec<(String, String)> = cmd
3209	.get_envs()
3210	.filter_map(\|(k, v)\| {
3211	Some((
3212	k.to_string_lossy().into_owned(),
3213	v?.to_string_lossy().into_owned(),
3214	))
3215	})
3216	.collect();
3217	assert!(
3218	envs.iter()
3219	.any(\|(k, v)\| k == "CODEWHALE_RUNTIME_TOKEN" && v == token),
3220	"token must be carried via CODEWHALE_RUNTIME_TOKEN: {envs:?}"
3221	);
3222	assert!(
3223	envs.iter()
3224	.any(\|(k, v)\| k == "DEEPSEEK_RUNTIME_TOKEN" && v == token),
3225	"legacy alias DEEPSEEK_RUNTIME_TOKEN must also carry the token: {envs:?}"
3226	);
3227	}
3228
3229	#[test]
3230	fn generated_auth_status_does_not_render_token() {
3231	let rendered = app_server_auth_status_lines(false).join("\n");
3232
3233	assert!(!rendered.contains("Authorization: Bearer"));
3234	assert!(rendered.contains("not printed"));
3235	assert!(rendered.contains("CODEWHALE_APP_SERVER_TOKEN"));
3236	}
3237
3238	#[test]
3239	fn auth_token_explicit_is_preserved() {
3240	let options = AppServerOptions {
3241	listen: "127.0.0.1:0".parse().expect("addr"),
3242	config_path: None,
3243	auth_token: Some("my-secret".to_string()),
3244	insecure_no_auth: false,
3245	cors_origins: Vec::new(),
3246	};
3247	let token = resolve_auth_token(&options).unwrap();
3248	assert_eq!(token.as_deref(), Some("my-secret"));
3249	}
3250
3251	#[test]
3252	fn auth_token_explicit_allows_non_loopback_bind() {
3253	let options = AppServerOptions {
3254	listen: "0.0.0.0:8787".parse().expect("socket addr"),
3255	config_path: None,
3256	auth_token: Some("my-secret".to_string()),
3257	insecure_no_auth: false,
3258	cors_origins: Vec::new(),
3259	};
3260	let token = resolve_auth_token(&options).unwrap();
3261	assert_eq!(token.as_deref(), Some("my-secret"));
3262	}
3263
3264	#[test]
3265	fn insecure_no_auth_on_loopback_returns_none() {
3266	let options = AppServerOptions {
3267	listen: "127.0.0.1:0".parse().expect("addr"),
3268	config_path: None,
3269	auth_token: None,
3270	insecure_no_auth: true,
3271	cors_origins: Vec::new(),
3272	};
3273	let token = resolve_auth_token(&options).unwrap();
3274	assert!(token.is_none());
3275	}
3276
3277	#[test]
3278	fn insecure_no_auth_on_non_loopback_fails_fast() {
3279	let options = AppServerOptions {
3280	listen: "0.0.0.0:8787".parse().expect("socket addr"),
3281	config_path: None,
3282	auth_token: None,
3283	insecure_no_auth: true,
3284	cors_origins: Vec::new(),
3285	};
3286
3287	let err = resolve_auth_token(&options).expect_err("non-loopback unauth should fail");
3288	assert!(
3289	err.to_string()
3290	.contains("refusing unauthenticated app-server bind")
3291	);
3292	}
3293
3294	// ── cors_layer ─────────────────────────────────────────────────────
3295
3296	#[test]
3297	fn cors_layer_includes_default_origins() {
3298	let layer = cors_layer(&[]);
3299	// Just verify it doesn't panic and creates successfully
3300	let _ = layer;
3301	}
3302
3303	#[test]
3304	fn cors_layer_adds_extra_origins() {
3305	let extras = vec!["https://example.com".to_string()];
3306	let layer = cors_layer(&extras);
3307	let _ = layer;
3308	}
3309
3310	#[test]
3311	fn cors_layer_skips_empty_origins() {
3312	let extras = vec!["".to_string(), " ".to_string()];
3313	let layer = cors_layer(&extras);
3314	let _ = layer;
3315	}
3316
3317	// ── JsonRpc helpers ────────────────────────────────────────────────
3318
3319	#[test]
3320	fn params_or_object_returns_object_for_null() {
3321	let result = params_or_object(Value::Null);
3322	assert_eq!(result, json!({}));
3323	}
3324
3325	#[test]
3326	fn params_or_object_passthrough_for_non_null() {
3327	let input = json!({"key": "value"});
3328	let result = params_or_object(input.clone());
3329	assert_eq!(result, input);
3330	}
3331
3332	#[test]
3333	fn jsonrpc_result_format() {
3334	let result = jsonrpc_result(Some(json!(1)), json!({"ok": true}));
3335	assert_eq!(result["jsonrpc"], "2.0");
3336	assert_eq!(result["id"], 1);
3337	assert_eq!(result["result"]["ok"], true);
3338	}
3339
3340	#[test]
3341	fn jsonrpc_result_null_id() {
3342	let result = jsonrpc_result(None, json!(null));
3343	assert_eq!(result["id"], Value::Null);
3344	}
3345
3346	#[test]
3347	fn jsonrpc_error_format() {
3348	let err = jsonrpc_error(Some(json!(2)), JsonRpcError::internal("oops"));
3349	assert_eq!(err["jsonrpc"], "2.0");
3350	assert_eq!(err["id"], 2);
3351	assert_eq!(err["error"]["code"], -32603);
3352	assert_eq!(err["error"]["message"], "oops");
3353	}
3354
3355	#[test]
3356	fn jsonrpc_error_codes() {
3357	assert_eq!(JsonRpcError::parse_error("").code, -32700);
3358	assert_eq!(JsonRpcError::invalid_request("").code, -32600);
3359	assert_eq!(JsonRpcError::method_not_found("x").code, -32601);
3360	assert_eq!(JsonRpcError::invalid_params("").code, -32602);
3361	assert_eq!(JsonRpcError::internal("").code, -32603);
3362	}
3363
3364	// ── AppServerOptions ───────────────────────────────────────────────
3365
3366	#[test]
3367	fn app_server_options_debug_does_not_leak_token() {
3368	let options = AppServerOptions {
3369	listen: "127.0.0.1:8080".parse().expect("addr"),
3370	config_path: None,
3371	auth_token: Some("secret-token".to_string()),
3372	insecure_no_auth: false,
3373	cors_origins: vec!["https://example.com".to_string()],
3374	};
3375	let debug = format!("{options:?}");
3376	assert!(!debug.contains("secret-token"));
3377	assert!(debug.contains("<redacted>"));
3378	assert!(debug.contains("8080"));
3379	}
3380
3381	// ── Default CORS origins ──────────────────────────────────────────
3382
3383	#[test]
3384	fn default_cors_origins_include_common_dev_ports() {
3385	assert!(DEFAULT_CORS_ORIGINS.contains(&"http://localhost:3000"));
3386	assert!(DEFAULT_CORS_ORIGINS.contains(&"http://localhost:5173"));
3387	assert!(DEFAULT_CORS_ORIGINS.contains(&"tauri://localhost"));
3388	}
3389	}
3390

3390 lines RUST