diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index e5b7212ca5..b4abc81177 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -482,17 +482,27 @@ async fn on_event_updates_status_from_task_started() { #[tokio::test] async fn on_event_updates_status_from_task_complete() { - let status = agent_status_from_event(&EventMsg::TurnComplete(TurnCompleteEvent { - turn_id: "turn-1".to_string(), - started_at: None, - last_agent_message: Some("done".to_string()), - error: None, - completed_at: None, - duration_ms: None, - time_to_first_token_ms: None, - })); - let expected = AgentStatus::Completed(Some("done".to_string())); - assert_eq!(status, Some(expected)); + for (error, expected) in [ + (None, AgentStatus::Completed(Some("done".to_string()))), + ( + Some(ErrorEvent { + message: "denied".to_string(), + codex_error_info: None, + }), + AgentStatus::Errored("denied".to_string()), + ), + ] { + let status = agent_status_from_event(&EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: "turn-1".to_string(), + started_at: None, + last_agent_message: Some("done".to_string()), + error, + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + })); + assert_eq!(status, Some(expected)); + } } #[tokio::test] diff --git a/codex-rs/core/src/agent/status.rs b/codex-rs/core/src/agent/status.rs index 43be718865..0ec7e96d4b 100644 --- a/codex-rs/core/src/agent/status.rs +++ b/codex-rs/core/src/agent/status.rs @@ -6,7 +6,10 @@ use codex_protocol::protocol::EventMsg; pub(crate) fn agent_status_from_event(msg: &EventMsg) -> Option { match msg { EventMsg::TurnStarted(_) => Some(AgentStatus::Running), - EventMsg::TurnComplete(ev) => Some(AgentStatus::Completed(ev.last_agent_message.clone())), + EventMsg::TurnComplete(ev) => Some(match &ev.error { + Some(error) => AgentStatus::Errored(error.message.clone()), + None => AgentStatus::Completed(ev.last_agent_message.clone()), + }), EventMsg::TurnAborted(ev) => match ev.reason { codex_protocol::protocol::TurnAbortReason::Interrupted | codex_protocol::protocol::TurnAbortReason::BudgetLimited => { diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 3a586260ab..d4a5ee04eb 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -40,6 +40,7 @@ use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookRunSummary; use codex_protocol::protocol::HookSource; use codex_protocol::protocol::HookStartedEvent; +use codex_protocol::protocol::InternalSessionSource; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::WarningEvent; @@ -356,6 +357,10 @@ pub(crate) async fn run_turn_stop_hooks( // Internal/synthetic subagents do not expose user-configured lifecycle // hooks, so there is no Stop or SubagentStop request to dispatch. SessionSource::SubAgent(_) => return StopOutcome::default(), + SessionSource::Internal(InternalSessionSource::MemoryConsolidation) => ( + StopHookTarget::MemoryConsolidation, + sess.hook_transcript_path().await, + ), _ => (StopHookTarget::Stop, sess.hook_transcript_path().await), }; let request_metadata = turn_context diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 792b2bf4c8..850166360d 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -96,11 +96,13 @@ use codex_protocol::protocol::AgentReasoningSectionBreakEvent; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::InternalSessionSource; use codex_protocol::protocol::PlanDeltaEvent; use codex_protocol::protocol::RawResponseCompletedEvent; use codex_protocol::protocol::ReasoningContentDeltaEvent; use codex_protocol::protocol::ReasoningRawContentDeltaEvent; use codex_protocol::protocol::SafetyBufferingEvent; +use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::TurnDiffEvent; use codex_protocol::protocol::WarningEvent; use codex_protocol::user_input::UserInput; @@ -506,6 +508,16 @@ pub(crate) async fn run_turn( last_agent_message.clone(), ) .await; + if matches!( + turn_context.session_source, + SessionSource::Internal(InternalSessionSource::MemoryConsolidation) + ) && (stop_outcome.should_block || stop_outcome.should_stop) + { + // Do not feed managed rejections back into an unattended memory loop. + return Err(CodexErr::InvalidRequest( + "Memory consolidation was rejected by a Stop hook.".to_string(), + )); + } if stop_outcome.should_block { if let Some(hook_prompt_message) = build_hook_prompt_message(&stop_outcome.continuation_fragments) diff --git a/codex-rs/hooks/src/engine/mod_tests.rs b/codex-rs/hooks/src/engine/mod_tests.rs index 574939f700..ed25847148 100644 --- a/codex-rs/hooks/src/engine/mod_tests.rs +++ b/codex-rs/hooks/src/engine/mod_tests.rs @@ -2219,6 +2219,58 @@ fn executor_stop_hooks_register_only_the_first_environment_and_handler() { assert_eq!(engine.handlers, expected_handlers); } +#[tokio::test] +async fn memory_consolidation_stop_preserves_policy_and_executor_cleanup() { + for (source, runs_policy) in [ + (HookSource::User, false), + (HookSource::Project, false), + (HookSource::SessionFlags, false), + (HookSource::Plugin, false), + (HookSource::System, true), + (HookSource::Mdm, true), + (HookSource::CloudRequirements, true), + (HookSource::CloudManagedConfig, true), + (HookSource::LegacyManagedConfigFile, true), + (HookSource::LegacyManagedConfigMdm, true), + (HookSource::Unknown, true), + ] { + let (mut engine, calls, mut request, expected_executor_call, _source) = + executor_stop_hook_fixture(); + request.target = StopHookTarget::MemoryConsolidation; + let policy_call = HookMcpCall { + server: "security".to_string(), + tool: "check".to_string(), + environment_id: None, + metadata: None, + input: Default::default(), + timeout: Duration::from_secs(5), + }; + let mut handler = engine.handlers[0].clone(); + handler.source_path = cwd().join("hooks.json").into(); + handler.source = source; + handler.kind = ConfiguredHandlerKind::McpTool { + server: policy_call.server.clone(), + tool: policy_call.tool.clone(), + input: policy_call.input.clone(), + }; + engine.handlers.push(handler); + assert_eq!( + engine.preview_stop(&request).len(), + usize::from(runs_policy) + ); + + let outcome = engine.run_stop(request).await; + let expected_calls = if runs_policy { + vec![policy_call, expected_executor_call] + } else { + vec![expected_executor_call] + }; + wait_for_mcp_calls(&calls, expected_calls.len()).await; + assert_eq!(*calls.lock().expect("lock MCP calls"), expected_calls); + assert_eq!(outcome.should_block, runs_policy); + } +} + #[tokio::test] async fn executor_stop_hooks_do_not_delay_stop_completion() { struct BlockingMcpExecutor { diff --git a/codex-rs/hooks/src/events/stop.rs b/codex-rs/hooks/src/events/stop.rs index b29e691911..89a59a65ef 100644 --- a/codex-rs/hooks/src/events/stop.rs +++ b/codex-rs/hooks/src/events/stop.rs @@ -8,6 +8,7 @@ use codex_protocol::protocol::HookOutputEntry; use codex_protocol::protocol::HookOutputEntryKind; use codex_protocol::protocol::HookRunStatus; use codex_protocol::protocol::HookRunSummary; +use codex_protocol::protocol::HookSource; use codex_utils_absolute_path::AbsolutePathBuf; use serde_json::Map; use serde_json::Value; @@ -40,6 +41,8 @@ pub struct StopRequest { #[derive(Debug, Clone)] pub enum StopHookTarget { Stop, + /// Internal memory work runs policy and executor hooks, not project completion checks. + MemoryConsolidation, SubagentStop { agent_id: String, agent_type: String, @@ -50,17 +53,44 @@ pub enum StopHookTarget { impl StopHookTarget { fn event_name(&self) -> HookEventName { match self { - Self::Stop => HookEventName::Stop, + Self::Stop | Self::MemoryConsolidation => HookEventName::Stop, Self::SubagentStop { .. } => HookEventName::SubagentStop, } } fn matcher_input(&self) -> Option<&str> { match self { - Self::Stop => None, + Self::Stop | Self::MemoryConsolidation => None, Self::SubagentStop { agent_type, .. } => Some(agent_type.as_str()), } } + + fn select_handlers(&self, handlers: &[ConfiguredHandler]) -> Vec { + dispatcher::select_handlers(handlers, self.event_name(), self.matcher_input()) + .into_iter() + .filter(|handler| { + !matches!(self, Self::MemoryConsolidation) + || matches!( + handler.source_path, + HandlerSourcePath::ExecutorScoped { .. } + ) + || match handler.source { + HookSource::User + | HookSource::Project + | HookSource::SessionFlags + | HookSource::Plugin => false, + HookSource::System + | HookSource::Mdm + | HookSource::CloudRequirements + | HookSource::CloudManagedConfig + | HookSource::LegacyManagedConfigFile + | HookSource::LegacyManagedConfigMdm + // Required hooks can have unknown attribution; retain them fail-closed. + | HookSource::Unknown => true, + } + }) + .collect() + } } #[derive(Debug, Default)] @@ -86,23 +116,17 @@ pub(crate) fn preview( handlers: &[ConfiguredHandler], request: &StopRequest, ) -> Vec { - dispatcher::select_handlers( - handlers, - request.target.event_name(), - request.target.matcher_input(), - ) - .into_iter() - .filter(|handler| matches!(handler.source_path, HandlerSourcePath::Local(_))) - .map(|handler| dispatcher::running_summary(&handler)) - .collect() + request + .target + .select_handlers(handlers) + .into_iter() + .filter(|handler| matches!(handler.source_path, HandlerSourcePath::Local(_))) + .map(|handler| dispatcher::running_summary(&handler)) + .collect() } pub(crate) async fn run(engine: &ClaudeHooksEngine, request: StopRequest) -> StopOutcome { - let matched = dispatcher::select_handlers( - &engine.handlers, - request.target.event_name(), - request.target.matcher_input(), - ); + let matched = request.target.select_handlers(&engine.handlers); if matched.is_empty() { return StopOutcome { hook_events: Vec::new(), @@ -114,8 +138,17 @@ pub(crate) async fn run(engine: &ClaudeHooksEngine, request: StopRequest) -> Sto }; } + // Memory workers terminate on managed rejection rather than continuing the turn, + // so their executor cleanup must also run when a managed hook blocks completion. + let (executor_cleanup, matched): (Vec<_>, Vec<_>) = matched.into_iter().partition(|handler| { + matches!(request.target, StopHookTarget::MemoryConsolidation) + && matches!( + handler.source_path, + HandlerSourcePath::ExecutorScoped { .. } + ) + }); let input_json = match request.target { - StopHookTarget::Stop => { + StopHookTarget::Stop | StopHookTarget::MemoryConsolidation => { let input = StopCommandInput { session_id: request.session_id.to_string(), turn_id: request.turn_id.clone(), @@ -181,14 +214,27 @@ pub(crate) async fn run(engine: &ClaudeHooksEngine, request: StopRequest) -> Sto let results = dispatcher::execute_handlers_with_metadata( engine, matched, - input_json, + input_json.clone(), request.cwd.as_path(), - Some(request.turn_id), + Some(request.turn_id.clone()), request.request_metadata.as_ref(), parse_completed, ) .await; + if !executor_cleanup.is_empty() { + dispatcher::execute_handlers_with_metadata( + engine, + executor_cleanup, + input_json, + request.cwd.as_path(), + Some(request.turn_id), + request.request_metadata.as_ref(), + parse_completed, + ) + .await; + } + let aggregate = aggregate_results(results.iter().map(|result| &result.data)); StopOutcome { diff --git a/codex-rs/memories/write/src/phase2.rs b/codex-rs/memories/write/src/phase2.rs index b3c0fbb245..a69190b83b 100644 --- a/codex-rs/memories/write/src/phase2.rs +++ b/codex-rs/memories/write/src/phase2.rs @@ -322,6 +322,8 @@ mod agent { agent_config.ephemeral = true; agent_config.memories.generate_memories = false; agent_config.memories.use_memories = false; + // Background memory work must not send user-facing completion notifications. + agent_config.notify = None; agent_config.include_apps_instructions = false; agent_config.mcp_servers = Constrained::allow_only(HashMap::new()); // Approval policy diff --git a/codex-rs/memories/write/src/startup_tests.rs b/codex-rs/memories/write/src/startup_tests.rs index 729dfb4a24..b14767a25e 100644 --- a/codex-rs/memories/write/src/startup_tests.rs +++ b/codex-rs/memories/write/src/startup_tests.rs @@ -6,6 +6,7 @@ use crate::runtime::MemoryStartupContext; use crate::start_memories_startup_task; use crate::storage::rebuild_raw_memories_file_from_memories; use crate::storage::sync_rollout_summaries_from_memories; +use codex_config::test_support::CloudConfigBundleFixture; use codex_config::types::MemoriesConfig; use codex_features::Feature; use codex_git_utils::diff_since_latest_init; @@ -31,6 +32,8 @@ use codex_rollout::RolloutItem; use codex_rollout::RolloutLine; use codex_state::Phase2JobClaimOutcome; use codex_utils_absolute_path::test_support::PathExt; +use core_test_support::fs_wait; +use core_test_support::hooks::trust_discovered_hooks; use core_test_support::responses::ResponseMock; use core_test_support::responses::ResponsesRequest; use core_test_support::responses::ev_assistant_message; @@ -173,6 +176,150 @@ async fn memories_startup_fails_consolidation_when_worker_creates_extension_syml Ok(()) } +#[tokio::test] +async fn memories_startup_phase2_scopes_stop_hooks() -> anyhow::Result<()> { + for (pin_hooks, managed_response) in [ + (false, None), + (true, None), + (true, Some(r#"{"decision":"block","reason":"denied"}"#)), + (true, Some(r#"{"continue":false,"stopReason":"denied"}"#)), + ] { + let server = start_mock_server().await; + let home = Arc::new(TempDir::new()?); + let db = init_state_db(&home).await?; + let root = memory_root(&home.path().abs()); + seed_stage1_output( + db.as_ref(), + home.path(), + chrono::Utc::now(), + "raw memory", + "rollout summary", + "stop-hooks", + ) + .await?; + seed_required_memory_artifacts(&root).await?; + + let python = if cfg!(windows) { "python" } else { "python3" }; + let hook_script = home.path().join("memory_hook.py"); + let hook_log = home.path().join("stop"); + let notify_log = home.path().join("notify"); + let managed_log = home.path().join("managed"); + let response = managed_response.unwrap_or("{}"); + tokio::fs::write( + &hook_script, + format!( + r#"import json +import sys +from pathlib import Path + +kind = sys.argv[1] +with Path(__file__).with_name(kind).open("a", newline="\n") as log: + log.write("hook invoked\n") +if kind == "managed": + print({response:?}) +elif kind == "stop" and Path(json.load(sys.stdin)["cwd"]).name == "memories": + print('{{"decision":"block","reason":"Keep working on the project."}}') +"# + ), + ) + .await?; + tokio::fs::write( + home.path().join("hooks.json"), + serde_json::json!({"hooks": {"Stop": [{"hooks": [{ + "type": "command", + "command": format!("{python} \"{}\" stop", hook_script.display()), + }]}]}}) + .to_string(), + ) + .await?; + let mut requirements = if pin_hooks { + "[features]\nhooks = true\n" + } else { + "" + } + .to_string(); + if managed_response.is_some() { + let command = + serde_json::to_string(&format!("{python} \"{}\" managed", hook_script.display()))?; + requirements.push_str(&format!( + "\n[[hooks.Stop]]\n[[hooks.Stop.hooks]]\ntype = \"command\"\ncommand = {command}\n" + )); + } + let test = test_codex() + .with_home(home) + .with_cloud_config_bundle( + CloudConfigBundleFixture::loader_with_enterprise_requirement(requirements), + ) + .with_config(move |config| { + config + .features + .enable(Feature::Sqlite) + .expect("enable SQLite"); + config.memories = startup_test_memories_config(); + config.notify = Some(vec![ + python.to_string(), + hook_script.display().to_string(), + "notify".to_string(), + ]); + trust_discovered_hooks(config); + }) + // Command hooks run on the app host, so the parent needs a host-local cwd. + .build(&server) + .await?; + let phase2 = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-phase2"), + ev_assistant_message("msg-phase2", "phase2 complete"), + ev_completed("resp-phase2"), + ]), + ) + .await; + + trigger_memories_startup(&test).await; + wait_for_single_request(&phase2).await; + if managed_response.is_some() { + assert_eq!( + wait_for_phase2_job_to_finish(db.as_ref()).await?, + Phase2JobClaimOutcome::SkippedRetryUnavailable + ); + assert_eq!( + tokio::fs::read_to_string(managed_log).await?, + "hook invoked\n" + ); + } else { + wait_for_phase2_workspace_reset(db.as_ref(), &root).await?; + } + phase2.single_request(); + assert!( + !hook_log.exists(), + "memory worker must not run the user Stop hook" + ); + assert!( + !notify_log.exists(), + "memory worker must not send legacy notifications" + ); + + if managed_response.is_none() { + let parent = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-parent"), + ev_assistant_message("msg-parent", "parent complete"), + ev_completed("resp-parent"), + ]), + ) + .await; + test.submit_turn("parent task").await?; + parent.single_request(); + assert_eq!(tokio::fs::read_to_string(hook_log).await?, "hook invoked\n"); + fs_wait::wait_for_path_exists(notify_log, Duration::from_secs(10)).await?; + } + shutdown_test_codex(&test).await?; + } + Ok(()) +} + #[tokio::test] async fn memories_startup_phase2_tracks_workspace_diff_across_runs() -> anyhow::Result<()> { let server = start_mock_server().await;