lib.rs - CodeWhale

lib.rs

1	use codewhale_protocol::runtime::{MAX_RUNTIME_IMAGE_BODY_BYTES, RuntimeImageInput};
2	use std::collections::{HashMap, VecDeque};
3	use std::net::SocketAddr;
4	use std::path::{Path, PathBuf};
5	use std::process::{Child, Command, Stdio};
6	use std::sync::Arc;
7	use std::time::{Duration, Instant};
8
9	use anyhow::{Context, Result, anyhow, bail};
10	use axum::extract::{DefaultBodyLimit, Request, State};
11	use axum::http::{HeaderValue, Method, StatusCode, header};
12	use axum::middleware::{self, Next};
13	use axum::response::{IntoResponse, Response};
14	use axum::routing::{get, post};
15	use axum::{Json, Router};
16	use codewhale_agent::ModelRegistry;
17	use codewhale_config::ConfigStore;
18	use codewhale_core::Runtime;
19	use codewhale_hooks::{HookDispatcher, JsonlHookSink, StdoutHookSink, UnixSocketHookSink};
20	use codewhale_mcp::McpManager;
21	use codewhale_protocol::{
22	AppRequest, AppResponse, EventFrame, PromptRequest, PromptResponse, ResponseChannel,
23	ThreadGoalClearParams, ThreadGoalGetParams, ThreadGoalSetParams, ThreadRequest, ThreadResponse,
24	};
25	use codewhale_state::StateStore;
26	use codewhale_tools::{ToolCall, ToolRegistry};
27	use serde::de::DeserializeOwned;
28	use serde::{Deserialize, Serialize};
29	use serde_json::{Value, json};
30	use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader};
31	use tokio::sync::{Mutex, RwLock};
32	use tower_http::cors::CorsLayer;
33	use uuid::Uuid;
34
35	mod chat_completions;
36	pub mod daemon_socket;
37
38	/// Legacy DeepSeek-era naming kept for external compatibility.
39	///
40	/// CodeWhale began life as DeepSeek-TUI; existing health probes, SDK
41	/// harnesses, and on-disk layouts still key off these names. Every remaining
42	/// legacy reference in this crate routes through this shim so a future
43	/// coordinated migration touches exactly one place (repo policy: preserve
44	/// legacy migration care).
45	mod legacy_deepseek_compat {
46	use std::path::PathBuf;
47
48	/// Service name advertised by the HTTP and stdio health probes.
49	pub(crate) const SERVICE_NAME: &str = "deepseek-app-server";
50
51	/// Fallback hook-event log location used when no config path is
52	/// provided (legacy `.deepseek/` dot-directory layout).
53	pub(crate) fn default_events_log_path() -> PathBuf {
54	PathBuf::from(".deepseek/events.jsonl")
55	}
56	}
57
58	/// Upper bound on JSON request bodies accepted by the HTTP app-server.
59	const MAX_HTTP_BODY_BYTES: usize = 16 * 1024 * 1024;
60	const MAX_SSE_FRAME_BYTES: usize = 16 * 1024 * 1024;
61
62	const DEFAULT_CORS_ORIGINS: &[&str] = &[
63	"http://localhost",
64	"http://localhost:1420",
65	"http://localhost:3000",
66	"http://localhost:5173",
67	"http://127.0.0.1",
68	"http://127.0.0.1:1420",
69	"tauri://localhost",
70	];
71
72	#[derive(Clone)]
73	pub struct AppServerOptions {
74	pub listen: SocketAddr,
75	pub config_path: Option<PathBuf>,
76	pub auth_token: Option<String>,
77	pub insecure_no_auth: bool,
78	pub cors_origins: Vec<String>,
79	}
80
81	impl std::fmt::Debug for AppServerOptions {
82	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83	f.debug_struct("AppServerOptions")
84	.field("listen", &self.listen)
85	.field("config_path", &self.config_path)
86	.field(
87	"auth_token",
88	&self.auth_token.as_ref().map(\|_\| "<redacted>"),
89	)
90	.field("insecure_no_auth", &self.insecure_no_auth)
91	.field("cors_origins", &self.cors_origins)
92	.finish()
93	}
94	}
95
96	/// Cached app-server→runtime bridge handle.
97	///
98	/// The outer [`AppState::runtime_bridge`] mutex guards only the cache slot;
99	/// this inner mutex serializes traffic on one bridge (single child process
100	/// plus per-thread seq bookkeeping requires ordered access).
101	type SharedRuntimeBridge = Arc<Mutex<RuntimeBridge>>;
102
103	#[derive(Clone)]
104	struct AppState {
105	config_path: Option<PathBuf>,
106	config: Arc<RwLock<codewhale_config::ConfigToml>>,
107	/// Read/write split mirrors [`Runtime`]'s own receivers: `&self`
108	/// operations (tool calls, status, MCP startup) share a read guard and
109	/// run concurrently; `&mut self` turns (prompt/thread) and config pushes
110	/// take the write guard because the runtime genuinely requires
111	/// exclusivity there.
112	runtime: Arc<RwLock<Runtime>>,
113	registry: ModelRegistry,
114	auth_token: Option<String>,
115	/// Cached bridge to the real runtime API. Shared by every surface that
116	/// executes a turn — stdio `thread/message`, HTTP `/thread` messages, and
117	/// both `/prompt` transports — because there is exactly one turn engine.
118	runtime_bridge: Arc<Mutex<Option<SharedRuntimeBridge>>>,
119	/// Client-facing thread key → durable runtime thread id.
120	///
121	/// Runtime threads are persisted by the child's on-disk store
122	/// (`RuntimeThreadStore` under the session/task data dir), so a mapping
123	/// stays valid across a bridge restart: the next child resolves the same
124	/// thread ids. Keeping this on `AppState` rather than `RuntimeBridge` is
125	/// the point — `invalidate_runtime_bridge` drops the child but must not
126	/// orphan live stdio threads onto silently minted replacements (#6246).
127	/// Callers already serialize on the bridge mutex, so the map needs no
128	/// ordering guarantees of its own.
129	runtime_thread_map: Arc<Mutex<HashMap<String, String>>>,
130	stdio_thread_hints: Arc<Mutex<HashMap<String, RuntimeThreadHint>>>,
131	/// Turns currently streaming over stdio, keyed by stdio thread id.
132	///
133	/// Deliberately kept outside the bridge mutex: a streaming turn holds
134	/// that mutex for its entire duration, so anything reachable only through
135	/// it cannot be used to stop the turn. This holds its own copy of what an
136	/// interrupt needs, so a cancel never waits on the turn it is cancelling.
137	in_flight_turns: Arc<Mutex<HashMap<String, InFlightTurn>>>,
138	}
139
140	/// Everything needed to interrupt a running turn without the bridge lock.
141	#[derive(Debug, Clone)]
142	struct InFlightTurn {
143	base_url: String,
144	auth_token: Option<String>,
145	/// Thread id as the runtime knows it, not the stdio-facing id.
146	runtime_thread_id: String,
147	turn_id: String,
148	}
149
150	type TurnRegistry = Arc<Mutex<HashMap<String, InFlightTurn>>>;
151
152	#[derive(Debug, Clone, Serialize, Deserialize)]
153	struct ToolCallRequest {
154	call: ToolCall,
155	#[serde(default)]
156	cwd: Option<PathBuf>,
157	}
158
159	#[derive(Debug, Deserialize)]
160	struct JsonRpcRequest {
161	#[serde(default)]
162	jsonrpc: Option<String>,
163	#[serde(default)]
164	id: Option<Value>,
165	method: String,
166	#[serde(default)]
167	params: Value,
168	}
169
170	/// Server error: the app-server could not reach the runtime that executes
171	/// turns. Kept in the JSON-RPC implementation-defined server range
172	/// (-32000..-32099) alongside `thread_not_found` (-32004).
173	const RUNTIME_UNAVAILABLE_CODE: i64 = -32005;
174	/// Server error: the named thread does not exist.
175	const THREAD_NOT_FOUND_CODE: i64 = -32004;
176	/// Server error: a daemon-socket client tried to act before `daemon/attach`.
177	/// Only the unix listener raises it; gated so the Windows build (where the
178	/// listener is a typed-unsupported stub) does not fail `warnings = "deny"`
179	/// on dead code.
180	#[cfg(unix)]
181	const ATTACH_REQUIRED_CODE: i64 = -32010;
182	/// Server error: a `daemon/attach` claim lost to a live owner.
183	#[cfg(unix)]
184	const DAEMON_ALREADY_CLAIMED_CODE: i64 = -32011;
185	/// Server error: only the owning client may `shutdown` the daemon.
186	const NOT_DAEMON_OWNER_CODE: i64 = -32012;
187	/// Server error: the client refused the daemon's version at attach time.
188	#[cfg(unix)]
189	const DAEMON_VERSION_SKEW_CODE: i64 = -32013;
190	/// Server error: `daemon/attach` sent twice on one connection.
191	const ALREADY_ATTACHED_CODE: i64 = -32014;
192
193	#[derive(Debug)]
194	struct JsonRpcError {
195	code: i64,
196	message: String,
197	data: Option<Value>,
198	}
199
200	#[derive(Debug)]
201	struct StdioDispatchResult {
202	result: Value,
203	should_exit: bool,
204	}
205
206	#[derive(Debug)]
207	struct RuntimeBridge {
208	base_url: String,
209	client: reqwest::Client,
210	auth_token: Option<String>,
211	child: Option<Child>,
212	/// Per-child SSE replay cursors, keyed by runtime thread id. Event
213	/// sequence numbering is per child process, so this resets with the
214	/// bridge: a replacement child replays from seq 0 and the turn filter
215	/// in `stream_turn_events` drops everything but the live turn.
216	last_seq_by_thread: HashMap<String, u64>,
217	}
218
219	#[derive(Debug, Clone, Default)]
220	struct RuntimeThreadHint {
221	model: Option<String>,
222	workspace: Option<PathBuf>,
223	}
224
225	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
226	enum TurnTerminalStatus {
227	Completed,
228	Failed,
229	Interrupted,
230	Canceled,
231	}
232
233	/// Structured capture of one bridged turn, for callers that must return
234	/// the turn instead of streaming it (HTTP `/prompt`, HTTP `/thread` messages).
235	///
236	/// The stdio path streams the same events to its writer and needs none of
237	/// this, so it passes `None` and pays nothing.
238	#[derive(Debug, Default)]
239	struct TurnTranscript {
240	/// Concatenated `agent_message` deltas — the model's actual output.
241	text: String,
242	/// The model the runtime reports for the thread that ran the turn.
243	model: Option<String>,
244	/// The same frames the stdio path writes, in order.
245	events: Vec<EventFrame>,
246	}
247
248	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
249	enum AppTransport {
250	Http,
251	Stdio,
252	/// Unix-domain-socket daemon transport (`daemon_socket`). Speaks the
253	/// stdio JSON-RPC protocol verbatim after a `daemon/attach` handshake.
254	Socket,
255	}
256
257	impl AppTransport {
258	/// Wire label reported by `healthz` / `capabilities`.
259	fn label(self) -> &'static str {
260	match self {
261	Self::Http => "http",
262	Self::Stdio => "stdio",
263	Self::Socket => "unix-socket",
264	}
265	}
266	}
267
268	/// Whether the peer driving a JSON-RPC loop may stop the whole server.
269	///
270	/// The process-owned stdio loop always may (its peer is the supervisor).
271	/// On the daemon socket only the client that claimed the daemon may; every
272	/// other attached client is refused with `not_daemon_owner` — the brief's
273	/// "never terminate a daemon the app did not spawn", enforced server-side.
274	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
275	enum ShutdownAuthority {
276	Granted,
277	Denied,
278	}
279
280	/// Per-connection policy for [`run_stdio_loop`].
281	#[derive(Debug, Clone, Copy)]
282	struct StdioLoopPolicy {
283	transport: AppTransport,
284	shutdown: ShutdownAuthority,
285	}
286
287	impl StdioLoopPolicy {
288	/// The loop owned by the process's own stdin/stdout.
289	const fn process_stdio() -> Self {
290	Self {
291	transport: AppTransport::Stdio,
292	shutdown: ShutdownAuthority::Granted,
293	}
294	}
295	}
296
297	/// Why [`run_stdio_loop`] returned.
298	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
299	enum StdioLoopExit {
300	/// The peer closed its write side; nothing asked the server to stop.
301	InputClosed,
302	/// The peer sent an honoured `shutdown`.
303	Shutdown,
304	}
305
306	#[derive(Debug, Deserialize)]
307	struct ConfigGetParams {
308	key: String,
309	}
310
311	#[derive(Debug, Deserialize)]
312	struct ConfigSetParams {
313	key: String,
314	value: String,
315	}
316
317	#[derive(Debug, Deserialize)]
318	struct ThreadIdParams {
319	thread_id: String,
320	}
321
322	#[derive(Debug, Deserialize)]
323	struct ThreadMessageParams {
324	#[serde(default, rename = "maxOutputTokens", alias = "max_output_tokens")]
325	max_output_tokens: Option<std::num::NonZeroU32>,
326	thread_id: String,
327	input: String,
328	#[serde(default)]
329	images: Vec<RuntimeImageInput>,
330	}
331
332	#[derive(Debug, Deserialize)]
333	struct ThreadInterruptParams {
334	thread_id: String,
335	}
336
337	pub async fn run(options: AppServerOptions) -> Result<()> {
338	let auth_token = resolve_auth_token(&options)?;
339	let state = build_state(options.config_path.clone(), auth_token)?;
340	let app = app_router(state, &options.cors_origins);
341
342	let listener = tokio::net::TcpListener::bind(options.listen).await?;
343	axum::serve(listener, app)
344	.with_graceful_shutdown(shutdown_signal())
345	.await?;
346	Ok(())
347	}
348
349	async fn shutdown_signal() {
350	let ctrl_c = async {
351	let _ = tokio::signal::ctrl_c().await;
352	};
353
354	#[cfg(unix)]
355	let terminate = async {
356	match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
357	Ok(mut signal) => {
358	signal.recv().await;
359	}
360	Err(_) => std::future::pending::<()>().await,
361	}
362	};
363
364	#[cfg(not(unix))]
365	let terminate = std::future::pending::<()>();
366
367	tokio::select! {
368	_ = ctrl_c => {}
369	_ = terminate => {}
370	}
371	}
372
373	fn app_router(state: AppState, cors_origins: &[String]) -> Router {
374	let protected_routes = Router::new()
375	.route(
376	"/thread",
377	post(thread_handler).layer(axum::extract::DefaultBodyLimit::max(
378	MAX_RUNTIME_IMAGE_BODY_BYTES,
379	)),
380	)
381	.route("/app", post(app_handler))
382	.route(
383	"/prompt",
384	post(prompt_handler).layer(axum::extract::DefaultBodyLimit::max(
385	MAX_RUNTIME_IMAGE_BODY_BYTES,
386	)),
387	)
388	.route("/tool", post(tool_handler))
389	.route("/jobs", get(jobs_handler))
390	.route("/mcp/startup", post(mcp_startup_handler))
391	.route(
392	"/v1/chat/completions",
393	post(chat_completions::chat_completions_handler),
394	)
395	.route_layer(middleware::from_fn_with_state(
396	state.clone(),
397	require_app_server_token,
398	));
399
400	Router::new()
401	.route("/healthz", get(healthz))
402	.merge(protected_routes)
403	.layer(DefaultBodyLimit::max(MAX_HTTP_BODY_BYTES))
404	.layer(cors_layer(cors_origins))
405	.with_state(state)
406	}
407
408	pub async fn run_stdio(config_path: Option<PathBuf>) -> Result<()> {
409	let state = build_state_with_transport(config_path, None, AppTransport::Stdio)?;
410	let reader = BufReader::new(tokio::io::stdin()).lines();
411	let writer = tokio::io::BufWriter::new(tokio::io::stdout());
412	run_stdio_loop(
413	&state,
414	reader,
415	writer,
416	StdioLoopPolicy::process_stdio(),
417	None::<()>,
418	)
419	.await
420	.map(\|_exit\| ())
421	}
422
423	/// The stdio JSON-RPC loop, generic over its transport so it can be driven by
424	/// a duplex pipe in tests rather than the process's real stdin/stdout.
425	async fn run_stdio_loop<R, W, C>(
426	state: &AppState,
427	mut reader: tokio::io::Lines<R>,
428	mut writer: W,
429	policy: StdioLoopPolicy,
430	// Dropped the moment input closes, not when the in-flight turn ends. The
431	// socket transport passes its owner claim here: a `thread/message` can run
432	// for minutes, and an owner who disconnects mid-turn must not keep the
433	// daemon claimed for the rest of it, or a relaunched client is locked out
434	// with `daemon_already_claimed` and cannot even shut the daemon down.
435	// Process stdio has no claim and passes `None`.
436	mut input_claim: Option<C>,
437	) -> Result<StdioLoopExit>
438	where
439	R: AsyncBufRead + Unpin,
440	W: AsyncWrite + Unpin,
441	C: Send,
442	{
443	// Work that arrived while a turn was streaming. The turn owns the writer
444	// for its whole duration, so these wait for it rather than interleaving
445	// into the middle of a response.
446	let mut pending: VecDeque<PendingStdioWork> = VecDeque::new();
447	let mut stdin_open = true;
448
449	loop {
450	let request = match pending.pop_front() {
451	Some(PendingStdioWork::Response(response)) => {
452	write_stdio_line(&mut writer, &response).await?;
453	continue;
454	}
455	Some(PendingStdioWork::Request(request)) => request,
456	None => {
457	if !stdin_open {
458	return Ok(StdioLoopExit::InputClosed);
459	}
460	let Some(line) = reader.next_line().await? else {
461	return Ok(StdioLoopExit::InputClosed);
462	};
463	match parse_stdio_line(&line) {
464	ParsedStdioLine::Blank => continue,
465	ParsedStdioLine::Rejected(response) => {
466	write_stdio_line(&mut writer, &response).await?;
467	continue;
468	}
469	ParsedStdioLine::Request(request) => request,
470	}
471	}
472	};
473
474	let id = request.id.clone();
475	if request.method == "shutdown" && policy.shutdown == ShutdownAuthority::Denied {
476	write_stdio_line(
477	&mut writer,
478	&jsonrpc_error(id, JsonRpcError::not_daemon_owner()),
479	)
480	.await?;
481	continue;
482	}
483	let dispatched = if request.method == "thread/message" {
484	// A turn can run for minutes. Keep reading stdin while it streams
485	// so an interrupt (or a shutdown) can actually reach it — with a
486	// plain `await` here, nothing could be read until it finished.
487	let dispatch = dispatch_stdio_request_with_writer(
488	state,
489	&mut writer,
490	&request.method,
491	request.params,
492	policy.transport,
493	);
494	tokio::pin!(dispatch);
495	loop {
496	tokio::select! {
497	outcome = &mut dispatch => break outcome,
498	line = reader.next_line(), if stdin_open => {
499	match line? {
500	None => {
501	stdin_open = false;
502	// Release the claim here, not after `dispatch`
503	// resolves.
504	drop(input_claim.take());
505	}
506	Some(line) => {
507	handle_line_during_turn(state, &line, &mut pending, policy).await;
508	}
509	}
510	}
511	}
512	}
513	} else {
514	dispatch_stdio_request_with_writer(
515	state,
516	&mut writer,
517	&request.method,
518	request.params,
519	policy.transport,
520	)
521	.await
522	};
523
524	match dispatched {
525	Ok(dispatch) => {
526	write_stdio_line(&mut writer, &jsonrpc_result(id, dispatch.result)).await?;
527	if dispatch.should_exit {
528	return Ok(StdioLoopExit::Shutdown);
529	}
530	}
531	Err(err) => {
532	write_stdio_line(&mut writer, &jsonrpc_error(id, err)).await?;
533	}
534	}
535	}
536	}
537
538	/// Work deferred until a streaming turn releases the writer.
539	enum PendingStdioWork {
540	/// Already answered (an interrupt acted immediately); just needs writing.
541	Response(Value),
542	/// Not started yet; runs normally once the turn is done.
543	Request(JsonRpcRequest),
544	}
545
546	enum ParsedStdioLine {
547	Blank,
548	Request(JsonRpcRequest),
549	Rejected(Value),
550	}
551
552	fn parse_stdio_line(line: &str) -> ParsedStdioLine {
553	if line.len() > MAX_RUNTIME_IMAGE_BODY_BYTES {
554	return ParsedStdioLine::Rejected(jsonrpc_error(
555	None,
556	JsonRpcError::invalid_params("request exceeds the 8 MiB transport limit"),
557	));
558	}
559	if line.trim().is_empty() {
560	return ParsedStdioLine::Blank;
561	}
562	let request: JsonRpcRequest = match serde_json::from_str(line) {
563	Ok(value) => value,
564	Err(err) => {
565	return ParsedStdioLine::Rejected(jsonrpc_error(
566	None,
567	JsonRpcError::parse_error(format!("invalid json: {err}")),
568	));
569	}
570	};
571	if request
572	.jsonrpc
573	.as_deref()
574	.is_some_and(\|version\| version != "2.0")
575	{
576	return ParsedStdioLine::Rejected(jsonrpc_error(
577	request.id,
578	JsonRpcError::invalid_request("jsonrpc version must be 2.0"),
579	));
580	}
581	ParsedStdioLine::Request(request)
582	}
583
584	/// Triage a request that arrived mid-turn.
585	///
586	/// Cancellation is the whole point of reading here, so `thread/interrupt`
587	/// runs immediately and only its reply waits for the writer. `shutdown` also
588	/// interrupts immediately — otherwise it would block on the bridge mutex the
589	/// turn is holding — and then queues so the turn can unwind first. Everything
590	/// else simply queues: it was never urgent, and running it now would race the
591	/// turn for the writer.
592	async fn handle_line_during_turn(
593	state: &AppState,
594	line: &str,
595	pending: &mut VecDeque<PendingStdioWork>,
596	policy: StdioLoopPolicy,
597	) {
598	let request = match parse_stdio_line(line) {
599	ParsedStdioLine::Blank => return,
600	ParsedStdioLine::Rejected(response) => {
601	pending.push_back(PendingStdioWork::Response(response));
602	return;
603	}
604	ParsedStdioLine::Request(request) => request,
605	};
606
607	match request.method.as_str() {
608	"thread/interrupt" => {
609	let id = request.id.clone();
610	let response = match parse_params::<ThreadInterruptParams>(params_or_object(
611	request.params.clone(),
612	)) {
613	Ok(parsed) => match interrupt_stdio_turn(state, &parsed.thread_id).await {
614	Ok(interrupted) => jsonrpc_result(
615	id,
616	json!({ "thread_id": parsed.thread_id, "interrupted": interrupted }),
617	),
618	Err(err) => jsonrpc_error(id, err),
619	},
620	Err(err) => jsonrpc_error(id, err),
621	};
622	pending.push_back(PendingStdioWork::Response(response));
623	}
624	"shutdown" if policy.shutdown == ShutdownAuthority::Denied => {
625	// A non-owner may not even interrupt the live turns: that is the
626	// first half of what shutdown does.
627	pending.push_back(PendingStdioWork::Response(jsonrpc_error(
628	request.id,
629	JsonRpcError::not_daemon_owner(),
630	)));
631	}
632	"shutdown" => {
633	let _ = interrupt_all_stdio_turns(state).await;
634	pending.push_back(PendingStdioWork::Request(request));
635	}
636	_ => pending.push_back(PendingStdioWork::Request(request)),
637	}
638	}
639
640	async fn write_stdio_line<W: AsyncWrite + Unpin>(writer: &mut W, response: &Value) -> Result<()> {
641	writer.write_all(&serde_json::to_vec(response)?).await?;
642	writer.write_all(b"\n").await?;
643	writer.flush().await?;
644	Ok(())
645	}
646
647	async fn healthz() -> Json<Value> {
648	Json(json!({
649	"status": "ok",
650	"protocol": "v2",
651	"service": legacy_deepseek_compat::SERVICE_NAME
652	}))
653	}
654
655	/// Render a routing failure as a typed HTTP error body.
656	///
657	/// Deliberately not a success-shaped payload with the error stuffed into a
658	/// content field: a client must be able to tell "the model said this" from
659	/// "nothing ran".
660	fn http_error_from_jsonrpc(err: JsonRpcError) -> (StatusCode, Json<Value>) {
661	let (status, code) = match err.code {
662	-32600 \| -32602 => (StatusCode::BAD_REQUEST, "invalid_request"),
663	THREAD_NOT_FOUND_CODE => (StatusCode::NOT_FOUND, "thread_not_found"),
664	RUNTIME_UNAVAILABLE_CODE => (StatusCode::SERVICE_UNAVAILABLE, "runtime_unavailable"),
665	_ => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"),
666	};
667	(
668	status,
669	Json(json!({
670	"error": {
671	"code": code,
672	"jsonrpc_code": err.code,
673	"message": err.message,
674	}
675	})),
676	)
677	}
678
679	async fn thread_handler(State(state): State<AppState>, Json(req): Json<ThreadRequest>) -> Response {
680	// A message is a turn, and turns belong to the runtime — not to the
681	// bookkeeping `Runtime` behind the other thread operations. This mirrors
682	// the interception stdio `thread/message` has always done.
683	if let ThreadRequest::Message {
684	thread_id,
685	input,
686	images,
687	max_output_tokens,
688	} = req
689	{
690	return match run_http_thread_message(&state, thread_id, input, images, max_output_tokens)
691	.await
692	{
693	Ok(res) => (StatusCode::OK, Json(res)).into_response(),
694	Err(err) => http_error_from_jsonrpc(err).into_response(),
695	};
696	}
697	let mut runtime = state.runtime.write().await;
698	match runtime.handle_thread(req).await {
699	Ok(res) => (StatusCode::OK, Json(res)).into_response(),
700	Err(err) => (
701	StatusCode::INTERNAL_SERVER_ERROR,
702	Json(ThreadResponse {
703	thread_id: "error".to_string(),
704	status: format!("error:{err}"),
705	thread: None,
706	threads: Vec::new(),
707	goal: None,
708	model: None,
709	model_provider: None,
710	cwd: None,
711	approval_policy: None,
712	sandbox: None,
713	events: Vec::new(),
714	data: json!({}),
715	}),
716	)
717	.into_response(),
718	}
719	}
720
721	/// `POST /prompt` — runs a genuine model turn through the runtime bridge.
722	///
723	/// Note what this handler does not do: it never takes the `Runtime` write
724	/// lock. The old implementation held it across the whole request while doing
725	/// no model work at all.
726	async fn prompt_handler(State(state): State<AppState>, Json(req): Json<PromptRequest>) -> Response {
727	let mut sink = tokio::io::sink();
728	match run_prompt_turn(&state, &mut sink, req).await {
729	Ok(res) => (StatusCode::OK, Json(res)).into_response(),
730	Err(err) => http_error_from_jsonrpc(err).into_response(),
731	}
732	}
733
734	async fn tool_handler(
735	State(state): State<AppState>,
736	Json(req): Json<ToolCallRequest>,
737	) -> (StatusCode, Json<Value>) {
738	let cwd = req
739	.cwd
740	.unwrap_or_else(\|\| std::env::current_dir().unwrap_or_else(\|_\| PathBuf::from(".")));
741	// Resolve approval policy from config instead of hardcoding.
742	let approval_mode = {
743	let cfg = state.config.read().await;
744	cfg.approval_policy
745	.as_deref()
746	.and_then(\|p\| match p.trim().to_ascii_lowercase().as_str() {
747	"auto" \| "yolo" => Some(codewhale_execpolicy::AskForApproval::UnlessTrusted),
748	"never" \| "deny" => Some(codewhale_execpolicy::AskForApproval::Never),
749	_ => None,
750	})
751	.unwrap_or(codewhale_execpolicy::AskForApproval::OnRequest)
752	};
753	// `invoke_tool` takes `&self`, so long-running tool executions share a
754	// read guard: they run concurrently with each other and with status
755	// reads instead of serializing every request behind one Mutex.
756	let runtime = state.runtime.read().await;
757	match runtime.invoke_tool(req.call, approval_mode, &cwd).await {
758	Ok(value) => (StatusCode::OK, Json(value)),
759	Err(err) => (
760	StatusCode::INTERNAL_SERVER_ERROR,
761	Json(json!({ "ok": false, "error": err.to_string() })),
762	),
763	}
764	}
765
766	async fn jobs_handler(State(state): State<AppState>) -> Json<AppResponse> {
767	let runtime = state.runtime.read().await;
768	Json(runtime.app_status())
769	}
770
771	async fn mcp_startup_handler(State(state): State<AppState>) -> Json<Value> {
772	let runtime = state.runtime.read().await;
773	let summary = runtime.mcp_startup().await;
774	Json(json!({
775	"ok": true,
776	"summary": summary
777	}))
778	}
779
780	async fn app_handler(
781	State(state): State<AppState>,
782	Json(req): Json<AppRequest>,
783	) -> (StatusCode, Json<AppResponse>) {
784	let response = process_app_request(&state, req, AppTransport::Http).await;
785	(app_response_status(&response), Json(response))
786	}
787
788	fn app_response_status(response: &AppResponse) -> StatusCode {
789	if response.ok {
790	return StatusCode::OK;
791	}
792	if response.data.get("request_id").is_some() {
793	StatusCode::CONFLICT
794	} else if response
795	.data
796	.get("error")
797	.and_then(Value::as_str)
798	.is_some_and(\|err\| err.contains("failed to load config"))
799	{
800	StatusCode::INTERNAL_SERVER_ERROR
801	} else {
802	StatusCode::BAD_REQUEST
803	}
804	}
805
806	fn build_state(config_path: Option<PathBuf>, auth_token: Option<String>) -> Result<AppState> {
807	build_state_with_transport(config_path, auth_token, AppTransport::Http)
808	}
809
810	fn build_state_with_transport(
811	config_path: Option<PathBuf>,
812	auth_token: Option<String>,
813	transport: AppTransport,
814	) -> Result<AppState> {
815	let has_explicit_config_path = config_path.is_some();
816	let store = ConfigStore::load(config_path)?;
817	let config_path = has_explicit_config_path.then(\|\| store.path().to_path_buf());
818	let config = store.config.clone();
819	let exec_policy = store.exec_policy_engine();
820	let registry = ModelRegistry::default();
821
822	let state_db_path = config_path
823	.as_ref()
824	.and_then(\|p\| p.parent().map(\|parent\| parent.join("state.db")));
825	let state_store = StateStore::open(state_db_path)?;
826
827	let mut hooks = HookDispatcher::default();
828	// Stdio carries JSON-RPC on stdout: printing raw hook events there
829	// corrupts the protocol stream (#5165). HTTP mode keeps the stdout
830	// sink for local development visibility.
831	if transport == AppTransport::Http {
832	hooks.add_sink(Arc::new(StdoutHookSink));
833	}
834	let hook_log_path = config_path
835	.as_ref()
836	.and_then(\|p\| p.parent().map(\|parent\| parent.join("events.jsonl")))
837	.unwrap_or_else(legacy_deepseek_compat::default_events_log_path);
838	hooks.add_sink(Arc::new(JsonlHookSink::new(hook_log_path)));
839
840	if let Some(socket_path) = config
841	.hook_sinks
842	.as_ref()
843	.and_then(\|sinks\| sinks.unix_socket_path.as_ref())
844	.filter(\|path\| !path.as_os_str().is_empty())
845	{
846	hooks.add_sink(Arc::new(UnixSocketHookSink::new(socket_path.clone())));
847	}
848
849	let runtime = Runtime::new(
850	config.clone(),
851	state_store,
852	Arc::new(ToolRegistry::default()),
853	Arc::new(McpManager::default()),
854	exec_policy,
855	hooks,
856	);
857
858	Ok(AppState {
859	config_path,
860	config: Arc::new(RwLock::new(config)),
861	runtime: Arc::new(RwLock::new(runtime)),
862	registry,
863	auth_token,
864	runtime_bridge: Arc::new(Mutex::new(None)),
865	runtime_thread_map: Arc::new(Mutex::new(HashMap::new())),
866	stdio_thread_hints: Arc::new(Mutex::new(HashMap::new())),
867	in_flight_turns: Arc::new(Mutex::new(HashMap::new())),
868	})
869	}
870
871	fn resolve_auth_token(options: &AppServerOptions) -> Result<Option<String>> {
872	let configured = options.auth_token.as_ref().map(\|token\| token.trim());
873	if let Some(token) = configured
874	&& token.is_empty()
875	{
876	bail!("app-server auth token cannot be empty");
877	}
878	let has_explicit_token = configured.is_some();
879
880	if options.insecure_no_auth {
881	if !options.listen.ip().is_loopback() {
882	bail!("refusing unauthenticated app-server bind on non-loopback address");
883	}
884	eprintln!("warning: app-server HTTP auth disabled by --insecure-no-auth");
885	return Ok(None);
886	}
887
888	if !has_explicit_token && !options.listen.ip().is_loopback() {
889	bail!(
890	"refusing non-loopback app-server bind without explicit auth token; pass --auth-token or set CODEWHALE_APP_SERVER_TOKEN"
891	);
892	}
893
894	let token = configured
895	.map(str::to_string)
896	.unwrap_or_else(\|\| format!("cwapp_{}", Uuid::new_v4().simple()));
897	for line in app_server_auth_status_lines(has_explicit_token) {
898	eprintln!("{line}");
899	}
900	Ok(Some(token))
901	}
902
903	fn app_server_auth_status_lines(has_explicit_token: bool) -> Vec<&'static str> {
904	if has_explicit_token {
905	return vec!["app-server auth: bearer token required for HTTP routes."];
906	}
907	vec![
908	"app-server auth: generated bearer token for this process (not printed).",
909	" Pass --auth-token or set CODEWHALE_APP_SERVER_TOKEN when another client needs to connect.",
910	]
911	}
912
913	fn cors_layer(extra_origins: &[String]) -> CorsLayer {
914	let mut origins: Vec<HeaderValue> = DEFAULT_CORS_ORIGINS
915	.iter()
916	.filter_map(\|origin\| HeaderValue::from_str(origin).ok())
917	.collect();
918	for raw in extra_origins {
919	let trimmed = raw.trim();
920	if trimmed.is_empty() {
921	continue;
922	}
923	match HeaderValue::from_str(trimmed) {
924	Ok(value) if !origins.contains(&value) => origins.push(value),
925	Ok(_) => {}
926	Err(err) => {
927	eprintln!("warning: ignoring invalid app-server CORS origin `{trimmed}`: {err}")
928	}
929	}
930	}
931
932	CorsLayer::new()
933	.allow_origin(origins)
934	.allow_methods([Method::GET, Method::POST, Method::OPTIONS])
935	.allow_headers([header::AUTHORIZATION, header::CONTENT_TYPE])
936	}
937
938	async fn require_app_server_token(
939	State(state): State<AppState>,
940	req: Request,
941	next: Next,
942	) -> Response {
943	let Some(expected) = state.auth_token.as_deref() else {
944	return next.run(req).await;
945	};
946	let authorized = req
947	.headers()
948	.get(header::AUTHORIZATION)
949	.and_then(\|value\| value.to_str().ok())
950	.and_then(\|raw\| raw.strip_prefix("Bearer "))
951	.is_some_and(\|token\| constant_time_eq(token.as_bytes(), expected.as_bytes()));
952
953	if authorized {
954	next.run(req).await
955	} else {
956	(
957	StatusCode::UNAUTHORIZED,
958	Json(json!({
959	"error": {
960	"message": "app-server bearer token required",
961	"status": StatusCode::UNAUTHORIZED.as_u16(),
962	}
963	})),
964	)
965	.into_response()
966	}
967	}
968
969	/// Compares the full length of both inputs regardless of where they first
970	/// differ, so auth failures don't leak the matching prefix length via timing.
971	fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
972	let mut diff = a.len() ^ b.len();
973	for i in 0..a.len().max(b.len()) {
974	let x = a.get(i).copied().unwrap_or(0);
975	let y = b.get(i).copied().unwrap_or(0);
976	diff \|= usize::from(x ^ y);
977	}
978	diff == 0
979	}
980
981	fn params_or_object(params: Value) -> Value {
982	if params.is_null() { json!({}) } else { params }
983	}
984
985	fn parse_params<T: DeserializeOwned>(params: Value) -> std::result::Result<T, JsonRpcError> {
986	serde_json::from_value(params).map_err(\|err\| JsonRpcError::invalid_params(err.to_string()))
987	}
988
989	fn jsonrpc_result(id: Option<Value>, result: Value) -> Value {
990	json!({
991	"jsonrpc": "2.0",
992	"id": id.unwrap_or(Value::Null),
993	"result": result
994	})
995	}
996
997	fn jsonrpc_error(id: Option<Value>, err: JsonRpcError) -> Value {
998	json!({
999	"jsonrpc": "2.0",
1000	"id": id.unwrap_or(Value::Null),
1001	"error": {
1002	"code": err.code,
1003	"message": err.message,
1004	"data": err.data
1005	}
1006	})
1007	}
1008
1009	impl JsonRpcError {
1010	fn parse_error(message: impl Into<String>) -> Self {
1011	Self {
1012	code: -32700,
1013	message: message.into(),
1014	data: None,
1015	}
1016	}
1017
1018	fn invalid_request(message: impl Into<String>) -> Self {
1019	Self {
1020	code: -32600,
1021	message: message.into(),
1022	data: None,
1023	}
1024	}
1025
1026	fn method_not_found(method: &str) -> Self {
1027	Self {
1028	code: -32601,
1029	message: format!("unsupported method: {method}"),
1030	data: None,
1031	}
1032	}
1033
1034	fn invalid_params(message: impl Into<String>) -> Self {
1035	Self {
1036	code: -32602,
1037	message: message.into(),
1038	data: None,
1039	}
1040	}
1041
1042	/// Server error (-32000..-32099): the turn engine could not be reached,
1043	/// or refused to start the turn — either way nothing ran. Distinct from
1044	/// `internal` because the caller can retry this one once a runtime is up.
1045	fn runtime_unavailable(message: impl Into<String>) -> Self {
1046	let message = message.into();
1047	Self {
1048	code: RUNTIME_UNAVAILABLE_CODE,
1049	message: message.clone(),
1050	data: Some(json!({
1051	"error": "runtime_unavailable",
1052	"detail": message,
1053	})),
1054	}
1055	}
1056
1057	/// Server error (-32000..-32099): the named thread does not exist.
1058	fn thread_not_found(thread_id: &str) -> Self {
1059	Self {
1060	code: THREAD_NOT_FOUND_CODE,
1061	message: format!("thread not found: {thread_id}"),
1062	data: Some(json!({
1063	"error": "thread_not_found",
1064	"thread_id": thread_id,
1065	})),
1066	}
1067	}
1068
1069	fn internal(message: impl Into<String>) -> Self {
1070	Self {
1071	code: -32603,
1072	message: message.into(),
1073	data: None,
1074	}
1075	}
1076
1077	/// Server error (-32000..-32099): the daemon-socket connection has not
1078	/// completed `daemon/attach`, so nothing but `healthz` is allowed yet.
1079	#[cfg(unix)]
1080	fn attach_required(method: &str) -> Self {
1081	Self {
1082	code: ATTACH_REQUIRED_CODE,
1083	message: format!("send daemon/attach before `{method}`"),
1084	data: Some(json!({
1085	"error": "attach_required",
1086	"method": method,
1087	"attach_method": daemon_socket::ATTACH_METHOD,
1088	})),
1089	}
1090	}
1091
1092	/// Server error (-32000..-32099): a `claim` attach lost to a live owner.
1093	#[cfg(unix)]
1094	fn daemon_already_claimed(owner: &Value) -> Self {
1095	Self {
1096	code: DAEMON_ALREADY_CLAIMED_CODE,
1097	message: "daemon already claimed by another client; attach with mode=attach"
1098	.to_string(),
1099	data: Some(json!({
1100	"error": "daemon_already_claimed",
1101	"owner": owner,
1102	})),
1103	}
1104	}
1105
1106	/// Server error (-32000..-32099): only the owner may stop the daemon.
1107	fn not_daemon_owner() -> Self {
1108	Self {
1109	code: NOT_DAEMON_OWNER_CODE,
1110	message: "only the client that claimed this daemon may shut it down".to_string(),
1111	data: Some(json!({ "error": "not_daemon_owner" })),
1112	}
1113	}
1114
1115	/// Server error (-32000..-32099): the client's expected daemon version
1116	/// does not match the running binary (bundle skew).
1117	#[cfg(unix)]
1118	fn daemon_version_skew(expected: &str, actual: &str) -> Self {
1119	Self {
1120	code: DAEMON_VERSION_SKEW_CODE,
1121	message: format!(
1122	"daemon version {actual} does not match the client's expected {expected}"
1123	),
1124	data: Some(json!({
1125	"error": "daemon_version_skew",
1126	"expected": expected,
1127	"actual": actual,
1128	})),
1129	}
1130	}
1131
1132	/// Server error (-32000..-32099): `daemon/attach` after attaching.
1133	fn already_attached() -> Self {
1134	Self {
1135	code: ALREADY_ATTACHED_CODE,
1136	message: "this connection is already attached".to_string(),
1137	data: Some(json!({ "error": "already_attached" })),
1138	}
1139	}
1140	}
1141
1142	async fn handle_thread_request(
1143	state: &AppState,
1144	req: ThreadRequest,
1145	) -> std::result::Result<ThreadResponse, JsonRpcError> {
1146	let mut runtime = state.runtime.write().await;
1147	runtime
1148	.handle_thread(req)
1149	.await
1150	.map_err(\|err\| JsonRpcError::internal(err.to_string()))
1151	}
1152
1153	/// One turn's worth of routing decisions, shared by every surface that runs
1154	/// a turn through the bridge.
1155	struct BridgedTurn<'a> {
1156	max_output_tokens: Option<std::num::NonZeroU32>,
1157	/// Client-facing thread id; the bridge maps it to a runtime thread.
1158	thread_key: &'a str,
1159	input: &'a str,
1160	images: &'a [RuntimeImageInput],
1161	/// Model for the runtime thread when this call is the one that creates
1162	/// it. An existing thread keeps the model it was created with.
1163	model_override: Option<String>,
1164	/// Publish the live turn so a concurrent `thread/interrupt` can cancel
1165	/// it. Only stdio has a mid-turn channel, so only stdio sets this.
1166	interruptible: bool,
1167	/// Forget the thread mapping once the turn ends. Set for one-shot
1168	/// prompts, whose synthetic thread key no client can name again.
1169	ephemeral: bool,
1170	}
1171
1172	/// Execute exactly one turn on the real runtime.
1173	///
1174	/// This is the only way any app-server surface runs a model: `/prompt`,
1175	/// `prompt/request`, `prompt/run`, stdio `thread/message`, and HTTP `/thread`
1176	/// messages all land here. There is no local fallback that fabricates a
1177	/// response — if the runtime cannot be reached the caller gets
1178	/// [`JsonRpcError::runtime_unavailable`] and nothing is written to history.
1179	async fn run_bridged_turn<W: AsyncWrite + Unpin>(
1180	state: &AppState,
1181	writer: &mut W,
1182	turn: BridgedTurn<'_>,
1183	transcript: Option<&mut TurnTranscript>,
1184	) -> std::result::Result<Value, JsonRpcError> {
1185	let mut hint = {
1186	let hints = state.stdio_thread_hints.lock().await;
1187	hints.get(turn.thread_key).cloned()
1188	};
1189	if let Some(model) = turn.model_override {
1190	hint.get_or_insert_with(RuntimeThreadHint::default).model = Some(model);
1191	}
1192	let bridge = acquire_runtime_bridge(state).await?;
1193	// The inner bridge lock is held for the whole turn: one child process
1194	// serves all threads and per-thread seq tracking requires ordered
1195	// access. The cache slot itself stays unlocked, so config updates and
1196	// bridge invalidation are never queued behind a streaming turn.
1197	let mut bridge = bridge.lock().await;
1198	let mut thread_map = state.runtime_thread_map.lock().await;
1199	if turn.max_output_tokens.is_some() {
1200	let info = bridge
1201	.request_json(
1202	bridge.authed(
1203	bridge
1204	.client
1205	.get(format!("{}/v1/runtime/info", bridge.base_url)),
1206	),
1207	)
1208	.await
1209	.map_err(\|err\| JsonRpcError::runtime_unavailable(err.to_string()))?;
1210	if info
1211	.pointer("/capabilities/turn_output_token_limit")
1212	.and_then(Value::as_bool)
1213	!= Some(true)
1214	{
1215	return Err(JsonRpcError::invalid_params(
1216	"Runtime does not support maxOutputTokens",
1217	));
1218	}
1219	if !thread_map.contains_key(turn.thread_key) {
1220	bridge
1221	.require_output_limited_model(hint.as_ref().and_then(\|hint\| hint.model.as_deref()))
1222	.await
1223	.map_err(\|err\| JsonRpcError::invalid_params(err.to_string()))?;
1224	}
1225	}
1226	let runtime_thread_id = bridge
1227	.ensure_runtime_thread(&mut thread_map, turn.thread_key, hint)
1228	.await
1229	.map_err(\|err\| JsonRpcError::runtime_unavailable(err.to_string()))?;
1230	// The mapping is settled for this turn; drop the guard so a long stream
1231	// never holds the map hostage. `forget_thread` re-locks below.
1232	drop(thread_map);
1233	let registration = turn
1234	.interruptible
1235	.then(\|\| (state.in_flight_turns.clone(), turn.thread_key.to_string()));
1236	let result = bridge
1237	.message_thread(
1238	&runtime_thread_id,
1239	turn.input,
1240	turn.images,
1241	turn.max_output_tokens,
1242	writer,
1243	registration,
1244	transcript,
1245	)
1246	.await;
1247	if turn.ephemeral {
1248	// Drop the mapping while we still hold the bridge lock, so a
1249	// long-lived app-server does not accumulate one entry per one-shot
1250	// prompt.
1251	let mut thread_map = state.runtime_thread_map.lock().await;
1252	bridge.forget_thread(&mut thread_map, turn.thread_key);
1253	}
1254	result.map_err(\|err\| JsonRpcError::internal(err.to_string()))
1255	}
1256
1257	/// Run a prompt as a genuine model turn and return what the model actually
1258	/// said.
1259	///
1260	/// `writer` receives the same streaming frames stdio `thread/message` emits;
1261	/// HTTP callers pass a sink and read the frames back out of
1262	/// [`PromptResponse::events`].
1263	async fn run_prompt_turn<W: AsyncWrite + Unpin>(
1264	state: &AppState,
1265	writer: &mut W,
1266	req: PromptRequest,
1267	) -> std::result::Result<PromptResponse, JsonRpcError> {
1268	if req.prompt.trim().is_empty() {
1269	return Err(JsonRpcError::invalid_params("prompt must not be empty"));
1270	}
1271	// The turn engine has no threadless mode, so a prompt without a thread
1272	// gets a fresh one. Keying it on a uuid keeps a one-shot prompt out of
1273	// any caller's history and out of the way of concurrent prompts.
1274	let ephemeral = req.thread_id.is_none();
1275	let thread_key = req
1276	.thread_id
1277	.clone()
1278	.unwrap_or_else(\|\| format!("prompt-{}", Uuid::new_v4()));
1279
1280	let mut transcript = TurnTranscript::default();
1281	run_bridged_turn(
1282	state,
1283	writer,
1284	BridgedTurn {
1285	max_output_tokens: req.max_output_tokens,
1286	thread_key: &thread_key,
1287	input: &req.prompt,
1288	images: &req.images,
1289	model_override: req.model.clone(),
1290	// `thread/interrupt` addresses client-facing thread ids. A
1291	// one-shot prompt has none to hand back, and a caller-supplied
1292	// thread id is already interruptible through `thread/message`.
1293	interruptible: false,
1294	ephemeral,
1295	},
1296	Some(&mut transcript),
1297	)
1298	.await?;
1299
1300	// Report the model the runtime actually ran, never a locally resolved
1301	// guess. The fallbacks only matter for a runtime that omits the field.
1302	let model = match transcript.model {
1303	Some(model) => model,
1304	None => match req.model {
1305	Some(model) => model,
1306	None => state
1307	.config
1308	.read()
1309	.await
1310	.model
1311	.clone()
1312	.unwrap_or_else(\|\| "unknown".to_string()),
1313	},
1314	};
1315
1316	Ok(PromptResponse {
1317	output: transcript.text,
1318	model,
1319	events: transcript.events,
1320	})
1321	}
1322
1323	async fn handle_prompt_request<W: AsyncWrite + Unpin>(
1324	state: &AppState,
1325	writer: &mut W,
1326	req: PromptRequest,
1327	) -> std::result::Result<PromptResponse, JsonRpcError> {
1328	run_prompt_turn(state, writer, req).await
1329	}
1330
1331	/// HTTP `/thread` with a `Message` body: same engine as stdio
1332	/// `thread/message`, but the turn is collected rather than streamed because
1333	/// this transport is request/response.
1334	async fn run_http_thread_message(
1335	state: &AppState,
1336	thread_id: String,
1337	input: String,
1338	images: Vec<RuntimeImageInput>,
1339	max_output_tokens: Option<std::num::NonZeroU32>,
1340	) -> std::result::Result<ThreadResponse, JsonRpcError> {
1341	let mut transcript = TurnTranscript::default();
1342	let mut sink = tokio::io::sink();
1343	let result = run_bridged_turn(
1344	state,
1345	&mut sink,
1346	BridgedTurn {
1347	max_output_tokens,
1348	thread_key: &thread_id,
1349	input: &input,
1350	images: &images,
1351	model_override: None,
1352	interruptible: false,
1353	ephemeral: false,
1354	},
1355	Some(&mut transcript),
1356	)
1357	.await?;
1358
1359	Ok(ThreadResponse {
1360	thread_id,
1361	// The turn ran to a terminal state before this response was built,
1362	// which is exactly what the old `accepted` did not mean.
1363	status: "completed".to_string(),
1364	thread: None,
1365	threads: Vec::new(),
1366	goal: None,
1367	model: transcript.model,
1368	model_provider: None,
1369	cwd: None,
1370	approval_policy: None,
1371	sandbox: None,
1372	events: transcript.events,
1373	data: result.get("data").cloned().unwrap_or_else(\|\| json!({})),
1374	})
1375	}
1376
1377	async fn handle_stdio_thread_message<W: AsyncWrite + Unpin>(
1378	state: &AppState,
1379	writer: &mut W,
1380	parsed: ThreadMessageParams,
1381	) -> std::result::Result<Value, JsonRpcError> {
1382	let mut result = run_bridged_turn(
1383	state,
1384	writer,
1385	BridgedTurn {
1386	max_output_tokens: parsed.max_output_tokens,
1387	thread_key: &parsed.thread_id,
1388	input: &parsed.input,
1389	images: &parsed.images,
1390	model_override: None,
1391	interruptible: true,
1392	ephemeral: false,
1393	},
1394	None,
1395	)
1396	.await?;
1397	if let Some(object) = result.as_object_mut() {
1398	object.insert("thread_id".to_string(), Value::String(parsed.thread_id));
1399	}
1400	Ok(result)
1401	}
1402
1403	/// Resuming or forking a thread the runtime reports as `missing` must fail
1404	/// with a named not-found error. Recording the null model/workspace of that
1405	/// response as a stdio hint would clobber any previously cached hint for
1406	/// the same thread id (#5171).
1407	fn ensure_thread_found(response: &ThreadResponse) -> std::result::Result<(), JsonRpcError> {
1408	if response.status == "missing" {
1409	return Err(JsonRpcError::thread_not_found(&response.thread_id));
1410	}
1411	Ok(())
1412	}
1413
1414	async fn record_stdio_thread_hint(state: &AppState, response: &ThreadResponse) {
1415	let mut hints = state.stdio_thread_hints.lock().await;
1416	hints.insert(
1417	response.thread_id.clone(),
1418	RuntimeThreadHint {
1419	model: response.model.clone(),
1420	workspace: response.cwd.clone(),
1421	},
1422	);
1423	}
1424
1425	/// Fetch the cached stdio→runtime bridge, spawning one on first use.
1426	///
1427	/// The cache-slot lock is held only for the lookup/insert — never across
1428	/// the child spawn or any request traffic — so [`invalidate_runtime_bridge`]
1429	/// and other slot users are never blocked behind a slow bridge operation.
1430	async fn acquire_runtime_bridge(
1431	state: &AppState,
1432	) -> std::result::Result<SharedRuntimeBridge, JsonRpcError> {
1433	if let Some(bridge) = state.runtime_bridge.lock().await.as_ref() {
1434	return Ok(bridge.clone());
1435	}
1436	let bridge = Arc::new(Mutex::new(
1437	RuntimeBridge::start(state.config_path.as_deref())
1438	.await
1439	.map_err(\|err\| JsonRpcError::runtime_unavailable(err.to_string()))?,
1440	));
1441	let mut slot = state.runtime_bridge.lock().await;
1442	// Prefer a bridge cached by a concurrent caller while we were spawning;
1443	// dropping our unused one kills the extra child via `Drop`.
1444	Ok(slot.get_or_insert_with(\|\| bridge.clone()).clone())
1445	}
1446
1447	/// Ask the runtime to interrupt a turn that is streaming right now.
1448	///
1449	/// Everything this needs was copied out of the bridge when the turn started,
1450	/// so it never touches the bridge mutex the turn is holding. Returns whether
1451	/// a live turn was found for `thread_id`.
1452	/// Interrupt one in-flight turn over HTTP, from an owned snapshot.
1453	///
1454	/// Split from [`interrupt_stdio_turn`] so teardown paths can run many
1455	/// concurrently (#6211 R8b) — each future owns its snapshot and never holds
1456	/// the turn registry across the request.
1457	async fn interrupt_turn_request(turn: &InFlightTurn) -> std::result::Result<bool, JsonRpcError> {
1458	let mut request = codewhale_release::platform_http_client_builder()
1459	.timeout(Duration::from_secs(10))
1460	.build()
1461	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?
1462	.post(format!(
1463	"{}/v1/threads/{}/turns/{}/interrupt",
1464	turn.base_url, turn.runtime_thread_id, turn.turn_id
1465	));
1466	if let Some(token) = turn.auth_token.as_deref() {
1467	request = request.bearer_auth(token);
1468	}
1469	request
1470	.send()
1471	.await
1472	.and_then(reqwest::Response::error_for_status)
1473	.map_err(\|err\| JsonRpcError::internal(format!("interrupt failed: {err}")))?;
1474	Ok(true)
1475	}
1476
1477	async fn interrupt_stdio_turn(
1478	state: &AppState,
1479	thread_id: &str,
1480	) -> std::result::Result<bool, JsonRpcError> {
1481	let Some(turn) = state.in_flight_turns.lock().await.get(thread_id).cloned() else {
1482	return Ok(false);
1483	};
1484	interrupt_turn_request(&turn).await
1485	}
1486
1487	/// Interrupt every in-flight turn concurrently, reporting how many were
1488	/// reached (#6211 R8b).
1489	///
1490	/// The teardown paths used to await each turn's interrupt in sequence, so
1491	/// their latency grew with the number of live turns — up to the 10s
1492	/// per-request timeout apiece. Each interrupt now owns its snapshot and runs
1493	/// as an independent task; individual failures are ignored exactly as the
1494	/// sequential loop ignored them, and the registry lock is never held across
1495	/// the requests.
1496	async fn interrupt_all_stdio_turns(state: &AppState) -> usize {
1497	let turns: Vec<InFlightTurn> = {
1498	let map = state.in_flight_turns.lock().await;
1499	map.values().cloned().collect()
1500	};
1501	let mut set = tokio::task::JoinSet::new();
1502	for turn in turns {
1503	set.spawn(async move { interrupt_turn_request(&turn).await });
1504	}
1505	let mut interrupted = 0usize;
1506	while let Some(joined) = set.join_next().await {
1507	if matches!(joined, Ok(Ok(true))) {
1508	interrupted += 1;
1509	}
1510	}
1511	interrupted
1512	}
1513
1514	/// Drop the cached runtime bridge so the next stdio thread message spawns a
1515	/// fresh child that re-reads the persisted config. An in-flight message
1516	/// keeps its own [`SharedRuntimeBridge`] clone and finishes against the old
1517	/// child, which is killed when the last clone drops.
1518	///
1519	/// The stdio→runtime thread map is deliberately not here: it lives on
1520	/// [`AppState::runtime_thread_map`] because runtime threads are durable —
1521	/// the fresh child resolves the same ids from its on-disk store (#6246).
1522	/// Only per-child state (`last_seq_by_thread`) dies with the bridge.
1523	async fn invalidate_runtime_bridge(state: &AppState) {
1524	let mut bridge = state.runtime_bridge.lock().await;
1525	*bridge = None;
1526	}
1527
1528	impl RuntimeBridge {
1529	async fn start(config_path: Option<&Path>) -> Result<Self> {
1530	install_rustls_crypto_provider();
1531	let port = reserve_runtime_port()?;
1532	let auth_token = format!("cwrt_{}", Uuid::new_v4().simple());
1533	let child = Self::runtime_command(config_path, port, &auth_token)?
1534	.spawn()
1535	.context("failed to start runtime API bridge")?;
1536	let mut bridge = Self {
1537	base_url: format!("http://127.0.0.1:{port}"),
1538	client: codewhale_release::platform_http_client_builder()
1539	.build()
1540	.context("failed to build runtime API client")?,
1541	auth_token: Some(auth_token),
1542	child: Some(child),
1543	last_seq_by_thread: HashMap::new(),
1544	};
1545	bridge.wait_until_ready().await?;
1546	Ok(bridge)
1547	}
1548
1549	fn runtime_command(config_path: Option<&Path>, port: u16, auth_token: &str) -> Result<Command> {
1550	let current_exe = std::env::current_exe().ok();
1551	let mut command = if let Some(path) = current_exe {
1552	Command::new(path)
1553	} else {
1554	Command::new("codewhale")
1555	};
1556	// Pass the runtime auth token out-of-band via env (not argv) so local
1557	// `ps` cannot read credential material from the child command line.
1558	// The TUI/runtime server already accepts CODEWHALE_RUNTIME_TOKEN /
1559	// DEEPSEEK_RUNTIME_TOKEN when --auth-token is absent.
1560	command
1561	.arg("app-server")
1562	.arg("--http")
1563	.arg("--host")
1564	.arg("127.0.0.1")
1565	.arg("--port")
1566	.arg(port.to_string())
1567	.env("CODEWHALE_RUNTIME_TOKEN", auth_token)
1568	.env("DEEPSEEK_RUNTIME_TOKEN", auth_token)
1569	.stdin(Stdio::null())
1570	.stdout(Stdio::null())
1571	.stderr(Stdio::null());
1572	if let Some(config_path) = config_path {
1573	command.arg("--config").arg(config_path);
1574	}
1575	Ok(command)
1576	}
1577
1578	async fn wait_until_ready(&mut self) -> Result<()> {
1579	let deadline = Instant::now() + Duration::from_secs(15);
1580	loop {
1581	if let Some(child) = self.child.as_mut()
1582	&& let Some(status) = child.try_wait()?
1583	{
1584	return Err(anyhow!(
1585	"runtime API bridge exited before becoming ready (status {status})"
1586	));
1587	}
1588
1589	match self
1590	.client
1591	.get(format!("{}/health", self.base_url))
1592	.send()
1593	.await
1594	{
1595	Ok(response) if response.status().is_success() => return Ok(()),
1596	_ if Instant::now() >= deadline => {
1597	bail!(
1598	"timed out waiting for runtime API bridge at {}/health",
1599	self.base_url
1600	)
1601	}
1602	_ => tokio::time::sleep(Duration::from_millis(50)).await,
1603	}
1604	}
1605	}
1606
1607	fn authed(&self, builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
1608	match self.auth_token.as_deref() {
1609	Some(token) => builder.bearer_auth(token),
1610	None => builder,
1611	}
1612	}
1613
1614	async fn request_json(&self, builder: reqwest::RequestBuilder) -> Result<Value> {
1615	let response = builder.send().await?;
1616	let status = response.status();
1617	let body = response.text().await?;
1618	if !status.is_success() {
1619	let detail = body.trim();
1620	if detail.is_empty() {
1621	bail!("runtime API returned {status}");
1622	}
1623	bail!("runtime API returned {status}: {detail}");
1624	}
1625	serde_json::from_str(&body).with_context(\|\| format!("invalid runtime API json: {body}"))
1626	}
1627
1628	/// Read the existing Runtime catalog before a one-shot prompt can create
1629	/// its thread. Existing threads are checked by canonical turn admission.
1630	async fn require_output_limited_model(&self, requested_model: Option<&str>) -> Result<()> {
1631	let providers = self
1632	.request_json(self.authed(self.client.get(format!("{}/v1/providers", self.base_url))))
1633	.await?;
1634	let current = providers
1635	.get("current")
1636	.and_then(Value::as_str)
1637	.context("Runtime provider is unavailable")?;
1638	let provider = providers
1639	.get("providers")
1640	.and_then(Value::as_array)
1641	.and_then(\|providers\| {
1642	providers
1643	.iter()
1644	.find(\|provider\| provider.get("id").and_then(Value::as_str) == Some(current))
1645	})
1646	.context("Runtime provider is unavailable")?;
1647	let model = requested_model
1648	.or_else(\|\| provider.get("default_model").and_then(Value::as_str))
1649	.context("maxOutputTokens requires an exact model")?;
1650	if model.trim().is_empty() \|\| model.eq_ignore_ascii_case("auto") {
1651	bail!("maxOutputTokens requires an exact model");
1652	}
1653	if !current
1654	.bytes()
1655	.all(\|byte\| byte.is_ascii_lowercase() \|\| byte.is_ascii_digit() \|\| byte == b'-')
1656	{
1657	bail!("Runtime provider identity is invalid");
1658	}
1659	let mut cursor = None;
1660	let mut seen = std::collections::HashSet::new();
1661	loop {
1662	let mut url =
1663	reqwest::Url::parse(&format!("{}/v1/providers/{current}/models", self.base_url))?;
1664	url.query_pairs_mut().append_pair("limit", "250");
1665	if let Some(cursor) = cursor.as_deref() {
1666	url.query_pairs_mut().append_pair("cursor", cursor);
1667	}
1668	let catalog = self.request_json(self.authed(self.client.get(url))).await?;
1669	if let Some(entry) =
1670	catalog
1671	.get("models")
1672	.and_then(Value::as_array)
1673	.and_then(\|models\| {
1674	models
1675	.iter()
1676	.find(\|entry\| entry.get("id").and_then(Value::as_str) == Some(model))
1677	})
1678	{
1679	if entry.get("output_token_limit").and_then(Value::as_str) == Some("supported") {
1680	return Ok(());
1681	}
1682	bail!("The selected Runtime model does not support maxOutputTokens");
1683	}
1684	let next = catalog
1685	.get("nextCursor")
1686	.and_then(Value::as_str)
1687	.context("Output-limit support is unknown for the selected Runtime model")?;
1688	if !seen.insert(next.to_string()) {
1689	bail!("Runtime model catalog cursor repeated");
1690	}
1691	cursor = Some(next.to_string());
1692	}
1693	}
1694
1695	/// Resolve `stdio_thread_id` to a runtime thread, minting one only when
1696	/// `thread_map` has no entry. The map lives on [`AppState`] and outlives
1697	/// this bridge, so a thread created under a previous child keeps its id
1698	/// here as long as the store it was persisted to is shared (#6246).
1699	async fn ensure_runtime_thread(
1700	&mut self,
1701	thread_map: &mut HashMap<String, String>,
1702	stdio_thread_id: &str,
1703	hint: Option<RuntimeThreadHint>,
1704	) -> Result<String> {
1705	if let Some(runtime_thread_id) = thread_map.get(stdio_thread_id) {
1706	return Ok(runtime_thread_id.clone());
1707	}
1708	let hint = hint.unwrap_or_default();
1709	let runtime_thread_id = self
1710	.create_runtime_thread(hint.model, hint.workspace)
1711	.await?;
1712	thread_map.insert(stdio_thread_id.to_string(), runtime_thread_id.clone());
1713	Ok(runtime_thread_id)
1714	}
1715
1716	/// Drop a thread mapping (and its seq cursor) once no caller can name
1717	/// the client-facing key again.
1718	fn forget_thread(&mut self, thread_map: &mut HashMap<String, String>, stdio_thread_id: &str) {
1719	if let Some(runtime_thread_id) = thread_map.remove(stdio_thread_id) {
1720	self.last_seq_by_thread.remove(&runtime_thread_id);
1721	}
1722	}
1723
1724	async fn create_runtime_thread(
1725	&mut self,
1726	model: Option<String>,
1727	workspace: Option<PathBuf>,
1728	) -> Result<String> {
1729	let record = self
1730	.request_json(
1731	self.authed(self.client.post(format!("{}/v1/threads", self.base_url)))
1732	.json(&json!({
1733	"model": model,
1734	"workspace": workspace,
1735	"mode": "agent",
1736	"archived": false,
1737	})),
1738	)
1739	.await?;
1740	let thread_id = extract_runtime_thread_id(&record)?.to_string();
1741	self.last_seq_by_thread
1742	.entry(thread_id.clone())
1743	.or_insert(0);
1744	Ok(thread_id)
1745	}
1746
1747	/// Run one turn to completion, streaming its events to `writer`.
1748	///
1749	/// `registration` is `Some` on the stdio path: it publishes the live turn
1750	/// so an `thread/interrupt` arriving mid-stream can reach the runtime
1751	/// without waiting on the bridge mutex this call holds.
1752	async fn message_thread<W: AsyncWrite + Unpin>(
1753	&mut self,
1754	thread_id: &str,
1755	input: &str,
1756	images: &[RuntimeImageInput],
1757	max_output_tokens: Option<std::num::NonZeroU32>,
1758	writer: &mut W,
1759	registration: Option<(TurnRegistry, String)>,
1760	mut transcript: Option<&mut TurnTranscript>,
1761	) -> Result<Value> {
1762	let mut request = json!({ "prompt": input });
1763	if !images.is_empty() {
1764	let info = self
1765	.request_json(
1766	self.authed(
1767	self.client
1768	.get(format!("{}/v1/runtime/info", self.base_url)),
1769	),
1770	)
1771	.await?;
1772	if info
1773	.pointer("/capabilities/turn_image_inputs")
1774	.and_then(Value::as_bool)
1775	!= Some(true)
1776	{
1777	bail!(
1778	"Runtime image input is unavailable; update the Runtime before sending attachments"
1779	);
1780	}
1781	request["images"] = json!(images);
1782	}
1783	if let Some(limit) = max_output_tokens {
1784	request["maxOutputTokens"] = json!(limit);
1785	}
1786	let turn = self
1787	.request_json(
1788	self.authed(
1789	self.client
1790	.post(format!("{}/v1/threads/{thread_id}/turns", self.base_url)),
1791	)
1792	.json(&request),
1793	)
1794	.await?;
1795	let turn_id = turn
1796	.pointer("/turn/id")
1797	.and_then(Value::as_str)
1798	.ok_or_else(\|\| anyhow!("runtime API turn response missing turn.id"))?
1799	.to_string();
1800	let response_id = format!("{thread_id}:{turn_id}");
1801
1802	if let Some(transcript) = transcript.as_deref_mut() {
1803	transcript.model = turn
1804	.pointer("/thread/model")
1805	.and_then(Value::as_str)
1806	.map(str::to_string);
1807	transcript.events.push(EventFrame::ResponseStart {
1808	response_id: response_id.clone(),
1809	});
1810	}
1811
1812	emit_stdio_event(
1813	writer,
1814	json!({
1815	"type": "response_start",
1816	"response_id": response_id,
1817	}),
1818	)
1819	.await?;
1820
1821	// Publish the turn only for the streaming window, and take it back
1822	// before any `?` below: a turn that has already finished must never
1823	// look cancellable.
1824	if let Some((registry, key)) = registration.as_ref() {
1825	registry.lock().await.insert(
1826	key.clone(),
1827	InFlightTurn {
1828	base_url: self.base_url.clone(),
1829	auth_token: self.auth_token.clone(),
1830	runtime_thread_id: thread_id.to_string(),
1831	turn_id: turn_id.clone(),
1832	},
1833	);
1834	}
1835
1836	let since_seq = self.last_seq_by_thread.get(thread_id).copied().unwrap_or(0);
1837	let stream_result = self
1838	.stream_turn_events(
1839	thread_id,
1840	&turn_id,
1841	&response_id,
1842	writer,
1843	since_seq,
1844	transcript.as_deref_mut(),
1845	)
1846	.await;
1847
1848	if let Some((registry, key)) = registration.as_ref() {
1849	registry.lock().await.remove(key);
1850	}
1851
1852	let _ = emit_stdio_event(
1853	writer,
1854	json!({
1855	"type": "response_end",
1856	"response_id": response_id,
1857	}),
1858	)
1859	.await;
1860	if let Some(transcript) = transcript {
1861	transcript.events.push(EventFrame::ResponseEnd {
1862	response_id: response_id.clone(),
1863	});
1864	}
1865
1866	let (last_seq, status, error) = stream_result?;
1867	self.last_seq_by_thread
1868	.insert(thread_id.to_string(), last_seq);
1869
1870	match status {
1871	TurnTerminalStatus::Completed => Ok(json!({
1872	"thread_id": thread_id,
1873	"status": "accepted",
1874	"thread": Value::Null,
1875	"threads": [],
1876	"model": Value::Null,
1877	"model_provider": Value::Null,
1878	"cwd": Value::Null,
1879	"approval_policy": Value::Null,
1880	"sandbox": Value::Null,
1881	"events": [],
1882	"data": { "turn_id": turn_id },
1883	})),
1884	TurnTerminalStatus::Failed => Err(anyhow!(
1885	"{}",
1886	error.unwrap_or_else(\|\| "turn failed".to_string())
1887	)),
1888	TurnTerminalStatus::Interrupted => Err(anyhow!(
1889	"{}",
1890	error.unwrap_or_else(\|\| "turn interrupted".to_string())
1891	)),
1892	TurnTerminalStatus::Canceled => Err(anyhow!(
1893	"{}",
1894	error.unwrap_or_else(\|\| "turn canceled".to_string())
1895	)),
1896	}
1897	}
1898
1899	async fn stream_turn_events<W: AsyncWrite + Unpin>(
1900	&self,
1901	thread_id: &str,
1902	turn_id: &str,
1903	response_id: &str,
1904	writer: &mut W,
1905	since_seq: u64,
1906	mut transcript: Option<&mut TurnTranscript>,
1907	) -> Result<(u64, TurnTerminalStatus, Option<String>)> {
1908	let mut response = self
1909	.authed(self.client.get(format!(
1910	"{}/v1/threads/{thread_id}/events?since_seq={since_seq}",
1911	self.base_url
1912	)))
1913	.send()
1914	.await?
1915	.error_for_status()?;
1916
1917	let mut buffer = Vec::new();
1918	let mut last_seq = since_seq;
1919
1920	while let Some(chunk) = response.chunk().await? {
1921	buffer.extend_from_slice(&chunk);
1922	if buffer.len() > MAX_SSE_FRAME_BYTES {
1923	bail!(
1924	"runtime SSE frame exceeded {MAX_SSE_FRAME_BYTES} bytes without a frame delimiter"
1925	);
1926	}
1927	while let Some(frame_bytes) = take_sse_frame(&mut buffer) {
1928	let Some((event_name, frame_data)) = parse_sse_frame(&frame_bytes) else {
1929	continue;
1930	};
1931	let envelope: Value = serde_json::from_str(&frame_data)
1932	.with_context(\|\| format!("invalid SSE json for {event_name}: {frame_data}"))?;
1933	if let Some(seq) = envelope.get("seq").and_then(Value::as_u64) {
1934	last_seq = last_seq.max(seq);
1935	}
1936	if envelope.get("turn_id").and_then(Value::as_str) != Some(turn_id) {
1937	continue;
1938	}
1939	let payload = envelope.get("payload").cloned().unwrap_or(Value::Null);
1940	match event_name.as_str() {
1941	"item.delta" => {
1942	let kind = payload
1943	.get("kind")
1944	.and_then(Value::as_str)
1945	.unwrap_or_default();
1946	if kind == "agent_message"
1947	&& let Some(delta) = payload.get("delta").and_then(Value::as_str)
1948	&& !delta.is_empty()
1949	{
1950	emit_stdio_event(
1951	writer,
1952	json!({
1953	"type": "response_delta",
1954	"response_id": response_id,
1955	"delta": delta,
1956	}),
1957	)
1958	.await?;
1959	if let Some(transcript) = transcript.as_deref_mut() {
1960	transcript.text.push_str(delta);
1961	transcript.events.push(EventFrame::ResponseDelta {
1962	response_id: response_id.to_string(),
1963	delta: delta.to_string(),
1964	channel: ResponseChannel::Text,
1965	});
1966	}
1967	}
1968	}
1969	"turn.completed" => {
1970	let status = turn_terminal_status(&payload);
1971	let error = payload
1972	.pointer("/turn/error")
1973	.and_then(Value::as_str)
1974	.map(str::to_string);
1975	return Ok((last_seq, status, error));
1976	}
1977	_ => {}
1978	}
1979	}
1980	}
1981
1982	bail!("runtime event stream ended before turn.completed")
1983	}
1984
1985	#[cfg(test)]
1986	fn from_base_url_for_test(base_url: String) -> Self {
1987	install_rustls_crypto_provider();
1988	Self {
1989	base_url,
1990	client: codewhale_release::platform_http_client_builder()
1991	.timeout(Duration::from_secs(5))
1992	.build()
1993	.expect("build reqwest test client"),
1994	auth_token: None,
1995	child: None,
1996	last_seq_by_thread: HashMap::new(),
1997	}
1998	}
1999	}
2000
2001	impl RuntimeBridge {
2002	/// Kills the managed runtime child and reaps it on a detached thread so
2003	/// neither an explicit shutdown nor Drop blocks a Tokio runtime thread.
2004	fn shutdown_child(&mut self) {
2005	if let Some(mut child) = self.child.take() {
2006	let _ = child.kill();
2007	std::thread::spawn(move \|\| {
2008	let _ = child.wait();
2009	});
2010	}
2011	}
2012	}
2013
2014	impl Drop for RuntimeBridge {
2015	fn drop(&mut self) {
2016	self.shutdown_child();
2017	}
2018	}
2019
2020	fn reserve_runtime_port() -> Result<u16> {
2021	let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
2022	Ok(listener.local_addr()?.port())
2023	}
2024
2025	fn install_rustls_crypto_provider() {
2026	let _ = rustls::crypto::ring::default_provider().install_default();
2027	}
2028
2029	fn extract_runtime_thread_id(record: &Value) -> Result<&str> {
2030	record
2031	.get("id")
2032	.and_then(Value::as_str)
2033	.ok_or_else(\|\| anyhow!("runtime API thread response missing id"))
2034	}
2035
2036	fn turn_terminal_status(payload: &Value) -> TurnTerminalStatus {
2037	match payload
2038	.pointer("/turn/status")
2039	.and_then(Value::as_str)
2040	.unwrap_or("completed")
2041	.to_ascii_lowercase()
2042	.as_str()
2043	{
2044	"failed" => TurnTerminalStatus::Failed,
2045	"interrupted" => TurnTerminalStatus::Interrupted,
2046	"canceled" \| "cancelled" => TurnTerminalStatus::Canceled,
2047	_ => TurnTerminalStatus::Completed,
2048	}
2049	}
2050
2051	async fn emit_stdio_event<W: AsyncWrite + Unpin>(writer: &mut W, event: Value) -> Result<()> {
2052	writer.write_all(&serde_json::to_vec(&event)?).await?;
2053	writer.write_all(b"\n").await?;
2054	writer.flush().await?;
2055	Ok(())
2056	}
2057
2058	fn take_sse_frame(buffer: &mut Vec<u8>) -> Option<Vec<u8>> {
2059	if let Some(pos) = buffer.windows(4).position(\|window\| window == b"\r\n\r\n") {
2060	return Some(buffer.drain(..pos + 4).collect());
2061	}
2062	buffer
2063	.windows(2)
2064	.position(\|window\| window == b"\n\n")
2065	.map(\|pos\| buffer.drain(..pos + 2).collect())
2066	}
2067
2068	fn parse_sse_frame(frame_bytes: &[u8]) -> Option<(String, String)> {
2069	let text = String::from_utf8(frame_bytes.to_vec()).ok()?;
2070	let mut event_name = None;
2071	let mut data_lines = Vec::new();
2072	for raw_line in text.lines() {
2073	let line = raw_line.trim_end_matches('\r');
2074	if let Some(value) = line.strip_prefix("event:") {
2075	event_name = Some(value.trim().to_string());
2076	} else if let Some(value) = line.strip_prefix("data:") {
2077	data_lines.push(value.trim_start().to_string());
2078	}
2079	}
2080	match (event_name, data_lines.is_empty()) {
2081	(Some(event), false) => Some((event, data_lines.join("\n"))),
2082	_ => None,
2083	}
2084	}
2085
2086	#[cfg(test)]
2087	async fn dispatch_stdio_request(
2088	state: &AppState,
2089	method: &str,
2090	params: Value,
2091	) -> std::result::Result<StdioDispatchResult, JsonRpcError> {
2092	let mut sink = tokio::io::sink();
2093	dispatch_stdio_request_with_writer(state, &mut sink, method, params, AppTransport::Stdio).await
2094	}
2095
2096	async fn dispatch_stdio_app_request(
2097	state: &AppState,
2098	request: AppRequest,
2099	transport: AppTransport,
2100	) -> std::result::Result<StdioDispatchResult, JsonRpcError> {
2101	let response = Box::pin(process_app_request(state, request, transport)).await;
2102	Ok(StdioDispatchResult {
2103	result: serde_json::to_value(response)
2104	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2105	should_exit: false,
2106	})
2107	}
2108
2109	async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
2110	state: &AppState,
2111	writer: &mut W,
2112	method: &str,
2113	params: Value,
2114	transport: AppTransport,
2115	) -> std::result::Result<StdioDispatchResult, JsonRpcError> {
2116	let outcome = match method {
2117	"healthz" \| "app/healthz" => StdioDispatchResult {
2118	result: json!({
2119	"status": "ok",
2120	"service": legacy_deepseek_compat::SERVICE_NAME,
2121	"transport": transport.label()
2122	}),
2123	should_exit: false,
2124	},
2125	"capabilities" => {
2126	let mut methods = vec![
2127	"healthz",
2128	"thread/capabilities",
2129	"thread/request",
2130	"thread/create",
2131	"thread/start",
2132	"thread/resume",
2133	"thread/fork",
2134	"thread/list",
2135	"thread/read",
2136	"thread/set_name",
2137	"thread/goal/set",
2138	"thread/goal/get",
2139	"thread/goal/clear",
2140	"thread/archive",
2141	"thread/unarchive",
2142	"thread/message",
2143	"thread/interrupt",
2144	"app/capabilities",
2145	"app/request",
2146	"app/config/get",
2147	"app/config/set",
2148	"app/config/unset",
2149	"app/config/list",
2150	"app/config/reload",
2151	"app/models",
2152	"app/thread_loaded_list",
2153	"prompt/capabilities",
2154	"prompt/request",
2155	"prompt/run",
2156	"shutdown",
2157	];
2158	if transport == AppTransport::Socket {
2159	// The daemon handshake exists only on the socket transport;
2160	// stdio/HTTP clients never see it, so the stdio pin is unchanged.
2161	methods.insert(1, daemon_socket::ATTACH_METHOD);
2162	}
2163	StdioDispatchResult {
2164	result: json!({
2165	"transport": transport.label(),
2166	"families": ["thread/", "app/", "prompt/*"],
2167	"turn_image_inputs": true,
2168	"methods": methods,
2169	}),
2170	should_exit: false,
2171	}
2172	}
2173	"thread/capabilities" => StdioDispatchResult {
2174	result: json!({
2175	"turn_image_inputs": true,
2176	"methods": [
2177	"thread/request",
2178	"thread/create",
2179	"thread/start",
2180	"thread/resume",
2181	"thread/fork",
2182	"thread/list",
2183	"thread/read",
2184	"thread/set_name",
2185	"thread/goal/set",
2186	"thread/goal/get",
2187	"thread/goal/clear",
2188	"thread/archive",
2189	"thread/unarchive",
2190	"thread/message",
2191	"thread/interrupt"
2192	]
2193	}),
2194	should_exit: false,
2195	},
2196	"thread/request" => {
2197	let request: ThreadRequest = parse_params(params)?;
2198	if let ThreadRequest::Message {
2199	thread_id,
2200	input,
2201	images,
2202	max_output_tokens,
2203	} = request
2204	{
2205	let response = handle_stdio_thread_message(
2206	state,
2207	writer,
2208	ThreadMessageParams {
2209	thread_id,
2210	input,
2211	images,
2212	max_output_tokens,
2213	},
2214	)
2215	.await?;
2216	return Ok(StdioDispatchResult {
2217	result: response,
2218	should_exit: false,
2219	});
2220	}
2221	let should_record_hint = matches!(
2222	&request,
2223	ThreadRequest::Create { .. }
2224	\| ThreadRequest::Start(_)
2225	\| ThreadRequest::Resume(_)
2226	\| ThreadRequest::Fork(_)
2227	);
2228	let response = handle_thread_request(state, request).await?;
2229	if should_record_hint {
2230	record_stdio_thread_hint(state, &response).await;
2231	}
2232	StdioDispatchResult {
2233	result: serde_json::to_value(response)
2234	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2235	should_exit: false,
2236	}
2237	}
2238	"thread/create" => {
2239	#[derive(Debug, Deserialize)]
2240	struct CreateParams {
2241	#[serde(default)]
2242	metadata: Value,
2243	}
2244	let parsed: CreateParams = parse_params(params_or_object(params))?;
2245	let response = handle_thread_request(
2246	state,
2247	ThreadRequest::Create {
2248	metadata: parsed.metadata,
2249	},
2250	)
2251	.await?;
2252	record_stdio_thread_hint(state, &response).await;
2253	StdioDispatchResult {
2254	result: serde_json::to_value(response)
2255	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2256	should_exit: false,
2257	}
2258	}
2259	"thread/start" => {
2260	let request = ThreadRequest::Start(parse_params(params_or_object(params))?);
2261	let response = handle_thread_request(state, request).await?;
2262	record_stdio_thread_hint(state, &response).await;
2263	StdioDispatchResult {
2264	result: serde_json::to_value(response)
2265	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2266	should_exit: false,
2267	}
2268	}
2269	"thread/resume" => {
2270	let request = ThreadRequest::Resume(parse_params(params_or_object(params))?);
2271	let response = handle_thread_request(state, request).await?;
2272	ensure_thread_found(&response)?;
2273	record_stdio_thread_hint(state, &response).await;
2274	StdioDispatchResult {
2275	result: serde_json::to_value(response)
2276	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2277	should_exit: false,
2278	}
2279	}
2280	"thread/fork" => {
2281	let request = ThreadRequest::Fork(parse_params(params_or_object(params))?);
2282	let response = handle_thread_request(state, request).await?;
2283	ensure_thread_found(&response)?;
2284	record_stdio_thread_hint(state, &response).await;
2285	StdioDispatchResult {
2286	result: serde_json::to_value(response)
2287	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2288	should_exit: false,
2289	}
2290	}
2291	"thread/list" => {
2292	let request = ThreadRequest::List(parse_params(params_or_object(params))?);
2293	let response = handle_thread_request(state, request).await?;
2294	StdioDispatchResult {
2295	result: serde_json::to_value(response)
2296	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2297	should_exit: false,
2298	}
2299	}
2300	"thread/read" => {
2301	let request = ThreadRequest::Read(parse_params(params_or_object(params))?);
2302	let response = handle_thread_request(state, request).await?;
2303	StdioDispatchResult {
2304	result: serde_json::to_value(response)
2305	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2306	should_exit: false,
2307	}
2308	}
2309	"thread/set_name" \| "thread/set-name" => {
2310	let request = ThreadRequest::SetName(parse_params(params_or_object(params))?);
2311	let response = handle_thread_request(state, request).await?;
2312	StdioDispatchResult {
2313	result: serde_json::to_value(response)
2314	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2315	should_exit: false,
2316	}
2317	}
2318	"thread/goal/set" \| "thread/goal_set" \| "thread/goal-set" => {
2319	let request = ThreadRequest::GoalSet(parse_params::<ThreadGoalSetParams>(
2320	params_or_object(params),
2321	)?);
2322	let response = handle_thread_request(state, request).await?;
2323	StdioDispatchResult {
2324	result: serde_json::to_value(response)
2325	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2326	should_exit: false,
2327	}
2328	}
2329	"thread/goal/get" \| "thread/goal_get" \| "thread/goal-get" => {
2330	let request = ThreadRequest::GoalGet(parse_params::<ThreadGoalGetParams>(
2331	params_or_object(params),
2332	)?);
2333	let response = handle_thread_request(state, request).await?;
2334	StdioDispatchResult {
2335	result: serde_json::to_value(response)
2336	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2337	should_exit: false,
2338	}
2339	}
2340	"thread/goal/clear" \| "thread/goal_clear" \| "thread/goal-clear" => {
2341	let request = ThreadRequest::GoalClear(parse_params::<ThreadGoalClearParams>(
2342	params_or_object(params),
2343	)?);
2344	let response = handle_thread_request(state, request).await?;
2345	StdioDispatchResult {
2346	result: serde_json::to_value(response)
2347	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2348	should_exit: false,
2349	}
2350	}
2351	"thread/archive" => {
2352	let parsed: ThreadIdParams = parse_params(params_or_object(params))?;
2353	let response = handle_thread_request(
2354	state,
2355	ThreadRequest::Archive {
2356	thread_id: parsed.thread_id,
2357	},
2358	)
2359	.await?;
2360	StdioDispatchResult {
2361	result: serde_json::to_value(response)
2362	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2363	should_exit: false,
2364	}
2365	}
2366	"thread/unarchive" => {
2367	let parsed: ThreadIdParams = parse_params(params_or_object(params))?;
2368	let response = handle_thread_request(
2369	state,
2370	ThreadRequest::Unarchive {
2371	thread_id: parsed.thread_id,
2372	},
2373	)
2374	.await?;
2375	StdioDispatchResult {
2376	result: serde_json::to_value(response)
2377	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2378	should_exit: false,
2379	}
2380	}
2381	"thread/message" => {
2382	let parsed: ThreadMessageParams = parse_params(params_or_object(params))?;
2383	let response = handle_stdio_thread_message(state, writer, parsed).await?;
2384	StdioDispatchResult {
2385	result: response,
2386	should_exit: false,
2387	}
2388	}
2389	"app/capabilities" => {
2390	dispatch_stdio_app_request(state, AppRequest::Capabilities, transport).await?
2391	}
2392	"app/request" => {
2393	let request: AppRequest = parse_params(params)?;
2394	dispatch_stdio_app_request(state, request, transport).await?
2395	}
2396	"app/config/get" => {
2397	let parsed: ConfigGetParams = parse_params(params_or_object(params))?;
2398	dispatch_stdio_app_request(state, AppRequest::ConfigGet { key: parsed.key }, transport)
2399	.await?
2400	}
2401	"app/config/set" => {
2402	let parsed: ConfigSetParams = parse_params(params_or_object(params))?;
2403	dispatch_stdio_app_request(
2404	state,
2405	AppRequest::ConfigSet {
2406	key: parsed.key,
2407	value: parsed.value,
2408	},
2409	transport,
2410	)
2411	.await?
2412	}
2413	"app/config/unset" => {
2414	let parsed: ConfigGetParams = parse_params(params_or_object(params))?;
2415	dispatch_stdio_app_request(
2416	state,
2417	AppRequest::ConfigUnset { key: parsed.key },
2418	transport,
2419	)
2420	.await?
2421	}
2422	"app/config/list" => {
2423	dispatch_stdio_app_request(state, AppRequest::ConfigList, transport).await?
2424	}
2425	"app/config/reload" => {
2426	dispatch_stdio_app_request(state, AppRequest::ConfigReload, transport).await?
2427	}
2428	"app/models" => dispatch_stdio_app_request(state, AppRequest::Models, transport).await?,
2429	"app/thread_loaded_list" \| "app/thread-loaded-list" => {
2430	dispatch_stdio_app_request(state, AppRequest::ThreadLoadedList, transport).await?
2431	}
2432	"prompt/capabilities" => StdioDispatchResult {
2433	result: json!({
2434	"methods": ["prompt/request", "prompt/run"]
2435	}),
2436	should_exit: false,
2437	},
2438	"prompt/request" \| "prompt/run" => {
2439	let request: PromptRequest = parse_params(params)?;
2440	let response = handle_prompt_request(state, writer, request).await?;
2441	StdioDispatchResult {
2442	result: serde_json::to_value(response)
2443	.map_err(\|err\| JsonRpcError::internal(err.to_string()))?,
2444	should_exit: false,
2445	}
2446	}
2447	"thread/interrupt" => {
2448	let parsed: ThreadInterruptParams = parse_params(params_or_object(params))?;
2449	let interrupted = interrupt_stdio_turn(state, &parsed.thread_id).await?;
2450	StdioDispatchResult {
2451	result: json!({
2452	"thread_id": parsed.thread_id,
2453	"interrupted": interrupted,
2454	}),
2455	should_exit: false,
2456	}
2457	}
2458	"shutdown" => {
2459	// A turn streaming right now holds the bridge mutex, so taking it
2460	// to kill the child would block until that turn ends — the exact
2461	// deadlock that made shutdown useless against a runaway turn.
2462	// Interrupt live turns first; they release the mutex promptly.
2463	let _ = interrupt_all_stdio_turns(state).await;
2464	if let Some(bridge) = state.runtime_bridge.lock().await.take() {
2465	bridge.lock().await.shutdown_child();
2466	}
2467	StdioDispatchResult {
2468	result: json!({"ok": true, "status": "stopped"}),
2469	should_exit: true,
2470	}
2471	}
2472	daemon_socket::ATTACH_METHOD if transport == AppTransport::Socket => {
2473	return Err(JsonRpcError::already_attached());
2474	}
2475	_ => return Err(JsonRpcError::method_not_found(method)),
2476	};
2477	Ok(outcome)
2478	}
2479
2480	async fn process_app_request(
2481	state: &AppState,
2482	req: AppRequest,
2483	_transport: AppTransport,
2484	) -> AppResponse {
2485	match req {
2486	AppRequest::Capabilities => AppResponse {
2487	ok: true,
2488	data: json!({
2489	"routes": ["/thread", "/app", "/prompt", "/tool", "/jobs", "/mcp/startup"],
2490	"config": ["get", "set", "unset", "list", "reload"],
2491	"events": ["response_start", "response_delta", "response_end", "tool_call_start", "tool_call_result", "mcp_startup_update", "mcp_startup_complete"],
2492	"transport": "stdio+http",
2493	"config_path": state.config_path.as_ref().map(\|p\| p.display().to_string()),
2494	}),
2495	events: Vec::new(),
2496	},
2497	AppRequest::ConfigGet { key } => {
2498	let cfg = state.config.read().await;
2499	let value = cfg.get_display_value(&key);
2500	AppResponse {
2501	ok: true,
2502	data: json!({ "key": key, "value": value }),
2503	events: Vec::new(),
2504	}
2505	}
2506	AppRequest::ConfigSet { key, value } => {
2507	let (result, snapshot) = {
2508	let mut cfg = state.config.write().await;
2509	let result = cfg.set_value(&key, &value);
2510	(result, cfg.clone())
2511	};
2512	let ok = result.is_ok();
2513	let message = result.err().map(\|e\| e.to_string());
2514	// Only propagate a mutation that actually happened. `set_value`
2515	// leaves the config untouched on an unknown key or invalid value,
2516	// so this is a no-op from the caller's point of view — but
2517	// `apply_config_update` invalidates the cached stdio bridge
2518	// regardless, and dropping the last reference kills the running
2519	// child runtime along with its thread map. A single typo'd key
2520	// would orphan every in-flight thread on that bridge.
2521	if ok {
2522	apply_config_update(state, snapshot, None, true).await;
2523	}
2524	AppResponse {
2525	ok,
2526	data: json!({ "key": key, "value": value, "error": message }),
2527	events: Vec::new(),
2528	}
2529	}
2530	AppRequest::ConfigUnset { key } => {
2531	let (result, snapshot) = {
2532	let mut cfg = state.config.write().await;
2533	let result = cfg.unset_value(&key);
2534	(result, cfg.clone())
2535	};
2536	let ok = result.is_ok();
2537	let message = result.err().map(\|e\| e.to_string());
2538	// See ConfigSet: a failed unset changed nothing and must not tear
2539	// down the runtime bridge.
2540	if ok {
2541	apply_config_update(state, snapshot, None, true).await;
2542	}
2543	AppResponse {
2544	ok,
2545	data: json!({ "key": key, "error": message }),
2546	events: Vec::new(),
2547	}
2548	}
2549	AppRequest::ConfigList => {
2550	let cfg = state.config.read().await;
2551	AppResponse {
2552	ok: true,
2553	data: json!({ "values": cfg.list_values() }),
2554	events: Vec::new(),
2555	}
2556	}
2557	AppRequest::ConfigReload => {
2558	// Re-read both `config.toml` and the sibling `permissions.toml`
2559	// from disk (the headless equivalent of the TUI
2560	// `reload_runtime_config` codepath) and push the fresh
2561	// snapshots into `state.config` and the live `Runtime`.
2562	//
2563	// `ConfigStore::load` resolves the same default config path
2564	// that `build_state` used at startup when `config_path` is
2565	// `None`, so a `None` here reloads from the same on-disk file
2566	// the server booted from.
2567	let store = match ConfigStore::load(state.config_path.clone()) {
2568	Ok(store) => store,
2569	Err(e) => {
2570	return AppResponse {
2571	ok: false,
2572	data: json!({ "error": format!("failed to load config: {e}") }),
2573	events: Vec::new(),
2574	};
2575	}
2576	};
2577	let new_config = store.config.clone();
2578	let new_exec_policy = store.exec_policy_engine();
2579
2580	// Disk is already the source of truth here, so nothing to
2581	// persist; the exec policy rides along so the runtime picks up
2582	// external `permissions.toml` edits too.
2583	apply_config_update(state, new_config, Some(new_exec_policy), false).await;
2584
2585	AppResponse {
2586	ok: true,
2587	data: json!({ "reloaded": true }),
2588	events: Vec::new(),
2589	}
2590	}
2591	AppRequest::Models => AppResponse {
2592	ok: true,
2593	data: json!({ "models": state.registry.list() }),
2594	events: Vec::new(),
2595	},
2596	AppRequest::ThreadLoadedList => {
2597	let mut runtime = state.runtime.write().await;
2598	let response = runtime
2599	.handle_thread(codewhale_protocol::ThreadRequest::List(
2600	codewhale_protocol::ThreadListParams {
2601	include_archived: false,
2602	limit: Some(50),
2603	},
2604	))
2605	.await;
2606	match response {
2607	Ok(thread_resp) => AppResponse {
2608	ok: true,
2609	data: json!({ "threads": thread_resp.threads }),
2610	events: thread_resp.events,
2611	},
2612	Err(err) => AppResponse {
2613	ok: false,
2614	data: json!({ "error": err.to_string() }),
2615	events: Vec::new(),
2616	},
2617	}
2618	}
2619	AppRequest::SubmitUserInput { request_id, .. } => {
2620	// This transport cannot deliver a clarification answer, and
2621	// saying otherwise was the bug: the previous implementation
2622	// reported `resolved: true` and filed the answers in a map with
2623	// no reader anywhere in this crate.
2624	//
2625	// It cannot be made to work here. `handle_line_during_turn`
2626	// executes exactly one method while a turn is streaming —
2627	// `thread/interrupt`. Everything else, `app/request` included,
2628	// queues until the turn ends, so an answer sent over this
2629	// transport would wait on the very turn that is waiting for it.
2630	// The runtime API owns the pending request and can resume the
2631	// turn, so that is where the reply belongs.
2632	AppResponse {
2633	ok: false,
2634	data: json!({
2635	"error": "user_input_reply_unsupported",
2636	"request_id": request_id,
2637	"message": concat!(
2638	"the app-server control transport cannot deliver clarification answers: ",
2639	"only `thread/interrupt` runs while a turn is streaming, so an answer sent ",
2640	"here would queue behind the turn waiting for it. Reply on the runtime API ",
2641	"instead: POST /v1/user-input/{thread_id}/{request_id}."
2642	),
2643	}),
2644	events: Vec::new(),
2645	}
2646	}
2647	}
2648	}
2649
2650	/// Propagate a new config snapshot to every place that must observe it:
2651	/// optionally persist it to disk, install it in the shared `state.config`,
2652	/// push it into the live [`Runtime`], and invalidate the cached stdio
2653	/// bridge so the next stdio request spawns a fresh child that reads the
2654	/// new on-disk config. The stdio→runtime thread map survives: runtime
2655	/// threads are durable, so the fresh child adopts the existing mappings
2656	/// rather than minting replacements (#6246). Shared by `ConfigSet` /
2657	/// `ConfigUnset` / `ConfigReload`.
2658	///
2659	/// `exec_policy` is `Some` only on the reload path, which re-reads
2660	/// `permissions.toml` from disk; set/unset intentionally leave the live
2661	/// exec policy alone (use `ConfigReload` to pick up external permission
2662	/// edits). `persist` is false on the reload path because disk is already
2663	/// the source of truth there.
2664	async fn apply_config_update(
2665	state: &AppState,
2666	snapshot: codewhale_config::ConfigToml,
2667	exec_policy: Option<codewhale_execpolicy::ExecPolicyEngine>,
2668	persist: bool,
2669	) {
2670	if persist && let Err(e) = persist_config(state, snapshot.clone()).await {
2671	tracing::error!("Failed to persist config update: {e}");
2672	}
2673	{
2674	let mut cfg = state.config.write().await;
2675	*cfg = snapshot.clone();
2676	}
2677	// Sync into the live Runtime so the next turn picks up the change
2678	// without a restart. MCP server connections are NOT refreshed here —
2679	// see `Runtime::reload_config_and_policy` for the headless boundary;
2680	// the TUI's explicit `/mcp reload` operation is a separate path.
2681	{
2682	let mut runtime = state.runtime.write().await;
2683	match exec_policy {
2684	Some(policy) => runtime.reload_config_and_policy(snapshot, policy),
2685	None => runtime.update_config(snapshot),
2686	}
2687	}
2688	invalidate_runtime_bridge(state).await;
2689	}
2690
2691	async fn persist_config(state: &AppState, config: codewhale_config::ConfigToml) -> Result<()> {
2692	if state.config_path.is_none() {
2693	return Ok(());
2694	}
2695	let mut store = ConfigStore::load(state.config_path.clone())?;
2696	store.config = config;
2697	store.save()
2698	}
2699
2700	/// Install the process-wide rustls crypto provider once for tests that build
2701	/// an HTTP client. Production installs it at startup; each test must do the
2702	/// same instead of relying on another test in the process having run first
2703	/// (nextest runs every test in its own process).
2704	#[cfg(test)]
2705	pub(crate) fn install_test_crypto_provider() {
2706	static INIT: std::sync::OnceLock<()> = std::sync::OnceLock::new();
2707	INIT.get_or_init(\|\| {
2708	let _ = rustls::crypto::ring::default_provider().install_default();
2709	});
2710	}
2711
2712	#[cfg(test)]
2713	mod tests {
2714	use super::*;
2715	use axum::body::{Body, to_bytes};
2716	use axum::extract::{Path as AxumPath, Query};
2717	use axum::http::header;
2718	use codewhale_protocol::AppRequest;
2719	use std::collections::HashMap;
2720	use std::fs;
2721	use tokio::io::AsyncReadExt;
2722	use tower::ServiceExt;
2723
2724	fn app_with_config(auth_token: Option<&str>) -> (Router, tempfile::TempDir) {
2725	let tmp = tempfile::tempdir().expect("tempdir");
2726	let config_path = tmp.path().join("config.toml");
2727	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2728	let state = build_state(
2729	Some(config_path),
2730	auth_token.map(std::string::ToString::to_string),
2731	)
2732	.expect("state");
2733	(app_router(state, &[]), tmp)
2734	}
2735
2736	#[test]
2737	fn build_state_keeps_resolved_explicit_config_path() {
2738	let tmp = tempfile::tempdir().expect("tempdir");
2739	let config_dir = tmp.path().join("config-dir");
2740	fs::create_dir_all(&config_dir).expect("config dir");
2741	let config_path = config_dir.join("config.toml");
2742	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2743
2744	let state = build_state(Some(config_path.clone()), None).expect("state");
2745
2746	assert_eq!(
2747	state.config_path.as_deref(),
2748	Some(
2749	config_path
2750	.canonicalize()
2751	.expect("canonical config")
2752	.as_path()
2753	)
2754	);
2755	}
2756
2757	#[tokio::test]
2758	async fn stdio_transport_never_registers_the_stdout_hook_sink() {
2759	let tmp = tempfile::tempdir().expect("tempdir");
2760	let config_path = tmp.path().join("config.toml");
2761	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2762
2763	let http_state =
2764	build_state_with_transport(Some(config_path.clone()), None, AppTransport::Http)
2765	.expect("http state");
2766	let stdio_state = build_state_with_transport(Some(config_path), None, AppTransport::Stdio)
2767	.expect("stdio state");
2768
2769	let http_sinks = http_state.runtime.read().await.hooks.sink_count();
2770	let stdio_sinks = stdio_state.runtime.read().await.hooks.sink_count();
2771	assert_eq!(
2772	http_sinks,
2773	stdio_sinks + 1,
2774	"HTTP mode keeps StdoutHookSink + JsonlHookSink; stdio must drop the stdout sink (#5165)"
2775	);
2776	}
2777
2778	async fn response_body_json(response: Response) -> Value {
2779	let bytes = to_bytes(response.into_body(), usize::MAX)
2780	.await
2781	.expect("body bytes");
2782	serde_json::from_slice(&bytes).expect("json response")
2783	}
2784
2785	#[tokio::test]
2786	async fn http_app_routes_require_bearer_token_when_auth_enabled() {
2787	let (app, _tmp) = app_with_config(Some("test-token"));
2788	let response = app
2789	.oneshot(
2790	Request::builder()
2791	.method(Method::POST)
2792	.uri("/app")
2793	.header(header::CONTENT_TYPE, "application/json")
2794	.body(Body::from(
2795	serde_json::to_vec(&AppRequest::ConfigGet {
2796	key: "api_key".to_string(),
2797	})
2798	.expect("request json"),
2799	))
2800	.expect("request"),
2801	)
2802	.await
2803	.expect("response");
2804
2805	assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
2806	}
2807
2808	#[tokio::test]
2809	async fn http_config_get_redacts_sensitive_values_after_auth() {
2810	let (app, _tmp) = app_with_config(Some("test-token"));
2811	let response = app
2812	.oneshot(
2813	Request::builder()
2814	.method(Method::POST)
2815	.uri("/app")
2816	.header(header::AUTHORIZATION, "Bearer test-token")
2817	.header(header::CONTENT_TYPE, "application/json")
2818	.body(Body::from(
2819	serde_json::to_vec(&AppRequest::ConfigGet {
2820	key: "api_key".to_string(),
2821	})
2822	.expect("request json"),
2823	))
2824	.expect("request"),
2825	)
2826	.await
2827	.expect("response");
2828
2829	assert_eq!(response.status(), StatusCode::OK);
2830	let body = response_body_json(response).await;
2831	assert_eq!(body["data"]["value"], "sk-d***cret");
2832	}
2833
2834	#[tokio::test]
2835	async fn cors_does_not_allow_arbitrary_origins() {
2836	let (app, _tmp) = app_with_config(Some("test-token"));
2837	let response = app
2838	.oneshot(
2839	Request::builder()
2840	.method(Method::GET)
2841	.uri("/healthz")
2842	.header(header::ORIGIN, "https://attacker.example")
2843	.body(Body::empty())
2844	.expect("request"),
2845	)
2846	.await
2847	.expect("response");
2848
2849	assert_eq!(response.status(), StatusCode::OK);
2850	assert!(
2851	response
2852	.headers()
2853	.get(header::ACCESS_CONTROL_ALLOW_ORIGIN)
2854	.is_none()
2855	);
2856	}
2857
2858	#[tokio::test]
2859	async fn build_state_loads_permissions_into_runtime_policy() {
2860	let tmp = tempfile::tempdir().expect("tempdir");
2861	let config_path = tmp.path().join("config.toml");
2862	fs::write(&config_path, "api_key = \"sk-deepseek-secret\"\n").expect("write config");
2863	fs::write(
2864	tmp.path().join("permissions.toml"),
2865	r#"
2866	[[rules]]
2867	tool = "exec_shell"
2868	command = "cargo test"
2869	"#,
2870	)
2871	.expect("write permissions");
2872
2873	let state = build_state(Some(config_path), None).expect("state");
2874	let runtime = state.runtime.read().await;
2875	let decision = runtime
2876	.exec_policy
2877	.check(codewhale_execpolicy::ExecPolicyContext {
2878	command: "cargo test --workspace",
2879	cwd: "/workspace",
2880	tool: Some("exec_shell"),
2881	path: None,
2882	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
2883	sandbox_mode: Some("workspace-write"),
2884	})
2885	.expect("policy check");
2886
2887	assert!(decision.allow);
2888	assert!(decision.requires_approval);
2889	assert_eq!(
2890	decision.matched_rule.as_deref(),
2891	Some("tool=exec_shell command=cargo test")
2892	);
2893	}
2894
2895	#[tokio::test]
2896	async fn config_reload_refreshes_runtime_config_and_exec_policy_from_disk() {
2897	let tmp = tempfile::tempdir().expect("tempdir");
2898	let config_path = tmp.path().join("config.toml");
2899	fs::write(
2900	&config_path,
2901	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
2902	)
2903	.expect("write config");
2904	// No permissions.toml at startup → exec_policy starts empty.
2905	let state = build_state(Some(config_path.clone()), None).expect("state");
2906
2907	// Sanity: initial runtime sees the on-disk model and has no rule.
2908	{
2909	let runtime = state.runtime.read().await;
2910	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-chat"));
2911	let decision = runtime
2912	.exec_policy
2913	.check(codewhale_execpolicy::ExecPolicyContext {
2914	command: "cargo test",
2915	cwd: "/workspace",
2916	tool: Some("exec_shell"),
2917	path: None,
2918	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
2919	sandbox_mode: Some("workspace-write"),
2920	})
2921	.expect("policy check");
2922	assert!(decision.matched_rule.is_none());
2923	}
2924
2925	// Edit both files on disk: new model + a permission rule.
2926	fs::write(
2927	&config_path,
2928	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-reasoner\"\n",
2929	)
2930	.expect("rewrite config");
2931	fs::write(
2932	tmp.path().join("permissions.toml"),
2933	r#"
2934	[[rules]]
2935	tool = "exec_shell"
2936	command = "cargo test"
2937	"#,
2938	)
2939	.expect("write permissions");
2940
2941	// ConfigReload must re-read both files and push them into the
2942	// live Runtime without a restart.
2943	let response =
2944	process_app_request(&state, AppRequest::ConfigReload, AppTransport::Stdio).await;
2945	assert!(response.ok, "reload should succeed");
2946	assert_eq!(response.data["reloaded"], true);
2947
2948	// The shared config lock reflects the new model.
2949	{
2950	let cfg = state.config.read().await;
2951	assert_eq!(cfg.model.as_deref(), Some("deepseek-reasoner"));
2952	}
2953	// The live Runtime reflects both the new model and the new rule.
2954	{
2955	let runtime = state.runtime.read().await;
2956	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-reasoner"));
2957	let decision = runtime
2958	.exec_policy
2959	.check(codewhale_execpolicy::ExecPolicyContext {
2960	command: "cargo test --workspace",
2961	cwd: "/workspace",
2962	tool: Some("exec_shell"),
2963	path: None,
2964	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
2965	sandbox_mode: Some("workspace-write"),
2966	})
2967	.expect("policy check");
2968	assert!(decision.allow);
2969	assert!(decision.requires_approval);
2970	assert_eq!(
2971	decision.matched_rule.as_deref(),
2972	Some("tool=exec_shell command=cargo test")
2973	);
2974	}
2975	}
2976
2977	#[tokio::test]
2978	async fn config_set_propagates_to_runtime_config_without_touching_exec_policy() {
2979	let tmp = tempfile::tempdir().expect("tempdir");
2980	let config_path = tmp.path().join("config.toml");
2981	fs::write(
2982	&config_path,
2983	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
2984	)
2985	.expect("write config");
2986	let state = build_state(Some(config_path.clone()), None).expect("state");
2987
2988	// Set a new model via the API. Only config.toml is touched; no
2989	// permissions.toml exists, so exec_policy must stay empty.
2990	let response = process_app_request(
2991	&state,
2992	AppRequest::ConfigSet {
2993	key: "model".to_string(),
2994	value: "deepseek-reasoner".to_string(),
2995	},
2996	AppTransport::Stdio,
2997	)
2998	.await;
2999	assert!(response.ok, "set should succeed");
3000
3001	// Live runtime sees the new model.
3002	{
3003	let runtime = state.runtime.read().await;
3004	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-reasoner"));
3005	// exec_policy was empty at startup and must remain empty.
3006	let decision = runtime
3007	.exec_policy
3008	.check(codewhale_execpolicy::ExecPolicyContext {
3009	command: "cargo test",
3010	cwd: "/workspace",
3011	tool: Some("exec_shell"),
3012	path: None,
3013	ask_for_approval: codewhale_execpolicy::AskForApproval::UnlessTrusted,
3014	sandbox_mode: Some("workspace-write"),
3015	})
3016	.expect("policy check");
3017	assert!(decision.matched_rule.is_none());
3018	}
3019	// The on-disk file was persisted.
3020	let persisted = fs::read_to_string(&config_path).expect("read config");
3021	assert!(persisted.contains("deepseek-reasoner"));
3022	}
3023
3024	/// A bridge stand-in with no child process: this test only cares about
3025	/// whether the cache slot survives, not about talking to a runtime.
3026	fn sentinel_bridge() -> SharedRuntimeBridge {
3027	Arc::new(Mutex::new(RuntimeBridge {
3028	base_url: "http://127.0.0.1:0".to_string(),
3029	client: codewhale_release::tls::reqwest_client(),
3030	auth_token: None,
3031	child: None,
3032	last_seq_by_thread: HashMap::new(),
3033	}))
3034	}
3035
3036	#[tokio::test]
3037	async fn failed_config_set_keeps_the_stdio_bridge() {
3038	crate::install_test_crypto_provider();
3039	// #4737: `set_value` rejects an invalid value before assigning, so the
3040	// request is a no-op — but `apply_config_update` ran anyway and
3041	// invalidated the cached bridge, dropping the child runtime along with
3042	// its thread map. A single bad value orphaned every in-flight stdio
3043	// thread, behind a response that correctly reported `ok: false`.
3044	//
3045	// Only `set_value` is exercised: an unknown key lands in `extras` and
3046	// succeeds, and `unset_value` has no failing input today, so its
3047	// identical guard has nothing to assert against.
3048	let tmp = tempfile::tempdir().expect("tempdir");
3049	let config_path = tmp.path().join("config.toml");
3050	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
3051	let state = build_state(Some(config_path.clone()), None).expect("state");
3052	*state.runtime_bridge.lock().await = Some(sentinel_bridge());
3053	state
3054	.runtime_thread_map
3055	.lock()
3056	.await
3057	.insert("stdio-1".to_string(), "runtime-1".to_string());
3058
3059	let response = process_app_request(
3060	&state,
3061	AppRequest::ConfigSet {
3062	key: "telemetry".to_string(),
3063	value: "not-a-bool".to_string(),
3064	},
3065	AppTransport::Stdio,
3066	)
3067	.await;
3068	assert!(!response.ok, "invalid value must fail: {response:?}");
3069
3070	assert!(
3071	state.runtime_bridge.lock().await.is_some(),
3072	"bridge must survive a failed config/set",
3073	);
3074	assert_eq!(
3075	state
3076	.runtime_thread_map
3077	.lock()
3078	.await
3079	.get("stdio-1")
3080	.map(String::as_str),
3081	Some("runtime-1"),
3082	"the live thread map must be intact",
3083	);
3084	}
3085
3086	#[tokio::test]
3087	async fn successful_config_set_still_invalidates_the_stdio_bridge() {
3088	crate::install_test_crypto_provider();
3089	// The other half of #4737: a mutation that did happen must still
3090	// rebuild the bridge, or the runtime keeps serving the old config.
3091	let tmp = tempfile::tempdir().expect("tempdir");
3092	let config_path = tmp.path().join("config.toml");
3093	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
3094	let state = build_state(Some(config_path.clone()), None).expect("state");
3095	*state.runtime_bridge.lock().await = Some(sentinel_bridge());
3096
3097	let response = process_app_request(
3098	&state,
3099	AppRequest::ConfigSet {
3100	key: "model".to_string(),
3101	value: "deepseek-reasoner".to_string(),
3102	},
3103	AppTransport::Stdio,
3104	)
3105	.await;
3106	assert!(response.ok, "valid set should succeed: {response:?}");
3107	assert!(
3108	state.runtime_bridge.lock().await.is_none(),
3109	"a successful config change must invalidate the cached bridge",
3110	);
3111	}
3112
3113	/// A stub runtime that records which thread ids turns ran on and how
3114	/// many threads it was asked to mint.
3115	#[derive(Clone)]
3116	struct RecordingRuntime {
3117	created: Arc<std::sync::atomic::AtomicUsize>,
3118	turn_threads: Arc<Mutex<Vec<String>>>,
3119	}
3120
3121	async fn spawn_recording_runtime() -> (String, RecordingRuntime, tokio::task::JoinHandle<()>) {
3122	async fn create_thread(State(f): State<RecordingRuntime>) -> Json<Value> {
3123	f.created.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
3124	Json(json!({ "id": "thr_minted" }))
3125	}
3126	async fn create_turn(
3127	State(f): State<RecordingRuntime>,
3128	AxumPath(thread_id): AxumPath<String>,
3129	) -> Json<Value> {
3130	f.turn_threads.lock().await.push(thread_id);
3131	Json(json!({ "turn": { "id": "turn_recorded" } }))
3132	}
3133	async fn thread_events(
3134	AxumPath(_thread_id): AxumPath<String>,
3135	) -> ([(header::HeaderName, &'static str); 1], String) {
3136	(
3137	[(header::CONTENT_TYPE, "text/event-stream")],
3138	sse_frame(
3139	"turn.completed",
3140	json!({
3141	"seq": 1,
3142	"turn_id": "turn_recorded",
3143	"payload": { "turn": { "status": "completed" } }
3144	}),
3145	),
3146	)
3147	}
3148
3149	let fixture = RecordingRuntime {
3150	created: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
3151	turn_threads: Arc::new(Mutex::new(Vec::new())),
3152	};
3153	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
3154	.await
3155	.expect("bind recording runtime");
3156	let addr = listener.local_addr().expect("listener addr");
3157	let app = Router::new()
3158	.route("/v1/threads", post(create_thread))
3159	.route("/v1/threads/{id}/turns", post(create_turn))
3160	.route("/v1/threads/{id}/events", get(thread_events))
3161	.with_state(fixture.clone());
3162	let server = tokio::spawn(async move {
3163	axum::serve(listener, app)
3164	.await
3165	.expect("serve recording runtime");
3166	});
3167	(format!("http://{addr}"), fixture, server)
3168	}
3169
3170	#[tokio::test]
3171	async fn config_update_keeps_the_stdio_thread_mapping() {
3172	crate::install_test_crypto_provider();
3173	// #6246: `apply_config_update` rebuilds the bridge child, but runtime
3174	// threads are durable — the fresh child resolves the same ids. The
3175	// bug dropped the stdio→runtime map with the old bridge, so the next
3176	// `thread/message` silently minted a new runtime thread instead of
3177	// resuming the mapped one.
3178	let (base_url, fixture, server) = spawn_recording_runtime().await;
3179	let (state, _tmp) = capability_test_state();
3180	state
3181	.runtime_thread_map
3182	.lock()
3183	.await
3184	.insert("stdio-keep".to_string(), "thr_keep".to_string());
3185	seed_bridge_at(&state, base_url.clone()).await;
3186
3187	// An unrelated config snapshot still rebuilds the bridge child.
3188	let snapshot = state.config.read().await.clone();
3189	apply_config_update(&state, snapshot, None, false).await;
3190	assert!(
3191	state.runtime_bridge.lock().await.is_none(),
3192	"config update must drop the cached bridge",
3193	);
3194	assert_eq!(
3195	state
3196	.runtime_thread_map
3197	.lock()
3198	.await
3199	.get("stdio-keep")
3200	.map(String::as_str),
3201	Some("thr_keep"),
3202	"the thread mapping must survive the bridge rebuild",
3203	);
3204
3205	// The fresh child (seeded here in place of `RuntimeBridge::start`,
3206	// which cannot spawn in-process) must resume the mapped thread.
3207	seed_bridge_at(&state, base_url).await;
3208	let result = dispatch_stdio_request(
3209	&state,
3210	"thread/message",
3211	json!({ "thread_id": "stdio-keep", "input": "next" }),
3212	)
3213	.await
3214	.expect("thread/message on the mapped thread");
3215
3216	assert_eq!(result.result["status"], json!("accepted"));
3217	assert_eq!(
3218	fixture.turn_threads.lock().await.as_slice(),
3219	["thr_keep".to_string()],
3220	"the turn must run on the pre-existing runtime thread",
3221	);
3222	assert_eq!(
3223	fixture.created.load(std::sync::atomic::Ordering::SeqCst),
3224	0,
3225	"no new runtime thread may be minted for a mapped stdio thread",
3226	);
3227
3228	server.abort();
3229	let _ = server.await;
3230	}
3231
3232	#[tokio::test]
3233	async fn config_unset_propagates_to_runtime_config() {
3234	let tmp = tempfile::tempdir().expect("tempdir");
3235	let config_path = tmp.path().join("config.toml");
3236	fs::write(
3237	&config_path,
3238	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
3239	)
3240	.expect("write config");
3241	let state = build_state(Some(config_path.clone()), None).expect("state");
3242
3243	// Sanity: runtime starts with the on-disk model.
3244	{
3245	let runtime = state.runtime.read().await;
3246	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-chat"));
3247	}
3248
3249	// Unset the model via the API. This walks a separate code path
3250	// from ConfigSet (unset_value + update_config), so it needs its
3251	// own regression coverage.
3252	let response = process_app_request(
3253	&state,
3254	AppRequest::ConfigUnset {
3255	key: "model".to_string(),
3256	},
3257	AppTransport::Stdio,
3258	)
3259	.await;
3260	assert!(response.ok, "unset should succeed");
3261
3262	// Live runtime sees the cleared model.
3263	{
3264	let runtime = state.runtime.read().await;
3265	assert!(runtime.config.model.is_none());
3266	}
3267	// Shared config lock agrees.
3268	{
3269	let cfg = state.config.read().await;
3270	assert!(cfg.model.is_none());
3271	}
3272	// The on-disk file no longer carries the model value.
3273	let persisted = fs::read_to_string(&config_path).expect("read config");
3274	assert!(!persisted.contains("deepseek-chat"));
3275	}
3276
3277	#[tokio::test]
3278	async fn config_reload_returns_error_when_disk_config_is_invalid() {
3279	let tmp = tempfile::tempdir().expect("tempdir");
3280	let config_path = tmp.path().join("config.toml");
3281	fs::write(
3282	&config_path,
3283	"api_key = \"sk-deepseek-secret\"\nmodel = \"deepseek-chat\"\n",
3284	)
3285	.expect("write config");
3286	let state = build_state(Some(config_path.clone()), None).expect("state");
3287
3288	// Corrupt the on-disk config so ConfigStore::load fails to parse.
3289	fs::write(&config_path, "api_key = \"unterminated\n").expect("corrupt config");
3290
3291	let response =
3292	process_app_request(&state, AppRequest::ConfigReload, AppTransport::Stdio).await;
3293	assert!(!response.ok, "reload of corrupt config must fail");
3294	let err = response.data["error"]
3295	.as_str()
3296	.expect("error message present")
3297	.to_string();
3298	assert!(
3299	err.contains("failed to load config"),
3300	"error should mention load failure, got: {err}"
3301	);
3302
3303	// Live state is untouched: the early-return on load error must
3304	// not have clobbered runtime.config or state.config.
3305	{
3306	let runtime = state.runtime.read().await;
3307	assert_eq!(runtime.config.model.as_deref(), Some("deepseek-chat"));
3308	}
3309	{
3310	let cfg = state.config.read().await;
3311	assert_eq!(cfg.model.as_deref(), Some("deepseek-chat"));
3312	}
3313	}
3314
3315	async fn seed_test_bridge(state: &AppState) -> SharedRuntimeBridge {
3316	let bridge = Arc::new(Mutex::new(RuntimeBridge::from_base_url_for_test(
3317	"http://127.0.0.1:9".to_string(),
3318	)));
3319	*state.runtime_bridge.lock().await = Some(bridge.clone());
3320	bridge
3321	}
3322
3323	#[tokio::test]
3324	async fn config_set_invalidates_cached_stdio_bridge() {
3325	let tmp = tempfile::tempdir().expect("tempdir");
3326	let config_path = tmp.path().join("config.toml");
3327	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
3328	let state = build_state(Some(config_path), None).expect("state");
3329	seed_test_bridge(&state).await;
3330
3331	let response = process_app_request(
3332	&state,
3333	AppRequest::ConfigSet {
3334	key: "model".to_string(),
3335	value: "deepseek-reasoner".to_string(),
3336	},
3337	AppTransport::Stdio,
3338	)
3339	.await;
3340	assert!(response.ok, "set should succeed");
3341
3342	// The cached bridge child must be dropped so the next stdio request
3343	// spawns a fresh runtime that reads the persisted config.
3344	assert!(state.runtime_bridge.lock().await.is_none());
3345	}
3346
3347	#[tokio::test]
3348	async fn config_reload_invalidates_cached_stdio_bridge() {
3349	let tmp = tempfile::tempdir().expect("tempdir");
3350	let config_path = tmp.path().join("config.toml");
3351	fs::write(&config_path, "model = \"deepseek-chat\"\n").expect("write config");
3352	let state = build_state(Some(config_path), None).expect("state");
3353	seed_test_bridge(&state).await;
3354
3355	let response =
3356	process_app_request(&state, AppRequest::ConfigReload, AppTransport::Stdio).await;
3357	assert!(response.ok, "reload should succeed");
3358
3359	assert!(state.runtime_bridge.lock().await.is_none());
3360	}
3361
3362	#[tokio::test]
3363	async fn stdio_bridge_invalidation_not_blocked_by_in_flight_turn() {
3364	let (state, _tmp) = capability_test_state();
3365	let bridge = seed_test_bridge(&state).await;
3366
3367	// Simulate a long streaming turn holding the inner bridge lock.
3368	let _in_flight = bridge.lock().await;
3369
3370	// Invalidation only touches the cache slot, so it must complete
3371	// without waiting for the in-flight turn to release the bridge.
3372	tokio::time::timeout(Duration::from_secs(1), invalidate_runtime_bridge(&state))
3373	.await
3374	.expect("invalidation must not wait on bridge traffic");
3375	assert!(state.runtime_bridge.lock().await.is_none());
3376	}
3377
3378	#[tokio::test]
3379	async fn runtime_read_paths_run_concurrently() {
3380	// Tool/status/mcp handlers take read guards; two must coexist so a
3381	// long-running tool call cannot serialize unrelated requests. With
3382	// the old `Mutex<Runtime>` this pattern would deadlock.
3383	let (state, _tmp) = capability_test_state();
3384	let first = state.runtime.read().await;
3385	let second = state.runtime.read().await;
3386	assert!(first.app_status().ok);
3387	assert!(second.app_status().ok);
3388	}
3389
3390	#[tokio::test]
3391	async fn health_probes_advertise_legacy_deepseek_service_name() {
3392	// External probes still key off the DeepSeek-era service name; both
3393	// transports must serve it from the single compat shim.
3394	let (app, _tmp) = app_with_config(None);
3395	let response = app
3396	.oneshot(
3397	Request::builder()
3398	.method(Method::GET)
3399	.uri("/healthz")
3400	.body(Body::empty())
3401	.expect("request"),
3402	)
3403	.await
3404	.expect("response");
3405	let body = response_body_json(response).await;
3406	assert_eq!(body["service"], legacy_deepseek_compat::SERVICE_NAME);
3407	assert_eq!(body["service"], "deepseek-app-server");
3408
3409	let (state, _tmp) = capability_test_state();
3410	let stdio = dispatch_stdio_request(&state, "healthz", json!({}))
3411	.await
3412	.expect("stdio healthz");
3413	assert_eq!(
3414	stdio.result["service"],
3415	legacy_deepseek_compat::SERVICE_NAME
3416	);
3417	}
3418
3419	#[test]
3420	fn non_loopback_bind_without_auth_fails_fast() {
3421	let options = AppServerOptions {
3422	listen: "0.0.0.0:8787".parse().expect("socket addr"),
3423	config_path: None,
3424	auth_token: None,
3425	insecure_no_auth: false,
3426	cors_origins: Vec::new(),
3427	};
3428
3429	let err =
3430	resolve_auth_token(&options).expect_err("non-loopback generated auth should fail");
3431	assert!(err.to_string().contains("without explicit auth token"));
3432	}
3433
3434	#[tokio::test]
3435	async fn stdio_transport_redacts_config_get_secrets() {
3436	let tmp = tempfile::tempdir().expect("tempdir");
3437	let config_path = tmp.path().join("config.toml");
3438	fs::write(&config_path, "").expect("write config");
3439	let state = build_state(Some(config_path), None).expect("state");
3440	{
3441	let mut cfg = state.config.write().await;
3442	cfg.api_key = Some("sk-deepseek-secret".to_string());
3443	}
3444
3445	let response = process_app_request(
3446	&state,
3447	AppRequest::ConfigGet {
3448	key: "api_key".to_string(),
3449	},
3450	AppTransport::Stdio,
3451	)
3452	.await;
3453
3454	assert_eq!(response.data["value"], "sk-d***cret");
3455	}
3456
3457	#[tokio::test]
3458	async fn stdio_thread_goal_methods_round_trip_persisted_goal() {
3459	let tmp = tempfile::tempdir().expect("tempdir");
3460	let config_path = tmp.path().join("config.toml");
3461	fs::write(&config_path, "").expect("write config");
3462	let state = build_state(Some(config_path), None).expect("state");
3463
3464	let capabilities = dispatch_stdio_request(&state, "thread/capabilities", json!({}))
3465	.await
3466	.expect("thread capabilities");
3467	assert!(
3468	capabilities.result["methods"]
3469	.as_array()
3470	.expect("methods")
3471	.iter()
3472	.any(\|method\| method == "thread/goal/set")
3473	);
3474
3475	let started = dispatch_stdio_request(&state, "thread/start", json!({}))
3476	.await
3477	.expect("start thread");
3478	let thread_id = started.result["thread_id"]
3479	.as_str()
3480	.expect("thread id")
3481	.to_string();
3482
3483	let set = dispatch_stdio_request(
3484	&state,
3485	"thread/goal/set",
3486	json!({
3487	"thread_id": thread_id,
3488	"objective": "Release 0.8.59",
3489	"token_budget": 59000
3490	}),
3491	)
3492	.await
3493	.expect("set goal");
3494	assert_eq!(set.result["status"], "ok");
3495	assert_eq!(set.result["goal"]["objective"], "Release 0.8.59");
3496	assert_eq!(set.result["goal"]["status"], "active");
3497
3498	let got = dispatch_stdio_request(
3499	&state,
3500	"thread/goal/get",
3501	json!({
3502	"thread_id": thread_id
3503	}),
3504	)
3505	.await
3506	.expect("get goal");
3507	assert_eq!(got.result["goal"]["token_budget"], 59000);
3508
3509	let cleared = dispatch_stdio_request(
3510	&state,
3511	"thread/goal/clear",
3512	json!({
3513	"thread_id": thread_id
3514	}),
3515	)
3516	.await
3517	.expect("clear goal");
3518	assert_eq!(cleared.result["status"], "cleared");
3519	assert_eq!(cleared.result["data"]["cleared"], true);
3520	}
3521
3522	#[tokio::test]
3523	async fn stdio_resume_of_missing_thread_fails_without_clobbering_the_hint() {
3524	let tmp = tempfile::tempdir().expect("tempdir");
3525	let config_path = tmp.path().join("config.toml");
3526	fs::write(&config_path, "").expect("write config");
3527	let state = build_state(Some(config_path), None).expect("state");
3528
3529	// A cached hint for a thread the runtime no longer knows: the exact
3530	// clobber scenario from #5171.
3531	let workspace = tmp.path().join("ws");
3532	{
3533	let mut hints = state.stdio_thread_hints.lock().await;
3534	hints.insert(
3535	"ghost-thread".to_string(),
3536	RuntimeThreadHint {
3537	model: Some("deepseek-v4-pro".to_string()),
3538	workspace: Some(workspace.clone()),
3539	},
3540	);
3541	}
3542
3543	let err = dispatch_stdio_request(
3544	&state,
3545	"thread/resume",
3546	json!({ "thread_id": "ghost-thread" }),
3547	)
3548	.await
3549	.expect_err("resuming a missing thread must fail with a named not-found error");
3550	assert_eq!(err.code, -32004);
3551	assert!(err.message.contains("ghost-thread"), "{}", err.message);
3552
3553	let fork_err = dispatch_stdio_request(
3554	&state,
3555	"thread/fork",
3556	json!({ "thread_id": "ghost-thread" }),
3557	)
3558	.await
3559	.expect_err("forking a missing thread must fail with a named not-found error");
3560	assert_eq!(fork_err.code, -32004);
3561
3562	let hints = state.stdio_thread_hints.lock().await;
3563	let hint = hints.get("ghost-thread").expect("cached hint survives");
3564	assert_eq!(hint.model.as_deref(), Some("deepseek-v4-pro"));
3565	assert_eq!(hint.workspace.as_deref(), Some(workspace.as_path()));
3566	}
3567
3568	fn sse_frame(event: &str, payload: Value) -> String {
3569	format!("event: {event}\ndata: {payload}\n\n")
3570	}
3571	/// A runtime whose turn never ends on its own — only an interrupt stops
3572	/// it. That is the shape of the runaway turn this protects against.
3573	async fn spawn_uninterruptible_until_asked_runtime() -> (
3574	String,
3575	Arc<tokio::sync::Notify>,
3576	tokio::task::JoinHandle<()>,
3577	) {
3578	use axum::body::Body;
3579	use axum::extract::Path as AxumPath;
3580
3581	let interrupted = Arc::new(tokio::sync::Notify::new());
3582
3583	async fn create_turn(AxumPath(_thread_id): AxumPath<String>) -> Json<Value> {
3584	Json(json!({ "turn": { "id": "turn_runaway" } }))
3585	}
3586	async fn create_thread() -> Json<Value> {
3587	Json(json!({ "id": "thr_runaway" }))
3588	}
3589	async fn interrupt(
3590	State(notify): State<Arc<tokio::sync::Notify>>,
3591	AxumPath((_thread_id, _turn_id)): AxumPath<(String, String)>,
3592	) -> Json<Value> {
3593	notify.notify_waiters();
3594	Json(json!({ "ok": true }))
3595	}
3596	async fn thread_events(
3597	State(notify): State<Arc<tokio::sync::Notify>>,
3598	AxumPath(_thread_id): AxumPath<String>,
3599	) -> ([(header::HeaderName, &'static str); 1], Body) {
3600	// Hold the event response open until something interrupts the
3601	// turn. Nothing else can end it, which is the point.
3602	notify.notified().await;
3603	let body = [
3604	sse_frame(
3605	"item.delta",
3606	json!({
3607	"seq": 1,
3608	"turn_id": "turn_runaway",
3609	"payload": { "kind": "agent_message", "delta": "thinking" }
3610	}),
3611	),
3612	sse_frame(
3613	"turn.completed",
3614	json!({
3615	"seq": 2,
3616	"turn_id": "turn_runaway",
3617	"payload": { "turn": { "status": "interrupted" } }
3618	}),
3619	),
3620	]
3621	.concat();
3622	(
3623	[(header::CONTENT_TYPE, "text/event-stream")],
3624	Body::from(body),
3625	)
3626	}
3627
3628	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
3629	.await
3630	.expect("bind test listener");
3631	let addr = listener.local_addr().expect("listener addr");
3632	let app = Router::new()
3633	.route("/v1/threads", post(create_thread))
3634	.route("/v1/threads/{thread_id}/turns", post(create_turn))
3635	.route(
3636	"/v1/threads/{thread_id}/turns/{turn_id}/interrupt",
3637	post(interrupt),
3638	)
3639	.route("/v1/threads/{thread_id}/events", get(thread_events))
3640	.with_state(interrupted.clone());
3641	let server = tokio::spawn(async move {
3642	axum::serve(listener, app)
3643	.await
3644	.expect("serve test runtime");
3645	});
3646	(format!("http://{addr}"), interrupted, server)
3647	}
3648
3649	#[tokio::test]
3650	async fn interrupt_stops_a_turn_that_would_otherwise_stream_forever() {
3651	let (base_url, _notify, server) = spawn_uninterruptible_until_asked_runtime().await;
3652	let (state, _tmp) = capability_test_state();
3653	*state.runtime_bridge.lock().await = Some(Arc::new(Mutex::new(
3654	RuntimeBridge::from_base_url_for_test(base_url),
3655	)));
3656
3657	let (client, server_side) = tokio::io::duplex(16 * 1024);
3658	let (client_reader, mut client_writer) = tokio::io::split(client);
3659
3660	let loop_state = state.clone();
3661	let loop_handle = tokio::spawn(async move {
3662	let (rx, tx) = tokio::io::split(server_side);
3663	run_stdio_loop(
3664	&loop_state,
3665	BufReader::new(rx).lines(),
3666	tx,
3667	StdioLoopPolicy::process_stdio(),
3668	None::<()>,
3669	)
3670	.await
3671	});
3672
3673	// Start the runaway turn.
3674	client_writer
3675	.write_all(
3676	b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"thread/message\",\
3677	\"params\":{\"thread_id\":\"thr_a\",\"input\":\"go\"}}\n",
3678	)
3679	.await
3680	.expect("send thread/message");
3681
3682	// Wait until the turn is genuinely in flight before cancelling, so the
3683	// test exercises mid-stream cancellation rather than a race.
3684	tokio::time::timeout(Duration::from_secs(10), async {
3685	loop {
3686	if state.in_flight_turns.lock().await.contains_key("thr_a") {
3687	return;
3688	}
3689	tokio::time::sleep(Duration::from_millis(10)).await;
3690	}
3691	})
3692	.await
3693	.expect("turn should register itself as in flight");
3694
3695	// The read loop must accept this while the turn holds the bridge.
3696	client_writer
3697	.write_all(
3698	b"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"thread/interrupt\",\
3699	\"params\":{\"thread_id\":\"thr_a\"}}\n",
3700	)
3701	.await
3702	.expect("send thread/interrupt");
3703	client_writer
3704	.write_all(b"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"shutdown\"}\n")
3705	.await
3706	.expect("send shutdown");
3707
3708	let finished = tokio::time::timeout(Duration::from_secs(20), loop_handle)
3709	.await
3710	.expect("the loop must exit rather than hang on the runaway turn");
3711	finished.expect("join loop").expect("loop result");
3712
3713	let mut output = String::new();
3714	let mut lines = BufReader::new(client_reader);
3715	lines
3716	.read_to_string(&mut output)
3717	.await
3718	.expect("read stdio output");
3719
3720	let responses: Vec<Value> = output
3721	.lines()
3722	.filter_map(\|line\| serde_json::from_str::<Value>(line).ok())
3723	.collect();
3724	let by_id = \|id: u64\| {
3725	responses
3726	.iter()
3727	.find(\|value\| value["id"] == json!(id))
3728	.unwrap_or_else(\|\| panic!("no response for id {id} in {output}"))
3729	.clone()
3730	};
3731
3732	// The turn ended as interrupted rather than running to completion.
3733	assert!(
3734	by_id(1)["error"].is_object(),
3735	"the interrupted turn should report an error, got: {}",
3736	by_id(1)
3737	);
3738	assert_eq!(by_id(2)["result"]["interrupted"], json!(true));
3739	assert_eq!(by_id(3)["result"]["status"], json!("stopped"));
3740
3741	server.abort();
3742	let _ = server.await;
3743	}
3744
3745	#[tokio::test]
3746	async fn interrupting_an_idle_thread_is_not_an_error() {
3747	let (state, _tmp) = capability_test_state();
3748	let response = dispatch_stdio_request(
3749	&state,
3750	"thread/interrupt",
3751	json!({ "thread_id": "thr_nothing_running" }),
3752	)
3753	.await
3754	.expect("interrupt dispatch");
3755	assert_eq!(response.result["interrupted"], json!(false));
3756	}
3757
3758	#[tokio::test]
3759	async fn output_cap_bridge_checks_support_before_creation_and_forwards_each_surface() {
3760	use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
3761	#[derive(Clone)]
3762	struct Fixture {
3763	supported: Arc<AtomicBool>,
3764	created: Arc<AtomicUsize>,
3765	requests: Arc<Mutex<Vec<Value>>>,
3766	}
3767	async fn info(State(f): State<Fixture>, headers: axum::http::HeaderMap) -> Json<Value> {
3768	assert_eq!(
3769	headers.get(header::AUTHORIZATION).unwrap(),
3770	"Bearer fixture-output-cap"
3771	);
3772	Json(
3773	json!({"capabilities":{"turn_output_token_limit":f.supported.load(Ordering::SeqCst)}}),
3774	)
3775	}
3776	async fn providers() -> Json<Value> {
3777	Json(
3778	json!({"current":"custom","providers":[{"id":"custom","default_model":"fixture-model"}]}),
3779	)
3780	}
3781	async fn models() -> Json<Value> {
3782	Json(
3783	json!({"models":[{"id":"fixture-model","output_token_limit":"supported"},{"id":"uncapped-transport","output_token_limit":"unsupported"}]}),
3784	)
3785	}
3786	async fn create_thread(State(f): State<Fixture>) -> Json<Value> {
3787	let n = f.created.fetch_add(1, Ordering::SeqCst);
3788	Json(json!({"id":format!("thr_cap_{n}")}))
3789	}
3790	async fn create_turn(State(f): State<Fixture>, Json(body): Json<Value>) -> Json<Value> {
3791	f.requests.lock().await.push(body);
3792	Json(json!({"turn":{"id":"turn_cap"}}))
3793	}
3794	async fn events() -> impl IntoResponse {
3795	(
3796	[(header::CONTENT_TYPE, "text/event-stream")],
3797	sse_frame(
3798	"turn.completed",
3799	json!({
3800	"seq":1,"turn_id":"turn_cap","payload":{"turn":{"status":"completed"}}
3801	}),
3802	),
3803	)
3804	}
3805	let fixture = Fixture {
3806	supported: Arc::new(AtomicBool::new(false)),
3807	created: Arc::new(AtomicUsize::new(0)),
3808	requests: Arc::new(Mutex::new(Vec::new())),
3809	};
3810	let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
3811	let addr = listener.local_addr().unwrap();
3812	let router = Router::new()
3813	.route("/v1/runtime/info", get(info))
3814	.route("/v1/providers", get(providers))
3815	.route("/v1/providers/custom/models", get(models))
3816	.route("/v1/threads", post(create_thread))
3817	.route("/v1/threads/{id}/turns", post(create_turn))
3818	.route("/v1/threads/{id}/events", get(events))
3819	.with_state(fixture.clone());
3820	let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
3821	let (state, _tmp) = capability_test_state();
3822	let mut bridge = RuntimeBridge::from_base_url_for_test(format!("http://{addr}"));
3823	bridge.auth_token = Some("fixture-output-cap".into());
3824	*state.runtime_bridge.lock().await = Some(Arc::new(Mutex::new(bridge)));
3825	let result = dispatch_stdio_request(
3826	&state,
3827	"prompt/run",
3828	json!({"prompt":"review","maxOutputTokens":1500}),
3829	)
3830	.await;
3831	assert!(result.unwrap_err().message.contains("does not support"));
3832	assert_eq!(fixture.created.load(Ordering::SeqCst), 0);
3833	assert!(fixture.requests.lock().await.is_empty());
3834	fixture.supported.store(true, Ordering::SeqCst);
3835	for model in ["auto", "uncapped-transport", "unknown-model"] {
3836	assert!(
3837	dispatch_stdio_request(
3838	&state,
3839	"prompt/run",
3840	json!({"prompt":"review","model":model,"maxOutputTokens":1500})
3841	)
3842	.await
3843	.is_err()
3844	);
3845	}
3846	assert_eq!(fixture.created.load(Ordering::SeqCst), 0);
3847	for (method, params) in [
3848	(
3849	"prompt/run",
3850	json!({"prompt":"review","maxOutputTokens":1500}),
3851	),
3852	(
3853	"thread/message",
3854	json!({"thread_id":"stdio-cap","input":"review","maxOutputTokens":1500}),
3855	),
3856	(
3857	"thread/request",
3858	json!({"kind":"message","thread_id":"request-cap","input":"review","maxOutputTokens":1500}),
3859	),
3860	] {
3861	dispatch_stdio_request(&state, method, params)
3862	.await
3863	.expect("existing app-server caller forwards allowance");
3864	}
3865	run_http_thread_message(
3866	&state,
3867	"http-cap".into(),
3868	"review".into(),
3869	Vec::new(),
3870	std::num::NonZeroU32::new(1500),
3871	)
3872	.await
3873	.unwrap();
3874	let requests = fixture.requests.lock().await.clone();
3875	assert_eq!(requests.len(), 4);
3876	assert!(
3877	requests
3878	.iter()
3879	.all(\|request\| request["maxOutputTokens"] == 1500)
3880	);
3881	let count = fixture.created.load(Ordering::SeqCst);
3882	for invalid in [
3883	json!(0),
3884	json!(-1),
3885	json!(1.5),
3886	json!("1500"),
3887	json!(4_294_967_296u64),
3888	] {
3889	assert!(
3890	dispatch_stdio_request(
3891	&state,
3892	"prompt/run",
3893	json!({"prompt":"review","maxOutputTokens":invalid})
3894	)
3895	.await
3896	.is_err()
3897	);
3898	}
3899	assert_eq!(fixture.created.load(Ordering::SeqCst), count);
3900	assert_eq!(fixture.requests.lock().await.len(), 4);
3901	server.abort();
3902	let _ = server.await;
3903	}
3904
3905	#[tokio::test]
3906	async fn stdio_runtime_bridge_streams_response_delta_events() {
3907	async fn create_turn(AxumPath(thread_id): AxumPath<String>) -> Json<Value> {
3908	Json(json!({
3909	"thread": { "id": thread_id },
3910	"turn": { "id": "turn_test" },
3911	}))
3912	}
3913
3914	async fn thread_events(
3915	AxumPath(thread_id): AxumPath<String>,
3916	Query(query): Query<HashMap<String, String>>,
3917	) -> ([(header::HeaderName, &'static str); 1], String) {
3918	assert_eq!(thread_id, "thr_test");
3919	assert_eq!(query.get("since_seq").map(String::as_str), Some("0"));
3920
3921	let body = [
3922	sse_frame(
3923	"item.delta",
3924	json!({
3925	"seq": 1,
3926	"turn_id": "turn_test",
3927	"payload": {
3928	"kind": "agent_message",
3929	"delta": "hello"
3930	}
3931	}),
3932	),
3933	sse_frame(
3934	"turn.completed",
3935	json!({
3936	"seq": 2,
3937	"turn_id": "turn_test",
3938	"payload": {
3939	"turn": {
3940	"status": "completed"
3941	}
3942	}
3943	}),
3944	),
3945	]
3946	.concat();
3947
3948	([(header::CONTENT_TYPE, "text/event-stream")], body)
3949	}
3950
3951	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
3952	.await
3953	.expect("bind test listener");
3954	let addr = listener.local_addr().expect("listener addr");
3955	let app = Router::new()
3956	.route("/v1/threads/{thread_id}/turns", post(create_turn))
3957	.route("/v1/threads/{thread_id}/events", get(thread_events));
3958
3959	let server = tokio::spawn(async move {
3960	axum::serve(listener, app)
3961	.await
3962	.expect("serve test runtime");
3963	});
3964
3965	let mut bridge = RuntimeBridge::from_base_url_for_test(format!("http://{addr}"));
3966	let (mut reader, mut writer) = tokio::io::duplex(4096);
3967
3968	let result = bridge
3969	.message_thread("thr_test", "hello", &[], None, &mut writer, None, None)
3970	.await
3971	.expect("message_thread should succeed");
3972	drop(writer);
3973
3974	let mut stdout = Vec::new();
3975	reader
3976	.read_to_end(&mut stdout)
3977	.await
3978	.expect("read stdio output");
3979	server.abort();
3980	let _ = server.await;
3981
3982	let lines: Vec<Value> = String::from_utf8(stdout)
3983	.expect("utf8 output")
3984	.lines()
3985	.map(\|line\| serde_json::from_str(line).expect("json line"))
3986	.collect();
3987
3988	assert_eq!(
3989	result.get("status").and_then(Value::as_str),
3990	Some("accepted")
3991	);
3992	assert_eq!(
3993	result.pointer("/data/turn_id").and_then(Value::as_str),
3994	Some("turn_test")
3995	);
3996	assert_eq!(bridge.last_seq_by_thread.get("thr_test"), Some(&2));
3997
3998	let event_types: Vec<&str> = lines
3999	.iter()
4000	.map(\|line\| {
4001	line.get("type")
4002	.and_then(Value::as_str)
4003	.expect("event type")
4004	})
4005	.collect();
4006	assert_eq!(
4007	event_types,
4008	vec!["response_start", "response_delta", "response_end"]
4009	);
4010	assert_eq!(lines[1]["delta"], "hello");
4011	}
4012
4013	#[tokio::test]
4014	async fn stdio_runtime_bridge_applies_thread_start_hints() {
4015	async fn create_thread(Json(body): Json<Value>) -> Json<Value> {
4016	assert_eq!(body["model"], "deepseek-v4");
4017	assert_eq!(body["workspace"], "/tmp/codewhale-stdio");
4018	Json(json!({
4019	"id": "thr_runtime",
4020	"model": body["model"].clone(),
4021	"workspace": body["workspace"].clone(),
4022	}))
4023	}
4024
4025	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4026	.await
4027	.expect("bind test listener");
4028	let addr = listener.local_addr().expect("listener addr");
4029	let app = Router::new().route("/v1/threads", post(create_thread));
4030
4031	let server = tokio::spawn(async move {
4032	axum::serve(listener, app)
4033	.await
4034	.expect("serve test runtime");
4035	});
4036
4037	let mut bridge = RuntimeBridge::from_base_url_for_test(format!("http://{addr}"));
4038	let mut thread_map = HashMap::new();
4039	let runtime_id = bridge
4040	.ensure_runtime_thread(
4041	&mut thread_map,
4042	"legacy_thread",
4043	Some(RuntimeThreadHint {
4044	model: Some("deepseek-v4".to_string()),
4045	workspace: Some(PathBuf::from("/tmp/codewhale-stdio")),
4046	}),
4047	)
4048	.await
4049	.expect("runtime thread");
4050	server.abort();
4051	let _ = server.await;
4052
4053	assert_eq!(runtime_id, "thr_runtime");
4054	assert_eq!(
4055	thread_map.get("legacy_thread").map(String::as_str),
4056	Some("thr_runtime")
4057	);
4058	}
4059
4060	// ── prompt routing runs a real turn ────────────────────────────────
4061	//
4062	// `/prompt`, `prompt/request` and `prompt/run` used to return HTTP 200
4063	// with a stringified echo of the caller's own routing metadata, having
4064	// called no model at all. These stand up the in-crate stub runtime and
4065	// assert the response is what the model streamed — not an echo — and
4066	// that an unreachable runtime is an explicit typed failure.
4067
4068	/// Prompts the stub runtime was actually asked to run.
4069	type StubPrompts = Arc<Mutex<Vec<String>>>;
4070
4071	/// A minimal but honest runtime: it creates threads, starts turns, and
4072	/// streams `agent_message` deltas followed by `turn.completed`.
4073	async fn spawn_stub_runtime() -> (String, StubPrompts, tokio::task::JoinHandle<()>) {
4074	async fn create_thread(Json(body): Json<Value>) -> Json<Value> {
4075	Json(json!({
4076	"id": "thr_stub",
4077	"model": body["model"].as_str().unwrap_or("stub-model-v1"),
4078	}))
4079	}
4080
4081	async fn create_turn(
4082	State(prompts): State<StubPrompts>,
4083	AxumPath(thread_id): AxumPath<String>,
4084	Json(body): Json<Value>,
4085	) -> Json<Value> {
4086	prompts
4087	.lock()
4088	.await
4089	.push(body["prompt"].as_str().unwrap_or_default().to_string());
4090	Json(json!({
4091	"thread": { "id": thread_id, "model": "stub-model-v1" },
4092	"turn": { "id": "turn_stub" },
4093	}))
4094	}
4095
4096	async fn thread_events(
4097	AxumPath(_thread_id): AxumPath<String>,
4098	) -> ([(header::HeaderName, &'static str); 1], String) {
4099	let body = [
4100	sse_frame(
4101	"item.delta",
4102	json!({
4103	"seq": 1,
4104	"turn_id": "turn_stub",
4105	"payload": { "kind": "agent_message", "delta": "the answer" }
4106	}),
4107	),
4108	sse_frame(
4109	"item.delta",
4110	json!({
4111	"seq": 2,
4112	"turn_id": "turn_stub",
4113	"payload": { "kind": "agent_message", "delta": " is 4" }
4114	}),
4115	),
4116	sse_frame(
4117	"turn.completed",
4118	json!({
4119	"seq": 3,
4120	"turn_id": "turn_stub",
4121	"payload": { "turn": { "status": "completed" } }
4122	}),
4123	),
4124	]
4125	.concat();
4126	([(header::CONTENT_TYPE, "text/event-stream")], body)
4127	}
4128
4129	let prompts: StubPrompts = Arc::new(Mutex::new(Vec::new()));
4130	let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4131	.await
4132	.expect("bind stub runtime");
4133	let addr = listener.local_addr().expect("listener addr");
4134	let app = Router::new()
4135	.route("/v1/threads", post(create_thread))
4136	.route("/v1/threads/{thread_id}/turns", post(create_turn))
4137	.route("/v1/threads/{thread_id}/events", get(thread_events))
4138	.with_state(prompts.clone());
4139	let server = tokio::spawn(async move {
4140	let _ = axum::serve(listener, app).await;
4141	});
4142	(format!("http://{addr}"), prompts, server)
4143	}
4144
4145	async fn seed_bridge_at(state: &AppState, base_url: String) -> SharedRuntimeBridge {
4146	let bridge = Arc::new(Mutex::new(RuntimeBridge::from_base_url_for_test(base_url)));
4147	*state.runtime_bridge.lock().await = Some(bridge.clone());
4148	bridge
4149	}
4150
4151	#[tokio::test]
4152	async fn prompt_request_executes_a_genuine_model_turn() {
4153	let (state, _tmp) = capability_test_state();
4154	let (base_url, prompts, server) = spawn_stub_runtime().await;
4155	seed_bridge_at(&state, base_url).await;
4156
4157	let (mut reader, mut writer) = tokio::io::duplex(4096);
4158	let dispatched = dispatch_stdio_request_with_writer(
4159	&state,
4160	&mut writer,
4161	"prompt/request",
4162	json!({ "prompt": "what is 2+2" }),
4163	AppTransport::Stdio,
4164	)
4165	.await
4166	.expect("prompt/request dispatch");
4167	drop(writer);
4168
4169	let response: PromptResponse =
4170	serde_json::from_value(dispatched.result).expect("prompt response");
4171
4172	// The model's words, not a restatement of the request.
4173	assert_eq!(response.output, "the answer is 4");
4174	assert!(
4175	!response.output.contains("what is 2+2"),
4176	"prompt echo leaked into the output: {}",
4177	response.output
4178	);
4179	assert_eq!(response.model, "stub-model-v1");
4180	assert_eq!(
4181	prompts.lock().await.as_slice(),
4182	["what is 2+2".to_string()],
4183	"the prompt must reach the runtime's turn endpoint"
4184	);
4185
4186	// Real streaming frames, not three canned ones.
4187	let deltas: Vec<String> = response
4188	.events
4189	.iter()
4190	.filter_map(\|event\| match event {
4191	EventFrame::ResponseDelta { delta, .. } => Some(delta.clone()),
4192	_ => None,
4193	})
4194	.collect();
4195	assert_eq!(deltas, vec!["the answer".to_string(), " is 4".to_string()]);
4196	assert!(matches!(
4197	response.events.first(),
4198	Some(EventFrame::ResponseStart { .. })
4199	));
4200	assert!(matches!(
4201	response.events.last(),
4202	Some(EventFrame::ResponseEnd { .. })
4203	));
4204
4205	// The stdio transport sees the same turn stream `thread/message` emits.
4206	let mut stdout = Vec::new();
4207	reader.read_to_end(&mut stdout).await.expect("read stdout");
4208	let stdout = String::from_utf8(stdout).expect("utf8 stdout");
4209	assert!(
4210	stdout.contains("\"type\":\"response_delta\"") && stdout.contains("the answer"),
4211	"stdio prompt turn must stream its deltas, got: {stdout}"
4212	);
4213
4214	// A prompt without a thread_id must not leave a mapping behind.
4215	assert!(
4216	state.runtime_thread_map.lock().await.is_empty(),
4217	"one-shot prompt threads must not accumulate in the map"
4218	);
4219
4220	server.abort();
4221	let _ = server.await;
4222	}
4223
4224	#[tokio::test]
4225	async fn prompt_without_a_reachable_runtime_fails_explicitly() {
4226	let (state, _tmp) = capability_test_state();
4227	// Port 9 (discard) refuses immediately: no runtime is listening.
4228	seed_bridge_at(&state, "http://127.0.0.1:9".to_string()).await;
4229
4230	let err = dispatch_stdio_request(&state, "prompt/run", json!({ "prompt": "hello" }))
4231	.await
4232	.expect_err("a prompt with no reachable runtime must fail, not echo");
4233	assert_eq!(err.code, RUNTIME_UNAVAILABLE_CODE);
4234
4235	let (status, Json(body)) = http_error_from_jsonrpc(err);
4236	assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
4237	assert_eq!(body["error"]["code"], "runtime_unavailable");
4238	assert!(
4239	body.get("output").is_none(),
4240	"a failure must not be shaped like a PromptResponse: {body}"
4241	);
4242	}
4243
4244	#[tokio::test]
4245	async fn empty_prompt_is_rejected_before_any_runtime_work() {
4246	let (state, _tmp) = capability_test_state();
4247	let err = dispatch_stdio_request(&state, "prompt/request", json!({ "prompt": " " }))
4248	.await
4249	.expect_err("an empty prompt must be rejected");
4250	assert_eq!(err.code, -32602);
4251	assert!(
4252	state.runtime_bridge.lock().await.is_none(),
4253	"a rejected prompt must not start a runtime"
4254	);
4255	}
4256
4257	#[tokio::test]
4258	async fn http_thread_message_runs_the_turn_instead_of_queueing_it() {
4259	let (state, _tmp) = capability_test_state();
4260	let (base_url, prompts, server) = spawn_stub_runtime().await;
4261	seed_bridge_at(&state, base_url).await;
4262
4263	let response = run_http_thread_message(
4264	&state,
4265	"thr_http".to_string(),
4266	"go".to_string(),
4267	Vec::new(),
4268	None,
4269	)
4270	.await
4271	.expect("http thread message");
4272
4273	assert_eq!(response.status, "completed");
4274	assert_eq!(response.thread_id, "thr_http");
4275	assert_eq!(response.data["turn_id"], "turn_stub");
4276	assert_eq!(prompts.lock().await.as_slice(), ["go".to_string()]);
4277	assert!(
4278	response
4279	.events
4280	.iter()
4281	.any(\|event\| matches!(event, EventFrame::ResponseDelta { .. })),
4282	"a completed turn must carry the deltas it streamed"
4283	);
4284
4285	server.abort();
4286	let _ = server.await;
4287	}
4288
4289	#[tokio::test]
4290	async fn http_thread_message_without_a_runtime_is_a_typed_error() {
4291	let (state, _tmp) = capability_test_state();
4292	seed_bridge_at(&state, "http://127.0.0.1:9".to_string()).await;
4293
4294	let err = run_http_thread_message(
4295	&state,
4296	"thr_http".to_string(),
4297	"go".to_string(),
4298	Vec::new(),
4299	None,
4300	)
4301	.await
4302	.expect_err("no runtime means no turn");
4303	assert_eq!(err.code, RUNTIME_UNAVAILABLE_CODE);
4304	}
4305
4306	#[tokio::test]
4307	async fn submit_user_input_refuses_instead_of_claiming_resolution() {
4308	let (state, _tmp) = capability_test_state();
4309	let response = process_app_request(
4310	&state,
4311	AppRequest::SubmitUserInput {
4312	request_id: "user-input-1".to_string(),
4313	answers: Vec::new(),
4314	},
4315	AppTransport::Stdio,
4316	)
4317	.await;
4318
4319	assert!(!response.ok, "this transport cannot deliver the answer");
4320	assert_eq!(response.data["error"], "user_input_reply_unsupported");
4321	assert!(
4322	response.data.get("resolved").is_none(),
4323	"nothing was resolved: {}",
4324	response.data
4325	);
4326	assert!(
4327	response.data["message"]
4328	.as_str()
4329	.expect("message")
4330	.contains("/v1/user-input/"),
4331	"the refusal must name the transport that can accept the answer"
4332	);
4333	assert!(
4334	!response.data["message"]
4335	.as_str()
4336	.expect("message")
4337	.contains(" "),
4338	"the refusal must not expose source-formatting whitespace"
4339	);
4340	}
4341
4342	// ── capability drift guard ─────────────────────────────────────────
4343	//
4344	// The stdio `capabilities` method is the benchmark/SDK contract: external
4345	// harnesses probe it (without spending model tokens) to learn what the
4346	// app-server can do. Pin the advertised method set so any change forces a
4347	// deliberate update here, in the dispatcher, and in docs/RUNTIME_API.md.
4348
4349	/// Methods advertised by the top-level `capabilities` probe, in order.
4350	const EXPECTED_CAPABILITY_METHODS: &[&str] = &[
4351	"healthz",
4352	"thread/capabilities",
4353	"thread/request",
4354	"thread/create",
4355	"thread/start",
4356	"thread/resume",
4357	"thread/fork",
4358	"thread/list",
4359	"thread/read",
4360	"thread/set_name",
4361	"thread/goal/set",
4362	"thread/goal/get",
4363	"thread/goal/clear",
4364	"thread/archive",
4365	"thread/unarchive",
4366	"thread/message",
4367	"thread/interrupt",
4368	"app/capabilities",
4369	"app/request",
4370	"app/config/get",
4371	"app/config/set",
4372	"app/config/unset",
4373	"app/config/list",
4374	"app/config/reload",
4375	"app/models",
4376	"app/thread_loaded_list",
4377	"prompt/capabilities",
4378	"prompt/request",
4379	"prompt/run",
4380	"shutdown",
4381	];
4382
4383	fn capability_test_state() -> (AppState, tempfile::TempDir) {
4384	let tmp = tempfile::tempdir().expect("tempdir");
4385	let config_path = tmp.path().join("config.toml");
4386	fs::write(&config_path, "").expect("write config");
4387	let state = build_state(Some(config_path), None).expect("state");
4388	(state, tmp)
4389	}
4390
4391	#[tokio::test]
4392	async fn capabilities_method_set_is_stable() {
4393	let (state, _tmp) = capability_test_state();
4394	let caps = dispatch_stdio_request(&state, "capabilities", json!({}))
4395	.await
4396	.expect("capabilities dispatch");
4397	let methods: Vec<String> = caps.result["methods"]
4398	.as_array()
4399	.expect("methods array")
4400	.iter()
4401	.map(\|m\| m.as_str().expect("method string").to_string())
4402	.collect();
4403	assert_eq!(
4404	methods, EXPECTED_CAPABILITY_METHODS,
4405	"app-server stdio capability set drifted; update the dispatcher, this \
4406	snapshot, and docs/RUNTIME_API.md together"
4407	);
4408	}
4409
4410	/// The socket transport advertises the `daemon/attach` handshake right
4411	/// after `healthz`; the stdio pin above must stay untouched by it.
4412	#[tokio::test]
4413	async fn socket_transport_advertises_daemon_attach() {
4414	let (state, _tmp) = capability_test_state();
4415	let mut sink = tokio::io::sink();
4416	let caps = dispatch_stdio_request_with_writer(
4417	&state,
4418	&mut sink,
4419	"capabilities",
4420	json!({}),
4421	AppTransport::Socket,
4422	)
4423	.await
4424	.expect("capabilities dispatch");
4425	assert_eq!(caps.result["transport"], json!("unix-socket"));
4426	let methods: Vec<String> = caps.result["methods"]
4427	.as_array()
4428	.expect("methods array")
4429	.iter()
4430	.map(\|m\| m.as_str().expect("method string").to_string())
4431	.collect();
4432	let mut expected: Vec<String> = EXPECTED_CAPABILITY_METHODS
4433	.iter()
4434	.map(\|m\| m.to_string())
4435	.collect();
4436	expected.insert(1, daemon_socket::ATTACH_METHOD.to_string());
4437	assert_eq!(methods, expected);
4438	}
4439
4440	#[tokio::test]
4441	async fn every_advertised_capability_is_dispatchable() {
4442	let (state, _tmp) = capability_test_state();
4443	// Empty params: methods may fail validation (-32602), but none may report
4444	// method-not-found (-32601). Required fields (e.g. PromptRequest.prompt)
4445	// make the prompt routes fail at parse time, so no model tokens are spent.
4446	for method in EXPECTED_CAPABILITY_METHODS {
4447	if let Err(err) = dispatch_stdio_request(&state, method, json!({})).await {
4448	assert_ne!(
4449	err.code,
4450	JsonRpcError::method_not_found(method).code,
4451	"advertised capability `{method}` is not dispatchable"
4452	);
4453	}
4454	}
4455	}
4456
4457	// ── resolve_auth_token ─────────────────────────────────────────────
4458
4459	#[test]
4460	fn auth_token_empty_string_fails() {
4461	let options = AppServerOptions {
4462	listen: "127.0.0.1:0".parse().expect("addr"),
4463	config_path: None,
4464	auth_token: Some(" ".to_string()),
4465	insecure_no_auth: false,
4466	cors_origins: Vec::new(),
4467	};
4468	let err = resolve_auth_token(&options).expect_err("empty token should fail");
4469	assert!(err.to_string().contains("cannot be empty"));
4470	}
4471
4472	#[test]
4473	fn auth_token_generated_when_none_provided() {
4474	let options = AppServerOptions {
4475	listen: "127.0.0.1:0".parse().expect("addr"),
4476	config_path: None,
4477	auth_token: None,
4478	insecure_no_auth: false,
4479	cors_origins: Vec::new(),
4480	};
4481	let token = resolve_auth_token(&options).unwrap();
4482	assert!(token.is_some());
4483	assert!(token.unwrap().starts_with("cwapp_"));
4484	}
4485
4486	#[test]
4487	fn runtime_bridge_command_keeps_auth_token_out_of_argv() {
4488	// FR001-C001: runtime auth token must not appear on the child argv
4489	// (visible via local `ps`); pass it via env instead.
4490	let token = "cwrt_unit_test_secret_token_not_for_argv";
4491	let cmd = RuntimeBridge::runtime_command(None, 18787, token).expect("command");
4492	let argv: Vec<String> = cmd
4493	.get_args()
4494	.map(\|a\| a.to_string_lossy().into_owned())
4495	.collect();
4496	assert!(
4497	!argv
4498	.iter()
4499	.any(\|a\| a.contains(token) \|\| a == "--auth-token"),
4500	"auth token must not be present in child argv: {argv:?}"
4501	);
4502	let envs: Vec<(String, String)> = cmd
4503	.get_envs()
4504	.filter_map(\|(k, v)\| {
4505	Some((
4506	k.to_string_lossy().into_owned(),
4507	v?.to_string_lossy().into_owned(),
4508	))
4509	})
4510	.collect();
4511	assert!(
4512	envs.iter()
4513	.any(\|(k, v)\| k == "CODEWHALE_RUNTIME_TOKEN" && v == token),
4514	"token must be carried via CODEWHALE_RUNTIME_TOKEN: {envs:?}"
4515	);
4516	assert!(
4517	envs.iter()
4518	.any(\|(k, v)\| k == "DEEPSEEK_RUNTIME_TOKEN" && v == token),
4519	"legacy alias DEEPSEEK_RUNTIME_TOKEN must also carry the token: {envs:?}"
4520	);
4521	}
4522
4523	#[test]
4524	fn generated_auth_status_does_not_render_token() {
4525	let rendered = app_server_auth_status_lines(false).join("\n");
4526
4527	assert!(!rendered.contains("Authorization: Bearer"));
4528	assert!(rendered.contains("not printed"));
4529	assert!(rendered.contains("CODEWHALE_APP_SERVER_TOKEN"));
4530	}
4531
4532	#[test]
4533	fn auth_token_explicit_is_preserved() {
4534	let options = AppServerOptions {
4535	listen: "127.0.0.1:0".parse().expect("addr"),
4536	config_path: None,
4537	auth_token: Some("my-secret".to_string()),
4538	insecure_no_auth: false,
4539	cors_origins: Vec::new(),
4540	};
4541	let token = resolve_auth_token(&options).unwrap();
4542	assert_eq!(token.as_deref(), Some("my-secret"));
4543	}
4544
4545	#[test]
4546	fn auth_token_explicit_allows_non_loopback_bind() {
4547	let options = AppServerOptions {
4548	listen: "0.0.0.0:8787".parse().expect("socket addr"),
4549	config_path: None,
4550	auth_token: Some("my-secret".to_string()),
4551	insecure_no_auth: false,
4552	cors_origins: Vec::new(),
4553	};
4554	let token = resolve_auth_token(&options).unwrap();
4555	assert_eq!(token.as_deref(), Some("my-secret"));
4556	}
4557
4558	#[test]
4559	fn insecure_no_auth_on_loopback_returns_none() {
4560	let options = AppServerOptions {
4561	listen: "127.0.0.1:0".parse().expect("addr"),
4562	config_path: None,
4563	auth_token: None,
4564	insecure_no_auth: true,
4565	cors_origins: Vec::new(),
4566	};
4567	let token = resolve_auth_token(&options).unwrap();
4568	assert!(token.is_none());
4569	}
4570
4571	#[test]
4572	fn insecure_no_auth_on_non_loopback_fails_fast() {
4573	let options = AppServerOptions {
4574	listen: "0.0.0.0:8787".parse().expect("socket addr"),
4575	config_path: None,
4576	auth_token: None,
4577	insecure_no_auth: true,
4578	cors_origins: Vec::new(),
4579	};
4580
4581	let err = resolve_auth_token(&options).expect_err("non-loopback unauth should fail");
4582	assert!(
4583	err.to_string()
4584	.contains("refusing unauthenticated app-server bind")
4585	);
4586	}
4587
4588	// ── cors_layer ─────────────────────────────────────────────────────
4589
4590	#[test]
4591	fn cors_layer_includes_default_origins() {
4592	let layer = cors_layer(&[]);
4593	// Just verify it doesn't panic and creates successfully
4594	let _ = layer;
4595	}
4596
4597	#[test]
4598	fn cors_layer_adds_extra_origins() {
4599	let extras = vec!["https://example.com".to_string()];
4600	let layer = cors_layer(&extras);
4601	let _ = layer;
4602	}
4603
4604	#[test]
4605	fn cors_layer_skips_empty_origins() {
4606	let extras = vec!["".to_string(), " ".to_string()];
4607	let layer = cors_layer(&extras);
4608	let _ = layer;
4609	}
4610
4611	// ── JsonRpc helpers ────────────────────────────────────────────────
4612
4613	#[test]
4614	fn params_or_object_returns_object_for_null() {
4615	let result = params_or_object(Value::Null);
4616	assert_eq!(result, json!({}));
4617	}
4618
4619	#[test]
4620	fn params_or_object_passthrough_for_non_null() {
4621	let input = json!({"key": "value"});
4622	let result = params_or_object(input.clone());
4623	assert_eq!(result, input);
4624	}
4625
4626	#[test]
4627	fn jsonrpc_result_format() {
4628	let result = jsonrpc_result(Some(json!(1)), json!({"ok": true}));
4629	assert_eq!(result["jsonrpc"], "2.0");
4630	assert_eq!(result["id"], 1);
4631	assert_eq!(result["result"]["ok"], true);
4632	}
4633
4634	#[test]
4635	fn jsonrpc_result_null_id() {
4636	let result = jsonrpc_result(None, json!(null));
4637	assert_eq!(result["id"], Value::Null);
4638	}
4639
4640	#[test]
4641	fn jsonrpc_error_format() {
4642	let err = jsonrpc_error(Some(json!(2)), JsonRpcError::internal("oops"));
4643	assert_eq!(err["jsonrpc"], "2.0");
4644	assert_eq!(err["id"], 2);
4645	assert_eq!(err["error"]["code"], -32603);
4646	assert_eq!(err["error"]["message"], "oops");
4647	}
4648
4649	#[test]
4650	fn jsonrpc_error_codes() {
4651	assert_eq!(JsonRpcError::parse_error("").code, -32700);
4652	assert_eq!(JsonRpcError::invalid_request("").code, -32600);
4653	assert_eq!(JsonRpcError::method_not_found("x").code, -32601);
4654	assert_eq!(JsonRpcError::invalid_params("").code, -32602);
4655	assert_eq!(JsonRpcError::internal("").code, -32603);
4656	}
4657
4658	// ── AppServerOptions ───────────────────────────────────────────────
4659
4660	#[test]
4661	fn app_server_options_debug_does_not_leak_token() {
4662	let options = AppServerOptions {
4663	listen: "127.0.0.1:8080".parse().expect("addr"),
4664	config_path: None,
4665	auth_token: Some("secret-token".to_string()),
4666	insecure_no_auth: false,
4667	cors_origins: vec!["https://example.com".to_string()],
4668	};
4669	let debug = format!("{options:?}");
4670	assert!(!debug.contains("secret-token"));
4671	assert!(debug.contains("<redacted>"));
4672	assert!(debug.contains("8080"));
4673	}
4674
4675	// ── Default CORS origins ──────────────────────────────────────────
4676
4677	#[test]
4678	fn default_cors_origins_include_common_dev_ports() {
4679	assert!(DEFAULT_CORS_ORIGINS.contains(&"http://localhost:3000"));
4680	assert!(DEFAULT_CORS_ORIGINS.contains(&"http://localhost:5173"));
4681	assert!(DEFAULT_CORS_ORIGINS.contains(&"tauri://localhost"));
4682	}
4683	#[tokio::test]
4684	async fn runtime_image_daemon_bridge_checks_transport_and_forwards_exact_wire() {
4685	async fn capture(
4686	State(seen): State<Arc<Mutex<Vec<Value>>>>,
4687	Json(body): Json<Value>,
4688	) -> (StatusCode, Json<Value>) {
4689	seen.lock().await.push(body);
4690	(
4691	StatusCode::BAD_REQUEST,
4692	Json(json!({"error":"fixture stops before an Engine"})),
4693	)
4694	}
4695	for supported in [false, true] {
4696	let seen = Arc::new(Mutex::new(Vec::new()));
4697	let app = Router::new()
4698	.route(
4699	"/v1/runtime/info",
4700	get(move \|\| async move {
4701	Json(json!({"capabilities":{"turn_image_inputs":supported}}))
4702	}),
4703	)
4704	.route("/v1/threads/{id}/turns", post(capture))
4705	.with_state(seen.clone());
4706	let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4707	let addr = listener.local_addr().unwrap();
4708	let server = tokio::spawn(async move {
4709	axum::serve(listener, app).await.unwrap();
4710	});
4711	let mut bridge = RuntimeBridge::from_base_url_for_test(format!("http://{addr}"));
4712	let images = vec![RuntimeImageInput {
4713	mime: "image/png".into(),
4714	data_base64: "fixture-bytes-validated-by-Core".into(),
4715	}];
4716	let mut writer = tokio::io::sink();
4717	assert!(
4718	bridge
4719	.message_thread(
4720	"thr_fixture",
4721	"look",
4722	&images,
4723	None,
4724	&mut writer,
4725	None,
4726	None
4727	)
4728	.await
4729	.is_err()
4730	);
4731	let requests = seen.lock().await;
4732	assert_eq!(requests.len(), usize::from(supported));
4733	if supported {
4734	assert_eq!(requests[0], json!({"prompt":"look","images":images}));
4735	}
4736	server.abort();
4737	}
4738	}
4739
4740	#[test]
4741	fn runtime_image_daemon_all_input_families_preserve_images() {
4742	let image = json!({"mime":"image/png","dataBase64":"AQ=="});
4743	let thread: ThreadMessageParams = serde_json::from_value(
4744	json!({"thread_id":"thr_fixture","input":"look","images":[image.clone()]}),
4745	)
4746	.unwrap();
4747	let prompt: PromptRequest =
4748	serde_json::from_value(json!({"prompt":"look","images":[image.clone()]})).unwrap();
4749	let generic: ThreadRequest = serde_json::from_value(
4750	json!({"kind":"message","thread_id":"thr_fixture","input":"look","images":[image]}),
4751	)
4752	.unwrap();
4753	assert_eq!(thread.images, prompt.images);
4754	let ThreadRequest::Message { images, .. } = generic else {
4755	panic!("message");
4756	};
4757	assert_eq!(thread.images, images);
4758	assert!(matches!(
4759	parse_stdio_line(&" ".repeat(MAX_RUNTIME_IMAGE_BODY_BYTES + 1)),
4760	ParsedStdioLine::Rejected(_)
4761	));
4762	}
4763	}
4764

4764 lines RUST