runtime_chat_relay.rs - CodeWhale

runtime_chat_relay.rs

1	//! Isolated native Runtime execution for account-owned Chat relay commands.
2	//!
3	//! The managed control plane supplies only opaque tenant/thread/turn bindings
4	//! and an exact non-secret provider route. Provider credentials and local
5	//! paths stay inside the Runtime. Every Chat thread is a dedicated
6	//! [`RuntimeThreadManager`] thread with an empty model-visible tool allowlist;
7	//! the active interactive TUI thread is never reused.
8
9	use std::{
10	collections::{BTreeMap, HashSet},
11	fs::{self, File},
12	path::{Path, PathBuf},
13	sync::Arc,
14	time::{Duration, Instant},
15	};
16
17	use anyhow::{Context, Result, bail};
18	use parking_lot::Mutex;
19	use serde::{Deserialize, Serialize};
20	use serde_json::{Value, json};
21	use sha2::{Digest, Sha256};
22
23	use crate::{
24	config::{Config, MemoryBackend, MemoryConfig, SkillsConfig},
25	plugins::PluginRegistry,
26	runtime_threads::{
27	CreateThreadRequest, RuntimeEventRecord, RuntimeThreadManager, RuntimeThreadManagerConfig,
28	RuntimeTurnStatus, StartTurnRequest,
29	},
30	};
31
32	#[cfg(test)]
33	use crate::config::ContextConfig;
34
35	const STATE_SCHEMA_VERSION: u32 = 2;
36	const MAX_RELAY_ID_BYTES: usize = 240;
37	const MAX_OPERATION_KEY_BYTES: usize = 128;
38	const STATE_FILE: &str = "runtime-chat-bindings.json";
39	const SCOPE_LOCK_FILE: &str = "runtime-chat.owner.lock";
40	const SAFE_CHAT_SYSTEM_PROMPT: &str = "You are Codewhale Chat. Answer the user's request directly and conversationally. This is an isolated chat-only session: no local project, workspace, memory, skill, account, credential, path, or runtime context is available or implied. Do not claim to inspect or change local files, run tools, or perform work execution.";
41
42	#[cfg(test)]
43	static TEST_STATE_PERSIST_FAILURES: std::sync::Mutex<Vec<(PathBuf, usize)>> =
44	std::sync::Mutex::new(Vec::new());
45
46	#[cfg(test)]
47	fn inject_state_persist_failures(path: &Path, count: usize) {
48	assert!(count > 0);
49	TEST_STATE_PERSIST_FAILURES
50	.lock()
51	.unwrap_or_else(std::sync::PoisonError::into_inner)
52	.push((path.to_path_buf(), count));
53	}
54
55	#[cfg(test)]
56	fn take_state_persist_failure(path: &Path) -> bool {
57	let mut failures = TEST_STATE_PERSIST_FAILURES
58	.lock()
59	.unwrap_or_else(std::sync::PoisonError::into_inner);
60	let Some(index) = failures.iter().position(\|(target, _)\| target == path) else {
61	return false;
62	};
63	if failures[index].1 > 1 {
64	failures[index].1 -= 1;
65	} else {
66	failures.remove(index);
67	}
68	true
69	}
70
71	#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
72	#[serde(rename_all = "camelCase", deny_unknown_fields)]
73	pub(crate) struct RuntimeChatPrompt {
74	#[serde(default, skip_serializing_if = "Option::is_none")]
75	pub max_output_tokens: Option<std::num::NonZeroU32>,
76	#[serde(rename = "type")]
77	pub command_type: String,
78	pub run_id: String,
79	pub turn_id: String,
80	pub operation_key: String,
81	pub runtime_binding_id: String,
82	pub runtime_thread_id: String,
83	pub prompt: String,
84	#[serde(default, skip_serializing_if = "Vec::is_empty")]
85	pub images: Vec<codewhale_protocol::runtime::RuntimeImageInput>,
86	#[serde(default, skip_serializing_if = "Option::is_none")]
87	pub system_prompt: Option<String>,
88	pub model: String,
89	pub model_provider: String,
90	pub model_provider_id: String,
91	#[serde(default, skip_serializing_if = "Option::is_none")]
92	pub reasoning_effort: Option<String>,
93	pub allowed_tools: Vec<String>,
94	pub mode: String,
95	pub requested_mode: String,
96	pub workspace: RuntimeChatWorkspace,
97	}
98
99	#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
100	#[serde(rename_all = "camelCase", deny_unknown_fields)]
101	pub(crate) struct RuntimeChatWorkspace {
102	pub id: String,
103	pub target_ref: String,
104	}
105
106	#[derive(Debug, Clone, PartialEq, Eq)]
107	pub(crate) struct RuntimeChatControlScope {
108	pub(crate) runtime_binding_id: String,
109	pub(crate) runtime_thread_id: String,
110	}
111
112	#[derive(Debug, Clone)]
113	pub(crate) struct RuntimeChatProjection {
114	pub run_id: String,
115	pub native_thread_id: String,
116	pub native_seq: u64,
117	pub source_event_id: String,
118	pub virtual_thread_id: String,
119	pub virtual_turn_id: String,
120	pub event: &'static str,
121	pub timestamp: String,
122	pub payload: Value,
123	}
124
125	#[derive(Clone)]
126	pub(crate) struct RuntimeChatRelayHost {
127	manager: Arc<RuntimeThreadManager>,
128	config: Arc<Config>,
129	state: Arc<Mutex<RelayState>>,
130	state_path: Arc<PathBuf>,
131	target_ref: Arc<String>,
132	session_id: Arc<String>,
133	_scope_lock: Arc<RelayScopeLock>,
134	apply_lock: Arc<tokio::sync::Mutex<()>>,
135	inference_ownership: Arc<Mutex<Option<crate::client::RuntimeChatInferenceOwnership>>>,
136	claimed_projections: Arc<Mutex<HashSet<(String, u64)>>>,
137	authorized_run_id: Arc<Mutex<Option<String>>>,
138	}
139
140	#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
141	#[serde(rename_all = "camelCase", deny_unknown_fields)]
142	struct RelayState {
143	schema_version: u32,
144	#[serde(default, skip_serializing_if = "Option::is_none")]
145	owner_scope_fingerprint: Option<String>,
146	#[serde(default)]
147	bindings: Vec<RelayThreadBinding>,
148	}
149
150	impl Default for RelayState {
151	fn default() -> Self {
152	Self {
153	schema_version: STATE_SCHEMA_VERSION,
154	owner_scope_fingerprint: None,
155	bindings: Vec::new(),
156	}
157	}
158	}
159
160	#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
161	#[serde(rename_all = "camelCase", deny_unknown_fields)]
162	struct RelayThreadBinding {
163	run_id: String,
164	runtime_binding_id: String,
165	virtual_thread_id: String,
166	native_thread_id: String,
167	model: String,
168	model_provider: String,
169	model_provider_id: String,
170	first_operation_fingerprint: String,
171	system_prompt_fingerprint: Option<String>,
172	#[serde(default)]
173	turns: BTreeMap<String, RelayTurnBinding>,
174	#[serde(default)]
175	projected_native_seq: u64,
176	}
177
178	#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
179	#[serde(rename_all = "camelCase", deny_unknown_fields)]
180	struct RelayTurnBinding {
181	native_turn_id: String,
182	operation_fingerprint: String,
183	request_fingerprint: String,
184	#[serde(default)]
185	terminal_projected: bool,
186	/// True only when the deterministic reservation was durably written but
187	/// native start was proven to have rejected before accepting provider work.
188	/// This distinguishes a retryable reservation from an already-projected
189	/// terminal turn, whose exact replay must remain settled.
190	#[serde(default)]
191	start_rejected: bool,
192	}
193
194	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
195	enum TurnReservationDisposition {
196	New,
197	Reopened,
198	ExistingUnsettled,
199	ExistingTerminal,
200	}
201
202	impl RelayState {
203	fn validate(&self) -> Result<()> {
204	if self.schema_version != STATE_SCHEMA_VERSION {
205	bail!("Runtime Chat binding state uses an unsupported schema");
206	}
207	if let Some(owner) = self.owner_scope_fingerprint.as_deref() {
208	validate_fingerprint(owner)?;
209	} else if !self.bindings.is_empty() {
210	bail!("Runtime Chat binding state has no account owner");
211	}
212	let mut binding_ids = HashSet::new();
213	let mut virtual_threads = HashSet::new();
214	let mut native_threads = HashSet::new();
215	for binding in &self.bindings {
216	validate_relay_id(&binding.run_id, "run id")?;
217	validate_relay_id(&binding.runtime_binding_id, "binding id")?;
218	validate_virtual_thread_id(&binding.virtual_thread_id)?;
219	validate_native_record_id(&binding.native_thread_id, "native thread id")?;
220	validate_route_id(&binding.model_provider, "provider id")?;
221	validate_route_id(&binding.model_provider_id, "model-provider id")?;
222	validate_model_id(&binding.model)?;
223	validate_fingerprint(&binding.first_operation_fingerprint)?;
224	if let Some(fingerprint) = binding.system_prompt_fingerprint.as_deref() {
225	validate_fingerprint(fingerprint)?;
226	}
227	if !binding_ids.insert(binding.runtime_binding_id.clone())
228	\|\| !virtual_threads.insert(binding.virtual_thread_id.clone())
229	\|\| !native_threads.insert(binding.native_thread_id.clone())
230	{
231	bail!("Runtime Chat binding state contains duplicate thread authority");
232	}
233	for (virtual_turn_id, turn) in &binding.turns {
234	validate_virtual_turn_id(virtual_turn_id)?;
235	validate_native_record_id(&turn.native_turn_id, "native turn id")?;
236	validate_fingerprint(&turn.operation_fingerprint)?;
237	validate_fingerprint(&turn.request_fingerprint)?;
238	}
239	}
240	Ok(())
241	}
242	}
243
244	impl RuntimeChatRelayHost {
245	pub(crate) fn open(
246	config: Config,
247	_plugin_registry: Arc<PluginRegistry>,
248	private_root: PathBuf,
249	target_ref: String,
250	session_id: String,
251	) -> Result<Self, String> {
252	validate_owner_component(&target_ref, "target")?;
253	validate_owner_component(&session_id, "session")?;
254	let private_dir = scoped_private_dir(&private_root, &target_ref, &session_id);
255	fs::create_dir_all(&private_dir)
256	.map_err(\|_\| "Runtime Chat could not prepare private local state.".to_string())?;
257	#[cfg(unix)]
258	{
259	use std::os::unix::fs::PermissionsExt;
260	fs::set_permissions(&private_dir, fs::Permissions::from_mode(0o700))
261	.map_err(\|_\| "Runtime Chat could not protect private local state.".to_string())?;
262	}
263	let scope_lock =
264	RelayScopeLock::acquire(&private_dir.join(SCOPE_LOCK_FILE)).map_err(\|error\| {
265	// Only WouldBlock is genuine contention. Any other lock
266	// failure is a local IO fault, and misreporting it as
267	// ownership hides the cause (#5735's flake evidence).
268	let contention = error
269	.downcast_ref::<std::io::Error>()
270	.is_some_and(\|io\| io.kind() == std::io::ErrorKind::WouldBlock);
271	if contention {
272	"Another Codewhale process already owns this Runtime Chat account session."
273	.to_string()
274	} else {
275	format!("Runtime Chat could not take its owner lock: {error:#}")
276	}
277	})?;
278	let state_path = private_dir.join(STATE_FILE);
279	let state = load_state(&state_path).map_err(\|_\| {
280	"The saved Runtime Chat binding state could not be trusted.".to_string()
281	})?;
282	state.validate().map_err(\|_\| {
283	"The saved Runtime Chat binding state could not be trusted.".to_string()
284	})?;
285
286	// Reuse the native Runtime thread engine and durable records, but keep
287	// account Chat in its own private store and empty workspace. The
288	// dedicated system-prompt override below is the model-visible boundary;
289	// this workspace/config hardening also prevents local project, memory,
290	// instruction, and skill sources from becoming fallback context.
291	let (execution_config, chat_workspace) =
292	isolated_chat_execution_config(&config, &private_dir)?;
293	let relay_plugin_registry = Arc::new(PluginRegistry::empty(&chat_workspace));
294	let task_data_dir = private_dir.join("tasks");
295	let mut manager_cfg = RuntimeThreadManagerConfig::from_task_data_dir(task_data_dir);
296	manager_cfg.data_dir = private_dir.join("runtime");
297	let manager = RuntimeThreadManager::open_with_plugin_registry(
298	execution_config,
299	chat_workspace,
300	manager_cfg,
301	relay_plugin_registry,
302	)
303	.map_err(\|_\| "Runtime Chat could not open its isolated native thread store.".to_string())?;
304
305	Ok(Self {
306	manager: Arc::new(manager),
307	config: Arc::new(config),
308	state: Arc::new(Mutex::new(state)),
309	state_path: Arc::new(state_path),
310	target_ref: Arc::new(target_ref),
311	session_id: Arc::new(session_id),
312	_scope_lock: Arc::new(scope_lock),
313	apply_lock: Arc::new(tokio::sync::Mutex::new(())),
314	inference_ownership: Arc::new(Mutex::new(None)),
315	claimed_projections: Arc::new(Mutex::new(HashSet::new())),
316	authorized_run_id: Arc::new(Mutex::new(None)),
317	})
318	}
319
320	pub(crate) fn catalog(&self, challenge: &str) -> Result<Value, String> {
321	self.ensure_account_bound()?;
322	crate::runtime_api::runtime_chat_relay_catalog(&self.config, challenge)
323	}
324
325	pub(crate) fn catalog_payload_fingerprint(payload: &Value) -> Result<String, String> {
326	serde_json::to_vec(&canonical_json_value(payload))
327	.map(\|bytes\| hex_digest(Sha256::digest(bytes)))
328	.map_err(\|_\| "Runtime Chat could not fingerprint its safe catalog.".to_string())
329	}
330
331	pub(crate) fn bind_account(&self, account_ref: &str, target_ref: &str) -> Result<(), String> {
332	validate_owner_component(account_ref, "account")?;
333	if target_ref != self.target_ref.as_str() {
334	return Err("The Runtime Chat account owner does not match this target.".to_string());
335	}
336	let owner = owner_scope_fingerprint(account_ref, target_ref, &self.session_id);
337	self.persist_state_update(
338	"Runtime Chat could not persist its account ownership.",
339	\|state\| bind_owner_scope(state, &owner),
340	)
341	}
342
343	pub(crate) fn authorize_run(&self, run_id: &str) -> Result<(), String> {
344	validate_relay_id(run_id, "run id")
345	.map_err(\|_\| "The Runtime Chat attachment has an invalid run identity.".to_string())?;
346	self.ensure_account_bound()?;
347	let durable_other_run_is_unsettled = self.state.lock().bindings.iter().any(\|binding\| {
348	binding.run_id != run_id && binding.turns.values().any(\|turn\| !turn.terminal_projected)
349	});
350	if durable_other_run_is_unsettled {
351	return Err(
352	"Finish or interrupt the active Runtime Chat turn before attaching another run."
353	.to_string(),
354	);
355	}
356	if self.has_any_unsettled_turns() && self.inference_ownership.lock().is_none() {
357	let ownership = crate::client::try_acquire_runtime_chat_inference_ownership()
358	.ok_or_else(\|\| {
359	"Finish the active local turn before recovering Runtime Chat.".to_string()
360	})?;
361	*self.inference_ownership.lock() = Some(ownership);
362	}
363	*self.authorized_run_id.lock() = Some(run_id.to_string());
364	Ok(())
365	}
366
367	#[cfg(test)]
368	pub(crate) fn has_unsettled_authorized_turns(&self) -> bool {
369	self.authorized_run_id
370	.lock()
371	.as_deref()
372	.is_some_and(\|run_id\| self.has_unsettled_turns_for_run(run_id))
373	}
374
375	pub(crate) fn has_any_unsettled_turns(&self) -> bool {
376	self.state
377	.lock()
378	.bindings
379	.iter()
380	.any(\|binding\| binding.turns.values().any(\|turn\| !turn.terminal_projected))
381	}
382
383	#[cfg(test)]
384	async fn ensure_inference_ownership(&self) {
385	if self.inference_ownership.lock().is_some() {
386	return;
387	}
388	let ownership = crate::client::acquire_runtime_chat_inference_ownership().await;
389	let mut current = self.inference_ownership.lock();
390	if current.is_none() {
391	*current = Some(ownership);
392	}
393	}
394
395	fn try_ensure_inference_ownership(&self) -> Result<(), String> {
396	if self.inference_ownership.lock().is_some() {
397	return Ok(());
398	}
399	let ownership =
400	crate::client::try_acquire_runtime_chat_inference_ownership().ok_or_else(\|\| {
401	"Finish the active local provider work before starting Runtime Chat.".to_string()
402	})?;
403	let mut current = self.inference_ownership.lock();
404	if current.is_none() {
405	*current = Some(ownership);
406	}
407	Ok(())
408	}
409
410	pub(crate) fn recover_inference_ownership_for_pending_delivery(&self) -> Result<(), String> {
411	if self.inference_ownership.lock().is_some() {
412	return Ok(());
413	}
414	let ownership =
415	crate::client::try_acquire_runtime_chat_inference_ownership().ok_or_else(\|\| {
416	"Finish the active local turn before recovering Runtime Chat delivery.".to_string()
417	})?;
418	*self.inference_ownership.lock() = Some(ownership);
419	Ok(())
420	}
421
422	pub(crate) fn release_inference_ownership_if_settled(&self) {
423	if !self.has_any_unsettled_turns() {
424	self.inference_ownership.lock().take();
425	}
426	}
427
428	pub(crate) fn is_exact_prompt_replay(
429	&self,
430	command: &RuntimeChatPrompt,
431	) -> Result<bool, String> {
432	command.validate_shape()?;
433	let operation_fingerprint = fingerprint(&command.operation_key);
434	let request_fingerprint = runtime_chat_request_fingerprint(command)?;
435	let state = self.state.lock();
436	let by_binding = state
437	.bindings
438	.iter()
439	.position(\|binding\| binding.runtime_binding_id == command.runtime_binding_id);
440	let by_thread = state
441	.bindings
442	.iter()
443	.position(\|binding\| binding.virtual_thread_id == command.runtime_thread_id);
444	let binding = match (by_binding, by_thread) {
445	(None, None) => return Ok(false),
446	(Some(binding_index), Some(thread_index)) if binding_index == thread_index => {
447	&state.bindings[binding_index]
448	}
449	_ => {
450	return Err(
451	"The Runtime Chat thread authority does not match its binding.".to_string(),
452	);
453	}
454	};
455	if binding.run_id != command.run_id {
456	return Err("The Runtime Chat replay belongs to another run.".to_string());
457	}
458	let Some(turn) = binding.turns.get(&command.turn_id) else {
459	return Ok(false);
460	};
461	if turn.operation_fingerprint == operation_fingerprint
462	&& turn.request_fingerprint == request_fingerprint
463	{
464	return Ok(true);
465	}
466	Err("The Runtime Chat turn binding does not match its replay.".to_string())
467	}
468
469	pub(crate) fn scope_matches(&self, target_ref: &str, session_id: &str) -> bool {
470	self.target_ref.as_str() == target_ref && self.session_id.as_str() == session_id
471	}
472
473	#[cfg(test)]
474	pub(crate) fn configured_default_model_for_tests(&self) -> String {
475	self.config.default_model()
476	}
477
478	#[cfg(test)]
479	pub(crate) fn inference_ownership_is_held_for_tests(&self) -> bool {
480	self.inference_ownership.lock().is_some()
481	}
482
483	#[cfg(test)]
484	pub(crate) async fn acquire_inference_ownership_for_tests(&self) {
485	self.ensure_inference_ownership().await;
486	}
487
488	#[cfg(test)]
489	pub(crate) fn install_unsettled_turn_for_tests(
490	&self,
491	run_id: &str,
492	native_thread_id: &str,
493	virtual_thread_id: &str,
494	virtual_turn_id: &str,
495	) -> Result<(), String> {
496	self.insert_binding(RelayThreadBinding {
497	run_id: run_id.to_string(),
498	runtime_binding_id: "binding_test_fixture".to_string(),
499	virtual_thread_id: virtual_thread_id.to_string(),
500	native_thread_id: native_thread_id.to_string(),
501	model: "model-1".to_string(),
502	model_provider: "custom".to_string(),
503	model_provider_id: "local-provider".to_string(),
504	first_operation_fingerprint: fingerprint("operation-test-fixture"),
505	system_prompt_fingerprint: None,
506	turns: BTreeMap::from([(
507	virtual_turn_id.to_string(),
508	RelayTurnBinding {
509	native_turn_id: "turn_test_fixture".to_string(),
510	operation_fingerprint: fingerprint("operation-test-fixture"),
511	request_fingerprint: fingerprint("request-test-fixture"),
512	terminal_projected: false,
513	start_rejected: false,
514	},
515	)]),
516	projected_native_seq: 0,
517	})
518	}
519
520	#[cfg(test)]
521	pub(crate) async fn install_prompt_replay_for_tests(
522	&self,
523	command: &RuntimeChatPrompt,
524	terminal_projected: bool,
525	) -> Result<(), String> {
526	command.validate_shape()?;
527	let operation_fingerprint = fingerprint(&command.operation_key);
528	let request_fingerprint = runtime_chat_request_fingerprint(command)?;
529	let native_thread_id = self
530	.manager
531	.create_thread(CreateThreadRequest {
532	model: Some(command.model.clone()),
533	model_provider: Some(command.model_provider.clone()),
534	model_provider_id: Some(command.model_provider_id.clone()),
535	reasoning_effort: command.reasoning_effort.clone(),
536	allowed_tools: Some(Vec::new()),
537	workspace: None,
538	mode: Some("agent".to_string()),
539	permission_posture: Some("ask".to_string()),
540	allow_shell: Some(false),
541	trust_mode: Some(false),
542	auto_approve: Some(false),
543	archived: false,
544	system_prompt: Some(dedicated_chat_system_prompt(
545	command.system_prompt.as_deref(),
546	)),
547	task_id: None,
548	dynamic_tools: Vec::new(),
549	environments: Vec::new(),
550	})
551	.await
552	.map_err(\|_\| "Runtime Chat could not create its replay fixture.".to_string())?
553	.id;
554	let native_turn_id = reserved_native_turn_id(
555	&native_thread_id,
556	&command.runtime_binding_id,
557	&command.turn_id,
558	&operation_fingerprint,
559	);
560	self.insert_binding(RelayThreadBinding {
561	run_id: command.run_id.clone(),
562	runtime_binding_id: command.runtime_binding_id.clone(),
563	virtual_thread_id: command.runtime_thread_id.clone(),
564	native_thread_id,
565	model: command.model.clone(),
566	model_provider: command.model_provider.clone(),
567	model_provider_id: command.model_provider_id.clone(),
568	first_operation_fingerprint: operation_fingerprint.clone(),
569	system_prompt_fingerprint: command.system_prompt.as_deref().map(fingerprint),
570	turns: BTreeMap::from([(
571	command.turn_id.clone(),
572	RelayTurnBinding {
573	native_turn_id,
574	operation_fingerprint,
575	request_fingerprint,
576	terminal_projected,
577	start_rejected: false,
578	},
579	)]),
580	projected_native_seq: if terminal_projected { 1 } else { 0 },
581	})
582	}
583
584	fn has_unsettled_turns_for_run(&self, run_id: &str) -> bool {
585	self.state.lock().bindings.iter().any(\|binding\| {
586	binding.run_id == run_id && binding.turns.values().any(\|turn\| !turn.terminal_projected)
587	})
588	}
589
590	pub(crate) async fn apply_prompt(&self, command: &RuntimeChatPrompt) -> Result<(), String> {
591	self.ensure_account_bound()?;
592	if self.authorized_run_id.lock().as_deref() != Some(command.run_id.as_str()) {
593	return Err("The Runtime Chat command is not authorized for this run.".to_string());
594	}
595	let _apply = self.apply_lock.lock().await;
596	command.validate_shape()?;
597	let operation_fingerprint = fingerprint(&command.operation_key);
598	let request_fingerprint = runtime_chat_request_fingerprint(command)?;
599	let system_prompt_fingerprint = command.system_prompt.as_deref().map(fingerprint);
600
601	let existing = {
602	let state = self.state.lock();
603	let by_binding = state
604	.bindings
605	.iter()
606	.position(\|binding\| binding.runtime_binding_id == command.runtime_binding_id);
607	let by_thread = state
608	.bindings
609	.iter()
610	.position(\|binding\| binding.virtual_thread_id == command.runtime_thread_id);
611	match (by_binding, by_thread) {
612	(None, None) => None,
613	(Some(binding_index), Some(thread_index)) if binding_index == thread_index => {
614	Some(state.bindings[binding_index].clone())
615	}
616	_ => {
617	return Err(
618	"The Runtime Chat thread authority does not match its binding.".to_string(),
619	);
620	}
621	}
622	};
623	let exact_operation_replay = existing.as_ref().is_some_and(\|binding\| {
624	binding.turns.get(&command.turn_id).is_some_and(\|turn\| {
625	turn.operation_fingerprint == operation_fingerprint
626	&& turn.request_fingerprint == request_fingerprint
627	})
628	});
629	if !exact_operation_replay {
630	self.validate_route(command)?;
631	}
632	if !exact_operation_replay && self.has_unsettled_turns_for_run(&command.run_id) {
633	return Err(
634	"Finish or interrupt the active Runtime Chat turn before starting another."
635	.to_string(),
636	);
637	}
638
639	let binding = if let Some(binding) = existing {
640	validate_existing_binding(
641	&binding,
642	command,
643	&operation_fingerprint,
644	system_prompt_fingerprint.as_deref(),
645	)?;
646	self.manager
647	.get_thread(&binding.native_thread_id)
648	.await
649	.map_err(\|_\| "The isolated Runtime Chat thread is unavailable.".to_string())?;
650	binding
651	} else {
652	let thread = self
653	.manager
654	.create_thread(CreateThreadRequest {
655	model: Some(command.model.clone()),
656	model_provider: Some(command.model_provider.clone()),
657	model_provider_id: Some(command.model_provider_id.clone()),
658	reasoning_effort: command.reasoning_effort.clone(),
659	allowed_tools: Some(Vec::new()),
660	workspace: None,
661	// The native engine's existing Act loop executes a pure
662	// chat turn once its model-visible tool catalog is empty.
663	// `chat` remains the account wire mode, not a second loop.
664	mode: Some("agent".to_string()),
665	permission_posture: Some("ask".to_string()),
666	allow_shell: Some(false),
667	trust_mode: Some(false),
668	auto_approve: Some(false),
669	archived: false,
670	system_prompt: Some(dedicated_chat_system_prompt(
671	command.system_prompt.as_deref(),
672	)),
673	task_id: None,
674	dynamic_tools: Vec::new(),
675	environments: Vec::new(),
676	})
677	.await
678	.map_err(\|_\| {
679	"Runtime Chat could not create an isolated native thread.".to_string()
680	})?;
681	let binding = RelayThreadBinding {
682	runtime_binding_id: command.runtime_binding_id.clone(),
683	run_id: command.run_id.clone(),
684	virtual_thread_id: command.runtime_thread_id.clone(),
685	native_thread_id: thread.id,
686	model: command.model.clone(),
687	model_provider: command.model_provider.clone(),
688	model_provider_id: command.model_provider_id.clone(),
689	first_operation_fingerprint: operation_fingerprint.clone(),
690	system_prompt_fingerprint,
691	turns: BTreeMap::new(),
692	projected_native_seq: 0,
693	};
694	if let Err(error) = self.insert_binding(binding.clone()) {
695	let _ = self
696	.manager
697	.discard_empty_thread(&binding.native_thread_id)
698	.await;
699	return Err(error);
700	}
701	binding
702	};
703
704	// Reject virtual-turn and operation-key drift before asking the native
705	// manager to start anything. RuntimeThreadManager independently
706	// enforces the same operation-key idempotency for crash replay.
707	if let Some(existing_turn) = binding.turns.get(&command.turn_id)
708	&& (existing_turn.operation_fingerprint != operation_fingerprint
709	\|\| existing_turn.request_fingerprint != request_fingerprint)
710	{
711	return Err("The Runtime Chat turn binding does not match its replay.".to_string());
712	}
713	if binding.turns.iter().any(\|(virtual_turn_id, turn)\| {
714	virtual_turn_id != &command.turn_id
715	&& turn.operation_fingerprint == operation_fingerprint
716	}) {
717	return Err("The Runtime Chat operation key belongs to another turn.".to_string());
718	}
719
720	// Preallocate and persist the exact native id before the manager may
721	// submit anything to a provider. RuntimeThreadManager binds the same id
722	// inside its operation-key transaction, eliminating the crash window
723	// between native acceptance and relay correlation.
724	let reserved_native_turn_id = reserved_native_turn_id(
725	&binding.native_thread_id,
726	&command.runtime_binding_id,
727	&command.turn_id,
728	&operation_fingerprint,
729	);
730	let reservation = self.reserve_turn(
731	&command.runtime_binding_id,
732	&command.runtime_thread_id,
733	&command.turn_id,
734	&reserved_native_turn_id,
735	&operation_fingerprint,
736	&request_fingerprint,
737	)?;
738	if matches!(reservation, TurnReservationDisposition::ExistingTerminal) {
739	return Ok(());
740	}
741
742	// The exclusive provider-request lease is acquired only after all
743	// command/route/binding validation and deterministic reservation are
744	// durable, but before native Runtime can resolve Auto or dispatch any
745	// provider request. At the common client seam it also blocks advisor,
746	// detached-subagent, compaction, purge, and interactive requests that
747	// could otherwise feed this attached CWC run.
748	if let Err(error) = self.try_ensure_inference_ownership() {
749	if !matches!(reservation, TurnReservationDisposition::ExistingUnsettled) {
750	self.finish_turn_reservation(
751	&command.runtime_binding_id,
752	&command.runtime_thread_id,
753	&command.turn_id,
754	)?;
755	}
756	self.release_inference_ownership_if_settled();
757	return Err(error);
758	}
759
760	let turn = match self
761	.manager
762	.start_turn_with_reserved_id(
763	&binding.native_thread_id,
764	StartTurnRequest {
765	max_output_tokens: command.max_output_tokens,
766	prompt: command.prompt.clone(),
767	images: command.images.clone(),
768	operation_key: Some(command.operation_key.clone()),
769	input_summary: None,
770	model: Some(command.model.clone()),
771	reasoning_effort: command.reasoning_effort.clone(),
772	allowed_tools: Some(Vec::new()),
773	mode: Some("agent".to_string()),
774	permission_posture: Some("ask".to_string()),
775	allow_shell: Some(false),
776	trust_mode: Some(false),
777	auto_approve: Some(false),
778	dynamic_tools: Vec::new(),
779	environment_id: None,
780	},
781	&reserved_native_turn_id,
782	)
783	.await
784	{
785	Ok(turn) => turn,
786	Err(error) => {
787	// A process may crash after the relay reservation is durable
788	// but before Runtime persists its operation binding/turn. On
789	// exact replay that appears as ExistingUnsettled. Settle it
790	// only when the native thread snapshot positively proves the
791	// deterministic turn id was never accepted; any unreadable or
792	// present native record stays fail-closed and projectable.
793	let native_turn_is_durable = self
794	.manager
795	.get_thread_detail(&binding.native_thread_id)
796	.await
797	.map(\|detail\| {
798	detail
799	.turns
800	.iter()
801	.any(\|turn\| turn.id == reserved_native_turn_id)
802	})
803	.unwrap_or(true);
804	if !native_turn_is_durable {
805	self.finish_turn_reservation(
806	&command.runtime_binding_id,
807	&command.runtime_thread_id,
808	&command.turn_id,
809	)?;
810	}
811	self.release_inference_ownership_if_settled();
812	return Err(sanitized_runtime_error("start", &error));
813	}
814	};
815	debug_assert_eq!(turn.id, reserved_native_turn_id);
816	Ok(())
817	}
818
819	pub(crate) async fn interrupt(
820	&self,
821	run_id: &str,
822	scope: &RuntimeChatControlScope,
823	virtual_turn_id: &str,
824	) -> Result<(), String> {
825	self.ensure_account_bound()?;
826	validate_relay_id(run_id, "run id")
827	.map_err(\|_\| "The Runtime Chat interrupt run is invalid.".to_string())?;
828	validate_relay_id(&scope.runtime_binding_id, "binding id")
829	.map_err(\|_\| "The Runtime Chat interrupt binding is invalid.".to_string())?;
830	validate_virtual_thread_id(&scope.runtime_thread_id)
831	.map_err(\|_\| "The Runtime Chat interrupt thread is invalid.".to_string())?;
832	validate_virtual_turn_id(virtual_turn_id)
833	.map_err(\|_\| "The Runtime Chat interrupt turn is invalid.".to_string())?;
834	let (native_thread_id, native_turn_id) = {
835	let state = self.state.lock();
836	resolve_interrupt_target(&state, run_id, scope, virtual_turn_id)?
837	};
838
839	let detail = self
840	.manager
841	.get_thread_detail(&native_thread_id)
842	.await
843	.map_err(\|_\| "The isolated Runtime Chat thread is unavailable.".to_string())?;
844	let turn = detail
845	.turns
846	.iter()
847	.find(\|turn\| turn.id == native_turn_id)
848	.ok_or_else(\|\| "The isolated Runtime Chat turn is unavailable.".to_string())?;
849	if !matches!(
850	turn.status,
851	RuntimeTurnStatus::Queued \| RuntimeTurnStatus::InProgress
852	) {
853	// Exact replay after the terminal boundary is already applied.
854	return Ok(());
855	}
856	self.manager
857	.interrupt_turn(&native_thread_id, &native_turn_id)
858	.await
859	.map(\|_\| ())
860	.map_err(\|error\| sanitized_runtime_error("interrupt", &error))
861	}
862
863	/// Return at most one not-yet-journaled projection per poll. The caller
864	/// journals it before advancing `projected_native_seq`, so a crash can
865	/// cause a safe replay but never silently lose an accepted native event.
866	pub(crate) async fn pending_projections(&self) -> Result<Vec<RuntimeChatProjection>, String> {
867	let Some(authorized_run_id) = self.authorized_run_id.lock().clone() else {
868	return Ok(Vec::new());
869	};
870	let bindings = {
871	let state = self.state.lock();
872	if state.owner_scope_fingerprint.is_none() {
873	return Ok(Vec::new());
874	}
875	state
876	.bindings
877	.iter()
878	.filter(\|binding\| binding.run_id == authorized_run_id)
879	.cloned()
880	.collect::<Vec<_>>()
881	};
882	for binding in bindings {
883	let events = self
884	.manager
885	.events_since_async(
886	&binding.native_thread_id,
887	Some(binding.projected_native_seq),
888	)
889	.await
890	.map_err(\|_\| "Runtime Chat could not read its native event ledger.".to_string())?;
891	for event in events {
892	let key = (binding.native_thread_id.clone(), event.seq);
893	if self.claimed_projections.lock().contains(&key) {
894	break;
895	}
896	match project_native_event(&binding, &event) {
897	ProjectionDecision::Project(projection) => {
898	self.claimed_projections.lock().insert(key);
899	// Return immediately after acquiring the claim. No
900	// later binding read can fail and accidentally drop a
901	// previously acquired in-process claim.
902	return Ok(vec![*projection]);
903	}
904	ProjectionDecision::Skip => {
905	self.advance_projection_cursor(&binding.native_thread_id, event.seq, None)?;
906	}
907	ProjectionDecision::WaitForTurnBinding => break,
908	}
909	}
910	}
911	Ok(Vec::new())
912	}
913
914	pub(crate) fn mark_projected(
915	&self,
916	native_thread_id: &str,
917	native_seq: u64,
918	virtual_turn_id: &str,
919	event: &str,
920	) -> Result<(), String> {
921	self.claimed_projections
922	.lock()
923	.remove(&(native_thread_id.to_string(), native_seq));
924	self.advance_projection_cursor(
925	native_thread_id,
926	native_seq,
927	(event == "turn.completed").then_some(virtual_turn_id),
928	)
929	}
930
931	pub(crate) fn release_projection(&self, native_thread_id: &str, native_seq: u64) {
932	self.claimed_projections
933	.lock()
934	.remove(&(native_thread_id.to_string(), native_seq));
935	}
936
937	pub(crate) fn release_all_projection_claims(&self) {
938	self.claimed_projections.lock().clear();
939	}
940
941	#[cfg(test)]
942	pub(crate) fn install_projection_claim_for_tests(
943	&self,
944	native_thread_id: &str,
945	native_seq: u64,
946	) {
947	self.claimed_projections
948	.lock()
949	.insert((native_thread_id.to_string(), native_seq));
950	}
951
952	#[cfg(test)]
953	pub(crate) fn projection_is_claimed_for_tests(
954	&self,
955	native_thread_id: &str,
956	native_seq: u64,
957	) -> bool {
958	self.claimed_projections
959	.lock()
960	.contains(&(native_thread_id.to_string(), native_seq))
961	}
962
963	#[cfg(test)]
964	async fn native_turn_count_for_binding_for_tests(&self, runtime_binding_id: &str) -> usize {
965	let native_thread_id = self
966	.state
967	.lock()
968	.bindings
969	.iter()
970	.find(\|binding\| binding.runtime_binding_id == runtime_binding_id)
971	.map(\|binding\| binding.native_thread_id.clone())
972	.expect("test binding exists");
973	self.manager
974	.get_thread_detail(&native_thread_id)
975	.await
976	.expect("test native thread detail")
977	.turns
978	.len()
979	}
980
981	fn validate_route(&self, command: &RuntimeChatPrompt) -> Result<(), String> {
982	let catalog = self.catalog("a2345678901234567890123456789012")?;
983	let provider = catalog
984	.get("providers")
985	.and_then(Value::as_array)
986	.and_then(\|providers\| providers.first())
987	.ok_or_else(\|\| "The active Runtime Chat route is unavailable.".to_string())?;
988	let route_matches = provider.get("id").and_then(Value::as_str)
989	== Some(command.model_provider.as_str())
990	&& provider.get("modelProviderId").and_then(Value::as_str)
991	== Some(command.model_provider_id.as_str())
992	&& provider
993	.get("models")
994	.and_then(Value::as_array)
995	.is_some_and(\|models\| {
996	models.iter().any(\|model\| {
997	model.get("id").and_then(Value::as_str) == Some(command.model.as_str())
998	&& (command.images.is_empty()
999	\|\| model.get("imageInput").and_then(Value::as_str)
1000	== Some("supported"))
1001	})
1002	});
1003	if command.max_output_tokens.is_some()
1004	&& !provider
1005	.get("models")
1006	.and_then(Value::as_array)
1007	.is_some_and(\|models\| {
1008	models.iter().any(\|model\| {
1009	model.get("id").and_then(Value::as_str) == Some(command.model.as_str())
1010	&& model.get("outputTokenLimit").and_then(Value::as_str)
1011	== Some("supported")
1012	})
1013	})
1014	{
1015	return Err(
1016	"The selected Runtime Chat route does not support maxOutputTokens.".to_string(),
1017	);
1018	}
1019	if !route_matches {
1020	return Err(
1021	"The requested Runtime Chat route is not the active ready route.".to_string(),
1022	);
1023	}
1024	Ok(())
1025	}
1026
1027	fn insert_binding(&self, binding: RelayThreadBinding) -> Result<(), String> {
1028	self.persist_state_update(
1029	"Runtime Chat could not persist its private thread binding.",
1030	\|state\| {
1031	if state.owner_scope_fingerprint.is_none() {
1032	return Err("The Runtime Chat account owner is not established.".to_string());
1033	}
1034	if state.bindings.iter().any(\|existing\| {
1035	existing.runtime_binding_id == binding.runtime_binding_id
1036	\|\| existing.virtual_thread_id == binding.virtual_thread_id
1037	}) {
1038	return Err(
1039	"The Runtime Chat thread binding changed while it was being created."
1040	.to_string(),
1041	);
1042	}
1043	state.bindings.push(binding);
1044	Ok(())
1045	},
1046	)
1047	}
1048
1049	fn reserve_turn(
1050	&self,
1051	runtime_binding_id: &str,
1052	virtual_thread_id: &str,
1053	virtual_turn_id: &str,
1054	native_turn_id: &str,
1055	operation_fingerprint: &str,
1056	request_fingerprint: &str,
1057	) -> Result<TurnReservationDisposition, String> {
1058	validate_native_record_id(native_turn_id, "native turn id")
1059	.map_err(public_validation_error)?;
1060	validate_fingerprint(request_fingerprint).map_err(public_validation_error)?;
1061	self.persist_state_update(
1062	"Runtime Chat could not persist its private turn reservation.",
1063	\|state\| {
1064	let binding = state
1065	.bindings
1066	.iter_mut()
1067	.find(\|binding\| {
1068	binding.runtime_binding_id == runtime_binding_id
1069	&& binding.virtual_thread_id == virtual_thread_id
1070	})
1071	.ok_or_else(\|\| "The Runtime Chat thread binding disappeared.".to_string())?;
1072	if let Some(existing) = binding.turns.get_mut(virtual_turn_id) {
1073	if existing.operation_fingerprint == operation_fingerprint
1074	&& existing.request_fingerprint == request_fingerprint
1075	&& existing.native_turn_id == native_turn_id
1076	{
1077	// A retry after a proven pre-dispatch failure reopens
1078	// the same deterministic reservation. Any accepted or
1079	// replayed native turn is therefore unsettled again
1080	// until its terminal event is durably projected.
1081	let disposition = if existing.terminal_projected && existing.start_rejected
1082	{
1083	TurnReservationDisposition::Reopened
1084	} else if existing.terminal_projected {
1085	TurnReservationDisposition::ExistingTerminal
1086	} else {
1087	TurnReservationDisposition::ExistingUnsettled
1088	};
1089	if matches!(disposition, TurnReservationDisposition::Reopened) {
1090	existing.terminal_projected = false;
1091	existing.start_rejected = false;
1092	}
1093	return Ok(disposition);
1094	}
1095	return Err(
1096	"The Runtime Chat turn binding does not match its replay.".to_string()
1097	);
1098	}
1099	if binding
1100	.turns
1101	.values()
1102	.any(\|turn\| turn.operation_fingerprint == operation_fingerprint)
1103	{
1104	return Err(
1105	"The Runtime Chat operation key belongs to another turn.".to_string()
1106	);
1107	}
1108	binding.turns.insert(
1109	virtual_turn_id.to_string(),
1110	RelayTurnBinding {
1111	native_turn_id: native_turn_id.to_string(),
1112	operation_fingerprint: operation_fingerprint.to_string(),
1113	request_fingerprint: request_fingerprint.to_string(),
1114	terminal_projected: false,
1115	start_rejected: false,
1116	},
1117	);
1118	Ok(TurnReservationDisposition::New)
1119	},
1120	)
1121	}
1122
1123	fn finish_turn_reservation(
1124	&self,
1125	runtime_binding_id: &str,
1126	virtual_thread_id: &str,
1127	virtual_turn_id: &str,
1128	) -> Result<(), String> {
1129	self.persist_state_update(
1130	"Runtime Chat could not settle its rejected turn reservation.",
1131	\|state\| {
1132	let turn = state
1133	.bindings
1134	.iter_mut()
1135	.find(\|binding\| {
1136	binding.runtime_binding_id == runtime_binding_id
1137	&& binding.virtual_thread_id == virtual_thread_id
1138	})
1139	.and_then(\|binding\| binding.turns.get_mut(virtual_turn_id))
1140	.ok_or_else(\|\| "The Runtime Chat turn reservation disappeared.".to_string())?;
1141	turn.terminal_projected = true;
1142	turn.start_rejected = true;
1143	Ok(())
1144	},
1145	)
1146	}
1147
1148	fn advance_projection_cursor(
1149	&self,
1150	native_thread_id: &str,
1151	native_seq: u64,
1152	terminal_virtual_turn_id: Option<&str>,
1153	) -> Result<(), String> {
1154	self.persist_state_update(
1155	"Runtime Chat could not persist its event cursor.",
1156	\|state\| {
1157	let binding = state
1158	.bindings
1159	.iter_mut()
1160	.find(\|binding\| binding.native_thread_id == native_thread_id)
1161	.ok_or_else(\|\| "The Runtime Chat projection binding is unknown.".to_string())?;
1162	binding.projected_native_seq = binding.projected_native_seq.max(native_seq);
1163	if let Some(virtual_turn_id) = terminal_virtual_turn_id {
1164	let turn = binding.turns.get_mut(virtual_turn_id).ok_or_else(\|\| {
1165	"The Runtime Chat terminal projection has no turn binding.".to_string()
1166	})?;
1167	turn.terminal_projected = true;
1168	turn.start_rejected = false;
1169	}
1170	Ok(())
1171	},
1172	)
1173	}
1174
1175	fn persist_state_update<T>(
1176	&self,
1177	persistence_error: &'static str,
1178	update: impl FnOnce(&mut RelayState) -> Result<T, String>,
1179	) -> Result<T, String> {
1180	let mut current = self.state.lock();
1181	let mut candidate = current.clone();
1182	let output = update(&mut candidate)?;
1183	persist_state(&self.state_path, &candidate).map_err(\|_\| persistence_error.to_string())?;
1184	*current = candidate;
1185	Ok(output)
1186	}
1187
1188	fn ensure_account_bound(&self) -> Result<(), String> {
1189	if self.state.lock().owner_scope_fingerprint.is_none() {
1190	return Err("The Runtime Chat account owner is not established.".to_string());
1191	}
1192	Ok(())
1193	}
1194	}
1195
1196	impl RuntimeChatPrompt {
1197	pub(crate) fn validate_shape(&self) -> Result<(), String> {
1198	crate::image_attach::prepare_runtime_images(&self.images)
1199	.map_err(\|error\| error.to_string())?;
1200	if !self.images.is_empty() && self.model.trim().eq_ignore_ascii_case("auto") {
1201	return Err(
1202	"Image inputs require an exact named model; Auto is unavailable for images."
1203	.to_string(),
1204	);
1205	}
1206	if self.command_type != "prompt.request" {
1207	return Err("Codewhale sent an unsupported Runtime Chat command.".to_string());
1208	}
1209	validate_relay_id(&self.run_id, "run id").map_err(public_validation_error)?;
1210	validate_virtual_turn_id(&self.turn_id).map_err(public_validation_error)?;
1211	validate_operation_key(&self.operation_key).map_err(public_validation_error)?;
1212	validate_relay_id(&self.runtime_binding_id, "binding id")
1213	.map_err(public_validation_error)?;
1214	validate_virtual_thread_id(&self.runtime_thread_id).map_err(public_validation_error)?;
1215	validate_model_id(&self.model).map_err(public_validation_error)?;
1216	validate_route_id(&self.model_provider, "provider id").map_err(public_validation_error)?;
1217	validate_route_id(&self.model_provider_id, "model-provider id")
1218	.map_err(public_validation_error)?;
1219	if self.prompt.trim().is_empty()
1220	\|\| self.prompt.len() > 128 * 1024
1221	\|\| self.prompt.contains('\0')
1222	{
1223	return Err("The Runtime Chat prompt is empty or oversized.".to_string());
1224	}
1225	if let Some(system_prompt) = self.system_prompt.as_deref()
1226	&& (system_prompt.trim().is_empty()
1227	\|\| system_prompt.len() > 64_000
1228	\|\| system_prompt.contains('\0'))
1229	{
1230	return Err("The Runtime Chat system instructions are invalid.".to_string());
1231	}
1232	if !self.allowed_tools.is_empty() {
1233	return Err("Runtime-backed Chat does not grant work-execution tools.".to_string());
1234	}
1235	if self.mode != "chat" \|\| self.requested_mode != "chat" {
1236	return Err("Runtime relay turns must use Chat mode.".to_string());
1237	}
1238	if let Some(reasoning) = self.reasoning_effort.as_deref()
1239	&& crate::reasoning_preference::ReasoningEffort::parse_strict(reasoning).is_err()
1240	{
1241	return Err("The Runtime Chat reasoning effort is invalid.".to_string());
1242	}
1243	validate_relay_id(&self.workspace.id, "workspace id").map_err(public_validation_error)?;
1244	validate_relay_id(&self.workspace.target_ref, "target reference")
1245	.map_err(public_validation_error)?;
1246	Ok(())
1247	}
1248	}
1249
1250	impl RuntimeChatControlScope {
1251	pub(crate) fn validate_for_turn(&self, virtual_turn_id: &str) -> Result<(), String> {
1252	validate_relay_id(&self.runtime_binding_id, "binding id")
1253	.map_err(public_validation_error)?;
1254	validate_virtual_thread_id(&self.runtime_thread_id).map_err(public_validation_error)?;
1255	validate_virtual_turn_id(virtual_turn_id).map_err(public_validation_error)
1256	}
1257	}
1258
1259	fn validate_existing_binding(
1260	binding: &RelayThreadBinding,
1261	command: &RuntimeChatPrompt,
1262	operation_fingerprint: &str,
1263	system_prompt_fingerprint: Option<&str>,
1264	) -> Result<(), String> {
1265	if binding.model != command.model
1266	\|\| binding.run_id != command.run_id
1267	\|\| binding.model_provider != command.model_provider
1268	\|\| binding.model_provider_id != command.model_provider_id
1269	{
1270	return Err("The Runtime Chat thread is pinned to a different provider route.".to_string());
1271	}
1272	if command.system_prompt.is_some()
1273	&& (binding.first_operation_fingerprint != operation_fingerprint
1274	\|\| binding.system_prompt_fingerprint.as_deref() != system_prompt_fingerprint)
1275	{
1276	return Err(
1277	"Runtime Chat system instructions are allowed only on the first turn.".to_string(),
1278	);
1279	}
1280	Ok(())
1281	}
1282
1283	fn resolve_interrupt_target(
1284	state: &RelayState,
1285	run_id: &str,
1286	scope: &RuntimeChatControlScope,
1287	virtual_turn_id: &str,
1288	) -> Result<(String, String), String> {
1289	let binding = state
1290	.bindings
1291	.iter()
1292	.find(\|binding\| {
1293	binding.runtime_binding_id == scope.runtime_binding_id
1294	&& binding.virtual_thread_id == scope.runtime_thread_id
1295	})
1296	.ok_or_else(\|\| "The Runtime Chat interrupt binding is unknown.".to_string())?;
1297	if binding.run_id != run_id {
1298	return Err("The Runtime Chat interrupt binding belongs to another run.".to_string());
1299	}
1300	let turn = binding
1301	.turns
1302	.get(virtual_turn_id)
1303	.ok_or_else(\|\| "The Runtime Chat interrupt turn is unknown.".to_string())?;
1304	Ok((
1305	binding.native_thread_id.clone(),
1306	turn.native_turn_id.clone(),
1307	))
1308	}
1309
1310	fn dedicated_chat_system_prompt(account_instructions: Option<&str>) -> String {
1311	match account_instructions
1312	.map(str::trim)
1313	.filter(\|value\| !value.is_empty())
1314	{
1315	Some(instructions) => format!(
1316	"{SAFE_CHAT_SYSTEM_PROMPT}\n\n<account_chat_instructions>\n{instructions}\n</account_chat_instructions>"
1317	),
1318	None => SAFE_CHAT_SYSTEM_PROMPT.to_string(),
1319	}
1320	}
1321
1322	enum ProjectionDecision {
1323	Project(Box<RuntimeChatProjection>),
1324	Skip,
1325	WaitForTurnBinding,
1326	}
1327
1328	fn project_native_event(
1329	binding: &RelayThreadBinding,
1330	event: &RuntimeEventRecord,
1331	) -> ProjectionDecision {
1332	let Some(native_turn_id) = event.turn_id.as_deref() else {
1333	return ProjectionDecision::Skip;
1334	};
1335	let Some((virtual_turn_id, _)) = binding
1336	.turns
1337	.iter()
1338	.find(\|(_, turn)\| turn.native_turn_id == native_turn_id)
1339	else {
1340	return ProjectionDecision::WaitForTurnBinding;
1341	};
1342	let base = \|\| RuntimeChatProjection {
1343	run_id: binding.run_id.clone(),
1344	native_thread_id: binding.native_thread_id.clone(),
1345	native_seq: event.seq,
1346	source_event_id: source_event_id(&binding.native_thread_id, event.seq),
1347	virtual_thread_id: binding.virtual_thread_id.clone(),
1348	virtual_turn_id: virtual_turn_id.clone(),
1349	event: "",
1350	timestamp: event.timestamp.to_rfc3339(),
1351	payload: Value::Null,
1352	};
1353	match event.event.as_str() {
1354	"turn.started" => {
1355	let mut projection = base();
1356	projection.event = "turn.started";
1357	projection.payload = json!({
1358	"turn": {
1359	"model": event.payload.pointer("/turn/effective_model")
1360	.and_then(Value::as_str)
1361	.unwrap_or(binding.model.as_str()),
1362	"mode": "chat",
1363	},
1364	});
1365	ProjectionDecision::Project(Box::new(projection))
1366	}
1367	"item.delta"
1368	if event.payload.get("kind").and_then(Value::as_str) == Some("agent_message") =>
1369	{
1370	let Some(delta) = event.payload.get("delta").and_then(Value::as_str) else {
1371	return ProjectionDecision::Skip;
1372	};
1373	let mut projection = base();
1374	projection.event = "item.delta";
1375	projection.payload = json!({ "kind": "agent_message", "delta": delta });
1376	ProjectionDecision::Project(Box::new(projection))
1377	}
1378	"turn.completed" => {
1379	let turn = event
1380	.payload
1381	.get("turn")
1382	.cloned()
1383	.unwrap_or_else(\|\| json!({}));
1384	let status = turn
1385	.get("status")
1386	.and_then(Value::as_str)
1387	.unwrap_or("failed");
1388	let mut projection = base();
1389	projection.event = "turn.completed";
1390	projection.payload = json!({
1391	"turn": {
1392	"status": status,
1393	"usage": turn.get("usage").cloned().unwrap_or_else(\|\| json!({})),
1394	"effective_model": turn.get("effective_model").and_then(Value::as_str)
1395	.unwrap_or(binding.model.as_str()),
1396	"effective_provider": turn.get("effective_provider").and_then(Value::as_str)
1397	.unwrap_or(binding.model_provider.as_str()),
1398	"effective_billing_surface": turn.get("effective_billing_surface")
1399	.and_then(Value::as_str).unwrap_or("provider_byok"),
1400	"effective_billing_mode": turn.get("effective_billing_mode")
1401	.and_then(Value::as_str).unwrap_or("local"),
1402	},
1403	});
1404	ProjectionDecision::Project(Box::new(projection))
1405	}
1406	_ => ProjectionDecision::Skip,
1407	}
1408	}
1409
1410	fn validate_owner_component(value: &str, label: &str) -> Result<(), String> {
1411	if value.trim() != value \|\| value.is_empty() \|\| value.len() > 512 \|\| value.contains('\0') {
1412	return Err(format!("The Runtime Chat {label} owner is invalid."));
1413	}
1414	Ok(())
1415	}
1416
1417	fn scoped_private_dir(root: &Path, target_ref: &str, session_id: &str) -> PathBuf {
1418	let mut hasher = Sha256::new();
1419	hasher.update(b"codewhale.runtime-chat-scope.v1\0");
1420	hasher.update(target_ref.as_bytes());
1421	hasher.update(b"\0");
1422	hasher.update(session_id.as_bytes());
1423	root.join(format!("scope-{}", hex_digest(hasher.finalize())))
1424	}
1425
1426	fn owner_scope_fingerprint(account_ref: &str, target_ref: &str, session_id: &str) -> String {
1427	let mut hasher = Sha256::new();
1428	hasher.update(b"codewhale.runtime-chat-owner.v1\0");
1429	hasher.update(account_ref.as_bytes());
1430	hasher.update(b"\0");
1431	hasher.update(target_ref.as_bytes());
1432	hasher.update(b"\0");
1433	hasher.update(session_id.as_bytes());
1434	hex_digest(hasher.finalize())
1435	}
1436
1437	fn bind_owner_scope(state: &mut RelayState, owner: &str) -> Result<(), String> {
1438	validate_fingerprint(owner)
1439	.map_err(\|_\| "The Runtime Chat account owner is invalid.".to_string())?;
1440	match state.owner_scope_fingerprint.as_deref() {
1441	Some(existing) if existing == owner => Ok(()),
1442	Some(_) => Err("This Runtime Chat session belongs to another account.".to_string()),
1443	None if state.bindings.is_empty() => {
1444	state.owner_scope_fingerprint = Some(owner.to_string());
1445	Ok(())
1446	}
1447	None => Err("The saved Runtime Chat state has no trusted account owner.".to_string()),
1448	}
1449	}
1450
1451	fn source_event_id(native_thread_id: &str, native_seq: u64) -> String {
1452	let mut hasher = Sha256::new();
1453	hasher.update(b"codewhale.runtime-chat-source-event.v1\0");
1454	hasher.update(native_thread_id.as_bytes());
1455	hasher.update(b"\0");
1456	hasher.update(native_seq.to_be_bytes());
1457	format!("native_event_{}", hex_digest(hasher.finalize()))
1458	}
1459
1460	fn reserved_native_turn_id(
1461	native_thread_id: &str,
1462	runtime_binding_id: &str,
1463	virtual_turn_id: &str,
1464	operation_fingerprint: &str,
1465	) -> String {
1466	let mut hasher = Sha256::new();
1467	hasher.update(b"codewhale.runtime-chat-native-turn.v1\0");
1468	hasher.update(native_thread_id.as_bytes());
1469	hasher.update(b"\0");
1470	hasher.update(runtime_binding_id.as_bytes());
1471	hasher.update(b"\0");
1472	hasher.update(virtual_turn_id.as_bytes());
1473	hasher.update(b"\0");
1474	hasher.update(operation_fingerprint.as_bytes());
1475	format!("turn_{}", hex_digest(hasher.finalize()))
1476	}
1477
1478	fn hex_digest(digest: impl AsRef<[u8]>) -> String {
1479	digest
1480	.as_ref()
1481	.iter()
1482	.map(\|byte\| format!("{byte:02x}"))
1483	.collect()
1484	}
1485
1486	fn canonical_json_value(value: &Value) -> Value {
1487	match value {
1488	Value::Array(values) => Value::Array(values.iter().map(canonical_json_value).collect()),
1489	Value::Object(values) => {
1490	let mut keys = values.keys().collect::<Vec<_>>();
1491	keys.sort_unstable();
1492	let mut canonical = serde_json::Map::new();
1493	for key in keys {
1494	canonical.insert(key.clone(), canonical_json_value(&values[key]));
1495	}
1496	Value::Object(canonical)
1497	}
1498	_ => value.clone(),
1499	}
1500	}
1501
1502	fn isolated_chat_execution_config(
1503	config: &Config,
1504	private_dir: &Path,
1505	) -> Result<(Config, PathBuf), String> {
1506	let workspace = private_dir.join("chat-workspace");
1507	let context_dir = private_dir.join("prompt-context");
1508	let skills_dir = context_dir.join("skills-empty");
1509	fs::create_dir_all(&workspace)
1510	.and_then(\|()\| fs::create_dir_all(&skills_dir))
1511	.map_err(\|_\| "Runtime Chat could not prepare its isolated Chat context.".to_string())?;
1512	#[cfg(unix)]
1513	{
1514	use std::os::unix::fs::PermissionsExt;
1515	for path in [&workspace, &context_dir, &skills_dir] {
1516	fs::set_permissions(path, fs::Permissions::from_mode(0o700)).map_err(\|_\| {
1517	"Runtime Chat could not protect its isolated Chat context.".to_string()
1518	})?;
1519	}
1520	}
1521
1522	let mut execution = config.clone();
1523	execution.runtime_chat_isolated = true;
1524	execution.skills_dir = Some(skills_dir.to_string_lossy().into_owned());
1525	execution.instructions = None;
1526	execution.notes_path = Some(
1527	context_dir
1528	.join("notes-disabled")
1529	.to_string_lossy()
1530	.into_owned(),
1531	);
1532	execution.mcp_config_path = Some(
1533	context_dir
1534	.join("mcp-disabled.json")
1535	.to_string_lossy()
1536	.into_owned(),
1537	);
1538	execution.memory = Some(MemoryConfig {
1539	enabled: Some(false),
1540	backend: Some(MemoryBackend::Off),
1541	});
1542	execution.memory_path = Some(
1543	context_dir
1544	.join("memory-disabled.md")
1545	.to_string_lossy()
1546	.into_owned(),
1547	);
1548	execution.context.project_pack = Some(false);
1549	execution
1550	.skills
1551	.get_or_insert_with(SkillsConfig::default)
1552	.scan_codewhale_only = Some(true);
1553	Ok((execution, workspace))
1554	}
1555
1556	#[derive(Debug)]
1557	struct RelayScopeLock {
1558	_file: File,
1559	}
1560
1561	impl RelayScopeLock {
1562	fn acquire(path: &Path) -> Result<Self> {
1563	if fs::symlink_metadata(path).is_ok_and(\|metadata\| metadata.file_type().is_symlink()) {
1564	bail!("refusing a symlinked Runtime Chat owner lock");
1565	}
1566	let mut options = fs::OpenOptions::new();
1567	options.read(true).write(true).create(true);
1568	#[cfg(unix)]
1569	{
1570	use std::os::unix::fs::OpenOptionsExt as _;
1571	options.mode(0o600).custom_flags(libc::O_NOFOLLOW);
1572	}
1573	#[cfg(windows)]
1574	{
1575	use std::os::windows::fs::OpenOptionsExt as _;
1576	use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OPEN_REPARSE_POINT;
1577	options.custom_flags(FILE_FLAG_OPEN_REPARSE_POINT);
1578	}
1579	let file = options.open(path).context("open Runtime Chat owner lock")?;
1580	if !file
1581	.metadata()
1582	.context("inspect Runtime Chat owner lock")?
1583	.file_type()
1584	.is_file()
1585	{
1586	bail!("Runtime Chat owner lock is not a regular file");
1587	}
1588	#[cfg(unix)]
1589	{
1590	use std::os::unix::fs::PermissionsExt as _;
1591	file.set_permissions(fs::Permissions::from_mode(0o600))
1592	.context("protect Runtime Chat owner lock")?;
1593	}
1594	// Same-process drop-then-reopen can observe WouldBlock for a brief
1595	// window while the previous fd is still closing (#5735). Retry only
1596	// that contention; a lock that stays held is still ownership.
1597	let deadline = Instant::now() + Duration::from_millis(25);
1598	loop {
1599	match Self::try_lock_exclusive(&file) {
1600	Ok(()) => break,
1601	Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
1602	if Instant::now() >= deadline {
1603	return Err(error).context("acquire Runtime Chat owner lock");
1604	}
1605	std::thread::yield_now();
1606	std::thread::sleep(Duration::from_millis(1));
1607	}
1608	Err(error) => {
1609	return Err(error).context("acquire Runtime Chat owner lock");
1610	}
1611	}
1612	}
1613	Ok(Self { _file: file })
1614	}
1615
1616	fn try_lock_exclusive(file: &File) -> std::io::Result<()> {
1617	#[cfg(unix)]
1618	{
1619	use std::os::fd::AsRawFd as _;
1620	// SAFETY: `file` owns a valid descriptor for the duration of the
1621	// call and remains alive in `Self` for the full lock lifetime.
1622	if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX \| libc::LOCK_NB) } != 0 {
1623	return Err(std::io::Error::last_os_error());
1624	}
1625	Ok(())
1626	}
1627	#[cfg(windows)]
1628	{
1629	use std::os::windows::io::AsRawHandle as _;
1630	use windows_sys::Win32::Storage::FileSystem::LockFile;
1631	// SAFETY: `file` owns a valid handle that remains alive in `Self`.
1632	if unsafe { LockFile(file.as_raw_handle() as _, 0, 0, u32::MAX, u32::MAX) } == 0 {
1633	return Err(std::io::Error::last_os_error());
1634	}
1635	Ok(())
1636	}
1637	#[cfg(not(any(unix, windows)))]
1638	{
1639	let _ = file;
1640	Ok(())
1641	}
1642	}
1643	}
1644
1645	impl Drop for RelayScopeLock {
1646	fn drop(&mut self) {
1647	// close() also releases, but unlocking first lets a same-process
1648	// reopen proceed without racing the previous fd's teardown (#5735).
1649	#[cfg(unix)]
1650	{
1651	use std::os::fd::AsRawFd as _;
1652	// SAFETY: Drop runs only while `_file` still owns this descriptor.
1653	unsafe {
1654	libc::flock(self._file.as_raw_fd(), libc::LOCK_UN);
1655	}
1656	}
1657	#[cfg(windows)]
1658	{
1659	use std::os::windows::io::AsRawHandle as _;
1660	use windows_sys::Win32::Storage::FileSystem::UnlockFile;
1661	// SAFETY: Drop runs only while `_file` still owns this handle.
1662	unsafe {
1663	UnlockFile(self._file.as_raw_handle() as _, 0, 0, u32::MAX, u32::MAX);
1664	}
1665	}
1666	}
1667	}
1668
1669	fn load_state(path: &Path) -> Result<RelayState> {
1670	match fs::read(path) {
1671	Ok(bytes) => serde_json::from_slice(&bytes).context("decode Runtime Chat binding state"),
1672	Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(RelayState::default()),
1673	Err(error) => Err(error).context("read Runtime Chat binding state"),
1674	}
1675	}
1676
1677	fn persist_state(path: &Path, state: &RelayState) -> Result<()> {
1678	state.validate()?;
1679	#[cfg(test)]
1680	if take_state_persist_failure(path) {
1681	bail!("injected Runtime Chat state persistence failure");
1682	}
1683	let body = serde_json::to_vec(state).context("encode Runtime Chat binding state")?;
1684	crate::utils::write_atomic(path, &body).context("persist Runtime Chat binding state")
1685	}
1686
1687	fn validate_relay_id(value: &str, label: &str) -> Result<()> {
1688	if value.is_empty()
1689	\|\| value.len() > MAX_RELAY_ID_BYTES
1690	\|\| value.contains("..")
1691	\|\| value.contains("://")
1692	\|\| !value.bytes().all(\|byte\| {
1693	byte.is_ascii_alphanumeric()
1694	\|\| matches!(byte, b'.' \| b'_' \| b':' \| b'@' \| b'/' \| b'+' \| b'~' \| b'-')
1695	})
1696	{
1697	bail!("invalid Runtime Chat {label}");
1698	}
1699	Ok(())
1700	}
1701
1702	fn validate_operation_key(value: &str) -> Result<()> {
1703	validate_relay_id(value, "operation key")?;
1704	if value.len() > MAX_OPERATION_KEY_BYTES {
1705	bail!("Runtime Chat operation key is too long");
1706	}
1707	Ok(())
1708	}
1709
1710	fn validate_virtual_thread_id(value: &str) -> Result<()> {
1711	if value.len() != 37
1712	\|\| !value.starts_with("local_thread_")
1713	\|\| !value[13..]
1714	.bytes()
1715	.all(\|byte\| byte.is_ascii_digit() \|\| matches!(byte, b'a'..=b'f'))
1716	{
1717	bail!("invalid Runtime Chat virtual thread id");
1718	}
1719	Ok(())
1720	}
1721
1722	fn validate_virtual_turn_id(value: &str) -> Result<()> {
1723	if value.len() != 35
1724	\|\| !value.starts_with("local_turn_")
1725	\|\| !value[11..]
1726	.bytes()
1727	.all(\|byte\| byte.is_ascii_digit() \|\| matches!(byte, b'a'..=b'f'))
1728	{
1729	bail!("invalid Runtime Chat virtual turn id");
1730	}
1731	Ok(())
1732	}
1733
1734	fn validate_native_record_id(value: &str, label: &str) -> Result<()> {
1735	if value.len() < 5
1736	\|\| value.len() > 80
1737	\|\| !value
1738	.bytes()
1739	.all(\|byte\| byte.is_ascii_alphanumeric() \|\| byte == b'_')
1740	{
1741	bail!("invalid {label}");
1742	}
1743	Ok(())
1744	}
1745
1746	fn validate_route_id(value: &str, label: &str) -> Result<()> {
1747	if !crate::runtime_api::runtime_chat_route_id_is_safe(value) {
1748	bail!("invalid Runtime Chat {label}");
1749	}
1750	Ok(())
1751	}
1752
1753	fn validate_model_id(value: &str) -> Result<()> {
1754	if !crate::runtime_api::runtime_chat_model_id_is_safe(value) {
1755	bail!("invalid Runtime Chat model id");
1756	}
1757	Ok(())
1758	}
1759
1760	fn validate_fingerprint(value: &str) -> Result<()> {
1761	if value.len() != 64 \|\| !value.bytes().all(\|byte\| byte.is_ascii_hexdigit()) {
1762	bail!("invalid Runtime Chat fingerprint");
1763	}
1764	Ok(())
1765	}
1766
1767	fn fingerprint(value: &str) -> String {
1768	hex_digest(Sha256::digest(value.as_bytes()))
1769	}
1770
1771	fn runtime_chat_request_fingerprint(command: &RuntimeChatPrompt) -> Result<String, String> {
1772	serde_json::to_value(command)
1773	.map(\|value\| canonical_json_value(&value))
1774	.and_then(\|value\| serde_json::to_vec(&value))
1775	.map(\|bytes\| hex_digest(Sha256::digest(bytes)))
1776	.map_err(\|_\| "Runtime Chat could not fingerprint its validated request.".to_string())
1777	}
1778
1779	fn public_validation_error(_error: anyhow::Error) -> String {
1780	"The Runtime Chat command contains an invalid opaque identity.".to_string()
1781	}
1782
1783	fn sanitized_runtime_error(action: &str, error: &anyhow::Error) -> String {
1784	let text = error.to_string().to_ascii_lowercase();
1785	if text.contains("operation_key") \|\| text.contains("operation key") {
1786	return "The Runtime Chat operation key does not match its original turn.".to_string();
1787	}
1788	format!("The isolated Runtime Chat turn could not {action}.")
1789	}
1790
1791	#[cfg(test)]
1792	mod tests {
1793	use super::*;
1794	use chrono::Utc;
1795	use std::time::Duration;
1796
1797	fn open_host(root: &Path) -> RuntimeChatRelayHost {
1798	RuntimeChatRelayHost::open(
1799	Config::default(),
1800	Arc::new(PluginRegistry::empty(root)),
1801	root.to_path_buf(),
1802	"target_fixture".to_string(),
1803	"session_fixture".to_string(),
1804	)
1805	.expect("open Runtime Chat host")
1806	}
1807
1808	fn binding() -> RelayThreadBinding {
1809	RelayThreadBinding {
1810	run_id: "run_fixture".to_string(),
1811	runtime_binding_id: "binding_fixture".to_string(),
1812	virtual_thread_id: format!("local_thread_{}", "a".repeat(24)),
1813	native_thread_id: "thr_fixture".to_string(),
1814	model: "model-1".to_string(),
1815	model_provider: "custom".to_string(),
1816	model_provider_id: "local-provider".to_string(),
1817	first_operation_fingerprint: fingerprint("operation-1"),
1818	system_prompt_fingerprint: None,
1819	turns: BTreeMap::from([(
1820	format!("local_turn_{}", "b".repeat(24)),
1821	RelayTurnBinding {
1822	native_turn_id: "turn_native".to_string(),
1823	operation_fingerprint: fingerprint("operation-1"),
1824	request_fingerprint: fingerprint("request-1"),
1825	terminal_projected: false,
1826	start_rejected: false,
1827	},
1828	)]),
1829	projected_native_seq: 0,
1830	}
1831	}
1832
1833	#[test]
1834	fn persisted_binding_state_rejects_duplicate_or_nonopaque_authority() {
1835	let one = binding();
1836	let state = RelayState {
1837	schema_version: STATE_SCHEMA_VERSION,
1838	owner_scope_fingerprint: Some(fingerprint("owner")),
1839	bindings: vec![one.clone()],
1840	};
1841	state.validate().unwrap();
1842	let duplicate = RelayState {
1843	schema_version: STATE_SCHEMA_VERSION,
1844	owner_scope_fingerprint: Some(fingerprint("owner")),
1845	bindings: vec![one.clone(), one],
1846	};
1847	assert!(duplicate.validate().is_err());
1848	}
1849
1850	#[test]
1851	fn projection_rewrites_native_ids_to_virtual_chat_ids_and_keeps_route_receipt() {
1852	let binding = binding();
1853	let event = RuntimeEventRecord {
1854	schema_version: 2,
1855	seq: 9,
1856	timestamp: Utc::now(),
1857	thread_id: binding.native_thread_id.clone(),
1858	turn_id: Some("turn_native".to_string()),
1859	item_id: None,
1860	event: "turn.completed".to_string(),
1861	payload: json!({
1862	"turn": {
1863	"status": "completed",
1864	"usage": { "input_tokens": 2, "output_tokens": 3 },
1865	"effective_model": "model-1",
1866	"effective_provider": "custom",
1867	"effective_billing_surface": "provider_byok",
1868	"effective_billing_mode": "local",
1869	}
1870	}),
1871	};
1872	let ProjectionDecision::Project(projected) = project_native_event(&binding, &event) else {
1873	panic!("terminal event must project");
1874	};
1875	assert_eq!(projected.virtual_thread_id, binding.virtual_thread_id);
1876	assert!(projected.virtual_turn_id.starts_with("local_turn_"));
1877	assert_eq!(projected.event, "turn.completed");
1878	assert_eq!(
1879	projected.source_event_id,
1880	source_event_id(&binding.native_thread_id, event.seq)
1881	);
1882	assert_eq!(projected.source_event_id.len(), 77);
1883	assert_eq!(projected.payload["turn"]["effective_billing_mode"], "local");
1884	assert!(!projected.payload.to_string().contains("thr_fixture"));
1885	}
1886
1887	#[test]
1888	fn chat_command_shape_requires_empty_tools_and_exact_chat_modes() {
1889	let mut prompt = RuntimeChatPrompt {
1890	images: Vec::new(),
1891	max_output_tokens: None,
1892	command_type: "prompt.request".to_string(),
1893	run_id: "run_fixture".to_string(),
1894	turn_id: format!("local_turn_{}", "b".repeat(24)),
1895	operation_key: "operation-1".to_string(),
1896	runtime_binding_id: "binding_fixture".to_string(),
1897	runtime_thread_id: format!("local_thread_{}", "a".repeat(24)),
1898	prompt: "hello".to_string(),
1899	system_prompt: None,
1900	model: "model-1".to_string(),
1901	model_provider: "custom".to_string(),
1902	model_provider_id: "local-provider".to_string(),
1903	reasoning_effort: Some("high".to_string()),
1904	allowed_tools: Vec::new(),
1905	mode: "chat".to_string(),
1906	requested_mode: "chat".to_string(),
1907	workspace: RuntimeChatWorkspace {
1908	id: "workspace_fixture".to_string(),
1909	target_ref: "target_fixture".to_string(),
1910	},
1911	};
1912	prompt.validate_shape().unwrap();
1913	for reasoning in ["minimal", "ultra"] {
1914	prompt.reasoning_effort = Some(reasoning.into());
1915	prompt.validate_shape().unwrap();
1916	}
1917	prompt.reasoning_effort = Some("invented-effort".into());
1918	assert!(prompt.validate_shape().is_err());
1919	prompt.reasoning_effort = None;
1920	prompt.allowed_tools.push("bash".to_string());
1921	assert!(prompt.validate_shape().is_err());
1922	prompt.allowed_tools.clear();
1923	prompt.requested_mode = "work".to_string();
1924	assert!(prompt.validate_shape().is_err());
1925	}
1926
1927	#[tokio::test]
1928	async fn active_participant_rejects_new_and_exact_replay_without_blocking_or_starting() {
1929	let root = tempfile::tempdir().unwrap();
1930	let config = Config {
1931	provider: Some("ollama".to_string()),
1932	default_text_model: Some("relay-local:fixture".to_string()),
1933	..Config::default()
1934	};
1935	let host = RuntimeChatRelayHost::open(
1936	config,
1937	Arc::new(PluginRegistry::empty(root.path())),
1938	root.path().to_path_buf(),
1939	"target_fixture".to_string(),
1940	"session_fixture".to_string(),
1941	)
1942	.unwrap();
1943	host.bind_account("account_fixture", "target_fixture")
1944	.unwrap();
1945	host.authorize_run("run_fixture").unwrap();
1946	let prompt = RuntimeChatPrompt {
1947	images: Vec::new(),
1948	max_output_tokens: None,
1949	command_type: "prompt.request".to_string(),
1950	run_id: "run_fixture".to_string(),
1951	turn_id: format!("local_turn_{}", "e".repeat(24)),
1952	operation_key: "operation-gate-fixture".to_string(),
1953	runtime_binding_id: "binding-gate-fixture".to_string(),
1954	runtime_thread_id: format!("local_thread_{}", "f".repeat(24)),
1955	prompt: "hello".to_string(),
1956	system_prompt: None,
1957	model: "relay-local:fixture".to_string(),
1958	model_provider: "ollama".to_string(),
1959	model_provider_id: "ollama".to_string(),
1960	reasoning_effort: None,
1961	allowed_tools: Vec::new(),
1962	mode: "chat".to_string(),
1963	requested_mode: "chat".to_string(),
1964	workspace: RuntimeChatWorkspace {
1965	id: "workspace_fixture".to_string(),
1966	target_ref: "target_fixture".to_string(),
1967	},
1968	};
1969
1970	let participant = crate::client::acquire_remote_control_inference_participant().await;
1971	let direct_error = host.try_ensure_inference_ownership().unwrap_err();
1972	assert!(
1973	direct_error.contains("active local provider work"),
1974	"{direct_error}"
1975	);
1976	for attempt in 0..2 {
1977	let error = tokio::time::timeout(Duration::from_secs(1), host.apply_prompt(&prompt))
1978	.await
1979	.expect("Runtime Chat admission must not deadlock behind the UI-owned writer")
1980	.unwrap_err();
1981	assert!(error.contains("active local provider work"), "{error}");
1982	assert!(
1983	!host.has_any_unsettled_turns(),
1984	"rejected attempt {attempt} must not leave a phantom reservation"
1985	);
1986	}
1987	assert_eq!(
1988	host.native_turn_count_for_binding_for_tests(&prompt.runtime_binding_id)
1989	.await,
1990	0,
1991	"neither the new command nor its exact replay may reach provider-backed native start"
1992	);
1993	drop(participant);
1994	host.try_ensure_inference_ownership().unwrap();
1995	host.release_inference_ownership_if_settled();
1996	}
1997
1998	#[tokio::test]
1999	async fn recovered_reservation_without_native_turn_settles_after_definitive_start_rejection() {
2000	let root = tempfile::tempdir().unwrap();
2001	let host = open_host(root.path());
2002	host.bind_account("account_fixture", "target_fixture")
2003	.unwrap();
2004	host.authorize_run("run_fixture").unwrap();
2005	let provider = host.config.api_provider();
2006	let model_provider_id = host
2007	.config
2008	.active_provider_identity(provider)
2009	.unwrap()
2010	.persisted_id()
2011	.unwrap_or_else(\|\| provider.as_str())
2012	.to_string();
2013	let prompt = RuntimeChatPrompt {
2014	images: Vec::new(),
2015	max_output_tokens: None,
2016	command_type: "prompt.request".to_string(),
2017	run_id: "run_fixture".to_string(),
2018	turn_id: format!("local_turn_{}", "8".repeat(24)),
2019	operation_key: "operation-crash-window".to_string(),
2020	runtime_binding_id: "binding-crash-window".to_string(),
2021	runtime_thread_id: format!("local_thread_{}", "9".repeat(24)),
2022	prompt: "hello".to_string(),
2023	system_prompt: None,
2024	model: host.config.default_model(),
2025	model_provider: provider.as_str().to_string(),
2026	model_provider_id,
2027	reasoning_effort: None,
2028	allowed_tools: Vec::new(),
2029	mode: "chat".to_string(),
2030	requested_mode: "chat".to_string(),
2031	workspace: RuntimeChatWorkspace {
2032	id: "workspace_fixture".to_string(),
2033	target_ref: "target_fixture".to_string(),
2034	},
2035	};
2036	host.install_prompt_replay_for_tests(&prompt, false)
2037	.await
2038	.unwrap();
2039	assert!(host.has_any_unsettled_turns());
2040	assert_eq!(
2041	host.native_turn_count_for_binding_for_tests(&prompt.runtime_binding_id)
2042	.await,
2043	0
2044	);
2045
2046	let error = host.apply_prompt(&prompt).await.unwrap_err();
2047	assert!(error.contains("could not start"), "{error}");
2048	assert!(
2049	!host.has_any_unsettled_turns(),
2050	"a proven pre-native crash reservation must not hold ownership forever"
2051	);
2052	assert!(!host.inference_ownership_is_held_for_tests());
2053	assert_eq!(
2054	host.native_turn_count_for_binding_for_tests(&prompt.runtime_binding_id)
2055	.await,
2056	0
2057	);
2058	}
2059
2060	#[test]
2061	fn catalog_is_challenge_bound_active_ready_and_secret_free() {
2062	let mut config = Config {
2063	provider: Some("ollama".to_string()),
2064	default_text_model: Some(crate::config::DEFAULT_OLLAMA_MODEL.to_string()),
2065	..Config::default()
2066	};
2067	config.api_key = Some("must-not-cross".to_string());
2068	config.base_url = Some("http://127.0.0.1:11434/v1".to_string());
2069	let challenge = "c".repeat(32);
2070	assert!(crate::runtime_api::runtime_chat_relay_catalog(&config, &challenge).is_err());
2071	config.default_text_model = Some("relay-local:fixture".to_string());
2072	let catalog = crate::runtime_api::runtime_chat_relay_catalog(&config, &challenge).unwrap();
2073	assert_eq!(catalog["protocol"], "codewhale.runtime-chat-relay.v1");
2074	assert_eq!(catalog["challenge"], challenge);
2075	assert_eq!(catalog["runtime"]["service"], "codewhale-runtime-api");
2076	assert_eq!(catalog["runtime"]["apiVersion"], "1.0");
2077	assert_eq!(catalog["runtime"]["capabilities"]["stable_event_ids"], true);
2078	assert_eq!(catalog["providers"].as_array().unwrap().len(), 1);
2079	assert_eq!(
2080	catalog["providers"][0]["models"][0]["imageInput"],
2081	"unknown"
2082	);
2083	assert_eq!(
2084	catalog["runtime"]["capabilities"]["turn_image_inputs"],
2085	true
2086	);
2087	let serialized = catalog.to_string();
2088	assert!(!serialized.contains("must-not-cross"));
2089	assert!(!serialized.contains("127.0.0.1"));
2090	assert!(!serialized.contains("baseUrl"));
2091	assert!(!serialized.contains("endpoint"));
2092	}
2093
2094	#[test]
2095	fn interrupt_resolution_is_bound_to_the_current_run() {
2096	let binding = binding();
2097	let turn_id = binding.turns.keys().next().unwrap().clone();
2098	let scope = RuntimeChatControlScope {
2099	runtime_binding_id: binding.runtime_binding_id.clone(),
2100	runtime_thread_id: binding.virtual_thread_id.clone(),
2101	};
2102	let state = RelayState {
2103	schema_version: STATE_SCHEMA_VERSION,
2104	owner_scope_fingerprint: Some(fingerprint("owner")),
2105	bindings: vec![binding],
2106	};
2107	assert!(resolve_interrupt_target(&state, "run_fixture", &scope, &turn_id).is_ok());
2108	assert!(resolve_interrupt_target(&state, "run_other", &scope, &turn_id).is_err());
2109	}
2110
2111	#[test]
2112	fn scoped_state_is_exclusive_account_bound_and_restart_stable() {
2113	let root = tempfile::tempdir().unwrap();
2114	let first_path = scoped_private_dir(root.path(), "target_fixture", "session_fixture");
2115	let same_path = scoped_private_dir(root.path(), "target_fixture", "session_fixture");
2116	let other_path = scoped_private_dir(root.path(), "target_fixture", "session_other");
2117	assert_eq!(first_path, same_path);
2118	assert_ne!(first_path, other_path);
2119	assert!(!first_path.to_string_lossy().contains("target_fixture"));
2120
2121	fs::create_dir_all(&first_path).unwrap();
2122	let lock_path = first_path.join(SCOPE_LOCK_FILE);
2123	let first_lock = RelayScopeLock::acquire(&lock_path).unwrap();
2124	assert!(RelayScopeLock::acquire(&lock_path).is_err());
2125
2126	let owner = owner_scope_fingerprint("account_one", "target_fixture", "session_fixture");
2127	let mut state = RelayState::default();
2128	bind_owner_scope(&mut state, &owner).unwrap();
2129	bind_owner_scope(&mut state, &owner).unwrap();
2130	assert!(
2131	bind_owner_scope(
2132	&mut state,
2133	&owner_scope_fingerprint("account_other", "target_fixture", "session_fixture")
2134	)
2135	.is_err()
2136	);
2137	state.bindings.push(binding());
2138	let state_path = first_path.join(STATE_FILE);
2139	persist_state(&state_path, &state).unwrap();
2140	assert_eq!(load_state(&state_path).unwrap(), state);
2141	assert!(
2142	fs::read_dir(&first_path)
2143	.unwrap()
2144	.filter_map(Result::ok)
2145	.all(\|entry\| !entry.file_name().to_string_lossy().ends_with(".tmp"))
2146	);
2147
2148	drop(first_lock);
2149	RelayScopeLock::acquire(&lock_path).unwrap();
2150	}
2151
2152	#[test]
2153	fn isolated_chat_prompt_drops_local_project_memory_and_skill_context() {
2154	let root = tempfile::tempdir().unwrap();
2155	let config = Config {
2156	skills_dir: Some("/Users/alice/CANARY_SKILLS".to_string()),
2157	instructions: Some(vec!["/Users/alice/CANARY_AGENTS.md".to_string()]),
2158	memory_path: Some("/Users/alice/CANARY_MEMORY.md".to_string()),
2159	memory: Some(MemoryConfig {
2160	enabled: Some(true),
2161	backend: Some(MemoryBackend::Native),
2162	}),
2163	context: ContextConfig {
2164	project_pack: Some(true),
2165	..ContextConfig::default()
2166	},
2167	..Config::default()
2168	};
2169
2170	let (execution, workspace) = isolated_chat_execution_config(&config, root.path()).unwrap();
2171	assert!(workspace.starts_with(root.path()));
2172	assert_ne!(workspace, PathBuf::from("/Users/alice"));
2173	assert!(!execution.memory_enabled());
2174	assert!(execution.instructions_paths().is_empty());
2175	assert!(!execution.project_context_pack_enabled());
2176	assert!(execution.skills_dir().starts_with(root.path()));
2177	assert!(execution.memory_path().starts_with(root.path()));
2178	assert!(execution.mcp_config_path().starts_with(root.path()));
2179	assert!(execution.notes_path().starts_with(root.path()));
2180	assert!(execution.skills_config().scan_codewhale_only());
2181
2182	let prompt = dedicated_chat_system_prompt(None);
2183	assert_eq!(prompt, SAFE_CHAT_SYSTEM_PROMPT);
2184	for canary in [
2185	"CANARY_SKILLS",
2186	"CANARY_AGENTS",
2187	"CANARY_MEMORY",
2188	"/Users/alice",
2189	] {
2190	assert!(!prompt.contains(canary));
2191	}
2192	let account_prompt = dedicated_chat_system_prompt(Some("Reply in short paragraphs."));
2193	assert!(account_prompt.starts_with(SAFE_CHAT_SYSTEM_PROMPT));
2194	assert!(account_prompt.contains("<account_chat_instructions>"));
2195	}
2196
2197	#[test]
2198	fn unsafe_model_ids_are_rejected_and_never_cross_the_catalog() {
2199	for model in [
2200	"/Users/alice/private-model",
2201	"~/.ssh/provider-model",
2202	"sk-live-secret",
2203	"../secrets/model",
2204	"redacted-local-path",
2205	"alice:hunter2@internal.example:443/model",
2206	"internal.example:443/model",
2207	"internal:443/model",
2208	] {
2209	assert!(validate_model_id(model).is_err(), "accepted {model:?}");
2210	let config = Config {
2211	provider: Some("ollama".to_string()),
2212	default_text_model: Some(model.to_string()),
2213	..Config::default()
2214	};
2215	if let Ok(catalog) =
2216	crate::runtime_api::runtime_chat_relay_catalog(&config, &"c".repeat(32))
2217	{
2218	let ids = catalog["providers"][0]["models"]
2219	.as_array()
2220	.unwrap()
2221	.iter()
2222	.filter_map(\|entry\| entry["id"].as_str())
2223	.collect::<Vec<_>>();
2224	assert!(!ids.contains(&model), "catalog leaked {model:?}");
2225	}
2226	}
2227	validate_model_id("anthropic/claude-sonnet-5").unwrap();
2228
2229	for provider_id in ["ghp_secret-provider", "hf_secret-provider", "glpat-secret"] {
2230	assert!(!crate::runtime_api::runtime_chat_route_id_is_safe(
2231	provider_id
2232	));
2233	let config = Config {
2234	provider: Some(provider_id.to_string()),
2235	providers: Some(crate::config::ProvidersConfig {
2236	custom: std::collections::HashMap::from([(
2237	provider_id.to_string(),
2238	crate::config::ProviderConfig {
2239	kind: Some("openai-compatible".to_string()),
2240	api_key: Some("fixture-key".to_string()),
2241	base_url: Some("https://example.test/v1".to_string()),
2242	model: Some("safe-model".to_string()),
2243	..Default::default()
2244	},
2245	)]),
2246	..Default::default()
2247	}),
2248	..Config::default()
2249	};
2250	let error = crate::runtime_api::runtime_chat_relay_catalog(&config, &"c".repeat(32))
2251	.unwrap_err();
2252	assert!(!error.contains(provider_id));
2253	}
2254	}
2255
2256	#[test]
2257	fn native_source_event_id_is_stable_across_transport_replay() {
2258	let first = source_event_id("thr_fixture", 41);
2259	assert_eq!(first, source_event_id("thr_fixture", 41));
2260	assert_ne!(first, source_event_id("thr_fixture", 42));
2261	assert_ne!(first, source_event_id("thr_other", 41));
2262	assert!(first.starts_with("native_event_"));
2263	assert!(
2264	first
2265	.bytes()
2266	.all(\|byte\| byte.is_ascii_lowercase() \|\| byte.is_ascii_digit() \|\| byte == b'_')
2267	);
2268	}
2269
2270	#[test]
2271	fn failed_state_writes_never_become_in_memory_authority_and_exact_retry_reopens() {
2272	let root = tempfile::tempdir().unwrap();
2273	let host = open_host(root.path());
2274
2275	inject_state_persist_failures(&host.state_path, 1);
2276	assert!(
2277	host.bind_account("account_fixture", "target_fixture")
2278	.is_err()
2279	);
2280	assert!(host.state.lock().owner_scope_fingerprint.is_none());
2281	host.bind_account("account_fixture", "target_fixture")
2282	.unwrap();
2283
2284	let mut relay_binding = binding();
2285	relay_binding.turns.clear();
2286	inject_state_persist_failures(&host.state_path, 1);
2287	assert!(host.insert_binding(relay_binding.clone()).is_err());
2288	assert!(host.state.lock().bindings.is_empty());
2289	host.insert_binding(relay_binding.clone()).unwrap();
2290
2291	let virtual_turn_id = format!("local_turn_{}", "c".repeat(24));
2292	let operation_fingerprint = fingerprint("operation-state-fault");
2293	let request_fingerprint = fingerprint("request-state-fault");
2294	let native_turn_id = reserved_native_turn_id(
2295	&relay_binding.native_thread_id,
2296	&relay_binding.runtime_binding_id,
2297	&virtual_turn_id,
2298	&operation_fingerprint,
2299	);
2300	inject_state_persist_failures(&host.state_path, 1);
2301	assert!(
2302	host.reserve_turn(
2303	&relay_binding.runtime_binding_id,
2304	&relay_binding.virtual_thread_id,
2305	&virtual_turn_id,
2306	&native_turn_id,
2307	&operation_fingerprint,
2308	&request_fingerprint,
2309	)
2310	.is_err()
2311	);
2312	assert!(
2313	!host.state.lock().bindings[0]
2314	.turns
2315	.contains_key(&virtual_turn_id)
2316	);
2317	host.reserve_turn(
2318	&relay_binding.runtime_binding_id,
2319	&relay_binding.virtual_thread_id,
2320	&virtual_turn_id,
2321	&native_turn_id,
2322	&operation_fingerprint,
2323	&request_fingerprint,
2324	)
2325	.unwrap();
2326
2327	inject_state_persist_failures(&host.state_path, 1);
2328	assert!(
2329	host.advance_projection_cursor(&relay_binding.native_thread_id, 7, None)
2330	.is_err()
2331	);
2332	assert_eq!(host.state.lock().bindings[0].projected_native_seq, 0);
2333	host.advance_projection_cursor(&relay_binding.native_thread_id, 7, None)
2334	.unwrap();
2335
2336	drop(host);
2337	let reopened = open_host(root.path());
2338	let state = reopened.state.lock();
2339	assert!(state.owner_scope_fingerprint.is_some());
2340	assert_eq!(state.bindings.len(), 1);
2341	assert_eq!(state.bindings[0].projected_native_seq, 7);
2342	assert_eq!(
2343	state.bindings[0].turns[&virtual_turn_id].native_turn_id,
2344	native_turn_id
2345	);
2346	}
2347
2348	#[test]
2349	fn conflicting_replay_never_settles_a_live_turn_reservation() {
2350	let root = tempfile::tempdir().unwrap();
2351	let host = open_host(root.path());
2352	host.bind_account("account_fixture", "target_fixture")
2353	.unwrap();
2354	let mut relay_binding = binding();
2355	relay_binding.turns.clear();
2356	host.insert_binding(relay_binding.clone()).unwrap();
2357
2358	let virtual_turn_id = format!("local_turn_{}", "d".repeat(24));
2359	let operation_fingerprint = fingerprint("same-operation");
2360	let original_request = fingerprint("original-request");
2361	let changed_request = fingerprint("changed-request");
2362	let native_turn_id = reserved_native_turn_id(
2363	&relay_binding.native_thread_id,
2364	&relay_binding.runtime_binding_id,
2365	&virtual_turn_id,
2366	&operation_fingerprint,
2367	);
2368	assert_eq!(
2369	host.reserve_turn(
2370	&relay_binding.runtime_binding_id,
2371	&relay_binding.virtual_thread_id,
2372	&virtual_turn_id,
2373	&native_turn_id,
2374	&operation_fingerprint,
2375	&original_request,
2376	)
2377	.unwrap(),
2378	TurnReservationDisposition::New
2379	);
2380	assert!(
2381	host.reserve_turn(
2382	&relay_binding.runtime_binding_id,
2383	&relay_binding.virtual_thread_id,
2384	&virtual_turn_id,
2385	&native_turn_id,
2386	&operation_fingerprint,
2387	&changed_request,
2388	)
2389	.is_err()
2390	);
2391	assert!(host.has_any_unsettled_turns());
2392	assert!(!host.state.lock().bindings[0].turns[&virtual_turn_id].terminal_projected);
2393	}
2394
2395	#[test]
2396	fn rejected_start_retry_reopens_the_same_deterministic_reservation() {
2397	let root = tempfile::tempdir().unwrap();
2398	let host = open_host(root.path());
2399	host.bind_account("account_fixture", "target_fixture")
2400	.unwrap();
2401	let mut relay_binding = binding();
2402	relay_binding.turns.clear();
2403	host.insert_binding(relay_binding.clone()).unwrap();
2404
2405	let virtual_turn_id = format!("local_turn_{}", "e".repeat(24));
2406	let operation_fingerprint = fingerprint("retry-operation");
2407	let request_fingerprint = fingerprint("retry-request");
2408	let native_turn_id = reserved_native_turn_id(
2409	&relay_binding.native_thread_id,
2410	&relay_binding.runtime_binding_id,
2411	&virtual_turn_id,
2412	&operation_fingerprint,
2413	);
2414	assert_eq!(
2415	host.reserve_turn(
2416	&relay_binding.runtime_binding_id,
2417	&relay_binding.virtual_thread_id,
2418	&virtual_turn_id,
2419	&native_turn_id,
2420	&operation_fingerprint,
2421	&request_fingerprint,
2422	)
2423	.unwrap(),
2424	TurnReservationDisposition::New
2425	);
2426	host.finish_turn_reservation(
2427	&relay_binding.runtime_binding_id,
2428	&relay_binding.virtual_thread_id,
2429	&virtual_turn_id,
2430	)
2431	.unwrap();
2432	assert!(!host.has_any_unsettled_turns());
2433	assert_eq!(
2434	host.reserve_turn(
2435	&relay_binding.runtime_binding_id,
2436	&relay_binding.virtual_thread_id,
2437	&virtual_turn_id,
2438	&native_turn_id,
2439	&operation_fingerprint,
2440	&request_fingerprint,
2441	)
2442	.unwrap(),
2443	TurnReservationDisposition::Reopened
2444	);
2445	assert!(host.has_any_unsettled_turns());
2446	}
2447
2448	#[test]
2449	fn exact_replay_of_a_projected_terminal_turn_stays_settled() {
2450	let root = tempfile::tempdir().unwrap();
2451	let host = open_host(root.path());
2452	host.bind_account("account_fixture", "target_fixture")
2453	.unwrap();
2454	let mut relay_binding = binding();
2455	relay_binding.turns.clear();
2456	host.insert_binding(relay_binding.clone()).unwrap();
2457
2458	let virtual_turn_id = format!("local_turn_{}", "f".repeat(24));
2459	let operation_fingerprint = fingerprint("terminal-replay-operation");
2460	let request_fingerprint = fingerprint("terminal-replay-request");
2461	let native_turn_id = reserved_native_turn_id(
2462	&relay_binding.native_thread_id,
2463	&relay_binding.runtime_binding_id,
2464	&virtual_turn_id,
2465	&operation_fingerprint,
2466	);
2467	assert_eq!(
2468	host.reserve_turn(
2469	&relay_binding.runtime_binding_id,
2470	&relay_binding.virtual_thread_id,
2471	&virtual_turn_id,
2472	&native_turn_id,
2473	&operation_fingerprint,
2474	&request_fingerprint,
2475	)
2476	.unwrap(),
2477	TurnReservationDisposition::New
2478	);
2479	host.advance_projection_cursor(&relay_binding.native_thread_id, 11, Some(&virtual_turn_id))
2480	.unwrap();
2481	assert!(!host.has_any_unsettled_turns());
2482
2483	assert_eq!(
2484	host.reserve_turn(
2485	&relay_binding.runtime_binding_id,
2486	&relay_binding.virtual_thread_id,
2487	&virtual_turn_id,
2488	&native_turn_id,
2489	&operation_fingerprint,
2490	&request_fingerprint,
2491	)
2492	.unwrap(),
2493	TurnReservationDisposition::ExistingTerminal
2494	);
2495	assert!(
2496	!host.has_any_unsettled_turns(),
2497	"an idempotent replay of already-projected output cannot reopen provider work"
2498	);
2499	}
2500
2501	#[tokio::test]
2502	async fn restart_refuses_a_new_run_until_the_durable_old_turn_is_projected() {
2503	let root = tempfile::tempdir().unwrap();
2504	{
2505	let host = open_host(root.path());
2506	host.bind_account("account_fixture", "target_fixture")
2507	.unwrap();
2508	host.insert_binding(binding()).unwrap();
2509	host.authorize_run("run_fixture").unwrap();
2510	assert!(host.has_unsettled_authorized_turns());
2511	}
2512
2513	let reopened = open_host(root.path());
2514	reopened
2515	.bind_account("account_fixture", "target_fixture")
2516	.unwrap();
2517	assert!(reopened.authorize_run("run_other").is_err());
2518	reopened.authorize_run("run_fixture").unwrap();
2519	let relay_binding = reopened.state.lock().bindings[0].clone();
2520	let virtual_turn_id = relay_binding.turns.keys().next().unwrap().clone();
2521	reopened
2522	.mark_projected(
2523	&relay_binding.native_thread_id,
2524	9,
2525	&virtual_turn_id,
2526	"turn.completed",
2527	)
2528	.unwrap();
2529	assert!(!reopened.has_unsettled_authorized_turns());
2530	reopened.authorize_run("run_other").unwrap();
2531	}
2532	#[test]
2533	fn runtime_image_relay_hash_preserves_text_and_binds_order() {
2534	let legacy = json!({"type":"prompt.request","runId":"run_fixture","turnId":format!("local_turn_{}", "b".repeat(24)),"operationKey":"operation-1","runtimeBindingId":"binding_fixture","runtimeThreadId":format!("local_thread_{}", "a".repeat(24)),"prompt":"look","model":"deepseek-v4-flash-vision-exp","modelProvider":"deepseek","modelProviderId":"deepseek","allowedTools":[],"mode":"chat","requestedMode":"chat","workspace":{"id":"workspace_fixture","targetRef":"target_fixture"}});
2535	let mut command: RuntimeChatPrompt = serde_json::from_value(legacy.clone()).unwrap();
2536	command.validate_shape().unwrap();
2537	let expected = hex_digest(Sha256::digest(
2538	serde_json::to_vec(&canonical_json_value(&legacy)).unwrap(),
2539	));
2540	assert_eq!(
2541	runtime_chat_request_fingerprint(&command).unwrap(),
2542	expected
2543	);
2544	let mut with_empty = legacy;
2545	with_empty["images"] = json!([]);
2546	let empty: RuntimeChatPrompt = serde_json::from_value(with_empty).unwrap();
2547	assert_eq!(runtime_chat_request_fingerprint(&empty).unwrap(), expected);
2548	command.images = vec![
2549	crate::image_attach::tests::runtime_image_fixture(1),
2550	crate::image_attach::tests::runtime_image_fixture(2),
2551	];
2552	command.validate_shape().unwrap();
2553	let first = runtime_chat_request_fingerprint(&command).unwrap();
2554	command.images.reverse();
2555	assert_ne!(runtime_chat_request_fingerprint(&command).unwrap(), first);
2556	command.model = "auto".into();
2557	assert!(command.validate_shape().is_err());
2558	}
2559	}
2560

2560 lines RUST