Expose retained command sessions in incremental world state (#59633)

## Why

A Code Mode script can start a command without returning its session handle. Make retained sessions visible to the model so it can find running commands and collect exited results with `write_stdin`.

## What changed

- Add the disabled-by-default `command_sessions_world_state` feature.
- Publish snapshots of up to 16 retained sessions with their IDs, running or exited status, and whether they are in use. Report omitted sessions and suppress unchanged updates.
- Preserve session visibility when `write_stdin` is unavailable and refresh snapshots after compaction. Clear stale session handles after restart or when the feature is disabled.

## Testing

Add an end-to-end scenario and model-visible history snapshot covering a dropped command handle, unchanged state, unavailable polling, compaction, exit and result collection, and stale-handle cleanup after restart.

GitOrigin-RevId: 45704d8c6403149997231aafb8496c12879aff23

codex-rs/core/src/context/world_state/command_sessions.rs

@@ -0,0 +1,150 @@
+//! Bounded command snapshots appended only when state changes.
+//! Live process state is authoritative; persisted snapshots only suppress duplicate updates.
+
+use super::PreviousSectionState;
+use super::SectionTransition;
+use super::WorldStateSection;
+use super::WorldStateUpdate;
+use crate::context::ContextualUserFragment;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ContentItemKind;
+use codex_protocol::models::input_text;
+use serde::Deserialize;
+use serde::Serialize;
+use std::collections::BTreeMap;
+
+const MAX_SESSIONS: usize = 16;
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub(crate) enum CommandSessionStatus {
+    Running,
+    Exited,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub(crate) struct CommandSession {
+    pub(crate) session_id: i32,
+    pub(crate) status: CommandSessionStatus,
+    pub(crate) in_use: bool,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+pub(crate) struct CommandSessionsState {
+    sessions: BTreeMap<i32, CommandSession>,
+    omitted: usize,
+    #[serde(default)]
+    pub(crate) polling_unavailable: bool,
+}
+
+impl CommandSessionsState {
+    pub(crate) fn new(sessions: impl Iterator<Item = CommandSession>) -> Self {
+        let mut state = Self::default();
+        for session in sessions {
+            state.sessions.insert(session.session_id, session);
+            if state.sessions.len() > MAX_SESSIONS {
+                state.sessions.pop_last();
+                state.omitted += 1;
+            }
+        }
+        state
+    }
+}
+
+impl WorldStateSection for CommandSessionsState {
+    const ID: &'static str = "command_sessions";
+    type Snapshot = Self;
+
+    fn has_retained_fragment_matcher() -> bool {
+        true
+    }
+
+    fn matches_retained_fragment(role: &str, text: &str) -> bool {
+        role == "developer" && Self::matches_text(text)
+    }
+
+    fn render_diff(
+        &self,
+        previous: PreviousSectionState<'_, Self::Snapshot>,
+    ) -> SectionTransition<Self::Snapshot> {
+        if matches!(previous, PreviousSectionState::Known(previous) if previous == self) {
+            return (None, Vec::new());
+        }
+        (
+            Some(self.clone()),
+            vec![WorldStateUpdate::fragment(self.clone())],
+        )
+    }
+}
+
+/// Clears stale handles when disabled, or preserves live handles when polling is unavailable.
+#[derive(Default)]
+pub(crate) struct DisabledCommandSessionsState(pub(crate) CommandSessionsState);
+
+impl WorldStateSection for DisabledCommandSessionsState {
+    const ID: &'static str = CommandSessionsState::ID;
+    type Snapshot = CommandSessionsState;
+
+    fn matches_legacy_fragment(role: &str, text: &str) -> bool {
+        CommandSessionsState::matches_retained_fragment(role, text)
+    }
+
+    fn render_diff(
+        &self,
+        previous: PreviousSectionState<'_, Self::Snapshot>,
+    ) -> SectionTransition<Self::Snapshot> {
+        match previous {
+            PreviousSectionState::Absent if self.0.sessions.is_empty() => (None, Vec::new()),
+            PreviousSectionState::Absent
+            | PreviousSectionState::Unknown
+            | PreviousSectionState::Known(_) => self.0.render_diff(previous),
+        }
+    }
+}
+
+impl ContextualUserFragment for CommandSessionsState {
+    fn role(&self) -> &'static str {
+        "developer"
+    }
+
+    fn content_kind(&self) -> ContentItemKind {
+        ContentItemKind("exec.command_sessions".to_string())
+    }
+
+    fn markers(&self) -> (&'static str, &'static str) {
+        Self::type_markers()
+    }
+
+    fn type_markers() -> (&'static str, &'static str) {
+        ("<command_sessions>", "</command_sessions>")
+    }
+
+    fn body(&self) -> ContentItem {
+        // Only harness-assigned IDs and fixed lifecycle labels belong in this
+        // developer fragment. At most 16 entries keep it below 2 KiB.
+        let mut text = String::from(
+            "Current command sessions for this thread (supersedes earlier snapshots). Exited sessions retain results for write_stdin. In-use sessions are temporarily busy with a tool interaction or result finalization.\n",
+        );
+        if self.polling_unavailable {
+            text.push_str("write_stdin is unavailable with the current tool set.\n");
+        }
+        if self.sessions.is_empty() && self.omitted == 0 {
+            text.push_str("None.");
+        }
+        for session in self.sessions.values() {
+            let id = session.session_id;
+            let status = match session.status {
+                CommandSessionStatus::Running => "running",
+                CommandSessionStatus::Exited => "exited; poll to collect",
+            };
+            let in_use = if session.in_use { "; in use" } else { "" };
+            text.push_str(&format!("{id}: {status}{in_use}\n"));
+        }
+        if self.omitted > 0 {
+            let omitted = self.omitted;
+            text.push_str(&format!(
+                "{omitted} additional sessions omitted; this snapshot is incomplete."
+            ));
+        }
+        input_text(format!("\n{}\n", text.trim_end()))
+    }
+}

codex-rs/core/src/context/world_state/mod.rs

@@ -1,6 +1,7 @@
 mod agents_md;
 mod apps_instructions;
 mod collaboration_mode;
+mod command_sessions;
 mod compact_permissions;
 mod context_window_guidance;
 mod environment;
@@ -36,12 +37,17 @@ use serde_json::Value;
 use sha1::Digest;
 use sha1::Sha1;
 use std::collections::BTreeMap;
+use std::collections::BTreeSet;
 use std::fmt;
 use std::sync::Arc;
 
 pub(crate) use agents_md::AgentsMdState;
 pub(crate) use apps_instructions::AppsInstructionsState;
 pub(crate) use collaboration_mode::CollaborationModeState;
+pub(crate) use command_sessions::CommandSession;
+pub(crate) use command_sessions::CommandSessionStatus;
+pub(crate) use command_sessions::CommandSessionsState;
+pub(crate) use command_sessions::DisabledCommandSessionsState;
 pub(crate) use compact_permissions::CompactPermissionsState;
 pub(crate) use context_window_guidance::ContextWindowGuidanceState;
 pub(crate) use environment::EnvironmentsState;
@@ -432,6 +438,9 @@ pub(crate) struct WorldState {
 #[serde(transparent)]
 pub(crate) struct WorldStateSnapshot {
     sections: BTreeMap<String, Value>,
+    // Rendering may need a refresh while persistence still needs the exact prior state.
+    #[serde(skip)]
+    invalidated_sections: BTreeSet<String>,
 }
 
 impl From<&Map<String, Value>> for WorldStateSnapshot {
@@ -441,6 +450,7 @@ impl From<&Map<String, Value>> for WorldStateSnapshot {
                 .iter()
                 .map(|(key, value)| (key.clone(), value.clone()))
                 .collect(),
+            invalidated_sections: BTreeSet::new(),
         }
     }
 }
@@ -448,6 +458,11 @@ impl From<&Map<String, Value>> for WorldStateSnapshot {
 impl WorldStateSnapshot {
     pub(crate) fn remove_section(&mut self, id: &str) {
         self.sections.remove(id);
+        self.invalidated_sections.remove(id);
+    }
+
+    pub(crate) fn invalidate_section(&mut self, id: &str) {
+        self.invalidated_sections.insert(id.to_string());
     }
 
     pub(crate) fn into_object(self) -> Map<String, Value> {
@@ -455,6 +470,9 @@ impl WorldStateSnapshot {
     }
 
     fn section(&self, id: &str) -> Option<&Value> {
+        if self.invalidated_sections.contains(id) {
+            return None;
+        }
         let snapshot = self.sections.get(id)?;
         if id == PluginsInstructionsState::ID
             && snapshot == &Value::Bool(false)
@@ -506,6 +524,7 @@ impl WorldStateSnapshot {
     pub(crate) fn apply_merge_patch(&mut self, patch: &Map<String, Value>) {
         // Borrow existing keys; only newly inserted sections need owned keys.
         for (key, value) in patch {
+            self.invalidated_sections.remove(key);
             if value.is_null() {
                 self.sections.remove(key);
             } else if let Some(current) = self.sections.get_mut(key) {

codex-rs/core/src/context/world_state/world_state_tests.rs

@@ -202,6 +202,7 @@ fn extension_owned_section_uses_its_snapshot_and_renderer() {
     ));
     let previous = WorldStateSnapshot {
         sections: BTreeMap::from([("extension_test".to_string(), json!({"value": "before"}))]),
+        ..Default::default()
     };
 
     let rendered = world_state
@@ -331,6 +332,7 @@ fn missing_retained_fragment_is_rendered_again() {
 fn extension_snapshot_updates_preserve_skipped_state() {
     let previous = WorldStateSnapshot {
         sections: BTreeMap::from([("extension_test".to_string(), json!({"published": true}))]),
+        ..Default::default()
     };
     let mut removed = WorldState::default();
     removed.add_extension_section(WorldStateSectionContribution::new("extension_test", |_| {
@@ -367,6 +369,7 @@ fn unreadable_section_snapshot_is_treated_as_unknown() {
     });
     let previous = WorldStateSnapshot {
         sections: BTreeMap::from([("test".to_string(), json!({"invalid": true}))]),
+        ..Default::default()
     };
 
     let rendered = current.render_history_fragment_diff(Some(&previous), &[]).1;
@@ -403,12 +406,14 @@ fn snapshot_merge_patch_changes_and_removes_nested_values() {
             ),
             ("removed_section".to_string(), json!({"value": true})),
         ]),
+        ..Default::default()
     };
     let current = WorldStateSnapshot {
         sections: BTreeMap::from([(
             "kept".to_string(),
             json!({"same": true, "changed": "after"}),
         )]),
+        ..Default::default()
     };
 
     assert_eq!(

codex-rs/core/src/session/mod.rs

@@ -2033,7 +2033,9 @@ impl Session {
             state.last_started_turn_id = last_started_turn_id;
             state.turn_attribution = turn_attribution;
             state.restored_turn_routing = restored_turn_routing;
-            if let Some(world_state) = world_state_baseline {
+            if let Some(mut world_state) = world_state_baseline {
+                // Handles belong to this runtime; a saved checkpoint can also lag its fragment.
+                world_state.invalidate_section("command_sessions");
                 state.history.set_world_state_baseline(world_state);
             }
             let fallback_ids = state.auto_compact_window_ids();

codex-rs/core/src/session/world_state.rs

@@ -171,6 +171,29 @@ impl Session {
             String::new()
         };
         let mut world_state = WorldState::default();
+        if turn_context
+            .config
+            .features
+            .enabled(Feature::CommandSessionsWorldState)
+        {
+            let mut sessions =
+                crate::unified_exec::command_sessions_state(&self.services.unified_exec_manager)
+                    .await;
+            if step_context
+                .tool_router
+                .exposes_tool(&ToolName::plain("write_stdin"))
+            {
+                world_state.add_section(sessions);
+            } else {
+                sessions.polling_unavailable = true;
+                world_state.add_section(crate::context::world_state::DisabledCommandSessionsState(
+                    sessions,
+                ));
+            }
+        } else {
+            world_state
+                .add_section(crate::context::world_state::DisabledCommandSessionsState::default());
+        }
         let extension_metrics = super::extension_metrics::from_session_telemetry(
             step_context.session_telemetry.clone(),
         );

codex-rs/core/src/unified_exec/command_sessions.rs

@@ -0,0 +1,29 @@
+//! Read-only snapshots of retained command sessions, including uncollected exits.
+//! Handles stay private until their initial exec_command collection finishes.
+
+use super::UnifiedExecProcessManager;
+use crate::context::world_state::CommandSession;
+use crate::context::world_state::CommandSessionStatus;
+use crate::context::world_state::CommandSessionsState;
+use std::sync::atomic::Ordering;
+
+pub(crate) async fn command_sessions_state(
+    manager: &UnifiedExecProcessManager,
+) -> CommandSessionsState {
+    let store = manager.process_store.lock().await;
+    CommandSessionsState::new(
+        store
+            .processes
+            .values()
+            .filter(|entry| !entry.initial_exec_command_active.load(Ordering::Acquire))
+            .map(|entry| CommandSession {
+                session_id: entry.process_id,
+                status: if entry.process.has_exited() {
+                    CommandSessionStatus::Exited
+                } else {
+                    CommandSessionStatus::Running
+                },
+                in_use: entry.process.interaction_lock().try_lock_owned().is_err(),
+            }),
+    )
+}

codex-rs/core/src/unified_exec/mod.rs

@@ -44,6 +44,7 @@ use crate::tools::network_approval::DeferredNetworkApproval;
 use codex_core_plugins::PluginMetricsSidecar;
 
 mod async_watcher;
+mod command_sessions;
 mod errors;
 mod oneshot;
 pub(crate) mod output;
@@ -58,6 +59,7 @@ pub(crate) fn set_deterministic_process_ids_for_tests(enabled: bool) {
     process_manager::set_deterministic_process_ids_for_tests(enabled);
 }
 
+pub(crate) use command_sessions::command_sessions_state;
 pub(crate) use errors::UnifiedExecError;
 pub(crate) use process::UnifiedExecProcess;
 pub(crate) use stdin_approval::TerminalPermissions;