From 0c6b7c85942406ab1a63d7b39334f70344139ad6 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Tue, 15 Sep 2026 20:58:10 -0600 Subject: [PATCH] fix(TASK-035): close agent-bracket leaks and prove compression/role-hook seams - Open the top-level agent bracket only at committed dispatch sites so inspection-flag early returns (--list-sessions, --info, --macro, --rebuild-rag, --init-*, --build-tools) never leak agent.started - Close the agent bracket ahead of process::exit in shell_execute's execute branch, where the dispatch wrapper's close is unreachable - Extract the ask-loop auto-compression seam and cover the session.compressed success path at both call sites with sink tests - Prove Session.role_hooks never serializes and that injected role_hooks YAML is ignored on deserialization --- src/config/session.rs | 31 +++++++++++ src/main.rs | 22 +++++++- src/repl/mod.rs | 127 ++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 174 insertions(+), 6 deletions(-) diff --git a/src/config/session.rs b/src/config/session.rs index 6a0e244..ad7a47f 100644 --- a/src/config/session.rs +++ b/src/config/session.rs @@ -1266,6 +1266,37 @@ mod tests { assert!(!yaml.contains("token_usage")); } + #[test] + fn session_role_hooks_is_not_serialized() { + let content = "---\nhooks:\n turn.completed:\n - name: role-probe\n command: role-cmd\n---\nPrompt"; + let mut session = Session::default(); + session.set_role(Role::new("hooked", content)); + assert!(session.role_hooks().is_some()); + + let yaml = serde_yaml::to_string(&session).unwrap(); + assert!( + !yaml.contains("role_hooks"), + "the role-hooks snapshot must never reach session YAML" + ); + assert!( + !yaml.contains("role-cmd"), + "no hook command text may leak into session YAML under any key" + ); + } + + #[test] + fn session_deserialization_ignores_injected_role_hooks() { + // A session file is attacker-writable data: a `role_hooks:` key + // crafted into it must never deserialize into executable hook + // config ā€” hooks are only ever re-read from the role file itself. + let yaml = "model: ''\nmessages: []\nrole_hooks:\n session.started:\n - name: injected\n command: touch /tmp/pwned\n"; + let session: Session = serde_yaml::from_str(yaml).unwrap(); + assert!( + session.role_hooks().is_none(), + "injected role_hooks in session YAML must be ignored" + ); + } + #[test] fn session_cost_accumulates_and_survives_yaml_round_trip() { let mut session = Session::default(); diff --git a/src/main.rs b/src/main.rs index 1c197a9..84f7ef4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -388,7 +388,12 @@ async fn run( let app = Arc::clone(&ctx.app.config); ctx.use_agent(app.as_ref(), agent, session, abort_signal.clone()) .await?; - ctx.top_level_agent_started(); + // The top-level agent bracket does NOT open here: the inspection + // flags below (--list-sessions, --info, --macro, --rebuild-rag, + // --init-*) return early without running the agent, and an open + // bracket on those paths would leak `agent.started` with no + // terminal event. It opens at the dispatch sites further down, + // where an agent run is actually committed. } else { let app: Arc = Arc::clone(&ctx.app.config); if let Some(prompt) = &cli.temp_role { @@ -525,6 +530,9 @@ async fn run( return Ok(()); } if cli.execute && !is_repl { + // Dispatch committed: open the top-level agent bracket only now, + // past every inspection-flag early return above. + ctx.top_level_agent_started(); let result = async { let input = create_input(&ctx, text, &cli.file, abort_signal.clone()).await?; shell_execute(&mut ctx, &SHELL, input, abort_signal.clone()).await @@ -550,6 +558,12 @@ async fn run( return acp::run_acp_server(ctx, abort_signal).await; } + // Dispatch committed (headless directive or REPL): open the top-level + // agent bracket only now, past every inspection-flag early return + // above. The ACP server stays unbracketed deliberately. The headless + // arm closes the bracket below; the REPL closes it at teardown (or on + // an interactive agent switch). + ctx.top_level_agent_started(); match is_repl { false => { let result = async { @@ -692,6 +706,12 @@ async fn shell_execute( if code == 0 && app.save_shell_history { let _ = append_to_shell_history(&shell.name, &eval_str, code); } + // `process::exit` never returns, so the dispatch + // wrapper's bracket close in `run()` is unreachable + // from here: close the top-level agent bracket now. + // The agent run itself succeeded ā€” the exit code + // belongs to the user's shell command, not to us. + ctx.top_level_agent_finished(None); hooks::drain_pending(EXIT_HOOK_DRAIN_TIMEOUT).await; process::exit(code); } diff --git a/src/repl/mod.rs b/src/repl/mod.rs index 9c45845..091d6e0 100644 --- a/src/repl/mod.rs +++ b/src/repl/mod.rs @@ -1644,10 +1644,7 @@ async fn ask_inner( }; eprintln!("\nšŸ“¢ {}", color.italic().paint("Compressing the session."),); - match ctx.compress_session().await { - Ok(()) => hooks::fire(HookEvent::SessionCompressed, ctx, &[], None), - Err(err) => warn!("Failed to compress the session: {err}"), - } + auto_compress_session(ctx).await; if let Some(session) = ctx.session.as_mut() { session.set_compressing(false); } @@ -1698,6 +1695,18 @@ fn print_turn_divider() { println!(); } +/// Auto-compression seam for the ask loop: `session.compressed` fires only +/// when compression actually succeeded. A failed auto-compression is +/// survivable mid-session, so the error is logged and swallowed rather +/// than aborting the turn ā€” unlike `.compress session`, which surfaces +/// the failure to the user. +async fn auto_compress_session(ctx: &mut RequestContext) { + match ctx.compress_session().await { + Ok(()) => hooks::fire(HookEvent::SessionCompressed, ctx, &[], None), + Err(err) => warn!("Failed to compress the session: {err}"), + } +} + fn should_continue(ctx: &RequestContext) -> bool { let config = ctx.auto_continue_config(); ctx.app.config.function_calling_support @@ -2005,7 +2014,8 @@ pub fn split_args_text(line: &str, is_win: bool) -> (Vec, &str) { #[cfg(test)] mod tests { use super::*; - use crate::config::{AppState, Session, WorkingMode}; + use crate::client::{ClientConfig, Model}; + use crate::config::{AppState, RoleLike, Session, WorkingMode}; use crate::hooks::{HookDef, HooksMap, test_sink}; use anyhow::anyhow; use serial_test::serial; @@ -2180,6 +2190,113 @@ mod tests { ); } + /// A ctx whose compression can actually SUCCEED offline: `dry_run` + /// makes the summarization call echo instead of hitting the network, + /// the default openai client config makes the session model + /// resolvable, and the session already holds a user exchange so + /// `compress_session` has something to summarize. + fn compressible_ctx(marker: &str) -> RequestContext { + let mut hooks_map = HooksMap::default(); + hooks_map.insert( + "session.compressed".to_string(), + vec![HookDef { + name: format!("{marker}-session.compressed"), + command: "true".to_string(), + }], + ); + let mut app = AppState::test_default(); + app.config = Arc::new(AppConfig { + hooks: hooks_map, + dry_run: true, + clients: vec![ClientConfig::default()], + ..Default::default() + }); + let mut ctx = RequestContext::new(Arc::new(app), WorkingMode::Repl); + let mut session = Session::default(); + session.set_model(Model::new("openai", "test-compress-model")); + ctx.session = Some(session); + let input = Input::from_str(&ctx, "hello there", None).unwrap(); + ctx.session + .as_mut() + .unwrap() + .add_message(&input, "hi") + .unwrap(); + ctx + } + + fn session_compressed_count(marker: &str) -> usize { + test_sink::snapshot() + .iter() + .filter(|capture| capture.hook_name == format!("{marker}-session.compressed")) + .count() + } + + #[tokio::test] + #[serial] + async fn compress_command_success_fires_session_compressed_once() { + let _sink = test_sink::install(); + let marker = "compress-ok-w2p"; + let mut ctx = compressible_ctx(marker); + let abort_signal = create_abort_signal(); + + // Boxed: `run_repl_command`'s state machine is far larger than a test + // thread's stack. + let result = Box::pin(run_repl_command( + &mut ctx, + abort_signal, + ".compress session", + )) + .await; + + assert!( + result.is_ok(), + "dry-run compression must succeed: {result:?}" + ); + assert_eq!( + session_compressed_count(marker), + 1, + "a successful `.compress session` must fire session.compressed exactly once" + ); + } + + #[tokio::test] + #[serial] + async fn auto_compress_success_fires_session_compressed_once() { + // Covers the ask-loop call site through its extracted seam; + // driving `needs_compression` itself requires a full LLM turn and + // is exercised end-to-end by integration coverage, not here. + let _sink = test_sink::install(); + let marker = "auto-compress-ok-f8r"; + let mut ctx = compressible_ctx(marker); + + auto_compress_session(&mut ctx).await; + + assert_eq!( + session_compressed_count(marker), + 1, + "a successful auto-compression must fire session.compressed exactly once" + ); + } + + #[tokio::test] + #[serial] + async fn auto_compress_failure_fires_no_session_compressed_and_swallows_error() { + let _sink = test_sink::install(); + let marker = "auto-compress-fail-d3k"; + let mut ctx = compressible_ctx(marker); + // An empty session makes compress_session fail; the seam must + // swallow the error (no panic, no propagation) and fire nothing. + ctx.session = Some(Session::default()); + + auto_compress_session(&mut ctx).await; + + assert_eq!( + session_compressed_count(marker), + 0, + "a failed auto-compression must fire no session.compressed hook" + ); + } + #[test] fn test_process_command_line() { assert_eq!(parse_command(" ."), Some((".", None)));