fix(sdk): unavailable terminal event and reconnect budget for failed connects

crates/tracedecay-sdk/src/client.rs

@@ -373,7 +373,7 @@ impl Client {
             if !Operation::TERMINAL_STATES
                 .iter()
                 .copied()
-                .any(|state| terminal_state_name(state) == termination)
+                .any(|state| state.as_str() == termination)
             {
                 return Err(protocol(
                     response.status(),
@@ -718,18 +718,6 @@ fn admit_canonical_sdk_http_binding(
     }
 }
 
-const fn terminal_state_name(state: TerminalState) -> &'static str {
-    match state {
-        TerminalState::Completed => "completed",
-        TerminalState::Cancelled => "cancelled",
-        TerminalState::TimedOut => "timed_out",
-        TerminalState::Failed => "failed",
-        TerminalState::Unavailable => "unavailable",
-        TerminalState::EffectUnknown => "effect_unknown",
-        TerminalState::Partial => "partial",
-    }
-}
-
 const fn cancellation_point_name(point: CancellationPoint) -> &'static str {
     match point {
         CancellationPoint::BeforeAdmission => "before_admission",
@@ -803,10 +791,9 @@ pub(crate) fn problem_error(body: &Value, status: u16, missing_value: &'static s
 
 /// Whether an event name ends the operation's stream.
 fn is_terminal_event(event: &str) -> bool {
-    matches!(
-        event,
-        "completed" | "cancelled" | "timed_out" | "failed" | "partial" | "effect_unknown"
-    )
+    TerminalState::ALL
+        .iter()
+        .any(|state| state.as_str() == event)
 }
 
 fn protocol(status: u16, message: impl Into<String>) -> ClientError {
@@ -1039,6 +1026,10 @@ pub struct StreamResume {
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct StreamOptions {
     pub resume: Option<StreamResume>,
+    /// Reconnect attempts allowed after the first connection. Each attempt,
+    /// including one that fails to connect, spends this budget, and only
+    /// transport failures are retried. After the budget is spent the iterator
+    /// yields one `Err` and then `None`.
     pub max_reconnects: usize,
 }
 
@@ -1124,6 +1115,9 @@ impl OperationStream {
                 "daemon did not open a canonical event stream",
             ));
         }
+        self.pending_event = None;
+        self.pending_id = None;
+        self.pending_data = None;
         self.reader = Some(BufReader::new(response));
         Ok(())
     }
@@ -1396,30 +1390,47 @@ impl Iterator for OperationStream {
             return None;
         }
         loop {
-            match self.read_event() {
+            let mut disconnect = match self.read_event() {
                 Ok(Some(event)) => return Some(Ok(event)),
-                Ok(None) if self.reconnects < self.options.max_reconnects => {
-                    if self.next_sequence.is_none() || self.resume_token.is_none() {
-                        return Some(Err(statusless_protocol(
-                            "stream ended without a resumable frontier",
-                        )));
-                    }
-                    self.reconnects += 1;
-                    if let Err(error) = self.connect() {
-                        return Some(Err(error));
-                    }
-                }
                 Ok(None) => {
-                    return Some(Err(ClientError::Transport(
-                        "event stream ended before a terminal event".into(),
-                    )));
+                    ClientError::Transport("event stream ended before a terminal event".into())
+                }
+                Err(error @ ClientError::Transport(_)) => error,
+                Err(error) => return Some(Err(self.finish(error))),
+            };
+            // Every reconnect attempt, including one that fails to connect,
+            // spends budget. Only transport failures are retried; auth, problem,
+            // and protocol errors surface at once.
+            loop {
+                if self.reconnects >= self.options.max_reconnects {
+                    return Some(Err(self.finish(disconnect)));
+                }
+                if self.next_sequence.is_none() || self.resume_token.is_none() {
+                    return Some(Err(self.finish(statusless_protocol(
+                        "stream ended without a resumable frontier",
+                    ))));
+                }
+                self.reconnects += 1;
+                match self.connect() {
+                    Ok(()) => break,
+                    Err(error @ ClientError::Transport(_)) => disconnect = error,
+                    Err(error) => return Some(Err(self.finish(error))),
                 }
-                Err(error) => return Some(Err(error)),
             }
         }
     }
 }
 
+impl OperationStream {
+    /// Ends iteration after `error`, so a caller that keeps calling `next` sees
+    /// `None` instead of an endless error stream.
+    fn finish(&mut self, error: ClientError) -> ClientError {
+        self.terminal = true;
+        self.reader = None;
+        error
+    }
+}
+
 /// Parsed canonical application problem.
 #[derive(Clone, Debug, PartialEq)]
 pub struct ProblemError {

crates/tracedecay-sdk/src/codegen.rs

@@ -1025,14 +1025,25 @@ fn render_rust_operations(
 }
 
 fn render_types() -> String {
-    format!("{HEADER}{TYPES_SOURCE}")
+    format!("{HEADER}{TYPES_SOURCE}{}", operation_terminations_source())
+}
+
+/// The runtime list of terminal event names, derived from the catalog enum so
+/// SDK stream classification cannot drift from the server's terminal states.
+fn operation_terminations_source() -> String {
+    let names = TerminalState::ALL
+        .iter()
+        .map(|state| format!("\"{}\"", state.as_str()))
+        .collect::<Vec<_>>()
+        .join(", ");
+    format!("export const OPERATION_TERMINATIONS = [{names}] as const;\n")
 }
 
 fn render_dashboard_application(
     operations: &[Operation],
     dashboard_bindings: &BTreeMap<String, Vec<String>>,
 ) -> Result<String, Box<dyn Error>> {
-    let mut out = format!("{HEADER}{TYPES_SOURCE}");
+    let mut out = render_types();
     out.push_str(
         "\nexport interface ApplicationHttpBinding { readonly bindingIds: readonly [string, ...string[]]; readonly resultSchema: { readonly schemaId: string; readonly revision: number }; readonly terminalStates: readonly string[]; readonly cancellation: CanonicalCancellation; readonly receipt: \"operation\" | \"durable_effect\"; readonly reconciliation: \"not_required\" | \"required\" }\n\
          export const APPLICATION_HTTP_BINDINGS = {\n",
@@ -1100,7 +1111,7 @@ fn canonical_dashboard_bindings() -> Result<BTreeMap<String, Vec<String>>, Box<d
 const TYPES_SOURCE: &str = r##"export type UnknownValue = string & {};
 export type RetryDirective = "never" | "same_request" | "after_delay" | "after_revalidate" | "after_reconcile" | UnknownValue;
 export type LegalAction = "correct_request" | "reauthorize" | "refresh" | "retry" | "reconcile" | "reset" | "contact_administrator" | UnknownValue;
-export type OperationTermination = "completed" | "cancelled" | "timed_out" | "failed" | "unavailable" | "partial" | "effect_unknown" | UnknownValue;
+export type OperationTermination = (typeof OPERATION_TERMINATIONS)[number] | UnknownValue;
 export type EffectTermination = "completed" | "cancelled" | "timed_out" | "failed" | "partial" | "effect_unknown" | UnknownValue;
 export type ApplicationProblemKind = "invalid_request" | "not_found_or_not_authorized" | "conflict" | "partial_effect" | "stale" | "unsupported" | "unavailable" | "execution_failed" | "reset_required" | "saturated" | "cancelled" | "timed_out" | UnknownValue;
 export type ApplicationUnavailableClassV1 = "authority" | "backend_unavailable" | "backend_disconnected" | "backend_retryable" | UnknownValue;

crates/tracedecay-sdk/tests/sdk_suite/client.rs

@@ -13,6 +13,7 @@ use tracedecay_sdk::operations::{
     ApplicationFactStoreCurate, MultiRootScopeSetRead, OperationTransport, TypedOperation,
     WorkflowListDefinitions,
 };
+use tracedecay_tool_catalog::TerminalState;
 
 #[derive(Debug, Default)]
 struct RecordingMcpTransport {
@@ -823,6 +824,177 @@ fn near_limit_multiline_sse_frames_still_decode() {
     server.join().unwrap();
 }
 
+fn terminal_frame(event: &str, receipt_termination: &str) -> String {
+    format!(
+        concat!(
+            "event: {event}\nid: 0\n",
+            "data: {{\"event\":\"{event}\",\"data\":{{\"sequence\":0,\"terminal\":{{",
+            "\"termination\":\"{event}\",\"receipt\":{{\"started_at\":1,\"ended_at\":2,",
+            "\"effective_deadline\":{{\"expires_at\":3}},\"cancellation\":null,",
+            "\"budget\":{{\"units_consumed\":1,\"bytes_consumed\":1,\"elapsed_micros\":1}},",
+            "\"termination\":\"{receipt}\"}}}}}}}}\n\n"
+        ),
+        event = event,
+        receipt = receipt_termination,
+    )
+}
+
+#[test]
+fn every_server_terminal_state_ends_the_stream_without_reconnecting() {
+    for state in TerminalState::ALL {
+        let name = state.as_str();
+        let body = format!("{SSE_OPEN_FRAME}{}", terminal_frame(name, name));
+        let (base_url, server) = serve(vec![event_response(&body)]);
+        let client = Client::builder(ConnectionMode::local(&base_url, "project.sdk", "sdk-token"))
+            .build()
+            .unwrap();
+        let mut stream = client
+            .stream_operation(
+                "request.operation",
+                StreamOptions {
+                    resume: None,
+                    max_reconnects: 3,
+                },
+            )
+            .unwrap();
+
+        assert_eq!(stream.next().unwrap().unwrap().event, "open");
+        let terminal = stream.next().unwrap().unwrap();
+        assert_eq!(terminal.event, name);
+        assert!(terminal.terminal(), "{name} must be terminal");
+        assert!(stream.next().is_none(), "{name} must end iteration");
+        assert_eq!(server.join().unwrap().len(), 1, "{name} must not reconnect");
+    }
+}
+
+/// Serves one scripted reply per connection: `Some` writes the response and
+/// `None` reads the request and drops the connection, a transient transport
+/// failure. Joins to the number of connections accepted.
+fn serve_scripted(script: Vec<Option<String>>) -> (String, thread::JoinHandle<usize>) {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let address = listener.local_addr().unwrap();
+    let task = thread::spawn(move || {
+        let attempts = script.len();
+        for reply in script {
+            let (mut stream, _) = listener.accept().unwrap();
+            request(&mut stream);
+            if let Some(response) = reply {
+                stream.write_all(response.as_bytes()).unwrap();
+            }
+        }
+        attempts
+    });
+    (format!("http://{address}"), task)
+}
+
+fn scripted_stream(
+    script: Vec<Option<String>>,
+    max_reconnects: usize,
+) -> (
+    Result<tracedecay_sdk::client::OperationStream, ClientError>,
+    thread::JoinHandle<usize>,
+) {
+    let (base_url, server) = serve_scripted(script);
+    let client = Client::builder(ConnectionMode::local(&base_url, "project.sdk", "sdk-token"))
+        .build()
+        .unwrap();
+    let stream = client.stream_operation(
+        "request.operation",
+        StreamOptions {
+            resume: None,
+            max_reconnects,
+        },
+    );
+    (stream, server)
+}
+
+fn open_then_eof() -> Option<String> {
+    Some(event_response(SSE_OPEN_FRAME))
+}
+
+#[test]
+fn dropped_reconnects_are_retried_within_the_budget() {
+    let completed = terminal_frame("completed", "completed");
+    let (stream, server) = scripted_stream(
+        vec![
+            open_then_eof(),
+            None,
+            None,
+            Some(event_response(&completed)),
+        ],
+        3,
+    );
+    let events = stream
+        .unwrap()
+        .collect::<Result<Vec<_>, _>>()
+        .expect("two failed reconnects recover within three attempts");
+
+    assert_eq!(
+        events.iter().map(|e| e.event.as_str()).collect::<Vec<_>>(),
+        ["open", "completed"]
+    );
+    assert_eq!(server.join().unwrap(), 4);
+}
+
+#[test]
+fn exhausted_reconnect_budget_yields_one_error_then_ends() {
+    let (stream, server) = scripted_stream(vec![open_then_eof(), None, None], 2);
+    let mut stream = stream.unwrap();
+
+    assert_eq!(stream.next().unwrap().unwrap().event, "open");
+    assert!(matches!(
+        stream.next(),
+        Some(Err(ClientError::Transport(_)))
+    ));
+    assert!(stream.next().is_none(), "the error ends the iteration");
+    assert!(stream.next().is_none());
+    assert_eq!(server.join().unwrap(), 3);
+}
+
+#[test]
+fn the_first_connection_is_not_retried() {
+    let (stream, server) = scripted_stream(vec![None], 3);
+
+    assert!(matches!(stream, Err(ClientError::Transport(_))));
+    assert_eq!(server.join().unwrap(), 1);
+}
+
+#[test]
+fn authentication_failures_on_reconnect_are_not_retried() {
+    let denied = "HTTP/1.1 401 Unauthorized\r\ncontent-length: 0\r\nconnection: close\r\n\r\n";
+    let (stream, server) = scripted_stream(vec![open_then_eof(), Some(denied.to_owned())], 3);
+    let mut stream = stream.unwrap();
+
+    assert_eq!(stream.next().unwrap().unwrap().event, "open");
+    assert!(matches!(
+        stream.next(),
+        Some(Err(ClientError::Authentication(401)))
+    ));
+    assert!(stream.next().is_none());
+    assert_eq!(server.join().unwrap(), 2);
+}
+
+#[test]
+fn unavailable_terminal_with_a_disagreeing_receipt_is_a_protocol_error() {
+    let body = format!(
+        "{SSE_OPEN_FRAME}{}",
+        terminal_frame("unavailable", "completed")
+    );
+    let (base_url, server) = serve(vec![event_response(&body)]);
+    let client = Client::builder(ConnectionMode::local(&base_url, "project.sdk", "sdk-token"))
+        .build()
+        .unwrap();
+    let mut stream = client
+        .stream_operation("request.operation", StreamOptions::default())
+        .unwrap();
+    assert_eq!(stream.next().unwrap().unwrap().event, "open");
+    assert!(matches!(
+        stream.next(),
+        Some(Err(ClientError::Protocol { .. }))
+    ));
+    server.join().unwrap();
+}
+
 #[test]
 fn http_operations_refuse_the_mcp_execution_path_with_a_typed_error() {
     let mcp = Arc::new(RecordingMcpTransport::default());