fix(api): make WatchSandbox loss-aware and resumable

architecture/gateway.md

@@ -284,6 +284,37 @@ Domain objects use shared metadata: stable server-generated IDs, human-readable
 names, creation timestamps, and labels. Crate-level details live in
 `crates/openshell-core/README.md`.
 
+### Watch streams
+
+`WatchSandbox` merges three per-sandbox sources into one client stream: status
+snapshots, server/sandbox logs, and platform events. Logs and platform events
+are resumable; a shared per-sandbox counter stamps each with a monotonic
+`cursor`. Cursor-ordered delivery is guaranteed for the replay phase: on
+resume the buffered events from both sources are sorted by cursor before
+emission. Live events carry cursors and are monotonic within each source, but
+the two sources are read independently, so a client should order across sources
+by `cursor` rather than by arrival. Status snapshots and warnings are re-read on
+demand and carry `cursor = 0`.
+
+The gateway holds a bounded in-memory tail per sandbox. Loss is reported with
+two distinct, documented behaviors:
+
+- **Recoverable lag** — a broadcast receiver falls behind and the server skips
+  ahead. The stream emits a `SandboxStreamWarning` event and continues; the
+  client sees the gap as a cursor discontinuity.
+- **Unrecoverable gap** — a reconnect requests `resume_after_cursor` below the
+  oldest buffered cursor (the tail has been trimmed past it). The server sends a
+  snapshot, then terminates the stream with `OUT_OF_RANGE` carrying the
+  requested and earliest-available cursors so the client can restart cleanly.
+
+On resume the server replays only events after the client's cursor from both
+resumable sources, merged in cursor order, before entering live delivery. The
+broadcast receivers are subscribed before replay, so an event buffered during
+initialization could appear in both replay and the live receiver; the producer
+tracks the highest replayed cursor and suppresses live events at or below it, so
+each event is delivered once. Clients track the highest observed `cursor` and
+pass it as `resume_after_cursor` on reconnect.
+
 ## Persistence
 
 The gateway persistence layer is a protobuf object store. Domain services store

crates/openshell-cli/src/run.rs

@@ -708,6 +708,7 @@ pub async fn sandbox_create(
             log_since_ms: 0,
             log_sources: vec!["gateway".to_string()],
             log_min_level: String::new(),
+            resume_after_cursor: 0,
         })
         .await
         .into_diagnostic()?
@@ -2607,6 +2608,7 @@ async fn wait_for_lifecycle_phase(
             log_since_ms: 0,
             log_sources: Vec::new(),
             log_min_level: String::new(),
+            resume_after_cursor: 0,
         })
         .await
         .into_diagnostic()?
@@ -7072,6 +7074,7 @@ pub async fn sandbox_logs(
                 log_since_ms: since_ms,
                 log_sources: source_filter,
                 log_min_level: level.to_uppercase(),
+                resume_after_cursor: 0,
             })
             .await
             .into_diagnostic()?

crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs

@@ -489,6 +489,7 @@ impl OpenShell for TestOpenShell {
             let _ = tx
                 .send(Ok(SandboxStreamEvent {
                     payload: Some(sandbox_stream_event::Payload::Sandbox(provisioning)),
+                    cursor: 0,
                 }))
                 .await;
             if vm_error_after_started {
@@ -500,11 +501,13 @@ impl OpenShell for TestOpenShell {
                             message: "Started VM launcher".to_string(),
                             ..PlatformEvent::default()
                         })),
+                        cursor: 0,
                     }))
                     .await;
                 let _ = tx
                     .send(Ok(SandboxStreamEvent {
                         payload: Some(sandbox_stream_event::Payload::Sandbox(error)),
+                        cursor: 0,
                     }))
                     .await;
                 tokio::time::sleep(Duration::from_secs(5)).await;
@@ -524,12 +527,14 @@ impl OpenShell for TestOpenShell {
                                 source: "gateway".to_string(),
                                 fields: HashMap::new(),
                             })),
+                            cursor: 0,
                         }))
                         .await;
                 }
                 let _ = tx
                     .send(Ok(SandboxStreamEvent {
                         payload: Some(sandbox_stream_event::Payload::Sandbox(ready)),
+                        cursor: 0,
                     }))
                     .await;
                 return;
@@ -538,6 +543,7 @@ impl OpenShell for TestOpenShell {
                 let _ = tx
                     .send(Ok(SandboxStreamEvent {
                         payload: Some(sandbox_stream_event::Payload::Sandbox(completed)),
+                        cursor: 0,
                     }))
                     .await;
                 return;
@@ -552,6 +558,7 @@ impl OpenShell for TestOpenShell {
                             message: "Preparing rootfs".to_string(),
                             ..PlatformEvent::default()
                         })),
+                        cursor: 0,
                     }))
                     .await;
                 tokio::time::sleep(Duration::from_millis(600)).await;
@@ -563,12 +570,14 @@ impl OpenShell for TestOpenShell {
                             message: "Formatting root disk".to_string(),
                             ..PlatformEvent::default()
                         })),
+                        cursor: 0,
                     }))
                     .await;
                 tokio::time::sleep(Duration::from_millis(600)).await;
                 let _ = tx
                     .send(Ok(SandboxStreamEvent {
                         payload: Some(sandbox_stream_event::Payload::Sandbox(ready)),
+                        cursor: 0,
                     }))
                     .await;
                 return;
@@ -580,11 +589,13 @@ impl OpenShell for TestOpenShell {
                         message: "Sandbox scheduled".to_string(),
                         ..PlatformEvent::default()
                     })),
+                    cursor: 0,
                 }))
                 .await;
             let _ = tx
                 .send(Ok(SandboxStreamEvent {
                     payload: Some(sandbox_stream_event::Payload::Sandbox(ready)),
+                    cursor: 0,
                 }))
                 .await;
         });

crates/openshell-server/src/compute/mod.rs

@@ -2890,6 +2890,8 @@ impl ComputeRuntime {
                                     public_platform_event_from_driver(&event),
                                 ),
                             ),
+                            // Placeholder: platform_event_bus.publish() stamps the cursor.
+                            cursor: 0,
                         },
                     );
                 }
@@ -3386,8 +3388,9 @@ impl ComputeRuntime {
     }
 
     fn cleanup_sandbox_state(&self, sandbox_id: &str) {
+        // `tracing_log_bus.remove` also clears the platform event bus and resets
+        // the shared cursor allocator last (see its docs).
         self.tracing_log_bus.remove(sandbox_id);
-        self.tracing_log_bus.platform_event_bus.remove(sandbox_id);
         self.sandbox_watch_bus.remove(sandbox_id);
     }