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.
This commit is contained in:
Kang
2026-10-22 17:55:59 +08:00
parent 81f03c12ff
commit ae7a8a782b
12 changed files with 750 additions and 20 deletions
@@ -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<TerminalExecutionEvent>,
reply: oneshot::Sender<Result<Value, RpcError>>,
result: Result<Value, RpcError>,
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<Self>,
) {
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);
}
@@ -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)
});
@@ -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);
@@ -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 {
@@ -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
+1 -1
View File
@@ -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;
@@ -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<String> },
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<Mutex<HashMap<String, RecordingHistorySearchEvent>>>;
type StatusMailbox = Arc<Mutex<HashMap<(String, u8), RecordingStatus>>>;
@@ -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,
@@ -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<Vec<ResolvedKeywordHighlightRule>>,
},
@@ -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();
+105 -14
View File
@@ -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<AgentCapturedOutput>,
}
@@ -85,6 +88,7 @@ pub struct AgentCaptureCancelResult {
pub exit_code: Option<i32>,
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<MatchResult<'a>> {
fn try_match_end<'a>(
&mut self,
text: &'a str,
recording_text: &mut String,
) -> Option<MatchResult<'a>> {
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);
+57 -1
View File
@@ -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) {
@@ -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");
+12
View File
@@ -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.