remote_control.rs - CodeWhale

remote_control.rs

1	//! Account-owned remote control for the active TUI session.
2	//!
3	//! This is deliberately a typed relay, not a remote shell. The control plane
4	//! may send prompts, approval decisions, and run-control requests for the exact
5	//! enrolled target. Provider credentials, paths, environment variables, and
6	//! arbitrary command strings never cross this boundary.
7
8	use std::{
9	collections::{HashMap, HashSet},
10	path::Path,
11	time::{Duration, Instant, SystemTime, UNIX_EPOCH},
12	};
13
14	use reqwest::Url;
15	use reqwest::{Client, Method, StatusCode};
16	use serde::{Deserialize, Serialize};
17	use serde_json::{Value, json};
18	use sha2::{Digest, Sha256};
19	use tokio::sync::mpsc;
20
21	use crate::{
22	core::events::{Event as EngineEvent, TurnOutcomeStatus},
23	models::{ContentBlock, Message},
24	};
25
26	const PRODUCTION_CONTROL_PLANE: &str = "https://api.codewhale.net/";
27	const ENROLLMENT_SECRET_SLOT: &str = "cwc-remote-control-enrollment-v1";
28	const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(25);
29	const SYNC_INTERVAL: Duration = Duration::from_millis(1_200);
30	const MAX_RESPONSE_BYTES: usize = 1024 * 1024;
31	const MAX_RUNS: usize = 64;
32	const MAX_COMMANDS: usize = 128;
33	const CAPABILITIES: &[&str] = &["evidence-ledger", "fim", "git", "shell"];
34
35	#[derive(Debug, Clone, PartialEq, Eq)]
36	pub enum RemoteControlAction {
37	Start,
38	Stop,
39	}
40
41	#[derive(Debug, Clone)]
42	pub struct RemoteStart {
43	pub workspace_label: String,
44	pub target_ref: String,
45	pub session_id: String,
46	pub runtime_version: String,
47	pub runtime_commit: String,
48	}
49
50	#[derive(Debug, Clone)]
51	pub enum RemoteEvent {
52	Notice(String),
53	Connected {
54	account_ref: String,
55	runner_id: String,
56	target_ref: String,
57	},
58	Command {
59	run_id: String,
60	seq: u64,
61	command: RemoteCommand,
62	},
63	Failed(String),
64	Stopped,
65	OwnershipRestored {
66	approvals: Vec<PendingRemoteApproval>,
67	},
68	}
69
70	#[derive(Debug, Clone, PartialEq, Eq)]
71	pub enum RemoteCommand {
72	Prompt {
73	turn_id: String,
74	prompt: String,
75	},
76	Approval {
77	gate: String,
78	approved: bool,
79	},
80	Control {
81	action: RemoteControlRequest,
82	turn_id: Option<String>,
83	},
84	}
85
86	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87	pub enum RemoteControlRequest {
88	Interrupt,
89	Cancel,
90	}
91
92	#[derive(Debug, Clone)]
93	enum WorkerCommand {
94	Upload {
95	run_id: String,
96	acknowledgements: Vec<CommandAcknowledgement>,
97	envelopes: Vec<Value>,
98	},
99	Stop,
100	}
101
102	#[derive(Debug, Clone, Serialize)]
103	#[serde(rename_all = "camelCase")]
104	struct CommandAcknowledgement {
105	command_seq: u64,
106	command_type: String,
107	status: String,
108	#[serde(skip_serializing_if = "Option::is_none")]
109	turn_id: Option<String>,
110	#[serde(skip_serializing_if = "Option::is_none")]
111	error: Option<String>,
112	}
113
114	#[derive(Debug, Clone, Deserialize, Serialize)]
115	#[serde(rename_all = "camelCase", deny_unknown_fields)]
116	struct PersistedEnrollment {
117	schema_version: u64,
118	control_plane_base: String,
119	runner_enrollment_id: String,
120	account_ref: String,
121	device_id: String,
122	target_ref: String,
123	target_grant_ref: String,
124	runtime_version: String,
125	runtime_commit: String,
126	bootstrap_secret: String,
127	}
128
129	#[derive(Debug, Clone)]
130	struct LiveEnrollment {
131	persisted: PersistedEnrollment,
132	access_token: String,
133	}
134
135	#[derive(Debug, Clone)]
136	struct ActiveRelayRun {
137	run_id: String,
138	turn_id: String,
139	}
140
141	#[derive(Debug, Clone)]
142	pub struct PendingRemoteApproval {
143	pub tool_id: String,
144	pub tool_name: String,
145	pub description: String,
146	pub input: Value,
147	pub approval_key: String,
148	pub intent_summary: Option<String>,
149	}
150
151	#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
152	enum Status {
153	#[default]
154	Off,
155	Connecting,
156	Connected,
157	Stopping,
158	Failed,
159	}
160
161	/// UI-thread owner for remote-control state and typed transport channels.
162	pub struct RemoteControlController {
163	status: Status,
164	status_detail: String,
165	account_ref: Option<String>,
166	target_ref: Option<String>,
167	active_run: Option<ActiveRelayRun>,
168	event_seq: HashMap<String, u64>,
169	uploaded_snapshots: HashSet<String>,
170	pending_approvals: HashMap<String, PendingRemoteApproval>,
171	command_fingerprints: HashMap<(String, u64), String>,
172	worker_tx: Option<mpsc::UnboundedSender<WorkerCommand>>,
173	event_rx: Option<mpsc::UnboundedReceiver<RemoteEvent>>,
174	worker: Option<tokio::task::JoinHandle<()>>,
175	applying_remote_command: bool,
176	ownership_blocked_until: Option<Instant>,
177	}
178
179	impl Default for RemoteControlController {
180	fn default() -> Self {
181	Self {
182	status: Status::Off,
183	status_detail: "off".to_string(),
184	account_ref: None,
185	target_ref: None,
186	active_run: None,
187	event_seq: HashMap::new(),
188	uploaded_snapshots: HashSet::new(),
189	pending_approvals: HashMap::new(),
190	command_fingerprints: HashMap::new(),
191	worker_tx: None,
192	event_rx: None,
193	worker: None,
194	applying_remote_command: false,
195	ownership_blocked_until: None,
196	}
197	}
198	}
199
200	impl RemoteControlController {
201	pub fn start(&mut self, start: RemoteStart) -> Result<(), String> {
202	if matches!(
203	self.status,
204	Status::Connecting \| Status::Connected \| Status::Stopping
205	) {
206	return Err("Remote control is already active.".to_string());
207	}
208	if self.status == Status::Failed
209	&& self
210	.ownership_blocked_until
211	.is_some_and(\|deadline\| Instant::now() < deadline)
212	{
213	return Err(
214	"The previous remote lease may still be active; wait for ownership to return before reconnecting."
215	.to_string(),
216	);
217	}
218	if !valid_runtime_version(&start.runtime_version)
219	\|\| !valid_runtime_commit(&start.runtime_commit)
220	\|\| !valid_opaque_ref(&start.target_ref)
221	\|\| !valid_opaque_ref(&start.session_id)
222	{
223	return Err("This build or session does not have an enrollable identity.".to_string());
224	}
225	let (worker_tx, worker_rx) = mpsc::unbounded_channel();
226	let (event_tx, event_rx) = mpsc::unbounded_channel();
227	self.stop_worker();
228	self.status = Status::Connecting;
229	self.status_detail = "waiting for account authorization".to_string();
230	self.target_ref = Some(start.target_ref.clone());
231	self.worker_tx = Some(worker_tx);
232	self.event_rx = Some(event_rx);
233	self.worker = Some(tokio::spawn(async move {
234	if let Err(error) = relay_worker(start, worker_rx, event_tx.clone()).await {
235	let _ = event_tx.send(RemoteEvent::Failed(error));
236	}
237	}));
238	Ok(())
239	}
240
241	pub fn stop(&mut self) {
242	if self.status == Status::Connecting {
243	// The worker may have completed its server-side connect just before
244	// the UI consumed RemoteEvent::Connected. Aborting it cannot prove
245	// that no lease exists, so retain the ownership lock through the
246	// server expiry instead of returning local input immediately.
247	self.stop_worker();
248	self.status = Status::Failed;
249	self.status_detail =
250	"authorization cancelled; waiting for any server lease to expire safely"
251	.to_string();
252	self.ownership_blocked_until = Some(Instant::now() + Duration::from_secs(95));
253	} else if self.status == Status::Connected {
254	let queued = self
255	.worker_tx
256	.as_ref()
257	.is_some_and(\|tx\| tx.send(WorkerCommand::Stop).is_ok());
258	self.worker_tx = None;
259	if queued {
260	self.status = Status::Stopping;
261	self.status_detail = "confirming the runner is offline".to_string();
262	} else {
263	self.status = Status::Failed;
264	self.status_detail =
265	"waiting for the last server lease to expire safely".to_string();
266	self.ownership_blocked_until = Some(Instant::now() + Duration::from_secs(95));
267	}
268	}
269	if self.status == Status::Off {
270	self.account_ref = None;
271	self.active_run = None;
272	self.pending_approvals.clear();
273	self.command_fingerprints.clear();
274	self.ownership_blocked_until = None;
275	}
276	}
277
278	fn stop_worker(&mut self) {
279	if let Some(worker) = self.worker.take() {
280	worker.abort();
281	}
282	self.worker_tx = None;
283	self.event_rx = None;
284	}
285
286	pub fn try_next_event(&mut self) -> Option<RemoteEvent> {
287	if self.status == Status::Failed
288	&& self
289	.ownership_blocked_until
290	.is_some_and(\|deadline\| Instant::now() >= deadline)
291	{
292	let approvals = self
293	.pending_approvals
294	.drain()
295	.map(\|(_, value)\| value)
296	.collect();
297	self.stop_worker();
298	self.status = Status::Off;
299	self.status_detail = "off".to_string();
300	self.ownership_blocked_until = None;
301	return Some(RemoteEvent::OwnershipRestored { approvals });
302	}
303	let event = self.event_rx.as_mut()?.try_recv().ok()?;
304	match &event {
305	RemoteEvent::Connected {
306	account_ref,
307	target_ref,
308	..
309	} => {
310	self.status = Status::Connected;
311	self.status_detail = "web owns prompts and approvals".to_string();
312	self.ownership_blocked_until = None;
313	self.account_ref = Some(account_ref.clone());
314	self.target_ref = Some(target_ref.clone());
315	}
316	RemoteEvent::Failed(reason) => {
317	self.status = Status::Failed;
318	self.status_detail =
319	format!("{reason}; waiting for the last server lease to expire safely");
320	self.ownership_blocked_until = Some(Instant::now() + Duration::from_secs(95));
321	self.active_run = None;
322	// The worker may have failed after the UI queued a snapshot but before the
323	// control plane durably accepted it. A later attachment must be allowed to
324	// send a fresh bounded snapshot for that run.
325	self.uploaded_snapshots.clear();
326	}
327	RemoteEvent::Stopped => {
328	self.status = Status::Off;
329	self.status_detail = "off".to_string();
330	self.active_run = None;
331	self.ownership_blocked_until = None;
332	if !self.pending_approvals.is_empty() {
333	let approvals = self
334	.pending_approvals
335	.drain()
336	.map(\|(_, value)\| value)
337	.collect();
338	return Some(RemoteEvent::OwnershipRestored { approvals });
339	}
340	}
341	RemoteEvent::Notice(_)
342	\| RemoteEvent::Command { .. }
343	\| RemoteEvent::OwnershipRestored { .. } => {}
344	}
345	Some(event)
346	}
347
348	pub fn status_line(&self) -> String {
349	match self.status {
350	Status::Off => "Remote control: off".to_string(),
351	Status::Connecting => format!("Remote control: connecting · {}", self.status_detail),
352	Status::Connected => format!(
353	"Remote control: connected · account {} · {}",
354	self.account_ref.as_deref().unwrap_or("account"),
355	self.status_detail
356	),
357	Status::Stopping => {
358	"Remote control: stopping · confirming the runner is offline".to_string()
359	}
360	Status::Failed => format!("Remote control: disconnected · {}", self.status_detail),
361	}
362	}
363
364	pub fn blocks_local_input(&self) -> bool {
365	let server_may_still_own = match self.status {
366	Status::Connecting \| Status::Connected \| Status::Stopping => true,
367	Status::Failed => self
368	.ownership_blocked_until
369	.is_some_and(\|deadline\| Instant::now() < deadline),
370	Status::Off => false,
371	};
372	server_may_still_own && !self.applying_remote_command
373	}
374
375	pub fn set_applying_remote_command(&mut self, value: bool) {
376	self.applying_remote_command = value;
377	}
378
379	pub fn claim_command(
380	&mut self,
381	run_id: &str,
382	seq: u64,
383	command: &RemoteCommand,
384	) -> Result<bool, String> {
385	let fingerprint = command_fingerprint(command);
386	let key = (run_id.to_string(), seq);
387	if let Some(existing) = self.command_fingerprints.get(&key) {
388	if existing == &fingerprint {
389	return Ok(false);
390	}
391	return Err(
392	"The control plane reused a command sequence with different content.".to_string(),
393	);
394	}
395	self.command_fingerprints.insert(key, fingerprint);
396	Ok(true)
397	}
398
399	pub fn activate_prompt(&mut self, run_id: &str, turn_id: &str) {
400	self.active_run = Some(ActiveRelayRun {
401	run_id: run_id.to_string(),
402	turn_id: turn_id.to_string(),
403	});
404	}
405
406	pub fn active_run_matches(&self, run_id: &str) -> bool {
407	self.active_run
408	.as_ref()
409	.is_some_and(\|active\| active.run_id == run_id)
410	}
411
412	pub fn upload_snapshot(&mut self, run_id: &str, messages: &[Message]) {
413	if !self.uploaded_snapshots.insert(run_id.to_string()) {
414	return;
415	}
416	let projected = project_session_messages(messages);
417	self.upload_envelope(
418	run_id,
419	"session.snapshot",
420	None,
421	json!({ "messages": projected }),
422	);
423	}
424
425	pub fn acknowledge(
426	&self,
427	run_id: &str,
428	seq: u64,
429	command: &RemoteCommand,
430	status: &str,
431	error: Option<String>,
432	) {
433	let Some(tx) = &self.worker_tx else {
434	return;
435	};
436	let _ = tx.send(WorkerCommand::Upload {
437	run_id: run_id.to_string(),
438	acknowledgements: vec![CommandAcknowledgement {
439	command_seq: seq,
440	command_type: command.kind().to_string(),
441	status: status.to_string(),
442	turn_id: command.turn_id().map(ToString::to_string),
443	error: error.map(\|value\| value.chars().take(800).collect()),
444	}],
445	envelopes: Vec::new(),
446	});
447	}
448
449	pub fn record_remote_approval(
450	&mut self,
451	tool_id: &str,
452	tool_name: &str,
453	description: &str,
454	input: &Value,
455	approval_key: &str,
456	intent_summary: Option<&str>,
457	) -> String {
458	let gate = projected_approval_id(tool_id);
459	self.pending_approvals.insert(
460	gate.clone(),
461	PendingRemoteApproval {
462	tool_id: tool_id.to_string(),
463	tool_name: tool_name.to_string(),
464	description: description.to_string(),
465	input: input.clone(),
466	approval_key: approval_key.to_string(),
467	intent_summary: intent_summary.map(ToString::to_string),
468	},
469	);
470	if let Some(active) = self.active_run.clone() {
471	self.upload_envelope(
472	&active.run_id,
473	"approval.required",
474	Some(&active.turn_id),
475	json!({
476	"id": gate,
477	"approval_id": gate,
478	"tool_name": tool_name,
479	"description": description,
480	}),
481	);
482	}
483	gate
484	}
485
486	pub fn take_pending_approval(&mut self, gate: &str) -> Option<String> {
487	self.pending_approvals
488	.remove(gate)
489	.map(\|approval\| approval.tool_id)
490	}
491
492	pub fn observe_engine_event(&mut self, event: &EngineEvent) {
493	let Some(active) = self.active_run.clone() else {
494	return;
495	};
496	match event {
497	EngineEvent::MessageDelta { content, .. } => self.upload_envelope(
498	&active.run_id,
499	"item.delta",
500	Some(&active.turn_id),
501	json!({ "kind": "agent_message", "delta": content }),
502	),
503	EngineEvent::ToolCallStarted { id, name, .. } => self.upload_envelope(
504	&active.run_id,
505	"item.started",
506	Some(&active.turn_id),
507	json!({ "tool": { "id": id, "name": name, "input": {} } }),
508	),
509	EngineEvent::ToolCallComplete { id, result, .. } => {
510	let (event_name, status) = if result.is_ok() {
511	("item.completed", "completed")
512	} else {
513	("item.failed", "failed")
514	};
515	self.upload_envelope(
516	&active.run_id,
517	event_name,
518	Some(&active.turn_id),
519	json!({
520	"item": {
521	"id": id,
522	"kind": "tool_call",
523	"status": status,
524	"summary": "",
525	"detail": "",
526	}
527	}),
528	);
529	}
530	EngineEvent::TurnStarted { turn_id, route, .. } => {
531	self.active_run = Some(ActiveRelayRun {
532	run_id: active.run_id.clone(),
533	turn_id: turn_id.clone(),
534	});
535	self.upload_envelope(
536	&active.run_id,
537	"turn.started",
538	Some(turn_id),
539	json!({
540	"turn": {
541	"model": route.as_ref().map(\|value\| value.model.as_str()).unwrap_or(""),
542	"mode": "",
543	}
544	}),
545	);
546	}
547	EngineEvent::TurnComplete { usage, status, .. } => {
548	let status = match status {
549	TurnOutcomeStatus::Completed => "completed",
550	TurnOutcomeStatus::Interrupted => "interrupted",
551	TurnOutcomeStatus::Failed => "failed",
552	};
553	self.upload_envelope(
554	&active.run_id,
555	"turn.completed",
556	Some(&active.turn_id),
557	json!({ "turn": { "status": status, "usage": usage } }),
558	);
559	self.active_run = None;
560	}
561	_ => {}
562	}
563	}
564
565	fn upload_envelope(
566	&mut self,
567	run_id: &str,
568	event: &str,
569	turn_id: Option<&str>,
570	payload: Value,
571	) {
572	let seq = self.event_seq.entry(run_id.to_string()).or_insert(0);
573	*seq = seq.saturating_add(1);
574	let Some(tx) = &self.worker_tx else {
575	return;
576	};
577	let _ = tx.send(WorkerCommand::Upload {
578	run_id: run_id.to_string(),
579	acknowledgements: Vec::new(),
580	envelopes: vec![json!({
581	"schema_version": 1,
582	"seq": *seq,
583	"event": event,
584	"kind": event,
585	"turn_id": turn_id,
586	"timestamp": chrono::Utc::now().to_rfc3339(),
587	"payload": payload,
588	})],
589	});
590	}
591	}
592
593	impl Drop for RemoteControlController {
594	fn drop(&mut self) {
595	self.stop_worker();
596	}
597	}
598
599	impl RemoteCommand {
600	fn kind(&self) -> &'static str {
601	match self {
602	Self::Prompt { .. } => "prompt.request",
603	Self::Approval { .. } => "approval.decision",
604	Self::Control { .. } => "run.control",
605	}
606	}
607
608	fn turn_id(&self) -> Option<&str> {
609	match self {
610	Self::Prompt { turn_id, .. } => Some(turn_id),
611	Self::Control { turn_id, .. } => turn_id.as_deref(),
612	Self::Approval { .. } => None,
613	}
614	}
615	}
616
617	pub fn target_ref(workspace: &Path, session_id: &str) -> String {
618	let mut hasher = Sha256::new();
619	hasher.update(workspace.to_string_lossy().as_bytes());
620	hasher.update(b"\0");
621	hasher.update(session_id.as_bytes());
622	format!("target_{}", &bytes_to_hex(&hasher.finalize())[..32])
623	}
624
625	fn project_session_messages(messages: &[Message]) -> Vec<Value> {
626	messages
627	.iter()
628	.filter_map(\|message\| {
629	let role = match message.role.as_str() {
630	"user" => "user",
631	"assistant" => "assistant",
632	_ => return None,
633	};
634	let text = message
635	.content
636	.iter()
637	.filter_map(\|block\| match block {
638	ContentBlock::Text { text, .. } => Some(text.as_str()),
639	_ => None,
640	})
641	.collect::<Vec<_>>()
642	.join("\n");
643	(!text.trim().is_empty()).then(\|\| {
644	json!({
645	"role": role,
646	"text": text.chars().take(16 * 1024).collect::<String>(),
647	})
648	})
649	})
650	.rev()
651	.take(64)
652	.collect::<Vec<_>>()
653	.into_iter()
654	.rev()
655	.collect()
656	}
657
658	fn projected_approval_id(raw: &str) -> String {
659	let mut hasher = Sha256::new();
660	hasher.update(b"local-runtime:approval\0");
661	hasher.update(raw.as_bytes());
662	format!("local_approval_{}", &bytes_to_hex(&hasher.finalize())[..24])
663	}
664
665	fn command_fingerprint(command: &RemoteCommand) -> String {
666	let canonical = format!("{command:?}");
667	bytes_to_hex(&Sha256::digest(canonical.as_bytes()))
668	}
669
670	fn bytes_to_hex(bytes: &[u8]) -> String {
671	bytes.iter().map(\|byte\| format!("{byte:02x}")).collect()
672	}
673
674	async fn relay_worker(
675	start: RemoteStart,
676	mut worker_rx: mpsc::UnboundedReceiver<WorkerCommand>,
677	event_tx: mpsc::UnboundedSender<RemoteEvent>,
678	) -> Result<(), String> {
679	let base = runner_control_plane_base()?;
680	let client = Client::builder()
681	.https_only(!cfg!(debug_assertions))
682	.redirect(reqwest::redirect::Policy::none())
683	.timeout(Duration::from_secs(20))
684	.build()
685	.map_err(\|_\| "Remote control could not initialize secure networking.".to_string())?;
686
687	let mut enrollment = match load_persisted_enrollment()? {
688	Some(saved) if saved.matches(&start, &base) => {
689	match refresh_enrollment(&client, saved).await {
690	Ok(enrollment) => enrollment,
691	Err(error) if error == "runner_enrollment_revoked" => {
692	delete_persisted_enrollment();
693	enroll_device(&client, &base, &start, &event_tx).await?
694	}
695	Err(error) => return Err(error),
696	}
697	}
698	Some(_) => {
699	delete_persisted_enrollment();
700	enroll_device(&client, &base, &start, &event_tx).await?
701	}
702	None => enroll_device(&client, &base, &start, &event_tx).await?,
703	};
704
705	let mut runner_id = connect_runner(&client, &enrollment, &start).await?;
706	let _ = event_tx.send(RemoteEvent::Connected {
707	account_ref: enrollment.persisted.account_ref.clone(),
708	runner_id: runner_id.clone(),
709	target_ref: start.target_ref.clone(),
710	});
711	let mut last_heartbeat = Instant::now() - HEARTBEAT_INTERVAL;
712	let mut command_cursor: HashMap<String, u64> = HashMap::new();
713	let mut delivered: HashMap<(String, u64), String> = HashMap::new();
714
715	loop {
716	tokio::select! {
717	command = worker_rx.recv() => {
718	match command {
719	Some(WorkerCommand::Upload { run_id, acknowledgements, envelopes }) => {
720	runner_request(
721	&client,
722	&enrollment,
723	Method::POST,
724	&["api", "local-runners", &runner_id, "runs", &run_id, "events"],
725	&[],
726	Some(json!({ "acknowledgements": acknowledgements, "envelopes": envelopes })),
727	).await?;
728	}
729	Some(WorkerCommand::Stop) \| None => {
730	// Do not return local input until the control plane has
731	// durably released this lease. If the confirmation
732	// cannot be delivered, the UI keeps ownership locked
733	// through the server-side lease expiry instead.
734	post_heartbeat(&client, &enrollment, &runner_id, &start, "offline").await?;
735	let _ = event_tx.send(RemoteEvent::Stopped);
736	return Ok(());
737	}
738	}
739	}
740	() = tokio::time::sleep(SYNC_INTERVAL) => {
741	if enrollment_needs_refresh(&enrollment) {
742	enrollment = refresh_enrollment(&client, enrollment.persisted.clone()).await?;
743	runner_id = connect_runner(&client, &enrollment, &start).await?;
744	}
745	if last_heartbeat.elapsed() >= HEARTBEAT_INTERVAL {
746	post_heartbeat(&client, &enrollment, &runner_id, &start, "active").await?;
747	last_heartbeat = Instant::now();
748	}
749	let runs = list_runs(&client, &enrollment, &runner_id).await?;
750	for run_id in runs {
751	let since = command_cursor.get(&run_id).copied().unwrap_or(0);
752	for listed in list_commands(&client, &enrollment, &runner_id, &run_id, since).await? {
753	let seq = listed.seq;
754	if !listed.ack_status.is_empty() {
755	if listed.ack_status == "accepted" {
756	recover_run(
757	&client,
758	&enrollment,
759	&runner_id,
760	&run_id,
761	"accepted command has no terminal acknowledgement after runner restart",
762	).await?;
763	}
764	command_cursor.insert(run_id.clone(), seq);
765	continue;
766	}
767	let command = parse_remote_command(&listed.command, &run_id)?;
768	let fingerprint = command_fingerprint(&command);
769	let key = (run_id.clone(), seq);
770	if let Some(existing) = delivered.get(&key) {
771	if existing != &fingerprint {
772	return Err("The control plane replayed a changed command sequence.".to_string());
773	}
774	} else {
775	delivered.insert(key, fingerprint);
776	upload_command_accepted(
777	&client,
778	&enrollment,
779	&runner_id,
780	&run_id,
781	seq,
782	&command,
783	).await?;
784	event_tx.send(RemoteEvent::Command {
785	run_id: run_id.clone(),
786	seq,
787	command,
788	}).map_err(\|_\| "The terminal remote-control owner stopped.".to_string())?;
789	}
790	command_cursor.insert(run_id.clone(), seq.max(since));
791	}
792	}
793	}
794	}
795	}
796	}
797
798	impl PersistedEnrollment {
799	fn matches(&self, start: &RemoteStart, base: &str) -> bool {
800	self.schema_version == 1
801	&& self.control_plane_base == base
802	&& self.target_ref == start.target_ref
803	&& self.runtime_version == start.runtime_version
804	&& self.runtime_commit == start.runtime_commit
805	&& valid_opaque_ref(&self.runner_enrollment_id)
806	&& valid_opaque_ref(&self.account_ref)
807	&& valid_opaque_ref(&self.device_id)
808	&& valid_opaque_ref(&self.target_grant_ref)
809	&& valid_secret(&self.bootstrap_secret)
810	}
811	}
812
813	async fn enroll_device(
814	client: &Client,
815	base: &str,
816	start: &RemoteStart,
817	event_tx: &mpsc::UnboundedSender<RemoteEvent>,
818	) -> Result<LiveEnrollment, String> {
819	let device_id = format!("device_{}", uuid::Uuid::new_v4().simple());
820	let value = public_request(
821	client,
822	Method::POST,
823	control_plane_url(base, &["api", "runner", "device", "start"], &[])?,
824	json!({
825	"deviceId": device_id,
826	"deviceLabel": "Codewhale terminal",
827	"targetRef": start.target_ref,
828	"targetLabel": start.workspace_label,
829	"runtimeVersion": start.runtime_version,
830	"runtimeCommit": start.runtime_commit,
831	"capabilities": CAPABILITIES,
832	}),
833	)
834	.await?;
835	let device_code = secret_field(&value, "deviceCode")?;
836	let user_code = string_field(&value, "userCode")?;
837	let verification_uri = string_field(&value, "verificationUriComplete")?;
838	let interval = value
839	.get("interval")
840	.and_then(Value::as_u64)
841	.filter(\|value\| (1..=30).contains(value))
842	.ok_or_else(\|\| {
843	"Codewhale returned an invalid device authorization interval.".to_string()
844	})?;
845	let expires_in = value
846	.get("expiresIn")
847	.and_then(Value::as_u64)
848	.filter(\|value\| (60..=1800).contains(value))
849	.ok_or_else(\|\| "Codewhale returned an invalid device authorization expiry.".to_string())?;
850	validate_authorization_url(&verification_uri, &user_code)?;
851	let _ = event_tx.send(RemoteEvent::Notice(format!(
852	"Authorize this terminal at {verification_uri} (code {user_code})."
853	)));
854	let _ = webbrowser::open(&verification_uri);
855	let deadline = Instant::now() + Duration::from_secs(expires_in);
856	loop {
857	if Instant::now() >= deadline {
858	return Err("Remote-control authorization expired; run /rc again.".to_string());
859	}
860	tokio::time::sleep(Duration::from_secs(interval)).await;
861	let response = client
862	.post(control_plane_url(
863	base,
864	&["api", "runner", "device", "token"],
865	&[],
866	)?)
867	.json(&json!({ "deviceCode": device_code }))
868	.send()
869	.await
870	.map_err(\|_\| "Remote-control authorization could not reach Codewhale.".to_string())?;
871	if response.status() == StatusCode::ACCEPTED {
872	continue;
873	}
874	if !response.status().is_success() {
875	return Err("Remote-control authorization was rejected.".to_string());
876	}
877	let exchange = read_bounded_json(response).await?;
878	let enrollment = enrollment_from_exchange(exchange, base, &device_id, start)?;
879	save_persisted_enrollment(&enrollment.persisted)?;
880	return Ok(enrollment);
881	}
882	}
883
884	fn enrollment_from_exchange(
885	value: Value,
886	base: &str,
887	device_id: &str,
888	start: &RemoteStart,
889	) -> Result<LiveEnrollment, String> {
890	if value.get("status").and_then(Value::as_str) != Some("approved") {
891	return Err("Codewhale returned an invalid runner credential.".to_string());
892	}
893	let record = value
894	.get("enrollment")
895	.filter(\|value\| value.is_object())
896	.ok_or_else(\|\| "Codewhale returned an invalid runner credential.".to_string())?;
897	let enrollment_id = opaque_field(record, "id")?;
898	let account_ref = opaque_field(record, "userId")?;
899	let returned_device = opaque_field(record, "deviceId")?;
900	if returned_device != device_id
901	\|\| record.get("runtimeVersion").and_then(Value::as_str)
902	!= Some(start.runtime_version.as_str())
903	\|\| record.get("runtimeCommit").and_then(Value::as_str)
904	!= Some(start.runtime_commit.as_str())
905	\|\| !exact_capabilities(record.get("capabilities"))
906	{
907	return Err("The runner credential does not match this terminal.".to_string());
908	}
909	let target_grant_ref = record
910	.get("targetGrants")
911	.and_then(Value::as_array)
912	.and_then(\|grants\| {
913	grants.iter().find(\|grant\| {
914	grant.get("targetRef").and_then(Value::as_str) == Some(start.target_ref.as_str())
915	&& grant
916	.get("revokedAt")
917	.and_then(Value::as_str)
918	.unwrap_or_default()
919	.is_empty()
920	})
921	})
922	.and_then(\|grant\| grant.get("grantId"))
923	.and_then(Value::as_str)
924	.filter(\|value\| valid_opaque_ref(value))
925	.ok_or_else(\|\| "Codewhale returned no grant for this session.".to_string())?
926	.to_string();
927	Ok(LiveEnrollment {
928	persisted: PersistedEnrollment {
929	schema_version: 1,
930	control_plane_base: base.to_string(),
931	runner_enrollment_id: enrollment_id,
932	account_ref,
933	device_id: returned_device,
934	target_ref: start.target_ref.clone(),
935	target_grant_ref,
936	runtime_version: start.runtime_version.clone(),
937	runtime_commit: start.runtime_commit.clone(),
938	bootstrap_secret: secret_field(&value, "bootstrapSecret")?,
939	},
940	access_token: access_token(&value)?,
941	})
942	}
943
944	async fn refresh_enrollment(
945	client: &Client,
946	persisted: PersistedEnrollment,
947	) -> Result<LiveEnrollment, String> {
948	let url = control_plane_url(
949	&persisted.control_plane_base,
950	&["api", "runner", "enrollments", "token"],
951	&[],
952	)?;
953	let response = client
954	.post(url)
955	.json(&json!({
956	"enrollmentId": persisted.runner_enrollment_id,
957	"bootstrapSecret": persisted.bootstrap_secret,
958	}))
959	.send()
960	.await
961	.map_err(\|_\| "Remote-control credential refresh could not reach Codewhale.".to_string())?;
962	if matches!(
963	response.status(),
964	StatusCode::UNAUTHORIZED \| StatusCode::FORBIDDEN
965	) {
966	return Err("runner_enrollment_revoked".to_string());
967	}
968	if !response.status().is_success() {
969	return Err("Remote-control credential refresh was rejected.".to_string());
970	}
971	let value = read_bounded_json(response).await?;
972	let record = value
973	.get("enrollment")
974	.filter(\|value\| value.is_object())
975	.ok_or_else(\|\| "Codewhale returned an invalid refreshed credential.".to_string())?;
976	if record.get("id").and_then(Value::as_str) != Some(persisted.runner_enrollment_id.as_str())
977	\|\| record.get("userId").and_then(Value::as_str) != Some(persisted.account_ref.as_str())
978	\|\| record.get("deviceId").and_then(Value::as_str) != Some(persisted.device_id.as_str())
979	\|\| record.get("runtimeVersion").and_then(Value::as_str)
980	!= Some(persisted.runtime_version.as_str())
981	\|\| record.get("runtimeCommit").and_then(Value::as_str)
982	!= Some(persisted.runtime_commit.as_str())
983	\|\| !exact_capabilities(record.get("capabilities"))
984	{
985	return Err("Codewhale returned a mismatched refreshed credential.".to_string());
986	}
987	Ok(LiveEnrollment {
988	persisted,
989	access_token: access_token(&value)?,
990	})
991	}
992
993	async fn connect_runner(
994	client: &Client,
995	enrollment: &LiveEnrollment,
996	start: &RemoteStart,
997	) -> Result<String, String> {
998	let value = runner_request(
999	client,
1000	enrollment,
1001	Method::POST,
1002	&["api", "local-runners", "connect"],
1003	&[],
1004	Some(json!({
1005	"deviceId": enrollment.persisted.device_id,
1006	"targetRef": start.target_ref,
1007	"displayLabel": start.workspace_label,
1008	"runtimeVersion": start.runtime_version,
1009	"runtimeCommit": start.runtime_commit,
1010	"capabilities": CAPABILITIES,
1011	"status": "active",
1012	})),
1013	)
1014	.await?;
1015	value
1016	.get("runner")
1017	.and_then(\|value\| value.get("id"))
1018	.and_then(Value::as_str)
1019	.filter(\|value\| valid_opaque_ref(value))
1020	.map(ToString::to_string)
1021	.ok_or_else(\|\| "Codewhale returned an invalid runner lease.".to_string())
1022	}
1023
1024	async fn post_heartbeat(
1025	client: &Client,
1026	enrollment: &LiveEnrollment,
1027	runner_id: &str,
1028	start: &RemoteStart,
1029	status: &str,
1030	) -> Result<(), String> {
1031	runner_request(
1032	client,
1033	enrollment,
1034	Method::POST,
1035	&["api", "local-runners", runner_id, "heartbeat"],
1036	&[],
1037	Some(json!({
1038	"runtimeVersion": start.runtime_version,
1039	"runtimeCommit": start.runtime_commit,
1040	"capabilities": CAPABILITIES,
1041	"status": status,
1042	})),
1043	)
1044	.await
1045	.map(\|_\| ())
1046	}
1047
1048	async fn list_runs(
1049	client: &Client,
1050	enrollment: &LiveEnrollment,
1051	runner_id: &str,
1052	) -> Result<Vec<String>, String> {
1053	let value = runner_request(
1054	client,
1055	enrollment,
1056	Method::GET,
1057	&["api", "local-runners", runner_id, "runs"],
1058	&[],
1059	None,
1060	)
1061	.await?;
1062	let runs = value
1063	.get("runs")
1064	.and_then(Value::as_array)
1065	.filter(\|runs\| runs.len() <= MAX_RUNS)
1066	.ok_or_else(\|\| "Codewhale returned an invalid runner run list.".to_string())?;
1067	runs.iter()
1068	.map(\|run\| {
1069	run.get("id")
1070	.and_then(Value::as_str)
1071	.filter(\|value\| valid_opaque_ref(value))
1072	.map(ToString::to_string)
1073	.ok_or_else(\|\| "Codewhale returned an invalid runner run.".to_string())
1074	})
1075	.collect()
1076	}
1077
1078	async fn list_commands(
1079	client: &Client,
1080	enrollment: &LiveEnrollment,
1081	runner_id: &str,
1082	run_id: &str,
1083	since: u64,
1084	) -> Result<Vec<ListedCommand>, String> {
1085	let value = runner_request(
1086	client,
1087	enrollment,
1088	Method::GET,
1089	&[
1090	"api",
1091	"local-runners",
1092	runner_id,
1093	"runs",
1094	run_id,
1095	"commands",
1096	],
1097	&[
1098	("since_seq", since.to_string()),
1099	("include_accepted", "1".to_string()),
1100	],
1101	None,
1102	)
1103	.await?;
1104	let commands = value
1105	.get("commands")
1106	.and_then(Value::as_array)
1107	.filter(\|commands\| commands.len() <= MAX_COMMANDS)
1108	.ok_or_else(\|\| "Codewhale returned an invalid command list.".to_string())?;
1109	commands
1110	.iter()
1111	.map(\|item\| {
1112	let seq = item
1113	.get("seq")
1114	.and_then(Value::as_u64)
1115	.filter(\|value\| *value > since)
1116	.ok_or_else(\|\| "Codewhale returned an invalid command sequence.".to_string())?;
1117	let command = item
1118	.get("command")
1119	.filter(\|value\| value.is_object())
1120	.cloned()
1121	.ok_or_else(\|\| "Codewhale returned an invalid typed command.".to_string())?;
1122	Ok(ListedCommand {
1123	seq,
1124	command,
1125	ack_status: item
1126	.get("ackStatus")
1127	.and_then(Value::as_str)
1128	.unwrap_or_default()
1129	.to_string(),
1130	})
1131	})
1132	.collect()
1133	}
1134
1135	struct ListedCommand {
1136	seq: u64,
1137	command: Value,
1138	ack_status: String,
1139	}
1140
1141	async fn upload_command_accepted(
1142	client: &Client,
1143	enrollment: &LiveEnrollment,
1144	runner_id: &str,
1145	run_id: &str,
1146	seq: u64,
1147	command: &RemoteCommand,
1148	) -> Result<(), String> {
1149	runner_request(
1150	client,
1151	enrollment,
1152	Method::POST,
1153	&["api", "local-runners", runner_id, "runs", run_id, "events"],
1154	&[],
1155	Some(json!({
1156	"acknowledgements": [{
1157	"commandSeq": seq,
1158	"commandType": command.kind(),
1159	"status": "accepted",
1160	"turnId": command.turn_id(),
1161	}],
1162	"envelopes": [],
1163	})),
1164	)
1165	.await
1166	.map(\|_\| ())
1167	}
1168
1169	async fn recover_run(
1170	client: &Client,
1171	enrollment: &LiveEnrollment,
1172	runner_id: &str,
1173	run_id: &str,
1174	reason: &str,
1175	) -> Result<(), String> {
1176	runner_request(
1177	client,
1178	enrollment,
1179	Method::POST,
1180	&[
1181	"api",
1182	"local-runners",
1183	runner_id,
1184	"runs",
1185	run_id,
1186	"recovery",
1187	],
1188	&[],
1189	Some(json!({ "reason": reason })),
1190	)
1191	.await
1192	.map(\|_\| ())
1193	}
1194
1195	fn parse_remote_command(value: &Value, expected_run_id: &str) -> Result<RemoteCommand, String> {
1196	if value.get("runId").and_then(Value::as_str) != Some(expected_run_id) {
1197	return Err("A remote command targeted a different run.".to_string());
1198	}
1199	match value.get("type").and_then(Value::as_str) {
1200	Some("prompt.request") => {
1201	let turn_id = value
1202	.get("turnId")
1203	.and_then(Value::as_str)
1204	.filter(\|value\| valid_opaque_ref(value))
1205	.ok_or_else(\|\| "A remote prompt had no valid turn id.".to_string())?;
1206	let prompt = value
1207	.get("prompt")
1208	.and_then(Value::as_str)
1209	.map(str::trim)
1210	.filter(\|value\| !value.is_empty() && value.len() <= 128 * 1024)
1211	.ok_or_else(\|\| "A remote prompt was empty or oversized.".to_string())?;
1212	Ok(RemoteCommand::Prompt {
1213	turn_id: turn_id.to_string(),
1214	prompt: prompt.to_string(),
1215	})
1216	}
1217	Some("approval.decision") => {
1218	let gate = value
1219	.get("gate")
1220	.and_then(Value::as_str)
1221	.filter(\|value\| valid_opaque_ref(value))
1222	.ok_or_else(\|\| "A remote approval had no valid gate id.".to_string())?;
1223	let approved = match value.get("decision").and_then(Value::as_str) {
1224	Some("approved") => true,
1225	Some("denied") => false,
1226	_ => return Err("A remote approval had an invalid decision.".to_string()),
1227	};
1228	Ok(RemoteCommand::Approval {
1229	gate: gate.to_string(),
1230	approved,
1231	})
1232	}
1233	Some("run.control") => {
1234	let action = match value.get("action").and_then(Value::as_str) {
1235	Some("interrupt") => RemoteControlRequest::Interrupt,
1236	Some("cancel") => RemoteControlRequest::Cancel,
1237	_ => return Err("A remote run-control command had an invalid action.".to_string()),
1238	};
1239	let turn_id = value
1240	.get("turnId")
1241	.and_then(Value::as_str)
1242	.map(ToString::to_string);
1243	Ok(RemoteCommand::Control { action, turn_id })
1244	}
1245	_ => Err("Codewhale sent an unsupported remote command.".to_string()),
1246	}
1247	}
1248
1249	async fn runner_request(
1250	client: &Client,
1251	enrollment: &LiveEnrollment,
1252	method: Method,
1253	segments: &[&str],
1254	query: &[(&str, String)],
1255	body: Option<Value>,
1256	) -> Result<Value, String> {
1257	let url = control_plane_url(&enrollment.persisted.control_plane_base, segments, query)?;
1258	let mut request = client
1259	.request(method, url)
1260	.bearer_auth(&enrollment.access_token);
1261	if let Some(body) = body {
1262	request = request.json(&body);
1263	}
1264	let response = request
1265	.send()
1266	.await
1267	.map_err(\|_\| "Remote control lost its secure connection.".to_string())?;
1268	if matches!(
1269	response.status(),
1270	StatusCode::UNAUTHORIZED \| StatusCode::FORBIDDEN
1271	) {
1272	delete_persisted_enrollment();
1273	return Err("The remote-control enrollment was revoked.".to_string());
1274	}
1275	if !response.status().is_success() {
1276	return Err(format!(
1277	"The remote-control server rejected a request ({}).",
1278	response.status()
1279	));
1280	}
1281	read_bounded_json(response).await
1282	}
1283
1284	async fn public_request(
1285	client: &Client,
1286	method: Method,
1287	url: Url,
1288	body: Value,
1289	) -> Result<Value, String> {
1290	let response = client
1291	.request(method, url)
1292	.json(&body)
1293	.send()
1294	.await
1295	.map_err(\|_\| "Remote control could not reach Codewhale.".to_string())?;
1296	if !response.status().is_success() {
1297	return Err(format!(
1298	"Codewhale rejected remote-control enrollment ({}).",
1299	response.status()
1300	));
1301	}
1302	read_bounded_json(response).await
1303	}
1304
1305	async fn read_bounded_json(response: reqwest::Response) -> Result<Value, String> {
1306	if response
1307	.content_length()
1308	.is_some_and(\|length\| length > MAX_RESPONSE_BYTES as u64)
1309	{
1310	return Err("Codewhale returned an oversized remote-control response.".to_string());
1311	}
1312	let bytes = response
1313	.bytes()
1314	.await
1315	.map_err(\|_\| "Codewhale returned an unreadable response.".to_string())?;
1316	if bytes.len() > MAX_RESPONSE_BYTES {
1317	return Err("Codewhale returned an oversized remote-control response.".to_string());
1318	}
1319	serde_json::from_slice(&bytes)
1320	.map_err(\|_\| "Codewhale returned an invalid remote-control response.".to_string())
1321	}
1322
1323	fn runner_control_plane_base() -> Result<String, String> {
1324	if cfg!(debug_assertions)
1325	&& let Ok(value) = std::env::var("CWC_RUNNER_CONTROL_PLANE_BASE")
1326	{
1327	let parsed =
1328	Url::parse(&value).map_err(\|_\| "The runner control plane is invalid.".to_string())?;
1329	let loopback = parsed.scheme() == "http"
1330	&& matches!(parsed.host_str(), Some("127.0.0.1" \| "localhost"))
1331	&& parsed.path() == "/"
1332	&& parsed.query().is_none()
1333	&& parsed.fragment().is_none();
1334	if loopback {
1335	return Ok(parsed.to_string());
1336	}
1337	return Err(
1338	"Debug remote control only accepts an explicit loopback control plane.".to_string(),
1339	);
1340	}
1341	Ok(PRODUCTION_CONTROL_PLANE.to_string())
1342	}
1343
1344	fn control_plane_url(
1345	base: &str,
1346	segments: &[&str],
1347	query: &[(&str, String)],
1348	) -> Result<Url, String> {
1349	let mut url =
1350	Url::parse(base).map_err(\|_\| "The runner control plane is invalid.".to_string())?;
1351	{
1352	let mut path = url
1353	.path_segments_mut()
1354	.map_err(\|_\| "The runner control plane is invalid.".to_string())?;
1355	path.pop_if_empty();
1356	for segment in segments {
1357	path.push(segment);
1358	}
1359	}
1360	if !query.is_empty() {
1361	let mut pairs = url.query_pairs_mut();
1362	for (key, value) in query {
1363	pairs.append_pair(key, value);
1364	}
1365	}
1366	Ok(url)
1367	}
1368
1369	fn load_persisted_enrollment() -> Result<Option<PersistedEnrollment>, String> {
1370	let Some(raw) = codewhale_secrets::Secrets::auto_detect()
1371	.get(ENROLLMENT_SECRET_SLOT)
1372	.map_err(\|error\| format!("Could not read the saved remote-control enrollment: {error}"))?
1373	else {
1374	return Ok(None);
1375	};
1376	serde_json::from_str(&raw)
1377	.map(Some)
1378	.map_err(\|_\| "The saved remote-control enrollment is invalid.".to_string())
1379	}
1380
1381	fn save_persisted_enrollment(enrollment: &PersistedEnrollment) -> Result<(), String> {
1382	let raw = serde_json::to_string(enrollment)
1383	.map_err(\|_\| "Could not encode the remote-control enrollment.".to_string())?;
1384	codewhale_secrets::Secrets::auto_detect()
1385	.set(ENROLLMENT_SECRET_SLOT, &raw)
1386	.map_err(\|error\| format!("Could not securely save the remote-control enrollment: {error}"))
1387	}
1388
1389	fn delete_persisted_enrollment() {
1390	if let Err(error) = codewhale_secrets::Secrets::auto_detect().delete(ENROLLMENT_SECRET_SLOT) {
1391	tracing::warn!("could not delete revoked remote-control enrollment: {error}");
1392	}
1393	}
1394
1395	fn enrollment_needs_refresh(enrollment: &LiveEnrollment) -> bool {
1396	jwt_expiry(&enrollment.access_token)
1397	.is_none_or(\|expiry\| expiry <= epoch_seconds().saturating_add(60))
1398	}
1399
1400	fn jwt_expiry(token: &str) -> Option<u64> {
1401	use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
1402	let payload = URL_SAFE_NO_PAD.decode(token.split('.').nth(1)?).ok()?;
1403	serde_json::from_slice::<Value>(&payload)
1404	.ok()?
1405	.get("exp")?
1406	.as_u64()
1407	}
1408
1409	fn access_token(value: &Value) -> Result<String, String> {
1410	let token = value
1411	.get("credential")
1412	.and_then(\|value\| value.get("accessToken"))
1413	.and_then(Value::as_str)
1414	.filter(\|value\| {
1415	(64..=8192).contains(&value.len()) && !value.chars().any(char::is_whitespace)
1416	})
1417	.ok_or_else(\|\| "Codewhale returned an invalid runner access token.".to_string())?
1418	.to_string();
1419	if jwt_expiry(&token).is_none_or(\|expiry\| expiry <= epoch_seconds()) {
1420	return Err("Codewhale returned an expired runner access token.".to_string());
1421	}
1422	Ok(token)
1423	}
1424
1425	fn exact_capabilities(value: Option<&Value>) -> bool {
1426	let Some(items) = value.and_then(Value::as_array) else {
1427	return false;
1428	};
1429	let mut actual = items.iter().filter_map(Value::as_str).collect::<Vec<_>>();
1430	actual.sort_unstable();
1431	actual == CAPABILITIES
1432	}
1433
1434	fn validate_authorization_url(value: &str, user_code: &str) -> Result<(), String> {
1435	let url = Url::parse(value)
1436	.map_err(\|_\| "Codewhale returned an invalid authorization URL.".to_string())?;
1437	let pairs = url.query_pairs().collect::<Vec<_>>();
1438	if url.scheme() != "https"
1439	\|\| url.host_str() != Some("app.codewhale.net")
1440	\|\| url.path() != "/runner/authorize"
1441	\|\| url.port().is_some()
1442	\|\| !url.username().is_empty()
1443	\|\| url.password().is_some()
1444	\|\| url.fragment().is_some()
1445	\|\| pairs.len() != 1
1446	\|\| pairs[0].0 != "user_code"
1447	\|\| pairs[0].1 != user_code
1448	{
1449	return Err("Codewhale returned an invalid authorization URL.".to_string());
1450	}
1451	Ok(())
1452	}
1453
1454	fn string_field(value: &Value, field: &str) -> Result<String, String> {
1455	value
1456	.get(field)
1457	.and_then(Value::as_str)
1458	.map(str::trim)
1459	.filter(\|value\| !value.is_empty() && value.len() <= 2048)
1460	.map(ToString::to_string)
1461	.ok_or_else(\|\| format!("Codewhale returned an invalid {field}."))
1462	}
1463
1464	fn secret_field(value: &Value, field: &str) -> Result<String, String> {
1465	value
1466	.get(field)
1467	.and_then(Value::as_str)
1468	.filter(\|value\| valid_secret(value))
1469	.map(ToString::to_string)
1470	.ok_or_else(\|\| format!("Codewhale returned an invalid {field}."))
1471	}
1472
1473	fn opaque_field(value: &Value, field: &str) -> Result<String, String> {
1474	value
1475	.get(field)
1476	.and_then(Value::as_str)
1477	.filter(\|value\| valid_opaque_ref(value))
1478	.map(ToString::to_string)
1479	.ok_or_else(\|\| format!("Codewhale returned an invalid {field}."))
1480	}
1481
1482	fn valid_opaque_ref(value: &str) -> bool {
1483	(3..=160).contains(&value.len())
1484	&& value
1485	.bytes()
1486	.all(\|byte\| byte.is_ascii_alphanumeric() \|\| matches!(byte, b'_' \| b'-' \| b'.'))
1487	}
1488
1489	fn valid_secret(value: &str) -> bool {
1490	(32..=8192).contains(&value.len()) && !value.chars().any(char::is_whitespace)
1491	}
1492
1493	fn valid_runtime_version(value: &str) -> bool {
1494	semver::Version::parse(value).is_ok() && value.len() <= 64
1495	}
1496
1497	fn valid_runtime_commit(value: &str) -> bool {
1498	value.len() == 40 && value.bytes().all(\|byte\| byte.is_ascii_hexdigit())
1499	}
1500
1501	fn epoch_seconds() -> u64 {
1502	SystemTime::now()
1503	.duration_since(UNIX_EPOCH)
1504	.unwrap_or_default()
1505	.as_secs()
1506	}
1507
1508	#[cfg(test)]
1509	mod tests {
1510	use super::*;
1511	use wiremock::{
1512	Mock, MockServer, ResponseTemplate,
1513	matchers::{body_json, method, path, query_param},
1514	};
1515
1516	#[test]
1517	fn target_identity_is_stable_without_exposing_the_path() {
1518	let target = target_ref(Path::new("/Users/alice/private/project"), "session-123");
1519	assert!(target.starts_with("target_"));
1520	assert_eq!(target.len(), 39);
1521	assert!(!target.contains("alice"));
1522	assert_eq!(
1523	target,
1524	target_ref(Path::new("/Users/alice/private/project"), "session-123")
1525	);
1526	}
1527
1528	#[test]
1529	fn typed_command_parser_rejects_shell_and_cross_run_content() {
1530	let prompt = parse_remote_command(
1531	&json!({
1532	"type": "prompt.request",
1533	"runId": "run-1",
1534	"turnId": "turn-1",
1535	"prompt": "Continue",
1536	}),
1537	"run-1",
1538	)
1539	.unwrap();
1540	assert_eq!(
1541	prompt,
1542	RemoteCommand::Prompt {
1543	turn_id: "turn-1".to_string(),
1544	prompt: "Continue".to_string(),
1545	}
1546	);
1547	assert!(
1548	parse_remote_command(
1549	&json!({
1550	"type": "shell",
1551	"runId": "run-1",
1552	"command": "rm -rf /",
1553	}),
1554	"run-1"
1555	)
1556	.is_err()
1557	);
1558	assert!(
1559	parse_remote_command(
1560	&json!({
1561	"type": "prompt.request",
1562	"runId": "run-other",
1563	"turnId": "turn-1",
1564	"prompt": "Continue",
1565	}),
1566	"run-1"
1567	)
1568	.is_err()
1569	);
1570	}
1571
1572	#[test]
1573	fn approval_projection_matches_control_plane_namespace() {
1574	assert_eq!(projected_approval_id("tool-call-1").len(), 39);
1575	assert!(projected_approval_id("tool-call-1").starts_with("local_approval_"));
1576	assert_ne!(
1577	projected_approval_id("tool-call-1"),
1578	projected_approval_id("tool-call-2")
1579	);
1580	}
1581
1582	#[test]
1583	fn authorization_url_is_exact_and_cannot_redirect_or_add_parameters() {
1584	assert!(
1585	validate_authorization_url(
1586	"https://app.codewhale.net/runner/authorize?user_code=ABCD-EFGH-JKLM",
1587	"ABCD-EFGH-JKLM",
1588	)
1589	.is_ok()
1590	);
1591	for spoofed in [
1592	"http://app.codewhale.net/runner/authorize?user_code=ABCD-EFGH-JKLM",
1593	"https://app.codewhale.net.evil.example/runner/authorize?user_code=ABCD-EFGH-JKLM",
1594	"https://app.codewhale.net/runner/authorize?user_code=ABCD-EFGH-JKLM&next=https://evil.example",
1595	"https://app.codewhale.net/runner/authorize?user_code=WRONG-CODE",
1596	] {
1597	assert!(validate_authorization_url(spoofed, "ABCD-EFGH-JKLM").is_err());
1598	}
1599	}
1600
1601	#[test]
1602	fn command_sequences_are_content_bound_and_replay_safe() {
1603	let mut controller = RemoteControlController::default();
1604	let prompt = RemoteCommand::Prompt {
1605	turn_id: "turn-1".to_string(),
1606	prompt: "Continue".to_string(),
1607	};
1608	assert_eq!(controller.claim_command("run-1", 1, &prompt), Ok(true));
1609	assert_eq!(controller.claim_command("run-1", 1, &prompt), Ok(false));
1610	assert!(
1611	controller
1612	.claim_command(
1613	"run-1",
1614	1,
1615	&RemoteCommand::Prompt {
1616	turn_id: "turn-1".to_string(),
1617	prompt: "Changed".to_string(),
1618	},
1619	)
1620	.is_err()
1621	);
1622	}
1623
1624	#[test]
1625	fn disconnected_remote_owner_keeps_local_input_locked_until_lease_expiry() {
1626	let mut controller = RemoteControlController::default();
1627	controller.status = Status::Failed;
1628	controller.ownership_blocked_until = Some(Instant::now() + Duration::from_secs(90));
1629	assert!(controller.blocks_local_input());
1630	controller.ownership_blocked_until = Some(Instant::now() - Duration::from_secs(1));
1631	assert!(!controller.blocks_local_input());
1632	}
1633
1634	#[test]
1635	fn stop_after_lease_expiry_preserves_pending_approvals_for_restoration() {
1636	let mut controller = RemoteControlController::default();
1637	controller.status = Status::Failed;
1638	controller.ownership_blocked_until = Some(Instant::now() - Duration::from_secs(1));
1639	controller.pending_approvals.insert(
1640	"approval_fixture".to_string(),
1641	PendingRemoteApproval {
1642	tool_id: "tool_fixture".to_string(),
1643	tool_name: "edit".to_string(),
1644	description: "Edit fixture".to_string(),
1645	input: Value::Null,
1646	approval_key: "approval_fixture".to_string(),
1647	intent_summary: Some("fixture".to_string()),
1648	},
1649	);
1650
1651	controller.stop();
1652	assert_eq!(controller.status, Status::Failed);
1653	assert_eq!(controller.pending_approvals.len(), 1);
1654
1655	let event = controller.try_next_event();
1656	assert!(matches!(
1657	event,
1658	Some(RemoteEvent::OwnershipRestored { approvals })
1659	if approvals.len() == 1
1660	&& approvals[0].approval_key == "approval_fixture"
1661	&& approvals[0].tool_id == "tool_fixture"
1662	));
1663	assert_eq!(controller.status, Status::Off);
1664	assert!(controller.pending_approvals.is_empty());
1665	}
1666
1667	#[test]
1668	fn cancelling_a_connect_keeps_input_locked_and_reconnect_blocked() {
1669	let mut controller = RemoteControlController::default();
1670	controller.status = Status::Connecting;
1671	controller.stop();
1672
1673	assert_eq!(controller.status, Status::Failed);
1674	assert!(controller.blocks_local_input());
1675	let result = controller.start(RemoteStart {
1676	workspace_label: "fixture".to_string(),
1677	target_ref: "target_fixture".to_string(),
1678	session_id: "session_fixture".to_string(),
1679	runtime_version: "0.9.1".to_string(),
1680	runtime_commit: "a".repeat(40),
1681	});
1682	assert!(result.is_err());
1683	assert!(result.unwrap_err().contains("previous remote lease"));
1684	}
1685
1686	#[test]
1687	fn failed_worker_allows_snapshot_retry_without_releasing_ownership() {
1688	let mut controller = RemoteControlController::default();
1689	controller.status = Status::Connected;
1690	controller.uploaded_snapshots.insert("run-1".to_string());
1691	let (event_tx, event_rx) = mpsc::unbounded_channel();
1692	controller.event_rx = Some(event_rx);
1693	event_tx
1694	.send(RemoteEvent::Failed("fixture disconnect".to_string()))
1695	.unwrap();
1696
1697	assert!(matches!(
1698	controller.try_next_event(),
1699	Some(RemoteEvent::Failed(_))
1700	));
1701	assert!(controller.uploaded_snapshots.is_empty());
1702	assert!(controller.blocks_local_input());
1703	}
1704
1705	#[tokio::test]
1706	async fn cwc_runner_wire_contract_preserves_pending_and_recovery_commands() {
1707	crate::tls::ensure_rustls_crypto_provider();
1708	let server = MockServer::start().await;
1709	Mock::given(method("GET"))
1710	.and(path("/api/local-runners/runner-1/runs/run-1/commands"))
1711	.and(query_param("since_seq", "0"))
1712	.and(query_param("include_accepted", "1"))
1713	.respond_with(ResponseTemplate::new(200).set_body_json(json!({
1714	"commands": [{
1715	"seq": 1,
1716	"deliveryStatus": "pending",
1717	"ackStatus": "",
1718	"command": {
1719	"type": "prompt.request",
1720	"runId": "run-1",
1721	"turnId": "turn-1",
1722	"prompt": "Continue from the web."
1723	}
1724	}, {
1725	"seq": 2,
1726	"deliveryStatus": "acknowledged",
1727	"ackStatus": "accepted",
1728	"command": {
1729	"type": "run.control",
1730	"runId": "run-1",
1731	"action": "interrupt"
1732	}
1733	}]
1734	})))
1735	.expect(1)
1736	.mount(&server)
1737	.await;
1738	Mock::given(method("POST"))
1739	.and(path("/api/local-runners/runner-1/runs/run-1/events"))
1740	.and(body_json(json!({
1741	"acknowledgements": [{
1742	"commandSeq": 1,
1743	"commandType": "prompt.request",
1744	"status": "accepted",
1745	"turnId": "turn-1"
1746	}],
1747	"envelopes": []
1748	})))
1749	.respond_with(ResponseTemplate::new(200).set_body_json(json!({
1750	"accepted": [],
1751	"count": 1,
1752	"cursor": 0
1753	})))
1754	.expect(1)
1755	.mount(&server)
1756	.await;
1757
1758	let enrollment = LiveEnrollment {
1759	persisted: PersistedEnrollment {
1760	schema_version: 1,
1761	control_plane_base: format!("{}/", server.uri()),
1762	runner_enrollment_id: "enrollment-1".to_string(),
1763	account_ref: "account-1".to_string(),
1764	device_id: "device-1".to_string(),
1765	target_ref: "target-1".to_string(),
1766	target_grant_ref: "grant-1".to_string(),
1767	runtime_version: "0.9.1".to_string(),
1768	runtime_commit: "a".repeat(40),
1769	bootstrap_secret: "b".repeat(43),
1770	},
1771	access_token: "fixture-runner-access-token".to_string(),
1772	};
1773	let client = Client::builder()
1774	.redirect(reqwest::redirect::Policy::none())
1775	.build()
1776	.expect("fixture client");
1777
1778	let listed = list_commands(&client, &enrollment, "runner-1", "run-1", 0)
1779	.await
1780	.expect("CWC command list");
1781	assert_eq!(listed.len(), 2);
1782	assert_eq!(listed[0].ack_status, "");
1783	assert_eq!(listed[1].ack_status, "accepted");
1784	let prompt =
1785	parse_remote_command(&listed[0].command, "run-1").expect("typed prompt command");
1786	upload_command_accepted(
1787	&client,
1788	&enrollment,
1789	"runner-1",
1790	"run-1",
1791	listed[0].seq,
1792	&prompt,
1793	)
1794	.await
1795	.expect("durable accepted acknowledgement");
1796	}
1797	}
1798

1798 lines RUST