fix(session): export reviewed Stop notification fence

PUBLIC_EXPORT_MANIFEST.json

@@ -2593,10 +2593,10 @@
       "sha256": "d235803e4919c54d3b24baeacaed5899e10eb26b93956d0cebeaa57fdd9bde41"
     },
     {
-      "bytes": 45052,
+      "bytes": 45592,
       "mode": "100644",
       "path": "apps/synth_desktop/src-tauri/src/session/codex/event_pump.rs",
-      "sha256": "dbf4a0502422869d27794af7b2d1f9d31f5a3ae2b6285d043e4c88173fe152fb"
+      "sha256": "4ff1fdfe2c70a4cf8e9e3162dc9d92fd88d9c5c70a888cd2921e5b8a40a265e0"
     },
     {
       "bytes": 47523,
@@ -2617,10 +2617,10 @@
       "sha256": "1c4dde3dd33c466a18e0e491b6b5cb7868a4db9e5b39be217bc6e279baa8b370"
     },
     {
-      "bytes": 78960,
+      "bytes": 80126,
       "mode": "100644",
       "path": "apps/synth_desktop/src-tauri/src/session/codex/manager.rs",
-      "sha256": "f0e60028c487e972ff3d83a3417bbf142e3e0098daa245c367cc1936a97614b4"
+      "sha256": "cf315b1835377ea7f74b9e54383398d51746acdb6f6e3bd94b9d108a0a817763"
     },
     {
       "bytes": 874,
@@ -2629,10 +2629,10 @@
       "sha256": "8302dca4d00ba0b9a03bc87931228156d3861584ff69647cfa4222000d37fb61"
     },
     {
-      "bytes": 25197,
+      "bytes": 25336,
       "mode": "100644",
       "path": "apps/synth_desktop/src-tauri/src/session/codex/proto.rs",
-      "sha256": "8a531b4ad2ab7bb18ef2a14da6e4a3a2dbb3d6c64a9c8c8481190aa99772747a"
+      "sha256": "1f50e1bd6783e4c8280c0e2a3983e631e5308eda2f6b7957733c331b0709626e"
     },
     {
       "bytes": 18116,
@@ -7690,7 +7690,7 @@
   "policySha256": "f5ef8fbb4560ee9e401c26c94da6f6189732be76291915027bd7ee3f5f5684dd",
   "schema": "workshop.public-export-manifest.v1",
   "source": {
-    "commit": "005a84d4ff3ac2b12e288e735e911088b639127a",
-    "tree": "837e0769d96681a82cc9b0c5b181cdaf1796146d"
+    "commit": "543aec726c8d54565d82faaf9f35399e89e879d0",
+    "tree": "a7f439abe88e252972ccc0edd5d6784d3796c606"
   }
 }

apps/synth_desktop/src-tauri/src/session/codex/event_pump.rs

@@ -111,6 +111,7 @@ pub(crate) struct SpawnServerRequest<'a> {
 /// Shared pump state cloned into the stdout reader task.
 #[derive(Clone)]
 pub(crate) struct EventPumpState {
+    pub notification_closed: Arc<Mutex<bool>>,
     pub records: Arc<RwLock<HashMap<String, CodexSessionRecord>>>,
     pub state_path: PathBuf,
     pub persistence: SessionPersistence,
@@ -415,6 +416,10 @@ async fn read_stdout<R: tauri::Runtime, T: AsyncRead + Unpin>(
             }
             continue;
         }
+        let notification_guard = persistence.notification_closed.lock().await;
+        if *notification_guard {
+            continue;
+        }
         let raw_method = message["method"].as_str().unwrap_or_default();
         let mut params = message.get("params").cloned().unwrap_or(Value::Null);
         crate::codex_oauth::redact_event_value(&mut params);
@@ -642,7 +647,8 @@ async fn read_stdout<R: tauri::Runtime, T: AsyncRead + Unpin>(
             .await;
         }
     }
-    if settlement.ready_for_eof_completion() {
+    let notification_guard = persistence.notification_closed.lock().await;
+    if !*notification_guard && settlement.ready_for_eof_completion() {
         // The child closed stdout after tools settled and an assistant item
         // completed, without `turn/completed` or `phase: final_answer`. That
         // is process-exit evidence, not a mid-turn commentary gap.
@@ -656,6 +662,7 @@ async fn read_stdout<R: tauri::Runtime, T: AsyncRead + Unpin>(
             .await;
         apply_codex_terminal(&app, &session_id, &persistence, "turn/completed", params).await;
     }
+    drop(notification_guard);
     let owned_attachment = {
         let mut sessions = persistence.sessions.write().await;
         let owns_current = sessions
@@ -1024,6 +1031,11 @@ pub(crate) fn normalized_turn_method<'a>(method: &'a str, params: &Value) -> &'a
         return method;
     }
     let turn = params.get("turn").unwrap_or(params);
+    if turn.get("status").and_then(Value::as_str).is_some_and(|status| {
+        matches!(status.to_ascii_lowercase().as_str(), "interrupted" | "cancelled" | "canceled")
+    }) {
+        return "turn/interrupted";
+    }
     let status_is_failure = turn
         .get("status")
         .and_then(Value::as_str)

apps/synth_desktop/src-tauri/src/session/codex/manager.rs

@@ -271,6 +271,7 @@ impl CodexManager {
         install_local_laguna_catalog(&home, &request)?;
         ensure_home(&home, &request)?;
         let attachment_id = uuid::Uuid::new_v4();
+        let notification_closed = Arc::new(Mutex::new(false));
         let server = spawn_server(
             app.clone(),
             SpawnServerRequest {
@@ -283,6 +284,7 @@ impl CodexManager {
                         == super::home::ProviderClass::OpenaiCodexOauth,
             },
             EventPumpState {
+                notification_closed: notification_closed.clone(),
                 records: self.records.clone(),
                 state_path: self.state_path.clone(),
                 persistence: self.persistence.clone(),
@@ -371,6 +373,7 @@ impl CodexManager {
             .ok_or_else(|| anyhow!("Codex {method} response missing thread id: {result}"))?;
         let mcp_reload_pending = server.persistent;
         let session = Arc::new(Session {
+            notification_closed,
             attachment_id,
             server,
             thread_id: thread_id.clone(),
@@ -1192,6 +1195,9 @@ impl CodexManager {
         let Some(turn_id) = session.turn_id.read().await.clone() else {
             return Ok(());
         };
+        // Drain any in-flight projection, then reject late notifications while
+        // still allowing RPC acknowledgements through the stdout reader.
+        *session.notification_closed.lock().await = true;
         // Terminalize durable state before asking the provider. Its own
         // turn/interrupted notification can race the request acknowledgement;
         // recording cancellation first keeps a deliberate Stop distinct from
@@ -1222,6 +1228,27 @@ impl CodexManager {
                 ),
             )
             .await?;
+        // Finalize the interrupted turn exactly once, retaining usage already
+        // observed before Stop. Late notifications cannot contaminate it.
+        let measurements = super::telemetry::finalize_performance_tracker(
+            &self.persistence,
+            &self.performance_trackers,
+            &self.receipts(),
+            session_id,
+            RunStatus::Interrupted.as_str(),
+            None,
+        )
+        .await;
+        for measurement in measurements {
+            self.persistence
+                .notify_codex_event(
+                    &app,
+                    session_id.to_owned(),
+                    super::generation_speed::MEASUREMENT_EVENT,
+                    serde_json::to_value(measurement)?,
+                )
+                .await;
+        }
         // Give the provider a bounded opportunity to stop leases and seal
         // partial evidence. An acknowledgement is not proof that a child tool
         // died, so the owned process group is fenced below in every case.

apps/synth_desktop/src-tauri/src/session/codex/proto.rs

@@ -611,6 +611,8 @@ pub(crate) fn default_sandbox() -> String {
 }
 
 pub(crate) struct Session {
+    /// Serializes cancellation against notification projection for this attachment.
+    pub(crate) notification_closed: Arc<Mutex<bool>>,
     pub(crate) attachment_id: uuid::Uuid,
     pub(crate) server: Arc<AppServer>,
     pub(crate) thread_id: String,