root/goodbuddy
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fix: reuse runtime processes across project sessions

Browse Source 
Multiple projects could spawn duplicate heavyweight runtimes, while settings
changes and remote reconnection could cancel or strand native conversations.
Share compatible local OpenCode and DeepSeek Harness process owners and
Agent-owned ACP connections, retaining workspace routing, per-session modes,
model attribution, question ownership, targeted cancellation and idle cleanup.

Restore remote pending questions and terminal task projections across reconnect
and desktop restart, and preserve accepted ordinary requests during settings
replacement while cancelling expert-team parents with their replaced scheduler.
Extend regression coverage and Agent type checks. Record measured resource and
real-model validation results, including failures and limitations.

Also include the conversation media-generation design drafts (not an implemented
feature), documentation entry links and dedicated Linux test-directory guidance.
Desktop and Agent changes must be delivered together for shared process results.

Validation: typecheck and lint passed; staged content and documentation links
checked. Prior full regression and live-model evidence are recorded in progress
notes; no new full test suite or model calls were run for this commit.

Release note: 多项目会话现在复用兼容的 OpenCode 和 DeepSeek Harness 进程，减少重复启动并保持工作区、工具和取消隔离。修复远端问答在断线、重启及配置更新后的恢复与状态同步问题，Desktop 与 Agent 需配套更新。
This commit is contained in:
mesalogo
2026-09-15 08:14:42 +08:00
parent b755951ecd
commit e5aa9ed688
65 changed files with 4116 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent-daemon/agent-acp-connection.test.ts
+63
View File
@@ -0,0 +1,63 @@
import { describe, expect, it, vi } from 'vitest'
import { AgentSideConnection, ndJsonStream, PROTOCOL_VERSION, type Client } from '@agentclientprotocol/sdk'
import { AgentAcpConnection } from './agent-acp-connection'
import type { RuntimeAcpProcessOutput, RuntimeAcpProcessOwner } from './runtime-acp-backend'
describe('AgentAcpConnection', () => {
it('uses one decoder and handshake and routes interleaved Sessions without closing peers', async () => {
let output!: (value: RuntimeAcpProcessOutput) => void | Promise<void>
let input!: ReadableStreamDefaultController<Uint8Array>
const unsubscribe = vi.fn()
const process: RuntimeAcpProcessOwner = {
identity: { launchId: 'launch', processId: 'process', supervisorIdentityDigest: `sha256:${'a'.repeat(64)}` },
beginPrompt: vi.fn(), completePrompt: vi.fn(), stop: vi.fn(),
reconcile: vi.fn(),
subscribeOutput: vi.fn(listener => { output = listener; return unsubscribe }),
subscribeExit: vi.fn(() => unsubscribe),
writeStdin: async data => input.enqueue(data)
}
const initialize = vi.fn(async () => ({ protocolVersion: PROTOCOL_VERSION }))
const server = new AgentSideConnection(() => ({
initialize, newSession: async () => ({ sessionId: 'session' }),
prompt: async () => ({ stopReason: 'end_turn' as const }),
cancel: async () => undefined
}), ndJsonStream(new WritableStream({
write: async data => { await output({ stream: 'stdout', data }) }
}), new ReadableStream({ start: controller => { input = controller } })))
const transport = new AgentAcpConnection(process)
const one: Client = { sessionUpdate: vi.fn(), requestPermission: vi.fn(async () => ({
outcome: { outcome: 'selected', optionId: 'allow' }
})) }
const two: Client = { sessionUpdate: vi.fn(), requestPermission: vi.fn(async () => ({
outcome: { outcome: 'cancelled' }
})) }
const release = transport.register('one', one)
transport.register('two', two)
try {
await Promise.all([1, 2].map(() => transport.initialize({ protocolVersion: PROTOCOL_VERSION })))
expect(initialize).toHaveBeenCalledOnce()
for (const sessionId of ['one', 'two', 'one', 'two']) {
await server.sessionUpdate({
sessionId, update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: sessionId } }
})
}
await vi.waitFor(() => expect(two.sessionUpdate).toHaveBeenCalledTimes(2))
expect(one.sessionUpdate).toHaveBeenCalledTimes(2)
const permission = (sessionId: string) => server.requestPermission({
sessionId, options: [{ kind: 'allow_once', optionId: 'allow', name: 'Allow' }],
toolCall: { toolCallId: 'tool', title: 'write', kind: 'edit' }
})
expect(await permission('one')).toEqual({ outcome: { outcome: 'selected', optionId: 'allow' } })
expect(await permission('two')).toEqual({ outcome: { outcome: 'cancelled' } })
release()
expect(await permission('one')).toEqual({ outcome: { outcome: 'cancelled' } })
expect(await permission('two')).toEqual({ outcome: { outcome: 'cancelled' } })
expect(process.subscribeOutput).toHaveBeenCalledOnce()
expect(unsubscribe).not.toHaveBeenCalled()
} finally {
transport.dispose()
input.close()
}
expect(unsubscribe).toHaveBeenCalledTimes(2)
})
})
src/agent-daemon/agent-acp-connection.ts
+108
View File
@@ -0,0 +1,108 @@
import {
ClientSideConnection,
ndJsonStream,
type Client,
type InitializeRequest,
type InitializeResponse
} from '@agentclientprotocol/sdk'
import type { RuntimeAcpProcessOwner } from './runtime-acp-backend'
/** One decoder and handshake per native process, with Session-scoped clients. */
export class AgentAcpConnection {
readonly connection: ClientSideConnection
readonly exited: Promise<never>
private initialization?: Promise<InitializeResponse>
private readonly sessions = new Map<string, Client>()
private readonly unsubscribeOutput: () => void
private readonly unsubscribeExit: () => void
private closed = false
private closeInput?: () => void
private bridgeOrigin?: string
constructor(
process: RuntimeAcpProcessOwner,
fallback?: () => Client,
createConnection?: (client: () => Client, stream: ReturnType<typeof ndJsonStream>) => ClientSideConnection
) {
let controller: ReadableStreamDefaultController<Uint8Array> | undefined
let rejectExit!: (error: Error) => void
this.exited = new Promise<never>((_resolve, reject) => { rejectExit = reject })
void this.exited.catch(() => undefined)
const input = new ReadableStream<Uint8Array>({
start(value) { controller = value }
})
this.closeInput = () => {
controller?.close()
controller = undefined
}
this.unsubscribeOutput = process.subscribeOutput((output) => {
if (!this.closed && output.stream === 'stdout' && output.data.byteLength > 0) {
controller?.enqueue(output.data.slice())
}
}) ?? (() => undefined)
this.unsubscribeExit = process.subscribeExit?.(() => {
if (this.closed) return
const error = new Error('ACP Runtime process exited')
controller?.error(error)
controller = undefined
rejectExit(error)
}) ?? (() => undefined)
const output = new WritableStream<Uint8Array>({
write: async (data) => { await process.writeStdin(data) }
})
const client = (): Client => ({
sessionUpdate: async (notification) => {
await (this.sessions.get(notification.sessionId) ?? fallback?.())
?.sessionUpdate(notification)
},
requestPermission: async (request) =>
await (this.sessions.get(request.sessionId) ?? fallback?.())
?.requestPermission(request) ?? { outcome: { outcome: 'cancelled' } },
extMethod: async () => ({}),
extNotification: async (method, params) => {
if (
method === 'goodbuddy/modelBridgeReady' &&
typeof params.origin === 'string' &&
/^http:\/\/127\.0\.0\.1:\d+\/[A-Za-z0-9_-]{43}$/u.test(params.origin)
) this.bridgeOrigin = params.origin
}
})
const stream = ndJsonStream(output, input)
this.connection = createConnection?.(client, stream) ?? new ClientSideConnection(client, stream)
}
initialize(request: InitializeRequest): Promise<InitializeResponse> {
return this.initialization ??= this.connection.initialize(request)
}
register(sessionId: string, client: Client): () => void {
if (this.sessions.has(sessionId)) throw new Error('ACP Session already has an owner')
this.sessions.set(sessionId, client)
return () => {
if (this.sessions.get(sessionId) === client) this.sessions.delete(sessionId)
}
}
async setModelRoute(
sessionId: string,
operationId: string,
socketPath?: string,
workMode?: 'ask' | 'execute'
): Promise<void> {
if (!this.bridgeOrigin) throw new Error('Shared model bridge is not ready')
const response = await fetch(`${this.bridgeOrigin}/session`, {
method: 'POST', headers: { 'content-type': 'application/json' },
body: JSON.stringify({ sessionId, operationId, ...(socketPath ? { socketPath, workMode } : { release: true }) }),
signal: AbortSignal.timeout(10_000)
})
if (!response.ok) throw new Error(`Shared model bridge route failed (${response.status})`)
}
dispose(): void {
if (this.closed) return
this.closed = true
this.sessions.clear()
this.unsubscribeOutput()
this.unsubscribeExit()
this.closeInput?.()
}
}
src/agent-daemon/agent-owned-acp-prompt.test.ts
+64 -11
View File
@@ -23,6 +23,64 @@ afterEach(() => {
})
describe('AgentOwnedAcpPrompt', () => {
it('applies each prompt mode to the retained session and its permission decisions', async () => {
const root = mkdtempSync(join(tmpdir(), 'goodbuddy-owned-acp-modes-'))
temporary.push(root)
const transcript = new SemanticPromptStore(join(root, 'prompts.sqlite'))
let client!: Client
let finish!: (response: PromptResponse) => void
const newSession = vi.fn(async () => ({ sessionId: 'session-1' }))
const prepareSession = vi.fn(async () => undefined)
const completePrompt = vi.fn(async () => undefined)
const owner = new AgentOwnedAcpPrompt({
bindingId: 'binding-1', controllerId: 'controller-1',
workspaceDirectory: '/workspace', process: new MemoryProcess(),
transcript, prepareSession, completePrompt,
createConnection: factory => {
client = factory()
return {
initialize: async () => ({ protocolVersion: 1, agentCapabilities: {} }),
newSession,
prompt: () => new Promise<PromptResponse>(resolve => { finish = resolve })
} as unknown as ClientSideConnection
}
})
try {
for (const [index, workMode] of (['execute', 'ask', 'execute'] as const).entries()) {
const operationId = `operation-${index}`
transcript.prepare({
bindingId: 'binding-1', operationId, requestId: operationId,
controllerId: 'controller-1', preparationDigest: `sha256:${'a'.repeat(64)}`,
promptSequence: index
})
await owner.start({
bindingId: 'binding-1', operationId, requestId: operationId,
prompt: [{ type: 'text', text: 'Continue with the selected mode' }]
}, workMode)
expect(prepareSession).toHaveBeenLastCalledWith('session-1', operationId, workMode)
const result = await client.requestPermission!({
sessionId: 'session-1',
toolCall: { toolCallId: `tool-${index}`, kind: 'execute', title: 'Run a test' },
options: [
{ kind: 'allow_once', optionId: 'allow', name: 'Allow' },
{ kind: 'reject_once', optionId: 'reject', name: 'Reject' }
]
})
expect(result.outcome).toEqual({
outcome: 'selected', optionId: workMode === 'execute' ? 'allow' : 'reject'
})
finish({ stopReason: 'end_turn' })
await vi.waitFor(() => expect(transcript.attach('binding-1', operationId, 'controller-1'))
.toMatchObject({ state: 'completed' }))
}
expect(newSession).toHaveBeenCalledOnce()
expect(completePrompt).toHaveBeenCalledTimes(3)
} finally {
owner.close()
transcript.close()
}
})
it('keeps the original prompt promise through Desktop detach and recovers the final response', async () => {
const root = mkdtempSync(join(tmpdir(), 'goodbuddy-owned-acp-'))
temporary.push(root)
@@ -76,7 +134,6 @@ describe('AgentOwnedAcpPrompt', () => {
bindingId: 'binding-1',
controllerId: 'controller-1',
workspaceDirectory: '/workspace',
workMode: 'execute',
expectedModel,
process,
transcript,
@@ -101,7 +158,7 @@ describe('AgentOwnedAcpPrompt', () => {
requestId: 'operation-1',
prompt: [{ type: 'text' as const, text: 'finish independently' }]
}
await expect(owner.start(request)).resolves.toMatchObject({
await expect(owner.start(request, 'execute')).resolves.toMatchObject({
state: 'running',
sessionId: 'session-1'
})
@@ -141,7 +198,7 @@ describe('AgentOwnedAcpPrompt', () => {
limit: 10
}).events.map((event) => event.kind)
).toEqual(['session-update', 'prompt-terminal'])
await expect(owner.start(request)).resolves.toMatchObject({
await expect(owner.start(request, 'execute')).resolves.toMatchObject({
state: 'completed'
})
expect(prompt).toHaveBeenCalledOnce()
@@ -170,7 +227,6 @@ describe('AgentOwnedAcpPrompt', () => {
bindingId: 'binding-1',
controllerId: 'controller-1',
workspaceDirectory: '/workspace',
workMode: 'execute',
process: new MemoryProcess(),
transcript,
completePrompt,
@@ -191,7 +247,7 @@ describe('AgentOwnedAcpPrompt', () => {
operationId: 'operation-1',
requestId: 'operation-1',
prompt: [{ type: 'text', text: 'finish independently' }]
})
}, 'execute')
await vi.waitFor(() =>
expect(
transcript.attach('binding-1', 'operation-1', 'controller-1')
@@ -239,7 +295,6 @@ describe('AgentOwnedAcpPrompt', () => {
bindingId: 'binding-1',
controllerId: 'controller-1',
workspaceDirectory: '/workspace',
workMode: 'execute',
process,
transcript,
completePrompt,
@@ -261,7 +316,7 @@ describe('AgentOwnedAcpPrompt', () => {
operationId: 'operation-1',
requestId: 'operation-1',
prompt: [{ type: 'text', text: 'wait for Runtime exit' }]
})
}, 'execute')
await process.emitExit()
await vi.waitFor(() =>
@@ -315,7 +370,6 @@ describe('AgentOwnedAcpPrompt', () => {
bindingId: 'binding-1',
controllerId: 'controller-1',
workspaceDirectory: '/workspace',
workMode: 'execute',
process: new MemoryProcess(),
transcript,
completePrompt: async () => undefined,
@@ -338,7 +392,7 @@ describe('AgentOwnedAcpPrompt', () => {
operationId: 'operation-1',
requestId: 'operation-1',
prompt: [{ type: 'text', text: 'run one tool' }]
})
}, 'execute')
await expect(
client.requestPermission({
sessionId: 'session-1',
@@ -388,7 +442,6 @@ describe('AgentOwnedAcpPrompt', () => {
bindingId: 'binding-1',
controllerId: 'controller-1',
workspaceDirectory: '/workspace',
workMode: 'ask',
process: new MemoryProcess(),
transcript,
completePrompt: async () => undefined,
@@ -411,7 +464,7 @@ describe('AgentOwnedAcpPrompt', () => {
operationId: 'operation-1',
requestId: 'operation-1',
prompt: [{ type: 'text', text: 'inspect one file' }]
})
}, 'ask')
const permission = (
kind: 'read' | 'edit'
): Parameters<Client['requestPermission']>[0] => ({
src/agent-daemon/agent-owned-acp-prompt.ts
+50 -65
View File
@@ -16,6 +16,7 @@ import type {
RemoteOwnedPromptStartResult
} from '../shared/remote-agent-contracts'
import type { RuntimeAcpProcessOwner } from './runtime-acp-backend'
import { AgentAcpConnection } from './agent-acp-connection'
import {
SemanticPromptStore,
SemanticPromptStoreError
@@ -25,9 +26,10 @@ export type AgentOwnedAcpPromptOptions = {
bindingId: string
controllerId: string
workspaceDirectory: string
workMode: 'ask' | 'execute'
expectedModel?: string
process: RuntimeAcpProcessOwner
transport?: AgentAcpConnection
prepareSession?: (sessionId: string, operationId: string, workMode: 'ask' | 'execute') => Promise<void>
transcript: SemanticPromptStore
completePrompt: (
operationId: string,
@@ -44,18 +46,18 @@ export type AgentOwnedAcpPromptOptions = {
}
/**
* Owns the ACP ClientSideConnection and the original prompt Promise. No
* transport-connection signal is part of this lifetime.
* Owns a native Session and its original prompt Promise. Desktop connection
* signals are not part of this lifetime; the native transport may be shared.
*/
export class AgentOwnedAcpPrompt {
readonly #options: AgentOwnedAcpPromptOptions
readonly #connection: ClientSideConnection
readonly #input: ReadableStream<Uint8Array>
readonly #unsubscribeOutput: () => void
readonly #unsubscribeExit: () => void
readonly #transport: AgentAcpConnection
#unregisterSession?: () => void
readonly #processExit: Promise<never>
#sessionId?: string
#promptPromise?: Promise<void>
#promptSettled?: Promise<void>
#closed = false
#initialized = false
readonly #questions = new Map<string, {
@@ -64,56 +66,23 @@ export class AgentOwnedAcpPrompt {
#acceptQuestions = false
#active?: {
operationId: string
workMode: 'ask' | 'execute'
}
constructor(options: AgentOwnedAcpPromptOptions) {
this.#options = options
let controller: ReadableStreamDefaultController<Uint8Array> | undefined
let rejectProcessExit!: (error: Error) => void
this.#processExit = new Promise<never>((_resolve, reject) => {
rejectProcessExit = reject
})
void this.#processExit.catch(() => undefined)
this.#input = new ReadableStream<Uint8Array>({
start(value) {
controller = value
}
})
const output = new WritableStream<Uint8Array>({
write: async (chunk) => {
await options.process.writeStdin(chunk)
}
})
const unsubscribe = options.process.subscribeOutput((event) => {
if (
event.stream === 'stdout' &&
event.data.byteLength > 0 &&
!this.#closed
) {
controller?.enqueue(event.data.slice())
}
})
this.#unsubscribeOutput = unsubscribe ?? (() => undefined)
const unsubscribeExit = options.process.subscribeExit?.(() => {
if (this.#closed) {
return
}
const error = new Error('ACP Runtime process exited')
this.#acceptQuestions = false
this.#questions.clear()
controller?.error(error)
controller = undefined
rejectProcessExit(error)
})
this.#unsubscribeExit = unsubscribeExit ?? (() => undefined)
const stream = ndJsonStream(output, this.#input)
this.#connection =
options.createConnection?.(() => this.#client(), stream) ??
new ClientSideConnection(() => this.#client(), stream)
this.#transport = options.transport ?? new AgentAcpConnection(
options.process, () => this.#client(), options.createConnection
)
this.#connection = this.#transport.connection
this.#processExit = this.#transport.exited
}
get sessionId(): string | undefined { return this.#sessionId }
async start(
request: RemoteOwnedPromptStartRequest
request: RemoteOwnedPromptStartRequest,
workMode: 'ask' | 'execute'
): Promise<RemoteOwnedPromptStartResult> {
if (
request.bindingId !== this.#options.bindingId
@@ -142,7 +111,7 @@ export class AgentOwnedAcpPrompt {
if (!this.#initialized) {
const initialization = await this.#whileProcessAlive(
this.#connection.initialize({
this.#transport.initialize({
protocolVersion: PROTOCOL_VERSION,
clientCapabilities: {
terminal: false,
@@ -170,7 +139,7 @@ export class AgentOwnedAcpPrompt {
})
)
} else if (
initialization.agentCapabilities?.sessionCapabilities?.resume === true
initialization.agentCapabilities?.sessionCapabilities?.resume != null
) {
await this.#whileProcessAlive(
this.#connection.resumeSession({
@@ -210,6 +179,7 @@ export class AgentOwnedAcpPrompt {
throw new Error('ACP Runtime did not select the prepared model')
}
}
this.#unregisterSession = this.#transport.register(this.#sessionId!, this.#client())
this.#initialized = true
} else if (
request.acpSessionId !== undefined &&
@@ -217,8 +187,10 @@ export class AgentOwnedAcpPrompt {
) {
throw new Error('ACP session identity cannot change within a binding')
}
await this.#options.prepareSession?.(this.#sessionId!, request.operationId, workMode)
this.#active = {
operationId: request.operationId
operationId: request.operationId,
workMode
}
this.#acceptQuestions = true
const begun = this.#options.transcript.begin({
@@ -229,13 +201,14 @@ export class AgentOwnedAcpPrompt {
})
// Keep this exact Promise alive on Agent. The caller receives only the
// durable operation identity and may disconnect immediately.
this.#promptPromise = this.#whileProcessAlive(
const promptOperation = this.#whileProcessAlive(
this.#connection.prompt({
sessionId: this.#sessionId,
sessionId: this.#sessionId!,
prompt: request.prompt
})
)
.then(
this.#promptSettled = promptOperation.then(() => undefined, () => undefined)
this.#promptPromise = promptOperation.then(
async (response) => {
const proposed =
response.stopReason === 'cancelled'
@@ -305,6 +278,17 @@ export class AgentOwnedAcpPrompt {
this.#clear()
}
async releaseSession(): Promise<void> {
await this.cancel()
if (this.#sessionId && !this.#closed) {
await this.#connection.closeSession({ sessionId: this.#sessionId })
}
// Wait for the original native Prompt, not its completion callback, which
// may be queued behind the caller's backend control operation.
await this.#promptSettled
this.close()
}
async respondToQuestion(request: RemoteQuestionResponse): Promise<void> {
const pending = this.#questions.get(request.questionId)
if (this.#closed || !pending || this.#active?.operationId !== request.operationId ||
@@ -384,14 +368,15 @@ export class AgentOwnedAcpPrompt {
#handlePermission(
request: RequestPermissionRequest
): RequestPermissionResponse {
const active = this.#requireActive()
const selected =
this.#options.workMode === 'execute'
? request.options.find((option) => option.kind === 'allow_always') ??
request.options.find((option) => option.kind === 'allow_once')
: request.toolCall.kind === 'read'
active.workMode === 'execute'
? request.options.find((option) => option.kind === 'allow_once') ??
request.options.find((option) => option.kind === 'allow_always')
: request.toolCall.kind === 'read' || request.toolCall.kind === 'search'
? request.options.find((option) => option.kind === 'allow_once')
: request.options.find((option) => option.kind === 'reject_always') ??
request.options.find((option) => option.kind === 'reject_once')
: request.options.find((option) => option.kind === 'reject_once') ??
request.options.find((option) => option.kind === 'reject_always')
const response: RequestPermissionResponse =
selected === undefined
? { outcome: { outcome: 'cancelled' } }
@@ -403,7 +388,7 @@ export class AgentOwnedAcpPrompt {
}
this.#options.transcript.append({
bindingId: this.#options.bindingId,
operationId: this.#requireActive().operationId,
operationId: active.operationId,
kind: 'permission-decision',
payload: {
sessionId: request.sessionId,
@@ -434,15 +419,15 @@ export class AgentOwnedAcpPrompt {
this.#closed = true
this.#acceptQuestions = false
this.#questions.clear()
this.#unsubscribeOutput()
this.#unsubscribeExit()
this.#unregisterSession?.()
if (!this.#options.transport) this.#transport.dispose()
}
async #whileProcessAlive<T>(operation: Promise<T>): Promise<T> {
return await Promise.race([operation, this.#processExit])
}
#requireActive(): NonNullable<AgentOwnedAcpPrompt['#active']> {
#requireActive(): { operationId: string; workMode: 'ask' | 'execute' } {
if (this.#active === undefined) {
throw new Error('ACP notification has no active prompt')
}
src/agent-daemon/cli.ts
+5
View File
@@ -186,6 +186,7 @@ async function runModelBridgeHelper(
'model',
'supports-image-input',
'work-mode',
'shared-sessions',
'opencode-entrypoint'
])
requireOptions(options, [
@@ -212,6 +213,9 @@ async function runModelBridgeHelper(
if (workMode !== 'ask' && workMode !== 'execute') {
throw new Error('Invalid model bridge work-mode option')
}
if (options['shared-sessions'] !== undefined && options['shared-sessions'] !== 'true') {
throw new Error('Invalid model bridge shared-sessions option')
}
return await (
dependencies.runModelBridgeHelper ??
runOpenCodeModelBridgeHelper
@@ -221,6 +225,7 @@ async function runModelBridgeHelper(
model: options.model!,
supportsImageInput: imageInput === 'true',
workMode,
...(options['shared-sessions'] === 'true' ? { sharedSessions: true } : {}),
opencodeEntrypoint: options['opencode-entrypoint']!
})
}
src/agent-daemon/diagnostic-log.ts
+1 -1
View File
@@ -513,7 +513,7 @@ function diagnosticErrorName(error: unknown): string {
return error instanceof Error ? 'Error' : 'NonError'
}
const SAFE_DIAGNOSTIC_ERROR_CODES = new Set([
const SAFE_DIAGNOSTIC_ERROR_CODES: ReadonlySet<string> = new Set([
'EACCES',
'EADDRINUSE',
'ECONNREFUSED',
src/agent-daemon/direct-linux-stdio-process-owner.test.ts
+26
View File
@@ -24,6 +24,32 @@ afterEach(() => {
})
describe('direct Linux stdio Runtime ownership', () => {
it('keeps shared stdin available across Session completion without a cumulative prompt quota', async () => {
const registry = createRegistry()
const child = fakeChild()
const owner = await launchDirectLinuxStdioProcessOwner({
manifest: manifest(), profile: profile(), sharedSessions: true,
identity: { launchId: 'shared-launch', processId: 'shared-process' },
installationId: 'installation-1', registry, platform: 'linux',
deadlineAt: '2030-01-01T00:00:00.000Z', maximumInputBytes: 16,
spawn: () => { queueMicrotask(() => child.emit('spawn')); return child },
randomOwnerToken: () => 'a'.repeat(32),
readProcessIdentity: async () => identity()
})
try {
for (let index = 0; index < 8; index++) {
owner.beginPrompt({ deadlineAt: '2030-01-01T00:00:00.000Z', maximumInputBytes: 16 })
await owner.writeStdin(Uint8Array.from(Buffer.from('12345678')))
await owner.completePrompt()
}
await owner.writeStdin(Uint8Array.from(Buffer.from('cancel-peer')))
} finally {
child.exitCode = 0
child.emit('close', 0, null)
registry.close()
}
})
it('spawns Ask directly without systemd using detached fixed stdio', async () => {
const registry = createRegistry()
const child = fakeChild()
src/agent-daemon/direct-linux-stdio-process-owner.ts
+12 -4
View File
@@ -108,6 +108,7 @@ export type DirectLinuxStdioProcessOwnerOptions = {
bridgeDirectory: string
socketPath: string
policy: ModelBridgePolicy
sharedSessions?: boolean
}
}
identity: { launchId: string; processId: string }
@@ -115,6 +116,7 @@ export type DirectLinuxStdioProcessOwnerOptions = {
registry: RuntimeOwnerRegistry
deadlineAt: string
maximumInputBytes: number
sharedSessions?: boolean
signal?: AbortSignal
platform?: NodeJS.Platform
spawn?: DirectLinuxStdioSpawn
@@ -151,6 +153,7 @@ export class DirectLinuxStdioProcessOwner {
readonly processIdentity: LinuxRuntimeProcessIdentity
readonly #child: DirectLinuxStdioChild
readonly #sharedSessions: boolean
readonly #registry: RuntimeOwnerRegistry
readonly #now: () => number
readonly #readIdentity: typeof readLinuxRuntimeProcessIdentity
@@ -194,6 +197,7 @@ export class DirectLinuxStdioProcessOwner {
listPidNamespaceMembers: typeof listLinuxPidNamespaceMembers
sendSignal: typeof process.kill
maximumInputBytes: number
sharedSessions?: boolean
maximumPendingStdinBytes: number
maximumStdinWriteBytes: number
maximumOutputBytes: number
@@ -208,6 +212,7 @@ export class DirectLinuxStdioProcessOwner {
this.ownerId = input.ownerId
this.processIdentity = Object.freeze({ ...input.processIdentity })
this.#child = input.child
this.#sharedSessions = input.sharedSessions === true
this.#registry = input.registry
this.#now = input.now
this.#readIdentity = input.readProcessIdentity
@@ -222,7 +227,7 @@ export class DirectLinuxStdioProcessOwner {
input.maximumPromptRuntimeMilliseconds
this.#maximumOutputQueueChunks = input.maximumOutputQueueChunks
this.#maximumListeners = input.maximumListeners
this.#scheduleDeadline(input.deadlineAt)
if (!this.#sharedSessions) this.#scheduleDeadline(input.deadlineAt)
input.child.stdout.on('data', (chunk: Buffer | Uint8Array) => {
this.#acceptOutput(chunk)
})
@@ -245,13 +250,13 @@ export class DirectLinuxStdioProcessOwner {
}
if (
payload.byteLength > this.#maximumStdinWriteBytes ||
this.#inputBytes + payload.byteLength > this.#maximumInputBytes ||
(!this.#sharedSessions && this.#inputBytes + payload.byteLength > this.#maximumInputBytes) ||
this.#pendingInputBytes + payload.byteLength > this.#maximumPendingStdinBytes
) {
throw capacityError('Runtime stdin quota reached')
}
const copy = Buffer.from(payload)
this.#inputBytes += copy.byteLength
if (!this.#sharedSessions) this.#inputBytes += copy.byteLength
this.#pendingInputBytes += copy.byteLength
const write = this.#writeTail.then(async () => {
if (this.#closed || this.#stdinClosed || !this.#promptActive) {
@@ -268,6 +273,7 @@ export class DirectLinuxStdioProcessOwner {
}
beginPrompt(input: { deadlineAt: string; maximumInputBytes: number }): void {
if (this.#sharedSessions && !this.#closed && !this.#stdinClosed) return
if (this.#closed || this.#stdinClosed || this.#promptActive) {
throw new DirectLinuxStdioProcessOwnerError(
this.#promptActive ? 'A Runtime prompt is already active' : 'Runtime process is closed',
@@ -293,6 +299,7 @@ export class DirectLinuxStdioProcessOwner {
}
async completePrompt(): Promise<void> {
if (this.#sharedSessions) return
if (!this.#promptActive) return
await this.#writeTail
this.#promptActive = false
@@ -655,6 +662,7 @@ export async function launchDirectLinuxStdioProcessOwner(
options.listPidNamespaceMembers ?? listLinuxPidNamespaceMembers,
sendSignal: options.sendSignal ?? process.kill,
maximumInputBytes,
sharedSessions: options.sharedSessions,
maximumPendingStdinBytes,
maximumStdinWriteBytes,
maximumOutputBytes: manifest.limits.maximumPromptOutputBytes,
@@ -975,7 +983,7 @@ function defaultSpawn(
args: readonly string[],
options: Parameters<DirectLinuxStdioSpawn>[2]
): DirectLinuxStdioChild {
return nodeSpawn(executable, [...args], options)
return nodeSpawn(executable, [...args], { ...options, stdio: [...options.stdio] })
}
async function waitForSpawn(
src/agent-daemon/model-bridge-helper.ts
+77 -18
View File
@@ -132,24 +132,27 @@ const FORWARDED_HEADER_NAMES = [
export class ModelBridgeLoopbackProxy {
readonly #exchange: ModelBridgeExchange
readonly #sharedSessions: boolean
readonly #sessionRoutes = new Map<string, { operationId: string; socketPath: string; workMode: 'ask' | 'execute' }>()
readonly #routeToken: string
readonly #maximumConnections: number
readonly #requestTimeoutMs: number
readonly #sockets = new Set<Socket>()
readonly #dispatchWaiters: DispatchWaiter[] = []
readonly #dispatches = new Map<string, DispatchWaiter[]>()
#server?: Server
#origin?: string
#dispatchBusy = false
#closing = false
#closePromise?: Promise<void>
constructor(options: {
exchange: ModelBridgeExchange
sharedSessions?: boolean
routeToken?: string
maximumConnections?: number
requestTimeoutMs?: number
}) {
this.#exchange = options.exchange
this.#sharedSessions = options.sharedSessions === true
this.#routeToken = parseModelBridgeRouteToken(
options.routeToken ??
randomBytes(32).toString('base64url')
@@ -259,6 +262,7 @@ export class ModelBridgeLoopbackProxy {
)
} finally {
this.#sockets.clear()
this.#sessionRoutes.clear()
this.#closing = false
}
}
@@ -280,6 +284,9 @@ export class ModelBridgeLoopbackProxy {
let responseFlushed = false
let deliveryAcknowledged = false
let dispatchHeld = false
let dispatchKey = ''
let sessionId: string | undefined
let sessionRoute: { operationId: string; socketPath: string } | undefined
const onAborted = (): void => {
cancellation.abort(
new ModelBridgeBrokerError('request-cancelled')
@@ -303,14 +310,58 @@ export class ModelBridgeLoopbackProxy {
timeout?.unref?.()
try {
if (this.#sharedSessions && incoming.url?.startsWith(`/${this.#routeToken}/session`)) {
const url = new URL(incoming.url, 'http://localhost')
if (url.pathname !== `/${this.#routeToken}/session`) throw new HttpRequestError(404, 'path-not-allowed')
if (incoming.method === 'GET') {
const route = this.#sessionRoutes.get(url.searchParams.get('sessionId') ?? '')
if (!route) throw new HttpRequestError(409, 'session-inactive')
outgoing.setHeader('content-type', 'application/json')
outgoing.end(JSON.stringify({ operationId: route.operationId, workMode: route.workMode }))
return
}
if (incoming.method !== 'POST') throw new HttpRequestError(405, 'method-not-allowed')
const value = JSON.parse((await readBoundedBody(incoming)).toString('utf8')) as {
sessionId?: string; operationId?: string; socketPath?: string; release?: boolean; workMode?: 'ask' | 'execute'
}
if (
typeof value.sessionId !== 'string' || value.sessionId.length > 128 ||
typeof value.operationId !== 'string' || value.operationId.length > 128
) throw new HttpRequestError(400, 'request-invalid')
if (value.release) {
if (this.#sessionRoutes.get(value.sessionId)?.operationId === value.operationId) {
this.#sessionRoutes.delete(value.sessionId)
}
} else {
if (value.workMode !== 'ask' && value.workMode !== 'execute') throw new HttpRequestError(400, 'request-invalid')
const socketPath = normalizedAbsolutePath(value.socketPath ?? '', 'Model bridge socket')
this.#sessionRoutes.set(value.sessionId, { operationId: value.operationId, socketPath, workMode: value.workMode })
}
outgoing.end('{}')
return
}
const request = await parseHttpRequest(
incoming,
this.#routeToken
)
await this.#acquireDispatch(cancellation.signal)
let exchange = this.#exchange
if (this.#sharedSessions) {
const sessionHeader = incoming.headers['x-goodbuddy-session']
const operationId = incoming.headers['x-goodbuddy-operation']
sessionId = typeof sessionHeader === 'string' ? sessionHeader : undefined
const route = sessionId ? this.#sessionRoutes.get(sessionId) : undefined
if (!route || route.operationId !== operationId) throw new HttpRequestError(409, 'session-inactive')
sessionRoute = route
dispatchKey = `${sessionId}\0${operationId}`
exchange = createUnixModelBridgeExchange({ socketPath: route.socketPath })
}
await this.#acquireDispatch(dispatchKey, cancellation.signal)
dispatchHeld = true
if (sessionId && this.#sessionRoutes.get(sessionId) !== sessionRoute) {
throw new HttpRequestError(409, 'session-inactive')
}
dispatched = normalizeExchangeResult(
await raceWithAbort(this.#exchange(request, {
await raceWithAbort(exchange(request, {
requestId: `model-${randomUUID()}`,
signal: cancellation.signal
}), cancellation.signal)
@@ -351,34 +402,36 @@ export class ModelBridgeLoopbackProxy {
incoming.off('aborted', onAborted)
outgoing.off('close', onResponseClose)
if (dispatchHeld) {
this.#releaseDispatch()
this.#releaseDispatch(dispatchKey)
}
}
}
async #acquireDispatch(signal: AbortSignal): Promise<void> {
async #acquireDispatch(key: string, signal: AbortSignal): Promise<void> {
signal.throwIfAborted()
if (!this.#dispatchBusy) {
this.#dispatchBusy = true
const queue = this.#dispatches.get(key)
if (!queue) {
this.#dispatches.set(key, [])
return
}
await new Promise<void>((resolve, reject) => {
const waiter: DispatchWaiter = { resolve, reject, signal }
waiter.abort = (): void => {
const index = this.#dispatchWaiters.indexOf(waiter)
const index = queue.indexOf(waiter)
if (index >= 0) {
this.#dispatchWaiters.splice(index, 1)
queue.splice(index, 1)
}
reject(signal.reason)
}
signal.addEventListener('abort', waiter.abort, { once: true })
this.#dispatchWaiters.push(waiter)
queue.push(waiter)
})
}
#releaseDispatch(): void {
while (this.#dispatchWaiters.length > 0) {
const waiter = this.#dispatchWaiters.shift()!
#releaseDispatch(key: string): void {
const queue = this.#dispatches.get(key)
while (queue && queue.length > 0) {
const waiter = queue.shift()!
if (waiter.abort !== undefined) {
waiter.signal.removeEventListener('abort', waiter.abort)
}
@@ -389,7 +442,7 @@ export class ModelBridgeLoopbackProxy {
waiter.resolve()
return
}
this.#dispatchBusy = false
this.#dispatches.delete(key)
}
}
@@ -399,6 +452,7 @@ export async function runOpenCodeModelBridgeHelper(options: {
model: string
supportsImageInput: boolean
workMode: 'ask' | 'execute'
sharedSessions?: boolean
opencodeEntrypoint: string
environment?: Readonly<NodeJS.ProcessEnv>
spawn?: ModelBridgeHelperSpawn
@@ -412,19 +466,19 @@ export async function runOpenCodeModelBridgeHelper(options: {
'OpenCode entrypoint'
)
const exchange = createUnixModelBridgeExchange({ socketPath })
const proxy = new ModelBridgeLoopbackProxy({ exchange })
const proxy = new ModelBridgeLoopbackProxy({ exchange, sharedSessions: options.sharedSessions })
const origin = await proxy.listen()
let pluginDirectory: string | undefined
try {
pluginDirectory = await mkdtemp(join(tmpdir(), 'goodbuddy-opencode-plugin-'))
const pluginPath = join(pluginDirectory, 'subagent.mjs')
await writeFile(pluginPath, openCodeSubagentPluginSource(), 'utf8')
await writeFile(pluginPath, openCodeSubagentPluginSource(options.sharedSessions ? origin : undefined), 'utf8')
const config = createOpenCodeModelBridgeProviderConfig({
protocol: options.protocol,
model: options.model,
loopbackOrigin: origin,
supportsImageInput: options.supportsImageInput,
workMode: options.workMode
workMode: options.sharedSessions ? 'ask' : options.workMode
})
const environment = credentialFreeHelperEnvironment(
options.environment ?? process.env,
@@ -433,6 +487,11 @@ export async function runOpenCodeModelBridgeHelper(options: {
plugin: [pathToFileURL(pluginPath).href]
})
)
if (options.sharedSessions) {
process.stdout.write(JSON.stringify({
jsonrpc: '2.0', method: 'goodbuddy/modelBridgeReady', params: { origin }
}) + '\n')
}
const child = (options.spawn ?? defaultHelperSpawn)(
opencodeEntrypoint,
['acp'],
src/agent-daemon/model-bridge.test.ts
+44
View File
@@ -615,6 +615,50 @@ describe('model bridge loopback helper', () => {
describe('model bridge Unix broker', () => {
const runOnUnix = process.platform === 'win32' ? it.skip : it
runOnUnix('routes shared Sessions concurrently and rejects queued work after its operation ends', async () => {
let unblock!: () => void
const blocked = new Promise<void>(resolve => { unblock = resolve })
const dispatchOne = vi.fn(async () => { await blocked; return validResponse })
const dispatchTwo = vi.fn(async () => validResponse)
const brokers = [dispatchOne, dispatchTwo].map(dispatch => new ModelBridgeBrokerServer({
scratchDirectory: privateTemporaryDirectory(), dispatch
}))
const fallback = vi.fn(async () => validResponse)
const proxy = new ModelBridgeLoopbackProxy({ exchange: fallback, sharedSessions: true })
await Promise.all(brokers.map(broker => broker.listen()))
const origin = await proxy.listen()
const route = (sessionId: string, operationId: string, index: number) => fetch(`${origin}/session`, {
method: 'POST', body: JSON.stringify({ sessionId, operationId, socketPath: brokers[index]!.socketPath, workMode: 'execute' })
})
const prompt = (sessionId: string, operationId: string) => fetch(`${origin}/v1/messages`, {
method: 'POST', headers: { 'content-type': 'application/json', 'x-goodbuddy-session': sessionId, 'x-goodbuddy-operation': operationId },
body: '{"model":"fixture"}'
})
try {
expect((await route('one', 'operation-1', 0)).ok).toBe(true)
expect((await route('two', 'operation-2', 1)).ok).toBe(true)
const first = prompt('one', 'operation-1')
await vi.waitFor(() => expect(dispatchOne).toHaveBeenCalledOnce())
const queued = prompt('one', 'operation-1')
expect((await prompt('two', 'operation-2')).status).toBe(201)
await route('one', 'next-operation', 0)
unblock()
expect((await first).status).toBe(201)
expect((await queued).status).toBe(409)
expect((await prompt('one', 'operation-1')).status).toBe(409)
expect((await prompt('one', 'next-operation')).status).toBe(201)
expect(dispatchOne).toHaveBeenCalledTimes(2)
expect(fallback).not.toHaveBeenCalled()
for (const dispatch of [dispatchOne, dispatchTwo]) {
expect(dispatch.mock.calls[0]?.[0]).not.toHaveProperty('headers.x-goodbuddy-session')
}
} finally {
unblock()
await proxy.close()
await Promise.all(brokers.map(broker => broker.close()))
}
})
it('allows a delayed connection with the default unlimited request timeout', async () => {
const socket = new Duplex({
read() {},
src/agent-daemon/opencode-runtime-profile.ts
+3
View File
@@ -41,6 +41,7 @@ export function createOpenCodeLaunchProfile(input: {
bridgeDirectory: string;
socketPath: string;
policy: ModelBridgePolicy;
sharedSessions?: boolean;
};
}): OpenCodeLaunchProfile {
const manifest = remoteRuntimeBundleManifestSchema.parse(input.manifest);
@@ -90,6 +91,7 @@ export function createOpenCodeLaunchProfile(input: {
"Model bridge socket",
),
policy: modelBridgePolicySchema.parse(input.modelBridge.policy),
sharedSessions: input.modelBridge.sharedSessions === true,
};
if (
modelBridge !== undefined &&
@@ -132,6 +134,7 @@ export function createOpenCodeLaunchProfile(input: {
modelBridge.policy.supportsImageInput ? "true" : "false",
"--work-mode",
input.workMode,
...(modelBridge.sharedSessions ? ["--shared-sessions", "true"] : []),
"--opencode-entrypoint",
executablePath,
],
src/agent-daemon/opencode-subagent-plugin.test.ts
+59
View File
@@ -5,6 +5,65 @@ import { describe, expect, it, vi } from 'vitest'
import { openCodeSubagentPluginSource } from './opencode-subagent-plugin'
describe('OpenCode ACP child event plugin', () => {
it('attributes child and compaction model rounds to the frozen root operation', async () => {
const operations = new Map([['one', 'operation-1'], ['two', 'operation-2']])
const fetchRoute = vi.fn(async (url: string) => {
const session = new URL(url).searchParams.get('sessionId')!
return { ok: operations.has(session), json: async () => ({ operationId: operations.get(session), workMode: 'execute' }) }
})
const factory = runInNewContext(
`(${openCodeSubagentPluginSource('http://127.0.0.1:12345/fixture').replace(/^import .*\n/gmu, '').replace('export default ', '')})`,
{ process: { stdout: { write: vi.fn() } }, createServer, randomBytes, fetch: fetchRoute }
)
const patch = vi.fn()
const hooks = await factory({ client: {
_client: { get: vi.fn(), post: vi.fn(), patch },
session: { get: async ({ path }: { path: { id: string } }) => ({
data: {
id: path.id, parentID: path.id === 'child' ? 'one' : undefined,
permission: path.id === 'child' ? [{ permission: 'task', pattern: '*', action: 'deny' }] : undefined
}
}) }
} })
const headers = async (sessionID: string, id: string) => {
const output = { headers: {} }
await hooks['chat.headers']({ sessionID, message: { id } }, output)
return output.headers
}
try {
await hooks['chat.message']({ sessionID: 'one' }, { message: { id: 'first' } })
expect(patch).toHaveBeenCalledWith({
url: '/session/one', body: { permission: [{ permission: '*', pattern: '*', action: 'allow' }] }, throwOnError: true
})
expect(await headers('child', 'child-message')).toEqual({
'x-goodbuddy-session': 'one', 'x-goodbuddy-operation': 'operation-1'
})
await hooks['chat.message']({ sessionID: 'child' }, { message: { id: 'child-message' } })
expect(patch).toHaveBeenLastCalledWith({
url: '/session/child', body: { permission: [
{ permission: '*', pattern: '*', action: 'allow' },
{ permission: 'task', pattern: '*', action: 'deny' }
] }, throwOnError: true
})
expect(await headers('two', 'compact')).toEqual({
'x-goodbuddy-session': 'two', 'x-goodbuddy-operation': 'operation-2'
})
operations.set('one', 'next-operation')
expect(await headers('one', 'first')).toEqual({
'x-goodbuddy-session': 'one', 'x-goodbuddy-operation': 'operation-1'
})
await hooks.event({ event: { type: 'session.status', properties: { sessionID: 'one', status: { type: 'idle' } } } })
expect(await headers('one', 'next-message')).toEqual({
'x-goodbuddy-session': 'one', 'x-goodbuddy-operation': 'next-operation'
})
operations.delete('two')
await expect(headers('two', 'stale-new-message')).rejects.toThrow('no longer active')
expect(await headers('two', 'compact')).toEqual({
'x-goodbuddy-session': 'two', 'x-goodbuddy-operation': 'operation-2'
})
} finally { await hooks.dispose() }
})
it('authenticates native replies, checks live ownership, rejects stale events and closes its endpoint', async () => {
const write = vi.fn()
const question = { id: 'q', sessionID: 'grandchild', questions: [
src/agent-daemon/opencode-subagent-plugin.ts
+66 -3
View File
@@ -2,10 +2,10 @@
* OpenCode's ACP adapter omits native questions and child sessions. Native
* events use ACP metadata; question replies use a process-local capability.
*/
export function openCodeSubagentPluginSource(): string {
export function openCodeSubagentPluginSource(modelBridgeOrigin?: string): string {
return `import { createServer } from 'node:http'
import { randomBytes } from 'node:crypto'
export default (input) => (${plugin.toString()})(input, { createServer, randomBytes })`
export default (input) => (${plugin.toString()})(input, { createServer, randomBytes, modelBridgeOrigin: ${JSON.stringify(modelBridgeOrigin)} })`
}
async function plugin(input?: {
@@ -13,17 +13,41 @@ async function plugin(input?: {
_client: {
get(input: { url: string; throwOnError: true }): Promise<{ data: unknown }>
post(input: { url: string; body?: unknown; throwOnError: true }): Promise<unknown>
patch(input: { url: string; body: unknown; throwOnError: true }): Promise<unknown>
}
session: { get(input: { path: { id: string }; throwOnError: true }): Promise<{
data: { id: string; parentID?: string }
data: { id: string; parentID?: string; permission?: Array<{ permission: string; pattern: string; action: string }> }
}> }
}
}, modules?: {
createServer: typeof import('node:http').createServer
randomBytes: typeof import('node:crypto').randomBytes
modelBridgeOrigin?: string
}) {
const tasks = new Map<string, { sessionId: string; callId: string }>()
const pending = new Map<string, { sessionId: string; root: string }>()
const modelMessages = new Map<string, { sessionId: string; operationId: string; workMode: 'ask' | 'execute' }>()
const modelRoute = async (sessionId: string, messageId: string) => {
const key = `${sessionId}\0${messageId}`
const previous = modelMessages.get(key)
if (previous) return previous
if (!input || !modules?.modelBridgeOrigin) return undefined
let root = sessionId
const seen = new Set<string>()
for (;;) {
if (seen.has(root)) throw new Error('OpenCode Session parent cycle')
seen.add(root)
const session = await input.client.session.get({ path: { id: root }, throwOnError: true })
if (!session.data.parentID) break
root = session.data.parentID
}
const response = await fetch(`${modules.modelBridgeOrigin}/session?sessionId=${encodeURIComponent(root)}`)
if (!response.ok) throw new Error('GoodBuddy model operation is no longer active')
const { operationId, workMode } = await response.json() as { operationId: string; workMode: 'ask' | 'execute' }
const route = { sessionId: root, operationId, workMode }
modelMessages.set(key, route)
return route
}
let endpoint: string | undefined
let closed = false
let closeServer: (() => void) | undefined
@@ -82,8 +106,44 @@ async function plugin(input?: {
closed = true
pending.clear()
tasks.clear()
modelMessages.clear()
closeServer?.()
},
'chat.message': async (
request: { sessionID: string },
output: { message: { id: string } }
) => {
const route = await modelRoute(request.sessionID, output.message.id)
if (!route) return
// Native ACP only forwards permissions for its registered root Sessions.
// Use native Session rules so child tools use Execute and Ask's
// unavailable tools are omitted instead of interrupting the prompt.
const permission = route.workMode === 'execute'
? [{ permission: '*', pattern: '*', action: 'allow' }]
: [
{ permission: '*', pattern: '*', action: 'deny' },
...['read', 'glob', 'grep', 'list', 'lsp', 'webfetch', 'websearch', 'codesearch', 'question', 'external_directory']
.map(permission => ({ permission, pattern: '*', action: 'allow' }))
]
if (route.sessionId !== request.sessionID) {
const child = await input!.client.session.get({ path: { id: request.sessionID }, throwOnError: true })
// OpenCode copies parent denies, not parent allows, into children.
// Keep its explicit subagent restrictions after the request rules.
permission.push(...(child.data.permission ?? []).filter(rule => rule.permission !== '*'))
}
await input!.client._client.patch({
url: `/session/${encodeURIComponent(request.sessionID)}`, body: { permission }, throwOnError: true
})
},
'chat.headers': async (
request: { sessionID: string; message: { id: string } },
output: { headers: Record<string, string> }
) => {
const route = await modelRoute(request.sessionID, request.message.id)
if (!route) return
output.headers['x-goodbuddy-session'] = route.sessionId
output.headers['x-goodbuddy-operation'] = route.operationId
},
event: async ({ event }: { event: {
type: string
properties: {
@@ -105,6 +165,9 @@ async function plugin(input?: {
const properties = event.properties
if (closed) return
if (event.type === 'session.status' && properties.status?.type === 'idle' && properties.sessionID) {
for (const [key, route] of modelMessages) {
if (route.sessionId === properties.sessionID) modelMessages.delete(key)
}
for (const [id, question] of pending) {
if (question.root !== properties.sessionID && question.sessionId !== properties.sessionID) continue
pending.delete(id)
src/agent-daemon/remote-question-binary.test.ts
+2 -2
View File
@@ -74,7 +74,7 @@ it.skipIf(!existsSync(binary)).each(['parent', 'child', 'reject', 'cancel'] as c
controllerId: 'controller', preparationDigest: `sha256:${'a'.repeat(64)}`, promptSequence: 0 })
let complete = false
const owner = new AgentOwnedAcpPrompt({ bindingId: 'binding', controllerId: 'controller',
workspaceDirectory: root, workMode: 'execute', transcript,
workspaceDirectory: root, transcript,
process: {
writeStdin: async (data: Uint8Array) => { child.stdin.write(data) },
subscribeOutput: (listener: (event: { stream: string; data: Uint8Array }) => void) => {
@@ -88,7 +88,7 @@ it.skipIf(!existsSync(binary)).each(['parent', 'child', 'reject', 'cancel'] as c
})
try {
await owner.start({ bindingId: 'binding', operationId: 'operation', requestId: 'operation',
prompt: [{ type: 'text', text: 'Ask the native question tool for the decision.' }] })
prompt: [{ type: 'text', text: 'Ask the native question tool for the decision.' }] }, 'execute')
let questionId: string | undefined
await expect.poll(async () => {
const page = transcript.page({ bindingId: 'binding', operationId: 'operation',
src/agent-daemon/runtime-acp-backend.test.ts
+134 -2
View File
@@ -1,4 +1,7 @@
import { describe, expect, it, vi } from 'vitest'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import {
type AcpJournalCursor,
type RemotePromptOperationPreparation,
@@ -7,7 +10,7 @@ import {
import type { RemoteRuntimeBundleManifest } from '../shared/remote-runtime-launch-contracts'
import type { ControllerLease } from './controller-registry'
import { AgentModelGatewayError } from './agent-model-gateway'
import { EventJournalCapacityError } from './event-journal'
import { EventJournal, EventJournalCapacityError } from './event-journal'
import type { ProtocolMethodContext } from './protocol-server'
import type {
ModelBridgeBrokerDispatch,
@@ -280,6 +283,8 @@ function harness(input: {
uniqueProcesses?: boolean
outputGate?: Promise<void>
agentOwned?: boolean
shareOwnedProcesses?: boolean
journal?: RuntimeAcpJournal
modelGateway?: {
dispatch: ReturnType<typeof vi.fn>
finalizePrompt: ReturnType<typeof vi.fn>
@@ -319,7 +324,9 @@ function harness(input: {
}
: undefined
const backend = new RuntimeAcpBackend({
journal,
shareOwnedProcesses: input.shareOwnedProcesses,
sharedProcessIdleMs: 20,
journal: input.journal ?? journal,
resolveRuntimeBundle: vi.fn(async () => resolved),
loadRegisteredRuntimeBundle: vi.fn(async () => verifiedBundle()),
resolveWorkspace: vi.fn(async (preparation) => ({
@@ -444,6 +451,66 @@ function harness(input: {
}
describe('RuntimeAcpBackend', () => {
it.each([false, true])('allows mode changes only on a shared Session (shared=%s)', async (shared) => {
const fixture = harness({ agentOwned: true, shareOwnedProcesses: shared, workMode: 'execute' })
try {
await open(fixture)
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation())
await invoke(fixture, 'runtime/completePrompt', {
bindingId: 'binding-1', operationId: 'request-1', requestId: 'request-1'
})
const next = fixture.preparation({
operationId: 'request-2', requestId: 'request-2', promptSequence: 1, workMode: 'ask'
})
if (shared) {
await expect(invoke(fixture, 'runtime/preparePrompt', next)).resolves.toMatchObject({ workMode: 'ask' })
await invoke(fixture, 'runtime/completePrompt', {
bindingId: 'binding-1', operationId: 'request-2', requestId: 'request-2'
})
await expect(invoke(fixture, 'runtime/preparePrompt', fixture.preparation({
operationId: 'request-3', requestId: 'request-3', promptSequence: 2, workMode: 'execute'
}))).resolves.toMatchObject({ workMode: 'execute' })
} else {
await expect(invoke(fixture, 'runtime/preparePrompt', next)).rejects.toThrow('work mode cannot change')
}
expect(fixture.launches).toHaveLength(1)
} finally {
await fixture.backend.dispose()
}
})
it('shares Agent-owned workspaces and modes, retaining peers and reclaiming only an unused process', async () => {
const fixture = harness({ agentOwned: true, shareOwnedProcesses: true })
const channels: Array<{ bindingId: string; channelId: string; channelEpoch: string }> = []
try {
for (let index = 0; index < 4; index++) {
const bindingId = `shared-${index}`
const workspaceIdentity = `workspace-${index % 2}`
const channel = await invoke(fixture, 'runtime/openAcpChannel', {
...fixture.openRequest, bindingId, workspaceIdentity
}) as (typeof channels)[number]
channels.push({ bindingId, channelId: channel.channelId, channelEpoch: channel.channelEpoch })
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({
bindingId, workspaceIdentity, channelEpoch: channel.channelEpoch,
workMode: index % 2 ? 'execute' : 'ask',
operationId: `operation-${index}`, requestId: `operation-${index}`
}))
}
expect(fixture.launches).toHaveLength(1)
for (const channel of channels.slice(0, 3)) {
await invoke(fixture, 'runtime/closeAcpChannel', {
...channel, reason: 'released'
})
}
await new Promise(resolve => setTimeout(resolve, 40))
expect(fixture.process.stops).toEqual([])
await invoke(fixture, 'runtime/closeAcpChannel', { ...channels[3], reason: 'released' })
await vi.waitFor(() => expect(fixture.process.stops).toEqual(['binding-closed']))
} finally {
await fixture.backend.dispose()
}
})
it('rejects Agent-first unknown blob channels before prompt authority exists', async () => {
const fixture = harness()
await expect(
@@ -1536,6 +1603,71 @@ describe('RuntimeAcpBackend', () => {
).rejects.toMatchObject({ code: 'stale-controller' })
})
it('resumes shared semantic bindings with no binary output in the real journal', async () => {
const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-empty-acp-replay-'))
const journal = new EventJournal(join(directory, 'events.sqlite'))
const fixture = harness({
agentOwned: true,
shareOwnedProcesses: true,
journal
})
const channels: Array<{
bindingId: string
channelId: string
channelEpoch: string
}> = []
try {
for (let index = 0; index < 4; index++) {
const bindingId = `empty-${index}`
const workspaceIdentity = `workspace-${index % 2}`
const channel = await invoke(fixture, 'runtime/openAcpChannel', {
...fixture.openRequest, bindingId, workspaceIdentity
}) as (typeof channels)[number]
channels.push({
bindingId,
channelId: channel.channelId,
channelEpoch: channel.channelEpoch
})
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({
bindingId, workspaceIdentity,
channelEpoch: channel.channelEpoch,
operationId: `operation-${index}`,
requestId: `operation-${index}`,
workMode: index % 2 ? 'execute' : 'ask'
}))
expect(journal.getAcpCursor(
bindingId, channel.channelEpoch, 'runtime-to-main'
)).toBeUndefined()
}
fixture.context.abort.abort()
await new Promise<void>(resolve => setImmediate(resolve))
const resumed = protocolContext({
generation: 2,
connectionId: 'connection-2'
})
for (const channel of channels) {
await fixture.backend.methods['runtime/resumeAcpChannel']!(
channel, resumed
)
await expect(fixture.backend.methods['runtime/replayAcpChannel']!(
{ ...channel, acknowledgedSequence: '1' }, resumed
)).rejects.toThrow('ACK exceeds durable output')
await expect(fixture.backend.methods['runtime/replayAcpChannel']!(
{ ...channel, acknowledgedSequence: '0' }, resumed
)).resolves.toMatchObject({
...channel, replayedThroughSequence: '0', live: true
})
}
expect(fixture.launches).toHaveLength(1)
expect(fixture.process.stops).toEqual([])
expect(fixture.process.writes).toEqual([])
} finally {
await fixture.backend.dispose()
journal.close()
await rm(directory, { recursive: true, force: true })
}
})
it('fails closed on supervised process identity conflict', async () => {
const fixture = harness()
await open(fixture)
src/agent-daemon/runtime-acp-backend.ts
+170 -17
View File
@@ -59,6 +59,7 @@ import type {
ProtocolMethodHandler
} from './protocol-server'
import { AgentOwnedAcpPrompt } from './agent-owned-acp-prompt'
import { AgentAcpConnection } from './agent-acp-connection'
import type { SemanticPromptStore } from './semantic-prompt-store'
import {
AgentModelGatewayError,
@@ -83,6 +84,7 @@ const DEFAULT_MAXIMUM_PENDING_OUTPUT_BYTES_GLOBAL =
const DEFAULT_MAXIMUM_OPERATIONS_PER_BINDING = 1_000
const DEFAULT_DISPOSE_TIMEOUT_MS = 10_000
const OWNED_PROMPT_CANCEL_GRACE_MS = 1_000
const SHARED_PROCESS_IDLE_MS = 60_000
const OWNED_PROMPT_START_TIMEOUT_MS = 2 * 60_000
const ACP_JOURNAL_RETRY_INITIAL_MILLISECONDS = 25
const ACP_JOURNAL_RETRY_MAXIMUM_MILLISECONDS = 1_000
@@ -228,6 +230,7 @@ export type RuntimeAcpProcessLaunch = {
workMode: 'ask' | 'execute'
deadlineAt: string
budget: RemotePromptOperationPreparation['budget']
sharedSessions?: boolean
modelBridge?: {
socketPath: string
policy: ModelBridgePolicy
@@ -249,6 +252,8 @@ export type RuntimeAcpBackendLimits = {
}
export type RuntimeAcpBackendOptions = {
shareOwnedProcesses?: boolean
sharedProcessIdleMs?: number
journal: RuntimeAcpJournal
resolveRuntimeBundle(
runtimeId: string,
@@ -298,6 +303,7 @@ type PreparedOperation = {
budget: RemotePromptOperationPreparation['budget']
completion?: z.infer<typeof acpCompletePromptResultSchema>
terminalState?: 'completed' | 'failed' | 'cancelled' | 'interrupted'
processTree?: 'running' | 'empty'
modelProfile?: AgentPromptModelProfile
promptSequence: number
ownedPromptStarted?: boolean
@@ -314,6 +320,7 @@ type BindingState = {
state: 'open' | 'running' | 'stopping' | 'closed' | 'outcome-unknown'
workMode?: 'ask' | 'execute'
process?: RuntimeAcpProcessOwner
sharedProcess?: SharedProcess
unsubscribeOutput?: () => void
unsubscribeExit?: () => void
stopRequested: boolean
@@ -338,12 +345,23 @@ type BindingState = {
poisoned: boolean
}
type SharedProcess = {
key: string
process: RuntimeAcpProcessOwner
transport: AgentAcpConnection
bindings: Set<BindingState>
unsubscribeExit?: () => void
idleTimer?: NodeJS.Timeout
exited: boolean
}
export class RuntimeAcpBackend {
readonly methods: Readonly<Record<string, ProtocolMethodHandler>>
readonly #options: RuntimeAcpBackendOptions
readonly #limits: RuntimeAcpBackendLimits
readonly #bindings = new Map<string, BindingState>()
readonly #processIdentityBindings = new Map<string, string>()
readonly #sharedProcesses = new Map<string, SharedProcess>()
readonly #boundConnectionSignals = new WeakSet<AbortSignal>()
#nextChannelEpoch: bigint
#lifetimeBindings = 0
@@ -636,6 +654,14 @@ export class RuntimeAcpBackend {
result.status === 'rejected'
)
.map((result) => result.reason)
const sharedResults = await Promise.allSettled(
[...this.#sharedProcesses.values()].map(group =>
this.#stopSharedProcess(group, 'binding-closed')
)
)
for (const result of sharedResults) {
if (result.status === 'rejected') failures.push(result.reason)
}
if (failures.length > 0) {
throw new AggregateError(
failures,
@@ -930,11 +956,15 @@ export class RuntimeAcpBackend {
'identity'
)
}
this.#options.journal.acknowledgeAcpFromMain?.({
bindingId: binding.request.bindingId,
channelEpoch: binding.openResult.channelEpoch,
sequence: request.acknowledgedSequence
})
// Agent-owned prompts may have semantic events but no binary journal.
// In that case the validated zero cursor has nothing to acknowledge.
if (initialCursor !== undefined) {
this.#options.journal.acknowledgeAcpFromMain?.({
bindingId: binding.request.bindingId,
channelEpoch: binding.openResult.channelEpoch,
sequence: request.acknowledgedSequence
})
}
if (this.#options.journal.replayAcpFrames === undefined) {
throw new RuntimeAcpBackendError(
'Durable ACP replay is unavailable',
@@ -1011,7 +1041,8 @@ export class RuntimeAcpBackend {
}
if (
binding.workMode !== undefined &&
binding.workMode !== preparation.workMode
binding.workMode !== preparation.workMode &&
binding.sharedProcess === undefined
) {
throw new RuntimeAcpBackendError(
'Runtime process work mode cannot change',
@@ -1124,12 +1155,22 @@ export class RuntimeAcpBackend {
if (binding.process === undefined) {
let process: RuntimeAcpProcessOwner
const sharedKey =
this.#options.shareOwnedProcesses &&
preparation.modelProfile && this.#options.modelGateway &&
this.#options.semanticPrompts && binding.modelBridgePolicy
? canonicalJson([
verified.manifest.bundleDigest,
verified.manifest.adapterDigest,
binding.modelBridgePolicy
])
: undefined
this.#options.diagnostics?.tryRecord('runtime.starting', {
runtimeId: verified.manifest.runtimeId,
workMode: preparation.workMode
})
try {
process = await this.#options.launchProcess({
const launch: RuntimeAcpProcessLaunch = {
manifest: verified.manifest,
bundle: verified,
workspace,
@@ -1137,6 +1178,7 @@ export class RuntimeAcpBackend {
workMode: preparation.workMode,
deadlineAt: preparation.deadlineAt,
budget: preparation.budget,
...(sharedKey === undefined ? {} : { sharedSessions: true }),
...(binding.modelBridgePolicy === undefined ||
binding.modelBridgeBroker === undefined
? {}
@@ -1146,7 +1188,15 @@ export class RuntimeAcpBackend {
policy: binding.modelBridgePolicy
}
})
})
}
if (sharedKey !== undefined) {
const group = await this.#acquireSharedProcess(sharedKey, launch)
binding.sharedProcess = group
group.bindings.add(binding)
process = group.process
} else {
process = await this.#options.launchProcess(launch)
}
assertProcessIdentity(process.identity)
} catch (error) {
this.#options.diagnostics?.tryRecord('runtime.start.failed', {
@@ -1166,7 +1216,7 @@ export class RuntimeAcpBackend {
const identityOwner = this.#processIdentityBindings.get(processKey)
if (
identityOwner !== undefined &&
identityOwner !== binding.request.bindingId
identityOwner !== (binding.sharedProcess?.key ?? binding.request.bindingId)
) {
await Promise.resolve(
process.stop({
@@ -1182,7 +1232,7 @@ export class RuntimeAcpBackend {
}
this.#processIdentityBindings.set(
processKey,
binding.request.bindingId
binding.sharedProcess?.key ?? binding.request.bindingId
)
binding.process = process
binding.workspaceDirectory = workspace.workspaceDirectory
@@ -1208,7 +1258,7 @@ export class RuntimeAcpBackend {
binding.unsubscribeOutput = unsubscribe
}
}
const unsubscribeExit = process.subscribeExit?.(() => {
const unsubscribeExit = binding.sharedProcess ? undefined : process.subscribeExit?.(() => {
void this.#enqueueControl(async () => {
await this.#handleProcessExit(binding)
}).catch(() => undefined)
@@ -1230,7 +1280,7 @@ export class RuntimeAcpBackend {
runtimeId: verified.manifest.runtimeId,
workMode: preparation.workMode
})
} else {
} else if (!binding.sharedProcess) {
try {
await binding.process.beginPrompt({
deadlineAt: preparation.deadlineAt,
@@ -1246,6 +1296,7 @@ export class RuntimeAcpBackend {
)
}
}
binding.workMode = preparation.workMode
binding.inputBytes = 0
if (!binding.operations.has(preparation.operationId)) {
binding.operations.set(preparation.operationId, {
@@ -1389,12 +1440,14 @@ export class RuntimeAcpBackend {
'conflict'
)
}
if (Buffer.byteLength(canonicalJson(request.prompt)) > prepared.budget.maximumInputBytes) {
throw new RuntimeAcpBackendError('Prompt input quota reached', 'capacity')
}
if (binding.ownedAcp === undefined) {
binding.ownedAcp = new AgentOwnedAcpPrompt({
bindingId: binding.request.bindingId,
controllerId: binding.controllerId,
workspaceDirectory: binding.workspaceDirectory,
workMode: binding.workMode!,
...(binding.modelBridgePolicy === undefined
? {}
: {
@@ -1404,6 +1457,14 @@ export class RuntimeAcpBackend {
)
}),
process: binding.process,
...(binding.sharedProcess ? {
transport: binding.sharedProcess.transport,
prepareSession: async (sessionId: string, operationId: string, workMode: 'ask' | 'execute') => {
await binding.sharedProcess!.transport.setModelRoute(
sessionId, operationId, binding.modelBridgeBroker!.socketPath, workMode
)
}
} : {}),
transcript: this.#options.semanticPrompts,
completePrompt: async (operationId, status) => {
await this.#enqueueControl(async () => {
@@ -1418,7 +1479,7 @@ export class RuntimeAcpBackend {
binding.poisoned ? 'outcome-unknown' : status
})
}
const result = await binding.ownedAcp.start(request)
const result = await binding.ownedAcp.start(request, prepared.acceptance.workMode)
prepared.ownedPromptStarted = true
if (binding.ownedPromptStartTimer !== undefined) {
clearTimeout(binding.ownedPromptStartTimer)
@@ -1532,7 +1593,7 @@ export class RuntimeAcpBackend {
}
await this.#closeModelBridge(binding, true)
try {
await binding.process.completePrompt()
if (!binding.sharedProcess) await binding.process.completePrompt()
} catch {
binding.state = 'outcome-unknown'
prepared.modelProfile = undefined
@@ -1556,6 +1617,7 @@ export class RuntimeAcpBackend {
: status === 'failed'
? 'failed'
: 'completed'
prepared.processTree = 'running'
if (status === 'completed') {
prepared.completion = acpCompletePromptResultSchema.parse({
bindingId: binding.request.bindingId,
@@ -1759,6 +1821,11 @@ export class RuntimeAcpBackend {
binding.modelBridgeBroker = undefined
binding.modelBridgeClient = undefined
let brokerError: unknown
if (binding.sharedProcess && binding.ownedAcp?.sessionId && binding.activeOperationId) {
await binding.sharedProcess.transport.setModelRoute(
binding.ownedAcp.sessionId, binding.activeOperationId
).catch(() => undefined)
}
try {
await broker?.close()
} catch (error) {
@@ -1902,7 +1969,7 @@ export class RuntimeAcpBackend {
status: 'terminal',
terminalState: prepared.terminalState,
processTree:
prepared.terminalState === 'completed' ? 'running' : 'empty'
prepared.processTree ?? (prepared.terminalState === 'completed' ? 'running' : 'empty')
}
}
if (binding.activeOperationId !== operation.operationId) {
@@ -2113,6 +2180,25 @@ export class RuntimeAcpBackend {
this.#finishBinding(binding, 'interrupted')
return
}
if (binding.sharedProcess) {
const group = binding.sharedProcess
const prepared = binding.activeOperationId
? binding.operations.get(binding.activeOperationId) : undefined
try {
await rejectAfter(
binding.ownedAcp?.releaseSession() ?? Promise.resolve(),
OWNED_PROMPT_CANCEL_GRACE_MS * 3,
'Shared Runtime Session release timed out'
)
if (prepared) prepared.processTree = group.exited ? 'empty' : 'running'
this.#finishBinding(binding, reason === 'user-cancelled' ? 'cancelled' : 'interrupted')
} catch {
// A non-responsive native control plane affects the whole process.
// Ordinary Session cancellation never reaches this fallback.
await this.#stopSharedProcess(group, reason)
}
return
}
binding.state = 'stopping'
if (!binding.stopRequested) {
binding.stopRequested = true
@@ -2277,7 +2363,12 @@ export class RuntimeAcpBackend {
}
#finishProcess(binding: BindingState): void {
if (binding.process !== undefined) {
const group = binding.sharedProcess
if (group) {
group.bindings.delete(binding)
binding.sharedProcess = undefined
this.#scheduleSharedIdle(group)
} else if (binding.process !== undefined) {
this.#processIdentityBindings.delete(
processIdentityKey(binding.process.identity)
)
@@ -2305,6 +2396,68 @@ export class RuntimeAcpBackend {
}
}
async #acquireSharedProcess(key: string, launch: RuntimeAcpProcessLaunch): Promise<SharedProcess> {
const existing = this.#sharedProcesses.get(key)
if (existing && !existing.exited) {
if (existing.idleTimer) clearTimeout(existing.idleTimer)
existing.idleTimer = undefined
return existing
}
const process = await this.#options.launchProcess(launch)
assertProcessIdentity(process.identity)
const group: SharedProcess = {
key, process, transport: new AgentAcpConnection(process),
bindings: new Set(), exited: false
}
group.unsubscribeExit = process.subscribeExit?.(() => {
void this.#enqueueControl(() => this.#handleSharedExit(group)).catch(() => undefined)
}) ?? undefined
this.#sharedProcesses.set(key, group)
return group
}
#scheduleSharedIdle(group: SharedProcess): void {
if (group.bindings.size || group.exited || this.#disposed || group.idleTimer) return
group.idleTimer = setTimeout(() => {
group.idleTimer = undefined
void this.#enqueueControl(async () => {
if (!group.bindings.size) await this.#stopSharedProcess(group, 'binding-closed')
}).catch(() => undefined)
}, this.#options.sharedProcessIdleMs ?? SHARED_PROCESS_IDLE_MS)
group.idleTimer.unref?.()
}
async #stopSharedProcess(group: SharedProcess, reason: RuntimeAcpProcessStopReason): Promise<void> {
if (group.exited) return
if (group.idleTimer) clearTimeout(group.idleTimer)
group.idleTimer = undefined
await group.process.stop({
reason, deadlineAt: new Date(this.#now() + 10_000).toISOString()
})
const result = await group.process.reconcile()
if (!sameProcessIdentity(result.identity, group.process.identity) || result.processTree !== 'empty') {
throw new RuntimeAcpBackendError('Shared Runtime process stop did not reconcile', 'process')
}
await this.#handleSharedExit(group)
}
async #handleSharedExit(group: SharedProcess): Promise<void> {
if (group.exited) return
group.exited = true
if (group.idleTimer) clearTimeout(group.idleTimer)
group.unsubscribeExit?.()
group.transport.dispose()
if (this.#sharedProcesses.get(group.key) === group) this.#sharedProcesses.delete(group.key)
this.#processIdentityBindings.delete(processIdentityKey(group.process.identity))
for (const binding of [...group.bindings]) {
if (binding.activeOperationId) {
const prepared = binding.operations.get(binding.activeOperationId)
if (prepared) prepared.processTree = 'empty'
}
await this.#handleProcessExit(binding)
}
}
#scheduleDeadline(binding: BindingState, deadlineAt: string): void {
if (binding.deadlineTimer !== undefined) {
clearTimeout(binding.deadlineTimer)
src/agent-daemon/runtime-composition.ts
+8 -3
View File
@@ -245,6 +245,7 @@ export async function createProductionRuntimeProtocol(
stateDirectory
)
backend = new RuntimeAcpBackend({
shareOwnedProcesses: true,
journal: options.events,
resolveRuntimeBundle: (runtimeId, bundleDigest) =>
currentRegistry().resolve(
@@ -254,7 +255,9 @@ export async function createProductionRuntimeProtocol(
),
loadRegisteredRuntimeBundle: async (resolved) => {
return await loadRegistered(
resolved.entry,
// This composition's resolver above returns the complete registry
// entry; the backend itself only needs its public identity fields.
resolved.entry as RuntimeRegistryEntry,
resolved.bundleDirectory
)
},
@@ -283,7 +286,8 @@ export async function createProductionRuntimeProtocol(
bridgeDirectory:
launch.workspace.bridgeDirectory!,
socketPath: launch.modelBridge.socketPath,
policy: launch.modelBridge.policy
policy: launch.modelBridge.policy,
sharedSessions: launch.sharedSessions
}
})
},
@@ -294,7 +298,8 @@ export async function createProductionRuntimeProtocol(
installationId,
registry: ownerRegistry,
deadlineAt: launch.deadlineAt,
maximumInputBytes: launch.budget.maximumInputBytes
maximumInputBytes: launch.budget.maximumInputBytes,
sharedSessions: launch.sharedSessions
}),
outputSink: options.outputSink,
semanticPrompts,