diff --git a/apps/desktop/renderer-architecture.json b/apps/desktop/renderer-architecture.json index 51b23b7387..7c1619c81b 100644 --- a/apps/desktop/renderer-architecture.json +++ b/apps/desktop/renderer-architecture.json @@ -216,7 +216,6 @@ "src/renderer/theme.ts", "src/renderer/titlebar-dim-color.ts", "src/renderer/titlebar-modal-sync.ts", - "src/renderer/transient-message-projection.ts", "src/renderer/turn-footer-actions.ts", "src/renderer/use-active-execution-boundary.ts", "src/renderer/use-app-shell-composer-quotes.ts", @@ -611,12 +610,13 @@ "createAppShellSessionEventHandlers" ], "dependencyPaths": { + "./application/contracts/message-queue-projection.js": 1, "./locales/conversation-copy.js": 1, "./model-connection-errors.js": 1, "@maka/ui": 1 }, "importSpecifiers": 11, - "nonTriviaTokens": 2974 + "nonTriviaTokens": 2771 }, "src/renderer/app-shell-session-start-actions.ts": { "importDeclarations": 2, @@ -2272,9 +2272,8 @@ "unresolvedDependencies": 0, "actionFactories": [], "dependencyPaths": { - "./new-task-reload-intent.js": 1, - "./transient-message-projection.js": 1, - "@maka/runtime-host/protocol": 1 + "./application/contracts/transient-message-projection.js": 1, + "./new-task-reload-intent.js": 1 } }, "src/renderer/session-workspace-errors.ts": { @@ -4128,15 +4127,6 @@ "./theme": 1 } }, - "src/renderer/transient-message-projection.ts": { - "bridgePaths": {}, - "environmentCapabilities": {}, - "hookCalls": {}, - "lifecycleMethods": {}, - "unresolvedDependencies": 0, - "actionFactories": [], - "dependencyPaths": {} - }, "src/renderer/turn-footer-actions.ts": { "bridgePaths": {}, "environmentCapabilities": {}, 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 159cbb3311..42364188d5 100644 --- a/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts +++ b/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts @@ -29,6 +29,7 @@ import type { PermissionMode } from '@maka/core/permission'; import type { SessionChangedEvent, SessionSummary, + StoredMessage, TurnRecord, } from '@maka/core/session'; import type { ContextCompactResult } from '@maka/runtime-host/protocol'; @@ -134,8 +135,10 @@ async function renderProbe( modelChoices?: readonly ChatModelChoice[]; ready?: (container: Element) => boolean; onSend?: (send: (text: string) => Promise) => void; + onQueue?: (queue: (text: string) => Promise) => void; onSteer?: (steer: (text: string) => Promise) => void; onStop?: (stop: () => Promise) => void; + onDeleteQueuedEntry?: (deleteEntry: (entryId: string) => Promise) => void; onSetPermissionMode?: (set: (mode: PermissionMode) => Promise) => void; confirmBypass?: () => Promise; onContextCompactionError?: (sessionId: string, error: unknown) => void; @@ -159,8 +162,10 @@ async function renderProbe( const children = options.ownership ? createElement(QuoteCompanionOwnershipProbe, { onSend: options.onSend ?? (() => undefined), + onQueue: options.onQueue, onSteer: options.onSteer, onStop: options.onStop, + onDeleteQueuedEntry: options.onDeleteQueuedEntry, onSetPermissionMode: options.onSetPermissionMode, onContextCompactionError: options.onContextCompactionError, pendingQuotes: options.pendingQuotes, @@ -199,8 +204,10 @@ async function renderOwnershipProbe( } = {}, ) { let send!: (text: string) => Promise; + let queue!: (text: string) => Promise; let steer!: (text: string) => Promise; let stop!: () => Promise; + let deleteQueuedEntry!: (entryId: string) => Promise; let setPermissionMode!: (mode: PermissionMode) => Promise; let eventHandler: ((event: SessionEvent) => void) | undefined; const subscribeEvents = sideChat.subscribeEvents; @@ -219,8 +226,10 @@ async function renderOwnershipProbe( { ownership: true, onSend: (value) => (send = value), + onQueue: (value) => (queue = value), onSteer: (value) => (steer = value), onStop: (value) => (stop = value), + onDeleteQueuedEntry: (value) => (deleteQueuedEntry = value), onSetPermissionMode: (value) => (setPermissionMode = value), ...options, }, @@ -228,8 +237,10 @@ async function renderOwnershipProbe( return { ...rendered, send: (text: string) => send(text), + queue: (text: string) => queue(text), steer: (text: string) => steer(text), stop: () => stop(), + deleteQueuedEntry: (entryId: string) => deleteQueuedEntry(entryId), setPermissionMode: (mode: PermissionMode) => setPermissionMode(mode), emit(event: SessionEvent) { assert.ok(eventHandler); @@ -477,7 +488,7 @@ test('dispatches /compact to the committed companion fork without sending model sendCalls += 1; return { ok: false as const, reason: 'seed only' }; }, - steer: async () => { + submitFollowUp: async () => { steerCalls += 1; return { kind: 'started' as const, turnId: 'unexpected-steer' }; }, @@ -501,6 +512,10 @@ test('dispatches the exact /compact Composer command before steering or ordinary calls.push('compact'); return true; }, + queue: async () => { + calls.push('queue'); + return true; + }, steer: async () => { calls.push('steer'); return true; @@ -515,6 +530,31 @@ test('dispatches the exact /compact Composer command before steering or ordinary assert.deepEqual(calls, ['compact']); }); +test('routes running Side Conversation submissions like the main conversation', async () => { + const calls: string[] = []; + const input = { + text: 'follow up', + streaming: true, + compact: async () => true, + queue: async (text: string) => { + calls.push(`queue:${text}`); + return true; + }, + steer: async (text: string) => { + calls.push(`steer:${text}`); + return true; + }, + send: async () => { + calls.push('send'); + return true; + }, + }; + + assert.equal(await dispatchQuoteCompanionInput(input), true); + assert.equal(await dispatchQuoteCompanionInput({ ...input, followUpMode: 'steer' }), true); + assert.deepEqual(calls, ['queue:follow up', 'steer:follow up']); +}); + test('keeps an async companion compaction exclusive until its terminal event', async () => { let compactCalls = 0; let sendCalls = 0; @@ -1523,12 +1563,12 @@ test('keeps the same Side Conversation admission across a recoverable subscripti }); test('keeps the active Side Conversation streaming when Stop retracts a queued steer', async () => { - const pendingSteer = deferred<{ kind: 'queued'; messageId: string }>(); + const pendingSteer = deferred<{ kind: 'queued' }>(); 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) => { + submitFollowUp: async (_sessionId, _placement, _text, requestedAdmissionId) => { steerCalls += 1; admissionId = requestedAdmissionId; return pendingSteer.promise; @@ -1569,19 +1609,19 @@ test('keeps the active Side Conversation streaming when Stop retracts a queued s assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); await act(async () => { - pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + pendingSteer.resolve({ kind: 'queued' }); 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 }>(); + const pendingSteer = deferred<{ kind: 'queued' }>(); let admissionId: string | undefined; const stoppedTargets: SideChatStopTarget[] = []; const { send, steer, stop } = await renderOwnershipProbe({ send: async () => ({ ok: true as const, turnId: 'old-turn' }), - steer: async (_sessionId, _text, requestedAdmissionId) => { + submitFollowUp: async (_sessionId, _placement, _text, requestedAdmissionId) => { admissionId = requestedAdmissionId; return pendingSteer.promise; }, @@ -1617,21 +1657,21 @@ test('stops the active Side Conversation after retracting its queued steer', asy { kind: 'turn', turnId: 'old-turn' }, ]); await act(async () => { - pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + pendingSteer.resolve({ kind: 'queued' }); 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 pendingSteer = deferred<{ kind: 'queued' }>(); const queuedStop = deferred(); const activeStop = deferred(); let admissionId: string | undefined; const stoppedTargets: SideChatStopTarget[] = []; const { emit, send, steer, stop } = await renderOwnershipProbe({ send: async () => ({ ok: true as const, turnId: 'old-turn' }), - steer: async (_sessionId, _text, requestedAdmissionId) => { + submitFollowUp: async (_sessionId, _placement, _text, requestedAdmissionId) => { admissionId = requestedAdmissionId; return pendingSteer.promise; }, @@ -1679,18 +1719,18 @@ test('does not let an older Stop failure release a newer active Turn Stop', asyn activeStop.resolve(undefined); await Promise.all([activeStopResult, duplicateStopResult]); await act(async () => { - pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + pendingSteer.resolve({ kind: 'queued' }); 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 }>(); + const pendingSteer = deferred<{ kind: 'queued' }>(); 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) => { + submitFollowUp: async (_sessionId, _placement, _text, requestedAdmissionId) => { admissionId = requestedAdmissionId; return pendingSteer.promise; }, @@ -1706,6 +1746,11 @@ test('continues projecting the active Turn while a steer awaits Host admission', await Promise.resolve(); }); await waitUntil(() => admissionId !== undefined); + assert.equal(container.firstElementChild?.getAttribute('data-transient-count'), '2'); + assert.equal( + container.firstElementChild?.getAttribute('data-transient-texts'), + 'initial prompt|queue this steer', + ); await act(async () => { emit(textDeltaEvent('old-turn-text', 'old-turn', 1, 'still streaming')); await Promise.resolve(); @@ -1716,10 +1761,855 @@ test('continues projecting the active Turn while a steer awaits Host admission', assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); await act(async () => { - pendingSteer.resolve({ kind: 'queued', messageId: admissionId as string }); + pendingSteer.resolve({ kind: 'queued' }); + assert.equal(await steerResult, true); + await Promise.resolve(); + }); +}); + +test('keeps an outcome-unknown Side Conversation steer addressable by message identity', async () => { + let admissionId: string | undefined; + const stoppedTargets: SideChatStopTarget[] = []; + const { send, steer, stop } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async (_sessionId, placement, _text, requestedAdmissionId) => { + assert.equal(placement, 'current_turn'); + admissionId = requestedAdmissionId; + return { kind: 'outcome_unknown' as const }; + }, + stop: async (_sessionId, target) => { + stoppedTargets.push(target); + return undefined; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + await act(async () => { + assert.equal(await steer('uncertain steer'), true); + await stop(); + await Promise.resolve(); + }); + + assert.deepEqual(stoppedTargets, [{ kind: 'admission', messageId: admissionId }]); +}); + +test('recovers a queued Side Conversation steer from the Host queue projection', async () => { + let admissionId: string | undefined; + const pendingSteer = deferred<{ kind: 'queued' }>(); + const { container, emit, send, steer } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async (_sessionId, _placement, _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('queued follow-up'); + await Promise.resolve(); + }); + await waitUntil(() => admissionId !== undefined); + await act(async () => { + emit( + queueUpdateEvent('queued-steer', 'old-turn', 1, [ + { + entryId: 'queued-steer-entry', + messageId: admissionId as string, + content: { text: 'queued follow-up' }, + placement: 'current_turn', + state: 'queued', + }, + ]), + ); + pendingSteer.resolve({ kind: 'queued' }); assert.equal(await steerResult, true); await Promise.resolve(); }); + + assert.equal( + container.firstElementChild?.getAttribute('data-transient-texts'), + 'initial prompt|queued follow-up', + ); + assert.equal(container.firstElementChild?.getAttribute('data-queue-texts'), 'queued follow-up'); + + await act(async () => { + emit({ + type: 'steering_message', + id: 'steering-consumed', + turnId: 'old-turn', + ts: 2, + messageId: admissionId as string, + content: { text: 'queued follow-up' }, + }); + await Promise.resolve(); + }); + + assert.equal(container.firstElementChild?.getAttribute('data-transient-texts'), 'initial prompt'); + assert.equal(container.firstElementChild?.getAttribute('data-queue-texts'), ''); +}); + +test('retracts a queued Side Conversation message without stopping the active turn', async () => { + let messageId: string | undefined; + const retracted: string[] = []; + const { container, emit, send, queue, deleteQueuedEntry } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async (_sessionId, _placement, _text, requestedMessageId) => { + messageId = requestedMessageId; + return { kind: 'queued' as const }; + }, + retractQueueEntry: async (_sessionId, entryId) => { + retracted.push(entryId); + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + await act(async () => { + assert.equal(await queue('remove me'), true); + emit( + queueUpdateEvent('queued-follow-up', 'old-turn', 1, [], [ + { + entryId: 'follow-up-entry', + messageId: messageId as string, + content: { text: 'remove me' }, + placement: 'next_turn', + state: 'queued', + }, + ]), + ); + await Promise.resolve(); + }); + + await act(async () => { + await deleteQueuedEntry('follow-up-entry'); + await Promise.resolve(); + }); + + assert.deepEqual(retracted, ['follow-up-entry']); + assert.equal(container.firstElementChild?.getAttribute('data-transient-texts'), 'initial prompt'); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'old-turn'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); +}); + +test('queues multiple Side Conversation follow-ups while the active turn keeps streaming', async () => { + const submissions: Array<{ placement: string; text: string; messageId: string }> = []; + const { container, send, queue } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async (_sessionId, placement, text, messageId) => { + submissions.push({ placement, text, messageId }); + return { kind: 'queued' as const }; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + await act(async () => { + assert.equal(await queue('first follow-up'), true); + await Promise.resolve(); + }); + await act(async () => { + assert.equal(await queue('second follow-up'), true); + await Promise.resolve(); + }); + + assert.deepEqual( + submissions.map(({ placement, text }) => ({ placement, text })), + [ + { placement: 'next_turn', text: 'first follow-up' }, + { placement: 'next_turn', text: 'second follow-up' }, + ], + ); + assert.equal( + container.firstElementChild?.getAttribute('data-transient-texts'), + 'initial prompt|first follow-up|second follow-up', + ); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'old-turn'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); +}); + +test('adopts a queued Side Conversation follow-up that starts after the active turn settles', async () => { + let followUpMessageId: string | undefined; + const pendingFollowUp = deferred<{ kind: 'started'; turnId: string }>(); + const { container, emit, send, queue } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async (_sessionId, placement, _text, messageId) => { + assert.equal(placement, 'next_turn'); + followUpMessageId = messageId; + return pendingFollowUp.promise; + }, + readSettledMessages: async () => ({ messages: [], settled: true }), + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + let followUpResult!: Promise; + await act(async () => { + followUpResult = queue('start after settlement'); + await Promise.resolve(); + }); + await waitUntil(() => followUpMessageId !== undefined); + await act(async () => { + emit(completeEvent('old-complete', 'old-turn', 1)); + pendingFollowUp.resolve({ kind: 'started', turnId: 'new-turn' }); + assert.equal(await followUpResult, true); + await Promise.resolve(); + }); + + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'new-turn'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); + assert.equal( + container.firstElementChild?.getAttribute('data-transient-texts'), + 'initial prompt|start after settlement', + ); + await act(async () => { + emit(textDeltaEvent('new-turn-text', 'new-turn', 2, 'new answer')); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-live-text'), 'new answer'); +}); + +test('reconciles a queued Side Conversation follow-up that settles before its started receipt', async () => { + let followUpMessageId: string | undefined; + let durableMessages: StoredMessage[] = []; + const pendingFollowUp = deferred<{ kind: 'started'; turnId: string }>(); + const { container, emit, send, queue } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async (_sessionId, placement, _text, messageId) => { + assert.equal(placement, 'next_turn'); + followUpMessageId = messageId; + return pendingFollowUp.promise; + }, + readSettledMessages: async () => ({ messages: durableMessages, settled: true }), + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + let followUpResult!: Promise; + await act(async () => { + followUpResult = queue('late follow-up'); + await Promise.resolve(); + }); + await waitUntil(() => followUpMessageId !== undefined); + await act(async () => { + emit(completeEvent('old-complete', 'old-turn', 1)); + await Promise.resolve(); + }); + await waitUntil(() => container.firstElementChild?.getAttribute('data-streaming') === 'false'); + + durableMessages = [ + { + type: 'user', + id: followUpMessageId as string, + turnId: 'new-turn', + ts: 2, + text: 'late follow-up', + }, + { + type: 'assistant', + id: 'new-assistant', + turnId: 'new-turn', + ts: 3, + text: 'new answer', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'new-complete-state', + turnId: 'new-turn', + ts: 4, + status: 'completed', + }, + ]; + await act(async () => { + emit( + messageAdmittedEvent( + 'new-turn-admission', + 'new-turn', + 2, + followUpMessageId as string, + ), + ); + emit(textDeltaEvent('new-turn-text', 'new-turn', 2, 'new answer')); + emit(completeEvent('new-complete', 'new-turn', 3)); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'false'); + + await act(async () => { + pendingFollowUp.resolve({ kind: 'started', turnId: 'new-turn' }); + assert.equal(await followUpResult, true); + await Promise.resolve(); + }); + + assert.equal( + container.firstElementChild?.getAttribute('data-message-texts'), + 'late follow-up|new answer', + ); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), ''); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'false'); +}); + +test('does not re-arm a settled follow-up after it leaves the bounded transcript tail', async () => { + let followUpMessageId: string | undefined; + let durableMessages: StoredMessage[] = []; + const turnBSettlement = deferred<{ + messages: StoredMessage[]; + settled: boolean; + }>(); + const pendingFollowUp = deferred<{ kind: 'started'; turnId: string }>(); + const { container, emit, send, queue } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'turn-a' }), + submitFollowUp: async (_sessionId, placement, _text, messageId) => { + assert.equal(placement, 'next_turn'); + followUpMessageId = messageId; + return pendingFollowUp.promise; + }, + readSettledMessages: async (_sessionId, options) => { + if (options?.requiredAssistantMessageId !== undefined) { + return turnBSettlement.promise; + } + return { messages: durableMessages, settled: true }; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + let followUpResult!: Promise; + await act(async () => { + followUpResult = queue('late follow-up'); + await Promise.resolve(); + }); + await waitUntil(() => followUpMessageId !== undefined); + await act(async () => { + emit(completeEvent('complete-a', 'turn-a', 1)); + await Promise.resolve(); + }); + await waitUntil(() => container.firstElementChild?.getAttribute('data-streaming') === 'false'); + + durableMessages = [ + { + type: 'user', + id: followUpMessageId as string, + turnId: 'turn-b', + ts: 2, + text: 'late follow-up', + }, + { + type: 'assistant', + id: 'assistant-b', + turnId: 'turn-b', + ts: 3, + text: 'answer B', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'complete-b-state', + turnId: 'turn-b', + ts: 4, + status: 'completed', + }, + ]; + await act(async () => { + emit(messageAdmittedEvent('admission-b', 'turn-b', 2, followUpMessageId as string)); + emit(textDeltaEvent('text-b', 'turn-b', 3, 'answer B')); + emit(completeEvent('complete-b', 'turn-b', 4)); + await Promise.resolve(); + }); + await act(async () => { + turnBSettlement.resolve({ messages: durableMessages, settled: true }); + await Promise.resolve(); + }); + await waitUntil( + () => container.firstElementChild?.getAttribute('data-message-texts') + === 'late follow-up|answer B', + ); + await waitUntil(() => container.firstElementChild?.getAttribute('data-streaming') === 'false'); + + // The next bounded snapshot contains only the later terminal Turn C. The + // panel has already observed and retained B's terminal state, so B's delayed + // started receipt must not make it live again merely because it left the tail. + durableMessages = [ + { + type: 'turn_state', + id: 'complete-c-state', + turnId: 'turn-c', + ts: 5, + status: 'completed', + }, + ]; + await act(async () => { + emit(messageAdmittedEvent('admission-c', 'turn-c', 5, 'message-c')); + await Promise.resolve(); + }); + await waitUntil(() => container.firstElementChild?.getAttribute('data-live-turn-id') === 'turn-c'); + await act(async () => { + emit(completeEvent('complete-c', 'turn-c', 6)); + await Promise.resolve(); + }); + await waitUntil(() => container.firstElementChild?.getAttribute('data-live-turn-id') === ''); + + await act(async () => { + pendingFollowUp.resolve({ kind: 'started', turnId: 'turn-b' }); + assert.equal(await followUpResult, true); + await Promise.resolve(); + }); + + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), ''); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'false'); + assert.equal( + container.firstElementChild?.getAttribute('data-message-texts'), + 'late follow-up|answer B', + ); +}); + +test('does not let a late Side Conversation started receipt replace a newer active turn', async () => { + const pendingFollowUp = deferred<{ kind: 'started'; turnId: string }>(); + const { container, emit, send, queue } = await renderOwnershipProbe({ + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async () => pendingFollowUp.promise, + readSettledMessages: async () => ({ messages: [], settled: true }), + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + let followUpResult!: Promise; + await act(async () => { + followUpResult = queue('late follow-up'); + await Promise.resolve(); + }); + await act(async () => { + emit(messageAdmittedEvent('newer-admission', 'newer-turn', 2, 'newer-message')); + emit(textDeltaEvent('newer-text', 'newer-turn', 3, 'newer answer')); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'newer-turn'); + + await act(async () => { + pendingFollowUp.resolve({ kind: 'started', turnId: 'late-turn' }); + assert.equal(await followUpResult, true); + await Promise.resolve(); + }); + + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'newer-turn'); + assert.equal(container.firstElementChild?.getAttribute('data-live-text'), 'newer answer'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); +}); + +test('keeps the settled prior turn visible while a queued successor is running', async () => { + let firstMessageId: string | undefined; + let followUpMessageId: string | undefined; + const oldTurnSettlement = deferred<{ + messages: StoredMessage[]; + settled: boolean; + }>(); + const pendingFollowUp = deferred<{ kind: 'started'; turnId: string }>(); + let stoppedTarget: SideChatStopTarget; + const { container, emit, send, queue, stop } = await renderOwnershipProbe({ + send: async (_sessionId, command) => { + firstMessageId = command.turnId; + return { ok: true as const, turnId: 'old-turn' }; + }, + submitFollowUp: async (_sessionId, placement, _text, messageId) => { + assert.equal(placement, 'next_turn'); + followUpMessageId = messageId; + return pendingFollowUp.promise; + }, + readSettledMessages: async (_sessionId, options) => + options?.requiredAssistantMessageId === 'assistant-message' + ? oldTurnSettlement.promise + : { messages: [], settled: true }, + stop: async (_sessionId, target) => { + stoppedTarget = target; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + emit(textDeltaEvent('old-turn-text', 'old-turn', 1, 'old answer')); + await Promise.resolve(); + }); + let followUpResult!: Promise; + await act(async () => { + followUpResult = queue('start next'); + await Promise.resolve(); + }); + await waitUntil(() => followUpMessageId !== undefined); + await act(async () => { + emit(completeEvent('old-complete', 'old-turn', 2)); + pendingFollowUp.resolve({ kind: 'started', turnId: 'new-turn' }); + assert.equal(await followUpResult, true); + await Promise.resolve(); + }); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'new-turn'); + + await act(async () => { + oldTurnSettlement.resolve({ + messages: [ + { + type: 'user', + id: firstMessageId as string, + turnId: 'old-turn', + ts: 1, + text: 'initial prompt', + }, + { + type: 'assistant', + id: 'assistant-message', + turnId: 'old-turn', + ts: 2, + text: 'old answer', + modelId: 'test-model', + }, + ], + settled: true, + }); + await Promise.resolve(); + }); + + await waitUntil( + () => container.firstElementChild?.getAttribute('data-message-texts') === 'initial prompt|old answer', + ); + assert.equal(container.firstElementChild?.getAttribute('data-live-turn-id'), 'new-turn'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'true'); + await act(async () => { + await stop(); + await Promise.resolve(); + }); + assert.deepEqual(stoppedTarget, { kind: 'turn', turnId: 'new-turn' }); +}); + +test('retires a cancelled queued Side Conversation message after observation reseeds', async () => { + let queuedMessageId: string | undefined; + let markSeeded: (() => void) | undefined; + let seedCount = 0; + const queriedMessageIds: string[][] = []; + const { container, emit, send, queue } = await renderOwnershipProbe({ + subscribeEvents: (_sessionId, _handler, onSeeded) => { + markSeeded = onSeeded; + if (seedCount === 0) { + seedCount += 1; + onSeeded?.(); + } + return () => undefined; + }, + send: async () => ({ ok: true as const, turnId: 'old-turn' }), + submitFollowUp: async (_sessionId, placement, _text, messageId) => { + assert.equal(placement, 'next_turn'); + queuedMessageId = messageId; + return { kind: 'queued' as const }; + }, + queryMessageExecutions: async (_sessionId, messageIds) => { + queriedMessageIds.push([...messageIds]); + return { + resolutions: messageIds.map((messageId) => + messageId === queuedMessageId + ? { messageId, state: 'cancelled' as const } + : { messageId, state: 'pending' as const }), + }; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + emit(textDeltaEvent('old-turn-text', 'old-turn', 1, 'still streaming')); + await Promise.resolve(); + }); + await act(async () => { + assert.equal(await queue('cancelled while disconnected'), true); + emit( + queueUpdateEvent('queued-follow-up', 'old-turn', 2, [], [ + { + entryId: 'follow-up-entry', + messageId: queuedMessageId as string, + content: { text: 'cancelled while disconnected' }, + placement: 'next_turn', + state: 'queued', + }, + ]), + ); + markSeeded?.(); + await Promise.resolve(); + }); + await waitUntil( + () => container.firstElementChild?.getAttribute('data-transient-texts') === 'initial prompt', + ); + + assert.equal(container.firstElementChild?.getAttribute('data-queue-texts'), ''); + assert.ok(queriedMessageIds.some((messageIds) => messageIds.includes(queuedMessageId as string))); +}); + +test('retires durable Side Conversation identities before observation recovery queries', async () => { + let rootMessageId: string | undefined; + let followUpMessageId: string | undefined; + let markSeeded: (() => void) | undefined; + let durableMessages: StoredMessage[] = []; + const queriedMessageIds: string[][] = []; + const { container, emit, send, queue } = await renderOwnershipProbe({ + subscribeEvents: (_sessionId, _handler, onSeeded) => { + markSeeded = onSeeded; + onSeeded?.(); + return () => undefined; + }, + send: async (_sessionId, command) => { + rootMessageId = command.turnId; + return { ok: true as const, turnId: 'turn-a' }; + }, + submitFollowUp: async (_sessionId, placement, _text, messageId) => { + assert.equal(placement, 'next_turn'); + followUpMessageId = messageId; + return { kind: 'queued' as const }; + }, + readSettledMessages: async () => ({ messages: durableMessages, settled: true }), + queryMessageExecutions: async (_sessionId, messageIds) => { + queriedMessageIds.push([...messageIds]); + return { + resolutions: messageIds.map((messageId) => ({ + messageId, + state: 'pending' as const, + })), + }; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + await act(async () => { + assert.equal(await queue('follow-up'), true); + await Promise.resolve(); + }); + const durableRootMessageId = rootMessageId; + const durableFollowUpMessageId = followUpMessageId; + assert.ok(durableRootMessageId); + assert.ok(durableFollowUpMessageId); + durableMessages = [ + { + type: 'user', + id: durableRootMessageId, + turnId: 'turn-a', + ts: 1, + text: 'initial prompt', + }, + { + type: 'assistant', + id: 'assistant-a', + turnId: 'turn-a', + ts: 2, + text: 'answer A', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'complete-a', + turnId: 'turn-a', + ts: 3, + status: 'completed', + }, + { + type: 'user', + id: durableFollowUpMessageId, + turnId: 'turn-b', + ts: 4, + text: 'follow-up', + }, + { + type: 'assistant', + id: 'assistant-b', + turnId: 'turn-b', + ts: 5, + text: 'answer B', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'complete-b', + turnId: 'turn-b', + ts: 6, + status: 'completed', + }, + ]; + + await act(async () => { + emit(completeEvent('event-complete-a', 'turn-a', 3)); + emit(messageAdmittedEvent('admission-b', 'turn-b', 4, durableFollowUpMessageId)); + emit(completeEvent('event-complete-b', 'turn-b', 6)); + await Promise.resolve(); + }); + await waitUntil( + () => container.firstElementChild?.getAttribute('data-message-texts') + === 'initial prompt|answer A|follow-up|answer B', + ); + assert.equal(container.firstElementChild?.getAttribute('data-transient-count'), '0'); + + queriedMessageIds.length = 0; + await act(async () => { + markSeeded?.(); + await Promise.resolve(); + }); + assert.deepEqual(queriedMessageIds, []); +}); + +test('recovers every queued Side Conversation successor across one observation gap', async () => { + let rootMessageId: string | undefined; + let firstFollowUpId: string | undefined; + let secondFollowUpId: string | undefined; + let markSeeded: (() => void) | undefined; + let durableMessages: StoredMessage[] = []; + const executionQueries: string[][] = []; + const { container, emit, send, queue } = await renderOwnershipProbe({ + subscribeEvents: (_sessionId, _handler, onSeeded) => { + markSeeded = onSeeded; + onSeeded?.(); + return () => undefined; + }, + send: async (_sessionId, command) => { + rootMessageId = command.turnId; + return { ok: true as const, turnId: 'turn-a' }; + }, + submitFollowUp: async (_sessionId, placement, text, messageId) => { + assert.equal(placement, 'next_turn'); + if (text === 'follow-up one') firstFollowUpId = messageId; + else secondFollowUpId = messageId; + return { kind: 'queued' as const }; + }, + readSettledMessages: async () => ({ messages: durableMessages, settled: true }), + queryMessageExecutions: async (_sessionId, messageIds) => { + executionQueries.push([...messageIds]); + return { + resolutions: messageIds.map((messageId) => ({ + messageId, + state: 'owned' as const, + turnId: messageId === firstFollowUpId ? 'turn-b' : 'turn-c', + runId: messageId === firstFollowUpId ? 'run-b' : 'run-c', + })), + }; + }, + }); + + await act(async () => { + assert.equal(await send('initial prompt'), true); + await Promise.resolve(); + }); + await act(async () => { + assert.equal(await queue('follow-up one'), true); + assert.equal(await queue('follow-up two'), true); + await Promise.resolve(); + }); + const durableFirstFollowUpId = firstFollowUpId; + const durableSecondFollowUpId = secondFollowUpId; + const durableRootMessageId = rootMessageId; + assert.ok(durableRootMessageId); + assert.ok(durableFirstFollowUpId); + assert.ok(durableSecondFollowUpId); + durableMessages = [ + { + type: 'user', + id: durableRootMessageId, + turnId: 'turn-a', + ts: 0, + text: 'initial prompt', + }, + { + type: 'assistant', + id: 'assistant-a', + turnId: 'turn-a', + ts: 1, + text: 'answer A', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'complete-a', + turnId: 'turn-a', + ts: 2, + status: 'completed', + }, + { + type: 'user', + id: durableFirstFollowUpId, + turnId: 'turn-b', + ts: 3, + text: 'follow-up one', + }, + { + type: 'assistant', + id: 'assistant-b', + turnId: 'turn-b', + ts: 4, + text: 'answer B', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'complete-b', + turnId: 'turn-b', + ts: 5, + status: 'completed', + }, + { + type: 'user', + id: durableSecondFollowUpId, + turnId: 'turn-c', + ts: 6, + text: 'follow-up two', + }, + { + type: 'assistant', + id: 'assistant-c', + turnId: 'turn-c', + ts: 7, + text: 'answer C', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'complete-c', + turnId: 'turn-c', + ts: 8, + status: 'completed', + }, + ]; + + await act(async () => { + // Replacement currently replays only the latest terminal root admission. + emit(messageAdmittedEvent('admission-c', 'turn-c', 6, durableSecondFollowUpId)); + emit(completeEvent('event-complete-c', 'turn-c', 8)); + markSeeded?.(); + await Promise.resolve(); + }); + + await waitUntil( + () => container.firstElementChild?.getAttribute('data-message-texts') + === 'initial prompt|answer A|follow-up one|answer B|follow-up two|answer C', + ); + assert.ok( + executionQueries.some((messageIds) => + messageIds.includes(durableFirstFollowUpId)), + ); + assert.equal(container.firstElementChild?.getAttribute('data-transient-count'), '0'); + assert.equal(container.firstElementChild?.getAttribute('data-streaming'), 'false'); }); test('fails a send when observation seed rejects and resubscribes for retry', async () => { @@ -2012,8 +2902,10 @@ function QuoteCompanionProbe(props: { function QuoteCompanionOwnershipProbe(props: { onSend: (send: (text: string) => Promise) => void; + onQueue?: (queue: (text: string) => Promise) => void; onSteer?: (steer: (text: string) => Promise) => void; onStop?: (stop: () => Promise) => void; + onDeleteQueuedEntry?: (deleteEntry: (entryId: string) => Promise) => void; onSetPermissionMode?: (set: (mode: PermissionMode) => Promise) => void; onContextCompactionError?: (sessionId: string, error: unknown) => void; pendingQuotes?: readonly StagedCompanionQuote[]; @@ -2033,8 +2925,10 @@ function QuoteCompanionOwnershipProbe(props: { onContextCompactionError: props.onContextCompactionError, }); props.onSend(companion.send); + props.onQueue?.(companion.queue); props.onSteer?.(companion.steer); props.onStop?.(companion.stop); + props.onDeleteQueuedEntry?.(companion.deleteQueuedEntry); props.onSetPermissionMode?.(companion.setPermissionMode); return createElement('div', { 'data-companion-id': companion.companionSession?.id ?? '', @@ -2047,6 +2941,11 @@ function QuoteCompanionOwnershipProbe(props: { 'data-permission-mode': companion.permissionMode ?? '', 'data-transient-count': String(companion.transientMessages.length), 'data-transient-text': companion.transientMessages[0]?.text ?? '', + 'data-transient-texts': companion.transientMessages.map((message) => message.text).join('|'), + 'data-message-texts': companion.messages + .flatMap((message) => 'text' in message && typeof message.text === 'string' ? [message.text] : []) + .join('|'), + 'data-queue-texts': companion.queuedMessages?.map((entry) => entry.content.text).join('|') ?? '', }); } 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 c8a73a0652..2ed7fcb2cc 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 @@ -27,6 +27,7 @@ import type { IpcMain } from "electron"; import { SIDE_CONVERSATION_SESSION_LABEL } from '@maka/core/side-conversation'; import { type AttachmentRef } from '@maka/core/events'; import { + MESSAGE_QUEUE_MAX_ENTRIES, SESSION_CONTINUITY_SCHEMA_VERSION, type SessionCatalogProjection, } from "@maka/runtime-host/protocol"; @@ -773,14 +774,18 @@ test("submits an ordinary composer message once under its stable message identit test('returns Host-owned cancellation proof to the renderer', async () => { const ipc = ipcHarness(); + const queriedMessageIds: string[][] = []; registerExecutionIpc( { client: executionClient({ - queryMessages: async (input) => ({ - cancelledMessageIds: input.messageIds.filter( + queryMessages: async (input) => { + queriedMessageIds.push([...input.messageIds]); + return { + cancelledMessageIds: input.messageIds.filter( (messageId) => messageId === 'message-cancelled', - ), - }), + ), + }; + }, }), }, ipc, @@ -793,6 +798,109 @@ test('returns Host-owned cancellation proof to the renderer', async () => { ]), { cancelledMessageIds: ['message-cancelled'] }, ); + assert.deepEqual(queriedMessageIds, [['message-accepted', 'message-cancelled']]); +}); + +test('batches cancellation proof queries at the Runtime Host protocol boundary', async () => { + const ipc = ipcHarness(); + const queriedMessageIds: string[][] = []; + registerExecutionIpc( + { + client: executionClient({ + queryMessages: async (input) => { + queriedMessageIds.push([...input.messageIds]); + return { cancelledMessageIds: input.messageIds.slice(-1) }; + }, + }), + }, + ipc, + ); + const messageIds = Array.from({ length: 65 }, (_, index) => `message-${index}`); + + assert.deepEqual( + await ipc.invoke('sessions:queryCancelledMessages', 'session-1', messageIds), + { cancelledMessageIds: ['message-63', 'message-64'] }, + ); + assert.deepEqual(queriedMessageIds.map((ids) => ids.length), [64, 1]); +}); + +test('rejects duplicate cancellation proof identities across protocol batches', async () => { + const ipc = ipcHarness(); + let queryCount = 0; + registerExecutionIpc( + { + client: executionClient({ + queryMessages: async () => { + queryCount += 1; + return { cancelledMessageIds: [] }; + }, + }), + }, + ipc, + ); + const messageIds = Array.from({ length: 65 }, (_, index) => `message-${index}`); + messageIds[64] = messageIds[0] as string; + + await assert.rejects( + ipc.invoke('sessions:queryCancelledMessages', 'session-1', messageIds), + /Duplicate Message identities/u, + ); + assert.equal(queryCount, 0); +}); + +test('rejects invalid or unbounded Desktop cancellation proof queries before dispatch', async () => { + const ipc = ipcHarness(); + let queryCount = 0; + registerExecutionIpc( + { + client: executionClient({ + queryMessages: async () => { + queryCount += 1; + return { cancelledMessageIds: [] }; + }, + }), + }, + ipc, + ); + + await assert.rejects( + ipc.invoke('sessions:queryCancelledMessages', 'session-1', ['valid-message', 42]), + /Invalid Message identity/u, + ); + await assert.rejects( + ipc.invoke('sessions:queryCancelledMessages', 'session-1', ['not a protocol identity']), + /Invalid Message identity/u, + ); + await assert.rejects( + ipc.invoke( + 'sessions:queryCancelledMessages', + 'session-1', + Array.from({ length: 4_097 }, (_, index) => `message-${index}`), + ), + /Invalid Message identities/u, + ); + assert.equal(queryCount, 0); +}); + +test('rejects duplicate cancellation proofs returned across protocol batches', async () => { + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + queryMessages: async () => ({ cancelledMessageIds: ['message-0'] }), + }), + }, + ipc, + ); + + await assert.rejects( + ipc.invoke( + 'sessions:queryCancelledMessages', + 'session-1', + Array.from({ length: 65 }, (_, index) => `message-${index}`), + ), + /Duplicate cancelled Message identities/u, + ); }); test('returns Host-owned Message execution resolutions to the renderer', async () => { @@ -834,6 +942,117 @@ test('returns Host-owned Message execution resolutions to the renderer', async ( ); }); +test('batches Message execution queries at the Runtime Host protocol boundary', async () => { + const ipc = ipcHarness(); + const queriedMessageIds: string[][] = []; + registerExecutionIpc( + { + client: executionClient({ + queryMessageExecutions: async (input) => { + queriedMessageIds.push([...input.messageIds]); + assert.ok(input.messageIds.length <= MESSAGE_QUEUE_MAX_ENTRIES); + return { + resolutions: input.messageIds.map((messageId) => { + if (messageId === 'message-0') { + return { + messageId, + state: 'owned' as const, + turnId: 'turn-0', + runId: 'run-0', + }; + } + if (messageId === 'message-64') { + return { messageId, state: 'cancelled' as const }; + } + return { messageId, state: 'pending' as const }; + }), + }; + }, + }), + }, + ipc, + ); + const messageIds = Array.from({ length: 65 }, (_, index) => `message-${index}`); + + const result = await ipc.invoke( + 'sessions:queryMessageExecutions', + 'session-1', + messageIds, + ) as { resolutions: unknown[] }; + + assert.deepEqual(queriedMessageIds.map((ids) => ids.length), [64, 1]); + assert.deepEqual(result.resolutions, messageIds.map((messageId) => { + if (messageId === 'message-0') { + return { + messageId, + state: 'owned', + turnId: 'turn-0', + runId: 'run-0', + }; + } + if (messageId === 'message-64') return { messageId, state: 'cancelled' }; + return { messageId, state: 'pending' }; + })); +}); + +test('rejects duplicate Message execution identities before protocol batching', async () => { + const ipc = ipcHarness(); + let queryCount = 0; + registerExecutionIpc( + { + client: executionClient({ + queryMessageExecutions: async () => { + queryCount += 1; + return { resolutions: [] }; + }, + }), + }, + ipc, + ); + const messageIds = Array.from({ length: 65 }, (_, index) => `message-${index}`); + messageIds[64] = messageIds[0] as string; + + await assert.rejects( + ipc.invoke('sessions:queryMessageExecutions', 'session-1', messageIds), + /Duplicate Message identities/u, + ); + assert.equal(queryCount, 0); +}); + +test('rejects invalid or unbounded Desktop Message execution queries before dispatch', async () => { + const ipc = ipcHarness(); + let queryCount = 0; + registerExecutionIpc( + { + client: executionClient({ + queryMessageExecutions: async () => { + queryCount += 1; + return { resolutions: [] }; + }, + }), + }, + ipc, + ); + + await assert.rejects( + ipc.invoke('sessions:queryMessageExecutions', 'session-1', ['valid-message', 42]), + /Invalid Message identity/u, + ); + await assert.rejects( + ipc.invoke('sessions:queryMessageExecutions', 'session-1', ['not a protocol identity']), + /Invalid Message identity/u, + ); + await assert.rejects( + ipc.invoke( + 'sessions:queryMessageExecutions', + 'session-1', + Array.from({ length: 4_097 }, (_, index) => `message-${index}`), + ), + /Invalid Message identities/u, + ); + assert.equal(queryCount, 0); +}); + test('submits a slash Skill message and reports the Host Skill outcome', async () => { const submits: unknown[] = []; const ipc = ipcHarness(); 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 549ef02d70..5b22b88ef9 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 @@ -2436,6 +2436,110 @@ test("reconciles terminal, Goal, interaction, and sidecar state after subscripti await observer.close(); }); +test('replays durable admission before a terminal successor on subscription recovery', async () => { + const firstEvents = new AsyncFrameQueue(); + const secondEvents = new AsyncFrameQueue(); + const recoveredSessions: string[] = []; + const terminalTranscript: StoredMessage[] = [ + { + type: 'user', + id: 'follow-up-message', + turnId: 'turn-2', + ts: 20, + text: 'Continue', + steeringEventId: 'follow-up-steering', + }, + { + type: 'assistant', + id: 'assistant-2', + turnId: 'turn-2', + ts: 30, + text: 'Done', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'terminal-2', + turnId: 'turn-2', + ts: 40, + status: 'completed', + }, + ]; + let openCount = 0; + const observer = new RuntimeHostSessionObserver({ + client: { + listSessionTurns: async () => [{ + turnId: 'turn-1', + status: 'completed' as const, + statusSource: 'recorded' as const, + }], + openSession: async () => { + openCount += 1; + if (openCount === 1) { + return runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: firstEvents, + async close() { + firstEvents.end(); + }, + }); + } + return runtimeHostSessionFixture({ + snapshot: continuitySnapshot({ + projectionRevision: 2, + rootTurn: { + sessionId: 'session-1', + turnId: 'turn-2', + runId: 'run-2', + status: 'completed', + terminalEventId: 'terminal-2', + }, + }), + transcript: Promise.resolve(terminalTranscript), + loadTranscriptOverlay: async () => terminalTranscript, + events: secondEvents, + async close() { + secondEvents.end(); + }, + }); + }, + }, + emitSessionsChanged() {}, + emitSubscriptionRecovered: (sessionId) => { + recoveredSessions.push(sessionId); + }, + now: () => 50, + }); + const target = eventTarget(23); + await observer.observe('session-1', 'observer-1', target, true); + + firstEvents.push({ + kind: 'subscription.closed', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 1, + reason: 'slow_consumer', + }); + await waitFor(() => recoveredSessions.length === 1); + + assert.deepEqual( + target.events + .filter((event) => event.turnId === 'turn-2') + .map((event) => event.type), + ['message_admission', 'text_complete', 'complete'], + ); + const admission = target.events.find( + (event): event is Extract => + event.type === 'message_admission', + ); + assert.deepEqual( + admission && { messageId: admission.messageId, turnId: admission.turnId }, + { messageId: 'follow-up-message', turnId: 'turn-2' }, + ); + await observer.close(); +}); + test("shares one Host subscription and one delivery per renderer target", async () => { const events = new AsyncFrameQueue(); let openCount = 0; diff --git a/apps/desktop/src/main/__tests__/transient-message-projection.test.ts b/apps/desktop/src/main/__tests__/transient-message-projection.test.ts index 7755365af8..ed5e3f0736 100644 --- a/apps/desktop/src/main/__tests__/transient-message-projection.test.ts +++ b/apps/desktop/src/main/__tests__/transient-message-projection.test.ts @@ -21,11 +21,12 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import type { StoredMessage } from '@maka/core/session'; import type { TransientUserMessageProjection } from '@maka/ui'; +import { deriveMessageQueueProjection } from '../../renderer/application/contracts/message-queue-projection.js'; import { mergeTransientMessageProjection, projectQueuedTransientMessages, reconcileTransientMessages, -} from '../../renderer/transient-message-projection.js'; +} from '../../renderer/application/contracts/transient-message-projection.js'; /** The durable Message that replaces the transient row above. */ function canonicalSend(): StoredMessage { @@ -140,6 +141,60 @@ test('uses the Host queue snapshot order for already-present transient messages' ); }); +test('derives one queue projection for main and Side Conversation consumers', () => { + const projection = deriveMessageQueueProjection({ + type: 'queue_update', + id: 'queue-1', + turnId: 'turn-1', + ts: 7, + steering: ['in flight', 'steer'], + followup: ['next'], + steeringEntries: [ + { + entryId: 'in-flight', + messageId: 'message-in-flight', + content: { text: 'in flight' }, + placement: 'current_turn', + state: 'in_flight', + }, + { + entryId: 'steer', + messageId: 'message-steer', + content: { text: 'raw', displayText: 'steer', quotes: [{ text: 'context' }] }, + placement: 'current_turn', + state: 'queued', + }, + ], + followupEntries: [ + { + entryId: 'next', + messageId: 'message-next', + content: { text: 'next' }, + placement: 'next_turn', + state: 'queued', + }, + ], + }); + + assert.deepEqual(projection.entries.map((entry) => entry.entryId), ['steer', 'next']); + assert.deepEqual(projection.transientMessages, [ + { + id: 'message-steer', + transientPlacement: 'current_turn', + hostTurnId: 'turn-1', + ts: 7, + text: 'steer', + quotes: [{ text: 'context' }], + }, + { + id: 'message-next', + transientPlacement: 'next_turn', + ts: 7, + text: 'next', + }, + ]); +}); + test('keeps a Host-bound current Turn when a later IPC result has no Turn identity', () => { const hostBound = { ...transient, id: 'message-current', hostTurnId: 'host-turn' }; const lateIpcUpdate = { ...transient, id: 'message-current', text: 'uploaded content' }; diff --git a/apps/desktop/src/main/__tests__/workbar-services-adapter.test.ts b/apps/desktop/src/main/__tests__/workbar-services-adapter.test.ts index 6e92fabf11..31df8a5a3e 100644 --- a/apps/desktop/src/main/__tests__/workbar-services-adapter.test.ts +++ b/apps/desktop/src/main/__tests__/workbar-services-adapter.test.ts @@ -97,6 +97,30 @@ describe('createDesktopWorkbarServices', () => { ); }); + it('reports Side Conversation readiness again after an observation reseed', () => { + const { bridge, calls } = createBridgeRecorder(); + const services = createDesktopWorkbarServices(bridge, { + readSettledMessages: async () => ({ messages: [], settled: true }), + }); + let readyCount = 0; + + services.sideChat.subscribeEvents('fork', () => undefined, () => { + readyCount += 1; + })(); + + const subscribe = calls.find((call) => call.name === 'sessions.subscribeEvents'); + assert.ok(subscribe); + const initialReady = subscribe.args[2] as (() => void) | undefined; + const observationSeed = subscribe.args[3] as + | ((phase: 'pending' | 'ready') => void) + | undefined; + initialReady?.(); + observationSeed?.('pending'); + observationSeed?.('ready'); + + assert.equal(readyCount, 2); + }); + it('maps every Workbar capability to the existing Desktop bridge', async () => { const { bridge, calls } = createBridgeRecorder(); const settledReads: unknown[][] = []; @@ -166,7 +190,23 @@ describe('createDesktopWorkbarServices', () => { text: 'hello', }); await services.sideChat.stop('fork'); - await services.sideChat.steer('fork', 'more'); + const nextFollowUp = await services.sideChat.submitFollowUp( + 'fork', + 'next_turn', + 'later', + 'message-next', + ); + const currentFollowUp = await services.sideChat.submitFollowUp( + 'fork', + 'current_turn', + 'more', + 'message-current', + ); + await services.sideChat.queryMessageExecutions('fork', ['message-next']); + await services.sideChat.retractQueueEntry('fork', 'entry-1'); + await services.sideChat.promoteQueueEntry('fork', 'entry-2'); + await services.sideChat.updateQueueEntry('fork', 'entry-3', 4, 'updated'); + await services.sideChat.reorderQueueEntries('fork', ['entry-3', 'entry-2']); await services.sideChat.setPermissionMode('fork', 'ask'); await services.sideChat.regenerateTurn('fork', { sourceTurnId: 'turn-2', @@ -223,6 +263,12 @@ describe('createDesktopWorkbarServices', () => { 'sessions.send', 'sessions.stop', 'sessions.submitMessage', + 'sessions.submitMessage', + 'sessions.queryMessageExecutions', + 'sessions.retractQueueEntry', + 'sessions.promoteQueueEntry', + 'sessions.updateQueueEntry', + 'sessions.reorderQueueEntries', 'sessions.setPermissionMode', 'sessions.regenerateTurn', 'sessions.respondToSandboxBoundary', @@ -249,5 +295,22 @@ describe('createDesktopWorkbarServices', () => { 's', { requiredAssistantMessageId: 'message' }, ]); + const followUpCalls = calls.filter((call) => call.name === 'sessions.submitMessage'); + assert.deepEqual(followUpCalls[0]?.args, [ + 'fork', + 'next_turn', + { messageId: 'message-next', text: 'later' }, + ]); + assert.deepEqual(followUpCalls[1]?.args, [ + 'fork', + 'current_turn', + { messageId: 'message-current', text: 'more' }, + ]); + assert.deepEqual(nextFollowUp, { kind: 'queued' }); + assert.deepEqual(currentFollowUp, { kind: 'queued' }); + assert.deepEqual( + calls.find((call) => call.name === 'sessions.queryMessageExecutions')?.args, + ['fork', ['message-next']], + ); }); }); 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 02022899b9..49ca61d927 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 @@ -25,6 +25,10 @@ import { RuntimeHostOperationError, RuntimeHostRequestInterruptedError, } from '@maka/runtime-host/client'; +import { + MESSAGE_QUEUE_MAX_ENTRIES, + type TurnMessageExecutionResolution, +} from '@maka/runtime-host/protocol'; import { type SessionChangedEvent, type SessionChangedReason, @@ -115,6 +119,7 @@ type RuntimeHostSessionExecutionClient = Pick< /** No Skill was named, so the Host resolved none. */ const EMPTY_SKILL_INVOCATION = { loaded: [], failed: [], receipts: [] } as const; +const DESKTOP_MESSAGE_QUERY_MAX_ENTRIES = 4_096; async function submitMessageWithReconnect( client: Pick, @@ -261,16 +266,62 @@ export function registerRuntimeHostSessionExecutionIpc( ipcMain.handle( 'sessions:queryCancelledMessages', async (_event, sessionId: string, messageIds: unknown) => { - if (!Array.isArray(messageIds)) throw new Error('Invalid Message identities'); - return deps.client.queryMessages({ sessionId, messageIds }); + const normalizedSessionId = requiredId(sessionId, 'Session'); + if ( + !Array.isArray(messageIds) + || messageIds.length > DESKTOP_MESSAGE_QUERY_MAX_ENTRIES + ) { + throw new Error('Invalid Message identities'); + } + const normalizedMessageIds = messageIds.map(requiredMessageId); + if (new Set(normalizedMessageIds).size !== normalizedMessageIds.length) { + throw new Error('Duplicate Message identities'); + } + // Keep the transport limit at the Runtime Host seam so renderer callers + // can query their complete optimistic projection as one operation. + const cancelledMessageIds: string[] = []; + for ( + let from = 0; + from < normalizedMessageIds.length; + from += MESSAGE_QUEUE_MAX_ENTRIES + ) { + const result = await deps.client.queryMessages({ + sessionId: normalizedSessionId, + messageIds: normalizedMessageIds.slice(from, from + MESSAGE_QUEUE_MAX_ENTRIES), + }); + cancelledMessageIds.push(...result.cancelledMessageIds); + } + if (new Set(cancelledMessageIds).size !== cancelledMessageIds.length) { + throw new Error('Duplicate cancelled Message identities'); + } + return { cancelledMessageIds }; }, ); ipcMain.handle( 'sessions:queryMessageExecutions', async (_event, sessionId: string, messageIds: unknown) => { - if (!Array.isArray(messageIds)) throw new Error('Invalid Message identities'); - return deps.client.queryMessageExecutions({ sessionId, messageIds }); + const normalizedSessionId = requiredId(sessionId, 'Session'); + if (!Array.isArray(messageIds) || messageIds.length > DESKTOP_MESSAGE_QUERY_MAX_ENTRIES) { + throw new Error('Invalid Message identities'); + } + const normalizedMessageIds = messageIds.map(requiredMessageId); + if (new Set(normalizedMessageIds).size !== normalizedMessageIds.length) { + throw new Error('Duplicate Message identities'); + } + const resolutions: TurnMessageExecutionResolution[] = []; + for ( + let from = 0; + from < normalizedMessageIds.length; + from += MESSAGE_QUEUE_MAX_ENTRIES + ) { + const result = await deps.client.queryMessageExecutions({ + sessionId: normalizedSessionId, + messageIds: normalizedMessageIds.slice(from, from + MESSAGE_QUEUE_MAX_ENTRIES), + }); + resolutions.push(...result.resolutions); + } + return { resolutions }; }, ); @@ -952,6 +1003,13 @@ function requiredId(value: unknown, label: string): string { return value; } +function requiredMessageId(value: unknown): string { + if (typeof value !== 'string' || !/^[A-Za-z0-9_-]{1,128}$/u.test(value)) { + throw new Error('Invalid Message identity'); + } + return value; +} + function requiredText(value: unknown, label: string): string { if ( typeof value !== "string" || diff --git a/apps/desktop/src/main/runtime-host-session-observer.ts b/apps/desktop/src/main/runtime-host-session-observer.ts index 011fdbe5da..718177ab3a 100644 --- a/apps/desktop/src/main/runtime-host-session-observer.ts +++ b/apps/desktop/src/main/runtime-host-session-observer.ts @@ -1569,7 +1569,7 @@ function replacementProjection( } terminalEvents.push(...stored); } else if (isTerminalTurn(root)) { - terminalEvents.push(...projector.seedTerminal(root)); + terminalEvents.push(...projector.seedActive(false), ...projector.seedTerminal(root)); } } if ( @@ -1577,7 +1577,7 @@ function replacementProjection( isTerminalTurn(root) && (!previousRoot || previousRoot.runId !== root.runId) ) { - terminalEvents.push(...projector.seedTerminal(root)); + terminalEvents.push(...projector.seedActive(false), ...projector.seedTerminal(root)); } return { terminalEvents, diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 85e6b0a70e..8c0338e469 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -250,6 +250,11 @@ export type DesktopSessionStopResult = | { kind: 'interrupted'; retractedMessageIds: string[] } | undefined; +/** Cancellation proof aggregated across every Runtime Host query batch. */ +export interface DesktopMessageCancellationQueryResult { + readonly cancelledMessageIds: readonly string[]; +} + export type DesktopReviseBeforeTurnInput = ReviseBeforeTurnInput & { /** Stable target identity for retrying one Desktop copy action. */ copyId: string; @@ -1140,7 +1145,7 @@ export interface MakaBridge { queryCancelledMessages( sessionId: string, messageIds: readonly string[], - ): Promise; + ): Promise; queryMessageExecutions( sessionId: string, messageIds: readonly string[], diff --git a/apps/desktop/src/renderer/app-shell-session-events.ts b/apps/desktop/src/renderer/app-shell-session-events.ts index 90335ba8c8..0357a1bf4c 100644 --- a/apps/desktop/src/renderer/app-shell-session-events.ts +++ b/apps/desktop/src/renderer/app-shell-session-events.ts @@ -33,6 +33,7 @@ import { type TransientUserMessageProjection, } from '@maka/ui'; import type { RefreshMessagesOptions } from './app-shell-chat-actions.js'; +import { deriveMessageQueueProjection } from './application/contracts/message-queue-projection.js'; import type { MessageQueueUiState } from './app-shell-session-ui-state.js'; import { isNoRealConnectionEvent, @@ -324,26 +325,11 @@ export function createAppShellSessionEventHandlers(options: { ); switch (event.type) { - case 'queue_update': + case 'queue_update': { + const queue = deriveMessageQueueProjection(event); projectQueuedTransientMessages?.( sessionId, - (event.steeringEntries ?? []).concat(event.followupEntries ?? []) - .filter((entry) => entry.state === 'queued') - .map((entry) => ({ - id: entry.messageId, - transientPlacement: entry.placement, - ...(entry.placement === 'current_turn' && { hostTurnId: event.turnId }), - ts: event.ts, - text: entry.content.displayText ?? entry.content.text, - ...(entry.content.attachments && { attachments: [...entry.content.attachments] }), - ...(entry.content.directoryReferences && { - directoryReferences: entry.content.directoryReferences, - }), - ...(entry.content.quotes && { quotes: [...entry.content.quotes] }), - ...(entry.content.inlineReferences && { - inlineReferences: [...entry.content.inlineReferences], - }), - })), + queue.transientMessages, ); setMessageQueueBySession?.((current) => { if (!event.steering.length && !event.followup.length) { @@ -356,14 +342,12 @@ export function createAppShellSessionEventHandlers(options: { ...current, [sessionId]: { queueRevision: event.queueRevision, - entries: [ - ...(event.steeringEntries ?? []).filter((entry) => entry.state === 'queued'), - ...(event.followupEntries ?? []), - ].map((entry) => structuredClone(entry)), + entries: queue.entries, }, }; }); break; + } case 'message_admission': if (event.outcome === 'retracted') removeTransientMessage?.(sessionId, event.messageId); break; diff --git a/apps/desktop/src/renderer/application/contracts/message-queue-projection.ts b/apps/desktop/src/renderer/application/contracts/message-queue-projection.ts new file mode 100644 index 0000000000..2857930a94 --- /dev/null +++ b/apps/desktop/src/renderer/application/contracts/message-queue-projection.ts @@ -0,0 +1,59 @@ +/* + * 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 type { + MessageQueueEntryProjection, + QueueUpdateEvent, +} from '@maka/core/events'; +import type { TransientUserMessageProjection } from '@maka/ui'; + +export interface MessageQueueProjection { + readonly entries: readonly MessageQueueEntryProjection[]; + readonly transientMessages: readonly TransientUserMessageProjection[]; +} + +/** One presentation contract for Host queue snapshots in every chat surface. */ +export function deriveMessageQueueProjection( + event: QueueUpdateEvent, +): MessageQueueProjection { + const entries = [ + ...(event.steeringEntries ?? []), + ...(event.followupEntries ?? []), + ] + .filter((entry) => entry.state === 'queued') + .map((entry) => structuredClone(entry)); + return { + entries, + transientMessages: entries.map((entry) => ({ + id: entry.messageId, + transientPlacement: entry.placement, + ...(entry.placement === 'current_turn' && { hostTurnId: event.turnId }), + ts: event.ts, + text: entry.content.displayText ?? entry.content.text, + ...(entry.content.attachments && { attachments: [...entry.content.attachments] }), + ...(entry.content.directoryReferences && { + directoryReferences: entry.content.directoryReferences, + }), + ...(entry.content.quotes && { quotes: [...entry.content.quotes] }), + ...(entry.content.inlineReferences && { + inlineReferences: [...entry.content.inlineReferences], + }), + })), + }; +} diff --git a/apps/desktop/src/renderer/transient-message-projection.ts b/apps/desktop/src/renderer/application/contracts/transient-message-projection.ts similarity index 100% rename from apps/desktop/src/renderer/transient-message-projection.ts rename to apps/desktop/src/renderer/application/contracts/transient-message-projection.ts diff --git a/apps/desktop/src/renderer/features/workbar/ports.ts b/apps/desktop/src/renderer/features/workbar/ports.ts index 4ee8b1cd65..08b200f620 100644 --- a/apps/desktop/src/renderer/features/workbar/ports.ts +++ b/apps/desktop/src/renderer/features/workbar/ports.ts @@ -18,6 +18,7 @@ */ import type { + MessageQueuePlacement, QuoteRef, SessionEvent, ShellRunUpdate, @@ -47,6 +48,7 @@ import type { Result } from '@maka/core/result'; import type { ContextCompactResult, ContextDiagnosticsResult, + TurnMessageExecutionQueryResult, } from '@maka/runtime-host/protocol'; import type { MergedUsageSummary } from '@maka/core/usage-ledger-merge'; import type { @@ -192,9 +194,9 @@ export type SideChatSendResult = | { 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 } +export type SideChatFollowUpResult = + | { kind: 'queued' } + | { kind: 'outcome_unknown' } | { kind: 'started'; turnId: string }; export type SideChatStopTarget = @@ -237,7 +239,25 @@ export interface SideChatSessionPort { sessionId: string, target?: SideChatStopTarget, ): Promise<{ kind: 'retracted'; messageId: string } | undefined>; - steer(sessionId: string, text: string, admissionId?: string): Promise; + submitFollowUp( + sessionId: string, + placement: MessageQueuePlacement, + text: string, + admissionId: string, + ): Promise; + queryMessageExecutions( + sessionId: string, + messageIds: readonly string[], + ): Promise; + retractQueueEntry(sessionId: string, entryId: string): Promise; + promoteQueueEntry(sessionId: string, entryId: string): Promise; + updateQueueEntry( + sessionId: string, + entryId: string, + expectedQueueRevision: number, + text: string, + ): Promise; + reorderQueueEntries(sessionId: string, entryIds: readonly string[]): Promise; setPermissionMode( sessionId: string, mode: PermissionMode, @@ -262,7 +282,8 @@ export interface SideChatSessionPort { subscribeEvents( sessionId: string, handler: (event: SessionEvent) => void, - onSeeded?: () => void, + /** Called after the initial observation seed and each reconnect seed. */ + onReady?: () => 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 11c1e45a56..5767a3e22b 100644 --- a/apps/desktop/src/renderer/features/workbar/testing.ts +++ b/apps/desktop/src/renderer/features/workbar/testing.ts @@ -137,9 +137,16 @@ export function createFakeWorkbarServices( }, send: async () => ({ ok: false, reason: 'not configured' }), stop: async () => undefined, - steer: async () => { - throw new Error('Fake sideChat.steer is not configured'); + submitFollowUp: async () => { + throw new Error('Fake sideChat.submitFollowUp is not configured'); }, + queryMessageExecutions: async (_sessionId, messageIds) => ({ + resolutions: messageIds.map((messageId) => ({ messageId, state: 'pending' as const })), + }), + retractQueueEntry: async () => undefined, + promoteQueueEntry: async () => undefined, + updateQueueEntry: async () => undefined, + reorderQueueEntries: async () => undefined, setPermissionMode: async () => { throw new Error('Fake sideChat.setPermissionMode is not configured'); }, diff --git a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-context-compaction.ts b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-context-compaction.ts index 527e258200..4822d11a0a 100644 --- a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-context-compaction.ts +++ b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-context-compaction.ts @@ -17,7 +17,7 @@ * under the License. */ -import type { ContextCompactionOutcome } from '@maka/core/events'; +import type { ContextCompactionOutcome, FollowUpMode } from '@maka/core/events'; import type { UiLocale } from '@maka/core/ui-locale'; import type { ContextCompactResult } from '@maka/runtime-host/protocol'; @@ -47,12 +47,18 @@ export function isExactCompactCommand(input: string): boolean { export function dispatchQuoteCompanionInput(input: { text: string; streaming: boolean; + followUpMode?: FollowUpMode; compact(): Promise; + queue(text: string): Promise; steer(text: string): Promise; send(): Promise; }): Promise { if (isExactCompactCommand(input.text)) return input.compact(); - if (input.streaming) return input.steer(input.text); + if (input.streaming) { + return input.followUpMode === 'steer' + ? input.steer(input.text) + : input.queue(input.text); + } return input.send(); } diff --git a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-panel.tsx b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-panel.tsx index d49fafb4a0..a5ae63ba47 100644 --- a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-panel.tsx +++ b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-panel.tsx @@ -339,11 +339,13 @@ export function QuoteCompanionPanel(props: { )} + onSend={(text, metadata) => dispatchQuoteCompanionInput({ text, streaming: companion.streaming, + followUpMode: metadata?.followUpMode, compact: companion.compact, + queue: companion.queue, steer: companion.steer, send: async () => { try { @@ -373,6 +375,12 @@ export function QuoteCompanionPanel(props: { hidden={Boolean(activeInteraction)} streaming={companion.streaming} processing={companion.processing} + queuedMessages={companion.queuedMessages} + queuedMessageRevision={companion.queuedMessageRevision} + onPromoteQueuedEntry={companion.promoteQueuedEntry} + onUpdateQueuedEntry={companion.updateQueuedEntry} + onDeleteQueuedEntry={companion.deleteQueuedEntry} + onReorderQueuedEntries={companion.reorderQueuedEntries} draftKey={draftKey} disabled={!companion.modelReady} onPickAttachments={pickAttachments} 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 1a65ba4024..e59e762fbc 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 @@ -34,6 +34,8 @@ import type { ClientCapabilityRequestEvent, ContextCompactionOutcome, FormRequestEvent, + MessageQueueEntryProjection, + MessageQueuePlacement, QuoteRef, SessionEvent, UserQuestionRequestEvent, @@ -63,7 +65,13 @@ import { type EnsureCompanionForkResult, } from './quote-companion-core.js'; import { isExactCompactCommand } from './quote-companion-context-compaction.js'; +import { deriveMessageQueueProjection } from '../../../../application/contracts/message-queue-projection.js'; import { mergeSettledMessages } from '../../../../settled-message-merge.js'; +import { + mergeTransientMessageProjection, + projectQueuedTransientMessages, + reconcileTransientMessages, +} from '../../../../application/contracts/transient-message-projection.js'; import { getDesktopConversationCopy } from '../../../../locales/conversation-copy.js'; import { snapshotCompanionQuotes, @@ -165,6 +173,9 @@ export interface UseQuoteCompanionResult { * before the durable transcript echoes them back. Reconciled away once the * durable message with the same id lands. Pass straight to `ChatView`. */ transientMessages: readonly TransientUserMessageProjection[]; + /** Host-authoritative pending steering and follow-up messages. */ + queuedMessages: readonly MessageQueueEntryProjection[]; + queuedMessageRevision: number | undefined; liveTurn: LiveTurnProjection | undefined; streaming: boolean; processing: boolean; @@ -188,6 +199,16 @@ export interface UseQuoteCompanionResult { send: (text: string, attachmentItems?: WorkbarIngestInput[]) => Promise; /** Insert text into the active companion turn at the next model step. */ steer: (text: string) => Promise; + /** Queue text for the next companion turn while the current turn continues. */ + queue: (text: string) => Promise; + promoteQueuedEntry: (entryId: string) => Promise; + updateQueuedEntry: ( + entryId: string, + expectedQueueRevision: number, + text: string, + ) => Promise; + deleteQueuedEntry: (entryId: string) => Promise; + reorderQueuedEntries: (entryIds: readonly string[]) => Promise; setPermissionMode: (mode: PermissionMode) => Promise; regenerate: (turnId: string) => Promise; stop: () => Promise; @@ -202,6 +223,19 @@ function requiredAssistantMessageId(projection: LiveTurnProjection | undefined): return [...(projection?.steps ?? [])].reverse().find((step) => step.text)?.stepId; } +function transcriptRecordsTerminalTurn( + messages: readonly StoredMessage[], + turnId: string, +): boolean { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (message?.type === 'turn_state' && message.turnId === turnId) { + return message.status !== 'running'; + } + } + return false; +} + /** * Companion for the quote side panel. On the first question it FORKS the main * session (`branchFromTurn` from the latest SETTLED turn) into a child that @@ -274,6 +308,8 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const compactionTurnIdRef = useRef(null); const pendingCompactionTerminalRef = useRef(null); const [allMessages, setAllMessages] = useState([]); + const allMessagesRef = useRef(allMessages); + allMessagesRef.current = allMessages; // Renderer-only user bubble shown the instant a send dispatches. The durable // transcript only echoes the just-sent question back mid-turn on a single // best-effort refresh (and otherwise not until the turn settles), so without @@ -283,6 +319,13 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const [pendingUserMessages, setPendingUserMessages] = useState< TransientUserMessageProjection[] >([]); + const pendingUserMessagesRef = useRef>( + new Map(), + ); + const [messageQueue, setMessageQueue] = useState<{ + readonly entries: readonly MessageQueueEntryProjection[]; + readonly queueRevision?: number; + }>({ entries: [] }); const [liveTurn, setLiveTurn] = useState(undefined); const liveTurnRef = useRef(liveTurn); liveTurnRef.current = liveTurn; @@ -362,16 +405,119 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan submitLockRef.current = locked; }, []); + const syncPendingUserMessages = useCallback(() => { + setPendingUserMessages([...pendingUserMessagesRef.current.values()]); + }, []); + + const addPendingUserMessage = useCallback((message: TransientUserMessageProjection) => { + const current = pendingUserMessagesRef.current.get(message.id); + pendingUserMessagesRef.current.set( + message.id, + current ? mergeTransientMessageProjection(current, message) : message, + ); + syncPendingUserMessages(); + }, [syncPendingUserMessages]); + + const reconcilePendingUserMessages = useCallback((durable: readonly StoredMessage[]) => { + reconcileTransientMessages(pendingUserMessagesRef.current, durable); + syncPendingUserMessages(); + }, [syncPendingUserMessages]); + + const mergeDurableMessages = useCallback((messages: readonly StoredMessage[]) => { + const next = mergeSettledMessages(allMessagesRef.current, messages); + allMessagesRef.current = next; + setAllMessages(next); + reconcilePendingUserMessages( + next.filter( + (message) => message.turnId !== undefined && ownTurnIdsRef.current.has(message.turnId), + ), + ); + }, [reconcilePendingUserMessages]); + // Retire the optimistic bubble for a message id. Called when a send is - // retracted/abandoned; the success path retires it implicitly by reconciling - // against the durable transcript (see the `transientMessages` derivation). + // retracted/abandoned; the success path retires it through the shared + // durable-transient reconciliation rule. const dropOptimisticUserMessage = useCallback((messageId: string) => { - setPendingUserMessages((current) => { - const next = current.filter((message) => message.id !== messageId); - return next.length === current.length ? current : next; + if (!pendingUserMessagesRef.current.delete(messageId)) return; + syncPendingUserMessages(); + }, [syncPendingUserMessages]); + + const dropQueuedMessage = useCallback((messageId: string) => { + setMessageQueue((current) => { + const entries = current.entries.filter((entry) => entry.messageId !== messageId); + return entries.length === current.entries.length ? current : { ...current, entries }; }); }, []); + const recordOwnedTurn = useCallback((turnId: string) => { + hasContentRef.current = true; + setHasContent(true); + ownTurnIdsRef.current.add(turnId); + setOwnTurnTick((tick) => tick + 1); + reconcilePendingUserMessages( + allMessagesRef.current.filter( + (message) => message.turnId !== undefined && ownTurnIdsRef.current.has(message.turnId), + ), + ); + }, [reconcilePendingUserMessages]); + + const adoptOwnedTurn = useCallback((turnId: string) => { + recordOwnedTurn(turnId); + activeTurnIdRef.current = turnId; + setLiveTurn((current) => + current?.turnId === turnId ? current : armLiveTurn(turnId), + ); + }, [recordOwnedTurn]); + + const projectMessageQueue = useCallback( + (event: Extract) => { + const queue = deriveMessageQueueProjection(event); + setMessageQueue({ entries: queue.entries, queueRevision: event.queueRevision }); + projectQueuedTransientMessages(pendingUserMessagesRef.current, queue.transientMessages); + syncPendingUserMessages(); + }, + [syncPendingUserMessages], + ); + + const reconcilePendingMessageExecutions = useCallback(async (forkId: string) => { + const messageIds = [...pendingUserMessagesRef.current.keys()]; + if (messageIds.length === 0) return; + try { + const { resolutions } = await sideChat.queryMessageExecutions(forkId, messageIds); + if (!mountedRef.current || companionIdRef.current !== forkId) return; + const cancelled = new Set(); + let ownershipChanged = false; + for (const resolution of resolutions) { + if (resolution.state === 'cancelled') { + cancelled.add(resolution.messageId); + } else if (resolution.state === 'owned') { + const previousSize = ownTurnIdsRef.current.size; + ownTurnIdsRef.current.add(resolution.turnId); + ownershipChanged ||= ownTurnIdsRef.current.size !== previousSize; + } + } + if (ownershipChanged) { + hasContentRef.current = true; + setHasContent(true); + setOwnTurnTick((tick) => tick + 1); + } + for (const messageId of cancelled) pendingUserMessagesRef.current.delete(messageId); + const renderable = allMessagesRef.current.filter( + (message) => message.turnId !== undefined && ownTurnIdsRef.current.has(message.turnId), + ); + reconcileTransientMessages(pendingUserMessagesRef.current, renderable); + syncPendingUserMessages(); + if (cancelled.size > 0) { + setMessageQueue((current) => ({ + ...current, + entries: current.entries.filter((entry) => !cancelled.has(entry.messageId)), + })); + } + } catch { + // A failed proof query leaves presentation intact until canonical proof arrives. + } + }, [mountedRef, sideChat, syncPendingUserMessages]); + const applyOwnedEvent = useCallback( (forkId: string, event: SessionEvent) => { const terminal: PendingCompactionTerminal | undefined = @@ -422,9 +568,14 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan : {}), }) .then(({ messages: next }) => { - if (!mountedRef.current || activeTurnIdRef.current !== settledTurnId) return; - setAllMessages((current) => mergeSettledMessages(current, next)); - setLiveTurn((prev) => (prev ? reconcileTerminalLiveTurn(prev, next) : prev)); + if (!mountedRef.current) return; + mergeDurableMessages(next); + if (activeTurnIdRef.current !== settledTurnId) return; + setLiveTurn((prev) => + prev?.turnId === settledTurnId + ? reconcileTerminalLiveTurn(prev, next) + : prev, + ); activeTurnIdRef.current = null; stopRequestRef.current = null; }) @@ -439,7 +590,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan }); } }, - [mountedRef, sideChat], + [mergeDurableMessages, mountedRef, sideChat], ); const bindAdmittedTurn = useCallback( @@ -454,13 +605,10 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan // Host admission is the durable-content boundary. Even if a concurrent // Stop interrupts the Run before send() settles, this fork now owns a // persisted user message and must never be replaced as an empty copy. - hasContentRef.current = true; - setHasContent(true); activeTurnIdRef.current = turnId; - ownTurnIdsRef.current.add(turnId); + recordOwnedTurn(turnId); admission.consumeOnAdmission?.(); setError(null); - setOwnTurnTick((tick) => tick + 1); if (!(options.preserveLiveTurn && liveTurnRef.current?.turnId === turnId)) { setLiveTurn(armLiveTurn(turnId)); } @@ -468,7 +616,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan if (event.turnId === turnId) applyOwnedEvent(forkId, event); } }, - [applyOwnedEvent, setPendingAdmission], + [applyOwnedEvent, recordOwnedTurn, setPendingAdmission], ); // A Message whose admission answer was lost is still reconcilable: the Host @@ -530,6 +678,73 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan [bindAdmittedTurn, releaseAdmission], ); + const reconcileStartedFollowUpTurn = useCallback(async ( + forkId: string, + turnId: string, + predecessorTurnId: string | null, + ): Promise => { + // Fence the subscription before the canonical read. This does not trigger + // a render by itself, but it lets a terminal event arriving during the read + // settle the returned Turn instead of being filtered as not-yet-owned. + // A still-later follow-up may already have activated another Turn, though; + // a delayed receipt for this Turn must not replace that newer authority. + if ( + activeTurnIdRef.current === null + || activeTurnIdRef.current === predecessorTurnId + || activeTurnIdRef.current === turnId + ) { + activeTurnIdRef.current = turnId; + } + const retainOrArmTurn = () => { + if ( + activeTurnIdRef.current === turnId + && !settlingTurnIdsRef.current.has(turnId) + ) { + adoptOwnedTurn(turnId); + } else { + recordOwnedTurn(turnId); + } + return true; + }; + let messages: StoredMessage[]; + try { + ({ messages } = await sideChat.readSettledMessages(forkId)); + if (!mountedRef.current || companionIdRef.current !== forkId) { + if (activeTurnIdRef.current === turnId) activeTurnIdRef.current = null; + return false; + } + } catch { + if (!mountedRef.current || companionIdRef.current !== forkId) { + if (activeTurnIdRef.current === turnId) activeTurnIdRef.current = null; + return false; + } + // Without canonical terminal proof, the Host's started receipt remains + // the best available authority and preserves the existing live path. + return retainOrArmTurn(); + } + + if ( + !transcriptRecordsTerminalTurn(messages, turnId) + && !transcriptRecordsTerminalTurn(allMessagesRef.current, turnId) + ) { + return retainOrArmTurn(); + } + + // A reconnect retry can replay the original `turn_started` receipt after + // the Turn's text and terminal event have already passed this renderer or + // left the bounded transcript tail. Retained terminal state is authoritative: + // keep the durable transcript without re-arming a Run that has no future + // terminal event to settle it. + if (activeTurnIdRef.current === turnId) { + activeTurnIdRef.current = null; + stopRequestRef.current = null; + } + recordOwnedTurn(turnId); + mergeDurableMessages(messages); + setLiveTurn((current) => current?.turnId === turnId ? undefined : current); + return true; + }, [adoptOwnedTurn, mergeDurableMessages, mountedRef, recordOwnedTurn, sideChat]); + // 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). @@ -555,17 +770,57 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan void sideChat.readSettledMessages(forkId) .then(({ messages }) => { if (mountedRef.current) { - setAllMessages((current) => mergeSettledMessages(current, messages)); + mergeDurableMessages(messages); } }) .catch(() => { if (mountedRef.current) setError(copyRef.current.errors.settlementFailed); }); + const observationSeeded = () => { + resolveReady(); + void sideChat.readSettledMessages(forkId) + .then(({ messages }) => { + if (!mountedRef.current || companionIdRef.current !== forkId) return; + mergeDurableMessages(messages); + void reconcilePendingMessageExecutions(forkId); + }) + .catch(() => { + void reconcilePendingMessageExecutions(forkId); + }); + }; const unsubscribe = sideChat.subscribeEvents( forkId, (event: SessionEvent) => { if (!mountedRef.current) return; + if (event.type === 'queue_update') { + projectMessageQueue(event); + return; + } const admission = pendingAdmissionRef.current; + if (event.type === 'steering_message') { + dropOptimisticUserMessage(event.messageId); + dropQueuedMessage(event.messageId); + if (event.turnId !== activeTurnIdRef.current) adoptOwnedTurn(event.turnId); + applyOwnedEvent(forkId, event); + return; + } else if (event.type === 'message_admission' && event.outcome === 'retracted') { + dropOptimisticUserMessage(event.messageId); + dropQueuedMessage(event.messageId); + if (admission?.messageId === event.messageId) { + admission.events.push(event); + resolveAdmission(forkId, admission, admission.messageId, true); + } + return; + } else if (event.type === 'message_admission' && event.outcome === 'admitted') { + dropQueuedMessage(event.messageId); + if (admission?.messageId === event.messageId) { + admission.events.push(event); + resolveAdmission(forkId, admission, admission.messageId, true); + } else { + adoptOwnedTurn(event.turnId); + } + return; + } if (event.type === 'error' && event.recoverable) { if (admission) { // Observation failure does not prove whether Host admitted the @@ -599,7 +854,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan } applyOwnedEvent(forkId, event); }, - resolveReady, + observationSeeded, rejectReady, ); let disposed = false; @@ -612,7 +867,19 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan resolveReady(); }; return ready; - }, [applyOwnedEvent, mountedRef, resolveAdmission, sideChat]); + }, [ + applyOwnedEvent, + adoptOwnedTurn, + dropOptimisticUserMessage, + dropQueuedMessage, + mountedRef, + mergeDurableMessages, + projectMessageQueue, + reconcilePendingMessageExecutions, + reconcileUnknownAdmission, + resolveAdmission, + sideChat, + ]); const commitFork = useCallback( (session: SessionSummary) => { @@ -666,7 +933,11 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan companionRef.current = undefined; clearPermissionModeIntent(existing.id); setCompanion(undefined); + allMessagesRef.current = []; setAllMessages([]); + pendingUserMessagesRef.current.clear(); + setPendingUserMessages([]); + setMessageQueue({ entries: [] }); onForkVisibilityChangeRef.current?.({ type: 'cleanup-succeeded', sessionId: existing.id, @@ -888,16 +1159,14 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan events: [], consumeOnAdmission: () => onQuotesConsumed(quoteSnapshot), }; - setPendingUserMessages((current) => [ - ...current.filter((message) => message.id !== turnId), - { - id: turnId, - text: trimmed, - ts: Date.now(), - transientPlacement: 'current_turn', - ...(quoteSnapshot.quotes.length > 0 ? { quotes: quoteSnapshot.quotes } : {}), - }, - ]); + const optimisticMessage: TransientUserMessageProjection = { + id: turnId, + text: trimmed, + ts: Date.now(), + transientPlacement: 'current_turn', + ...(quoteSnapshot.quotes.length > 0 ? { quotes: quoteSnapshot.quotes } : {}), + }; + addPendingUserMessage(optimisticMessage); // Setup can still fail before the send is in flight (fork unavailable, // fail-closed permission write, or a lost subscription). Retire the // optimistic bubble and release the lock so a failed first send never @@ -1018,7 +1287,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan }) .then(({ messages: next }) => { if (mountedRef.current) { - setAllMessages((current) => mergeSettledMessages(current, next)); + mergeDurableMessages(next); } }) .catch(() => {}); @@ -1062,7 +1331,9 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan sideChat, bindAdmittedTurn, compact, + addPendingUserMessage, dropOptimisticUserMessage, + mergeDurableMessages, releaseAdmission, resolveAdmission, setPendingAdmission, @@ -1120,7 +1391,10 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan } }, [releaseAdmission, resolveAdmission, sideChat]); - const steer = useCallback(async (text: string): Promise => { + const submitFollowUp = useCallback(async ( + text: string, + placement: MessageQueuePlacement, + ): Promise => { const id = companionIdRef.current; const trimmed = text.trim(); if ( @@ -1128,7 +1402,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan !id || !trimmed || !turnInFlight || - pendingAdmissionRef.current + (placement === 'current_turn' && pendingAdmissionRef.current !== null) ) { return false; } @@ -1137,35 +1411,71 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan messageId: admissionId, events: [], }; - setPendingAdmission(admission); + const predecessorTurnId = activeTurnIdRef.current; + const optimisticMessage: TransientUserMessageProjection = { + id: admissionId, + text: trimmed, + ts: Date.now(), + transientPlacement: placement, + ...(placement === 'current_turn' && activeTurnIdRef.current + ? { hostTurnId: activeTurnIdRef.current } + : {}), + }; + addPendingUserMessage(optimisticMessage); + if (placement === 'current_turn') setPendingAdmission(admission); try { - const outcome = await sideChat.steer(id, trimmed, admissionId); + const outcome = await sideChat.submitFollowUp(id, placement, trimmed, admissionId); if (!mountedRef.current) return false; - if ((await admission.stopPromise) === 'confirmed') return false; + if (placement === 'current_turn' && (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') { + if (placement === 'current_turn') { + bindAdmittedTurn(id, outcome.turnId, { preserveLiveTurn: true }); + } else { + // The active Turn can settle between the local streaming check and + // Host admission. In that race a nominal next-turn follow-up starts + // immediately. Reconcile first because a reconnect can replay this + // receipt after the Host-named Turn has already settled. + if (!(await reconcileStartedFollowUpTurn(id, outcome.turnId, predecessorTurnId))) { + return false; + } + } + } else if (resolveAdmission(id, admission, admissionId, true)?.kind === 'retracted') { return false; + } else if ( + outcome.kind === 'queued' && + placement === 'current_turn' && + pendingAdmissionRef.current === admission + ) { + // A queued follow-up no longer owns the Composer's single in-flight + // admission slot. Its optimistic row and the Host queue projection + // remain until delivery/retraction, while later follow-ups may queue too. + setPendingAdmission(null); } setError(null); return true; } catch { if (mountedRef.current) { - if (pendingAdmissionRef.current === admission) { + if (placement === 'current_turn' && pendingAdmissionRef.current === admission) { releaseAdmission(admission, copyRef.current.errors.sendFailed); } else if ( admissionOutcomeForMessage(admission.events, admission.messageId)?.kind !== 'retracted' ) { + dropOptimisticUserMessage(admission.messageId); setError(copyRef.current.errors.sendFailed); } } return false; } }, [ + addPendingUserMessage, bindAdmittedTurn, + reconcileStartedFollowUpTurn, + dropOptimisticUserMessage, mountedRef, releaseAdmission, resolveAdmission, @@ -1174,6 +1484,54 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan turnInFlight, ]); + const steer = useCallback( + (text: string) => submitFollowUp(text, 'current_turn'), + [submitFollowUp], + ); + const queue = useCallback( + (text: string) => submitFollowUp(text, 'next_turn'), + [submitFollowUp], + ); + + const runQueueEntryAction = useCallback( + async (action: (sessionId: string) => Promise): Promise => { + const id = companionIdRef.current; + if (!id) return; + try { + await action(id); + } catch (error) { + if (mountedRef.current) setError(copyRef.current.errors.respondFailed); + throw error; + } + }, + [mountedRef], + ); + + const promoteQueuedEntry = useCallback( + (entryId: string) => runQueueEntryAction((id) => sideChat.promoteQueueEntry(id, entryId)), + [runQueueEntryAction, sideChat], + ); + const updateQueuedEntry = useCallback( + (entryId: string, expectedQueueRevision: number, text: string) => + runQueueEntryAction((id) => + sideChat.updateQueueEntry(id, entryId, expectedQueueRevision, text), + ), + [runQueueEntryAction, sideChat], + ); + const deleteQueuedEntry = useCallback( + async (entryId: string): Promise => { + const messageId = messageQueue.entries.find((entry) => entry.entryId === entryId)?.messageId; + await runQueueEntryAction((id) => sideChat.retractQueueEntry(id, entryId)); + if (messageId) dropOptimisticUserMessage(messageId); + }, + [dropOptimisticUserMessage, messageQueue.entries, runQueueEntryAction, sideChat], + ); + const reorderQueuedEntries = useCallback( + (entryIds: readonly string[]) => + runQueueEntryAction((id) => sideChat.reorderQueueEntries(id, entryIds)), + [runQueueEntryAction, sideChat], + ); + const setPermissionMode = useCallback( (mode: PermissionMode): Promise => { const id = companionIdRef.current; @@ -1290,15 +1648,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const messages = allMessages.filter( (message) => message.turnId !== undefined && ownTurnIdsRef.current.has(message.turnId), ); - // Drop the optimistic bubble only once its durable twin will actually RENDER, - // i.e. it is in `messages` (own-turn filtered) — not merely settled into - // `allMessages`. Building this from `allMessages` could retire the transient on - // an `outcome_unknown` settle while the durable message is still filtered out of - // the render, blinking the question away until `reconcileUnknownAdmission` binds. - const durableMessageIds = new Set(messages.map((message) => message.id)); - const transientMessages = pendingUserMessages.filter( - (message) => !durableMessageIds.has(message.id), - ); + const transientMessages = pendingUserMessages; // Inherited model (read-only): the fork's once created, else the source's. const activeModel = companion ? { llmConnectionSlug: companion.llmConnectionSlug, model: companion.model } @@ -1330,6 +1680,8 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan hasContent, messages, transientMessages, + queuedMessages: messageQueue.entries, + queuedMessageRevision: messageQueue.queueRevision, liveTurn, streaming, processing, @@ -1345,6 +1697,11 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan compact, send, steer, + queue, + promoteQueuedEntry, + updateQueuedEntry, + deleteQueuedEntry, + reorderQueuedEntries, setPermissionMode, regenerate, stop, 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 39a6e490c5..29e2aac6da 100644 --- a/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts +++ b/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts @@ -47,6 +47,27 @@ export function createDesktopWorkbarServices( bridge: DesktopWorkbarBridge = window.maka, dependencies: DesktopWorkbarServiceDependencies = DEFAULT_DEPENDENCIES, ): WorkbarServices { + const submitSideChatFollowUp: WorkbarServices['sideChat']['submitFollowUp'] = async ( + sessionId, + placement, + text, + admissionId, + ) => { + const result = await bridge.sessions.submitMessage(sessionId, placement, { + messageId: admissionId, + text, + }); + if (!result.ok) { + if (result.reason === 'outcome_unknown') { + return { kind: 'outcome_unknown' }; + } + throw new Error('Runtime Host refused the follow-up Message'); + } + return result.disposition === 'turn_started' && result.turnId + ? { kind: 'started', turnId: result.turnId } + : { kind: 'queued' }; + }; + return { review: { read: (input) => bridge.gitReview.read(input), @@ -114,27 +135,17 @@ export function createDesktopWorkbarServices( ); return result?.kind === 'retracted' ? result : undefined; }, - // Steering is a Message placed at the current Turn's boundary, so it - // rides the one admission channel. Runtime Host names the outcome; this - // adapter only renames it for the Side Conversation port. - steer: async (sessionId, text, admissionId) => { - const messageId = admissionId ?? crypto.randomUUID(); - const result = await bridge.sessions.submitMessage(sessionId, 'current_turn', { - messageId, - text, - }); - if (!result.ok) { - if (result.reason === 'outcome_unknown') { - return { kind: 'outcome_unknown', messageId }; - } - // No Turn opened and nothing was queued; the caller surfaces it as a - // failed send rather than waiting for an admission that never lands. - throw new Error('Runtime Host refused the steering Message'); - } - return result.disposition === 'turn_started' && result.turnId - ? { kind: 'started', turnId: result.turnId } - : { kind: 'queued', messageId }; - }, + submitFollowUp: submitSideChatFollowUp, + queryMessageExecutions: (sessionId, messageIds) => + bridge.sessions.queryMessageExecutions(sessionId, messageIds), + retractQueueEntry: (sessionId, entryId) => + bridge.sessions.retractQueueEntry(sessionId, entryId), + promoteQueueEntry: (sessionId, entryId) => + bridge.sessions.promoteQueueEntry(sessionId, entryId), + updateQueueEntry: (sessionId, entryId, expectedQueueRevision, text) => + bridge.sessions.updateQueueEntry(sessionId, entryId, expectedQueueRevision, text), + reorderQueueEntries: (sessionId, entryIds) => + bridge.sessions.reorderQueueEntries(sessionId, entryIds), setPermissionMode: (sessionId, mode) => bridge.sessions.setPermissionMode(sessionId, mode), regenerateTurn: (sessionId, input) => @@ -147,8 +158,16 @@ export function createDesktopWorkbarServices( bridge.sessions.respondToUserQuestion(sessionId, response), respondToUserForm: (sessionId, response) => bridge.sessions.respondToUserForm(sessionId, response), - subscribeEvents: (sessionId, handler, onSeeded, onSeedError) => - bridge.sessions.subscribeEvents(sessionId, handler, onSeeded, undefined, onSeedError), + subscribeEvents: (sessionId, handler, onReady, onSeedError) => + bridge.sessions.subscribeEvents( + sessionId, + handler, + onReady, + (phase) => { + if (phase === 'ready') onReady?.(); + }, + onSeedError, + ), subscribeSessionChanges: (handler) => bridge.sessions.subscribeChanges(handler), }, }; diff --git a/apps/desktop/src/renderer/session-workspace-actions.ts b/apps/desktop/src/renderer/session-workspace-actions.ts index 44412d10ab..2efc612d18 100644 --- a/apps/desktop/src/renderer/session-workspace-actions.ts +++ b/apps/desktop/src/renderer/session-workspace-actions.ts @@ -35,14 +35,13 @@ import type { StoredMessage } from '@maka/core/session'; import type { TransientUserMessageProjection } from '@maka/ui'; -import { MESSAGE_QUEUE_MAX_ENTRIES } from '@maka/runtime-host/protocol'; import { clearNewTaskReloadIntent, markNewTaskReloadIntent } from './new-task-reload-intent.js'; import type { DesktopTranscriptRangeController } from './desktop-transcript-range-store.js'; import { mergeTransientMessageProjection, projectQueuedTransientMessages as applyQueuedTransientProjection, reconcileTransientMessages, -} from './transient-message-projection.js'; +} from './application/contracts/transient-message-projection.js'; type RefBox = { current: T }; @@ -164,21 +163,14 @@ export function createSessionWorkspaceActions(deps: { const pending = transientMessagesBySessionRef.current.get(sessionId); if (!pending || pending.size === 0) return; try { - // A legal Host queue already fills the protocol's per-query cap, and an - // unreconciled root Message sits beside it, so asking about every row at - // once fails the whole proof and retires nothing. const messageIds = [...pending.keys()]; - const cancelled: string[] = []; - for (let from = 0; from < messageIds.length; from += MESSAGE_QUEUE_MAX_ENTRIES) { - const result = await window.maka.sessions.queryCancelledMessages( - sessionId, - messageIds.slice(from, from + MESSAGE_QUEUE_MAX_ENTRIES), - ); - cancelled.push(...result.cancelledMessageIds); - } + const { cancelledMessageIds } = await window.maka.sessions.queryCancelledMessages( + sessionId, + messageIds, + ); const current = transientMessagesBySessionRef.current.get(sessionId); if (!current) return; - for (const messageId of cancelled) current.delete(messageId); + for (const messageId of cancelledMessageIds) current.delete(messageId); if (current.size === 0) transientMessagesBySessionRef.current.delete(sessionId); reprojectActiveTransients(sessionId); } catch { diff --git a/apps/desktop/stories/session-workbar.stories.tsx b/apps/desktop/stories/session-workbar.stories.tsx index b228c31896..110de5570a 100644 --- a/apps/desktop/stories/session-workbar.stories.tsx +++ b/apps/desktop/stories/session-workbar.stories.tsx @@ -891,7 +891,14 @@ function bridge(options: { }), send: async () => ({ ok: true, turnId: 'story-side-chat-turn' }), stop: async () => undefined, - steer: async () => ({ kind: 'started', turnId: 'story-side-chat-turn' }), + submitFollowUp: async () => ({ kind: 'started', turnId: 'story-side-chat-turn' }), + queryMessageExecutions: async (_sessionId, messageIds) => ({ + resolutions: messageIds.map((messageId) => ({ messageId, state: 'pending' as const })), + }), + retractQueueEntry: async () => undefined, + promoteQueueEntry: async () => undefined, + updateQueueEntry: async () => undefined, + reorderQueueEntries: async () => undefined, setPermissionMode: async (_sessionId, mode) => ({ ...SIDE_CHAT_SESSION, permissionMode: mode, diff --git a/docs/side-conversation.md b/docs/side-conversation.md index 4fa8f25810..71e0f426e4 100644 --- a/docs/side-conversation.md +++ b/docs/side-conversation.md @@ -50,8 +50,9 @@ The generic side-conversation entry extends that foundation: - only instructions submitted in the side chat are active; explicit side-chat actions may use the inherited permission profile, and the permission can be changed from the side Composer; -- a running side turn accepts a Steer message at the next model step while Stop - remains available; +- while a side turn is running, Enter queues a follow-up for the next turn and + Shift+Enter steers the active turn at its next model step; both appear + immediately, while Stop remains available; - the side Composer shares `/` Skill discovery, `@` file references, files, quotes, and draft ownership with the main Composer; - settled side answers expose Copy, Info, and Regenerate without navigating the @@ -305,8 +306,8 @@ Maka now has the first usable slice of the same architecture: - multiple numbered side-chat tabs with independent drafts, forks, streams, and quote queues; - the same Composer shell as the main conversation, including a functional - attachment menu, Skill and file mentions, inherited permission menu, and - mid-turn Steer submission; + attachment menu, Skill and file mentions, inherited permission menu, + Enter-to-queue / Shift+Enter-to-Steer routing, and Host-backed queue controls; - the same answer metadata surface for Copy, Info, and Regenerate, with Branch withheld because it would navigate outside the temporary side-tab lifecycle; - no content-area close action: Side Chat lifetime belongs exclusively to tab