From 8cd23bada1495025120052d3c700dcfdbea64f63 Mon Sep 17 00:00:00 2001 From: lofyer Date: Thu, 13 Aug 2026 01:36:00 +0800 Subject: [PATCH] fix: serialize runtime cleanup --- src/main/agent/opencode-runtime.test.ts | 104 ++++++++++++++++++ src/main/agent/opencode-runtime.ts | 52 ++++++++- src/main/assistant/subagent-scheduler.test.ts | 46 ++++++++ src/main/assistant/subagent-scheduler.ts | 8 +- 4 files changed, 206 insertions(+), 4 deletions(-) diff --git a/src/main/agent/opencode-runtime.test.ts b/src/main/agent/opencode-runtime.test.ts index 66ee3ce..3b3c89a 100644 --- a/src/main/agent/opencode-runtime.test.ts +++ b/src/main/agent/opencode-runtime.test.ts @@ -15,6 +15,7 @@ import type { createOpencodeClient } from '@opencode-ai/sdk/v2' import type spawn from 'cross-spawn' import { describe, expect, it, vi } from 'vitest' import type { KnowledgeMcpGateway } from './knowledge-mcp-gateway' +import type { RuntimeEvent } from './runtime' import { OpenCodeRuntime, type OpenCodeRuntimeDependencies @@ -1080,6 +1081,109 @@ describe('OpenCodeRuntime embedded launcher', () => { expect(runtime.requiresToolApproval).toBe(false) }) + it('serializes external runs that share one conversation session', async () => { + const child = fakeChild() + let releaseFirst!: () => void + const firstGate = new Promise((resolve) => { + releaseFirst = resolve + }) + let subscriptionCount = 0 + const promptAsync = vi.fn().mockResolvedValue({ + data: true, + error: undefined + }) + const client = { + session: { + create: vi.fn().mockResolvedValue({ + data: { id: 'session-1' }, + error: undefined + }), + update: vi.fn().mockResolvedValue({ + data: { id: 'session-1' }, + error: undefined + }), + promptAsync, + abort: vi.fn().mockResolvedValue({ + data: true, + error: undefined + }) + }, + event: { + subscribe: vi.fn().mockImplementation(async () => { + subscriptionCount += 1 + const current = subscriptionCount + return { + stream: (async function* () { + if (current === 1) { + await firstGate + } + yield { + type: 'session.idle', + properties: { sessionID: 'session-1' } + } + })() + } + }) + }, + tool: { + ids: vi.fn().mockResolvedValue({ + data: [], + error: undefined + }) + } + } as unknown as ReturnType + const { deps } = dependencies(child, { + createClient: vi.fn( + () => client + ) as unknown as typeof createOpencodeClient + }) + const runtime = new OpenCodeRuntime( + options({ + baseUrl: 'http://127.0.0.1:4096', + embedded: false + }), + deps + ) + const request = { + requestId: '00000000-0000-4000-8000-000000000101', + conversationId: 'shared-conversation', + prompt: 'first', + workMode: 'execute' as const + } + const collect = async ( + stream: AsyncGenerator + ): Promise => { + const events: RuntimeEvent[] = [] + for await (const event of stream) { + events.push(event) + } + return events + } + const first = collect(runtime.run( + request, + new AbortController().signal + )) + await vi.waitFor(() => expect(promptAsync).toHaveBeenCalledTimes(1)) + const second = collect( + runtime.run( + { + ...request, + requestId: '00000000-0000-4000-8000-000000000102', + prompt: 'second' + }, + new AbortController().signal + ) + ) + await Promise.resolve() + expect(promptAsync).toHaveBeenCalledTimes(1) + + releaseFirst() + await first + await second + expect(promptAsync).toHaveBeenCalledTimes(2) + await runtime.dispose() + }) + it('loads assigned Skills before prompting', async () => { const child = fakeChild() const promptAsync = vi.fn().mockResolvedValue({ error: undefined }) diff --git a/src/main/agent/opencode-runtime.ts b/src/main/agent/opencode-runtime.ts index 1440b77..6a7ae62 100644 --- a/src/main/agent/opencode-runtime.ts +++ b/src/main/agent/opencode-runtime.ts @@ -544,6 +544,7 @@ export class OpenCodeRuntime implements AgentRuntime { } >() private embeddedRunTail: Promise = Promise.resolve() + private readonly conversationRunTails = new Map>() private readonly dependencies: OpenCodeRuntimeDependencies constructor( @@ -594,6 +595,47 @@ export class OpenCodeRuntime implements AgentRuntime { } } + private async acquireConversationRun( + conversationId: string, + signal: AbortSignal + ): Promise<() => void> { + signal.throwIfAborted() + const previous = + this.conversationRunTails.get(conversationId) ?? Promise.resolve() + let releaseGate!: () => void + const gate = new Promise((resolve) => { + releaseGate = resolve + }) + const tail = previous.then( + () => gate, + () => gate + ) + this.conversationRunTails.set(conversationId, tail) + let abort!: () => void + const aborted = new Promise((_resolve, reject) => { + abort = () => reject(signal.reason) + }) + signal.addEventListener('abort', abort, { once: true }) + try { + await Promise.race([previous, aborted]) + signal.throwIfAborted() + return () => { + releaseGate() + if (this.conversationRunTails.get(conversationId) === tail) { + this.conversationRunTails.delete(conversationId) + } + } + } catch (error) { + releaseGate() + if (this.conversationRunTails.get(conversationId) === tail) { + this.conversationRunTails.delete(conversationId) + } + throw error + } finally { + signal.removeEventListener('abort', abort) + } + } + private terminate(child: SpawnedProcess): void { if (child.exitCode !== null) { return @@ -1030,13 +1072,18 @@ export class OpenCodeRuntime implements AgentRuntime { request: AgentExecutionRequest, signal: AbortSignal ): AsyncGenerator { - const release = this.usesEmbeddedPermissionMediation() + const releaseEmbedded = this.usesEmbeddedPermissionMediation() ? await this.acquireEmbeddedRun(signal) : undefined + const releaseConversation = await this.acquireConversationRun( + request.conversationId, + signal + ) try { yield* this.runUnlocked(request, signal) } finally { - release?.() + releaseConversation() + releaseEmbedded?.() } } @@ -1643,6 +1690,7 @@ export class OpenCodeRuntime implements AgentRuntime { this.clientInitialization = undefined this.sessions.clear() this.sessionInitializations.clear() + this.conversationRunTails.clear() await server?.close() } diff --git a/src/main/assistant/subagent-scheduler.test.ts b/src/main/assistant/subagent-scheduler.test.ts index f1bf0b7..ec18f60 100644 --- a/src/main/assistant/subagent-scheduler.test.ts +++ b/src/main/assistant/subagent-scheduler.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest' +import { vi } from 'vitest' import { SubagentScheduler } from './subagent-scheduler' describe('SubagentScheduler', () => { @@ -51,4 +52,49 @@ describe('SubagentScheduler', () => { await expect(blocker).rejects.toThrow('120 秒') scheduler.dispose() }) + + it('holds its concurrency slot until aborted work finishes cleanup', async () => { + const scheduler = new SubagentScheduler({ + concurrency: 1, + queueLimit: 1, + timeoutMs: 1_000 + }) + const controller = new AbortController() + let finishCleanup!: () => void + const cleanupGate = new Promise((resolve) => { + finishCleanup = resolve + }) + const started: string[] = [] + const first = scheduler.schedule(async (signal) => { + started.push('first') + await new Promise((resolve) => { + signal.addEventListener('abort', () => resolve(), { once: true }) + }) + await cleanupGate + return 'first' + }, controller.signal) + const second = scheduler.schedule(async () => { + started.push('second') + return 'second' + }) + + await vi.waitFor(() => expect(started).toEqual(['first'])) + controller.abort(new Error('cancelled')) + await expect(first).rejects.toThrow('cancelled') + await Promise.resolve() + expect(started).toEqual(['first']) + + let idle = false + const idlePromise = scheduler.waitForIdle().then(() => { + idle = true + }) + await Promise.resolve() + expect(idle).toBe(false) + + finishCleanup() + await expect(second).resolves.toBe('second') + await idlePromise + expect(started).toEqual(['first', 'second']) + scheduler.dispose() + }) }) diff --git a/src/main/assistant/subagent-scheduler.ts b/src/main/assistant/subagent-scheduler.ts index d4ac21b..b5e76c6 100644 --- a/src/main/assistant/subagent-scheduler.ts +++ b/src/main/assistant/subagent-scheduler.ts @@ -131,8 +131,12 @@ export class SubagentScheduler { } controller.signal.addEventListener('abort', onAbort, { once: true }) }) - void Promise.race([workPromise, abortPromise]) - .then(entry.resolve, entry.reject) + void Promise.race([workPromise, abortPromise]).then( + entry.resolve, + entry.reject + ) + void workPromise + .catch(() => undefined) .finally(() => { clearTimeout(timeout) entry.signal?.removeEventListener('abort', forwardAbort)