feat(pvisor): add Codex SandboxReplay resume bridge

crates/persisting-overlaynet/src/interception.rs

@@ -198,11 +198,19 @@ impl InterceptionMetrics {
     }
 
     pub(crate) fn tcp_flow_closed(&self) {
-        let _ = self.counters.active_tcp_flows.try_update(
-            Ordering::Relaxed,
-            Ordering::Relaxed,
-            |active| active.checked_sub(1),
-        );
+        let active_tcp_flows = &self.counters.active_tcp_flows;
+        let mut current = active_tcp_flows.load(Ordering::Relaxed);
+        while current > 0 {
+            match active_tcp_flows.compare_exchange_weak(
+                current,
+                current - 1,
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+            ) {
+                Ok(_) => break,
+                Err(next) => current = next,
+            }
+        }
     }
 
     pub(crate) fn tcp_flow_denied(&self) {

crates/persisting-pvisor/src/cli/replay.rs

@@ -69,6 +69,8 @@ pub struct ReplayArgs {
     #[arg(long, value_name = "DIR")]
     output_dir: Option<PathBuf>,
 
+    /// Model-router/run session key. Codex native continuation identity is
+    /// derived from the trajectory and is never taken from this field.
     #[arg(long)]
     session_id: Option<String>,
 

crates/persisting-replay/assets/pi_agent_runner.mjs

@@ -4,6 +4,11 @@ import fs from "node:fs";
 import path from "node:path";
 import { pathToFileURL } from "node:url";
 
+// Pi runtimes may be launched with Node 16/18 in older sandboxes where the
+// global structuredClone helper is unavailable. JSON is sufficient for the
+// native event objects we copy here and keeps replay portable across runtimes.
+const clone = globalThis.structuredClone ?? ((value) => JSON.parse(JSON.stringify(value)));
+
 function load(filename) {
   return JSON.parse(fs.readFileSync(filename, "utf8"));
 }
@@ -181,15 +186,15 @@ async function run(request) {
 
   for (const event of events) {
     if (event?.type === "message_end" && event?.message?.role === "user") {
-      sessionManager.appendMessage(structuredClone(event.message));
-      reconstructedEvents.push(structuredClone(event));
+      sessionManager.appendMessage(clone(event.message));
+      reconstructedEvents.push(clone(event));
       continue;
     }
     if (event?.type !== "turn_end" || event?.message?.role !== "assistant") continue;
     const calls = toolCalls(event.message);
     if (calls.length > 0 && replayedBatches >= request.after_step) break;
     prefixTurns += 1;
-    const assistant = structuredClone(event.message);
+    const assistant = clone(event.message);
     sessionManager.appendMessage(assistant);
     const freshResults = [];
     for (const call of calls) {
@@ -198,7 +203,7 @@ async function run(request) {
       observations.push(fresh.observation);
       sessionManager.appendMessage(fresh.message);
     }
-    reconstructedEvents.push({ ...structuredClone(event), toolResults: freshResults });
+    reconstructedEvents.push({ ...clone(event), toolResults: freshResults });
     if (calls.length > 0) {
       replayedBatches += 1;
       if (replayedBatches === request.after_step) break;
@@ -259,7 +264,7 @@ async function run(request) {
   let terminalError = null;
   const remaining = request.max_steps == null ? null : request.max_steps - prefixTurns;
   const unsubscribe = session.subscribe((event) => {
-    liveEvents.push(structuredClone(event));
+    liveEvents.push(clone(event));
     if (event.type === "turn_end") {
       continuedSteps += 1;
       if (event.message?.stopReason === "error") {