fix(core): outlast transient transport faults and restart an interrupted stream

README.md

@@ -45,7 +45,7 @@ The design starts with one question: **What is the smallest capability Open Max
 | Lifecycle policy and events | Process hooks and permission files |
 | Compaction integration | The built-in model summary plus the `compaction` hook event |
 | Model endpoints | `providers.json`, including local servers, gateways, and private proxies |
-| Shortcuts or a completely different UI | Prompt templates, or a custom frontend speaking `openmax-stdio/5` |
+| Shortcuts or a completely different UI | Prompt templates, or a custom frontend speaking `openmax-stdio/6` |
 
 These are deliberate boundaries, not placeholders for hidden orchestration products. Open Max does not carry an MCP host, nested-agent scheduler, plan mode, background-job product, built-in TODO database, user-keybinding engine, pluggable compactor, or TUI plugin ABI. The agent composes those richer workflows from the same host tools a developer can inspect, edit, test, and remove.
 
@@ -198,7 +198,7 @@ Sessions, settings, tools, and skills stay under `~/.openmax/` and your project
 - [Configuration](docs/configuration.md): settings, approvals, providers, project trust
 - [Usage](docs/usage.md): CLI flags, keybindings, slash commands
 - [Extending](docs/extending.md): tools, skills, templates, hooks, permissions, validation, freezing
-- [stdio protocol](docs/stdio-protocol.md): the `openmax-stdio/5` contract for custom frontends
+- [stdio protocol](docs/stdio-protocol.md): the `openmax-stdio/6` contract for custom frontends
 
 ## Development
 

crates/core/src/agent.rs

@@ -2293,6 +2293,13 @@ impl TokenBatcher {
         match delta {
             StreamDelta::Content(t) => self.content.push_str(&t),
             StreamDelta::Reasoning(t) => self.thinking.push_str(&t),
+            // Flush first so the wire keeps the order the client saw: the
+            // reasoning already streamed, then the notice that it is void.
+            StreamDelta::Retry { attempt, max_attempts, reason } => {
+                self.flush();
+                self.core.send_agent(&self.session_id, AgentEvent::Retry { attempt, max_attempts, reason });
+                return;
+            }
         }
         if self.last_flush.elapsed() >= FLUSH_INTERVAL {
             self.flush();

crates/core/src/spec.rs

@@ -849,14 +849,14 @@ spending a read on the address - lexical ranking cannot separate those two,
 because they are about the same thing in the same words.
 "#;
 
-const STDIO: &str = r#"# stdio protocol (openmax-stdio/5)
+const STDIO: &str = r#"# stdio protocol (openmax-stdio/6)
 
 `openmax --stdio` speaks line-delimited JSON both ways: commands on stdin,
 `AgentEvent` envelopes on stdout. This is the stable contract for custom
 frontends, editor integrations, and one openmax driving another.
 
 Handshake: the first stdout line is
-{"type":"hello","proto":"openmax-stdio/5","protocol_version":5,"session_id":"...","version":"...","project":"/abs/path","continued":false}.
+{"type":"hello","proto":"openmax-stdio/6","protocol_version":6,"session_id":"...","version":"...","project":"/abs/path","continued":false}.
 `protocol_version` is compared as an integer; any wire change bumps it.
 
 Commands, one JSON object per line:
@@ -898,7 +898,10 @@ the frozen tool schemas sent on every request, context_tokens),
 refreeze receipt, or a policy/providers/settings/approval notice - surfaced
 here so a frontend can render what the model sees; `call_id` links it to the
 tool result it rode, or is empty for a note inserted before the next prompt
-like a turn-start receipt), `diff` (call_id,
+like a turn-start receipt), `retry` (attempt, max_attempts, reason: the
+model request is being resent after a transport failure, a 429, or a stream
+that died before any reply text; thinking already streamed for that attempt
+is void), `diff` (call_id,
 path, diff, added, removed), `approval_request` (approval_id, name, summary,
 detail, reason, source_path, source_sha, and an optional `env`), `approval_settled` (approval_id,
 outcome), `refrozen` (tools, skills, changes: the refreeze receipt naming
@@ -946,9 +949,10 @@ only guaranteed terminator. A command that starts no turn (empty text, an
 untrusted project) still gets one, with stop_reason `refused`, after the
 `protocol_error` that says why. A turn that dies unexpectedly reports
 `error` and then `done` with stop_reason `error`; a provider stream that ends
-mid-answer with no completion signal reports its partial `message_done`, then
-`error`, then `done` with stop_reason `truncated`, and no tool call it carried
-is run. The single exception is a
+mid-answer with no completion signal is resent (each resend announced by a
+`retry`) while no reply text has streamed, and otherwise reports its partial
+`message_done`, then `error`, then `done` with stop_reason `truncated`, and
+no tool call it carried is run. The single exception is a
 `user` sent while a turn is in flight: that is refused with a
 `protocol_error` and no `done`, because the running turn owns the next one.
 
@@ -1338,6 +1342,7 @@ mod tests {
             },
             AgentEvent::ToolEnd { call_id: String::new(), ok: true, output: String::new() },
             AgentEvent::HarnessNote { call_id: String::new(), text: String::new() },
+            AgentEvent::Retry { attempt: 0, max_attempts: 0, reason: String::new() },
             AgentEvent::Diff {
                 call_id: String::new(),
                 path: String::new(),

crates/core/src/types.rs

@@ -108,6 +108,13 @@ pub enum AgentEvent {
     /// links it to the tool result it rides when it rode one, else empty (a
     /// note inserted before the next prompt, e.g. a turn-start receipt).
     HarnessNote { call_id: String, text: String },
+    /// The model request is being resent: `attempt` of `max_attempts` goes
+    /// out after a backoff, `reason` having ended the previous one (a
+    /// transport failure, a 429, or a stream that died before any reply
+    /// text). Emitted before the wait; a cancel during it ends the turn and
+    /// the attempt never goes out. Thinking streamed for the failed attempt
+    /// is void; no Token preceded it.
+    Retry { attempt: u32, max_attempts: u32, reason: String },
     Diff { call_id: String, path: String, diff: String, added: usize, removed: usize },
     /// Mutating tool waiting on the user. `detail` is a short args preview
     /// (paths, command head) for the TUI card; may be empty.
@@ -138,7 +145,7 @@ pub enum AgentEvent {
         /// as structured data, not folded into `detail`, so the frontend
         /// controls its own un-clippable placement. Additive and defaulted:
         /// a stream without the key deserializes unchanged, and the key is
-        /// omitted from the wire whenever it is empty, so `openmax-stdio/5`
+        /// omitted from the wire whenever it is empty, so `openmax-stdio/6`
         /// bytes are byte-identical for every call that grants no env.
         #[serde(default, skip_serializing_if = "Vec::is_empty")]
         env: Vec<String>,
@@ -220,7 +227,7 @@ mod tests {
 
     /// Golden wire format for every `AgentEvent`, wrapped in its envelope
     /// exactly as `--stdio` and `--print --json` emit it. These strings are
-    /// the `openmax-stdio/5` contract: session_id first, then the `type`
+    /// the `openmax-stdio/6` contract: session_id first, then the `type`
     /// discriminator, then variant fields in declaration order. A change here
     /// is a protocol break and must bump `PROTO_VERSION`.
     /// `every_agent_event_variant_is_pinned_here` fails to compile if a variant
@@ -345,6 +352,10 @@ mod tests {
             env(AgentEvent::SchemasOverBudget { schema_tokens: 6800, budget_tokens: 2150 }),
             r#"{"session_id":"s1","type":"schemas_over_budget","schema_tokens":6800,"budget_tokens":2150}"#
         );
+        assert_eq!(
+            env(AgentEvent::Retry { attempt: 2, max_attempts: 8, reason: "request failed: connection reset".into() }),
+            r#"{"session_id":"s1","type":"retry","attempt":2,"max_attempts":8,"reason":"request failed: connection reset"}"#
+        );
 
         assert_eq!(
             env(AgentEvent::HookFailed {
@@ -428,6 +439,7 @@ mod tests {
                 AgentEvent::ToolStart { .. } => {}
                 AgentEvent::ToolEnd { .. } => {}
                 AgentEvent::HarnessNote { .. } => {}
+                AgentEvent::Retry { .. } => {}
                 AgentEvent::Diff { .. } => {}
                 AgentEvent::ApprovalRequest { .. } => {}
                 AgentEvent::ApprovalSettled { .. } => {}

crates/tui/src/app.rs

@@ -2747,6 +2747,17 @@ impl App {
                     ));
                 }
             }
+            AgentEvent::Retry { attempt, max_attempts, reason } => {
+                // The reasoning shown so far belongs to the attempt that
+                // failed; the fresh attempt starts its own. The reason can
+                // carry a backend body; the note is one line.
+                self.thinking_tail.clear();
+                self.thinking_source.clear();
+                self.thinking_wrapped.clear();
+                self.thinking_chars = 0;
+                self.note(&format!("{}; retrying ({attempt} of {max_attempts})", open_max_core::text::one_line(&reason)));
+                self.dirty.mark_tail();
+            }
             AgentEvent::SchemasOverBudget { schema_tokens, budget_tokens } => {
                 // Says what it costs and what to do, not how compaction reacts:
                 // that depends on whether any room is left at all.
@@ -5499,6 +5510,29 @@ mod tests {
         assert_eq!(home_shortened("/srv/data", Some("/")), "/srv/data");
     }
 
+    /// The reasoning tail shown during a turn belongs to one attempt. A
+    /// retry starts the reply over, so what the failed attempt streamed is
+    /// dropped before the fresh attempt's reasoning arrives; otherwise the
+    /// two would read as one thought.
+    #[test]
+    fn a_retry_drops_the_failed_attempts_reasoning_tail() {
+        let (mut app, dir) = app_fixture();
+        app.running = true;
+        app.turn_started = Some(std::time::Instant::now());
+        app.on_agent_event(AgentEvent::Thinking { text: "abandoned line".into() });
+        assert_eq!(app.thinking_tail, "abandoned line");
+        app.on_agent_event(AgentEvent::Retry {
+            attempt: 2,
+            max_attempts: 8,
+            reason: "the stream ended before the reply finished".into(),
+        });
+        assert!(app.thinking_tail.is_empty());
+        assert_eq!(app.thinking_chars, 0);
+        app.on_agent_event(AgentEvent::Thinking { text: "fresh line".into() });
+        assert_eq!(app.thinking_tail, "fresh line");
+        fs::remove_dir_all(dir).unwrap();
+    }
+
     #[test]
     fn streaming_output_grows_above_the_fixed_prompt() {
         let (mut app, dir) = app_fixture();

crates/tui/src/headless.rs

@@ -264,6 +264,11 @@ async fn run_turn_events(
                     );
                 }
             }
+            AgentEvent::Retry { attempt, max_attempts, reason } => {
+                if !json {
+                    let _ = writeln!(stderr, "openmax: {}; retrying ({attempt} of {max_attempts})", one_line(reason));
+                }
+            }
             AgentEvent::SchemasOverBudget { schema_tokens, budget_tokens } => {
                 // Advisory: the turn still runs, so the exit code is untouched.
                 if !json {

crates/tui/src/stdio.rs

@@ -4,7 +4,7 @@
 //! speak line-delimited JSON (an editor plugin, an orchestrator, another
 //! openmax) can drive a complete interactive session, approvals included.
 //!
-//! Protocol (`openmax-stdio/5`), one JSON object per line. The normative
+//! Protocol (`openmax-stdio/6`), one JSON object per line. The normative
 //! reference (every field of every line) is `docs/stdio-protocol.md`, kept in
 //! step with `openmax --spec stdio`; `crates/core/src/types.rs` golden tests
 //! pin the event wire.
@@ -18,7 +18,7 @@
 //!   {"cmd":"quit"}                                   finish the turn, then exit
 //!
 //! stdout lines:
-//!   {"type":"hello","proto":"openmax-stdio/5","protocol_version":5,"session_id":"...","version":"...","project":"..."}
+//!   {"type":"hello","proto":"openmax-stdio/6","protocol_version":6,"session_id":"...","version":"...","project":"..."}
 //!   AgentEvent envelopes exactly as `--print --json` emits them
 //!   {"type":"protocol_error","message":"..."}        bad input; session unharmed
 //!
@@ -46,11 +46,11 @@ use open_max_core::types::{AgentEvent, AgentEventEnvelope};
 use serde::Deserialize;
 use tokio::sync::mpsc;
 
-pub const PROTO: &str = "openmax-stdio/5";
+pub const PROTO: &str = "openmax-stdio/6";
 /// Machine-comparable protocol major. A client negotiates on this integer;
 /// `PROTO` embeds the same number as a human-readable id (checked in tests).
 /// Bump on any wire change (event field, command shape, framing line).
-pub const PROTO_VERSION: u32 = 5;
+pub const PROTO_VERSION: u32 = 6;
 
 // Unknown `cmd` values are protocol errors; extra fields on a known command
 // are ignored (lenient by design, so clients can annotate lines freely).
@@ -465,7 +465,7 @@ fn transcript_value(
     })
 }
 
-/// Validate one JSONL line against the `openmax-stdio/5` contract using the
+/// Validate one JSONL line against the `openmax-stdio/6` contract using the
 /// authoritative types (`Command` for stdin, `AgentEvent` for stdout events),
 /// so there is no second schema to drift. Returns a short label on success
 /// (`cmd user`, `event token`, `hello`) or a human reason on failure.

crates/tui/tests/cli.rs

@@ -602,8 +602,8 @@ fn stdio_handshake_speaks_the_contract() {
     reader.read_line(&mut hello).unwrap();
     let hello: serde_json::Value = serde_json::from_str(&hello).unwrap();
     assert_eq!(hello["type"], "hello");
-    assert_eq!(hello["proto"], "openmax-stdio/5");
-    assert_eq!(hello["protocol_version"], 5);
+    assert_eq!(hello["proto"], "openmax-stdio/6");
+    assert_eq!(hello["protocol_version"], 6);
     assert!(hello["session_id"].is_string());
 
     writeln!(stdin, r#"{{"cmd":"quit"}}"#).unwrap();
@@ -1084,11 +1084,14 @@ fn a_truncated_stream_reports_truncation_instead_of_a_clean_stop() {
 /// call, so the arguments parse and nothing looks broken. A stream with no
 /// completion signal is not a response the model asked to act on (more calls
 /// may have been coming, or this one may still have been under revision), so
-/// the call must not run.
+/// the call must not run. Reply text streams first, so this is the
+/// interruption the client does not start over.
 #[test]
 fn a_truncated_stream_never_runs_the_tool_call_it_carried() {
     let (project, home) = fresh_dirs("truncated-native-call");
-    let (base_url, _requests, _server) = spawn_scripted_server(vec![(WRITE_CALL_SSE.to_string(), false)]);
+    let prose = "data: {\"choices\":[{\"delta\":{\"content\":\"writing it\"},\"finish_reason\":null}]}\n\n";
+    let (base_url, _requests, _server) =
+        spawn_scripted_server(vec![(format!("{prose}{WRITE_CALL_SSE}"), false)]);
     // auto, so a refusal here is the truncation and not the approval gate.
     write_settings_with_mode(&home, &base_url, "auto");
 

docs/stdio-protocol.md

@@ -1,4 +1,4 @@
-# stdio protocol (`openmax-stdio/5`)
+# stdio protocol (`openmax-stdio/6`)
 
 `openmax --stdio` speaks line-delimited JSON both ways, so any process that
 reads and writes JSONL (an editor plugin, an orchestrator, another openmax) can
@@ -15,7 +15,7 @@ This file is the normative reference for every field of every line.
 The first stdout line is:
 
 ```json
-{"type":"hello","proto":"openmax-stdio/5","protocol_version":5,"session_id":"...","version":"0.2.0","project":"/abs/path","continued":false}
+{"type":"hello","proto":"openmax-stdio/6","protocol_version":6,"session_id":"...","version":"0.2.0","project":"/abs/path","continued":false}
 ```
 
 `protocol_version` is an integer a client compares directly; `proto` carries
@@ -72,6 +72,7 @@ one).
 | `tool_start` | `call_id`, `name`, `args` (object) |
 | `tool_end` | `call_id`, `ok` (bool), `output` |
 | `harness_note` | `call_id`, `text` (a note the harness wrote into the model's transcript: a refreeze receipt, or a policy, providers, settings, or approval notice. `call_id` links it to the tool result it rode; it is empty for a note inserted before the next prompt, such as a turn-start receipt) |
+| `retry` | `attempt`, `max_attempts`, `reason` (the model request is being resent: `attempt` is the one that goes out of the `max_attempts` budget after a backoff wait, `reason` having ended the previous one with a transport failure, a 429, or a stream that died before any reply text arrived. Emitted before the wait; a `cancel` during it ends the turn and the attempt never goes out. `thinking` lines already emitted for that attempt are void and the reply starts over; no `token` line precedes a retried stream. A request gives up before the budget when three attempts in a row never reached the endpoint) |
 | `diff` | `call_id`, `path`, `diff`, `added`, `removed` |
 | `approval_request` | `approval_id`, `name`, `summary`, `detail`, `reason` (`gate`, or `unapproved_source` which unattended clients must never auto-approve), `source_path`, `source_sha`, and optional `env` (see below) |
 | `approval_settled` | `approval_id`, `outcome` (`approved`, `declined`, `timed_out`, or `cancelled`) |
@@ -137,7 +138,7 @@ followed by `done` with `stop_reason` `refused`, so a client that blocks on
 | `stop_reason` | Meaning |
 | --- | --- |
 | provider `finish_reason` | Passed through verbatim on a normal turn, commonly `stop` or `length`. Treat any unlisted value as a normal end |
-| `truncated` | The provider stream ended with no completion signal; the reply is incomplete, any tool calls it carried were refused, and an `error` line precedes it |
+| `truncated` | The provider stream ended with no completion signal (after reply text had streamed, or on the last retry) or exceeded a client limit; the reply is incomplete, any tool calls it carried were refused, and an `error` line precedes it |
 | `max_iterations` | The turn hit the tool-call ceiling |
 | `budget_exhausted` | The per-turn `max_agent_tokens` cap refused the next request at admission; nothing was sent, and resubmitting continues the work |
 | `unverified` | A blocking `turn_end` hook refused the completion more times than the harness honors (8), or its refusal could not be persisted; the reply stands unverified |