mesalogo
7adae97baa
Wait for model delivery acknowledgements before completing remote prompts and preserve completed outcomes when idle Runtime cleanup fails. Explicit messages after an unknown outcome now open a fresh binding with history instead of reusing the failed session; recovery remains attach-only. Run obsolete payload cleanup through the current Agent after activation, retaining history and unresolved process ownership. Live legacy daemons and shared Runtime versions remain deferred. Includes focused regressions, local OpenCode simulation, and isolated Linux validation evidence. Typecheck and lint passed; the full test run timed out. Release note: Remote conversations can continue after an uncertain result, and completed answers no longer fail solely because Runtime cleanup fails. Updates remove verified inactive Agent payloads and owned orphan processes while retaining active or uncertain installations and conversation history.
2148 lines
68 KiB
TypeScript
2148 lines
68 KiB
TypeScript
import { describe, expect, it, vi } from 'vitest'
|
|
import { Client } from '@modelcontextprotocol/sdk/client/index.js'
|
|
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'
|
|
import { decodeRemoteImageToolMessage, encodeRemoteImageToolMessage, remoteImageToolCallSchema } from '../shared/remote-image-tool-contracts'
|
|
import type { AgentFrame } from '../shared/agent-protocol'
|
|
import { mkdtemp, rm } from 'node:fs/promises'
|
|
import { tmpdir } from 'node:os'
|
|
import { join } from 'node:path'
|
|
import {
|
|
type AcpJournalCursor,
|
|
type RemotePromptOperationPreparation,
|
|
UNBOUNDED_REMOTE_PROMPT_DEADLINE
|
|
} from '../shared/remote-agent-contracts'
|
|
import type { RemoteRuntimeBundleManifest } from '../shared/remote-runtime-launch-contracts'
|
|
import type { ControllerLease } from './controller-registry'
|
|
import { AgentModelGatewayError } from './agent-model-gateway'
|
|
import { EventJournal, EventJournalCapacityError } from './event-journal'
|
|
import { SemanticPromptStore } from './semantic-prompt-store'
|
|
import { AgentAcpConnection } from './agent-acp-connection'
|
|
import type { ProtocolMethodContext } from './protocol-server'
|
|
import type {
|
|
ModelBridgeBrokerDispatch,
|
|
ModelBridgeBrokerServer
|
|
} from './model-bridge-broker'
|
|
import {
|
|
RuntimeAcpBackend,
|
|
type RuntimeAcpJournal,
|
|
type RuntimeAcpProcessIdentity,
|
|
type RuntimeAcpProcessOutput,
|
|
type RuntimeAcpProcessOwner,
|
|
type RuntimeAcpProcessReconciliation,
|
|
type RuntimeAcpResolvedBundle,
|
|
type RuntimeAcpVerifiedBundle
|
|
} from './runtime-acp-backend'
|
|
|
|
const digest = (character: string): string =>
|
|
`sha256:${character.repeat(64)}`
|
|
|
|
class MemoryAcpJournal implements RuntimeAcpJournal {
|
|
readonly frames: Array<{
|
|
controllerId: string
|
|
bindingId: string
|
|
channelEpoch: string
|
|
direction: 'main-to-runtime' | 'runtime-to-main'
|
|
sequence: string
|
|
payload: Uint8Array
|
|
}> = []
|
|
readonly delivered: string[] = []
|
|
readonly retired: Array<{
|
|
controllerId: string
|
|
bindingId: string
|
|
channelEpoch: string
|
|
}> = []
|
|
mainAckSequence = '0'
|
|
failAppend = false
|
|
capacityBlocked = false
|
|
appendAttempts = 0
|
|
|
|
appendAcpFrame(input: (typeof this.frames)[number]): { created: boolean } {
|
|
this.appendAttempts += 1
|
|
if (this.capacityBlocked) {
|
|
throw new EventJournalCapacityError(
|
|
'journal capacity',
|
|
'stream',
|
|
false
|
|
)
|
|
}
|
|
if (this.failAppend) {
|
|
throw new Error('journal capacity')
|
|
}
|
|
const existing = this.frames.find(
|
|
(frame) =>
|
|
frame.bindingId === input.bindingId &&
|
|
frame.channelEpoch === input.channelEpoch &&
|
|
frame.direction === input.direction &&
|
|
frame.sequence === input.sequence
|
|
)
|
|
if (existing !== undefined) {
|
|
if (
|
|
Buffer.compare(
|
|
Buffer.from(existing.payload),
|
|
Buffer.from(input.payload)
|
|
) !== 0
|
|
) {
|
|
throw new Error('sequence conflict')
|
|
}
|
|
return { created: false }
|
|
}
|
|
const latest = this.frames
|
|
.filter(
|
|
(frame) =>
|
|
frame.bindingId === input.bindingId &&
|
|
frame.channelEpoch === input.channelEpoch &&
|
|
frame.direction === input.direction
|
|
)
|
|
.at(-1)
|
|
if (BigInt(input.sequence) !== BigInt(latest?.sequence ?? '0') + 1n) {
|
|
throw new Error('non-contiguous sequence')
|
|
}
|
|
this.frames.push({ ...input, payload: input.payload.slice() })
|
|
return { created: true }
|
|
}
|
|
|
|
markAcpDelivered(input: {
|
|
bindingId: string
|
|
channelEpoch: string
|
|
sequence: string
|
|
}): void {
|
|
this.delivered.push(input.sequence)
|
|
}
|
|
|
|
acknowledgeAcpFromMain(input: { sequence: string }): void {
|
|
this.mainAckSequence = input.sequence
|
|
}
|
|
|
|
retireAcpBinding(input: {
|
|
controllerId: string
|
|
bindingId: string
|
|
channelEpoch: string
|
|
}): void {
|
|
this.retired.push(input)
|
|
}
|
|
|
|
replayAcpFrames(input: {
|
|
bindingId: string
|
|
channelEpoch: string
|
|
direction: 'runtime-to-main'
|
|
afterSequence: string
|
|
limit?: number
|
|
}) {
|
|
return this.frames
|
|
.filter(
|
|
(frame) =>
|
|
frame.bindingId === input.bindingId &&
|
|
frame.channelEpoch === input.channelEpoch &&
|
|
frame.direction === input.direction &&
|
|
BigInt(frame.sequence) > BigInt(input.afterSequence)
|
|
)
|
|
.slice(0, input.limit)
|
|
.map((frame) => ({
|
|
sequence: frame.sequence,
|
|
payload: frame.payload.slice(),
|
|
createdAt: 1
|
|
}))
|
|
}
|
|
|
|
getAcpCursor(
|
|
bindingId: string,
|
|
channelEpoch: string,
|
|
direction: 'main-to-runtime' | 'runtime-to-main'
|
|
): AcpJournalCursor | undefined {
|
|
const frames = this.frames.filter(
|
|
(frame) =>
|
|
frame.bindingId === bindingId &&
|
|
frame.channelEpoch === channelEpoch &&
|
|
frame.direction === direction
|
|
)
|
|
if (frames.length === 0) {
|
|
return undefined
|
|
}
|
|
const journaledSequence = frames.at(-1)!.sequence
|
|
return {
|
|
bindingId,
|
|
channelEpoch,
|
|
direction,
|
|
journaledSequence,
|
|
deliveredSequence:
|
|
direction === 'main-to-runtime'
|
|
? this.delivered.at(-1) ?? '0'
|
|
: '0',
|
|
mainAckSequence:
|
|
direction === 'runtime-to-main'
|
|
? this.mainAckSequence
|
|
: '0'
|
|
}
|
|
}
|
|
}
|
|
|
|
class FakeProcess implements RuntimeAcpProcessOwner {
|
|
readonly identity: RuntimeAcpProcessIdentity
|
|
readonly writes: Uint8Array[] = []
|
|
readonly stops: string[] = []
|
|
readonly prompts: Array<{
|
|
deadlineAt: string
|
|
maximumInputBytes: number
|
|
}> = []
|
|
completions = 0
|
|
onWrite?: (payload: Uint8Array) => void
|
|
reconciliation: RuntimeAcpProcessReconciliation
|
|
#listener?: (
|
|
output: RuntimeAcpProcessOutput
|
|
) => void | Promise<void>
|
|
readonly #exitListeners = new Set<
|
|
() => void | Promise<void>
|
|
>()
|
|
|
|
constructor(identity = '1') {
|
|
this.identity = {
|
|
launchId: `launch-${identity}`,
|
|
processId: `process-${identity}`,
|
|
supervisorIdentityDigest: digest(identity)
|
|
}
|
|
this.reconciliation = {
|
|
identity: this.identity,
|
|
state: 'running',
|
|
processTree: 'running'
|
|
}
|
|
}
|
|
|
|
beginPrompt(input: {
|
|
deadlineAt: string
|
|
maximumInputBytes: number
|
|
}): void {
|
|
this.prompts.push(input)
|
|
this.reconciliation = {
|
|
identity: this.identity,
|
|
state: 'running',
|
|
processTree: 'running'
|
|
}
|
|
}
|
|
|
|
completePrompt(): void {
|
|
this.completions += 1
|
|
}
|
|
|
|
writeStdin(payload: Uint8Array): void {
|
|
this.writes.push(payload.slice())
|
|
this.onWrite?.(payload)
|
|
}
|
|
|
|
stop(input: { reason: string }): void {
|
|
this.stops.push(input.reason)
|
|
this.reconciliation = {
|
|
identity: this.identity,
|
|
state: 'cancelled',
|
|
processTree: 'empty'
|
|
}
|
|
}
|
|
|
|
reconcile(): RuntimeAcpProcessReconciliation {
|
|
return this.reconciliation
|
|
}
|
|
|
|
subscribeOutput(
|
|
listener: (
|
|
output: RuntimeAcpProcessOutput
|
|
) => void | Promise<void>
|
|
): () => void {
|
|
this.#listener = listener
|
|
return () => {
|
|
this.#listener = undefined
|
|
}
|
|
}
|
|
|
|
subscribeExit(listener: () => void | Promise<void>): () => void {
|
|
this.#exitListeners.add(listener)
|
|
return () => this.#exitListeners.delete(listener)
|
|
}
|
|
|
|
async emit(
|
|
stream: 'stdout' | 'stderr',
|
|
value: string
|
|
): Promise<void> {
|
|
await this.#listener?.({
|
|
stream,
|
|
data: Buffer.from(value)
|
|
})
|
|
}
|
|
|
|
async emitExit(
|
|
state: RuntimeAcpProcessReconciliation['state'] = 'failed'
|
|
): Promise<void> {
|
|
this.reconciliation = {
|
|
identity: this.identity,
|
|
state,
|
|
processTree: 'empty'
|
|
}
|
|
for (const listener of [...this.#exitListeners]) {
|
|
await listener()
|
|
}
|
|
}
|
|
}
|
|
|
|
type Harness = ReturnType<typeof harness>
|
|
|
|
function deferred<T = void>() {
|
|
let resolve!: (value: T) => void
|
|
let reject!: (error: unknown) => void
|
|
const promise = new Promise<T>((yes, no) => { resolve = yes; reject = no })
|
|
return { promise, resolve, reject }
|
|
}
|
|
|
|
async function ownedHarness(shared: boolean) {
|
|
const root = await mkdtemp(join(tmpdir(), 'goodbuddy-terminal-race-'))
|
|
const transcript = new SemanticPromptStore(join(root, 'prompts.sqlite'))
|
|
const prompts: Array<() => Promise<void>> = []
|
|
const processes: FakeProcess[] = []
|
|
const setRoute = vi.spyOn(AgentAcpConnection.prototype, 'setModelRoute').mockResolvedValue()
|
|
const fixture = harness({
|
|
agentOwned: true, shareOwnedProcesses: shared, semanticStore: transcript, uniqueProcesses: true,
|
|
configureProcess: process => {
|
|
processes.push(process)
|
|
process.onWrite = payload => {
|
|
const request = JSON.parse(Buffer.from(payload).toString())
|
|
if (request.id === undefined) return
|
|
const reply = (result: unknown) => process.emit('stdout', `${JSON.stringify({ jsonrpc: '2.0', id: request.id, result })}\n`)
|
|
if (request.method === 'session/prompt') {
|
|
prompts.push(() => reply({ stopReason: 'end_turn' }))
|
|
return
|
|
}
|
|
let result: unknown = {}
|
|
if (request.method === 'initialize') result = { protocolVersion: 1, agentCapabilities: {} }
|
|
if (request.method === 'session/new') result = { sessionId: `session-${processes.length}` }
|
|
if (request.method === 'session/set_config_option') result = {
|
|
configOptions: [{ id: 'model', name: 'Model', type: 'select', currentValue: request.params.value, options: [] }]
|
|
}
|
|
void reply(result)
|
|
}
|
|
}
|
|
})
|
|
await open(fixture)
|
|
return {
|
|
...fixture, transcript, prompts, processes,
|
|
start: async (operationId = 'request-1', deadlineAt = UNBOUNDED_REMOTE_PROMPT_DEADLINE) => {
|
|
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({ operationId, requestId: operationId, deadlineAt }))
|
|
await invoke(fixture, 'runtime/startPrompt', {
|
|
bindingId: 'binding-1', operationId, requestId: operationId, prompt: [{ type: 'text', text: 'Finish' }]
|
|
})
|
|
},
|
|
page: (operationId = 'request-1') => transcript.page({
|
|
bindingId: 'binding-1', operationId, controllerId: fixture.context.controller.controllerId,
|
|
afterSequence: '0', limit: 10
|
|
}),
|
|
cleanup: async () => {
|
|
try { await fixture.backend.dispose() } finally {
|
|
setRoute.mockRestore()
|
|
transcript.close()
|
|
await rm(root, { recursive: true, force: true })
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
function harness(input: {
|
|
workMode?: 'ask' | 'execute'
|
|
now?: number
|
|
maximumOutputBytes?: number
|
|
bridgeCloseError?: boolean
|
|
uniqueProcesses?: boolean
|
|
outputGate?: Promise<void>
|
|
agentOwned?: boolean
|
|
shareOwnedProcesses?: boolean
|
|
semanticStore?: SemanticPromptStore
|
|
configureProcess?: (process: FakeProcess) => void
|
|
journal?: RuntimeAcpJournal
|
|
modelGateway?: {
|
|
dispatch: ReturnType<typeof vi.fn>
|
|
finalizePrompt: ReturnType<typeof vi.fn>
|
|
}
|
|
launchError?: Error
|
|
} = {}) {
|
|
const workMode = input.workMode ?? 'ask'
|
|
let now = input.now ?? 1_000
|
|
const journal = new MemoryAcpJournal()
|
|
const process = new FakeProcess()
|
|
input.configureProcess?.(process)
|
|
const outputFrames: string[] = []
|
|
const launches: Array<{
|
|
workMode: 'ask' | 'execute'
|
|
scratch: string
|
|
}> = []
|
|
const lifecycle: string[] = []
|
|
const diagnostics = {
|
|
tryRecord: vi.fn()
|
|
}
|
|
const bridgeCloses = vi.fn(async () => {
|
|
if (input.bridgeCloseError) {
|
|
throw new Error('broker close failed')
|
|
}
|
|
})
|
|
const blobSink = vi.fn<(frame: AgentFrame) => Promise<void>>(async () => undefined)
|
|
let bridgeDispatch: ModelBridgeBrokerDispatch | undefined
|
|
const launchModelBridges: unknown[] = []
|
|
const resolved = resolvedBundle()
|
|
const semanticPrompts = input.agentOwned
|
|
? {
|
|
prepare: vi.fn(() => ({ created: true })),
|
|
append: vi.fn(),
|
|
findStarted: vi.fn(),
|
|
attach: vi.fn(),
|
|
page: vi.fn(),
|
|
acknowledge: vi.fn()
|
|
}
|
|
: undefined
|
|
const backend = new RuntimeAcpBackend({
|
|
shareOwnedProcesses: input.shareOwnedProcesses,
|
|
sharedProcessIdleMs: 20,
|
|
journal: input.journal ?? journal,
|
|
resolveRuntimeBundle: vi.fn(async () => resolved),
|
|
loadRegisteredRuntimeBundle: vi.fn(async () => verifiedBundle()),
|
|
resolveWorkspace: vi.fn(async (preparation) => ({
|
|
workspaceIdentity: preparation.workspaceIdentity,
|
|
workspaceDirectory: '/workspace',
|
|
scratchDirectory: '/scratch',
|
|
bridgeDirectory: '/bridge'
|
|
})),
|
|
launchProcess: vi.fn(async (launch) => {
|
|
lifecycle.push('launch')
|
|
if (input.launchError !== undefined) {
|
|
throw input.launchError
|
|
}
|
|
launchModelBridges.push(launch.modelBridge)
|
|
launches.push({
|
|
workMode: launch.workMode,
|
|
scratch: launch.scratch
|
|
})
|
|
if (input.uniqueProcesses && launches.length > 1) {
|
|
const next = new FakeProcess(String(launches.length))
|
|
input.configureProcess?.(next)
|
|
return next
|
|
}
|
|
return process
|
|
}),
|
|
now: () => now,
|
|
diagnostics,
|
|
outputSink: vi.fn(async (frame) => {
|
|
await input.outputGate
|
|
outputFrames.push(Buffer.from(frame.payload).toString())
|
|
}),
|
|
blobSink,
|
|
createModelBridgeBroker: ({ dispatch }) => {
|
|
bridgeDispatch = dispatch
|
|
return {
|
|
socketPath: '/bridge/model-bridge.sock',
|
|
listen: vi.fn(async () => {
|
|
lifecycle.push('bridge-listen')
|
|
}),
|
|
close: bridgeCloses
|
|
} as unknown as ModelBridgeBrokerServer
|
|
},
|
|
...(semanticPrompts
|
|
? {
|
|
semanticPrompts: (input.semanticStore ?? semanticPrompts) as never,
|
|
modelGateway:
|
|
(input.modelGateway ?? {
|
|
dispatch: vi.fn(),
|
|
finalizePrompt: vi.fn()
|
|
}) as never
|
|
}
|
|
: {})
|
|
})
|
|
const context = protocolContext()
|
|
const openRequest = {
|
|
bindingId: 'binding-1',
|
|
runtimeId: 'opencode',
|
|
runtimeBundleDigest: digest('1'),
|
|
workspaceIdentity: 'workspace-identity-1'
|
|
}
|
|
|
|
return {
|
|
backend,
|
|
context,
|
|
journal,
|
|
process,
|
|
launches,
|
|
lifecycle,
|
|
diagnostics,
|
|
bridgeCloses,
|
|
blobSink,
|
|
get bridgeDispatch() {
|
|
return bridgeDispatch
|
|
},
|
|
launchModelBridges,
|
|
outputFrames,
|
|
semanticPrompts,
|
|
openRequest,
|
|
preparation: (
|
|
overrides: Partial<RemotePromptOperationPreparation> = {}
|
|
): RemotePromptOperationPreparation => ({
|
|
bindingId: openRequest.bindingId,
|
|
operationId: 'request-1',
|
|
requestId: 'request-1',
|
|
workMode,
|
|
controllerId: context.controller.controllerId,
|
|
controllerGeneration: context.controller.generation,
|
|
connectionGeneration: context.controller.generation,
|
|
channelEpoch: '1000',
|
|
hostId: 'host-1',
|
|
hostRevision: 1,
|
|
hostKeyGeneration: 1,
|
|
workspaceIdentity: openRequest.workspaceIdentity,
|
|
agentInstallationId: 'installation-1',
|
|
runtimeId: openRequest.runtimeId,
|
|
runtimeBundleDigest: openRequest.runtimeBundleDigest,
|
|
runtimeAdapterDigest: digest('2'),
|
|
modelProfile: {
|
|
profileId: 'profile-1',
|
|
modelProfileDigest: digest('3'),
|
|
provider: 'openai',
|
|
baseUrl: 'https://model.example/v1',
|
|
model: 'test-model',
|
|
protocol: 'openai-responses',
|
|
authentication: 'api-key',
|
|
apiKey: 'test-api-key',
|
|
capabilities: { imageInput: false },
|
|
limits: {
|
|
maximumOutputTokens: 4_096,
|
|
requestTimeoutMilliseconds: 60_000
|
|
}
|
|
},
|
|
promptSequence: 0,
|
|
deadlineAt: new Date(now + 60_000).toISOString(),
|
|
budget: {
|
|
maximumInputBytes: 1_024,
|
|
maximumOutputBytes: input.maximumOutputBytes ?? 1_024
|
|
},
|
|
...overrides
|
|
}),
|
|
setNow(value: number): void {
|
|
now = value
|
|
}
|
|
}
|
|
}
|
|
|
|
describe('RuntimeAcpBackend', () => {
|
|
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)
|
|
const enteredStop = deferred()
|
|
const releaseStop = deferred()
|
|
const stop = vi.spyOn(fixture.process, 'stop').mockImplementationOnce(async () => {
|
|
enteredStop.resolve()
|
|
await releaseStop.promise
|
|
if (failure === 'stop') throw new Error('idle stop failed')
|
|
})
|
|
try {
|
|
await fixture.start()
|
|
await fixture.prompts[0]!()
|
|
await enteredStop.promise
|
|
expect(fixture.page()).toMatchObject({ state: 'completed', events: [{ kind: 'prompt-terminal', payload: { status: 'completed' } }] })
|
|
releaseStop.resolve()
|
|
await expect(invoke(fixture, 'runtime/reconcilePrompt', {
|
|
bindingId: 'binding-1', operationId: 'request-1', requestId: 'request-1'
|
|
})).resolves.toMatchObject({ status: 'terminal', terminalState: 'completed' })
|
|
expect(fixture.diagnostics.tryRecord).toHaveBeenCalledWith('daemon.stop.failed', expect.objectContaining({
|
|
reason: 'idle-runtime-cleanup', error: expect.any(Error)
|
|
}))
|
|
expect(fixture.page().events).toHaveLength(1)
|
|
expect(fixture.page().state).toBe('completed')
|
|
} finally {
|
|
releaseStop.resolve()
|
|
stop.mockRestore()
|
|
await fixture.cleanup()
|
|
}
|
|
}
|
|
})
|
|
|
|
it('waits for bridge delivery before committing the owned terminal or stopping Runtime', async () => {
|
|
const fixture = await ownedHarness(true)
|
|
const enteredClose = deferred()
|
|
const delivered = deferred()
|
|
fixture.bridgeCloses.mockImplementationOnce(async () => {
|
|
enteredClose.resolve()
|
|
await delivered.promise
|
|
})
|
|
try {
|
|
await fixture.start()
|
|
await fixture.prompts[0]!()
|
|
await enteredClose.promise
|
|
expect(fixture.bridgeCloses).toHaveBeenCalledWith({ waitForDelivery: true })
|
|
expect(fixture.page().state).toBe('running')
|
|
expect(fixture.process.stops).toEqual([])
|
|
delivered.resolve()
|
|
await invoke(fixture, 'runtime/getAcpCursors', { bindingId: 'binding-1' })
|
|
expect(fixture.page().state).toBe('completed')
|
|
expect(fixture.process.stops).toEqual(['binding-closed'])
|
|
} finally { delivered.resolve(); await fixture.cleanup() }
|
|
})
|
|
|
|
it.each([false, true])('keeps transcript and reconciliation unknown when completion delivery fails (shared=%s)', async shared => {
|
|
const fixture = await ownedHarness(shared)
|
|
fixture.bridgeCloses.mockRejectedValueOnce(new Error('delivery ACK timed out'))
|
|
try {
|
|
await fixture.start()
|
|
await fixture.prompts[0]!()
|
|
await vi.waitFor(() => expect(fixture.page().state).toBe('outcome-unknown'))
|
|
await expect(invoke(fixture, 'runtime/reconcilePrompt', {
|
|
bindingId: 'binding-1', operationId: 'request-1', requestId: 'request-1'
|
|
})).resolves.toMatchObject({ status: 'outcome-unknown' })
|
|
expect(fixture.page().events).toHaveLength(1)
|
|
} finally { await fixture.cleanup() }
|
|
})
|
|
|
|
it('binds MCP image calls to authenticated blob framing and closes each prompt endpoint', async () => {
|
|
const fixture = harness({ workMode: 'execute' })
|
|
const client = new Client({ name: 'image-backend-test', version: '1' })
|
|
try {
|
|
await open(fixture)
|
|
const accepted = await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({
|
|
imageTool: { channelId: 'image-channel', channelEpoch: '1', description: 'Image catalog' }
|
|
})) as { imageToolUrl: string }
|
|
await client.connect(new StreamableHTTPClientTransport(new URL(accepted.imageToolUrl)))
|
|
expect((await client.listTools()).tools.map(tool => tool.name)).toEqual(['generate_image'])
|
|
const result = client.callTool({ name: 'generate_image', arguments: { intent: 'create', prompt: 'Blue' } })
|
|
await vi.waitFor(() => expect(fixture.blobSink).toHaveBeenCalledOnce())
|
|
const frame = fixture.blobSink.mock.calls[0]![0]
|
|
expect(frame.header).toMatchObject({ channelId: 'image-channel', channelEpoch: '1', sequence: '1', direction: 'agent-to-main' })
|
|
expect(fixture.backend.authorizeBlobFrame(frame, fixture.context.controller)).toBe(true)
|
|
expect(fixture.backend.authorizeBlobFrame(frame, { ...fixture.context.controller, connectionId: 'foreign' })).toBe(false)
|
|
const call = remoteImageToolCallSchema.parse(decodeRemoteImageToolMessage(frame.payload))
|
|
const payload = encodeRemoteImageToolMessage({ callId: call.callId, error: 'Controlled test failure' })
|
|
const reply = { header: { ...frame.header, direction: 'main-to-agent' as const, payloadLength: payload.byteLength }, payload }
|
|
await expect(fixture.backend.onBlobFrame(reply, { ...fixture.context, channelId: 'image-channel', controller: { ...fixture.context.controller, connectionId: 'foreign' } })).rejects.toThrow('authority')
|
|
await fixture.backend.onBlobFrame(reply, { ...fixture.context, channelId: 'image-channel' })
|
|
expect(await result).toMatchObject({ isError: true })
|
|
await invoke(fixture, 'runtime/completePrompt', { bindingId: 'binding-1', operationId: 'request-1', requestId: 'request-1' })
|
|
await expect(fetch(accepted.imageToolUrl)).rejects.toThrow()
|
|
const next = await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({
|
|
requestId: 'request-2', operationId: 'request-2', promptSequence: 1,
|
|
imageTool: { channelId: 'image-channel-2', channelEpoch: '2', description: 'Refreshed image catalog' }
|
|
})) as { imageToolUrl: string }
|
|
expect(next.imageToolUrl).not.toBe(accepted.imageToolUrl)
|
|
} finally { await client.close(); await fixture.backend.dispose() }
|
|
})
|
|
|
|
it('rejects image tool preparation in Ask before launching a process or bridge', async () => {
|
|
const fixture = harness()
|
|
try {
|
|
await open(fixture)
|
|
await expect(invoke(fixture, 'runtime/preparePrompt', fixture.preparation({
|
|
imageTool: { channelId: 'image-channel', channelEpoch: '1', description: 'Images' }
|
|
}))).rejects.toThrow('Execute')
|
|
expect(fixture.launches).toHaveLength(0)
|
|
expect(fixture.lifecycle).not.toContain('bridge-listen')
|
|
} finally { await fixture.backend.dispose() }
|
|
})
|
|
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.each([false, true])('reclaims completed owned processes, preserves detached peers and loads native history (shared=%s)', async shared => {
|
|
const root = await mkdtemp(join(tmpdir(), 'goodbuddy-idle-runtime-'))
|
|
const transcript = new SemanticPromptStore(join(root, 'prompts.sqlite'))
|
|
const pending = new Map<string, () => Promise<void>>()
|
|
const processes: FakeProcess[] = []
|
|
const loaded: string[] = []
|
|
const setRoute = vi.spyOn(AgentAcpConnection.prototype, 'setModelRoute').mockResolvedValue()
|
|
let sessionSequence = 0
|
|
const fixture = harness({
|
|
agentOwned: true, shareOwnedProcesses: shared, uniqueProcesses: true, semanticStore: transcript,
|
|
configureProcess: process => {
|
|
processes.push(process)
|
|
process.onWrite = payload => {
|
|
const request = JSON.parse(Buffer.from(payload).toString())
|
|
if (request.id === undefined) return
|
|
const reply = (result: unknown) => process.emit('stdout', `${JSON.stringify({ jsonrpc: '2.0', id: request.id, result })}\n`)
|
|
if (request.method === 'session/prompt') {
|
|
pending.set(request.params.sessionId, () => reply({ stopReason: 'end_turn' }))
|
|
return
|
|
}
|
|
let result: unknown = {}
|
|
if (request.method === 'initialize') result = { protocolVersion: 1, agentCapabilities: { loadSession: true } }
|
|
if (request.method === 'session/new') result = { sessionId: `native-${++sessionSequence}` }
|
|
if (request.method === 'session/load') loaded.push(request.params.sessionId)
|
|
if (request.method === 'session/set_config_option') result = {
|
|
configOptions: [{ id: 'model', name: 'Model', type: 'select', currentValue: request.params.value, options: [] }]
|
|
}
|
|
void reply(result)
|
|
}
|
|
}
|
|
})
|
|
const disconnect = new AbortController()
|
|
fixture.context.signal = disconnect.signal
|
|
const start = async (index: number, operationId: string, sessionId?: string) => {
|
|
const bindingId = `owned-${index}`
|
|
const channel = await invoke(fixture, 'runtime/openAcpChannel', { ...fixture.openRequest, bindingId }) as { channelEpoch: string }
|
|
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation({ bindingId, channelEpoch: channel.channelEpoch,
|
|
operationId, requestId: operationId, promptSequence: sessionId ? 1 : 0 }))
|
|
return await invoke(fixture, 'runtime/startPrompt', { bindingId, operationId, requestId: operationId,
|
|
...(sessionId ? { acpSessionId: sessionId } : {}), prompt: [{ type: 'text', text: 'Continue' }] }) as { sessionId: string }
|
|
}
|
|
try {
|
|
const first = await start(1, 'operation-1')
|
|
const second = await start(2, 'operation-2')
|
|
expect(processes).toHaveLength(shared ? 1 : 2)
|
|
await pending.get(first.sessionId)!()
|
|
await vi.waitFor(() => expect(transcript.attach('owned-1', 'operation-1', fixture.context.controller.controllerId).state).toBe('completed'))
|
|
expect(processes[0]!.stops).toHaveLength(shared ? 0 : 1)
|
|
expect(processes.at(-1)!.stops).toHaveLength(0)
|
|
await pending.get(second.sessionId)!()
|
|
await vi.waitFor(() => expect(transcript.attach('owned-2', 'operation-2', fixture.context.controller.controllerId).state).toBe('completed'))
|
|
expect(processes.every(process => process.stops.length === 1)).toBe(true)
|
|
const resumed = await start(1, 'operation-3', first.sessionId)
|
|
expect(loaded).toEqual([first.sessionId])
|
|
expect(resumed.sessionId).toBe(first.sessionId)
|
|
await processes.at(-1)!.emit('stdout', `${JSON.stringify({
|
|
jsonrpc: '2.0', method: 'session/update', params: {
|
|
sessionId: resumed.sessionId,
|
|
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: '' },
|
|
_meta: { goodbuddyQuestion: {
|
|
endpoint: `http://127.0.0.1:12345/${'a'.repeat(43)}`,
|
|
question: { id: 'question-1', sessionID: resumed.sessionId,
|
|
questions: [{ header: 'Continue', question: 'Proceed?', options: [] }] }
|
|
} }
|
|
}
|
|
}
|
|
})}\n`)
|
|
await vi.waitFor(() => expect(transcript.page({
|
|
bindingId: 'owned-1', operationId: 'operation-3',
|
|
controllerId: fixture.context.controller.controllerId, afterSequence: '0', limit: 10
|
|
}).events.some(event => JSON.stringify(event.payload).includes('question-1'))).toBe(true))
|
|
disconnect.abort()
|
|
await expect(fixture.backend.drain()).resolves.toBe(false)
|
|
expect(processes.at(-1)!.stops).toHaveLength(0)
|
|
await pending.get(resumed.sessionId)!()
|
|
await vi.waitFor(() => expect(transcript.attach('owned-1', 'operation-3', fixture.context.controller.controllerId).state).toBe('completed'))
|
|
await expect(fixture.backend.drain()).resolves.toBe(true)
|
|
fixture.context.signal = new AbortController().signal
|
|
await expect(start(1, 'operation-4', first.sessionId)).rejects.toThrow('draining')
|
|
expect(transcript.attach('owned-1', 'operation-1', fixture.context.controller.controllerId).sessionId).toBe(first.sessionId)
|
|
} finally {
|
|
await fixture.backend.dispose()
|
|
setRoute.mockRestore()
|
|
transcript.close()
|
|
await rm(root, { recursive: true, force: true })
|
|
}
|
|
})
|
|
|
|
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(
|
|
fixture.backend.onBlobFrame(
|
|
{
|
|
header: {
|
|
protocolMajor: 1,
|
|
protocolMinor: 0,
|
|
connectionId: fixture.context.controller.connectionId,
|
|
generation: fixture.context.controller.generation,
|
|
channelId: 'unknown-model-channel',
|
|
channelEpoch: '1',
|
|
direction: 'main-to-agent',
|
|
sequence: '1',
|
|
kind: 'blob',
|
|
payloadLength: 1
|
|
},
|
|
payload: Uint8Array.of(1)
|
|
},
|
|
{
|
|
...fixture.context,
|
|
channelId: 'unknown-model-channel'
|
|
}
|
|
)
|
|
).rejects.toMatchObject({ code: 'identity' })
|
|
})
|
|
|
|
it('starts the exact prepared model bridge before launch and revokes it on completion', async () => {
|
|
const fixture = harness()
|
|
await invoke(
|
|
fixture,
|
|
'runtime/openAcpChannel',
|
|
fixture.openRequest
|
|
)
|
|
const modelBridge = {
|
|
version: 'goodbuddy-model-bridge-v1' as const,
|
|
channelId: 'model-channel-1',
|
|
channelEpoch: '20',
|
|
policy: {
|
|
protocol: 'openai-responses' as const,
|
|
model: 'private-model',
|
|
modelProfileDigest: digest('8'),
|
|
supportsImageInput: false
|
|
}
|
|
}
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({ modelBridge })
|
|
)
|
|
expect(fixture.lifecycle).toEqual([
|
|
'bridge-listen',
|
|
'launch'
|
|
])
|
|
expect(fixture.launchModelBridges).toEqual([
|
|
{
|
|
socketPath: '/bridge/model-bridge.sock',
|
|
policy: modelBridge.policy
|
|
}
|
|
])
|
|
expect(fixture.bridgeDispatch).toEqual(expect.any(Function))
|
|
|
|
await invoke(fixture, 'runtime/completePrompt', {
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-1',
|
|
requestId: 'request-1'
|
|
})
|
|
expect(fixture.bridgeCloses).toHaveBeenCalledOnce()
|
|
})
|
|
|
|
it('starts the Agent-local model bridge from a prompt profile without a Main blob channel', async () => {
|
|
const fixture = harness({ agentOwned: true })
|
|
await invoke(
|
|
fixture,
|
|
'runtime/openAcpChannel',
|
|
fixture.openRequest
|
|
)
|
|
const preparation = fixture.preparation()
|
|
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
preparation
|
|
)
|
|
|
|
expect(fixture.lifecycle).toEqual([
|
|
'bridge-listen',
|
|
'launch'
|
|
])
|
|
expect(fixture.launchModelBridges).toEqual([
|
|
{
|
|
socketPath: '/bridge/model-bridge.sock',
|
|
policy: {
|
|
protocol: preparation.modelProfile!.protocol,
|
|
model: preparation.modelProfile!.model,
|
|
modelProfileDigest:
|
|
preparation.modelProfile!.modelProfileDigest,
|
|
supportsImageInput:
|
|
preparation.modelProfile!.capabilities.imageInput
|
|
}
|
|
}
|
|
])
|
|
expect(fixture.bridgeDispatch).toEqual(expect.any(Function))
|
|
expect(fixture.blobSink).not.toHaveBeenCalled()
|
|
})
|
|
|
|
it('allows more than 100 Agent-local model calls in one prompt', async () => {
|
|
const dispatch = vi.fn(async () => ({
|
|
response: {
|
|
status: 200,
|
|
headers: {},
|
|
bodyBase64: Buffer.from('{}').toString('base64')
|
|
},
|
|
acknowledgeDelivery: vi.fn(async () => undefined),
|
|
failDelivery: vi.fn()
|
|
}))
|
|
const fixture = harness({
|
|
agentOwned: true,
|
|
modelGateway: {
|
|
dispatch,
|
|
finalizePrompt: vi.fn()
|
|
}
|
|
})
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
|
|
for (let index = 0; index < 101; index += 1) {
|
|
const exchange = await fixture.bridgeDispatch!(
|
|
{
|
|
method: 'POST',
|
|
path: '/v1/responses',
|
|
headers: { 'content-type': 'application/json' },
|
|
bodyBase64: Buffer.from('{}').toString('base64')
|
|
},
|
|
{
|
|
requestId: `provider-request-${index}`,
|
|
signal: new AbortController().signal
|
|
}
|
|
)
|
|
if (!('acknowledgeDelivery' in exchange)) {
|
|
throw new Error('Expected an acknowledged model exchange')
|
|
}
|
|
await exchange.acknowledgeDelivery()
|
|
}
|
|
|
|
expect(dispatch).toHaveBeenCalledTimes(101)
|
|
})
|
|
|
|
it('runs open, prepare, ACP delivery, durable output, cursors, and close', async () => {
|
|
const fixture = harness()
|
|
const opened = await invoke(
|
|
fixture,
|
|
'runtime/openAcpChannel',
|
|
fixture.openRequest
|
|
)
|
|
expect(opened).toMatchObject({
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000'
|
|
})
|
|
|
|
const acceptance = await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
expect(acceptance).toMatchObject({
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-1',
|
|
workMode: 'ask',
|
|
requestId: 'request-1'
|
|
})
|
|
expect(fixture.launches).toEqual([
|
|
{ workMode: 'ask', scratch: '/scratch' }
|
|
])
|
|
|
|
fixture.journal.appendAcpFrame({
|
|
controllerId: fixture.context.controller.controllerId,
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
direction: 'main-to-runtime',
|
|
sequence: '1',
|
|
payload: Buffer.from('input')
|
|
})
|
|
await fixture.backend.onAcpFrame(
|
|
acpFrame(fixture.context.controller, 'input'),
|
|
{ ...fixture.context, channelId: 'binding-1' }
|
|
)
|
|
expect(
|
|
Buffer.from(fixture.process.writes[0]!).toString()
|
|
).toBe('input')
|
|
expect(fixture.journal.delivered).toEqual([])
|
|
fixture.journal.markAcpDelivered({
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
sequence: '1'
|
|
})
|
|
expect(fixture.journal.delivered).toEqual(['1'])
|
|
|
|
await fixture.process.emit('stderr', 'private diagnostic')
|
|
await fixture.process.emit('stdout', 'output')
|
|
expect(fixture.outputFrames).toEqual(['output'])
|
|
expect(
|
|
fixture.journal.frames.filter(
|
|
(frame) => frame.direction === 'runtime-to-main'
|
|
)
|
|
).toHaveLength(1)
|
|
|
|
const cursors = await invoke(
|
|
fixture,
|
|
'runtime/getAcpCursors',
|
|
{ bindingId: 'binding-1' }
|
|
)
|
|
expect(cursors).toEqual({
|
|
lastOutboundJournaledSequence: '1',
|
|
lastOutboundDeliveredSequence: '1',
|
|
lastInboundJournaledSequence: '1',
|
|
lastMainAckSequence: '0'
|
|
})
|
|
|
|
await expect(
|
|
invoke(fixture, 'runtime/closeAcpChannel', {
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
reason: 'released'
|
|
})
|
|
).resolves.toEqual({
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
closed: true
|
|
})
|
|
expect(fixture.process.stops).toEqual(['binding-closed'])
|
|
expect(fixture.journal.retired).toEqual([
|
|
{
|
|
controllerId: 'controller-1',
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1000'
|
|
}
|
|
])
|
|
})
|
|
|
|
it.each(['ask', 'execute'] as const)(
|
|
'accepts %s as the complete prompt authorization',
|
|
async (workMode) => {
|
|
const fixture = harness({ workMode })
|
|
await open(fixture)
|
|
await expect(
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
).resolves.toMatchObject({ workMode })
|
|
}
|
|
)
|
|
|
|
it('is idempotent for stable identities and rejects conflicting payloads', async () => {
|
|
const fixture = harness()
|
|
const firstOpen = await open(fixture)
|
|
await expect(open(fixture)).resolves.toEqual(firstOpen)
|
|
const first = await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
await expect(
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
).resolves.toEqual(first)
|
|
expect(fixture.launches).toHaveLength(1)
|
|
|
|
await expect(
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
deadlineAt: new Date(62_000).toISOString()
|
|
})
|
|
)
|
|
).rejects.toMatchObject({ code: 'conflict' })
|
|
await expect(
|
|
invoke(fixture, 'runtime/openAcpChannel', {
|
|
...fixture.openRequest,
|
|
workspaceIdentity: 'workspace-other'
|
|
})
|
|
).rejects.toMatchObject({ code: 'conflict' })
|
|
})
|
|
|
|
it('completes two prompts idempotently while keeping one process alive', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
const completion = {
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-1',
|
|
requestId: 'request-1'
|
|
}
|
|
const first = await invoke(
|
|
fixture,
|
|
'runtime/completePrompt',
|
|
completion
|
|
)
|
|
await expect(
|
|
invoke(fixture, 'runtime/completePrompt', completion)
|
|
).resolves.toEqual(first)
|
|
await expect(
|
|
invoke(fixture, 'runtime/reconcilePrompt', completion)
|
|
).resolves.toEqual({
|
|
status: 'terminal',
|
|
terminalState: 'completed',
|
|
processTree: 'running'
|
|
})
|
|
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
operationId: 'request-2',
|
|
requestId: 'request-2',
|
|
budget: {
|
|
maximumInputBytes: 2_048,
|
|
maximumOutputBytes: 1_024
|
|
}
|
|
})
|
|
)
|
|
await invoke(fixture, 'runtime/completePrompt', {
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-2',
|
|
requestId: 'request-2'
|
|
})
|
|
|
|
expect(fixture.launches).toHaveLength(1)
|
|
expect(fixture.process.prompts).toHaveLength(1)
|
|
expect(fixture.process.prompts[0]!.maximumInputBytes).toBe(2_048)
|
|
expect(fixture.process.completions).toBe(2)
|
|
expect(fixture.process.stops).toEqual([])
|
|
})
|
|
|
|
it('rejects completion conflicts and unsafe stable reopen', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
await expect(
|
|
invoke(fixture, 'runtime/completePrompt', {
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-other',
|
|
requestId: 'request-other'
|
|
})
|
|
).rejects.toMatchObject({ code: 'identity' })
|
|
|
|
await expect(
|
|
fixture.backend.methods['runtime/openAcpChannel']!(
|
|
fixture.openRequest,
|
|
protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
)
|
|
).rejects.toMatchObject({ code: 'stale-controller' })
|
|
})
|
|
|
|
it('reopens a detached idle binding after a proven same-controller takeover', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
await invoke(fixture, 'runtime/completePrompt', {
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-1',
|
|
requestId: 'request-1'
|
|
})
|
|
fixture.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
|
|
const replacement = protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
replacement.controllerTakeoverProven = true
|
|
await expect(
|
|
fixture.backend.methods['runtime/openAcpChannel']!(
|
|
fixture.openRequest,
|
|
replacement
|
|
)
|
|
).resolves.toMatchObject({
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1001'
|
|
})
|
|
expect(fixture.process.stops).toEqual(['binding-closed'])
|
|
expect(fixture.journal.retired).toContainEqual({
|
|
controllerId: 'controller-1',
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1000'
|
|
})
|
|
})
|
|
|
|
it('reattaches a detached Agent-owned operation without replacing its process or raw output channel', async () => {
|
|
const fixture = harness({ agentOwned: true })
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
await fixture.process.emit('stdout', 'semantic-only-output')
|
|
expect(fixture.outputFrames).toEqual([])
|
|
|
|
fixture.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
const replacement = protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
replacement.controllerTakeoverProven = true
|
|
const reopened = await fixture.backend.methods[
|
|
'runtime/openAcpChannel'
|
|
]!(fixture.openRequest, replacement)
|
|
|
|
expect(reopened).toMatchObject({
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1000'
|
|
})
|
|
expect(fixture.process.stops).toEqual([])
|
|
await fixture.backend.dispose()
|
|
})
|
|
|
|
it('revokes an Agent-owned preparation that never starts', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const fixture = harness({ agentOwned: true })
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
deadlineAt: UNBOUNDED_REMOTE_PROMPT_DEADLINE
|
|
})
|
|
)
|
|
|
|
await vi.advanceTimersByTimeAsync(2 * 60_000)
|
|
await vi.waitFor(() =>
|
|
expect(fixture.semanticPrompts?.append).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
terminalState: 'outcome-unknown'
|
|
})
|
|
)
|
|
)
|
|
expect(fixture.process.stops).toContain('identity-conflict')
|
|
await fixture.backend.dispose()
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
|
|
it('records a failed semantic terminal when an owned Runtime exits before a model round', async () => {
|
|
const fixture = harness({ agentOwned: true })
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
|
|
await fixture.process.emitExit('failed')
|
|
await vi.waitFor(() =>
|
|
expect(fixture.semanticPrompts?.append).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
terminalState: 'failed',
|
|
payload: expect.objectContaining({
|
|
error: expect.objectContaining({
|
|
name: 'RuntimeExited'
|
|
})
|
|
})
|
|
})
|
|
)
|
|
)
|
|
expect(fixture.bridgeCloses).toHaveBeenCalledOnce()
|
|
await vi.waitFor(() =>
|
|
expect(fixture.diagnostics.tryRecord).toHaveBeenCalledWith(
|
|
'runtime.exited',
|
|
{
|
|
runtimeId: 'opencode',
|
|
workMode: 'ask',
|
|
outcome: 'failed'
|
|
}
|
|
)
|
|
)
|
|
expect(fixture.diagnostics.tryRecord).toHaveBeenCalledWith(
|
|
'runtime.starting',
|
|
{
|
|
runtimeId: 'opencode',
|
|
workMode: 'ask'
|
|
}
|
|
)
|
|
expect(fixture.diagnostics.tryRecord).toHaveBeenCalledWith(
|
|
'runtime.started',
|
|
{
|
|
runtimeId: 'opencode',
|
|
workMode: 'ask'
|
|
}
|
|
)
|
|
await fixture.backend.dispose()
|
|
})
|
|
|
|
it('records a redacted Runtime launch failure at the unified preparation point', async () => {
|
|
const launchError = new Error(
|
|
'accepted Prompt and API key must never be persisted'
|
|
)
|
|
const fixture = harness({ launchError })
|
|
await open(fixture)
|
|
|
|
await expect(
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
).rejects.toThrow('Runtime process launch failed')
|
|
|
|
expect(fixture.diagnostics.tryRecord).toHaveBeenCalledWith(
|
|
'runtime.start.failed',
|
|
{
|
|
runtimeId: 'opencode',
|
|
workMode: 'ask',
|
|
error: launchError
|
|
}
|
|
)
|
|
await fixture.backend.dispose()
|
|
})
|
|
|
|
it('records a semantic terminal before stopping an owned prompt at its deadline', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const fixture = harness({ agentOwned: true })
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
deadlineAt: new Date(1_100).toISOString()
|
|
})
|
|
)
|
|
|
|
await vi.advanceTimersByTimeAsync(100)
|
|
await vi.waitFor(() =>
|
|
expect(fixture.semanticPrompts?.append).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
terminalState: 'failed',
|
|
payload: expect.objectContaining({
|
|
error: expect.objectContaining({
|
|
name: 'PromptDeadlineExceeded'
|
|
})
|
|
})
|
|
})
|
|
)
|
|
)
|
|
expect(fixture.process.stops).toContain('deadline-exceeded')
|
|
await fixture.backend.dispose()
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
|
|
it('terminalizes outcome-unknown after an uncertain Agent gateway dispatch', async () => {
|
|
const modelGateway = {
|
|
dispatch: vi.fn(async () => {
|
|
throw new AgentModelGatewayError(
|
|
'timeout',
|
|
'Provider dispatch outcome is unknown'
|
|
)
|
|
}),
|
|
finalizePrompt: vi.fn()
|
|
}
|
|
const fixture = harness({
|
|
agentOwned: true,
|
|
modelGateway
|
|
})
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
|
|
await expect(
|
|
fixture.bridgeDispatch?.(
|
|
{
|
|
method: 'POST',
|
|
path: '/v1/responses',
|
|
headers: { 'content-type': 'application/json' },
|
|
bodyBase64: Buffer.from(
|
|
JSON.stringify({ model: 'test-model' })
|
|
).toString('base64')
|
|
},
|
|
{
|
|
requestId: 'model-request-1',
|
|
signal: new AbortController().signal
|
|
}
|
|
)
|
|
).rejects.toMatchObject({ code: 'timeout' })
|
|
await vi.waitFor(() =>
|
|
expect(fixture.semanticPrompts?.append).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
terminalState: 'outcome-unknown'
|
|
})
|
|
)
|
|
)
|
|
expect(fixture.process.stops).toContain('identity-conflict')
|
|
await fixture.backend.dispose()
|
|
})
|
|
|
|
it('drains the detached old epoch before replacing an idle binding', async () => {
|
|
let releaseOutput!: () => void
|
|
const outputGate = new Promise<void>((resolve) => {
|
|
releaseOutput = resolve
|
|
})
|
|
const fixture = harness({ outputGate })
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
const pendingOutput = fixture.process.emit('stdout', 'pending')
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
await invoke(fixture, 'runtime/completePrompt', {
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-1',
|
|
requestId: 'request-1'
|
|
})
|
|
fixture.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
|
|
const replacement = protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
replacement.controllerTakeoverProven = true
|
|
let replaced = false
|
|
const reopen = fixture.backend.methods[
|
|
'runtime/openAcpChannel'
|
|
]!(fixture.openRequest, replacement).then((result) => {
|
|
replaced = true
|
|
return result
|
|
})
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
expect(replaced).toBe(false)
|
|
|
|
releaseOutput()
|
|
await pendingOutput
|
|
await expect(reopen).resolves.toMatchObject({
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1001'
|
|
})
|
|
expect(fixture.journal.frames).toContainEqual(
|
|
expect.objectContaining({
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
sequence: '1'
|
|
})
|
|
)
|
|
})
|
|
|
|
it('does not adopt an idle binding without exact takeover proof', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
fixture.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
|
|
await expect(
|
|
fixture.backend.methods['runtime/openAcpChannel']!(
|
|
fixture.openRequest,
|
|
protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
)
|
|
).rejects.toMatchObject({ code: 'stale-controller' })
|
|
})
|
|
|
|
it('reopens a definitively closed stable binding with a fresh epoch and generation', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
await invoke(fixture, 'runtime/closeAcpChannel', {
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
reason: 'released'
|
|
})
|
|
const reopened = await fixture.backend.methods[
|
|
'runtime/openAcpChannel'
|
|
]!(
|
|
fixture.openRequest,
|
|
protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
)
|
|
expect(reopened).toMatchObject({
|
|
bindingId: 'binding-1',
|
|
channelEpoch: '1001'
|
|
})
|
|
})
|
|
|
|
it('rejects cross-controller access and expired preparations', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await expect(
|
|
fixture.backend.methods['runtime/getAcpCursors']!(
|
|
{ bindingId: 'binding-1' },
|
|
protocolContext({
|
|
controllerId: 'controller-other',
|
|
connectionId: 'connection-other'
|
|
})
|
|
)
|
|
).rejects.toMatchObject({ code: 'stale-controller' })
|
|
await expect(
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
deadlineAt: new Date(1_000).toISOString()
|
|
})
|
|
)
|
|
).rejects.toMatchObject({ code: 'deadline' })
|
|
})
|
|
|
|
it('keeps an unbounded Runtime prompt alive without a wall-clock timer', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
deadlineAt: UNBOUNDED_REMOTE_PROMPT_DEADLINE
|
|
})
|
|
)
|
|
|
|
await vi.advanceTimersByTimeAsync(30 * 24 * 60 * 60_000)
|
|
expect(fixture.process.stops).toEqual([])
|
|
await fixture.process.emit('stdout', 'still-running')
|
|
expect(fixture.outputFrames).toEqual(['still-running'])
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
|
|
it('does not stop the process when cumulative output exceeds the display budget', async () => {
|
|
const fixture = harness({ maximumOutputBytes: 3 })
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
await expect(
|
|
fixture.process.emit('stdout', 'four')
|
|
).resolves.toBeUndefined()
|
|
expect(fixture.process.stops).toEqual([])
|
|
expect(fixture.outputFrames).toEqual(['four'])
|
|
})
|
|
|
|
it('waits for durable journal capacity instead of stopping Runtime output', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
fixture.journal.capacityBlocked = true
|
|
const output = fixture.process.emit('stdout', 'later')
|
|
await new Promise<void>((resolve) => setTimeout(resolve, 10))
|
|
expect(fixture.process.stops).toEqual([])
|
|
expect(fixture.outputFrames).toEqual([])
|
|
|
|
fixture.journal.capacityBlocked = false
|
|
await output
|
|
|
|
expect(fixture.process.stops).toEqual([])
|
|
expect(fixture.outputFrames).toEqual(['later'])
|
|
})
|
|
|
|
it('backs off while durable journal capacity remains unavailable', async () => {
|
|
vi.useFakeTimers()
|
|
try {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
fixture.journal.capacityBlocked = true
|
|
const output = fixture.process.emit('stdout', 'later')
|
|
|
|
await vi.advanceTimersByTimeAsync(775)
|
|
expect(fixture.journal.appendAttempts).toBeLessThanOrEqual(6)
|
|
expect(fixture.process.stops).toEqual([])
|
|
|
|
fixture.journal.capacityBlocked = false
|
|
await vi.advanceTimersByTimeAsync(800)
|
|
await output
|
|
|
|
expect(fixture.outputFrames).toEqual(['later'])
|
|
await fixture.backend.dispose()
|
|
} finally {
|
|
vi.useRealTimers()
|
|
}
|
|
})
|
|
|
|
it('stops with output quota when the detached durable journal is exhausted', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
fixture.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
fixture.journal.failAppend = true
|
|
await expect(
|
|
fixture.process.emit('stdout', 'offline')
|
|
).rejects.toMatchObject({ code: 'output-quota' })
|
|
expect(fixture.process.stops).toEqual(['output-quota'])
|
|
expect(fixture.outputFrames).toEqual([])
|
|
})
|
|
|
|
it('escalates cancellation but detaches without stopping on disconnect', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
await expect(
|
|
invoke(fixture, 'runtime/escalateCancellation', {
|
|
bindingId: 'binding-1',
|
|
sessionId: 'session-1',
|
|
operationId: 'request-1',
|
|
requestId: 'request-1',
|
|
reason: 'requested'
|
|
})
|
|
).resolves.toEqual({ bindingId: 'binding-1', stopped: true })
|
|
expect(fixture.process.stops).toEqual(['user-cancelled'])
|
|
|
|
const disconnected = harness()
|
|
await open(disconnected)
|
|
await invoke(
|
|
disconnected,
|
|
'runtime/preparePrompt',
|
|
disconnected.preparation()
|
|
)
|
|
disconnected.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
expect(disconnected.process.stops).toEqual([])
|
|
await disconnected.process.emit('stdout', 'offline')
|
|
expect(disconnected.outputFrames).toEqual([])
|
|
expect(
|
|
disconnected.journal.frames.filter(
|
|
(frame) => frame.direction === 'runtime-to-main'
|
|
).map((frame) => Buffer.from(frame.payload).toString())
|
|
).toEqual(['offline'])
|
|
})
|
|
|
|
it('retains the owned cancellation outcome when the semantic terminal already exists', async () => {
|
|
const fixture = harness({ agentOwned: true })
|
|
let terminalWritten = false
|
|
fixture.semanticPrompts!.append.mockImplementation(() => {
|
|
if (terminalWritten) throw new Error('Prompt is already terminal')
|
|
terminalWritten = true
|
|
})
|
|
await open(fixture)
|
|
await invoke(fixture, 'runtime/preparePrompt', fixture.preparation())
|
|
await invoke(fixture, 'runtime/escalateCancellation', {
|
|
bindingId: 'binding-1', sessionId: 'session-1',
|
|
operationId: 'request-1', requestId: 'request-1', reason: 'requested'
|
|
})
|
|
await expect(invoke(fixture, 'runtime/reconcilePrompt', {
|
|
bindingId: 'binding-1', operationId: 'request-1', requestId: 'request-1'
|
|
})).resolves.toEqual({ status: 'terminal', terminalState: 'cancelled', processTree: 'empty' })
|
|
await fixture.backend.dispose()
|
|
})
|
|
|
|
it('closes an idle model bridge on detach without poisoning or stopping Runtime work', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
modelBridge: {
|
|
version: 'goodbuddy-model-bridge-v1',
|
|
channelId: 'model-channel-idle',
|
|
channelEpoch: '20',
|
|
policy: {
|
|
protocol: 'openai-responses',
|
|
model: 'private-model',
|
|
modelProfileDigest: digest('8'),
|
|
supportsImageInput: false
|
|
}
|
|
}
|
|
})
|
|
)
|
|
|
|
fixture.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
expect(fixture.bridgeCloses).toHaveBeenCalledOnce()
|
|
expect(fixture.process.stops).toEqual([])
|
|
await fixture.process.emit('stdout', 'idle-bridge-offline')
|
|
expect(fixture.outputFrames).toEqual([])
|
|
|
|
const resumedContext = protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
resumedContext.controllerTakeoverProven = true
|
|
await expect(
|
|
fixture.backend.methods['runtime/openAcpChannel']!(
|
|
fixture.openRequest,
|
|
resumedContext
|
|
)
|
|
).rejects.toMatchObject({ code: 'stale-controller' })
|
|
await expect(
|
|
fixture.backend.methods['runtime/resumeAcpChannel']!(
|
|
{
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000'
|
|
},
|
|
resumedContext
|
|
)
|
|
).resolves.toMatchObject({
|
|
bindingId: 'binding-1',
|
|
cursors: { lastInboundJournaledSequence: '1' }
|
|
})
|
|
})
|
|
|
|
it('poisons an in-flight Provider exchange on detach without stopping Runtime work', async () => {
|
|
const fixture = harness({ bridgeCloseError: true })
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
modelBridge: {
|
|
version: 'goodbuddy-model-bridge-v1',
|
|
channelId: 'model-channel-active',
|
|
channelEpoch: '21',
|
|
policy: {
|
|
protocol: 'openai-responses',
|
|
model: 'private-model',
|
|
modelProfileDigest: digest('8'),
|
|
supportsImageInput: false
|
|
}
|
|
}
|
|
})
|
|
)
|
|
const providerExchange = fixture.bridgeDispatch!(
|
|
{
|
|
method: 'POST',
|
|
path: '/v1/responses',
|
|
headers: { 'content-type': 'application/json' },
|
|
bodyBase64: Buffer.from(
|
|
JSON.stringify({ model: 'private-model' })
|
|
).toString('base64')
|
|
},
|
|
{
|
|
requestId: 'provider-request-1',
|
|
signal: new AbortController().signal
|
|
}
|
|
)
|
|
await vi.waitFor(() =>
|
|
expect(fixture.blobSink).toHaveBeenCalledOnce()
|
|
)
|
|
|
|
fixture.context.abort.abort()
|
|
await expect(providerExchange).rejects.toMatchObject({
|
|
code: 'cancelled'
|
|
})
|
|
expect(fixture.process.stops).toEqual([])
|
|
await fixture.process.emit('stdout', 'poisoned-bridge-offline')
|
|
expect(
|
|
fixture.journal.frames
|
|
.filter((frame) => frame.direction === 'runtime-to-main')
|
|
.map((frame) => Buffer.from(frame.payload).toString())
|
|
).toEqual(['poisoned-bridge-offline'])
|
|
|
|
await expect(
|
|
fixture.backend.methods['runtime/resumeAcpChannel']!(
|
|
{
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000'
|
|
},
|
|
protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
)
|
|
).rejects.toMatchObject({ code: 'conflict' })
|
|
})
|
|
|
|
it('resumes the exact detached binding and replays ACK-exclusive output before live output', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
fixture.context.abort.abort()
|
|
await new Promise<void>((resolve) => setImmediate(resolve))
|
|
await fixture.process.emit('stdout', 'offline-1')
|
|
await fixture.process.emit('stdout', 'offline-2')
|
|
expect(fixture.outputFrames).toEqual([])
|
|
|
|
const resumedContext = protocolContext({
|
|
generation: 2,
|
|
connectionId: 'connection-2'
|
|
})
|
|
await expect(
|
|
fixture.backend.methods['runtime/resumeAcpChannel']!(
|
|
{
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000'
|
|
},
|
|
resumedContext
|
|
)
|
|
).resolves.toMatchObject({
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
cursors: {
|
|
lastInboundJournaledSequence: '2',
|
|
lastMainAckSequence: '0'
|
|
}
|
|
})
|
|
await expect(
|
|
fixture.backend.methods['runtime/replayAcpChannel']!(
|
|
{
|
|
bindingId: 'binding-1',
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
acknowledgedSequence: '1'
|
|
},
|
|
resumedContext
|
|
)
|
|
).resolves.toMatchObject({
|
|
replayedThroughSequence: '2',
|
|
live: true
|
|
})
|
|
await fixture.process.emit('stdout', 'live-3')
|
|
expect(fixture.outputFrames).toEqual(['offline-2', 'live-3'])
|
|
|
|
await expect(
|
|
fixture.backend.methods['runtime/getAcpCursors']!(
|
|
{ bindingId: 'binding-1' },
|
|
fixture.context
|
|
)
|
|
).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)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
fixture.process.reconciliation = {
|
|
identity: {
|
|
...fixture.process.identity,
|
|
processId: 'different-process'
|
|
},
|
|
state: 'running',
|
|
processTree: 'running'
|
|
}
|
|
await expect(
|
|
invoke(fixture, 'runtime/reconcilePrompt', {
|
|
bindingId: 'binding-1',
|
|
operationId: 'request-1',
|
|
requestId: 'request-1'
|
|
})
|
|
).rejects.toMatchObject({ code: 'identity' })
|
|
expect(fixture.process.stops).toEqual(['identity-conflict'])
|
|
})
|
|
|
|
it('boundedly stops and reconciles every owned process on disposal', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
|
|
await expect(fixture.backend.dispose(100)).resolves.toBeUndefined()
|
|
expect(fixture.process.stops).toEqual(['binding-closed'])
|
|
await expect(
|
|
invoke(
|
|
fixture,
|
|
'runtime/openAcpChannel',
|
|
fixture.openRequest
|
|
)
|
|
).rejects.toMatchObject({ code: 'closed' })
|
|
})
|
|
|
|
it('rejects a process identity reused by another binding', async () => {
|
|
const fixture = harness()
|
|
await open(fixture)
|
|
await invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
)
|
|
const secondOpen = {
|
|
...fixture.openRequest,
|
|
bindingId: 'binding-2',
|
|
workspaceIdentity: 'workspace-identity-2'
|
|
}
|
|
await invoke(
|
|
fixture,
|
|
'runtime/openAcpChannel',
|
|
secondOpen
|
|
)
|
|
await expect(
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
bindingId: 'binding-2',
|
|
operationId: 'request-2',
|
|
requestId: 'request-2',
|
|
channelEpoch: '1001',
|
|
workspaceIdentity: 'workspace-identity-2'
|
|
})
|
|
)
|
|
).rejects.toMatchObject({ code: 'identity' })
|
|
expect(fixture.process.stops).toEqual(['identity-conflict'])
|
|
})
|
|
|
|
it('runs independent conversations for two workspaces on one controller', async () => {
|
|
const fixture = harness({ uniqueProcesses: true })
|
|
await open(fixture)
|
|
const secondOpen = {
|
|
...fixture.openRequest,
|
|
bindingId: 'binding-2',
|
|
workspaceIdentity: 'workspace-identity-2'
|
|
}
|
|
await invoke(
|
|
fixture,
|
|
'runtime/openAcpChannel',
|
|
secondOpen
|
|
)
|
|
|
|
await Promise.all([
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation()
|
|
),
|
|
invoke(
|
|
fixture,
|
|
'runtime/preparePrompt',
|
|
fixture.preparation({
|
|
bindingId: 'binding-2',
|
|
operationId: 'request-2',
|
|
requestId: 'request-2',
|
|
channelEpoch: '1001',
|
|
workspaceIdentity: 'workspace-identity-2'
|
|
})
|
|
)
|
|
])
|
|
|
|
expect(fixture.launches).toHaveLength(2)
|
|
expect(fixture.process.stops).toEqual([])
|
|
})
|
|
})
|
|
|
|
async function open(fixture: Harness): Promise<Record<string, unknown>> {
|
|
return (await invoke(
|
|
fixture,
|
|
'runtime/openAcpChannel',
|
|
fixture.openRequest
|
|
)) as Record<string, unknown>
|
|
}
|
|
|
|
async function invoke(
|
|
fixture: Harness,
|
|
method: string,
|
|
params: unknown
|
|
): Promise<unknown> {
|
|
return await fixture.backend.methods[method]!(params, fixture.context)
|
|
}
|
|
|
|
function protocolContext(
|
|
overrides: Partial<ControllerLease> = {}
|
|
): ProtocolMethodContext & { abort: AbortController } {
|
|
const abort = new AbortController()
|
|
return {
|
|
controller: {
|
|
controllerId: 'controller-1',
|
|
connectionId: 'connection-1',
|
|
generation: 1,
|
|
leaseExpiresAt: 1_000_000,
|
|
capabilityGeneration: 1,
|
|
ownedObjects: {},
|
|
...overrides
|
|
},
|
|
channelId: 'control-channel',
|
|
signal: abort.signal,
|
|
abort
|
|
}
|
|
}
|
|
|
|
function resolvedBundle(): RuntimeAcpResolvedBundle {
|
|
return {
|
|
entry: {
|
|
runtimeId: 'opencode',
|
|
bundleDigest: digest('1'),
|
|
acpCapabilitiesDigest: digest('6')
|
|
},
|
|
bundleDirectory: '/runtime/opencode'
|
|
}
|
|
}
|
|
|
|
function verifiedBundle(): RuntimeAcpVerifiedBundle {
|
|
return {
|
|
bundleDirectory: '/runtime/opencode',
|
|
executablePath: '/runtime/opencode/bin/opencode',
|
|
manifest: {
|
|
runtimeId: 'opencode',
|
|
provider: 'opencode',
|
|
bundleDigest: digest('1'),
|
|
adapterDigest: digest('2'),
|
|
acpCapabilitiesDigest: digest('6'),
|
|
limits: {
|
|
maximumPromptRuntimeMilliseconds: 60_000,
|
|
maximumPromptInputBytes: 16 * 1024 * 1024,
|
|
maximumPromptOutputBytes: 10_000
|
|
}
|
|
} as RemoteRuntimeBundleManifest,
|
|
manifestDigest: digest('5')
|
|
}
|
|
}
|
|
|
|
function acpFrame(
|
|
controller: ControllerLease,
|
|
value: string
|
|
) {
|
|
const payload = Buffer.from(value)
|
|
return {
|
|
header: {
|
|
protocolMajor: 1,
|
|
protocolMinor: 0,
|
|
connectionId: controller.connectionId,
|
|
generation: controller.generation,
|
|
channelId: 'binding-1',
|
|
channelEpoch: '1000',
|
|
direction: 'main-to-agent' as const,
|
|
sequence: '1',
|
|
kind: 'acp' as const,
|
|
payloadLength: payload.byteLength
|
|
},
|
|
payload
|
|
}
|
|
}
|