Add thread history projection observers

codex-rs/Cargo.toml

@@ -111,6 +111,7 @@ members = [
     "terminal-detection",
     "test-binary-support",
     "thread-manager-sample",
+    "thread-store-protocol",
     "thread-store",
     "uds",
     "codex-experimental-api-macros",
@@ -217,6 +218,7 @@ codex-stdio-to-uds = { path = "stdio-to-uds" }
 codex-terminal-detection = { path = "terminal-detection" }
 codex-test-binary-support = { path = "test-binary-support" }
 codex-thread-store = { path = "thread-store" }
+codex-thread-store-protocol = { path = "thread-store-protocol" }
 codex-tools = { path = "tools" }
 codex-tui = { path = "tui" }
 codex-uds = { path = "uds" }

codex-rs/app-server-protocol/Cargo.toml

@@ -18,6 +18,7 @@ clap = { workspace = true, features = ["derive"] }
 codex-experimental-api-macros = { workspace = true }
 codex-protocol = { workspace = true }
 codex-shell-command = { workspace = true }
+codex-thread-store-protocol = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 schemars = { workspace = true }
 serde = { workspace = true, features = ["derive"] }

codex-rs/app-server-protocol/src/protocol/lifecycle_projection.rs

@@ -0,0 +1,154 @@
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RolloutItem;
+use codex_thread_store_protocol::LifecycleMutation;
+use codex_thread_store_protocol::StoredLifecycleProjectionState;
+use codex_thread_store_protocol::ThreadHistoryMutation;
+use codex_thread_store_protocol::ThreadHistoryMutationMetadata;
+use codex_thread_store_protocol::ThreadHistoryProjectionObserver;
+use serde_json::json;
+
+#[cfg(test)]
+#[path = "lifecycle_projection_tests.rs"]
+mod tests;
+
+/// Observes append batches and emits only thread lifecycle mutations.
+pub struct LifecycleProjectionObserver {
+    current_turn_id: Option<String>,
+}
+
+impl Default for LifecycleProjectionObserver {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl LifecycleProjectionObserver {
+    pub fn new() -> Self {
+        Self {
+            current_turn_id: None,
+        }
+    }
+
+    pub fn from_stored_state(state: &StoredLifecycleProjectionState) -> Self {
+        Self {
+            current_turn_id: state.current_turn_id.clone(),
+        }
+    }
+
+    pub fn observe_append(
+        &mut self,
+        persisted_rollout_items: &[RolloutItem],
+        projection_source_events: &[RolloutItem],
+    ) -> Vec<ThreadHistoryMutation> {
+        source_rollout_items(persisted_rollout_items, projection_source_events)
+            .iter()
+            .filter_map(|rollout_item| self.lifecycle_mutation(rollout_item))
+            .collect()
+    }
+
+    fn lifecycle_mutation(&mut self, rollout_item: &RolloutItem) -> Option<ThreadHistoryMutation> {
+        let RolloutItem::EventMsg(event) = rollout_item else {
+            return None;
+        };
+        let (event_type, turn_id, payload) = match event {
+            EventMsg::TurnStarted(payload) => {
+                self.current_turn_id = Some(payload.turn_id.clone());
+                (
+                    "turn.started",
+                    Some(payload.turn_id.clone()),
+                    json!({
+                        "startedAt": payload.started_at,
+                    }),
+                )
+            }
+            EventMsg::TurnComplete(payload) => {
+                self.clear_current_turn_if_matches(&payload.turn_id);
+                (
+                    "turn.completed",
+                    Some(payload.turn_id.clone()),
+                    json!({
+                        "completedAt": payload.completed_at,
+                        "durationMs": payload.duration_ms,
+                    }),
+                )
+            }
+            EventMsg::TurnAborted(payload) => {
+                let turn_id = payload
+                    .turn_id
+                    .clone()
+                    .or_else(|| self.current_turn_id.clone());
+                if let Some(turn_id) = &turn_id {
+                    self.clear_current_turn_if_matches(turn_id);
+                }
+                (
+                    "turn.cancelled",
+                    turn_id,
+                    json!({
+                        "reason": payload.reason,
+                        "completedAt": payload.completed_at,
+                        "durationMs": payload.duration_ms,
+                    }),
+                )
+            }
+            EventMsg::Error(payload) if payload.affects_turn_status() => (
+                "turn.failed",
+                Some(self.current_turn_id.clone()?),
+                json!({
+                    "message": payload.message,
+                    "codexErrorInfo": payload.codex_error_info,
+                }),
+            ),
+            EventMsg::ThreadRolledBack(payload) => (
+                "thread.rolled_back",
+                None,
+                json!({
+                    "numTurns": payload.num_turns,
+                }),
+            ),
+            _ => return None,
+        };
+        Some(ThreadHistoryMutation::Lifecycle(LifecycleMutation {
+            metadata: mutation_metadata(),
+            payload: json!({
+                "eventType": event_type,
+                "turnId": turn_id,
+                "payload": payload,
+            }),
+        }))
+    }
+
+    fn clear_current_turn_if_matches(&mut self, turn_id: &str) {
+        if self.current_turn_id.as_deref() == Some(turn_id) {
+            self.current_turn_id = None;
+        }
+    }
+}
+
+impl ThreadHistoryProjectionObserver for LifecycleProjectionObserver {
+    fn observe_append(
+        &mut self,
+        persisted_rollout_items: &[RolloutItem],
+        projection_source_events: &[RolloutItem],
+    ) -> Vec<ThreadHistoryMutation> {
+        LifecycleProjectionObserver::observe_append(
+            self,
+            persisted_rollout_items,
+            projection_source_events,
+        )
+    }
+}
+
+fn mutation_metadata() -> ThreadHistoryMutationMetadata {
+    ThreadHistoryMutationMetadata { schema_version: 1 }
+}
+
+fn source_rollout_items<'a>(
+    persisted_rollout_items: &'a [RolloutItem],
+    projection_source_events: &'a [RolloutItem],
+) -> &'a [RolloutItem] {
+    if projection_source_events.is_empty() {
+        persisted_rollout_items
+    } else {
+        projection_source_events
+    }
+}

codex-rs/app-server-protocol/src/protocol/lifecycle_projection_tests.rs

@@ -0,0 +1,152 @@
+use super::*;
+use codex_protocol::protocol::ErrorEvent;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::ThreadRolledBackEvent;
+use codex_protocol::protocol::TurnCompleteEvent;
+use codex_protocol::protocol::TurnStartedEvent;
+use codex_protocol::protocol::UserMessageEvent;
+use codex_thread_store_protocol::StoredLifecycleProjectionState;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+
+#[test]
+fn emits_lifecycle_mutations_from_lifecycle_events_only() {
+    let mut observer = LifecycleProjectionObserver::new();
+    let persisted_rollout_items = vec![
+        RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: "turn-a".into(),
+            trace_id: None,
+            started_at: Some(10),
+            model_context_window: None,
+            collaboration_mode_kind: Default::default(),
+        })),
+        RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
+            client_id: None,
+            message: "hello".into(),
+            images: None,
+            text_elements: Vec::new(),
+            local_images: Vec::new(),
+            ..Default::default()
+        })),
+        RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
+            turn_id: "turn-a".into(),
+            last_agent_message: None,
+            completed_at: Some(11),
+            duration_ms: Some(1_000),
+            time_to_first_token_ms: None,
+        })),
+        RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
+            num_turns: 1,
+        })),
+    ];
+    let mutations = observer.observe_append(&persisted_rollout_items, &[]);
+
+    assert_eq!(
+        lifecycle_payloads(mutations),
+        vec![
+            json!({
+                "eventType": "turn.started",
+                "turnId": "turn-a",
+                "payload": {
+                    "startedAt": 10,
+                },
+            }),
+            json!({
+                "eventType": "turn.completed",
+                "turnId": "turn-a",
+                "payload": {
+                    "completedAt": 11,
+                    "durationMs": 1_000,
+                },
+            }),
+            json!({
+                "eventType": "thread.rolled_back",
+                "turnId": null,
+                "payload": {
+                    "numTurns": 1,
+                },
+            }),
+        ]
+    );
+}
+
+#[test]
+fn emits_turn_failed_for_turn_affecting_errors() {
+    let mut observer = LifecycleProjectionObserver::new();
+    let persisted_rollout_items = vec![
+        RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: "turn-a".into(),
+            trace_id: None,
+            started_at: Some(10),
+            model_context_window: None,
+            collaboration_mode_kind: Default::default(),
+        })),
+        RolloutItem::EventMsg(EventMsg::Error(ErrorEvent {
+            message: "boom".into(),
+            codex_error_info: None,
+        })),
+    ];
+    let mutations = observer.observe_append(&persisted_rollout_items, &[]);
+
+    assert_eq!(
+        lifecycle_payloads(mutations),
+        vec![
+            json!({
+                "eventType": "turn.started",
+                "turnId": "turn-a",
+                "payload": {
+                    "startedAt": 10,
+                },
+            }),
+            json!({
+                "eventType": "turn.failed",
+                "turnId": "turn-a",
+                "payload": {
+                    "message": "boom",
+                    "codexErrorInfo": null,
+                },
+            }),
+        ]
+    );
+}
+
+#[test]
+fn restores_current_turn_for_later_turn_failure() {
+    let mut observer =
+        LifecycleProjectionObserver::from_stored_state(&StoredLifecycleProjectionState {
+            current_turn_id: Some("turn-a".into()),
+        });
+
+    let mutations = observer.observe_append(
+        &[],
+        &[RolloutItem::EventMsg(EventMsg::Error(ErrorEvent {
+            message: "boom".into(),
+            codex_error_info: None,
+        }))],
+    );
+
+    assert_eq!(
+        lifecycle_payloads(mutations),
+        vec![json!({
+            "eventType": "turn.failed",
+            "turnId": "turn-a",
+            "payload": {
+                "message": "boom",
+                "codexErrorInfo": null,
+            },
+        })]
+    );
+}
+
+fn lifecycle_payloads(mutations: Vec<ThreadHistoryMutation>) -> Vec<serde_json::Value> {
+    mutations
+        .into_iter()
+        .map(|mutation| match mutation {
+            ThreadHistoryMutation::Lifecycle(mutation) => mutation.payload,
+            ThreadHistoryMutation::ThreadItem(_) | ThreadHistoryMutation::TurnSummary(_) => {
+                panic!("lifecycle observer emitted non-lifecycle mutation")
+            }
+        })
+        .collect()
+}

codex-rs/app-server-protocol/src/protocol/mod.rs

@@ -4,8 +4,12 @@
 pub mod common;
 pub mod event_mapping;
 pub mod item_builders;
+mod lifecycle_projection;
 mod mappers;
 mod serde_helpers;
 pub mod thread_history;
+mod thread_history_builder;
+mod thread_item_projection;
+mod turn_summary_projection;
 pub mod v1;
 pub mod v2;