[Bridge] Armed a stream from a Nexus read that waits and finds nothing.

crates/streams/nexus/temporal_streams.nexusrpc.yaml

@@ -189,6 +189,13 @@ $defs:
           producer attempts the reader already saw, so its superseded and
           stale marks are exact across calls and across handler Workers. A
           state that doesn't parse is refused as a cursor mistake.
+      arm:
+        type: boolean
+        description: >-
+          For a reader that sleeps until its caller hears of progress. When
+          the call waits its whole wait_ms and collects nothing, the endpoint
+          arms the stream, so the next append notifies the stream's callers.
+          A reader with no callback leaves it out.
     required: [stream]
     additionalProperties: false
 
@@ -280,6 +287,12 @@ $defs:
           also when the answer was cut short by its byte budget. It holds at
           most 16 KiB, which is about 22 KiB as base64, so it can ride in a
           Workflow's History. A longer state is refused.
+      armed:
+        type: boolean
+        description: >-
+          The call asked to arm, collected nothing and armed the stream, so
+          the caller waits for progress. Never true with records or with done.
+          An empty answer without it falls back to waiting for progress too.
     required: [records, next_token, done]
     additionalProperties: true
 

crates/streams/src/nexus/mod.rs

@@ -307,6 +307,7 @@ impl NexusCalls {
                     next_token: latest.cursor,
                     done: false,
                     state: None,
+                    armed: false,
                 }),
                 Err(error) => failed(failure::of_stream(&error)),
             };
@@ -640,6 +641,7 @@ impl Budget {
             next_token: read.response.cursor.clone(),
             done: false,
             state: Some(Base64::default()),
+            armed: false,
         });
         let state = crossing_size(read.state.encoded_len().min(MAX_STATE_BYTES));
         Self {
@@ -714,6 +716,8 @@ fn finish_read(
         wires.push(wire);
         kept.extend(stored);
     }
+    // Core's read no longer arms, so the contract's `armed` stays false until it leaves the IDL.
+    let armed = false;
     let (next_token, done) = if cut {
         // A supersession belongs to the record after it, so it goes into the next answer too.
         if wires.last().is_some_and(|wire| wire.superseded.is_some()) {
@@ -734,6 +738,7 @@ fn finish_read(
         next_token,
         done,
         state: Some(Base64(state.encode_to_vec())),
+        armed,
     })
 }
 
@@ -807,6 +812,7 @@ fn read_failure(error: StreamError, state: Vec<u8>) -> NexusStep {
             next_token: cursor,
             done: false,
             state: (!state.is_empty()).then_some(Base64(state)),
+            armed: false,
         }),
         _ => failed(failure::of_stream(&error)),
     }

crates/streams/src/nexus/wire.rs

@@ -96,6 +96,8 @@ pub(crate) struct ReadInput {
     /// Left as base64, so an oversized state is refused before it is decoded.
     #[serde(default)]
     pub(crate) state: Option<String>,
+    #[serde(default)]
+    pub(crate) arm: Option<bool>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -105,6 +107,8 @@ pub(crate) struct ReadOutput {
     pub(crate) done: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) state: Option<Base64>,
+    #[serde(skip_serializing_if = "std::ops::Not::not")]
+    pub(crate) armed: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]

crates/streams/tests/nexus_contract.rs

@@ -58,6 +58,8 @@ fn full_input(contract: &Value, name: &str, workflow_id: &str) -> Value {
             ("ReadInput", "wait_ms") => json!(0),
             ("ReadInput", "latest_only") => json!(false),
             ("ReadInput", "state") => Value::Null,
+            // So the answer carries `armed`, which the output check holds to the contract.
+            ("ReadInput", "arm") => json!(true),
             (model, member) => panic!("the contract added {model}.{member}; teach this test"),
         };
         out.insert(member, value);

crates/streams/tests/nexus_service.rs

@@ -699,6 +699,20 @@ mod cases {
         assert!(small.get("error").is_none());
         assert_eq!(texts(&answer), ["small"]);
     }
+    fn arming(wf: &str, after: &str) -> Value {
+        let mut input = read_input(wf, after);
+        input["arm"] = json!(true);
+        input
+    }
+
+    pub(crate) async fn a_read_that_asks_to_arm_is_never_armed(store: Arc<dyn StreamStore>) {
+        let calls = service(store, NexusOptions::default());
+        let wf = workflow();
+        let answer = read(&calls, arming(&wf, "")).await.unwrap();
+        assert!(records(&answer).is_empty());
+        assert!(answer.get("armed").is_none(), "{answer}");
+        assert_eq!(answer["done"], false);
+    }
 }
 
 macro_rules! lanes {
@@ -738,6 +752,7 @@ macro_rules! all_cases {
             a_record_above_the_budget_still_crosses_alone,
             an_append_above_the_record_limit_is_refused,
             a_stored_record_above_the_limit_crosses_as_an_error,
+            a_read_that_asks_to_arm_is_never_armed,
         );
     };
 }