[app-server] feat: add turn/diff/updated event (#7279)

This is the V2 version of `EventMsg::TurnDiff`.

I decided to expose this as a `turn/*` notification as opposed to an
Item to make it more explicit that the diff is accumulated throughout a
turn (every `apply_patch` call updates the running diff). Also, I don't
think it's worth persisting this diff as an Item because it can always
be recomputed from the actual `FileChange` Items.

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

@@ -504,6 +504,7 @@ server_notification_definitions! {
     ThreadStarted => "thread/started" (v2::ThreadStartedNotification),
     TurnStarted => "turn/started" (v2::TurnStartedNotification),
     TurnCompleted => "turn/completed" (v2::TurnCompletedNotification),
+    TurnDiffUpdated => "turn/diff/updated" (v2::TurnDiffUpdatedNotification),
     ItemStarted => "item/started" (v2::ItemStartedNotification),
     ItemCompleted => "item/completed" (v2::ItemCompletedNotification),
     AgentMessageDelta => "item/agentMessage/delta" (v2::AgentMessageDeltaNotification),

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

@@ -1141,6 +1141,16 @@ pub struct TurnCompletedNotification {
     pub turn: Turn,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+/// Notification that the turn-level unified diff has changed.
+/// Contains the latest aggregated diff across all file changes in the turn.
+pub struct TurnDiffUpdatedNotification {
+    pub turn_id: String,
+    pub diff: String,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -38,6 +38,7 @@ use codex_app_server_protocol::ServerRequestPayload;
 use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::Turn;
 use codex_app_server_protocol::TurnCompletedNotification;
+use codex_app_server_protocol::TurnDiffUpdatedNotification;
 use codex_app_server_protocol::TurnError;
 use codex_app_server_protocol::TurnInterruptResponse;
 use codex_app_server_protocol::TurnStatus;
@@ -53,6 +54,7 @@ use codex_core::protocol::McpToolCallBeginEvent;
 use codex_core::protocol::McpToolCallEndEvent;
 use codex_core::protocol::Op;
 use codex_core::protocol::ReviewDecision;
+use codex_core::protocol::TurnDiffEvent;
 use codex_core::review_format::format_review_findings_block;
 use codex_protocol::ConversationId;
 use codex_protocol::protocol::ReviewOutputEvent;
@@ -549,11 +551,31 @@ pub(crate) async fn apply_bespoke_event_handling(
             handle_turn_interrupted(conversation_id, event_id, &outgoing, &turn_summary_store)
                 .await;
         }
+        EventMsg::TurnDiff(turn_diff_event) => {
+            handle_turn_diff(&event_id, turn_diff_event, api_version, outgoing.as_ref()).await;
+        }
 
         _ => {}
     }
 }
 
+async fn handle_turn_diff(
+    event_id: &str,
+    turn_diff_event: TurnDiffEvent,
+    api_version: ApiVersion,
+    outgoing: &OutgoingMessageSender,
+) {
+    if let ApiVersion::V2 = api_version {
+        let notification = TurnDiffUpdatedNotification {
+            turn_id: event_id.to_string(),
+            diff: turn_diff_event.unified_diff,
+        };
+        outgoing
+            .send_server_notification(ServerNotification::TurnDiffUpdated(notification))
+            .await;
+    }
+}
+
 async fn emit_turn_completed_with_status(
     conversation_id: ConversationId,
     event_id: String,
@@ -1481,4 +1503,56 @@ mod tests {
 
         assert_eq!(notification, expected);
     }
+
+    #[tokio::test]
+    async fn test_handle_turn_diff_emits_v2_notification() -> Result<()> {
+        let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let outgoing = OutgoingMessageSender::new(tx);
+        let unified_diff = "--- a\n+++ b\n".to_string();
+
+        handle_turn_diff(
+            "turn-1",
+            TurnDiffEvent {
+                unified_diff: unified_diff.clone(),
+            },
+            ApiVersion::V2,
+            &outgoing,
+        )
+        .await;
+
+        let msg = rx
+            .recv()
+            .await
+            .ok_or_else(|| anyhow!("should send one notification"))?;
+        match msg {
+            OutgoingMessage::AppServerNotification(ServerNotification::TurnDiffUpdated(
+                notification,
+            )) => {
+                assert_eq!(notification.turn_id, "turn-1");
+                assert_eq!(notification.diff, unified_diff);
+            }
+            other => bail!("unexpected message: {other:?}"),
+        }
+        assert!(rx.try_recv().is_err(), "no extra messages expected");
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_handle_turn_diff_is_noop_for_v1() -> Result<()> {
+        let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY);
+        let outgoing = OutgoingMessageSender::new(tx);
+
+        handle_turn_diff(
+            "turn-1",
+            TurnDiffEvent {
+                unified_diff: "diff".to_string(),
+            },
+            ApiVersion::V1,
+            &outgoing,
+        )
+        .await;
+
+        assert!(rx.try_recv().is_err(), "no messages expected");
+        Ok(())
+    }
 }