[Bridge] Removed the arm from the Nexus read.

crates/streams/nexus/temporal_streams.nexusrpc.yaml

@@ -205,13 +205,6 @@ $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.
       floor_holder:
         type: string
         description: >-
@@ -354,12 +347,6 @@ $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.
       replay:
         $ref: "#/$defs/ReplayRange"
         description: >-

crates/streams/src/nexus/mod.rs

@@ -331,7 +331,6 @@ impl NexusCalls {
                     next_token: latest.cursor,
                     done: false,
                     state: None,
-                    armed: false,
                     replay: None,
                 }),
                 Err(error) => failed(failure::of_stream(&error)),
@@ -730,7 +729,6 @@ impl Budget {
             next_token: read.response.cursor.clone(),
             done: false,
             state: Some(Base64::default()),
-            armed: false,
             replay: None,
         });
         let state = crossing_size(read.state.encoded_len().min(MAX_STATE_BYTES));
@@ -796,8 +794,6 @@ fn finish_read(
         wires.push(wire);
         kept.push(record);
     }
-    // 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()) {
@@ -825,7 +821,6 @@ fn finish_read(
         next_token,
         done,
         state: Some(Base64(state.encode_to_vec())),
-        armed,
         replay,
     })
 }
@@ -864,7 +859,6 @@ fn finish_read_range(
         next_token: response.cursor,
         done: false,
         state: None,
-        armed: false,
         replay: None,
     })
 }
@@ -996,7 +990,6 @@ fn read_failure(error: StreamError, state: Vec<u8>, positions: bool) -> NexusSte
                 records_digest: Base64(records_digest(&[])),
             }),
             state: (!state.is_empty()).then_some(Base64(state)),
-            armed: false,
         }),
         _ => failed(failure::of_stream(&error)),
     }

crates/streams/src/nexus/wire.rs

@@ -97,8 +97,6 @@ pub(crate) struct ReadInput {
     #[serde(default)]
     pub(crate) state: Option<String>,
     #[serde(default)]
-    pub(crate) arm: Option<bool>,
-    #[serde(default)]
     pub(crate) floor_holder: Option<String>,
 }
 
@@ -126,8 +124,6 @@ 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,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) replay: Option<ReplayRange>,
 }
@@ -276,7 +272,6 @@ mod tests {
             next_token: "c".to_string(),
             done: false,
             state: None,
-            armed: false,
             replay: Some(ReplayRange {
                 chain_first_run_id: "run-1".to_string(),
                 after_token: String::new(),

crates/streams/tests/nexus_contract.rs

@@ -58,8 +58,6 @@ 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),
             ("ReadInput", "floor_holder") => json!("ns/run-1/reader"),
             (model, member) => panic!("the contract added {model}.{member}; teach this test"),
         };

crates/streams/tests/nexus_service.rs

@@ -702,19 +702,13 @@ 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>) {
+    pub(crate) async fn a_read_that_asks_to_arm_is_refused(store: Arc<dyn StreamStore>) {
+        // The contract has no arm, and a read refuses members it doesn't know.
         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);
+        let mut input = read_input(&workflow(), "");
+        input["arm"] = json!(true);
+        let refused = read(&calls, input).await.unwrap_err();
+        assert_eq!(kind_of(&refused).0, "BAD_REQUEST");
     }
 }
 
@@ -755,7 +749,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,
+            a_read_that_asks_to_arm_is_refused,
         );
     };
 }