Replay provider streams interrupted after partial output

.config/nextest.toml

@@ -139,7 +139,7 @@ binary_id(cyber-account)
 | binary_id(cyber-server::pty)
 | binary_id(cyber-server::requests)
 | binary_id(cyber-server::resume_children)
-| binary_id(cyber-server::runtime) & (test(loaded_) | test(registration) | test(complete_tool_call_is_durable_before_the_stream_finishes) | test(transport_failure_preserves_tool_outcomes_without_replaying_actions) | test(delete_) | test(shutdown_))
+| binary_id(cyber-server::runtime) & (test(loaded_) | test(registration) | test(a_complete_tool_call_is_recorded_when_its_stream_completes) | test(a_mid_stream_retry_records_only_the_replayed_attempt) | test(transport_failure_preserves_tool_outcomes_without_replaying_actions) | test(delete_) | test(shutdown_))
 | binary_id(cyber-server::structured_results)
 | binary_id(cyber-server::subagent_names)
 | binary_id(cyber-server::subtask_cancellation)

ROADMAP.md

@@ -43,7 +43,7 @@ The first release prioritizes reliable editing, recovery and measurable coding q
 
 **Provider base prompts and output styles (`system-context`): Status: implemented** (change `add-output-styles`). Base prompts follow the catalog model's family rather than the provider id; built-in, user and trust-gated project output styles are selected per primary Session (`/output-style`, `cyber exec --output-style`, `output_style`) and apply through a new Context Epoch. Plugin-distributed styles and `cyber import` of `.claude/output-styles` are deferred.
 
-**Model protocol fidelity (`session-runtime`, `provider-catalog`, `system-context`, `builtin-tools`, `tool-registry`): Status: in progress** (change `improve-model-protocol-fidelity`). Reasoning is kept as ordered items with opaque provider state and replayed natively to the model that produced it; per-model behavior profiles replace family checks; requests gain the Responses fields and routing headers the models use; streams have an idle timeout; known provider error codes map to typed reasons with server-advised delays; the base prompts are rewritten per family, versioned, and checked clean-room against a reference product's prompts. Live tool-loop tests for OpenAI, Anthropic, DeepSeek and Gemini 3 are written and run with keys (`cargo test -p cyber-llm --test live -- --ignored`); they have not yet been run. The profile's `edit_tool` picks the edit tools, with a freeform `apply_patch` constrained by our own envelope grammar on OpenAI Responses; hunks match tolerantly (whitespace, typographic punctuation) and fail as ambiguous rather than guess; a shell `apply_patch` heredoc runs as the patch tool when no such program is installed; and the profile's token budget with `middle` truncation keeps both ends of long output. Remaining: a run of the live tests with provider keys and the before/after pinned-repository eval on `openai/gpt-6-sol`. The Responses WebSocket transport, zstd bodies and server-side compaction are separate follow-up changes.
+**Model protocol fidelity (`session-runtime`, `provider-catalog`, `system-context`, `builtin-tools`, `tool-registry`): Status: in progress** (change `improve-model-protocol-fidelity`). Reasoning is kept as ordered items with opaque provider state and replayed natively to the model that produced it; per-model behavior profiles replace family checks; requests gain the Responses fields and routing headers the models use; streams have an idle timeout; known provider error codes map to typed reasons with server-advised delays; the base prompts are rewritten per family, versioned, and checked clean-room against a reference product's prompts. Live tool-loop tests for OpenAI, Anthropic, DeepSeek and Gemini 3 are written and run with keys (`cargo test -p cyber-llm --test live -- --ignored`); they have not yet been run. The profile's `edit_tool` picks the edit tools, with a freeform `apply_patch` constrained by our own envelope grammar on OpenAI Responses; hunks match tolerantly (whitespace, typographic punctuation) and fail as ambiguous rather than guess; a shell `apply_patch` heredoc runs as the patch tool when no such program is installed; and the profile's token budget with `middle` truncation keeps both ends of long output. A retryable stream failure after partial output now discards the partial answer and replays the request, within the same retry budget and only while none of its tool calls has been dispatched (change `retry-interrupted-streams`, which followed a pilot task lost to a mid-answer transport error). Remaining: a run of the live tests with provider keys and the before/after pinned-repository eval on `openai/gpt-6-sol`. The Responses WebSocket transport, zstd bodies and server-side compaction are separate follow-up changes.
 
 **Exit criteria**
 - Runs a 50-turn coding task end-to-end with OpenAI, Anthropic and a local model through the compatible adapter. Each P0 adapter also passes the published live coding baseline in `harness-evaluation`; turn count alone is insufficient.

crates/cyber-app/tests/server.rs

@@ -2310,10 +2310,11 @@ fn read_json_request(stream: &mut std::net::TcpStream) -> Option<serde_json::Val
 }
 
 /// `tool-registry` → Eager execution during streaming through the production App and
-/// its OpenAI-compatible adapter: the first of two streamed calls is durable as soon as
-/// the second one starts, while the provider stream is still open.
+/// its OpenAI-compatible adapter: while the stream is open it may still be replayed
+/// (`provider-catalog` → Retry policy), so a completed call is not durable yet; both calls
+/// are recorded, in order, when the stream completes.
 #[tokio::test]
-async fn openai_compatible_call_is_durable_while_its_stream_is_still_open() {
+async fn openai_compatible_calls_are_recorded_when_their_stream_completes() {
     use cyber_server::runtime::{Admission, CreateSession, Delivery};
     use serde_json::json;
     use std::io::Write;
@@ -2325,7 +2326,8 @@ async fn openai_compatible_call_is_durable_while_its_stream_is_still_open() {
     let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
     let url = format!("http://{}/v1", listener.local_addr().unwrap());
     let release = Arc::new(AtomicBool::new(false));
-    let open = release.clone();
+    let streamed = Arc::new(AtomicBool::new(false));
+    let (open, sent) = (release.clone(), streamed.clone());
     let provider = std::thread::spawn(move || {
         let chunk = |delta: serde_json::Value, finish: Option<&str>| {
             format!(
@@ -2365,6 +2367,7 @@ async fn openai_compatible_call_is_durable_while_its_stream_is_still_open() {
                 chunk(read(1, "call_b", "b.txt"), None)
             );
             let _ = stream.flush();
+            sent.store(true, Ordering::SeqCst);
             while !open.load(Ordering::SeqCst) {
                 std::thread::sleep(Duration::from_millis(5));
             }
@@ -2413,19 +2416,19 @@ async fn openai_compatible_call_is_durable_while_its_stream_is_still_open() {
             .map(|e| e.data["call_id"].as_str().unwrap().to_owned())
             .collect()
     };
-    let early = tokio::time::timeout(Duration::from_secs(20), async {
-        loop {
-            let calls = called(&application.store);
-            if !calls.is_empty() {
-                break calls;
-            }
+    tokio::time::timeout(Duration::from_secs(20), async {
+        while !streamed.load(Ordering::SeqCst) {
             tokio::time::sleep(Duration::from_millis(10)).await;
         }
     })
     .await
-    .expect("no call became durable while the stream was open");
-    assert_eq!(early, ["call_a"], "only the completed call is recorded");
-    assert!(!release.load(Ordering::SeqCst));
+    .expect("the provider did not stream the calls");
+    // `call_a` is complete once `call_b` starts; give the Drain time to have seen it.
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    assert!(
+        called(&application.store).is_empty(),
+        "nothing is durable while the stream can be replayed"
+    );
     release.store(true, Ordering::SeqCst);
     application.runtime.wait_idle(&id).await;
     assert_eq!(called(&application.store), ["call_a", "call_b"]);

crates/cyber-cli/src/commands/exec/mod.rs

@@ -527,6 +527,7 @@ async fn handle(
             out.error(&message);
             run.error = Some(message);
         }
+        "session.retry" => out.retry(&event.data),
         "session.deleted" => {
             run.error = Some("the Session was deleted".into());
             return Step::End;

crates/cyber-cli/src/commands/exec/report.rs

@@ -291,6 +291,17 @@ pub struct Out {
     review: Option<Review>,
 }
 
+/// The `retry` stream-json payload of a `session.retry` event.
+fn retry_payload(data: &Value) -> Value {
+    json!({
+        "message_id": data["message_id"],
+        "attempt": data["attempt"],
+        "delay_ms": data["delay_ms"],
+        "error": data["error"],
+        "discarded": data["discarded"] == true,
+    })
+}
+
 fn now_ms() -> i64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -554,6 +565,24 @@ impl Out {
         }
     }
 
+    /// `session.retry`: a provider retry. A `discarded` attempt never reached the output,
+    /// which holds recorded content only (`provider-catalog` → Retry policy).
+    pub fn retry(&self, data: &Value) {
+        let payload = retry_payload(data);
+        self.note(&format!(
+            "retry {} in {} ms: {}{}",
+            payload["attempt"],
+            payload["delay_ms"],
+            data["error"].as_str().unwrap_or_default(),
+            if payload["discarded"] == true {
+                " (partial answer discarded)"
+            } else {
+                ""
+            }
+        ));
+        self.line("retry", payload);
+    }
+
     /// A durable session event.
     pub fn durable(&mut self, run: &mut Run, kind: &str, data: &Value) {
         let base = kind.rsplit_once('.').map_or(kind, |(b, _)| b);
@@ -1039,6 +1068,30 @@ mod tests {
         );
     }
 
+    /// A discarded attempt is reported as a `retry` line; only recorded content becomes
+    /// `assistant_text`, so the run's result is the replayed answer alone.
+    #[test]
+    fn a_discarded_attempt_is_a_retry_line_and_never_the_result() {
+        let retry = json!({ "session_id": "ses_1", "message_id": "msg_1", "attempt": 1,
+            "delay_ms": 900, "error": "transport: reset", "discarded": true });
+        assert_eq!(
+            retry_payload(&retry),
+            json!({ "message_id": "msg_1", "attempt": 1, "delay_ms": 900,
+                "error": "transport: reset", "discarded": true })
+        );
+        let args = Wrapper::parse_from(["exec", "hi"]).args;
+        let mut out = Out::new(Format::Json, &args, &json!({ "id": "ses_1" }));
+        let mut run = Run::new(&args, Instant::now());
+        out.retry(&retry);
+        out.durable(
+            &mut run,
+            "session.text.ended.1",
+            &json!({ "message_id": "msg_1", "text": "Done." }),
+        );
+        assert_eq!(run.text, "Done.");
+        assert_eq!(out.summary(&run, 0)["result"], "Done.");
+    }
+
     #[test]
     fn only_forbidden_refusals_become_review_denials() {
         let api = |tag: &str| ClientError::Api {

crates/cyber-llm/examples/turn.rs

@@ -75,8 +75,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
             adapter.as_ref(),
             &request,
             &RetryPolicy::default(),
-            |n, d, e| {
-                eprintln!("retry {n} in {d:?}: {e}");
+            |retry| {
+                eprintln!(
+                    "retry {} in {:?}: {}",
+                    retry.attempt, retry.delay, retry.error
+                );
             },
         )
         .await?;

crates/cyber-llm/src/lib.rs

@@ -20,7 +20,7 @@ mod types;
 
 pub use adapters::{Adapter, ApiKind, Endpoint, EventStream};
 pub use error::{ErrorKind, LlmError, classify, local_clock};
-pub use retry::{RetryPolicy, open_with_retry};
+pub use retry::{RetryNotice, RetryPolicy, RetryingStream, open_with_retry};
 pub use sse::{SseEvent, SseParser};
 pub use turn::{TurnOutput, collect};
 pub use types::{

crates/cyber-llm/src/retry.rs

@@ -1,16 +1,26 @@
 //! Retry policy (`provider-catalog` → Retry policy).
 //!
-//! Retries happen only before any assistant output: a stream that already produced output
-//! and then fails is returned to the caller, which settles the Turn as interrupted.
+//! A retryable failure before any assistant output reopens the request silently. One after
+//! partial output discards that attempt: the stream yields [`LlmEvent::Discarded`] and then
+//! the events of a fresh attempt of the same request. Retries before and after output share
+//! one budget. Nothing is replayed after protocol completion, since a completed stream ends.
+//!
+//! The stream never knows about tool dispatch: a consumer SHALL NOT start a tool call of the
+//! answer while the stream can still be replayed, which the Drain guarantees by dispatching
+//! only after the stream ended.
 
+use std::pin::Pin;
 use std::time::Duration;
 
-use futures::stream::{self, StreamExt};
+use futures::stream::{self, Stream, StreamExt};
 
 use crate::adapters::{Adapter, EventStream};
 use crate::error::LlmError;
 use crate::types::{LlmEvent, LlmRequest};
 
+/// A stream that may replay its request; it borrows the adapter it reopens with.
+pub type RetryingStream<'a> = Pin<Box<dyn Stream<Item = Result<LlmEvent, LlmError>> + Send + 'a>>;
+
 #[derive(Debug, Clone)]
 pub struct RetryPolicy {
     pub max_retries: u32,
@@ -44,30 +54,120 @@ impl RetryPolicy {
     }
 }
 
-/// Open a stream, retrying retryable failures that happen before any output.
-/// `on_retry(attempt, delay, error)` is called before each wait.
-pub async fn open_with_retry(
-    adapter: &dyn Adapter,
+/// One retry, announced before its wait.
+#[derive(Debug)]
+pub struct RetryNotice<'e> {
+    /// 1-based retry number.
+    pub attempt: u32,
+    pub delay: Duration,
+    pub error: &'e LlmError,
+    /// The failed attempt had streamed output, which the consumer must drop.
+    pub discarded: bool,
+}
+
+/// Open a stream, retrying retryable failures before any output, and replaying the request
+/// when a retryable failure interrupts partial output. `on_retry` is called before each wait.
+pub async fn open_with_retry<'a>(
+    adapter: &'a dyn Adapter,
     request: &LlmRequest,
     policy: &RetryPolicy,
-    mut on_retry: impl FnMut(u32, Duration, &LlmError),
-) -> Result<EventStream, LlmError> {
-    let mut attempt = 0;
-    loop {
-        let failure = match adapter.stream(request.clone()).await {
-            Ok(stream) => match first_output(stream).await {
-                Ok(stream) => return Ok(stream),
+    on_retry: impl FnMut(&RetryNotice<'_>) + Send + 'a,
+) -> Result<RetryingStream<'a>, LlmError> {
+    let mut replay = Replay {
+        adapter,
+        request: request.clone(),
+        policy: policy.clone(),
+        on_retry,
+        attempt: 0,
+    };
+    let first = replay.open().await?;
+    Ok(stream::unfold((replay, Phase::Live(first)), step).boxed())
+}
+
+struct Replay<'a, F> {
+    adapter: &'a dyn Adapter,
+    request: LlmRequest,
+    policy: RetryPolicy,
+    on_retry: F,
+    /// Retries used so far.
+    attempt: u32,
+}
+
+enum Phase {
+    Live(EventStream),
+    /// The attempt failed after output and was discarded; the next one opens after the wait.
+    Reopen(LlmError),
+    Done,
+}
+
+impl<F: FnMut(&RetryNotice<'_>)> Replay<'_, F> {
+    /// Open an attempt that reaches its first output, retrying failures before it.
+    async fn open(&mut self) -> Result<EventStream, LlmError> {
+        loop {
+            let failure = match self.adapter.stream(self.request.clone()).await {
+                Ok(stream) => match first_output(stream).await {
+                    Ok(stream) => return Ok(stream),
+                    Err(e) => e,
+                },
                 Err(e) => e,
-            },
-            Err(e) => e,
-        };
-        if !failure.kind.is_retryable() || attempt >= policy.max_retries {
+            };
+            self.wait(failure, false).await?;
+        }
+    }
+
+    fn may_retry(&self, failure: &LlmError) -> bool {
+        failure.kind.is_retryable() && self.attempt < self.policy.max_retries
+    }
+
+    /// Announce the next retry and wait for it, or give the failure back.
+    async fn wait(&mut self, failure: LlmError, discarded: bool) -> Result<(), LlmError> {
+        if !self.may_retry(&failure) {
             return Err(failure);
         }
-        attempt += 1;
-        let delay = policy.delay(attempt, &failure);
-        on_retry(attempt, delay, &failure);
+        self.attempt += 1;
+        let delay = self.policy.delay(self.attempt, &failure);
+        (self.on_retry)(&RetryNotice {
+            attempt: self.attempt,
+            delay,
+            error: &failure,
+            discarded,
+        });
         tokio::time::sleep(delay).await;
+        Ok(())
+    }
+}
+
+type Item = Result<LlmEvent, LlmError>;
+
+/// The next item of a replaying stream.
+async fn step<F: FnMut(&RetryNotice<'_>)>(
+    (mut replay, phase): (Replay<'_, F>, Phase),
+) -> Option<(Item, (Replay<'_, F>, Phase))> {
+    match phase {
+        Phase::Done => None,
+        Phase::Live(mut stream) => match stream.next().await {
+            None => None,
+            Some(Ok(event)) => Some((Ok(event), (replay, Phase::Live(stream)))),
+            // The attempt already streamed output (`first_output`), so a retry discards it.
+            Some(Err(e)) if replay.may_retry(&e) => {
+                Some((Ok(LlmEvent::Discarded), (replay, Phase::Reopen(e))))
+            }
+            Some(Err(e)) => Some((Err(e), (replay, Phase::Done))),
+        },
+        Phase::Reopen(failure) => {
+            let reopened = match replay.wait(failure, true).await {
+                Ok(()) => replay.open().await,
+                Err(e) => Err(e),
+            };
+            match reopened {
+                Ok(mut stream) => {
+                    // `first_output` returns a stream with at least one item unless it ended.
+                    let first = stream.next().await?;
+                    Some((first, (replay, Phase::Live(stream))))
+                }
+                Err(e) => Some((Err(e), (replay, Phase::Done))),
+            }
+        }
     }
 }
 

crates/cyber-llm/src/turn.rs

@@ -1,8 +1,7 @@
 //! Collect a stream into the result of one Turn.
 
-use futures::StreamExt;
+use futures::{Stream, StreamExt};
 
-use crate::adapters::EventStream;
 use crate::error::LlmError;
 use crate::types::{FinishReason, LlmEvent, ToolCall, Usage};
 
@@ -15,9 +14,10 @@ pub struct TurnOutput {
     pub finish: Option<FinishReason>,
 }
 
-/// Drain a stream, calling `on_event` for each event (for live rendering).
+/// Drain a stream, calling `on_event` for each event (for live rendering). A discarded
+/// attempt's output is dropped; its usage was spent and stays counted.
 pub async fn collect(
-    mut stream: EventStream,
+    mut stream: impl Stream<Item = Result<LlmEvent, LlmError>> + Unpin,
     mut on_event: impl FnMut(&LlmEvent),
 ) -> Result<TurnOutput, LlmError> {
     let mut out = TurnOutput::default();
@@ -36,6 +36,12 @@ pub async fn collect(
             LlmEvent::Usage(usage) => out.usage.add(&usage),
             LlmEvent::Finish { reason } => out.finish = Some(reason),
             LlmEvent::ToolCallDelta { .. } => {}
+            LlmEvent::Discarded => {
+                out = TurnOutput {
+                    usage: out.usage,
+                    ..TurnOutput::default()
+                };
+            }
         }
     }
     Ok(out)