From ae7a8a782bb09bb202ea0e346fd9a94e460ed86e Mon Sep 17 00:00:00 2001 From: Kang Date: Thu, 22 Oct 2026 17:55:59 +0800 Subject: [PATCH] feat(recording): enhance terminal execution recording capabilities (#955) - Introduced `ExecutionRecordingEvent` to manage recording boundaries and output during terminal executions. - Updated `TerminalExecutionState` to capture and submit execution recordings, including command boundaries and output text. - Enhanced `NyaTermApp` to record automated terminal commands and manage recording state effectively. - Added tests to validate the new recording functionality, ensuring accurate capture of command execution and output. - Improved documentation on session recording to reflect new features and usage scenarios. --- .../capability_runtime/terminal_execution.rs | 48 ++- .../terminal_execution/tests.rs | 310 ++++++++++++++++++ .../shell/event_pump/session_events.rs | 7 + .../src/features/terminal/state.rs | 11 + .../terminal_runtime/view_io/input.rs | 44 ++- crates/nyaterm-desktop/src/models/mod.rs | 2 +- .../nyaterm-desktop/src/models/recording.rs | 64 ++++ .../src/models/terminal/mod.rs | 41 +++ crates/nyaterm-terminal/src/agent_capture.rs | 119 ++++++- crates/nyaterm-transport/src/recording.rs | 58 +++- .../src/recording/audit_tests.rs | 54 +++ docs/session-recording.md | 12 + 12 files changed, 750 insertions(+), 20 deletions(-) diff --git a/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution.rs b/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution.rs index 06b4b2658..72a4d7c08 100644 --- a/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution.rs +++ b/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution.rs @@ -19,6 +19,7 @@ use crate::features::NyaTermApp; use crate::features::runtime_jobs::await_blocking_result; use crate::features::terminal::ai_execution::{TerminalExecutionEvent, TerminalExecutionStatus}; use crate::models::SessionLaunchConfig; +use crate::models::recording::ExecutionRecordingEvent; use super::{CapabilityCaller, CapabilityRequest}; use crate::features::mcp::server::rpc_failure; @@ -62,6 +63,8 @@ struct TerminalExecutionCompletion { event: Option, reply: oneshot::Sender>, result: Result, + recording_tail: String, + recording_boundary: bool, } /// Owns every automatic terminal execution, independently of the visible AI scope. @@ -137,6 +140,7 @@ impl TerminalExecutionState { raw.feed(&text, Instant::now()); return AgentCaptureProcessResult { visible_text: text, + recording_text: String::new(), completed: Vec::new(), }; } @@ -157,6 +161,7 @@ impl TerminalExecutionState { } AgentCaptureProcessResult { visible_text: text, + recording_text: String::new(), completed: Vec::new(), } } @@ -223,6 +228,8 @@ impl TerminalExecutionState { return None; } let mut task = self.tasks.remove(session_id).expect("matched task"); + let recording_boundary = task.request.policy.background || task.capture.is_some(); + let mut recording_tail = String::new(); let duration_ms = task .started .elapsed() @@ -274,6 +281,7 @@ impl TerminalExecutionState { .or_default() .push(filter); if let Some(partial) = capture.cancel(&task.capture_marker_id) { + recording_tail = partial.recording_text; partial_truncated = partial.source_truncated; match &mut result { Ok(value) => { @@ -328,6 +336,8 @@ impl TerminalExecutionState { task.worker_cancellation.cancel(); Some(TerminalExecutionCompletion { event, + recording_tail, + recording_boundary, reply: task.request.reply, result: result.and_then(|value| { serde_json::to_value(value) @@ -512,7 +522,10 @@ impl NyaTermApp { .flatten(); let capture = wrapped.as_ref().map(|_| { let mut capture = AgentOutputCaptureProcessor::new(); - capture.register(capture_marker_id.clone()); + capture.register_with_end_newline( + capture_marker_id.clone(), + profile != AiExecutionProfile::Cmd, + ); capture }); let raw_capture = (profile == AiExecutionProfile::SendOnly).then(|| { @@ -555,6 +568,7 @@ impl NyaTermApp { self.sync_session_event_bridge_session_policy(&session_id); self.ensure_terminal_execution_clock(cx); if let Some(target) = target { + self.record_automated_terminal_command(&session_id, &command); let task = self .blocking_jobs .submit_task("terminal-background-execute", move |_| { @@ -619,9 +633,14 @@ impl NyaTermApp { }) .detach(); } else { + let recorded_command = wrapped.as_ref().map(|_| command.as_str()); let input = wrapped.unwrap_or_else(|| format!("{command}\r")); // The automatic writer bypasses the user-input gate, never broadcast/history. - let result = self.write_automated_terminal_input(&session_id, input.as_bytes()); + let result = self.write_automated_terminal_input( + &session_id, + input.as_bytes(), + recorded_command, + ); if result.is_err() { self.finish_terminal_execution( &session_id, @@ -653,6 +672,15 @@ impl NyaTermApp { status: TerminalExecutionStatus, cx: &mut Context, ) { + let background_output = self + .mcp + .capabilities + .terminal_executions + .tasks + .get(session_id) + .filter(|task| task.capture_marker_id == id && task.request.policy.background) + .and_then(|_| result.as_ref().ok()) + .map(|result| result.output.clone()); let Some(completion) = self .mcp .capabilities @@ -661,6 +689,22 @@ impl NyaTermApp { else { return; }; + if let Some(output) = background_output { + self.terminal + .submit_execution_recording(session_id, ExecutionRecordingEvent::Output(output)); + } + if !completion.recording_tail.is_empty() { + self.terminal.submit_execution_recording( + session_id, + ExecutionRecordingEvent::Output(completion.recording_tail), + ); + } + if completion.recording_boundary { + self.terminal.submit_execution_recording( + session_id, + ExecutionRecordingEvent::Boundary { command: None }, + ); + } if let Some(event) = completion.event { self.present_terminal_execution(event); } diff --git a/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution/tests.rs b/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution/tests.rs index 02c28e0b2..0b332538d 100644 --- a/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution/tests.rs +++ b/crates/nyaterm-desktop/src/features/capability_runtime/terminal_execution/tests.rs @@ -50,6 +50,287 @@ fn result() -> TerminalExecuteResult { } } +fn start_execution_recording( + app: &crate::features::NyaTermApp, + session_id: &str, + path: &std::path::Path, + mode: nyaterm_transport::RecordingMode, + include_input: bool, +) { + use nyaterm_transport::{ + ExistingFileBehavior, RecordingContext, RecordingProfile, RecordingRotationPolicy, + }; + app.recording + .writer() + .start( + session_id.into(), + RecordingContext { + session_id: session_id.into(), + session_name: "fixture".into(), + connection_id: None, + connection_name: None, + group_path: None, + protocol: "local".into(), + host: None, + port: None, + username: None, + started_at: time::OffsetDateTime::now_utc(), + }, + RecordingProfile { + mode, + base_path: path.parent().unwrap().into(), + path_template: "unused.log".into(), + include_timestamps: false, + include_io_labels: true, + include_session_metadata: false, + rotation: RecordingRotationPolicy::Session, + existing_file_behavior: ExistingFileBehavior::Overwrite, + include_binary_transfer_payloads: false, + include_input, + }, + Some(path.into()), + 1024 * 1024, + ) + .unwrap(); +} + +#[gpui::test] +fn automatic_execution_records_logical_commands_and_streamed_output_without_changing_raw( + cx: &mut gpui::TestAppContext, +) { + use crate::features::test_support::app_with_visible_local_session; + use crate::models::SessionLaunchConfig; + use crate::test_support::TestConfigDir; + use nyaterm_core::agent_capture::build_agent_capture_command; + use nyaterm_core::models::connection::AiExecutionProfile; + use nyaterm_transport::{RecordingMode, SessionEvent, TelnetSessionConfig}; + use std::io::Read; + use std::net::TcpListener; + cx.executor().allow_parking(); + let directory = TestConfigDir::new("nyaterm-execution-recording"); + let app = app_with_visible_local_session(cx, directory.path(), "fixture"); + for (index, (profile, mode, include_input, sensitive, ending)) in [ + ( + AiExecutionProfile::Posix, + RecordingMode::Transcript, + true, + false, + "complete", + ), + ( + AiExecutionProfile::Powershell, + RecordingMode::Transcript, + true, + false, + "complete", + ), + ( + AiExecutionProfile::Cmd, + RecordingMode::Transcript, + true, + false, + "complete", + ), + ( + AiExecutionProfile::Posix, + RecordingMode::Transcript, + false, + false, + "complete", + ), + ( + AiExecutionProfile::Posix, + RecordingMode::Transcript, + true, + true, + "complete", + ), + ( + AiExecutionProfile::Posix, + RecordingMode::Raw, + true, + false, + "complete", + ), + ( + AiExecutionProfile::Posix, + RecordingMode::Transcript, + true, + false, + "cancel", + ), + ( + AiExecutionProfile::Posix, + RecordingMode::Transcript, + true, + false, + "timeout", + ), + ( + AiExecutionProfile::SendOnly, + RecordingMode::Transcript, + true, + false, + "complete", + ), + ] + .into_iter() + .enumerate() + { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let config = TelnetSessionConfig { + host: "127.0.0.1".into(), + port: listener.local_addr().unwrap().port(), + raw_tcp: true, + ..Default::default() + }; + let (manager, id) = cx.update_entity(&app, |app, _| { + let manager = app.session.manager_handle(); + let info = manager.create_telnet_session(config.clone()).unwrap(); + let mut metadata = app.session.metadata("fixture").unwrap().clone(); + metadata.launch_config = SessionLaunchConfig::Telnet(config); + metadata.ai_execution_profile = profile; + app.session.register_session_metadata(&info.id, metadata); + app.terminal + .seed_session_view(info.id.clone(), String::new(), "UTF-8"); + (manager, info.id) + }); + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let path = directory.path().join(format!("{index}.log")); + let (mut receiver, marker, cancellation) = cx.update_entity(&app, |app, cx| { + start_execution_recording(app, &id, &path, mode, include_input); + if mode == RecordingMode::Transcript { + app.submit_terminal_decoded_output(&id, "before>".into()); + } + app.observe_recording_prompt_output(&id, if sensitive { "Password: " } else { "$ " }); + let (mut request, receiver) = execution_request("audit", "external", &id, false); + request.command = "echo fixture".into(); + request.policy.max_lines = 0; + let cancellation = request.cancellation.clone(); + app.start_terminal_execution(request, cx); + let marker = app.mcp.capabilities.terminal_executions.tasks[&id] + .capture_marker_id + .clone(); + (receiver, marker, cancellation) + }); + let wire = build_agent_capture_command(profile, &marker, "echo fixture") + .unwrap_or_else(|| "echo fixture\r".into()); + let mut sent = vec![0; wire.len()]; + stream.read_exact(&mut sent).unwrap(); + assert_eq!(sent, wire.as_bytes()); + let output = if profile == AiExecutionProfile::SendOnly { + "echo fixture\r\nfixture\r\nRouter#".to_string() + } else if ending == "complete" { + format!( + "{wire}\r\n__DF_CMD_START_{marker}__\r\necho fixture\r\n\r\n__DF_CMD_END_{marker}_0__\r\nprompt>" + ) + } else { + format!("{wire}\r\n__DF_CMD_START_{marker}__\r\npartial\r\n") + }; + let completion = cx.update_entity(&app, |app, cx| { + app.session.extend_pending_events([ + SessionEvent::OutputBoundary { + session_id: id.clone(), + token: marker.clone(), + }, + SessionEvent::Output { + session_id: id.clone(), + data: output.as_bytes().to_vec(), + }, + ]); + app.drain_session_events(cx); + if ending != "complete" { + if ending == "cancel" { + cancellation.cancel(); + } else { + app.mcp + .capabilities + .terminal_executions + .tasks + .get_mut(&id) + .unwrap() + .started -= Duration::from_secs(31); + } + app.poll_terminal_executions(cx); + app.session.extend_pending_events([SessionEvent::Output { + session_id: id.clone(), + data: format!("late\r\n__DF_CMD_END_{marker}_0__\r\nprompt>").into_bytes(), + }]); + app.drain_session_events(cx); + } else if profile == AiExecutionProfile::SendOnly { + app.finish_terminal_execution( + &id, + &marker, + Ok(result()), + TerminalExecutionStatus::Completed, + cx, + ); + } + if mode == RecordingMode::Transcript { + app.submit_terminal_decoded_output(&id, "after>".into()); + } + app.terminal.recording_output_fence()().unwrap(); + app.recording.writer().stop_complete(id.clone()).unwrap() + }); + let result = receiver.try_recv().unwrap(); + if ending == "cancel" { + assert!(result.is_err()); + } else { + assert_eq!( + result.unwrap()["timedOut"], + ending == "timeout", + "case {index}" + ); + } + if mode == RecordingMode::Raw { + assert_eq!(std::fs::read(&path).unwrap(), output.as_bytes()); + assert_eq!( + std::fs::read(completion.input_file_path.unwrap()).unwrap(), + sent + ); + } else { + let text = std::fs::read_to_string(&path).unwrap(); + assert!(!text.contains("__DF_CMD_"), "case {index}"); + assert!(!text.contains("┌ AI #"), "case {index}"); + assert!( + text.find("before>").unwrap() < text.find("after>").unwrap(), + "case {index}" + ); + if profile != AiExecutionProfile::SendOnly && ending == "complete" { + assert!(text.contains("[OUTPUT] after>\n"), "case {index}"); + } + assert_eq!( + text.matches("[INPUT] echo fixture").count(), + usize::from(include_input && !sensitive), + "case {index}" + ); + if ending == "complete" { + let expected = if profile == AiExecutionProfile::SendOnly { + "fixture" + } else { + "echo fixture" + }; + assert_eq!( + text.matches(&format!("[OUTPUT] {expected}\n")).count(), + 1, + "case {index}: {text}" + ); + } else { + assert_eq!( + text.matches("[OUTPUT] partial\n").count(), + 1, + "case {index}" + ); + assert_eq!(text.matches("[OUTPUT] late\n").count(), 1, "case {index}"); + } + } + manager.close(&id).unwrap(); + } +} + #[cfg(windows)] #[gpui::test] fn shared_entry_routes_native_codex_claude_and_external_calls_on_a_hidden_terminal( @@ -79,6 +360,13 @@ fn shared_entry_routes_native_codex_claude_and_external_calls_on_a_hidden_termin let mut metadata = app.session.metadata("visible").unwrap().clone(); metadata.ai_execution_profile = AiExecutionProfile::Cmd; app.session.register_session_metadata(&session.id, metadata); + start_execution_recording( + app, + &session.id, + &directory.path().join("execution.log"), + nyaterm_transport::RecordingMode::Transcript, + true, + ); }); let mut credentials = Vec::new(); for (source, background) in [ @@ -167,6 +455,28 @@ fn shared_entry_routes_native_codex_claude_and_external_calls_on_a_hidden_termin ); }); } + cx.update_entity(&app, |app, _| { + app.terminal.recording_output_fence()().unwrap(); + app.recording + .writer() + .stop_complete(session.id.clone()) + .unwrap(); + }); + let transcript = std::fs::read_to_string(directory.path().join("execution.log")).unwrap(); + assert!(!transcript.contains("__DF_CMD_")); + assert!(!transcript.contains("┌ AI #")); + assert_eq!( + transcript + .matches("[INPUT] echo %NYATERM_CAPTURE_FIXTURE%") + .count(), + 3 + ); + assert_eq!(transcript.matches("[OUTPUT] current-pty").count(), 3); + assert_eq!( + transcript.matches("[INPUT] echo background-value").count(), + 1 + ); + assert_eq!(transcript.matches("[OUTPUT] background-value").count(), 1); let text = cx.update_entity(&app, |app, _| { app.terminal.request_frame_text_for_test(&session.id) }); diff --git a/crates/nyaterm-desktop/src/features/shell/event_pump/session_events.rs b/crates/nyaterm-desktop/src/features/shell/event_pump/session_events.rs index 76f606426..fceff5875 100644 --- a/crates/nyaterm-desktop/src/features/shell/event_pump/session_events.rs +++ b/crates/nyaterm-desktop/src/features/shell/event_pump/session_events.rs @@ -11,6 +11,7 @@ use crate::features::shell::event_pump::helpers::{ terminal_log_plain_text, terminal_output_dropped_marker, }; use crate::features::{NyaTermApp, formatting::short_id}; +use crate::models::recording::ExecutionRecordingEvent; #[derive(Clone, Copy)] enum SessionOutputDrainStep { @@ -659,6 +660,12 @@ impl NyaTermApp { let stage_duration = stage_started_at.elapsed(); drain_timings.ai_capture += stage_duration; chunk_timings.ai_capture += stage_duration; + if !result.recording_text.is_empty() { + self.terminal.submit_execution_recording( + session_id, + ExecutionRecordingEvent::Output(result.recording_text), + ); + } if !result.visible_text.is_empty() { let stage_started_at = Instant::now(); self.submit_terminal_decoded_output(session_id, result.visible_text); diff --git a/crates/nyaterm-desktop/src/features/terminal/state.rs b/crates/nyaterm-desktop/src/features/terminal/state.rs index bca5332cf..1864a8c81 100644 --- a/crates/nyaterm-desktop/src/features/terminal/state.rs +++ b/crates/nyaterm-desktop/src/features/terminal/state.rs @@ -19,6 +19,7 @@ use super::editing_state::TerminalEditingState; use super::terminal_surface::TerminalScrollbarDragState; use super::terminal_surface_entity::TerminalSurface; use crate::features::{FontResolutionStatus, terminal::ResolvedAppearanceFont}; +use crate::models::recording::ExecutionRecordingEvent; use crate::models::{ ActionLinkMenuState, ActionLinkTooltipState, RecordingHistorySearchEvent, RecordingHistorySearchKey, TerminalFrameEvent, TerminalFramePipeline, TerminalSearchMode, @@ -197,6 +198,16 @@ pub(in crate::features) struct TerminalOverlayVisibility { } impl TerminalFeatureState { + pub(in crate::features) fn submit_execution_recording( + &self, + session_id: &str, + event: ExecutionRecordingEvent, + ) { + self.view + .frame_pipeline + .submit_execution_recording(session_id, event); + } + pub(in crate::features) fn recording_output_fence( &self, ) -> impl FnOnce() -> Result<(), String> + Send + 'static { diff --git a/crates/nyaterm-desktop/src/features/terminal/terminal_runtime/view_io/input.rs b/crates/nyaterm-desktop/src/features/terminal/terminal_runtime/view_io/input.rs index 621f9300a..99a63daa5 100644 --- a/crates/nyaterm-desktop/src/features/terminal/terminal_runtime/view_io/input.rs +++ b/crates/nyaterm-desktop/src/features/terminal/terminal_runtime/view_io/input.rs @@ -7,6 +7,7 @@ use nyaterm_core::{ }; use crate::features::NyaTermApp; +use crate::models::recording::ExecutionRecordingEvent; use crate::terminal::{ TerminalKeyMode, terminal_key_bytes_with_mode, terminal_key_release_bytes_with_mode, }; @@ -864,22 +865,43 @@ impl NyaTermApp { { return Err("An automatic command is capturing this terminal".into()); } - self.write_terminal_input_recorded_impl(session_id, bytes) + self.write_terminal_input_recorded_impl(session_id, bytes, None) } pub(in crate::features) fn write_automated_terminal_input( &mut self, session_id: &str, bytes: &[u8], + command: Option<&str>, ) -> Result<(), String> { self.commands.cancel_staged_quick_command(session_id); - self.write_terminal_input_recorded_impl(session_id, bytes) + self.write_terminal_input_recorded_impl(session_id, bytes, command) + } + + pub(in crate::features) fn record_automated_terminal_command( + &self, + session_id: &str, + command: &str, + ) { + let include = self + .recording + .writer() + .capture_policy(session_id) + .include_input + && !self.recording_input_is_sensitive(session_id); + self.terminal.submit_execution_recording( + session_id, + ExecutionRecordingEvent::Boundary { + command: include.then(|| command.to_string()), + }, + ); } fn write_terminal_input_recorded_impl( &mut self, session_id: &str, bytes: &[u8], + command: Option<&str>, ) -> Result<(), String> { if self.security.screen_locked() { return Err("screen locked".to_string()); @@ -909,6 +931,24 @@ impl NyaTermApp { // Covers typed Enter, paste, and autofill through the shared write path. self.terminal.assist.credential_prompt_awaiting_output = true; } + if let Some(command) = command { + if self.recording.writer().capture_policy(session_id).mode + == nyaterm_transport::RecordingMode::Raw + { + if self + .recording + .writer() + .capture_policy(session_id) + .include_input + && !self.recording_input_is_sensitive(session_id) + { + self.recording.write_raw_input(session_id, encoded); + } + } else { + self.record_automated_terminal_command(session_id, command); + } + return Ok(()); + } if disposition.record_logical_input && self .recording diff --git a/crates/nyaterm-desktop/src/models/mod.rs b/crates/nyaterm-desktop/src/models/mod.rs index 8c6a01150..072dce521 100644 --- a/crates/nyaterm-desktop/src/models/mod.rs +++ b/crates/nyaterm-desktop/src/models/mod.rs @@ -9,7 +9,7 @@ mod layout_state; mod navigation; mod network; mod prompts; -mod recording; +pub(crate) mod recording; mod remote; mod security; mod session; diff --git a/crates/nyaterm-desktop/src/models/recording.rs b/crates/nyaterm-desktop/src/models/recording.rs index d5fe238d3..85187420d 100644 --- a/crates/nyaterm-desktop/src/models/recording.rs +++ b/crates/nyaterm-desktop/src/models/recording.rs @@ -25,6 +25,29 @@ const RECORDING_WRITE_QUEUE_COMMAND_LIMIT: u64 = 4096; const STATUS_INTERVAL: Duration = Duration::from_millis(250); const FLUSH_INTERVAL: Duration = Duration::from_secs(1); +pub(crate) enum ExecutionRecordingEvent { + Boundary { command: Option }, + Output(String), +} + +impl std::fmt::Debug for ExecutionRecordingEvent { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(match self { + Self::Boundary { .. } => "ExecutionBoundary { .. }", + Self::Output(_) => "ExecutionOutput { .. }", + }) + } +} + +impl ExecutionRecordingEvent { + pub(crate) fn payload_bytes(&self) -> usize { + match self { + Self::Boundary { command } => command.as_ref().map_or(0, String::len), + Self::Output(text) => text.len(), + } + } +} + type SearchMailbox = Arc>>; type StatusMailbox = Arc>>; @@ -301,6 +324,29 @@ impl ReplySender { } impl RecordingWriteHandle { + pub(crate) fn write_execution_event( + &self, + session_id: &str, + mut event: ExecutionRecordingEvent, + ) { + let policy = self.capture_policy(session_id); + if policy.mode == RecordingMode::Raw { + return; + } + if !policy.include_input + && let ExecutionRecordingEvent::Boundary { command } = &mut event + { + *command = None; + } + self.enqueue_payload( + session_id, + event.payload_bytes(), + RecordingWriteCommand::Execution { + session_id: session_id.into(), + event, + }, + ); + } pub(crate) fn report_output_gap(&self, session_id: &str, bytes: usize) { self.report_dropped_payload(session_id, bytes); } @@ -671,6 +717,10 @@ impl RecordingHistorySearchKey { } #[derive(Debug)] enum RecordingWriteCommand { + Execution { + session_id: String, + event: ExecutionRecordingEvent, + }, Shutdown, Start { session_id: String, @@ -860,6 +910,20 @@ fn run_recording_writer( Err(_) => break, }; match command { + RecordingWriteCommand::Execution { session_id, event } => { + if let Some(id) = current_recording_id(&session_id, &rekeyed) { + match &event { + ExecutionRecordingEvent::Boundary { command } => { + manager.execution_boundary(id, command.as_deref()) + } + ExecutionRecordingEvent::Output(text) => { + manager.write_execution_output(id, text) + } + } + publisher.status(&manager, &handle, id, false); + } + handle.release_payload(&session_id, event.payload_bytes()); + } RecordingWriteCommand::Shutdown => break, RecordingWriteCommand::Start { session_id, diff --git a/crates/nyaterm-desktop/src/models/terminal/mod.rs b/crates/nyaterm-desktop/src/models/terminal/mod.rs index 346282737..f257c3603 100644 --- a/crates/nyaterm-desktop/src/models/terminal/mod.rs +++ b/crates/nyaterm-desktop/src/models/terminal/mod.rs @@ -30,6 +30,7 @@ use crate::{ }; use super::RecordingWriteHandle; +use crate::models::recording::ExecutionRecordingEvent; mod selection; pub(crate) use selection::{TerminalBufferCellPos, TerminalCellPos, TerminalSelection}; @@ -1644,6 +1645,38 @@ impl TerminalFramePipeline { }); } + /// Recording-only events share the frame queue with ordinary decoded output, + /// so they cannot overtake output still being processed by this worker. + pub(crate) fn submit_execution_recording( + &self, + session_id: &str, + event: ExecutionRecordingEvent, + ) { + if let ExecutionRecordingEvent::Output(text) = event { + let mut remaining = text.as_str(); + while !remaining.is_empty() { + let mut end = remaining.len().min(64 * 1024); + while !remaining.is_char_boundary(end) { + end -= 1; + } + let _ = self + .command_tx + .send(TerminalFrameCommand::ExecutionRecording { + session_id: session_id.into(), + event: ExecutionRecordingEvent::Output(remaining[..end].into()), + }); + remaining = &remaining[end..]; + } + } else { + let _ = self + .command_tx + .send(TerminalFrameCommand::ExecutionRecording { + session_id: session_id.into(), + event, + }); + } + } + pub(crate) fn submit_synthetic_output(&self, session_id: String, text: String) { if !text.is_empty() { let _ = self @@ -1914,6 +1947,10 @@ impl Default for TerminalFramePipeline { #[derive(Debug)] enum TerminalFrameCommand { + ExecutionRecording { + session_id: String, + event: ExecutionRecordingEvent, + }, SetKeywordRules { rules: Arc>, }, @@ -3555,6 +3592,7 @@ fn terminal_frame_command_priority_snapshot_insert_before(command: &TerminalFram fn terminal_frame_command_output_bytes(command: &TerminalFrameCommand) -> usize { match command { + TerminalFrameCommand::ExecutionRecording { event, .. } => event.payload_bytes(), TerminalFrameCommand::Output { data, .. } => data.len(), TerminalFrameCommand::DecodedOutput { text, .. } | TerminalFrameCommand::SyntheticOutput { text, .. } => text.len(), @@ -3714,6 +3752,9 @@ fn run_terminal_frame_processor( push_event(&event_queue, TerminalFrameEvent::Output(event)); } } + TerminalFrameCommand::ExecutionRecording { session_id, event } => { + recording_writer.write_execution_event(&session_id, event); + } TerminalFrameCommand::SyntheticOutput { session_id, text } => { if let Some(session) = sessions.get_mut(&session_id) { let started_at = Instant::now(); diff --git a/crates/nyaterm-terminal/src/agent_capture.rs b/crates/nyaterm-terminal/src/agent_capture.rs index b32d92221..ae2e6797a 100644 --- a/crates/nyaterm-terminal/src/agent_capture.rs +++ b/crates/nyaterm-terminal/src/agent_capture.rs @@ -75,6 +75,9 @@ pub struct AgentCapturedOutput { #[derive(Debug, Clone, PartialEq, Eq)] pub struct AgentCaptureProcessResult { pub visible_text: String, + /// Command output before the bounded RPC result is truncated. Never includes + /// wrapper echoes, markers, or the post-command prompt. + pub recording_text: String, pub completed: Vec, } @@ -85,6 +88,7 @@ pub struct AgentCaptureCancelResult { pub exit_code: Option, pub duration_ms: u64, pub source_truncated: bool, + pub recording_text: String, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -100,6 +104,9 @@ struct ActiveCapture { phase: CapturePhase, start_time: Instant, source_truncated: bool, + start_newline: bool, + recording_tail: String, + end_newline: bool, } /// Control strings are opaque to marker matching. CMD titles can contain copies @@ -212,6 +219,11 @@ impl AgentOutputCaptureProcessor { } pub fn register(&mut self, marker_id: String) { + self.register_with_end_newline(marker_id, true); + } + + /// POSIX and PowerShell insert a newline before END; CMD only echoes END. + pub fn register_with_end_newline(&mut self, marker_id: String, end_newline: bool) { self.active.insert( marker_id, ActiveCapture { @@ -219,6 +231,9 @@ impl AgentOutputCaptureProcessor { phase: CapturePhase::WaitingForStart, start_time: Instant::now(), source_truncated: false, + start_newline: true, + recording_tail: String::new(), + end_newline, }, ); } @@ -255,6 +270,7 @@ impl AgentOutputCaptureProcessor { exit_code: None, duration_ms: capture.start_time.elapsed().as_millis() as u64, source_truncated: capture.source_truncated, + recording_text: capture.recording_tail, }) } @@ -262,6 +278,7 @@ impl AgentOutputCaptureProcessor { if !self.has_active() { return AgentCaptureProcessResult { visible_text: text.to_string(), + recording_text: String::new(), completed: self.drain_completed(), }; } @@ -270,6 +287,7 @@ impl AgentOutputCaptureProcessor { if self.active.is_empty() { return AgentCaptureProcessResult { visible_text: filtered.text, + recording_text: String::new(), completed: self.drain_completed(), }; } @@ -285,6 +303,7 @@ impl AgentOutputCaptureProcessor { }; // Forward only terminal queries during capture so shell startup can finish. let mut visible_text = filtered.queries; + let mut recording_text = String::new(); while !remaining.is_empty() { if let Some(result) = self.try_match_start(remaining) { @@ -292,7 +311,7 @@ impl AgentOutputCaptureProcessor { continue; } - if let Some(result) = self.try_match_end(remaining) { + if let Some(result) = self.try_match_end(remaining, &mut recording_text) { visible_text.push_str(result.before); remaining = result.after; continue; @@ -301,7 +320,7 @@ impl AgentOutputCaptureProcessor { if let Some(capture_id) = self.any_in_phase(CapturePhase::Capturing) { if let Some(pos) = remaining.find(MARKER_PREFIX) { if let Some(cap) = self.active.get_mut(&capture_id) { - append_capture_output(cap, &remaining[..pos]); + append_capture_output(cap, &remaining[..pos], &mut recording_text); } let candidate = &remaining[pos..]; if self.is_possible_marker_prefix(candidate) { @@ -309,7 +328,7 @@ impl AgentOutputCaptureProcessor { remaining = ""; } else if pos == 0 { if let Some(cap) = self.active.get_mut(&capture_id) { - append_capture_output(cap, MARKER_PREFIX); + append_capture_output(cap, MARKER_PREFIX, &mut recording_text); } remaining = &remaining[MARKER_PREFIX.len()..]; } else { @@ -320,7 +339,7 @@ impl AgentOutputCaptureProcessor { .possible_marker_tail_start(remaining) .unwrap_or(remaining.len()); if let Some(cap) = self.active.get_mut(&capture_id) { - append_capture_output(cap, &remaining[..tail_start]); + append_capture_output(cap, &remaining[..tail_start], &mut recording_text); } self.pending_marker_tail.push_str(&remaining[tail_start..]); remaining = ""; @@ -352,6 +371,7 @@ impl AgentOutputCaptureProcessor { AgentCaptureProcessResult { visible_text, + recording_text, completed: self.drain_completed(), } } @@ -379,15 +399,17 @@ impl AgentOutputCaptureProcessor { }) .min_by_key(|(_, end)| *end)?; self.active.get_mut(&marker_id)?.phase = CapturePhase::Capturing; - let after_marker = &text[marker_end..]; - let after = after_marker - .strip_prefix("\r\n") - .or_else(|| after_marker.strip_prefix('\n')) - .unwrap_or(after_marker); - Some(MatchResult { before: "", after }) + Some(MatchResult { + before: "", + after: &text[marker_end..], + }) } - fn try_match_end<'a>(&mut self, text: &'a str) -> Option> { + fn try_match_end<'a>( + &mut self, + text: &'a str, + recording_text: &mut String, + ) -> Option> { let (marker_id, start, marker_end, exit_code) = self .active .iter() @@ -405,7 +427,13 @@ impl AgentOutputCaptureProcessor { }) .min_by_key(|(_, start, _, _)| *start)?; let capture = self.active.get_mut(&marker_id)?; - append_capture_output(capture, &text[..start]); + append_capture_output(capture, &text[..start], recording_text); + // Delay at most one newline so the wrapper's separator is excluded, + // even when it and the END marker arrive in different chunks. + let tail = std::mem::take(&mut capture.recording_tail); + if !capture.end_newline { + recording_text.push_str(&tail); + } self.completed.push_back(AgentCapturedOutput { marker_id: marker_id.clone(), output: std::mem::take(&mut capture.buffer).trim().to_string(), @@ -459,7 +487,29 @@ impl AgentOutputCaptureProcessor { } } -fn append_capture_output(capture: &mut ActiveCapture, text: &str) { +fn append_capture_output(capture: &mut ActiveCapture, text: &str, recording_text: &mut String) { + let text = if capture.start_newline { + let text = text.trim_start_matches('\r'); + if text.is_empty() { + return; + } + capture.start_newline = false; + text.strip_prefix('\n').unwrap_or(text) + } else { + text + }; + if !text.is_empty() { + recording_text.push_str(&std::mem::take(&mut capture.recording_tail)); + recording_text.push_str(text); + let tail_len = if recording_text.ends_with("\r\n") { + 2 + } else if recording_text.ends_with(['\r', '\n']) { + 1 + } else { + 0 + }; + capture.recording_tail = recording_text.split_off(recording_text.len() - tail_len); + } let available = MAX_CAPTURE_BYTES.saturating_sub(capture.buffer.len()); let mut end = available.min(text.len()); while !text.is_char_boundary(end) { @@ -604,6 +654,45 @@ mod tests { } } + #[test] + fn recording_stream_preserves_whitespace_and_removes_split_wrapper_framing() { + let stream = + "wrapper\r\n__DF_CMD_START_log__\r\n 猫🐱 \r\n\r\n__DF_CMD_END_log_0__\r\nprompt>"; + for split in stream.char_indices().map(|(i, _)| i) { + let mut processor = AgentOutputCaptureProcessor::new(); + processor.register("log".into()); + let first = processor.process(&stream[..split]); + let second = processor.process(&stream[split..]); + let recorded = first.recording_text + &second.recording_text; + assert_eq!(recorded, " 猫🐱 \r\n", "split {split}"); + } + } + + #[test] + fn cancellation_flushes_only_the_unrecorded_tail_and_late_markers_stay_hidden() { + let mut processor = AgentOutputCaptureProcessor::new(); + processor.register("log".into()); + let first = processor.process("__DF_CMD_START_log__\npartial\n__DF_CMD_EN"); + let mut retired = processor.retire("log"); + let cancelled = processor.cancel("log").unwrap(); + assert_eq!( + first.recording_text + &cancelled.recording_text, + "partial\n" + ); + let (late, done) = retired.process("D_log_0__\nlate output\n"); + assert!(done); + assert_eq!(late, "\nlate output\n"); + } + + #[test] + fn cmd_recording_preserves_empty_output_lines_without_an_end_separator() { + let mut processor = AgentOutputCaptureProcessor::new(); + processor.register_with_end_newline("cmd".into(), false); + let first = processor.process("__DF_CMD_START_cmd__\r\n\r"); + let second = processor.process("\n__DF_CMD_END_cmd_0__\r\nC:\\>"); + assert_eq!(first.recording_text + &second.recording_text, "\r\n"); + } + #[test] fn unrelated_and_malformed_markers_do_not_finish_or_restart_capture() { let mut processor = AgentOutputCaptureProcessor::new(); @@ -625,7 +714,9 @@ mod tests { let mut processor = AgentOutputCaptureProcessor::new(); processor.register("limit".into()); processor.process("__DF_CMD_START_limit__\n"); - processor.process(&"猫".repeat(MAX_CAPTURE_BYTES / 3 + 5)); + let body = "猫".repeat(MAX_CAPTURE_BYTES / 3 + 5); + let streamed = processor.process(&body); + assert_eq!(streamed.recording_text, body); let result = processor.process("\n__DF_CMD_END_limit_0__\nPS>"); let output = &result.completed[0]; assert!(output.source_truncated); diff --git a/crates/nyaterm-transport/src/recording.rs b/crates/nyaterm-transport/src/recording.rs index 505f48770..2cb586818 100644 --- a/crates/nyaterm-transport/src/recording.rs +++ b/crates/nyaterm-transport/src/recording.rs @@ -491,6 +491,7 @@ struct SessionCaptureState { suppress_next_newline: bool, next_line_id: u64, sanitizer: RecordingSanitizer, + execution_sanitizer: RecordingSanitizer, history_truncated: bool, history_include_input: bool, batch_timestamp: String, @@ -519,6 +520,7 @@ impl SessionCaptureState { suppress_next_newline: false, next_line_id: 1, sanitizer: RecordingSanitizer::default(), + execution_sanitizer: RecordingSanitizer::default(), history_truncated: false, history_include_input: false, batch_timestamp: chrono_timestamp(), @@ -594,6 +596,7 @@ impl SessionCaptureState { self.flush_output_lines(true); self.dropped_bytes = self.dropped_bytes.saturating_add(bytes as u64); self.sanitizer.reset(); + self.execution_sanitizer.reset(); self.pending_input_escape = None; self.live_echo_buffer.clear(); self.submitted_line_echo = None; @@ -793,13 +796,43 @@ impl SessionCaptureState { } } - for chunk in utf8_fragments(&sanitized, self.fragment_limit()) { + self.append_output(&sanitized); + } + + fn append_output(&mut self, text: &str) { + for chunk in utf8_fragments(text, self.fragment_limit()) { self.output_buffer.push_str(chunk); self.flush_output_lines(false); self.trim_records(); } } + fn execution_boundary(&mut self, command: Option<&str>) { + self.flush_output_lines(true); + self.execution_sanitizer.reset(); + if self.includes_input() + && !self + .recording + .as_ref() + .is_some_and(|file| file.profile.mode == RecordingMode::Raw) + && let Some(command) = command + { + self.batch_timestamp = chrono_timestamp(); + let text = RecordingSanitizer::default().feed(command); + for line in text.lines() { + self.append_record("INPUT", line.to_string()); + } + } + } + + fn write_execution_output(&mut self, text: &str) { + self.batch_timestamp = chrono_timestamp(); + let text = self.execution_sanitizer.feed(text); + // Background execution can overlap manual typing and output. Its output + // must neither consume keyboard echo nor share partial control strings. + self.append_output(&text); + } + fn finish(&mut self) { self.commit_partial_input(); self.flush_output_lines(true); @@ -1465,6 +1498,29 @@ impl RecordingManager { state.write_input(data); } + /// Delimit automatic execution without treating its command as typed input. + /// `None` also finishes an execution and flushes an unterminated output line. + pub fn execution_boundary(&self, session_id: &str, command: Option<&str>) { + let budget = *lock_recover(&self.memory_limit_bytes); + let mut sessions = lock_recover(&self.sessions); + let state = sessions + .entry(session_id.into()) + .or_insert_with(|| SessionCaptureState::new(budget)); + state.history_include_input = *lock_recover(&self.history_include_input); + state.set_memory_limit(budget); + state.execution_boundary(command); + } + + pub fn write_execution_output(&self, session_id: &str, text: &str) { + let budget = *lock_recover(&self.memory_limit_bytes); + let mut sessions = lock_recover(&self.sessions); + let state = sessions + .entry(session_id.into()) + .or_insert_with(|| SessionCaptureState::new(budget)); + state.set_memory_limit(budget); + state.write_execution_output(text); + } + pub fn resync_input_line(&self, session_id: &str, terminal_line: &str) { let mut sessions = lock_recover(&self.sessions); if let Some(state) = sessions.get_mut(session_id) { diff --git a/crates/nyaterm-transport/src/recording/audit_tests.rs b/crates/nyaterm-transport/src/recording/audit_tests.rs index 6999cbfc7..9ef216b09 100644 --- a/crates/nyaterm-transport/src/recording/audit_tests.rs +++ b/crates/nyaterm-transport/src/recording/audit_tests.rs @@ -60,6 +60,60 @@ fn default_capture_never_stores_no_echo_user_input() { ); } +#[test] +fn execution_boundaries_preserve_identical_output_and_separate_unterminated_lines() { + let dir = TestTempDir::new("nyaterm-recording-execution"); + for include_input in [false, true] { + let manager = RecordingManager::new(); + let (context, mut settings) = profile(dir.path(), RecordingMode::Transcript); + settings.include_input = include_input; + settings.include_io_labels = true; + let path = dir.join(format!("{include_input}.log")); + manager + .start_with_profile("audit", context, settings, Some(path.clone())) + .unwrap(); + manager.write_output("audit", "before>"); + manager.execution_boundary("audit", Some("same\nsecond")); + manager.write_execution_output("audit", "same\nsecond"); + manager.execution_boundary("audit", None); + manager.write_output("audit", "after>"); + manager.stop("audit").unwrap(); + let text = fs::read_to_string(path).unwrap(); + let expected = if include_input { + "[OUTPUT] before>\n[INPUT] same\n[INPUT] second\n[OUTPUT] same\n[OUTPUT] second\n[OUTPUT] after>\n" + } else { + "[OUTPUT] before>\n[OUTPUT] same\n[OUTPUT] second\n[OUTPUT] after>\n" + }; + assert_eq!(text, expected); + } +} + +#[test] +fn background_execution_preserves_manual_input_echo_and_partial_control_strings() { + let dir = TestTempDir::new("nyaterm-recording-interleaved-execution"); + let manager = RecordingManager::new(); + let (context, mut settings) = profile(dir.path(), RecordingMode::Transcript); + settings.include_input = true; + settings.include_io_labels = true; + let path = dir.join("interleaved.log"); + manager + .start_with_profile("audit", context, settings, Some(path.clone())) + .unwrap(); + manager.write_input("audit", b"man"); + manager.execution_boundary("audit", Some("background")); + manager.write_execution_output("audit", "man\n"); + manager.write_output("audit", "man\x1b]hidden-title"); + manager.execution_boundary("audit", None); + manager.write_output("audit", "\x07"); + manager.write_input("audit", b"ual\r"); + manager.write_output("audit", "manual\r\nmanual-result\n"); + manager.stop("audit").unwrap(); + assert_eq!( + fs::read_to_string(path).unwrap(), + "[INPUT] background\n[OUTPUT] man\n[INPUT] manual\n[OUTPUT] manual-result\n" + ); +} + #[test] fn split_escape_strings_are_removed_from_transcript() { let dir = TestTempDir::new("nyaterm-recording-split-escape"); diff --git a/docs/session-recording.md b/docs/session-recording.md index 72dce0f40..a910dbc26 100644 --- a/docs/session-recording.md +++ b/docs/session-recording.md @@ -37,6 +37,18 @@ peers. Dedicated sensitive-input adapters and terminal protocol replies are always excluded. Unknown hidden-input prompts cannot reliably be detected; this limitation is shown beside the input switch. +Automatic terminal execution records the original command as INPUT, subject to +the same input switch and sensitive-prompt checks. Internal shell wrappers and +capture markers are excluded from Transcript. Captured output is recorded +incrementally, independently of the bounded MCP result and AI card line limit; +SendOnly output continues through the ordinary recording path once. Cancellation +and timeout retain partial output, and late output keeps recording with internal +markers removed. Background SSH/local execution records its submitted command +and returned stdout/stderr without inserting the AI presentation into the log. +Raw still records only session wire bytes, including shell wrappers, and excludes +background execution that does not pass through that session. Prompt detection +does not redact arbitrary secrets embedded in commands or command output. + The binary-transfer switch selects either session bytes before transfer filters or terminal bytes after the existing X/Y/ZMODEM and trzsz filters. The Raw capture marker on terminal submissions prevents duplicate capture after decoding.