feat: keep managed SSH prompts running on Hosts

Managed SSH prompts previously depended on Desktop-owned provider and transcript state, so accepted work could stop or become uncertain when the Desktop, network, or power disappeared. The Agent now owns accepted OpenCode prompt execution, model and tool rounds, cancellation, and the durable semantic transcript, while Desktop remains authoritative for permanent projects, conversations, tasks, and messages.

Recovered work attaches to the original conversation without replay. The project switcher also groups managed SSH projects by saved Host and shows live Agent connectivity beside each group without opening a connection merely to read status. Remote project subtitles now show only the path, and the Agent lock advances to 0.11.14 with Runtime ACP v4.

Release note: 托管 SSH 任务在桌面端、网络或电源中断后可继续在主机上运行,并能回到原会话精确恢复;项目菜单现按主机分组并显示实时连接状态。
This commit is contained in:
mesalogo
2026-08-31 11:04:41 +08:00
parent f6f8d5e053
commit 561de29f00
75 changed files with 13439 additions and 1327 deletions
@@ -36,6 +36,13 @@ import {
import {
remotePromptOperationAcceptanceSchema,
remotePromptOperationPreparationSchema,
remoteOwnedPromptAttachRequestSchema,
remoteOwnedPromptStartRequestSchema,
remoteOwnedPromptStartResultSchema,
remoteSemanticTranscriptAckRequestSchema,
remoteSemanticTranscriptAckResultSchema,
remoteSemanticTranscriptPageRequestSchema,
remoteSemanticTranscriptPageResultSchema,
remoteGitDiffRequestSchema,
remoteGitDiffResultSchema,
remoteGitStatusRequestSchema,
@@ -166,6 +173,22 @@ export const AGENT_PROTOCOL_METHODS = {
params: remotePromptOperationPreparationSchema,
result: remotePromptOperationAcceptanceSchema
},
'runtime/startPrompt': {
params: remoteOwnedPromptStartRequestSchema,
result: remoteOwnedPromptStartResultSchema
},
'runtime/attachPrompt': {
params: remoteOwnedPromptAttachRequestSchema,
result: remoteOwnedPromptStartResultSchema
},
'runtime/pagePromptTranscript': {
params: remoteSemanticTranscriptPageRequestSchema,
result: remoteSemanticTranscriptPageResultSchema
},
'runtime/ackPromptTranscript': {
params: remoteSemanticTranscriptAckRequestSchema,
result: remoteSemanticTranscriptAckResultSchema
},
'runtime/completePrompt': {
params: acpCompletePromptRequestSchema,
result: acpCompletePromptResultSchema
@@ -238,4 +238,88 @@ describe('ControllerStateStore', () => {
)?.acpBindings[0]?.cursors.lastOutboundJournaledSequence
).toBe('12')
})
it('preserves advanced cursors when reopening the same recovery channel', async () => {
const file = new MemoryStateFile()
const store = new ControllerStateStore(file)
await store.updateConnection(connection)
await store.updateAcpBinding(connection.cacheKey, 'binding-1', {
...connection.acpBindings[0]!,
cursors: {
lastOutboundJournaledSequence: '12',
lastOutboundDeliveredSequence: '11',
lastInboundJournaledSequence: '14',
lastMainAckSequence: '13'
}
})
await store.updateAcpBinding(connection.cacheKey, 'binding-1', {
...connection.acpBindings[0]!,
cursors: {
lastOutboundJournaledSequence: '0',
lastOutboundDeliveredSequence: '0',
lastInboundJournaledSequence: '0',
lastMainAckSequence: '0'
}
})
await expect(
store.getConnection(connection.cacheKey)
).resolves.toMatchObject({
acpBindings: [
{
bindingId: 'binding-1',
cursors: {
lastOutboundJournaledSequence: '12',
lastOutboundDeliveredSequence: '11',
lastInboundJournaledSequence: '14',
lastMainAckSequence: '13'
}
}
]
})
})
it('starts zero cursors when reopening a binding on a new channel epoch', async () => {
const file = new MemoryStateFile()
const store = new ControllerStateStore(file)
await store.updateConnection(connection)
await store.updateAcpBinding(connection.cacheKey, 'binding-1', {
...connection.acpBindings[0]!,
cursors: {
lastOutboundJournaledSequence: '12',
lastOutboundDeliveredSequence: '11',
lastInboundJournaledSequence: '14',
lastMainAckSequence: '13'
}
})
await store.updateAcpBinding(connection.cacheKey, 'binding-1', {
...connection.acpBindings[0]!,
channelEpoch: '2',
cursors: {
lastOutboundJournaledSequence: '0',
lastOutboundDeliveredSequence: '0',
lastInboundJournaledSequence: '0',
lastMainAckSequence: '0'
}
})
await expect(
store.getConnection(connection.cacheKey)
).resolves.toMatchObject({
acpBindings: [
{
bindingId: 'binding-1',
channelEpoch: '2',
cursors: {
lastOutboundJournaledSequence: '0',
lastOutboundDeliveredSequence: '0',
lastInboundJournaledSequence: '0',
lastMainAckSequence: '0'
}
}
]
})
})
})
+38 -21
View File
@@ -241,7 +241,7 @@ export class ControllerStateStore {
): Promise<void> {
assertCacheKey(cacheKey)
const parsedBindingId = agentIdentifierSchema.parse(bindingId)
const parsed =
let parsed =
binding === undefined
? undefined
: acpRecoveryBindingSchema.parse(binding)
@@ -261,28 +261,45 @@ export class ControllerStateStore {
)
if (
parsed !== undefined &&
previous !== undefined &&
(
parsed.channelId !== previous.channelId ||
parsed.channelEpoch !== previous.channelEpoch ||
Object.keys(parsed.cursors).some(
(key) =>
BigInt(
parsed.cursors[
key as keyof typeof parsed.cursors
]
) <
BigInt(
previous.cursors[
key as keyof typeof previous.cursors
]
)
)
)
previous !== undefined
) {
throw new Error(
'Controller ACP recovery identity or cursors cannot regress'
const candidate = parsed
const channelChanged =
candidate.channelId !== previous.channelId
const epochChanged =
candidate.channelEpoch !== previous.channelEpoch
const isOpenClaim = Object.values(
candidate.cursors
).every((value) => value === '0')
const cursorsRegress = Object.keys(candidate.cursors).some(
(key) =>
BigInt(
candidate.cursors[
key as keyof typeof candidate.cursors
]
) <
BigInt(
previous.cursors[
key as keyof typeof previous.cursors
]
)
)
if (
channelChanged ||
(epochChanged && !isOpenClaim)
) {
throw new Error(
'Controller ACP recovery identity or cursors cannot regress'
)
}
if (!epochChanged && cursorsRegress) {
if (!isOpenClaim) {
throw new Error(
'Controller ACP recovery identity or cursors cannot regress'
)
}
parsed = previous
}
}
connection.acpBindings = connection.acpBindings.filter(
(candidate) => candidate.bindingId !== parsedBindingId
@@ -1,21 +1,6 @@
import { describe, expect, it, vi } from 'vitest'
import {
runtimeSessionBindingSchema,
type RuntimeSessionBinding
} from '../../shared/remote-agent-contracts'
import type { RemoteModelGateway } from '../agent/remote-model-gateway'
import { RemoteModelGatewayError } from '../agent/remote-model-gateway'
import { ModelCallCapacityError } from '../agent/model-call-operation-store'
import { MemoryRuntimeSessionBindingStore } from '../agent/runtime-session-binding-store'
import { describe, expect, it } from 'vitest'
import type { ResolvedModelProfile } from '../runtime-settings-store'
import {
createManagedModelBridge,
reconcileStartupModelCalls
} from './managed-model-bridge'
import { MainModelBridgeDispatchError } from './main-model-bridge-dispatcher'
const digest = (character: string): string =>
`sha256:${character.repeat(64)}`
import { createManagedModelBridge } from './managed-model-bridge'
const profile: ResolvedModelProfile = {
id: '00000000-0000-4000-8000-000000000001',
@@ -24,307 +9,65 @@ const profile: ResolvedModelProfile = {
modelName: 'private-model',
protocol: 'openai-responses',
authentication: 'api-key',
supportsImageInput: false,
apiKey: 'main-only-secret'
supportsImageInput: true,
maximumOutputTokens: 8_192,
apiKey: 'prompt-scoped-secret'
}
const response = {
status: 200,
headers: { 'content-type': 'application/json' },
bodyBase64: Buffer.from('{}').toString('base64')
}
describe('managed Agent prompt model setup', () => {
it('validates and clones a bounded Agent prompt profile', () => {
const setup = createManagedModelBridge({ profile })
function bridgeFixture(options?: {
bindingStore?: MemoryRuntimeSessionBindingStore
}) {
const bindingStore =
options?.bindingStore ??
new MemoryRuntimeSessionBindingStore()
const dispatch = vi.fn(async () => response)
const markResponseDelivered = vi.fn()
const finalizePrompt = vi.fn()
const gateway = {
dispatch,
markResponseDelivered,
finalizePrompt
} as unknown as RemoteModelGateway
const bridge = createManagedModelBridge({
profile,
gateway,
bindingStore
})
return {
bindingStore,
dispatch,
markResponseDelivered,
finalizePrompt,
bridge
}
}
function binding(
policy: ReturnType<typeof bridgeFixture>['bridge']['policy'],
state: RuntimeSessionBinding['state'] = 'prompt-running'
): RuntimeSessionBinding {
return runtimeSessionBindingSchema.parse({
bindingId: 'binding-1',
controllerId: 'controller-1',
controllerGeneration: 2,
conversationId: 'conversation-1',
hostId: 'host-1',
hostRevision: 1,
hostKeyGeneration: 1,
workspaceIdentity: 'workspace-1',
agentInstallationId: 'agent-1',
daemonBootIdAtOpen: 'boot-1',
runtimeId: 'opencode',
runtimeBundleDigest: digest('a'),
runtimeAdapterDigest: digest('b'),
modelBridgeVersion: 'goodbuddy-model-bridge-v1',
modelBridgePolicy: policy,
acpSessionId: 'session-1',
acpCapabilitiesDigest: digest('c'),
state,
activePromptOperationId:
state === 'prompt-running' || state === 'outcome-unknown'
? 'prompt-1'
: undefined,
channelEpoch: '1',
lastOutboundJournaledSequence: '0',
lastOutboundDeliveredSequence: '0',
lastInboundJournaledSequence: '0',
lastMainAckSequence: '0'
})
}
function dispatchInput(
policy: ReturnType<typeof bridgeFixture>['bridge']['policy']
) {
return {
identity: {
bindingId: 'binding-1',
promptOperationId: 'prompt-1',
requestId: 'local-model-request-1',
roundIndex: 0,
modelProfileDigest: policy.modelProfileDigest,
messageId: 'message-1'
},
policy,
request: {
method: 'POST' as const,
path: '/v1/responses' as const,
headers: { 'content-type': 'application/json' },
bodyBase64: Buffer.from(
JSON.stringify({ model: 'private-model' })
).toString('base64')
},
requestDigest: digest('e')
}
}
describe('managed model bridge composition', () => {
it('does not impose call-count or token quotas', () => {
const fixture = bridgeFixture()
expect(fixture.bridge.policy).toEqual({
protocol: 'openai-responses',
model: 'private-model',
expect(setup.profile).toEqual({
profileId: profile.id,
modelProfileDigest: expect.stringMatching(
/^sha256:[a-f0-9]{64}$/u
),
supportsImageInput: false
provider: 'openai',
baseUrl: 'https://provider.example/v1',
model: 'private-model',
protocol: 'openai-responses',
authentication: 'api-key',
apiKey: 'prompt-scoped-secret',
capabilities: { imageInput: true },
limits: {
maximumOutputTokens: 8_192,
maximumModelCalls: 100,
maximumTotalOutputTokens: 819_200,
requestTimeoutMilliseconds: 60_000
}
})
expect(setup.policy).toEqual({
protocol: 'openai-responses',
model: 'private-model',
modelProfileDigest: setup.profile.modelProfileDigest,
supportsImageInput: true
})
expect(JSON.stringify(setup.policy)).not.toContain(
'prompt-scoped-secret'
)
profile.apiKey = 'changed-after-creation'
expect(setup.profile.apiKey).toBe('prompt-scoped-secret')
})
it('keeps credentials in Main and maps exact dispatch plus delivery identity', async () => {
const fixture = bridgeFixture()
await fixture.bindingStore.put(binding(fixture.bridge.policy))
const input = dispatchInput(fixture.bridge.policy)
await expect(
fixture.bridge.channel.dispatch(
input,
new AbortController().signal
)
).resolves.toEqual(response)
expect(fixture.dispatch).toHaveBeenCalledWith(
expect.objectContaining({
requestId: 'local-model-request-1',
bindingId: 'binding-1',
promptOperationId: 'prompt-1',
promptSequence: 0,
roundIndex: 0,
modelProfile: expect.objectContaining({
apiKey: 'main-only-secret'
})
}),
input.request,
expect.any(AbortSignal)
)
expect(JSON.stringify(fixture.bridge.policy)).not.toContain(
'main-only-secret'
)
expect(JSON.stringify(fixture.bridge.policy)).not.toContain(
'provider.example'
)
await fixture.bridge.channel.onDelivered(
input.identity,
input.requestDigest
)
expect(fixture.markResponseDelivered).toHaveBeenCalledWith({
bindingId: 'binding-1',
promptOperationId: 'prompt-1',
roundIndex: 0
})
await fixture.bridge.channel.finalizePrompt(input.identity)
expect(fixture.finalizePrompt).toHaveBeenCalledWith({
bindingId: 'binding-1',
promptOperationId: 'prompt-1',
promptSequence: 0
})
})
it('classifies only known pre-dispatch gateway failures as definitive', async () => {
const fixture = bridgeFixture()
await fixture.bindingStore.put(binding(fixture.bridge.policy))
vi.mocked(fixture.dispatch).mockRejectedValueOnce(
new RemoteModelGatewayError(
'request-policy-mismatch',
'safe failure'
)
)
await expect(
fixture.bridge.channel.dispatch(
dispatchInput(fixture.bridge.policy),
new AbortController().signal
)
).rejects.toMatchObject({
code: 'request-policy-mismatch',
outcomeUnknown: false,
postDispatch: false
})
vi.mocked(fixture.dispatch).mockRejectedValueOnce(
new ModelCallCapacityError()
)
await expect(
fixture.bridge.channel.dispatch(
dispatchInput(fixture.bridge.policy),
new AbortController().signal
)
).rejects.toMatchObject({
code: 'model-ledger-capacity',
outcomeUnknown: false,
postDispatch: false
})
vi.mocked(fixture.dispatch).mockRejectedValueOnce(
new RemoteModelGatewayError(
'network-error',
'must not escape'
)
)
await expect(
fixture.bridge.channel.dispatch(
dispatchInput(fixture.bridge.policy),
new AbortController().signal
)
).rejects.toBeInstanceOf(MainModelBridgeDispatchError)
await expect(
fixture.bridge.channel.dispatch(
{
...dispatchInput(fixture.bridge.policy),
policy: {
...fixture.bridge.policy,
model: 'different-model'
}
},
new AbortController().signal
)
).rejects.toMatchObject({
code: 'model-policy-mismatch',
outcomeUnknown: false
})
})
it('durably poisons the exact active binding', async () => {
const fixture = bridgeFixture()
await fixture.bindingStore.put(binding(fixture.bridge.policy))
await fixture.bridge.channel.poison(
'binding-1',
'prompt-1',
'delivery-unknown'
)
await expect(
fixture.bindingStore.getById('binding-1')
).resolves.toMatchObject({ state: 'outcome-unknown' })
})
it('paginates startup uncertainty and poisons only matching live prompts', async () => {
const fixture = bridgeFixture()
await fixture.bindingStore.put(binding(fixture.bridge.policy))
const firstCursor = {
updatedAt: 10,
callOperationId: 'call-1'
}
const listStartupRecords = vi
.fn()
.mockReturnValueOnce({
records: [
{
status: 'completed',
identity: {
bindingId: 'binding-1',
promptOperationId: 'prompt-1'
}
}
],
nextCursor: firstCursor
it('rejects unusable profiles before opening a remote binding', () => {
expect(() =>
createManagedModelBridge({
profile: {
...profile,
authentication: 'api-key',
apiKey: undefined
}
})
.mockReturnValueOnce({ records: [] })
await reconcileStartupModelCalls({
gatewayStore: { listStartupRecords },
bindingStore: fixture.bindingStore
})
expect(listStartupRecords).toHaveBeenNthCalledWith(2, {
cursor: firstCursor,
limit: 100
})
await expect(
fixture.bindingStore.getById('binding-1')
).resolves.toMatchObject({ state: 'outcome-unknown' })
})
it('marks a prepared but undispatched startup call as interrupted', async () => {
const fixture = bridgeFixture()
await fixture.bindingStore.put(binding(fixture.bridge.policy))
await reconcileStartupModelCalls({
gatewayStore: {
listStartupRecords: () => ({
records: [
{
status: 'prepared',
identity: {
bindingId: 'binding-1',
promptOperationId: 'prompt-1'
}
}
]
})
},
bindingStore: fixture.bindingStore
})
await expect(
fixture.bindingStore.getById('binding-1')
).resolves.toMatchObject({
state: 'interrupted',
activePromptOperationId: undefined
})
).toThrow(/usable text model profile/iu)
expect(() =>
createManagedModelBridge({
profile: {
...profile,
protocol: 'openai-images-generations'
}
})
).toThrow(/usable text model profile/iu)
})
})
+57 -221
View File
@@ -1,247 +1,83 @@
import { isDeepStrictEqual } from 'node:util'
import {
agentPromptModelProfileSchema,
modelBridgePolicySchema,
type AgentPromptModelProfile,
type ModelBridgePolicy
} from '../../shared/model-bridge-contracts'
import { runtimeSessionBindingSchema } from '../../shared/remote-agent-contracts'
import {
RemoteModelGateway,
RemoteModelGatewayError,
createResolvedModelProfileDigest
} from '../agent/remote-model-gateway'
import { ModelCallCapacityError } from '../agent/model-call-operation-store'
import type { RuntimeSessionBindingStore } from '../agent/runtime-session-binding-store'
import { createResolvedModelProfileDigest } from '../agent/remote-model-gateway'
import type { ResolvedModelProfile } from '../runtime-settings-store'
import {
MainModelBridgeDispatchError,
type MainModelBridgeDispatchInput
} from './main-model-bridge-dispatcher'
import type { ProtocolRemoteRuntimeChannelOptions } from './protocol-remote-runtime-channel'
const PRE_DISPATCH_GATEWAY_ERRORS = new Set([
'invalid-provider-endpoint',
'missing-provider-credential',
'model-profile-digest-mismatch',
'protocol-path-mismatch',
'request-policy-mismatch'
])
const DEFAULT_MAXIMUM_OUTPUT_TOKENS = 32_000
const MAXIMUM_MODEL_CALLS_PER_PROMPT = 100
const MODEL_REQUEST_TIMEOUT_MILLISECONDS = 60_000
/**
* Prompt-scoped model material passed directly to the Agent. The policy is
* safe to persist; the profile (and its credential) is not.
*/
export type ManagedModelBridge = {
policy: ModelBridgePolicy
channel: NonNullable<ProtocolRemoteRuntimeChannelOptions['modelBridge']>
profile: AgentPromptModelProfile
}
export function createManagedModelBridge(options: {
profile: ResolvedModelProfile
gateway: RemoteModelGateway
bindingStore: RuntimeSessionBindingStore
}): ManagedModelBridge {
const profile = structuredClone(options.profile)
const selected = structuredClone(options.profile)
if (
profile.protocol === 'openai-images-generations' ||
(profile.authentication === 'api-key' && !profile.apiKey)
selected.protocol === 'openai-images-generations' ||
(selected.authentication === 'api-key' && !selected.apiKey)
) {
throw new Error(
'Managed remote OpenCode requires a usable text model profile'
)
}
const maximumOutputTokens = Math.min(
1_000_000,
Math.max(
1,
selected.maximumOutputTokens ??
DEFAULT_MAXIMUM_OUTPUT_TOKENS
)
)
const modelProfileDigest =
createResolvedModelProfileDigest(selected)
const profile = agentPromptModelProfileSchema.parse({
profileId: selected.id,
modelProfileDigest,
provider:
selected.protocol === 'anthropic-messages'
? 'anthropic'
: 'openai',
baseUrl: selected.baseUrl,
model: selected.modelName,
protocol: selected.protocol,
authentication: selected.authentication,
...(selected.apiKey === undefined
? {}
: { apiKey: selected.apiKey }),
capabilities: {
imageInput: selected.supportsImageInput === true
},
limits: {
maximumOutputTokens,
maximumModelCalls: MAXIMUM_MODEL_CALLS_PER_PROMPT,
maximumTotalOutputTokens: Math.min(
10_000_000,
maximumOutputTokens * MAXIMUM_MODEL_CALLS_PER_PROMPT
),
requestTimeoutMilliseconds:
MODEL_REQUEST_TIMEOUT_MILLISECONDS
}
})
const policy = modelBridgePolicySchema.parse({
protocol: profile.protocol,
model: profile.modelName,
modelProfileDigest: createResolvedModelProfileDigest(profile),
supportsImageInput: profile.supportsImageInput === true
model: profile.model,
modelProfileDigest: profile.modelProfileDigest,
supportsImageInput: profile.capabilities.imageInput
})
const assertActiveBinding = async (
input: Pick<
MainModelBridgeDispatchInput['identity'],
'bindingId' | 'promptOperationId'
>
) => {
const binding = await options.bindingStore.getById(input.bindingId)
if (
binding === undefined ||
binding.state !== 'prompt-running' ||
binding.activePromptOperationId !== input.promptOperationId ||
binding.modelBridgeVersion !== 'goodbuddy-model-bridge-v1' ||
!isDeepStrictEqual(binding.modelBridgePolicy, policy)
) {
throw new MainModelBridgeDispatchError(
'binding-policy-mismatch',
{ outcomeUnknown: false, postDispatch: false }
)
}
return binding
}
return {
policy,
channel: {
dispatch: async (input, signal) => {
const binding = await assertActiveBinding(input.identity)
if (!isDeepStrictEqual(input.policy, policy)) {
throw new MainModelBridgeDispatchError(
'model-policy-mismatch',
{ outcomeUnknown: false, postDispatch: false }
)
}
try {
return await options.gateway.dispatch(
{
requestId: input.identity.requestId,
bindingId: input.identity.bindingId,
promptOperationId:
input.identity.promptOperationId,
promptSequence: binding.promptSequence,
roundIndex: input.identity.roundIndex,
modelProfileDigest: policy.modelProfileDigest,
modelProfile: profile
},
input.request,
signal
)
} catch (error) {
if (
error instanceof MainModelBridgeDispatchError
) {
throw error
}
if (
error instanceof RemoteModelGatewayError &&
PRE_DISPATCH_GATEWAY_ERRORS.has(error.code)
) {
throw new MainModelBridgeDispatchError(error.code, {
outcomeUnknown: false,
postDispatch: false,
cause: error
})
}
if (error instanceof ModelCallCapacityError) {
throw new MainModelBridgeDispatchError(
'model-ledger-capacity',
{
outcomeUnknown: false,
postDispatch: false,
cause: error
}
)
}
throw new MainModelBridgeDispatchError(
error instanceof RemoteModelGatewayError
? error.code
: 'dispatch-failed',
{
outcomeUnknown: true,
postDispatch: true,
cause: error
}
)
}
},
onDelivered: async (identity) => {
await assertActiveBinding(identity)
options.gateway.markResponseDelivered({
bindingId: identity.bindingId,
promptOperationId: identity.promptOperationId,
roundIndex: identity.roundIndex
})
},
finalizePrompt: async (identity) => {
const binding = await assertActiveBinding(identity)
options.gateway.finalizePrompt({
bindingId: identity.bindingId,
promptOperationId: identity.promptOperationId,
promptSequence: binding.promptSequence
})
},
poison: async (
bindingId,
promptOperationId
) => {
const binding =
await options.bindingStore.getById(bindingId)
if (
binding === undefined ||
binding.state === 'closed' ||
binding.state === 'outcome-unknown' ||
binding.activePromptOperationId !== promptOperationId
) {
return
}
await options.bindingStore.put(
runtimeSessionBindingSchema.parse({
...binding,
state: 'outcome-unknown'
})
)
}
}
profile: structuredClone(profile),
policy
}
}
export async function reconcileStartupModelCalls(options: {
gatewayStore: {
listStartupRecords(input?: {
cursor?: {
updatedAt: number
callOperationId: string
}
limit?: number
}): {
records: Array<{
status:
| 'prepared'
| 'dispatched'
| 'completed'
| 'failed-definitive'
| 'outcome-unknown'
identity: {
bindingId: string
promptOperationId: string
}
}>
nextCursor?: {
updatedAt: number
callOperationId: string
}
}
}
bindingStore: RuntimeSessionBindingStore
}): Promise<void> {
let cursor:
| { updatedAt: number; callOperationId: string }
| undefined
do {
const page = options.gatewayStore.listStartupRecords({
...(cursor === undefined ? {} : { cursor }),
limit: 100
})
for (const record of page.records) {
const binding = await options.bindingStore.getById(
record.identity.bindingId
)
if (
binding === undefined ||
binding.state === 'closed' ||
binding.state === 'outcome-unknown' ||
binding.activePromptOperationId !==
record.identity.promptOperationId
) {
continue
}
await options.bindingStore.put(
runtimeSessionBindingSchema.parse({
...binding,
state:
record.status === 'prepared'
? 'interrupted'
: 'outcome-unknown',
...(record.status === 'prepared'
? { activePromptOperationId: undefined }
: {})
})
)
}
cursor = page.nextCursor
} while (cursor !== undefined)
}
@@ -10,6 +10,8 @@ const remoteHarness = vi.hoisted(() => ({
channelFactory: (bindingId: string) => Promise<unknown>
}
dispose: ReturnType<typeof vi.fn>
beginDrain: ReturnType<typeof vi.fn>
forceShutdown: ReturnType<typeof vi.fn>
runCalls: number
}>,
runGate: undefined as Promise<void> | undefined
@@ -23,6 +25,8 @@ vi.mock('../agent/acp-remote-runtime', () => ({
readonly runtimeId = 'opencode'
readonly supportsToolExecution = true
readonly dispose = vi.fn(async () => undefined)
readonly beginDrain = vi.fn(async () => undefined)
readonly forceShutdown = vi.fn(async () => undefined)
runCalls = 0
constructor(
@@ -61,11 +65,7 @@ vi.mock('../agent/acp-remote-runtime', () => ({
}
async releaseConversation() {}
async beginDrain() {}
async waitForDrain() {}
async forceShutdown() {
await this.dispose()
}
}
}))
@@ -215,11 +215,21 @@ function harness(
modelProfileDigest: digest('9'),
supportsImageInput: false
},
channel: {
dispatch: vi.fn(),
onDelivered: vi.fn(),
finalizePrompt: vi.fn(),
poison: vi.fn()
profile: {
profileId: 'profile-1',
modelProfileDigest: digest('9'),
provider: 'openai',
baseUrl: 'https://provider.example/v1',
model: 'private-model',
protocol: 'openai-responses',
authentication: 'none',
capabilities: { imageInput: false },
limits: {
maximumOutputTokens: 4_096,
maximumModelCalls: 100,
maximumTotalOutputTokens: 409_600,
requestTimeoutMilliseconds: 60_000
}
}
},
...overrides
@@ -289,12 +299,7 @@ describe('createManagedRemoteAcpRuntime', () => {
fixture.options.agentServices.connectionManager.acquire
).toHaveBeenCalledWith('host-1', expect.objectContaining({
requiredCapabilities: [
{ name: 'runtime/acp', exactVersion: 3, critical: true },
{
name: 'runtime/model-bridge',
exactVersion: 1,
critical: true
}
{ name: 'runtime/acp', exactVersion: 4, critical: true }
]
}))
@@ -429,6 +434,36 @@ describe('createManagedRemoteAcpRuntime', () => {
await runtime.dispose()
})
it('abandons connected runtime ownership on application exit without remote cleanup', async () => {
const fixture = harness()
const runtime = await createManagedRemoteAcpRuntime(
fixture.options
)
let finishRun = (): void => undefined
remoteHarness.runGate = new Promise<void>((resolve) => {
finishRun = resolve
})
const stream = runtime.run(
request('request-detach'),
new AbortController().signal
)
await expect(stream.next()).resolves.toMatchObject({
value: { type: 'status' }
})
await runtime.detachForApplicationExit?.()
await runtime.dispose()
expect(remoteHarness.instances[0]!.beginDrain).not.toHaveBeenCalled()
expect(remoteHarness.instances[0]!.forceShutdown).not.toHaveBeenCalled()
expect(remoteHarness.instances[0]!.dispose).not.toHaveBeenCalled()
expect(fixture.workspaceClose).not.toHaveBeenCalled()
expect(fixture.releaseConnection).not.toHaveBeenCalled()
finishRun()
await collect(stream)
})
it('rejects capabilities that differ from the current Runtime', async () => {
const fixture = harness()
vi.mocked(
@@ -481,12 +516,7 @@ function capabilities() {
return {
generation: 7,
capabilities: [
{ name: 'runtime/acp', version: 3, critical: true },
{
name: 'runtime/model-bridge',
version: 1,
critical: true
}
{ name: 'runtime/acp', version: 4, critical: true }
],
runtimes: [
{
@@ -29,12 +29,7 @@ import type {
const RUNTIME_ACP_CAPABILITY = {
name: 'runtime/acp',
exactVersion: 3,
critical: true
} as const
const RUNTIME_MODEL_BRIDGE_CAPABILITY = {
name: 'runtime/model-bridge',
exactVersion: 1,
exactVersion: 4,
critical: true
} as const
@@ -106,8 +101,7 @@ export async function createManagedRemoteAcpRuntime(
{
...agent,
requiredCapabilities: [
RUNTIME_ACP_CAPABILITY,
RUNTIME_MODEL_BRIDGE_CAPABILITY
RUNTIME_ACP_CAPABILITY
]
}
)
@@ -130,8 +124,7 @@ export async function createManagedRemoteAcpRuntime(
workspaceBinding,
agent,
activeConnection,
runtime,
options.modelBridge
runtime
)
const initialTransport =
await resources.prepareTransport(capabilities)
@@ -160,6 +153,7 @@ export async function createManagedRemoteAcpRuntime(
resources.openChannel(transport, bindingId),
bindingStore: options.bindingStore,
modelBridgePolicy: options.modelBridge.policy,
modelProfile: options.modelBridge.profile,
assertHostCurrent: (identity) => {
if (
identity.controllerId !== controllerId ||
@@ -213,6 +207,7 @@ class ManagedRemoteRuntimeResources {
| undefined
#workspaceTail: Promise<void> = Promise.resolve()
#disposePromise?: Promise<void>
#detachedForApplicationExit = false
constructor(
private readonly workspaceBinding: RemoteWorkspaceProjectBinding,
private readonly agent: AgentInstallationIdentity,
@@ -223,8 +218,7 @@ class ManagedRemoteRuntimeResources {
>
>
>,
private readonly runtime: ManagedRemoteRuntimeIdentity,
private readonly modelBridge: ManagedModelBridge
private readonly runtime: ManagedRemoteRuntimeIdentity
) {}
get workspaceIdentity(): string {
@@ -297,8 +291,7 @@ class ManagedRemoteRuntimeResources {
runtimeId: 'opencode',
runtimeBundleDigest: this.runtime.bundleDigest,
workspaceIdentity: this.workspaceIdentity
},
modelBridge: this.modelBridge.channel
}
})
try {
assertSameTransport(
@@ -388,10 +381,18 @@ class ManagedRemoteRuntimeResources {
}
dispose(): Promise<void> {
if (this.#detachedForApplicationExit) {
return Promise.resolve()
}
this.#disposePromise ??= this.#dispose()
return this.#disposePromise
}
detachForApplicationExit(): void {
this.#detachedForApplicationExit = true
this.#workspace = undefined
}
async #dispose(): Promise<void> {
await this.#workspaceTail
const errors: unknown[] = []
@@ -426,6 +427,7 @@ class ManagedRemoteAcpRuntime implements AgentRuntime {
#activeRuns = 0
#draining = false
#disposed = false
#detachedForApplicationExit = false
constructor(
private remote: AcpRemoteRuntime,
@@ -543,6 +545,9 @@ class ManagedRemoteAcpRuntime implements AgentRuntime {
}
forceShutdown(): Promise<void> {
if (this.#detachedForApplicationExit) {
return Promise.resolve()
}
this.#disposed = true
return this.#transitionTail
.then(() => this.remote.forceShutdown())
@@ -550,11 +555,21 @@ class ManagedRemoteAcpRuntime implements AgentRuntime {
}
dispose(): Promise<void> {
if (this.#detachedForApplicationExit) {
return Promise.resolve()
}
this.#disposed = true
this.#disposePromise ??= this.#dispose()
return this.#disposePromise
}
detachForApplicationExit(): void {
this.#disposed = true
this.#draining = true
this.#detachedForApplicationExit = true
this.resources.detachForApplicationExit()
}
async #dispose(): Promise<void> {
await this.#transitionTail
const results = await Promise.allSettled([
@@ -685,9 +700,6 @@ function assertRuntimeAdvertised(
const capability = capabilities.capabilities.find(
(entry) => entry.name === RUNTIME_ACP_CAPABILITY.name
)
const modelBridgeCapability = capabilities.capabilities.find(
(entry) => entry.name === RUNTIME_MODEL_BRIDGE_CAPABILITY.name
)
const advertised = capabilities.runtimes.filter(
(entry) => entry.runtimeId === 'opencode'
)
@@ -696,10 +708,6 @@ function assertRuntimeAdvertised(
capability.version !==
RUNTIME_ACP_CAPABILITY.exactVersion ||
!capability.critical ||
modelBridgeCapability === undefined ||
modelBridgeCapability.version !==
RUNTIME_MODEL_BRIDGE_CAPABILITY.exactVersion ||
!modelBridgeCapability.critical ||
advertised.length !== 1 ||
advertised[0]!.version !== runtime.runtimeVersion ||
advertised[0]!.bundleDigest !== runtime.bundleDigest ||
@@ -1,7 +1,5 @@
import { join } from 'node:path'
import { SqliteRuntimeSessionBindingStore } from '../agent/runtime-session-binding-store'
import { ModelCallOperationStore } from '../agent/model-call-operation-store'
import { RemoteModelGateway } from '../agent/remote-model-gateway'
import type { ResolvedModelProfile } from '../runtime-settings-store'
import type { AgentRuntimeSelection } from '../../shared/runtime-selection-contracts'
import type { SshHostStore } from '../ssh/ssh-host-store'
@@ -22,13 +20,11 @@ import {
type RemoteRuntimeActivator
} from './remote-runtime-installation-manager'
import {
createManagedModelBridge,
reconcileStartupModelCalls
createManagedModelBridge
} from './managed-model-bridge'
import { boundedDiagnostic } from './bounded-diagnostic'
const RUNTIME_BINDING_DATABASE_NAME = 'remote-runtime-bindings.sqlite'
const MODEL_CALL_DATABASE_NAME = 'remote-model-calls-v2.sqlite'
export type ManagedRemoteExecutionServicesOptions = {
sshHostStore: SshHostStore
@@ -62,8 +58,6 @@ export class ManagedRemoteExecutionServices {
readonly runtimeValidator: ManagedRemoteProjectRuntimeValidator
readonly workspaceAccessFactory: ManagedRemoteWorkspaceAccessFactory
readonly bindingStore: SqliteRuntimeSessionBindingStore
readonly modelCallStore: ModelCallOperationStore
readonly modelGateway: RemoteModelGateway
#disposePromise?: Promise<void>
readonly #readiness: Promise<void>
@@ -144,16 +138,7 @@ export class ManagedRemoteExecutionServices {
this.bindingStore = new SqliteRuntimeSessionBindingStore(
join(options.userDataPath, RUNTIME_BINDING_DATABASE_NAME)
)
this.modelCallStore = new ModelCallOperationStore(
join(options.userDataPath, MODEL_CALL_DATABASE_NAME)
)
this.modelGateway = new RemoteModelGateway({
store: this.modelCallStore
})
this.#readiness = reconcileStartupModelCalls({
gatewayStore: this.modelCallStore,
bindingStore: this.bindingStore
})
this.#readiness = Promise.resolve()
}
createProjectSaveService(
@@ -192,9 +177,7 @@ export class ManagedRemoteExecutionServices {
) {
await this.#readiness
const modelBridge = createManagedModelBridge({
profile: options.modelProfile,
gateway: this.modelGateway,
bindingStore: this.bindingStore
profile: options.modelProfile
})
return createManagedRemoteAcpRuntime({
...options,
@@ -228,7 +211,6 @@ export class ManagedRemoteExecutionServices {
}
await settle(() => this.runtimeInstallationManager.dispose())
await settle(() => this.bindingStore.dispose())
await settle(() => this.modelCallStore.dispose())
if (errors.length > 0) {
throw new AggregateError(
errors,
@@ -40,7 +40,7 @@ function advertisedCapabilities(): DaemonCapabilities {
generation: 2,
capabilities: [
{ name: 'workspace/read', version: 1, critical: true },
{ name: 'runtime/acp', version: 3, critical: true },
{ name: 'runtime/acp', version: 4, critical: true },
{
name: 'runtime/model-bridge',
version: 1,
@@ -250,7 +250,7 @@ describe('ManagedRemoteProjectRuntimeValidator', () => {
...advertisedCapabilities(),
capabilities: [
{ name: 'workspace/read', version: 1, critical: true },
{ name: 'runtime/acp', version: 3, critical: true }
{ name: 'runtime/acp', version: 4, critical: true }
]
}
],
@@ -260,7 +260,7 @@ describe('ManagedRemoteProjectRuntimeValidator', () => {
...advertisedCapabilities(),
capabilities: [
{ name: 'workspace/read', version: 1, critical: true },
{ name: 'runtime/acp', version: 3, critical: true },
{ name: 'runtime/acp', version: 4, critical: true },
{
name: 'runtime/model-bridge',
version: 3,
@@ -276,7 +276,7 @@ describe('ManagedRemoteProjectRuntimeValidator', () => {
capabilities: [
{
name: 'runtime/acp',
version: 3,
version: 4,
critical: false
}
]
@@ -12,7 +12,7 @@ import type {
} from './remote-runtime-installation-manager'
const RUNTIME_ACP_CAPABILITY = 'runtime/acp'
const RUNTIME_ACP_CAPABILITY_VERSION = 3
const RUNTIME_ACP_CAPABILITY_VERSION = 4
const RUNTIME_MODEL_BRIDGE_CAPABILITY = 'runtime/model-bridge'
const RUNTIME_MODEL_BRIDGE_CAPABILITY_VERSION = 1
@@ -320,7 +320,7 @@ function validCapabilities(): DaemonCapabilities {
return {
generation: 11,
capabilities: [
{ name: 'runtime/acp', version: 3, critical: true }
{ name: 'runtime/acp', version: 4, critical: true }
],
runtimes: [
{
@@ -406,9 +406,78 @@ function defaultResponse(
processTree: 'empty'
}
}
if ((method as string) === 'runtime/startPrompt') {
const identity = params as {
bindingId: string
operationId: string
requestId: string
}
return {
bindingId: identity.bindingId,
operationId: identity.operationId,
requestId: identity.requestId,
sessionId: 'session-owned-1',
state: 'running',
latestSemanticSequence: '0'
}
}
if ((method as string) === 'runtime/attachPrompt') {
return {
...(params as Record<string, unknown>),
sessionId: 'session-owned-1',
state: 'running',
latestSemanticSequence: '1'
}
}
if ((method as string) === 'runtime/pagePromptTranscript') {
return {
bindingId: 'binding-1',
operationId: 'operation-1',
events: [],
latestSequence: '1',
acknowledgedSequence: '0',
state: 'running',
sessionId: 'session-owned-1',
hasMore: false
}
}
if ((method as string) === 'runtime/ackPromptTranscript') {
return params
}
return params
}
function promptRecoveryResponse(
method: RuntimeProtocolMethod,
params: unknown,
deadlineAt: string
): unknown {
if (method === 'runtime/resumeAcpChannel') {
return {
bindingId: 'binding-1',
channelId: 'binding-1',
channelEpoch: '9',
deadlineAt,
cursors: {
lastOutboundJournaledSequence: '0',
lastOutboundDeliveredSequence: '0',
lastInboundJournaledSequence: '0',
lastMainAckSequence: '0'
}
}
}
if (method === 'runtime/replayAcpChannel') {
return {
bindingId: 'binding-1',
channelId: 'binding-1',
channelEpoch: '9',
replayedThroughSequence: '0',
live: true
}
}
return defaultResponse(method, params)
}
async function openChannel(fixture = connectionFixture()) {
const channel = await createProtocolRemoteRuntimeChannel({
connection: fixture.connection,
@@ -420,6 +489,177 @@ async function openChannel(fixture = connectionFixture()) {
}
describe('ProtocolRemoteRuntimeChannel', () => {
it('maps the complete Agent-owned prompt RPC set', async () => {
const fixture = await openChannel()
const identity = {
bindingId: 'binding-1',
operationId: 'operation-1',
requestId: 'operation-1'
}
await expect(
fixture.channel.startOwnedPrompt({
...identity,
prompt: [{ type: 'text', text: 'hello' }]
})
).resolves.toMatchObject({
sessionId: 'session-owned-1',
state: 'running'
})
await expect(
fixture.channel.attachOwnedPrompt(identity)
).resolves.toMatchObject({
latestSemanticSequence: '1'
})
await expect(
fixture.channel.pageOwnedPromptTranscript({
bindingId: identity.bindingId,
operationId: identity.operationId,
afterSequence: '0',
limit: 10
})
).resolves.toMatchObject({
latestSequence: '1',
state: 'running'
})
await expect(
fixture.channel.ackOwnedPromptTranscript({
bindingId: identity.bindingId,
operationId: identity.operationId,
acknowledgedSequence: '1'
})
).resolves.toMatchObject({
acknowledgedSequence: '1'
})
expect(
fixture.client.requests.map((request) => request.method as string)
).toEqual(expect.arrayContaining([
'runtime/startPrompt',
'runtime/attachPrompt',
'runtime/pagePromptTranscript',
'runtime/ackPromptTranscript'
]))
await fixture.channel.close()
})
it('attaches the exact operation when a prompt start response is lost', async () => {
const fixture = await openChannel()
const deadlineAt = new Date(Date.now() + 10_000).toISOString()
fixture.channel.setRecoveryBoundary(
deadlineAt,
new AbortController().signal
)
const resumed = new FakeRuntimeClient(8)
resumed.responder = (method, params) =>
promptRecoveryResponse(method, params, deadlineAt)
fixture.setReconnectClient(resumed)
let starts = 0
fixture.client.responder = async (method, params) => {
if (method === 'runtime/startPrompt' && starts++ === 0) {
throw new Error('start response lost')
}
return defaultResponse(method, params)
}
await expect(
fixture.channel.startOwnedPrompt({
bindingId: 'binding-1',
operationId: 'operation-1',
requestId: 'operation-1',
prompt: [{ type: 'text', text: 'hello' }]
})
).resolves.toMatchObject({
operationId: 'operation-1',
sessionId: 'session-owned-1'
})
expect(
resumed.requests.map((request) => request.method)
).toContain('runtime/attachPrompt')
expect(
resumed.requests.find(
(request) => request.method === 'runtime/attachPrompt'
)?.params
).toEqual({
bindingId: 'binding-1',
operationId: 'operation-1',
requestId: 'operation-1'
})
await fixture.channel.close()
})
it('reissues one identical start after attach proves the first was not accepted', async () => {
const fixture = await openChannel()
const deadlineAt = new Date(Date.now() + 10_000).toISOString()
fixture.channel.setRecoveryBoundary(
deadlineAt,
new AbortController().signal
)
const resumed = new FakeRuntimeClient(8)
fixture.setReconnectClient(resumed)
let starts = 0
fixture.client.responder = async (method, params) => {
if (method === 'runtime/startPrompt' && starts++ === 0) {
throw new Error('start request not delivered')
}
return promptRecoveryResponse(method, params, deadlineAt)
}
resumed.responder = async (method, params) => {
if (method === 'runtime/attachPrompt') {
throw new AgentRpcError(-32009, 'missing prompt', {
code: 'not-found'
})
}
return promptRecoveryResponse(method, params, deadlineAt)
}
const start = {
bindingId: 'binding-1',
operationId: 'operation-1',
requestId: 'operation-1',
prompt: [{ type: 'text' as const, text: 'hello' }]
}
await expect(
fixture.channel.startOwnedPrompt(start)
).resolves.toMatchObject({
operationId: 'operation-1',
sessionId: 'session-owned-1'
})
const promptStarts = [
...fixture.client.requests,
...resumed.requests
].filter((request) => request.method === 'runtime/startPrompt')
expect(promptStarts).toHaveLength(2)
expect(promptStarts[0]?.params).toEqual(start)
expect(promptStarts[1]?.params).toEqual(start)
await fixture.channel.close()
})
it('preserves autonomous RPC rejection identity for recovery decisions', async () => {
const fixture = await openChannel()
fixture.client.responder = async (method, params) => {
if (method === 'runtime/attachPrompt') {
throw new AgentRpcError(-32009, 'missing prompt', {
code: 'not-found'
})
}
return defaultResponse(method, params)
}
await expect(
fixture.channel.attachOwnedPrompt({
bindingId: 'binding-1',
operationId: 'operation-1',
requestId: 'operation-1'
})
).rejects.toMatchObject({
reason: 'remote',
remoteMethod: 'runtime/attachPrompt',
remoteServiceCode: 'not-found',
remoteRequestOutcome: 'rejected'
})
await fixture.channel.close()
})
it('opens before registering binary and streams bytes in both directions', async () => {
const fixture = await openChannel()
expect(fixture.client.events.slice(0, 2)).toEqual([
@@ -478,6 +718,23 @@ describe('ProtocolRemoteRuntimeChannel', () => {
runtimeId: 'opencode',
runtimeBundleDigest,
runtimeAdapterDigest: digest('d'),
modelProfile: {
profileId: 'profile-1',
modelProfileDigest: digest('e'),
provider: 'openai',
baseUrl: 'https://provider.example/v1',
model: 'test-model',
protocol: 'openai-responses',
authentication: 'none',
capabilities: { imageInput: false },
limits: {
maximumOutputTokens: 4_096,
maximumModelCalls: 100,
maximumTotalOutputTokens: 409_600,
requestTimeoutMilliseconds: 60_000
}
},
promptSequence: 0,
deadlineAt: new Date(Date.now() + 10_000).toISOString(),
budget: {
maximumInputBytes: 1_024,
@@ -21,8 +21,15 @@ import {
import { AgentFrameError } from '../../shared/agent-protocol/frame'
import {
assertRemotePromptAcceptanceMatchesPreparation,
remoteOwnedPromptAttachRequestSchema,
remoteOwnedPromptStartRequestSchema,
remoteOwnedPromptStartResultSchema,
remotePromptOperationAcceptanceSchema,
remotePromptOperationPreparationSchema,
remoteSemanticTranscriptAckRequestSchema,
remoteSemanticTranscriptAckResultSchema,
remoteSemanticTranscriptPageRequestSchema,
remoteSemanticTranscriptPageResultSchema,
UNBOUNDED_REMOTE_PROMPT_DEADLINE
} from '../../shared/remote-agent-contracts'
import {
@@ -38,9 +45,16 @@ import type {
RemotePromptOperationReconciliationResult,
RemoteRuntimeChannel,
RemoteRuntimeChannelCapabilities,
RemoteSemanticTranscriptAckRequest,
RemoteSemanticTranscriptAckResult,
RemoteSemanticTranscriptPageRequest,
RemoteSemanticTranscriptPageResult,
RuntimeSessionBindingCursors
} from '../agent/remote-runtime-channel'
import { StaleRemoteRuntimeGenerationError } from '../agent/remote-runtime-channel'
import {
isDefinitiveRemoteRuntimeRequestRejection,
StaleRemoteRuntimeGenerationError
} from '../agent/remote-runtime-channel'
import {
AgentProtocolClientError,
AgentRpcError,
@@ -62,7 +76,7 @@ import type {
} from './main-model-bridge-dispatcher'
const RUNTIME_ACP_CAPABILITY = 'runtime/acp'
const RUNTIME_ACP_CAPABILITY_VERSION = 3
const RUNTIME_ACP_CAPABILITY_VERSION = 4
const RUNTIME_MODEL_BRIDGE_CAPABILITY = 'runtime/model-bridge'
const RUNTIME_MODEL_BRIDGE_CAPABILITY_VERSION = 1
const DEFAULT_CONTROL_TIMEOUT_MS = 15_000
@@ -80,6 +94,11 @@ export type RuntimeProtocolParams<M extends RuntimeProtocolMethod> =
AgentProtocolParams<M>
export type RuntimeProtocolResult<M extends RuntimeProtocolMethod> =
AgentProtocolResult<M>
type OwnedRuntimeProtocolMethod =
| 'runtime/startPrompt'
| 'runtime/attachPrompt'
| 'runtime/pagePromptTranscript'
| 'runtime/ackPromptTranscript'
export type RuntimeProtocolRequestOptions = AgentProtocolRequestOptions
@@ -401,6 +420,7 @@ export class ProtocolRemoteRuntimeChannel
readonly capabilities: RemoteRuntimeChannelCapabilities = {
cancellationEscalation: true,
promptOperationReconciliation: true,
ownedPrompt: true,
modelBridge: false
}
readonly closed: Promise<void>
@@ -416,6 +436,7 @@ export class ProtocolRemoteRuntimeChannel
#paused = false
#pauseWaiters = new Set<() => void>()
#pendingControlRequests = 0
#pendingUnsafeControlRequests = 0
#activeBinarySends = 0
#recovering = false
#recoveryPromise?: Promise<void>
@@ -594,6 +615,126 @@ export class ProtocolRemoteRuntimeChannel
return acceptance
}
async startOwnedPrompt(
request: z.input<typeof remoteOwnedPromptStartRequestSchema>
): Promise<z.infer<typeof remoteOwnedPromptStartResultSchema>> {
const parsed = remoteOwnedPromptStartRequestSchema.parse(request)
this.#assertOwnedPromptIdentity(parsed)
try {
return this.#parseOwnedPromptIdentity(
parsed,
await this.#ownedRequest(
'runtime/startPrompt',
parsed
)
)
} catch (error) {
if (
error instanceof ProtocolRemoteRuntimeChannelError &&
error.reason === 'transport' &&
this.isCurrentGeneration()
) {
const identity = remoteOwnedPromptAttachRequestSchema.parse({
bindingId: parsed.bindingId,
operationId: parsed.operationId,
requestId: parsed.requestId
})
try {
return await this.attachOwnedPrompt(identity)
} catch (attachError) {
if (
attachError instanceof ProtocolRemoteRuntimeChannelError &&
attachError.remoteServiceCode === 'not-found' &&
isDefinitiveRemoteRuntimeRequestRejection(
attachError,
'runtime/attachPrompt'
)
) {
return this.#parseOwnedPromptIdentity(
parsed,
await this.#ownedRequest(
'runtime/startPrompt',
parsed
)
)
}
throw attachError
}
}
throw error
}
}
async attachOwnedPrompt(
request: z.input<typeof remoteOwnedPromptAttachRequestSchema>
): Promise<z.infer<typeof remoteOwnedPromptStartResultSchema>> {
const parsed = remoteOwnedPromptAttachRequestSchema.parse(request)
this.#assertOwnedPromptIdentity(parsed)
return this.#parseOwnedPromptIdentity(
parsed,
await this.#ownedIdempotentRequest(
'runtime/attachPrompt',
parsed
)
)
}
async pageOwnedPromptTranscript(
request: RemoteSemanticTranscriptPageRequest
): Promise<RemoteSemanticTranscriptPageResult> {
const parsed = remoteSemanticTranscriptPageRequestSchema.parse(
request
)
this.#assertOwnedPromptIdentity(parsed)
const result = parseRemoteResult(
remoteSemanticTranscriptPageResultSchema,
await this.#ownedIdempotentRequest(
'runtime/pagePromptTranscript',
parsed
),
'Remote Runtime semantic transcript page is invalid'
)
if (
result.bindingId !== parsed.bindingId ||
result.operationId !== parsed.operationId
) {
throw new ProtocolRemoteRuntimeChannelError(
'Remote Runtime semantic transcript identity is invalid',
'binding-mismatch'
)
}
return result
}
async ackOwnedPromptTranscript(
request: RemoteSemanticTranscriptAckRequest
): Promise<RemoteSemanticTranscriptAckResult> {
const parsed = remoteSemanticTranscriptAckRequestSchema.parse(
request
)
this.#assertOwnedPromptIdentity(parsed)
const result = parseRemoteResult(
remoteSemanticTranscriptAckResultSchema,
await this.#ownedIdempotentRequest(
'runtime/ackPromptTranscript',
parsed
),
'Remote Runtime semantic transcript ACK is invalid'
)
if (
result.bindingId !== parsed.bindingId ||
result.operationId !== parsed.operationId ||
BigInt(result.acknowledgedSequence) <
BigInt(parsed.acknowledgedSequence)
) {
throw new ProtocolRemoteRuntimeChannelError(
'Remote Runtime semantic transcript ACK identity is invalid',
'binding-mismatch'
)
}
return result
}
setRecoveryBoundary(
deadlineAt: string,
signal: AbortSignal
@@ -906,6 +1047,7 @@ export class ProtocolRemoteRuntimeChannel
)
}
this.#pendingControlRequests += 1
this.#pendingUnsafeControlRequests += 1
try {
const result = await requestWithTimeout(
this.#state.client,
@@ -923,6 +1065,71 @@ export class ProtocolRemoteRuntimeChannel
throw redactProtocolFailure(error, method)
} finally {
this.#pendingControlRequests -= 1
this.#pendingUnsafeControlRequests -= 1
}
}
async #ownedRequest<M extends OwnedRuntimeProtocolMethod>(
method: M,
params: RuntimeProtocolParams<M>
): Promise<RuntimeProtocolResult<M>> {
await this.#waitForRecovery()
this.#assertCurrent()
if (
this.#pendingControlRequests >=
this.#state.maximumPendingControlRequests
) {
throw new ProtocolRemoteRuntimeChannelError(
'Remote Runtime control request limit reached',
'capacity'
)
}
this.#pendingControlRequests += 1
try {
const result = await requestWithTimeout(
this.#state.client,
method,
params,
this.#state.controlTimeoutMs,
this.#lifetime.signal
)
this.#assertCurrent()
return result
} catch (error) {
if (error instanceof StaleRemoteRuntimeGenerationError) {
throw error
}
if (
!this.#closing &&
!this.#lifetime.signal.aborted &&
!(error instanceof AgentRpcError)
) {
await this.#transportInterrupted().catch(() => undefined)
await this.#waitForRecovery().catch(() => undefined)
}
throw redactProtocolFailure(error, method)
} finally {
this.#pendingControlRequests -= 1
}
}
async #ownedIdempotentRequest<
M extends Exclude<OwnedRuntimeProtocolMethod, 'runtime/startPrompt'>
>(
method: M,
params: RuntimeProtocolParams<M>
): Promise<RuntimeProtocolResult<M>> {
try {
return await this.#ownedRequest(method, params)
} catch (error) {
if (
error instanceof ProtocolRemoteRuntimeChannelError &&
error.reason === 'transport' &&
this.isCurrentGeneration()
) {
return await this.#ownedRequest(method, params)
}
throw error
}
}
@@ -970,6 +1177,49 @@ export class ProtocolRemoteRuntimeChannel
}
}
#assertOwnedPromptIdentity(request: {
bindingId: string
operationId: string
requestId?: string
}): void {
this.#assertBinding(request.bindingId)
if (
request.requestId !== undefined &&
request.operationId !== request.requestId
) {
throw new ProtocolRemoteRuntimeChannelError(
'Remote Runtime owned prompt operation identity is invalid',
'binding-mismatch'
)
}
}
#parseOwnedPromptIdentity(
request: {
bindingId: string
operationId: string
requestId: string
},
raw: unknown
): z.infer<typeof remoteOwnedPromptStartResultSchema> {
const result = parseRemoteResult(
remoteOwnedPromptStartResultSchema,
raw,
'Remote Runtime owned prompt response is invalid'
)
if (
result.bindingId !== request.bindingId ||
result.operationId !== request.operationId ||
result.requestId !== request.requestId
) {
throw new ProtocolRemoteRuntimeChannelError(
'Remote Runtime owned prompt response identity is invalid',
'binding-mismatch'
)
}
return result
}
async #waitUntilResumed(): Promise<void> {
while (this.#paused && !this.#closing) {
await new Promise<void>((resolve) => {
@@ -1053,7 +1303,7 @@ export class ProtocolRemoteRuntimeChannel
this.#recoverySignal.aborted ||
recoveryDeadlineExpired ||
this.#activeBinarySends > 0 ||
this.#pendingControlRequests > 0 ||
this.#pendingUnsafeControlRequests > 0 ||
providerDeliveryMayBeUncertain ||
this.#state.connection.reconnect === undefined
) {
@@ -231,6 +231,76 @@ function harness(
}
describe('RemoteAgentConnectionManager', () => {
it('publishes the aggregate live connection state for each Host', async () => {
let finishTransport!: () => void
const gate = new Promise<void>((resolve) => {
finishTransport = resolve
})
const test = harness(gate, undefined, 1_000)
const states: string[] = []
const unsubscribe =
test.manager.onHostConnectionStateChange(({ state }) => {
states.push(state)
})
expect(
test.manager.getHostConnectionState('host-1')
).toBe('disconnected')
const acquiring = test.manager.acquire('host-1', installation)
await vi.waitFor(() =>
expect(states).toEqual(['connecting'])
)
finishTransport()
const lease = await acquiring
expect(
test.manager.getHostConnectionState('host-1')
).toBe('ready')
test.clients[0]!.transportClose()
expect(
test.manager.getHostConnectionState('host-1')
).toBe('error')
lease.release()
await test.manager.dispose()
expect(states).toEqual([
'connecting',
'ready',
'error',
'disconnected'
])
unsubscribe()
})
it('clears a retained connection error when its Host is invalidated', async () => {
const test = harness()
const states: string[] = []
test.manager.onHostConnectionStateChange(({ state }) => {
states.push(state)
})
test.connectTransport.mockRejectedValueOnce(
new Error('attach failed')
)
await expect(
test.manager.acquire('host-1', installation)
).rejects.toThrow()
expect(
test.manager.getHostConnectionState('host-1')
).toBe('error')
await test.manager.invalidateHost('host-1')
expect(
test.manager.getHostConnectionState('host-1')
).toBe('disconnected')
expect(states).toEqual([
'connecting',
'error',
'disconnected'
])
await test.manager.dispose()
})
it('deduplicates concurrent connects and ref-counts the shared connection', async () => {
const test = harness()
const [first, second] = await Promise.all([
@@ -32,6 +32,10 @@ import {
type ControllerAcpRecoveryBinding,
type ControllerConnectionState
} from './controller-state-store'
import type {
SshHostAgentConnectionState,
SshHostAgentConnectionStatus
} from '../../shared/ssh-host-contracts'
const DEFAULT_IDLE_TIMEOUT_MS = 60_000
const DEFAULT_ATTACH_TIMEOUT_MS = 10_000
@@ -168,6 +172,13 @@ export class RemoteAgentConnectionManager {
readonly #attachTimeoutMs: number
readonly #identityKey = randomBytes(32)
readonly #entries = new Map<string, ConnectionEntry>()
readonly #hostConnectionStates = new Map<
string,
SshHostAgentConnectionState
>()
readonly #hostConnectionListeners = new Set<
(status: SshHostAgentConnectionStatus) => void
>()
#disposed = false
constructor(options: {
@@ -261,6 +272,7 @@ export class RemoteAgentConnectionManager {
}
entry.connectPromise = this.#connect(entry, target, false)
this.#entries.set(identity.cacheKey, entry)
this.#syncHostConnectionState(hostId)
} else if (
entry.connectionState === 'terminal' ||
entry.connectionState === 'disposed'
@@ -306,6 +318,41 @@ export class RemoteAgentConnectionManager {
}
}
getHostConnectionState(
hostId: string
): SshHostAgentConnectionState {
return this.#hostConnectionStates.get(hostId) ?? 'disconnected'
}
#deriveHostConnectionState(
hostId: string,
emptyState: SshHostAgentConnectionState = 'disconnected'
): SshHostAgentConnectionState {
const states = [...this.#entries.values()]
.filter((entry) => entry.hostId === hostId)
.map((entry) => entry.connectionState)
if (states.includes('ready')) {
return 'ready'
}
if (states.includes('connecting')) {
return 'connecting'
}
if (
states.includes('offline') ||
states.includes('terminal')
) {
return 'error'
}
return emptyState
}
onHostConnectionStateChange(
listener: (status: SshHostAgentConnectionStatus) => void
): () => void {
this.#hostConnectionListeners.add(listener)
return () => this.#hostConnectionListeners.delete(listener)
}
async invalidateHost(hostId: string): Promise<void> {
for (const entry of [...this.#entries.values()]) {
if (entry.hostId === hostId) {
@@ -317,6 +364,7 @@ export class RemoteAgentConnectionManager {
this.#disposeEntry(entry)
}
}
this.#syncHostConnectionState(hostId)
this.#sshPool.disposeHost(hostId)
await this.#controllerState.invalidateHost(hostId)
}
@@ -333,6 +381,7 @@ export class RemoteAgentConnectionManager {
)
this.#disposeEntry(entry)
}
this.#hostConnectionListeners.clear()
await this.#controllerState.flush()
this.#controllerState.dispose()
}
@@ -447,6 +496,7 @@ export class RemoteAgentConnectionManager {
previous,
previous !== undefined
)
this.#syncHostConnectionState(entry.hostId)
for (const listener of entry.clientChangeListeners) {
try {
listener()
@@ -477,6 +527,7 @@ export class RemoteAgentConnectionManager {
if (entry.refs === 0) {
this.#entries.delete(entry.key)
}
this.#syncHostConnectionState(entry.hostId, 'error')
}
throw classified
} finally {
@@ -834,6 +885,7 @@ export class RemoteAgentConnectionManager {
}
entry.controller = new AbortController()
entry.connectionState = 'connecting'
this.#syncHostConnectionState(entry.hostId)
entry.connectPromise = this.#connect(entry, target, true)
await entry.connectPromise
})()
@@ -862,6 +914,7 @@ export class RemoteAgentConnectionManager {
error
)
entry.connectionState = 'offline'
this.#syncHostConnectionState(entry.hostId)
active?.unsubscribeClose()
active?.sshLease.release()
if (entry.refs === 0 && entry.waiters === 0) {
@@ -986,6 +1039,34 @@ export class RemoteAgentConnectionManager {
entry.connection?.sshLease.release()
entry.connection = undefined
entry.clientChangeListeners.clear()
this.#syncHostConnectionState(entry.hostId)
}
#syncHostConnectionState(
hostId: string,
emptyState: SshHostAgentConnectionState = 'disconnected'
): void {
const previousState = this.getHostConnectionState(hostId)
const state = this.#deriveHostConnectionState(
hostId,
emptyState
)
if (state === previousState) {
return
}
if (state === 'disconnected') {
this.#hostConnectionStates.delete(hostId)
} else {
this.#hostConnectionStates.set(hostId, state)
}
const status = { hostId, state }
for (const listener of this.#hostConnectionListeners) {
try {
listener(status)
} catch {
// One observer cannot interrupt connection ownership.
}
}
}
}