fix: serialize runtime cleanup

This commit is contained in:
lofyer
2026-08-13 01:36:00 +08:00
parent fd1ff92927
commit 8cd23bada1
4 changed files with 206 additions and 4 deletions
+104
View File
@@ -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<void>((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<typeof createOpencodeClient>
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<RuntimeEvent, void, void>
): Promise<RuntimeEvent[]> => {
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 })
+50 -2
View File
@@ -544,6 +544,7 @@ export class OpenCodeRuntime implements AgentRuntime {
}
>()
private embeddedRunTail: Promise<void> = Promise.resolve()
private readonly conversationRunTails = new Map<string, Promise<void>>()
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<void>((resolve) => {
releaseGate = resolve
})
const tail = previous.then(
() => gate,
() => gate
)
this.conversationRunTails.set(conversationId, tail)
let abort!: () => void
const aborted = new Promise<never>((_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<RuntimeEvent, void, void> {
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()
}
@@ -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<void>((resolve) => {
finishCleanup = resolve
})
const started: string[] = []
const first = scheduler.schedule(async (signal) => {
started.push('first')
await new Promise<void>((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()
})
})
+6 -2
View File
@@ -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)