fix: resolve runtime, review, channel and chat state failures

Keep remote startup cancellable without blocking other sessions and surface local OpenCode submission failures promptly. Preserve supervisor project isolation, pause behavior and valid story destinations.

Restore generated-image channel delivery, Telegram reconnection on unchanged Save and DingTalk business-error retries. Preserve virtualized message disclosures using sparse state and release stale supervisor UI requests.

Add production-path regressions, Electron checks, isolated SQLite validation and audit evidence. Fix the SDK cleanup test fixture lifetime under garbage collection without weakening its assertions.

Validation: 6136 tests passed, 87 skipped; typecheck, lint and diff checks passed. Seven real text-model calls, current-source Linux x64 Host scenarios and large-history Electron comparisons completed.

Release note: Fix stuck runtime requests, inconsistent review results and channel delivery failures, and retain expanded details while browsing long conversations. Remote runtime fixes require an Agent build containing these changes.
This commit is contained in:
mesalogo
2026-10-05 10:25:43 +08:00
parent 126dbc57e2
commit bbf24c282a
49 changed files with 1710 additions and 106 deletions
@@ -23,6 +23,33 @@ afterEach(() => {
})
describe('AgentOwnedAcpPrompt', () => {
it('coalesces identical starts and rejects conflicting starts while initialization is pending', async () => {
const root = mkdtempSync(join(tmpdir(), 'goodbuddy-owned-start-'))
temporary.push(root)
const transcript = new SemanticPromptStore(join(root, 'prompts.sqlite'))
transcript.prepare({ bindingId: 'binding-1', operationId: 'operation-1', requestId: 'operation-1',
controllerId: 'controller-1', preparationDigest: `sha256:${'a'.repeat(64)}`, promptSequence: 0 })
let initialize!: (value: { protocolVersion: number }) => void
let finish!: (value: PromptResponse) => void
const newSession = vi.fn(async () => ({ sessionId: 'session-1' }))
const prompt = vi.fn(() => new Promise<PromptResponse>(resolve => { finish = resolve }))
const owner = new AgentOwnedAcpPrompt({ bindingId: 'binding-1', controllerId: 'controller-1',
workspaceDirectory: '/workspace', process: new MemoryProcess(), transcript, completePrompt: async () => undefined,
createConnection: () => ({ initialize: () => new Promise(resolve => { initialize = resolve }), newSession, prompt }) as unknown as ClientSideConnection })
const request = { bindingId: 'binding-1', operationId: 'operation-1', requestId: 'operation-1', prompt: [{ type: 'text' as const, text: 'test' }] }
try {
const first = owner.start(request)
const replay = owner.start(request)
await expect(owner.start({ ...request, prompt: [{ type: 'text', text: 'different' }] })).rejects.toMatchObject({ code: 'conflict' })
initialize({ protocolVersion: 1 })
expect(await first).toEqual(await replay)
expect(newSession).toHaveBeenCalledOnce()
expect(prompt).toHaveBeenCalledOnce()
finish({ stopReason: 'end_turn' })
await vi.waitFor(() => expect(transcript.attach('binding-1', 'operation-1', 'controller-1').state).toBe('completed'))
} finally { owner.close(); transcript.close() }
})
it('refreshes prompt tools on the retained session and authorizes execution', async () => {
const root = mkdtempSync(join(tmpdir(), 'goodbuddy-owned-acp-modes-'))
temporary.push(root)
+56 -12
View File
@@ -18,6 +18,7 @@ import type {
} from '../shared/remote-agent-contracts'
import type { RuntimeAcpProcessOwner } from './runtime-acp-backend'
import { AgentAcpConnection } from './agent-acp-connection'
import { raceWithAbort } from './async-utils'
import {
SemanticPromptStore,
SemanticPromptStoreError
@@ -63,6 +64,11 @@ export class AgentOwnedAcpPrompt {
#promptSettled?: Promise<void>
#closed = false
#initialized = false
#starting?: {
digest: string
controller: AbortController
promise: Promise<RemoteOwnedPromptStartResult>
}
#hasMcpServers = false
readonly #questions = new Map<string, {
endpoint: string; operationId: string; questionCount: number; notification: SessionNotification
@@ -86,6 +92,7 @@ export class AgentOwnedAcpPrompt {
async start(
request: RemoteOwnedPromptStartRequest
): Promise<RemoteOwnedPromptStartResult> {
if (this.#closed) throw new Error('ACP Session is closed')
if (
request.bindingId !== this.#options.bindingId
) {
@@ -95,6 +102,12 @@ export class AgentOwnedAcpPrompt {
)
}
const startDigest = digest(canonicalJson(request))
if (this.#starting) {
if (this.#starting.digest !== startDigest) {
throw new SemanticPromptStoreError('Another ACP prompt is starting', 'conflict')
}
return this.#starting.promise
}
const existing = this.#options.transcript.findStarted({
bindingId: request.bindingId,
operationId: request.operationId,
@@ -111,9 +124,39 @@ export class AgentOwnedAcpPrompt {
)
}
const controller = new AbortController()
const promise = this.#start(request, startDigest, controller.signal)
this.#starting = { digest: startDigest, controller, promise }
try {
return await promise
} finally {
this.#starting = undefined
}
}
async #start(
request: RemoteOwnedPromptStartRequest,
startDigest: string,
signal: AbortSignal
): Promise<RemoteOwnedPromptStartResult> {
const whileStarting = async <T>(operation: Promise<T>): Promise<T> => {
const result = await raceWithAbort(this.#whileProcessAlive(operation), signal)
signal.throwIfAborted()
return result
}
const retainSession = async (operation: Promise<string>): Promise<string> =>
whileStarting(operation.then(sessionId => {
if (signal.aborted) {
// A native session can be created after its owner was cancelled.
void this.#connection.closeSession({ sessionId }).catch(() => undefined)
} else {
this.#sessionId = sessionId
}
return sessionId
}))
const mcpServers = this.#options.mcpServers?.() ?? []
if (!this.#initialized) {
const initialization = await this.#whileProcessAlive(
const initialization = await whileStarting(
this.#transport.initialize({
protocolVersion: PROTOCOL_VERSION,
clientCapabilities: {
@@ -134,38 +177,36 @@ export class AgentOwnedAcpPrompt {
}
if (request.acpSessionId !== undefined && this.#sessionId === undefined) {
if (initialization.agentCapabilities?.loadSession === true) {
await this.#whileProcessAlive(
this.#sessionId = await retainSession(
this.#connection.loadSession({
sessionId: request.acpSessionId,
cwd: this.#options.workspaceDirectory,
mcpServers
})
}).then(() => request.acpSessionId!)
)
} else if (
initialization.agentCapabilities?.sessionCapabilities?.resume != null
) {
await this.#whileProcessAlive(
this.#sessionId = await retainSession(
this.#connection.resumeSession({
sessionId: request.acpSessionId,
cwd: this.#options.workspaceDirectory,
mcpServers
})
}).then(() => request.acpSessionId!)
)
} else {
throw new Error('ACP Runtime cannot resume the requested session')
}
this.#sessionId = request.acpSessionId
} else if (this.#sessionId === undefined) {
const created = await this.#whileProcessAlive(
this.#sessionId = await retainSession(
this.#connection.newSession({
cwd: this.#options.workspaceDirectory,
mcpServers
})
}).then(created => created.sessionId)
)
this.#sessionId = created.sessionId
}
if (this.#options.expectedModel !== undefined) {
const configured = await this.#whileProcessAlive(
const configured = await whileStarting(
this.#connection.setSessionConfigOption({
sessionId: this.#sessionId!,
configId: 'model',
@@ -190,13 +231,14 @@ export class AgentOwnedAcpPrompt {
) {
throw new Error('ACP session identity cannot change within a binding')
} else if (mcpServers.length > 0 || this.#hasMcpServers) {
await this.#whileProcessAlive(this.#connection.resumeSession({
await whileStarting(this.#connection.resumeSession({
sessionId: this.#sessionId!, cwd: this.#options.workspaceDirectory,
mcpServers
}))
}
this.#hasMcpServers = mcpServers.length > 0
await this.#options.prepareSession?.(this.#sessionId!, request.operationId)
await whileStarting(Promise.resolve(this.#options.prepareSession?.(this.#sessionId!, request.operationId)))
signal.throwIfAborted()
this.#active = {
operationId: request.operationId
}
@@ -251,6 +293,7 @@ export class AgentOwnedAcpPrompt {
}
async cancel(): Promise<void> {
this.#starting?.controller.abort(new Error('ACP prompt startup cancelled'))
this.#acceptQuestions = false
this.#questions.clear()
if (this.#sessionId !== undefined && !this.#closed) {
@@ -396,6 +439,7 @@ export class AgentOwnedAcpPrompt {
return
}
this.#closed = true
this.#starting?.controller.abort(new Error('ACP Session is closed'))
this.#acceptQuestions = false
this.#questions.clear()
this.#unregisterSession?.()
+24
View File
@@ -44,6 +44,30 @@ afterEach(() => {
})
describe('AgentProtocolServer connection bounds', () => {
it('dispatches cancellation and peer requests before a native startup response', async () => {
const gate = deferred<void>()
const calls: string[] = []
const harness = createHarness({ methods: {
'runtime/startPrompt': async () => { calls.push('start'); await gate.promise; return { started: true } },
'runtime/escalateCancellation': async () => { calls.push('cancel'); return { stopped: true } },
'runtime/getAcpCursors': async () => { calls.push('peer'); return {} }
} })
try {
harness.socket.receive(Buffer.concat([
controlRequest(harness.controller, 'start', '1', 1, 'runtime/startPrompt'),
controlRequest(harness.controller, 'cancel', '1', 1, 'runtime/escalateCancellation'),
controlRequest(harness.controller, 'peer', '1', 1, 'runtime/getAcpCursors')
].map(Buffer.from)))
await new Promise(resolve => setImmediate(resolve))
expect(calls).toEqual(['start', 'cancel', 'peer'])
expect(harness.socket.writes).toHaveLength(4)
gate.resolve()
await waitFor(() => harness.socket.writes.length === 6)
expect(harness.socket.maximumConcurrentWrites).toBe(1)
expect(harness.socket.destroyCalled).toBe(false)
} finally { gate.resolve(); harness.close() }
})
it('accepts a frame fragmented into single-byte socket chunks', async () => {
const harness = createHarness()
const frame = controlRequest(
+15 -5
View File
@@ -866,11 +866,19 @@ class ProtocolConnection {
if (!('id' in request)) {
await this.#invokeNotification(request, channel)
} else {
await this.#invokeRequest(request, channel)
await this.#sendChannelClose(channel)
this.#channels.close(channel.channelId)
this.#channelKinds.delete(channel.channelId)
return
const completion = this.#invokeRequest(request, channel).then(async () => {
if (this.#closed) return
await this.#sendChannelClose(channel)
this.#channels.close(channel.channelId)
this.#channelKinds.delete(channel.channelId)
})
if (request.method === 'runtime/startPrompt') {
// Startup holds its admitted request channel (and its existing capacity
// slot), but must leave the socket free for cancellation and peer work.
void completion.catch(error => this.#close(protocolFailureCategory('dispatch', error)))
} else {
await completion
}
}
}
@@ -923,6 +931,7 @@ class ProtocolConnection {
controllerTakeoverProven: this.#takeoverProven,
signal: this.#abortController.signal
})
if (this.#closed) return
if (request.method === 'runtime/resumeAcpChannel') {
this.#bindResumedAcpChannel(result)
}
@@ -941,6 +950,7 @@ class ProtocolConnection {
result: result ?? null
})
} catch (error) {
if (this.#closed) return
const serviceCode = typedErrorCode(error)
await this.#sendJson(channel, {
jsonrpc: '2.0',
@@ -518,6 +518,84 @@ function harness(input: {
}
describe('RuntimeAcpBackend', () => {
it.each(['watchdog', 'dispose'])('interrupts stalled native initialization through %s', async action => {
vi.useFakeTimers()
const fixture = harness({ agentOwned: true, shareOwnedProcesses: true })
try {
await open(fixture)
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({ deadlineAt: UNBOUNDED_REMOTE_PROMPT_DEADLINE }))
let settled = false
const starting = invoke(fixture, 'runtime/startPrompt', {
bindingId: 'binding-1', operationId: 'request-1', requestId: 'request-1', prompt: [{ type: 'text', text: 'test' }]
}).catch(() => undefined).finally(() => { settled = true })
await vi.advanceTimersByTimeAsync(1)
expect(fixture.process.writes).toHaveLength(1)
let disposalError: unknown
const disposal = action === 'dispose' ? fixture.backend.dispose().catch((error: unknown) => { disposalError = error }) : undefined
await vi.advanceTimersByTimeAsync(action === 'watchdog' ? 121000 : 10000)
const observed = { settled, stops: fixture.process.stops.length, disposalError }
const initialize = JSON.parse(Buffer.from(fixture.process.writes[0]!).toString())
await fixture.process.emit('stdout', `${JSON.stringify({ jsonrpc: '2.0', id: initialize.id, error: { code: -32603, message: 'release fixture' } })}\n`)
await Promise.all([starting, disposal])
await vi.advanceTimersByTimeAsync(100)
expect(observed).toEqual({ settled: true, stops: 1, disposalError: undefined })
expect(fixture.process.writes.map(payload => JSON.parse(Buffer.from(payload).toString()).method)).toEqual(['initialize'])
} finally {
vi.useRealTimers()
await fixture.backend.dispose().catch(() => undefined)
}
})
it.each(['cancel', 'deadline'])('keeps a shared peer responsive during native startup and %s', async reason => {
const fixture = await ownedHarness(true)
let releaseStartup: (() => Promise<void>) | undefined
try {
await fixture.start()
const process = fixture.processes[0]!
const originalWrite = process.onWrite!
process.onWrite = payload => {
const request = JSON.parse(Buffer.from(payload).toString())
if (request.method === 'session/new') {
releaseStartup = () => process.emit('stdout', `${JSON.stringify({ jsonrpc: '2.0', id: request.id, result: { sessionId: 'late-session' } })}\n`)
} else originalWrite(payload)
}
const opened = await invoke(fixture, 'runtime/openAcpChannel', { ...fixture.openRequest, bindingId: 'starting' }) as { channelEpoch: string }
vi.useFakeTimers()
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({ bindingId: 'starting', channelEpoch: opened.channelEpoch,
operationId: 'starting-request', requestId: 'starting-request',
deadlineAt: reason === 'deadline' ? new Date(1100).toISOString() : UNBOUNDED_REMOTE_PROMPT_DEADLINE }))
let startSettled = false
const starting = invoke(fixture, 'runtime/startPrompt', { bindingId: 'starting', operationId: 'starting-request', requestId: 'starting-request', prompt: [{ type: 'text', text: 'test' }] })
.catch(() => undefined).finally(() => { startSettled = true })
await vi.advanceTimersByTimeAsync(1)
expect(releaseStartup).toBeDefined()
let peerSettled = false
const peer = invoke(fixture, 'runtime/getAcpCursors', { bindingId: 'binding-1' }).then(() => { peerSettled = true })
let cancelled: Promise<unknown> | undefined
if (reason === 'cancel') cancelled = invoke(fixture, 'runtime/escalateCancellation', {
bindingId: 'starting', operationId: 'starting-request', requestId: 'starting-request', sessionId: 'opening', reason: 'requested'
})
await vi.advanceTimersByTimeAsync(200)
const observed = { peerSettled, startSettled, stops: process.stops.length }
// Release the fake native response even on the old implementation.
await releaseStartup!()
await vi.advanceTimersByTimeAsync(5000)
await Promise.all([starting, peer, cancelled])
expect(observed).toEqual({ peerSettled: true, startSettled: true, stops: 0 })
expect(fixture.transcript.attach('binding-1', 'request-1', fixture.context.controller.controllerId).state).toBe('running')
const writes = process.writes.map(payload => JSON.parse(Buffer.from(payload).toString()))
expect(writes.filter(request => request.method === 'session/prompt')).toHaveLength(1)
expect(writes).toContainEqual(expect.objectContaining({ method: 'session/close', params: { sessionId: 'late-session' } }))
await fixture.prompts[0]!()
await vi.advanceTimersByTimeAsync(1)
} finally {
await releaseStartup?.()
await fixture.prompts[0]?.()
vi.useRealTimers()
await fixture.cleanup()
}
})
it.each([false, true])('commits completion before idle cleanup and preserves it on cleanup failure (shared=%s)', async shared => {
for (const failure of ['stop', 'reconcile'] as const) {
const fixture = await ownedHarness(shared)
+18 -10
View File
@@ -440,7 +440,8 @@ export class RuntimeAcpBackend {
'runtime/preparePrompt': (params, context) =>
this.#enqueuePublicControl(() => this.#prepare(params, context)),
'runtime/startPrompt': (params, context) =>
this.#enqueuePublicControl(() => this.#startOwnedPrompt(params, context)),
this.#enqueuePublicControl(() => this.#startOwnedPrompt(params, context))
.then(({ started }) => started),
'runtime/attachPrompt': (params, context) =>
this.#enqueuePublicControl(() => this.#attachOwnedPrompt(params, context)),
'runtime/respondToQuestion': (params, context) =>
@@ -1441,7 +1442,7 @@ export class RuntimeAcpBackend {
return completion
}
async #startOwnedPrompt(
#startOwnedPrompt(
params: unknown,
context: ProtocolMethodContext
) {
@@ -1464,7 +1465,7 @@ export class RuntimeAcpBackend {
clearTimeout(binding.ownedPromptStartTimer)
binding.ownedPromptStartTimer = undefined
}
return existing
return { started: Promise.resolve(existing) }
}
if (
this.#options.semanticPrompts === undefined ||
@@ -1523,13 +1524,20 @@ export class RuntimeAcpBackend {
binding.poisoned ? 'outcome-unknown' : status
})
}
const result = await binding.ownedAcp.start(request)
prepared.ownedPromptStarted = true
if (binding.ownedPromptStartTimer !== undefined) {
clearTimeout(binding.ownedPromptStartTimer)
binding.ownedPromptStartTimer = undefined
}
return result
const owner = binding.ownedAcp
// Native initialization must not hold the global control queue: cancellation,
// watchdogs and other bindings still need to make progress while it waits.
const started = owner.start(request).then(result => {
if (binding.ownedAcp === owner && binding.activeOperationId === request.operationId) {
prepared.ownedPromptStarted = true
if (binding.ownedPromptStartTimer !== undefined) {
clearTimeout(binding.ownedPromptStartTimer)
binding.ownedPromptStartTimer = undefined
}
}
return result
})
return { started }
}
#attachOwnedPrompt(