data plane: resume process SSE streams from the guest output journal

crates/apps/spaced/host/cli/src/main.rs

@@ -288,7 +288,7 @@ async fn run() -> anyhow::Result<()> {
         }
         Commands::Connect { process_id } => {
             let mut stream = client
-                .connect_process_stream_with_cursor(process_id, None, None)
+                .connect_process_stream_with_cursor(process_id, None, None, false)
                 .await?;
             while let Some(event) = stream.message().await? {
                 print_process_event(event).await?;

crates/libs/api/src/routes/public_docs.rs

@@ -71,7 +71,7 @@ fn public_openapi_value() -> Value {
     let data_plane_events_item = json!({
         "get": {
             "summary": "Stream process output or file-watch events over SSE",
-            "description": "Streams output from process or file-watch sessions.\n\n### Connect\n\n- Use the session token, not your management API token. Send it in `Authorization: Bearer ...` or `Sec-WebSocket-Protocol`; query-string tokens are rejected.\n- Every frame after `attached` has an increasing SSE `id`.\n\n### Reconnect\n\n- Send the last received `id` as `Last-Event-ID`. The stream resumes live; missed output is not replayed.\n- The first identified frame is `resync_required`, with reason `cursor_replay_unavailable` and id `Last-Event-ID + 1`.\n- Keep output already received. If the process exited while disconnected, reconnecting still reports `process_exited`.\n- For file watches, re-list the watched path after `resync_required` before using new events.",
+            "description": "Streams output from process or file-watch sessions.\n\n### Connect\n\n- Use the session token, not your management API token. Send it in `Authorization: Bearer ...` or `Sec-WebSocket-Protocol`; query-string tokens are rejected.\n- SSE `id`s increase. Every file-watch frame after `attached` has one. On a process stream only `stdout`, `stderr` and `resync_required` frames do; other frames carry none, so the last `id` received stays the one to resume from.\n- Echo the `id` exactly as received: process stream ids are not consecutive. They stay below 2^53, so a JavaScript `Number` holds them exactly.\n\n### Reconnect\n\n- Send the last received `id` as `Last-Event-ID`. An `id` is only valid for the session whose stream sent it.\n- A process stream resumes with the output after that frame while the sandbox retains it, a bounded amount per process. Without `Last-Event-ID` it starts from the beginning of the output, which can repeat output already received.\n- When the output to resume from is no longer retained, the stream sends `resync_required` with reason `cursor_out_of_window`, then continues with the oldest output still retained (with live output on sandboxes from older templates). When `Last-Event-ID` is past the process's output, it sends the same frame and replays the output from its start. Keep output already received.\n- An `id` that does not resume exactly (one from an earlier SPACE version, one altered in transit, or one sent after the stream replayed the output from its start behind `cursor_out_of_window` or continued with live output) replays the retained output behind `resync_required` with reason `cursor_replay_unavailable`.\n- Every `id` a resumed process stream sends is above `Last-Event-ID`.\n- Process stream ids carry the output position for a process's first 2^33 - 1 output chunks (about 8.6 billion; some 99 days at 1,000 separate writes a second). After that, each resume replays the output still retained behind `resync_required`.\n- If the process exited while disconnected, reconnecting still reports `process_exited`.\n- A file-watch stream resumes live: its first identified frame is `resync_required`, with reason `cursor_replay_unavailable` and id `Last-Event-ID + 1`. Re-list the watched path before using new events.",
             "responses": {
                 "200": { "description": "Server-sent event stream using DataPlaneServerFrame JSON payloads" },
                 "401": { "description": "Invalid or expired data-plane token" },

crates/libs/api/src/routes/public_docs/schema_contract.rs

@@ -88,7 +88,7 @@ fn openapi_identifier(value: &str) -> String {
 
 pub(super) fn insert_command_event_stream_schema(spec: &mut Value) {
     let response = json!({
-        "description": "### Read the stream\n\n- Each SSE event carries one `DataPlaneServerFrame` JSON object. Ignore comment heartbeats.\n- Decode stdout and stderr's `data_base64` using standard base64.\n- Success requires `process_exited` with exit code zero and no error or missing output. HTTP 200 or a closed stream alone is not success.\n\n### Example output\n\nExcerpt, not a complete stream:\n\n```text\nid: 2\nevent: stdout\ndata: {\"type\":\"stdout\",\"seq\":0,\"data_base64\":\"SGVsbG8gZnJvbSBTUEFDRQo=\"}\n\nid: 3\nevent: process_exited\ndata: {\"type\":\"process_exited\",\"process_id\":\"proc_00000000-0000-0000-0000-000000000000\",\"exit_code\":0,\"status\":\"exited\"}\n\n```",
+        "description": "### Read the stream\n\n- Each SSE event carries one `DataPlaneServerFrame` JSON object. Ignore comment heartbeats.\n- Decode stdout and stderr's `data_base64` using standard base64.\n- Success requires `process_exited` with exit code zero and no error or missing output. HTTP 200 or a closed stream alone is not success.\n\n### Example output\n\nExcerpt, not a complete stream:\n\n```text\nid: 4503599627632640\nevent: stdout\ndata: {\"type\":\"stdout\",\"seq\":1,\"data_base64\":\"SGVsbG8gZnJvbSBTUEFDRQo=\"}\n\nevent: process_exited\ndata: {\"type\":\"process_exited\",\"process_id\":\"proc_00000000-0000-0000-0000-000000000000\",\"exit_code\":0,\"status\":\"exited\"}\n\n```",
         "content": {
             "text/event-stream": {
                 "schema": { "type": "string" },
@@ -117,7 +117,7 @@ pub(super) fn insert_command_event_stream_schema(spec: &mut Value) {
             "name": "Last-Event-ID",
             "in": "header",
             "required": false,
-            "description": "SSE `id` of the last frame received.\n\n- Resumes the live stream; missed output is not replayed.\n- The first identified frame is `resync_required` (`cursor_replay_unavailable`), with id `Last-Event-ID + 1`.\n- Values above the `int64` maximum return 400.",
+            "description": "SSE `id` of the last frame received.\n\n- Echo the `id` exactly as received, on the same session's stream.\n- A process stream resumes with the output after that frame while it is retained; otherwise `resync_required` precedes the output it continues with: `cursor_out_of_window` when the output is evicted or the `id` is past it, `cursor_replay_unavailable` when the `id` does not resume exactly.\n- A file-watch stream resumes live; its first identified frame is `resync_required` (`cursor_replay_unavailable`), with id `Last-Event-ID + 1`.\n- Values above the `int64` maximum, and process stream values too high to number a later frame above, return 400.",
             "schema": {
                 "type": "integer",
                 "format": "int64",

crates/libs/common/core/src/lib.rs

@@ -160,6 +160,29 @@ pub mod spaced {
     /// as opposed to the sandbox or its runtime being unknown to a hop.
     pub const GUEST_PATH_NOT_FOUND: &str = "guest-path-not-found";
 
+    /// [`ERROR_REASON_METADATA_KEY`] value of an OUT_OF_RANGE that ends a
+    /// process output stream whose cursor is past the output the process has
+    /// written, as opposed to a message too large for a hop to decode.
+    pub const PROCESS_OUTPUT_CURSOR_AHEAD: &str = "process-output-cursor-ahead";
+
+    /// [`ERROR_REASON_METADATA_KEY`] value of a RESOURCE_EXHAUSTED that ends a
+    /// process output stream whose cursor is before the oldest output the
+    /// guest still retains, as opposed to a hop's refusal or a transport reset
+    /// with the same code. [`PROCESS_OUTPUT_FIRST_SEQUENCE_METADATA_KEY`] says
+    /// where the retained output starts.
+    pub const PROCESS_OUTPUT_LAGGED: &str = "process-output-lagged";
+
+    /// Metadata key, on a [`PROCESS_OUTPUT_LAGGED`] status, of the cursor of
+    /// the oldest output the guest still retains, in decimal. Absent when the
+    /// guest no longer holds the process's output at all.
+    pub const PROCESS_OUTPUT_FIRST_SEQUENCE_METADATA_KEY: &str =
+        "space-process-output-first-sequence";
+
+    /// Message of the guest's [`PROCESS_OUTPUT_LAGGED`] status. Guests older
+    /// than that reason send it untagged, so it identifies their lag status.
+    pub const PROCESS_OUTPUT_LAGGED_MESSAGE: &str =
+        "process output stream fell behind retained output buffer";
+
     /// [`ERROR_REASON_METADATA_KEY`] value of a request refused before any of
     /// its work started: an admission limit, or a guest or registry shutting
     /// down. This request did nothing, so the caller may retry it, after the

crates/libs/data/plane/server/Cargo.toml

@@ -64,6 +64,7 @@ pplx-space-public-api = { workspace = true, features = ["serde"] }
 pplx-space-sbx-mgr-client = { workspace = true }
 
 [dev-dependencies]
+pplx-space-metrics = { workspace = true, features = ["test-capture"] }
 tokio = { workspace = true, features = ["test-util"] }
 pplx-space-test-env = { workspace = true }
 aws-lc-rs = { workspace = true }

crates/libs/data/plane/server/src/ecw_connect/tests.rs

@@ -43,7 +43,9 @@ fn process_events_encode_connect_wire_shape() {
             process_id: "proc-1".to_string(),
             pid: None,
             event: Some(spaced_process_event::Event::Connected(
-                SpacedProcessConnected {},
+                SpacedProcessConnected {
+                    output_cursor: None,
+                },
             )),
         },
         true,

crates/libs/data/plane/server/src/ecw_process.rs

@@ -1,4 +1,4 @@
 use std::{
     collections::BTreeMap,
     convert::Infallible,
     fmt::Display,
@@ -406,6 +406,7 @@
             process_id,
             timeout,
             output_cursor: None,
+            resume_from_oldest_retained: false,
         }),
     };
     let route_state = state.clone();

crates/libs/data/plane/server/src/ecw_process/stream.rs

@@ -449,6 +449,7 @@ pub(super) fn build_process_start_resume_request(
                 .as_ref()
                 .and_then(|operation| operation.timeout),
             output_cursor: Some(output_cursor),
+            resume_from_oldest_retained: false,
         }),
     }
 }

crates/libs/data/plane/server/src/ecw_process/tests.rs

@@ -73,7 +73,9 @@ fn connected_process_event(process_id: &ProcessId) -> SpacedProcessEvent {
         process_id: process_id.to_string(),
         pid: Some(123),
         event: Some(spaced_ops::process_event::Event::Connected(
-            spaced_ops::ProcessConnected {},
+            spaced_ops::ProcessConnected {
+                output_cursor: None,
+            },
         )),
     }
 }

crates/libs/data/plane/server/src/native/auth.rs

@@ -96,9 +96,10 @@ pub(crate) fn reject_query_tokens(query: Option<&str>) -> Result<(), AttachError
 }
 
 /// Largest `Last-Event-ID` a client may resume from. It is the OpenAPI
-/// parameter's `int64` maximum, and it leaves room for the `last + 1` id the
-/// resume contract promises, so the stream can never saturate onto an id the
-/// client already acknowledged.
+/// parameter's `int64` maximum, and it leaves room for the `last + 1` id a
+/// file-watch resume sends, so that stream can never saturate onto an id the
+/// client already acknowledged. Process streams bound their ids by their own
+/// layout (see `ProcessEventId`) and refuse a higher id themselves.
 pub(crate) const MAX_LAST_EVENT_ID: u64 = i64::MAX as u64;
 
 pub(crate) fn last_event_id(headers: &HeaderMap) -> Result<Option<u64>, &'static str> {