Introduce ChatStreamInterruptedException: handle broken streams with retries, teammate failure reporting, and related updates to tests and documentation.

Author: NikolayPianikov

docs/34-asides-and-team-messages.md

@@ -168,6 +168,20 @@ model's account of the stall to the main branch as a `blocker` from that branch,
 lead (see the stall rule in [Hidden model instructions](22-hidden-model-instructions-and-run-completion.md)).
 A teammate's branch is recognised by its first message: a team message the main branch sent.
 
+### Failures
+
+A team runs with nobody watching each branch, so a failure must neither stop the team silently
+nor wait for someone to press Retry:
+
+- **A broken stream is retried by the agent.** When a model step's stream breaks off after it has
+  started — falls silent past the idle timeout, ends in the middle of a tool call, or loses its
+  connection (`ChatStreamInterruptedException`, `IOException`) — the agent asks for the step again,
+  up to three times in a row, after 2, 4 and 6 seconds. Nothing of the broken step was persisted
+  or executed, so this is the same as Retry. This applies to every chat, not only teams.
+- **A teammate's failed turn is reported to the lead** as a `blocker` beginning with "Failed:";
+  `team-coordinate` resumes that branch once with `app_runs` Resume and tells the user if it fails
+  again. A turn stopped for lack of progress is reported the same way.
+
 ### Known limits
 
 - Branches share the project's working directory. The charter assigns owned paths to each role;

src/AI.Server/Application/Chat/ChatStreamInterruptedException.cs

@@ -0,0 +1,8 @@
+namespace AI.Application.Chat;
+
+/// <summary>
+/// A response stream that started and then broke off before its finish: it fell silent, or ended
+/// in the middle of a tool call. Nothing the model said in it has been acted on, so the step can be
+/// asked for again; see <see cref="Tools.ChatAgent"/>.
+/// </summary>
+public sealed class ChatStreamInterruptedException(string message) : InvalidOperationException(message);

src/AI.Server/Application/Runs/ChatRunDispatcher.cs

@@ -386,11 +386,12 @@ private static List<QueuedRunMessage> TakenAsides(ChatDetails chat, Guid? head,
     }
 
     /// <summary>
-    /// A teammate whose turn stopped for lack of progress tells the lead, who would otherwise learn
-    /// of it only by looking. A teammate's branch is one that a team message of the main branch
+    /// A teammate whose turn stopped — for lack of progress, or failed — tells the lead, who would
+    /// otherwise learn of it only by looking, and a team waits on a lead that does not look. A
+    /// teammate's branch is one that a team message of the main branch
     /// started (docs/34-asides-and-team-messages.md); any other branch keeps the report to itself.
     /// </summary>
-    private async Task ReportStalledTeammateAsync(Runtime runtime, string report, CancellationToken token)
+    private async Task ReportToLeadAsync(Runtime runtime, string report, CancellationToken token)
     {
         var (projectId, chatId, branchId) = (runtime.State.ProjectId, runtime.State.ChatId, runtime.State.BranchId);
         if (branchId == chatId) return;
@@ -405,7 +406,7 @@ private async Task ReportStalledTeammateAsync(Runtime runtime, string report, Ca
         try
         {
             await SubmitCoreAsync(projectId, chatId, new SubmitChatMessageRequest(id, id,
-                    "Stopped: several steps in a row produced nothing new.\n\n" + report, ChatSubmitMode.Send, chatId),
+                    report, ChatSubmitMode.Send, chatId),
                 true, token, new ChatMessageSender(chatId, branchId, "blocker"));
         }
         catch (Exception error) when (error is InvalidOperationException or ArgumentException or IOException
@@ -695,7 +696,8 @@ await contextBuilder.BuildAsync(chat, ResumeHead(chat, runtime.State.BranchId, q
                     draftToolCall: (name, ct) => ReportDraftToolCallAsync(runtime, name, ct),
                     overlayPromptsAllowed: queued.Interactive,
                     asides: ct => TakeAsidesIntoTurnAsync(runtime, ct),
-                    stalledReport: (report, ct) => ReportStalledTeammateAsync(runtime, report, ct));
+                    stalledReport: (report, ct) => ReportToLeadAsync(runtime,
+                        "Stopped: several steps in a row produced nothing new.\n\n" + report, ct));
                 var suggestTitle = false;
                 Guid? answeredHead = null;
                 using (await synchronization.EnterAsync(runtime.State.ChatId, token))
@@ -771,6 +773,15 @@ await contextBuilder.BuildAsync(chat, ResumeHead(chat, runtime.State.BranchId, q
                 try { await chatMutations.MarkActivityCoreAsync(runtime.State.ProjectId, runtime.State.ChatId, CancellationToken.None); }
                 catch (Exception markError) when (markError is IOException or UnauthorizedAccessException
                     or InvalidOperationException or Domain.Common.DomainException) { }
+                // Started outside this lease: telling the lead submits into the main branch, which
+                // takes the same chat lock.
+                if (!_shutdown.IsCancellationRequested)
+                {
+                    var failure = Describe(error);
+                    _ = Task.Run(() => ReportToLeadAsync(runtime,
+                        $"Failed: {failure}\nThe branch is stopped until someone retries or discards its turn.",
+                        _shutdown.Token));
+                }
             }
             if (error is OperationCanceledException && runtime.ResumeRequested && !_shutdown.IsCancellationRequested) runtime.State.Resume();
             try { await SaveAsync(runtime, await chats.GetAsync(runtime.State.ProjectId, runtime.State.ChatId, CancellationToken.None), CancellationToken.None); }

src/AI.Server/Application/Tools/ChatAgent.cs

@@ -159,6 +159,7 @@ await persist(ToolMessage(batch[remaining].Id,
         }
 
         var empty = 0;
+        var interrupted = 0;
         var truncated = 0;
         var continuedAnswer = new StringBuilder();
         var stalledSteps = 0;
@@ -377,17 +378,30 @@ async Task<string> StalledReportAsync(IReadOnlyList<ChatCompletionMessage> messa
             }
             await Usage(plan);
             if (!plan.Fits) throw new ContextWindowExceededException(plan);
-            await foreach (var chunk in completion.StreamAsync(
-                               request with { ContextMessages = plan.Messages, Tools = available }, token))
+            try
             {
-                chunkCount++;
-                if (chunk.ToolCalls is { } received) calls.AddRange(received);
-                if (chunk.FinishReason is { Length: > 0 } reason) finish = reason;
-                if (chunk is { ToolCallsStarted: true, ToolCallName: { Length: > 0 } starting } && draftToolCall is not null)
-                    await draftToolCall(starting, token);
-                if (chunk.Content.Length == 0) continue;
-                content.Append(chunk.Content);
-                await Draft(chunk.Content);
+                await foreach (var chunk in completion.StreamAsync(
+                                   request with { ContextMessages = plan.Messages, Tools = available }, token))
+                {
+                    chunkCount++;
+                    if (chunk.ToolCalls is { } received) calls.AddRange(received);
+                    if (chunk.FinishReason is { Length: > 0 } reason) finish = reason;
+                    if (chunk is { ToolCallsStarted: true, ToolCallName: { Length: > 0 } starting } && draftToolCall is not null)
+                        await draftToolCall(starting, token);
+                    if (chunk.Content.Length == 0) continue;
+                    content.Append(chunk.Content);
+                    await Draft(chunk.Content);
+                }
+                interrupted = 0;
+            }
+            // A stream that broke off part-way has left nothing behind: the step is persisted only
+            // once it is complete, so none of its calls ran. Asking for the step again is what the
+            // person would do with Retry, and a team branch has nobody watching to press it.
+            catch (Exception error) when (StreamBroke(error, token) && ++interrupted <= MaxInterruptedSteps)
+            {
+                await Draft(null);
+                await Task.Delay(TimeSpan.FromSeconds(2 * interrupted), token);
+                continue;
             }
             if (calls.Count == 0 && content.Length == 0)
             {
@@ -594,6 +608,20 @@ await AbandonAsync(calls, index,
     /// </summary>
     private const int MaxEmptyTurns = 2;
 
+    /// <summary>
+    /// How many times in a row a step whose stream broke off is asked for again before the run
+    /// fails. A provider that keeps dropping the same request is reported rather than retried forever.
+    /// </summary>
+    private const int MaxInterruptedSteps = 3;
+
+    /// <summary>
+    /// Whether a step failed because its stream broke off after it had started — fell silent, ended
+    /// mid tool call, or lost its connection — rather than because the endpoint refused it.
+    /// </summary>
+    private static bool StreamBroke(Exception error, CancellationToken token) =>
+        !token.IsCancellationRequested && error is ChatStreamInterruptedException or IOException
+            or HttpRequestException { StatusCode: null };
+
     /// <summary>
     /// Summarizes what fills a context that is filling up: the earlier turns, kept as a history
     /// checkpoint, or — when they are already summarized or too small — the completed steps of a

src/AI.Server/Infrastructure/Chat/ChatCompletionSseParser.cs

@@ -67,7 +67,7 @@ public async IAsyncEnumerable<ChatCompletionChunk> ParseAsync(
 
             if (line is null)
             {
-                if (calls.Count > 0) throw new InvalidOperationException("Tool call stream ended before completion.");
+                if (calls.Count > 0) throw new ChatStreamInterruptedException("Tool call stream ended before completion.");
                 break;
             }
 
@@ -83,7 +83,7 @@ public async IAsyncEnumerable<ChatCompletionChunk> ParseAsync(
             var data = line[5..].TrimStart();
             if (data == "[DONE]")
             {
-                if (calls.Count > 0) throw new InvalidOperationException("Tool call stream has no completion marker.");
+                if (calls.Count > 0) throw new ChatStreamInterruptedException("Tool call stream has no completion marker.");
                 break;
             }
 
@@ -182,8 +182,8 @@ public async IAsyncEnumerable<ChatCompletionChunk> ParseAsync(
     /// whatever text had arrived — often only "Now I'll create the files." — the run's final
     /// answer, and the run reported success with the work not done.
     /// </summary>
-    private InvalidOperationException Stalled(int pendingToolCalls) => pendingToolCalls > 0
-        ? new InvalidOperationException("Tool call stream timed out before completion.")
-        : new InvalidOperationException(
+    private ChatStreamInterruptedException Stalled(int pendingToolCalls) => pendingToolCalls > 0
+        ? new ChatStreamInterruptedException("Tool call stream timed out before completion.")
+        : new ChatStreamInterruptedException(
             $"The model stopped sending its answer for {policy.StreamIdleTimeout.TotalSeconds:0} seconds before finishing it.");
 }

src/AI.Server/Skills/BuiltIn/team-coordinate/SKILL.md

@@ -23,6 +23,9 @@ owned paths, deliverables and the protocol (docs/34-asides-and-team-messages.md)
      sender's branch, intent `Answer`, mode Send: it wakes the teammate.
    - `blocker`: unblock it — answer, adjust the scope, or move the work to another teammate. A
      change to scope, owned paths or a contract is a decision (below).
+     A `blocker` that starts with "Failed:" is the application reporting that the teammate's turn
+     failed (an endpoint or stream error, not the teammate's judgement): resume that branch once
+     with `app_runs` Resume; if it fails again, tell the user which branch is stopped and why.
    - `decision` proposed by a teammate: accept or reject it as a decision.
    - `status`: nothing to send. Note it for the status table.
    - `done`: check the deliverable against its definition of done — read the changes, run the

tests/AI.Server.Tests/Infrastructure/Storage/ChatExecutionTests.cs

@@ -441,6 +441,51 @@ await fixture.Dispatcher.SubmitFromRunAsync(fixture.ProjectId, fixture.ChatId,
         chat!.Messages.Single(message => message.Id == question).Delivery.ShouldBe(MessageDelivery.Turn);
     }
 
+    [Fact]
+    public async Task AStepWhoseStreamBrokeOffShouldBeAskedForAgainWithoutFailingTheRun()
+    {
+        await using var fixture = await Fixture.CreateAsync();
+        await fixture.SetChatAutomationAsync(new ChatAutomationSettings(AutoTitle: false, SuggestReplies: false));
+        fixture.Completion.NextPrelude = "Reading the files";
+        await fixture.SubmitAsync(new SubmitChatMessageRequest(Guid.NewGuid(), Guid.NewGuid(), "Summarize the repository"));
+
+        (await fixture.NextCallAsync()).Answer.SetException(
+            new AI.Application.Chat.ChatStreamInterruptedException("Tool call stream timed out before completion."));
+        (await fixture.NextCallAsync()).Answer.SetResult("The summary.");
+
+        await fixture.WaitAsync(run => run.Status == ChatRunStatus.Completed);
+        var chat = await fixture.Chats.GetAsync(fixture.ProjectId, fixture.ChatId, CancellationToken.None);
+        chat!.Messages[^1].Content.ShouldBe("The summary.");
+        // The broken attempt left nothing behind, its half-streamed text included.
+        chat.Messages.ShouldNotContain(message => message.Content.Contains("Reading the files"));
+    }
+
+    [Fact]
+    public async Task ATeammatesFailedTurnShouldBeReportedToTheLead()
+    {
+        await using var fixture = await Fixture.CreateAsync();
+        await fixture.SetChatAutomationAsync(new ChatAutomationSettings(AutoTitle: false, SuggestReplies: false));
+        await fixture.SubmitAsync(new SubmitChatMessageRequest(Guid.NewGuid(), Guid.NewGuid(), "Team charter"));
+        (await fixture.NextCallAsync()).Answer.SetResult("Forking the team.");
+        await fixture.WaitAsync(run => run.Status == ChatRunStatus.Completed);
+        var head = (await fixture.Chats.GetAsync(fixture.ProjectId, fixture.ChatId, CancellationToken.None))!.Messages[^1].Id;
+        var brief = Guid.NewGuid();
+        await fixture.Dispatcher.SubmitFromRunAsync(fixture.ProjectId, fixture.ChatId,
+            new SubmitChatMessageRequest(brief, brief, "You are backend.", ChatSubmitMode.Fork, fixture.ChatId,
+                MessageParentMode.Explicit, head),
+            new AI.Domain.Chats.ChatMessageSender(fixture.ChatId, fixture.ChatId, "decision"), CancellationToken.None);
+
+        (await fixture.NextCallAsync()).Answer.SetException(new InvalidOperationException("The endpoint refused the request."));
+
+        // The lead is woken by the report rather than waiting for a teammate that will never answer.
+        var lead = await fixture.NextCallAsync();
+        var report = lead.Request.ContextMessages!.Last(item => item.Role == "user" && !item.IsContextSummary);
+        report.ForModel.ShouldContain("(blocker)");
+        report.ForModel.ShouldContain("Failed: The endpoint refused the request.");
+        lead.Answer.SetResult("Resuming the backend branch.");
+        await fixture.WaitAsync(run => run.BranchId == fixture.ChatId && run.Status == ChatRunStatus.Completed);
+    }
+
     [Fact]
     public async Task MessageFromAnotherBranchsRunShouldCarryItsSender()
     {