diff --git a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts index 308b9dce42..fa8cb4a3fc 100644 --- a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts @@ -582,5 +582,6 @@ function continuitySnapshot() { followup: [], }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }; } diff --git a/apps/desktop/src/main/__tests__/permission-response-ipc-boundary.test.ts b/apps/desktop/src/main/__tests__/permission-response-ipc-boundary.test.ts index bbc6bbad36..e5c6a2585f 100644 --- a/apps/desktop/src/main/__tests__/permission-response-ipc-boundary.test.ts +++ b/apps/desktop/src/main/__tests__/permission-response-ipc-boundary.test.ts @@ -251,9 +251,14 @@ describe('permission response IPC boundary', () => { normalizeStopSessionInput({ source: 'stop_button', expectedTurnId: 'turn-workhub', + expectedAdmissionId: 'side-admission', extra: true, }), - { source: 'stop_button', expectedTurnId: 'turn-workhub' }, + { + source: 'stop_button', + expectedTurnId: 'turn-workhub', + expectedAdmissionId: 'side-admission', + }, ); assert.throws(() => normalizeStopSessionInput(null), /stop session input/); assert.throws(() => normalizeStopSessionInput({ source: 'toolbar' }), /stop session source/); @@ -261,5 +266,9 @@ describe('permission response IPC boundary', () => { () => normalizeStopSessionInput({ expectedTurnId: '' }), /expectedTurnId/, ); + assert.throws( + () => normalizeStopSessionInput({ expectedAdmissionId: '' }), + /expectedAdmissionId/, + ); }); }); diff --git a/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts b/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts index 8553509e2d..262600e919 100644 --- a/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts +++ b/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts @@ -121,7 +121,7 @@ describe('quote companion disposal fencing', () => { ...defaults.sideChat, send: async () => { sends += 1; - return { ok: true as const }; + return { ok: true as const, turnId: 'side-chat-turn' }; }, }; @@ -140,7 +140,7 @@ describe('quote companion disposal fencing', () => { }); it('does not consume quotes or report success when disposal wins the send race', async () => { - const pendingSend = deferred<{ ok: true }>(); + const pendingSend = deferred<{ ok: true; turnId: string }>(); let disposed = false; let consumed = 0; const defaults = createFakeWorkbarServices(); @@ -158,7 +158,7 @@ describe('quote companion disposal fencing', () => { ); disposed = true; - pendingSend.resolve({ ok: true }); + pendingSend.resolve({ ok: true, turnId: 'side-chat-turn' }); assert.deepEqual(await turn, { status: 'disposed' }); assert.equal(consumed, 0); @@ -179,7 +179,7 @@ describe('quote companion disposal fencing', () => { }, send: async () => { sends += 1; - return { ok: true as const }; + return { ok: true as const, turnId: 'side-chat-turn' }; }, }; const turn = performCompanionTurn( diff --git a/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts b/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts index 4070741470..c92153b392 100644 --- a/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts +++ b/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts @@ -22,6 +22,7 @@ import { afterEach, test } from 'node:test'; import { parseHTML } from 'linkedom'; import { act, createElement } from 'react'; import { createRoot, type Root } from 'react-dom/client'; +import type { SessionEvent } from '@maka/core/events'; import type { SessionChangedEvent, SessionSummary, TurnRecord } from '@maka/core/session'; import { createFakeWorkbarServices, @@ -44,18 +45,69 @@ const originalGlobals = { let mountedRoot: Root | undefined; const SOURCE_SESSION = session('source-session'); -afterEach(async () => { - if (mountedRoot) { - await act(async () => { - mountedRoot?.unmount(); - await Promise.resolve(); - }); - } - mountedRoot = undefined; - Object.assign(globalThis, originalGlobals); -}); +function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((settle, fail) => { + resolve = settle; + reject = fail; + }); + return { promise, reject, resolve }; +} -test('retries a busy Side Conversation at the newest settled boundary and clears its banner', async () => { +type QueueUpdate = Extract; +type QueueEntry = NonNullable[number]; + +function completeEvent(id: string, turnId: string, ts: number): SessionEvent { + return { type: 'complete', id, turnId, ts, stopReason: 'end_turn' }; +} + +function textDeltaEvent(id: string, turnId: string, ts: number, text: string): SessionEvent { + return { type: 'text_delta', id, messageId: 'assistant-message', turnId, ts, text }; +} + +function queueUpdateEvent( + id: string, + turnId: string, + ts: number, + steeringEntries: readonly QueueEntry[] = [], + followupEntries: readonly QueueEntry[] = [], +): QueueUpdate { + return { + type: 'queue_update', + id, + turnId, + ts, + queueRevision: 1, + steering: steeringEntries.map((entry) => entry.content.text), + followup: followupEntries.map((entry) => entry.content.text), + steeringEntries: [...steeringEntries], + followupEntries: [...followupEntries], + }; +} + +function messageAdmittedEvent( + id: string, + turnId: string, + ts: number, + messageId: string, +): SessionEvent { + return { type: 'message_admission', id, messageId, turnId, ts, outcome: 'admitted' }; +} + +function recoverableErrorEvent(id: string, turnId: string, ts: number): SessionEvent { + return { + type: 'error', + id, + turnId, + ts, + recoverable: true, + reason: 'connection_closed', + message: 'connection closed', + }; +} + +function installDom() { const parsed = parseHTML('
'); const { document, window } = parsed; Object.assign(globalThis, { @@ -67,16 +119,110 @@ test('retries a busy Side Conversation at the newest settled boundary and clears Node: window.Node, IS_REACT_ACT_ENVIRONMENT: true, }); + const container = document.querySelector('#root'); + assert.ok(container); + return container; +} - let listCount = 0; - let sessionChange: ((event: SessionChangedEvent) => void) | undefined; - let releaseRetry: (() => void) | undefined; - const branchInputs: Array<{ sourceTurnId: string; copyId: string }> = []; +async function renderProbe( + sideChat: Partial, + options: { + ownership?: boolean; + sourceSession?: SessionSummary; + ready?: (container: Element) => boolean; + onSend?: (send: (text: string) => Promise) => void; + onSteer?: (steer: (text: string) => Promise) => void; + onStop?: (stop: () => Promise) => void; + } = {}, +) { + const container = installDom(); const defaults = createFakeWorkbarServices(); const services: WorkbarServices = { ...defaults, sideChat: { ...defaults.sideChat, + listTurns: async () => [settledTurn('source-turn')], + branchFromTurn: async () => ({ ok: true as const, session: session('side-conversation') }), + ...sideChat, + }, + }; + const root = createRoot(container); + mountedRoot = root; + const children = options.ownership + ? createElement(QuoteCompanionOwnershipProbe, { + onSend: options.onSend ?? (() => undefined), + onSteer: options.onSteer, + onStop: options.onStop, + }) + : createElement(QuoteCompanionProbe, { sourceSession: options.sourceSession }); + + await act(async () => { + root.render(createElement(WorkbarServicesProvider, { services, children })); + await Promise.resolve(); + }); + await waitUntil( + () => + options.ready?.(container) ?? + container.firstElementChild?.getAttribute('data-companion-id') === 'side-conversation', + ); + return { container, root, services }; +} + +async function renderOwnershipProbe(sideChat: Partial) { + let send!: (text: string) => Promise; + let steer!: (text: string) => Promise; + let stop!: () => Promise; + let eventHandler: ((event: SessionEvent) => void) | undefined; + const subscribeEvents = sideChat.subscribeEvents; + const rendered = await renderProbe( + { + ...sideChat, + subscribeEvents: (sessionId, handler, onSeeded, onSeedError) => { + eventHandler = handler; + if (subscribeEvents) { + return subscribeEvents(sessionId, handler, onSeeded, onSeedError); + } + onSeeded?.(); + return () => undefined; + }, + }, + { + ownership: true, + onSend: (value) => (send = value), + onSteer: (value) => (steer = value), + onStop: (value) => (stop = value), + }, + ); + return { + ...rendered, + send: (text: string) => send(text), + steer: (text: string) => steer(text), + stop: () => stop(), + emit(event: SessionEvent) { + assert.ok(eventHandler); + eventHandler(event); + }, + }; +} + +afterEach(async () => { + if (mountedRoot) { + await act(async () => { + mountedRoot?.unmount(); + await Promise.resolve(); + }); + } + mountedRoot = undefined; + Object.assign(globalThis, originalGlobals); +}); + +test('retries a busy Side Conversation at the newest settled boundary and clears its banner', async () => { + let listCount = 0; + let sessionChange: ((event: SessionChangedEvent) => void) | undefined; + let releaseRetry: (() => void) | undefined; + const branchInputs: Array<{ sourceTurnId: string; copyId: string }> = []; + const { container } = await renderProbe( + { listTurns: async () => { listCount += 1; return listCount === 1 @@ -100,22 +246,8 @@ test('retries a busy Side Conversation at the newest settled boundary and clears }; }, }, - }; - const container = document.querySelector('#root'); - assert.ok(container); - const root = createRoot(container); - mountedRoot = root; - - await act(async () => { - root.render( - createElement(WorkbarServicesProvider, { - services, - children: createElement(QuoteCompanionProbe), - }), - ); - await Promise.resolve(); - }); - await waitUntil(() => branchInputs.length === 1 && sessionChange !== undefined); + { ready: () => branchInputs.length === 1 && sessionChange !== undefined }, + ); assert.match(container.textContent, /main conversation or a linked task is still running/i); const probe = container.firstElementChild; assert.ok(probe); @@ -152,24 +284,9 @@ test('retries a busy Side Conversation at the newest settled boundary and clears }); test('does not restart foreground setup when the source Session object refreshes', async () => { - const parsed = parseHTML('
'); - const { document, window } = parsed; - Object.assign(globalThis, { - document, - window, - HTMLElement: window.HTMLElement, - HTMLIFrameElement: window.HTMLIFrameElement ?? class HTMLIFrameElement {}, - Event: window.Event, - Node: window.Node, - IS_REACT_ACT_ENVIRONMENT: true, - }); - let branchCount = 0; - const defaults = createFakeWorkbarServices(); - const services: WorkbarServices = { - ...defaults, - sideChat: { - ...defaults.sideChat, + const { container, root, services } = await renderProbe( + { listTurns: async () => [settledTurn('settled-turn')], branchFromTurn: async () => { branchCount += 1; @@ -179,37 +296,676 @@ test('does not restart foreground setup when the source Session object refreshes return await new Promise(() => undefined); }, }, - }; - const container = document.querySelector('#root'); - assert.ok(container); - const root = createRoot(container); - mountedRoot = root; + { sourceSession: session('source-session'), ready: () => branchCount === 1 }, + ); + const probe = container.firstElementChild; + assert.ok(probe); + await waitUntil( + () => branchCount === 1 && probe.getAttribute('data-preparing') === 'false', + ); - const render = (sourceSession: SessionSummary) => + await act(async () => { root.render( createElement(WorkbarServicesProvider, { services, - children: createElement(QuoteCompanionProbe, { sourceSession }), + children: createElement(QuoteCompanionProbe, { + sourceSession: session('source-session'), + }), }), ); + await Promise.resolve(); + }); + + assert.equal(branchCount, 1); + assert.equal(probe.getAttribute('data-preparing'), 'false'); +}); + +test('keeps Side Conversation events owned by the Host-admitted turn across an admission race', async () => { + const pendingSend = deferred<{ ok: true; turnId: string }>(); + const { container, emit, send } = await renderOwnershipProbe({ + send: async () => pendingSend.promise, + }); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('new prompt'); + await Promise.resolve(); + }); + + await act(async () => { + emit(completeEvent('late-old-terminal', 'old-turn', 1)); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'true'); + + await act(async () => { + emit(textDeltaEvent('new-text-before-response', 'host-admitted-turn', 2, 'answer')); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'true'); + + await act(async () => { + pendingSend.resolve({ ok: true, turnId: 'host-admitted-turn' }); + assert.equal(await sendResult, true); + await Promise.resolve(); + }); + + const probe = container.firstElementChild; + assert.ok(probe); + assert.equal(probe.getAttribute('data-live-turn-id'), 'host-admitted-turn'); + assert.equal(probe.getAttribute('data-live-text'), 'answer'); + assert.equal(probe.getAttribute('data-streaming'), 'true'); + assert.equal(probe.getAttribute('data-processing'), 'false'); +}); + +test('binds a busy-raced Side Conversation send through its Host-admitted message identity', async () => { + let admissionId: string | undefined; + const pendingSend = deferred<{ + ok: true; + steered: true; + turnId: string; + messageId: string; + }>(); + const { container, emit, send } = await renderOwnershipProbe({ + send: async (_sessionId, command) => { + admissionId = command.turnId; + return pendingSend.promise; + }, + }); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('steer the active turn'); + await Promise.resolve(); + }); + await act(async () => { + emit(completeEvent('late-old-terminal', 'old-turn', 1)); + emit( + queueUpdateEvent('accepted-queue', 'host-active-turn', 2, [ + { + entryId: 'accepted-entry', + messageId: admissionId as string, + content: { text: 'steer the active turn' }, + placement: 'current_turn', + state: 'queued', + }, + ]), + ); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'true'); + assert.notEqual( + container.firstElementChild?.getAttribute('data-live-turn-id'), + 'host-active-turn', + ); + + await act(async () => { + pendingSend.resolve({ + ok: true, + steered: true, + turnId: 'requested-turn-is-not-the-owner', + messageId: admissionId as string, + }); + assert.equal(await sendResult, true); + await Promise.resolve(); + }); + assert.notEqual( + container.firstElementChild?.getAttribute('data-live-turn-id'), + 'host-active-turn', + ); + await act(async () => { + emit( + messageAdmittedEvent( + 'accepted-admission', + 'host-active-turn', + 2.5, + admissionId as string, + ), + ); + emit(textDeltaEvent('accepted-text', 'host-active-turn', 3, 'answer after steering')); + await Promise.resolve(); + }); + + const probe = container.firstElementChild; + assert.ok(probe); + assert.equal(probe.getAttribute('data-live-turn-id'), 'host-active-turn'); + assert.equal(probe.getAttribute('data-live-text'), 'answer after steering'); + assert.equal(probe.getAttribute('data-streaming'), 'true'); + assert.equal(probe.getAttribute('data-processing'), 'false'); +}); + +test('replays queued Side Conversation text after Host assigns the ticket to a successor Turn', async () => { + let admissionId: string | undefined; + const pendingSend = deferred<{ + ok: false; + reason: 'outcome_unknown'; + messageId: string; + }>(); + const { container, emit, send } = await renderOwnershipProbe({ + send: async (_sessionId, command) => { + admissionId = command.turnId; + return pendingSend.promise; + }, + }); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('continue in the successor turn'); + await Promise.resolve(); + }); + await act(async () => { + emit( + messageAdmittedEvent( + 'successor-admission', + 'successor-root', + 1, + admissionId as string, + ), + ); + emit(queueUpdateEvent('successor-queue', 'successor-root', 2)); + emit(textDeltaEvent('successor-text', 'successor-root', 3, 'answer from successor')); + await Promise.resolve(); + }); await act(async () => { - render(session('source-session')); + pendingSend.resolve({ + ok: false, + reason: 'outcome_unknown', + messageId: admissionId as string, + }); + assert.equal(await sendResult, true); await Promise.resolve(); }); + const probe = container.firstElementChild; assert.ok(probe); + assert.equal(probe.getAttribute('data-live-turn-id'), 'successor-root'); + assert.equal(probe.getAttribute('data-live-text'), 'answer from successor'); + assert.equal(probe.getAttribute('data-processing'), 'false'); +}); + +test('clears a stopped Side Conversation admission when its live retraction is lost', async () => { + let admissionId: string | undefined; + const pendingStop = deferred<{ kind: 'retracted'; messageId: string }>(); + const pendingSend = deferred<{ + ok: true; + steered: true; + turnId: string; + messageId: string; + }>(); + const { container, send, stop } = await renderOwnershipProbe({ + send: async (_sessionId, command) => { + admissionId = command.turnId; + return pendingSend.promise; + }, + stop: async (_sessionId, expectedAdmissionId) => { + assert.equal(expectedAdmissionId, admissionId); + return pendingStop.promise; + }, + }); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('stop this queued send'); + await Promise.resolve(); + }); + let stopResult!: Promise; + await act(async () => { + stopResult = stop(); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'true'); + + await act(async () => { + pendingStop.resolve({ kind: 'retracted', messageId: admissionId as string }); + await stopResult; + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'false'); + + await act(async () => { + pendingSend.resolve({ + ok: true, + steered: true, + turnId: 'old-turn', + messageId: admissionId as string, + }); + assert.equal(await sendResult, false); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'false'); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), ''); +}); + +test('keeps a Side Conversation admission when Host stop outcome is unknown', async () => { + let admissionId: string | undefined; + const pendingStop = deferred(); + const { container, emit, send, stop } = await renderOwnershipProbe({ + send: async (_sessionId, command) => { + admissionId = command.turnId; + return { + ok: false as const, + reason: 'outcome_unknown' as const, + messageId: admissionId as string, + }; + }, + stop: async () => pendingStop.promise, + }); + + await act(async () => { + assert.equal(await send('keep this admission'), true); + await Promise.resolve(); + }); + let stopResult!: Promise; + await act(async () => { + stopResult = stop(); + await Promise.resolve(); + }); + await act(async () => { + emit( + messageAdmittedEvent( + 'admitted-during-unknown-stop', + 'admitted-after-unknown-stop', + 1, + admissionId as string, + ), + ); + emit( + textDeltaEvent( + 'text-during-unknown-stop', + 'admitted-after-unknown-stop', + 2, + 'answer', + ), + ); + await Promise.resolve(); + }); + await act(async () => { + pendingStop.reject(new Error('Host stop result is unknown')); + await stopResult; + await Promise.resolve(); + }); await waitUntil( - () => branchCount === 1 && probe.getAttribute('data-preparing') === 'false', + () => + container.firstElementChild?.getAttribute('data-live-turn-id') === + 'admitted-after-unknown-stop', ); + assert.equal( + container.firstElementChild?.getAttribute('data-live-turn-id'), + 'admitted-after-unknown-stop', + ); + assert.equal(container.firstElementChild?.getAttribute('data-live-text'), 'answer'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'false'); +}); +test('stops a bound Side Conversation by its exact Host Turn identity', async () => { + let stoppedAdmissionId: string | undefined; + const { send, stop } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'host-turn-1' }), + stop: async (_sessionId, admissionId) => { + stoppedAdmissionId = admissionId; + }, + }); + await act(async () => { + assert.equal(await send('start this exact turn'), true); + await Promise.resolve(); + }); await act(async () => { - render(session('source-session')); + await stop(); await Promise.resolve(); }); + assert.equal(stoppedAdmissionId, 'host-turn-1'); +}); - assert.equal(branchCount, 1); - assert.equal(probe.getAttribute('data-preparing'), 'false'); +test('releases a queued Side Conversation admission from the Host queue retract', async () => { + let admissionId: string | undefined; + const pendingSend = deferred<{ + ok: true; + steered: true; + turnId: string; + messageId: string; + }>(); + const { container, emit, send } = await renderOwnershipProbe({ + send: async (_sessionId, command) => { + admissionId = command.turnId; + return pendingSend.promise; + }, + }); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('retract this queued send'); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'true'); + + await act(async () => { + emit({ + type: 'message_admission', + id: 'retracted-admission', + turnId: 'old-turn', + ts: 1, + messageId: admissionId as string, + outcome: 'retracted', + }); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'false'); + + await act(async () => { + pendingSend.resolve({ + ok: true, + steered: true, + turnId: 'not-the-owner', + messageId: admissionId as string, + }); + assert.equal(await sendResult, false); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'false'); +}); + +test('keeps the same Side Conversation admission across a recoverable subscription error', async () => { + let subscriptionCount = 0; + const pendingSend = deferred<{ ok: true; turnId: string }>(); + const { container, emit, send } = await renderOwnershipProbe({ + subscribeEvents: (_sessionId, _handler, onSeeded) => { + subscriptionCount += 1; + onSeeded?.(); + return () => undefined; + }, + send: async () => pendingSend.promise, + }); + assert.equal(subscriptionCount, 1); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('survive a recoverable stream error'); + await Promise.resolve(); + }); + await waitUntil(() => container.firstElementChild?.getAttribute('data-processing') === 'true'); + await act(async () => { + emit(recoverableErrorEvent('recoverable-subscription-error', 'old-turn', 1)); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-processing'), 'true'); + assert.equal(subscriptionCount, 1); + + await act(async () => { + pendingSend.resolve({ ok: true, turnId: 'late-turn' }); + assert.equal(await sendResult, true); + await Promise.resolve(); + }); + await act(async () => { + emit(completeEvent('late-complete', 'late-turn', 2)); + await Promise.resolve(); + }); + await waitUntil(() => container.firstElementChild?.getAttribute('data-processing') === 'false'); +}); + +test('keeps the active Side Conversation streaming when Stop retracts a queued steer', async () => { + const pendingSteer = deferred<{ kind: 'queued'; messageId: string }>(); + let admissionId: string | undefined; + let steerCalls = 0; + const { container, emit, send, steer, stop } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + steer: async (_sessionId, _text, requestedAdmissionId) => { + steerCalls += 1; + admissionId = requestedAdmissionId; + return pendingSteer.promise; + }, + stop: async () => undefined, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + await waitUntil(() => container.firstElementChild?.getAttribute('data-streaming') === 'true'); + let steerResult!: Promise; + await act(async () => { + steerResult = steer('queue this steer'); + await Promise.resolve(); + }); + await waitUntil(() => steerCalls === 1); + assert.ok(admissionId); + await act(async () => { + await stop(); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'old-turn'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); + + await act(async () => { + emit({ + type: 'message_admission', + id: 'queued-steer-retracted', + turnId: 'old-turn', + ts: 1, + messageId: admissionId as string, + outcome: 'retracted', + }); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); + + await act(async () => { + pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + assert.equal(await steerResult, false); + await Promise.resolve(); + }); +}); + +test('stops the active Side Conversation after retracting its queued steer', async () => { + const pendingSteer = deferred<{ kind: 'queued'; messageId: string }>(); + let admissionId: string | undefined; + const stoppedIds: Array = []; + const { send, steer, stop } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + steer: async (_sessionId, _text, requestedAdmissionId) => { + admissionId = requestedAdmissionId; + return pendingSteer.promise; + }, + stop: async (_sessionId, expectedId) => { + stoppedIds.push(expectedId); + return expectedId && expectedId === admissionId + ? { kind: 'retracted' as const, messageId: expectedId } + : undefined; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + let steerResult!: Promise; + await act(async () => { + steerResult = steer('queue this steer'); + await Promise.resolve(); + }); + await waitUntil(() => admissionId !== undefined); + await act(async () => { + await stop(); + await Promise.resolve(); + }); + await act(async () => { + await stop(); + await Promise.resolve(); + }); + + assert.deepEqual(stoppedIds, [admissionId, 'old-turn']); + await act(async () => { + pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + assert.equal(await steerResult, false); + await Promise.resolve(); + }); +}); + +test('does not let an older Stop failure release a newer active Turn Stop', async () => { + const pendingSteer = deferred<{ kind: 'queued'; messageId: string }>(); + const queuedStop = deferred(); + const activeStop = deferred(); + let admissionId: string | undefined; + const stoppedIds: Array = []; + const { emit, send, steer, stop } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + steer: async (_sessionId, _text, requestedAdmissionId) => { + admissionId = requestedAdmissionId; + return pendingSteer.promise; + }, + stop: async (_sessionId, expectedId) => { + stoppedIds.push(expectedId); + return stoppedIds.length === 1 ? queuedStop.promise : activeStop.promise; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + let steerResult!: Promise; + await act(async () => { + steerResult = steer('queue this steer'); + await Promise.resolve(); + }); + await waitUntil(() => admissionId !== undefined); + const queuedStopResult = stop(); + await act(async () => { + emit({ + type: 'message_admission', + id: 'queued-steer-retracted-before-stop-reply', + turnId: 'old-turn', + ts: 1, + messageId: admissionId as string, + outcome: 'retracted', + }); + await Promise.resolve(); + }); + const activeStopResult = stop(); + await act(async () => { + queuedStop.reject(new Error('old Stop reply was lost')); + await queuedStopResult; + await Promise.resolve(); + }); + const duplicateStopResult = stop(); + await Promise.resolve(); + + assert.deepEqual(stoppedIds, [admissionId, 'old-turn']); + activeStop.resolve(undefined); + await Promise.all([activeStopResult, duplicateStopResult]); + await act(async () => { + pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + assert.equal(await steerResult, false); + await Promise.resolve(); + }); +}); + +test('continues projecting the active Turn while a steer awaits Host admission', async () => { + const pendingSteer = deferred<{ kind: 'queued'; messageId: string }>(); + let admissionId: string | undefined; + const { container, emit, send, steer } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + steer: async (_sessionId, _text, requestedAdmissionId) => { + admissionId = requestedAdmissionId; + return pendingSteer.promise; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + let steerResult!: Promise; + await act(async () => { + steerResult = steer('queue this steer'); + await Promise.resolve(); + }); + await waitUntil(() => admissionId !== undefined); + await act(async () => { + emit(textDeltaEvent('old-turn-text', 'old-turn', 1, 'still streaming')); + await Promise.resolve(); + }); + + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'old-turn'); + assert.equal(container.firstElementChild?.getAttribute('data-live-text'), 'still streaming'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); + + await act(async () => { + pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + assert.equal(await steerResult, true); + await Promise.resolve(); + }); +}); + +test('fails a send when observation seed rejects and resubscribes for retry', async () => { + let sendCalls = 0; + let subscriptionCount = 0; + let rejectSeed: ((error: unknown) => void) | undefined; + let markSeeded: (() => void) | undefined; + const { send } = await renderOwnershipProbe({ + subscribeEvents: (_sessionId, _handler, onSeeded, onSeedError) => { + subscriptionCount += 1; + if (subscriptionCount === 1) rejectSeed = onSeedError; + else markSeeded = onSeeded; + return () => undefined; + }, + send: async () => { + sendCalls += 1; + return { ok: true as const, turnId: 'retry-turn' }; + }, + }); + assert.ok(rejectSeed); + + let failedResult!: Promise; + await act(async () => { + failedResult = send('observer failure'); + rejectSeed?.(new Error('observer failed')); + assert.equal(await failedResult, false); + }); + assert.equal(sendCalls, 0); + assert.equal(subscriptionCount, 2); + assert.ok(markSeeded); + + await act(async () => { + markSeeded?.(); + await Promise.resolve(); + }); + let retryResult!: Promise; + await act(async () => { + retryResult = send('retry after observer failure'); + assert.equal(await retryResult, true); + }); + assert.equal(sendCalls, 1); +}); + +test('releases a send waiting for observation when the Side Conversation is disposed', async () => { + let sendCalls = 0; + let unsubscribed = false; + const { root, send } = await renderOwnershipProbe({ + subscribeEvents: () => () => { + unsubscribed = true; + }, + send: async () => { + sendCalls += 1; + return { ok: true as const, turnId: 'disposed-turn' }; + }, + }); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('dispose while observing'); + await Promise.resolve(); + }); + await act(async () => { + root.unmount(); + await Promise.resolve(); + }); + + assert.equal(await sendResult, false); + assert.equal(sendCalls, 0); + assert.equal(unsubscribed, true); + mountedRoot = undefined; }); function QuoteCompanionProbe(props: { sourceSession?: SessionSummary }) { @@ -227,6 +983,31 @@ function QuoteCompanionProbe(props: { sourceSession?: SessionSummary }) { }, companion.error); } +function QuoteCompanionOwnershipProbe(props: { + onSend: (send: (text: string) => Promise) => void; + onSteer?: (steer: (text: string) => Promise) => void; + onStop?: (stop: () => Promise) => void; +}) { + const companion = useQuoteCompanion({ + panelId: 'ownership-panel', + pendingQuotes: [], + sourceSession: SOURCE_SESSION, + locale: 'en', + onQuotesConsumed: () => undefined, + }); + props.onSend(companion.send); + props.onSteer?.(companion.steer); + props.onStop?.(companion.stop); + return createElement('div', { + 'data-companion-id': companion.companionSession?.id ?? '', + 'data-error': companion.error ?? '', + 'data-live-turn-id': companion.liveTurn?.turnId ?? '', + 'data-live-text': companion.liveTurn?.steps.find((step) => step.text)?.text?.text ?? '', + 'data-streaming': String(companion.streaming), + 'data-processing': String(companion.processing), + }); +} + function session(id: string): SessionSummary { return { id, diff --git a/apps/desktop/src/main/__tests__/runtime-host-bot-session-adapter.test.ts b/apps/desktop/src/main/__tests__/runtime-host-bot-session-adapter.test.ts index 6482a9a2bf..2dc9bd14ea 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-bot-session-adapter.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-bot-session-adapter.test.ts @@ -467,6 +467,7 @@ function continuitySnapshot(rootTurn: TurnSnapshot | null): SessionContinuitySna goal: null, queue: { hostEpoch: 'host-1', queueRevision: 0, steering: [], followup: [] }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }; } diff --git a/apps/desktop/src/main/__tests__/runtime-host-client.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client.test.ts index 01fd41646e..748a679bc4 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client.test.ts @@ -181,6 +181,7 @@ function subscription( goal: null, queue: { hostEpoch: 'host-1', queueRevision: 0, steering: [], followup: [] }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }, loadTranscript: async (_decodeMessage: (value: unknown) => T) => { lifecycle.push(`${sessionId}:transcript`); diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts index ea2d4b24b7..af1ab014cc 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts @@ -1119,6 +1119,7 @@ function continuitySnapshot( followup: [], }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], ...overrides, }; } diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts index 4514ea326b..708350c8f2 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts @@ -30,7 +30,10 @@ import { SESSION_CONTINUITY_SCHEMA_VERSION, type SessionCatalogProjection, } from "@maka/runtime-host/protocol"; -import { RuntimeHostOperationError } from '@maka/runtime-host/client'; +import { + RuntimeHostOperationError, + RuntimeHostRequestInterruptedError, +} from '@maka/runtime-host/client'; import { createAttachmentApprovalRegistry } from "../attachment-approval.js"; import type { DesktopRuntimeHostSession } from "../runtime-host-client.js"; import { @@ -40,6 +43,13 @@ import { import { RuntimeHostSessionObserver } from "../runtime-host-session-observer.js"; import { runtimeHostSessionFixture } from "./runtime-host-session-test-fixture.js"; +test('registers Session observation as one reconnectable operation', () => { + const ipc = ipcHarness(); + registerExecutionIpc({ client: executionClient({}) }, ipc); + + assert.equal(ipc.reconnectableChannels.has('sessions:observe'), true); +}); + test("keeps synthetic E2E interactions visible through Host hydration and retires their answer", async () => { const observer = observerWithSnapshot(); const ipc = ipcHarness(); @@ -627,6 +637,232 @@ test("queues a mid-turn send as steering when the Host reports the session busy" ]); }); +test("retries a dispatched normal send with its original Turn identity", async () => { + const starts: unknown[] = []; + let reconnectQueries = 0; + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + getSession: async () => { + reconnectQueries += 1; + return sideConversationSession(); + }, + startTurn: async (input) => { + starts.push(input); + if (starts.length === 1) { + throw new RuntimeHostRequestInterruptedError( + "turn.start", + "command", + "dispatched", + "connection_lost", + ); + } + return { + kind: "started", + turn: { + sessionId: input.sessionId, + turnId: input.turnId, + runId: "run-1", + status: "running", + }, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; + }, + }), + newId: () => "turn-1", + }, + ipc, + ); + + const result = await ipc.invoke("sessions:send", "session-1", { + type: "send", + text: "keep this Turn identity", + }); + + assert.equal(reconnectQueries, 2, 'initial Session lookup plus reconnect probe'); + assert.deepEqual(starts, [ + { + sessionId: "session-1", + turnId: "turn-1", + content: { text: "keep this Turn identity", inlineReferences: [] }, + }, + { + sessionId: "session-1", + turnId: "turn-1", + content: { text: "keep this Turn identity", inlineReferences: [] }, + }, + ]); + assert.deepEqual(result, { + ok: true, + turnId: "turn-1", + attachments: [], + inlineReferences: [], + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }); +}); + +test("does not add admission retry semantics to an ordinary send", async () => { + let starts = 0; + let sessionQueries = 0; + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + getSession: async () => { + sessionQueries += 1; + return session(); + }, + startTurn: async () => { + starts += 1; + throw new RuntimeHostRequestInterruptedError( + "turn.start", + "command", + "dispatched", + "connection_lost", + ); + }, + }), + newId: () => "turn-1", + }, + ipc, + ); + + await assert.rejects( + ipc.invoke("sessions:send", "session-1", { + type: "send", + text: "preserve the ordinary send contract", + }), + RuntimeHostRequestInterruptedError, + ); + assert.equal(starts, 1); + assert.equal(sessionQueries, 1, "only the initial Session lookup runs"); +}); + +test("retries a dispatched busy fallback with its original message identity", async () => { + const submits: unknown[] = []; + let reconnectQueries = 0; + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + getSession: async (sessionId) => { + reconnectQueries += 1; + return sessionId === 'side-session' + ? sideConversationSession(sessionId) + : session(); + }, + startTurn: async () => { + throw new RuntimeHostOperationError( + "turn.start", + "session_busy", + "Session already has an active root Turn", + ); + }, + submitMessage: async (input) => { + submits.push(input); + if ( + input.messageId === "turn-unknown" || + input.content.text === "ordinary chat keeps the existing failure contract" + ) { + throw new RuntimeHostOperationError( + "turn.message.submit", + "outcome_unknown", + "Message disposition cannot be proven in this Host Epoch", + ); + } + if (submits.length === 1) { + throw new RuntimeHostRequestInterruptedError( + "turn.message.submit", + "command", + "dispatched", + "connection_lost", + ); + } + return { disposition: "steering", queueRevision: 1 }; + }, + }), + newId: () => "id-1", + }, + ipc, + ); + + const result = await ipc.invoke("sessions:send", "side-session", { + type: "send", + turnId: "turn-1", + text: "keep this message identity", + }); + + assert.equal(reconnectQueries, 2, 'initial Session lookup plus reconnect probe'); + assert.deepEqual(submits, [ + { + sessionId: "side-session", + messageId: "turn-1", + content: { text: "keep this message identity", inlineReferences: [] }, + placement: "current_turn", + }, + { + sessionId: "side-session", + messageId: "turn-1", + content: { text: "keep this message identity", inlineReferences: [] }, + placement: "current_turn", + }, + ]); + assert.deepEqual(result, { + ok: true, + steered: true, + turnId: "turn-1", + messageId: "turn-1", + attachments: [], + inlineReferences: [], + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }); + await assert.rejects( + ipc.invoke("sessions:send", "session-1", { + type: "send", + turnId: "turn-unknown", + text: "ordinary chat keeps the existing failure contract", + }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'outcome_unknown', + ); + assert.deepEqual( + await ipc.invoke("sessions:send", "side-session", { + type: "send", + turnId: "turn-unknown", + text: "keep waiting for the Host outcome", + }), + { + ok: false, + reason: "outcome_unknown", + messageId: "turn-unknown", + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }, + ); +}); + +test("returns the Host-started Turn identity when a direct steer races idle", async () => { + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + getSession: async () => session(), + submitMessage: async () => ({ + disposition: "turn_started", + turnId: "host-started-turn", + }), + }), + newId: () => "steer-message-id", + }, + ipc, + ); + + assert.deepEqual(await ipc.invoke("sessions:steer", "session-1", "continue now"), { + kind: "started", + turnId: "host-started-turn", + }); +}); + test("starts the turn from the queued message when the busy race resolves idle", async () => { const changes: unknown[] = []; const submits: unknown[] = []; @@ -920,11 +1156,19 @@ test("routes per-entry queue mutations to the Runtime Host", async () => { test("binds steer and stop to Host-owned queue and active Turn identities", async () => { const submits: unknown[] = []; const interrupts: unknown[] = []; + const retractions: unknown[] = []; const stopLifecycle: string[] = []; let sequence = 0; const client = executionClient({ submitMessage: async (input) => { submits.push(input); + if (input.messageId === 'unknown-ticket') { + throw new RuntimeHostOperationError( + 'turn.message.submit', + 'outcome_unknown', + 'Message disposition cannot be proven in this Host Epoch', + ); + } return { disposition: "steering", queueRevision: 2 }; }, interruptTurn: async (input) => { @@ -943,17 +1187,40 @@ test("binds steer and stop to Host-owned queue and active Turn identities", asyn }, }; }, + retractQueueEntry: async (input) => { + retractions.push(input); + return { queueRevision: 3 }; + }, + }); + const observer = observerWithSnapshot({ + queue: { + hostEpoch: 'host-1', + queueRevision: 2, + steering: [ + { + entryId: 'entry-1', + messageId: 'steer-ticket-1', + content: { text: 'Continue' }, + placement: 'current_turn', + state: 'queued', + }, + { + entryId: 'entry-2', + messageId: 'in-flight-ticket', + content: { text: 'Already accepted' }, + placement: 'current_turn', + state: 'in_flight', + }, + ], + followup: [], + }, + rootTurnSourceMessageIds: ['successor-ticket'], }); - const observer = observerWithSnapshot(); const ipc = ipcHarness(); registerExecutionIpc( { client, observer, - attachmentApprovals: createAttachmentApprovalRegistry(), - emitSessionsChanged() {}, - stat: async () => ({ size: 0 }), - resizeImage: async (bytes) => bytes, beforeStop() { stopLifecycle.push("teardown"); }, @@ -963,29 +1230,69 @@ test("binds steer and stop to Host-owned queue and active Turn identities", asyn ); assert.deepEqual( - await ipc.invoke("sessions:steer", "session-1", " Continue "), + await ipc.invoke("sessions:steer", "session-1", " Continue ", "steer-ticket-1"), { kind: "queued", + messageId: "steer-ticket-1", }, ); + assert.deepEqual( + await ipc.invoke('sessions:steer', 'session-1', 'Continue', 'unknown-ticket'), + { kind: 'outcome_unknown', messageId: 'unknown-ticket' }, + ); + assert.deepEqual( + await ipc.invoke("sessions:stop", "session-1", { + source: "stop_button", + expectedAdmissionId: "steer-ticket-1", + }), + { kind: 'retracted', messageId: 'steer-ticket-1' }, + ); + assert.deepEqual(retractions, [ + { + sessionId: 'session-1', + entryId: 'entry-1', + retractId: 'id-1', + }, + ]); + assert.deepEqual(stopLifecycle, []); + for (const expectedAdmissionId of ['in-flight-ticket', 'successor-ticket']) { + await ipc.invoke('sessions:stop', 'session-1', { + source: 'stop_button', + expectedAdmissionId, + }); + } + assert.deepEqual(stopLifecycle, ['teardown', 'interrupt', 'teardown', 'interrupt']); await ipc.invoke("sessions:stop", "session-1", { source: "stop_button", expectedTurnId: "turn-unrelated", }); - assert.deepEqual(stopLifecycle, []); + assert.deepEqual(stopLifecycle, ['teardown', 'interrupt', 'teardown', 'interrupt']); await ipc.invoke("sessions:stop", "session-1", { source: "stop_button", expectedTurnId: "turn-1", }); - assert.deepEqual(stopLifecycle, ["teardown", "interrupt"]); + assert.deepEqual(stopLifecycle, [ + 'teardown', + 'interrupt', + 'teardown', + 'interrupt', + 'teardown', + 'interrupt', + ]); assert.deepEqual(submits, [ { sessionId: "session-1", - messageId: "id-1", + messageId: "steer-ticket-1", content: { text: "Continue" }, placement: "current_turn", }, + { + sessionId: 'session-1', + messageId: 'unknown-ticket', + content: { text: 'Continue' }, + placement: 'current_turn', + }, ]); assert.deepEqual(interrupts, [ { @@ -994,6 +1301,18 @@ test("binds steer and stop to Host-owned queue and active Turn identities", asyn turnId: "turn-1", runId: "run-1", }, + { + sessionId: 'session-1', + interruptId: 'id-3', + turnId: 'turn-1', + runId: 'run-1', + }, + { + sessionId: 'session-1', + interruptId: 'id-4', + turnId: 'turn-1', + runId: 'run-1', + }, ]); await observer.close(); }); @@ -1041,12 +1360,15 @@ function unusedObserver(): RuntimeHostSessionObserver { }); } -function observerWithSnapshot(): RuntimeHostSessionObserver { - return observerWithTranscript([]); +function observerWithSnapshot( + overrides: Partial = {}, +): RuntimeHostSessionObserver { + return observerWithTranscript([], overrides); } function observerWithTranscript( transcript: readonly import('@maka/core/session').StoredMessage[], + overrides: Partial = {}, ): RuntimeHostSessionObserver { let finishEvents!: () => void; const eventsFinished = new Promise((resolve) => { @@ -1079,6 +1401,8 @@ function observerWithTranscript( followup: [], }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], + ...overrides, }, activeAssistantStreams: [], transcript: Promise.resolve([...transcript]), @@ -1100,15 +1424,24 @@ type IpcHandler = Parameters["handle"]>[1]; function ipcHarness() { const handlers = new Map(); + const reconnectableChannels = new Set(); const sender = Object.assign(new EventEmitter(), { id: 9, send() {} }); + const register = (channel: string, handler: IpcHandler) => { + assert.equal( + handlers.has(channel), + false, + `duplicate handler: ${channel}`, + ); + handlers.set(channel, handler); + }; return { + reconnectableChannels, handle(channel: string, handler: IpcHandler) { - assert.equal( - handlers.has(channel), - false, - `duplicate handler: ${channel}`, - ); - handlers.set(channel, handler); + register(channel, handler); + }, + handleReconnectableRead(channel: string, handler: IpcHandler) { + reconnectableChannels.add(channel); + register(channel, handler); }, async invoke(channel: string, ...args: unknown[]): Promise { const handler = handlers.get(channel); @@ -1125,22 +1458,21 @@ function ipcHarness() { } function registerExecutionIpc( - deps: Omit< - RuntimeHostSessionExecutionIpcDeps, - 'sessionCopyCleanup' | 'onBackgroundError' | 'observations' - > & - Partial< - Pick< - RuntimeHostSessionExecutionIpcDeps, - 'sessionCopyCleanup' | 'onBackgroundError' | 'observations' - > - >, - ipcMain: Pick, + deps: Pick & + Partial>, + ipcMain: Pick & { handleReconnectableRead?: IpcMain['handle'] }, ): (sessionId: string) => Promise { + const observer = deps.observer ?? unusedObserver(); return registerRuntimeHostSessionExecutionIpc( { + observer, + attachmentApprovals: createAttachmentApprovalRegistry(), + emitSessionsChanged() {}, + stat: async () => ({ size: 0 }), + resizeImage: async (bytes) => bytes, + beforeStop() {}, ...deps, - observations: deps.observations ?? deps.observer, + observations: deps.observations ?? observer, sessionCopyCleanup: deps.sessionCopyCleanup ?? unusedSessionCopyCleanup(), onBackgroundError: deps.onBackgroundError ?? (() => undefined), }, @@ -1161,9 +1493,9 @@ function unusedSessionCopyCleanup(): RuntimeHostSessionExecutionIpcDeps['session }; } -function session(cwd = "/workspace"): SessionCatalogProjection { +function session(cwd = "/workspace", id = 'session-1'): SessionCatalogProjection { return { - id: "session-1", + id, revision: 1, workspace: { target: { kind: 'host_path', path: cwd }, @@ -1187,3 +1519,7 @@ function session(cwd = "/workspace"): SessionCatalogProjection { orchestrationMode: "default", }; } + +function sideConversationSession(id = 'session-1'): SessionCatalogProjection { + return { ...session('/workspace', id), labels: [SIDE_CONVERSATION_SESSION_LABEL] }; +} diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts index 29e4ebcde1..e8c21a387a 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts @@ -2570,6 +2570,7 @@ function continuitySnapshot( followup: [], }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], ...overrides, }; } diff --git a/apps/desktop/src/main/__tests__/seed-completion.test.ts b/apps/desktop/src/main/__tests__/seed-completion.test.ts deleted file mode 100644 index 37482395a8..0000000000 --- a/apps/desktop/src/main/__tests__/seed-completion.test.ts +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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 assert from 'node:assert/strict'; -import { test } from 'node:test'; -import { notifyWhenSeeded } from '../../preload/seed-completion.js'; - -test('does not notify after the subscription is disposed', async () => { - let resolveSeed!: () => void; - const seed = new Promise((resolve) => { - resolveSeed = resolve; - }); - let notifications = 0; - - const dispose = notifyWhenSeeded(seed, () => { - notifications++; - }); - dispose(); - resolveSeed(); - await seed; - await Promise.resolve(); - - assert.equal(notifications, 0); -}); diff --git a/apps/desktop/src/main/permission-response-guard.ts b/apps/desktop/src/main/permission-response-guard.ts index 1dd307674a..4c39f944b5 100644 --- a/apps/desktop/src/main/permission-response-guard.ts +++ b/apps/desktop/src/main/permission-response-guard.ts @@ -65,6 +65,7 @@ interface NormalizedSendSessionCommand { type NormalizedStopSessionInput = { source?: 'stop_button'; expectedTurnId?: string; + expectedAdmissionId?: string; }; export function normalizeSandboxBoundaryResponse(input: unknown): SandboxBoundaryResponse { @@ -324,9 +325,17 @@ export function normalizeStopSessionInput(input: unknown): NormalizedStopSession 'Invalid stop session expectedTurnId', MAX_TURN_ID_LENGTH, ); + const expectedAdmissionId = value.expectedAdmissionId === undefined + ? undefined + : normalizeRequiredString( + value.expectedAdmissionId, + 'Invalid stop session expectedAdmissionId', + MAX_TURN_ID_LENGTH, + ); return { ...(value.source ? { source: 'stop_button' as const } : {}), ...(expectedTurnId ? { expectedTurnId } : {}), + ...(expectedAdmissionId ? { expectedAdmissionId } : {}), }; } diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts index bd961f7fdd..533c8786ec 100644 --- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts @@ -20,8 +20,12 @@ import { randomUUID } from "node:crypto"; import type { IpcMainInvokeEvent } from "electron"; import { MAX_ATTACHMENT_COUNT } from '@maka/core/attachments'; -import { RuntimeHostOperationError } from '@maka/runtime-host/client'; +import { + RuntimeHostOperationError, + RuntimeHostRequestInterruptedError, +} from '@maka/runtime-host/client'; import { SKILL_INVOCATION_TOKEN_SOURCE } from '@maka/core/skill-invocation-token'; +import { isSideConversationSession } from '@maka/core/side-conversation'; import { type SessionChangedEvent, type SessionChangedReason, @@ -56,6 +60,7 @@ import { type RuntimeHostTranscriptTarget, } from "./runtime-host-session-observer.js"; import type { DesktopTranscriptRangeRequest } from '../preload/transcript-contract.js'; +import type { DesktopSessionStopResult } from '../preload/bridge-contract.js'; import { toDesktopHostSessionSummary } from "./runtime-host-session-catalog-ipc-main.js"; import { mergeWorkspaceFileInlineReferences } from "./session-workspace-inline-references.js"; @@ -63,6 +68,24 @@ type SideConversationBranchResult = | { readonly ok: true; readonly session: ReturnType } | { readonly ok: false; readonly reason: 'session_busy' | 'operation_unavailable' }; +async function retryDispatchedCommand( + command: () => Promise, + waitForReconnect: () => Promise, +): Promise { + try { + return await command(); + } catch (error) { + if ( + !(error instanceof RuntimeHostRequestInterruptedError) || + error.dispatch !== 'dispatched' + ) { + throw error; + } + await waitForReconnect(); + return command(); + } +} + type RuntimeHostSessionExecutionClient = Pick< DesktopRuntimeHostClient, | "answerInteraction" @@ -88,6 +111,23 @@ type RuntimeHostSessionExecutionClient = Pick< | "updateSessionConfiguration" >; +async function submitMessageWithReconnect( + client: Pick, + input: Parameters[0], +): Promise> | undefined> { + try { + return await retryDispatchedCommand( + () => client.submitMessage(input), + () => client.getSession(input.sessionId), + ); + } catch (error) { + if (error instanceof RuntimeHostOperationError && error.code === 'outcome_unknown') { + return undefined; + } + throw error; + } +} + export interface RuntimeHostSessionExecutionIpcDeps { client: RuntimeHostSessionExecutionClient; observer: RuntimeHostSessionObserver; @@ -150,15 +190,21 @@ export function registerRuntimeHostSessionExecutionIpc( const newId = deps.newId ?? randomUUID; const stopSession = createRuntimeHostSessionStop(deps, newId); - ipcMain.handle( + handleReconnectableRead( + ipcMain, "sessions:observe", async (event, sessionId: unknown, observerId: unknown) => { const normalizedSessionId = requiredId(sessionId, "Session"); const normalizedObserverId = requiredId(observerId, "Session observer"); + const session = await deps.client.getSession(normalizedSessionId); + if (!session) { + throw new Error(`Runtime Host Session not found: ${normalizedSessionId}`); + } await deps.observations.observe( normalizedSessionId, normalizedObserverId, event.sender as RuntimeHostSessionObserverTarget, + isSideConversationSession(session.labels), ); }, ); @@ -216,6 +262,7 @@ export function registerRuntimeHostSessionExecutionIpc( const session = await deps.client.getSession(sessionId); if (!session) throw new Error(`Runtime Host Session not found: ${sessionId}`); + const sideConversation = isSideConversationSession(session.labels); const turnId = command.turnId ?? newId(); let attachments = retainedAttachmentsForSession( sessionId, @@ -276,7 +323,12 @@ export function registerRuntimeHostSessionExecutionIpc( }; let startResult; try { - startResult = await deps.client.startTurn(startInput); + startResult = sideConversation + ? await retryDispatchedCommand( + () => deps.client.startTurn(startInput), + () => deps.client.getSession(sessionId), + ) + : await deps.client.startTurn(startInput); } catch (error) { // The renderer routes text at a session it sees as running to // `sessions:steer`, but its view can lag the Host: another window, a @@ -297,15 +349,27 @@ export function registerRuntimeHostSessionExecutionIpc( ) { throw error; } - const submitted = await deps.client.submitMessage({ + // Preserve the renderer's command identity in the durable message so + // a lost IPC reply can be reconciled as root-vs-steering later. + const messageId = turnId; + const emptySkillInvocation = { loaded: [], failed: [], receipts: [] }; + const submitInput = { sessionId, - // Preserve the renderer's command identity in the durable message so - // a lost IPC reply can be reconciled as root-vs-steering later. - messageId: turnId, + messageId, content: startInput.content, - placement: "current_turn", - }); - const emptySkillInvocation = { loaded: [], failed: [], receipts: [] }; + placement: 'current_turn' as const, + }; + const submitted = sideConversation + ? await submitMessageWithReconnect(deps.client, submitInput) + : await deps.client.submitMessage(submitInput); + if (!submitted) { + return { + ok: false as const, + reason: 'outcome_unknown' as const, + messageId, + skillInvocation: emptySkillInvocation, + }; + } if (submitted.disposition === "turn_started") { deps.emitSessionsChanged("status-change", sessionId, { turnId: submitted.turnId, @@ -325,6 +389,7 @@ export function registerRuntimeHostSessionExecutionIpc( ok: true as const, steered: true as const, turnId, + ...(sideConversation ? { messageId } : {}), attachments, inlineReferences, skillInvocation: emptySkillInvocation, @@ -351,15 +416,22 @@ export function registerRuntimeHostSessionExecutionIpc( ipcMain.handle( "sessions:steer", - async (_event, sessionId: string, text: unknown) => { + async (_event, sessionId: string, text: unknown, admissionId: unknown) => { const content = steeringContent(text); - await deps.client.submitMessage({ + const messageId = admissionId === undefined ? newId() : requiredId(admissionId, "Admission"); + const submitted = await submitMessageWithReconnect(deps.client, { sessionId, - messageId: newId(), + messageId, content: { text: content }, placement: "current_turn", }); - return { kind: "queued" as const }; + if (!submitted) return { kind: 'outcome_unknown' as const, messageId }; + return submitted.disposition === "turn_started" + ? { kind: "started" as const, turnId: submitted.turnId } + : { + kind: "queued" as const, + messageId, + }; }, ); ipcMain.handle( @@ -512,7 +584,7 @@ export function registerRuntimeHostSessionExecutionIpc( "sessions:stop", async (_event, sessionId: string, input: unknown) => { const normalized = normalizeStopSessionInput(input); - return stopSession(sessionId, normalized.expectedTurnId); + return stopSession(sessionId, normalized); }, ); @@ -685,7 +757,9 @@ export function registerRuntimeHostSessionExecutionIpc( return toDesktopHostSessionSummary(revision); }, ); - return stopSession; + return async (sessionId) => { + await stopSession(sessionId); + }; } function normalizeTranscriptRangeRequest(input: unknown): DesktopTranscriptRangeRequest { @@ -736,8 +810,39 @@ function createRuntimeHostSessionStop( "beforeStop" | "client" | "observer" | "emitSessionsChanged" >, newId: () => string = randomUUID, -): (sessionId: string, expectedTurnId?: string) => Promise { - return async (sessionId, expectedTurnId) => { +): ( + sessionId: string, + target?: { readonly expectedTurnId?: string; readonly expectedAdmissionId?: string }, +) => Promise { + return async (sessionId, target = {}) => { + let expectedTurnId = target.expectedTurnId; + if (target.expectedAdmissionId) { + const observed = await deps.observer.snapshot(sessionId); + const root = observed.rootTurn; + const entry = [...observed.queue.steering, ...observed.queue.followup].find( + (candidate) => candidate.messageId === target.expectedAdmissionId, + ); + if (entry?.state === 'queued') { + await deps.client.retractQueueEntry({ + sessionId, + entryId: entry.entryId, + retractId: newId(), + }); + deps.emitSessionsChanged('status-change', sessionId); + return { kind: 'retracted', messageId: entry.messageId }; + } + if ( + root && + !isTerminalStatus(root.status) && + (root.turnId === target.expectedAdmissionId || + observed.rootTurnSourceMessageIds.includes(target.expectedAdmissionId) || + entry?.state === 'in_flight') + ) { + expectedTurnId = root.turnId; + } else { + throw new Error('Host admission outcome is unknown'); + } + } if (expectedTurnId) { const observed = (await deps.observer.snapshot(sessionId)).rootTurn; if ( diff --git a/apps/desktop/src/main/runtime-host-session-observation-registry.ts b/apps/desktop/src/main/runtime-host-session-observation-registry.ts index e60ca88983..0e95057c59 100644 --- a/apps/desktop/src/main/runtime-host-session-observation-registry.ts +++ b/apps/desktop/src/main/runtime-host-session-observation-registry.ts @@ -76,6 +76,7 @@ function requireTranscriptSource( interface SessionObservationRegistration { readonly sessionId: string; + readonly messageAdmissions: boolean; readonly target: RuntimeHostSessionObserverTarget; readonly destroyedListener: () => void; readonly ready: ObservationReadiness; @@ -151,6 +152,7 @@ export class RuntimeHostSessionObservationRegistry { registration.sessionId, observerId, bindTarget(registration.target), + registration.messageAdmissions, ); if ( this.#source !== source || @@ -218,11 +220,16 @@ export class RuntimeHostSessionObservationRegistry { sessionId: string, observerId: string, target: RuntimeHostSessionObserverTarget, + messageAdmissions = false, ): Promise { this.#assertOpen(); const previous = this.#registrations.get(observerId); if (previous) { - if (previous.sessionId !== sessionId || previous.target.id !== target.id) { + if ( + previous.sessionId !== sessionId || + previous.target.id !== target.id || + previous.messageAdmissions !== messageAdmissions + ) { throw new Error("Runtime Host Session observer identity was reused"); } return previous.ready.promise; @@ -235,6 +242,7 @@ export class RuntimeHostSessionObservationRegistry { void ready.promise.catch(() => undefined); const registration: SessionObservationRegistration = { sessionId, + messageAdmissions, target, destroyedListener, ready, @@ -246,7 +254,12 @@ export class RuntimeHostSessionObservationRegistry { const source = this.#source; if (!source) return registration.ready.promise; try { - await source.observe(sessionId, observerId, this.#bindTarget(target)); + await source.observe( + sessionId, + observerId, + this.#bindTarget(target), + messageAdmissions, + ); if ( this.#source === source && this.#registrations.get(observerId) === registration diff --git a/apps/desktop/src/main/runtime-host-session-observer.ts b/apps/desktop/src/main/runtime-host-session-observer.ts index 390ee5f34e..273dfca234 100644 --- a/apps/desktop/src/main/runtime-host-session-observer.ts +++ b/apps/desktop/src/main/runtime-host-session-observer.ts @@ -118,6 +118,7 @@ interface ObservedSessionState { snapshot?: SessionContinuitySnapshot; projector?: RuntimeHostSessionProjector; transcriptAccess: number; + messageAdmissions: boolean; closing: boolean; } @@ -425,6 +426,7 @@ export class RuntimeHostSessionObserver { sessionId: string, observerId: string, target: RuntimeHostSessionObserverTarget, + messageAdmissions = false, ): Promise { this.#assertOpen(); const previous = this.#observers.get(observerId); @@ -438,6 +440,10 @@ export class RuntimeHostSessionObserver { return; } const state = this.#state(sessionId); + if (messageAdmissions && !state.messageAdmissions) { + state.messageAdmissions = true; + state.projector?.enableMessageAdmissions(); + } let group = state.targets.get(target.id); if (!group) { const destroyedListener = () => { @@ -622,6 +628,7 @@ export class RuntimeHostSessionObserver { subscriptionOwner, pendingTranscriptConsumers: 0, transcriptAccess: 0, + messageAdmissions: false, closing: false, }; this.#states.set(sessionId, state); @@ -800,6 +807,7 @@ export class RuntimeHostSessionObserver { subscription.replica.projectionSeed, this.#now, subscription.activeAssistantStreams, + state.messageAdmissions, ); const terminalTurnIds = new Set(); for (const turnId of state.watchedTurnIds) { diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index d478543721..293268ad20 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -48,7 +48,6 @@ import type { SessionCommand, SessionEvent, ShellRunUpdate, - QueueEnqueueOutcome, } from '@maka/core/events'; import type { UserQuestionResponse } from '@maka/core/user-question'; import type { PermissionMode } from '@maka/core/permission'; @@ -207,6 +206,10 @@ export type DesktopSideConversationBranchResult = | { ok: true; session: DesktopSessionSummary } | { ok: false; reason: 'session_busy' | 'operation_unavailable' }; +export type DesktopSessionStopResult = + | { kind: 'retracted'; messageId: string } + | undefined; + export type DesktopReviseBeforeTurnInput = ReviseBeforeTurnInput & { /** Stable target identity for retrying one Desktop copy action. */ copyId: string; @@ -721,7 +724,7 @@ export interface MakaBridge { sessionId: string, command: | SessionCommand - | { + | { type: 'send'; turnId: string; text: string; @@ -743,7 +746,18 @@ export interface MakaBridge { * The send raced a root Turn another client opened first and was * queued into it as steering instead of starting `turnId`. */ - steered?: true; + steered?: never; + messageId?: never; + attachments: import('@maka/core/events').AttachmentRef[]; + inlineReferences: import('@maka/core/events').InlineReference[]; + skillInvocation: import('@maka/runtime/skill-invocation').SkillInvocationResult; + } + | { + ok: true; + turnId: string; + steered: true; + /** Host admission identity for the message queued as steering. */ + messageId: string; attachments: import('@maka/core/events').AttachmentRef[]; inlineReferences: import('@maka/core/events').InlineReference[]; skillInvocation: import('@maka/runtime/skill-invocation').SkillInvocationResult; @@ -753,12 +767,30 @@ export interface MakaBridge { reason: 'skill_invocation_failed'; skillInvocation: import('@maka/runtime/skill-invocation').SkillInvocationResult; } + | { + ok: false; + reason: 'outcome_unknown'; + messageId: string; + skillInvocation: import('@maka/runtime/skill-invocation').SkillInvocationResult; + } >; stop( sessionId: string, - input?: { source?: 'stop_button'; expectedTurnId?: string }, - ): Promise; - steer(sessionId: string, text: string): Promise; + input?: { + source?: 'stop_button'; + expectedTurnId?: string; + expectedAdmissionId?: string; + }, + ): Promise; + steer( + sessionId: string, + text: string, + admissionId?: string, + ): Promise< + | { kind: 'queued'; messageId: string } + | { kind: 'outcome_unknown'; messageId: string } + | { kind: 'started'; turnId: string } + >; enqueue( sessionId: string, placement: 'current_turn' | 'next_turn', @@ -825,6 +857,7 @@ export interface MakaBridge { handler: (event: SessionEvent) => void, onSeeded?: () => void, onObservationSeed?: (phase: 'pending' | 'ready') => void, + onSeedError?: (error: unknown) => void, ): () => void; subscribeChanges(handler: (event: SessionChangedEvent) => void): () => void; archive(sessionId: string, options?: { revisionFamily?: boolean }): Promise; diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 656472bf20..661baf0b69 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -20,7 +20,6 @@ import { contextBridge, ipcRenderer } from 'electron'; import { encodeIngestItems } from './attachment-ingest-payload.js'; import { collectThreadSearchResponses } from './multi-host-thread-search.js'; -import { notifyWhenSeeded } from './seed-completion.js'; import { releaseSessionObservation } from './session-observation-release.js'; import type { MakaBridge, @@ -31,6 +30,7 @@ import type { RendererIngestInput, DesktopBranchFromTurnInput, DesktopSideConversationBranchResult, + DesktopSessionStopResult, DesktopReviseBeforeTurnInput, AppUpdateInstallRequest, AppUpdateInstallResult, @@ -116,7 +116,6 @@ import type { SessionCommand, SessionEvent, ShellRunUpdate, - QueueEnqueueOutcome, } from '@maka/core/events'; import type { UserQuestionResponse } from '@maka/core/user-question'; import type { PermissionMode } from '@maka/core/permission'; @@ -1528,7 +1527,7 @@ const makaBridge = { sessionId: string, command: | SessionCommand - | { + | { type: 'send'; turnId: string; text: string; @@ -1553,6 +1552,12 @@ const makaBridge = { reason: 'skill_invocation_failed'; skillInvocation: import('@maka/runtime/skill-invocation').SkillInvocationResult; } + | { + ok: false; + reason: 'outcome_unknown'; + messageId: string; + skillInvocation: import('@maka/runtime/skill-invocation').SkillInvocationResult; + } > { const session = await runtimeHostSessionRef(sessionId); const send = async (input: SessionCommand | Record) => { @@ -1589,12 +1594,24 @@ const makaBridge = { }, stop( sessionId: string, - input?: { source?: 'stop_button'; expectedTurnId?: string }, - ): Promise { + input?: { + source?: 'stop_button'; + expectedTurnId?: string; + expectedAdmissionId?: string; + }, + ): Promise { return invokeSessionRuntimeHost('sessions:stop', sessionId, input); }, - steer(sessionId: string, text: string): Promise { - return invokeSessionRuntimeHost('sessions:steer', sessionId, text); + steer( + sessionId: string, + text: string, + admissionId?: string, + ): Promise< + | { kind: 'queued'; messageId: string } + | { kind: 'outcome_unknown'; messageId: string } + | { kind: 'started'; turnId: string } + > { + return invokeSessionRuntimeHost('sessions:steer', sessionId, text, admissionId); }, async enqueue( sessionId: string, @@ -1729,6 +1746,7 @@ const makaBridge = { handler: (event: SessionEvent) => void, onSeeded?: () => void, onObservationSeed?: (phase: 'pending' | 'ready') => void, + onSeedError?: (error: unknown) => void, ): () => void { const observerId = crypto.randomUUID(); let disposed = false; @@ -1736,15 +1754,23 @@ const makaBridge = { let unsubscribeObservationSeed = () => {}; const observeDispatch = runtimeHostSessionRef(sessionId).then((session) => { if (disposed) return { completion: Promise.resolve() }; - unsubscribeEvents = subscribeRuntimeHostEvent( + const profileId = runtimeHostMetadata.get(session.scope.hostId)?.profileId; + if (!profileId) throw new Error('The Runtime Host profile for this task is unavailable'); + // Keep the renderer listener across Host target epochs. The observer + // registry restores this observer on the replacement target. Profile + // identity admits that replacement without accepting another Host's + // same-named Session channel. + unsubscribeEvents = subscribeEveryRuntimeHostEvent( `sessions:event:${session.sessionId}`, - session.scope, - (event: SessionEvent) => handler(projectDesktopSessionEvent(session.scope, event)), + (scope, event: SessionEvent) => { + if (runtimeHostMetadata.get(scope.hostId)?.profileId !== profileId) return; + handler(projectDesktopSessionEvent(scope, event)); + }, ); - unsubscribeObservationSeed = subscribeRuntimeHostEvent( + unsubscribeObservationSeed = subscribeEveryRuntimeHostEvent( 'sessions:observation-seed', - session.scope, - (payload: { sessionId?: string; phase?: string }) => { + (scope, payload: { sessionId?: string; phase?: string }) => { + if (runtimeHostMetadata.get(scope.hostId)?.profileId !== profileId) return; if (payload.sessionId !== session.sessionId) return; if (payload.phase === 'pending' || payload.phase === 'ready') { onObservationSeed?.(payload.phase); @@ -1761,10 +1787,16 @@ const makaBridge = { }; }); const observing = observeDispatch.then(({ completion }) => completion); - const disposeSeedNotification = notifyWhenSeeded(observing, onSeeded); + void observing.then( + () => { + if (!disposed) onSeeded?.(); + }, + (error: unknown) => { + if (!disposed) onSeedError?.(error); + }, + ); return () => { disposed = true; - disposeSeedNotification(); unsubscribeObservationSeed(); unsubscribeEvents(); void releaseSessionObservation(observeDispatch, () => diff --git a/apps/desktop/src/preload/seed-completion.ts b/apps/desktop/src/preload/seed-completion.ts deleted file mode 100644 index 464abe4b7b..0000000000 --- a/apps/desktop/src/preload/seed-completion.ts +++ /dev/null @@ -1,34 +0,0 @@ -/* - * 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. - */ - -export function notifyWhenSeeded( - seed: Promise, - notify: (() => void) | undefined, -): () => void { - let disposed = false; - void seed - .then(() => { - if (!disposed) notify?.(); - }) - .catch(() => undefined); - - return () => { - disposed = true; - }; -} diff --git a/apps/desktop/src/renderer/features/workbar/ports.ts b/apps/desktop/src/renderer/features/workbar/ports.ts index fb14676742..37e97ff710 100644 --- a/apps/desktop/src/renderer/features/workbar/ports.ts +++ b/apps/desktop/src/renderer/features/workbar/ports.ts @@ -18,7 +18,6 @@ */ import type { - QueueEnqueueOutcome, QuoteRef, SessionEvent, ShellRunUpdate, @@ -197,8 +196,15 @@ export interface WorkbarAttachmentsService { } export type SideChatSendResult = - | { ok: true } - | { ok: false; reason?: string }; + | { ok: true; turnId: string; steered?: false } + | { ok: true; turnId: string; steered: true; messageId: string } + | { ok: false; reason: 'outcome_unknown'; messageId: string } + | { ok: false; reason?: string; messageId?: never }; + +export type SideChatSteerResult = + | { kind: 'queued'; messageId: string } + | { kind: 'outcome_unknown'; messageId: string } + | { kind: 'started'; turnId: string }; export interface SideChatSessionPort { listSessions(): Promise; @@ -231,8 +237,11 @@ export interface SideChatSessionPort { attachmentItems?: WorkbarIngestInput[]; }, ): Promise; - stop(sessionId: string): Promise; - steer(sessionId: string, text: string): Promise; + stop( + sessionId: string, + admissionId?: string, + ): Promise<{ kind: 'retracted'; messageId: string } | undefined>; + steer(sessionId: string, text: string, admissionId?: string): Promise; setPermissionMode( sessionId: string, mode: PermissionMode, @@ -249,6 +258,8 @@ export interface SideChatSessionPort { subscribeEvents( sessionId: string, handler: (event: SessionEvent) => void, + onSeeded?: () => void, + onSeedError?: (error: unknown) => void, ): WorkbarUnsubscribe; subscribeSessionChanges(handler: (event: SessionChangedEvent) => void): WorkbarUnsubscribe; } diff --git a/apps/desktop/src/renderer/features/workbar/testing.ts b/apps/desktop/src/renderer/features/workbar/testing.ts index f2e3e940b5..da788175e0 100644 --- a/apps/desktop/src/renderer/features/workbar/testing.ts +++ b/apps/desktop/src/renderer/features/workbar/testing.ts @@ -136,7 +136,10 @@ export function createFakeWorkbarServices( regenerateTurn: async () => undefined, respondToSandboxBoundary: async () => undefined, respondToUserQuestion: async () => undefined, - subscribeEvents: noopSubscription, + subscribeEvents: (_sessionId, _handler, onSeeded) => { + onSeeded?.(); + return noopSubscription(); + }, subscribeSessionChanges: noopSubscription, }, ...overrides, diff --git a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts index dceb846301..626f6d6dce 100644 --- a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts +++ b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts @@ -31,6 +31,7 @@ import type { SessionSummary, TurnRecord } from '@maka/core/session'; import type { UiLocale } from '@maka/core/ui-locale'; import type { SideChatSessionPort, + SideChatSendResult, WorkbarIngestInput, } from '../../ports.js'; import { @@ -167,19 +168,22 @@ export async function dismissCompanionCopy( } /** - * The shared Composer's `streaming` input means "a turn is interruptible", not - * merely "a text delta has arrived". Keep the companion interruptible from the - * optimistic waiting projection through live output; `processing` only chooses - * the quieter pre-first-token presentation inside that same in-flight window. + * The shared Composer's `streaming` input means Host work is interruptible, not + * merely that a text delta has arrived. A pending admission remains stoppable + * before it owns a Turn; an admitted Turn remains stoppable through live output. */ export function deriveCompanionComposerState( - turnInFlight: boolean, + hasPendingAdmission: boolean, + activeTurnId: string | null, liveTurn: LiveTurnProjection | undefined, ): { streaming: boolean; processing: boolean } { - const streaming = turnInFlight && liveTurn?.terminal !== true; + const activeTurnStreaming = activeTurnId !== null && liveTurn?.terminal !== true; + const streaming = hasPendingAdmission || activeTurnStreaming; return { streaming, - processing: streaming && (!liveTurn || liveTurn.phase === 'waiting'), + processing: + streaming && + (!activeTurnStreaming || !liveTurn || liveTurn.phase === 'waiting'), }; } @@ -283,7 +287,9 @@ export async function ensureCompanionFork( } export type CompanionTurnResult = - | { status: 'sent'; forkId: string } + | { status: 'sent'; forkId: string; turnId: string; steered?: false } + | { status: 'sent'; forkId: string; turnId: string; steered: true; messageId: string } + | { status: 'pending'; forkId: string; messageId: string } | { status: 'disposed' } | { status: 'error'; code: CompanionErrorCode }; @@ -335,7 +341,7 @@ export async function performCompanionTurn( if (createdForkId) scheduleCompanionCleanup(deps, createdForkId); return { status: 'disposed' }; } - let result: { ok: true } | { ok: false; reason?: string }; + let result: SideChatSendResult; try { result = await deps.api.send(forkId, { type: 'send', @@ -353,10 +359,15 @@ export async function performCompanionTurn( // run was started, so surface the error and keep the quotes for retry rather // than reporting success and hanging in the processing state. if (!result.ok) { + if (result.reason === 'outcome_unknown' && result.messageId) { + return { status: 'pending', forkId, messageId: result.messageId }; + } return { status: 'error', code: 'send_rejected' }; } deps.onQuotesConsumed(); - return { status: 'sent', forkId }; + return result.steered + ? { status: 'sent', forkId, turnId: result.turnId, steered: true, messageId: result.messageId } + : { status: 'sent', forkId, turnId: result.turnId }; } export function isCompanionTurnTerminal(event: SessionEvent): boolean { diff --git a/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts b/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts index 7c4bcb5be8..cc5c5fec10 100644 --- a/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts +++ b/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts @@ -61,6 +61,37 @@ import { } from './quote-companion-panel-state.js'; import type { CompanionForkVisibilityEvent } from './quote-companion-visibility.js'; +type PendingAdmission = { + messageId: string; + events: SessionEvent[]; + consumeOnAdmission?: () => void; + stopPromise?: Promise<'confirmed' | 'unknown'>; +}; + +type AdmissionOutcome = + | { kind: 'admitted'; turnId: string } + | { kind: 'retracted' }; + +function admissionOutcomeForMessage( + events: readonly SessionEvent[], + messageId: string, +): AdmissionOutcome | undefined { + const admitted = events.find( + (event) => + event.type === 'message_admission' && + event.outcome === 'admitted' && + event.messageId === messageId, + ); + if (admitted) return { kind: 'admitted', turnId: admitted.turnId }; + const retracted = events.some( + (event) => + event.type === 'message_admission' && + event.outcome === 'retracted' && + event.messageId === messageId, + ); + return retracted ? { kind: 'retracted' } : undefined; +} + export interface UseQuoteCompanionInput { /** Stable owner for the currently mounted panel generation. */ panelId: string; @@ -157,9 +188,11 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const sourceSessionIdRef = useRef(sourceSession?.id); sourceSessionIdRef.current = sourceSessionId; const forkSetupPromiseRef = useRef | null>(null); - const stopRequestedRef = useRef(false); + const stopRequestRef = useRef | null>(null); const activeTurnIdRef = useRef(null); - const turnInFlightRef = useRef(false); + const pendingAdmissionRef = useRef(null); + const subscriptionReadyRef = useRef>(Promise.resolve()); + const submitLockRef = useRef(false); const settlingTurnIdsRef = useRef>(new Set()); const onForkVisibilityChangeRef = useRef(onForkVisibilityChange); onForkVisibilityChangeRef.current = onForkVisibilityChange; @@ -173,7 +206,13 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const liveTurnRef = useRef(liveTurn); liveTurnRef.current = liveTurn; const [interactions, setInteractions] = useState({}); - const [turnInFlight, setTurnInFlight] = useState(false); + const [pendingAdmission, setPendingAdmissionState] = useState(null); + const { streaming, processing } = deriveCompanionComposerState( + pendingAdmission !== null, + activeTurnIdRef.current, + liveTurn, + ); + const turnInFlight = streaming; const [preparing, setPreparing] = useState(Boolean(sourceSession)); const [permissionModePending, setPermissionModePending] = useState(false); const [regeneratePendingTurnId, setRegeneratePendingTurnId] = useState( @@ -192,25 +231,17 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const mountedRef = useMountedRef(); const dismissalGuardRef = useRef(createCompanionDismissalGuard()); - // Subscribe to the fork's event stream + load its transcript. Called - // synchronously the moment the fork is committed, BEFORE the run starts, so - // no boundary request / complete can be missed (the stream has no replay). - const subscribeToFork = useCallback((forkId: string) => { - void sideChat.readSettledMessages(forkId) - .then(({ messages }) => { - if (mountedRef.current) { - setAllMessages((current) => mergeSettledMessages(current, messages)); - } - }) - .catch(() => { - if (mountedRef.current) setError(copyRef.current.errors.settlementFailed); - }); - unsubscribeRef.current = sideChat.subscribeEvents(forkId, (event: SessionEvent) => { - if (!mountedRef.current) return; + const setPendingAdmission = useCallback((admission: PendingAdmission | null) => { + pendingAdmissionRef.current = admission; + setPendingAdmissionState(admission); + }, []); + + const applyOwnedEvent = useCallback( + (forkId: string, event: SessionEvent) => { const effect = companionRunEventEffect( event, activeTurnIdRef.current, - stopRequestedRef.current, + stopRequestRef.current !== null, localeRef.current, ); if (effect.kind === 'ignore') return; @@ -237,24 +268,155 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan setAllMessages((current) => mergeSettledMessages(current, next)); setLiveTurn((prev) => (prev ? reconcileTerminalLiveTurn(prev, next) : prev)); activeTurnIdRef.current = null; - turnInFlightRef.current = false; - stopRequestedRef.current = false; - setTurnInFlight(false); + stopRequestRef.current = null; }) .catch(() => { if (!mountedRef.current || activeTurnIdRef.current !== settledTurnId) return; activeTurnIdRef.current = null; - turnInFlightRef.current = false; - stopRequestedRef.current = false; - setTurnInFlight(false); + stopRequestRef.current = null; setError((current) => current ?? copyRef.current.errors.settlementFailed); }) .finally(() => { settlingTurnIdsRef.current.delete(settledTurnId); }); } + }, + [mountedRef, sideChat], + ); + + const bindAdmittedTurn = useCallback( + ( + forkId: string, + turnId: string, + options: { readonly preserveLiveTurn?: boolean } = {}, + ) => { + const admission = pendingAdmissionRef.current; + if (!admission) return; + setPendingAdmission(null); + activeTurnIdRef.current = turnId; + ownTurnIdsRef.current.add(turnId); + admission.consumeOnAdmission?.(); + setError(null); + setOwnTurnTick((tick) => tick + 1); + if (!(options.preserveLiveTurn && liveTurnRef.current?.turnId === turnId)) { + setLiveTurn(armLiveTurn(turnId)); + } + for (const event of admission.events) { + if (event.turnId === turnId) applyOwnedEvent(forkId, event); + } + }, + [applyOwnedEvent, setPendingAdmission], + ); + + const releaseAdmission = useCallback( + (admission: PendingAdmission, message?: string) => { + if (pendingAdmissionRef.current !== admission) return; + setPendingAdmission(null); + if (stopRequestRef.current === admission.stopPromise) stopRequestRef.current = null; + if (!activeTurnIdRef.current) setLiveTurn(undefined); + if (message) setError(message); + }, + [setPendingAdmission], + ); + + const resolveAdmission = useCallback( + ( + forkId: string, + admission: PendingAdmission, + messageId: string, + preserveLiveTurn = false, + ): AdmissionOutcome | undefined => { + const outcome = admissionOutcomeForMessage(admission.events, messageId); + if (outcome?.kind === 'admitted') { + bindAdmittedTurn(forkId, outcome.turnId, { preserveLiveTurn }); + } else if (outcome?.kind === 'retracted') { + releaseAdmission(admission); + } + return outcome; + }, + [bindAdmittedTurn, releaseAdmission], + ); + + // Subscribe to the fork's event stream + load its transcript. Called + // synchronously the moment the fork is committed, BEFORE the run starts, so + // no boundary request / complete can be missed (the stream has no replay). + const subscribeToFork = useCallback((forkId: string): Promise => { + let resolveReady!: () => void; + let rejectReady!: (error: unknown) => void; + let readySettled = false; + const ready = new Promise((resolve, reject) => { + resolveReady = () => { + if (readySettled) return; + readySettled = true; + resolve(); + }; + rejectReady = (error: unknown) => { + if (readySettled) return; + readySettled = true; + reject(error); + }; }); - }, [mountedRef, sideChat]); + // A subscription can fail before the first send. Keep that failure + // observable to a later send without creating an unhandled rejection now. + void ready.catch(() => undefined); + void sideChat.readSettledMessages(forkId) + .then(({ messages }) => { + if (mountedRef.current) { + setAllMessages((current) => mergeSettledMessages(current, messages)); + } + }) + .catch(() => { + if (mountedRef.current) setError(copyRef.current.errors.settlementFailed); + }); + const unsubscribe = sideChat.subscribeEvents( + forkId, + (event: SessionEvent) => { + if (!mountedRef.current) return; + const admission = pendingAdmissionRef.current; + if (event.type === 'error' && event.recoverable) { + if (admission) { + // Observation failure does not prove whether Host admitted the + // dispatched command. Keep its identity until Host events or the + // command result provide an authoritative outcome. + setError(copyRef.current.errors.sendFailed); + return; + } + setError(copyRef.current.errors.sendFailed); + const retry = Promise.reject(new Error(event.message)); + void retry.catch(() => undefined); + subscriptionReadyRef.current = retry; + return; + } + if (admission) { + if ( + event.type === 'message_admission' && + event.messageId === admission.messageId + ) { + admission.events.push(event); + resolveAdmission(forkId, admission, admission.messageId, true); + } else if (event.turnId === activeTurnIdRef.current) { + applyOwnedEvent(forkId, event); + } else { + admission.events.push(event); + } + return; + } + applyOwnedEvent(forkId, event); + }, + resolveReady, + rejectReady, + ); + let disposed = false; + unsubscribeRef.current = () => { + if (disposed) return; + disposed = true; + unsubscribe(); + // A send waiting for observation readiness must finish when the panel is + // disposed; its mounted check below then turns this into a clean no-op. + resolveReady(); + }; + return ready; + }, [applyOwnedEvent, mountedRef, resolveAdmission, sideChat]); const commitFork = useCallback( (session: SessionSummary) => { @@ -262,7 +424,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan companionIdRef.current = session.id; companionRef.current = session; setCompanion(session); - subscribeToFork(session.id); + subscriptionReadyRef.current = subscribeToFork(session.id); }, [subscribeToFork], ); @@ -406,21 +568,43 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan attachmentItems?: WorkbarIngestInput[], ): Promise => { const trimmed = text.trim(); - if (!mountedRef.current || !trimmed || turnInFlightRef.current || !sourceSession) { + if ( + !mountedRef.current || + !trimmed || + submitLockRef.current || + activeTurnIdRef.current || + pendingAdmissionRef.current || + !sourceSession + ) { return false; } - // Close the same-frame double-submit window before the first await. The - // visible in-flight state still begins only when the run is armed. - turnInFlightRef.current = true; + // Close the same-frame double-submit window before fork readiness can yield. + submitLockRef.current = true; setError(null); const turnId = crypto.randomUUID(); const quoteSnapshot = snapshotCompanionQuotes(panelId, pendingQuotes); const label = (quoteSnapshot.quotes[0]?.text ?? trimmed).slice(0, 24); const fork = await ensureFork(`${copyRef.current.namePrefix}${label}`); if (fork.status !== 'ready') { - turnInFlightRef.current = false; + submitLockRef.current = false; return false; } + try { + await subscriptionReadyRef.current; + } catch { + if (mountedRef.current) { + unsubscribeRef.current?.(); + subscriptionReadyRef.current = subscribeToFork(fork.session.id); + setError(copyRef.current.errors.sendFailed); + } + submitLockRef.current = false; + return false; + } + if (!mountedRef.current) { + submitLockRef.current = false; + return false; + } + let sendAdmission: PendingAdmission | undefined; const result = await performCompanionTurn({ api: sideChat, sourceSession, @@ -441,17 +625,37 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan onForkCommitted: () => {}, // Arm the optimistic live turn right before the send. onBeforeSend: () => { - stopRequestedRef.current = false; - activeTurnIdRef.current = turnId; - turnInFlightRef.current = true; - setTurnInFlight(true); + stopRequestRef.current = null; + const admission: PendingAdmission = { + messageId: turnId, + events: [], + }; + sendAdmission = admission; + setPendingAdmission(admission); + submitLockRef.current = false; setLiveTurn(armLiveTurn(turnId)); - ownTurnIdsRef.current.add(turnId); - setOwnTurnTick((tick) => tick + 1); }, onQuotesConsumed: () => onQuotesConsumed(quoteSnapshot), }); - if (result.status === 'sent') { + if (result.status === 'sent' || result.status === 'pending') { + const admission = sendAdmission; + if (!admission) return false; + if (result.status === 'pending') { + admission.consumeOnAdmission = () => onQuotesConsumed(quoteSnapshot); + const wasPending = pendingAdmissionRef.current === admission; + const outcome = resolveAdmission(result.forkId, admission, result.messageId); + if (outcome?.kind === 'admitted' && !wasPending) admission.consumeOnAdmission(); + if (outcome?.kind === 'retracted') { + return false; + } + } else if (result.steered) { + if (resolveAdmission(result.forkId, admission, result.messageId)?.kind === 'retracted') { + return false; + } + } else { + bindAdmittedTurn(result.forkId, result.turnId); + } + if ((await admission.stopPromise) === 'confirmed') return false; setHasContent(true); // Surface the just-sent user message immediately, and reflect any // automatic connection/model rebound in the read-only model label. @@ -485,12 +689,11 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan }; setError(byCode[result.code]); activeTurnIdRef.current = null; - turnInFlightRef.current = false; - setTurnInFlight(false); + if (sendAdmission) releaseAdmission(sendAdmission); setLiveTurn(undefined); } // 'disposed' → the panel unmounted mid-create; nothing to update. - turnInFlightRef.current = false; + submitLockRef.current = false; return false; }, [ @@ -501,36 +704,108 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan ensureFork, mountedRef, sideChat, + bindAdmittedTurn, + releaseAdmission, + resolveAdmission, + setPendingAdmission, ], ); const stop = useCallback(async (): Promise => { const id = companionIdRef.current; - if (!id) return; - stopRequestedRef.current = true; + if (!id || stopRequestRef.current) return; + const admission = pendingAdmissionRef.current; + if (admission) { + const stopPromise = sideChat.stop(id, admission.messageId).then( + (outcome) => { + if ( + outcome?.kind === 'retracted' && + outcome.messageId === admission.messageId + ) { + releaseAdmission(admission); + } + return 'confirmed' as const; + }, + () => { + // A rejected Stop tells us nothing about whether the Host stopped + // the Turn. Keep the admission alive so a late Host outcome can + // still bind its own Turn; the user can retry Stop after this. + if (pendingAdmissionRef.current === admission) { + admission.stopPromise = undefined; + resolveAdmission(id, admission, admission.messageId, true); + } + if (stopRequestRef.current === stopPromise) stopRequestRef.current = null; + return 'unknown' as const; + }, + ); + admission.stopPromise = stopPromise; + stopRequestRef.current = stopPromise; + await stopPromise; + return; + } + const stopPromise = sideChat.stop(id, activeTurnIdRef.current ?? undefined); + stopRequestRef.current = stopPromise; try { - await sideChat.stop(id); + await stopPromise; } catch { - stopRequestedRef.current = false; + if (stopRequestRef.current === stopPromise) stopRequestRef.current = null; // best-effort; the terminal event still reconciles state } - }, [sideChat]); + }, [releaseAdmission, resolveAdmission, sideChat]); const steer = useCallback(async (text: string): Promise => { const id = companionIdRef.current; const trimmed = text.trim(); - if (!mountedRef.current || !id || !trimmed || !turnInFlight) return false; + if ( + !mountedRef.current || + !id || + !trimmed || + !turnInFlight || + pendingAdmissionRef.current + ) { + return false; + } + const admissionId = crypto.randomUUID(); + const admission: PendingAdmission = { + messageId: admissionId, + events: [], + }; + setPendingAdmission(admission); try { - const outcome = await sideChat.steer(id, trimmed); + const outcome = await sideChat.steer(id, trimmed, admissionId); if (!mountedRef.current) return false; - if (outcome.kind !== 'queued') return false; + if ((await admission.stopPromise) === 'confirmed') return false; + if (admissionOutcomeForMessage(admission.events, admission.messageId)?.kind === 'retracted') { + return false; + } + if (outcome.kind === 'started') { + bindAdmittedTurn(id, outcome.turnId, { preserveLiveTurn: true }); + } else if (resolveAdmission(id, admission, outcome.messageId, true)?.kind === 'retracted') { + return false; + } setError(null); return true; } catch { - if (mountedRef.current) setError(copyRef.current.errors.sendFailed); + if (mountedRef.current) { + if (pendingAdmissionRef.current === admission) { + releaseAdmission(admission, copyRef.current.errors.sendFailed); + } else if ( + admissionOutcomeForMessage(admission.events, admission.messageId)?.kind !== 'retracted' + ) { + setError(copyRef.current.errors.sendFailed); + } + } return false; } - }, [mountedRef, sideChat, turnInFlight]); + }, [ + bindAdmittedTurn, + mountedRef, + releaseAdmission, + resolveAdmission, + setPendingAdmission, + sideChat, + turnInFlight, + ]); const setPermissionMode = useCallback( async (mode: PermissionMode): Promise => { @@ -560,10 +835,8 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan if (!id || turnInFlight || regeneratePendingTurnId) return false; setRegeneratePendingTurnId(turnId); const regenerationTurnId = crypto.randomUUID(); - stopRequestedRef.current = false; + stopRequestRef.current = null; activeTurnIdRef.current = regenerationTurnId; - turnInFlightRef.current = true; - setTurnInFlight(true); setError(null); setLiveTurn(armLiveTurn(regenerationTurnId)); ownTurnIdsRef.current.add(regenerationTurnId); @@ -577,8 +850,6 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan } catch { if (mountedRef.current) { activeTurnIdRef.current = null; - turnInFlightRef.current = false; - setTurnInFlight(false); setLiveTurn(undefined); setError(copyRef.current.errors.sendFailed); } @@ -621,7 +892,6 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const messages = allMessages.filter( (message) => message.turnId !== undefined && ownTurnIdsRef.current.has(message.turnId), ); - const { streaming, processing } = deriveCompanionComposerState(turnInFlight, liveTurn); // Inherited model (read-only): the fork's once created, else the source's. const activeModel = companion ? { llmConnectionSlug: companion.llmConnectionSlug, model: companion.model } diff --git a/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts b/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts index 86c9d0ef00..b3aebaa499 100644 --- a/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts +++ b/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts @@ -120,8 +120,12 @@ export function createDesktopWorkbarServices( abandonSessionCopy: (sourceSessionId, copyId) => bridge.sessions.abandonSessionCopy(sourceSessionId, copyId), send: (sessionId, command) => bridge.sessions.send(sessionId, command), - stop: (sessionId) => bridge.sessions.stop(sessionId), - steer: (sessionId, text) => bridge.sessions.steer(sessionId, text), + stop: (sessionId, admissionId) => + bridge.sessions.stop( + sessionId, + admissionId ? { source: 'stop_button', expectedAdmissionId: admissionId } : undefined, + ), + steer: (sessionId, text, admissionId) => bridge.sessions.steer(sessionId, text, admissionId), setPermissionMode: (sessionId, mode) => bridge.sessions.setPermissionMode(sessionId, mode), regenerateTurn: (sessionId, input) => @@ -130,8 +134,8 @@ export function createDesktopWorkbarServices( bridge.sessions.respondToSandboxBoundary(sessionId, response), respondToUserQuestion: (sessionId, response) => bridge.sessions.respondToUserQuestion(sessionId, response), - subscribeEvents: (sessionId, handler) => - bridge.sessions.subscribeEvents(sessionId, handler), + subscribeEvents: (sessionId, handler, onSeeded, onSeedError) => + bridge.sessions.subscribeEvents(sessionId, handler, onSeeded, undefined, onSeedError), subscribeSessionChanges: (handler) => bridge.sessions.subscribeChanges(handler), }, }; diff --git a/apps/desktop/src/renderer/workhub-session-port.ts b/apps/desktop/src/renderer/workhub-session-port.ts index 4d66344637..35558b5fab 100644 --- a/apps/desktop/src/renderer/workhub-session-port.ts +++ b/apps/desktop/src/renderer/workhub-session-port.ts @@ -68,7 +68,7 @@ export interface WorkHubDesktopSessionBridge { stop( sessionId: string, input?: { source?: 'stop_button'; expectedTurnId?: string }, - ): Promise; + ): Promise; subscribeChanges(handler: () => void): () => void; } diff --git a/apps/desktop/stories/session-workbar.stories.tsx b/apps/desktop/stories/session-workbar.stories.tsx index 600fd6ebfa..8d80ea4879 100644 --- a/apps/desktop/stories/session-workbar.stories.tsx +++ b/apps/desktop/stories/session-workbar.stories.tsx @@ -709,9 +709,9 @@ function bridge(options: { branchFromTurn: async () => ({ ok: true, session: SIDE_CHAT_SESSION }), cleanupSessionCopy: async () => undefined, abandonSessionCopy: async () => undefined, - send: async () => ({ ok: true }), + send: async () => ({ ok: true, turnId: 'story-side-chat-turn' }), stop: async () => undefined, - steer: async () => ({ kind: 'queued' }), + steer: async () => ({ kind: 'started', turnId: 'story-side-chat-turn' }), setPermissionMode: async (_sessionId, mode) => ({ ...SIDE_CHAT_SESSION, permissionMode: mode, @@ -719,7 +719,10 @@ function bridge(options: { regenerateTurn: async () => undefined, respondToSandboxBoundary: async () => undefined, respondToUserQuestion: async () => undefined, - subscribeEvents: unsubscribe, + subscribeEvents: (_sessionId, _handler, onSeeded) => { + onSeeded?.(); + return unsubscribe(); + }, subscribeSessionChanges: unsubscribe, }, }); diff --git a/packages/cli/src/__tests__/runtime-host-run-command.test.ts b/packages/cli/src/__tests__/runtime-host-run-command.test.ts index 94165620fc..3ba1fa5910 100644 --- a/packages/cli/src/__tests__/runtime-host-run-command.test.ts +++ b/packages/cli/src/__tests__/runtime-host-run-command.test.ts @@ -1554,6 +1554,7 @@ function continuitySnapshot( queue: { hostEpoch: 'host-1', queueRevision: 0, steering: [], followup: [] }, interactions: { pending: [] }, ...overrides, + rootTurnSourceMessageIds: overrides.rootTurnSourceMessageIds ?? [], }; } 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..37b5873595 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1824,6 +1824,7 @@ function continuitySnapshot( queue: { hostEpoch: 'host-1', queueRevision: 0, steering: [], followup: [] }, interactions: { pending: [] }, ...overrides, + rootTurnSourceMessageIds: overrides.rootTurnSourceMessageIds ?? [], }; } diff --git a/packages/core/src/backend-types.ts b/packages/core/src/backend-types.ts index 8990812e8b..46262ca9f2 100644 --- a/packages/core/src/backend-types.ts +++ b/packages/core/src/backend-types.ts @@ -191,6 +191,7 @@ export type BackendSessionEvent = Exclude< { type: | 'queue_update' + | 'message_admission' | 'permission_request' | 'permission_answer_ack' | 'permission_closure_ack' diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts index 4a58287e49..bea135e5fd 100644 --- a/packages/core/src/events.ts +++ b/packages/core/src/events.ts @@ -486,6 +486,7 @@ export type SessionEvent = | PlanSubmittedEvent | TokenUsageEvent | SteeringMessageEvent + | MessageAdmissionEvent | QueueUpdateEvent | ProviderRetryEvent | ErrorEvent @@ -1064,6 +1065,18 @@ export interface SteeringMessageEvent extends BaseEvent { submittedContentDigest?: `sha256:${string}`; } +/** + * Transient Host projection fact: a submitted message now belongs to this + * Turn. It is emitted by the session projector, not by a backend or durable + * event ledger, so a client can bind a queued admission without guessing from + * timing or Turn ids returned by a stale command response. + */ +export interface MessageAdmissionEvent extends BaseEvent { + type: 'message_admission'; + messageId: string; + outcome: 'admitted' | 'retracted'; +} + /** * Result of enqueuing a steering / followup message. `fallback` means there was * no active run to attach to (the turn just ended) and the caller should open a diff --git a/packages/runtime-host/src/__tests__/agent-graph-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/agent-graph-two-client-uds.test.ts index e05eb35488..44c1343bbe 100644 --- a/packages/runtime-host/src/__tests__/agent-graph-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/agent-graph-two-client-uds.test.ts @@ -365,6 +365,7 @@ function canonical(hostEpoch: string): CanonicalSessionProjection { goal: null, queue: { hostEpoch, queueRevision: 0, steering: [], followup: [] }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }; } diff --git a/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts b/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts index bb7a568554..742d9cd9d1 100644 --- a/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts +++ b/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts @@ -74,6 +74,7 @@ test('projects the canonical root lifecycle and the attachment queue from real S goal: null, queue: { hostEpoch: 'epoch-1', queueRevision: 0, steering: [], followup: [] }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }); const admitted = await rootAdmissions.admitRootTurn({ @@ -154,6 +155,51 @@ test('projects the canonical root lifecycle and the attachment queue from real S }); }); +test('projects root admission source message tickets into continuity', async () => { + await withStores(async (root, stores) => { + const session = await stores.sessionStore.create(sessionInput(root)); + const rootAdmissions = new RootAdmissionOwner(stores.agentRunStore); + await rootAdmissions.recoverSession(session.id); + const messages = createMessages(session.id, stores); + const reader = new CanonicalSessionProjectionReader({ + stores, + rootAdmissions, + messages, + }); + await rootAdmissions.admitRootTurn({ + sessionId: session.id, + turnId: 'turn-successor', + proposedRunId: 'run-successor', + proposedUserMessageId: 'source-1', + execution: { kind: 'external_message' }, + normalizedInput: { text: 'first\n\nsecond' }, + sourceMessages: [ + { + messageId: 'source-1', + content: { text: 'first' }, + placement: 'current_turn', + disposition: 'steering', + }, + { + messageId: 'source-2', + content: { text: 'second' }, + placement: 'next_turn', + disposition: 'followup', + }, + ], + admittedAt: 10, + }); + + const canonical = await reader.read(session.id); + assert.deepEqual(canonical?.rootTurnSourceMessageIds, ['source-1', 'source-2']); + assert.deepEqual(createSessionContinuitySnapshot(canonical!, 1).rootTurnSourceMessageIds, [ + 'source-1', + 'source-2', + ]); + await messages.close(); + }); +}); + test('projects pending Interactions and preflights their combined snapshot capacity', async () => { await withStores(async (root, stores) => { const session = await stores.sessionStore.create(sessionInput(root)); diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index a80204812e..48020f9895 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -1436,6 +1436,7 @@ function canonicalProjection(sessionId: string): CanonicalSessionProjection { followup: [], }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }; } diff --git a/packages/runtime-host/src/__tests__/goal-protocol.test.ts b/packages/runtime-host/src/__tests__/goal-protocol.test.ts index f922dad0ce..519a7918b2 100644 --- a/packages/runtime-host/src/__tests__/goal-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/goal-protocol.test.ts @@ -124,6 +124,7 @@ test('Goal projection is part of the exact Session continuity schema', () => { goal, queue: { hostEpoch: 'epoch-1', queueRevision: 0, steering: [], followup: [] }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }); assert.deepEqual(snapshot.goal, goal); assert.throws(() => diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index c4f2fad83b..d5c8102105 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -295,6 +295,9 @@ describe('Runtime Host bootstrap protocol', () => { }), isInvalidFrame, ); + const { rootTurnSourceMessageIds: _missing, ...missingAdmissionTickets } = + continuitySnapshot('epoch-1'); + assert.throws(() => decodeSessionContinuitySnapshot(missingAdmissionTickets), isInvalidFrame); const waiting = { ...continuitySnapshot('epoch-1'), rootTurn: { @@ -1796,5 +1799,6 @@ function continuitySnapshot(hostEpoch: string) { followup: [], }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }; } diff --git a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts index 4ca2ec4d36..c3921be3a7 100644 --- a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts @@ -2078,6 +2078,7 @@ function canonical( followup: [], }, interactions: overrides.interactions ?? { pending: [] }, + rootTurnSourceMessageIds: [], }; } diff --git a/packages/runtime-host/src/__tests__/session-projector.test.ts b/packages/runtime-host/src/__tests__/session-projector.test.ts index e2138d3f78..492b0967ee 100644 --- a/packages/runtime-host/src/__tests__/session-projector.test.ts +++ b/packages/runtime-host/src/__tests__/session-projector.test.ts @@ -19,6 +19,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; +import type { SessionEvent } from '@maka/core/events'; import type { StoredMessage } from '@maka/core/session'; import { createRuntimeHostSessionProjectionSeed, @@ -72,6 +73,278 @@ test('applies authoritative replacement once and does not complete it again at T ); }); +test('emits the Host admission fact when a queued message enters a successor Turn', () => { + const ordinary = new RuntimeHostSessionProjector( + withRootSourceMessageIds(snapshot(), ['ordinary-ticket']), + createRuntimeHostSessionProjectionSeed([], snapshot()), + () => 10, + ); + assert.equal( + ordinary.seedActive(false).some((event) => event.type === 'message_admission'), + false, + ); + + const rejoined = new RuntimeHostSessionProjector( + withRootSourceMessageIds(snapshot(), ['rejoined-ticket']), + createRuntimeHostSessionProjectionSeed([], snapshot()), + () => 10, + [], + true, + ); + assert.deepEqual( + rejoined + .seedActive(false) + .filter( + (event): event is Extract => + event.type === 'message_admission' && event.outcome === 'admitted', + ) + .map((event) => ({ + turnId: event.turnId, + messageId: event.messageId, + })), + [{ turnId: 'turn-1', messageId: 'rejoined-ticket' }], + ); + + const previous = snapshot({ + queue: { + hostEpoch: 'host-1', + queueRevision: 1, + steering: [ + { + entryId: 'entry-1', + messageId: 'ticket-1', + content: { text: 'continue in successor' }, + placement: 'current_turn', + state: 'queued', + }, + ], + followup: [], + }, + }); + const projector = new RuntimeHostSessionProjector( + previous, + createRuntimeHostSessionProjectionSeed([], previous), + () => 10, + [], + true, + ); + const next = withRootSourceMessageIds( + snapshot({ + projectionRevision: 2, + queue: { + hostEpoch: 'host-1', + queueRevision: 2, + steering: [], + followup: [], + }, + rootTurn: { + sessionId: 'session-1', + turnId: 'turn-2', + runId: 'run-2', + status: 'running', + }, + }), + ['ticket-1'], + ); + + const events = projector.accept({ + kind: 'subscription.session_projection', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 1, + snapshot: next, + }).events; + + assert.deepEqual( + events + .filter((event) => event.type === 'message_admission' || event.type === 'queue_update') + .map((event) => event.type), + ['message_admission', 'queue_update'], + ); + + assert.deepEqual( + events + .filter( + (event): event is Extract => + event.type === 'message_admission' && event.outcome === 'admitted', + ) + .map((event) => ({ + turnId: event.turnId, + messageId: event.messageId, + })), + [{ turnId: 'turn-2', messageId: 'ticket-1' }], + ); + + const retracted = new RuntimeHostSessionProjector( + previous, + createRuntimeHostSessionProjectionSeed([], previous), + () => 10, + [], + true, + ).accept({ + kind: 'subscription.session_projection', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 1, + snapshot: snapshot({ + projectionRevision: 2, + queue: next.queue, + }), + }).events; + assert.deepEqual( + retracted + .filter( + (event): event is Extract => + event.type === 'message_admission' && event.outcome === 'retracted', + ) + .map((event) => event.messageId), + ['ticket-1'], + ); +}); + +test('emits the Host admission fact when a queued message enters the active Turn', () => { + const previous = snapshot({ + queue: { + hostEpoch: 'host-1', + queueRevision: 1, + steering: [ + { + entryId: 'entry-1', + messageId: 'ticket-1', + content: { text: 'continue here' }, + placement: 'current_turn', + state: 'queued', + }, + ], + followup: [], + }, + }); + const projector = new RuntimeHostSessionProjector( + previous, + createRuntimeHostSessionProjectionSeed([], previous), + () => 10, + [], + true, + ); + + const events = projector.accept({ + kind: 'subscription.session_projection', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 1, + snapshot: snapshot({ + projectionRevision: 2, + queue: { + hostEpoch: 'host-1', + queueRevision: 2, + steering: [ + { + entryId: 'entry-1', + messageId: 'ticket-1', + content: { text: 'continue here' }, + placement: 'current_turn', + state: 'in_flight', + }, + ], + followup: [], + }, + }), + }).events; + + assert.deepEqual( + events + .filter( + (event): event is Extract => + event.type === 'message_admission', + ) + .map((event) => ({ + outcome: event.outcome, + turnId: event.turnId, + messageId: event.messageId, + })), + [{ outcome: 'admitted', turnId: 'turn-1', messageId: 'ticket-1' }], + ); +}); + +test('reseeds the Host admission fact for a message already in the active Turn', () => { + const current = snapshot({ + queue: { + hostEpoch: 'host-1', + queueRevision: 2, + steering: [ + { + entryId: 'entry-1', + messageId: 'ticket-1', + content: { text: 'continue here' }, + placement: 'current_turn', + state: 'in_flight', + }, + ], + followup: [], + }, + }); + const projector = new RuntimeHostSessionProjector( + current, + createRuntimeHostSessionProjectionSeed([], current), + () => 10, + [], + true, + ); + + assert.deepEqual( + projector + .seedActive(false) + .filter( + (event): event is Extract => + event.type === 'message_admission', + ) + .map((event) => ({ + outcome: event.outcome, + turnId: event.turnId, + messageId: event.messageId, + })), + [{ outcome: 'admitted', turnId: 'turn-1', messageId: 'ticket-1' }], + ); +}); + +test('reseeds the Host admission fact after an active Turn message leaves the queue', () => { + const current = snapshot(); + const projector = new RuntimeHostSessionProjector( + current, + createRuntimeHostSessionProjectionSeed( + [ + { + type: 'user', + id: 'ticket-1', + turnId: 'turn-1', + ts: 1, + text: 'continue here', + steeringEventId: 'steering-event-1', + }, + ], + current, + ), + () => 10, + [], + true, + ); + + assert.deepEqual( + projector + .seedActive(false) + .filter( + (event): event is Extract => + event.type === 'message_admission', + ) + .map((event) => ({ + outcome: event.outcome, + turnId: event.turnId, + messageId: event.messageId, + })), + [{ outcome: 'admitted', turnId: 'turn-1', messageId: 'ticket-1' }], + ); +}); + test('reseeds the latest provider retry when the active Turn still carries one', () => { const retry = { phase: 'scheduled' as const, @@ -379,10 +652,18 @@ function snapshot(overrides: Partial = {}): SessionCo followup: [], }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], ...overrides, }; } +function withRootSourceMessageIds( + value: SessionContinuitySnapshot, + rootTurnSourceMessageIds: readonly string[], +): SessionContinuitySnapshot { + return { ...value, rootTurnSourceMessageIds }; +} + function assistant(id: string, text: string): Extract { return { type: 'assistant', 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..40c0ef67f1 100644 --- a/packages/runtime-host/src/__tests__/session-subscription-client.test.ts +++ b/packages/runtime-host/src/__tests__/session-subscription-client.test.ts @@ -1255,6 +1255,7 @@ function openResult( goal: null, queue: { hostEpoch, queueRevision: 1, steering: [], followup: [] }, interactions: { pending: [] }, + rootTurnSourceMessageIds: [], }, }; } diff --git a/packages/runtime-host/src/adapter/session-projector.ts b/packages/runtime-host/src/adapter/session-projector.ts index 09760af34a..bb1618e4e2 100644 --- a/packages/runtime-host/src/adapter/session-projector.ts +++ b/packages/runtime-host/src/adapter/session-projector.ts @@ -42,7 +42,10 @@ interface AssistantAccumulator { } export interface RuntimeHostSessionProjectionSeed { - readonly durableInFlightMessageIds: readonly string[]; + readonly durableSteeringMessages: readonly { + readonly messageId: string; + readonly turnId: string; + }[]; readonly activeAssistantMessages: readonly Extract[]; } @@ -50,13 +53,13 @@ export function createRuntimeHostSessionProjectionSeed( transcript: readonly StoredMessage[], snapshot: SessionContinuitySnapshot, ): RuntimeHostSessionProjectionSeed { - const inFlightMessageIds = new Set( - rootQueueInFlight(snapshot.queue).map((entry) => entry.messageId), - ); return { - durableInFlightMessageIds: transcript - .filter((message) => inFlightMessageIds.has(message.id)) - .map((message) => message.id), + durableSteeringMessages: transcript + .filter( + (message): message is Extract => + message.type === 'user' && message.steeringEventId !== undefined, + ) + .map((message) => ({ messageId: message.id, turnId: message.turnId })), activeAssistantMessages: snapshot.rootTurn === null ? [] @@ -84,17 +87,24 @@ export class RuntimeHostSessionProjector { #snapshot: SessionContinuitySnapshot; readonly #now: () => number; readonly #transcriptIds: Set; + readonly #durableSteeringTurnByMessage: ReadonlyMap; readonly #accumulators = new Map(); + #projectMessageAdmissions: boolean; constructor( snapshot: SessionContinuitySnapshot, seed: RuntimeHostSessionProjectionSeed, now: () => number = Date.now, activeAssistantStreams: readonly SessionAssistantStreamIdentity[] = [], + projectMessageAdmissions = false, ) { this.#snapshot = structuredClone(snapshot); this.#now = now; - this.#transcriptIds = new Set(seed.durableInFlightMessageIds); + this.#durableSteeringTurnByMessage = new Map( + seed.durableSteeringMessages.map(({ messageId, turnId }) => [messageId, turnId]), + ); + this.#transcriptIds = new Set(this.#durableSteeringTurnByMessage.keys()); + this.#projectMessageAdmissions = projectMessageAdmissions; const root = snapshot.rootTurn; if (!root) return; for (const message of seed.activeAssistantMessages) { @@ -139,10 +149,32 @@ export class RuntimeHostSessionProjector { return structuredClone(this.#snapshot); } + enableMessageAdmissions(): void { + this.#projectMessageAdmissions = true; + } + seedActive(includeAssistantText: boolean): SessionEvent[] { const root = this.#snapshot.rootTurn; - if (!root || isRuntimeHostTerminalTurn(root)) return []; + if (!root) return []; const events: SessionEvent[] = []; + if (this.#projectMessageAdmissions) { + events.push( + ...projectMessageAdmissionEvents( + root, + [ + ...new Set([ + ...this.#snapshot.rootTurnSourceMessageIds, + ...rootQueueInFlight(this.#snapshot.queue).map((entry) => entry.messageId), + ...[...this.#durableSteeringTurnByMessage] + .filter(([, turnId]) => turnId === root.turnId) + .map(([messageId]) => messageId), + ]), + ], + this.#now(), + ), + ); + } + if (isRuntimeHostTerminalTurn(root)) return events; let seededAssistantText = false; if (includeAssistantText) { for (const accumulator of this.#accumulators.values()) { @@ -192,7 +224,12 @@ export class RuntimeHostSessionProjector { } seedTerminal(turn: RuntimeHostTerminalTurn): SessionEvent[] { - return this.#terminalEvents(turn, true); + return [ + ...(this.#projectMessageAdmissions + ? projectMessageAdmissionEvents(turn, this.#snapshot.rootTurnSourceMessageIds, this.#now()) + : []), + ...this.#terminalEvents(turn, true), + ]; } seedStoredTerminal(turnId: string, transcript: readonly StoredMessage[]): SessionEvent[] { @@ -356,8 +393,16 @@ export class RuntimeHostSessionProjector { events.push(...projectRuntimeHostInteractionRequest(interaction, this.#now())); } const root = next.rootTurn; + const enteredActiveTurn = + root && queueChanged(previousSnapshot.queue, next.queue) + ? newlyInFlight(previousSnapshot.queue, next.queue) + : []; + if (this.#projectMessageAdmissions) { + events.push(...projectMessageRetractionEvents(previousSnapshot, next, this.#now())); + events.push(...projectNewMessageAdmissionEvents(previousSnapshot, next, this.#now())); + } if (root && queueChanged(previousSnapshot.queue, next.queue)) { - for (const entry of newlyInFlight(previousSnapshot.queue, next.queue)) { + for (const entry of enteredActiveTurn) { events.push({ type: 'steering_message', id: `host-queue:${next.queue.hostEpoch}:${next.queue.queueRevision}:${entry.entryId}`, @@ -437,6 +482,67 @@ function emptyUpdate(events: readonly SessionEvent[]): RuntimeHostProjectionUpda return { events, resolvedInteractions: [] }; } +function projectMessageAdmissionEvents( + root: TurnSnapshot, + messageIds: readonly string[], + ts: number, +): SessionEvent[] { + return messageIds.map((messageId) => ({ + type: 'message_admission' as const, + id: `host-admission:${root.runId}:${messageId}`, + turnId: root.turnId, + ts, + messageId, + outcome: 'admitted' as const, + })); +} + +function projectNewMessageAdmissionEvents( + previous: SessionContinuitySnapshot, + next: SessionContinuitySnapshot, + ts: number, +): SessionEvent[] { + const root = next.rootTurn; + if (!root) return []; + const previousIds = + previous.rootTurn?.runId === root.runId + ? new Set(previous.rootTurnSourceMessageIds) + : new Set(); + const messageIds = new Set( + next.rootTurnSourceMessageIds.filter((messageId) => !previousIds.has(messageId)), + ); + for (const entry of newlyInFlight(previous.queue, next.queue)) messageIds.add(entry.messageId); + return projectMessageAdmissionEvents(root, [...messageIds], ts); +} + +function projectMessageRetractionEvents( + previous: SessionContinuitySnapshot, + next: SessionContinuitySnapshot, + ts: number, +): SessionEvent[] { + const root = next.rootTurn ?? previous.rootTurn; + if (!root || previous.queue.hostEpoch !== next.queue.hostEpoch) return []; + const retained = new Set( + [...next.queue.steering, ...next.queue.followup].map((entry) => entry.messageId), + ); + const admitted = new Set(next.rootTurnSourceMessageIds); + return [...previous.queue.steering, ...previous.queue.followup] + .filter( + (entry) => + entry.state === 'queued' && + !retained.has(entry.messageId) && + !admitted.has(entry.messageId), + ) + .map((entry) => ({ + type: 'message_admission' as const, + id: `host-retraction:${next.queue.hostEpoch}:${next.queue.queueRevision}:${entry.messageId}`, + turnId: root.turnId, + ts, + messageId: entry.messageId, + outcome: 'retracted' as const, + })); +} + export function projectRuntimeHostInteractionRequest( interaction: InteractionPendingSnapshot, now: number, diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 19d5dd1f5b..327dbfa347 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -91,7 +91,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 48 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 49 as const; +// 49: Session continuity snapshots carry Host-owned root source message +// tickets so clients can bind queued admissions to successor Turns. // 48: Session branch creation accepts an explicit Side Conversation intent. // Older peers reject the strict input shape or cannot apply its snapshot semantics. // 47: Project registration can carry an explicit location preference. Epoch-46 diff --git a/packages/runtime-host/src/protocol/session-continuity.ts b/packages/runtime-host/src/protocol/session-continuity.ts index c8ccb5055a..4a50f11f19 100644 --- a/packages/runtime-host/src/protocol/session-continuity.ts +++ b/packages/runtime-host/src/protocol/session-continuity.ts @@ -87,6 +87,8 @@ export interface SessionContinuitySnapshot { goal: GoalProjection | null; queue: SessionMessageQueueProjection; interactions: SessionInteractionProjection; + /** Host-owned source message tickets admitted into the root Turn. */ + rootTurnSourceMessageIds: readonly string[]; } export interface SubscriptionOpenInput { @@ -515,7 +517,8 @@ export function decodeSessionContinuitySnapshot(value: unknown): SessionContinui 'Session continuity snapshot', SESSION_CONTINUITY_SNAPSHOT_MAX_BYTES, ); - const record = requireExactRecord(value, 'Session continuity snapshot', [ + const record = requireRecord(value, 'Session continuity snapshot'); + assertAllowedKeys(record, 'Session continuity snapshot', [ 'schemaVersion', 'session', 'projectionRevision', @@ -523,6 +526,17 @@ export function decodeSessionContinuitySnapshot(value: unknown): SessionContinui 'goal', 'queue', 'interactions', + 'rootTurnSourceMessageIds', + ]); + assertRequiredKeys(record, 'Session continuity snapshot', [ + 'schemaVersion', + 'session', + 'projectionRevision', + 'rootTurn', + 'goal', + 'queue', + 'interactions', + 'rootTurnSourceMessageIds', ]); if (record.schemaVersion !== SESSION_CONTINUITY_SCHEMA_VERSION) { throw invalidProtocolFrame('Unsupported Session continuity snapshot schema'); @@ -545,9 +559,17 @@ export function decodeSessionContinuitySnapshot(value: unknown): SessionContinui goal, queue: decodeSessionMessageQueueProjection(record.queue), interactions, + rootTurnSourceMessageIds: decodeRootTurnSourceMessageIds(record.rootTurnSourceMessageIds), }; } +function decodeRootTurnSourceMessageIds(value: unknown): string[] { + if (!Array.isArray(value)) throw invalidProtocolFrame('Invalid root Turn source message ids'); + return value.map((messageId, index) => + requireId(messageId, `root Turn source message id ${index}`), + ); +} + function decodeSubscriptionOpenInput(value: unknown): SubscriptionOpenInput { const record = requireExactRecord(value, 'subscription.open input', ['sessionId', 'transcript']); const transcript = requireRecord(record.transcript, 'subscription transcript policy'); diff --git a/packages/runtime-host/src/server/canonical-session-projection.ts b/packages/runtime-host/src/server/canonical-session-projection.ts index 2594acca4c..76cabef6ee 100644 --- a/packages/runtime-host/src/server/canonical-session-projection.ts +++ b/packages/runtime-host/src/server/canonical-session-projection.ts @@ -54,6 +54,7 @@ export interface CanonicalSessionProjection { readonly goal: GoalProjection | null; readonly queue: SessionMessageQueueProjection; readonly interactions: SessionInteractionProjection; + readonly rootTurnSourceMessageIds: readonly string[]; } export interface CanonicalSessionProjectionCandidate { @@ -98,6 +99,7 @@ export class CanonicalSessionProjectionReader { } let rootTurn: TurnSnapshot | null = null; + let rootTurnSourceMessageIds: readonly string[] = []; if (admission) { const durableAdmission = await this.#stores.agentRunStore.readRootTurnAdmission( sessionId, @@ -108,6 +110,7 @@ export class CanonicalSessionProjectionReader { } this.#rootAdmissions.assertKnownAdmission(durableAdmission); rootTurn = await readCanonicalTurnSnapshot(this.#stores, durableAdmission); + rootTurnSourceMessageIds = durableAdmission.sourceMessages.map(({ messageId }) => messageId); } const interactions = projectSessionInteractions( @@ -124,7 +127,14 @@ export class CanonicalSessionProjectionReader { createdAt: header.createdAt, isArchived: header.isArchived, }; - return { session, rootTurn, goal, queue, interactions }; + return { + session, + rootTurn, + goal, + queue, + interactions, + rootTurnSourceMessageIds, + }; } async fitsCandidate( @@ -192,6 +202,7 @@ function sessionContinuitySnapshotInput( goal: canonical.goal, queue: canonical.queue, interactions: canonical.interactions, + rootTurnSourceMessageIds: canonical.rootTurnSourceMessageIds, }; } diff --git a/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts b/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts index bef67c4f58..045f94d21b 100644 --- a/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts +++ b/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts @@ -624,6 +624,24 @@ const projectionRunHeader: AgentRunHeader = { }; describe('SessionEvent projection coverage', () => { + test('keeps Host admission facts out of durable Runtime events', () => { + assert.throws( + () => + mapSessionEventToRuntimeEvent( + { + type: 'message_admission', + id: 'message-admission-1', + turnId: 'turn-1', + ts: 1, + messageId: 'message-1', + outcome: 'admitted', + }, + ctx, + ), + /message_admission is not a backend event/, + ); + }); + // The contract is over what a reader can actually meet: every mapped event // AgentRun admits to the ledger has to project. It asserts on the unclaimed // codes at either severity, not on the hard one alone — a control fact whose diff --git a/packages/runtime/src/session-event-runtime-mapper.ts b/packages/runtime/src/session-event-runtime-mapper.ts index e282d2e69e..3044d110f9 100644 --- a/packages/runtime/src/session-event-runtime-mapper.ts +++ b/packages/runtime/src/session-event-runtime-mapper.ts @@ -129,12 +129,10 @@ export function mapSessionEventToRuntimeEvent( ctx: RuntimeEventMapContext, memory: SessionEventMapMemory = createSessionEventMapMemory(), ): RuntimeEvent { - if (event.type === 'queue_update') { - // Not backend-mappable by design: the kernel is queue_update's only - // legal producer and pushes it directly into the turn stream. The flow - // drops a backend-yielded one at the ingress (see run()), so reaching - // this line means a caller bypassed that authority boundary. - throw new Error('queue_update is not a backend event: the kernel is its only legal producer'); + if (event.type === 'queue_update' || event.type === 'message_admission') { + // These are Host/kernel projection facts, not backend events. The live + // ingress drops them, so reaching this line bypassed that authority boundary. + throw new Error(`${event.type} is not a backend event`); } if (isLegacyPermissionSessionEvent(event)) { throw new Error(`${event.type} is a legacy permission event and is not backend-mappable`); @@ -144,7 +142,11 @@ export function mapSessionEventToRuntimeEvent( } export function isLiveBackendSessionEvent(event: SessionEvent): event is BackendSessionEvent { - return event.type !== 'queue_update' && !isLegacyPermissionSessionEvent(event); + return ( + event.type !== 'queue_update' && + event.type !== 'message_admission' && + !isLegacyPermissionSessionEvent(event) + ); } function isLegacyPermissionSessionEvent(event: SessionEvent): event is Extract<