diff --git a/docs/quality/recurring-defects.md b/docs/quality/recurring-defects.md index f6c04361..6632d964 100644 --- a/docs/quality/recurring-defects.md +++ b/docs/quality/recurring-defects.md @@ -54,6 +54,26 @@ `testQuery`，导致每次提问多一次 detail 请求，且 `EXTERNAL_KB_CAPABILITY_UNAVAILABLE` 会出现在正常对话中。 +## 事件迭代结束前释放底层订阅 + +**症状**：并行 OpenCode 会话已结束，但部分 SSE 响应流仍未关闭；只检查最终取消信号 +会误判回收成功。 + +**原因**：OpenCode SDK `1.18.29` 在迭代器 `return()` 时移除取消监听并释放 reader lock， +但不主动取消 reader。外层 `for await` 先结束 SDK 迭代器，外层 `finally` 再 abort 的 +顺序过晚，带缓冲的响应流可能继续存活。聊天和原生 Compact 均曾使用该顺序。 + +**规则**：两条 SDK 消费路径共用 `consumeEventSubscription`，先 abort 对应订阅，再结束 +底层迭代器；请求取消和 Server 生命周期仍保留各自作用域。不要修改依赖包、放宽回收 +断言或关闭共享 Server 来代替单次订阅回收。 + +**防复发**：Runtime 回归检查正常结束和消费方提前返回时，底层迭代器清理前取消已经发生。 +`opencode-runtime-lifecycle.test.ts` 使用真实二进制与 SDK，在 1、4、8 路并行、单请求 +取消和原生 Compact 后检查打开的响应流归零，并验证并行请求不被误取消。 + +**历史**：`b03c256` 已增加请求级订阅取消；2026-09-13 候选检查仍复现 3 条响应流未关闭， +复查发现聊天与 Compact 都需要调整上述顺序。 + ## 快照比对代替取消 **症状**：为检测“操作期间配置被改动”而反复读回状态并 `JSON.stringify` 比较。 diff --git a/src/main/agent/opencode-runtime-lifecycle.test.ts b/src/main/agent/opencode-runtime-lifecycle.test.ts index 041f4326..c88f0408 100644 --- a/src/main/agent/opencode-runtime-lifecycle.test.ts +++ b/src/main/agent/opencode-runtime-lifecycle.test.ts @@ -15,7 +15,7 @@ const binaryPath = join( ) it.skipIf(!existsSync(binaryPath))( - 'closes real event streams after parallel runs without cancelling a peer', + 'closes real event streams after parallel runs and compaction without cancelling a peer', async () => { const workspace = await mkdtemp(join(tmpdir(), 'goodbuddy-opencode-lifecycle-')) let cancelledController: AbortController | undefined @@ -94,14 +94,18 @@ it.skipIf(!existsSync(binaryPath))( } }) }) - const run = async (controller = new AbortController(), cancel = false): Promise => { + const run = async ( + controller = new AbortController(), + cancel = false, + conversationId = crypto.randomUUID() + ): Promise => { const requestId = crypto.randomUUID() const token = gateway.grant(requestId, [crypto.randomUUID()], controller.signal)! const deadline = setTimeout(() => controller.abort(), 30_000) let text = '' try { for await (const event of runtime.run({ - requestId, conversationId: crypto.randomUUID(), workMode: 'ask', + requestId, conversationId, workMode: 'ask', prompt: cancel ? 'CANCEL_LIFECYCLE' : 'Reply LIFECYCLE_OK without tools.', knowledgeCapabilityToken: token }, controller.signal)) { @@ -126,6 +130,18 @@ it.skipIf(!existsSync(binaryPath))( expect(run()).resolves.toBe('LIFECYCLE_OK') ]) await expect.poll(() => openEventBodies).toBe(0) + const conversationId = crypto.randomUUID() + await expect(run(new AbortController(), false, conversationId)).resolves.toBe('LIFECYCLE_OK') + await expect(runtime.compactConversation({ + requestId: crypto.randomUUID(), + conversationId, + runtimeSelection: { provider: 'opencode' }, + history: [], + historyMessageIds: [] + }, new AbortController().signal)).resolves.toMatchObject({ + result: { compacted: true } + }) + await expect.poll(() => openEventBodies).toBe(0) } finally { await runtime.dispose() await gateway.dispose() diff --git a/src/main/agent/opencode-runtime.test.ts b/src/main/agent/opencode-runtime.test.ts index ea7e672d..b800abd0 100644 --- a/src/main/agent/opencode-runtime.test.ts +++ b/src/main/agent/opencode-runtime.test.ts @@ -1737,6 +1737,7 @@ describe("OpenCodeRuntime embedded launcher", () => { const setup = runClient([]); const controller = new AbortController(); let subscriptionSignal: AbortSignal | undefined; + let abortedBeforeIteratorClose: boolean | undefined; const subscribe = setup.event.subscribe as unknown as ReturnType< typeof vi.fn >; @@ -1745,15 +1746,19 @@ describe("OpenCodeRuntime embedded launcher", () => { subscriptionSignal = options.signal; return { stream: (async function* () { - if (outcome === "cancelled") { - controller.abort(new DOMException("Cancelled", "AbortError")); - return; + try { + if (outcome === "cancelled") { + controller.abort(new DOMException("Cancelled", "AbortError")); + return; + } + if (outcome === "failed") return; + yield { + type: "session.idle", + properties: { sessionID: "session-1" }, + }; + } finally { + abortedBeforeIteratorClose = subscriptionSignal?.aborted; } - if (outcome === "failed") return; - yield { - type: "session.idle", - properties: { sessionID: "session-1" }, - }; })(), }; }, @@ -1784,6 +1789,9 @@ describe("OpenCodeRuntime embedded launcher", () => { await collect(); } expect(subscriptionSignal?.aborted).toBe(true); + if (outcome !== "failed") { + expect(abortedBeforeIteratorClose).toBe(true); + } expect(controller.signal.aborted).toBe(outcome === "cancelled"); } finally { await runtime.dispose(); diff --git a/src/main/agent/opencode-runtime.ts b/src/main/agent/opencode-runtime.ts index c8b4487b..384eec18 100644 --- a/src/main/agent/opencode-runtime.ts +++ b/src/main/agent/opencode-runtime.ts @@ -295,6 +295,24 @@ function eventSubscriptionOptions(signal: AbortSignal) { }; } +async function* consumeEventSubscription( + stream: AsyncGenerator, + controller: AbortController, +): AsyncGenerator { + try { + while (true) { + const next = await stream.next(); + if (next.done) return; + yield next.value; + } + } finally { + // The SDK removes its abort listener on iterator return without cancelling + // the reader. Abort first, while that listener can still close the body. + controller.abort(); + await stream.return(); + } +} + function awaitWithAbort( operation: Promise, signal: AbortSignal, @@ -2466,7 +2484,10 @@ export class OpenCodeRuntime implements AgentRuntime { const reportedMessageIds = new Set(); const childTaskIdsBySession = new Map(); let waitingForRetry = false; - for await (const event of subscription.stream) { + for await (const event of consumeEventSubscription( + subscription.stream, + subscriptionController, + )) { const childProgress = subagentProgress.update(event); if ( childProgress && @@ -3101,7 +3122,10 @@ export class OpenCodeRuntime implements AgentRuntime { const usageEvents: RuntimeModelUsageEvent[] = []; const reportedMessageIds = new Set(); const usageCapture = (async () => { - for await (const event of subscription.stream) { + for await (const event of consumeEventSubscription( + subscription.stream, + subscriptionController, + )) { if ( event.type === "message.updated" && event.properties.sessionID === sessionId &&