| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| import { withTimeout } from '@maka/core/test-only/async-primitives'; |
| import assert from 'node:assert/strict'; |
| import { fork, type ChildProcess } from 'node:child_process'; |
| import { randomUUID } from 'node:crypto'; |
| import { |
| appendFile, |
| chmod, |
| mkdir, |
| mkdtemp, |
| readFile, |
| readdir, |
| rm, |
| writeFile, |
| } from 'node:fs/promises'; |
| import { createServer, type Server } from 'node:http'; |
| import { connect, type Socket } from 'node:net'; |
| import { tmpdir } from 'node:os'; |
| import { dirname, join } from 'node:path'; |
| import { test } from 'node:test'; |
| import { |
| TOOL_BOUNDARY_PROTOCOL_V1, |
| runtimeEventHasModelVisibleContent, |
| } from '@maka/core/runtime-event'; |
| import { canonicalToolArgsHash } from '@maka/core/tool-args-identity'; |
| import { |
| runtimeInvocationOutcome, |
| type RuntimeInvocationRecord, |
| } from '@maka/core/runtime-invocation'; |
| import { runtimeInvocationFailureClass } from '@maka/runtime/runtime-event-read-model'; |
| import type { MessageContent, AttachmentRef } from '@maka/core/events'; |
| import type { ConnectionCatalogEntry } from '@maka/core/runtime-policy'; |
| import type { StoredMessage } from '@maka/core/session'; |
| import { isTerminalRuntimeEvent } from '@maka/core/runtime-event'; |
| import type { RuntimeEvent, RuntimeInvocationLineage } from '@maka/core/runtime-event'; |
| import { |
| buildRecoveredTerminalRuntimeEvent, |
| classifyTerminalRuntimeLedger, |
| commitTerminalRunWithRuntimeFact, |
| } from '@maka/runtime/terminal-run-commit'; |
| import { |
| FAKE_ASK_USER_QUESTION_DURING_DRAIN_PROMPT, |
| FAKE_ASK_USER_QUESTION_PROMPT, |
| FAKE_WAIT_FOR_STEERING_PROMPT, |
| } from '@maka/runtime/test-only/fake-backend'; |
| import { type MakaTool, type MakaToolContext } from '@maka/runtime/tool-runtime'; |
| import { openInteractiveArtifactStoreForWrite } from '@maka/storage/artifact-stores'; |
| import { |
| openInteractiveExecutionStoresForRead, |
| openInteractiveExecutionStoresForWrite, |
| } from '@maka/storage/execution-stores'; |
| import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores'; |
| import { |
| resolveRootControlNamespace, |
| resolveStorageRoot, |
| tryAcquireInteractiveRootOwner, |
| tryAcquireInteractiveRootReader, |
| type StorageRootCapability, |
| } from '@maka/storage/root-authority'; |
| import { |
| connectRuntimeHost, |
| RuntimeHostOperationError, |
| RuntimeHostSubscriptionError, |
| type RuntimeHostConnection, |
| type RuntimeHostSessionSubscription, |
| } from '../client/index.js'; |
| import { |
| decodeHostFrame, |
| RUNTIME_HOST_PROTOCOL_VERSION, |
| type ConnectionCatalogQueryResult, |
| type InteractionPendingSnapshot, |
| type SubscriptionFrame, |
| type TurnMessageSubmitInput, |
| type TurnSnapshot, |
| type TurnStartResult, |
| } from '../protocol/index.js'; |
| import { SessionAdmissionGate } from '../server/session-admission-gate.js'; |
| import { FramedTransport } from '../transport/framed-transport.js'; |
| import { readLedgerMessages } from './fixtures/ledger-transcript.js'; |
| |
| import { |
| CONNECTION_EFFECT_MODEL_IDS, |
| PROCESS_TIMEOUT_MS, |
| SubscriptionProbe, |
| assertJsonLines, |
| connectClient, |
| quoteRefs, |
| requireStartedTurn, |
| operationError, |
| quotedContent, |
| sendStartWithoutReadingResponse, |
| startConnectionEffectProvider, |
| userRuntimeContent, |
| waitForDurableMessageConflict, |
| waitForPendingInteraction, |
| waitForRunningTurn, |
| waitForTerminalTurn, |
| waitForTurn, |
| withExecutionRoot, |
| } from './fixtures/execution-host-suite.js'; |
| |
| test('subscribed Clients share one canonical queue and ordered root handoff', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const host = await fixture.startHost(); |
| const desktop = await connectClient(fixture.root); |
| const tui = await connectClient(fixture.root); |
| const desktopSubscription = await desktop.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| const tuiSubscription = await tui.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| const desktopProbe = new SubscriptionProbe(desktopSubscription); |
| const tuiProbe = new SubscriptionProbe(tuiSubscription); |
| for (const subscription of [desktopSubscription, tuiSubscription]) { |
| assert.equal(subscription.hostEpoch, host.hostEpoch); |
| assert.equal(subscription.snapshot.rootTurn, null); |
| assert.equal(subscription.snapshot.projectionRevision, 1); |
| assert.equal(subscription.snapshot.queue.hostEpoch, host.hostEpoch); |
| } |
| |
| const firstTurnId = randomUUID(); |
| const started = requireStartedTurn( |
| await desktop.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId: firstTurnId, |
| content: { text: `continuity root ${'x'.repeat(540)}` }, |
| }), |
| ); |
| for (const probe of [desktopProbe, tuiProbe]) { |
| const liveDelta = await probe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_delta' && frame.delta.turnId === firstTurnId, |
| 'continuity did not publish the live assistant delta', |
| ); |
| assert.equal(liveDelta.kind, 'subscription.session_delta'); |
| if (liveDelta.kind === 'subscription.session_delta') { |
| assert.equal(liveDelta.delta.runId, started.runId); |
| } |
| } |
| |
| const desktopFollowupId = randomUUID(); |
| const desktopFollowupContent = { text: 'continue from the desktop' }; |
| const desktopQueued = await desktop.request('turn.message.submit', { |
| originHostEpoch: host.hostEpoch, |
| sessionId: fixture.sessionId, |
| messageId: desktopFollowupId, |
| content: desktopFollowupContent, |
| placement: 'next_turn', |
| }); |
| assert.equal(desktopQueued.disposition, 'followup'); |
| const tuiFollowupId = randomUUID(); |
| const tuiFollowupContent = { text: 'continue from the terminal' }; |
| const tuiQueued = await tui.request('turn.message.submit', { |
| originHostEpoch: host.hostEpoch, |
| sessionId: fixture.sessionId, |
| messageId: tuiFollowupId, |
| content: tuiFollowupContent, |
| placement: 'next_turn', |
| }); |
| assert.equal(tuiQueued.disposition, 'followup'); |
| for (const probe of [desktopProbe, tuiProbe]) { |
| const queueProjection = await probe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_projection' && |
| frame.snapshot.queue.followup.some((entry) => entry.messageId === desktopFollowupId) && |
| frame.snapshot.queue.followup.some((entry) => entry.messageId === tuiFollowupId), |
| 'continuity did not publish both accepted follow-ups', |
| ); |
| assert.equal(queueProjection.kind, 'subscription.session_projection'); |
| } |
| |
| await desktop.close(); |
| await desktopProbe.waitForFailure('connection_closed'); |
| assert.equal((await tui.status()).connections, 1); |
| const terminal = await tuiProbe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_projection' && |
| frame.snapshot.rootTurn?.turnId === firstTurnId && |
| frame.snapshot.rootTurn.status === 'completed', |
| 'continuity did not publish the terminal root cut', |
| ); |
| assert.equal(terminal.kind, 'subscription.session_projection'); |
| const successor = await tuiProbe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_projection' && |
| frame.snapshot.rootTurn !== null && |
| frame.snapshot.rootTurn.turnId !== firstTurnId, |
| 'continuity did not publish the successor root', |
| ); |
| assert.equal(successor.kind, 'subscription.session_projection'); |
| if (successor.kind !== 'subscription.session_projection' || !successor.snapshot.rootTurn) { |
| return; |
| } |
| assert.equal(successor.snapshot.rootTurn.sessionId, fixture.sessionId); |
| assert.ok(tuiProbe.indexOf(terminal) < tuiProbe.indexOf(successor)); |
| await tuiSubscription.close(); |
| await tuiProbe.done; |
| await waitForTerminalTurn(tui, fixture.sessionId, successor.snapshot.rootTurn.turnId); |
| await tui.close(); |
| await fixture.stopHost(host); |
| const chain = await fixture.readAdmissionChain(); |
| assert.equal(chain.length, 3); |
| assert.deepEqual( |
| chain.slice(0, 2).map((admission) => admission.turnId), |
| [firstTurnId, successor.snapshot.rootTurn.turnId], |
| ); |
| assert.equal(chain[2]?.previousRootTurnId, successor.snapshot.rootTurn.turnId); |
| assert.deepEqual( |
| chain.slice(1).map((admission) => admission.sourceMessages.map((source) => source.messageId)), |
| [[desktopFollowupId], [tuiFollowupId]], |
| ); |
| assert.deepEqual(chain[1]?.normalizedInput, desktopFollowupContent); |
| assert.deepEqual(chain[2]?.normalizedInput, tuiFollowupContent); |
| }); |
| }); |
| |
| test('a quote-only queued message survives the wire snapshot, the admission chain, and a Host restart (#4804)', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const probe = new SubscriptionProbe( |
| await client.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }), |
| ); |
| |
| // The root turn occupies the session so the quote-only submit queues as |
| // a follow-up instead of opening a successor. |
| const rootTurnId = randomUUID(); |
| requireStartedTurn( |
| await client.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId: rootTurnId, |
| content: { text: `continuity root ${'x'.repeat(540)}` }, |
| }), |
| ); |
| |
| // â‘ The framed-client submit admits the quote-only Message. |
| const messageId = randomUUID(); |
| const queued = await client.request('turn.message.submit', { |
| originHostEpoch: host.hostEpoch, |
| sessionId: fixture.sessionId, |
| messageId, |
| content: quotedContent('the deploy failed at step three'), |
| placement: 'next_turn', |
| }); |
| assert.equal(queued.disposition, 'followup'); |
| |
| // ② The wire queue snapshot carries the entry with its excerpt — the |
| // read-back that a queue-snapshot decoder gap would break. |
| const projection = (await probe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_projection' && |
| frame.snapshot.queue.followup.some((entry) => entry.messageId === messageId), |
| 'the queued quote-only message never reached the wire snapshot', |
| )) as Extract<SubscriptionFrame, { kind: 'subscription.session_projection' }>; |
| const wireEntry = projection.snapshot.queue.followup.find( |
| (entry) => entry.messageId === messageId, |
| ); |
| assert.match( |
| wireEntry?.content.text ?? '', |
| /the deploy failed at step three/, |
| 'the excerpt survives wire serialization', |
| ); |
| |
| // â‘¢ A Host restart re-opens the stores and re-publishes the durable |
| // queue entry with the quote intact — close/reopen the whole chain. |
| await fixture.killHost(host); |
| await client.closed; |
| const secondHost = await fixture.startHost(); |
| const second = await connectClient(fixture.root); |
| const recoveredSubscription = await second.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| // The restart promotes the queued follow-up into a successor root Turn |
| // that runs to completion; the durable user message must carry the |
| // quote excerpt — the full submit -> admission -> wire snapshot -> |
| // reopen round trip me2seeks asked to pin (#5125 review, item 3). |
| const probe2 = new SubscriptionProbe(recoveredSubscription); |
| const successor = await probe2.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_projection' && |
| frame.snapshot.rootTurn !== null && |
| frame.snapshot.rootTurn.turnId !== rootTurnId, |
| 'no successor root was recovered after the Host restart', |
| ); |
| if (successor.kind !== 'subscription.session_projection' || !successor.snapshot.rootTurn) |
| return; |
| await waitForTerminalTurn(second, fixture.sessionId, successor.snapshot.rootTurn.turnId); |
| await second.close(); |
| await probe2.done; |
| await fixture.stopHost(secondHost); |
| assert.deepEqual( |
| (await fixture.readSessionUserMessages()) |
| .filter((message) => message.id === messageId) |
| .map((message) => message.id), |
| [messageId], |
| ); |
| }); |
| }); |
| |
| // The root-start admission path, end to end against the real Host and stores: |
| // the wire decoder, the durable admission authority and the replayed event |
| // must agree that structured content carries the turn, or a quote-only |
| // turn.start is refused (or stored invisible) one layer below any |
| // decoder-level assertion (#4804, #4815 review). |
| test('a quote-only turn.start forms a durable Turn whose user event stays model-visible (#4804)', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const turnId = randomUUID(); |
| const content: MessageContent = { text: '', quotes: quoteRefs('root-start') }; |
| |
| const started = requireStartedTurn( |
| await client.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId, |
| content, |
| }), |
| ); |
| await waitForTerminalTurn(client, fixture.sessionId, turnId); |
| assert.equal(started.turnId, turnId); |
| await client.close(); |
| await fixture.stopHost(host); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.runs.length, 1); |
| assert.deepEqual( |
| ledger.userMessages.map((message) => message.quotes), |
| [content.quotes], |
| ); |
| const userEvent = ledger.runtimeEvents.find( |
| (event) => event.role === 'user' && event.content?.kind === 'text', |
| ); |
| assert.ok(userEvent, 'the admitted turn persisted a user RuntimeEvent'); |
| if (!userEvent || userEvent.content?.kind !== 'text') return; |
| assert.deepEqual(userEvent.content.quotes, content.quotes); |
| // The persisted event passes the exact predicate that gates model replay: |
| // admission, durability and visibility decide by one rule. |
| assert.equal(runtimeEventHasModelVisibleContent(userEvent), true); |
| }); |
| }); |
| |
| test('an attachment-only turn.start forms a durable Turn whose user event stays model-visible (#4804)', async () => { |
| await withExecutionRoot(async (fixture) => { |
| // Stage the canonical Artifact first — the ingest step every real client |
| // performs before a hosted Turn may reference the attachment. |
| const owner = await tryAcquireInteractiveRootOwner(fixture.capability); |
| assert.ok(owner); |
| if (!owner) return; |
| let attachmentRef: AttachmentRef; |
| try { |
| const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); |
| try { |
| const artifact = await artifacts.create({ |
| sessionId: fixture.sessionId, |
| turnId: 'staging', |
| name: 'chart.png', |
| kind: 'image', |
| source: 'user_upload', |
| mimeType: 'image/png', |
| content: 'fake-png-bytes', |
| }); |
| attachmentRef = { |
| kind: 'image', |
| name: artifact.name, |
| mimeType: 'image/png', |
| bytes: artifact.sizeBytes, |
| ref: { |
| kind: 'session_file', |
| sessionId: fixture.sessionId, |
| relativePath: artifact.id, |
| }, |
| }; |
| } finally { |
| artifacts.close(); |
| } |
| } finally { |
| await owner.close(); |
| } |
| |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const turnId = randomUUID(); |
| const content: MessageContent = { text: '', attachments: [attachmentRef] }; |
| |
| const started = requireStartedTurn( |
| await client.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId, |
| content, |
| }), |
| ); |
| await waitForTerminalTurn(client, fixture.sessionId, turnId); |
| assert.equal(started.turnId, turnId); |
| await client.close(); |
| await fixture.stopHost(host); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.runs.length, 1); |
| assert.deepEqual( |
| ledger.userMessages.map((message) => message.attachments), |
| [content.attachments], |
| ); |
| const userEvent = ledger.runtimeEvents.find( |
| (event) => event.role === 'user' && event.content?.kind === 'text', |
| ); |
| assert.ok(userEvent, 'the admitted turn persisted a user RuntimeEvent'); |
| if (!userEvent || userEvent.content?.kind !== 'text') return; |
| assert.deepEqual(userEvent.content.attachments, content.attachments); |
| assert.equal(runtimeEventHasModelVisibleContent(userEvent), true); |
| }); |
| }); |
| |
| test('production UDS admission commits one transcript before the root handoff', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const messageId = randomUUID(); |
| const started = await client.request('turn.message.submit', { |
| originHostEpoch: host.hostEpoch, |
| sessionId: fixture.sessionId, |
| messageId, |
| content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, |
| placement: 'current_turn', |
| }); |
| assert.equal(started.disposition, 'turn_started'); |
| if (started.disposition !== 'turn_started') return; |
| const active = await client.request('turn.query', { |
| sessionId: fixture.sessionId, |
| turnId: started.turnId, |
| }); |
| await client.request('turn.stop', { |
| sessionId: fixture.sessionId, |
| turnId: started.turnId, |
| runId: active.runId, |
| }); |
| await client.close(); |
| await fixture.stopHost(host); |
| const ledger = await fixture.readTurn(started.turnId); |
| assert.deepEqual( |
| ledger.userMessages |
| .filter((message) => message.id === messageId) |
| .map((message) => message.id), |
| [messageId], |
| ); |
| }); |
| }); |
| |
| test('a Host crash after queue admission recovers the durable successor once', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const firstHost = await fixture.startHost(); |
| const first = await connectClient(fixture.root); |
| const started = requireStartedTurn( |
| await first.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId: randomUUID(), |
| content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, |
| }), |
| ); |
| const messageId = randomUUID(); |
| const queued = await first.request('turn.message.submit', { |
| originHostEpoch: firstHost.hostEpoch, |
| sessionId: fixture.sessionId, |
| messageId, |
| content: { text: 'recover this accepted successor' }, |
| placement: 'next_turn', |
| }); |
| assert.equal(queued.disposition, 'followup'); |
| await fixture.killHost(firstHost); |
| await first.closed; |
| |
| const secondHost = await fixture.startHost(); |
| const second = await connectClient(fixture.root); |
| const subscription = await second.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| const probe = new SubscriptionProbe(subscription); |
| const successor = await probe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_projection' && |
| frame.snapshot.rootTurn !== null && |
| frame.snapshot.rootTurn.turnId !== started.turnId, |
| 'durable successor was not recovered after the Host crash', |
| ); |
| assert.equal(successor.kind, 'subscription.session_projection'); |
| if (successor.kind !== 'subscription.session_projection' || !successor.snapshot.rootTurn) |
| return; |
| await waitForTerminalTurn(second, fixture.sessionId, successor.snapshot.rootTurn.turnId); |
| await subscription.close(); |
| await probe.done; |
| await second.close(); |
| await fixture.stopHost(secondHost); |
| assert.deepEqual( |
| (await fixture.readSessionUserMessages()) |
| .filter((message) => message.id === messageId) |
| .map((message) => message.id), |
| [messageId], |
| ); |
| }); |
| }); |
| |
| test('restart replays an atomically admitted root without duplicating its transcript', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const turnId = randomUUID(); |
| const messageId = randomUUID(); |
| const content = { text: 'recover the root after admission before Run creation' }; |
| await fixture.seedAtomicRootAdmissionWithoutRun({ turnId, messageId, content }); |
| |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const terminal = await waitForTerminalTurn(client, fixture.sessionId, turnId); |
| assert.equal(terminal.status, 'completed'); |
| await client.close(); |
| await fixture.stopHost(host); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.deepEqual( |
| ledger.userMessages |
| .filter((message) => message.id === messageId) |
| .map((message) => message.id), |
| [messageId], |
| ); |
| }); |
| }); |
| |
| test('concurrent root admission for one Session has a single winner', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const host = await fixture.startHost(); |
| const first = await connectClient(fixture.root); |
| const second = await connectClient(fixture.root); |
| const turnIds = [randomUUID(), randomUUID()] as const; |
| |
| const outcomes = await Promise.allSettled([ |
| first.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId: turnIds[0], |
| content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, |
| }), |
| second.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId: turnIds[1], |
| content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, |
| }), |
| ]); |
| const winners = outcomes.filter( |
| (outcome): outcome is PromiseFulfilledResult<TurnStartResult> => |
| outcome.status === 'fulfilled', |
| ); |
| const rejected = outcomes.filter( |
| (outcome): outcome is PromiseRejectedResult => outcome.status === 'rejected', |
| ); |
| assert.equal(winners.length, 1); |
| assert.equal(rejected.length, 1); |
| assert.ok(rejected[0]?.reason instanceof RuntimeHostOperationError); |
| assert.equal(rejected[0]?.reason.code, 'session_busy'); |
| |
| const winnerResult = winners[0]?.value; |
| assert.ok(winnerResult); |
| const winner = requireStartedTurn(winnerResult); |
| await first.request('turn.stop', { |
| sessionId: fixture.sessionId, |
| turnId: winner.turnId, |
| runId: winner.runId, |
| }); |
| await first.close(); |
| await second.close(); |
| await fixture.stopHost(host); |
| |
| const chain = await fixture.readAdmissionChain(); |
| assert.equal(chain.length, 1); |
| assert.equal(chain[0]?.turnId, winner.turnId); |
| assert.equal(chain[0]?.previousRootTurnId, null); |
| }); |
| }); |
| |
| test('an archived Session rejects a new Turn before durable admission', async () => { |
| await withExecutionRoot(async (fixture) => { |
| await fixture.archiveSession(); |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const turnId = randomUUID(); |
| |
| await assert.rejects( |
| () => |
| client.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId, |
| content: { text: 'must not execute' }, |
| }), |
| operationError('session_archived'), |
| ); |
| assert.equal((await client.status()).state, 'ready'); |
| await client.close(); |
| await fixture.stopHost(host); |
| |
| assert.deepEqual(await fixture.readTurnFootprint(turnId), { |
| admitted: false, |
| runCount: 0, |
| userMessageCount: 0, |
| }); |
| }); |
| }); |
| |
| test('a killed Host is recovered exactly once before its successor becomes ready', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const firstHost = await fixture.startHost(); |
| const first = await connectClient(fixture.root); |
| const firstSubscription = await first.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| const firstProbe = new SubscriptionProbe(firstSubscription); |
| const turnId = randomUUID(); |
| const started = requireStartedTurn( |
| await first.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId, |
| content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, |
| }), |
| ); |
| await firstProbe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_projection' && |
| frame.snapshot.rootTurn?.runId === started.runId && |
| frame.snapshot.rootTurn.status !== 'admitted', |
| 'first Host did not publish the active root projection', |
| ); |
| const pending = await waitForPendingInteraction(firstSubscription, firstProbe, started.runId); |
| assert.equal(pending.sessionId, fixture.sessionId); |
| assert.equal(pending.turnId, turnId); |
| assert.equal(pending.runId, started.runId); |
| const questionRequest = pending.request; |
| assert.ok(questionRequest.kind === 'question'); |
| |
| await fixture.killHost(firstHost); |
| await first.closed; |
| await firstProbe.waitForFailure('connection_closed'); |
| const secondHost = await fixture.startHost(); |
| const second = await connectClient(fixture.root); |
| const recoveredSubscription = await second.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| const recovered = await second.request('turn.query', { |
| sessionId: fixture.sessionId, |
| turnId, |
| }); |
| assert.equal(recovered.status, 'failed'); |
| if (recovered.status === 'failed') assert.equal(recovered.failureClass, 'app_restarted'); |
| assert.notEqual(recoveredSubscription.hostEpoch, firstSubscription.hostEpoch); |
| assert.equal(recoveredSubscription.snapshot.projectionRevision, 1); |
| assert.deepEqual(recoveredSubscription.snapshot.rootTurn, recovered); |
| assert.equal(recoveredSubscription.snapshot.queue.hostEpoch, recoveredSubscription.hostEpoch); |
| assert.deepEqual(recoveredSubscription.snapshot.queue.steering, []); |
| assert.deepEqual(recoveredSubscription.snapshot.queue.followup, []); |
| const closed = await second.request('interaction.query', { |
| sessionId: fixture.sessionId, |
| interactionId: pending.interactionId, |
| }); |
| assert.equal(closed.sessionId, fixture.sessionId); |
| assert.equal(closed.turnId, turnId); |
| assert.equal(closed.runId, started.runId); |
| assert.equal(closed.status, 'closed'); |
| assert.equal(closed.outcome.kind, 'closure'); |
| if (closed.outcome.kind === 'closure') assert.equal(closed.outcome.reason, 'host_restarted'); |
| await assert.rejects( |
| () => |
| second.request('interaction.answer', { |
| sessionId: fixture.sessionId, |
| interactionId: pending.interactionId, |
| answer: { |
| kind: 'question', |
| answers: questionRequest.questions.map(() => null), |
| }, |
| }), |
| operationError('already_resolved'), |
| ); |
| await recoveredSubscription.close(); |
| await second.close(); |
| await fixture.stopHost(secondHost); |
| |
| const thirdHost = await fixture.startHost(); |
| const third = await connectClient(fixture.root); |
| const stable = await third.request('turn.query', { |
| sessionId: fixture.sessionId, |
| turnId, |
| }); |
| assert.deepEqual(stable, recovered); |
| assert.equal(stable.runId, started.runId); |
| assert.deepEqual( |
| await third.request('interaction.query', { |
| sessionId: fixture.sessionId, |
| interactionId: pending.interactionId, |
| }), |
| closed, |
| ); |
| await third.close(); |
| await fixture.stopHost(thirdHost); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.terminalEvents.length, 1); |
| assert.equal(ledger.classification.kind, 'fact'); |
| if (ledger.classification.kind === 'fact') { |
| assert.equal(ledger.classification.fact.failureClass, 'app_restarted'); |
| } |
| }); |
| }); |
| |
| test('graceful Host shutdown stops and drains an active Turn before releasing ownership', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const subscription = await client.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| const probe = new SubscriptionProbe(subscription); |
| const turnId = randomUUID(); |
| const started = requireStartedTurn( |
| await client.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId, |
| content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, |
| }), |
| ); |
| // What this pins is the drain of an ACTIVE Turn, and `turn.start` |
| // returning only says the Turn was admitted. Waiting for the question it |
| // is about to ask is what makes it active, so stopping before that would |
| // leave which state the Host drains up to how fast the machine is. |
| await waitForPendingInteraction(subscription, probe, started.runId); |
| |
| const exit = await fixture.stopHost(host); |
| assert.deepEqual(exit, { code: 0, signal: null }); |
| await client.closed; |
| await probe.waitForFailure('connection_closed'); |
| |
| const successor = await fixture.startHost(); |
| const observer = await connectClient(fixture.root); |
| const stable = await observer.request('turn.query', { |
| sessionId: fixture.sessionId, |
| turnId, |
| }); |
| assert.equal(stable.runId, started.runId); |
| assert.equal(stable.status, 'cancelled'); |
| await observer.close(); |
| await fixture.stopHost(successor); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.terminalEvents.length, 1); |
| assert.equal(ledger.classification.kind, 'fact'); |
| if (ledger.classification.kind === 'fact') { |
| assert.equal(ledger.classification.fact.runStatus, 'cancelled'); |
| assert.notEqual(ledger.classification.fact.failureClass, 'app_restarted'); |
| } |
| }); |
| }); |
| |
| test('Host shutdown contains a user-question admission rejected by Interaction drain', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| const subscription = await client.openSessionSubscription({ |
| sessionId: fixture.sessionId, |
| transcript: { kind: 'none' }, |
| }); |
| const probe = new SubscriptionProbe(subscription); |
| const turnId = randomUUID(); |
| const started = requireStartedTurn( |
| await client.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId, |
| content: { text: FAKE_ASK_USER_QUESTION_DURING_DRAIN_PROMPT }, |
| }), |
| ); |
| |
| await probe.waitFor( |
| (frame) => |
| frame.kind === 'subscription.session_event' && |
| frame.runId === started.runId && |
| frame.event.type === 'tool_start' && |
| frame.event.toolName === 'AskUserQuestion', |
| 'question scenario did not reach its admission checkpoint', |
| ); |
| |
| const exit = await fixture.stopHost(host, { type: 'shutdown_question_admission' }); |
| assert.deepEqual(exit, { code: 0, signal: null }); |
| await client.closed; |
| await probe.waitForFailure('connection_closed'); |
| await fixture.assertOwnerAvailable(); |
| assert.equal(await fixture.readPendingInteractionCount(), 0); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.terminalEvents.length, 1); |
| assert.equal(ledger.classification.kind, 'fact'); |
| if (ledger.classification.kind === 'fact') { |
| assert.equal(ledger.classification.fact.runStatus, 'cancelled'); |
| assert.notEqual(ledger.classification.fact.failureClass, 'app_restarted'); |
| } |
| }); |
| }); |
| |
| test('a durable admission without a Run resumes before the Host becomes ready', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const turnId = randomUUID(); |
| const quotes = quotedContent('recover pending admission'); |
| const { runId } = await fixture.seedAdmission(turnId, quotes); |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| |
| const recovered = await client.request('turn.query', { |
| sessionId: fixture.sessionId, |
| turnId, |
| }); |
| assert.equal(recovered.runId, runId); |
| assert.ok(recovered.status === 'running' || recovered.status === 'waiting_for_user'); |
| await assert.rejects( |
| () => |
| client.request('turn.start', { |
| sessionId: fixture.sessionId, |
| turnId: randomUUID(), |
| content: { text: 'must remain behind the recovered admission' }, |
| }), |
| operationError('session_busy'), |
| ); |
| const stopped = await client.request( |
| 'turn.stop', |
| { |
| sessionId: fixture.sessionId, |
| turnId, |
| runId, |
| }, |
| PROCESS_TIMEOUT_MS, |
| ); |
| assert.equal(stopped.status, 'cancelled'); |
| await client.close(); |
| await fixture.stopHost(host); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.runs.length, 1); |
| assert.equal(ledger.userMessages.length, 1); |
| assert.deepEqual(ledger.userMessages[0]?.quotes, quotes.quotes); |
| assert.deepEqual(userRuntimeContent(ledger.runtimeEvents)?.quotes, quotes.quotes); |
| assert.equal(ledger.terminalEvents.length, 1); |
| assert.equal(ledger.classification.kind, 'fact'); |
| if (ledger.classification.kind === 'fact') { |
| assert.notEqual(ledger.classification.fact.failureClass, 'app_restarted'); |
| } |
| }); |
| }); |
| |
| test('startup recovery compares an existing quoted UserMessage canonically', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const turnId = randomUUID(); |
| const content = quotedContent('recover existing message'); |
| const { runId, userMessageId } = await fixture.seedRunWithUserMessage(turnId, content); |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| |
| const recovered = await client.request('turn.query', { |
| sessionId: fixture.sessionId, |
| turnId, |
| }); |
| assert.equal(recovered.runId, runId); |
| assert.equal(recovered.status, 'failed'); |
| await client.close(); |
| await fixture.stopHost(host); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.userMessages.length, 1); |
| assert.equal(ledger.userMessages[0]?.id, userMessageId); |
| assert.deepEqual(ledger.userMessages[0]?.quotes, content.quotes); |
| assert.equal(ledger.terminalEvents.length, 1); |
| }); |
| }); |
| |
| test('startup recovery restores the admitted UserMessage before terminalizing its Run', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const turnId = randomUUID(); |
| const { runId, userMessageId } = await fixture.seedRunWithoutUserMessage( |
| turnId, |
| 'recover the admitted message', |
| ); |
| const host = await fixture.startHost(); |
| const client = await connectClient(fixture.root); |
| |
| const recovered = await client.request('turn.query', { |
| sessionId: fixture.sessionId, |
| turnId, |
| }); |
| assert.equal(recovered.runId, runId); |
| assert.equal(recovered.status, 'failed'); |
| if (recovered.status === 'failed') { |
| assert.equal(recovered.failureClass, 'app_restarted'); |
| } |
| await client.close(); |
| await fixture.stopHost(host); |
| |
| const ledger = await fixture.readTurn(turnId); |
| assert.equal(ledger.runs.length, 1); |
| assert.equal(ledger.userMessages.length, 1); |
| assert.equal(ledger.userMessages[0]?.id, userMessageId); |
| assert.equal(ledger.terminalEvents.length, 1); |
| }); |
| }); |
| |
| test('startup recovery canonically closes pending linked child admissions without inventing identity', async () => { |
| await withExecutionRoot(async (fixture) => { |
| const initial = await fixture.seedPendingChildAdmission('linked_child_initial'); |
| const resume = await fixture.seedPendingChildAdmission('linked_child_resume'); |
| const retry = await fixture.seedPendingChildAdmission('linked_child_provider_retry'); |
| const graph = await fixture.seedPendingChildAdmission('claimed_agent_graph_intent'); |
| |
| const firstHost = await fixture.startHost(); |
| await fixture.stopHost(firstHost); |
| const secondHost = await fixture.startHost(); |
| await fixture.stopHost(secondHost); |
| |
| const reader = await tryAcquireInteractiveRootReader(fixture.capability); |
| assert.ok(reader); |
| if (!reader) throw new Error('Unable to acquire recovery result reader'); |
| let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined; |
| try { |
| stores = await openInteractiveExecutionStoresForRead(reader.lease); |
| for (const recovered of [initial, resume, retry, graph]) { |
| const run: RuntimeInvocationRecord | undefined = ( |
| await stores.runtimeEventStore.listSessionInvocations(recovered.sessionId) |
| ).find((candidate) => candidate.runId === recovered.runId); |
| assert.ok(run); |
| assert.equal(runtimeInvocationOutcome(run), 'failed'); |
| assert.equal(runtimeInvocationFailureClass(run), 'app_restarted'); |
| const lineage: RuntimeInvocationLineage | undefined = run.opening.lineage; |
| assert.equal(lineage?.agentId, recovered.agentId); |
| assert.equal(lineage?.agentName, recovered.agentName); |
| assert.equal(run.opening.configuration.workspaceIdentity, undefined); |
| if (recovered.kind === 'linked_child_resume') { |
| assert.equal(lineage?.resumedFromRunId, recovered.sourceRunId); |
| assert.equal(lineage?.retriedFromRunId, undefined); |
| } else if (recovered.kind === 'linked_child_provider_retry') { |
| assert.equal(lineage?.retriedFromRunId, recovered.sourceRunId); |
| assert.equal(lineage?.resumedFromRunId, undefined); |
| } else { |
| assert.equal(lineage?.resumedFromRunId, undefined); |
| assert.equal(lineage?.retriedFromRunId, undefined); |
| } |
| const runtimeEvents = await stores.runtimeEventStore.readImmutableRuntimeEvents( |
| recovered.sessionId, |
| recovered.runId, |
| ); |
| const terminal = classifyTerminalRuntimeLedger(run, runtimeEvents); |
| assert.equal(terminal.kind, 'fact'); |
| if (terminal.kind === 'fact') { |
| assert.equal(terminal.fact.runStatus, 'failed'); |
| assert.equal(terminal.fact.failureClass, 'app_restarted'); |
| } |
| const userMessages: StoredMessage[] = ( |
| await readLedgerMessages(stores.runtimeEventStore, recovered.sessionId) |
| ).filter((message) => message.type === 'user' && message.turnId === recovered.turnId); |
| assert.equal(userMessages.length, recovered.kind === 'linked_child_provider_retry' ? 0 : 1); |
| if (recovered.kind !== 'linked_child_provider_retry') { |
| assert.equal(userMessages[0]?.id, recovered.userMessageId); |
| } |
| } |
| } finally { |
| await stores?.sessionStore.close?.(); |
| await reader.close(); |
| } |
| }); |
| }); |
