From 9bfff85f86d04aef0f6b27d9b41be5785b7cf3ef Mon Sep 17 00:00:00 2001 From: Josh Purtell Date: Thu, 10 Sep 2026 13:30:07 -0400 Subject: [PATCH] fix(session): export reviewed Stop notification fence Deterministic public export of source 543aec726c8d54565d82faaf9f35399e89e879d0: three production files plus manifest only. All 1281 manifest hashes verified, public audit passed, offline Rust check passed. Existing release tags and assets remain unchanged. --- PUBLIC_EXPORT_MANIFEST.json | 16 +++++------ .../src-tauri/src/session/codex/event_pump.rs | 14 +++++++++- .../src-tauri/src/session/codex/manager.rs | 27 +++++++++++++++++++ .../src-tauri/src/session/codex/proto.rs | 2 ++ 4 files changed, 50 insertions(+), 9 deletions(-) diff --git a/PUBLIC_EXPORT_MANIFEST.json b/PUBLIC_EXPORT_MANIFEST.json index bf63980fe..476e63eeb 100644 --- a/PUBLIC_EXPORT_MANIFEST.json +++ b/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" } } diff --git a/apps/synth_desktop/src-tauri/src/session/codex/event_pump.rs b/apps/synth_desktop/src-tauri/src/session/codex/event_pump.rs index e771c82dd..76ee8ba44 100644 --- a/apps/synth_desktop/src-tauri/src/session/codex/event_pump.rs +++ b/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>, pub records: Arc>>, pub state_path: PathBuf, pub persistence: SessionPersistence, @@ -415,6 +416,10 @@ async fn read_stdout( } 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( .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( .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) diff --git a/apps/synth_desktop/src-tauri/src/session/codex/manager.rs b/apps/synth_desktop/src-tauri/src/session/codex/manager.rs index 5bb285177..31f7a878b 100644 --- a/apps/synth_desktop/src-tauri/src/session/codex/manager.rs +++ b/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. diff --git a/apps/synth_desktop/src-tauri/src/session/codex/proto.rs b/apps/synth_desktop/src-tauri/src/session/codex/proto.rs index a7458a065..d81c12865 100644 --- a/apps/synth_desktop/src-tauri/src/session/codex/proto.rs +++ b/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>, pub(crate) attachment_id: uuid::Uuid, pub(crate) server: Arc, pub(crate) thread_id: String,