diff --git a/CHANGELOG.md b/CHANGELOG.md index 439539b854..8932cacc0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,9 @@ ### Changed +- Made typed `request()` the sole direct Runtime Host operation API; removed the 17 forwarding + aliases from direct and reconnecting connections while preserving status validation, + subscriptions, capabilities, listeners, lifecycle, and close behavior. - Collapsed the RuntimeRunner/Flow/Invocation shell into `RuntimeKernel`; backend dispatch, terminal coalescing, stop/drain, and durable continuation admission now have one production owner, immutable request snapshots remain enforced at AgentRun acceptance and backend dispatch, diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index de21a46047..ffc49d1099 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -1150,7 +1150,7 @@ export class DesktopRuntimeHostClient { } queryHostDiagnostics(): Promise> { - return this.connection.queryHostDiagnostics(2_000); + return this.connection.request('host.diagnostics.query', {}, 2_000); } prepareHostRetirement( diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 1b9dc7abb1..a6fb92e2db 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1606,7 +1606,6 @@ class FakeConnection { hostEpoch: 'host-1', request: (operation: K, input: OperationInput) => this.request(operation, input), - startTurn: (input) => this.request('turn.start', input), openSessionSubscription: async () => { const subscription = this.subscriptions[this.openedSubscriptions]; this.openedSubscriptions += 1; diff --git a/packages/cli/src/runtime-host-service-manager.ts b/packages/cli/src/runtime-host-service-manager.ts index 23903f2606..f727eaa754 100644 --- a/packages/cli/src/runtime-host-service-manager.ts +++ b/packages/cli/src/runtime-host-service-manager.ts @@ -1068,7 +1068,7 @@ export async function verifyRuntimeHostManagedServiceReady( const status = await connected.connection.status(Math.max(1, remaining)); if (status.state === 'ready') { const [diagnostics, service] = await Promise.all([ - connected.connection.queryHostDiagnostics(), + connected.connection.request('host.diagnostics.query', {}), backend.status(), ]); if (service.active && service.pid !== null && diagnostics.pid === service.pid) return; @@ -1118,7 +1118,7 @@ async function prepareRuntimeHostRetirement( } const hostEpoch = connected.connection.hostEpoch; try { - const diagnostics = await connected.connection.queryHostDiagnostics(); + const diagnostics = await connected.connection.request('host.diagnostics.query', {}); if (diagnostics.pid !== expectedPid) { throw new RuntimeHostServiceManagerError( 'retirement_failed', diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts index 6b7cf0f2a8..2eeebf0e8c 100644 --- a/packages/cli/src/runtime-host-session-driver.ts +++ b/packages/cli/src/runtime-host-session-driver.ts @@ -119,7 +119,7 @@ export interface RuntimeHostMakaSessionDriverInput { type RuntimeHostSessionDriverConnection = Pick< RuntimeHostConnection, - 'hostEpoch' | 'openSessionSubscription' | 'request' | 'startTurn' + 'hostEpoch' | 'openSessionSubscription' | 'request' >; export interface RuntimeHostMakaSessionDriver extends MakaSessionDriver { @@ -286,7 +286,7 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { ...(options.turnOrchestration ? { turnOrchestration: options.turnOrchestration } : {}), ...(options.maxSteps !== undefined ? { maxSteps: options.maxSteps } : {}), }; - const result = await this.#connection.startTurn(startInput); + const result = await this.#connection.request('turn.start', startInput); if (result.kind === 'blocked') { throw new SkillInvocationBlockedError(result.skillInvocation); } diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index a80204812e..5760ad57e2 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -102,7 +102,7 @@ test('concurrent responses remain framed and correlated in reverse completion or async ({ connectClient }) => { const client = await connectClient(); const requests = Array.from({ length: requestCount }, (_, index) => - client.queryTurn({ sessionId: 'session', turnId: `turn-${index}` }, 5_000), + client.request('turn.query', { sessionId: 'session', turnId: `turn-${index}` }, 5_000), ); try { await withTimeout( @@ -257,7 +257,7 @@ test('the Client backpressures a healthy request burst at the Host connection li async ({ connectClient }) => { const client = await connectClient(); const requests = Array.from({ length: requestCount }, (_, index) => - client.queryTurn({ sessionId: 'session', turnId: `burst-${index}` }, 5_000), + client.request('turn.query', { sessionId: 'session', turnId: `burst-${index}` }, 5_000), ); try { await withTimeout(firstWaveEntered.promise, 1_000, 'first request wave was not admitted'); @@ -656,7 +656,7 @@ test('an admitted operation settles without connection or residency leakage afte async ({ connectClient }) => { const client = await connectClient(); const requestFailure = client - .queryTurn({ sessionId: 'session', turnId: 'disconnect' }, 5_000) + .request('turn.query', { sessionId: 'session', turnId: 'disconnect' }, 5_000) .then( () => undefined, (error: unknown) => error, @@ -698,7 +698,7 @@ test('an admitted command reports an unknown outcome when its connection closes' }), async ({ connectClient }) => { const client = await connectClient(); - const command = client.startTurn({ + const command = client.request('turn.start', { sessionId: 'session', turnId: 'interrupted-command', content: { text: 'start' }, diff --git a/packages/runtime-host/src/__tests__/daily-review-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/daily-review-two-client-uds.test.ts index 36e128c21a..fac9c490a5 100644 --- a/packages/runtime-host/src/__tests__/daily-review-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/daily-review-two-client-uds.test.ts @@ -84,19 +84,19 @@ test('two Clients share Daily Review config, generation, and restart recovery', owner = undefined; [desktop, tui] = await Promise.all([connect(root), connect(root)]); - const initial = await desktop.queryDailyReview({ kind: 'config' }); + const initial = await desktop.request('daily-review.query', { kind: 'config' }); assert.deepEqual(initial, { kind: 'config', revision: 0, config: { enabled: false, executeTime: '08:00', modelKey: '' }, }); const mutations = await Promise.all([ - desktop.mutateDailyReview({ + desktop.request('daily-review.mutate', { kind: 'update_config', expectedRevision: 0, config: { enabled: false, executeTime: '09:00', modelKey: '' }, }), - tui.mutateDailyReview({ + tui.request('daily-review.mutate', { kind: 'update_config', expectedRevision: 0, config: { enabled: false, executeTime: '10:00', modelKey: '' }, @@ -105,8 +105,8 @@ test('two Clients share Daily Review config, generation, and restart recovery', assert.equal(mutations.filter((result) => result.kind === 'config_committed').length, 1); assert.equal(mutations.filter((result) => result.kind === 'revision_conflict').length, 1); assert.deepEqual( - await desktop.queryDailyReview({ kind: 'config' }), - await tui.queryDailyReview({ kind: 'config' }), + await desktop.request('daily-review.query', { kind: 'config' }), + await tui.request('daily-review.query', { kind: 'config' }), ); const run = { @@ -117,15 +117,15 @@ test('two Clients share Daily Review config, generation, and restart recovery', replaceExisting: false, }; const [desktopRun, tuiRun] = await Promise.all([ - desktop.mutateDailyReview(run), - tui.mutateDailyReview(run), + desktop.request('daily-review.mutate', run), + tui.request('daily-review.mutate', run), ]); assert.deepEqual(tuiRun, desktopRun); assert.equal(desktopRun.kind, 'archive'); if (desktopRun.kind !== 'archive') return; assert.equal(desktopRun.archive.status, 'no_data'); - const noModel = await desktop.mutateDailyReview({ + const noModel = await desktop.request('daily-review.mutate', { ...run, offsetDays: 0, modelKeyOverride: 'missing-provider::missing-model', @@ -135,7 +135,7 @@ test('two Clients share Daily Review config, generation, and restart recovery', assert.equal(noModel.archive.status, 'no_model'); assert.equal(noModel.archive.totals.requestCount, 1); - const enabled = await desktop.mutateDailyReview({ + const enabled = await desktop.request('daily-review.mutate', { kind: 'update_config', expectedRevision: 1, config: { @@ -164,14 +164,14 @@ test('two Clients share Daily Review config, generation, and restart recovery', owner = undefined; tui = await connect(root); assert.deepEqual( - await tui.queryDailyReview({ + await tui.request('daily-review.query', { kind: 'archive', archiveId: desktopRun.archive.id, }), { kind: 'archive', archive: desktopRun.archive }, ); assert.deepEqual( - await tui.queryDailyReview({ + await tui.request('daily-review.query', { kind: 'archive', archiveId: scheduled.id, }), @@ -201,7 +201,7 @@ async function waitForScheduledArchive( ): Promise { const deadline = Date.now() + 5_000; while (Date.now() < deadline) { - const page = await connection.queryDailyReview({ + const page = await connection.request('daily-review.query', { kind: 'archives', beforeArchiveId: null, limit: 10, @@ -209,7 +209,7 @@ async function waitForScheduledArchive( if (page.kind === 'archives') { const scheduled = page.archives.find((archive) => archive.trigger === 'cron'); if (scheduled) { - const result = await connection.queryDailyReview({ + const result = await connection.request('daily-review.query', { kind: 'archive', archiveId: scheduled.id, }); diff --git a/packages/runtime-host/src/__tests__/deep-research-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/deep-research-two-client-uds.test.ts index ef082b3813..54cc1dd802 100644 --- a/packages/runtime-host/src/__tests__/deep-research-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/deep-research-two-client-uds.test.ts @@ -88,8 +88,8 @@ test('two Clients and a restarted production Host share one Deep Research projec [desktop, tui] = await Promise.all([connect(root), connect(root)]); const [desktopProjection, tuiProjection] = await Promise.all([ - desktop.queryDeepResearch({ sessionId: session.id }), - tui.queryDeepResearch({ sessionId: session.id }), + desktop.request('deep-research.query', { sessionId: session.id }), + tui.request('deep-research.query', { sessionId: session.id }), ]); assert.deepEqual(tuiProjection, desktopProjection); assert.equal(desktopProjection.kind, 'snapshot'); @@ -128,7 +128,10 @@ test('two Clients and a restarted production Host share one Deep Research projec owner = undefined; tui = await connect(root); - assert.deepEqual(await tui.queryDeepResearch({ sessionId: session.id }), desktopProjection); + assert.deepEqual( + await tui.request('deep-research.query', { sessionId: session.id }), + desktopProjection, + ); } finally { await Promise.allSettled([desktop?.close(), tui?.close()]); await host?.close().catch(() => undefined); diff --git a/packages/runtime-host/src/__tests__/execution-host-continuation.test.ts b/packages/runtime-host/src/__tests__/execution-host-continuation.test.ts index 77b5d228b7..a0dae34fea 100644 --- a/packages/runtime-host/src/__tests__/execution-host-continuation.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-continuation.test.ts @@ -38,7 +38,7 @@ test('two Clients idempotently start one Host-owned safe-boundary continuation', let clientsClosed = false; let hostStopped = false; try { - const plan = await first.queryTurnResume({ sessionId: fixture.sessionId }); + const plan = await first.request('turn.resume.query', { sessionId: fixture.sessionId }); assert.deepEqual(plan, { sessionId: fixture.sessionId, disposition: 'ready', @@ -54,8 +54,8 @@ test('two Clients idempotently start one Host-owned safe-boundary continuation', sourceRuntimeEventHighWater: source.sourceRuntimeEventHighWater, }; const [firstStart, secondStart] = await Promise.all([ - first.startTurnResume(input), - second.startTurnResume(input), + first.request('turn.resume.start', input), + second.request('turn.resume.start', input), ]); assert.equal(firstStart.kind, 'started'); assert.equal(secondStart.kind, 'started'); @@ -64,10 +64,12 @@ test('two Clients idempotently start one Host-owned safe-boundary continuation', const terminal = await waitForTerminalTurn(first, fixture.sessionId, turnId); assert.equal(terminal.status, 'completed'); - const retry = await second.startTurnResume(input); + const retry = await second.request('turn.resume.start', input); assert.deepEqual(retry, { kind: 'started', turn: terminal }); - const settledPlan = await first.queryTurnResume({ sessionId: fixture.sessionId }); + const settledPlan = await first.request('turn.resume.query', { + sessionId: fixture.sessionId, + }); assert.deepEqual(settledPlan, { sessionId: fixture.sessionId, disposition: 'parked', @@ -125,7 +127,7 @@ test('startup repairs a continuation Run created before its durable start', asyn const host = await fixture.startHost(); const client = await connectClient(fixture.root); try { - const repaired = await client.queryTurn({ + const repaired = await client.request('turn.query', { sessionId: fixture.sessionId, turnId: crash.targetTurnId, }); @@ -133,7 +135,7 @@ test('startup repairs a continuation Run created before its durable start', asyn assert.equal(repaired.status, 'failed'); assert.equal(repaired.failureClass, 'continuation_abandoned_before_provider_dispatch'); assert.deepEqual( - await client.queryTurnResume({ + await client.request('turn.resume.query', { sessionId: fixture.sessionId, sourceRunId: crash.sourceRunId, expectedRuntimeEventHighWater: crash.sourceRuntimeEventHighWater, @@ -161,7 +163,7 @@ test('startup repairs a continuation claim committed before its target Run', asy let clientClosed = false; let hostStopped = false; try { - const repaired = await client.queryTurn({ + const repaired = await client.request('turn.query', { sessionId: fixture.sessionId, turnId: crash.targetTurnId, }); @@ -192,7 +194,7 @@ test('startup parks a provider-indeterminate continuation without blocking the H const host = await fixture.startHost(); const client = await connectClient(fixture.root); try { - const indeterminate = await client.queryTurn({ + const indeterminate = await client.request('turn.query', { sessionId: fixture.sessionId, turnId: crash.targetTurnId, }); @@ -204,7 +206,7 @@ test('startup parks a provider-indeterminate continuation without blocking the H reason: 'continuation_started_indeterminate' as const, }; assert.deepEqual( - await client.queryTurnResume({ + await client.request('turn.resume.query', { sessionId: fixture.sessionId, sourceRunId: crash.sourceRunId, expectedRuntimeEventHighWater: crash.sourceRuntimeEventHighWater, @@ -212,7 +214,7 @@ test('startup parks a provider-indeterminate continuation without blocking the H plan, ); assert.deepEqual( - await client.startTurnResume({ + await client.request('turn.resume.start', { sessionId: fixture.sessionId, turnId: crash.targetTurnId, sourceRunId: crash.sourceRunId, @@ -222,7 +224,7 @@ test('startup parks a provider-indeterminate continuation without blocking the H ); await assert.rejects( () => - client.stopTurn({ + client.request('turn.stop', { sessionId: fixture.sessionId, turnId: crash.targetTurnId, runId: crash.targetRunId, @@ -231,7 +233,7 @@ test('startup parks a provider-indeterminate continuation without blocking the H error instanceof RuntimeHostOperationError && error.code === 'operation_conflict', ); assert.deepEqual( - await client.queryTurnResume({ + await client.request('turn.resume.query', { sessionId: fixture.sessionId, sourceRunId: 'different-continuation-source', expectedRuntimeEventHighWater: crash.sourceRuntimeEventHighWater, @@ -244,7 +246,7 @@ test('startup parks a provider-indeterminate continuation without blocking the H ); await assert.rejects( () => - client.startTurn({ + client.request('turn.start', { sessionId: fixture.sessionId, turnId: 'turn-after-indeterminate-continuation', content: { text: 'Do not overtake the parked continuation.' }, @@ -253,7 +255,7 @@ test('startup parks a provider-indeterminate continuation without blocking the H ); const sibling = requireStartedTurn( - await client.startTurn({ + await client.request('turn.start', { sessionId: siblingSessionId, turnId: 'turn-unrelated-to-indeterminate-continuation', content: { text: 'Continue normally.' }, @@ -276,14 +278,14 @@ test('startup parks a provider-indeterminate continuation when resume is disable const host = await fixture.startHost(undefined, false); const client = await connectClient(fixture.root); try { - const indeterminate = await client.queryTurn({ + const indeterminate = await client.request('turn.query', { sessionId: fixture.sessionId, turnId: crash.targetTurnId, }); assert.equal(indeterminate.runId, crash.targetRunId); assert.equal(indeterminate.status === 'created' || indeterminate.status === 'running', true); assert.deepEqual( - await client.queryTurnResume({ + await client.request('turn.resume.query', { sessionId: fixture.sessionId, sourceRunId: crash.sourceRunId, expectedRuntimeEventHighWater: crash.sourceRuntimeEventHighWater, @@ -295,7 +297,7 @@ test('startup parks a provider-indeterminate continuation when resume is disable }, ); const sibling = requireStartedTurn( - await client.startTurn({ + await client.request('turn.start', { sessionId: siblingSessionId, turnId: 'turn-unrelated-to-disabled-indeterminate-continuation', content: { text: 'Continue normally.' }, @@ -317,7 +319,7 @@ test('startup parks a pre-claim continuation whose Client Capability is absent', const client = await connectClient(fixture.root); try { assert.deepEqual( - await client.queryTurnResume({ + await client.request('turn.resume.query', { sessionId: fixture.sessionId, sourceRunId: pending.sourceRunId, expectedRuntimeEventHighWater: pending.sourceRuntimeEventHighWater, @@ -330,7 +332,7 @@ test('startup parks a pre-claim continuation whose Client Capability is absent', ); await assert.rejects( () => - client.startTurn({ + client.request('turn.start', { sessionId: fixture.sessionId, turnId: 'turn-overtaking-client-capability-continuation', content: { text: 'Do not overtake the pending continuation.' }, @@ -338,7 +340,7 @@ test('startup parks a pre-claim continuation whose Client Capability is absent', (error) => error instanceof RuntimeHostOperationError && error.code === 'session_busy', ); assert.deepEqual( - await client.queryTurn({ + await client.request('turn.query', { sessionId: fixture.sessionId, turnId: pending.targetTurnId, }), @@ -351,7 +353,7 @@ test('startup parks a pre-claim continuation whose Client Capability is absent', ); await assert.rejects( () => - client.stopTurn({ + client.request('turn.stop', { sessionId: fixture.sessionId, turnId: pending.targetTurnId, runId: pending.targetRunId, @@ -360,7 +362,7 @@ test('startup parks a pre-claim continuation whose Client Capability is absent', error instanceof RuntimeHostOperationError && error.code === 'operation_conflict', ); assert.deepEqual( - await client.queryTurnResume({ + await client.request('turn.resume.query', { sessionId: fixture.sessionId, sourceRunId: pending.sourceRunId, expectedRuntimeEventHighWater: pending.sourceRuntimeEventHighWater, @@ -393,7 +395,7 @@ test('Runtime Host keeps safe-boundary continuation opt-in', async () => { reason: 'continuation_unavailable' as const, }; assert.deepEqual( - await client.queryTurnResume({ + await client.request('turn.resume.query', { sessionId: fixture.sessionId, sourceRunId: source.sourceRunId, expectedRuntimeEventHighWater: source.sourceRuntimeEventHighWater, @@ -401,7 +403,7 @@ test('Runtime Host keeps safe-boundary continuation opt-in', async () => { plan, ); assert.deepEqual( - await client.startTurnResume({ + await client.request('turn.resume.start', { sessionId: fixture.sessionId, turnId: targetTurnId, sourceRunId: source.sourceRunId, @@ -436,19 +438,29 @@ test('resume query previews the initiating Client Capability without binding it' const client = await connectClient(fixture.root); const provider = resumeFixtureProvider(serverId, [toolName]); try { - assert.deepEqual(await client.queryTurnResume({ sessionId: fixture.sessionId }), { - sessionId: fixture.sessionId, - disposition: 'parked', - reason: 'safety_check_failed', - }); + assert.deepEqual( + await client.request('turn.resume.query', { + sessionId: fixture.sessionId, + }), + { + sessionId: fixture.sessionId, + disposition: 'parked', + reason: 'safety_check_failed', + }, + ); await client.replaceClientCapabilities(provider); - assert.deepEqual(await client.queryTurnResume({ sessionId: fixture.sessionId }), { - sessionId: fixture.sessionId, - disposition: 'ready', - sourceRunId: source.sourceRunId, - sourceTurnId: source.sourceTurnId, - sourceRuntimeEventHighWater: source.sourceRuntimeEventHighWater, - }); + assert.deepEqual( + await client.request('turn.resume.query', { + sessionId: fixture.sessionId, + }), + { + sessionId: fixture.sessionId, + disposition: 'ready', + sourceRunId: source.sourceRunId, + sourceTurnId: source.sourceTurnId, + sourceRuntimeEventHighWater: source.sourceRuntimeEventHighWater, + }, + ); } finally { await client.close(); await fixture.stopHost(host); diff --git a/packages/runtime-host/src/__tests__/execution-host-message.test.ts b/packages/runtime-host/src/__tests__/execution-host-message.test.ts index 8c75c21e5a..fb7932c05b 100644 --- a/packages/runtime-host/src/__tests__/execution-host-message.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-message.test.ts @@ -120,7 +120,7 @@ test('steering becomes durable and ordered followups automatically start the nex const first = await connectClient(fixture.root); const second = await connectClient(fixture.root); const firstTurnId = randomUUID(); - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId: firstTurnId, content: { text: FAKE_WAIT_FOR_STEERING_PROMPT }, @@ -249,7 +249,7 @@ test('explicit retract is durable across connections and prevents successor admi const second = await connectClient(fixture.root); const turnId = randomUUID(); const started = requireStartedTurn( - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -280,7 +280,7 @@ test('explicit retract is durable across connections and prevents successor admi const retrying = await connectClient(fixture.root); assert.deepEqual(await retrying.request('queue.retract', retractInput), retracted); - const terminal = await first.stopTurn({ + const terminal = await first.request('turn.stop', { sessionId: fixture.sessionId, turnId, runId: started.runId, @@ -305,7 +305,7 @@ test('interrupt atomically retracts queued followup, stops the exact run, and is const second = await connectClient(fixture.root); const turnId = randomUUID(); const started = requireStartedTurn( - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, diff --git a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts index 6cc97ee966..5e7405f413 100644 --- a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts @@ -138,7 +138,7 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as const firstTurnId = randomUUID(); const started = requireStartedTurn( - await desktop.startTurn({ + await desktop.request('turn.start', { sessionId: fixture.sessionId, turnId: firstTurnId, content: { text: `continuity root ${'x'.repeat(540)}` }, @@ -223,12 +223,12 @@ test('concurrent root admission for one Session has a single winner', async () = const turnIds = [randomUUID(), randomUUID()] as const; const outcomes = await Promise.allSettled([ - first.startTurn({ + first.request('turn.start', { sessionId: fixture.sessionId, turnId: turnIds[0], content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, }), - second.startTurn({ + second.request('turn.start', { sessionId: fixture.sessionId, turnId: turnIds[1], content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -249,7 +249,7 @@ test('concurrent root admission for one Session has a single winner', async () = const winnerResult = winners[0]?.value; assert.ok(winnerResult); const winner = requireStartedTurn(winnerResult); - await first.stopTurn({ + await first.request('turn.stop', { sessionId: fixture.sessionId, turnId: winner.turnId, runId: winner.runId, @@ -274,7 +274,7 @@ test('an archived Session rejects a new Turn before durable admission', async () await assert.rejects( () => - client.startTurn({ + client.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: 'must not execute' }, @@ -304,7 +304,7 @@ test('a killed Host is recovered exactly once before its successor becomes ready const firstProbe = new SubscriptionProbe(firstSubscription); const turnId = randomUUID(); const started = requireStartedTurn( - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -333,7 +333,7 @@ test('a killed Host is recovered exactly once before its successor becomes ready sessionId: fixture.sessionId, transcript: { kind: 'none' }, }); - const recovered = await second.queryTurn({ + const recovered = await second.request('turn.query', { sessionId: fixture.sessionId, turnId, }); @@ -373,7 +373,7 @@ test('a killed Host is recovered exactly once before its successor becomes ready const thirdHost = await fixture.startHost(); const third = await connectClient(fixture.root); - const stable = await third.queryTurn({ + const stable = await third.request('turn.query', { sessionId: fixture.sessionId, turnId, }); @@ -404,7 +404,7 @@ test('graceful Host shutdown stops and drains an active Turn before releasing ow const client = await connectClient(fixture.root); const turnId = randomUUID(); const started = requireStartedTurn( - await client.startTurn({ + await client.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -417,7 +417,7 @@ test('graceful Host shutdown stops and drains an active Turn before releasing ow const successor = await fixture.startHost(); const observer = await connectClient(fixture.root); - const stable = await observer.queryTurn({ + const stable = await observer.request('turn.query', { sessionId: fixture.sessionId, turnId, }); @@ -444,7 +444,7 @@ test('a durable admission without a Run resumes before the Host becomes ready', const host = await fixture.startHost(); const client = await connectClient(fixture.root); - const recovered = await client.queryTurn({ + const recovered = await client.request('turn.query', { sessionId: fixture.sessionId, turnId, }); @@ -452,14 +452,15 @@ test('a durable admission without a Run resumes before the Host becomes ready', assert.ok(recovered.status === 'running' || recovered.status === 'waiting_for_user'); await assert.rejects( () => - client.startTurn({ + client.request('turn.start', { sessionId: fixture.sessionId, turnId: randomUUID(), content: { text: 'must remain behind the recovered admission' }, }), operationError('session_busy'), ); - const stopped = await client.stopTurn( + const stopped = await client.request( + 'turn.stop', { sessionId: fixture.sessionId, turnId, @@ -492,7 +493,7 @@ test('startup recovery compares an existing quoted UserMessage canonically', asy const host = await fixture.startHost(); const client = await connectClient(fixture.root); - const recovered = await client.queryTurn({ + const recovered = await client.request('turn.query', { sessionId: fixture.sessionId, turnId, }); @@ -519,7 +520,7 @@ test('startup recovery restores the admitted UserMessage before terminalizing it const host = await fixture.startHost(); const client = await connectClient(fixture.root); - const recovered = await client.queryTurn({ + const recovered = await client.request('turn.query', { sessionId: fixture.sessionId, turnId, }); diff --git a/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts b/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts index 86e95b4cf0..2e7e5207b1 100644 --- a/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts @@ -127,7 +127,7 @@ test('retry after a discarded turn.start response reuses the durable semantic ad dropped.abort(); const retried = requireStartedTurn( - await observer.startTurn({ + await observer.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text }, @@ -136,7 +136,7 @@ test('retry after a discarded turn.start response reuses the durable semantic ad assert.equal(retried.runId, committed.runId); await assert.rejects( () => - observer.startTurn({ + observer.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: `${text} changed` }, @@ -152,7 +152,7 @@ test('retry after a discarded turn.start response reuses the durable semantic ad const successorClient = await connectClient(fixture.root); assert.deepEqual( requireStartedTurn( - await successorClient.startTurn({ + await successorClient.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text }, @@ -161,7 +161,7 @@ test('retry after a discarded turn.start response reuses the durable semantic ad terminal, ); const successorTurnId = randomUUID(); - await successorClient.startTurn({ + await successorClient.request('turn.start', { sessionId: fixture.sessionId, turnId: successorTurnId, content: { text: 'successor must extend the recovered durable tip' }, @@ -189,7 +189,7 @@ test('startup recovery replays an admitted regenerate with its source lineage', const first = await connectClient(fixture.root); const sourceTurnId = randomUUID(); const regeneratedTurnId = randomUUID(); - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId: sourceTurnId, content: quotedContent('recover this regeneration'), @@ -224,7 +224,7 @@ test('a fresh quoted Turn preserves durable and Runtime handoff content', async const turnId = randomUUID(); const content = quotedContent('fresh quoted turn'); - await client.startTurn({ sessionId: fixture.sessionId, turnId, content }); + await client.request('turn.start', { sessionId: fixture.sessionId, turnId, content }); await waitForTerminalTurn(client, fixture.sessionId, turnId); await client.close(); await fixture.stopHost(host); @@ -329,7 +329,7 @@ test('stale Session operations return not_found across the SQLite-backed UDS Hos ); await assert.rejects( () => - client.startTurn({ + client.request('turn.start', { sessionId: staleSessionId, turnId: randomUUID(), content: { text: 'stale start' }, diff --git a/packages/runtime-host/src/__tests__/execution-host.test.ts b/packages/runtime-host/src/__tests__/execution-host.test.ts index 874c02a86c..f4ac9d9394 100644 --- a/packages/runtime-host/src/__tests__/execution-host.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host.test.ts @@ -560,7 +560,8 @@ test('two Clients share one execution after the starting Client disconnects', as const turnId = randomUUID(); const started = requireStartedTurn( - await first.startTurn( + await first.request( + 'turn.start', { sessionId: fixture.sessionId, turnId, @@ -586,7 +587,8 @@ test('two Clients share one execution after the starting Client disconnects', as const secondProbe = new SubscriptionProbe(secondSubscription); await assert.rejects( () => - second.startTurn( + second.request( + 'turn.start', { sessionId: fixture.sessionId, turnId: randomUUID(), @@ -611,13 +613,14 @@ test('two Clients share one execution after the starting Client disconnects', as }), pending, ); - const observed = await second.queryTurn({ + const observed = await second.request('turn.query', { sessionId: fixture.sessionId, turnId, }); assert.equal(observed.runId, started.runId); assert.ok(observed.status === 'running' || observed.status === 'waiting_for_user'); - const stopped = await second.stopTurn( + const stopped = await second.request( + 'turn.stop', { sessionId: fixture.sessionId, turnId, @@ -651,7 +654,8 @@ test('two Clients share one execution after the starting Client disconnects', as const nextTurnId = randomUUID(); const next = requireStartedTurn( - await second.startTurn( + await second.request( + 'turn.start', { sessionId: fixture.sessionId, turnId: nextTurnId, @@ -662,7 +666,8 @@ test('two Clients share one execution after the starting Client disconnects', as ); assert.deepEqual( requireStartedTurn( - await second.startTurn( + await second.request( + 'turn.start', { sessionId: fixture.sessionId, turnId, @@ -674,20 +679,21 @@ test('two Clients share one execution after the starting Client disconnects', as stopped, ); assert.deepEqual( - await second.stopTurn({ + await second.request('turn.stop', { sessionId: fixture.sessionId, turnId, runId: started.runId, }), stopped, ); - const nextObserved = await second.queryTurn({ + const nextObserved = await second.request('turn.query', { sessionId: fixture.sessionId, turnId: nextTurnId, }); assert.equal(nextObserved.runId, next.runId); assert.ok(nextObserved.status === 'running' || nextObserved.status === 'waiting_for_user'); - await second.stopTurn( + await second.request( + 'turn.stop', { sessionId: fixture.sessionId, turnId: nextTurnId, @@ -719,7 +725,8 @@ test('regenerate replays the durable source content with one recoverable root id const sourceTurnId = randomUUID(); const regeneratedTurnId = randomUUID(); try { - await client.startTurn( + await client.request( + 'turn.start', { sessionId: fixture.sessionId, turnId: sourceTurnId, @@ -729,7 +736,8 @@ test('regenerate replays the durable source content with one recoverable root id ); await waitForTerminalTurn(client, fixture.sessionId, sourceTurnId); - const started = await client.regenerateTurn( + const started = await client.request( + 'turn.regenerate', { sessionId: fixture.sessionId, sourceTurnId, @@ -740,7 +748,7 @@ test('regenerate replays the durable source content with one recoverable root id const terminal = await waitForTerminalTurn(client, fixture.sessionId, regeneratedTurnId); assert.equal(terminal.runId, started.runId); assert.deepEqual( - await client.regenerateTurn({ + await client.request('turn.regenerate', { sessionId: fixture.sessionId, sourceTurnId, turnId: regeneratedTurnId, @@ -772,14 +780,14 @@ test('regenerate rejects self-source and legacy target collisions without draini const firstHost = await fixture.startHost(); const first = await connectClient(fixture.root); const sourceTurnId = randomUUID(); - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId: sourceTurnId, content: { text: 'source request' }, }); await waitForTerminalTurn(first, fixture.sessionId, sourceTurnId); await assert.rejects( - first.regenerateTurn({ + first.request('turn.regenerate', { sessionId: fixture.sessionId, sourceTurnId, turnId: sourceTurnId, @@ -794,7 +802,7 @@ test('regenerate rejects self-source and legacy target collisions without draini const second = await connectClient(fixture.root); try { await assert.rejects( - second.regenerateTurn({ + second.request('turn.regenerate', { sessionId: fixture.sessionId, sourceTurnId, turnId: legacy.sourceTurnId, @@ -802,7 +810,7 @@ test('regenerate rejects self-source and legacy target collisions without draini operationError('operation_conflict'), ); const followingTurnId = randomUUID(); - await second.startTurn({ + await second.request('turn.start', { sessionId: fixture.sessionId, turnId: followingTurnId, content: { text: 'Host remains available' }, @@ -831,12 +839,17 @@ test('context actions share root admission and expose backend capability honestl const turnId = randomUUID(); const unavailableTurnId = randomUUID(); try { - assert.deepEqual(await first.queryContextDiagnostics({ sessionId: fixture.sessionId }), { - status: 'unavailable', - reason: 'no_completed_request', - }); + assert.deepEqual( + await first.request('context.diagnostics.query', { + sessionId: fixture.sessionId, + }), + { + status: 'unavailable', + reason: 'no_completed_request', + }, + ); const started = requireStartedTurn( - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -844,26 +857,26 @@ test('context actions share root admission and expose backend capability honestl ); await waitForRunningTurn(second, fixture.sessionId, turnId); await assert.rejects( - second.compactContext({ + second.request('context.compact', { sessionId: fixture.sessionId, turnId: randomUUID(), }), operationError('session_busy'), ); - await second.stopTurn({ + await second.request('turn.stop', { sessionId: fixture.sessionId, turnId, runId: started.runId, }); await assert.rejects( - second.compactContext({ + second.request('context.compact', { sessionId: fixture.sessionId, turnId: unavailableTurnId, }), operationError('operation_unavailable'), ); await assert.rejects( - second.queryContextDiagnostics({ sessionId: 'missing-session' }), + second.request('context.diagnostics.query', { sessionId: 'missing-session' }), operationError('not_found'), ); } finally { @@ -884,7 +897,7 @@ test('a disconnected Client leaves a durable Interaction that another Client can const first = await connectClient(fixture.root); const turnId = randomUUID(); const started = requireStartedTurn( - await first.startTurn({ + await first.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -978,7 +991,10 @@ test('a disconnected Client leaves a durable Interaction that another Client can }), winner, ); - assert.deepEqual(await observer.queryTurn({ sessionId: fixture.sessionId, turnId }), completed); + assert.deepEqual( + await observer.request('turn.query', { sessionId: fixture.sessionId, turnId }), + completed, + ); await observer.close(); await fixture.stopHost(secondHost); }); @@ -997,7 +1013,7 @@ test('two UDS Clients settle one hosted sandbox boundary and resume its exact Ru const probe = new SubscriptionProbe(subscription); const turnId = randomUUID(); const started = requireStartedTurn( - await starter.startTurn({ + await starter.request('turn.start', { sessionId: fixture.sessionId, turnId, content: { text: FAKE_ASK_SANDBOX_BOUNDARY_PROMPT }, @@ -1062,7 +1078,10 @@ test('two UDS Clients settle one hosted sandbox boundary and resume its exact Ru }), firstWinner, ); - assert.deepEqual(await observer.queryTurn({ sessionId: fixture.sessionId, turnId }), completed); + assert.deepEqual( + await observer.request('turn.query', { sessionId: fixture.sessionId, turnId }), + completed, + ); await observer.close(); await fixture.stopHost(secondHost); }); diff --git a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts index f44e88b8fa..34d4cb284a 100644 --- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts +++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts @@ -1184,7 +1184,7 @@ export async function waitForTurn( const deadline = Date.now() + PROCESS_TIMEOUT_MS; while (true) { try { - return await connection.queryTurn({ sessionId, turnId }); + return await connection.request('turn.query', { sessionId, turnId }); } catch (error) { if (!(error instanceof RuntimeHostOperationError) || error.code !== 'not_found') throw error; if (Date.now() >= deadline) throw new Error('Turn admission was not observed'); @@ -1275,7 +1275,7 @@ export async function waitForTerminalTurn( try { return await withTimeout( (async () => { - const current = await connection.queryTurn({ sessionId, turnId }); + const current = await connection.request('turn.query', { sessionId, turnId }); if (isTerminalTurnSnapshot(current)) return current; for await (const frame of subscription) { if (frame.kind !== 'subscription.session_projection') continue; @@ -1307,7 +1307,7 @@ export async function waitForRunningTurn( ): Promise { const deadline = Date.now() + PROCESS_TIMEOUT_MS; while (true) { - const snapshot = await connection.queryTurn({ sessionId, turnId }); + const snapshot = await connection.request('turn.query', { sessionId, turnId }); if (snapshot.status === 'running' || snapshot.status === 'waiting_for_user') return snapshot; if (Date.now() >= deadline) throw new Error('Turn did not become active'); await sleep(20); diff --git a/packages/runtime-host/src/__tests__/host-kernel.test.ts b/packages/runtime-host/src/__tests__/host-kernel.test.ts index d9f31bda78..4a2919b386 100644 --- a/packages/runtime-host/src/__tests__/host-kernel.test.ts +++ b/packages/runtime-host/src/__tests__/host-kernel.test.ts @@ -801,7 +801,7 @@ describe('non-serving Runtime Host kernel', () => { assert.equal(status.state, 'ready'); assert.equal(status.connections, 1); } - const diagnostics = await connected.connection.queryHostDiagnostics(); + const diagnostics = await connected.connection.request('host.diagnostics.query', {}); assert.equal(diagnostics.hostEpoch, winner.host.hostEpoch); assert.equal(diagnostics.state, 'ready'); assert.equal(diagnostics.pid, process.pid); diff --git a/packages/runtime-host/src/__tests__/owned-candidate.test.ts b/packages/runtime-host/src/__tests__/owned-candidate.test.ts index 118ace3309..87c8200ce3 100644 --- a/packages/runtime-host/src/__tests__/owned-candidate.test.ts +++ b/packages/runtime-host/src/__tests__/owned-candidate.test.ts @@ -220,7 +220,7 @@ test('a late owned candidate stays alive after another client adopts it', { assert.equal(released, 1); assert.equal(settled, 0); - const diagnostics = await adopted.connection.queryHostDiagnostics(); + const diagnostics = await adopted.connection.request('host.diagnostics.query', {}); assert.equal(diagnostics.pid, host.pid); assert.doesNotThrow(() => process.kill(host.pid, 0)); } finally { diff --git a/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts index 536972ef46..daf3c28b93 100644 --- a/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts @@ -31,7 +31,11 @@ import { type RuntimeHostConnection, type RuntimeHostSessionSubscription, } from '../client/index.js'; -import { RUNTIME_HOST_PROTOCOL_VERSION, type SubscriptionFrame } from '../protocol/index.js'; +import { + RUNTIME_HOST_PROTOCOL_VERSION, + type OperationOutput, + type SubscriptionFrame, +} from '../protocol/index.js'; import { FakeBackend } from '@maka/runtime/test-only/fake-backend'; import { createExecutionRuntimeHostComposition } from '../server/execution-composition.js'; import { RuntimeHostKernel, type RuntimeHostCompositionFactory } from '../server/host-kernel.js'; @@ -90,7 +94,10 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth transcript: { kind: 'none' }, }); - const first = await desktop.queryPlan({ kind: 'list_start', sessionId: session.id }); + const first = await desktop.request('plan.query', { + kind: 'list_start', + sessionId: session.id, + }); assert.equal(first.kind, 'page'); if (first.kind !== 'page') return; const approval = { @@ -101,7 +108,7 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth expectedStoreVersion: first.storeVersion, turnId: 'approve-turn', }; - const started = await tui.startPlanTurn(approval); + const started = await tui.request('plan.turn.start', approval); const approved = started.plan; assert.equal(approved.eventType, 'plan_approved'); assert.ok(approved.executionId); @@ -114,7 +121,7 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth assert.equal(changed.sessionId, session.id); assert.equal(changed.domain, 'plan'); - const shared = await tui.queryPlan({ kind: 'list_start', sessionId: session.id }); + const shared = await tui.request('plan.query', { kind: 'list_start', sessionId: session.id }); assert.equal(shared.kind, 'page'); if (shared.kind === 'page') { assert.equal(shared.activeExecutionId, approved.executionId); @@ -139,11 +146,14 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth owner = undefined; tui = await connect(root); - const replayed = await tui.startPlanTurn(approval); + const replayed = await tui.request('plan.turn.start', approval); assert.equal(replayed.plan.executionId, approved.executionId); assert.equal(replayed.plan.storeVersion, approved.storeVersion); assert.equal(replayed.turn.turnId, approval.turnId); - const recovered = await tui.queryPlan({ kind: 'list_start', sessionId: session.id }); + const recovered = await tui.request('plan.query', { + kind: 'list_start', + sessionId: session.id, + }); assert.equal(recovered.kind, 'page'); if (recovered.kind !== 'page') return; assert.equal(recovered.activeExecutionId, null); @@ -153,7 +163,7 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth assert.equal(execution.execution.status, 'interrupted'); await assert.rejects( - tui.startPlanTurn({ + tui.request('plan.turn.start', { kind: 'resume_execution', sessionId: session.id, executionId: execution.execution.executionId, @@ -162,7 +172,10 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth (error: unknown) => error instanceof Error && 'code' in error && error.code === 'operation_conflict', ); - const unchanged = await tui.queryPlan({ kind: 'list_start', sessionId: session.id }); + const unchanged = await tui.request('plan.query', { + kind: 'list_start', + sessionId: session.id, + }); assert.equal(unchanged.kind, 'page'); assert.equal( unchanged.kind === 'page' @@ -171,7 +184,7 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth 'interrupted', ); - const resumed = await tui.startPlanTurn({ + const resumed = await tui.request('plan.turn.start', { kind: 'resume_execution', sessionId: session.id, executionId: execution.execution.executionId, @@ -181,7 +194,10 @@ test('two Clients and a restarted production Host share one retry-safe Plan auth assert.equal(resumed.plan.executionId, execution.execution.executionId); assert.equal(resumed.turn.turnId, 'resume-turn'); await waitForTerminal(tui, resumed.turn); - const afterResume = await tui.queryPlan({ kind: 'list_start', sessionId: session.id }); + const afterResume = await tui.request('plan.query', { + kind: 'list_start', + sessionId: session.id, + }); assert.equal(afterResume.kind, 'page'); assert.equal( afterResume.kind === 'page' ? afterResume.activeExecutionId : undefined, @@ -204,7 +220,7 @@ async function connect(rootPath: string): Promise { async function waitForTerminal( connection: RuntimeHostConnection, - initial: Awaited>['turn'], + initial: OperationOutput<'plan.turn.start'>['turn'], ): Promise { let snapshot = initial; for (let attempt = 0; attempt < 100; attempt += 1) { @@ -216,7 +232,7 @@ async function waitForTerminal( return; } await new Promise((resolve) => setTimeout(resolve, 10)); - snapshot = await connection.queryTurn({ + snapshot = await connection.request('turn.query', { sessionId: snapshot.sessionId, turnId: snapshot.turnId, }); diff --git a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts index 68fc00d72c..77a250a90b 100644 --- a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts +++ b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts @@ -35,6 +35,7 @@ import { RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, type HostIncompatible, + type HostStatusResult, type OperationInput, type OperationKey, type OperationOutput, @@ -76,6 +77,47 @@ test('a reconnecting Client retries an interrupted query on the replacement conn await connection.close(); }); +test('a reconnecting Client retries status through the validated status surface', async () => { + let firstStatusCalls = 0; + let replacementStatusCalls = 0; + const first = connectionHarness( + 'first', + () => { + throw new Error('status must not use request()'); + }, + undefined, + undefined, + async () => { + firstStatusCalls += 1; + first.disconnect(); + throw interrupted('host.status', 'query', 'dispatched'); + }, + ); + const replacement = connectionHarness( + 'replacement', + () => { + throw new Error('status must not use request()'); + }, + undefined, + undefined, + async () => { + replacementStatusCalls += 1; + return hostStatus('replacement'); + }, + ); + const connection = await createRuntimeHostReconnectingConnection({ + initialConnection: first.connection, + connect: async () => replacement.connection, + }); + + assert.deepEqual(await connection.status(), hostStatus('replacement')); + assert.equal(firstStatusCalls, 1); + assert.equal(replacementStatusCalls, 1); + assert.deepEqual(first.operations, []); + assert.deepEqual(replacement.operations, []); + await connection.close(); +}); + test('a reconnecting Client never replays an admitted command with an unknown outcome', async () => { const first = connectionHarness('first', (operation) => { first.disconnect(); @@ -469,6 +511,7 @@ function connectionHarness( id: 'maka.interactive', revision: '1', }, + status: () => Promise = async () => hostStatus(id, composition), ) { let resolveClosed!: () => void; const closed = new Promise((resolve) => { @@ -484,6 +527,7 @@ function connectionHarness( compositionId: composition.id, compositionRevision: composition.revision, closed, + status, request: async (operation: DirectRequestOperationKey, input: unknown) => { operations.push(operation); return request(operation, input); @@ -508,6 +552,24 @@ function connectionHarness( }; } +function hostStatus( + id: string, + composition: { readonly id: string; readonly revision: string } = { + id: 'maka.interactive', + revision: '1', + }, +): HostStatusResult { + return { + hostEpoch: `host-${id}`, + compositionId: composition.id, + compositionRevision: composition.revision, + state: 'ready', + connections: 1, + activeOperations: 0, + activeResidencies: 0, + }; +} + function interrupted( operation: OperationKey, mode: 'query' | 'command' | 'control', diff --git a/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts index ce4bc3f478..665548b33d 100644 --- a/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/session-catalog-two-client-uds.test.ts @@ -351,12 +351,12 @@ test('two Clients share stable Session creation, CAS configuration, and catalog const secondCwd = join(base, 'workspace-second'); await Promise.all([mkdir(firstCwd), mkdir(secondCwd)]); const relocationOutcomes = await Promise.all([ - desktop.relocateSessionWorkspace({ + desktop.request('session.workspace.relocate', { sessionId: narrowedSession.id, expectedRevision: narrowedSession.revision, workspace: { kind: 'host_path', path: firstCwd }, }), - tui.relocateSessionWorkspace({ + tui.request('session.workspace.relocate', { sessionId: narrowedSession.id, expectedRevision: narrowedSession.revision, workspace: { kind: 'host_path', path: secondCwd }, diff --git a/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts index 21edde3183..1c2e6c24e3 100644 --- a/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts @@ -100,9 +100,9 @@ test('two Clients share one durable Session recap effect', async () => { desktop = await connect(root); tui = await connect(root); const input = { sessionId: 'session-1', effectId: 'effect-1', reason: 'manual' as const }; - const desktopResult = desktop.generateSessionRecap(input); + const desktopResult = desktop.request('session.recap.generate', input); await modelStarted.promise; - const tuiResult = tui.generateSessionRecap(input); + const tuiResult = tui.request('session.recap.generate', input); modelRelease.release(); const [first, second] = await Promise.all([desktopResult, tuiResult]); assert.deepEqual(first, { diff --git a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts index 5ccf51c26a..f5b678f143 100644 --- a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts @@ -247,7 +247,7 @@ async function verifyConcurrentRevisionAuthority( }); assert.equal(sourceGraph.status, 'completed'); assert.equal(sourceGraph.operators[0]?.childSessionId, graphChildSessionId); - await desktop.startTurn({ + await desktop.request('turn.start', { sessionId: GRAPH_REVISION_TARGET_ID, turnId: 'graph-revision-new-turn', content: { text: 'continue independently' }, @@ -422,7 +422,7 @@ async function verifyConcurrentRevisionAuthority( ); const busyTurn = requireStartedTurn( - await desktop.startTurn({ + await desktop.request('turn.start', { sessionId: busySessionId, turnId: 'busy-turn', content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -449,7 +449,8 @@ async function verifyConcurrentRevisionAuthority( assertionError = error; } try { - const stopped = await desktop.stopTurn( + const stopped = await desktop.request( + 'turn.stop', { sessionId: busySessionId, turnId: 'busy-turn', @@ -470,7 +471,7 @@ async function verifyConcurrentRevisionAuthority( if (assertionError !== undefined) throw assertionError; const activeSourceTurn = requireStartedTurn( - await desktop.startTurn({ + await desktop.request('turn.start', { sessionId: sourceSessionId, turnId: 'active-source-turn', content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, @@ -509,7 +510,8 @@ async function verifyConcurrentRevisionAuthority( assertionError = error; } try { - const stopped = await desktop.stopTurn( + const stopped = await desktop.request( + 'turn.stop', { sessionId: sourceSessionId, turnId: 'active-source-turn', @@ -566,7 +568,7 @@ async function verifyRestartRecoveryAndAdmission( assert.equal(admitted.kind, 'committed'); if (admitted.kind !== 'committed') assert.fail('Admitted revision must commit'); assert.equal(requireSessionProjection(admitted.session).revisionIndex, 3); - await restarted.startTurn({ + await restarted.request('turn.start', { sessionId: ADMITTED_REVISION_TARGET_ID, turnId: 'turn-3', content: { text: 'commit this revision' }, diff --git a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts index dfe5d1bcab..1a777646ef 100644 --- a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts +++ b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts @@ -181,11 +181,42 @@ test('isolates a sequence gap and continues requests on the same connection', as () => subscription[Symbol.asyncIterator]().next(), hasSubscriptionReason('sequence_gap'), ); + await assert.rejects( + // @ts-expect-error host.status must use the validated status() API. + () => connection.request('host.status', {}), + /status requires the validated status\(\) API/, + ); assert.equal((await connection.status()).hostEpoch, connection.hostEpoch); }, ); }); +test('fails the connection when status reports a different Host identity', async () => { + await withProtocolPeer( + async (transport, hostEpoch, rootId) => { + const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId); + const opened = openResult(hostEpoch, 'subscription-status-identity'); + await writeProtocolFrame(transport, { + requestId: request.requestId, + operation: 'subscription.open', + ok: true, + result: opened, + }); + await answerClose(transport, opened.subscriptionId); + await answerStatus(transport, 'different-host-epoch'); + }, + async (connection) => { + const subscription = await connection.openSessionSubscription({ + sessionId: 'session-1', + transcript: { kind: 'none' }, + }); + await subscription.close(); + await assert.rejects(() => connection.status(), /status for a different Host identity/); + await connection.closed; + }, + ); +}); + test('rejects epoch and Session correlation changes per subscription', async () => { for (const changed of ['epoch', 'session', 'graph'] as const) { await withProtocolPeer( @@ -671,7 +702,7 @@ test('fails the connection when overlay release is not confirmed', async () => { () => subscription.loadTranscript(decodeStoredMessage), hasSubscriptionReason('transcript_release_failed'), ); - await assert.rejects(() => connection.request('host.status', {})); + await assert.rejects(() => connection.status()); }, ); }); @@ -753,7 +784,7 @@ test('keeps the connection usable when close wins the overlay release race', asy text: 'overlay', }, ]); - assert.equal((await connection.request('host.status', {})).hostEpoch, connection.hostEpoch); + assert.equal((await connection.status()).hostEpoch, connection.hostEpoch); }, ); }); diff --git a/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts b/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts index a1fcc42901..1ad5bb3ac9 100644 --- a/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts +++ b/packages/runtime-host/src/__tests__/usage-pricing-client-correlation.test.ts @@ -187,7 +187,7 @@ describe('Usage/Pricing client response correlation', () => { const request = requestUnchecked(connection, mismatch.operation, mismatch.input); await assert.rejects(request, isInvalidFrame); await connection.closed; - await assert.rejects(requestUnchecked(connection, 'host.status', {}), isInvalidFrame); + await assert.rejects(connection.status(REQUEST_TIMEOUT_MS), isInvalidFrame); }, ); }); diff --git a/packages/runtime-host/src/client/connect-or-spawn.ts b/packages/runtime-host/src/client/connect-or-spawn.ts index 9b7f3b63c2..941620395a 100644 --- a/packages/runtime-host/src/client/connect-or-spawn.ts +++ b/packages/runtime-host/src/client/connect-or-spawn.ts @@ -245,7 +245,10 @@ export async function connectOwnedRuntimeHostWithDependencies( } const ownedConnection = result.connection; connection = ownedConnection; - const diagnostics = await abortable(() => ownedConnection.queryHostDiagnostics(), input.signal); + const diagnostics = await abortable( + () => ownedConnection.request('host.diagnostics.query', {}), + input.signal, + ); if (diagnostics.pid !== host.pid) { await connection.close(); connection = undefined; diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index 1daf690a69..1b0ce98a81 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -40,17 +40,6 @@ import { type ClientCapabilityUnregisterResult, type ClientHello, type ConfigurationChangedFrame, - type ContextCompactInput, - type ContextCompactResult, - type ContextDiagnosticsQueryInput, - type ContextDiagnosticsResult, - type DeepResearchQueryInput, - type DeepResearchQueryResult, - type DailyReviewMutateInput, - type DailyReviewMutateResult, - type DailyReviewQueryInput, - type DailyReviewQueryResult, - type HostDiagnosticsResult, type HostOperationErrorCode, type HostIncompatible, type HostRegistration, @@ -63,12 +52,6 @@ import { type OperationInput, type OperationKey, type OperationOutput, - type PlanControlInput, - type PlanControlResult, - type PlanQueryInput, - type PlanQueryResult, - type PlanTurnStartInput, - type PlanTurnStartResult, type ProjectCatalogChangedFrame, type ProtocolRange, type RequestFrame, @@ -77,20 +60,6 @@ import { type ScheduledTaskChangedFrame, type SubscriptionFrame, type SubscriptionOpenInput, - type SessionWorkspaceRelocateInput, - type SessionRecapGenerateInput, - type SessionRecapGenerateResult, - type SessionUpdateResult, - type TurnQueryInput, - type TurnRegenerateInput, - type TurnResumePlan, - type TurnResumeQueryInput, - type TurnResumeStartInput, - type TurnResumeStartResult, - type TurnSnapshot, - type TurnStartInput, - type TurnStartResult, - type TurnStopInput, requireClientInstanceId, requireHostCompositionId, requireHostGeneration, @@ -256,41 +225,6 @@ export interface RuntimeHostConnection { timeoutMs?: number, ): Promise>; status(timeoutMs?: number): Promise; - queryHostDiagnostics(timeoutMs?: number): Promise; - startTurn(input: TurnStartInput, timeoutMs?: number): Promise; - queryTurn(input: TurnQueryInput, timeoutMs?: number): Promise; - stopTurn(input: TurnStopInput, timeoutMs?: number): Promise; - regenerateTurn(input: TurnRegenerateInput, timeoutMs?: number): Promise; - queryContextDiagnostics( - input: ContextDiagnosticsQueryInput, - timeoutMs?: number, - ): Promise; - compactContext(input: ContextCompactInput, timeoutMs?: number): Promise; - relocateSessionWorkspace( - input: SessionWorkspaceRelocateInput, - timeoutMs?: number, - ): Promise; - generateSessionRecap( - input: SessionRecapGenerateInput, - timeoutMs?: number, - ): Promise; - queryPlan(input: PlanQueryInput, timeoutMs?: number): Promise; - controlPlan(input: PlanControlInput, timeoutMs?: number): Promise; - startPlanTurn(input: PlanTurnStartInput, timeoutMs?: number): Promise; - queryDeepResearch( - input: DeepResearchQueryInput, - timeoutMs?: number, - ): Promise; - queryDailyReview( - input: DailyReviewQueryInput, - timeoutMs?: number, - ): Promise; - mutateDailyReview( - input: DailyReviewMutateInput, - timeoutMs?: number, - ): Promise; - queryTurnResume(input: TurnResumeQueryInput, timeoutMs?: number): Promise; - startTurnResume(input: TurnResumeStartInput, timeoutMs?: number): Promise; openSessionSubscription( input: SubscriptionOpenInput, timeoutMs?: number, @@ -309,6 +243,7 @@ export interface RuntimeHostConnection { export type DirectRequestOperationKey = Exclude< OperationKey, + | 'host.status' | 'subscription.open' | 'subscription.close' | 'client.capability.replace' @@ -471,18 +406,15 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { input: OperationInput, timeoutMs?: number, ): Promise> { + if (isHostStatusOperation(operation)) { + return Promise.reject(new Error('Runtime Host status requires the validated status() API')); + } if (isClientCapabilityMutation(operation)) { return Promise.reject( new Error('Client Capability mutations require the dedicated capability channel'), ); } - return this.#requestOperation( - operation, - input, - timeoutMs ?? (operation === 'host.status' ? DEFAULT_LIVENESS_TIMEOUT_MS : undefined), - (result) => result, - operation === 'host.status' ? 'connection' : 'request', - ); + return this.#requestOperation(operation, input, timeoutMs, (result) => result, 'request'); } #requestOperation( @@ -576,7 +508,16 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { } async status(timeoutMs?: number): Promise { - const status = await this.request('host.status', {}, timeoutMs); + return this.#requestOperation( + 'host.status', + {}, + timeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS, + (status) => this.#validateHostStatusIdentity(status), + 'connection', + ); + } + + #validateHostStatusIdentity(status: HostStatusResult): HostStatusResult { if ( status.hostEpoch !== this.hostEpoch || status.compositionId !== this.compositionId || @@ -589,92 +530,6 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { return status; } - queryHostDiagnostics(timeoutMs?: number): Promise { - return this.request('host.diagnostics.query', {}, timeoutMs); - } - - startTurn(input: TurnStartInput, timeoutMs?: number): Promise { - return this.request('turn.start', input, timeoutMs); - } - - queryTurn(input: TurnQueryInput, timeoutMs?: number): Promise { - return this.request('turn.query', input, timeoutMs); - } - - stopTurn(input: TurnStopInput, timeoutMs?: number): Promise { - return this.request('turn.stop', input, timeoutMs); - } - - regenerateTurn(input: TurnRegenerateInput, timeoutMs?: number): Promise { - return this.request('turn.regenerate', input, timeoutMs); - } - - queryContextDiagnostics( - input: ContextDiagnosticsQueryInput, - timeoutMs?: number, - ): Promise { - return this.request('context.diagnostics.query', input, timeoutMs); - } - - compactContext(input: ContextCompactInput, timeoutMs?: number): Promise { - return this.request('context.compact', input, timeoutMs); - } - - relocateSessionWorkspace( - input: SessionWorkspaceRelocateInput, - timeoutMs?: number, - ): Promise { - return this.request('session.workspace.relocate', input, timeoutMs); - } - - generateSessionRecap( - input: SessionRecapGenerateInput, - timeoutMs?: number, - ): Promise { - return this.request('session.recap.generate', input, timeoutMs); - } - - queryPlan(input: PlanQueryInput, timeoutMs?: number): Promise { - return this.request('plan.query', input, timeoutMs); - } - - controlPlan(input: PlanControlInput, timeoutMs?: number): Promise { - return this.request('plan.control', input, timeoutMs); - } - - startPlanTurn(input: PlanTurnStartInput, timeoutMs?: number): Promise { - return this.request('plan.turn.start', input, timeoutMs); - } - - queryDeepResearch( - input: DeepResearchQueryInput, - timeoutMs?: number, - ): Promise { - return this.request('deep-research.query', input, timeoutMs); - } - - queryDailyReview( - input: DailyReviewQueryInput, - timeoutMs?: number, - ): Promise { - return this.request('daily-review.query', input, timeoutMs); - } - - mutateDailyReview( - input: DailyReviewMutateInput, - timeoutMs?: number, - ): Promise { - return this.request('daily-review.mutate', input, timeoutMs); - } - - queryTurnResume(input: TurnResumeQueryInput, timeoutMs?: number): Promise { - return this.request('turn.resume.query', input, timeoutMs); - } - - startTurnResume(input: TurnResumeStartInput, timeoutMs?: number): Promise { - return this.request('turn.resume.start', input, timeoutMs); - } - openSessionSubscription( input: SubscriptionOpenInput, timeoutMs?: number, @@ -961,9 +816,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { {}, DEFAULT_LIVENESS_TIMEOUT_MS, (status) => { - if (status.hostEpoch !== this.hostEpoch) { - throw new Error('Runtime Host returned status for a different Host Epoch'); - } + this.#validateHostStatusIdentity(status); try { this.#onLivenessProbe?.(); } catch { @@ -1082,6 +935,10 @@ function isClientCapabilityMutation(operation: unknown): boolean { return operation === 'client.capability.replace' || operation === 'client.capability.unregister'; } +function isHostStatusOperation(operation: unknown): boolean { + return operation === 'host.status'; +} + export async function connectRuntimeHost( input: ConnectRuntimeHostInput, ): Promise { diff --git a/packages/runtime-host/src/client/reconnecting-connection.ts b/packages/runtime-host/src/client/reconnecting-connection.ts index 84fc4e2ac1..07fb2d1ad9 100644 --- a/packages/runtime-host/src/client/reconnecting-connection.ts +++ b/packages/runtime-host/src/client/reconnecting-connection.ts @@ -21,44 +21,13 @@ import { HOST_OPERATION_SPECS, type ClientCapabilityReplaceResult, type ClientCapabilityUnregisterResult, - type ContextCompactInput, - type ContextCompactResult, - type ContextDiagnosticsQueryInput, - type ContextDiagnosticsResult, - type DeepResearchQueryInput, - type DeepResearchQueryResult, - type DailyReviewMutateInput, - type DailyReviewMutateResult, - type DailyReviewQueryInput, - type DailyReviewQueryResult, - type HostDiagnosticsResult, type HostStatusResult, type OperationInput, type OperationKey, type OperationOutput, - type PlanControlInput, - type PlanControlResult, - type PlanQueryInput, - type PlanQueryResult, - type PlanTurnStartInput, - type PlanTurnStartResult, type SessionCatalogChangedFrame, type ScheduledTaskChangedFrame, - type SessionWorkspaceRelocateInput, - type SessionRecapGenerateInput, - type SessionRecapGenerateResult, - type SessionUpdateResult, type SubscriptionOpenInput, - type TurnQueryInput, - type TurnRegenerateInput, - type TurnResumePlan, - type TurnResumeQueryInput, - type TurnResumeStartInput, - type TurnResumeStartResult, - type TurnSnapshot, - type TurnStartInput, - type TurnStartResult, - type TurnStopInput, } from '../protocol/index.js'; import type { ClientCapabilityProvider } from './client-capability.js'; import { @@ -189,94 +158,19 @@ class RuntimeHostReconnectingConnectionImpl implements RuntimeHostReconnectingCo return this.#request(operation, input, timeoutMs); } - status(timeoutMs?: number): Promise { - return this.#request('host.status', {}, timeoutMs); - } - - queryHostDiagnostics(timeoutMs?: number): Promise { - return this.#request('host.diagnostics.query', {}, timeoutMs); - } - - startTurn(input: TurnStartInput, timeoutMs?: number): Promise { - return this.#request('turn.start', input, timeoutMs); - } - - queryTurn(input: TurnQueryInput, timeoutMs?: number): Promise { - return this.#request('turn.query', input, timeoutMs); - } - - stopTurn(input: TurnStopInput, timeoutMs?: number): Promise { - return this.#request('turn.stop', input, timeoutMs); - } - - regenerateTurn(input: TurnRegenerateInput, timeoutMs?: number): Promise { - return this.#request('turn.regenerate', input, timeoutMs); - } - - queryContextDiagnostics( - input: ContextDiagnosticsQueryInput, - timeoutMs?: number, - ): Promise { - return this.#request('context.diagnostics.query', input, timeoutMs); - } - - compactContext(input: ContextCompactInput, timeoutMs?: number): Promise { - return this.#request('context.compact', input, timeoutMs); - } - - relocateSessionWorkspace( - input: SessionWorkspaceRelocateInput, - timeoutMs?: number, - ): Promise { - return this.#request('session.workspace.relocate', input, timeoutMs); - } - - generateSessionRecap( - input: SessionRecapGenerateInput, - timeoutMs?: number, - ): Promise { - return this.#request('session.recap.generate', input, timeoutMs); - } - - queryPlan(input: PlanQueryInput, timeoutMs?: number): Promise { - return this.#request('plan.query', input, timeoutMs); - } - - controlPlan(input: PlanControlInput, timeoutMs?: number): Promise { - return this.#request('plan.control', input, timeoutMs); - } - - startPlanTurn(input: PlanTurnStartInput, timeoutMs?: number): Promise { - return this.#request('plan.turn.start', input, timeoutMs); - } - - queryDeepResearch( - input: DeepResearchQueryInput, - timeoutMs?: number, - ): Promise { - return this.#request('deep-research.query', input, timeoutMs); - } - - queryDailyReview( - input: DailyReviewQueryInput, - timeoutMs?: number, - ): Promise { - return this.#request('daily-review.query', input, timeoutMs); - } - - mutateDailyReview( - input: DailyReviewMutateInput, - timeoutMs?: number, - ): Promise { - return this.#request('daily-review.mutate', input, timeoutMs); - } - - queryTurnResume(input: TurnResumeQueryInput, timeoutMs?: number): Promise { - return this.#request('turn.resume.query', input, timeoutMs); - } - - startTurnResume(input: TurnResumeStartInput, timeoutMs?: number): Promise { - return this.#request('turn.resume.start', input, timeoutMs); + async status(timeoutMs?: number): Promise { + const deadline = requestDeadline(timeoutMs); + let previous: RuntimeHostConnection | undefined; + while (true) { + const connection = await this.#waitForConnection('host.status', previous, deadline); + const remaining = deadline === undefined ? undefined : Math.max(1, deadline - Date.now()); + try { + return await connection.status(remaining); + } catch (error) { + if (!isRetryableQueryInterruption(error)) throw error; + previous = connection; + } + } } async openSessionSubscription(