From acc1ee8d0664cfd43d8271be93e102bd544b9416 Mon Sep 17 00:00:00 2001 From: wsp1911 Date: Tue, 14 Jul 2026 10:33:35 +0800 Subject: [PATCH] feat(terminal): persist user terminal history for agents Persist rotating plain-text transcripts for user-created terminal sessions, including command metadata, exit codes, cwd changes, and shell-attributed output. Configure transcript storage for desktop and CLI, recover transcript indexes across restarts, and retain active sessions plus the most recent completed sessions. Add progressive READ_TERMINAL prompt guidance so agents can inspect local terminal evidence when command or output context is missing. --- src/apps/cli/src/main.rs | 13 + src/apps/desktop/src/api/terminal_api.rs | 14 + .../prompt_builder/prompt_builder_impl.rs | 100 ++ .../agentic/agents/prompts/agentic_mode.md | 2 + src/crates/services/terminal/Cargo.toml | 1 + .../services/terminal/src/config/types.rs | 35 + src/crates/services/terminal/src/lib.rs | 3 +- .../services/terminal/src/session/manager.rs | 195 ++- .../terminal/src/shell/integration.rs | 332 ++++- .../services/terminal/src/transcript.rs | 1150 +++++++++++++++++ 10 files changed, 1749 insertions(+), 96 deletions(-) create mode 100644 src/crates/services/terminal/src/transcript.rs diff --git a/src/apps/cli/src/main.rs b/src/apps/cli/src/main.rs index 178e6a8f1e..47027a4a03 100644 --- a/src/apps/cli/src/main.rs +++ b/src/apps/cli/src/main.rs @@ -369,11 +369,24 @@ fn terminal_scripts_dir() -> std::path::PathBuf { } async fn initialize_terminal_service() { + use bitfun_core::infrastructure::try_get_path_manager_arc; use bitfun_core::service::runtime::RuntimeManager; use bitfun_core::service::terminal::{TerminalApi, TerminalConfig}; let mut terminal_config = TerminalConfig::default(); terminal_config.shell_integration.scripts_dir = Some(terminal_scripts_dir()); + match try_get_path_manager_arc() { + Ok(path_manager) => { + terminal_config.transcript.root_dir = + Some(path_manager.user_data_dir().join("terminals")); + } + Err(error) => { + tracing::warn!( + "Failed to configure terminal transcript storage; recording is disabled: {}", + error + ); + } + } if let Ok(runtime_manager) = RuntimeManager::new() { let current_path = std::env::var("PATH").ok(); diff --git a/src/apps/desktop/src/api/terminal_api.rs b/src/apps/desktop/src/api/terminal_api.rs index 32824a95d0..5ec3b3b1bc 100644 --- a/src/apps/desktop/src/api/terminal_api.rs +++ b/src/apps/desktop/src/api/terminal_api.rs @@ -7,6 +7,7 @@ use std::sync::Arc; use tauri::{AppHandle, Emitter, State}; use tokio::sync::Mutex; +use bitfun_core::infrastructure::try_get_path_manager_arc; use bitfun_core::service::remote_ssh::workspace_state::get_remote_workspace_manager; use bitfun_core::service::runtime::RuntimeManager; use bitfun_core::service::terminal::TerminalEvent; @@ -47,6 +48,19 @@ impl TerminalState { let scripts_dir = Self::get_scripts_dir(); config.shell_integration.scripts_dir = Some(scripts_dir); + match try_get_path_manager_arc() { + Ok(path_manager) => { + config.transcript.root_dir = + Some(path_manager.user_data_dir().join("terminals")); + } + Err(error) => { + warn!( + "Failed to configure terminal transcript storage; recording is disabled: {}", + error + ); + } + } + // Prepend BitFun-managed runtime dirs to PATH so Bash/Skill commands can // run on machines without preinstalled dev tools. if let Ok(runtime_manager) = RuntimeManager::new() { diff --git a/src/crates/assembly/core/src/agentic/agents/prompt_builder/prompt_builder_impl.rs b/src/crates/assembly/core/src/agentic/agents/prompt_builder/prompt_builder_impl.rs index e7283bddc5..8fc2eb25cc 100644 --- a/src/crates/assembly/core/src/agentic/agents/prompt_builder/prompt_builder_impl.rs +++ b/src/crates/assembly/core/src/agentic/agents/prompt_builder/prompt_builder_impl.rs @@ -5,6 +5,7 @@ use crate::agentic::tools::implementations::ExecCommandTool; use crate::agentic::util::remote_workspace_layout::build_remote_workspace_layout_preview; use crate::agentic::workspace::WorkspaceBackend; use crate::agentic::WorkspaceBinding; +use crate::infrastructure::try_get_path_manager_arc; use crate::service::bootstrap::build_workspace_persona_prompt; use crate::service::config::global::GlobalConfigManager; use crate::service::config::{get_app_language_code, get_global_config_service}; @@ -33,6 +34,7 @@ const PLACEHOLDER_VISUAL_MODE: &str = "{VISUAL_MODE}"; const PLACEHOLDER_SESSION_ID: &str = "{SESSION_ID}"; const PLACEHOLDER_DEEP_RESEARCH_REPORT_LINK: &str = "{DEEP_RESEARCH_REPORT_LINK}"; const PLACEHOLDER_MEMORY_ROOT: &str = "{MEMORY_ROOT}"; +const PLACEHOLDER_READ_TERMINAL: &str = "{READ_TERMINAL}"; #[derive(Debug, Clone)] pub struct PromptBuilderContext { @@ -414,6 +416,42 @@ Output Mermaid in fenced code blocks (```mermaid) so the UI can render them. } } + fn build_terminal_transcript_prompt_guidance(&self) -> String { + if self.context.remote_execution.is_some() { + return String::new(); + } + + match try_get_path_manager_arc() { + Ok(path_manager) => { + let agents_path = path_manager + .user_data_dir() + .join("terminals") + .join("AGENTS.md"); + format!( + "## User terminal history + +The user's terminal history may contain execution evidence that is missing from the conversation. Consult it when that evidence could materially affect the task, especially when: + +- The user refers to a command, terminal operation, or result they previously ran or observed. +- The user reports a command-line, build, test, script, or process problem without providing the exact command or enough output to diagnose it. + +Use the terminal history to recover relevant evidence before guessing or asking the user to repeat information that may already be recorded. Do not inspect it routinely when the request is unrelated to terminal activity or the conversation already contains sufficient command and output context. + +For instructions on locating and reading the transcripts, read: `{}` +", + agents_path.to_string_lossy().replace('\\', "/"), + ) + } + Err(error) => { + warn!( + "Failed to build terminal transcript prompt guidance; omitting it: {}", + error + ); + String::new() + } + } + } + /// Get user language preference instruction /// /// Read app.language from global config, generate simple language instruction @@ -454,6 +492,7 @@ Do not read from, modify, create, move, or delete files outside this workspace u /// - `{CLAW_WORKSPACE}` - Claw-specific workspace ownership and boundary rules /// - `{VISUAL_MODE}` - Visual mode instruction (Mermaid diagrams, read from global config) /// - `{MEMORY_ROOT}` - BitFun memory workspace root, used by internal memory agents + /// - `{READ_TERMINAL}` - Local user terminal transcript guidance /// /// If a placeholder is not in the template, corresponding content will not be added pub async fn build_prompt_from_template(&self, template: &str) -> BitFunResult { @@ -499,6 +538,11 @@ Do not read from, modify, create, move, or delete files outside this workspace u result = result.replace(PLACEHOLDER_VISUAL_MODE, &visual_mode); } + if result.contains(PLACEHOLDER_READ_TERMINAL) { + let read_terminal = self.build_terminal_transcript_prompt_guidance(); + result = result.replace(PLACEHOLDER_READ_TERMINAL, &read_terminal); + } + // Replace {SESSION_ID} — used by deep-research Pro mode to anchor a per-session // work_dir under .bitfun/sessions/{SESSION_ID}/research/. Falls back to a // timestamp slug when no session is bound (e.g. one-shot prompt builds in tests). @@ -803,6 +847,62 @@ mod tests { assert!(!runtime_context.contains("Local BitFun client OS:")); } + #[tokio::test] + async fn local_terminal_transcript_placeholder_includes_the_agents_path() { + let context = PromptBuilderContext::new("workspace/root", None, None); + let prompt = PromptBuilder::new(context) + .build_prompt_from_template("{READ_TERMINAL}") + .await + .expect("prompt should build"); + let expected_path = crate::infrastructure::try_get_path_manager_arc() + .expect("path manager should initialize") + .user_data_dir() + .join("terminals") + .join("AGENTS.md") + .to_string_lossy() + .replace('\\', "/"); + + assert!(prompt.contains(&expected_path)); + assert!(prompt.contains( + "The user refers to a command, terminal operation, or result they previously ran or observed." + )); + assert!(prompt + .contains("The user reports a command-line, build, test, script, or process problem")); + assert!(prompt.contains("Do not inspect it routinely")); + } + + #[tokio::test] + async fn remote_terminal_transcript_placeholder_is_omitted() { + let context = PromptBuilderContext::new("/workspace/project", None, None) + .with_remote_prompt_overlay( + RemoteExecutionHints { + connection_display_name: "dev-server".to_string(), + kernel_name: "Linux".to_string(), + hostname: "devbox".to_string(), + }, + None, + ); + let prompt = PromptBuilder::new(context) + .build_prompt_from_template("before\n{READ_TERMINAL}\nafter") + .await + .expect("prompt should build"); + + assert!(!prompt.contains("User terminal transcript")); + assert!(!prompt.contains("{READ_TERMINAL}")); + assert_eq!(prompt, "before\n\nafter"); + } + + #[tokio::test] + async fn template_without_terminal_transcript_placeholder_is_unchanged() { + let context = PromptBuilderContext::new("workspace/root", None, None); + let prompt = PromptBuilder::new(context) + .build_prompt_from_template("plain template") + .await + .expect("prompt should build"); + + assert_eq!(prompt, "plain template"); + } + #[tokio::test] async fn deep_research_report_link_defaults_to_workspace_relative_path() { let context = diff --git a/src/crates/assembly/core/src/agentic/agents/prompts/agentic_mode.md b/src/crates/assembly/core/src/agentic/agents/prompts/agentic_mode.md index 485effba1e..8ec8c71b89 100644 --- a/src/crates/assembly/core/src/agentic/agents/prompts/agentic_mode.md +++ b/src/crates/assembly/core/src/agentic/agents/prompts/agentic_mode.md @@ -46,6 +46,7 @@ When presenting options, state your recommendation and reasoning, keep choices c When presenting options or plans, never include time estimates - focus on what each option involves, not how long it might take. {VISUAL_MODE} + # Doing tasks The user will primarily request you perform software engineering tasks. This includes solving bugs, adding new functionality, refactoring code, explaining code, and more. For these tasks the following steps are recommended: - Read relevant code before proposing concrete changes to it. For broad design discussion, state assumptions and inspect files before editing. @@ -114,3 +115,4 @@ IMPORTANT: Whenever you mention a file path that the user might want to open, ma {LANGUAGE_PREFERENCE} +{READ_TERMINAL} \ No newline at end of file diff --git a/src/crates/services/terminal/Cargo.toml b/src/crates/services/terminal/Cargo.toml index 9cf9dfbca5..86e3f6d04d 100644 --- a/src/crates/services/terminal/Cargo.toml +++ b/src/crates/services/terminal/Cargo.toml @@ -49,6 +49,7 @@ dashmap = { workspace = true } dirs = { workspace = true } [dev-dependencies] +tempfile = { workspace = true } [target.'cfg(windows)'.dependencies] # Windows process-tree ownership diff --git a/src/crates/services/terminal/src/config/types.rs b/src/crates/services/terminal/src/config/types.rs index e80cac81a0..8dffc98925 100644 --- a/src/crates/services/terminal/src/config/types.rs +++ b/src/crates/services/terminal/src/config/types.rs @@ -34,6 +34,10 @@ pub struct TerminalConfig { /// Terminal dimensions pub default_cols: u16, pub default_rows: u16, + + /// Persistent plain-text transcript settings for user-created terminal sessions. + #[serde(default)] + pub transcript: TerminalTranscriptConfig, } impl Default for TerminalConfig { @@ -49,6 +53,37 @@ impl Default for TerminalConfig { shell_integration: ShellIntegrationConfig::default(), default_cols: 80, default_rows: 24, + transcript: TerminalTranscriptConfig::default(), + } + } +} + +/// Persistent plain-text transcript settings for user-created terminal sessions. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(default)] +pub struct TerminalTranscriptConfig { + /// Transcript root directory. `None` disables recording so reusable terminal-core + /// consumers do not choose a product storage location themselves. + #[serde(default)] + pub root_dir: Option, + + /// Maximum size of a single append-only segment before the next write rotates it. + pub segment_size_bytes: u64, + + /// Number of recent segments retained for one terminal session. + pub retained_segments_per_session: usize, + + /// Number of recent sessions retained unless more sessions are currently active. + pub max_recent_sessions: usize, +} + +impl Default for TerminalTranscriptConfig { + fn default() -> Self { + Self { + root_dir: None, + segment_size_bytes: 4 * 1024 * 1024, + retained_segments_per_session: 4, + max_recent_sessions: 10, } } } diff --git a/src/crates/services/terminal/src/lib.rs b/src/crates/services/terminal/src/lib.rs index 2b81834113..faac410bd9 100644 --- a/src/crates/services/terminal/src/lib.rs +++ b/src/crates/services/terminal/src/lib.rs @@ -22,6 +22,7 @@ pub mod pty; pub mod runtime_port; pub mod session; pub mod shell; +mod transcript; // Re-export main types for convenience pub use api::{ @@ -29,7 +30,7 @@ pub use api::{ ExecuteCommandResponse, GetHistoryRequest, GetHistoryResponse, ResizeRequest, SendCommandRequest, SessionResponse, ShellInfo, SignalRequest, TerminalApi, WriteRequest, }; -pub use config::{ShellConfig, TerminalConfig}; +pub use config::{ShellConfig, TerminalConfig, TerminalTranscriptConfig}; pub use events::{TerminalEvent, TerminalEventEmitter}; pub use exec::{ get_global_exec_process_manager, ExecCommandRequest as LocalExecCommandRequest, diff --git a/src/crates/services/terminal/src/session/manager.rs b/src/crates/services/terminal/src/session/manager.rs index ba7a6d6011..89d1933b0a 100644 --- a/src/crates/services/terminal/src/session/manager.rs +++ b/src/crates/services/terminal/src/session/manager.rs @@ -18,12 +18,27 @@ use crate::shell::{ CommandState, ScriptsManager, ShellDetector, ShellIntegration, ShellIntegrationEvent, ShellIntegrationManager, ShellType, }; +use crate::transcript::TranscriptRecorder; use crate::{TerminalError, TerminalResult}; use super::{SessionSource, SessionStatus, TerminalSession}; const COMMAND_TIMEOUT_INTERRUPT_GRACE_MS: Duration = Duration::from_millis(500); +async fn prepare_shell_integration_input( + session_integrations: &Arc>>, + session_id: &str, + clear_output: bool, +) { + let mut integrations = session_integrations.write().await; + if let Some(integration) = integrations.get_mut(session_id) { + integration.notify_input_written(); + if clear_output { + integration.clear_output(); + } + } +} + /// Why a command stream reached completion. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -170,6 +185,9 @@ pub struct SessionManager { /// Per-session output taps for real-time output streaming output_taps: Arc>>>, + + /// Persistent plain-text transcripts for manually created terminal sessions. + transcript_recorder: Option, } impl SessionManager { @@ -186,6 +204,7 @@ impl SessionManager { let integration_manager = Arc::new(ShellIntegrationManager::new()); let binding = Arc::new(super::TerminalSessionBinding::new()); let output_taps = Arc::new(DashMap::new()); + let transcript_recorder = TranscriptRecorder::from_config(&config.transcript); let manager = Self { config, @@ -198,6 +217,7 @@ impl SessionManager { binding, scripts_manager, output_taps, + transcript_recorder, }; // Start event forwarding @@ -222,6 +242,7 @@ impl SessionManager { let pty_to_session = self.pty_to_session.clone(); let session_integrations = self.session_integrations.clone(); let output_taps = self.output_taps.clone(); + let transcript_recorder = self.transcript_recorder.clone(); tokio::spawn(async move { loop { @@ -234,15 +255,15 @@ impl SessionManager { PtyServiceEvent::ResizeCompleted { id, .. } => *id, }; - // Retry the pty_to_session lookup a few times for - // non-Data events. create_session sets the mapping - // AFTER create_process returns, but event forwarding - // can deliver ProcessReady before the mapping exists. + // `create_session` can receive PTY output before it has stored the + // PTY-to-session mapping. Retry every event type so initial shell output + // reaches both terminal consumers and the durable transcript. The mapping is + // inserted only after transcript setup has completed. let session_id = { let mapping = pty_to_session.read().await; match mapping.get(&pty_id).cloned() { Some(sid) => Some(sid), - None if !matches!(event, PtyServiceEvent::ProcessData { .. }) => { + None => { drop(mapping); let mut found = None; for _ in 0..50 { @@ -255,7 +276,6 @@ impl SessionManager { } found } - None => None, } }; @@ -264,23 +284,75 @@ impl SessionManager { PtyServiceEvent::ProcessData { data, .. } => { let data_str = String::from_utf8_lossy(&data).to_string(); - // Update last activity and record to history - if let Some(session) = sessions.write().await.get_mut(&session_id) { + // Update last activity and record to history. + let is_manual_session = if let Some(session) = + sessions.write().await.get_mut(&session_id) + { session.touch(); // Record output to history for frontend recovery session.add_output(&data_str); - } + session.source == SessionSource::Manual + } else { + false + }; - // Process through shell integration - let si_events = { + // Process through shell integration. + let (has_shell_integration, si_events) = { let mut integrations = session_integrations.write().await; if let Some(integration) = integrations.get_mut(&session_id) { - integration.process_data(&data_str) + (true, integration.process_data(&data_str)) } else { - Vec::new() + (false, Vec::new()) } }; + if is_manual_session { + if let Some(recorder) = &transcript_recorder { + if has_shell_integration { + for event in &si_events { + match event { + ShellIntegrationEvent::CommandStarted { + command, + .. + } => match recorder + .record_command(&session_id, command) + { + Ok(()) => {} + Err(error) => { + warn!( + "Failed to record terminal transcript command: session_id={} error={}", + session_id, error + ); + } + }, + ShellIntegrationEvent::OutputData { + data, + .. + } => match recorder + .record_output(&session_id, data) + { + Ok(()) => {} + Err(error) => { + warn!( + "Failed to record terminal transcript output: session_id={} error={}", + session_id, error + ); + } + }, + _ => {} + } + } + } else if let Err(error) = + recorder.record_output(&session_id, &data_str) + { + warn!( + "Failed to record terminal transcript output: session_id={} error={}", + session_id, error + ); + } + } + } + // Emit shell integration events as terminal events after // releasing the integration map lock. for si_event in si_events { @@ -301,6 +373,20 @@ impl SessionManager { command_id, exit_code, } => { + if is_manual_session { + if let (Some(recorder), Some(exit_code)) = + (&transcript_recorder, exit_code) + { + if let Err(error) = recorder + .record_exit_code(&session_id, exit_code) + { + warn!( + "Failed to record terminal transcript exit code: session_id={} error={}", + session_id, error + ); + } + } + } let _ = event_emitter .emit(TerminalEvent::CommandFinished { session_id: session_id.clone(), @@ -310,6 +396,18 @@ impl SessionManager { .await; } ShellIntegrationEvent::CwdChanged { cwd } => { + if is_manual_session { + if let Some(recorder) = &transcript_recorder { + if let Err(error) = recorder + .record_cwd_changed(&session_id, &cwd) + { + warn!( + "Failed to record terminal transcript cwd change: session_id={} error={}", + session_id, error + ); + } + } + } if let Some(session) = sessions.write().await.get_mut(&session_id) { @@ -352,9 +450,27 @@ impl SessionManager { } } PtyServiceEvent::ProcessExit { exit_code, .. } => { - // Update session - if let Some(session) = sessions.write().await.get_mut(&session_id) { - session.set_exited(exit_code.map(|c| c as i32)); + // Update session. + let is_manual_session = if let Some(session) = + sessions.write().await.get_mut(&session_id) + { + session.set_exited(exit_code.map(|code| code as i32)); + session.source == SessionSource::Manual + } else { + false + }; + if is_manual_session { + if let Some(recorder) = &transcript_recorder { + if let Err(error) = recorder.finish_session( + &session_id, + exit_code.map(|code| code as i32), + ) { + warn!( + "Failed to finish terminal transcript: session_id={} error={}", + session_id, error + ); + } + } } TerminalEvent::Exit { @@ -642,7 +758,19 @@ impl SessionManager { } } - // Store PTY to session mapping + // Initialize durable capture before exposing this PTY to event forwarding, so a + // fast shell's initial prompt cannot be recorded before its transcript exists. + if let Some(recorder) = &self.transcript_recorder { + if let Err(error) = recorder.start_session(&session) { + warn!( + "Failed to start terminal transcript: session_id={} error={}", + session_id, error + ); + } + } + + // Store PTY to session mapping after transcript setup. Event forwarding retries early + // PTY events until this mapping is available. { let mut mapping = self.pty_to_session.write().await; mapping.insert(pty_id, session_id.clone()); @@ -979,14 +1107,6 @@ impl SessionManager { // Generate command ID let command_id = uuid::Uuid::new_v4().to_string(); - // Clear any previous output - { - let mut integrations = session_integrations.write().await; - if let Some(integration) = integrations.get_mut(&session_id) { - integration.clear_output(); - } - } - // Send started event send(CommandStreamEvent::Started { command_id: command_id.clone(), @@ -1000,6 +1120,10 @@ impl SessionManager { format!("{}\r", command) }; + // End any previous command's late-output attribution before the PTY can + // echo or render this new input. + prepare_shell_integration_input(&session_integrations, &session_id, true).await; + // Send the command if let Err(e) = pty_service.write(pty_id, cmd_to_send.as_bytes()).await { send(CommandStreamEvent::Error { @@ -1295,6 +1419,12 @@ impl SessionManager { .ok_or_else(|| TerminalError::SessionNotFound(session_id.to_string()))? }; + if !data.is_empty() { + // Clear post-command attribution before the PTY can emit input echo, + // PSReadLine predictions, or other rendering caused by this input. + prepare_shell_integration_input(&self.session_integrations, session_id, false).await; + } + self.pty_service.write(pty_id, data).await } @@ -1346,14 +1476,14 @@ impl SessionManager { /// Close a session pub async fn close_session(&self, session_id: &str, immediate: bool) -> TerminalResult<()> { - let pty_id = { + let (pty_id, is_manual_session) = { let mut sessions = self.sessions.write().await; let session = sessions .get_mut(session_id) .ok_or_else(|| TerminalError::SessionNotFound(session_id.to_string()))?; session.status = SessionStatus::Terminating; - session.pty_id + (session.pty_id, session.source == SessionSource::Manual) }; // Shutdown PTY if exists @@ -1367,6 +1497,17 @@ impl SessionManager { self.pty_service.shutdown(pty_id, immediate).await?; } + if is_manual_session { + if let Some(recorder) = &self.transcript_recorder { + if let Err(error) = recorder.finish_session(session_id, None) { + warn!( + "Failed to finish terminal transcript: session_id={} error={}", + session_id, error + ); + } + } + } + // Remove shell integration { let mut integrations = self.session_integrations.write().await; diff --git a/src/crates/services/terminal/src/shell/integration.rs b/src/crates/services/terminal/src/shell/integration.rs index 6c5c05012f..4ee461e1c2 100644 --- a/src/crates/services/terminal/src/shell/integration.rs +++ b/src/crates/services/terminal/src/shell/integration.rs @@ -56,10 +56,7 @@ pub enum CommandState { } impl CommandState { - /// Check if we should still collect output (executing or just finished) - /// - /// Note: This only checks the state itself. `ShellIntegration::should_collect_output()` - /// also considers the `post_command_collecting` flag for ConPTY late output. + /// Check if the command lifecycle can still produce command output. pub fn should_collect_output(&self) -> bool { matches!( self, @@ -68,6 +65,35 @@ impl CommandState { } } +/// Attribution state for output rendered after a command has semantically finished. +/// +/// PowerShell on Windows can emit the D/A/B integration markers before ConPTY +/// delivers the command's final rendered output. Keeping the command ID in this +/// state lets us recover that output without treating later input rendering as +/// part of the completed command. +#[derive(Debug, Clone, PartialEq, Default)] +enum PostCommandCapture { + #[default] + Inactive, + /// CommandFinished was received, but PromptStart has not been observed yet. + AwaitingPrompt { command_id: String }, + /// PromptStart was received; inspect the A-to-B region for prompt text. + DetectingReorder { command_id: String }, + /// No prompt text appeared between A and B, so trailing text is late output. + CollectingLateOutput { command_id: String }, +} + +impl PostCommandCapture { + fn command_id_for_output(&self) -> Option<&str> { + match self { + Self::AwaitingPrompt { command_id } | Self::CollectingLateOutput { command_id } => { + Some(command_id) + } + Self::Inactive | Self::DetectingReorder { .. } => None, + } + } +} + /// Event emitted by shell integration #[derive(Debug, Clone)] pub enum ShellIntegrationEvent { @@ -84,6 +110,8 @@ pub enum ShellIntegrationEvent { PropertyChanged { key: String, value: String }, /// Output data received during command execution OutputData { command_id: String, data: String }, + /// Plain terminal text with OSC integration control sequences removed. + PlainOutput { data: String }, } /// Shell integration parser and state tracker @@ -112,16 +140,9 @@ pub struct ShellIntegration { last_exit_code: Option, /// Flag indicating a command just finished (for output collection) command_just_finished: bool, - /// Flag for collecting late output after CommandFinished. - /// On Windows, ConPTY may deliver rendered output AFTER shell integration - /// sequences (CommandFinished/PromptStart/CommandInputStart). This flag - /// keeps output collection active until the next CommandExecutionStart. - post_command_collecting: bool, - /// When true, we are between PromptStart and CommandInputStart, - /// checking whether prompt text exists to detect ConPTY reordering. - detecting_conpty_reorder: bool, - /// Buffer for plain text that was NOT collected by shell integration - /// (i.e., output received while `should_collect()` returned false). + /// Attribution for output rendered after CommandFinished. + post_command_capture: PostCommandCapture, + /// Buffer for plain text that was not attributed to a command. /// This captures the terminal state after command execution, including /// prompts (e.g., `$ `, `dquote> `) and other non-command output. /// Cleared when a new command starts executing. @@ -144,8 +165,7 @@ impl ShellIntegration { in_osc: false, last_exit_code: None, command_just_finished: false, - post_command_collecting: false, - detecting_conpty_reorder: false, + post_command_capture: PostCommandCapture::Inactive, recent_plain_output: String::new(), } } @@ -185,11 +205,23 @@ impl ShellIntegration { self.has_rich_detection } - /// Check if output should be collected, considering both state and post-command flag. - /// On Windows ConPTY, rendered output may arrive after shell integration sequences - /// have already transitioned the state to Prompt/Input. - fn should_collect(&self) -> bool { - self.state.should_collect_output() || self.post_command_collecting + /// Resolve the command that should own text at the current parser position. + fn output_command_id(&self) -> Option<&str> { + match self.state { + CommandState::Executing => self.current_command_id.as_deref(), + CommandState::Finished { .. } | CommandState::Prompt | CommandState::Input => { + self.post_command_capture.command_id_for_output() + } + CommandState::Idle => None, + } + } + + /// Stop attributing post-command rendering before new input is written to the PTY. + /// + /// This intentionally leaves an actively executing command untouched so interactive + /// stdin does not stop collection of that command's later output. + pub fn notify_input_written(&mut self) { + self.post_command_capture = PostCommandCapture::Inactive; } /// Get accumulated output for current command @@ -213,6 +245,7 @@ impl ShellIntegration { pub fn process_data(&mut self, data: &str) -> Vec { let mut events = Vec::new(); let mut plain_output = String::new(); + let mut transcript_output = String::new(); let mut chars = data.chars().peekable(); while let Some(ch) = chars.next() { @@ -235,16 +268,12 @@ impl ShellIntegration { OscSequence::CommandFinished { .. } | OscSequence::PromptStart ); if should_flush && !plain_output.is_empty() { - if self.should_collect() { + if let Some(command_id) = self.output_command_id().map(str::to_owned) { self.output_buffer.push_str(&plain_output); - if let Some(cmd_id) = &self.current_command_id { - events.push(ShellIntegrationEvent::OutputData { - command_id: cmd_id.clone(), - data: std::mem::take(&mut plain_output), - }); - } else { - plain_output.clear(); - } + events.push(ShellIntegrationEvent::OutputData { + command_id, + data: std::mem::take(&mut plain_output), + }); } else { // Not collecting output (e.g., shell is showing prompt // or in continuation mode). Capture this text so the @@ -255,16 +284,22 @@ impl ShellIntegration { } // ConPTY reorder detection: at CommandInputStart, if no - // prompt text accumulated since PromptStart, ConPTY sent - // the sequences before the rendered output. Re-enable - // post-command collection so late output is captured. - if self.detecting_conpty_reorder - && matches!(seq, OscSequence::CommandInputStart) - { - if plain_output.is_empty() { - self.post_command_collecting = true; - } - self.detecting_conpty_reorder = false; + // prompt text accumulated since PromptStart, the integration + // markers overtook the rendered output. Preserve the finished + // command ID until the host writes the next input. + if matches!(seq, OscSequence::CommandInputStart) { + let capture = std::mem::take(&mut self.post_command_capture); + self.post_command_capture = match capture { + PostCommandCapture::DetectingReorder { command_id } + if plain_output.is_empty() => + { + PostCommandCapture::CollectingLateOutput { command_id } + } + PostCommandCapture::DetectingReorder { .. } => { + PostCommandCapture::Inactive + } + other => other, + }; } if let Some(event) = self.handle_sequence(seq) { @@ -286,31 +321,35 @@ impl ShellIntegration { } else { // Not an OSC sequence, include the ESC in output plain_output.push(ch); + transcript_output.push(ch); } } else { plain_output.push(ch); + transcript_output.push(ch); } } - // Accumulate plain output if we should collect output. - // Continue collecting after Finished via post_command_collecting flag, - // because ConPTY may deliver rendered output after shell integration sequences. + // Accumulate text only when it has a concrete command owner. This includes + // ConPTY late output after the parser has already transitioned to Prompt/Input. if !plain_output.is_empty() { - if self.should_collect() { + if let Some(command_id) = self.output_command_id().map(str::to_owned) { self.output_buffer.push_str(&plain_output); - - if let Some(cmd_id) = &self.current_command_id { - events.push(ShellIntegrationEvent::OutputData { - command_id: cmd_id.clone(), - data: plain_output, - }); - } + events.push(ShellIntegrationEvent::OutputData { + command_id, + data: plain_output, + }); } else { // Not collecting output — capture as recent terminal state self.recent_plain_output.push_str(&plain_output); } } + if !transcript_output.is_empty() { + events.push(ShellIntegrationEvent::PlainOutput { + data: transcript_output, + }); + } + events } @@ -398,16 +437,16 @@ impl ShellIntegration { fn handle_sequence(&mut self, seq: OscSequence) -> Option { match seq { OscSequence::PromptStart => { - // When we see the next prompt, the previous command is truly done - // Clear all state from previous command - if self.post_command_collecting { - // Temporarily disable post-command collection. - // If no prompt text appears between PromptStart and - // CommandInputStart, ConPTY reordering is detected and - // collection will be re-enabled at CommandInputStart. - self.post_command_collecting = false; - self.detecting_conpty_reorder = true; - } + // Temporarily stop collection while inspecting the A-to-B prompt region. + // If that region is empty, CommandInputStart will restore attribution + // using the retained finished-command ID. + let capture = std::mem::take(&mut self.post_command_capture); + self.post_command_capture = match capture { + PostCommandCapture::AwaitingPrompt { command_id } => { + PostCommandCapture::DetectingReorder { command_id } + } + _ => PostCommandCapture::Inactive, + }; self.current_command_id = None; self.current_command = None; self.state = CommandState::Prompt; @@ -424,8 +463,7 @@ impl ShellIntegration { // Clear previous command's exit code when new command starts self.last_exit_code = None; self.command_just_finished = false; - self.post_command_collecting = false; - self.detecting_conpty_reorder = false; + self.post_command_capture = PostCommandCapture::Inactive; // Generate command ID if we have a command if self.current_command.is_some() { @@ -447,11 +485,18 @@ impl ShellIntegration { // Save exit code - this survives state transitions self.last_exit_code = exit_code; self.command_just_finished = true; - // Keep collecting output after finish — ConPTY may deliver - // rendered output after the shell integration sequences. - self.post_command_collecting = true; - - // Emit event but keep command_id for output collection + // Retain the command ID separately because PromptStart clears the + // active command before ConPTY necessarily delivers rendered output. + self.post_command_capture = self + .current_command_id + .as_ref() + .map(|command_id| PostCommandCapture::AwaitingPrompt { + command_id: command_id.clone(), + }) + .unwrap_or(PostCommandCapture::Inactive); + + // Emit event but keep current_command_id until PromptStart so output + // arriving between D and A remains attributable. let event = self.current_command_id.as_ref().map(|cmd_id| { ShellIntegrationEvent::CommandFinished { command_id: cmd_id.clone(), @@ -732,6 +777,157 @@ mod tests { assert_eq!(integration.get_recent_plain_output(), "$ "); } + #[test] + fn conpty_reordered_output_retains_finished_command_id() { + let mut integration = ShellIntegration::new(); + + integration.process_data("\x1b]633;E;ls;nonce123\x07"); + let started_events = integration.process_data("\x1b]633;C\x07\r\n"); + let command_id = started_events + .iter() + .find_map(|event| match event { + ShellIntegrationEvent::CommandStarted { command_id, .. } => { + Some(command_id.clone()) + } + _ => None, + }) + .expect("command should start"); + + integration.process_data("first output\r\n"); + let events = integration.process_data(concat!( + "\x1b[?25l", + "\x1b]633;D;0\x07", + "\x1b]633;A\x07", + "\x1b]633;P;Cwd=C:\\\\workspace\x07", + "\x1b]633;B\x07", + "late output\r\nPS C:\\\\workspace> " + )); + + assert!(events.iter().any(|event| { + matches!( + event, + ShellIntegrationEvent::OutputData { + command_id: output_command_id, + data, + } if output_command_id == &command_id && data.contains("late output") + ) + })); + assert!(integration.get_output().contains("late output")); + } + + #[test] + fn conpty_reordered_output_can_arrive_in_a_later_chunk() { + let mut integration = ShellIntegration::new(); + + integration.process_data("\x1b]633;E;echo test;nonce123\x07"); + let started_events = integration.process_data("\x1b]633;C\x07"); + let command_id = started_events + .iter() + .find_map(|event| match event { + ShellIntegrationEvent::CommandStarted { command_id, .. } => { + Some(command_id.clone()) + } + _ => None, + }) + .expect("command should start"); + + integration.process_data( + "\x1b]633;D;0\x07\x1b]633;A\x07\x1b]633;P;Cwd=C:\\\\workspace\x07\x1b]633;B\x07", + ); + let events = integration.process_data("test\r\nPS C:\\\\workspace> "); + + assert!(events.iter().any(|event| { + matches!( + event, + ShellIntegrationEvent::OutputData { + command_id: output_command_id, + data, + } if output_command_id == &command_id && data.contains("test") + ) + })); + } + + #[test] + fn input_written_stops_post_command_output_attribution() { + let mut integration = ShellIntegration::new(); + + integration.process_data("\x1b]633;E;echo test;nonce123\x07"); + integration.process_data("\x1b]633;C\x07"); + integration.process_data( + "\x1b]633;D;0\x07\x1b]633;A\x07\x1b]633;P;Cwd=C:\\\\workspace\x07\x1b]633;B\x07", + ); + + let late_events = integration.process_data("test\r\nPS C:\\\\workspace> "); + assert!(late_events + .iter() + .any(|event| matches!(event, ShellIntegrationEvent::OutputData { .. }))); + + integration.notify_input_written(); + let input_events = integration.process_data("echo next\x1b[38;2;128;128;128m prediction"); + + assert!(!input_events + .iter() + .any(|event| matches!(event, ShellIntegrationEvent::OutputData { .. }))); + assert!(input_events.iter().any(|event| { + matches!(event, ShellIntegrationEvent::PlainOutput { data } if data.contains("prediction")) + })); + } + + #[test] + fn input_written_does_not_stop_executing_command_output() { + let mut integration = ShellIntegration::new(); + + integration.process_data("\x1b]633;E;interactive;nonce123\x07"); + integration.process_data("\x1b]633;C\x07"); + integration.notify_input_written(); + let events = integration.process_data("continued output\r\n"); + + assert!(events.iter().any(|event| { + matches!(event, ShellIntegrationEvent::OutputData { data, .. } if data == "continued output\r\n") + })); + } + + #[test] + fn plain_output_omits_integration_control_sequences() { + let mut integration = ShellIntegration::new(); + + let events = integration.process_data("PS> echo hello\r\n\x1b]633;D;0\x07hello\r\n"); + let plain_output: String = events + .into_iter() + .filter_map(|event| match event { + ShellIntegrationEvent::PlainOutput { data } => Some(data), + _ => None, + }) + .collect(); + + assert_eq!(plain_output, "PS> echo hello\r\nhello\r\n"); + } + + #[test] + fn command_output_excludes_pre_execution_input_rendering() { + let mut integration = ShellIntegration::new(); + let input_events = + integration.process_data("PS> echo hello\x1b[?25l\x1b[38;2;128;128;128m prediction"); + + assert!(input_events.iter().any(|event| { + matches!(event, ShellIntegrationEvent::PlainOutput { data } if data.contains("prediction")) + })); + assert!(!input_events + .iter() + .any(|event| matches!(event, ShellIntegrationEvent::OutputData { .. }))); + + let command_events = + integration.process_data("\x1b]633;E;echo hello;nonce123\x07\x1b]633;C\x07"); + assert!(command_events.iter().any(|event| { + matches!(event, ShellIntegrationEvent::CommandStarted { command, .. } if command == "echo hello") + })); + + let output_events = integration.process_data("hello\r\n"); + assert!(output_events.iter().any(|event| { + matches!(event, ShellIntegrationEvent::OutputData { data, .. } if data == "hello\r\n") + })); + } + #[test] fn continuation_prompt_is_recorded_as_recent_plain_output() { let mut integration = ShellIntegration::new(); diff --git a/src/crates/services/terminal/src/transcript.rs b/src/crates/services/terminal/src/transcript.rs new file mode 100644 index 0000000000..cf3a5f190d --- /dev/null +++ b/src/crates/services/terminal/src/transcript.rs @@ -0,0 +1,1150 @@ +//! Persistent plain-text transcripts for user-created terminal sessions. +//! For terminals with shell integration, the remaining text is command output: prompt and command-input rendering are omitted. +//! Terminals without shell integration retain raw terminal text. + +use std::collections::HashMap; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::SystemTime; + +use chrono::{DateTime, SecondsFormat, Utc}; +use rand::Rng; +use serde::{Deserialize, Serialize}; + +use crate::config::TerminalTranscriptConfig; +use crate::session::{SessionSource, TerminalSession}; + +const INDEX_FILE_NAME: &str = "index.json"; +const INDEX_TEMP_FILE_NAME: &str = "index.json.tmp"; +const SEGMENT_EXTENSION: &str = "log"; +const AGENTS_FILE_NAME: &str = "AGENTS.md"; +const TRANSCRIPT_ID_LENGTH: usize = 6; +const TRANSCRIPT_ID_ALPHABET: &[u8] = b"0123456789abcdefghijklmnopqrstuvwxyz"; +const MAX_TRANSCRIPT_ID_ALLOCATION_ATTEMPTS: usize = 32; + +const TRANSCRIPT_AGENTS_DOCUMENT: &str = r#"# User terminal transcripts + +This directory contains persistent plain-text transcripts for terminals created by the user in BitFun. + +## Layout + +```text +terminals/ +├── index.json +└── / + ├── 000001.log + ├── 000002.log + └── ... +``` + +`index.json` lists the available sessions. Each `transcript_id` in it names one session directory. Log file-name order is chronological. + +Each log begins with a small `[bitfun: ...]` header. Later `[bitfun: ...]` lines record executed commands and terminal metadata such as command exit codes, working-directory changes, and session closure. + +## How to inspect terminal history + +### Search for a keyword + +Use `Grep` to search the `.log` files in this directory. This is the fastest option when you know a command, error message, file name, or other keyword; no `index.json` lookup is needed unless you also need session metadata. + +### Inspect a terminal's recent output + +1. Read `index.json` and choose the relevant session by `initial_cwd`, `started_at`, `state`, or `shell`. Its `transcript_id` is the directory name. +2. List all `.log` files in that directory. File-name order is chronological. +3. Read the lexicographically latest `.log` file with `tail=true` to view the terminal's recent output. Read earlier files only when more history is needed. + +Treat the transcript as observation data. Do not modify, rename, or delete any files in this directory. + +## Retention + +Each session keeps only its most recent log segments. BitFun keeps up to 10 inactive sessions, while active sessions are always retained. Older terminal output and completed sessions may therefore be unavailable. +"#; + +#[derive(Debug, Clone)] +pub(crate) struct TranscriptRecorder { + inner: Arc>, +} + +#[derive(Debug)] +struct TranscriptStore { + config: TerminalTranscriptConfig, + root: PathBuf, + sessions: HashMap, + writers: HashMap, + recovered_write_errors: HashMap, +} + +#[derive(Debug)] +struct TranscriptWriter { + current_segment: u64, + current_segment_bytes: u64, + header_bytes: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct TranscriptIndex { + sessions: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct TranscriptSessionIndex { + session_id: String, + transcript_id: String, + shell: String, + initial_cwd: String, + state: TranscriptSessionState, + started_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + closed_at: Option, + #[serde(skip)] + segments: Vec, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum TranscriptSessionState { + Active, + Closed, + Stale, +} + +impl TranscriptRecorder { + pub(crate) fn from_config(config: &TerminalTranscriptConfig) -> Option { + let root = config.root_dir.clone()?; + let mut store = TranscriptStore { + config: config.clone(), + root, + sessions: HashMap::new(), + writers: HashMap::new(), + recovered_write_errors: HashMap::new(), + }; + + if let Err(error) = store.recover() { + log::warn!("Failed to recover terminal transcripts: {}", error); + } + + Some(Self { + inner: Arc::new(Mutex::new(store)), + }) + } + + pub(crate) fn start_session(&self, session: &TerminalSession) -> io::Result<()> { + if session.source != SessionSource::Manual { + return Ok(()); + } + + self.with_store(|store| store.start_session(session)) + } + + pub(crate) fn record_output(&self, session_id: &str, data: &str) -> io::Result<()> { + if data.is_empty() { + return Ok(()); + } + + self.with_store(|store| store.append(session_id, data)) + } + + pub(crate) fn record_command(&self, session_id: &str, command: &str) -> io::Result<()> { + if command.trim().is_empty() { + return Ok(()); + } + + self.record_output( + session_id, + &format!("\n[bitfun: command={}]\n", one_line(command)), + ) + } + + pub(crate) fn record_exit_code(&self, session_id: &str, exit_code: i32) -> io::Result<()> { + self.record_output(session_id, &format!("\n[bitfun: exit_code={exit_code}]\n")) + } + + pub(crate) fn record_cwd_changed(&self, session_id: &str, cwd: &str) -> io::Result<()> { + self.record_output(session_id, &format!("\n[bitfun: cwd={cwd}]\n")) + } + + pub(crate) fn finish_session( + &self, + session_id: &str, + exit_code: Option, + ) -> io::Result<()> { + self.with_store(|store| store.finish_session(session_id, exit_code)) + } + + fn with_store( + &self, + operation: impl FnOnce(&mut TranscriptStore) -> io::Result, + ) -> io::Result { + let mut store = self.inner.lock().map_err(|_| { + io::Error::new( + io::ErrorKind::Other, + "terminal transcript recorder lock is poisoned", + ) + })?; + operation(&mut store) + } +} + +impl TranscriptStore { + fn recover(&mut self) -> io::Result<()> { + fs::create_dir_all(&self.root)?; + self.write_agents_document()?; + + let index_path = self.index_path(); + let mut changed = false; + if let Ok(contents) = fs::read_to_string(&index_path) { + match serde_json::from_str::(&contents) { + Ok(index) => { + for mut session in index.sessions { + if !is_safe_session_id(&session.session_id) + || !is_transcript_id(&session.transcript_id) + { + log::warn!( + "Ignoring terminal transcript index entry with invalid IDs: session_id={} transcript_id={}", + one_line(&session.session_id), + one_line(&session.transcript_id) + ); + changed = true; + continue; + } + if session.state == TranscriptSessionState::Active { + session.state = TranscriptSessionState::Stale; + changed = true; + } + if self.sessions.contains_key(&session.session_id) + || self + .sessions + .values() + .any(|existing| existing.transcript_id == session.transcript_id) + { + log::warn!("Ignoring duplicate terminal transcript index entry"); + changed = true; + continue; + } + self.sessions.insert(session.session_id.clone(), session); + } + } + Err(error) => { + log::warn!("Ignoring unreadable terminal transcript index: {}", error); + changed = true; + } + } + } + + for entry in fs::read_dir(&self.root)? { + let entry = entry?; + if !entry.file_type()?.is_dir() { + continue; + } + + let transcript_id = entry.file_name().to_string_lossy().to_string(); + if !is_transcript_id(&transcript_id) { + log::warn!( + "Ignoring terminal transcript directory with invalid transcript ID: {}", + transcript_id + ); + continue; + } + if self + .sessions + .values() + .any(|session| session.transcript_id == transcript_id) + { + continue; + } + + if let Some(session) = Self::recover_session(&entry.path(), &transcript_id)? { + let session_id = session.session_id.clone(); + if self.sessions.contains_key(&session_id) { + log::warn!( + "Ignoring recovered terminal transcript with duplicate session ID: {}", + one_line(&session_id) + ); + changed = true; + continue; + } + self.sessions.insert(session_id, session); + changed = true; + } + } + + let session_count_before_retention = self.sessions.len(); + self.apply_session_retention()?; + if self.sessions.len() != session_count_before_retention { + changed = true; + } + + if changed || !index_path.exists() { + self.write_index()?; + } + + Ok(()) + } + + fn recover_session( + session_dir: &Path, + transcript_id: &str, + ) -> io::Result> { + let mut segments = Self::segment_names(session_dir)?; + if segments.is_empty() { + return Ok(None); + } + segments.sort(); + + let first_segment = session_dir.join(&segments[0]); + let header = Self::read_segment_header(&first_segment)?; + let started_at = fs::metadata(&first_segment) + .and_then(|metadata| metadata.modified()) + .map(format_system_time) + .unwrap_or_else(|_| Utc::now().to_rfc3339()); + let session_id = header + .get("session_id") + .filter(|session_id| is_safe_session_id(session_id)) + .cloned() + .unwrap_or_else(|| format!("recovered-{transcript_id}")); + + Ok(Some(TranscriptSessionIndex { + session_id, + transcript_id: transcript_id.to_string(), + shell: header + .get("shell") + .cloned() + .unwrap_or_else(|| "unknown".to_string()), + initial_cwd: header.get("initial_cwd").cloned().unwrap_or_default(), + state: TranscriptSessionState::Stale, + started_at, + closed_at: None, + segments, + })) + } + + fn start_session(&mut self, session: &TerminalSession) -> io::Result<()> { + if self.sessions.contains_key(&session.id) { + return Ok(()); + } + if !is_safe_session_id(&session.id) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "terminal session ID contains characters unsafe for transcript metadata", + )); + } + + let (transcript_id, session_dir) = self.allocate_transcript_dir()?; + let segment = segment_name(1); + let index = TranscriptSessionIndex { + session_id: session.id.clone(), + transcript_id: transcript_id.clone(), + shell: session.shell_type.to_string(), + initial_cwd: session.initial_cwd.clone(), + state: TranscriptSessionState::Active, + started_at: format_timestamp(session.created_at), + closed_at: None, + segments: vec![segment.clone()], + }; + let segment_path = session_dir.join(&segment); + let header_bytes = match (|| -> io::Result { + let mut file = File::create(&segment_path)?; + write_segment_header_from_index(&mut file, &index, 1, None)?; + file.flush()?; + Ok(file.metadata()?.len()) + })() { + Ok(header_bytes) => header_bytes, + Err(error) => { + if let Err(remove_error) = fs::remove_dir_all(&session_dir) { + log::warn!( + "Failed to clean up terminal transcript directory after start error: transcript_id={} error={}", + transcript_id, + remove_error + ); + } + return Err(error); + } + }; + + self.sessions.insert(session.id.clone(), index); + self.writers.insert( + session.id.clone(), + TranscriptWriter { + current_segment: 1, + current_segment_bytes: header_bytes, + header_bytes, + }, + ); + self.apply_session_retention()?; + self.write_index() + } + + fn append(&mut self, session_id: &str, data: &str) -> io::Result<()> { + if !self.sessions.contains_key(session_id) { + return Ok(()); + } + + if self + .sessions + .get(session_id) + .is_some_and(|session| session.state != TranscriptSessionState::Active) + { + return Ok(()); + } + + let result = self.append_inner(session_id, data); + match result { + Ok(()) => { + if let Some(error) = self.recovered_write_errors.remove(session_id) { + let recovery_marker = format!( + "\n[bitfun: recorder_recovered_after_error={}]\n", + one_line(&error) + ); + self.append_inner(session_id, &recovery_marker)?; + } + Ok(()) + } + Err(error) => { + self.recovered_write_errors + .insert(session_id.to_string(), error.to_string()); + Err(error) + } + } + } + + fn append_inner(&mut self, session_id: &str, data: &str) -> io::Result<()> { + let data_len = data.len() as u64; + let rotate_before_write = { + let writer = self.ensure_writer(session_id)?; + writer.current_segment_bytes > writer.header_bytes + && writer.current_segment_bytes.saturating_add(data_len) + > self.config.segment_size_bytes + }; + + if rotate_before_write { + self.rotate(session_id)?; + } + + let current_segment = self.ensure_writer(session_id)?.current_segment; + let segment_path = self + .session_dir(session_id)? + .join(segment_name(current_segment)); + + let mut file = OpenOptions::new().append(true).open(&segment_path)?; + file.write_all(data.as_bytes())?; + file.flush()?; + + let writer = self.ensure_writer(session_id)?; + debug_assert_eq!(writer.current_segment, current_segment); + writer.current_segment_bytes = writer.current_segment_bytes.saturating_add(data_len); + Ok(()) + } + + fn finish_session(&mut self, session_id: &str, exit_code: Option) -> io::Result<()> { + let Some(session) = self.sessions.get(session_id) else { + return Ok(()); + }; + if session.state != TranscriptSessionState::Active { + return Ok(()); + } + + let marker = match exit_code { + Some(code) => format!("\n[bitfun: session_closed exit_code={code}]\n"), + None => "\n[bitfun: session_closed]\n".to_string(), + }; + self.append(session_id, &marker)?; + + if let Some(session) = self.sessions.get_mut(session_id) { + session.state = TranscriptSessionState::Closed; + session.closed_at = Some(format_timestamp(Utc::now())); + } + self.writers.remove(session_id); + self.apply_session_retention()?; + self.write_index() + } + + fn rotate(&mut self, session_id: &str) -> io::Result<()> { + let next_segment = self.ensure_writer(session_id)?.current_segment + 1; + let previous_segment = segment_name(next_segment - 1); + let session = self.sessions.get(session_id).cloned().ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + "terminal transcript session missing", + ) + })?; + let session_dir = self.session_dir(session_id)?; + let next_name = segment_name(next_segment); + let next_path = session_dir.join(&next_name); + let mut file = File::create(next_path)?; + write_segment_header_from_index( + &mut file, + &session, + next_segment, + Some(&previous_segment), + )?; + file.flush()?; + let header_bytes = file.metadata()?.len(); + + let writer = self.ensure_writer(session_id)?; + writer.current_segment = next_segment; + writer.current_segment_bytes = header_bytes; + writer.header_bytes = header_bytes; + + let retained_segments = { + let session = self + .sessions + .get_mut(session_id) + .expect("session should exist while rotating"); + session.segments.push(next_name); + session.segments.clone() + }; + + let retained_segment_limit = self.config.retained_segments_per_session.max(1); + if retained_segments.len() > retained_segment_limit { + let remove_count = retained_segments.len() - retained_segment_limit; + let removed: Vec<_> = retained_segments.into_iter().take(remove_count).collect(); + for segment in &removed { + fs::remove_file(session_dir.join(segment))?; + } + let session = self + .sessions + .get_mut(session_id) + .expect("session should exist while trimming segments"); + session.segments.drain(..remove_count); + } + + self.write_index() + } + + fn ensure_writer(&mut self, session_id: &str) -> io::Result<&mut TranscriptWriter> { + if !self.writers.contains_key(session_id) { + let current_segment_name = self + .sessions + .get(session_id) + .ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + "terminal transcript session missing", + ) + })? + .segments + .last() + .cloned() + .ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + "terminal transcript session has no segments", + ) + })?; + let current_segment = parse_segment_number(¤t_segment_name).ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidData, + "invalid terminal transcript segment name", + ) + })?; + let segment_path = self.session_dir(session_id)?.join(¤t_segment_name); + let current_segment_bytes = fs::metadata(&segment_path) + .map(|metadata| metadata.len()) + .unwrap_or(0); + let header_bytes = segment_header_bytes(&segment_path).unwrap_or(0); + self.writers.insert( + session_id.to_string(), + TranscriptWriter { + current_segment, + current_segment_bytes, + header_bytes, + }, + ); + } + Ok(self + .writers + .get_mut(session_id) + .expect("terminal transcript writer should exist")) + } + + fn apply_session_retention(&mut self) -> io::Result<()> { + let active_sessions = self + .sessions + .values() + .filter(|session| session.state == TranscriptSessionState::Active) + .count(); + let max_sessions = self.config.max_recent_sessions.max(active_sessions); + + while self.sessions.len() > max_sessions { + let Some(session_id) = self + .sessions + .values() + .filter(|session| session.state != TranscriptSessionState::Active) + .min_by(|left, right| { + let left_time = left.closed_at.as_deref().unwrap_or(&left.started_at); + let right_time = right.closed_at.as_deref().unwrap_or(&right.started_at); + left_time + .cmp(right_time) + .then_with(|| left.session_id.cmp(&right.session_id)) + }) + .map(|session| session.session_id.clone()) + else { + break; + }; + + fs::remove_dir_all(self.session_dir(&session_id)?)?; + self.sessions.remove(&session_id); + self.writers.remove(&session_id); + self.recovered_write_errors.remove(&session_id); + } + + Ok(()) + } + + fn write_index(&self) -> io::Result<()> { + fs::create_dir_all(&self.root)?; + let mut sessions: Vec<_> = self.sessions.values().cloned().collect(); + sessions.sort_by(|left, right| { + right + .started_at + .cmp(&left.started_at) + .then_with(|| right.transcript_id.cmp(&left.transcript_id)) + }); + let index = TranscriptIndex { sessions }; + let serialized = serde_json::to_vec_pretty(&index).map_err(|error| { + io::Error::new( + io::ErrorKind::Other, + format!("serialize terminal transcript index: {error}"), + ) + })?; + + let temporary_path = self.root.join(INDEX_TEMP_FILE_NAME); + let mut temporary = File::create(&temporary_path)?; + temporary.write_all(&serialized)?; + temporary.write_all(b"\n")?; + temporary.sync_all()?; + drop(temporary); + replace_file(&temporary_path, &self.index_path()) + } + + fn write_agents_document(&self) -> io::Result<()> { + fs::write(self.root.join(AGENTS_FILE_NAME), TRANSCRIPT_AGENTS_DOCUMENT) + } + + fn index_path(&self) -> PathBuf { + self.root.join(INDEX_FILE_NAME) + } + + fn session_dir(&self, session_id: &str) -> io::Result { + let transcript_id = self + .sessions + .get(session_id) + .map(|session| session.transcript_id.as_str()) + .ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + "terminal transcript session missing", + ) + })?; + Ok(self.transcript_dir(transcript_id)) + } + + fn transcript_dir(&self, transcript_id: &str) -> PathBuf { + self.root.join(transcript_id) + } + + fn allocate_transcript_dir(&self) -> io::Result<(String, PathBuf)> { + self.allocate_transcript_dir_with(generate_transcript_id) + } + + fn allocate_transcript_dir_with( + &self, + mut next_id: impl FnMut() -> String, + ) -> io::Result<(String, PathBuf)> { + fs::create_dir_all(&self.root)?; + + for _ in 0..MAX_TRANSCRIPT_ID_ALLOCATION_ATTEMPTS { + let transcript_id = next_id(); + if !is_transcript_id(&transcript_id) + || self + .sessions + .values() + .any(|session| session.transcript_id == transcript_id) + { + continue; + } + + let directory = self.transcript_dir(&transcript_id); + match fs::create_dir(&directory) { + Ok(()) => return Ok((transcript_id, directory)), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue, + Err(error) => return Err(error), + } + } + + Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "failed to allocate a unique terminal transcript ID", + )) + } + + fn segment_names(session_dir: &Path) -> io::Result> { + let mut segments = Vec::new(); + for entry in fs::read_dir(session_dir)? { + let entry = entry?; + if !entry.file_type()?.is_file() { + continue; + } + let name = entry.file_name().to_string_lossy().to_string(); + if parse_segment_number(&name).is_some() { + segments.push(name); + } + } + Ok(segments) + } + + fn read_segment_header(path: &Path) -> io::Result> { + let file = File::open(path)?; + let mut values = HashMap::new(); + for line in BufReader::new(file).lines().take(16) { + let line = line?; + let Some(body) = line + .strip_prefix("[bitfun: ") + .and_then(|line| line.strip_suffix(']')) + else { + continue; + }; + let Some((key, value)) = body.split_once('=') else { + continue; + }; + values.insert(key.to_string(), value.to_string()); + } + Ok(values) + } +} + +fn write_segment_header_from_index( + file: &mut File, + session: &TranscriptSessionIndex, + segment: u64, + previous_segment: Option<&str>, +) -> io::Result<()> { + writeln!(file, "===== BitFun user terminal transcript =====")?; + writeln!(file, "[bitfun: session_id={}]", session.session_id)?; + writeln!(file, "[bitfun: transcript_id={}]", session.transcript_id)?; + writeln!(file, "[bitfun: shell={}]", session.shell)?; + writeln!(file, "[bitfun: initial_cwd={}]", session.initial_cwd)?; + writeln!(file, "[bitfun: started_at={}]", session.started_at)?; + writeln!(file, "[bitfun: segment={:06}]", segment)?; + if let Some(previous_segment) = previous_segment { + writeln!(file, "[bitfun: previous_segment={previous_segment}]")?; + } + writeln!(file)?; + Ok(()) +} + +fn segment_header_bytes(path: &Path) -> io::Result { + let file = File::open(path)?; + let mut reader = BufReader::new(file); + let mut header_bytes = 0_u64; + let mut line = String::new(); + + loop { + line.clear(); + let read = reader.read_line(&mut line)?; + if read == 0 { + break; + } + header_bytes = header_bytes.saturating_add(read as u64); + if line.trim_end_matches(['\r', '\n']).is_empty() { + break; + } + } + + Ok(header_bytes) +} + +fn replace_file(temporary_path: &Path, target_path: &Path) -> io::Result<()> { + match fs::rename(temporary_path, target_path) { + Ok(()) => Ok(()), + Err(_) if target_path.exists() => { + fs::remove_file(target_path)?; + fs::rename(temporary_path, target_path) + } + Err(error) => Err(error), + } +} + +fn is_safe_session_id(session_id: &str) -> bool { + !session_id.is_empty() + && session_id + .chars() + .all(|character| character.is_ascii_alphanumeric() || matches!(character, '-' | '_')) +} + +fn is_transcript_id(transcript_id: &str) -> bool { + transcript_id.len() == TRANSCRIPT_ID_LENGTH + && transcript_id + .bytes() + .all(|byte| byte.is_ascii_digit() || byte.is_ascii_lowercase()) +} + +fn generate_transcript_id() -> String { + let mut random = rand::thread_rng(); + (0..TRANSCRIPT_ID_LENGTH) + .map(|_| { + let index = random.gen_range(0..TRANSCRIPT_ID_ALPHABET.len()); + TRANSCRIPT_ID_ALPHABET[index] as char + }) + .collect() +} + +fn segment_name(segment: u64) -> String { + format!("{segment:06}.{SEGMENT_EXTENSION}") +} + +fn parse_segment_number(name: &str) -> Option { + let number = name.strip_suffix(&format!(".{SEGMENT_EXTENSION}"))?; + if number.len() != 6 || !number.bytes().all(|byte| byte.is_ascii_digit()) { + return None; + } + number.parse().ok() +} + +fn format_system_time(time: SystemTime) -> String { + format_timestamp(DateTime::::from(time)) +} + +fn format_timestamp(time: DateTime) -> String { + time.to_rfc3339_opts(SecondsFormat::Secs, true) +} + +fn one_line(value: &str) -> String { + value.replace(['\r', '\n'], " ") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::shell::ShellType; + use tempfile::TempDir; + + fn config(temp_dir: &TempDir) -> TerminalTranscriptConfig { + TerminalTranscriptConfig { + root_dir: Some(temp_dir.path().join("terminals")), + segment_size_bytes: 32, + retained_segments_per_session: 2, + max_recent_sessions: 10, + } + } + + fn session(id: &str) -> TerminalSession { + TerminalSession::new( + id.to_string(), + format!("Terminal {id}"), + ShellType::PowerShell, + "E:/workspace".to_string(), + 80, + 24, + SessionSource::Manual, + ) + } + + fn read_index(root: &Path) -> TranscriptIndex { + serde_json::from_str( + &fs::read_to_string(root.join(INDEX_FILE_NAME)).expect("index should be readable"), + ) + .expect("index should be valid JSON") + } + + fn transcript_id_for(root: &Path, session_id: &str) -> String { + read_index(root) + .sessions + .into_iter() + .find(|session| session.session_id == session_id) + .expect("session should be present in index") + .transcript_id + } + + #[test] + fn manual_session_creates_plain_text_segment_and_index() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let mut transcript_config = config(&temp_dir); + transcript_config.segment_size_bytes = 1024 * 1024; + let recorder = TranscriptRecorder::from_config(&transcript_config) + .expect("recorder should be configured"); + let session = session("manual-1"); + + recorder + .start_session(&session) + .expect("session transcript should start"); + recorder + .record_output(&session.id, "PS E:\\workspace> echo hello\r\nhello\r\n") + .expect("output should append"); + recorder + .finish_session(&session.id, Some(0)) + .expect("session transcript should finish"); + + let root = temp_dir.path().join("terminals"); + let index = read_index(&root); + assert_eq!(index.sessions.len(), 1); + let indexed_session = &index.sessions[0]; + assert_eq!(indexed_session.session_id, "manual-1"); + assert!(is_transcript_id(&indexed_session.transcript_id)); + assert!(indexed_session.segments.is_empty()); + assert!(!indexed_session.started_at.contains('.')); + assert!(indexed_session + .closed_at + .as_deref() + .is_some_and(|closed_at| !closed_at.contains('.'))); + + let segment = + fs::read_to_string(root.join(&indexed_session.transcript_id).join("000001.log")) + .expect("segment should be readable"); + assert!(!root.join("manual-1").exists()); + assert!(segment.contains("[bitfun: session_id=manual-1]")); + assert!(segment.contains(&format!( + "[bitfun: transcript_id={}]", + indexed_session.transcript_id + ))); + assert!(segment.contains("PS E:\\workspace> echo hello")); + assert!(segment.contains("hello")); + + let index_json: serde_json::Value = serde_json::from_str( + &fs::read_to_string(root.join(INDEX_FILE_NAME)).expect("index should be readable"), + ) + .expect("index should be valid JSON"); + let session_json = &index_json["sessions"][0]; + assert_eq!(session_json["session_id"], "manual-1"); + assert_eq!( + session_json["transcript_id"], + serde_json::Value::String(indexed_session.transcript_id.clone()) + ); + assert!(session_json.get("source").is_none()); + assert!(session_json.get("current_segment").is_none()); + assert!(session_json.get("dropped_before_segment").is_none()); + assert!(session_json.get("segments").is_none()); + + let agents_document = fs::read_to_string(root.join(AGENTS_FILE_NAME)) + .expect("transcript instructions should be readable"); + assert!(agents_document.contains("# User terminal transcripts")); + assert!(agents_document.contains("`index.json`")); + assert!(agents_document.contains("terminals/")); + assert!(agents_document.contains("└── /")); + assert!(agents_document.contains("### Search for a keyword")); + assert!(agents_document.contains("Use `Grep`")); + assert!(agents_document.contains("### Inspect a terminal's recent output")); + assert!(agents_document.contains("List all `.log` files")); + } + + #[test] + fn records_executed_commands_and_ignores_blank_commands() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let mut transcript_config = config(&temp_dir); + transcript_config.segment_size_bytes = 1024 * 1024; + let recorder = TranscriptRecorder::from_config(&transcript_config) + .expect("recorder should be configured"); + let session = session("manual-1"); + + recorder + .start_session(&session) + .expect("session transcript should start"); + recorder + .record_command(&session.id, "echo hello") + .expect("command should append"); + recorder + .record_command(&session.id, " \r\n\t") + .expect("blank command should be ignored"); + recorder + .record_output(&session.id, "hello\n") + .expect("output should append"); + + let root = temp_dir.path().join("terminals"); + let transcript_id = transcript_id_for(&root, &session.id); + let segment = fs::read_to_string(root.join(transcript_id).join("000001.log")) + .expect("segment should be readable"); + + assert!(segment.contains("[bitfun: command=echo hello]\nhello\n")); + assert_eq!(segment.matches("[bitfun: command=").count(), 1); + } + + #[test] + fn agent_session_does_not_create_transcript_files() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let recorder = TranscriptRecorder::from_config(&config(&temp_dir)) + .expect("recorder should be configured"); + let mut agent_session = session("agent-1"); + agent_session.source = SessionSource::Agent; + + recorder + .start_session(&agent_session) + .expect("agent session should be ignored"); + + let root = temp_dir.path().join("terminals"); + assert_eq!(read_index(&root).sessions.len(), 0); + assert_eq!( + fs::read_dir(root) + .expect("transcript root should be readable") + .filter_map(Result::ok) + .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir())) + .count(), + 0 + ); + } + + #[test] + fn rotation_keeps_only_recent_sealed_segments() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let mut config = config(&temp_dir); + config.segment_size_bytes = 5; + let recorder = + TranscriptRecorder::from_config(&config).expect("recorder should be configured"); + let session = session("manual-1"); + recorder + .start_session(&session) + .expect("session should start"); + + recorder + .record_output(&session.id, "12345") + .expect("first output should append"); + recorder + .record_output(&session.id, "6") + .expect("second output should rotate"); + recorder + .record_output(&session.id, "78901") + .expect("third output should rotate"); + + let root = temp_dir.path().join("terminals"); + let session_dir = root.join(transcript_id_for(&root, &session.id)); + assert!(!session_dir.join("000001.log").exists()); + assert!(session_dir.join("000002.log").exists()); + assert!(session_dir.join("000003.log").exists()); + } + + #[test] + fn completed_sessions_are_trimmed_but_active_sessions_are_retained() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let recorder = TranscriptRecorder::from_config(&config(&temp_dir)) + .expect("recorder should be configured"); + + for number in 0..11 { + let session = session(&format!("closed-{number:02}")); + recorder + .start_session(&session) + .expect("session should start"); + recorder + .finish_session(&session.id, Some(0)) + .expect("session should finish"); + } + + let root = temp_dir.path().join("terminals"); + let retained_directories = fs::read_dir(&root) + .expect("transcript root should be readable") + .filter_map(Result::ok) + .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir())) + .count(); + assert_eq!(retained_directories, 10); + + for number in 0..11 { + let session = session(&format!("active-{number:02}")); + recorder + .start_session(&session) + .expect("active session should start"); + } + + let active_directories = fs::read_dir(&root) + .expect("transcript root should be readable") + .filter_map(Result::ok) + .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir())) + .count(); + assert_eq!(active_directories, 11); + } + + #[test] + fn recovered_stale_sessions_are_trimmed_as_inactive() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let config = config(&temp_dir); + let recorder = + TranscriptRecorder::from_config(&config).expect("recorder should be configured"); + + for number in 0..11 { + let session = session(&format!("active-{number:02}")); + recorder + .start_session(&session) + .expect("session should start"); + } + drop(recorder); + + let recovered = TranscriptRecorder::from_config(&config).expect("recovery should succeed"); + let root = temp_dir.path().join("terminals"); + let index = read_index(&root); + + assert_eq!(index.sessions.len(), 10); + assert!(index + .sessions + .iter() + .all(|session| session.state == TranscriptSessionState::Stale)); + assert!(!index + .sessions + .iter() + .any(|session| session.session_id == "active-00")); + drop(recovered); + } + + #[test] + fn startup_recovery_replaces_unreadable_empty_index() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let root = temp_dir.path().join("terminals"); + fs::create_dir_all(&root).expect("transcript root should create"); + fs::write(root.join(INDEX_FILE_NAME), "not valid JSON") + .expect("invalid index should write"); + + let _recorder = TranscriptRecorder::from_config(&config(&temp_dir)) + .expect("recorder should be configured"); + + assert!(read_index(&root).sessions.is_empty()); + } + + #[test] + fn startup_recovery_rebuilds_missing_index_from_segments() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let config = config(&temp_dir); + let recorder = + TranscriptRecorder::from_config(&config).expect("recorder should be configured"); + let session = session("manual-1"); + recorder + .start_session(&session) + .expect("session should start"); + recorder + .record_output(&session.id, "hello") + .expect("output should append"); + drop(recorder); + + let root = temp_dir.path().join("terminals"); + fs::remove_file(root.join(INDEX_FILE_NAME)).expect("index should remove"); + let _recovered = TranscriptRecorder::from_config(&config).expect("recovery should succeed"); + + let index = read_index(&root); + assert_eq!(index.sessions.len(), 1); + assert_eq!(index.sessions[0].session_id, "manual-1"); + assert!(is_transcript_id(&index.sessions[0].transcript_id)); + assert!(root.join(&index.sessions[0].transcript_id).exists()); + assert_eq!(index.sessions[0].state, TranscriptSessionState::Stale); + } + + #[test] + fn transcript_id_allocation_retries_after_directory_collision() { + let temp_dir = TempDir::new().expect("temporary directory should create"); + let root = temp_dir.path().join("terminals"); + fs::create_dir_all(&root).expect("transcript root should create"); + fs::create_dir(root.join("aaaaaa")).expect("colliding directory should create"); + let store = TranscriptStore { + config: config(&temp_dir), + root: root.clone(), + sessions: HashMap::new(), + writers: HashMap::new(), + recovered_write_errors: HashMap::new(), + }; + let mut candidates = ["aaaaaa".to_string(), "bbbbbb".to_string()].into_iter(); + + let (transcript_id, directory) = store + .allocate_transcript_dir_with(|| candidates.next().expect("candidate should exist")) + .expect("second transcript ID should allocate"); + + assert_eq!(transcript_id, "bbbbbb"); + assert_eq!(directory, root.join("bbbbbb")); + assert!(directory.is_dir()); + } +}