fix(opencode-go): treat stream boundaries as liveness

lin-hongkuan · lin-hongkuan · commit b9b88e63c7fa · 2026-06-25T02:04:48.000+08:00
diff --git a/extensions/opencode-go/stream-termination.test.ts b/extensions/opencode-go/stream-termination.test.ts
@@ -182,6 +182,82 @@ describe("createOpencodeGoStalledStreamWrapper", () => {
     await consumer;
   });
 
+  it("keeps a live stream open when block boundaries arrive between deltas", async () => {
+    const { stream: baseStream, controller } = createFakeBaseStream();
+    let abortCalled = false;
+
+    const underlying = vi.fn((_model, _context, options) => {
+      if (options?.signal) {
+        options.signal.addEventListener("abort", () => {
+          abortCalled = true;
+        });
+      }
+      return baseStream;
+    });
+
+    const wrapper = createOpencodeGoStalledStreamWrapper(underlying as any, {
+      provider: "opencode-go",
+      idleTimeoutMs: 5_000,
+    });
+
+    const downstream = await Promise.resolve(
+      wrapper({ provider: "opencode-go", id: "glm-4.6" } as any, {} as any, {} as any),
+    );
+    expect(downstream).toBeDefined();
+    if (!downstream) {
+      return;
+    }
+
+    const received: AnyEvent[] = [];
+    const consumer = (async () => {
+      for await (const event of downstream) {
+        received.push(event);
+      }
+    })();
+
+    const partial = {
+      role: "assistant",
+      content: [{ type: "toolCall", partialJson: "{", arguments: {}, name: "gateway" }],
+      stopReason: undefined,
+    };
+    controller.emit({ type: "start", partial } as any);
+    controller.emit({ type: "toolcall_delta", contentIndex: 0, delta: "{", partial } as any);
+    await vi.advanceTimersByTimeAsync(0);
+
+    await vi.advanceTimersByTimeAsync(3_000);
+    controller.emit({
+      type: "toolcall_end",
+      contentIndex: 0,
+      toolCall: { type: "toolCall", id: "call-1", name: "gateway", arguments: {} },
+      partial,
+    } as any);
+
+    await vi.advanceTimersByTimeAsync(3_000);
+    controller.emit({ type: "toolcall_start", contentIndex: 1, partial } as any);
+
+    await vi.advanceTimersByTimeAsync(4_000);
+    expect(abortCalled).toBe(false);
+
+    controller.emit({
+      type: "done",
+      reason: "stop",
+      message: {
+        ...partial,
+        content: [{ type: "text", text: "final answer" }],
+        stopReason: "stop",
+      },
+    } as any);
+    controller.end();
+    await consumer;
+
+    expect(received.some((event) => event.type === "done")).toBe(true);
+    expect(
+      received.some(
+        (event) => event.type === "error" && (event as any).error?.stopReason === "error",
+      ),
+    ).toBe(false);
+  });
+
   it("uses a longer first-event timeout than the inter-event idle timeout", async () => {
     const { stream: baseStream } = createFakeBaseStream();
     let abortCalled = false;
diff --git a/extensions/opencode-go/stream-termination.ts b/extensions/opencode-go/stream-termination.ts
@@ -51,14 +51,23 @@ function resolveTimeoutMs(model: unknown, fallbackMs: number): number {
   return validTimeoutMs((model as { requestTimeoutMs?: unknown })?.requestTimeoutMs) ?? fallbackMs;
 }
 
-function isProviderProgressEvent(event: AssistantMessageEvent): boolean {
+function isProviderDeltaEvent(event: AssistantMessageEvent): boolean {
   return (
     event.type === "text_delta" ||
     event.type === "thinking_delta" ||
     event.type === "toolcall_delta"
   );
 }
 
+function isProviderBoundaryLivenessEvent(event: AssistantMessageEvent): boolean {
+  return (
+    event.type === "text_end" ||
+    event.type === "thinking_end" ||
+    event.type === "toolcall_start" ||
+    event.type === "toolcall_end"
+  );
+}
+
 function combineAbortSignals(signals: (AbortSignal | undefined)[]): {
   signal: AbortSignal;
   cleanup(): void;
@@ -200,8 +209,8 @@ function synthesizeMinimalAssistantMessage(
  *
  * Behavior:
  * - Provider-scoped: only applies when `model.provider === options.provider`.
- * - Idle-based: the timer covers stream creation, first event delivery, and
- *   every gap after provider progress begins; if no event arrives within
+ * - Idle-based: the timer covers stream creation, first provider delta, and
+ *   every gap after provider progress begins; if no live event arrives within
  *   `idleTimeoutMs`, the wrapper calls `controller.abort()` on the AbortSignal
  *   injected into the underlying call (so the OpenAI SDK request is genuinely
  *   interrupted, not just the iterator) and pushes a terminal `error` event
@@ -210,7 +219,8 @@ function synthesizeMinimalAssistantMessage(
  *   wrapper forwards the event, clears all timers, and ends the stream.
  *
  * The wrapper never shortens the natural end of a normal completion, because
- * provider progress refreshes the idle timer and a terminal event cancels it entirely.
+ * provider deltas and later block boundaries refresh the idle timer and a terminal event cancels
+ * it entirely.
  */
 export function createOpencodeGoStalledStreamWrapper(
   underlying: ProviderStreamFn,
@@ -246,6 +256,7 @@ export function createOpencodeGoStalledStreamWrapper(
     let idleTimer: ReturnType<typeof setTimeout> | undefined;
     let lastSeenPartial: AssistantMessage | undefined;
     let settled = false;
+    let sawProviderDelta = false;
     let baseIterator: AsyncIterator<AssistantMessageEvent> | undefined;
 
     const clearIdleTimer = () => {
@@ -353,7 +364,10 @@ export function createOpencodeGoStalledStreamWrapper(
           }
           trackPartial(event);
           output.push(event);
-          if (isProviderProgressEvent(event)) {
+          if (isProviderDeltaEvent(event)) {
+            sawProviderDelta = true;
+            armIdleTimer();
+          } else if (sawProviderDelta && isProviderBoundaryLivenessEvent(event)) {
             armIdleTimer();
           }
         }
