diff --git a/apps/desktop/e2e/workhub-reconstruction.spec.ts b/apps/desktop/e2e/workhub-reconstruction.spec.ts index 938df54eb7..0b29525c03 100644 --- a/apps/desktop/e2e/workhub-reconstruction.spec.ts +++ b/apps/desktop/e2e/workhub-reconstruction.spec.ts @@ -19,7 +19,7 @@ import { COMPOSER_INPUT, ensureSidebarExpanded, expect, test } from './fixtures'; -test('WorkHub rebuilds Session conversation after navigating away and back', async ({ +test('WorkHub rebuilds delegated execution feedback after navigating away and back', async ({ window: page, }) => { const initialPrompt = '检查支付回调重复投递时的幂等性'; @@ -64,6 +64,10 @@ test('WorkHub rebuilds Session conversation after navigating away and back', asy hasText: routedPrompt, }), ).toBeVisible(); + await expect( + page.locator('.workhub-projected-turn', { hasText: routedPrompt }) + .locator('.workhub-submitted-state'), + ).toHaveText('进行中'); }); test('WorkHub defers destructive correction until linked delegation exists', async ({ 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 b95338d162..8bdf462bed 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 @@ -630,6 +630,45 @@ test('returns Host-owned cancellation proof to the renderer', async () => { ); }); +test('returns Host-owned Message execution resolutions to the renderer', async () => { + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + queryMessageExecutions: async (input) => ({ + resolutions: input.messageIds.map((messageId) => messageId === 'message-cancelled' + ? { messageId, state: 'cancelled' as const } + : { + messageId, + state: 'owned' as const, + turnId: 'successor-turn', + runId: 'successor-run', + }), + }), + }), + }, + ipc, + ); + + assert.deepEqual( + await ipc.invoke('sessions:queryMessageExecutions', 'session-1', [ + 'message-delegated', + 'message-cancelled', + ]), + { + resolutions: [ + { + messageId: 'message-delegated', + state: 'owned', + turnId: 'successor-turn', + runId: 'successor-run', + }, + { messageId: 'message-cancelled', state: 'cancelled' }, + ], + }, + ); +}); + test('submits a slash Skill message and reports the Host Skill outcome', async () => { const submits: unknown[] = []; const ipc = ipcHarness(); @@ -1502,6 +1541,7 @@ function executionClient(overrides: Partial): ExecutionClient { interruptTurn: unavailable, listSessionTurnLandmarks: unavailable, listSessionTurns: unavailable, + queryMessageExecutions: unavailable, queryMessages: unavailable, queryTurnResume: unavailable, readExecutionBoundary: unavailable, diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index ff438b0540..943516eaa0 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -27,6 +27,7 @@ import { WORKHUB_ROUTING_STRATEGY_ID, type WorkHubSessionFacts, type WorkHubSessionPort, + type WorkHubCoordinationTurn, } from '../../renderer/workhub-controller.js'; const appShellUrl = [ @@ -76,6 +77,8 @@ function port(sessions: WorkHubSessionFacts[]): WorkHubSessionPort { return { list: async () => sessions, recentTurns: async () => [], + delegationFeedback: async (references) => + references.map(({ delegationId }) => ({ delegationId, state: 'accepted' })), routingEvidence: async () => [], create: async () => { throw new Error('create is not used by this read test'); @@ -90,6 +93,122 @@ function port(sessions: WorkHubSessionFacts[]): WorkHubSessionPort { }; } +function coordinationAssignmentTurn(): WorkHubCoordinationTurn { + return { + messageId: 'assignment-1', + turnId: 'action-1', + text: 'Continue payments', + state: 'completed', + assignment: { + delegationId: 'delegation-1', + targetSessionId: 'payment', + targetSessionName: 'Payments', + targetMessageId: 'payment-message', + targetTurnId: 'payment-turn', + feedbackState: 'accepted', + }, + updatedAt: 10, + }; +} + +test('conversation acknowledges a durable assignment before projecting target execution', async () => { + const sessions = port([session('payment')]); + let onSessionChanged: (() => void) | undefined; + let feedbackState: 'completed' | 'waiting_for_user' = 'completed'; + sessions.subscribe = (handler) => { + onSessionChanged = handler; + return () => { + onSessionChanged = undefined; + }; + }; + sessions.delegationFeedback = async (references) => + references.map(({ delegationId }) => ({ delegationId, state: feedbackState })); + const assignment = coordinationAssignmentTurn(); + const snapshots: string[] = []; + const controller = createGatedWorkHubController({ + sessions, + coordination: { + open: async (handler) => { + handler([assignment]); + return { close: async () => undefined }; + }, + answer: async (input) => ({ turnId: input.turnId }), + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [] }), + act: async () => ({ disposition: 'answer_here', coordinationTurnId: 'unused' }), + }, + }); + + const handle = await controller.openConversation((turns) => { + snapshots.push(turns[0]?.assignment?.feedbackState ?? 'missing'); + }, () => undefined); + await Promise.resolve(); + + assert.deepEqual(snapshots.slice(0, 2), ['accepted', 'completed']); + + feedbackState = 'waiting_for_user'; + onSessionChanged?.(); + await Promise.resolve(); + await Promise.resolve(); + assert.equal(snapshots.at(-1), 'waiting_for_user'); + + await handle.close(); +}); + +test('conversation feedback never lets an older refresh overwrite newer target state', async () => { + const sessions = port([session('payment')]); + let onSessionChanged: (() => void) | undefined; + sessions.subscribe = (handler) => { + onSessionChanged = handler; + return () => undefined; + }; + type Feedback = Awaited>; + const pending: Array<{ + references: Parameters[0]; + resolve(feedback: Feedback): void; + }> = []; + sessions.delegationFeedback = (references) => + new Promise((resolve) => pending.push({ references, resolve })); + const snapshots: string[] = []; + const controller = createGatedWorkHubController({ + sessions, + coordination: { + open: async (handler) => { + handler([coordinationAssignmentTurn()]); + return { close: async () => undefined }; + }, + answer: async (input) => ({ turnId: input.turnId }), + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'b'.repeat(64)}`, candidates: [] }), + act: async () => ({ disposition: 'answer_here', coordinationTurnId: 'unused' }), + }, + }); + + const handle = await controller.openConversation((turns) => { + snapshots.push(turns[0]?.assignment?.feedbackState ?? 'missing'); + }, () => undefined); + assert.equal(pending.length, 1); + onSessionChanged?.(); + assert.equal(pending.length, 2); + + pending[1]!.resolve(pending[1]!.references.map(({ delegationId }) => ({ + delegationId, + state: 'completed', + }))); + await Promise.resolve(); + await Promise.resolve(); + pending[0]!.resolve(pending[0]!.references.map(({ delegationId }) => ({ + delegationId, + state: 'failed', + }))); + await Promise.resolve(); + await Promise.resolve(); + + assert.equal(snapshots.at(-1), 'completed'); + assert.equal(snapshots.includes('failed'), false); + await handle.close(); +}); + test('read exposes existing ordinary Sessions as factual Work summaries', async () => { const controller = createWorkHubController({ sessions: port([ diff --git a/apps/desktop/src/main/__tests__/workhub-coordination-host-scope.test.ts b/apps/desktop/src/main/__tests__/workhub-coordination-host-scope.test.ts index 0a4e014fe2..5b3e18ac37 100644 --- a/apps/desktop/src/main/__tests__/workhub-coordination-host-scope.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-coordination-host-scope.test.ts @@ -44,6 +44,7 @@ test('WorkHub candidates follow the resolved Coordination Session Host only', as completeHostIds: ['host-a', 'host-b'], }), listTurns: async () => [], + queryMessageExecutions: async () => ({ resolutions: [] }), create: async () => { throw new Error('unscoped create must not be used'); }, diff --git a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts index 8f3eb7d0c2..a92dc154c0 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -56,6 +56,14 @@ const unusedTranscripts = { }, }; +const noMessageExecutions = async () => ({ + resolutions: [] as Array< + | { messageId: string; state: 'pending' } + | { messageId: string; state: 'cancelled' } + | { messageId: string; state: 'owned'; turnId: string; runId: string } + >, +}); + function transcriptsWith(messages: readonly StoredMessage[]) { return { open: async (sessionId: string, handler: (batch: DesktopTranscriptBatch) => void) => { @@ -148,8 +156,12 @@ test('projects the durable Coordination transcript into the WorkHub conversation text: 'Continue payments', state: 'completed', assignment: { + delegationId: 'payments-delegation', targetSessionId: 'payments', targetSessionName: 'Payments', + targetMessageId: 'payments-message', + targetTurnId: 'payments-turn', + feedbackState: 'accepted', }, updatedAt: 20, }]); @@ -291,6 +303,7 @@ test('desktop adapter rebuilds recent turns from the Session transcript and clos sessions: { list: async () => [], listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, @@ -366,6 +379,7 @@ test('desktop adapter cancels an unavailable transcript without hiding ready Ses sessions: { list: async () => [], listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, send: async () => { throw new Error('not used'); }, stop: async () => {}, @@ -448,6 +462,7 @@ test('desktop adapter projects Session catalog facts without owning copies', asy sessions: { list: async () => source, listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, @@ -506,6 +521,151 @@ test('desktop adapter projects Session catalog facts without owning copies', asy ]); }); +test('desktop adapter rebuilds delegation feedback from the Message-owned execution Turn', async () => { + const sessions = [ + desktopSession('accepted'), + desktopSession('running', { status: 'running', runningTurnIds: ['turn-running'] }), + desktopSession('waiting', { + status: 'waiting_for_user', + runningTurnIds: ['turn-waiting'], + }), + desktopSession('completed', { + status: 'waiting_for_user', + runningTurnIds: ['later-turn'], + }), + desktopSession('failed'), + desktopSession('aborted'), + desktopSession('cancelled'), + desktopSession('recovering'), + ]; + const turns = new Map>([ + ['running', [{ turnId: 'turn-running', status: 'running', statusSource: 'recorded' }]], + ['waiting', [{ turnId: 'turn-waiting', status: 'running', statusSource: 'recorded' }]], + ['completed', [{ turnId: 'turn-completed', status: 'completed', statusSource: 'recorded' }]], + ['failed', [{ turnId: 'turn-failed', status: 'failed', statusSource: 'recorded' }]], + ['aborted', [{ turnId: 'turn-aborted', status: 'aborted', statusSource: 'recorded' }]], + ]); + const adapter = createDesktopWorkHubSessionPort({ + transcripts: unusedTranscripts, + sessions: { + list: async () => sessions, + listTurns: async (sessionId) => { + if (sessionId === 'recovering') throw new Error('Host is recovering'); + return turns.get(sessionId) ?? []; + }, + queryMessageExecutions: async (sessionId, messageIds) => ({ + resolutions: sessionId === 'accepted' + ? messageIds.map((messageId) => ({ messageId, state: 'pending' as const })) + : sessionId === 'cancelled' + ? messageIds.map((messageId) => ({ messageId, state: 'cancelled' as const })) + : sessionId === 'recovering' + ? [] + : messageIds.map((messageId) => ({ + messageId, + state: 'owned' as const, + turnId: `turn-${sessionId}`, + runId: `run-${sessionId}`, + })), + }), + create: async () => { throw new Error('not used'); }, + send: async () => { throw new Error('not used'); }, + stop: async () => {}, + subscribeChanges: () => () => {}, + }, + projectName: () => 'Maka', + newTurnId: () => 'unused', + }); + const references = [ + ['accepted', 'turn-accepted'], + ['running', 'turn-running'], + ['waiting', 'turn-waiting'], + ['completed', 'turn-completed'], + ['failed', 'turn-failed'], + ['aborted', 'turn-aborted'], + ['cancelled', 'turn-cancelled'], + ['recovering', 'turn-recovering'], + ].map(([targetSessionId, targetTurnId]) => ({ + delegationId: `delegation-${targetSessionId}`, + targetSessionId: targetSessionId!, + targetMessageId: `message-${targetSessionId}`, + targetTurnId: targetTurnId!, + })); + + const feedback = await adapter.delegationFeedback(references); + + assert.deepEqual(feedback.map(({ delegationId, state }) => ({ delegationId, state })), [ + { delegationId: 'delegation-accepted', state: 'accepted' }, + { delegationId: 'delegation-running', state: 'running' }, + { delegationId: 'delegation-waiting', state: 'waiting_for_user' }, + { delegationId: 'delegation-completed', state: 'completed' }, + { delegationId: 'delegation-failed', state: 'failed' }, + { delegationId: 'delegation-aborted', state: 'aborted' }, + { delegationId: 'delegation-cancelled', state: 'aborted' }, + { delegationId: 'delegation-recovering', state: 'recovering' }, + ]); +}); + +test('desktop adapter follows a delegated Message into its successor Turn', async () => { + const targetSessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }); + const adapter = createDesktopWorkHubSessionPort({ + transcripts: transcriptsWith([{ + type: 'user', + id: 'payment-message', + turnId: 'successor-turn', + ts: 2, + text: 'Continue payment recovery', + steeringEventId: 'payment-message', + }]), + sessions: { + list: async () => [desktopSession(targetSessionId, { + status: 'running', + runningTurnIds: ['successor-turn'], + })], + listTurns: async () => [ + { + turnId: 'admission-turn', + status: 'completed', + statusSource: 'recorded', + }, + { + turnId: 'successor-turn', + status: 'running', + statusSource: 'recorded', + }, + ], + queryMessageExecutions: async (_sessionId, messageIds) => ({ + resolutions: messageIds.map((messageId) => ({ + messageId, + state: 'owned' as const, + turnId: 'successor-turn', + runId: 'successor-run', + })), + }), + create: async () => { throw new Error('not used'); }, + send: async () => { throw new Error('not used'); }, + stop: async () => {}, + subscribeChanges: () => () => {}, + }, + projectName: () => 'Maka', + newTurnId: () => 'unused', + }); + + const references = [{ + delegationId: 'payment-delegation', + targetSessionId, + targetTurnId: 'admission-turn', + targetMessageId: 'payment-message', + }]; + assert.deepEqual(await adapter.delegationFeedback(references), [{ + delegationId: 'payment-delegation', + state: 'running', + }]); +}); + test('desktop adapter preserves per-Host catalog coverage for ownership reconciliation', async () => { const localSessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'local' }); const adapter = createDesktopWorkHubSessionPort({ @@ -517,6 +677,7 @@ test('desktop adapter preserves per-Host catalog coverage for ownership reconcil completeHostIds: ['local-host'], }), listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, send: async () => { throw new Error('not used'); }, stop: async () => {}, @@ -546,6 +707,7 @@ test('desktop adapter delegates create, send, and invalidation to Session APIs', runningTurnIds: ['turn-new'], })], listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async (input) => { calls.push(['create', input]); return desktopSession('created', { name: input.name }); @@ -594,6 +756,7 @@ test('desktop adapter preserves when Session delivery steered an existing root T sessions: { list: async () => [], listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, @@ -626,6 +789,7 @@ test('desktop adapter distinguishes definite rejection from an unknown delivery sessions: { list: async () => [], listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, send: async () => { if (outcome === 'throw') throw new Error('transport disconnected'); @@ -719,6 +883,7 @@ test('desktop adapter reconciles lost replies from authoritative transcript iden sessions: { list: async () => [], listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, send: async () => { throw new Error('not used'); }, stop: async () => {}, @@ -745,6 +910,7 @@ test('desktop adapter binds stop to the root Turn owned by the WorkHub submissio sessions: { list: async () => [], listTurns: async () => [], + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, @@ -782,6 +948,7 @@ test('desktop adapter derives stable origin evidence from the existing Session l { userPromptPreview: '把风险按高、中、低分组' }, ]; }, + queryMessageExecutions: noMessageExecutions, create: async () => { throw new Error('not used'); }, diff --git a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts index b6f1eb03ec..a254f12be4 100644 --- a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts @@ -24,6 +24,7 @@ import { renderToStaticMarkup } from 'react-dom/server'; import { AstryxLocaleProvider, LocaleProvider } from '@maka/ui'; import { WorkHubCoordinationStatus, + WorkHubCoordinationTurnView, WorkHubProjectionRefreshGate, WorkHubSurfaceRouteGate, submitAndRecordWorkHubSurfaceInput, @@ -37,6 +38,8 @@ import { createLegacyWorkHubControllerForTests as createWorkHubController, WORKHUB_ROUTING_STRATEGY_ID, type WorkHubController, + type WorkHubCoordinationTurn, + type WorkHubDelegationExecutionState, type WorkHubSubmitInput, } from '../../renderer/workhub-controller.js'; import { WorkHubSendLease } from '../../renderer/workhub-send-lease.js'; @@ -107,6 +110,52 @@ test('Coordination lifecycle keeps a visible loading state and exposes failure r assert.match(failed, />Retry { + const states: Array<[WorkHubDelegationExecutionState, string]> = [ + ['accepted', 'Accepted'], + ['running', 'Running'], + ['waiting_for_user', 'Waiting for you'], + ['completed', 'Completed'], + ['failed', 'Failed'], + ['aborted', 'Aborted'], + ['recovering', 'Recovering'], + ]; + for (const [state, label] of states) { + const turn: WorkHubCoordinationTurn = { + messageId: 'assignment-1', + turnId: 'action-1', + text: 'Continue payments', + state: 'completed', + assignment: { + delegationId: 'delegation-1', + targetSessionId: 'payment', + targetSessionName: 'Payments', + targetMessageId: 'payment-message', + targetTurnId: 'payment-turn', + feedbackState: state, + }, + updatedAt: 10, + }; + const markup = renderToStaticMarkup( + createElement(LocaleProvider, { + locale: 'en', + children: createElement(AstryxLocaleProvider, { + children: createElement(WorkHubCoordinationTurnView, { + turn, + projection: { sessions: [], turns: [] }, + locale: 'en', + onOpenSession: () => undefined, + }), + }), + }), + ); + assert.match(markup, / @@ -749,6 +774,15 @@ function workHubCopy(locale: UiLocale) { delivery_failed: '输入未能送达，请重试。', }, scrollToBottom: '滚动到底部', archived: '已归档', states: { active: '活跃', running: '进行中', waiting_for_user: '等待你', blocked: '受阻', aborted: '已中止' }, + delegationStates: { + accepted: '已接收', + running: '进行中', + waiting_for_user: '等待你', + completed: '已完成', + failed: '失败', + aborted: '已中止', + recovering: '正在恢复', + }, turnStates: { running: '进行中', completed: '已完成', aborted: '已中止', failed: '失败' }, } as const; } @@ -782,6 +816,15 @@ function workHubCopy(locale: UiLocale) { delivery_failed: 'The input could not be delivered. Try again.', }, scrollToBottom: 'Scroll to bottom', archived: 'Archived', states: { active: 'Active', running: 'Running', waiting_for_user: 'Waiting for you', blocked: 'Blocked', aborted: 'Aborted' }, + delegationStates: { + accepted: 'Accepted', + running: 'Running', + waiting_for_user: 'Waiting for you', + completed: 'Completed', + failed: 'Failed', + aborted: 'Aborted', + recovering: 'Recovering', + }, turnStates: { running: 'Running', completed: 'Completed', aborted: 'Aborted', failed: 'Failed' }, } as const; } diff --git a/docs/architecture/workhub-coordination-session-adr.md b/docs/architecture/workhub-coordination-session-adr.md index eade3fbfb9..7b49723565 100644 --- a/docs/architecture/workhub-coordination-session-adr.md +++ b/docs/architecture/workhub-coordination-session-adr.md @@ -98,6 +98,7 @@ transcripts, such as: delegationId coordinationTurnId targetSessionId +targetMessageId targetTurnId disposition ``` @@ -131,6 +132,22 @@ so WorkHub does not own a second recovery state machine or compensation chain. The `delegation_assigned` record itself projects the visible WorkHub turn; the renderer does not append a second summary. +The first-response contract is hybrid. The atomic `delegation_assigned` record is +an immediate durable acknowledgement, so WorkHub confirms acceptance without +waiting for target execution. The target Message is the stable delegation +identity; `targetTurnId` records only its admission location. WorkHub asks the +target Message authority which Turn durably consumed or admitted that Message, +then joins the resolved Turn's recorded lifecycle and the target Session's exact +live-Turn membership to project `running`, `waiting_for_user`, `completed`, +`failed`, and `aborted`. This remains correct when an unconsumed steering Message +is folded into a successor Turn or recovery aggregates several pending Messages +under one new Turn. A durable cancellation tombstone for a retracted queued +Message resolves the delegation to `aborted`. If the target authority is +temporarily unreadable, WorkHub projects `recovering` rather than inventing a +terminal result. These execution states are never appended as mutable Coordination +records; Session change notifications invalidate the projection and opening +WorkHub after restart rebuilds it from the same link and target facts. + The renderer persists only a Host-scoped action id until acknowledgement. Composer draft text uses a separate storage key and lifecycle. A reload therefore preserves idempotency without freezing old text or coupling draft edits to Host authority. @@ -156,7 +173,8 @@ been committed. - Coordination Session role representation, lazy creation, durable lookup, recovery, per-Host UI resolution, persistent transcript, closed dispositions, and the Action Gate are implemented. Durable delegation linkage is encoded in - that transcript; target lifecycle projection, linked correction, and destructive + that transcript; target lifecycle projection and the hybrid first-response + contract are implemented as rebuildable reads. Linked correction and destructive replacement/Stop recovery remain later work. Reevaluate the per-Host decision if supported workflows require one WorkHub diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 48f24ad74a..4f31f1431f 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -130,6 +130,104 @@ test('consumes an atomically committed active-target admission exactly once', as assert.equal(fixture.drainRequests(), 0); }); +test('idle recovery preserves the exact root identity chosen with a durable admission', async () => { + const fixture = createFixture(); + fixture.setRootState({ kind: 'idle' }); + const content = { text: 'recover the linked WorkHub assignment' }; + await fixture.admissions.commitMessageAdmission({ + ...ROOT, + messageId: 'workhub-linked-message', + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 10, + }); + + await fixture.coordinator.consumePendingAdmissions([ROOT.sessionId]); + + assert.equal(fixture.recoveredBatches.length, 1); + assert.deepEqual(fixture.recoveredBatches[0]?.rootIdentity, { + turnId: ROOT.turnId, + runId: ROOT.runId, + }); +}); + +test('idle recovery does not reuse a predecessor identity for its queued successor', async () => { + const fixture = createFixture(); + fixture.setRootState({ kind: 'idle' }); + const content = { text: 'recover the queued successor' }; + await fixture.admissions.commitMessageAdmission({ + ...ROOT, + messageId: 'queued-successor-message', + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'next_turn', + placement: 'next_turn', + disposition: 'followup', + admittedAt: 10, + }); + + await fixture.coordinator.consumePendingAdmissions([ROOT.sessionId]); + + assert.equal(fixture.recoveredBatches.length, 1); + assert.equal(fixture.recoveredBatches[0]?.rootIdentity, undefined); +}); + +test('idle recovery resolves differently preassigned Messages to their shared successor Turn', async () => { + const fixture = createFixture(); + fixture.setRootState({ kind: 'idle' }); + for (const [messageId, turnId, runId] of [ + ['workhub-message-a', 'preassigned-turn-a', 'preassigned-run-a'], + ['workhub-message-b', 'preassigned-turn-b', 'preassigned-run-b'], + ] as const) { + const content = { text: `recover ${messageId}` }; + await fixture.admissions.commitMessageAdmission({ + sessionId: ROOT.sessionId, + turnId, + runId, + messageId, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 10, + }); + } + + await fixture.coordinator.consumePendingAdmissions([ROOT.sessionId]); + const resolved = await fixture.coordinator.handlers['turn.message.execution.query']( + { + sessionId: ROOT.sessionId, + messageIds: ['workhub-message-a', 'workhub-message-b'], + }, + operationContext(), + ); + + assert.equal(fixture.recoveredBatches[0]?.rootIdentity, undefined); + assert.deepEqual(resolved, { + ok: true, + result: { + resolutions: [ + { + messageId: 'workhub-message-a', + state: 'owned', + turnId: 'recovered-turn', + runId: 'durable-run', + }, + { + messageId: 'workhub-message-b', + state: 'owned', + turnId: 'recovered-turn', + runId: 'durable-run', + }, + ], + }, + }); +}); + test('idle submit starts exactly one root Turn and retry identity is connection-independent', async () => { const fixture = createFixture(); fixture.setRootState({ kind: 'idle' }); @@ -202,27 +300,75 @@ test('message query reports only durable cancellation proof', async () => { await submit(fixture, 'cancelled-message', 'discard me', 'next_turn'); await submit(fixture, 'accepted-message', 'waiting', 'next_turn'); await fixture.coordinator.cancelMessages(ROOT.sessionId, ['cancelled-message']); + const result = await fixture.coordinator.handlers['turn.message.query']( + { + sessionId: ROOT.sessionId, + messageIds: ['cancelled-message', 'accepted-message', 'unknown-message'], + }, + operationContext(), + ); + + assert.deepEqual(result, { + ok: true, + result: { cancelledMessageIds: ['cancelled-message'] }, + }); +}); + +test('message execution query reports the Turn that durably owns each Message', async () => { + const fixture = createFixture(); + const pendingContent = { text: 'not handed off yet' }; + await fixture.admissions.commitMessageAdmission({ + ...ROOT, + messageId: 'pending-message', + content: pendingContent, + submittedContentDigest: messageContentDigest(pendingContent), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 10, + }); fixture.receipts.set( 'handed-off-message', - sourceReceipt('handed-off-message', 'delivered', 'current_turn', 'steering'), + sourceReceipt( + 'handed-off-message', + 'delivered by successor', + 'current_turn', + 'steering', + 'successor-turn', + ), ); + fixture.events.push(steeringEvent('steered-message', 'consumed by admission Turn')); - const result = await fixture.coordinator.handlers['turn.message.query']( + const result = await fixture.coordinator.handlers['turn.message.execution.query']( { sessionId: ROOT.sessionId, - messageIds: [ - 'cancelled-message', - 'accepted-message', - 'handed-off-message', - 'unknown-message', - ], + messageIds: ['pending-message', 'handed-off-message', 'steered-message', 'unknown-message'], }, operationContext(), ); assert.deepEqual(result, { ok: true, - result: { cancelledMessageIds: ['cancelled-message'] }, + result: { + resolutions: [ + { + messageId: 'pending-message', + state: 'pending', + }, + { + messageId: 'handed-off-message', + state: 'owned', + turnId: 'successor-turn', + runId: 'durable-run', + }, + { + messageId: 'steered-message', + state: 'owned', + turnId: ROOT.turnId, + runId: ROOT.runId, + }, + ], + }, }); }); @@ -743,6 +889,18 @@ test('entry retract removes one queued entry, replays its outcome, and rejects s fixture.coordinator.projection(ROOT.sessionId).steering.map((entry) => entry.messageId), ['steer-1'], ); + assert.deepEqual( + await fixture.coordinator.handlers['turn.message.execution.query']( + { sessionId: ROOT.sessionId, messageIds: ['follow-1'] }, + operationContext(), + ), + { + ok: true, + result: { + resolutions: [{ messageId: 'follow-1', state: 'cancelled' }], + }, + }, + ); const retry = await fixture.coordinator.handlers['queue.entry.retract']( { diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 662fcd9e8d..9aa5d6ad42 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -324,6 +324,10 @@ describe('Runtime Host bootstrap protocol', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 50); }); + test('publishes a new compatibility epoch for Message execution ownership', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 61); + }); + test('publishes a new compatibility epoch for exact Session Connection identity', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 56); }); @@ -1157,9 +1161,14 @@ describe('Runtime Host bootstrap protocol', () => { operation: 'turn.message.query' as const, input: { sessionId: 'session-1', - messageIds: ['message-1', 'message-2'], + messageIds: ['message-1', 'message-2', 'message-3'], }, }; + const executionQuery = { + requestId: 'execution-query-request-1', + operation: 'turn.message.execution.query' as const, + input: query.input, + }; const submit = { requestId: 'submit-request-1', operation: 'turn.message.submit' as const, @@ -1188,6 +1197,36 @@ describe('Runtime Host bootstrap protocol', () => { }, }; assert.deepEqual(decodeClientFrame(query), query); + assert.deepEqual(decodeClientFrame(executionQuery), executionQuery); + const queried = { + requestId: executionQuery.requestId, + operation: executionQuery.operation, + ok: true as const, + result: { + resolutions: [ + { messageId: 'message-1', state: 'pending' as const }, + { + messageId: 'message-2', + state: 'owned' as const, + turnId: 'turn-2', + runId: 'run-2', + }, + { messageId: 'message-3', state: 'cancelled' as const }, + ], + }, + }; + assert.deepEqual(decodeHostFrame(queried), queried); + assert.throws( + () => + decodeHostFrame({ + ...queried, + result: { + ...queried.result, + resolutions: [...queried.result.resolutions, ...queried.result.resolutions], + }, + }), + isInvalidFrame, + ); assert.deepEqual(decodeClientFrame(submit), submit); assert.deepEqual(decodeClientFrame(retract), retract); assert.deepEqual(decodeClientFrame(interrupt), interrupt); diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index ec371b34ff..355b8a5bf4 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -55,7 +55,7 @@ import type { BackendCompactHistoryInput, BackendSendInput, } from '@maka/core/backend-types'; -import type { SessionEvent } from '@maka/core/events'; +import { messageContentDigest, type SessionEvent } from '@maka/core/events'; import { WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_SESSION_ROLE, @@ -179,6 +179,62 @@ test('turn.start rejects the reserved WorkHub Coordination Session identity', as } }); +test('recovered Messages retain their durably assigned root identity', async () => { + const fixture = await createFailureFixture({ + registerBackend: (backends) => + backends.register('ai-sdk', (context) => new FakeBackend(context)), + }); + const turnId = 'workhub-linked-turn'; + const runId = 'workhub-linked-run'; + const messageId = 'workhub-linked-message'; + const content = { text: 'continue the linked assignment' }; + const source = { + messageId, + content, + submittedContentDigest: messageContentDigest(content), + placement: 'current_turn' as const, + disposition: 'steering' as const, + }; + try { + await fixture.stores.sessionStore.commitMessageAdmission({ + sessionId: fixture.sessionId, + turnId, + runId, + messageId, + content, + submittedContentDigest: source.submittedContentDigest, + submittedPlacement: 'current_turn', + placement: source.placement, + disposition: source.disposition, + admittedAt: 1, + }); + + const outcome = await fixture.sessionAdmission.run(fixture.sessionId, (lease) => + fixture.coordinator.startRecoveredMessages( + { + sessionId: fixture.sessionId, + content, + submittedContent: content, + sources: [source], + rootIdentity: { turnId, runId }, + }, + lease, + ), + ); + + assert.deepEqual(outcome, { turnId }); + const admission = await fixture.stores.agentRunStore.readRootTurnAdmission( + fixture.sessionId, + turnId, + ); + assert.equal(admission?.runId, runId); + } finally { + await fixture.coordinator.close(); + await fixture.messages.close(); + await fixture.dispose(); + } +}); + test('turn.start rejects a corrupt Coordination role on an ordinary identity', async () => { const fixture = await createFailureFixture({ registerBackend: (backends) => diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 101bcd07fe..42a4c55671 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -94,7 +94,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 63 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 64 as const; +// 64: Message lifecycle queries expose durable execution ownership and +// cancellation. Older peers cannot decode or provide the closed proof list. // 63: Connection updates accept the full canonical enabled-model limit. // Older peers reject valid catalogs containing more than 64 enabled models. // 62: A Direct peer listener can expose owner-only Peer Mesh management diff --git a/packages/runtime-host/src/protocol/message.ts b/packages/runtime-host/src/protocol/message.ts index ea4df35d27..c30546427c 100644 --- a/packages/runtime-host/src/protocol/message.ts +++ b/packages/runtime-host/src/protocol/message.ts @@ -123,6 +123,25 @@ export interface TurnMessageQueryResult { readonly cancelledMessageIds: readonly string[]; } +export interface TurnMessageExecutionQueryInput { + readonly sessionId: string; + readonly messageIds: readonly string[]; +} + +export interface TurnMessageExecutionQueryResult { + readonly resolutions: readonly TurnMessageExecutionResolution[]; +} + +export type TurnMessageExecutionResolution = + | { readonly messageId: string; readonly state: 'pending' } + | { readonly messageId: string; readonly state: 'cancelled' } + | { + readonly messageId: string; + readonly state: 'owned'; + readonly turnId: string; + readonly runId: string; + }; + export interface QueueRetractInput { readonly originHostEpoch: string; readonly sessionId: string; @@ -202,6 +221,13 @@ export const MESSAGE_OPERATION_SPECS = { decodeInput: decodeTurnMessageQueryInput, decodeOutput: decodeTurnMessageQueryResult, }), + 'turn.message.execution.query': defineOperation({ + mode: 'query', + availability: 'ready', + errors: MESSAGE_OPERATION_ERRORS, + decodeInput: decodeTurnMessageExecutionQueryInput, + decodeOutput: decodeTurnMessageExecutionQueryResult, + }), 'turn.message.submit': defineOperation({ mode: 'command', availability: 'ready', @@ -341,6 +367,59 @@ function decodeTurnMessageQueryResult(value: unknown): TurnMessageQueryResult { return { cancelledMessageIds }; } +function decodeTurnMessageExecutionQueryInput(value: unknown): TurnMessageExecutionQueryInput { + return decodeTurnMessageQueryInput(value); +} + +function decodeTurnMessageExecutionQueryResult(value: unknown): TurnMessageExecutionQueryResult { + const record = requireExactRecord(value, 'turn.message.execution.query result', ['resolutions']); + if (!Array.isArray(record.resolutions) || record.resolutions.length > MESSAGE_QUEUE_MAX_ENTRIES) { + throw invalidProtocolFrame('Invalid turn.message.execution.query resolutions'); + } + const resolutions = record.resolutions.map((value): TurnMessageExecutionResolution => { + const resolution = requireRecord(value, 'turn.message.execution.query resolution'); + if (resolution.state === 'pending') { + assertExactKeys(resolution, 'turn.message.execution.query pending resolution', [ + 'messageId', + 'state', + ]); + return { + messageId: requireEntityId(resolution.messageId, 'messageId'), + state: 'pending', + }; + } + if (resolution.state === 'cancelled') { + assertExactKeys(resolution, 'turn.message.execution.query cancelled resolution', [ + 'messageId', + 'state', + ]); + return { + messageId: requireEntityId(resolution.messageId, 'messageId'), + state: 'cancelled', + }; + } + if (resolution.state === 'owned') { + assertExactKeys(resolution, 'turn.message.execution.query owned resolution', [ + 'messageId', + 'state', + 'turnId', + 'runId', + ]); + return { + messageId: requireEntityId(resolution.messageId, 'messageId'), + state: 'owned', + turnId: requireEntityId(resolution.turnId, 'turnId'), + runId: requireEntityId(resolution.runId, 'runId'), + }; + } + throw invalidProtocolFrame('Invalid turn.message.execution.query resolution state'); + }); + if (new Set(resolutions.map(({ messageId }) => messageId)).size !== resolutions.length) { + throw invalidProtocolFrame('Duplicate turn.message.execution.query messageId'); + } + return { resolutions }; +} + function decodeTurnMessageSubmitResult(value: unknown): TurnMessageSubmitResult { const record = requireRecord(value, 'turn.message.submit result'); if (record.disposition === 'turn_started') { diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 638f4aa4dc..8c048d2c96 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -318,6 +318,7 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'subscription.open', 'task.ledger.query', 'turn.interrupt', + 'turn.message.execution.query', 'turn.message.query', 'turn.message.submit', 'turn.query', diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 1db629328d..11a953a119 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -128,6 +128,17 @@ export interface HostMessageRecoveryBatch { readonly content: MessageContent; readonly submittedContent: MessageContent; readonly sources: readonly RootTurnSourceMessage[]; + /** + * A root identity durably chosen with the pending Messages. Recovery must + * preserve it so immutable links to the exact Turn keep naming the execution + * that is actually admitted. A batch only carries one when every pending + * current-Turn steering Message names the same root; next-Turn follow-ups + * name their predecessor and must receive a new successor identity. + */ + readonly rootIdentity?: { + readonly turnId: string; + readonly runId: string; + }; /** * What the recovered Message asked of its Turn. Only a lone Message can * carry one — exact-Turn intent needs an idle Session and opens its own root @@ -338,6 +349,7 @@ const HOST_EPOCH_PATTERN = /^[A-Za-z0-9_-]{1,128}$/u; export class HostMessageCoordinator implements RuntimeMessageAuthority { readonly handlers: MessageOperationHandlerMap = { 'turn.message.query': (input) => this.queryMessages(input), + 'turn.message.execution.query': (input) => this.queryMessageExecutions(input), 'turn.message.submit': (input, context) => this.submit(input, context), 'queue.retract': (input) => this.retract(input), 'queue.entry.retract': (input) => this.retractQueuedEntry(input), @@ -411,6 +423,71 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { return success({ cancelledMessageIds }); } + async queryMessageExecutions(input: { + sessionId: string; + messageIds: readonly string[]; + }): Promise< + MessageOutcome<{ + resolutions: Array< + | { messageId: string; state: 'pending' } + | { messageId: string; state: 'cancelled' } + | { messageId: string; state: 'owned'; turnId: string; runId: string } + >; + }> + > { + const resolutions: Array< + | { messageId: string; state: 'pending' } + | { messageId: string; state: 'cancelled' } + | { messageId: string; state: 'owned'; turnId: string; runId: string } + > = []; + for (const messageId of input.messageIds) { + const receipt = await this.#durableProof.readRootTurnSourceMessageReceipt( + input.sessionId, + messageId, + ); + if ( + receipt?.admission.sessionId === input.sessionId && + receipt.sourceMessage.messageId === messageId + ) { + // A root source receipt is the latest durable ownership proof and + // therefore outranks the steering location from which a Message may + // have been folded into this successor. + resolutions.push({ + messageId, + state: 'owned', + turnId: receipt.admission.turnId, + runId: receipt.admission.runId, + }); + continue; + } + const steering = await this.#durableProof.readImmutableSteeringMessageProof( + input.sessionId, + messageId, + ); + if ( + steering?.event.sessionId === input.sessionId && + steering.event.refs?.providerEventId === messageId + ) { + resolutions.push({ + messageId, + state: 'owned', + turnId: steering.event.turnId, + runId: steering.event.runId, + }); + continue; + } + if (await this.#admissions.hasCancelledMessageAdmission(input.sessionId, messageId)) { + resolutions.push({ messageId, state: 'cancelled' }); + continue; + } + const pending = await this.#admissions.readMessageAdmission(input.sessionId, messageId); + if (pending?.sessionId === input.sessionId && pending.messageId === messageId) { + resolutions.push({ messageId, state: 'pending' }); + } + } + return success({ resolutions }); + } + retireSessions(sessionIds: readonly string[]): void { for (const sessionId of new Set(sessionIds)) { const state = this.#sessions.get(sessionId); @@ -711,12 +788,14 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { 'Message recovery authority is unavailable', ); } + const rootIdentity = sharedPendingRootIdentity(pending); const started = await this.#root.startRecoveredMessages( { sessionId, content: aggregateMessageContents(pending.map((entry) => entry.content)), submittedContent: aggregateMessageContents(pending.map((entry) => entry.content)), sources: pending.map(pendingMessageSource), + ...(rootIdentity ? { rootIdentity } : {}), ...(pending.length === 1 && pending[0]!.submittedIntent ? { submittedIntent: pending[0]!.submittedIntent } : {}), @@ -2217,6 +2296,24 @@ function pendingMessageSource(admission: PendingMessageAdmission): RootTurnSourc }; } +function sharedPendingRootIdentity( + admissions: readonly PendingMessageAdmission[], +): HostMessageRecoveryBatch['rootIdentity'] { + const first = admissions[0]; + if (!first || first.placement !== 'current_turn' || first.disposition !== 'steering') { + return undefined; + } + return admissions.every( + (admission) => + admission.placement === 'current_turn' && + admission.disposition === 'steering' && + admission.turnId === first.turnId && + admission.runId === first.runId, + ) + ? { turnId: first.turnId, runId: first.runId } + : undefined; +} + function submittedProjectionContent(content: MessageContent): MessageContent { const normalized = normalizeMessageContent(content); const text = normalized.displayText ?? normalized.text; diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index 8208c5ad08..77af9e17a5 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -90,6 +90,7 @@ export type ConnectionEffectOperationKey = Extract< export type MessageOperationKey = Extract< OperationKey, | 'turn.message.query' + | 'turn.message.execution.query' | 'turn.message.submit' | 'queue.retract' | 'queue.entry.retract' diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index b8639b0bee..c9e7dfefe0 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -1169,7 +1169,14 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { const reservation = this.reserveRootTurn(input.sessionId); if (!reservation) return { error: 'Another root Turn is being admitted' }; try { - const turnId = randomUUID(); + // A steering admission can preassign a future root while the Session + // is idle. An identity already in the owned chain instead names the + // predecessor that accepted the Message and must not become its own + // successor. + const latestAdmission = this.rootAdmissionOwner.latestAdmission(input.sessionId); + const rootIdentity = + input.rootIdentity?.turnId === latestAdmission?.turnId ? undefined : input.rootIdentity; + const turnId = rootIdentity?.turnId ?? randomUUID(); // The recovered Message asked for this mode before the Host stopped; // admitting without it would run a different Turn than was requested. const turnOrchestration = input.submittedIntent?.turnOrchestration; @@ -1177,7 +1184,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { const admitted = await this.rootAdmissionOwner.admitRootTurn({ sessionId: input.sessionId, turnId, - proposedRunId: randomUUID(), + proposedRunId: rootIdentity?.runId ?? randomUUID(), proposedUserMessageId: input.sources.length === 1 ? input.sources[0]!.messageId : null, execution: { kind: 'external_message',