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

@@ -745,6 +745,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()?
@@ -3085,6 +3086,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()?
@@ -5525,6 +5527,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

@@ -607,6 +607,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 {
@@ -618,11 +619,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;
@@ -642,12 +645,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;
@@ -656,6 +661,7 @@ impl OpenShell for TestOpenShell {
                 let _ = tx
                     .send(Ok(SandboxStreamEvent {
                         payload: Some(sandbox_stream_event::Payload::Sandbox(completed)),
+                        cursor: 0,
                     }))
                     .await;
                 return;
@@ -670,6 +676,7 @@ impl OpenShell for TestOpenShell {
                             message: "Preparing rootfs".to_string(),
                             ..PlatformEvent::default()
                         })),
+                        cursor: 0,
                     }))
                     .await;
                 tokio::time::sleep(Duration::from_millis(600)).await;
@@ -681,12 +688,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;
@@ -698,11 +707,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-sdk/Cargo.toml

@@ -28,6 +28,7 @@ tokio-tungstenite = { workspace = true }
 tonic = { workspace = true, features = ["tls-native-roots"] }
 tower = { workspace = true }
 tracing = { workspace = true }
+async-stream = "0.3.6"
 
 [dev-dependencies]
 serde_json = { workspace = true }