openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-09 20:50:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Scope stop hooks for memory consolidation (#40587)

Browse Source 
## Why

Background memory consolidation must honor managed completion policies without
running project completion checks or sending user-facing notifications.

## What changed

- Add a memory-consolidation stop target that selects managed policy hooks and
  executor-scoped cleanup while excluding user, project, session, and plugin
  stop hooks.
- End memory consolidation with an error when a managed stop hook rejects it,
  and propagate that turn error into agent status.
- Disable legacy completion notifications for the background memory worker.

## Testing

- Cover hook selection, executor cleanup, managed rejection, agent error
  status, and notification suppression.

GitOrigin-RevId: be350679ddcde2dde8a408b0972f08facff01a37
This commit is contained in:
jif
2026-08-25 10:29:26 +00:00
committed by copyberry
parent 4fa6ad1730
commit 7c6eb0eef1
8 changed files with 308 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control_tests.rs
+21 -11
View File
@@ -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]
codex-rs/core/src/agent/status.rs
+4 -1
View File
@@ -6,7 +6,10 @@ use codex_protocol::protocol::EventMsg;
pub(crate) fn agent_status_from_event(msg: &EventMsg) -> Option<AgentStatus> {
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 => {
codex-rs/core/src/hook_runtime.rs
+5
View File
@@ -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
codex-rs/core/src/session/turn.rs
+12
View File
@@ -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)
codex-rs/hooks/src/engine/mod_tests.rs
+52
View File
@@ -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 {
codex-rs/hooks/src/events/stop.rs
+65 -19
View File
@@ -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<ConfiguredHandler> {
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<HookRunSummary> {
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 {
codex-rs/memories/write/src/phase2.rs
+2
View File
@@ -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
codex-rs/memories/write/src/startup_tests.rs
+147
View File
@@ -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;