diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index d54603a3d0..6de8984039 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -4051,14 +4051,16 @@ impl Session { } pub(crate) async fn hook_transcript_path(&self) -> Option { - self.ensure_rollout_materialized().await; - match self.current_rollout_path().await { - Ok(path) => path, + let rollout_path = match self.current_rollout_path().await { + Ok(Some(path)) => path, + Ok(None) => return None, Err(err) => { warn!("{err}"); - None + return None; } - } + }; + self.ensure_rollout_materialized().await; + Some(rollout_path) } pub(crate) async fn take_pending_session_start_source( diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 76f5f5484f..515d189e91 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -4290,7 +4290,7 @@ async fn wait_for_thread_rollback_failed(rx: &async_channel::Receiver) -> } } -async fn attach_thread_persistence(session: &mut Session) -> PathBuf { +async fn open_thread_persistence(session: &mut Session) -> PathBuf { let config = session.get_config().await; let live_thread = LiveThread::create( Arc::clone(&session.services.thread_store), @@ -4325,11 +4325,6 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf { .await .expect("create thread persistence"); session.services.live_thread = Some(live_thread); - session.ensure_rollout_materialized().await; - session - .flush_rollout() - .await - .expect("attached rollout should flush"); session .current_rollout_path() .await @@ -4337,6 +4332,16 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf { .expect("thread should have rollout path") } +async fn attach_thread_persistence(session: &mut Session) -> PathBuf { + let rollout_path = open_thread_persistence(session).await; + session.ensure_rollout_materialized().await; + session + .flush_rollout() + .await + .expect("attached rollout should flush"); + rollout_path +} + fn text_block(s: &str) -> serde_json::Value { json!({ "type": "text", @@ -9621,6 +9626,41 @@ async fn attach_in_memory_thread_store( store } +#[tokio::test] +async fn hook_transcript_path_does_not_persist_non_local_thread_store() { + let (mut session, _) = make_session_and_context().await; + let store = attach_in_memory_thread_store(&mut session).await; + + assert_eq!(session.hook_transcript_path().await, None); + assert_eq!( + store.calls().await, + codex_thread_store::InMemoryThreadStoreCalls { + create_thread: 1, + ..Default::default() + } + ); +} + +#[tokio::test] +async fn hook_transcript_path_materializes_lazy_local_thread() { + let (mut session, _) = make_session_and_context().await; + let rollout_path = open_thread_persistence(&mut session).await; + assert!(!rollout_path.exists()); + + assert_eq!( + session.hook_transcript_path().await, + Some(rollout_path.clone()) + ); + let (items, thread_id, parse_errors) = RolloutRecorder::load_rollout_items(&rollout_path) + .await + .expect("read materialized rollout"); + assert_eq!((thread_id, parse_errors), (Some(session.thread_id), 0)); + assert!(matches!( + items.as_slice(), + [RolloutItem::SessionMeta(meta)] if meta.meta.id == session.thread_id + )); +} + async fn wait_for_flush_count( store: &codex_thread_store::InMemoryThreadStore, expected_flushes: usize, diff --git a/codex-rs/core/tests/suite/hooks.rs b/codex-rs/core/tests/suite/hooks.rs index 7f7d87b8a4..865ba7e01b 100644 --- a/codex-rs/core/tests/suite/hooks.rs +++ b/codex-rs/core/tests/suite/hooks.rs @@ -6,6 +6,7 @@ use anyhow::Result; use codex_core::StartThreadOptions; use codex_core::config::Config; use codex_core::config::Constrained; +use codex_core::config::ThreadStoreConfig; use codex_features::Feature; use codex_model_provider_info::ModelProviderInfo; use codex_model_provider_info::built_in_model_providers; @@ -28,6 +29,7 @@ use codex_protocol::request_permissions::PermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile; use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::user_input::UserInput; +use codex_thread_store::InMemoryThreadStore; use codex_utils_absolute_path::AbsolutePathBuf; use core_test_support::TestTargetOs; use core_test_support::hooks::trust_discovered_hooks; @@ -772,9 +774,14 @@ raise SystemExit(2) Ok(()) } -fn write_session_start_hook_recording_transcript(home: &Path) -> Result<()> { +fn write_session_start_hook_recording_transcript(home: &Path, stop: bool) -> Result<()> { let script_path = home.join("session_start_hook.py"); let log_path = home.join("session_start_hook_log.jsonl"); + let stop_output = if stop { + r#"print(json.dumps({"continue": False, "stopReason": "integration assertion complete"}))"# + } else { + "" + }; let script = format!( r#"import json from pathlib import Path @@ -782,15 +789,18 @@ import sys payload = json.load(sys.stdin) transcript_path = payload.get("transcript_path") -record = {{ - "transcript_path": transcript_path, - "exists": Path(transcript_path).exists() if transcript_path else False, -}} +if transcript_path is not None and not Path(transcript_path).is_file(): + raise RuntimeError( + f"transcript did not exist when the hook ran: {{transcript_path}}" + ) with Path(r"{log_path}").open("a", encoding="utf-8") as handle: - handle.write(json.dumps(record) + "\n") + handle.write(json.dumps(payload) + "\n") + +{stop_output} "#, log_path = log_path.display(), + stop_output = stop_output, ); let hooks = serde_json::json!({ "hooks": { @@ -1328,7 +1338,7 @@ async fn session_start_hook_sees_materialized_transcript_path() -> Result<()> { let mut builder = test_codex() .with_pre_build_hook(|home| { - write_session_start_hook_recording_transcript(home) + write_session_start_hook_recording_transcript(home, /*stop*/ false) .expect("failed to write session start hook test fixture"); }) .with_config(trust_discovered_hooks); @@ -1338,14 +1348,42 @@ async fn session_start_hook_sees_materialized_transcript_path() -> Result<()> { let hook_inputs = read_session_start_hook_inputs(test.codex_home_path())?; assert_eq!(hook_inputs.len(), 1); - assert_eq!( - hook_inputs[0] - .get("transcript_path") - .and_then(Value::as_str) - .map(str::is_empty), - Some(false) + let transcript_path = hook_inputs[0]["transcript_path"] + .as_str() + .expect("local session start hook transcript_path should be a string"); + assert!( + Path::new(transcript_path).exists(), + "local session start hook transcript_path should be materialized", ); - assert_eq!(hook_inputs[0].get("exists"), Some(&Value::Bool(true))); + + Ok(()) +} + +#[tokio::test] +async fn session_start_hook_skips_persistence_for_non_local_thread_store() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let store_id = uuid::Uuid::new_v4().to_string(); + let store = InMemoryThreadStore::for_id(store_id.clone()); + let mut builder = test_codex() + .with_pre_build_hook(|home| { + write_session_start_hook_recording_transcript(home, /*stop*/ true) + .expect("failed to write session start hook test fixture"); + }) + .with_config(move |config| { + config.experimental_thread_store = ThreadStoreConfig::InMemory { id: store_id }; + trust_discovered_hooks(config); + }); + let test = builder.build(&server).await?; + + test.submit_turn("run the non-local session start hook") + .await?; + + let hook_inputs = read_session_start_hook_inputs(test.codex_home_path())?; + assert_eq!(hook_inputs.len(), 1); + assert_eq!(hook_inputs[0]["transcript_path"], Value::Null); + assert_eq!(store.calls().await.persist_thread, 0); Ok(()) }