diff --git a/.changeset/tool-result-per-call-settle.md b/.changeset/tool-result-per-call-settle.md
new file mode 100644
index 00000000..78e04ddf
--- /dev/null
+++ b/.changeset/tool-result-per-call-settle.md
@@ -0,0 +1,5 @@
+---
+'@openrouter/agent': patch
+---
+
+Broadcast each tool call's `tool.result` / `tool.call_output` (and the legacy `tool_result`) the moment that call settles, instead of holding every broadcast until the whole round's `Promise.allSettled` resolves. With serially executing calls (e.g. `maxConcurrency: 1`), stream consumers now see each call complete live rather than all at once when the round ends. Model-facing `function_call_output` assembly is unchanged: outputs are still collected in call order, so follow-up request input and prompt-cache behavior are identical. (DEV-1067)
diff --git a/packages/agent/src/lib/model-result.ts b/packages/agent/src/lib/model-result.ts
index 6c1ce1f7..ce70b13d 100644
--- a/packages/agent/src/lib/model-result.ts
+++ b/packages/agent/src/lib/model-result.ts
@@ -219,6 +219,66 @@ interface RunBinding {
   task: () => ToolTask | null;
 }
 
+/**
+ * Tagged outcome of one tool call in a round, produced by
+ * `executeSingleToolCall` and consumed by the settle-time finalizers in
+ * `executeToolRound`. `null` means the call is manual (client-executed) and
+ * yields no output this round.
+ */
+type SingleToolCallOutcome<TTools extends readonly Tool[]> =
+  | null
+  | {
+      type: 'parse_error';
+      output: models.FunctionCallOutputItem;
+    }
+  | {
+      type: 'hook_blocked';
+      output: models.FunctionCallOutputItem;
+    }
+  | {
+      type: 'paused';
+      toolCall: ParsedToolCall<Tool>;
+    }
+  | {
+      type: 'async';
+      toolCall: ParsedToolCall<Tool>;
+      tool: Tool;
+      invocation: AsyncToolInvocation;
+      controller: AbortController;
+      timeoutMs: number | undefined;
+      runBinding: RunBinding;
+      preliminaryResultsForCall: InferToolEventsUnion<TTools>[];
+    }
+  | {
+      type: 'execution';
+      toolCall: ParsedToolCall<Tool>;
+      tool: Tool;
+      result: {
+        result: unknown;
+        error?: Error;
+      };
+      preliminaryResultsForCall: InferToolEventsUnion<TTools>[];
+    };
+
+/**
+ * A round call after settle-time finalization: broadcasts already emitted,
+ * model-facing output (if any) computed. `executeToolRound`'s ordered
+ * assembly loop only sorts these into the round's result arrays.
+ */
+type FinalizedRoundCall =
+  | {
+      kind: 'skipped';
+    }
+  | {
+      kind: 'paused';
+      toolCall: ParsedToolCall<Tool>;
+    }
+  | {
+      kind: 'output';
+      output: models.FunctionCallOutputItem;
+      deferredTask?: PendingAsyncTool;
+    };
+
 /**
  * Extract the call identity echoed on a server-tool output item, for
  * doom-loop fingerprinting. Server tools never pass through the client
@@ -3631,41 +3691,7 @@ export class ModelResult<
   private async executeSingleToolCall(
     toolCall: ParsedToolCall<Tool>,
     turnContext: TurnContext,
-  ): Promise<
-    | null
-    | {
-        type: 'parse_error';
-        output: models.FunctionCallOutputItem;
-      }
-    | {
-        type: 'hook_blocked';
-        output: models.FunctionCallOutputItem;
-      }
-    | {
-        type: 'paused';
-        toolCall: ParsedToolCall<Tool>;
-      }
-    | {
-        type: 'async';
-        toolCall: ParsedToolCall<Tool>;
-        tool: Tool;
-        invocation: AsyncToolInvocation;
-        controller: AbortController;
-        timeoutMs: number | undefined;
-        runBinding: RunBinding;
-        preliminaryResultsForCall: InferToolEventsUnion<TTools>[];
-      }
-    | {
-        type: 'execution';
-        toolCall: ParsedToolCall<Tool>;
-        tool: Tool;
-        result: {
-          result: unknown;
-          error?: Error;
-        };
-        preliminaryResultsForCall: InferToolEventsUnion<TTools>[];
-      }
-  > {
+  ): Promise<SingleToolCallOutcome<TTools>> {
     // Universal task-tool dispatch: ONE static tool ("task") handles every
     // check/steer/result/cancel interaction with running tasks, addressed
     // by taskId — the wire surface stays constant regardless of how many
@@ -3944,153 +3970,176 @@ export class ModelResult<
     pausedCalls: ParsedToolCall<Tool>[];
     deferredTasks: PendingAsyncTool[];
   }> {
-    const toolCallPromises = toolCalls.map((toolCall) =>
-      this.executeSingleToolCall(toolCall, turnContext),
+    // Each call is finalized the moment ITS promise settles (DEV-1067): the
+    // tool.result / tool_result broadcasts and the tool.call_output event
+    // fire per call, so stream consumers see an early call complete while
+    // later calls in the round are still running. This also starts async
+    // invocations (grace-window race, registry tracking) at settle time
+    // rather than after the whole round, so N background calls never
+    // serialize into (N-1)×graceMs of stagger.
+    const finalizedPromises = toolCalls.map((toolCall) =>
+      this.executeSingleToolCall(toolCall, turnContext).then(
+        (value) => this.finalizeSettledRoundCall(value),
+        (reason: unknown) => this.finalizeRejectedRoundCall(toolCall, reason),
+      ),
     );
+    const finalizedCalls = await Promise.all(finalizedPromises);
 
-    const settledResults = await Promise.allSettled(toolCallPromises);
+    // Ordered assembly: broadcasts already happened per call above, but the
+    // MODEL-facing outputs are collected in call order so the follow-up
+    // request's input is deterministic (prompt-cache stability).
     const toolResults: models.FunctionCallOutputItem[] = [];
     const pausedCalls: ParsedToolCall<Tool>[] = [];
     const deferredTasks: PendingAsyncTool[] = [];
-
-    // Start ALL async invocations before consuming any outcome: the work
-    // (and its grace window) begins in handleAsyncInvocation, so awaiting
-    // it inside the ordered loop below would serialize N background calls
-    // into (N-1)×graceMs of stagger. Kicked off here in parallel; the
-    // ordered loop awaits the per-call promise, so OUTPUT order stays call
-    // order (prompt-cache stability).
-    const asyncOutcomes = new Map<
-      number,
-      Promise<{
-        output: models.FunctionCallOutputItem;
-        deferredTask?: PendingAsyncTool;
-      }>
-    >();
-    for (let i = 0; i < settledResults.length; i++) {
-      const settled = settledResults[i];
-      if (settled?.status === 'fulfilled' && settled.value?.type === 'async') {
-        asyncOutcomes.set(i, this.handleAsyncInvocation(settled.value));
+    for (const finalized of finalizedCalls) {
+      switch (finalized.kind) {
+        case 'skipped':
+          break;
+        case 'paused':
+          // HITL tool returned null — record the pause so the caller can
+          // break out of the outer loop before attempting a follow-up
+          // request with an incomplete set of outputs. The call will be
+          // surfaced via state (pendingToolCalls + status='awaiting_hitl')
+          // for manual resume.
+          pausedCalls.push(finalized.toolCall);
+          break;
+        case 'output':
+          toolResults.push(finalized.output);
+          if (finalized.deferredTask) {
+            deferredTasks.push(finalized.deferredTask);
+          }
+          break;
+        default:
+          finalized satisfies never;
       }
     }
 
-    for (let i = 0; i < settledResults.length; i++) {
-      const settled = settledResults[i];
-      const originalToolCall = toolCalls[i];
-      if (!settled || !originalToolCall) {
-        continue;
-      }
-
-      if (settled.status === 'rejected') {
-        const errorMessage =
-          settled.reason instanceof Error ? settled.reason.message : String(settled.reason);
+    return {
+      toolResults,
+      pausedCalls,
+      deferredTasks,
+    };
+  }
 
-        // `runToolWithHooks` is the single point of emission for PostToolUseFailure.
-        this.broadcastToolResult(
-          originalToolCall.id,
-          String(originalToolCall.name),
-          this.toolSourceByName(String(originalToolCall.name)),
-          {
-            error: errorMessage,
-          } as InferToolOutputsUnion<TTools>,
-        );
+  /**
+   * Finalize one settled tool call of a round, at the moment it settles:
+   * emit its tool.result / tool.call_output events and compute the
+   * model-facing output. Runs concurrently across the round's calls;
+   * `executeToolRound` re-orders the returned outputs into call order.
+   */
+  private async finalizeSettledRoundCall(
+    value: SingleToolCallOutcome<TTools>,
+  ): Promise<FinalizedRoundCall> {
+    // Manual (client-executed) tool — no output this round.
+    if (!value) {
+      return {
+        kind: 'skipped',
+      };
+    }
 
-        const rejectedOutput: models.FunctionCallOutputItem = {
-          type: 'function_call_output' as const,
-          id: `output_${originalToolCall.id}`,
-          callId: originalToolCall.id,
-          output: JSON.stringify({
-            error: errorMessage,
-          }),
-        };
-        toolResults.push(rejectedOutput);
-        this.turnBroadcaster?.push({
-          type: 'tool.call_output' as const,
-          output: rejectedOutput,
-          timestamp: Date.now(),
-        } satisfies ToolCallOutputEvent);
-        continue;
-      }
+    if (value.type === 'parse_error' || value.type === 'hook_blocked') {
+      this.pushToolCallOutputEvent(value.output);
+      return {
+        kind: 'output',
+        output: value.output,
+      };
+    }
 
-      const value = settled.value;
-      if (!value) {
-        continue;
-      }
+    if (value.type === 'paused') {
+      return {
+        kind: 'paused',
+        toolCall: value.toolCall,
+      };
+    }
 
-      if (value.type === 'parse_error' || value.type === 'hook_blocked') {
-        toolResults.push(value.output);
-        this.turnBroadcaster?.push({
-          type: 'tool.call_output' as const,
-          output: value.output,
-          timestamp: Date.now(),
-        } satisfies ToolCallOutputEvent);
-        continue;
-      }
+    if (value.type === 'async') {
+      // Background / deferred: the call escapes the round (grace-window
+      // race, registry tracking, placeholder synthesis).
+      const asyncOutcome = await this.handleAsyncInvocation(value);
+      this.pushToolCallOutputEvent(asyncOutcome.output);
+      return {
+        kind: 'output',
+        output: asyncOutcome.output,
+        ...(asyncOutcome.deferredTask && {
+          deferredTask: asyncOutcome.deferredTask,
+        }),
+      };
+    }
 
-      if (value.type === 'paused') {
-        // HITL tool returned null — record the pause so the caller can break
-        // out of the outer loop before attempting a follow-up request with an
-        // incomplete set of outputs. The call will be surfaced via state
-        // (pendingToolCalls + status='awaiting_hitl') for manual resume.
-        pausedCalls.push(value.toolCall);
-        continue;
-      }
+    const toolResult = (
+      value.result.error
+        ? {
+            error: value.result.error.message,
+          }
+        : value.result.result
+    ) as InferToolOutputsUnion<TTools>;
+    this.broadcastToolResult(
+      value.toolCall.id,
+      String(value.toolCall.name),
+      isMcpTool(value.tool) ? 'mcp' : 'client',
+      toolResult,
+      value.preliminaryResultsForCall.length > 0 ? value.preliminaryResultsForCall : undefined,
+    );
 
-      if (value.type === 'async') {
-        // Background / deferred: the call escapes the round. Its handling
-        // (grace-window race, registry tracking, placeholder synthesis)
-        // was started above in parallel with the round's other async
-        // calls; awaiting here keeps outputs in call order.
-        const asyncOutcome = await (asyncOutcomes.get(i) ?? this.handleAsyncInvocation(value));
-        toolResults.push(asyncOutcome.output);
-        this.turnBroadcaster?.push({
-          type: 'tool.call_output' as const,
-          output: asyncOutcome.output,
-          timestamp: Date.now(),
-        } satisfies ToolCallOutputEvent);
-        if (asyncOutcome.deferredTask) {
-          deferredTasks.push(asyncOutcome.deferredTask);
-        }
-        continue;
-      }
+    const outputForModel = await this.computeToolOutputForModel(value);
 
-      const toolResult = (
-        value.result.error
-          ? {
-              error: value.result.error.message,
-            }
-          : value.result.result
-      ) as InferToolOutputsUnion<TTools>;
-      this.broadcastToolResult(
-        value.toolCall.id,
-        String(value.toolCall.name),
-        isMcpTool(value.tool) ? 'mcp' : 'client',
-        toolResult,
-        value.preliminaryResultsForCall.length > 0 ? value.preliminaryResultsForCall : undefined,
-      );
+    const executedOutput: models.FunctionCallOutputItem = {
+      type: 'function_call_output' as const,
+      id: `output_${value.toolCall.id}`,
+      callId: value.toolCall.id,
+      output: outputForModel,
+    };
+    this.pushToolCallOutputEvent(executedOutput);
+    return {
+      kind: 'output',
+      output: executedOutput,
+    };
+  }
 
-      const outputForModel = await this.computeToolOutputForModel(value);
+  /**
+   * Finalize one REJECTED tool call of a round: synthesize the error output
+   * and emit its events at the moment the rejection settles.
+   */
+  private finalizeRejectedRoundCall(
+    toolCall: ParsedToolCall<Tool>,
+    reason: unknown,
+  ): FinalizedRoundCall {
+    const errorMessage = reason instanceof Error ? reason.message : String(reason);
 
-      const executedOutput: models.FunctionCallOutputItem = {
-        type: 'function_call_output' as const,
-        id: `output_${value.toolCall.id}`,
-        callId: value.toolCall.id,
-        output: outputForModel,
-      };
-      toolResults.push(executedOutput);
-      this.turnBroadcaster?.push({
-        type: 'tool.call_output' as const,
-        output: executedOutput,
-        timestamp: Date.now(),
-      } satisfies ToolCallOutputEvent);
-    }
+    // `runToolWithHooks` is the single point of emission for PostToolUseFailure.
+    this.broadcastToolResult(
+      toolCall.id,
+      String(toolCall.name),
+      this.toolSourceByName(String(toolCall.name)),
+      {
+        error: errorMessage,
+      } as InferToolOutputsUnion<TTools>,
+    );
 
+    const rejectedOutput: models.FunctionCallOutputItem = {
+      type: 'function_call_output' as const,
+      id: `output_${toolCall.id}`,
+      callId: toolCall.id,
+      output: JSON.stringify({
+        error: errorMessage,
+      }),
+    };
+    this.pushToolCallOutputEvent(rejectedOutput);
     return {
-      toolResults,
-      pausedCalls,
-      deferredTasks,
+      kind: 'output',
+      output: rejectedOutput,
     };
   }
 
+  /** Push a `tool.call_output` event to the unified turn broadcaster. */
+  private pushToolCallOutputEvent(output: models.FunctionCallOutputItem): void {
+    this.turnBroadcaster?.push({
+      type: 'tool.call_output' as const,
+      output,
+      timestamp: Date.now(),
+    } satisfies ToolCallOutputEvent);
+  }
+
   /**
    * Handle one async invocation from a round.
    *
diff --git a/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts b/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts
new file mode 100644
index 00000000..f828418b
--- /dev/null
+++ b/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts
@@ -0,0 +1,312 @@
+import type { OpenRouterCore } from '@openrouter/sdk/core';
+import type * as models from '@openrouter/sdk/models';
+import { describe, expect, it, vi } from 'vitest';
+import { z } from 'zod/v4';
+import type { GetResponseOptions } from '../../src/lib/model-result.js';
+import { ModelResult } from '../../src/lib/model-result.js';
+import { tool } from '../../src/lib/tool.js';
+import type { Tool } from '../../src/lib/tool-types.js';
+import { isToolResultEvent } from '../../src/lib/tool-types.js';
+
+type Internal = {
+  currentState: {
+    id: string;
+    messages: models.BaseInputsUnion[];
+    status: 'in_progress';
+    createdAt: number;
+    updatedAt: number;
+  } | null;
+  initPromise: Promise<void> | null;
+  getInitialResponse: () => Promise<models.OpenResponsesResult>;
+  makeFollowupRequest: (
+    currentResponse: models.OpenResponsesResult,
+    toolResults: models.FunctionCallOutputItem[],
+    turnNumber: number,
+  ) => Promise<models.OpenResponsesResult>;
+  shouldStopExecution: () => Promise<boolean>;
+  executeToolsIfNeeded: () => Promise<void>;
+  ensureTurnBroadcaster: () => {
+    createConsumer: () => AsyncIterableIterator<unknown>;
+    push: (event: unknown) => void;
+    complete: () => void;
+  };
+};
+
+function makeResponseWithToolCalls(
+  calls: Array<{
+    id: string;
+    name: string;
+    arguments: string;
+  }>,
+): models.OpenResponsesResult {
+  return {
+    id: 'resp_test',
+    object: 'response',
+    createdAt: 0,
+    model: 'test-model',
+    status: 'completed',
+    output: calls.map((c) => ({
+      type: 'function_call' as const,
+      id: c.id,
+      callId: c.id,
+      name: c.name,
+      arguments: c.arguments,
+      status: 'completed' as const,
+    })),
+    usage: {
+      inputTokens: 1,
+      outputTokens: 1,
+      totalTokens: 2,
+    },
+  } as unknown as models.OpenResponsesResult;
+}
+
+function makeFinalResponse(): models.OpenResponsesResult {
+  return {
+    id: 'resp_final',
+    object: 'response',
+    createdAt: 0,
+    model: 'test-model',
+    status: 'completed',
+    output: [
+      {
+        type: 'message',
+        id: 'msg_1',
+        role: 'assistant',
+        status: 'completed',
+        content: [
+          {
+            type: 'output_text',
+            text: 'done',
+          },
+        ],
+      },
+    ],
+    usage: {
+      inputTokens: 1,
+      outputTokens: 1,
+      totalTokens: 2,
+    },
+  } as unknown as models.OpenResponsesResult;
+}
+
+function buildModelResult(tools: readonly Tool[]): Internal {
+  const config: GetResponseOptions<readonly Tool[]> = {
+    request: {
+      model: 'test-model',
+      input: 'hello',
+    },
+    client: {} as OpenRouterCore,
+    tools,
+  };
+  const internal = new ModelResult<readonly Tool[]>(config) as unknown as Internal;
+  internal.currentState = {
+    id: 'conv',
+    messages: [],
+    status: 'in_progress',
+    createdAt: 0,
+    updatedAt: 0,
+  };
+  internal.initPromise = Promise.resolve();
+  internal.shouldStopExecution = async () => false;
+  return internal;
+}
+
+async function drainInto(
+  consumer: AsyncIterableIterator<unknown>,
+  events: unknown[],
+): Promise<void> {
+  for await (const event of consumer) {
+    events.push(event);
+  }
+}
+
+function isToolCallOutputEvent(event: unknown): event is {
+  type: 'tool.call_output';
+  output: models.FunctionCallOutputItem;
+} {
+  return (
+    typeof event === 'object' &&
+    event !== null &&
+    'type' in event &&
+    event.type === 'tool.call_output'
+  );
+}
+
+/**
+ * DEV-1067 regression: when a round has multiple tool calls, each call's
+ * tool.result / tool.call_output must be broadcast the moment THAT call
+ * settles — not held until the whole round's Promise.allSettled resolves —
+ * while the model-facing outputs stay in call order.
+ */
+describe('per-call broadcast on settlement (DEV-1067)', () => {
+  it('broadcasts a fast call before a slow sibling in the same round finishes', async () => {
+    let releaseSlow: () => void = () => undefined;
+    const slowGate = new Promise<void>((resolve) => {
+      releaseSlow = resolve;
+    });
+
+    const slow = tool({
+      name: 'slow',
+      inputSchema: z.object({}),
+      outputSchema: z.object({
+        done: z.boolean(),
+      }),
+      execute: async () => {
+        await slowGate;
+        return {
+          done: true,
+        };
+      },
+    });
+    const fast = tool({
+      name: 'fast',
+      inputSchema: z.object({}),
+      outputSchema: z.object({
+        done: z.boolean(),
+      }),
+      execute: async () => ({
+        done: true,
+      }),
+    });
+
+    const internal = buildModelResult([
+      slow,
+      fast,
+    ]);
+    // Call order: slow FIRST, fast second — so the fast call settling early
+    // is only observable if broadcasts are per-call, and the ordered
+    // model-facing assembly is only correct if it re-sorts to call order.
+    internal.getInitialResponse = async () =>
+      makeResponseWithToolCalls([
+        {
+          id: 'call_slow',
+          name: 'slow',
+          arguments: '{}',
+        },
+        {
+          id: 'call_fast',
+          name: 'fast',
+          arguments: '{}',
+        },
+      ]);
+
+    let followupToolResults: models.FunctionCallOutputItem[] | null = null;
+    internal.makeFollowupRequest = async (_currentResponse, toolResults) => {
+      followupToolResults = toolResults;
+      return makeFinalResponse();
+    };
+
+    const broadcaster = internal.ensureTurnBroadcaster();
+    const events: unknown[] = [];
+    const consumer = broadcaster.createConsumer();
+    const drainPromise = drainInto(consumer, events);
+
+    const executionPromise = internal.executeToolsIfNeeded();
+
+    // The fast call's completion must appear on the wire while the slow
+    // call is still running (its gate is unreleased).
+    await vi.waitFor(() => {
+      const toolResults = events.filter(isToolResultEvent);
+      expect(toolResults.some((e) => e.toolCallId === 'call_fast')).toBe(true);
+    });
+    expect(events.filter(isToolResultEvent).some((e) => e.toolCallId === 'call_slow')).toBe(false);
+    const fastOutputs = events.filter(isToolCallOutputEvent);
+    expect(fastOutputs.some((e) => e.output.callId === 'call_fast')).toBe(true);
+    expect(fastOutputs.some((e) => e.output.callId === 'call_slow')).toBe(false);
+
+    releaseSlow();
+    await executionPromise;
+    broadcaster.complete();
+    await drainPromise;
+
+    // Both calls completed on the wire, fast before slow (settlement order).
+    const toolResultIds = events.filter(isToolResultEvent).map((e) => e.toolCallId);
+    expect(toolResultIds).toEqual([
+      'call_fast',
+      'call_slow',
+    ]);
+
+    // Model-facing outputs stay in CALL order regardless of settlement
+    // order (prompt-cache stability).
+    expect(followupToolResults).not.toBeNull();
+    expect(followupToolResults?.map((o) => o.callId)).toEqual([
+      'call_slow',
+      'call_fast',
+    ]);
+  });
+
+  it('broadcasts a rejected call as it settles without waiting for the round', async () => {
+    let releaseSlow: () => void = () => undefined;
+    const slowGate = new Promise<void>((resolve) => {
+      releaseSlow = resolve;
+    });
+
+    const slow = tool({
+      name: 'slow',
+      inputSchema: z.object({}),
+      outputSchema: z.object({
+        done: z.boolean(),
+      }),
+      execute: async () => {
+        await slowGate;
+        return {
+          done: true,
+        };
+      },
+    });
+    const boom = tool({
+      name: 'boom',
+      inputSchema: z.object({}),
+      outputSchema: z.object({
+        ok: z.boolean(),
+      }),
+      execute: async () => {
+        throw new Error('explode');
+      },
+    });
+
+    const internal = buildModelResult([
+      slow,
+      boom,
+    ]);
+    internal.getInitialResponse = async () =>
+      makeResponseWithToolCalls([
+        {
+          id: 'call_slow',
+          name: 'slow',
+          arguments: '{}',
+        },
+        {
+          id: 'call_boom',
+          name: 'boom',
+          arguments: '{}',
+        },
+      ]);
+    internal.makeFollowupRequest = async () => makeFinalResponse();
+
+    const broadcaster = internal.ensureTurnBroadcaster();
+    const events: unknown[] = [];
+    const consumer = broadcaster.createConsumer();
+    const drainPromise = drainInto(consumer, events);
+
+    const executionPromise = internal.executeToolsIfNeeded();
+
+    await vi.waitFor(() => {
+      const toolResults = events.filter(isToolResultEvent);
+      expect(toolResults.some((e) => e.toolCallId === 'call_boom')).toBe(true);
+    });
+    const boomResult = events.filter(isToolResultEvent).find((e) => e.toolCallId === 'call_boom');
+    expect(boomResult).toMatchObject({
+      result: {
+        error: 'explode',
+      },
+    });
+    expect(events.filter(isToolResultEvent).some((e) => e.toolCallId === 'call_slow')).toBe(false);
+
+    releaseSlow();
+    await executionPromise;
+    broadcaster.complete();
+    await drainPromise;
+  });
+});