runtime_api.rs - CodeWhale

runtime_api.rs

1	//! Runtime HTTP/SSE API for local Codewhale automation.
2
3	use std::collections::{BTreeMap, BTreeSet};
4	use std::convert::Infallible;
5	use std::fs;
6	use std::net::{IpAddr, SocketAddr, UdpSocket};
7	use std::path::{Path as FsPath, PathBuf};
8	use std::sync::Arc;
9	use std::time::Duration;
10
11	use anyhow::{Context, Result, anyhow, bail};
12	use async_stream::stream;
13	use axum::extract::{Path, Query, Request, State};
14	use axum::http::header;
15	use axum::http::{HeaderName, HeaderValue, Method, StatusCode};
16	use axum::middleware;
17	use axum::response::Html;
18	use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
19	use axum::response::{IntoResponse, Response};
20	use axum::routing::{get, post};
21	use axum::{Json, Router};
22	use chrono::Utc;
23	use codewhale_protocol::runtime::{
24	DynamicToolCallResult, RUNTIME_API_VERSION, RUNTIME_EVENT_ENVELOPE_SCHEMA_VERSION,
25	RuntimeCapabilities, RuntimeEventEnvelope, RuntimeExperimentalCapabilities,
26	};
27	use codewhale_secrets::account::{
28	ACCOUNT_API_BASE_ENV, DEFAULT_ACCOUNT_API_BASE, RuntimeAccountInfo,
29	};
30	#[cfg(not(test))]
31	use codewhale_secrets::account::{AccountSessionStore, secure_account_session_secrets};
32	use serde::{Deserialize, Serialize};
33	use serde_json::{Value, json};
34	use tokio::net::TcpListener;
35	use tokio::sync::Mutex;
36	use tokio_util::sync::CancellationToken;
37	use tower_http::cors::CorsLayer;
38
39	#[cfg(test)]
40	use crate::dependencies::ExternalTool;
41
42	use crate::automation_manager::{
43	AutomationManager, AutomationRecord, AutomationRunRecord, AutomationSchedulerConfig,
44	CreateAutomationRequest, SharedAutomationManager, UpdateAutomationRequest, spawn_scheduler,
45	};
46	use crate::config::{
47	ApiProvider, Config, DEFAULT_TEXT_MODEL, normalize_model_name_for_provider, validate_route,
48	};
49	use crate::fleet::executor::{FleetExecutor, configured_codewhale_binary};
50	use crate::fleet::ledger::{FleetEventReplayError, FleetLedgerState, FleetTaskLedgerStatus};
51	use crate::fleet::manager::{
52	FleetManager, FleetStatusSnapshot, FleetWorkerInspection, FleetWorkerRuntimeProjection,
53	ManagedFleetRunDescriptor,
54	};
55	use crate::fleet::profile::canonical_public_role_name;
56	use crate::fleet::task_spec::FleetTaskSpecDocument;
57	use crate::fleet::worker_runtime::fleet_write_roots;
58	use crate::mcp::McpPool;
59	#[cfg(test)]
60	pub(super) use crate::models::{ContentBlock, Message};
61	use crate::runtime_threads::{
62	CompactThreadRequest, CreateThreadRequest, ExternalApprovalDecision,
63	MAX_RUNTIME_EVENT_REPLAY_TAIL, RuntimeThreadManager, RuntimeThreadManagerConfig,
64	SharedRuntimeThreadManager, StartTurnRequest, SteerTurnRequest, ThreadDetail, ThreadListFilter,
65	ThreadRecord, TurnItemKind, TurnRecord, UpdateThreadRequest, UsageGroupBy,
66	};
67	#[cfg(test)]
68	pub(super) use crate::runtime_threads::{RuntimeTurnStatus, TurnItemLifecycleStatus};
69	use crate::session_manager::default_sessions_dir;
70	#[cfg(test)]
71	pub(super) use crate::session_manager::{SavedSession, SessionMetadata};
72	use crate::skill_state::SkillStateStore;
73	use crate::task_manager::{
74	NewTaskRequest, SharedTaskManager, TaskManager, TaskManagerConfig, TaskRecord, TaskSummary,
75	};
76	use crate::tools::subagent::{
77	AgentWorkerRecord, SharedSubAgentManager, load_persisted_agent_worker_records,
78	new_shared_subagent_manager_with_timeout,
79	};
80	use codewhale_protocol::fleet::{
81	FleetArtifactKind, FleetEventReplay, FleetRun, FleetRunId, FleetRuntimeEvent,
82	FleetRuntimeTarget, FleetSecurityPolicy, FleetTaskSpec, FleetWorkerEventPayload,
83	FleetWorkerSpec, FleetWorkerStatus, FleetWorkflowDescriptor, FleetWorkflowKind,
84	};
85
86	mod auth;
87	mod sessions;
88	mod web;
89	mod workspace;
90	#[cfg(test)]
91	use self::auth::{ResolvedRuntimeAuth, token_from_cookie_header};
92	use self::auth::{
93	require_runtime_token, resolve_runtime_auth, runtime_auth_status_lines,
94	runtime_request_is_authorized,
95	};
96	use self::sessions::{
97	create_session_from_thread, delete_session, get_session, list_sessions, list_sessions_summary,
98	patch_session, resume_session_thread, save_current_session,
99	};
100	#[cfg(test)]
101	use self::sessions::{messages_from_thread_detail, session_to_detail};
102	#[cfg(test)]
103	use self::workspace::collect_workspace_status;
104	use self::workspace::{collect_workspace_git_metadata, workspace_status};
105
106	#[derive(Clone)]
107	pub struct RuntimeApiState {
108	config: Arc<parking_lot::RwLock<Config>>,
109	workspace: PathBuf,
110	plugin_discovery: Arc<crate::plugins::PluginDiscoveryContext>,
111	task_manager: SharedTaskManager,
112	runtime_threads: SharedRuntimeThreadManager,
113	cors_origins: Vec<String>,
114	sessions_dir: PathBuf,
115	/// Original `--config` path (if any) used to load the initial config.
116	/// Passed to `Config::load` on reload and to persistence helpers so
117	/// GUI-driven config changes target the same file the server was
118	/// started with, instead of falling back to the default discovery.
119	config_path: Option<PathBuf>,
120	/// Effective initial profile (`--profile` or `DEEPSEEK_PROFILE`).
121	/// Reload must retain this overlay so profile-scoped routes do not vanish.
122	config_profile: Option<String>,
123	automations: SharedAutomationManager,
124	sub_agent_manager: SharedSubAgentManager,
125	runtime_token: Option<String>,
126	skill_state: Arc<Mutex<SkillStateStore>>,
127	auth_required: bool,
128	bind_host: String,
129	bind_port: u16,
130	mobile_enabled: bool,
131	web: Option<web::RuntimeWebState>,
132	/// Executable used by Runtime API-owned Fleet manager loops. Stored on
133	/// state so tests and embedded callers can provide a hermetic worker.
134	fleet_codewhale_binary: String,
135	/// Shared McpPool reused for explicit live MCP discovery. Passive API
136	/// calls do not initialize this pool so dashboards cannot accidentally
137	/// become a second stdio-process owner. The outer mutex guards only the
138	/// lazily-initialized slot; slow per-pool work (connect_all) runs under
139	/// the inner handle so it cannot block slot reads.
140	mcp_pool: Arc<Mutex<Option<Arc<Mutex<McpPool>>>>>,
141	#[cfg(test)]
142	compat_stream_test_hook: Option<tokio::sync::mpsc::UnboundedSender<CompatStreamTestPoint>>,
143	}
144
145	#[cfg(test)]
146	enum CompatStreamTestPoint {
147	ThreadCreated {
148	thread_id: String,
149	resume: tokio::sync::oneshot::Sender<()>,
150	},
151	SubscribedBeforeReplay {
152	thread_id: String,
153	turn_id: String,
154	resume: tokio::sync::oneshot::Sender<()>,
155	},
156	ReplayLoaded {
157	thread_id: String,
158	turn_id: String,
159	resume: tokio::sync::oneshot::Sender<()>,
160	},
161	}
162
163	#[derive(Debug, Clone)]
164	pub struct RuntimeApiOptions {
165	pub host: String,
166	pub port: u16,
167	pub workers: usize,
168	/// Additional CORS origins to allow on top of the built-in defaults
169	/// (`http://localhost:{3000,1420}`, `http://127.0.0.1:{3000,1420}`,
170	/// `tauri://localhost`). Populated by `--cors-origin` (repeatable),
171	/// `CODEWHALE_CORS_ORIGINS` (comma-separated, `DEEPSEEK_CORS_ORIGINS`
172	/// as alias), and `[runtime_api] cors_origins` in `config.toml`.
173	/// Whalescale#255 / #561.
174	pub cors_origins: Vec<String>,
175	/// Optional bearer token required for `/v1/*` routes. If omitted here,
176	/// `run_http_server` checks `CODEWHALE_RUNTIME_TOKEN`, then
177	/// `DEEPSEEK_RUNTIME_TOKEN` as an alias.
178	pub auth_token: Option<String>,
179	/// Allow `/v1/*` routes without auth when no token is configured.
180	pub insecure_no_auth: bool,
181	/// Enables the built-in mobile control page at `/mobile`.
182	pub mobile: bool,
183	/// Enables the embedded local browser client and opens it after binding.
184	/// Web mode is always loopback-only and uses a one-time bootstrap cookie
185	/// exchange rather than exposing the Runtime token to the browser URL.
186	pub web: bool,
187	/// Show a QR code for the mobile URL in the terminal.
188	pub show_qr: bool,
189	/// Original `--config` path used to load the initial config. When
190	/// `Some`, GUI-driven config reloads and persistence target this file
191	/// instead of the default discovery path.
192	pub config_path: Option<PathBuf>,
193	/// Effective profile used to load the server's initial Config.
194	pub config_profile: Option<String>,
195	}
196
197	impl Default for RuntimeApiOptions {
198	fn default() -> Self {
199	Self {
200	host: "127.0.0.1".to_string(),
201	port: 7878,
202	workers: 2,
203	cors_origins: Vec::new(),
204	auth_token: None,
205	insecure_no_auth: false,
206	mobile: false,
207	web: false,
208	show_qr: false,
209	config_path: None,
210	config_profile: None,
211	}
212	}
213	}
214
215	#[derive(Debug, Deserialize)]
216	struct StreamTurnRequest {
217	prompt: String,
218	model: Option<String>,
219	mode: Option<String>,
220	permission_posture: Option<String>,
221	workspace: Option<PathBuf>,
222	allow_shell: Option<bool>,
223	trust_mode: Option<bool>,
224	auto_approve: Option<bool>,
225	}
226
227	#[derive(Debug, Serialize)]
228	struct HealthResponse {
229	status: &'static str,
230	service: &'static str,
231	mode: &'static str,
232	}
233
234	#[derive(Debug, Serialize)]
235	struct TasksResponse {
236	tasks: Vec<TaskSummary>,
237	counts: crate::task_manager::TaskCounts,
238	}
239
240	#[derive(Debug, Deserialize)]
241	struct TasksQuery {
242	limit: Option<usize>,
243	workspace: Option<PathBuf>,
244	}
245
246	#[derive(Debug, Deserialize)]
247	struct ThreadsQuery {
248	limit: Option<usize>,
249	include_archived: Option<bool>,
250	/// When `true`, returns archived threads only (overrides `include_archived`).
251	/// Whalescale#260 / #563.
252	archived_only: Option<bool>,
253	}
254
255	#[derive(Debug, Deserialize)]
256	struct ThreadSummaryQuery {
257	limit: Option<usize>,
258	search: Option<String>,
259	include_archived: Option<bool>,
260	/// When `true`, returns archived threads only (overrides `include_archived`).
261	/// Whalescale#260 / #563.
262	archived_only: Option<bool>,
263	}
264
265	fn resolve_thread_filter(
266	include_archived: Option<bool>,
267	archived_only: Option<bool>,
268	) -> ThreadListFilter {
269	if archived_only.unwrap_or(false) {
270	ThreadListFilter::ArchivedOnly
271	} else if include_archived.unwrap_or(false) {
272	ThreadListFilter::IncludeArchived
273	} else {
274	ThreadListFilter::ActiveOnly
275	}
276	}
277
278	#[derive(Debug, Serialize)]
279	struct ThreadSummary {
280	id: String,
281	title: String,
282	preview: String,
283	model: String,
284	mode: String,
285	workspace: PathBuf,
286	branch: Option<String>,
287	head: Option<String>,
288	dirty: bool,
289	archived: bool,
290	updated_at: chrono::DateTime<Utc>,
291	latest_turn_id: Option<String>,
292	latest_turn_status: Option<String>,
293	}
294
295	#[derive(Debug, Serialize)]
296	struct SkillEntry {
297	name: String,
298	description: String,
299	/// Native Skill locator. Reviewed plugin paths are deliberately omitted;
300	/// their bodies are available only through the authority-bound snapshot.
301	path: Option<PathBuf>,
302	source: String,
303	plugin_id: Option<String>,
304	plugin_generation: Option<u64>,
305	plugin_content_hash: Option<String>,
306	enabled: bool,
307	is_bundled: bool,
308	}
309
310	#[derive(Debug, Serialize)]
311	struct SkillsResponse {
312	directory: PathBuf,
313	directories: Vec<PathBuf>,
314	warnings: Vec<String>,
315	skills: Vec<SkillEntry>,
316	}
317
318	#[derive(Debug, Serialize)]
319	struct AgentRunsResponse {
320	runs: Vec<AgentWorkerRecord>,
321	}
322
323	#[derive(Debug, Deserialize)]
324	struct SetSkillEnabledRequest {
325	enabled: bool,
326	}
327
328	#[derive(Debug, Serialize)]
329	struct SetSkillEnabledResponse {
330	name: String,
331	enabled: bool,
332	}
333
334	#[derive(Debug, Deserialize)]
335	struct DecideApprovalBody {
336	decision: String,
337	#[serde(default)]
338	remember: bool,
339	}
340
341	#[derive(Debug, Serialize)]
342	struct DecideApprovalResponse {
343	ok: bool,
344	approval_id: String,
345	decision: String,
346	delivered: bool,
347	}
348
349	#[derive(Debug, Deserialize)]
350	struct SubmitUserInputBody {
351	answers: Vec<UserInputAnswerBody>,
352	}
353
354	#[derive(Debug, Deserialize)]
355	struct UserInputAnswerBody {
356	id: String,
357	label: String,
358	value: String,
359	}
360
361	#[derive(Debug, Serialize)]
362	struct SubmitUserInputResponse {
363	ok: bool,
364	input_id: String,
365	delivered: bool,
366	}
367
368	#[derive(Debug, Serialize)]
369	struct RuntimeInfoResponse {
370	service: &'static str,
371	runtime_api_version: &'static str,
372	codewhale_version: &'static str,
373	/// Full 40-character source commit embedded by the shared build script.
374	/// Desktop compatibility intentionally rejects `unknown` and abbreviated
375	/// values, so source archives without build provenance fail closed.
376	codewhale_commit: &'static str,
377	bind_host: String,
378	port: u16,
379	auth_required: bool,
380	transports: Vec<&'static str>,
381	capabilities: RuntimeCapabilities,
382	account: RuntimeAccountInfo,
383	experimental: RuntimeExperimentalCapabilities,
384	// Backward-compatible alias kept for existing clients.
385	version: &'static str,
386	}
387
388	fn default_runtime_capabilities() -> RuntimeCapabilities {
389	RuntimeCapabilities {
390	account_session: true,
391	threads: true,
392	turns: true,
393	turn_steer: true,
394	turn_interrupt: true,
395	event_replay: true,
396	external_tools: true,
397	environments: false,
398	worker_runtime: true,
399	fleet_run_create: true,
400	fleet_run_start: true,
401	fleet_event_replay: true,
402	fleet_event_stream: true,
403	fleet_local_target: true,
404	}
405	}
406
407	fn runtime_api_sub_agent_manager(workspace: &FsPath, workers: usize) -> SharedSubAgentManager {
408	let max_agents = workers.max(1);
409	new_shared_subagent_manager_with_timeout(
410	workspace.to_path_buf(),
411	max_agents,
412	max_agents,
413	Duration::from_secs(crate::config::DEFAULT_SUBAGENT_HEARTBEAT_TIMEOUT_SECS),
414	max_agents,
415	None,
416	)
417	}
418
419	#[derive(Debug, Serialize)]
420	struct McpServerEntry {
421	name: String,
422	enabled: bool,
423	required: bool,
424	command: Option<String>,
425	url: Option<String>,
426	connected: bool,
427	enabled_tools: Vec<String>,
428	disabled_tools: Vec<String>,
429	}
430
431	#[derive(Debug, Serialize)]
432	struct McpServersResponse {
433	servers: Vec<McpServerEntry>,
434	}
435
436	#[derive(Debug, Deserialize)]
437	struct McpToolsQuery {
438	server: Option<String>,
439	#[serde(default)]
440	connect: bool,
441	}
442
443	#[derive(Debug, Serialize)]
444	struct McpToolEntry {
445	server: String,
446	name: String,
447	prefixed_name: String,
448	description: Option<String>,
449	input_schema: Value,
450	}
451
452	#[derive(Debug, Serialize)]
453	struct McpToolsResponse {
454	tools: Vec<McpToolEntry>,
455	}
456
457	#[derive(Debug, Deserialize)]
458	struct AutomationRunsQuery {
459	limit: Option<usize>,
460	}
461
462	#[derive(Debug, Deserialize)]
463	struct ThreadEventsQuery {
464	since_seq: Option<u64>,
465	replay_limit: Option<usize>,
466	}
467
468	const DEFAULT_FLEET_EVENT_REPLAY_LIMIT: usize = 250;
469	const MAX_FLEET_EVENT_REPLAY_LIMIT: usize = 1_000;
470
471	#[derive(Debug, Deserialize)]
472	#[serde(deny_unknown_fields)]
473	struct CreateFleetRunRequest {
474	#[serde(default)]
475	name: Option<String>,
476	target: FleetRuntimeTarget,
477	roles: Vec<ManagedFleetRoleRequest>,
478	workflow: ManagedFleetWorkflowRequest,
479	#[serde(default, alias = "workers")]
480	worker_specs: Vec<FleetWorkerSpec>,
481	#[serde(default)]
482	labels: BTreeMap<String, String>,
483	#[serde(default)]
484	security_policy: Option<FleetSecurityPolicy>,
485	#[serde(default)]
486	max_workers: Option<usize>,
487	}
488
489	#[derive(Debug, Deserialize)]
490	#[serde(deny_unknown_fields)]
491	struct ManagedFleetRoleRequest {
492	name: String,
493	#[serde(default)]
494	agent_profile: Option<String>,
495	}
496
497	#[derive(Debug, Deserialize)]
498	#[serde(deny_unknown_fields)]
499	struct ManagedFleetWorkflowRequest {
500	id: String,
501	kind: FleetWorkflowKind,
502	#[serde(alias = "task_specs")]
503	tasks: Vec<FleetTaskSpec>,
504	}
505
506	#[derive(Debug, Deserialize)]
507	struct FleetEventsQuery {
508	after: Option<String>,
509	limit: Option<usize>,
510	}
511
512	#[derive(Debug, Serialize)]
513	struct StartTurnResponse {
514	thread: ThreadRecord,
515	turn: TurnRecord,
516	}
517
518	/// Start the runtime API server.
519	pub async fn run_http_server(
520	config: Config,
521	workspace: PathBuf,
522	plugin_discovery: Arc<crate::plugins::PluginDiscoveryContext>,
523	options: RuntimeApiOptions,
524	) -> Result<()> {
525	if options.port == 0 {
526	bail!("Port must be > 0");
527	}
528	if options.web && options.host != "127.0.0.1" {
529	bail!("Codewhale web is loopback-only and must bind to 127.0.0.1");
530	}
531	if options.web && options.insecure_no_auth {
532	bail!("Codewhale web requires Runtime authentication; remove --insecure");
533	}
534
535	let task_cfg = TaskManagerConfig::from_runtime(
536	&config,
537	workspace.clone(),
538	config.default_text_model.clone(),
539	Some(options.workers),
540	);
541	let runtime_threads = Arc::new(RuntimeThreadManager::open_with_plugin_registry(
542	config.clone(),
543	workspace.clone(),
544	RuntimeThreadManagerConfig::from_task_data_dir(task_cfg.data_dir.clone()),
545	plugin_discovery.registry_for_workspace(&workspace),
546	)?);
547	let task_manager =
548	TaskManager::start_with_runtime_manager(task_cfg, config.clone(), runtime_threads.clone())
549	.await?;
550	let automations = Arc::new(Mutex::new(AutomationManager::default_location()?));
551	runtime_threads.attach_automation_manager(automations.clone());
552	let scheduler_cancel = CancellationToken::new();
553	let scheduler_handle = spawn_scheduler(
554	automations.clone(),
555	task_manager.clone(),
556	scheduler_cancel.clone(),
557	AutomationSchedulerConfig::default(),
558	);
559
560	let sessions_dir = default_sessions_dir().unwrap_or_else(\|_\| fallback_sessions_dir());
561	let runtime_token_env = std::env::var("CODEWHALE_RUNTIME_TOKEN")
562	.ok()
563	.or_else(\|\| std::env::var("DEEPSEEK_RUNTIME_TOKEN").ok());
564	let resolved_auth = resolve_runtime_auth(
565	options.auth_token.clone(),
566	runtime_token_env,
567	options.insecure_no_auth,
568	);
569	let runtime_token = resolved_auth.token.clone();
570	let auth_enabled = runtime_token.is_some();
571	let (web, web_bootstrap) = if options.web {
572	runtime_token
573	.as_ref()
574	.context("Codewhale web requires a Runtime authentication token")?;
575	let (web, bootstrap) = web::RuntimeWebState::new();
576	(Some(web), Some(bootstrap))
577	} else {
578	(None, None)
579	};
580	let skill_state = SkillStateStore::load_default()
581	.context("load persistent Skill activation state for Runtime API")?;
582	let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, options.workers);
583	let state = RuntimeApiState {
584	config: Arc::new(parking_lot::RwLock::new(config.clone())),
585	workspace,
586	plugin_discovery,
587	task_manager,
588	runtime_threads,
589	cors_origins: options.cors_origins.clone(),
590	sessions_dir,
591	config_path: options.config_path.clone(),
592	config_profile: options.config_profile.clone(),
593	automations,
594	sub_agent_manager,
595	runtime_token: runtime_token.clone(),
596	skill_state: Arc::new(Mutex::new(skill_state)),
597	auth_required: auth_enabled,
598	bind_host: options.host.clone(),
599	bind_port: options.port,
600	mobile_enabled: options.mobile,
601	web,
602	fleet_codewhale_binary: configured_codewhale_binary(),
603	mcp_pool: Arc::new(Mutex::new(None)),
604	#[cfg(test)]
605	compat_stream_test_hook: None,
606	};
607	let app = build_router(state);
608
609	let addr: SocketAddr = format!("{}:{}", options.host, options.port)
610	.parse()
611	.with_context(\|\| format!("Invalid bind address '{}:{}'", options.host, options.port))?;
612	let listener = TcpListener::bind(addr)
613	.await
614	.with_context(\|\| format!("Failed to bind {addr}"))?;
615
616	let bound_addr = listener
617	.local_addr()
618	.context("Failed to read Runtime API listener address")?;
619	println!("Runtime API listening on http://{bound_addr}");
620	for line in runtime_auth_status_lines(&resolved_auth) {
621	println!("{line}");
622	}
623	if options.mobile {
624	print_mobile_urls(
625	bound_addr,
626	auth_enabled,
627	resolved_auth.generated,
628	options.show_qr,
629	);
630	}
631	if let Some(bootstrap) = web_bootstrap {
632	println!("Codewhale web enabled at http://{bound_addr}/");
633	let bootstrap_url = web::bootstrap_url(bound_addr, &bootstrap);
634	if let Err(error) = crate::utils::open_url(&bootstrap_url) {
635	scheduler_cancel.cancel();
636	scheduler_handle.abort();
637	return Err(error)
638	.context("Failed to open the Codewhale web client in the default browser");
639	}
640	}
641	let is_loopback = options.host == "127.0.0.1" \|\| options.host == "::1";
642	if is_loopback {
643	println!("Security: this server is local-first. Do not expose it to untrusted networks.");
644	} else {
645	println!(
646	"Security: bound to {host}; reachable from any peer that can route to this address.",
647	host = options.host
648	);
649	if !auth_enabled {
650	println!(
651	" WARNING: auth is disabled. Anyone on the network can call /v1/* without authentication."
652	);
653	}
654	println!(
655	" /v1/runtime/info reports bind_host={host:?}, port={port}, auth_required={auth}.",
656	host = options.host,
657	port = options.port,
658	auth = auth_enabled,
659	);
660	}
661	let serve_result = axum::serve(
662	listener,
663	app.into_make_service_with_connect_info::<SocketAddr>(),
664	)
665	.await
666	.map_err(\|e\| anyhow!("Runtime API server error: {e}"));
667	scheduler_cancel.cancel();
668	scheduler_handle.abort();
669	serve_result
670	}
671
672	fn fallback_sessions_dir() -> PathBuf {
673	if let Some(home) = codewhale_paths::codewhale_home_override().ok().flatten() {
674	return home.join("sessions");
675	}
676	codewhale_paths::legacy_deepseek_home()
677	.unwrap_or_else(\|\| PathBuf::from(codewhale_paths::LEGACY_APP_DIR))
678	.join("sessions")
679	}
680
681	pub fn build_router(state: RuntimeApiState) -> Router {
682	let api_routes = Router::new()
683	.route(
684	"/v1/sessions",
685	get(list_sessions)
686	.post(create_session_from_thread)
687	.put(save_current_session),
688	)
689	.route("/v1/sessions/summary", get(list_sessions_summary))
690	.route(
691	"/v1/sessions/{id}",
692	get(get_session).patch(patch_session).delete(delete_session),
693	)
694	.route(
695	"/v1/sessions/{id}/resume-thread",
696	post(resume_session_thread),
697	)
698	.route("/v1/workspace/status", get(workspace_status))
699	.route("/v1/agent-runs", get(list_agent_runs))
700	.route("/v1/agent-runs/{run_id}", get(get_agent_run))
701	.route(
702	"/v1/fleet/runs",
703	get(list_fleet_runs).post(create_fleet_run),
704	)
705	.route("/v1/fleet/runs/{run_id}", get(get_fleet_run))
706	.route(
707	"/v1/fleet/runs/{run_id}/workers",
708	get(list_fleet_run_workers),
709	)
710	.route("/v1/fleet/runs/{run_id}/start", post(start_fleet_run))
711	.route("/v1/fleet/runs/{run_id}/events", get(stream_fleet_events))
712	.route(
713	"/v1/fleet/runs/{run_id}/events/replay",
714	get(replay_fleet_events),
715	)
716	.route("/v1/fleet/runs/{run_id}/stop", post(stop_fleet_run))
717	.route("/v1/fleet/workers/{worker_id}", get(get_fleet_worker))
718	.route(
719	"/v1/fleet/workers/{worker_id}/interrupt",
720	post(interrupt_fleet_worker),
721	)
722	.route(
723	"/v1/fleet/workers/{worker_id}/stop",
724	post(stop_fleet_worker),
725	)
726	.route(
727	"/v1/fleet/workers/{worker_id}/restart",
728	post(restart_fleet_worker),
729	)
730	.route("/v1/stream", post(stream_turn))
731	.route("/v1/threads", get(list_threads).post(create_thread))
732	.route("/v1/threads/summary", get(list_threads_summary))
733	.route("/v1/threads/{id}", get(get_thread).patch(update_thread))
734	.route("/v1/threads/{id}/resume", post(resume_thread))
735	.route("/v1/threads/{id}/fork", post(fork_thread))
736	.route("/v1/threads/{id}/undo", post(undo_thread_turn))
737	.route("/v1/threads/{id}/patch-undo", post(patch_undo_thread_turn))
738	.route("/v1/threads/{id}/retry", post(retry_thread_turn))
739	.route("/v1/threads/{id}/turns", post(start_thread_turn))
740	.route(
741	"/v1/threads/{id}/turns/{turn_id}/steer",
742	post(steer_thread_turn),
743	)
744	.route(
745	"/v1/threads/{id}/turns/{turn_id}/interrupt",
746	post(interrupt_thread_turn),
747	)
748	.route(
749	"/v1/threads/{id}/turns/{turn_id}/tool-calls/{call_id}/result",
750	post(deliver_dynamic_tool_result),
751	)
752	.route("/v1/threads/{id}/compact", post(compact_thread))
753	.route("/v1/threads/{id}/events", get(stream_thread_events))
754	.route("/v1/approvals/{approval_id}", post(decide_approval))
755	.route(
756	"/v1/user-input/{thread_id}/{input_id}",
757	post(submit_user_input),
758	)
759	.route("/v1/tasks", get(list_tasks).post(create_task))
760	.route("/v1/tasks/{id}", get(get_task))
761	.route("/v1/tasks/{id}/cancel", post(cancel_task))
762	.route("/v1/skills", get(list_skills))
763	.route("/v1/skills/{name}", post(set_skill_enabled))
764	.route("/v1/apps/mcp/servers", get(list_mcp_servers))
765	.route("/v1/apps/mcp/tools", get(list_mcp_tools))
766	.route(
767	"/v1/automations",
768	get(list_automations).post(create_automation),
769	)
770	.route(
771	"/v1/automations/{id}",
772	get(get_automation)
773	.patch(update_automation)
774	.delete(delete_automation),
775	)
776	.route("/v1/automations/{id}/run", post(run_automation))
777	.route("/v1/automations/{id}/pause", post(pause_automation))
778	.route("/v1/automations/{id}/resume", post(resume_automation))
779	.route("/v1/automations/{id}/runs", get(list_automation_runs))
780	.route("/v1/usage", get(get_usage))
781	.route("/v1/snapshots", get(list_snapshots))
782	.route("/v1/snapshots/{id}/restore", post(restore_snapshot))
783	.route("/v1/providers", get(list_providers))
784	.route("/v1/providers/{id}/models", get(list_provider_models))
785	.route("/v1/providers/{id}/switch", post(switch_provider))
786	.route("/v1/config", get(get_config).post(set_config))
787	.route("/v1/config/reload", post(reload_config))
788	.route_layer(middleware::from_fn_with_state(
789	state.clone(),
790	require_runtime_token,
791	));
792
793	Router::new()
794	.route("/", get(web::web_page))
795	.route("/assets/codewhale-web.css", get(web::web_styles))
796	.route("/assets/codewhale-web.js", get(web::web_script))
797	.route(
798	"/__codewhale/bootstrap/{nonce}",
799	get(web::exchange_bootstrap),
800	)
801	.route("/health", get(health))
802	.route("/mobile", get(mobile_page))
803	.route("/mobile/", get(mobile_page))
804	.route("/v1/runtime/info", get(runtime_info))
805	.merge(api_routes)
806	.layer(cors_layer(&state.cors_origins))
807	.with_state(state)
808	}
809
810	async fn mobile_page(State(state): State<RuntimeApiState>, req: Request) -> Response {
811	if !state.mobile_enabled {
812	return (
813	StatusCode::NOT_FOUND,
814	"mobile control is disabled; start with `codewhale serve --mobile`",
815	)
816	.into_response();
817	}
818	let _ = req;
819	Html(MOBILE_HTML).into_response()
820	}
821
822	fn print_mobile_urls(addr: SocketAddr, auth_enabled: bool, generated_auth: bool, show_qr: bool) {
823	println!("Mobile control page enabled.");
824
825	let port = addr.port();
826	let qr_url = if addr.ip().is_unspecified() {
827	println!(" Local: http://127.0.0.1:{port}/mobile");
828	if let Some(ip) = detect_lan_ip() {
829	let lan_url = format!("http://{ip}:{port}/mobile");
830	println!(" LAN: {lan_url}");
831	lan_url
832	} else {
833	println!(" LAN: bind is 0.0.0.0; open http://<this-machine-ip>:{port}/mobile");
834	format!("http://127.0.0.1:{port}/mobile")
835	}
836	} else {
837	let url = format!("http://{addr}/mobile");
838	println!(" URL: {url}");
839	url
840	};
841	if auth_enabled {
842	if generated_auth {
843	println!(
844	" Auth uses an unprinted generated token; restart with CODEWHALE_RUNTIME_TOKEN or --auth-token to sign in from another client."
845	);
846	} else {
847	println!(" Enter the configured runtime token in the page connection field.");
848	}
849	}
850	println!("Mobile security: use only on a trusted LAN/VPN; this server does not provide TLS.");
851
852	if show_qr {
853	match qrcode::QrCode::new(qr_url.as_bytes()) {
854	Ok(qr) => {
855	let qr_str = qr.render::<qrcode::render::unicode::Dense1x2>().build();
856	println!("\n{qr_str}");
857	}
858	Err(e) => {
859	eprintln!("Warning: could not generate QR code: {e}");
860	}
861	}
862	}
863	}
864
865	#[cfg(test)]
866	fn url_query_component(value: &str) -> String {
867	let mut encoded = String::with_capacity(value.len());
868	for byte in value.bytes() {
869	match byte {
870	b'A'..=b'Z' \| b'a'..=b'z' \| b'0'..=b'9' \| b'-' \| b'.' \| b'_' \| b'~' => {
871	encoded.push(byte as char);
872	}
873	_ => {
874	use std::fmt::Write as _;
875	let _ = write!(encoded, "%{byte:02X}");
876	}
877	}
878	}
879	encoded
880	}
881
882	fn detect_lan_ip() -> Option<String> {
883	let socket = UdpSocket::bind("0.0.0.0:0").ok()?;
884	// UDP connect only selects the outbound interface locally; no packet is sent.
885	socket.connect("10.255.255.255:1").ok()?;
886	let addr = socket.local_addr().ok()?;
887	Some(addr.ip().to_string())
888	}
889
890	async fn health() -> Json<HealthResponse> {
891	Json(HealthResponse {
892	status: "ok",
893	service: "codewhale-runtime-api",
894	mode: "local",
895	})
896	}
897
898	async fn create_task(
899	State(state): State<RuntimeApiState>,
900	Json(mut req): Json<NewTaskRequest>,
901	) -> Result<(StatusCode, Json<TaskRecord>), ApiError> {
902	if req.prompt.trim().is_empty() {
903	return Err(ApiError::bad_request("prompt is required"));
904	}
905	if req.workspace.is_none() {
906	req.workspace = Some(state.workspace.clone());
907	}
908	if req.model.is_none() {
909	req.model = Some(
910	state
911	.config
912	.read()
913	.default_text_model
914	.clone()
915	.unwrap_or_else(\|\| DEFAULT_TEXT_MODEL.to_string()),
916	);
917	}
918	let task = state
919	.task_manager
920	.add_task(req)
921	.await
922	.map_err(\|e\| ApiError::bad_request(e.to_string()))?;
923	Ok((StatusCode::CREATED, Json(task)))
924	}
925
926	async fn create_thread(
927	State(state): State<RuntimeApiState>,
928	Json(mut req): Json<CreateThreadRequest>,
929	) -> Result<(StatusCode, Json<ThreadRecord>), ApiError> {
930	if req.workspace.is_none() {
931	req.workspace = Some(state.workspace.clone());
932	}
933	if req.mode.as_ref().is_none_or(\|m\| m.trim().is_empty()) {
934	req.mode = Some("agent".to_string());
935	}
936
937	let thread = state
938	.runtime_threads
939	.create_thread(req)
940	.await
941	.map_err(\|e\| ApiError::bad_request(e.to_string()))?;
942	Ok((StatusCode::CREATED, Json(thread)))
943	}
944
945	async fn list_threads(
946	State(state): State<RuntimeApiState>,
947	Query(query): Query<ThreadsQuery>,
948	) -> Result<Json<Vec<ThreadRecord>>, ApiError> {
949	let filter = resolve_thread_filter(query.include_archived, query.archived_only);
950	let threads = state
951	.runtime_threads
952	.list_threads(filter, query.limit)
953	.await
954	.map_err(\|e\| ApiError::internal(e.to_string()))?;
955	Ok(Json(threads))
956	}
957
958	async fn list_threads_summary(
959	State(state): State<RuntimeApiState>,
960	Query(query): Query<ThreadSummaryQuery>,
961	) -> Result<Json<Vec<ThreadSummary>>, ApiError> {
962	let limit = query.limit.unwrap_or(50).clamp(1, 500);
963	let search = query.search.as_deref().map(str::to_ascii_lowercase);
964	let filter = resolve_thread_filter(query.include_archived, query.archived_only);
965	let threads = state
966	.runtime_threads
967	.list_threads(filter, Some(limit))
968	.await
969	.map_err(\|e\| ApiError::internal(e.to_string()))?;
970
971	let mut summaries = Vec::new();
972	for thread in threads {
973	let detail = state
974	.runtime_threads
975	.get_thread_detail(&thread.id)
976	.await
977	.map_err(map_thread_err)?;
978	let latest_turn = detail.turns.last();
979	let latest_status =
980	latest_turn.map(\|turn\| format!("{:?}", turn.status).to_ascii_lowercase());
981
982	let title = thread
983	.title
984	.as_deref()
985	.map(str::trim)
986	.filter(\|t\| !t.is_empty())
987	.map(\|t\| truncate_text(t, 72))
988	.unwrap_or_else(\|\| {
989	latest_turn
990	.map(\|turn\| {
991	if turn.input_summary.trim().is_empty() {
992	"New Thread".to_string()
993	} else {
994	truncate_text(&turn.input_summary, 72)
995	}
996	})
997	.unwrap_or_else(\|\| "New Thread".to_string())
998	});
999
1000	let preview = detail
1001	.items
1002	.iter()
1003	.rev()
1004	.find_map(\|item\| match item.kind {
1005	TurnItemKind::AgentMessage \| TurnItemKind::UserMessage => {
1006	let text = item.detail.clone().unwrap_or_else(\|\| item.summary.clone());
1007	if text.trim().is_empty() {
1008	None
1009	} else {
1010	Some(truncate_text(&text, 140))
1011	}
1012	}
1013	_ => None,
1014	})
1015	.unwrap_or_else(\|\| title.clone());
1016
1017	if let Some(search) = &search {
1018	let haystack = format!(
1019	"{} {} {} {}",
1020	thread.id.to_ascii_lowercase(),
1021	title.to_ascii_lowercase(),
1022	preview.to_ascii_lowercase(),
1023	thread.model.to_ascii_lowercase()
1024	);
1025	if !haystack.contains(search) {
1026	continue;
1027	}
1028	}
1029
1030	let workspace_git = collect_workspace_git_metadata(&thread.workspace);
1031	summaries.push(ThreadSummary {
1032	id: thread.id,
1033	title,
1034	preview,
1035	model: thread.model,
1036	mode: thread.mode,
1037	branch: workspace_git.branch,
1038	head: workspace_git.head,
1039	dirty: workspace_git.dirty,
1040	workspace: thread.workspace,
1041	archived: thread.archived,
1042	updated_at: thread.updated_at,
1043	latest_turn_id: thread.latest_turn_id,
1044	latest_turn_status: latest_status,
1045	});
1046	}
1047
1048	if summaries.len() > limit {
1049	summaries.truncate(limit);
1050	}
1051
1052	Ok(Json(summaries))
1053	}
1054
1055	async fn list_agent_runs(
1056	State(state): State<RuntimeApiState>,
1057	) -> Result<Json<AgentRunsResponse>, ApiError> {
1058	let runs = load_persisted_agent_worker_records(&state.workspace).map_err(\|err\| {
1059	ApiError::internal(format!("Failed to load persisted agent run records: {err}"))
1060	})?;
1061	Ok(Json(AgentRunsResponse { runs }))
1062	}
1063
1064	async fn get_agent_run(
1065	State(state): State<RuntimeApiState>,
1066	Path(run_id): Path<String>,
1067	) -> Result<Json<AgentWorkerRecord>, ApiError> {
1068	let runs = load_persisted_agent_worker_records(&state.workspace).map_err(\|err\| {
1069	ApiError::internal(format!("Failed to load persisted agent run records: {err}"))
1070	})?;
1071	let run = runs
1072	.into_iter()
1073	.find(\|record\| {
1074	let effective_run_id = if record.spec.run_id.is_empty() {
1075	record.spec.worker_id.as_str()
1076	} else {
1077	record.spec.run_id.as_str()
1078	};
1079	effective_run_id == run_id \|\| record.spec.worker_id == run_id
1080	})
1081	.ok_or_else(\|\| ApiError::not_found(format!("agent run '{run_id}' not found")))?;
1082	Ok(Json(run))
1083	}
1084
1085	async fn create_fleet_run(
1086	State(state): State<RuntimeApiState>,
1087	Json(request): Json<CreateFleetRunRequest>,
1088	) -> Result<(StatusCode, Json<Value>), ApiError> {
1089	if request.target != FleetRuntimeTarget::ThisComputer {
1090	return Err(ApiError::not_implemented(format!(
1091	"Fleet target {:?} is not available in this local Runtime; choose this_computer",
1092	request.target
1093	)));
1094	}
1095	let (document, descriptor, max_workers) = prepare_managed_fleet_run(request)?;
1096	let manager = open_fleet_manager(&state)?;
1097	let report = manager
1098	.create_queued_run_with_descriptor(document, max_workers, descriptor)
1099	.map_err(\|error\| ApiError::bad_request(format!("Failed to create Fleet run: {error}")))?;
1100	let ledger_state = manager
1101	.rebuild_state()
1102	.map_err(\|error\| ApiError::internal(format!("Failed to rebuild Fleet state: {error}")))?;
1103	let run = ledger_state
1104	.runs
1105	.get(&report.run_id.0)
1106	.ok_or_else(\|\| ApiError::internal("Created Fleet run was missing from its ledger"))?;
1107	Ok((
1108	StatusCode::CREATED,
1109	Json(json!({
1110	"execution": "awaiting_start",
1111	"run": fleet_run_detail_json(&manager, run, &ledger_state)?,
1112	"warnings": report.warnings,
1113	})),
1114	))
1115	}
1116
1117	fn prepare_managed_fleet_run(
1118	request: CreateFleetRunRequest,
1119	) -> Result<(FleetTaskSpecDocument, ManagedFleetRunDescriptor, usize), ApiError> {
1120	if request.security_policy.is_some() {
1121	return Err(ApiError::not_implemented(
1122	"Managed Fleet security_policy overrides are not executable yet; use named roles and bounded task workspace/tool scopes",
1123	));
1124	}
1125	if !request.worker_specs.is_empty() {
1126	return Err(ApiError::not_implemented(
1127	"Managed Fleet custom worker_specs are not available yet; local Runtime worker IDs are generated per run so worker controls cannot collide across Fleets",
1128	));
1129	}
1130	if request.roles.is_empty() {
1131	return Err(ApiError::bad_request(
1132	"roles must declare at least one named Fleet role",
1133	));
1134	}
1135	if request.roles.len() > 128 {
1136	return Err(ApiError::bad_request(
1137	"roles cannot contain more than 128 entries",
1138	));
1139	}
1140	let workflow_id = managed_fleet_token("workflow.id", &request.workflow.id)?;
1141	let workflow_kind = request.workflow.kind;
1142	let name = request
1143	.name
1144	.as_deref()
1145	.map(str::trim)
1146	.filter(\|name\| !name.is_empty())
1147	.unwrap_or(workflow_id.as_str())
1148	.to_string();
1149	if name.len() > 256 \|\| name.chars().any(char::is_control) {
1150	return Err(ApiError::bad_request(
1151	"name must be one printable line no longer than 256 bytes",
1152	));
1153	}
1154
1155	let mut roles = BTreeMap::new();
1156	for role in request.roles {
1157	let normalized = canonical_public_role_name(&managed_fleet_token("role.name", &role.name)?);
1158	let agent_profile = role
1159	.agent_profile
1160	.as_deref()
1161	.map(\|profile\| managed_fleet_token("role.agent_profile", profile))
1162	.transpose()?;
1163	if roles.insert(normalized.clone(), agent_profile).is_some() {
1164	return Err(ApiError::bad_request(format!(
1165	"duplicate Fleet role '{normalized}'"
1166	)));
1167	}
1168	}
1169
1170	let mut tasks = request.workflow.tasks;
1171	let mut used_roles = BTreeSet::new();
1172	for task in &mut tasks {
1173	let worker = task.worker.as_mut().ok_or_else(\|\| {
1174	ApiError::bad_request(format!(
1175	"Fleet task '{}' must select one named role through worker.role",
1176	task.id
1177	))
1178	})?;
1179	let role = worker.role.as_deref().ok_or_else(\|\| {
1180	ApiError::bad_request(format!(
1181	"Fleet task '{}' must select one named role through worker.role",
1182	task.id
1183	))
1184	})?;
1185	let role = canonical_public_role_name(&managed_fleet_token("task.worker.role", role)?);
1186	let declared_profile = roles.get(&role).ok_or_else(\|\| {
1187	ApiError::bad_request(format!(
1188	"Fleet task '{}' references undeclared role '{role}'",
1189	task.id
1190	))
1191	})?;
1192	if let Some(profile) = declared_profile {
1193	match worker.agent_profile.as_deref() {
1194	Some(task_profile) if task_profile != profile => {
1195	return Err(ApiError::bad_request(format!(
1196	"Fleet task '{}' overrides role '{role}' agent_profile '{profile}' with '{task_profile}'",
1197	task.id
1198	)));
1199	}
1200	None => worker.agent_profile = Some(profile.clone()),
1201	Some(_) => {}
1202	}
1203	}
1204	worker.role = Some(role.clone());
1205	used_roles.insert(role);
1206	}
1207	let unused_roles = roles
1208	.keys()
1209	.filter(\|role\| !used_roles.contains(*role))
1210	.cloned()
1211	.collect::<Vec<_>>();
1212	if !unused_roles.is_empty() {
1213	return Err(ApiError::bad_request(format!(
1214	"Every declared Fleet role must own a Workflow task; unused roles: {}",
1215	unused_roles.join(", ")
1216	)));
1217	}
1218	reject_parallel_write_collisions(&tasks)?;
1219
1220	let default_workers = roles.len().min(tasks.len()).max(1);
1221	let max_workers = request.max_workers.unwrap_or(default_workers);
1222	if !(1..=128).contains(&max_workers) {
1223	return Err(ApiError::bad_request(
1224	"max_workers must be between 1 and 128",
1225	));
1226	}
1227	let role_names = roles.into_keys().collect::<Vec<_>>();
1228	Ok((
1229	FleetTaskSpecDocument {
1230	name: Some(name),
1231	labels: request.labels,
1232	security_policy: None,
1233	workers: Vec::new(),
1234	tasks,
1235	},
1236	ManagedFleetRunDescriptor {
1237	target: Some(request.target),
1238	workflow: Some(FleetWorkflowDescriptor {
1239	id: workflow_id,
1240	kind: workflow_kind,
1241	}),
1242	roles: role_names,
1243	},
1244	max_workers,
1245	))
1246	}
1247
1248	fn managed_fleet_token(field: &str, value: &str) -> Result<String, ApiError> {
1249	let value = value.trim();
1250	if value.is_empty()
1251	\|\| value.len() > 128
1252	\|\| !value
1253	.chars()
1254	.all(\|ch\| ch.is_ascii_alphanumeric() \|\| matches!(ch, '-' \| '_' \| '.'))
1255	{
1256	return Err(ApiError::bad_request(format!(
1257	"{field} must be a simple ASCII token no longer than 128 bytes"
1258	)));
1259	}
1260	Ok(value.to_string())
1261	}
1262
1263	fn reject_parallel_write_collisions(tasks: &[FleetTaskSpec]) -> Result<(), ApiError> {
1264	let mut claims: Vec<(String, String)> = Vec::new();
1265	for task in tasks {
1266	let write_roots = fleet_write_roots(task).map_err(\|error\| {
1267	ApiError::bad_request(format!(
1268	"Fleet task '{}' has an invalid write scope: {error}",
1269	task.id
1270	))
1271	})?;
1272	for normalized in write_roots {
1273	for (owner, existing) in &claims {
1274	if owner != &task.id && managed_paths_overlap(existing.as_str(), &normalized) {
1275	return Err(ApiError::bad_request(format!(
1276	"Parallel Workflow write scope collision: tasks '{owner}' and '{}' both claim overlapping paths",
1277	task.id
1278	)));
1279	}
1280	}
1281	claims.push((task.id.clone(), normalized));
1282	}
1283	}
1284	Ok(())
1285	}
1286
1287	fn managed_paths_overlap(left: &str, right: &str) -> bool {
1288	left == right
1289	\|\| left
1290	.strip_prefix(right)
1291	.is_some_and(\|suffix\| suffix.starts_with('/'))
1292	\|\| right
1293	.strip_prefix(left)
1294	.is_some_and(\|suffix\| suffix.starts_with('/'))
1295	}
1296
1297	async fn start_fleet_run(
1298	State(state): State<RuntimeApiState>,
1299	Path(run_id): Path<String>,
1300	) -> Result<(StatusCode, Json<Value>), ApiError> {
1301	let manager = open_fleet_manager(&state)?;
1302	let durable = manager
1303	.rebuild_state()
1304	.map_err(\|error\| ApiError::internal(format!("Failed to rebuild Fleet state: {error}")))?;
1305	let run = durable
1306	.runs
1307	.get(&run_id)
1308	.ok_or_else(\|\| ApiError::not_found(format!("fleet run '{run_id}' not found")))?;
1309	match run.target {
1310	Some(FleetRuntimeTarget::ThisComputer) => {}
1311	Some(target) => {
1312	return Err(ApiError::not_implemented(format!(
1313	"Fleet target {target:?} is not available in this local Runtime"
1314	)));
1315	}
1316	None => {
1317	return Err(ApiError::bad_request(
1318	"Fleet run has no explicit Runtime target and cannot be started through the managed API",
1319	));
1320	}
1321	}
1322	if run.workflow.is_none() \|\| run.roles.is_empty() {
1323	return Err(ApiError::bad_request(
1324	"Fleet run has no managed Workflow/role descriptor and cannot be started through the managed API",
1325	));
1326	}
1327	let run_id = FleetRunId::from(run_id);
1328	let report = manager.activate_run(&run_id).map_err(\|error\| {
1329	let message = format!("Failed to start Fleet run '{}': {error}", run_id.0);
1330	if message.contains("already terminal") {
1331	ApiError::conflict(message)
1332	} else {
1333	ApiError::bad_request(message)
1334	}
1335	})?;
1336	let max_workers = durable
1337	.runs
1338	.get(&run_id.0)
1339	.and_then(\|run\| run.max_workers)
1340	.unwrap_or_else(\|\| report.worker_ids.len().max(1));
1341	let workspace = state.workspace.clone();
1342	let codewhale_binary = state.fleet_codewhale_binary.clone();
1343	let execution_run_id = run_id.clone();
1344	tokio::spawn(async move {
1345	let mut executor = FleetExecutor::new(&workspace);
1346	if let Err(error) = manager
1347	.run_to_completion(
1348	&execution_run_id,
1349	max_workers,
1350	&mut executor,
1351	&codewhale_binary,
1352	None,
1353	Duration::from_millis(250),
1354	)
1355	.await
1356	{
1357	tracing::error!(
1358	run_id = %execution_run_id.0,
1359	error = %error,
1360	"Runtime API Fleet manager exited with an error"
1361	);
1362	}
1363	});
1364	Ok((
1365	StatusCode::ACCEPTED,
1366	Json(json!({
1367	"action": "start",
1368	"execution": "scheduled",
1369	"run_id": run_id.0,
1370	"target": "this_computer",
1371	"leased": report.leased,
1372	"queued": report.queued,
1373	"worker_ids": report.worker_ids,
1374	})),
1375	))
1376	}
1377
1378	async fn replay_fleet_events(
1379	State(state): State<RuntimeApiState>,
1380	Path(run_id): Path<String>,
1381	Query(query): Query<FleetEventsQuery>,
1382	) -> Result<Json<FleetEventReplay>, ApiError> {
1383	let (after, limit) = validate_fleet_events_query(query)?;
1384	let replay = load_fleet_event_replay(state, FleetRunId::from(run_id), after, limit)
1385	.await
1386	.map_err(map_fleet_replay_error)?;
1387	Ok(Json(replay))
1388	}
1389
1390	async fn stream_fleet_events(
1391	State(state): State<RuntimeApiState>,
1392	Path(run_id): Path<String>,
1393	Query(query): Query<FleetEventsQuery>,
1394	) -> Result<Sse<impl futures_util::Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
1395	let (after, limit) = validate_fleet_events_query(query)?;
1396	let run_id = FleetRunId::from(run_id);
1397	let initial = load_fleet_event_replay(state.clone(), run_id.clone(), after.clone(), limit)
1398	.await
1399	.map_err(map_fleet_replay_error)?;
1400	let event_stream = replay_live_fleet_events(state, run_id, after, limit, initial);
1401	Ok(Sse::new(event_stream).keep_alive(
1402	KeepAlive::new()
1403	.interval(Duration::from_secs(15))
1404	.text("keepalive"),
1405	))
1406	}
1407
1408	fn replay_live_fleet_events(
1409	state: RuntimeApiState,
1410	run_id: FleetRunId,
1411	mut after: Option<String>,
1412	limit: usize,
1413	initial: FleetEventReplay,
1414	) -> impl futures_util::Stream<Item = Result<SseEvent, Infallible>> {
1415	stream! {
1416	let mut page = initial;
1417	loop {
1418	if page.history_truncated {
1419	yield Ok(sse_json(
1420	"fleet.replay.truncated",
1421	json!({
1422	"run_id": run_id.0.clone(),
1423	"reload_projection": true,
1424	}),
1425	));
1426	}
1427	for event in page.events {
1428	after = Some(event.cursor.clone());
1429	yield Ok(fleet_sse_event(&event));
1430	}
1431	if !page.has_more {
1432	tokio::time::sleep(Duration::from_millis(250)).await;
1433	}
1434	match load_fleet_event_replay(
1435	state.clone(),
1436	run_id.clone(),
1437	after.clone(),
1438	limit,
1439	)
1440	.await
1441	{
1442	Ok(next) => page = next,
1443	Err(FleetEventReplayError::CursorUnavailable { .. }) => {
1444	yield Ok(sse_json(
1445	"fleet.replay.cursor_unavailable",
1446	json!({
1447	"run_id": run_id.0.clone(),
1448	"reload_projection": true,
1449	}),
1450	));
1451	return;
1452	}
1453	Err(error) => {
1454	tracing::warn!(
1455	run_id = %run_id.0,
1456	error = %error,
1457	"Fleet event stream stopped while reading durable history"
1458	);
1459	yield Ok(sse_json(
1460	"fleet.stream.error",
1461	json!({ "retryable": true }),
1462	));
1463	return;
1464	}
1465	}
1466	}
1467	}
1468	}
1469
1470	async fn load_fleet_event_replay(
1471	state: RuntimeApiState,
1472	run_id: FleetRunId,
1473	after: Option<String>,
1474	limit: usize,
1475	) -> std::result::Result<FleetEventReplay, FleetEventReplayError> {
1476	tokio::task::spawn_blocking(move \|\| {
1477	let manager =
1478	open_fleet_manager(&state).map_err(\|error\| FleetEventReplayError::Storage {
1479	message: error.message,
1480	})?;
1481	manager.replay_events(&run_id, after.as_deref(), limit)
1482	})
1483	.await
1484	.map_err(\|error\| FleetEventReplayError::Storage {
1485	message: format!("Fleet replay worker failed: {error}"),
1486	})?
1487	}
1488
1489	fn validate_fleet_events_query(
1490	query: FleetEventsQuery,
1491	) -> Result<(Option<String>, usize), ApiError> {
1492	let after = query
1493	.after
1494	.map(\|cursor\| cursor.trim().to_string())
1495	.filter(\|cursor\| !cursor.is_empty());
1496	if after.as_deref().is_some_and(\|cursor\| {
1497	cursor.len() > 96
1498	\|\| !cursor.starts_with("fev1_")
1499	\|\| !cursor
1500	.chars()
1501	.all(\|ch\| ch.is_ascii_alphanumeric() \|\| ch == '_')
1502	}) {
1503	return Err(ApiError::bad_request(
1504	"after is not a valid Fleet event cursor",
1505	));
1506	}
1507	let limit = query.limit.unwrap_or(DEFAULT_FLEET_EVENT_REPLAY_LIMIT);
1508	if !(1..=MAX_FLEET_EVENT_REPLAY_LIMIT).contains(&limit) {
1509	return Err(ApiError::bad_request(format!(
1510	"limit must be between 1 and {MAX_FLEET_EVENT_REPLAY_LIMIT}"
1511	)));
1512	}
1513	Ok((after, limit))
1514	}
1515
1516	fn map_fleet_replay_error(error: FleetEventReplayError) -> ApiError {
1517	let message = error.to_string();
1518	match error {
1519	FleetEventReplayError::UnknownRun { .. } => ApiError::not_found(message),
1520	FleetEventReplayError::CursorUnavailable { .. } => ApiError::conflict(message),
1521	FleetEventReplayError::Storage { .. } => ApiError::internal(message),
1522	}
1523	}
1524
1525	fn fleet_sse_event(event: &FleetRuntimeEvent) -> SseEvent {
1526	let data = serde_json::to_string(event).unwrap_or_else(\|_\| "{}".to_string());
1527	SseEvent::default()
1528	.id(event.cursor.clone())
1529	.event(event.event.clone())
1530	.data(data)
1531	}
1532
1533	async fn list_fleet_runs(State(state): State<RuntimeApiState>) -> Result<Json<Value>, ApiError> {
1534	let manager = open_fleet_manager(&state)?;
1535	let ledger_state = manager
1536	.rebuild_state()
1537	.map_err(\|err\| ApiError::internal(format!("Failed to rebuild fleet state: {err}")))?;
1538	let runs: Vec<_> = ledger_state
1539	.runs
1540	.values()
1541	.map(\|run\| fleet_run_summary_json(&manager, run, &ledger_state))
1542	.collect::<Result<Vec<_>, _>>()?;
1543	let status = manager
1544	.status()
1545	.map_err(\|err\| ApiError::internal(format!("Failed to read fleet status: {err}")))?;
1546	Ok(Json(json!({
1547	"status": fleet_status_json(&status),
1548	"runs": runs,
1549	})))
1550	}
1551
1552	async fn get_fleet_run(
1553	State(state): State<RuntimeApiState>,
1554	Path(run_id): Path<String>,
1555	) -> Result<Json<Value>, ApiError> {
1556	let manager = open_fleet_manager(&state)?;
1557	let ledger_state = manager
1558	.rebuild_state()
1559	.map_err(\|err\| ApiError::internal(format!("Failed to rebuild fleet state: {err}")))?;
1560	let run = ledger_state
1561	.runs
1562	.get(&run_id)
1563	.ok_or_else(\|\| ApiError::not_found(format!("fleet run '{run_id}' not found")))?;
1564	Ok(Json(fleet_run_detail_json(&manager, run, &ledger_state)?))
1565	}
1566
1567	async fn list_fleet_run_workers(
1568	State(state): State<RuntimeApiState>,
1569	Path(run_id): Path<String>,
1570	) -> Result<Json<Value>, ApiError> {
1571	let manager = open_fleet_manager(&state)?;
1572	let ledger_state = manager
1573	.rebuild_state()
1574	.map_err(\|err\| ApiError::internal(format!("Failed to rebuild fleet state: {err}")))?;
1575	let run = ledger_state
1576	.runs
1577	.get(&run_id)
1578	.ok_or_else(\|\| ApiError::not_found(format!("fleet run '{run_id}' not found")))?;
1579	let workers = run
1580	.worker_specs
1581	.iter()
1582	.map(\|worker\| {
1583	manager
1584	.inspect_worker(&worker.id)
1585	.map(\|inspection\| fleet_worker_json(&inspection))
1586	.map_err(\|err\| {
1587	ApiError::internal(format!(
1588	"Failed to inspect fleet worker {}: {err}",
1589	worker.id
1590	))
1591	})
1592	})
1593	.collect::<Result<Vec<_>, _>>()?;
1594	Ok(Json(json!({
1595	"run_id": run_id,
1596	"workers": workers,
1597	})))
1598	}
1599
1600	async fn get_fleet_worker(
1601	State(state): State<RuntimeApiState>,
1602	Path(worker_id): Path<String>,
1603	) -> Result<Json<Value>, ApiError> {
1604	let manager = open_fleet_manager(&state)?;
1605	let inspection = manager.inspect_worker(&worker_id).map_err(\|err\| {
1606	ApiError::not_found(format!("fleet worker '{worker_id}' not found: {err}"))
1607	})?;
1608	Ok(Json(fleet_worker_json(&inspection)))
1609	}
1610
1611	async fn interrupt_fleet_worker(
1612	State(state): State<RuntimeApiState>,
1613	Path(worker_id): Path<String>,
1614	) -> Result<Json<Value>, ApiError> {
1615	let manager = open_fleet_manager(&state)?;
1616	let inspection = manager.interrupt_worker(&worker_id).map_err(\|err\| {
1617	ApiError::bad_request(format!(
1618	"Failed to interrupt fleet worker '{worker_id}': {err}"
1619	))
1620	})?;
1621	Ok(Json(json!({
1622	"action": "interrupt",
1623	"worker": fleet_worker_json(&inspection),
1624	})))
1625	}
1626
1627	async fn stop_fleet_worker(
1628	State(state): State<RuntimeApiState>,
1629	Path(worker_id): Path<String>,
1630	) -> Result<Json<Value>, ApiError> {
1631	let manager = open_fleet_manager(&state)?;
1632	let inspection = manager.interrupt_worker(&worker_id).map_err(\|err\| {
1633	ApiError::bad_request(format!("Failed to stop fleet worker '{worker_id}': {err}"))
1634	})?;
1635	Ok(Json(json!({
1636	"action": "stop",
1637	"worker": fleet_worker_json(&inspection),
1638	})))
1639	}
1640
1641	async fn restart_fleet_worker(
1642	State(state): State<RuntimeApiState>,
1643	Path(worker_id): Path<String>,
1644	) -> Result<Json<Value>, ApiError> {
1645	let manager = open_fleet_manager(&state)?;
1646	let report = manager.restart_worker(&worker_id).map_err(\|err\| {
1647	ApiError::bad_request(format!(
1648	"Failed to restart fleet worker '{worker_id}': {err}"
1649	))
1650	})?;
1651	let worker = fleet_worker_json(&report.inspection);
1652	let run_id = report.run_id.clone();
1653	let max_workers = report.max_workers;
1654	let workspace = state.workspace.clone();
1655	let codewhale_binary = state.fleet_codewhale_binary.clone();
1656	tokio::spawn(async move {
1657	let mut executor = FleetExecutor::new(&workspace);
1658	if let Err(err) = manager
1659	.run_to_completion(
1660	&run_id,
1661	max_workers,
1662	&mut executor,
1663	&codewhale_binary,
1664	None,
1665	Duration::from_millis(250),
1666	)
1667	.await
1668	{
1669	tracing::error!(
1670	run_id = %run_id.0,
1671	error = %err,
1672	"Runtime API Fleet restart manager exited with an error"
1673	);
1674	}
1675	});
1676	Ok(Json(json!({
1677	"action": "restart",
1678	"execution": "scheduled",
1679	"run_id": report.run_id.0,
1680	"worker": worker,
1681	})))
1682	}
1683
1684	async fn stop_fleet_run(
1685	State(state): State<RuntimeApiState>,
1686	Path(run_id): Path<String>,
1687	) -> Result<Json<Value>, ApiError> {
1688	let manager = open_fleet_manager(&state)?;
1689	let run_id = FleetRunId::from(run_id);
1690	let stopped = manager.stop_run(&run_id).map_err(\|err\| {
1691	ApiError::bad_request(format!("Failed to stop fleet run '{}': {err}", run_id.0))
1692	})?;
1693	let status = manager
1694	.run_status(&run_id)
1695	.map_err(\|err\| ApiError::internal(format!("Failed to read fleet run status: {err}")))?;
1696	Ok(Json(json!({
1697	"action": "stop",
1698	"run_id": run_id.0,
1699	"stopped": stopped,
1700	"status": fleet_status_json(&status),
1701	})))
1702	}
1703
1704	fn open_fleet_manager(state: &RuntimeApiState) -> Result<FleetManager, ApiError> {
1705	let (exec_config, fleet_config, session_model, route_config) = {
1706	let config = state.config.read();
1707	let exec_config = config
1708	.fleet
1709	.as_ref()
1710	.map(\|fleet\| fleet.exec.clone())
1711	.unwrap_or_default();
1712	// The active session route is the operator: workers without a
1713	// task/profile model pin inherit the model the user picked in /model.
1714	(
1715	exec_config,
1716	config.fleet_config(),
1717	config.default_model(),
1718	config.clone(),
1719	)
1720	};
1721	FleetManager::open(&state.workspace)
1722	.map(\|manager\| {
1723	manager
1724	.with_exec_config(exec_config)
1725	.with_fleet_config(fleet_config)
1726	.with_sub_agent_manager(state.sub_agent_manager.clone())
1727	.with_session_model(session_model)
1728	.with_route_config(route_config)
1729	})
1730	.map_err(\|err\| ApiError::internal(format!("Failed to open fleet manager: {err}")))
1731	}
1732
1733	fn fleet_run_summary_json(
1734	manager: &FleetManager,
1735	run: &FleetRun,
1736	ledger_state: &FleetLedgerState,
1737	) -> Result<Value, ApiError> {
1738	let status = manager
1739	.run_status(&run.id)
1740	.map_err(\|err\| ApiError::internal(format!("Failed to read fleet run status: {err}")))?;
1741	let task_statuses = ledger_state
1742	.tasks
1743	.values()
1744	.filter(\|task\| task.entry.run_id == run.id)
1745	.map(\|task\| {
1746	json!({
1747	"task_id": task.entry.task_id.clone(),
1748	"status": fleet_task_status_label(task.status),
1749	"leased_to": task.leased_to.clone(),
1750	"attempts": task.entry.attempts,
1751	})
1752	})
1753	.collect::<Vec<_>>();
1754	Ok(json!({
1755	"id": run.id.0.clone(),
1756	"name": run.name.clone(),
1757	"lifecycle_status": ledger_state
1758	.run_status_overrides
1759	.get(&run.id.0)
1760	.unwrap_or(&run.status),
1761	"status": fleet_status_json(&status),
1762	"target": run.target,
1763	"workflow": run.workflow.clone(),
1764	"roles": run.roles.clone(),
1765	"task_count": run.task_specs.len(),
1766	"worker_count": run.worker_specs.len(),
1767	"tasks": task_statuses,
1768	"labels": run.labels.clone(),
1769	"created_at": run.created_at.clone(),
1770	"updated_at": run.updated_at.clone(),
1771	"completed_at": run.completed_at.clone(),
1772	}))
1773	}
1774
1775	fn fleet_run_detail_json(
1776	manager: &FleetManager,
1777	run: &FleetRun,
1778	ledger_state: &FleetLedgerState,
1779	) -> Result<Value, ApiError> {
1780	let mut value = fleet_run_summary_json(manager, run, ledger_state)?;
1781	if let Some(map) = value.as_object_mut() {
1782	map.insert("task_specs".to_string(), json!(run.task_specs.clone()));
1783	map.insert("worker_specs".to_string(), json!(run.worker_specs.clone()));
1784	}
1785	Ok(value)
1786	}
1787
1788	fn fleet_status_json(status: &FleetStatusSnapshot) -> Value {
1789	json!({
1790	"runs": status.runs,
1791	"queued": status.queued,
1792	"running": status.running,
1793	"completed": status.completed,
1794	"partial": status.partial,
1795	"failed": status.failed,
1796	"restarted": status.restarted,
1797	"escalated": status.escalated,
1798	"transport_failed": status.transport_failed,
1799	"task_failed": status.task_failed,
1800	"verifier_failed": status.verifier_failed,
1801	"cancelled": status.cancelled,
1802	"stale": status.stale,
1803	"workers": status
1804	.workers
1805	.iter()
1806	.map(\|(worker_id, status)\| {
1807	(
1808	worker_id.clone(),
1809	Value::String(worker_status_label(status).to_string()),
1810	)
1811	})
1812	.collect::<serde_json::Map<String, Value>>(),
1813	})
1814	}
1815
1816	fn fleet_worker_json(inspection: &FleetWorkerInspection) -> Value {
1817	json!({
1818	"worker_id": inspection.worker_id.clone(),
1819	"status": worker_status_label(&inspection.status),
1820	"run_id": inspection.current_run_id.as_ref().map(\|run_id\| run_id.0.clone()),
1821	"task_id": inspection.current_task_id.clone(),
1822	"objective": inspection.objective.clone(),
1823	"role": inspection.role.clone(),
1824	"host": inspection.host.clone(),
1825	"latest_heartbeat_at": inspection.latest_heartbeat_at.clone(),
1826	"latest_event": inspection.latest_event.as_ref().map(fleet_event_json),
1827	"artifacts": inspection.artifacts.iter().map(fleet_artifact_json).collect::<Vec<_>>(),
1828	"last_error": inspection.last_error.clone(),
1829	"alert_state": inspection.alert_state.clone(),
1830	"runtime_state": inspection.runtime_state.as_ref().map(fleet_worker_runtime_json),
1831	})
1832	}
1833
1834	fn fleet_worker_runtime_json(runtime: &FleetWorkerRuntimeProjection) -> Value {
1835	json!({
1836	"agent_status": runtime.agent_status.clone(),
1837	"steps_taken": runtime.steps_taken,
1838	"latest_message": runtime.latest_message.clone(),
1839	"error": runtime.error.clone(),
1840	"result_summary": runtime.result_summary.clone(),
1841	"has_session": runtime.has_session,
1842	})
1843	}
1844
1845	fn fleet_artifact_json(artifact: &codewhale_protocol::fleet::FleetArtifactRef) -> Value {
1846	json!({
1847	"kind": artifact_kind_label(&artifact.kind),
1848	"path": artifact.path.clone(),
1849	"checksum": artifact.checksum.clone(),
1850	"mime_type": artifact.mime_type.clone(),
1851	"size_bytes": artifact.size_bytes,
1852	})
1853	}
1854
1855	fn fleet_event_json(event: &codewhale_protocol::fleet::FleetWorkerEvent) -> Value {
1856	json!({
1857	"seq": event.seq,
1858	"run_id": event.run_id.0.clone(),
1859	"worker_id": event.worker_id.clone(),
1860	"task_id": event.task_id.clone(),
1861	"timestamp": event.timestamp.clone(),
1862	"label": fleet_event_label(&event.payload),
1863	"payload": event.payload.clone(),
1864	})
1865	}
1866
1867	fn worker_status_label(status: &FleetWorkerStatus) -> &'static str {
1868	match status {
1869	FleetWorkerStatus::Unknown => "unknown",
1870	FleetWorkerStatus::Online => "online",
1871	FleetWorkerStatus::Busy => "busy",
1872	FleetWorkerStatus::Offline => "offline",
1873	FleetWorkerStatus::Unhealthy => "unhealthy",
1874	FleetWorkerStatus::Draining => "draining",
1875	FleetWorkerStatus::Retired => "retired",
1876	}
1877	}
1878
1879	fn fleet_task_status_label(status: FleetTaskLedgerStatus) -> &'static str {
1880	match status {
1881	FleetTaskLedgerStatus::Enqueued => "enqueued",
1882	FleetTaskLedgerStatus::Leased => "leased",
1883	FleetTaskLedgerStatus::Completed => "completed",
1884	FleetTaskLedgerStatus::Failed => "failed",
1885	FleetTaskLedgerStatus::Cancelled => "cancelled",
1886	}
1887	}
1888
1889	fn artifact_kind_label(kind: &FleetArtifactKind) -> String {
1890	match kind {
1891	FleetArtifactKind::Log => "log".to_string(),
1892	FleetArtifactKind::Patch => "patch".to_string(),
1893	FleetArtifactKind::TestResult => "test_result".to_string(),
1894	FleetArtifactKind::Report => "report".to_string(),
1895	FleetArtifactKind::Checkpoint => "checkpoint".to_string(),
1896	FleetArtifactKind::Receipt => "receipt".to_string(),
1897	FleetArtifactKind::Other(value) => value.clone(),
1898	}
1899	}
1900
1901	fn fleet_event_label(payload: &FleetWorkerEventPayload) -> String {
1902	match payload {
1903	FleetWorkerEventPayload::Queued => "queued".to_string(),
1904	FleetWorkerEventPayload::Leased { .. } => "leased".to_string(),
1905	FleetWorkerEventPayload::Starting => "starting".to_string(),
1906	FleetWorkerEventPayload::Running => "running".to_string(),
1907	FleetWorkerEventPayload::ModelWait { model } => model
1908	.as_ref()
1909	.map(\|model\| format!("model_wait model={model}"))
1910	.unwrap_or_else(\|\| "model_wait".to_string()),
1911	FleetWorkerEventPayload::RunningTool { tool, call_id } => call_id
1912	.as_ref()
1913	.map(\|call_id\| format!("running_tool tool={tool} call_id={call_id}"))
1914	.unwrap_or_else(\|\| format!("running_tool tool={tool}")),
1915	FleetWorkerEventPayload::WorkflowEvent {
1916	workflow_run_id,
1917	event,
1918	} => event
1919	.get("type")
1920	.and_then(serde_json::Value::as_str)
1921	.map(\|kind\| format!("workflow_event run_id={workflow_run_id} type={kind}"))
1922	.unwrap_or_else(\|\| format!("workflow_event run_id={workflow_run_id}")),
1923	FleetWorkerEventPayload::Heartbeat { .. } => "heartbeat".to_string(),
1924	FleetWorkerEventPayload::Artifact(artifact) => {
1925	format!("artifact kind={}", artifact_kind_label(&artifact.kind))
1926	}
1927	FleetWorkerEventPayload::Completed { exit_code, summary } => match (exit_code, summary) {
1928	(Some(code), Some(summary)) => format!("completed exit_code={code} {summary}"),
1929	(Some(code), None) => format!("completed exit_code={code}"),
1930	(None, Some(summary)) => format!("completed {summary}"),
1931	(None, None) => "completed".to_string(),
1932	},
1933	FleetWorkerEventPayload::Failed {
1934	reason,
1935	recoverable,
1936	} => {
1937	format!("failed recoverable={recoverable} reason={reason}")
1938	}
1939	FleetWorkerEventPayload::Cancelled { cancelled_by } => cancelled_by
1940	.as_ref()
1941	.map(\|by\| format!("cancelled by={by}"))
1942	.unwrap_or_else(\|\| "cancelled".to_string()),
1943	FleetWorkerEventPayload::Interrupted { signal } => signal
1944	.as_ref()
1945	.map(\|signal\| format!("interrupted signal={signal}"))
1946	.unwrap_or_else(\|\| "interrupted".to_string()),
1947	FleetWorkerEventPayload::Stale { last_heartbeat_at } => last_heartbeat_at
1948	.as_ref()
1949	.map(\|ts\| format!("stale last_heartbeat_at={ts}"))
1950	.unwrap_or_else(\|\| "stale".to_string()),
1951	FleetWorkerEventPayload::Restarted { restart_count } => {
1952	format!("restarted count={restart_count}")
1953	}
1954	FleetWorkerEventPayload::Escalated { channel, alert_id } => alert_id
1955	.as_ref()
1956	.map(\|alert_id\| format!("escalated channel={channel} alert_id={alert_id}"))
1957	.unwrap_or_else(\|\| format!("escalated channel={channel}")),
1958	}
1959	}
1960
1961	async fn list_skills(
1962	State(state): State<RuntimeApiState>,
1963	) -> Result<Json<SkillsResponse>, ApiError> {
1964	let (skills_dir, mode) = {
1965	let config = state.config.read();
1966	let skills_dir = resolve_skills_dir(&config, &state.workspace);
1967	let mode = crate::skills::SkillDiscoveryMode::from_codewhale_only(
1968	config.skills_config().scan_codewhale_only(),
1969	);
1970	(skills_dir, mode)
1971	};
1972	let plugin_registry = state
1973	.plugin_discovery
1974	.registry_for_workspace(&state.workspace);
1975	let (registry, directories) = discover_skills_for_runtime_api(
1976	&state.workspace,
1977	&skills_dir,
1978	mode,
1979	Some(plugin_registry.as_ref()),
1980	);
1981	let mut skill_state = state.skill_state.lock().await;
1982	skill_state
1983	.refresh()
1984	.map_err(\|error\| ApiError::internal(format!("refresh skill state: {error}")))?;
1985	let skills = registry
1986	.list()
1987	.iter()
1988	.map(\|skill\| {
1989	let (path, source, plugin_id, plugin_generation, plugin_content_hash) =
1990	match &skill.source {
1991	crate::skills::SkillSource::Native => (
1992	Some(skill.path.clone()),
1993	"native".to_string(),
1994	None,
1995	None,
1996	None,
1997	),
1998	crate::skills::SkillSource::Plugin {
1999	plugin_id,
2000	plugin_name,
2001	authority,
2002	} => (
2003	None,
2004	format!("reviewed-plugin-snapshot:{plugin_name}"),
2005	Some(plugin_id.clone()),
2006	Some(authority.state_generation),
2007	Some(authority.content_hash.clone()),
2008	),
2009	};
2010	SkillEntry {
2011	name: skill.name.clone(),
2012	description: skill.description.clone(),
2013	path,
2014	source,
2015	plugin_id,
2016	plugin_generation,
2017	plugin_content_hash,
2018	enabled: skill_state.is_enabled(&skill.name),
2019	is_bundled: skill_entry_is_bundled(skill, &skills_dir),
2020	}
2021	})
2022	.collect();
2023	Ok(Json(SkillsResponse {
2024	directory: skills_dir,
2025	directories,
2026	warnings: registry.warnings().to_vec(),
2027	skills,
2028	}))
2029	}
2030
2031	async fn set_skill_enabled(
2032	State(state): State<RuntimeApiState>,
2033	Path(name): Path<String>,
2034	Json(req): Json<SetSkillEnabledRequest>,
2035	) -> Result<Json<SetSkillEnabledResponse>, ApiError> {
2036	let (skills_dir, mode) = {
2037	let config = state.config.read();
2038	let skills_dir = resolve_skills_dir(&config, &state.workspace);
2039	let mode = crate::skills::SkillDiscoveryMode::from_codewhale_only(
2040	config.skills_config().scan_codewhale_only(),
2041	);
2042	(skills_dir, mode)
2043	};
2044	let plugin_registry = state
2045	.plugin_discovery
2046	.registry_for_workspace(&state.workspace);
2047	let (registry, directories) = discover_skills_for_runtime_api(
2048	&state.workspace,
2049	&skills_dir,
2050	mode,
2051	Some(plugin_registry.as_ref()),
2052	);
2053	let exists = registry.list().iter().any(\|skill\| skill.name == name);
2054	if !exists {
2055	return Err(ApiError::not_found(format!(
2056	"skill '{name}' not found in searched directories: {}",
2057	format_skill_search_paths(&directories)
2058	)));
2059	}
2060
2061	let mut store = state.skill_state.lock().await;
2062	store
2063	.set_enabled(&name, req.enabled)
2064	.map_err(\|err\| ApiError::internal(format!("persist skill state: {err}")))?;
2065	Ok(Json(SetSkillEnabledResponse {
2066	name,
2067	enabled: req.enabled,
2068	}))
2069	}
2070
2071	async fn decide_approval(
2072	State(state): State<RuntimeApiState>,
2073	Path(approval_id): Path<String>,
2074	Json(req): Json<DecideApprovalBody>,
2075	) -> Result<Json<DecideApprovalResponse>, ApiError> {
2076	let decision = match req.decision.as_str() {
2077	"allow" => ExternalApprovalDecision::Allow {
2078	remember: req.remember,
2079	},
2080	"deny" => ExternalApprovalDecision::Deny {
2081	remember: req.remember,
2082	},
2083	other => {
2084	return Err(ApiError::bad_request(format!(
2085	"invalid decision '{other}'; expected \"allow\" or \"deny\""
2086	)));
2087	}
2088	};
2089	let delivered = state
2090	.runtime_threads
2091	.deliver_external_approval(&approval_id, decision);
2092	if !delivered {
2093	return Err(ApiError::not_found(format!(
2094	"no pending approval with id '{approval_id}'"
2095	)));
2096	}
2097	Ok(Json(DecideApprovalResponse {
2098	ok: true,
2099	approval_id,
2100	decision: req.decision,
2101	delivered,
2102	}))
2103	}
2104
2105	async fn submit_user_input(
2106	State(state): State<RuntimeApiState>,
2107	Path((thread_id, input_id)): Path<(String, String)>,
2108	Json(req): Json<SubmitUserInputBody>,
2109	) -> Result<Json<SubmitUserInputResponse>, ApiError> {
2110	use crate::tools::user_input::{UserInputAnswer, UserInputResponse};
2111	let answers: Vec<UserInputAnswer> = req
2112	.answers
2113	.into_iter()
2114	.map(\|a\| UserInputAnswer {
2115	id: a.id,
2116	label: a.label,
2117	value: a.value,
2118	})
2119	.collect();
2120	let response = UserInputResponse { answers };
2121	let delivered = state
2122	.runtime_threads
2123	.submit_user_input(&thread_id, &input_id, response)
2124	.await
2125	.map_err(map_thread_err)?;
2126	if !delivered {
2127	return Err(ApiError::not_found(format!(
2128	"no pending user-input request with id '{input_id}'"
2129	)));
2130	}
2131	Ok(Json(SubmitUserInputResponse {
2132	ok: true,
2133	input_id,
2134	delivered,
2135	}))
2136	}
2137
2138	async fn runtime_info(
2139	State(state): State<RuntimeApiState>,
2140	request: Request,
2141	) -> Json<RuntimeInfoResponse> {
2142	let version = env!("CARGO_PKG_VERSION");
2143	let commit = option_env!("CODEWHALE_BUILD_COMMIT").unwrap_or("unknown");
2144	let api_base = runtime_account_api_base();
2145	let account = runtime_account_info_for_request(
2146	runtime_request_is_authorized(&request, &state),
2147	&api_base,
2148	\|\| runtime_account_info(state.config_profile.as_deref(), &api_base),
2149	);
2150	Json(RuntimeInfoResponse {
2151	service: "codewhale-runtime-api",
2152	runtime_api_version: RUNTIME_API_VERSION,
2153	codewhale_version: version,
2154	codewhale_commit: commit,
2155	bind_host: state.bind_host.clone(),
2156	port: state.bind_port,
2157	auth_required: state.auth_required,
2158	transports: vec!["http", "sse"],
2159	capabilities: default_runtime_capabilities(),
2160	account,
2161	experimental: RuntimeExperimentalCapabilities::default(),
2162	version,
2163	})
2164	}
2165
2166	fn runtime_account_info(profile: Option<&str>, api_base: &str) -> RuntimeAccountInfo {
2167	#[cfg(test)]
2168	{
2169	let _ = profile;
2170	RuntimeAccountInfo::signed_out(api_base.to_string())
2171	}
2172
2173	#[cfg(not(test))]
2174	{
2175	secure_account_session_secrets()
2176	.and_then(\|secrets\| {
2177	AccountSessionStore::new(secrets, profile, api_base).runtime_info_at(Utc::now())
2178	})
2179	.unwrap_or_else(\|_\| RuntimeAccountInfo::signed_out(api_base.to_string()))
2180	}
2181	}
2182
2183	fn runtime_account_info_for_request(
2184	authorized: bool,
2185	api_base: &str,
2186	load: impl FnOnce() -> RuntimeAccountInfo,
2187	) -> RuntimeAccountInfo {
2188	if authorized {
2189	load()
2190	} else {
2191	RuntimeAccountInfo::signed_out(api_base.to_string())
2192	}
2193	}
2194
2195	fn runtime_account_api_base() -> String {
2196	std::env::var(ACCOUNT_API_BASE_ENV)
2197	.ok()
2198	.and_then(\|value\| normalize_runtime_account_api_base(&value))
2199	.unwrap_or_else(\|\| DEFAULT_ACCOUNT_API_BASE.to_string())
2200	}
2201
2202	fn normalize_runtime_account_api_base(value: &str) -> Option<String> {
2203	let mut url = reqwest::Url::parse(value.trim()).ok()?;
2204	if !url.username().is_empty()
2205	\|\| url.password().is_some()
2206	\|\| url.query().is_some()
2207	\|\| url.fragment().is_some()
2208	\|\| !matches!(url.path(), "" \| "/")
2209	{
2210	return None;
2211	}
2212	let host = url.host_str()?;
2213	let loopback = host.eq_ignore_ascii_case("localhost")
2214	\|\| host
2215	.trim_start_matches('[')
2216	.trim_end_matches(']')
2217	.parse::<IpAddr>()
2218	.is_ok_and(\|address\| address.is_loopback());
2219	if url.scheme() != "https" && !(url.scheme() == "http" && loopback) {
2220	return None;
2221	}
2222	url.set_path("/");
2223	Some(url.as_str().trim_end_matches('/').to_string())
2224	}
2225
2226	async fn list_mcp_servers(
2227	State(state): State<RuntimeApiState>,
2228	) -> Result<Json<McpServersResponse>, ApiError> {
2229	let mcp_config_path = state.config.read().mcp_config_path();
2230	let plugin_registry = state
2231	.plugin_discovery
2232	.registry_for_workspace(&state.workspace);
2233	let config = crate::mcp::load_config_with_workspace_and_plugins(
2234	&mcp_config_path,
2235	&state.workspace,
2236	plugin_registry.as_ref(),
2237	)
2238	.map_err(\|e\| ApiError::internal(format!("Failed to load MCP config: {e}")))?;
2239
2240	let mut servers = Vec::new();
2241	for (name, server_cfg) in config.servers {
2242	servers.push(McpServerEntry {
2243	name: name.clone(),
2244	enabled: server_cfg.is_enabled(),
2245	required: server_cfg.required,
2246	command: server_cfg.command.clone(),
2247	url: server_cfg.url.clone(),
2248	connected: false,
2249	enabled_tools: server_cfg.enabled_tools.clone(),
2250	disabled_tools: server_cfg.disabled_tools.clone(),
2251	});
2252	}
2253	servers.sort_by(\|a, b\| a.name.cmp(&b.name));
2254
2255	Ok(Json(McpServersResponse { servers }))
2256	}
2257
2258	async fn list_mcp_tools(
2259	State(state): State<RuntimeApiState>,
2260	Query(query): Query<McpToolsQuery>,
2261	) -> Result<Json<McpToolsResponse>, ApiError> {
2262	// Double-checked init: hold the state-level slot mutex only long enough
2263	// to grab (or lazily create) the pool handle. connect_all can stall on a
2264	// slow MCP server and must not run under the slot lock.
2265	let pool_handle = {
2266	let mut pool_slot = state.mcp_pool.lock().await;
2267	match pool_slot.as_ref() {
2268	Some(pool) => Some(Arc::clone(pool)),
2269	None if query.connect => {
2270	let mcp_config_path = state.config.read().mcp_config_path();
2271	let plugin_registry = state
2272	.plugin_discovery
2273	.registry_for_workspace(&state.workspace);
2274	let new_pool = McpPool::from_config_path_with_workspace_and_plugins(
2275	&mcp_config_path,
2276	&state.workspace,
2277	plugin_registry,
2278	)
2279	.map_err(\|e\| ApiError::internal(format!("Failed to load MCP config: {e}")))?;
2280	let handle = Arc::new(Mutex::new(new_pool));
2281	pool_slot.replace(Arc::clone(&handle));
2282	Some(handle)
2283	}
2284	None => None,
2285	}
2286	};
2287
2288	let Some(pool_handle) = pool_handle else {
2289	return Ok(Json(McpToolsResponse { tools: Vec::new() }));
2290	};
2291
2292	let mut pool = pool_handle.lock().await;
2293	if query.connect {
2294	let _errors = pool.connect_all().await;
2295	}
2296
2297	let mut tools = Vec::new();
2298	for (prefixed_name, tool) in pool.all_tools() {
2299	let Ok((server, name)) = pool.parse_prefixed_name(&prefixed_name) else {
2300	continue;
2301	};
2302
2303	if let Some(filter) = query.server.as_deref()
2304	&& server != filter
2305	{
2306	continue;
2307	}
2308
2309	tools.push(McpToolEntry {
2310	server: server.to_string(),
2311	name: name.to_string(),
2312	prefixed_name,
2313	description: tool.description.clone(),
2314	input_schema: tool.input_schema.clone(),
2315	});
2316	}
2317
2318	tools.sort_by(\|a, b\| a.server.cmp(&b.server).then_with(\|\| a.name.cmp(&b.name)));
2319
2320	Ok(Json(McpToolsResponse { tools }))
2321	}
2322
2323	async fn list_automations(
2324	State(state): State<RuntimeApiState>,
2325	) -> Result<Json<Vec<AutomationRecord>>, ApiError> {
2326	let manager = state.automations.lock().await;
2327	let automations = manager
2328	.list_automations()
2329	.map_err(\|e\| ApiError::internal(format!("Failed to list automations: {e}")))?;
2330	Ok(Json(automations))
2331	}
2332
2333	async fn create_automation(
2334	State(state): State<RuntimeApiState>,
2335	Json(req): Json<CreateAutomationRequest>,
2336	) -> Result<(StatusCode, Json<AutomationRecord>), ApiError> {
2337	let manager = state.automations.lock().await;
2338	let automation = manager
2339	.create_automation(req)
2340	.map_err(\|e\| ApiError::bad_request(e.to_string()))?;
2341	Ok((StatusCode::CREATED, Json(automation)))
2342	}
2343
2344	async fn get_automation(
2345	State(state): State<RuntimeApiState>,
2346	Path(id): Path<String>,
2347	) -> Result<Json<AutomationRecord>, ApiError> {
2348	let manager = state.automations.lock().await;
2349	let automation = manager.get_automation(&id).map_err(map_automation_err)?;
2350	Ok(Json(automation))
2351	}
2352
2353	async fn update_automation(
2354	State(state): State<RuntimeApiState>,
2355	Path(id): Path<String>,
2356	Json(req): Json<UpdateAutomationRequest>,
2357	) -> Result<Json<AutomationRecord>, ApiError> {
2358	let manager = state.automations.lock().await;
2359	let automation = manager
2360	.update_automation(&id, req)
2361	.map_err(map_automation_err)?;
2362	Ok(Json(automation))
2363	}
2364
2365	async fn delete_automation(
2366	State(state): State<RuntimeApiState>,
2367	Path(id): Path<String>,
2368	) -> Result<Json<AutomationRecord>, ApiError> {
2369	let manager = state.automations.lock().await;
2370	let automation = manager.delete_automation(&id).map_err(map_automation_err)?;
2371	Ok(Json(automation))
2372	}
2373
2374	async fn run_automation(
2375	State(state): State<RuntimeApiState>,
2376	Path(id): Path<String>,
2377	) -> Result<Json<AutomationRunRecord>, ApiError> {
2378	// run_now_shared drops the manager mutex across the task-manager await so
2379	// other automation endpoints stay responsive behind a slow enqueue.
2380	let run =
2381	crate::automation_manager::run_now_shared(&state.automations, &id, &state.task_manager)
2382	.await
2383	.map_err(map_automation_err)?;
2384	Ok(Json(run))
2385	}
2386
2387	async fn pause_automation(
2388	State(state): State<RuntimeApiState>,
2389	Path(id): Path<String>,
2390	) -> Result<Json<AutomationRecord>, ApiError> {
2391	let manager = state.automations.lock().await;
2392	let automation = manager.pause_automation(&id).map_err(map_automation_err)?;
2393	Ok(Json(automation))
2394	}
2395
2396	async fn resume_automation(
2397	State(state): State<RuntimeApiState>,
2398	Path(id): Path<String>,
2399	) -> Result<Json<AutomationRecord>, ApiError> {
2400	let manager = state.automations.lock().await;
2401	let automation = manager.resume_automation(&id).map_err(map_automation_err)?;
2402	Ok(Json(automation))
2403	}
2404
2405	async fn list_automation_runs(
2406	State(state): State<RuntimeApiState>,
2407	Path(id): Path<String>,
2408	Query(query): Query<AutomationRunsQuery>,
2409	) -> Result<Json<Vec<AutomationRunRecord>>, ApiError> {
2410	let manager = state.automations.lock().await;
2411	let runs = manager
2412	.list_runs(&id, query.limit)
2413	.map_err(map_automation_err)?;
2414	Ok(Json(runs))
2415	}
2416
2417	async fn get_thread(
2418	State(state): State<RuntimeApiState>,
2419	Path(id): Path<String>,
2420	) -> Result<Json<ThreadDetail>, ApiError> {
2421	let detail = state
2422	.runtime_threads
2423	.get_thread_detail(&id)
2424	.await
2425	.map_err(map_thread_err)?;
2426	Ok(Json(detail))
2427	}
2428
2429	async fn update_thread(
2430	State(state): State<RuntimeApiState>,
2431	Path(id): Path<String>,
2432	Json(req): Json<UpdateThreadRequest>,
2433	) -> Result<Json<ThreadRecord>, ApiError> {
2434	let thread = state
2435	.runtime_threads
2436	.update_thread(&id, req)
2437	.await
2438	.map_err(map_thread_err)?;
2439	Ok(Json(thread))
2440	}
2441
2442	async fn resume_thread(
2443	State(state): State<RuntimeApiState>,
2444	Path(id): Path<String>,
2445	) -> Result<Json<ThreadRecord>, ApiError> {
2446	let thread = state
2447	.runtime_threads
2448	.resume_thread(&id)
2449	.await
2450	.map_err(map_thread_err)?;
2451	Ok(Json(thread))
2452	}
2453
2454	async fn fork_thread(
2455	State(state): State<RuntimeApiState>,
2456	Path(id): Path<String>,
2457	) -> Result<(StatusCode, Json<ThreadRecord>), ApiError> {
2458	let thread = state
2459	.runtime_threads
2460	.fork_thread(&id)
2461	.await
2462	.map_err(map_thread_err)?;
2463	Ok((StatusCode::CREATED, Json(thread)))
2464	}
2465
2466	#[derive(Debug, Deserialize)]
2467	struct UndoTurnRequest {
2468	/// How many turns back to undo (default 0 = last turn only).
2469	#[serde(default)]
2470	depth: Option<usize>,
2471	}
2472
2473	#[derive(Debug, Serialize)]
2474	struct UndoTurnResponse {
2475	/// The new forked thread (with the last N turns removed).
2476	thread: ThreadRecord,
2477	/// The original user message text from the first dropped turn,
2478	/// so the GUI can pre-populate the input box.
2479	original_user_text: Option<String>,
2480	}
2481
2482	async fn undo_thread_turn(
2483	State(state): State<RuntimeApiState>,
2484	Path(id): Path<String>,
2485	Json(req): Json<UndoTurnRequest>,
2486	) -> Result<(StatusCode, Json<UndoTurnResponse>), ApiError> {
2487	let depth = req.depth.unwrap_or(0);
2488	let (forked_thread, original_user_text) = state
2489	.runtime_threads
2490	.fork_at_user_message(&id, depth)
2491	.await
2492	.map_err(map_thread_err)?;
2493	Ok((
2494	StatusCode::CREATED,
2495	Json(UndoTurnResponse {
2496	thread: forked_thread,
2497	original_user_text,
2498	}),
2499	))
2500	}
2501
2502	/// Result of the snapshot-based file rollback step of patch-undo, reported
2503	/// alongside the new forked thread.
2504	#[derive(Debug, Serialize)]
2505	struct PatchUndoResult {
2506	/// Whether files were restored from a snapshot.
2507	files_restored: bool,
2508	/// Human-readable summary of what was restored (diff stat).
2509	summary: Option<String>,
2510	/// The label of the restored snapshot (e.g. "tool:apply_patch" or "pre-turn:3").
2511	snapshot_label: Option<String>,
2512	}
2513
2514	#[derive(Debug, Serialize)]
2515	struct PatchUndoResponse {
2516	/// Result of the snapshot-based file rollback step.
2517	patch_result: PatchUndoResult,
2518	/// The new forked thread (with the last turn removed).
2519	thread: ThreadRecord,
2520	/// The original user text from the removed turn (for re-editing).
2521	original_user_text: Option<String>,
2522	}
2523
2524	async fn patch_undo_thread_turn(
2525	State(state): State<RuntimeApiState>,
2526	Path(id): Path<String>,
2527	Json(req): Json<UndoTurnRequest>,
2528	) -> Result<(StatusCode, Json<PatchUndoResponse>), ApiError> {
2529	let depth = req.depth.unwrap_or(0);
2530
2531	// Step 1: Try snapshot-based file rollback (patch_undo).
2532	let thread = state
2533	.runtime_threads
2534	.get_thread(&id)
2535	.await
2536	.map_err(map_thread_err)?;
2537	let patch_result = patch_undo_workspace_files(&thread.workspace, thread.session_id.as_deref());
2538
2539	// Step 2: Remove the last conversation turn (undo_conversation).
2540	let (forked_thread, original_user_text) = state
2541	.runtime_threads
2542	.fork_at_user_message(&id, depth)
2543	.await
2544	.map_err(map_thread_err)?;
2545
2546	Ok((
2547	StatusCode::CREATED,
2548	Json(PatchUndoResponse {
2549	patch_result,
2550	thread: forked_thread,
2551	original_user_text,
2552	}),
2553	))
2554	}
2555
2556	/// Restore the newest `tool:` or `pre-turn:` snapshot that differs from the
2557	/// current workspace — same target selection as the TUI's `patch_undo`.
2558	fn patch_undo_workspace_files(
2559	workspace: &FsPath,
2560	current_session_id: Option<&str>,
2561	) -> PatchUndoResult {
2562	let repo = match crate::snapshot::SnapshotRepo::open_or_init(workspace) {
2563	Ok(repo) => repo,
2564	Err(e) => {
2565	return PatchUndoResult {
2566	files_restored: false,
2567	summary: Some(format!("Snapshot repo unavailable: {e}")),
2568	snapshot_label: None,
2569	};
2570	}
2571	};
2572	let Some(current_session_id) = current_session_id else {
2573	return PatchUndoResult {
2574	files_restored: false,
2575	summary: Some(
2576	"No current session is bound to this thread; workspace files were not changed."
2577	.to_string(),
2578	),
2579	snapshot_label: None,
2580	};
2581	};
2582	let snapshots = match repo.list(100) {
2583	Ok(snapshots) => snapshots,
2584	Err(e) => {
2585	return PatchUndoResult {
2586	files_restored: false,
2587	summary: Some(format!("Failed to list snapshots: {e}")),
2588	snapshot_label: None,
2589	};
2590	}
2591	};
2592	let target = snapshots
2593	.iter()
2594	.filter(\|s\| s.label.starts_with("tool:") \|\| s.label.starts_with("pre-turn:"))
2595	.filter(\|s\| s.session_id.as_deref() == Some(current_session_id))
2596	.find(\|s\| matches!(repo.work_tree_matches_snapshot(&s.id), Ok(false)));
2597	let Some(target) = target else {
2598	return PatchUndoResult {
2599	files_restored: false,
2600	summary: Some(
2601	"No current-session tool or pre-turn snapshots differ from the current workspace."
2602	.to_string(),
2603	),
2604	snapshot_label: None,
2605	};
2606	};
2607	if let Err(e) = repo.restore(&target.id) {
2608	return PatchUndoResult {
2609	files_restored: false,
2610	summary: Some(format!("Restore failed: {e}")),
2611	snapshot_label: None,
2612	};
2613	}
2614
2615	// Compute a diff stat for the summary.
2616	use crate::dependencies::{ExternalTool as _, Git};
2617	let diff_stat = Git::command().and_then(\|mut git\| {
2618	git.args(["diff", "--stat"])
2619	.current_dir(workspace)
2620	.output()
2621	.ok()
2622	.and_then(\|o\| {
2623	let s = String::from_utf8_lossy(&o.stdout).trim().to_string();
2624	if s.is_empty() { None } else { Some(s) }
2625	})
2626	});
2627
2628	let short = &target.id.as_str()[..target.id.as_str().len().min(8)];
2629	let summary = match diff_stat {
2630	Some(ref stat) => format!(
2631	"Restored snapshot '{}' ({}). Files affected:\n{stat}",
2632	target.label, short
2633	),
2634	None => format!(
2635	"Restored snapshot '{}' ({}). No diff changes detected.",
2636	target.label, short
2637	),
2638	};
2639	PatchUndoResult {
2640	files_restored: true,
2641	summary: Some(summary),
2642	snapshot_label: Some(target.label.clone()),
2643	}
2644	}
2645
2646	#[derive(Debug, Deserialize)]
2647	struct RetryTurnRequest {
2648	/// How many turns back to retry (default 0 = last turn only).
2649	#[serde(default)]
2650	depth: Option<usize>,
2651	/// Override the user message text. If omitted, the original text
2652	/// from the dropped turn is re-used.
2653	#[serde(default)]
2654	prompt: Option<String>,
2655	}
2656
2657	#[derive(Debug, Serialize)]
2658	struct RetryTurnResponse {
2659	/// The new forked thread (with the last N turns removed).
2660	thread: ThreadRecord,
2661	/// The turn created by the retry.
2662	turn: TurnRecord,
2663	}
2664
2665	async fn retry_thread_turn(
2666	State(state): State<RuntimeApiState>,
2667	Path(id): Path<String>,
2668	Json(req): Json<RetryTurnRequest>,
2669	) -> Result<(StatusCode, Json<RetryTurnResponse>), ApiError> {
2670	let depth = req.depth.unwrap_or(0);
2671	let (forked_thread, original_user_text) = state
2672	.runtime_threads
2673	.fork_at_user_message(&id, depth)
2674	.await
2675	.map_err(map_thread_err)?;
2676
2677	let retry_prompt = req.prompt.or(original_user_text).unwrap_or_default();
2678	if retry_prompt.trim().is_empty() {
2679	return Err(ApiError::bad_request(
2680	"No user message to retry — the dropped turn had no user text",
2681	));
2682	}
2683
2684	let turn = state
2685	.runtime_threads
2686	.start_turn(
2687	&forked_thread.id,
2688	StartTurnRequest {
2689	prompt: retry_prompt,
2690	input_summary: None,
2691	model: None,
2692	mode: None,
2693	permission_posture: None,
2694	allow_shell: None,
2695	trust_mode: None,
2696	auto_approve: None,
2697	dynamic_tools: Vec::new(),
2698	environment_id: None,
2699	},
2700	)
2701	.await
2702	.map_err(map_thread_err)?;
2703
2704	Ok((
2705	StatusCode::CREATED,
2706	Json(RetryTurnResponse {
2707	thread: forked_thread,
2708	turn,
2709	}),
2710	))
2711	}
2712
2713	async fn start_thread_turn(
2714	State(state): State<RuntimeApiState>,
2715	Path(id): Path<String>,
2716	Json(req): Json<StartTurnRequest>,
2717	) -> Result<(StatusCode, Json<StartTurnResponse>), ApiError> {
2718	let turn = state
2719	.runtime_threads
2720	.start_turn(&id, req)
2721	.await
2722	.map_err(map_thread_err)?;
2723	let thread = state
2724	.runtime_threads
2725	.get_thread(&id)
2726	.await
2727	.map_err(map_thread_err)?;
2728	Ok((
2729	StatusCode::CREATED,
2730	Json(StartTurnResponse { thread, turn }),
2731	))
2732	}
2733
2734	async fn steer_thread_turn(
2735	State(state): State<RuntimeApiState>,
2736	Path((id, turn_id)): Path<(String, String)>,
2737	Json(req): Json<SteerTurnRequest>,
2738	) -> Result<Json<TurnRecord>, ApiError> {
2739	let turn = state
2740	.runtime_threads
2741	.steer_turn(&id, &turn_id, req)
2742	.await
2743	.map_err(map_thread_err)?;
2744	Ok(Json(turn))
2745	}
2746
2747	async fn interrupt_thread_turn(
2748	State(state): State<RuntimeApiState>,
2749	Path((id, turn_id)): Path<(String, String)>,
2750	) -> Result<Json<TurnRecord>, ApiError> {
2751	let turn = state
2752	.runtime_threads
2753	.interrupt_turn(&id, &turn_id)
2754	.await
2755	.map_err(map_thread_err)?;
2756	Ok(Json(turn))
2757	}
2758
2759	async fn deliver_dynamic_tool_result(
2760	State(state): State<RuntimeApiState>,
2761	Path((id, turn_id, call_id)): Path<(String, String, String)>,
2762	Json(result): Json<DynamicToolCallResult>,
2763	) -> Result<StatusCode, ApiError> {
2764	state
2765	.runtime_threads
2766	.get_thread(&id)
2767	.await
2768	.map_err(map_thread_err)?;
2769	if state
2770	.runtime_threads
2771	.deliver_dynamic_tool_result(&id, &turn_id, &call_id, result)
2772	.await
2773	.map_err(\|error\| ApiError::internal(error.to_string()))?
2774	{
2775	Ok(StatusCode::ACCEPTED)
2776	} else {
2777	Err(ApiError::not_found(format!(
2778	"No pending dynamic tool call '{call_id}'"
2779	)))
2780	}
2781	}
2782
2783	async fn compact_thread(
2784	State(state): State<RuntimeApiState>,
2785	Path(id): Path<String>,
2786	Json(req): Json<CompactThreadRequest>,
2787	) -> Result<(StatusCode, Json<StartTurnResponse>), ApiError> {
2788	let turn = state
2789	.runtime_threads
2790	.compact_thread(&id, req)
2791	.await
2792	.map_err(map_thread_err)?;
2793	let thread = state
2794	.runtime_threads
2795	.get_thread(&id)
2796	.await
2797	.map_err(map_thread_err)?;
2798	Ok((
2799	StatusCode::ACCEPTED,
2800	Json(StartTurnResponse { thread, turn }),
2801	))
2802	}
2803
2804	async fn list_tasks(
2805	State(state): State<RuntimeApiState>,
2806	Query(query): Query<TasksQuery>,
2807	) -> Result<Json<TasksResponse>, ApiError> {
2808	let tasks = state
2809	.task_manager
2810	.list_tasks_scoped(query.limit, query.workspace.as_deref())
2811	.await;
2812	let counts = state.task_manager.counts().await;
2813	Ok(Json(TasksResponse { tasks, counts }))
2814	}
2815
2816	async fn get_task(
2817	State(state): State<RuntimeApiState>,
2818	Path(id): Path<String>,
2819	) -> Result<Json<TaskRecord>, ApiError> {
2820	let task = state
2821	.task_manager
2822	.get_task(&id)
2823	.await
2824	.map_err(map_task_err)?;
2825	Ok(Json(task))
2826	}
2827
2828	async fn cancel_task(
2829	State(state): State<RuntimeApiState>,
2830	Path(id): Path<String>,
2831	) -> Result<Json<TaskRecord>, ApiError> {
2832	let cancellation = state
2833	.task_manager
2834	.cancel_task(&id)
2835	.await
2836	.map_err(map_task_err)?;
2837	Ok(Json(cancellation.task))
2838	}
2839
2840	async fn stream_thread_events(
2841	State(state): State<RuntimeApiState>,
2842	Path(id): Path<String>,
2843	Query(query): Query<ThreadEventsQuery>,
2844	) -> Result<Sse<impl futures_util::Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
2845	let _ = state
2846	.runtime_threads
2847	.get_thread(&id)
2848	.await
2849	.map_err(map_thread_err)?;
2850
2851	// Subscribe before reading durable history. An event emitted while replay
2852	// is loaded is then present in both places (and deduped below) or queued
2853	// live, never in an uncovered handoff window.
2854	let live = state.runtime_threads.subscribe_events();
2855	if query
2856	.replay_limit
2857	.is_some_and(\|limit\| limit > MAX_RUNTIME_EVENT_REPLAY_TAIL)
2858	{
2859	return Err(ApiError::bad_request(format!(
2860	"replay_limit cannot exceed {MAX_RUNTIME_EVENT_REPLAY_TAIL}"
2861	)));
2862	}
2863	let replay = state
2864	.runtime_threads
2865	.replay_events(&id, query.since_seq, query.replay_limit)
2866	.await
2867	.map_err(\|e\| ApiError::internal(e.to_string()))?;
2868
2869	let stream = replay_live_thread_events(
2870	state.runtime_threads.clone(),
2871	id,
2872	replay.base_seq,
2873	replay.batches,
2874	live,
2875	);
2876
2877	Ok(Sse::new(stream).keep_alive(
2878	KeepAlive::new()
2879	.interval(Duration::from_secs(15))
2880	.text("keepalive"),
2881	))
2882	}
2883
2884	fn replay_live_thread_events(
2885	runtime_threads: SharedRuntimeThreadManager,
2886	thread_id: String,
2887	mut last_seq: u64,
2888	mut backlog: tokio::sync::mpsc::Receiver<
2889	std::result::Result<Vec<crate::runtime_threads::RuntimeEventRecord>, String>,
2890	>,
2891	mut live: tokio::sync::broadcast::Receiver<crate::runtime_threads::RuntimeEventRecord>,
2892	) -> impl futures_util::Stream<Item = Result<SseEvent, Infallible>> {
2893	stream! {
2894	while let Some(batch) = backlog.recv().await {
2895	let events = match batch {
2896	Ok(events) => events,
2897	Err(error) => {
2898	tracing::warn!(
2899	thread_id = %thread_id,
2900	last_seq,
2901	%error,
2902	"Failed to replay Runtime web event stream from durable history"
2903	);
2904	return;
2905	}
2906	};
2907	for event in events {
2908	if event.thread_id != thread_id \|\| event.seq <= last_seq {
2909	continue;
2910	}
2911	let previous_seq = last_seq;
2912	last_seq = event.seq;
2913	let event_name = event.event.clone();
2914	yield Ok(sse_json(
2915	&event_name,
2916	runtime_event_payload_with_previous(event, previous_seq),
2917	));
2918	}
2919	}
2920
2921	'live: loop {
2922	match live.recv().await {
2923	Ok(event) => {
2924	if event.thread_id != thread_id \|\| event.seq <= last_seq {
2925	continue;
2926	}
2927	let previous_seq = last_seq;
2928	last_seq = event.seq;
2929	let event_name = event.event.clone();
2930	yield Ok(sse_json(
2931	&event_name,
2932	runtime_event_payload_with_previous(event, previous_seq),
2933	));
2934	}
2935	Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
2936	// Broadcast is only a wake-up path; durable history remains
2937	// authoritative. Catch up from the last delivered cursor so
2938	// receiver pressure cannot turn into a silent prompt loss.
2939	let mut recovered = match runtime_threads
2940	.replay_events(&thread_id, Some(last_seq), None)
2941	.await
2942	{
2943	Ok(replay) => replay.batches,
2944	Err(error) => {
2945	tracing::warn!(
2946	thread_id = %thread_id,
2947	last_seq,
2948	skipped,
2949	%error,
2950	"Failed to recover lagged Runtime web event stream from durable history"
2951	);
2952	break 'live;
2953	}
2954	};
2955	while let Some(batch) = recovered.recv().await {
2956	let events = match batch {
2957	Ok(events) => events,
2958	Err(error) => {
2959	tracing::warn!(
2960	thread_id = %thread_id,
2961	last_seq,
2962	skipped,
2963	%error,
2964	"Failed to recover lagged Runtime web event stream from durable history"
2965	);
2966	break 'live;
2967	}
2968	};
2969	for event in events {
2970	if event.thread_id != thread_id \|\| event.seq <= last_seq {
2971	continue;
2972	}
2973	let previous_seq = last_seq;
2974	last_seq = event.seq;
2975	let event_name = event.event.clone();
2976	yield Ok(sse_json(
2977	&event_name,
2978	runtime_event_payload_with_previous(event, previous_seq),
2979	));
2980	}
2981	}
2982	}
2983	Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
2984	}
2985	}
2986	}
2987	}
2988
2989	async fn stream_turn(
2990	State(state): State<RuntimeApiState>,
2991	Json(req): Json<StreamTurnRequest>,
2992	) -> Result<Sse<impl futures_util::Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
2993	if req.prompt.trim().is_empty() {
2994	return Err(ApiError::bad_request("prompt is required"));
2995	}
2996
2997	let model = req.model.clone().unwrap_or_else(\|\| {
2998	state
2999	.config
3000	.read()
3001	.default_text_model
3002	.clone()
3003	.unwrap_or_else(\|\| DEFAULT_TEXT_MODEL.to_string())
3004	});
3005	let workspace = req
3006	.workspace
3007	.clone()
3008	.unwrap_or_else(\|\| state.workspace.clone());
3009	let mode = req.mode.clone().unwrap_or_else(\|\| "agent".to_string());
3010	let permission_posture = req.permission_posture.clone();
3011	let allow_shell = req.allow_shell.unwrap_or(state.config.read().allow_shell());
3012	let trust_mode = req.trust_mode.unwrap_or(false);
3013	let auto_approve = req.auto_approve.unwrap_or(false);
3014	let prompt = req.prompt;
3015
3016	let thread = state
3017	.runtime_threads
3018	.create_thread(CreateThreadRequest {
3019	model: Some(model.clone()),
3020	workspace: Some(workspace.clone()),
3021	mode: Some(mode.clone()),
3022	permission_posture: permission_posture.clone(),
3023	allow_shell: Some(allow_shell),
3024	trust_mode: Some(trust_mode),
3025	auto_approve: Some(auto_approve),
3026	archived: true,
3027	system_prompt: None,
3028	task_id: None,
3029	..Default::default()
3030	})
3031	.await
3032	.map_err(\|e\| ApiError::internal(format!("Failed to create stream thread: {e}")))?;
3033
3034	#[cfg(test)]
3035	if let Some(hook) = &state.compat_stream_test_hook {
3036	let (resume, wait_for_resume) = tokio::sync::oneshot::channel();
3037	hook.send(CompatStreamTestPoint::ThreadCreated {
3038	thread_id: thread.id.clone(),
3039	resume,
3040	})
3041	.map_err(\|_\| ApiError::internal("Compatibility stream test hook closed"))?;
3042	wait_for_resume
3043	.await
3044	.map_err(\|_\| ApiError::internal("Compatibility stream test hook dropped resume"))?;
3045	}
3046
3047	let turn = state
3048	.runtime_threads
3049	.start_turn(
3050	&thread.id,
3051	StartTurnRequest {
3052	prompt,
3053	input_summary: None,
3054	model: Some(model.clone()),
3055	mode: Some(mode.clone()),
3056	permission_posture,
3057	allow_shell: Some(allow_shell),
3058	trust_mode: Some(trust_mode),
3059	auto_approve: Some(auto_approve),
3060	..Default::default()
3061	},
3062	)
3063	.await
3064	.map_err(\|e\| ApiError::internal(format!("Failed to start stream turn: {e}")))?;
3065
3066	// Subscribe before reading the durable replay. Events produced while the
3067	// replay is loaded then exist in at least one source, and the sequence
3068	// cursor below removes overlap without dropping the handoff edge.
3069	let mut live = state.runtime_threads.subscribe_events();
3070	let thread_id = thread.id.clone();
3071	let turn_id = turn.id.clone();
3072
3073	#[cfg(test)]
3074	if let Some(hook) = &state.compat_stream_test_hook {
3075	let (resume, wait_for_resume) = tokio::sync::oneshot::channel();
3076	hook.send(CompatStreamTestPoint::SubscribedBeforeReplay {
3077	thread_id: thread_id.clone(),
3078	turn_id: turn_id.clone(),
3079	resume,
3080	})
3081	.map_err(\|_\| ApiError::internal("Compatibility stream test hook closed"))?;
3082	wait_for_resume
3083	.await
3084	.map_err(\|_\| ApiError::internal("Compatibility stream test hook dropped resume"))?;
3085	}
3086
3087	let mut backlog = state
3088	.runtime_threads
3089	.replay_events(&thread.id, None, None)
3090	.await
3091	.map_err(\|e\| ApiError::internal(format!("Failed to load stream backlog: {e}")))?;
3092
3093	#[cfg(test)]
3094	if let Some(hook) = &state.compat_stream_test_hook {
3095	let (resume, wait_for_resume) = tokio::sync::oneshot::channel();
3096	hook.send(CompatStreamTestPoint::ReplayLoaded {
3097	thread_id: thread_id.clone(),
3098	turn_id: turn_id.clone(),
3099	resume,
3100	})
3101	.map_err(\|_\| ApiError::internal("Compatibility stream test hook closed"))?;
3102	wait_for_resume
3103	.await
3104	.map_err(\|_\| ApiError::internal("Compatibility stream test hook dropped resume"))?;
3105	}
3106
3107	let stream = stream! {
3108	let mut last_seq = 0;
3109	yield Ok(sse_json("turn.started", json!({
3110	"thread_id": thread.id,
3111	"turn_id": turn.id,
3112	"model": model,
3113	"mode": mode,
3114	"workspace": workspace,
3115	})));
3116
3117	while let Some(batch) = backlog.batches.recv().await {
3118	let events = match batch {
3119	Ok(events) => events,
3120	Err(error) => {
3121	tracing::warn!(
3122	thread_id = %thread_id,
3123	turn_id = %turn_id,
3124	%error,
3125	"Failed to replay compatibility stream from durable history"
3126	);
3127	yield Ok(sse_json("error", json!({
3128	"message": "failed to replay durable event stream",
3129	})));
3130	return;
3131	}
3132	};
3133	for event in events {
3134	let Some((mapped, terminal)) = take_compat_turn_event(
3135	&event,
3136	&thread_id,
3137	&turn_id,
3138	&mut last_seq,
3139	) else {
3140	continue;
3141	};
3142	if let Some(mapped) = mapped {
3143	yield Ok(mapped);
3144	}
3145	if terminal {
3146	yield Ok(sse_json("done", json!({})));
3147	return;
3148	}
3149	}
3150	}
3151
3152	loop {
3153	match live.recv().await {
3154	Ok(event) => {
3155	let Some((mapped, terminal)) = take_compat_turn_event(
3156	&event,
3157	&thread_id,
3158	&turn_id,
3159	&mut last_seq,
3160	) else {
3161	continue;
3162	};
3163	if let Some(mapped) = mapped {
3164	yield Ok(mapped);
3165	}
3166	if terminal {
3167	yield Ok(sse_json("done", json!({})));
3168	return;
3169	}
3170	}
3171	Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
3172	let mut recovered = match state.runtime_threads
3173	.replay_events(&thread_id, Some(last_seq), None)
3174	.await
3175	{
3176	Ok(replay) => replay.batches,
3177	Err(error) => {
3178	tracing::warn!(
3179	thread_id = %thread_id,
3180	turn_id = %turn_id,
3181	last_seq,
3182	skipped,
3183	%error,
3184	"Failed to recover lagged compatibility stream from durable history"
3185	);
3186	yield Ok(sse_json("error", json!({
3187	"message": "failed to recover lagged event stream",
3188	})));
3189	return;
3190	}
3191	};
3192	while let Some(batch) = recovered.recv().await {
3193	let events = match batch {
3194	Ok(events) => events,
3195	Err(error) => {
3196	tracing::warn!(
3197	thread_id = %thread_id,
3198	turn_id = %turn_id,
3199	last_seq,
3200	skipped,
3201	%error,
3202	"Failed to recover lagged compatibility stream from durable history"
3203	);
3204	yield Ok(sse_json("error", json!({
3205	"message": "failed to recover lagged event stream",
3206	})));
3207	return;
3208	}
3209	};
3210	for event in events {
3211	let Some((mapped, terminal)) = take_compat_turn_event(
3212	&event,
3213	&thread_id,
3214	&turn_id,
3215	&mut last_seq,
3216	) else {
3217	continue;
3218	};
3219	if let Some(mapped) = mapped {
3220	yield Ok(mapped);
3221	}
3222	if terminal {
3223	yield Ok(sse_json("done", json!({})));
3224	return;
3225	}
3226	}
3227	}
3228	}
3229	Err(tokio::sync::broadcast::error::RecvError::Closed) => {
3230	yield Ok(sse_json("error", json!({ "message": "event channel closed" })));
3231	return;
3232	}
3233	}
3234	}
3235	};
3236
3237	Ok(Sse::new(stream).keep_alive(
3238	KeepAlive::new()
3239	.interval(Duration::from_secs(15))
3240	.text("keepalive"),
3241	))
3242	}
3243
3244	fn take_compat_turn_event(
3245	event: &crate::runtime_threads::RuntimeEventRecord,
3246	thread_id: &str,
3247	turn_id: &str,
3248	last_seq: &mut u64,
3249	) -> Option<(Option<SseEvent>, bool)> {
3250	if event.thread_id != thread_id
3251	\|\| event.turn_id.as_deref() != Some(turn_id)
3252	\|\| event.seq <= *last_seq
3253	{
3254	return None;
3255	}
3256	*last_seq = event.seq;
3257	Some((
3258	map_compat_stream_event(event),
3259	event.event == "turn.completed",
3260	))
3261	}
3262
3263	fn runtime_event_payload(event: crate::runtime_threads::RuntimeEventRecord) -> serde_json::Value {
3264	let event_name = event.event.clone();
3265	let timestamp = event.timestamp.to_rfc3339();
3266	let schema_version = RUNTIME_EVENT_ENVELOPE_SCHEMA_VERSION;
3267	let envelope = RuntimeEventEnvelope {
3268	schema_version,
3269	seq: event.seq,
3270	event: event_name.clone(),
3271	kind: event_name,
3272	thread_id: event.thread_id,
3273	turn_id: event.turn_id,
3274	item_id: event.item_id,
3275	timestamp: timestamp.clone(),
3276	created_at: Some(timestamp),
3277	payload: event.payload,
3278	extra: Default::default(),
3279	};
3280	serde_json::to_value(envelope).expect("serialize runtime event envelope")
3281	}
3282
3283	fn runtime_event_payload_with_previous(
3284	event: crate::runtime_threads::RuntimeEventRecord,
3285	previous_seq: u64,
3286	) -> serde_json::Value {
3287	let mut payload = runtime_event_payload(event);
3288	if let Some(object) = payload.as_object_mut() {
3289	object.insert("previous_seq".to_string(), json!(previous_seq));
3290	}
3291	payload
3292	}
3293
3294	fn map_compat_stream_event(event: &crate::runtime_threads::RuntimeEventRecord) -> Option<SseEvent> {
3295	let payload = &event.payload;
3296	match event.event.as_str() {
3297	"item.delta" => {
3298	let kind = payload
3299	.get("kind")
3300	.and_then(\|v\| v.as_str())
3301	.unwrap_or_default();
3302	if kind == "agent_message" {
3303	let content = payload
3304	.get("delta")
3305	.and_then(\|v\| v.as_str())
3306	.unwrap_or_default();
3307	Some(sse_json("message.delta", json!({ "content": content })))
3308	} else if kind == "tool_call" {
3309	let output = payload
3310	.get("delta")
3311	.and_then(\|v\| v.as_str())
3312	.unwrap_or_default();
3313	Some(sse_json("tool.progress", json!({ "output": output })))
3314	} else {
3315	None
3316	}
3317	}
3318	"item.started" => {
3319	let tool = payload.get("tool")?;
3320	let id = tool.get("id").cloned().unwrap_or(Value::Null);
3321	let name = tool.get("name").cloned().unwrap_or(Value::Null);
3322	let input = tool.get("input").cloned().unwrap_or(Value::Null);
3323	Some(sse_json(
3324	"tool.started",
3325	json!({
3326	"id": id,
3327	"name": name,
3328	"input": input,
3329	}),
3330	))
3331	}
3332	"item.completed" \| "item.failed" => {
3333	let item = payload.get("item")?;
3334	let kind = item
3335	.get("kind")
3336	.and_then(\|v\| v.as_str())
3337	.unwrap_or_default();
3338	if kind == "tool_call" \|\| kind == "file_change" \|\| kind == "command_execution" {
3339	let id = item.get("id").cloned().unwrap_or(Value::Null);
3340	let success = event.event == "item.completed";
3341	let output = item.get("detail").cloned().unwrap_or_else(\|\| {
3342	Value::String(
3343	item.get("summary")
3344	.and_then(\|v\| v.as_str())
3345	.unwrap_or_default()
3346	.to_string(),
3347	)
3348	});
3349	Some(sse_json(
3350	"tool.completed",
3351	json!({
3352	"id": id,
3353	"success": success,
3354	"output": output,
3355	}),
3356	))
3357	} else if kind == "status" {
3358	let message = item
3359	.get("detail")
3360	.and_then(\|v\| v.as_str())
3361	.or_else(\|\| item.get("summary").and_then(\|v\| v.as_str()))
3362	.unwrap_or_default();
3363	Some(sse_json("status", json!({ "message": message })))
3364	} else if kind == "error" {
3365	let message = item
3366	.get("detail")
3367	.and_then(\|v\| v.as_str())
3368	.or_else(\|\| item.get("summary").and_then(\|v\| v.as_str()))
3369	.unwrap_or_default();
3370	Some(sse_json("error", json!({ "message": message })))
3371	} else {
3372	None
3373	}
3374	}
3375	"approval.required" => {
3376	let approval_id = payload
3377	.get("approval_id")
3378	.or_else(\|\| payload.get("id"))?
3379	.clone();
3380	Some(sse_json(
3381	"approval.required",
3382	json!({
3383	"id": approval_id,
3384	"approval_id": approval_id,
3385	"thread_id": event.thread_id,
3386	"turn_id": event.turn_id,
3387	"tool_name": payload.get("tool_name"),
3388	"description": payload.get("description"),
3389	"intent_summary": payload.get("intent_summary"),
3390	}),
3391	))
3392	}
3393	"approval.decided" => {
3394	let approval_id = payload
3395	.get("approval_id")
3396	.or_else(\|\| payload.get("id"))?
3397	.clone();
3398	Some(sse_json(
3399	"approval.decided",
3400	json!({
3401	"id": approval_id,
3402	"approval_id": approval_id,
3403	"thread_id": event.thread_id,
3404	"turn_id": event.turn_id,
3405	"decision": payload.get("decision"),
3406	"remember": payload.get("remember"),
3407	"auto": payload.get("auto"),
3408	"timeout": payload.get("timeout"),
3409	}),
3410	))
3411	}
3412	"approval.timeout" => {
3413	let approval_id = payload
3414	.get("approval_id")
3415	.or_else(\|\| payload.get("id"))?
3416	.clone();
3417	Some(sse_json(
3418	"approval.timeout",
3419	json!({
3420	"id": approval_id,
3421	"approval_id": approval_id,
3422	"thread_id": event.thread_id,
3423	"turn_id": event.turn_id,
3424	"timeout_secs": payload.get("timeout_secs"),
3425	}),
3426	))
3427	}
3428	"user_input.required" => {
3429	let input_id = payload
3430	.get("input_id")
3431	.or_else(\|\| payload.get("id"))?
3432	.clone();
3433	let request = payload.get("request")?.clone();
3434	Some(sse_json(
3435	"user_input.required",
3436	json!({
3437	"id": input_id,
3438	"input_id": input_id,
3439	"thread_id": event.thread_id,
3440	"turn_id": event.turn_id,
3441	"status": "required",
3442	"request": request,
3443	}),
3444	))
3445	}
3446	"user_input.answered" \| "user_input.canceled" => {
3447	let input_id = payload
3448	.get("input_id")
3449	.or_else(\|\| payload.get("id"))?
3450	.clone();
3451	let status = if event.event == "user_input.answered" {
3452	"submitted"
3453	} else {
3454	"canceled"
3455	};
3456	Some(sse_json(
3457	&event.event,
3458	json!({
3459	"id": input_id,
3460	"input_id": input_id,
3461	"thread_id": event.thread_id,
3462	"turn_id": event.turn_id,
3463	"status": status,
3464	"terminal": payload.get("terminal").and_then(Value::as_bool).unwrap_or(false),
3465	}),
3466	))
3467	}
3468	"sandbox.denied" => Some(sse_json("sandbox.denied", payload.clone())),
3469	"turn.completed" => {
3470	let usage = payload
3471	.get("turn")
3472	.and_then(\|turn\| turn.get("usage"))
3473	.cloned()
3474	.unwrap_or(json!(null));
3475	Some(sse_json("turn.completed", json!({ "usage": usage })))
3476	}
3477	_ => None,
3478	}
3479	}
3480
3481	fn sse_json(event: &str, payload: serde_json::Value) -> SseEvent {
3482	let data = serde_json::to_string(&payload).unwrap_or_else(\|_\| "{}".to_string());
3483	SseEvent::default().event(event).data(data)
3484	}
3485
3486	fn truncate_text(text: &str, max_chars: usize) -> String {
3487	let char_count = text.chars().count();
3488	if char_count <= max_chars {
3489	return text.to_string();
3490	}
3491	let truncated: String = text.chars().take(max_chars.saturating_sub(3)).collect();
3492	format!("{truncated}...")
3493	}
3494
3495	fn resolve_skills_dir(config: &Config, workspace: &std::path::Path) -> PathBuf {
3496	if config.skills_config().scan_codewhale_only() {
3497	if config.skills_dir.is_some() {
3498	return config.skills_dir();
3499	}
3500	if let Some(codewhale_skills_dir) = crate::skills::codewhale_workspace_skills_dir(workspace)
3501	&& let Ok(canonical_skills) = fs::canonicalize(&codewhale_skills_dir)
3502	{
3503	return canonical_skills;
3504	}
3505	return config.skills_dir();
3506	}
3507
3508	// Canonicalize the workspace once so the symlink-containment check below
3509	// compares like-for-like. If the workspace can't be canonicalized at all
3510	// (e.g. it doesn't exist on disk yet) fall back to the configured global
3511	// skills dir rather than risk constructing paths from a non-existent root.
3512	let canonical_workspace = match fs::canonicalize(workspace) {
3513	Ok(path) => path,
3514	Err(_) => return config.skills_dir(),
3515	};
3516	for candidate in [
3517	canonical_workspace.join(".agents").join("skills"),
3518	canonical_workspace.join("skills"),
3519	] {
3520	// Re-canonicalize the candidate so a `.agents/skills` symlink to e.g.
3521	// `/etc` cannot promote arbitrary filesystem locations into the
3522	// skills directory. The candidate must still resolve under the
3523	// canonicalized workspace root after symlink expansion.
3524	if let Ok(canon) = fs::canonicalize(&candidate)
3525	&& canon.starts_with(&canonical_workspace)
3526	&& canon.is_dir()
3527	{
3528	return canon;
3529	}
3530	}
3531	config.skills_dir()
3532	}
3533
3534	fn skills_search_directories(
3535	workspace: &FsPath,
3536	skills_dir: &FsPath,
3537	mode: crate::skills::SkillDiscoveryMode,
3538	) -> Vec<PathBuf> {
3539	crate::skills::skill_directories_for_workspace_and_dir(workspace, skills_dir, mode)
3540	}
3541
3542	fn discover_skills_for_runtime_api(
3543	workspace: &FsPath,
3544	skills_dir: &FsPath,
3545	mode: crate::skills::SkillDiscoveryMode,
3546	plugins: Option<&crate::plugins::PluginRegistry>,
3547	) -> (crate::skills::SkillRegistry, Vec<PathBuf>) {
3548	let directories = skills_search_directories(workspace, skills_dir, mode);
3549	let registry =
3550	crate::skills::discover_from_directories_with_plugins(directories.clone(), plugins);
3551	(registry, directories)
3552	}
3553
3554	fn skill_entry_is_bundled(skill: &crate::skills::Skill, skills_dir: &FsPath) -> bool {
3555	if !crate::skills::is_bundled_skill_name(&skill.name) {
3556	return false;
3557	}
3558
3559	let expected_path = skills_dir.join(&skill.name).join("SKILL.md");
3560	paths_refer_to_same_file(&skill.path, &expected_path)
3561	}
3562
3563	fn paths_refer_to_same_file(left: &FsPath, right: &FsPath) -> bool {
3564	match (fs::canonicalize(left), fs::canonicalize(right)) {
3565	(Ok(left), Ok(right)) => left == right,
3566	_ => left == right,
3567	}
3568	}
3569
3570	fn format_skill_search_paths(directories: &[PathBuf]) -> String {
3571	if directories.is_empty() {
3572	return "<none>".to_string();
3573	}
3574	directories
3575	.iter()
3576	.map(\|path\| path.display().to_string())
3577	.collect::<Vec<_>>()
3578	.join(", ")
3579	}
3580
3581	#[derive(Debug, Deserialize)]
3582	struct UsageQuery {
3583	/// ISO-8601 lower bound (inclusive). When omitted, no lower bound.
3584	since: Option<String>,
3585	/// ISO-8601 upper bound (inclusive). When omitted, no upper bound.
3586	until: Option<String>,
3587	/// Bucket key. One of `day` (default), `model`, `provider`, `thread`.
3588	group_by: Option<String>,
3589	}
3590
3591	fn parse_iso8601(raw: &str, field: &str) -> Result<chrono::DateTime<Utc>, ApiError> {
3592	chrono::DateTime::parse_from_rfc3339(raw)
3593	.map(\|dt\| dt.with_timezone(&Utc))
3594	.map_err(\|e\| ApiError::bad_request(format!("Invalid {field} (expected RFC 3339): {e}")))
3595	}
3596
3597	async fn get_usage(
3598	State(state): State<RuntimeApiState>,
3599	Query(query): Query<UsageQuery>,
3600	) -> Result<Json<Value>, ApiError> {
3601	let since = match query.since.as_deref() {
3602	Some(raw) => Some(parse_iso8601(raw, "since")?),
3603	None => None,
3604	};
3605	let until = match query.until.as_deref() {
3606	Some(raw) => Some(parse_iso8601(raw, "until")?),
3607	None => None,
3608	};
3609	if let (Some(s), Some(u)) = (since, until)
3610	&& s > u
3611	{
3612	return Err(ApiError::bad_request("since must be <= until".to_string()));
3613	}
3614	let group_by = match query.group_by.as_deref().unwrap_or("day") {
3615	"day" => UsageGroupBy::Day,
3616	"model" => UsageGroupBy::Model,
3617	"provider" => UsageGroupBy::Provider,
3618	"thread" => UsageGroupBy::Thread,
3619	other => {
3620	return Err(ApiError::bad_request(format!(
3621	"Unsupported group_by '{other}': expected one of day, model, provider, thread"
3622	)));
3623	}
3624	};
3625
3626	let aggregation = state
3627	.runtime_threads
3628	.aggregate_usage(since, until, group_by)
3629	.await
3630	.map_err(\|e\| ApiError::internal(e.to_string()))?;
3631	Ok(Json(json!(aggregation)))
3632	}
3633
3634	#[derive(Debug, Deserialize)]
3635	struct SnapshotsQuery {
3636	/// Maximum number of snapshots to return. Mirrors `/restore list [N]`.
3637	limit: Option<usize>,
3638	}
3639
3640	#[derive(Debug, Serialize)]
3641	struct SnapshotEntry {
3642	id: String,
3643	label: String,
3644	timestamp: i64,
3645	}
3646
3647	async fn list_snapshots(
3648	State(state): State<RuntimeApiState>,
3649	Query(query): Query<SnapshotsQuery>,
3650	) -> Result<Json<Vec<SnapshotEntry>>, ApiError> {
3651	Ok(Json(snapshot_entries_for_workspace(
3652	&state.workspace,
3653	query,
3654	)?))
3655	}
3656
3657	async fn restore_snapshot(
3658	State(state): State<RuntimeApiState>,
3659	Path(id): Path<String>,
3660	) -> Result<Json<Value>, ApiError> {
3661	restore_snapshot_for_workspace(&state.workspace, &id)?;
3662	Ok(Json(json!({
3663	"restored": id,
3664	})))
3665	}
3666
3667	fn restore_snapshot_for_workspace(workspace: &FsPath, id: &str) -> Result<(), ApiError> {
3668	let repo = crate::snapshot::SnapshotRepo::open_or_init(workspace)
3669	.map_err(\|e\| ApiError::internal(format!("Snapshot repo init failed: {e}")))?;
3670	let snapshot_id = crate::snapshot::SnapshotId(id.to_string());
3671	repo.restore(&snapshot_id)
3672	.map_err(\|e\| ApiError::internal(format!("Snapshot restore failed: {e}")))
3673	}
3674
3675	fn snapshot_entries_for_workspace(
3676	workspace: &FsPath,
3677	query: SnapshotsQuery,
3678	) -> Result<Vec<SnapshotEntry>, ApiError> {
3679	const DEFAULT_LIMIT: usize = 20;
3680	const MAX_LIMIT: usize = 100;
3681
3682	let limit = match query.limit.unwrap_or(DEFAULT_LIMIT) {
3683	1..=MAX_LIMIT => query.limit.unwrap_or(DEFAULT_LIMIT),
3684	other => {
3685	return Err(ApiError::bad_request(format!(
3686	"limit must be between 1 and {MAX_LIMIT}; got {other}",
3687	)));
3688	}
3689	};
3690	let repo = crate::snapshot::SnapshotRepo::open_or_init(workspace)
3691	.map_err(\|e\| ApiError::internal(format!("Snapshot repo unavailable: {e}")))?;
3692	let snapshots = repo
3693	.list(limit)
3694	.map_err(\|e\| ApiError::internal(format!("Failed to list snapshots: {e}")))?;
3695	Ok(snapshots
3696	.into_iter()
3697	.map(\|snapshot\| SnapshotEntry {
3698	id: snapshot.id.as_str().to_string(),
3699	label: snapshot.label,
3700	timestamp: snapshot.timestamp,
3701	})
3702	.collect())
3703	}
3704
3705	// ── Provider / Model catalog endpoints ──
3706
3707	/// Entry in `GET /v1/providers`.
3708	///
3709	/// Exposes the static provider registry so the GUI can render a dynamic
3710	/// provider picker instead of hard-coding `deepseek` only. The `id` matches
3711	/// `ApiProvider::as_str()` and is the value the GUI should send back via
3712	/// `POST /v1/config { key: "provider", value: <id> }`.
3713	#[derive(Debug, Clone, Serialize)]
3714	struct ProviderEntry {
3715	/// Stable identifier — matches `ApiProvider::as_str()` and the TOML
3716	/// `provider = "<id>"` key. Use this as the canonical value when
3717	/// persisting or comparing.
3718	id: String,
3719	/// Human-friendly name for picker UIs (e.g. "DeepSeek", "OpenAI").
3720	display_name: String,
3721	/// Default base URL for this provider ( informational; the live base URL
3722	/// may be overridden in config.toml).
3723	default_base_url: String,
3724	/// Default model id for this provider, if any. Empty for pass-through
3725	/// providers (Ollama / Custom) that expose no built-in catalog.
3726	default_model: String,
3727	/// Whether this provider exposes a built-in model list. When false, the
3728	/// GUI should render a free-text input instead of calling
3729	/// `/v1/providers/{id}/models`.
3730	has_model_catalog: bool,
3731	/// API key environment variable candidates, e.g. `["DEEPSEEK_API_KEY"]`.
3732	/// The GUI may surface these in a tooltip when auth is missing.
3733	env_vars: Vec<String>,
3734	}
3735
3736	#[derive(Debug, Clone, Serialize)]
3737	struct ProvidersResponse {
3738	/// Currently active provider id (matches `GET /v1/config`'s `provider`).
3739	current: String,
3740	providers: Vec<ProviderEntry>,
3741	}
3742
3743	/// Entry in `GET /v1/providers/{id}/models`.
3744	#[derive(Debug, Clone, Serialize)]
3745	struct ProviderModelEntry {
3746	/// Canonical model id (suitable for `default_text_model` or
3747	/// `POST /v1/threads/{id}` `model` field).
3748	id: String,
3749	}
3750
3751	#[derive(Debug, Clone, Serialize)]
3752	struct ProviderModelsResponse {
3753	provider: String,
3754	models: Vec<ProviderModelEntry>,
3755	}
3756
3757	fn push_unique_model(models: &mut Vec<String>, model: &str) {
3758	let model = model.trim();
3759	if !model.is_empty()
3760	&& !models
3761	.iter()
3762	.any(\|existing\| existing.eq_ignore_ascii_case(model))
3763	{
3764	models.push(model.to_string());
3765	}
3766	}
3767
3768	fn normalize_api_base_url(base_url: &str) -> String {
3769	base_url.trim().trim_end_matches('/').to_ascii_lowercase()
3770	}
3771
3772	fn provider_uses_custom_route_for_api(config: &Config, provider: ApiProvider) -> bool {
3773	config
3774	.provider_config_for(provider)
3775	.and_then(\|entry\| entry.base_url.as_deref())
3776	.is_some_and(\|base_url\| {
3777	normalize_api_base_url(base_url) != normalize_api_base_url(provider.default_base_url())
3778	})
3779	}
3780
3781	fn provider_models_for_api(
3782	config: &Config,
3783	active_provider: ApiProvider,
3784	provider: ApiProvider,
3785	) -> Vec<String> {
3786	let mut models = Vec::new();
3787	if let Some(model) = config
3788	.provider_config_for(provider)
3789	.and_then(\|entry\| entry.model.as_deref())
3790	{
3791	push_unique_model(&mut models, model);
3792	}
3793	if provider == active_provider {
3794	let active_model = config.default_model();
3795	if !active_model.trim().eq_ignore_ascii_case("auto") {
3796	push_unique_model(&mut models, &active_model);
3797	}
3798	if config.model_ids_pass_through() {
3799	return models;
3800	}
3801	}
3802	if provider_uses_custom_route_for_api(config, provider) {
3803	return models;
3804	}
3805	for model in crate::provider_lake::models_for_provider(config, active_provider, provider) {
3806	push_unique_model(&mut models, &model);
3807	}
3808	models
3809	}
3810
3811	fn provider_default_model_for_api(
3812	config: &Config,
3813	active_provider: ApiProvider,
3814	provider: ApiProvider,
3815	) -> String {
3816	if provider == active_provider {
3817	return config.default_model();
3818	}
3819	provider_models_for_api(config, active_provider, provider)
3820	.into_iter()
3821	.next()
3822	.unwrap_or_default()
3823	}
3824
3825	async fn list_providers(
3826	State(state): State<RuntimeApiState>,
3827	) -> Result<Json<ProvidersResponse>, ApiError> {
3828	let config = state.config.read().clone();
3829	let active_provider = config.api_provider();
3830	let current = active_provider.as_str().to_string();
3831	let mut providers = Vec::new();
3832	for api_provider in ApiProvider::sorted_for_display() {
3833	let default_model = provider_default_model_for_api(&config, active_provider, api_provider);
3834	let has_model_catalog =
3835	!crate::provider_lake::all_catalog_models_for_provider(api_provider).is_empty();
3836	providers.push(ProviderEntry {
3837	id: api_provider.as_str().to_string(),
3838	display_name: api_provider.display_name().to_string(),
3839	default_base_url: api_provider.default_base_url().to_string(),
3840	default_model,
3841	has_model_catalog,
3842	env_vars: api_provider
3843	.env_vars()
3844	.iter()
3845	.map(std::string::ToString::to_string)
3846	.collect(),
3847	});
3848	}
3849	Ok(Json(ProvidersResponse { current, providers }))
3850	}
3851
3852	#[derive(Debug, Deserialize)]
3853	struct ListProviderModelsParams {
3854	/// Optional filter: when provided, models whose id contains this
3855	/// substring (case-insensitive) are returned. Currently informational —
3856	/// the catalog is small enough to filter client-side.
3857	#[serde(default)]
3858	#[allow(dead_code)]
3859	filter: Option<String>,
3860	}
3861
3862	async fn list_provider_models(
3863	State(state): State<RuntimeApiState>,
3864	Path(id): Path<String>,
3865	_params: Query<ListProviderModelsParams>,
3866	) -> Result<Json<ProviderModelsResponse>, ApiError> {
3867	let config = state.config.read().clone();
3868	let active_provider = config.api_provider();
3869	let api_provider = ApiProvider::parse(&id)
3870	.ok_or_else(\|\| ApiError::bad_request(format!("Unknown provider id '{id}'")))?;
3871	// Reject requests for the legacy deepseek-cn alias that has no
3872	// ProviderKind metadata — the GUI should use `deepseek` instead.
3873	if api_provider == ApiProvider::DeepseekCN {
3874	return Err(ApiError::bad_request(
3875	"provider 'deepseek-cn' is a legacy alias; use 'deepseek' instead",
3876	));
3877	}
3878	let models = provider_models_for_api(&config, active_provider, api_provider)
3879	.into_iter()
3880	.map(\|id\| ProviderModelEntry { id: id.to_string() })
3881	.collect();
3882	Ok(Json(ProviderModelsResponse {
3883	provider: api_provider.as_str().to_string(),
3884	models,
3885	}))
3886	}
3887
3888	/// Request body for `POST /v1/providers/{id}/switch`.
3889	///
3890	/// Mirrors the TUI's `AppAction::SwitchProvider { provider, model }` payload
3891	/// (see `tui/ui.rs::switch_provider`). `model` is optional: when omitted,
3892	/// the runtime resolves the active model from `[providers.<id>].model` (or
3893	/// the provider's built-in default) and does not persist a `model` key,
3894	/// so the user's per-provider config is preserved. When provided, the model
3895	/// is normalized, persisted for the target provider, and (for DeepSeek
3896	/// providers) also pinned as `default_text_model`.
3897	#[derive(Debug, Deserialize, Default)]
3898	struct SwitchProviderRequest {
3899	#[serde(default)]
3900	model: Option<String>,
3901	}
3902
3903	/// Response for `POST /v1/providers/{id}/switch`.
3904	#[derive(Debug, Serialize)]
3905	struct SwitchProviderResponse {
3906	/// The provider id that was switched to (echoes the path).
3907	provider: String,
3908	/// The resolved active model after the switch. This is the model the
3909	/// runtime will use for new turns — either the user-supplied override
3910	/// or the value resolved from `[providers.<id>].model` / the
3911	/// provider's built-in default. The GUI should display this value,
3912	/// not `ProviderEntry.default_model`, to avoid showing the catalog
3913	/// default when the user has configured a different model.
3914	model: String,
3915	/// Human-readable status message for logging/toasts.
3916	message: String,
3917	/// Whether the new provider + model were persisted to config.toml.
3918	persisted: bool,
3919	}
3920
3921	/// `POST /v1/providers/{id}/switch` — switch the active provider, optionally
3922	/// overriding the model.
3923	///
3924	/// This is the GUI-facing counterpart of the TUI's `/provider` slash command
3925	/// (`commands/groups/core/provider.rs`) and `AppAction::SwitchProvider`
3926	/// (`tui/ui.rs::switch_provider`). It exists so the GUI does not have to
3927	/// simulate the switch with multiple `POST /v1/config` calls + a reload,
3928	/// which historically led to two bugs:
3929	///
3930	/// 1. The GUI persisted `model = <catalog default>` even when the user
3931	/// clicked the picker without choosing a model, clobbering a user-set
3932	/// `[providers.<id>].model` (e.g. `glm-2` overwritten with
3933	/// `deepseek-v4-pro`).
3934	/// 2. The GUI then displayed the catalog default instead of the actually
3935	/// resolved model, because it never asked the backend what model was
3936	/// selected.
3937	///
3938	/// Persistence mirrors `switch_provider` (ui.rs:9390-9410):
3939	/// - `provider` is always persisted (root `provider` key).
3940	/// - `model` is persisted only when `model_override.is_some()`, via
3941	/// `persist_provider_model_key` (writes `[providers.<id>].model`, or the
3942	/// root `default_text_model` for DeepSeek). The `Settings` provider-model
3943	/// map is updated the same way, including the DeepSeek-specific
3944	/// `default_model` pin.
3945	/// - Config is reloaded from disk and synced to active engines via
3946	/// `runtime_threads.reload_config`, exactly like `POST /v1/config/reload`.
3947	async fn switch_provider(
3948	State(state): State<RuntimeApiState>,
3949	Path(id): Path<String>,
3950	Json(req): Json<SwitchProviderRequest>,
3951	) -> Result<Json<SwitchProviderResponse>, ApiError> {
3952	use crate::config_persistence;
3953
3954	let target = ApiProvider::parse(&id)
3955	.ok_or_else(\|\| ApiError::bad_request(format!("Unknown provider id '{id}'")))?;
3956	// Reject the legacy deepseek-cn alias — same guard as list_provider_models.
3957	if target == ApiProvider::DeepseekCN {
3958	return Err(ApiError::bad_request(
3959	"provider 'deepseek-cn' is a legacy alias; use 'deepseek' instead",
3960	));
3961	}
3962
3963	// Normalize the optional model override against the target provider.
3964	// Mirrors `set_config`'s `model` branch, which validates against the
3965	// active route — except here we validate against the target provider,
3966	// because the active route is about to change.
3967	let model_override: Option<String> = match req.model.as_deref().map(str::trim) {
3968	None \| Some("") => None,
3969	Some(raw) => Some(normalize_runtime_config_model(target, raw)?),
3970	};
3971
3972	// Resolve the target provider identity before mutating config, so
3973	// persistence uses the same key the TUI's switch_provider would.
3974	let (provider_identity, _active_provider) = {
3975	let config = state.config.read();
3976	(config.provider_identity_for(target), config.api_provider())
3977	};
3978
3979	// Persist `provider` (always) + `model` (only when explicitly given).
3980	// This is the critical TUI-parity rule: a bare `/provider <id>` (no
3981	// model arg) MUST NOT write a `model` key, otherwise the user's
3982	// per-provider `[providers.<id>].model` config gets overwritten with
3983	// whatever the runtime resolves as the default.
3984	config_persistence::persist_root_string_key(
3985	state.config_path.as_deref(),
3986	"provider",
3987	&provider_identity,
3988	)
3989	.map_err(\|e\| ApiError::internal(format!("Failed to persist provider: {e}")))?;
3990
3991	if let Some(ref model) = model_override {
3992	config_persistence::persist_provider_model_key(
3993	state.config_path.as_deref(),
3994	target,
3995	&provider_identity,
3996	model,
3997	)
3998	.map_err(\|e\| ApiError::internal(format!("Failed to persist model: {e}")))?;
3999
4000	// Mirror the TUI's Settings update (ui.rs:9398-9406): record the
4001	// provider→model mapping, and for DeepSeek also pin the global
4002	// `default_model`. Failures here are non-fatal — the config.toml
4003	// write above is the source of truth.
4004	let _ = crate::settings::Settings::transact(\|settings\| {
4005	settings.set_model_for_provider(target.as_str(), model);
4006	if matches!(target, ApiProvider::Deepseek \| ApiProvider::DeepseekCN) {
4007	let _ = settings.set("default_model", model);
4008	}
4009	Ok(())
4010	});
4011	}
4012
4013	// Reload config from disk and sync to active engines. This matches
4014	// `POST /v1/config/reload` exactly: load → validate thread routes →
4015	// swap in the new config. A failure here means an active thread's
4016	// route is invalid under the new provider — surface it so the GUI can
4017	// tell the user to fix their config.
4018	let reloaded = Config::load(state.config_path.clone(), state.config_profile.as_deref())
4019	.map_err(\|e\| ApiError::internal(format!("Failed to reload config: {e}")))?;
4020	state
4021	.runtime_threads
4022	.reload_config(reloaded.clone())
4023	.await
4024	.map_err(\|err\| ApiError::bad_request(format!("Config reload rejected: {err}")))?;
4025	{
4026	let mut config = state.config.write();
4027	*config = reloaded;
4028	}
4029
4030	// Read the resolved active model + provider from the freshly reloaded
4031	// config. This is the value the GUI must display — NOT the catalog
4032	// default and NOT the previously-active model.
4033	let (active_provider, active_model) = {
4034	let config = state.config.read();
4035	(config.api_provider(), config.default_model())
4036	};
4037
4038	let message = if model_override.is_some() {
4039	format!(
4040	"Provider switched to {} (model: {}).",
4041	active_provider.as_str(),
4042	active_model
4043	)
4044	} else {
4045	format!(
4046	"Provider switched to {} (model: {}, resolved from config).",
4047	active_provider.as_str(),
4048	active_model
4049	)
4050	};
4051
4052	Ok(Json(SwitchProviderResponse {
4053	provider: active_provider.as_str().to_string(),
4054	model: active_model,
4055	message,
4056	persisted: true,
4057	}))
4058	}
4059
4060	// ── Config endpoints ──
4061
4062	/// GUI-relevant config snapshot returned by `GET /v1/config`.
4063	#[derive(Debug, Clone, Serialize)]
4064	struct GuiConfigResponse {
4065	model: String,
4066	provider: String,
4067	approval_mode: String,
4068	reasoning_effort: String,
4069	auto_compact: bool,
4070	cost_currency: String,
4071	default_mode: String,
4072	default_model: String,
4073	base_url: String,
4074	allow_shell: bool,
4075	mcp_config_path: String,
4076	subagents_enabled: bool,
4077	subagents_max_depth: u32,
4078	show_thinking: bool,
4079	thinking_default_expanded: bool,
4080	thinking_highlight: bool,
4081	show_tool_details: bool,
4082	inline_diffs: String,
4083	locale: String,
4084	max_history: usize,
4085	workspace_follow_symlinks: bool,
4086	calm_mode: bool,
4087	sandbox_mode: String,
4088	strict_tool_mode: bool,
4089	memory_enabled: bool,
4090	search_provider: String,
4091	prompt_suggestion: bool,
4092	}
4093
4094	/// Request body for `POST /v1/config` (set a single config key).
4095	#[derive(Debug, Deserialize)]
4096	struct SetConfigRequest {
4097	key: String,
4098	value: String,
4099	#[serde(default)]
4100	persist: bool,
4101	}
4102
4103	/// Response for `POST /v1/config` (set a single config key).
4104	#[derive(Debug, Serialize)]
4105	struct SetConfigResponse {
4106	key: String,
4107	value: String,
4108	message: String,
4109	persisted: bool,
4110	requires_reload: bool,
4111	}
4112
4113	fn persist_runtime_tui_setting(key: &str, value: &str) -> Result<(), ApiError> {
4114	// Validate against a throwaway copy first, so an invalid value is still a
4115	// 400 rather than an internal error raised from inside the transaction.
4116	let mut probe = crate::settings::Settings::load_persisted()
4117	.map_err(\|e\| ApiError::internal(format!("Failed to load settings: {e}")))?;
4118	probe
4119	.set(key, value)
4120	.map_err(\|e\| ApiError::bad_request(e.to_string()))?;
4121	// The write itself re-applies the key inside `Settings::transact`, so it
4122	// cannot save the stale snapshot above over a concurrent writer's field.
4123	crate::settings::Settings::transact(\|settings\| settings.set(key, value))
4124	.map_err(\|e\| ApiError::internal(format!("Failed to save settings: {e}")))
4125	}
4126
4127	/// Response for `POST /v1/config/reload`.
4128	#[derive(Debug, Serialize)]
4129	struct ReloadConfigResponse {
4130	message: String,
4131	}
4132
4133	async fn get_config(
4134	State(state): State<RuntimeApiState>,
4135	) -> Result<Json<GuiConfigResponse>, ApiError> {
4136	let config = state.config.read();
4137	let settings = crate::settings::Settings::load_persisted().unwrap_or_default();
4138	let mcp_config_path = config.mcp_config_path().display().to_string();
4139
4140	let model = config.default_model();
4141
4142	let provider = config.provider_identity_for(config.api_provider());
4143
4144	let approval_mode = config
4145	.approval_policy
4146	.as_deref()
4147	.unwrap_or("suggest")
4148	.to_string();
4149	let reasoning_effort = config.reasoning_effort().unwrap_or("auto").to_string();
4150	let cost_currency = settings.cost_currency.clone();
4151	let default_mode = settings.default_mode.as_str().to_string();
4152	// This field is the legacy root DeepSeek fallback, not the active
4153	// provider model above. Keeping the two explicit prevents a Z.ai model
4154	// update from silently rewriting a future DeepSeek route.
4155	let default_model = config
4156	.default_text_model
4157	.clone()
4158	.unwrap_or_else(\|\| DEFAULT_TEXT_MODEL.to_string());
4159	let base_url = config.deepseek_base_url().to_string();
4160
4161	Ok(Json(GuiConfigResponse {
4162	model,
4163	provider,
4164	approval_mode,
4165	reasoning_effort,
4166	auto_compact: settings.auto_compact,
4167	cost_currency,
4168	default_mode,
4169	default_model,
4170	base_url,
4171	allow_shell: config.allow_shell(),
4172	mcp_config_path,
4173	subagents_enabled: config.subagents_enabled(),
4174	subagents_max_depth: config.subagent_max_spawn_depth(),
4175	show_thinking: settings.show_thinking,
4176	thinking_default_expanded: settings.thinking_default_expanded,
4177	thinking_highlight: settings.thinking_highlight,
4178	show_tool_details: settings.show_tool_details,
4179	inline_diffs: settings.inline_diffs.clone(),
4180	locale: settings.locale.clone(),
4181	max_history: settings.max_input_history,
4182	workspace_follow_symlinks: settings.workspace_follow_symlinks,
4183	calm_mode: settings.calm_mode,
4184	sandbox_mode: config
4185	.sandbox_mode
4186	.clone()
4187	.unwrap_or_else(\|\| "workspace-write".to_string()),
4188	strict_tool_mode: config.strict_tool_mode.unwrap_or(false),
4189	memory_enabled: config.memory_enabled(),
4190	search_provider: config.search_provider().as_str().to_string(),
4191	prompt_suggestion: config.prompt_suggestion_enabled(),
4192	}))
4193	}
4194
4195	async fn set_config(
4196	State(state): State<RuntimeApiState>,
4197	Json(req): Json<SetConfigRequest>,
4198	) -> Result<Json<SetConfigResponse>, ApiError> {
4199	use crate::config_persistence;
4200
4201	let key = req.key.to_lowercase();
4202	let mut value = req.value;
4203	let persist = req.persist;
4204
4205	// Validate model keys even for dry-run requests. Model ids are provider
4206	// owned; accepting a DeepSeek id while Z.ai is active creates a saved
4207	// route that cannot execute after reload.
4208	let active_route = {
4209	let config = state.config.read();
4210	let provider = config.api_provider();
4211	(provider, config.provider_identity_for(provider))
4212	};
4213	match key.as_str() {
4214	"model" => {
4215	value = normalize_runtime_config_model(active_route.0, &value)?;
4216	}
4217	"default_model" => {
4218	value = normalize_runtime_config_model(ApiProvider::Deepseek, &value)?;
4219	}
4220	_ => {}
4221	}
4222
4223	// All persisted config keys require a reload to take effect in the
4224	// runtime (including syncing to active engines). The caller should
4225	// POST /v1/config/reload after persisting.
4226	let requires_reload = persist;
4227
4228	// Handle persistence directly via config_persistence.
4229	// The runtime's in-memory state is NOT mutated here; the caller
4230	// should POST /v1/config/reload after persisting to apply changes.
4231	if persist {
4232	let config_path = state.config_path.as_deref();
4233	let result: anyhow::Result<PathBuf> = match key.as_str() {
4234	"model" => config_persistence::persist_provider_model_key(
4235	config_path,
4236	active_route.0,
4237	&active_route.1,
4238	&value,
4239	),
4240	"default_model" => config_persistence::persist_root_string_key(
4241	config_path,
4242	"default_text_model",
4243	&value,
4244	),
4245	"reasoning_effort" => {
4246	config_persistence::persist_root_string_key(config_path, "reasoning_effort", &value)
4247	}
4248	"approval_mode" \| "approval_policy" => {
4249	config_persistence::persist_root_string_key(config_path, "approval_policy", &value)
4250	}
4251	"base_url" => config_persistence::persist_root_string_key(
4252	config_path,
4253	"deepseek_base_url",
4254	&value,
4255	),
4256	"provider" => {
4257	// Validate the provider id against the static registry so the
4258	// GUI gets a clear error instead of silently persisting an
4259	// unknown value that `Config::api_provider()` would later
4260	// ignore (falling back to DeepSeek).
4261	let parsed = ApiProvider::parse(&value).ok_or_else(\|\| {
4262	ApiError::bad_request(format!(
4263	"Unknown provider '{value}'. Call GET /v1/providers for the list of supported ids."
4264	))
4265	})?;
4266	let result =
4267	config_persistence::persist_root_string_key(config_path, "provider", &value);
4268	if result.is_ok() {
4269	// Keep the in-memory provider in step with the persisted
4270	// value so a following set_config(model) resolves the new
4271	// provider's table instead of clobbering the previous
4272	// provider's root default_text_model (#4658 follow-up).
4273	state.config.write().provider = Some(parsed.as_str().to_string());
4274	}
4275	result
4276	}
4277	"provider_url" \| "provider_base_url" => {
4278	let provider = state.config.read().api_provider();
4279	config_persistence::persist_provider_base_url_key(config_path, provider, &value)
4280	}
4281	"cost_currency"
4282	\| "default_mode"
4283	\| "auto_compact"
4284	\| "show_thinking"
4285	\| "thinking_default_expanded"
4286	\| "thinking_highlight"
4287	\| "show_tool_details"
4288	\| "inline_diffs"
4289	\| "calm_mode"
4290	\| "workspace_follow_symlinks"
4291	\| "locale"
4292	\| "max_history" => {
4293	persist_runtime_tui_setting(&key, &value)?;
4294	return Ok(Json(SetConfigResponse {
4295	key,
4296	value,
4297	message: "Config persisted. Call /v1/config/reload to apply.".to_string(),
4298	persisted: true,
4299	requires_reload,
4300	}));
4301	}
4302	"allow_shell" => {
4303	let enabled = value.parse::<bool>().map_err(\|_\| {
4304	ApiError::bad_request(format!(
4305	"Invalid value '{value}' for allow_shell: expected 'true' or 'false'"
4306	))
4307	})?;
4308	config_persistence::persist_root_bool_key(config_path, "allow_shell", enabled)
4309	}
4310	"mcp_config_path" => {
4311	config_persistence::persist_root_string_key(config_path, "mcp_config_path", &value)
4312	}
4313	"subagents_enabled" => {
4314	let enabled = value.parse::<bool>().map_err(\|_\| {
4315	ApiError::bad_request(format!(
4316	"Invalid value '{value}' for subagents_enabled: expected 'true' or 'false'"
4317	))
4318	})?;
4319	config_persistence::persist_subagents_bool_key(config_path, "enabled", enabled)
4320	}
4321	"subagents_max_depth" => {
4322	let raw = value.parse::<u64>().map_err(\|_\| {
4323	ApiError::bad_request(format!(
4324	"Invalid value '{value}' for subagents_max_depth: expected a non-negative integer"
4325	))
4326	})?;
4327	let clamped = raw.min(u64::from(codewhale_config::MAX_SPAWN_DEPTH_CEILING));
4328	config_persistence::persist_subagents_integer_key(config_path, "max_depth", clamped)
4329	}
4330	"sandbox_mode" => {
4331	let normalized = match value.to_lowercase().as_str() {
4332	"none" \| "off" \| "disabled" => "none".to_string(),
4333	"opensandbox" \| "external-sandbox" \| "external" => "opensandbox".to_string(),
4334	"workspace-write" \| "workspace_write" => "workspace-write".to_string(),
4335	"read-only" \| "read_only" => "read-only".to_string(),
4336	"danger-full-access" \| "danger_full_access" \| "full" => {
4337	"danger-full-access".to_string()
4338	}
4339	"workspace" \| "workspace-read-write" \| "workspace_read_write" => {
4340	"workspace-write".to_string()
4341	}
4342	_ => {
4343	return Err(ApiError::bad_request(format!(
4344	"Invalid sandbox_mode '{value}'. Supported: none, read-only, workspace-write, danger-full-access, opensandbox"
4345	)));
4346	}
4347	};
4348	config_persistence::persist_root_string_key(
4349	config_path,
4350	"sandbox_mode",
4351	&normalized,
4352	)
4353	}
4354	"strict_tool_mode" => {
4355	let enabled = value.parse::<bool>().map_err(\|_\| {
4356	ApiError::bad_request(format!(
4357	"Invalid value '{value}' for strict_tool_mode: expected 'true' or 'false'"
4358	))
4359	})?;
4360	config_persistence::persist_root_bool_key(config_path, "strict_tool_mode", enabled)
4361	}
4362	"memory_enabled" => {
4363	let enabled = value.parse::<bool>().map_err(\|_\| {
4364	ApiError::bad_request(format!(
4365	"Invalid value '{value}' for memory_enabled: expected 'true' or 'false'"
4366	))
4367	})?;
4368	config_persistence::persist_table_bool_key(
4369	config_path,
4370	"memory",
4371	"enabled",
4372	enabled,
4373	)
4374	}
4375	"search_provider" => {
4376	let normalized = value.to_lowercase();
4377	config_persistence::persist_table_string_key(
4378	config_path,
4379	"search",
4380	"provider",
4381	&normalized,
4382	)
4383	}
4384	"prompt_suggestion" => {
4385	let enabled = value.parse::<bool>().map_err(\|_\| {
4386	ApiError::bad_request(format!(
4387	"Invalid value '{value}' for prompt_suggestion: expected 'true' or 'false'"
4388	))
4389	})?;
4390	config_persistence::persist_root_bool_key(config_path, "prompt_suggestion", enabled)
4391	}
4392	_ => {
4393	return Err(ApiError::bad_request(format!(
4394	"Unknown config key '{key}'. Supported keys: model, default_model, reasoning_effort, approval_mode, base_url, provider, provider_url, cost_currency, default_mode, auto_compact, allow_shell, mcp_config_path, show_thinking, thinking_default_expanded, thinking_highlight, show_tool_details, inline_diffs, locale, max_history, calm_mode, workspace_follow_symlinks, subagents_enabled, subagents_max_depth, sandbox_mode, strict_tool_mode, memory_enabled, search_provider, prompt_suggestion"
4395	)));
4396	}
4397	};
4398
4399	if let Err(e) = result {
4400	return Err(ApiError::internal(format!(
4401	"Failed to persist config key '{key}': {e}"
4402	)));
4403	}
4404	}
4405
4406	Ok(Json(SetConfigResponse {
4407	key,
4408	value,
4409	message: if persist {
4410	"Config persisted. Call /v1/config/reload to apply.".to_string()
4411	} else {
4412	"Config not persisted (add persist: true to save)".to_string()
4413	},
4414	persisted: persist,
4415	requires_reload,
4416	}))
4417	}
4418
4419	fn normalize_runtime_config_model(provider: ApiProvider, value: &str) -> Result<String, ApiError> {
4420	let value = value.trim();
4421	validate_route(provider, value).map_err(ApiError::bad_request)?;
4422	if value.eq_ignore_ascii_case("auto") {
4423	return Ok("auto".to_string());
4424	}
4425	normalize_model_name_for_provider(provider, value).ok_or_else(\|\| {
4426	ApiError::bad_request(format!(
4427	"Invalid model '{value}' for provider '{}'.",
4428	provider.as_str()
4429	))
4430	})
4431	}
4432
4433	async fn reload_config(
4434	State(state): State<RuntimeApiState>,
4435	) -> Result<Json<ReloadConfigResponse>, ApiError> {
4436	let reloaded = Config::load(state.config_path.clone(), state.config_profile.as_deref())
4437	.map_err(\|e\| ApiError::internal(format!("Failed to reload config: {e}")))?;
4438	state
4439	.runtime_threads
4440	.reload_config(reloaded.clone())
4441	.await
4442	.map_err(\|err\| ApiError::bad_request(format!("Config reload rejected: {err}")))?;
4443	{
4444	let mut config = state.config.write();
4445	*config = reloaded;
4446	}
4447	Ok(Json(ReloadConfigResponse {
4448	message: "Config reloaded from disk; new turns will resolve the updated provider routes"
4449	.to_string(),
4450	}))
4451	}
4452
4453	const MOBILE_HTML: &str = include_str!("runtime_mobile.html");
4454
4455	/// Built-in dev origins always allowed by the runtime API (whalescale#255).
4456	const DEFAULT_CORS_ORIGINS: &[&str] = &[
4457	"http://localhost:3000",
4458	"http://127.0.0.1:3000",
4459	"http://localhost:1420",
4460	"http://127.0.0.1:1420",
4461	"tauri://localhost",
4462	];
4463
4464	fn cors_layer(extra_origins: &[String]) -> CorsLayer {
4465	let mut origins: Vec<HeaderValue> = DEFAULT_CORS_ORIGINS
4466	.iter()
4467	.filter_map(\|o\| HeaderValue::from_str(o).ok())
4468	.collect();
4469	for raw in extra_origins {
4470	let trimmed = raw.trim();
4471	if trimmed.is_empty() {
4472	continue;
4473	}
4474	match HeaderValue::from_str(trimmed) {
4475	Ok(value) if !origins.contains(&value) => origins.push(value),
4476	Ok(_) => {}
4477	Err(err) => tracing::warn!(
4478	"Ignoring invalid CORS origin '{trimmed}': {err}; expected scheme://host[:port]"
4479	),
4480	}
4481	}
4482	CorsLayer::new()
4483	.allow_origin(origins)
4484	.allow_methods([
4485	Method::GET,
4486	Method::POST,
4487	Method::PATCH,
4488	Method::DELETE,
4489	Method::OPTIONS,
4490	])
4491	.allow_headers([
4492	header::AUTHORIZATION,
4493	header::CONTENT_TYPE,
4494	header::ACCEPT,
4495	HeaderName::from_static("x-codewhale-runtime-token"),
4496	HeaderName::from_static("x-deepseek-runtime-token"),
4497	])
4498	}
4499
4500	fn map_task_err(err: anyhow::Error) -> ApiError {
4501	let message = err.to_string();
4502	if message.contains("not found") {
4503	ApiError::not_found(message)
4504	} else {
4505	ApiError::bad_request(message)
4506	}
4507	}
4508
4509	fn map_automation_err(err: anyhow::Error) -> ApiError {
4510	let message = err.to_string();
4511	if message.contains("Failed to read automation")
4512	\|\| message.contains("No such file or directory")
4513	{
4514	ApiError::not_found(message)
4515	} else {
4516	ApiError::bad_request(message)
4517	}
4518	}
4519
4520	fn map_thread_err(err: anyhow::Error) -> ApiError {
4521	let message = err.to_string();
4522	let lower = message.to_ascii_lowercase();
4523	if (lower.starts_with("thread '") && lower.ends_with("' not found"))
4524	\|\| lower.starts_with("thread not found:")
4525	{
4526	ApiError::not_found(message)
4527	} else if message.contains("already has an active turn")
4528	\|\| message.contains("No active turn")
4529	\|\| message.contains("is not active")
4530	{
4531	ApiError {
4532	status: StatusCode::CONFLICT,
4533	message,
4534	}
4535	} else {
4536	ApiError::bad_request(message)
4537	}
4538	}
4539
4540	#[derive(Debug, Clone)]
4541	struct ApiError {
4542	status: StatusCode,
4543	message: String,
4544	}
4545
4546	impl ApiError {
4547	fn bad_request(message: impl Into<String>) -> Self {
4548	Self {
4549	status: StatusCode::BAD_REQUEST,
4550	message: message.into(),
4551	}
4552	}
4553
4554	fn not_found(message: impl Into<String>) -> Self {
4555	Self {
4556	status: StatusCode::NOT_FOUND,
4557	message: message.into(),
4558	}
4559	}
4560
4561	fn conflict(message: impl Into<String>) -> Self {
4562	Self {
4563	status: StatusCode::CONFLICT,
4564	message: message.into(),
4565	}
4566	}
4567
4568	fn not_implemented(message: impl Into<String>) -> Self {
4569	Self {
4570	status: StatusCode::NOT_IMPLEMENTED,
4571	message: message.into(),
4572	}
4573	}
4574
4575	fn internal(message: impl Into<String>) -> Self {
4576	Self {
4577	status: StatusCode::INTERNAL_SERVER_ERROR,
4578	message: message.into(),
4579	}
4580	}
4581	}
4582
4583	impl IntoResponse for ApiError {
4584	fn into_response(self) -> Response {
4585	(
4586	self.status,
4587	Json(json!({
4588	"error": {
4589	"message": self.message,
4590	"status": self.status.as_u16(),
4591	}
4592	})),
4593	)
4594	.into_response()
4595	}
4596	}
4597
4598	#[cfg(test)]
4599	mod tests;
4600

4600 lines RUST