terminal_session.rs - CodeWhale

terminal_session.rs

1	//! Stateful, PTY-backed terminal sessions.
2	//!
3	//! Live PTY processes remain deliberately process-local, while a non-secret
4	//! durable summary records identity, last-known cwd, lifecycle state, and
5	//! replacement history. A later process reports the shell as stale/lost and
6	//! starts a new identity; it never claims to reattach from a reused PID.
7
8	#[cfg(unix)]
9	use std::collections::{HashMap, VecDeque};
10	#[cfg(unix)]
11	use std::io::Write;
12	#[cfg(unix)]
13	use std::path::{Path, PathBuf};
14	#[cfg(unix)]
15	use std::sync::{Arc, Mutex, OnceLock};
16	#[cfg(unix)]
17	use std::time::{Duration, Instant};
18
19	use async_trait::async_trait;
20	#[cfg(unix)]
21	use serde::{Deserialize, Serialize};
22	use serde_json::json;
23	#[cfg(unix)]
24	use sha2::{Digest, Sha256};
25	#[cfg(unix)]
26	use uuid::Uuid;
27
28	use super::spec::{
29	ApprovalRequirement, ToolCapability, ToolContext, ToolError, ToolResult, ToolSpec,
30	};
31	#[cfg(unix)]
32	use super::spec::{optional_u64, required_str};
33
34	#[cfg(unix)]
35	const BUFFER_LIMIT: usize = 512 * 1024;
36	#[cfg(unix)]
37	const OUTPUT_LIMIT: usize = 12 * 1024;
38	#[cfg(unix)]
39	const DEFAULT_TIMEOUT_SECS: u64 = 120;
40	#[cfg(unix)]
41	const MAX_TIMEOUT_SECS: u64 = 600;
42	#[cfg(unix)]
43	const CANCEL_CONFIRM_TIMEOUT: Duration = Duration::from_secs(2);
44	#[cfg(unix)]
45	const CANCEL_SENTINEL_RETRY_INTERVAL: Duration = Duration::from_millis(50);
46
47	#[cfg(unix)]
48	struct TerminalSession {
49	writer: Arc<Mutex<Box<dyn Write + Send>>>,
50	child: Box<dyn portable_pty::Child + Send>,
51	output: Arc<Mutex<OutputBuffer>>,
52	read_cursor: u64,
53	command: Option<CommandState>,
54	durable: DurableTerminalRecord,
55	durable_path: PathBuf,
56	}
57
58	#[cfg(unix)]
59	#[derive(Clone, Debug, Serialize, Deserialize)]
60	struct DurableTerminalRecord {
61	schema_version: u32,
62	session_id: String,
63	runtime_nonce: String,
64	process_id: u32,
65	workspace: PathBuf,
66	name: String,
67	shell: String,
68	last_known_cwd: PathBuf,
69	environment_summary: EnvironmentSummary,
70	state: DurableTerminalState,
71	updated_at: String,
72	#[serde(default, skip_serializing_if = "Option::is_none")]
73	previous: Option<Box<DurableTerminalRecord>>,
74	}
75
76	#[cfg(unix)]
77	#[derive(Clone, Debug, Serialize, Deserialize)]
78	struct EnvironmentSummary {
79	term: Option<String>,
80	virtual_env_active: bool,
81	conda_env_active: bool,
82	nix_shell_active: bool,
83	note: String,
84	}
85
86	#[cfg(unix)]
87	#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
88	#[serde(rename_all = "snake_case")]
89	enum DurableTerminalState {
90	Running,
91	Idle,
92	Canceled,
93	Failed,
94	StaleLost,
95	Reset,
96	}
97
98	#[cfg(unix)]
99	struct CommandState {
100	marker: String,
101	}
102
103	#[cfg(unix)]
104	#[derive(Default)]
105	struct OutputBuffer {
106	bytes: VecDeque<u8>,
107	total: u64,
108	}
109
110	#[cfg(unix)]
111	impl OutputBuffer {
112	fn append(&mut self, data: &[u8]) {
113	self.total = self.total.saturating_add(data.len() as u64);
114	self.bytes.extend(data);
115	while self.bytes.len() > BUFFER_LIMIT {
116	let _ = self.bytes.pop_front();
117	}
118	}
119
120	fn text(&self) -> String {
121	String::from_utf8_lossy(&self.bytes.iter().copied().collect::<Vec<_>>()).into_owned()
122	}
123	}
124
125	#[cfg(unix)]
126	type SharedSession = Arc<Mutex<TerminalSession>>;
127
128	#[cfg(unix)]
129	#[derive(Clone, Debug, Hash, PartialEq, Eq)]
130	struct SessionKey {
131	workspace: PathBuf,
132	name: String,
133	}
134
135	#[cfg(unix)]
136	static SESSIONS: OnceLock<Mutex<HashMap<SessionKey, SharedSession>>> = OnceLock::new();
137
138	#[cfg(unix)]
139	static RUNTIME_NONCE: OnceLock<String> = OnceLock::new();
140
141	#[cfg(unix)]
142	fn runtime_nonce() -> &'static str {
143	RUNTIME_NONCE.get_or_init(\|\| Uuid::new_v4().to_string())
144	}
145
146	#[cfg(unix)]
147	fn sessions() -> &'static Mutex<HashMap<SessionKey, SharedSession>> {
148	SESSIONS.get_or_init(\|\| Mutex::new(HashMap::new()))
149	}
150
151	#[cfg(unix)]
152	fn session_key(name: &str, workspace: &Path) -> SessionKey {
153	SessionKey {
154	workspace: workspace
155	.canonicalize()
156	.unwrap_or_else(\|_\| workspace.to_path_buf()),
157	name: name.to_string(),
158	}
159	}
160
161	#[cfg(unix)]
162	fn durable_path(name: &str, workspace: &Path) -> Result<PathBuf, String> {
163	let workspace = workspace
164	.canonicalize()
165	.unwrap_or_else(\|_\| workspace.to_path_buf());
166	let mut hasher = Sha256::new();
167	hasher.update(workspace.to_string_lossy().as_bytes());
168	hasher.update(b"\0");
169	hasher.update(name.as_bytes());
170	let digest = hasher
171	.finalize()
172	.iter()
173	.map(\|byte\| format!("{byte:02x}"))
174	.collect::<String>();
175	#[cfg(test)]
176	let state_dir = workspace.join(".codewhale-test-terminal-sessions");
177	#[cfg(not(test))]
178	let state_dir = codewhale_config::ensure_state_dir("terminal-sessions")
179	.map_err(\|error\| format!("failed to resolve terminal session state directory: {error}"))?;
180	std::fs::create_dir_all(&state_dir)
181	.map_err(\|error\| format!("failed to create terminal session state directory: {error}"))?;
182	Ok(state_dir.join(format!("{}.json", &digest[..32])))
183	}
184
185	#[cfg(unix)]
186	fn load_durable(path: &Path) -> Option<DurableTerminalRecord> {
187	let bytes = std::fs::read(path).ok()?;
188	serde_json::from_slice(&bytes).ok()
189	}
190
191	#[cfg(unix)]
192	fn persist_durable(path: &Path, record: &DurableTerminalRecord) -> Result<(), String> {
193	let payload = serde_json::to_vec_pretty(record)
194	.map_err(\|error\| format!("failed to encode terminal session state: {error}"))?;
195	crate::utils::write_atomic(path, &payload)
196	.map_err(\|error\| format!("failed to persist terminal session state: {error}"))
197	}
198
199	#[cfg(unix)]
200	fn create_session(name: &str, workspace: &std::path::Path) -> Result<SharedSession, String> {
201	let shell = std::env::var("SHELL").unwrap_or_else(\|_\| "/bin/sh".to_string());
202	let workspace = workspace
203	.canonicalize()
204	.unwrap_or_else(\|_\| workspace.to_path_buf());
205	let durable_path = durable_path(name, &workspace)?;
206	let previous = load_durable(&durable_path).map(\|mut record\| {
207	// A persisted shell is historical evidence only. A random per-process
208	// nonce makes PID reuse irrelevant and deliberately forbids reattach.
209	record.state = DurableTerminalState::StaleLost;
210	record.updated_at = chrono::Utc::now().to_rfc3339();
211	Box::new(record)
212	});
213	let pty = portable_pty::native_pty_system();
214	let pair = pty
215	.openpty(portable_pty::PtySize {
216	rows: 24,
217	cols: 120,
218	pixel_width: 0,
219	pixel_height: 0,
220	})
221	.map_err(\|e\| format!("failed to open PTY: {e}"))?;
222
223	let mut command = portable_pty::CommandBuilder::new(&shell);
224	command.arg("-i");
225	command.cwd(&workspace);
226	let child = pair
227	.slave
228	.spawn_command(command)
229	.map_err(\|e\| format!("failed to start shell {shell}: {e}"))?;
230	drop(pair.slave);
231
232	let reader = pair
233	.master
234	.try_clone_reader()
235	.map_err(\|e\| format!("failed to read PTY: {e}"))?;
236	let writer = pair
237	.master
238	.take_writer()
239	.map_err(\|e\| format!("failed to write PTY: {e}"))?;
240	let output = Arc::new(Mutex::new(OutputBuffer::default()));
241	let reader_output = Arc::clone(&output);
242	std::thread::spawn(move \|\| {
243	let mut reader = reader;
244	let mut buf = [0u8; 8192];
245	loop {
246	match std::io::Read::read(&mut reader, &mut buf) {
247	Ok(0) \| Err(_) => break,
248	Ok(n) => {
249	if let Ok(mut output) = reader_output.lock() {
250	output.append(&buf[..n]);
251	}
252	}
253	}
254	}
255	});
256
257	let durable = DurableTerminalRecord {
258	schema_version: 1,
259	session_id: Uuid::new_v4().to_string(),
260	runtime_nonce: runtime_nonce().to_string(),
261	process_id: std::process::id(),
262	workspace: workspace.clone(),
263	name: name.to_string(),
264	shell,
265	last_known_cwd: workspace,
266	environment_summary: EnvironmentSummary {
267	term: std::env::var("TERM").ok().filter(\|value\| !value.trim().is_empty()),
268	virtual_env_active: std::env::var_os("VIRTUAL_ENV").is_some(),
269	conda_env_active: std::env::var_os("CONDA_PREFIX").is_some(),
270	nix_shell_active: std::env::var_os("IN_NIX_SHELL").is_some(),
271	note: "Values and secrets are not persisted; in-shell environment changes are process-local."
272	.to_string(),
273	},
274	state: DurableTerminalState::Idle,
275	updated_at: chrono::Utc::now().to_rfc3339(),
276	previous,
277	};
278	persist_durable(&durable_path, &durable)?;
279
280	Ok(Arc::new(Mutex::new(TerminalSession {
281	writer: Arc::new(Mutex::new(writer)),
282	child,
283	output,
284	read_cursor: 0,
285	command: None,
286	durable,
287	durable_path,
288	})))
289	}
290
291	#[cfg(unix)]
292	fn get_or_create(name: &str, workspace: &std::path::Path) -> Result<SharedSession, String> {
293	let key = session_key(name, workspace);
294	let mut registry = sessions()
295	.lock()
296	.map_err(\|_\| "terminal session registry lock poisoned".to_string())?;
297	if let Some(session) = registry.get(&key) {
298	return Ok(Arc::clone(session));
299	}
300	let session = create_session(name, workspace)?;
301	registry.insert(key, Arc::clone(&session));
302	Ok(session)
303	}
304
305	#[cfg(unix)]
306	fn find(name: &str, workspace: &Path) -> Result<SharedSession, String> {
307	let live = sessions()
308	.lock()
309	.map_err(\|_\| "terminal session registry lock poisoned".to_string())?
310	.get(&session_key(name, workspace))
311	.cloned();
312	if let Some(live) = live {
313	return Ok(live);
314	}
315	if let Ok(path) = durable_path(name, workspace)
316	&& let Some(mut record) = load_durable(&path)
317	{
318	record.state = DurableTerminalState::StaleLost;
319	record.updated_at = chrono::Utc::now().to_rfc3339();
320	let _ = persist_durable(&path, &record);
321	return Err(format!(
322	"terminal session '{name}' is stale/lost after restart (last cwd: {}); run terminal/run with this name to start a replacement while preserving the historical summary",
323	record.last_known_cwd.display()
324	));
325	}
326	Err(format!(
327	"terminal session '{name}' does not exist in workspace {}",
328	workspace.display()
329	))
330	}
331
332	#[cfg(unix)]
333	fn write_bytes(session: &TerminalSession, bytes: &[u8]) -> Result<(), String> {
334	let mut writer = session
335	.writer
336	.lock()
337	.map_err(\|_\| "terminal PTY writer lock poisoned".to_string())?;
338	writer
339	.write_all(bytes)
340	.map_err(\|e\| format!("PTY write failed: {e}"))?;
341	writer.flush().map_err(\|e\| format!("PTY flush failed: {e}"))
342	}
343
344	#[cfg(unix)]
345	fn output_snapshot(session: &TerminalSession) -> String {
346	session
347	.output
348	.lock()
349	.map(\|output\| output.text())
350	.unwrap_or_default()
351	}
352
353	#[cfg(unix)]
354	fn take_output(session: &mut TerminalSession) -> String {
355	let Ok(output) = session.output.lock() else {
356	return String::new();
357	};
358	let retained_start = output.total.saturating_sub(output.bytes.len() as u64);
359	let start = session.read_cursor.max(retained_start);
360	let skip = usize::try_from(start.saturating_sub(retained_start)).unwrap_or(usize::MAX);
361	let bytes = output.bytes.iter().skip(skip).copied().collect::<Vec<_>>();
362	session.read_cursor = output.total;
363	String::from_utf8_lossy(&bytes).into_owned()
364	}
365
366	#[cfg(unix)]
367	fn prune_output(input: &str) -> String {
368	if input.len() <= OUTPUT_LIMIT {
369	return input.to_string();
370	}
371	let head = OUTPUT_LIMIT / 3;
372	let tail = OUTPUT_LIMIT - head;
373	let head_end = input
374	.char_indices()
375	.find(\|(index, _)\| *index >= head)
376	.map_or(input.len(), \|(index, _)\| index);
377	let tail_start = input
378	.char_indices()
379	.rev()
380	.find(\|(index, _)\| input.len() - *index <= tail)
381	.map_or(0, \|(index, _)\| index);
382	format!(
383	"{}\n… [output truncated: {} bytes omitted] …\n{}",
384	&input[..head_end],
385	input.len() - OUTPUT_LIMIT,
386	&input[tail_start..]
387	)
388	}
389
390	#[cfg(unix)]
391	fn completion(session: &TerminalSession) -> Option<(i32, String)> {
392	let state = session.command.as_ref()?;
393	let output = output_snapshot(session);
394	let marker = format!("\n{}:", state.marker);
395	let line = output
396	.rsplit(&marker)
397	.next()
398	.and_then(\|tail\| tail.lines().next())?;
399	let (status, cwd) = line.split_once(':')?;
400	Some((status.parse().ok()?, cwd.to_string()))
401	}
402
403	#[cfg(unix)]
404	fn start_command(session: &mut TerminalSession, command: &str) -> Result<(), String> {
405	if session.command.is_some() && completion(session).is_none() {
406	return Err("terminal session already has a running foreground command".to_string());
407	}
408	let marker = format!("__CODEWHALE_TERM_{}__", Uuid::new_v4().simple());
409	// The command must run in the CURRENT shell — a subshell would discard
410	// exactly the state (cd, exports, functions, activated envs) this tool
411	// exists to preserve (EXEC-001). The sentinel line is typed after the
412	// command; the tty line discipline holds it until the foreground command
413	// finishes reading input.
414	let wrapped = format!(
415	"{command}\n__cw_status=$?; printf '\\n{marker}:%s:%s\\n' \"$__cw_status\" \"$PWD\"\n"
416	);
417	write_bytes(session, wrapped.as_bytes())?;
418	session.command = Some(CommandState { marker });
419	session.durable.state = DurableTerminalState::Running;
420	session.durable.updated_at = chrono::Utc::now().to_rfc3339();
421	persist_durable(&session.durable_path, &session.durable)?;
422	Ok(())
423	}
424
425	#[cfg(unix)]
426	fn write_completion_sentinel(
427	session: &TerminalSession,
428	marker: &str,
429	status: i32,
430	) -> Result<(), String> {
431	let sentinel =
432	format!("__cw_status={status}; printf '\\n{marker}:%s:%s\\n' \"$__cw_status\" \"$PWD\"\n");
433	write_bytes(session, sentinel.as_bytes())
434	}
435
436	#[cfg(unix)]
437	#[cfg(test)]
438	fn wait_session(session: &mut TerminalSession, timeout: Duration) -> (Option<(i32, String)>, bool) {
439	let deadline = Instant::now() + timeout;
440	loop {
441	if let Some(done) = completion(session) {
442	return (Some(done), false);
443	}
444	if Instant::now() >= deadline {
445	return (None, true);
446	}
447	std::thread::sleep(Duration::from_millis(25));
448	}
449	}
450
451	/// Wait without monopolizing the per-session lock. `terminal/send` and
452	/// `terminal/cancel` must be able to acquire the lock while a foreground
453	/// command is active; otherwise interactive input and cancellation deadlock
454	/// behind the waiter.
455	#[cfg(unix)]
456	fn wait_shared_session(
457	session: &SharedSession,
458	timeout: Duration,
459	) -> Result<(Option<(i32, String)>, bool), ToolError> {
460	let deadline = Instant::now() + timeout;
461	loop {
462	let done = {
463	let session = session
464	.lock()
465	.map_err(\|_\| ToolError::execution_failed("terminal session lock poisoned"))?;
466	completion(&session)
467	};
468	if let Some(done) = done {
469	return Ok((Some(done), false));
470	}
471	if Instant::now() >= deadline {
472	return Ok((None, true));
473	}
474	std::thread::sleep(Duration::from_millis(25));
475	}
476	}
477
478	/// Interrupt a foreground command and wait until the persistent shell confirms
479	/// that it is ready again.
480	///
481	/// Canonical PTYs may flush queued input when they process ETX. Resubmit the
482	/// completion sentinel until the shell acknowledges it so a busy host cannot
483	/// strand the session merely because the one post-interrupt write raced that
484	/// flush.
485	#[cfg(unix)]
486	fn cancel_shared_session(session: &SharedSession) -> Result<(i32, String), ToolError> {
487	let marker = {
488	let session = session
489	.lock()
490	.map_err(\|_\| ToolError::execution_failed("terminal session lock poisoned"))?;
491	if session.command.is_none() \|\| completion(&session).is_some() {
492	return Err(ToolError::execution_failed(
493	"terminal session has no running foreground command",
494	));
495	}
496	write_bytes(&session, &[3]).map_err(ToolError::execution_failed)?;
497	session
498	.command
499	.as_ref()
500	.expect("running command checked above")
501	.marker
502	.clone()
503	};
504	let deadline = Instant::now() + CANCEL_CONFIRM_TIMEOUT;
505
506	loop {
507	std::thread::sleep(CANCEL_SENTINEL_RETRY_INTERVAL);
508	let session = session
509	.lock()
510	.map_err(\|_\| ToolError::execution_failed("terminal session lock poisoned"))?;
511	if let Some(done) = completion(&session) {
512	return Ok(done);
513	}
514	if Instant::now() >= deadline {
515	return Err(ToolError::execution_failed(
516	"terminal interrupt was sent, but cancellation was not confirmed within 2 seconds",
517	));
518	}
519	write_completion_sentinel(&session, &marker, 130).map_err(ToolError::execution_failed)?;
520	}
521	}
522
523	#[cfg(unix)]
524	fn session_result(
525	session: &mut TerminalSession,
526	done: Option<(i32, String)>,
527	timed_out: bool,
528	) -> ToolResult {
529	let output = prune_output(&take_output(session));
530	let finished = done.is_some();
531	let (exit_code, cwd) = done.map_or((None, String::new()), \|(code, cwd)\| (Some(code), cwd));
532	let status = if timed_out {
533	"timed_out"
534	} else if finished {
535	"completed"
536	} else {
537	"running"
538	};
539	if !cwd.is_empty() {
540	session.durable.last_known_cwd = PathBuf::from(&cwd);
541	}
542	session.durable.state = if timed_out \|\| !finished {
543	DurableTerminalState::Running
544	} else if exit_code == Some(0) {
545	DurableTerminalState::Idle
546	} else if exit_code == Some(130) {
547	DurableTerminalState::Canceled
548	} else {
549	DurableTerminalState::Failed
550	};
551	session.durable.updated_at = chrono::Utc::now().to_rfc3339();
552	let persistence_error = persist_durable(&session.durable_path, &session.durable).err();
553	let previous = session.durable.previous.as_deref().map(\|record\| {
554	json!({
555	"session_id": record.session_id,
556	"state": record.state,
557	"last_known_cwd": record.last_known_cwd,
558	"updated_at": record.updated_at,
559	})
560	});
561	ToolResult {
562	content: output,
563	// A successful send while the command is still running is itself a
564	// successful tool operation. Completed commands still report their
565	// real exit status, and timeouts remain unsuccessful.
566	success: !timed_out && (!finished \|\| exit_code == Some(0)),
567	metadata: Some(json!({
568	"status": status,
569	"exit_code": exit_code,
570	"cwd": cwd,
571	"session_persistent": true,
572	"durability": "live shell is process-local; identity and last-known summary persist",
573	"terminal_session_id": session.durable.session_id,
574	"terminal_state": session.durable.state,
575	"state_path": session.durable_path,
576	"previous_session": previous,
577	"persistence_error": persistence_error,
578	})),
579	}
580	}
581
582	#[cfg(unix)]
583	fn session_name(input: &serde_json::Value, required: bool) -> Result<&str, ToolError> {
584	match input.get("session").and_then(serde_json::Value::as_str) {
585	Some(name) if !name.is_empty() => Ok(name),
586	Some(_) => Err(ToolError::execution_failed("session must not be empty")),
587	None if required => Err(ToolError::missing_field("session")),
588	None => Ok("term-1"),
589	}
590	}
591
592	#[cfg(unix)]
593	fn timeout_secs(input: &serde_json::Value, key: &str) -> Result<Duration, ToolError> {
594	Ok(Duration::from_secs(
595	optional_u64(input, key, DEFAULT_TIMEOUT_SECS)?.clamp(1, MAX_TIMEOUT_SECS),
596	))
597	}
598
599	fn shell_allowed(context: &ToolContext) -> Result<(), ToolError> {
600	if context.shell_policy.allows_shell() {
601	Ok(())
602	} else {
603	Err(ToolError::execution_failed(
604	"Shell tools are disabled by the active permission profile.",
605	))
606	}
607	}
608
609	#[cfg(not(unix))]
610	fn unsupported() -> ToolResult {
611	ToolResult::error("Stateful terminal sessions are currently supported on Unix only.")
612	}
613
614	macro_rules! terminal_tool_common {
615	($name:literal, $description:literal) => {
616	fn name(&self) -> &'static str {
617	$name
618	}
619	fn description(&self) -> &'static str {
620	$description
621	}
622	fn capabilities(&self) -> Vec<ToolCapability> {
623	vec![
624	ToolCapability::ExecutesCode,
625	ToolCapability::RequiresApproval,
626	]
627	}
628	fn approval_requirement(&self) -> ApprovalRequirement {
629	ApprovalRequirement::Required
630	}
631	};
632	}
633
634	pub struct TerminalRunTool;
635	#[async_trait]
636	impl ToolSpec for TerminalRunTool {
637	terminal_tool_common!(
638	"terminal/run",
639	"Run a command in a persistent PTY shell session. cd, exports, shell functions, and activated environments persist across calls in this process. Identity and a non-secret last-known summary persist across restarts; prior shells are surfaced as stale/lost and are never reattached."
640	);
641	fn input_schema(&self) -> serde_json::Value {
642	json!({"type":"object","properties":{"command":{"type":"string"},"session":{"type":"string","default":"term-1"},"timeout_secs":{"type":"integer","default":120}},"required":["command"]})
643	}
644	async fn execute(
645	&self,
646	input: serde_json::Value,
647	context: &ToolContext,
648	) -> Result<ToolResult, ToolError> {
649	shell_allowed(context)?;
650	#[cfg(unix)]
651	{
652	let command = required_str(&input, "command")?.to_string();
653	let name = session_name(&input, false)?.to_string();
654	let session =
655	get_or_create(&name, &context.workspace).map_err(ToolError::execution_failed)?;
656	let timeout = timeout_secs(&input, "timeout_secs")?;
657	return tokio::task::spawn_blocking(move \|\| {
658	{
659	let mut session = session.lock().map_err(\|_\| {
660	ToolError::execution_failed("terminal session lock poisoned")
661	})?;
662	start_command(&mut session, &command).map_err(ToolError::execution_failed)?;
663	}
664	let (done, timed_out) = wait_shared_session(&session, timeout)?;
665	let mut session = session
666	.lock()
667	.map_err(\|_\| ToolError::execution_failed("terminal session lock poisoned"))?;
668	Ok(session_result(&mut session, done, timed_out))
669	})
670	.await
671	.map_err(\|e\| ToolError::execution_failed(e.to_string()))?;
672	}
673	#[cfg(not(unix))]
674	{
675	let _ = input;
676	Ok(unsupported())
677	}
678	}
679	}
680
681	pub struct TerminalSendTool;
682	#[async_trait]
683	impl ToolSpec for TerminalSendTool {
684	terminal_tool_common!(
685	"terminal/send",
686	"Send raw input to a live persistent terminal session. Use a literal ETX control byte to interrupt an interactive process. A prior-process shell is reported as stale/lost rather than reattached."
687	);
688	fn input_schema(&self) -> serde_json::Value {
689	json!({"type":"object","properties":{"session":{"type":"string"},"text":{"type":"string"},"wait_ms":{"type":"integer","default":250}},"required":["session","text"]})
690	}
691	async fn execute(
692	&self,
693	input: serde_json::Value,
694	context: &ToolContext,
695	) -> Result<ToolResult, ToolError> {
696	shell_allowed(context)?;
697	#[cfg(unix)]
698	{
699	let name = session_name(&input, true)?.to_string();
700	let text = required_str(&input, "text")?.as_bytes().to_vec();
701	let session = find(&name, &context.workspace).map_err(ToolError::execution_failed)?;
702	let wait = Duration::from_millis(optional_u64(&input, "wait_ms", 250)?.min(60_000));
703	return tokio::task::spawn_blocking(move \|\| {
704	let mut session = session
705	.lock()
706	.map_err(\|_\| ToolError::execution_failed("terminal session lock poisoned"))?;
707	write_bytes(&session, &text).map_err(ToolError::execution_failed)?;
708	std::thread::sleep(wait);
709	let done = completion(&session);
710	Ok(session_result(&mut session, done, false))
711	})
712	.await
713	.map_err(\|e\| ToolError::execution_failed(e.to_string()))?;
714	}
715	#[cfg(not(unix))]
716	{
717	let _ = input;
718	Ok(unsupported())
719	}
720	}
721	}
722
723	pub struct TerminalWaitTool;
724	#[async_trait]
725	impl ToolSpec for TerminalWaitTool {
726	terminal_tool_common!(
727	"terminal/wait",
728	"Wait for the current foreground command in a live persistent terminal session and return buffered output. A prior-process shell is reported as stale/lost rather than reattached."
729	);
730	fn input_schema(&self) -> serde_json::Value {
731	json!({"type":"object","properties":{"session":{"type":"string"},"timeout_secs":{"type":"integer","default":120}},"required":["session"]})
732	}
733	async fn execute(
734	&self,
735	input: serde_json::Value,
736	context: &ToolContext,
737	) -> Result<ToolResult, ToolError> {
738	shell_allowed(context)?;
739	#[cfg(unix)]
740	{
741	let name = session_name(&input, true)?.to_string();
742	let session = find(&name, &context.workspace).map_err(ToolError::execution_failed)?;
743	let timeout = timeout_secs(&input, "timeout_secs")?;
744	return tokio::task::spawn_blocking(move \|\| {
745	let (done, timed_out) = wait_shared_session(&session, timeout)?;
746	let mut session = session
747	.lock()
748	.map_err(\|_\| ToolError::execution_failed("terminal session lock poisoned"))?;
749	Ok(session_result(&mut session, done, timed_out))
750	})
751	.await
752	.map_err(\|e\| ToolError::execution_failed(e.to_string()))?;
753	}
754	#[cfg(not(unix))]
755	{
756	let _ = input;
757	Ok(unsupported())
758	}
759	}
760	}
761
762	pub struct TerminalCancelTool;
763	#[async_trait]
764	impl ToolSpec for TerminalCancelTool {
765	terminal_tool_common!(
766	"terminal/cancel",
767	"Interrupt the running foreground command with ETX. The live terminal session survives and can be reused; its non-secret summary persists."
768	);
769	fn input_schema(&self) -> serde_json::Value {
770	json!({"type":"object","properties":{"session":{"type":"string"}},"required":["session"]})
771	}
772	async fn execute(
773	&self,
774	input: serde_json::Value,
775	context: &ToolContext,
776	) -> Result<ToolResult, ToolError> {
777	shell_allowed(context)?;
778	#[cfg(unix)]
779	{
780	let name = session_name(&input, true)?.to_string();
781	let session = find(&name, &context.workspace).map_err(ToolError::execution_failed)?;
782	return tokio::task::spawn_blocking(move \|\| {
783	let done = cancel_shared_session(&session)?;
784	let mut session = session
785	.lock()
786	.map_err(\|_\| ToolError::execution_failed("terminal session lock poisoned"))?;
787	let mut result = session_result(&mut session, Some(done), false);
788	result.success = true;
789	if let Some(metadata) = result.metadata.as_mut() {
790	metadata["status"] = json!("canceled");
791	metadata["canceled"] = json!(true);
792	}
793	Ok(result)
794	})
795	.await
796	.map_err(\|e\| ToolError::execution_failed(e.to_string()))?;
797	}
798	#[cfg(not(unix))]
799	{
800	let _ = input;
801	Ok(unsupported())
802	}
803	}
804	}
805
806	pub struct TerminalResetTool;
807	#[async_trait]
808	impl ToolSpec for TerminalResetTool {
809	terminal_tool_common!(
810	"terminal/reset",
811	"Kill and recreate a persistent terminal session with a fresh environment. This loses live cd, exports, functions, activated environments, and running work while retaining the prior historical summary."
812	);
813	fn input_schema(&self) -> serde_json::Value {
814	json!({"type":"object","properties":{"session":{"type":"string"}},"required":["session"]})
815	}
816	async fn execute(
817	&self,
818	input: serde_json::Value,
819	context: &ToolContext,
820	) -> Result<ToolResult, ToolError> {
821	shell_allowed(context)?;
822	#[cfg(unix)]
823	{
824	let name = session_name(&input, true)?.to_string();
825	let old = find(&name, &context.workspace).map_err(ToolError::execution_failed)?;
826	let workspace = context.workspace.clone();
827	return tokio::task::spawn_blocking(move \|\| {
828	if let Ok(mut old) = old.lock() { let _ = old.child.kill(); }
829	if let Ok(mut old) = old.lock() {
830	old.durable.state = DurableTerminalState::Reset;
831	old.durable.updated_at = chrono::Utc::now().to_rfc3339();
832	let _ = persist_durable(&old.durable_path, &old.durable);
833	}
834	let fresh = create_session(&name, &workspace).map_err(ToolError::execution_failed)?;
835	sessions().lock().map_err(\|_\| ToolError::execution_failed("terminal session registry lock poisoned"))?.insert(session_key(&name, &workspace), fresh);
836	Ok(ToolResult { content: format!("Reset terminal session '{name}'. Lost shell state and any running command."), success: true, metadata: Some(json!({"session":name,"reset":true,"lost_state":["cwd","environment","functions","activated environments","running command"]})) })
837	}).await.map_err(\|e\| ToolError::execution_failed(e.to_string()))?;
838	}
839	#[cfg(not(unix))]
840	{
841	let _ = input;
842	Ok(unsupported())
843	}
844	}
845	}
846
847	#[cfg(all(test, unix))]
848	mod tests {
849	use super::*;
850
851	fn fresh(name: &str) -> SharedSession {
852	let session = get_or_create(name, std::path::Path::new("/tmp")).unwrap();
853	let mut session_guard = session.lock().unwrap();
854	let _ = session_guard.child.kill();
855	drop(session_guard);
856	let replacement = create_session(name, std::path::Path::new("/tmp")).unwrap();
857	sessions().lock().unwrap().insert(
858	session_key(name, Path::new("/tmp")),
859	Arc::clone(&replacement),
860	);
861	replacement
862	}
863
864	fn run(session: &SharedSession, command: &str, timeout: Duration) -> ToolResult {
865	let mut session = session.lock().unwrap();
866	start_command(&mut session, command).unwrap();
867	let (done, timed_out) = wait_session(&mut session, timeout);
868	session_result(&mut session, done, timed_out)
869	}
870
871	#[test]
872	#[cfg(unix)]
873	fn cd_persists_between_runs() {
874	let session = fresh("test-cd");
875	let _ = run(
876	&session,
877	"mkdir -p /tmp/cw-term-cd-proof && cd /tmp/cw-term-cd-proof",
878	Duration::from_secs(3),
879	);
880	// A separate run must still be inside the directory — this is the
881	// whole point of the stateful session (EXEC-001).
882	let result = run(&session, "pwd", Duration::from_secs(3));
883	assert!(result.content.contains("cw-term-cd-proof"), "{}", {
884	&result.content
885	});
886	}
887
888	#[test]
889	#[cfg(unix)]
890	fn export_persists_and_sessions_are_isolated() {
891	let one = fresh("test-env-one");
892	let two = fresh("test-env-two");
893	let _ = run(&one, "export CW_TERM_TEST=present", Duration::from_secs(3));
894	assert!(
895	run(&one, "printf %s $CW_TERM_TEST", Duration::from_secs(3))
896	.content
897	.contains("present")
898	);
899	assert!(
900	!run(
901	&two,
902	"printf %s ${CW_TERM_TEST-unset}",
903	Duration::from_secs(3)
904	)
905	.content
906	.contains("present")
907	);
908	}
909
910	#[test]
911	#[cfg(unix)]
912	fn reset_replaces_shell_environment() {
913	let session = fresh("test-reset");
914	let _ = run(
915	&session,
916	"export CW_TERM_RESET=present",
917	Duration::from_secs(3),
918	);
919	assert!(
920	run(&session, "printf %s $CW_TERM_RESET", Duration::from_secs(3))
921	.content
922	.contains("present")
923	);
924	let _ = session.lock().unwrap().child.kill();
925	let replacement = create_session("test-reset", std::path::Path::new("/tmp")).unwrap();
926	sessions().lock().unwrap().insert(
927	session_key("test-reset", Path::new("/tmp")),
928	Arc::clone(&replacement),
929	);
930	assert!(
931	!run(
932	&replacement,
933	"printf %s ${CW_TERM_RESET-unset}",
934	Duration::from_secs(3)
935	)
936	.content
937	.contains("present")
938	);
939	}
940
941	#[test]
942	#[cfg(unix)]
943	fn timeout_leaves_session_alive() {
944	let session = fresh("test-timeout");
945	let result = run(
946	&session,
947	"printf before; sleep 2",
948	Duration::from_millis(100),
949	);
950	assert_eq!(result.metadata.as_ref().unwrap()["status"], "timed_out");
951	let result = {
952	let mut session_guard = session.lock().unwrap();
953	let (done, timed_out) = wait_session(&mut session_guard, Duration::from_secs(3));
954	session_result(&mut session_guard, done, timed_out)
955	};
956	assert_eq!(result.metadata.as_ref().unwrap()["status"], "completed");
957	let result = run(&session, "printf after", Duration::from_secs(3));
958	assert!(result.content.contains("after"));
959	}
960
961	#[test]
962	#[cfg(unix)]
963	fn cancel_interrupts_sleep_and_session_survives() {
964	let session = fresh("test-cancel");
965	let worker = Arc::clone(&session);
966	{
967	let mut guard = worker.lock().unwrap();
968	start_command(&mut guard, "sleep 10").unwrap();
969	}
970	let handle = std::thread::spawn(move \|\| {
971	let (done, timed_out) = wait_shared_session(&worker, Duration::from_secs(30)).unwrap();
972	let mut guard = worker.lock().unwrap();
973	session_result(&mut guard, done, timed_out)
974	});
975	std::thread::sleep(Duration::from_millis(150));
976	let done = cancel_shared_session(&session).unwrap();
977	let result = handle.join().unwrap();
978	assert_eq!(done.0, 130);
979	assert_eq!(result.metadata.as_ref().unwrap()["exit_code"], 130);
980	assert!(
981	run(&session, "printf alive", Duration::from_secs(3))
982	.content
983	.contains("alive")
984	);
985	}
986
987	#[test]
988	#[cfg(unix)]
989	fn session_names_are_scoped_to_workspace() {
990	let first_workspace = tempfile::tempdir().unwrap();
991	let second_workspace = tempfile::tempdir().unwrap();
992	let first = get_or_create("shared-name", first_workspace.path()).unwrap();
993	let second = get_or_create("shared-name", second_workspace.path()).unwrap();
994	assert!(!Arc::ptr_eq(&first, &second));
995	assert!(find("shared-name", first_workspace.path()).is_ok());
996	assert!(find("shared-name", second_workspace.path()).is_ok());
997	assert!(find("shared-name", Path::new("/tmp")).is_err());
998	}
999
1000	#[test]
1001	#[cfg(unix)]
1002	fn durable_summary_marks_prior_process_shell_stale_and_preserves_history() {
1003	let workspace = tempfile::tempdir().unwrap();
1004	let canonical_workspace = workspace.path().canonicalize().unwrap();
1005	let name = format!("restart-proof-{}", Uuid::new_v4());
1006	let first = create_session(&name, workspace.path()).unwrap();
1007	let first_record = first.lock().unwrap().durable.clone();
1008	assert_eq!(first_record.state, DurableTerminalState::Idle);
1009	assert_eq!(first_record.last_known_cwd, canonical_workspace);
1010	assert!(!first_record.session_id.is_empty());
1011
1012	// Removing only the process-local registry entry models a restart.
1013	// The durable record remains, but cannot be used to reattach.
1014	sessions()
1015	.lock()
1016	.unwrap()
1017	.remove(&session_key(&name, workspace.path()));
1018	let stale = match find(&name, workspace.path()) {
1019	Ok(_) => panic!("persisted session must not be reattached"),
1020	Err(error) => error,
1021	};
1022	assert!(stale.contains("stale/lost"), "{stale}");
1023	assert!(stale.contains("start a replacement"), "{stale}");
1024
1025	let replacement = create_session(&name, workspace.path()).unwrap();
1026	let replacement = replacement.lock().unwrap();
1027	assert_ne!(replacement.durable.session_id, first_record.session_id);
1028	let previous = replacement.durable.previous.as_deref().unwrap();
1029	assert_eq!(previous.session_id, first_record.session_id);
1030	assert_eq!(previous.state, DurableTerminalState::StaleLost);
1031	assert_eq!(previous.last_known_cwd, canonical_workspace);
1032	assert_ne!(replacement.durable.runtime_nonce, "");
1033	}
1034
1035	#[test]
1036	#[cfg(unix)]
1037	fn output_is_capped_with_notice() {
1038	let session = fresh("test-output-cap");
1039	let result = run(&session, "yes x \| head -n 100000", Duration::from_secs(3));
1040	assert!(result.content.len() <= OUTPUT_LIMIT + 100);
1041	assert!(result.content.contains("output truncated"));
1042	}
1043	}
1044

1044 lines RUST