diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index f69322f8ec..bec80224c0 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1924,7 +1924,6 @@ dependencies = [ "codex-git-utils", "codex-hooks", "codex-instructions", - "codex-keyring-store", "codex-login", "codex-mcp", "codex-model-provider-info", diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index d2296c75c5..6c85612bc9 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -218,7 +218,9 @@ impl ThreadHistoryBuilder { RolloutItem::EventMsg(event) => self.handle_event(event), RolloutItem::Compacted(payload) => self.handle_compacted(payload), RolloutItem::ResponseItem(item) => self.handle_response_item(item), - RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {} + RolloutItem::TurnContext(_) + | RolloutItem::SessionMeta(_) + | RolloutItem::SessionState(_) => {} } } diff --git a/codex-rs/app-server/tests/suite/v2/client_metadata.rs b/codex-rs/app-server/tests/suite/v2/client_metadata.rs index c85febd7d4..e343e08470 100644 --- a/codex-rs/app-server/tests/suite/v2/client_metadata.rs +++ b/codex-rs/app-server/tests/suite/v2/client_metadata.rs @@ -18,7 +18,7 @@ use std::path::Path; use tempfile::TempDir; use tokio::time::timeout; -const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20); #[tokio::test] async fn turn_start_forwards_client_metadata_to_responses_request_v2() -> Result<()> { diff --git a/codex-rs/app-server/tests/suite/v2/collaboration_mode_list.rs b/codex-rs/app-server/tests/suite/v2/collaboration_mode_list.rs index 7c36827e6d..0dd4e8174a 100644 --- a/codex-rs/app-server/tests/suite/v2/collaboration_mode_list.rs +++ b/codex-rs/app-server/tests/suite/v2/collaboration_mode_list.rs @@ -21,7 +21,7 @@ use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::time::timeout; -const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10); +const DEFAULT_TIMEOUT: Duration = Duration::from_secs(20); /// Confirms the server returns the default collaboration mode presets in a stable order. #[tokio::test] diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index 30caa13761..e6818eabd7 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -47,7 +47,7 @@ use tokio_tungstenite::tungstenite::http::HeaderValue; use tokio_tungstenite::tungstenite::http::header::AUTHORIZATION; use tokio_tungstenite::tungstenite::http::header::ORIGIN; -pub(super) const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10); +pub(super) const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(20); pub(super) type WsClient = WebSocketStream>; type HmacSha256 = Hmac; diff --git a/codex-rs/app-server/tests/suite/v2/initialize.rs b/codex-rs/app-server/tests/suite/v2/initialize.rs index 165160468f..cfb08a1718 100644 --- a/codex-rs/app-server/tests/suite/v2/initialize.rs +++ b/codex-rs/app-server/tests/suite/v2/initialize.rs @@ -24,7 +24,7 @@ use std::time::Duration; use tempfile::TempDir; use tokio::time::timeout; -const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20); #[tokio::test] async fn initialize_uses_client_info_name_as_originator() -> Result<()> { diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index efc167c9d0..4c1b80498c 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -69,7 +69,7 @@ use wiremock::matchers::method; use wiremock::matchers::path; use wiremock::matchers::path_regex; -const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10); +const DEFAULT_TIMEOUT: Duration = Duration::from_secs(20); const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex."; const V2_STEERING_ACKNOWLEDGEMENT: &str = "This was sent to steer the previous background agent task."; @@ -1574,7 +1574,7 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<( create_shell_command_sse_response( realtime_tool_ok_command(), /*workdir*/ None, - Some(5000), + Some(10_000), "shell_call", )?, create_final_assistant_message_sse_response("shell tool finished")?, diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 7b3acd6460..e45bf3d258 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -147,7 +147,6 @@ codex-shell-escalation = { workspace = true } assert_cmd = { workspace = true } assert_matches = { workspace = true } codex-arg0 = { workspace = true } -codex-keyring-store = { workspace = true } codex-otel = { workspace = true } codex-utils-cargo-bin = { workspace = true } core_test_support = { workspace = true } diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 509a578f5c..d7a6403c93 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -116,6 +116,7 @@ fn keep_forked_rollout_item(item: &RolloutItem) -> bool { | ResponseItem::Compaction { .. } | ResponseItem::Other, ) => false, + RolloutItem::SessionState(_) => false, RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index d332853167..e3070c6401 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -186,7 +186,7 @@ async fn wait_for_subagent_notification(parent_thread: &Arc) -> boo sleep(Duration::from_millis(25)).await; } }; - timeout(Duration::from_secs(2), wait).await.is_ok() + timeout(Duration::from_secs(5), wait).await.is_ok() } async fn persist_thread_for_tree_resume(thread: &Arc, message: &str) { diff --git a/codex-rs/core/src/agent_identity.rs b/codex-rs/core/src/agent_identity.rs index d799f15846..71c759e59e 100644 --- a/codex-rs/core/src/agent_identity.rs +++ b/codex-rs/core/src/agent_identity.rs @@ -153,14 +153,16 @@ impl AgentIdentityManager { Ok(stored_identity) } - pub(crate) async fn task_matches_current_binding(&self, task: &RegisteredAgentTask) -> bool { + pub(crate) async fn task_matches_current_identity(&self, task: &RegisteredAgentTask) -> bool { if !self.feature_enabled { return false; } - self.current_auth_binding() + self.current_stored_identity() .await - .is_some_and(|(_, binding)| task.matches_binding(&binding)) + .is_some_and(|stored_identity| { + stored_identity.agent_runtime_id == task.agent_runtime_id + }) } async fn current_auth_binding(&self) -> Option<(CodexAuth, AgentIdentityBinding)> { @@ -177,6 +179,11 @@ impl AgentIdentityManager { binding.map(|binding| (auth, binding)) } + async fn current_stored_identity(&self) -> Option { + let (auth, binding) = self.current_auth_binding().await?; + self.load_stored_identity(&auth, &binding).ok().flatten() + } + async fn register_agent_identity( &self, binding: &AgentIdentityBinding, diff --git a/codex-rs/core/src/agent_identity/task_registration.rs b/codex-rs/core/src/agent_identity/task_registration.rs index 30d2567f28..9bbc364db3 100644 --- a/codex-rs/core/src/agent_identity/task_registration.rs +++ b/codex-rs/core/src/agent_identity/task_registration.rs @@ -2,6 +2,7 @@ use std::time::Duration; use anyhow::Context; use anyhow::Result; +use codex_protocol::protocol::SessionAgentTask; use crypto_box::SecretKey as Curve25519SecretKey; use ed25519_dalek::Signer as _; use serde::Deserialize; @@ -16,9 +17,6 @@ const AGENT_TASK_REGISTRATION_TIMEOUT: Duration = Duration::from_secs(15); #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct RegisteredAgentTask { - pub(crate) binding_id: String, - pub(crate) chatgpt_account_id: String, - pub(crate) chatgpt_user_id: Option, pub(crate) agent_runtime_id: String, pub(crate) task_id: String, pub(crate) registered_at: String, @@ -82,9 +80,6 @@ impl AgentIdentityManager { .await .with_context(|| format!("failed to parse agent task response from {url}"))?; let registered_task = RegisteredAgentTask { - binding_id: stored_identity.binding_id.clone(), - chatgpt_account_id: stored_identity.chatgpt_account_id.clone(), - chatgpt_user_id: stored_identity.chatgpt_user_id.clone(), agent_runtime_id: stored_identity.agent_runtime_id.clone(), task_id: decrypt_task_id_response( &stored_identity, @@ -107,18 +102,20 @@ impl AgentIdentityManager { } impl RegisteredAgentTask { - pub(super) fn matches_binding(&self, binding: &AgentIdentityBinding) -> bool { - binding.matches_parts( - &self.binding_id, - &self.chatgpt_account_id, - self.chatgpt_user_id.as_deref(), - ) + pub(crate) fn to_session_agent_task(&self) -> SessionAgentTask { + SessionAgentTask { + agent_runtime_id: self.agent_runtime_id.clone(), + task_id: self.task_id.clone(), + registered_at: self.registered_at.clone(), + } } - pub(crate) fn has_same_binding(&self, other: &Self) -> bool { - self.binding_id == other.binding_id - && self.chatgpt_account_id == other.chatgpt_account_id - && self.chatgpt_user_id == other.chatgpt_user_id + pub(crate) fn from_session_agent_task(task: SessionAgentTask) -> Self { + Self { + agent_runtime_id: task.agent_runtime_id, + task_id: task.task_id, + registered_at: task.registered_at, + } } } @@ -242,9 +239,6 @@ mod tests { assert_eq!( task, RegisteredAgentTask { - binding_id: "chatgpt-account-account-123".to_string(), - chatgpt_account_id: "account-123".to_string(), - chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent-123".to_string(), task_id: "task_123".to_string(), registered_at: task.registered_at.clone(), @@ -331,9 +325,6 @@ mod tests { assert_eq!( task, RegisteredAgentTask { - binding_id: "chatgpt-account-account-123".to_string(), - chatgpt_account_id: "account-123".to_string(), - chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent-123".to_string(), task_id: "task_123".to_string(), registered_at: task.registered_at.clone(), @@ -342,7 +333,7 @@ mod tests { } #[tokio::test] - async fn task_matches_current_binding_rejects_stale_auth_binding() { + async fn task_matches_current_identity_rejects_stale_registered_identity() { let auth_manager = AuthManager::from_auth_for_testing(make_chatgpt_auth("account-456", Some("user-456"))); let manager = AgentIdentityManager::new_for_tests( @@ -352,15 +343,12 @@ mod tests { SessionSource::Cli, ); let task = RegisteredAgentTask { - binding_id: "chatgpt-account-account-123".to_string(), - chatgpt_account_id: "account-123".to_string(), - chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent-123".to_string(), task_id: "task_123".to_string(), registered_at: "2026-03-23T12:00:00Z".to_string(), }; - assert!(!manager.task_matches_current_binding(&task).await); + assert!(!manager.task_matches_current_identity(&task).await); } async fn mount_human_biscuit( diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 6128f27035..a2398849df 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -121,6 +121,7 @@ use codex_protocol::protocol::RawResponseItemEvent; use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SessionStateUpdate; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnContextItem; @@ -845,6 +846,7 @@ pub(crate) struct Session { pub(crate) services: SessionServices, js_repl: Arc, next_internal_sub_id: AtomicU64, + agent_task_registration_lock: Mutex<()>, } #[derive(Clone, Debug)] @@ -1529,7 +1531,68 @@ impl Session { handlers::shutdown(self, self.next_internal_sub_id()).await; } - async fn cached_agent_task_for_current_binding(&self) -> Option { + fn latest_persisted_agent_task( + rollout_items: &[RolloutItem], + ) -> Option> { + rollout_items.iter().rev().find_map(|item| match item { + RolloutItem::SessionState(update) => Some( + update + .agent_task + .clone() + .map(RegisteredAgentTask::from_session_agent_task), + ), + _ => None, + }) + } + + async fn restore_persisted_agent_task(&self, rollout_items: &[RolloutItem]) { + let Some(agent_task) = Self::latest_persisted_agent_task(rollout_items).flatten() else { + return; + }; + + let mut state = self.state.lock().await; + state.set_agent_task(agent_task); + } + + async fn persist_agent_task_update(&self, agent_task: Option<&RegisteredAgentTask>) { + self.persist_rollout_items(&[RolloutItem::SessionState(SessionStateUpdate { + agent_task: agent_task.map(RegisteredAgentTask::to_session_agent_task), + })]) + .await; + } + + async fn clear_cached_agent_task(&self, agent_task: &RegisteredAgentTask) { + let cleared = { + let mut state = self.state.lock().await; + if state.agent_task().as_ref() == Some(agent_task) { + state.clear_agent_task(); + true + } else { + false + } + }; + if cleared { + self.persist_agent_task_update(/*agent_task*/ None).await; + } + } + + async fn cache_agent_task(&self, agent_task: RegisteredAgentTask) -> RegisteredAgentTask { + let changed = { + let mut state = self.state.lock().await; + if state.agent_task().as_ref() == Some(&agent_task) { + false + } else { + state.set_agent_task(agent_task.clone()); + true + } + }; + if changed { + self.persist_agent_task_update(Some(&agent_task)).await; + } + agent_task + } + + async fn cached_agent_task_for_current_identity(&self) -> Option { let agent_task = { let state = self.state.lock().await; state.agent_task() @@ -1538,7 +1601,7 @@ impl Session { if self .services .agent_identity_manager - .task_matches_current_binding(&agent_task) + .task_matches_current_identity(&agent_task) .await { debug!( @@ -1552,17 +1615,19 @@ impl Session { debug!( agent_runtime_id = %agent_task.agent_runtime_id, task_id = %agent_task.task_id, - "discarding cached agent task because auth binding changed" + "discarding cached agent task because the registered agent identity changed" ); - let mut state = self.state.lock().await; - if state.agent_task().as_ref() == Some(&agent_task) { - state.clear_agent_task(); - } + self.clear_cached_agent_task(&agent_task).await; None } async fn ensure_agent_task_registered(&self) -> anyhow::Result> { - if let Some(agent_task) = self.cached_agent_task_for_current_binding().await { + if let Some(agent_task) = self.cached_agent_task_for_current_identity().await { + return Ok(Some(agent_task)); + } + + let _guard = self.agent_task_registration_lock.lock().await; + if let Some(agent_task) = self.cached_agent_task_for_current_identity().await { return Ok(Some(agent_task)); } @@ -1575,31 +1640,18 @@ impl Session { if !self .services .agent_identity_manager - .task_matches_current_binding(&agent_task) + .task_matches_current_identity(&agent_task) .await { debug!( agent_runtime_id = %agent_task.agent_runtime_id, task_id = %agent_task.task_id, - "discarding newly registered agent task because auth binding changed" + "discarding newly registered agent task because the registered agent identity changed" ); continue; } - { - let mut state = self.state.lock().await; - if let Some(existing_agent_task) = state.agent_task() { - if existing_agent_task.has_same_binding(&agent_task) { - return Ok(Some(existing_agent_task)); - } - debug!( - agent_runtime_id = %existing_agent_task.agent_runtime_id, - task_id = %existing_agent_task.task_id, - "replacing cached agent task because auth binding changed" - ); - } - state.set_agent_task(agent_task.clone()); - } + let agent_task = self.cache_agent_task(agent_task).await; info!( thread_id = %self.conversation_id, @@ -2221,6 +2273,7 @@ impl Session { services, js_repl, next_internal_sub_id: AtomicU64::new(0), + agent_task_registration_lock: Mutex::new(()), }); if let Some(network_policy_decider_session) = network_policy_decider_session { let mut guard = network_policy_decider_session.write().await; @@ -2490,6 +2543,7 @@ impl Session { } InitialHistory::Resumed(resumed_history) => { let rollout_items = resumed_history.history; + self.restore_persisted_agent_task(&rollout_items).await; let previous_turn_settings = self .apply_rollout_reconstruction(&turn_context, &rollout_items) .await; diff --git a/codex-rs/core/src/codex/rollout_reconstruction.rs b/codex-rs/core/src/codex/rollout_reconstruction.rs index a4c042af0c..3e407c4cd7 100644 --- a/codex-rs/core/src/codex/rollout_reconstruction.rs +++ b/codex-rs/core/src/codex/rollout_reconstruction.rs @@ -207,7 +207,9 @@ impl Session { active_segment.get_or_insert_with(ActiveReplaySegment::default); active_segment.counts_as_user_turn |= is_user_turn_boundary(response_item); } - RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => {} + RolloutItem::EventMsg(_) + | RolloutItem::SessionMeta(_) + | RolloutItem::SessionState(_) => {} } if base_replacement_history.is_some() @@ -275,6 +277,7 @@ impl Session { history.drop_last_n_user_turns(rollback.num_turns); } RolloutItem::EventMsg(_) + | RolloutItem::SessionState(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {} } diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index c4c5000749..9e5957b003 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -1077,6 +1077,67 @@ async fn record_initial_history_reconstructs_resumed_transcript() { assert_eq!(expected, history.raw_items()); } +#[tokio::test] +async fn record_initial_history_restores_latest_persisted_agent_task() { + let (session, _turn_context) = make_session_and_context().await; + let expected = RegisteredAgentTask { + agent_runtime_id: "agent-123".to_string(), + task_id: "task-123".to_string(), + registered_at: "2026-03-23T12:00:00Z".to_string(), + }; + let rollout_items = vec![ + RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { + agent_task: Some(expected.to_session_agent_task()), + }), + RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { + agent_task: None, + }), + RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { + agent_task: Some(expected.to_session_agent_task()), + }), + ]; + + session + .record_initial_history(InitialHistory::Resumed(ResumedHistory { + conversation_id: ThreadId::default(), + history: rollout_items, + rollout_path: PathBuf::from("/tmp/resume.jsonl"), + })) + .await; + + assert_eq!(session.state.lock().await.agent_task(), Some(expected)); +} + +#[tokio::test] +async fn record_initial_history_honors_cleared_persisted_agent_task() { + let (session, _turn_context) = make_session_and_context().await; + let rollout_items = vec![ + RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { + agent_task: Some( + RegisteredAgentTask { + agent_runtime_id: "agent-123".to_string(), + task_id: "task-123".to_string(), + registered_at: "2026-03-23T12:00:00Z".to_string(), + } + .to_session_agent_task(), + ), + }), + RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { + agent_task: None, + }), + ]; + + session + .record_initial_history(InitialHistory::Resumed(ResumedHistory { + conversation_id: ThreadId::default(), + history: rollout_items, + rollout_path: PathBuf::from("/tmp/resume.jsonl"), + })) + .await; + + assert_eq!(session.state.lock().await.agent_task(), None); +} + #[tokio::test] async fn record_initial_history_new_defers_initial_context_until_first_turn() { let (session, _turn_context) = make_session_and_context().await; @@ -2954,6 +3015,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { services, js_repl, next_internal_sub_id: AtomicU64::new(0), + agent_task_registration_lock: Mutex::new(()), }; (session, turn_context) @@ -3804,6 +3866,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( services, js_repl, next_internal_sub_id: AtomicU64::new(0), + agent_task_registration_lock: Mutex::new(()), }); (session, turn_context, rx_event) diff --git a/codex-rs/core/src/state/session_tests.rs b/codex-rs/core/src/state/session_tests.rs index 6816c8731d..d5ae7fcde6 100644 --- a/codex-rs/core/src/state/session_tests.rs +++ b/codex-rs/core/src/state/session_tests.rs @@ -39,10 +39,7 @@ async fn set_agent_task_persists_plaintext_task_for_session_reuse() { let session_configuration = make_session_configuration_for_tests().await; let mut state = SessionState::new(session_configuration); let agent_task = RegisteredAgentTask { - binding_id: "chatgpt-account-account-123".to_string(), - chatgpt_account_id: "account-123".to_string(), - chatgpt_user_id: Some("user-123".to_string()), - agent_runtime_id: "agent_123".to_string(), + agent_runtime_id: "agent-123".to_string(), task_id: "task_123".to_string(), registered_at: "2026-03-23T12:00:00Z".to_string(), }; @@ -57,9 +54,6 @@ async fn clear_agent_task_removes_cached_task() { let session_configuration = make_session_configuration_for_tests().await; let mut state = SessionState::new(session_configuration); let agent_task = RegisteredAgentTask { - binding_id: "chatgpt-account-account-123".to_string(), - chatgpt_account_id: "account-123".to_string(), - chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent_123".to_string(), task_id: "task_123".to_string(), registered_at: "2026-03-23T12:00:00Z".to_string(), diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 65d6dab04c..8fb4dc6138 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2666,6 +2666,21 @@ impl fmt::Display for SubAgentSource { } } +/// Persisted agent-task details that let a resumed thread keep using the same backend task. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema, TS)] +pub struct SessionAgentTask { + pub agent_runtime_id: String, + pub task_id: String, + pub registered_at: String, +} + +/// Session-scoped state updates that can be appended after the canonical SessionMeta line. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema, TS, Default)] +pub struct SessionStateUpdate { + #[serde(default)] + pub agent_task: Option, +} + /// SessionMeta contains session-level data that doesn't correspond to a specific turn. /// /// NOTE: There used to be an `instructions` field here, which stored user_instructions, but we @@ -2735,6 +2750,7 @@ pub struct SessionMetaLine { #[serde(tag = "type", content = "payload", rename_all = "snake_case")] pub enum RolloutItem { SessionMeta(SessionMetaLine), + SessionState(SessionStateUpdate), ResponseItem(ResponseItem), Compacted(CompactedItem), TurnContext(TurnContextItem), diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index e7d3dae5de..78510b8cac 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -1063,6 +1063,9 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { // Not included in `head`; skip. } + RolloutItem::SessionState(_) => { + // Not included in `head`; skip. + } RolloutItem::EventMsg(ev) => { if let EventMsg::UserMessage(user) = ev { summary.saw_user_event = true; @@ -1115,6 +1118,7 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result {} } diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index e8a95839d7..be920b79e2 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -70,7 +70,8 @@ pub fn builder_from_items( ) -> Option { if let Some(session_meta) = items.iter().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => Some(meta_line), - RolloutItem::ResponseItem(_) + RolloutItem::SessionState(_) + | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -124,7 +125,8 @@ pub async fn extract_metadata_from_rollout( metadata, memory_mode: items.iter().rev().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), - RolloutItem::ResponseItem(_) + RolloutItem::SessionState(_) + | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 4b50781e76..6e6fb8b7a2 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -16,9 +16,10 @@ pub fn is_persisted_response_item(item: &RolloutItem, mode: EventPersistenceMode RolloutItem::ResponseItem(item) => should_persist_response_item(item), RolloutItem::EventMsg(ev) => should_persist_event_msg(ev, mode), // Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns). - RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => { - true - } + RolloutItem::Compacted(_) + | RolloutItem::TurnContext(_) + | RolloutItem::SessionMeta(_) + | RolloutItem::SessionState(_) => true, } } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 01ea10cb95..309865613c 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -683,6 +683,9 @@ impl RolloutRecorder { RolloutItem::Compacted(item) => { items.push(RolloutItem::Compacted(item)); } + RolloutItem::SessionState(update) => { + items.push(RolloutItem::SessionState(update)); + } RolloutItem::TurnContext(item) => { items.push(RolloutItem::TurnContext(item)); } @@ -1303,6 +1306,7 @@ async fn resume_candidate_matches_cwd( && let Some(latest_turn_context_cwd) = items.iter().rev().find_map(|item| match item { RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()), RolloutItem::SessionMeta(_) + | RolloutItem::SessionState(_) | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) => None, diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 91c6755ffe..9294637361 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -19,6 +19,7 @@ pub fn apply_rollout_item( ) { match item { RolloutItem::SessionMeta(meta_line) => apply_session_meta_from_item(metadata, meta_line), + RolloutItem::SessionState(_) => {} RolloutItem::TurnContext(turn_ctx) => apply_turn_context(metadata, turn_ctx), RolloutItem::EventMsg(event) => apply_event_msg(metadata, event), RolloutItem::ResponseItem(item) => apply_response_item(metadata, item), @@ -36,9 +37,10 @@ pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool { RolloutItem::EventMsg( EventMsg::TokenCount(_) | EventMsg::UserMessage(_) | EventMsg::ThreadNameUpdated(_), ) => true, - RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) => { - false - } + RolloutItem::SessionState(_) + | RolloutItem::EventMsg(_) + | RolloutItem::ResponseItem(_) + | RolloutItem::Compacted(_) => false, } } diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index e56ca3f386..0ae73bcc14 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -912,7 +912,8 @@ fn one_thread_id_from_rows( pub(super) fn extract_dynamic_tools(items: &[RolloutItem]) -> Option>> { items.iter().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.dynamic_tools.clone()), - RolloutItem::ResponseItem(_) + RolloutItem::SessionState(_) + | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -922,7 +923,8 @@ pub(super) fn extract_dynamic_tools(items: &[RolloutItem]) -> Option Option { items.iter().rev().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), - RolloutItem::ResponseItem(_) + RolloutItem::SessionState(_) + | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, diff --git a/notes/pr17387-agent-task-persistence-assumption.md b/notes/pr17387-agent-task-persistence-assumption.md new file mode 100644 index 0000000000..9b1059ebbc --- /dev/null +++ b/notes/pr17387-agent-task-persistence-assumption.md @@ -0,0 +1,15 @@ +# PR 17387 Agent Task Persistence Assumption + +The review request said to persist the registered agent task_id "in the same place that other +session details are stored" without specifying the exact persistence shape. + +Assumption used in this patch: + +- Persist session-scoped agent task state in the thread rollout file, alongside other resumable + thread/session data. +- Keep agent identity registration on startup, but register the backend task lazily when a session + first needs task-scoped auth. +- Model task updates as an append-only RolloutItem::SessionState record instead of mutating the + canonical first SessionMeta line, because resumed threads need later updates and clears to win. +- Do not carry auth-binding fields on the persisted session task; only persist the task fields + needed to resume the same backend task for the same stored registered agent identity.