diff --git a/.github/assets/workhub-turn-admission/receipts-dark.png b/.github/assets/workhub-turn-admission/receipts-dark.png new file mode 100644 index 0000000000..2d6fb9594a Binary files /dev/null and b/.github/assets/workhub-turn-admission/receipts-dark.png differ diff --git a/.github/assets/workhub-turn-admission/receipts-light.png b/.github/assets/workhub-turn-admission/receipts-light.png new file mode 100644 index 0000000000..7042dcdb20 Binary files /dev/null and b/.github/assets/workhub-turn-admission/receipts-light.png differ diff --git a/apps/desktop/src/main/__tests__/runtime-host-client-operations.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client-operations.test.ts index 302d644243..e5e97357da 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client-operations.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client-operations.test.ts @@ -100,7 +100,6 @@ test('resolves WorkHub coordination through the dedicated Host operation', async { sessionId: 'maka_workhub_coordination' }, { candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [] }, { disposition: 'answer_here', coordinationTurnId: 'action-turn' }, - { turnId: 'summary-turn' }, ]); assert.deepEqual(await client.resolveWorkHubCoordinationSession(), { @@ -118,14 +117,6 @@ test('resolves WorkHub coordination through the dedicated Host operation', async }), { disposition: 'answer_here', coordinationTurnId: 'action-turn' }, ); - assert.deepEqual( - await client.recordWorkHubCoordination({ - turnId: 'summary-turn', - userText: 'Request', - assistantText: 'Summary', - }), - { turnId: 'summary-turn' }, - ); assert.deepEqual(requests, [ { operation: 'workhub.coordination.resolve', input: {} }, { operation: 'workhub.coordination.candidates', input: {} }, @@ -137,14 +128,7 @@ test('resolves WorkHub coordination through the dedicated Host operation', async proposal: { disposition: 'answer_here' }, }, }, - { - operation: 'workhub.coordination.record', - input: { - turnId: 'summary-turn', - userText: 'Request', - assistantText: 'Summary', - }, - }, + ]); }); diff --git a/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts index 70a84dedf9..d3159f2b6b 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts @@ -25,7 +25,6 @@ import { registerRuntimeHostWorkHubIpc } from '../runtime-host-workhub-ipc-main. test('projects WorkHub coordination resolution through its dedicated IPC domain', async () => { const handlers = new Map unknown>(); let resolveCalls = 0; - const records: unknown[] = []; const actions: unknown[] = []; const changes: unknown[] = []; const createdSessionId = 'runtime-created-session'; @@ -35,14 +34,6 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain' resolveCalls += 1; return { sessionId: 'maka_workhub_coordination' }; }, - recordWorkHubCoordination: async (input: { - turnId: string; - userText: string; - assistantText: string; - }) => { - records.push(input); - return { turnId: input.turnId }; - }, listWorkHubCoordinationCandidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [], @@ -74,19 +65,7 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain' assert.ok(handler); assert.deepEqual(await handler({}), { sessionId: 'maka_workhub_coordination' }); assert.equal(resolveCalls, 1); - assert.deepEqual( - await handlers.get('workhub:record')?.({}, { - turnId: 'record', - userText: 'Request', - assistantText: 'Summary', - }), - { turnId: 'record' }, - ); - assert.deepEqual(records, [{ - turnId: 'record', - userText: 'Request', - assistantText: 'Summary', - }]); + assert.equal(handlers.has('workhub:record'), false); assert.deepEqual(await handlers.get('workhub:candidates')?.({}), { candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [], diff --git a/apps/desktop/src/main/__tests__/workhub-controller-fixture.ts b/apps/desktop/src/main/__tests__/workhub-controller-fixture.ts index 4f859c6817..4e92d0fcaa 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller-fixture.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller-fixture.ts @@ -83,7 +83,7 @@ export function createWorkHubController({ ...(routingStrategy ? { routingStrategy } : {}), coordination: { open: async (handler) => { handler(transcript); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => { const candidates = (await sessions.list()) .filter((entry) => entry.kind === 'ordinary' && !entry.archived) diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index b8f0fc0fb1..f37645ce2f 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -110,7 +110,7 @@ test('conversation acknowledges a durable assignment before projecting target ex handler([assignment]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [] }), act: async () => ({ disposition: 'answer_here', coordinationTurnId: 'unused' }), }, @@ -157,7 +157,7 @@ test('conversation feedback never lets an older refresh overwrite newer target s handler([assignment]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'b'.repeat(64)}`, candidates: [] }), act: async () => ({ disposition: 'answer_here', coordinationTurnId: 'unused' }), }, @@ -199,7 +199,7 @@ test('direct stop bypasses routing candidates and preserves a not_owned delegati handler([coordinationAssignmentTurn()]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => { candidateReads += 1; return { candidateSetId: `sha256:${'d'.repeat(64)}`, candidates: [] }; @@ -255,7 +255,7 @@ test('an anaphoric stop asks for a fresh named imperative without offering a rou handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('stop clarification must not read route candidates'), act: async () => assert.fail('anaphoric stop must not reach the Action Gate'), }, @@ -282,7 +282,7 @@ test('a named resume submits and reports what the Host did', async () => { handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'e'.repeat(64)}`, candidates: [{ candidateRef: 'candidate-payments', sessionId: 'payments', sessionName: 'Payments', latestDelegationActionId: 'source-action', workspace: { target: { kind: 'host_path' as const, path: '/workspace/payments' }, hostCwd: '/workspace/payments' }, state: 'active' as const, updatedAt: 1 }], @@ -327,7 +327,7 @@ test('an anaphoric resume asks for a named work item', async () => { handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('resume clarification must not read route candidates'), act: async () => assert.fail('anaphoric resume must not reach the Action Gate'), }, @@ -354,7 +354,7 @@ test('a resume the Host will not admit becomes its clarification', async () => { handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'e'.repeat(64)}`, candidates: [{ candidateRef: 'candidate-payments', sessionId: 'payments', sessionName: 'Payments', latestDelegationActionId: 'source-action', workspace: { target: { kind: 'host_path' as const, path: '/workspace/payments' }, hostCwd: '/workspace/payments' }, state: 'active' as const, updatedAt: 1 }], @@ -392,7 +392,7 @@ test('a resume identity conflict is not mislabeled as a missing target', async ( handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'e'.repeat(64)}`, candidates: [{ candidateRef: 'candidate-payments', sessionId: 'payments', sessionName: 'Payments', latestDelegationActionId: 'source-action', workspace: { target: { kind: 'host_path' as const, path: '/workspace/payments' }, hostCwd: '/workspace/payments' }, state: 'active' as const, updatedAt: 1 }], @@ -419,7 +419,7 @@ test('a Runtime Host without safe-boundary resume explains why it cannot resume' handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'e'.repeat(64)}`, candidates: [{ candidateRef: 'candidate-payments', sessionId: 'payments', sessionName: 'Payments', latestDelegationActionId: 'source-action', workspace: { target: { kind: 'host_path' as const, path: '/workspace/payments' }, hostCwd: '/workspace/payments' }, state: 'active' as const, updatedAt: 1 }], @@ -453,7 +453,7 @@ test('a recovering Runtime Host tells the user to retry resume', async () => { handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'e'.repeat(64)}`, candidates: [{ candidateRef: 'candidate-payments', sessionId: 'payments', sessionName: 'Payments', latestDelegationActionId: 'source-action', workspace: { target: { kind: 'host_path' as const, path: '/workspace/payments' }, hostCwd: '/workspace/payments' }, state: 'active' as const, updatedAt: 1 }], @@ -484,7 +484,7 @@ test('a named stop reports the Gate refusal instead of judging the target itself handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('stop clarification must not read route candidates'), act: async () => { submitted += 1; @@ -518,7 +518,7 @@ test('a stop that fails for any other reason is a fault, not a clarification', a handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('stop clarification must not read route candidates'), act: async () => { throw new WorkHubCoordinationFailure('persistence_failed', 'WorkHub stop state is unavailable'); @@ -548,7 +548,7 @@ test('stop-shaped ordinary work routes normally instead of looping on clarificat handler([]); return { close: async () => undefined }; }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'e'.repeat(64)}`, candidates: [{ @@ -1631,7 +1631,7 @@ test('submit keeps unmatched non-executable conversation in WorkHub', async () = sessions, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [], @@ -1677,7 +1677,7 @@ test('production submission delegates only through the Runtime-owned candidate r sessions, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'b'.repeat(64)}`, candidates: [{ @@ -1731,7 +1731,7 @@ test('production retry reaches durable Action Gate replay while target is waitin sessions, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'c'.repeat(64)}`, candidates: [{ @@ -1776,7 +1776,7 @@ test('production sends an explicit correction as a linked replacement', async () sessions, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'d'.repeat(64)}`, candidates: [ @@ -1901,7 +1901,7 @@ test('production natural-language corrections retain the prior delegation link', sessions, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId, candidates: candidates.map((candidate) => { @@ -1999,7 +1999,7 @@ test('production correction-shaped creation stays create_new without an existing sessions: port([]), coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'c'.repeat(64)}`, candidates: [], @@ -2030,9 +2030,6 @@ test('production clarification is persisted through the typed Action Gate dispos sessions: port([]), coordination: { open: async () => ({ close: async () => undefined }), - record: async () => { - throw new Error('legacy summary recording must not persist clarification'); - }, candidates: async () => ({ candidateSetId: `sha256:${'c'.repeat(64)}`, candidates: [], @@ -2047,7 +2044,7 @@ test('production clarification is persisted through the typed Action Gate dispos }, }); - assert.deepEqual(await controller.recordConversationTurn({ + assert.deepEqual(await controller.requestClarification({ turnId: 'clarification-action', userText: '继续稳定性问题', assistantText: '请选择目标 Session', @@ -2073,7 +2070,7 @@ test('production creation leaves Session identity and workspace authority to mai sessions, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'c'.repeat(64)}`, candidates: [], @@ -3212,7 +3209,7 @@ for (const createStrategy of [createWorkHubR24RoutingStrategy, () => createWorkH routingStrategy, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'e'.repeat(64)}`, candidates: [{ candidateRef: 'payments-ref', sessionId: 'payments', sessionName: 'Payments', latestDelegationActionId: 'source-action', workspace: { target: { kind: 'host_path' as const, path: '/workspace/payments' }, hostCwd: '/workspace/payments' }, state: 'active' as const, updatedAt: 1 }] }), act: async (input) => { assert.equal(input.proposal.disposition, 'resume_work'); 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 c6fbca465b..774d1ad654 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -448,7 +448,7 @@ test('Coordination transcript adapter never replays history and completes only t }; }, }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('conversation open must not read route candidates'), act: async () => ({ ok: true, @@ -532,7 +532,7 @@ test('Coordination transcript adapter retries latest-record completion in the sa }; }, }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('conversation open must not read route candidates'), act: async () => ({ ok: true, @@ -629,7 +629,7 @@ test('Coordination transcript adapter ignores a stale latest-record failure afte }; }, }, - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('conversation open must not read route candidates'), act: async () => ({ ok: true, @@ -1158,3 +1158,28 @@ test('desktop adapter derives stable origin evidence from the existing Session l assert.deepEqual(second, first); assert.equal(reads, 1); }); + + +test('admitted clarification and resume project receipts without assistant messages', () => { + const turns = projectWorkHubCoordinationTurns([ + { type: 'user', id: 'failed-user', turnId: 'request', ts: 0, text: 'Which task?' }, + { type: 'turn_state', id: 'failed-state', turnId: 'request', ts: 1, status: 'failed' }, + { type: 'user', id: 'u', turnId: 'retry-turn', ts: 1, text: 'Which task?' }, + { type: 'workhub_coordination', kind: 'action_receipt', schemaVersion: 1, + id: 'receipt', turnId: 'retry-turn', ts: 2, + receipt: { actionId: 'request', userText: 'Which task?', clarification: 'Please name a task.', + result: { disposition: 'clarify', coordinationTurnId: 'retry-turn' } } }, + { type: 'turn_state', id: 'done', turnId: 'retry-turn', ts: 3, status: 'completed' }, + { type: 'workhub_coordination', kind: 'action_receipt', schemaVersion: 1, + id: 'resume-receipt', turnId: 'resume-turn', ts: 4, + receipt: { actionId: 'resume', userText: 'Resume Payments', + result: { disposition: 'resume_work', outcome: 'resume_started', targetSessionId: 'payments', targetTurnId: 'target-turn' } } }, + { type: 'turn_state', id: 'resume-done', turnId: 'resume-turn', ts: 5, status: 'completed' }, + ]); + assert.equal(turns.length, 2); + assert.equal(turns[0]?.turnId, 'request'); + assert.equal(turns[0]?.result, 'Please name a task.'); + assert.equal(turns[0]?.state, 'completed'); + assert.equal(turns[1]?.result, undefined); + assert.deepEqual(turns[1]?.resume, { disposition: 'resume_work', outcome: 'resume_started', targetSessionId: 'payments', targetTurnId: 'target-turn' }); +}); 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 772fb55f39..34375bb78f 100644 --- a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts @@ -413,7 +413,7 @@ test('surface keeps clarification and successful routing in WorkHub', async () = handler([]); return { close: async () => undefined }; }, - recordConversationTurn: async ({ turnId }) => ({ turnId }), + requestClarification: async ({ turnId }) => ({ turnId }), resetVisitContext: () => {}, subscribe: () => () => {}, submit: async (input) => { @@ -474,7 +474,7 @@ test('ambiguous creation is durably clarified before a fresh imperative creates }, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [], @@ -556,7 +556,7 @@ test('surface leaves discussion in WorkHub instead of creating a task view', asy handler([]); return { close: async () => undefined }; }, - recordConversationTurn: async ({ turnId }) => ({ turnId }), + requestClarification: async ({ turnId }) => ({ turnId }), resetVisitContext: () => {}, subscribe: () => () => {}, submit: async (input) => ({ @@ -647,7 +647,7 @@ test('real Session projection creates new guide topics and preserves origin ambi sessions: port, coordination: { open: async () => ({ close: async () => undefined }), - record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: sessions.map((entry) => ({ @@ -838,7 +838,7 @@ test('successful delegated submission needs no renderer summary write', async () assert.equal(records, 0); }); -test('resume records ordinary conversation text without persisting execution fields', async () => { +test('resume relies on the admitted Host receipt without a second conversation write', async () => { const records: unknown[] = []; const controller = fakeController({ submit: async (input) => ({ @@ -854,11 +854,7 @@ test('resume records ordinary conversation text without persisting execution fie summary: () => 'Resume requested. See the target Session for current progress.', onSummaryError: () => assert.fail('conversation write must succeed'), }); - assert.deepEqual(records, [{ - turnId: 'resume-1', userText: 'Resume Payments', - assistantText: 'Resume requested. See the target Session for current progress.', disposition: 'summary', - }]); -}); + assert.deepEqual(records, []);}); test('lease retires only after an acknowledged submission', async () => { const { storage } = memoryStorage(); @@ -928,13 +924,13 @@ function memoryStorage() { function fakeController(input: { submit: WorkHubController['submit']; - record: WorkHubController['recordConversationTurn']; + record: WorkHubController['requestClarification']; }): WorkHubController { return { read: async () => ({ sessions: [], turns: [] }), submit: input.submit, openConversation: async () => ({ close: async () => undefined }), - recordConversationTurn: input.record, + requestClarification: input.record, subscribe: () => () => undefined, resetVisitContext: () => undefined, }; diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index 0eb623bba6..3008f4c603 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -987,12 +987,6 @@ export class DesktopRuntimeHostClient { - recordWorkHubCoordination( - input: OperationInput<"workhub.coordination.record">, - ): Promise> { - return this.request("workhub.coordination.record", input); - } - listExternalSessionSources(): Promise { return this.request("external-session.source.query", {}); } diff --git a/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts b/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts index 1ebca81bf6..1ec25bb135 100644 --- a/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts @@ -32,7 +32,6 @@ type RuntimeHostWorkHubClient = Pick< DesktopRuntimeHostClient, | 'actWorkHubCoordination' | 'listWorkHubCoordinationCandidates' - | 'recordWorkHubCoordination' | 'resolveWorkHubCoordinationSession' >; @@ -52,9 +51,6 @@ export function registerRuntimeHostWorkHubIpc( ipcMain.handle('workhub:resolveCoordinationSession', () => client.resolveWorkHubCoordinationSession(), ); - ipcMain.handle('workhub:record', (_event, input) => - client.recordWorkHubCoordination(input), - ); ipcMain.handle('workhub:candidates', () => client.listWorkHubCoordinationCandidates()); ipcMain.handle('workhub:act', async (_event, rawInput: RendererWorkHubActionInput) => { try { diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 85e6b0a70e..26b72601d0 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -1009,11 +1009,6 @@ export interface MakaBridge { workHub: { /** Resolve the active Runtime Host's stable coordination conversation. */ resolveCoordinationSession(): Promise; - /** Persist one deterministic clarification or routing summary. */ - record( - coordinationSessionId: string, - input: { turnId: string; userText: string; assistantText: string }, - ): Promise<{ turnId: string }>; /** Read one bounded, Host-issued candidate set for a coordination action. */ candidates( coordinationSessionId: string, diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 99790e2880..d9bed024b0 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -2006,16 +2006,6 @@ const makaBridge = { (scope) => ipcRenderer.invoke('workhub:resolveCoordinationSession', scope), ); }, - async record( - coordinationSessionId: string, - input: { turnId: string; userText: string; assistantText: string }, - ): Promise<{ turnId: string }> { - const scope = await resolveDesktopWorkHubCoordinationCreateScope( - coordinationSessionId, - runtimeHostSessionRef, - ); - return ipcRenderer.invoke('workhub:record', scope, input) as Promise<{ turnId: string }>; - }, async candidates( coordinationSessionId: string, ): Promise> { diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index ce23268337..95a6306aae 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -1309,8 +1309,6 @@ function AppShellContent({ coordination: createDesktopWorkHubCoordinationPort({ sessionId: workHubCoordinationSessionId ?? 'workhub-coordination-unresolved', transcripts: window.maka.transcripts, - record: (input) => - window.maka.workHub.record(workHubCoordinationSessionId!, input), candidates: () => window.maka.workHub.candidates(workHubCoordinationSessionId!), act: (input) => diff --git a/apps/desktop/src/renderer/workhub-controller.ts b/apps/desktop/src/renderer/workhub-controller.ts index 6eea15a9c6..722c79546d 100644 --- a/apps/desktop/src/renderer/workhub-controller.ts +++ b/apps/desktop/src/renderer/workhub-controller.ts @@ -141,6 +141,7 @@ export interface WorkHubCoordinationTurn { readonly targetSessionName: string; readonly outcome?: Extract['outcome']; }; + resume?: Extract; updatedAt: number; } @@ -258,11 +259,6 @@ export interface WorkHubCoordinationPort { handler: (turns: readonly WorkHubCoordinationTurn[]) => void, onError: (error: unknown) => void, ): Promise<{ close(): Promise }>; - record(input: { - turnId: string; - userText: string; - assistantText: string; - }): Promise<{ turnId: string }>; candidates(): Promise; act(input: Omit): Promise; } @@ -276,11 +272,11 @@ export interface WorkHubController { ) => void, onError: (error: unknown) => void, ): Promise<{ close(): Promise }>; - recordConversationTurn(input: { + requestClarification(input: { turnId: string; userText: string; assistantText: string; - disposition?: 'clarify' | 'summary'; + disposition: 'clarify'; }): Promise<{ turnId: string }>; subscribe(handler: () => void): () => void; resetVisitContext(): void; @@ -515,26 +511,16 @@ export function createWorkHubController(deps: { }, }; }, - async recordConversationTurn(input) { - if (input.disposition === 'clarify') { - const result = await coordination.act({ - actionId: input.turnId, - userText: input.userText, - proposal: { - disposition: 'clarify', - assistantText: input.assistantText, - }, - }); - if (result.disposition !== 'clarify') { - throw new Error('WorkHub Action Gate returned an unexpected disposition'); - } - return { turnId: result.coordinationTurnId }; - } - return coordination.record({ - turnId: input.turnId, + async requestClarification(input) { + const result = await coordination.act({ + actionId: input.turnId, userText: input.userText, - assistantText: input.assistantText, + proposal: { disposition: 'clarify', assistantText: input.assistantText }, }); + if (result.disposition !== 'clarify') { + throw new Error('WorkHub Action Gate returned an unexpected disposition'); + } + return { turnId: result.coordinationTurnId }; }, subscribe(handler) { return deps.sessions.subscribe(handler); diff --git a/apps/desktop/src/renderer/workhub-coordination-port.ts b/apps/desktop/src/renderer/workhub-coordination-port.ts index 954e68286b..3b61b02546 100644 --- a/apps/desktop/src/renderer/workhub-coordination-port.ts +++ b/apps/desktop/src/renderer/workhub-coordination-port.ts @@ -47,18 +47,12 @@ const WORKHUB_COORDINATION_LATEST_RECORD_MAX_BYTES = 512 * 1024; export function createDesktopWorkHubCoordinationPort(deps: { sessionId: string; transcripts: WorkHubDesktopTranscriptBridge; - record(input: { - turnId: string; - userText: string; - assistantText: string; - }): Promise<{ turnId: string }>; candidates(): Promise; act( input: Omit, ): Promise>; }): WorkHubCoordinationPort { return { - record: deps.record, candidates: deps.candidates, async act(input) { const outcome = await deps.act(input); @@ -157,6 +151,15 @@ export function projectWorkHubCoordinationTurns( const stateByTurnId = new Map( deriveTurnRecords(messages).map((turn) => [turn.turnId, projectState(turn.status)]), ); + const factualTurnIds = new Set(messages.flatMap((message) => { + if (message.type !== 'workhub_coordination') return []; + if (message.kind === 'action_receipt' && message.receipt.result.disposition === 'clarify') { + return [message.receipt.actionId]; + } + return message.kind === 'delegation_assigned' || message.kind === 'delegation_stop_requested' + ? [message.coordinationTurnId] + : []; + })); const turns: WorkHubCoordinationTurn[] = []; const latestUserIndexByTurnId = new Map(); const terminalLinkState = new Map(); @@ -173,6 +176,26 @@ export function projectWorkHubCoordinationTurns( } for (const message of messages) { + if (message.type === 'workhub_coordination' && message.kind === 'action_receipt') { + const { receipt } = message; + if (receipt.result.disposition === 'clarify' || receipt.result.disposition === 'resume_work') { + const earlier = latestUserIndexByTurnId.get(message.turnId); + const row = { + messageId: message.id, + turnId: receipt.actionId, + text: boundedWorkHubTimelineText(receipt.userText), + ...(receipt.result.disposition === 'resume_work' ? { resume: receipt.result } : {}), + ...(receipt.clarification + ? { result: boundedWorkHubTimelineText(receipt.clarification) } + : {}), + state: stateByTurnId.get(message.turnId) ?? 'running', + updatedAt: message.ts, + }; + if (earlier !== undefined) turns[earlier] = row; + else turns.push(row); + } + continue; + } if (message.type === 'workhub_coordination' && message.kind === 'delegation_stop_requested') { const resolution = stopResolutionByDelegationId.get(message.stopsDelegationId); turns.push({ @@ -211,6 +234,7 @@ export function projectWorkHubCoordinationTurns( continue; } if (message.type === 'user') { + if (factualTurnIds.has(message.turnId)) continue; const text = boundedWorkHubTimelineText(userFacingText(message)); if (!text) continue; turns.push({ diff --git a/apps/desktop/src/renderer/workhub-surface.tsx b/apps/desktop/src/renderer/workhub-surface.tsx index db8dd711ce..5ebf8b9f5c 100644 --- a/apps/desktop/src/renderer/workhub-surface.tsx +++ b/apps/desktop/src/renderer/workhub-surface.tsx @@ -173,21 +173,22 @@ export async function submitAndRecordWorkHubSurfaceInput(input: { // accepted and must not consume the immutable Coordination summary owned by // this action identity. A later same-identity retry may still be admitted. // Delegations project directly from the Host's atomic delegation_assigned - // record. Only local clarification still needs the generic summary path. + // record. Local clarification is submitted as its own admitted Host action. if ( result.kind === 'discussion' || result.kind === 'waiting' || result.kind === 'submitted' || - result.kind === 'stop' + result.kind === 'stop' || + result.kind === 'resume' ) { return result; } try { - await input.controller.recordConversationTurn({ + await input.controller.requestClarification({ turnId: input.request.requestId, userText: input.recordedUserText, assistantText: input.summary(result), - disposition: result.kind === 'clarification' ? 'clarify' : 'summary', + disposition: 'clarify', }); } catch (error) { input.onSummaryError(); @@ -632,6 +633,18 @@ export function WorkHubCoordinationTurnView(props: { copy={copy} onOpenSession={props.onOpenSession} /> + ) : props.turn.resume ? ( + candidate.target.sessionId === props.turn.resume!.targetSessionId, + )} + targetSessionId={props.turn.resume.targetSessionId} + heading={copy.resumeOutcomes[props.turn.resume.outcome]} + state={copy.resumeRequested} + result={undefined} + copy={copy} + onOpenSession={props.onOpenSession} + /> ) : assignment ? ( {} }; }, - recordConversationTurn: async ({ turnId }) => ({ turnId }), + requestClarification: async ({ turnId }) => ({ turnId }), subscribe: () => () => {}, resetVisitContext: () => {}, }; @@ -275,3 +275,20 @@ export const ConversationPromptAnchors: Story = { export const ConversationPromptAnchorsNarrow: Story = { ...ConversationPromptAnchors, }; + + +// Real path: a completed Coordination Run projects host action receipts through +// the shared transcript, with clarification text and a navigable resume target. +export const CoordinationActionReceipts: Story = { + render: () => , + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + await expect(await canvas.findByText('你是指支付回调幂等性任务吗？')).toBeVisible(); + await expect(await canvas.findByText('已让中断的工作继续：')).toBeVisible(); + }, +}; diff --git a/docs/architecture/workhub-coordination-session-adr.md b/docs/architecture/workhub-coordination-session-adr.md index 8b9a848019..e0e5686c78 100644 --- a/docs/architecture/workhub-coordination-session-adr.md +++ b/docs/architecture/workhub-coordination-session-adr.md @@ -48,7 +48,7 @@ durable conversation and execution substrate. The role is provisioned lazily when WorkHub first needs it and resolves to the same Session after Runtime Host or application restarts. The Session role representation, lookup, recovery, and per-Host UI resolution enforce this lifecycle contract. The -coordination transcript and disposition semantics remain separate later work. +coordination transcript and typed dispositions use that same Session substrate. The per-Host boundary is intentional. A Coordination Session coordinates only the ordinary Sessions belonging to the same Runtime Host. Switching Runtime Hosts @@ -73,6 +73,29 @@ It never acquires authority over an ordinary Session's execution or lifecycle. ## Dispositions and action admission +An admitted Coordination request owns a real root Turn and Run in the reserved +WorkHub Session. `answer_here` executes the existing model answer path. Action +Turns execute the Host operation through the same Runtime admission, execution +ownership, terminal commit, and recovery machinery; admission does not require an +extra model call. Intent, Resolver, and clarification can later invoke models +inside this coordination execution without changing target Session authority. + +A successful Action Run writes a host-authored, model-hidden +`RuntimeEvent.actions.coordination` receipt. The transcript projects it as an +`action_receipt`, not an invented assistant response. Clarification carries its +prompt; resume carries the target reference and admission acknowledgement. +The synthetic `workhub.coordination.record` operation is removed. Released history +remains readable without inventing admissions for old summary rows. + +A receipt acknowledges what the operation accepted; it is not the target's current +execution state. Re-delivery of a completed request returns that receipt, including +after restart, without repeating the effect. Failed attempts remain terminal; +a same-action retry gets a subsequent admitted Turn. An interrupted Host action is +closed by Runtime recovery and never replayed as a model answer. Target-owned +claims, assignment atomicity, and resume source-boundary checks still decide +whether an unfinished effect can continue. Transactional delegation/Stop facts +remain authoritative for their existing ownership and linkage projections. + Every WorkHub input resolves to exactly one proposed **disposition**: - `answer_here`: answer in the Coordination Session. @@ -277,8 +300,8 @@ lets the stop reach a terminal resolution. replacement. Its target comes from the shared Session Resolver port, whose first implementation is a temporary exact-name baseline; replacing it changes recall only, because admission revalidates opaque identity and expected state - rather than any display name. Pause, resume, and pronoun-based stop controls - remain later work. + rather than any display name. Named resume uses ordinary Session continuation admission. Pause and + pronoun-based stop controls remain later work. Reevaluate the per-Host decision if supported workflows require one WorkHub conversation to coordinate ordinary Sessions on multiple Runtime Hosts, or if Host diff --git a/packages/core/package.json b/packages/core/package.json index 45d46600f6..8f43f71bd1 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -157,7 +157,8 @@ "./unified-diff": "./dist/unified-diff.js", "./dev-single-instance": "./dist/dev-single-instance.js", "./maka-wordmark": "./dist/maka-wordmark.js", - "./test-only/async-primitives": "./dist/test-only/async-primitives.js" + "./test-only/async-primitives": "./dist/test-only/async-primitives.js", + "./workhub-action-result": "./dist/workhub-action-result.js" }, "scripts": { "clean": "node ../../scripts/clean-paths.mjs dist tsconfig.tsbuildinfo src/model-metadata.generated.ts", diff --git a/packages/core/src/__tests__/runtime-event.test.ts b/packages/core/src/__tests__/runtime-event.test.ts index 3bfb380177..0476d7dd74 100644 --- a/packages/core/src/__tests__/runtime-event.test.ts +++ b/packages/core/src/__tests__/runtime-event.test.ts @@ -893,3 +893,31 @@ describe('RuntimeEvent reference validation', () => { ); }); }); + +test('Coordination Runtime receipts survive decoding and reject unrecognized results', () => { + const coordination = { + actionId: 'action', + userText: 'Which task?', + clarification: 'Name a task.', + result: { disposition: 'clarify' as const, coordinationTurnId: 'turn-1' }, + }; + const event = baseEvent({ + role: 'system', + author: 'host', + modelVisibility: 'hidden', + actions: { coordination }, + }); + assert.deepEqual(decodeRuntimeEvent(event).actions?.coordination, coordination); + assert.throws(() => + decodeRuntimeEvent({ + ...event, + actions: { coordination: { ...coordination, result: { disposition: 'execute_anything' } } }, + }), + ); + assert.throws(() => + decodeRuntimeEvent({ + ...event, + actions: { coordination: { ...coordination, executionStatus: 'completed' } }, + }), + ); +}); diff --git a/packages/core/src/runtime-event.ts b/packages/core/src/runtime-event.ts index 9794423f43..3f1c718d51 100644 --- a/packages/core/src/runtime-event.ts +++ b/packages/core/src/runtime-event.ts @@ -33,6 +33,7 @@ * projection, or ledger logic lives here. Those arrive in later nodes. */ +import { isWorkHubActionReceipt, type WorkHubActionReceipt } from './workhub-action-result.js'; import { isModelRetryDecision, type ModelRetryDecision } from './model-failure.js'; import { @@ -508,6 +509,8 @@ export interface RuntimeEventPermissionClosureAccepted { * event without `actions.endInvocation` MUST assert a terminal `status`. */ export interface RuntimeEventActions { + /** Host coordination receipt linked to this admitted Run. */ + coordination?: WorkHubActionReceipt; /** Durable physical pause; does not complete or cancel the owning logical Turn. */ handoffPause?: RuntimeHandoffPause; /** Patch applied to invocation-scoped runtime state. */ @@ -806,6 +809,7 @@ const RUNTIME_ACTIONS_SHAPE = defineObjectShape()( [], [ 'handoffPause', + 'coordination', 'stateDelta', 'artifactDelta', 'permissionRequest', @@ -1234,6 +1238,7 @@ function isRuntimeEventActions(value: unknown): value is RuntimeEventActions { } return ( (value.handoffPause === undefined || isRuntimeHandoffPause(value.handoffPause)) && + (value.coordination === undefined || isWorkHubActionReceipt(value.coordination)) && (value.stateDelta === undefined || isRecord(value.stateDelta)) && (value.artifactDelta === undefined || (isRecord(value.artifactDelta) && diff --git a/packages/core/src/runtime-invocation.ts b/packages/core/src/runtime-invocation.ts index 9108319565..28a52cd72c 100644 --- a/packages/core/src/runtime-invocation.ts +++ b/packages/core/src/runtime-invocation.ts @@ -241,6 +241,7 @@ export type RootExecutionDescriptor = | { /** Tool-free conversational execution admitted only by WorkHub authority. */ kind: 'workhub_coordination'; + operation?: 'action'; inputDigest: `sha256:${string}`; } | { kind: 'regenerate'; sourceTurnId: string } diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 342ff47eaf..0383f43b04 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -17,6 +17,7 @@ * under the License. */ +import { isWorkHubActionReceipt, type WorkHubActionReceipt } from './workhub-action-result.js'; import { isModelRetryDecision, type ModelRetryDecision } from './model-failure.js'; import { @@ -1108,7 +1109,18 @@ export interface WorkHubActionClaim { export type WorkHubActionClaimOutcome = 'claimed' | 'same_claim' | 'conflict'; +export interface WorkHubCoordinationActionMessage { + type: 'workhub_coordination'; + kind: 'action_receipt'; + schemaVersion: 1; + id: string; + turnId: string; + ts: number; + receipt: WorkHubActionReceipt; +} + export type WorkHubCoordinationMessage = + | WorkHubCoordinationActionMessage | WorkHubDelegationAssignedMessage | WorkHubDelegationReplacementRequestedMessage | WorkHubDelegationReplacementAbortedMessage @@ -1585,6 +1597,16 @@ function decodeMessage( } function isWorkHubCoordinationMessage(message: Record): boolean { + if (message.kind === 'action_receipt') + return ( + hasMessageEnvelope(message, true) && + message.schemaVersion === 1 && + Object.keys(message).every((k) => + ['type', 'kind', 'schemaVersion', 'id', 'turnId', 'ts', 'receipt'].includes(k), + ) && + isWorkHubActionReceipt(message.receipt) + ); + if (message.kind === 'delegation_stop_requested') { return ( hasMessageEnvelope(message, true) && diff --git a/packages/core/src/workhub-action-result.ts b/packages/core/src/workhub-action-result.ts new file mode 100644 index 0000000000..feafef0426 --- /dev/null +++ b/packages/core/src/workhub-action-result.ts @@ -0,0 +1,115 @@ +/* + * 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 { isRecord } from './record-schema.js'; + +export type WorkHubActionResult = + | { readonly disposition: 'answer_here'; readonly coordinationTurnId: string } + | { readonly disposition: 'clarify'; readonly coordinationTurnId: string } + | { + readonly disposition: 'delegate_existing'; + readonly targetSessionId: string; + readonly targetTurnId: string; + readonly steered?: true; + } + | { + readonly disposition: 'create_new'; + readonly targetSessionId: string; + readonly targetTurnId: string; + readonly steered?: true; + } + | { + readonly disposition: 'replace'; + readonly replacementDisposition: 'delegate_existing' | 'create_new'; + readonly targetSessionId: string; + readonly targetTurnId: string; + readonly steered?: true; + } + | { + readonly disposition: 'stop_work'; + readonly outcome: 'cancelled_pending' | 'stop_delivered' | 'already_terminal' | 'not_owned'; + readonly targetSessionId: string; + readonly targetTurnId?: string; + } + | { + readonly disposition: 'resume_work'; + readonly outcome: 'resume_started' | 'already_running'; + readonly targetSessionId: string; + readonly targetTurnId?: string; + }; + +/** Coordination receipt, not a copy of target execution state. */ +export interface WorkHubActionReceipt { + actionId: string; + userText: string; + result: WorkHubActionResult; + clarification?: string; +} + +export function isWorkHubActionReceipt(value: unknown): value is WorkHubActionReceipt { + if ( + !isRecord(value) || + Object.keys(value).some( + (k) => !['actionId', 'userText', 'result', 'clarification'].includes(k), + ) || + typeof value.actionId !== 'string' || + !value.actionId || + typeof value.userText !== 'string' || + !value.userText.trim() || + (value.clarification !== undefined && typeof value.clarification !== 'string') + ) + return false; + const r = value.result; + if (!isRecord(r)) return false; + const text = (k: string) => typeof r[k] === 'string' && r[k] !== ''; + const keys = (allowed: string[]) => Object.keys(r).every((k) => allowed.includes(k)); + if (r.disposition === 'answer_here' || r.disposition === 'clarify') + return keys(['disposition', 'coordinationTurnId']) && text('coordinationTurnId'); + if ( + r.disposition === 'delegate_existing' || + r.disposition === 'create_new' || + r.disposition === 'replace' + ) + return ( + keys([ + 'disposition', + 'targetSessionId', + 'targetTurnId', + 'steered', + ...(r.disposition === 'replace' ? ['replacementDisposition'] : []), + ]) && + text('targetSessionId') && + text('targetTurnId') && + (r.steered === undefined || r.steered === true) && + (r.disposition !== 'replace' || + r.replacementDisposition === 'delegate_existing' || + r.replacementDisposition === 'create_new') + ); + if (r.disposition === 'stop_work' || r.disposition === 'resume_work') + return ( + keys(['disposition', 'outcome', 'targetSessionId', 'targetTurnId']) && + text('targetSessionId') && + (r.targetTurnId === undefined || text('targetTurnId')) && + (r.disposition === 'stop_work' + ? ['cancelled_pending', 'stop_delivered', 'already_terminal', 'not_owned'] + : ['resume_started', 'already_running'] + ).includes(String(r.outcome)) + ); + return false; +} diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index f20fa37f49..6bfea6f1be 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -17,7 +17,9 @@ * under the License. */ +import { workHubCoordinationTurnId } from '../server/workhub-coordination-action-gate.js'; import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; import { runtimeInvocationOutcome } from '@maka/core/runtime-invocation'; import { createRunCompositionSnapshot } from '@maka/core/run-composition'; import type { BackendSendInput } from '@maka/core/backend-types'; @@ -828,7 +830,17 @@ test('production composition commits automatic titles through Host-owned Session test('WorkHub creates new work through the production assignment composition', async () => { await withCompositionRoot(async ({ root, owner }) => { const connectionId = await configureFakeDefaultTarget(owner); - const { composition, manager } = await createCapturedExecutionComposition(owner); + let coordinationModelCalls = 0; + const { composition, manager } = await createCapturedExecutionComposition(owner, { + primaryBackendFactory: (backendContext) => + new (class extends FakeBackend { + override async *send(input: BackendSendInput): AsyncIterable { + if (backendContext.header.id === 'maka_workhub_coordination') + coordinationModelCalls += 1; + yield* super.send(input); + } + })(backendContext), + }); const context = { hostEpoch: 'execution-composition-test', connectionId: 'workhub-create-client', @@ -850,6 +862,76 @@ test('WorkHub creates new work through the production assignment composition', a assert.equal(created.ok, true, JSON.stringify(created)); if (!created.ok || created.result.disposition !== 'create_new') return; + const coordinationStores = await openInteractiveExecutionStoresForWrite(owner.lease); + const admission = await coordinationStores.agentRunStore.readRootTurnAdmission( + 'maka_workhub_coordination', + 'workhub-create-action', + ); + assert.ok(admission, 'Delegation must belong to an admitted Coordination Turn'); + assert.equal(admission.execution.kind, 'workhub_coordination'); + const events = await coordinationStores.runtimeEventStore.readImmutableRuntimeEvents( + admission.sessionId, + admission.runId, + ); + assert.equal( + events.find((event) => event.actions?.coordination)?.actions?.coordination?.result + .disposition, + 'create_new', + ); + assert.ok(events.some((event) => event.status === 'completed')); + assert.equal( + events.some((event) => event.role === 'model'), + false, + ); + const transcript = await coordinationStores.sessionStore.readMessages(admission.sessionId); + assert.ok( + transcript.some( + (message) => message.type === 'workhub_coordination' && message.kind === 'action_receipt', + ), + 'The shared transcript must expose the Runtime receipt to Desktop', + ); + const replayed = await composition.handlers['workhub.coordination.act']( + { + actionId: 'workhub-create-action', + userText: 'Fix login stability', + proposal: { disposition: 'create_new', title: 'Login stability' }, + create: { workspace: { kind: 'host_path', path: root } }, + }, + context, + ); + assert.deepEqual(replayed, created); + const clarifyInput = { + actionId: 'workhub-clarify', + userText: 'Which task?', + proposal: { disposition: 'clarify' as const, assistantText: 'Please name the task.' }, + }; + const clarified = await composition.handlers['workhub.coordination.act']( + clarifyInput, + context, + ); + assert.ok(clarified.ok, JSON.stringify(clarified)); + assert.equal(clarified.result.disposition, 'clarify'); + if (clarified.result.disposition === 'clarify') { + assert.ok( + await coordinationStores.agentRunStore.readRootTurnAdmission( + admission.sessionId, + clarified.result.coordinationTurnId, + ), + ); + } + assert.deepEqual( + await composition.handlers['workhub.coordination.act'](clarifyInput, context), + clarified, + ); + const changed = await composition.handlers['workhub.coordination.act']( + { + ...clarifyInput, + proposal: { disposition: 'clarify', assistantText: 'Different content' }, + }, + context, + ); + assert.equal(changed.ok, false); + assert.equal(coordinationModelCalls, 0, 'Actions must not start a model answer in WorkHub'); const targetSessionId = created.result.targetSessionId; const session = (await manager.listSessions()).find(({ id }) => id === targetSessionId); assert.equal(session?.name, 'Login stability'); @@ -883,6 +965,93 @@ test('WorkHub creates new work through the production assignment composition', a }); }); +test('interrupted Coordination admission recovers without a model and retries in a new Turn', async () => { + await withCompositionRoot(async ({ root, owner }) => { + await configureFakeDefaultTarget(owner); + const context = { + hostEpoch: 'coordination-recovery', + connectionId: 'client', + principal: 'local_os_user' as const, + acquireResidency: () => ({ release() {} }), + }; + const first = await createCapturedExecutionComposition(owner); + await first.composition.handlers['workhub.coordination.resolve']({}, context); + await first.composition.close(); + const request = { + actionId: 'orphaned-action', + userText: 'Which task?', + proposal: { disposition: 'clarify' as const, assistantText: 'Please name a task.' }, + }; + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + await stores.agentRunStore.admitRootTurn({ + sessionId: 'maka_workhub_coordination', + turnId: workHubCoordinationTurnId(request.actionId, 'clarify'), + proposedRunId: 'orphaned-run', + proposedUserMessageId: 'orphaned-user', + execution: { + kind: 'workhub_coordination', + operation: 'action', + inputDigest: `sha256:${createHash('sha256').update(JSON.stringify(request)).digest('hex')}`, + }, + previousRootTurnId: null, + normalizedInput: { text: request.userText }, + sourceMessages: [], + admittedAt: 1, + }); + await owner.close(); + const reopened = await tryAcquireInteractiveRootOwner( + await resolveStorageRoot({ path: root, kind: 'interactive' }), + ); + assert.ok(reopened); + let modelCalls = 0; + try { + const recovered = await createCapturedExecutionComposition(reopened, { + primaryBackendFactory: (backendContext) => + new (class extends FakeBackend { + override async *send(input: BackendSendInput): AsyncIterable { + modelCalls += 1; + yield* super.send(input); + } + })(backendContext), + }); + try { + const reopenedStores = await openInteractiveExecutionStoresForWrite(reopened.lease); + const oldRun = ( + await reopenedStores.runtimeEventStore.listSessionInvocations('maka_workhub_coordination') + ).find((run) => run.runId === 'orphaned-run'); + assert.ok(oldRun); + assert.equal(runtimeInvocationOutcome(oldRun), 'failed'); + const retried = await recovered.composition.handlers['workhub.coordination.act']( + request, + context, + ); + assert.ok(retried.ok, JSON.stringify(retried)); + assert.equal(retried.result.disposition, 'clarify'); + if (retried.result.disposition !== 'clarify') return; + assert.notEqual( + retried.result.coordinationTurnId, + workHubCoordinationTurnId(request.actionId, 'clarify'), + ); + const retryAdmission = await reopenedStores.agentRunStore.readRootTurnAdmission( + 'maka_workhub_coordination', + retried.result.coordinationTurnId, + ); + assert.ok(retryAdmission); + assert.notEqual(retryAdmission.runId, 'orphaned-run'); + assert.deepEqual( + await recovered.composition.handlers['workhub.coordination.act'](request, context), + retried, + ); + assert.equal(modelCalls, 0); + } finally { + await recovered.composition.close(); + } + } finally { + await reopened.close(); + } + }); +}); + test('WorkHub Resume and Stop follow logical lineage across repeated physical handoffs', async () => { await withCompositionRoot(async ({ root, owner }) => { const connectionId = await configureFakeDefaultTarget(owner); @@ -1055,8 +1224,15 @@ test('WorkHub Resume and Stop follow logical lineage across repeated physical ha }, }; const replayed = await composition.handlers['workhub.coordination.act'](retry, context); - assert.equal(replayed.ok, false, JSON.stringify(replayed)); - if (!replayed.ok) assert.equal(replayed.error.code, 'operation_conflict'); + // Re-delivery acknowledges the original Coordination Run; it must never + // resume a later interruption under the same action identity. + assert.equal(replayed.ok, true, JSON.stringify(replayed)); + const stillInterrupted = await composition.handlers['turn.query']( + { sessionId: target.id, turnId: continuation.turnId }, + context, + ); + assert.ok(stillInterrupted.ok); + assert.notEqual(stillInterrupted.result.status, 'running'); const fresh = await composition.handlers['workhub.coordination.act']( { ...retry, actionId: 'workhub-resume-again' }, context, @@ -1216,9 +1392,21 @@ test('WorkHub does not record resume while safe-boundary resume is disabled', as message: 'Safe-boundary resume is disabled for this Runtime Host', }, }); - await composition.close(); const stores = await openInteractiveExecutionStoresForWrite(owner.lease); assert.equal(await stores.sessionStore.readWorkHubActionClaim(actionId), undefined); + const clarified = await composition.handlers['workhub.coordination.act']( + { + actionId, + userText: 'Resume Payments', + proposal: { + disposition: 'clarify', + assistantText: 'Resume is unavailable on this Host.', + }, + }, + context, + ); + assert.ok(clarified.ok, JSON.stringify(clarified)); + assert.equal(clarified.result.disposition, 'clarify'); } finally { await composition.close(); } diff --git a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts index f6ae747120..06fddd5e77 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts @@ -39,10 +39,6 @@ import { } from '@maka/core/session'; import { createSessionStore, type SessionAuthorityStore } from '@maka/storage/session-store'; import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store'; -import { - WORKHUB_COORDINATION_SUMMARY_MAX_BYTES, - WORKHUB_COORDINATION_TEXT_MAX_BYTES, -} from '../protocol/index.js'; import type { ConnectionContext } from '../server/operation-dispatcher.js'; import type { RootTurnCoordinator } from '../server/root-turn-coordinator.js'; import { SessionAdmissionGate } from '../server/session-admission-gate.js'; @@ -430,70 +426,6 @@ describe('Host WorkHub Coordination coordinator', () => { } }); - test('records synthetic coordination summaries durably and retries idempotently', async () => { - const root = await mkdtemp(join(tmpdir(), 'maka-workhub-record-')); - const store = createSessionStore(root); - try { - const workhub = coordinator(root, store); - assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); - const input = { - turnId: 'summary-turn', - userText: 'Continue payment work', - assistantText: 'Submitted to Payment', - }; - assert.deepEqual(await workhub.handlers['workhub.coordination.record'](input, CONTEXT), { - ok: true, - result: { turnId: 'summary-turn' }, - }); - assert.deepEqual(await workhub.handlers['workhub.coordination.record'](input, CONTEXT), { - ok: true, - result: { turnId: 'summary-turn' }, - }); - const maximumInput = { - turnId: 'maximum-summary-turn', - // Each NUL is one UTF-8 input byte but six bytes once JSON-escaped in - // the durable transcript record. Retry lookup must budget for that - // worst case, not only the decoded text sizes. - userText: '\0'.repeat(WORKHUB_COORDINATION_TEXT_MAX_BYTES), - assistantText: '\0'.repeat(WORKHUB_COORDINATION_SUMMARY_MAX_BYTES), - }; - assert.deepEqual( - await workhub.handlers['workhub.coordination.record'](maximumInput, CONTEXT), - { ok: true, result: { turnId: 'maximum-summary-turn' } }, - ); - assert.deepEqual( - await workhub.handlers['workhub.coordination.record'](maximumInput, CONTEXT), - { ok: true, result: { turnId: 'maximum-summary-turn' } }, - ); - const messages = await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID); - assert.equal(messages.length, 6); - assert.deepEqual( - messages.slice(0, 3).map(({ type, turnId }) => ({ type, turnId })), - [ - { type: 'user', turnId: 'summary-turn' }, - { type: 'assistant', turnId: 'summary-turn' }, - { type: 'turn_state', turnId: 'summary-turn' }, - ], - ); - const conflict = await workhub.handlers['workhub.coordination.record']( - { ...input, assistantText: 'Different summary' }, - CONTEXT, - ); - assert.equal(conflict.ok, false); - if (!conflict.ok) assert.equal(conflict.error.code, 'operation_conflict'); - assert.equal((await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID)).length, 6); - const empty = await workhub.handlers['workhub.coordination.record']( - { ...input, turnId: 'empty-summary', assistantText: ' ' }, - CONTEXT, - ); - assert.equal(empty.ok, false); - if (!empty.ok) assert.equal(empty.error.code, 'operation_conflict'); - } finally { - await store.close?.(); - await rm(root, { recursive: true, force: true }); - } - }); - test('persists delegated action ownership and replays it after Host restart', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-workhub-delegation-')); const userText = 'Continue payment work. '.repeat(900); @@ -1921,90 +1853,11 @@ describe('Host WorkHub Coordination coordinator', () => { await rm(root, { recursive: true, force: true }); } }); - - test('refuses to merge a Turn identity shared across answer and record', async () => { - const root = await mkdtemp(join(tmpdir(), 'maka-workhub-turn-identity-')); - const store = createSessionStore(root); - const admission = new SessionAdmissionGate(); - const { executions } = coordinationExecutions(admission); - try { - const workhub = coordinator(root, store, () => undefined, undefined, executions, admission); - assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); - - // An answered Turn is owned by the root admission ledger. - assert.equal( - ( - await workhub.handlers['workhub.coordination.answer']( - { turnId: 'shared-turn', text: 'What is left on payments?' }, - CONTEXT, - ) - ).ok, - true, - ); - const recordAfterAnswer = await workhub.handlers['workhub.coordination.record']( - { turnId: 'shared-turn', userText: 'Continue payments', assistantText: 'Sent to Payments' }, - CONTEXT, - ); - assert.deepEqual(recordAfterAnswer, { - ok: false, - error: { - code: 'operation_conflict', - message: 'WorkHub Coordination Turn identity belongs to a different operation', - }, - }); - assert.deepEqual(await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID), []); - - // A recorded Turn is owned by the durable summary triplet. - assert.equal( - ( - await workhub.handlers['workhub.coordination.record']( - { - turnId: 'recorded-turn', - userText: 'Continue payments', - assistantText: 'Sent to Payments', - }, - CONTEXT, - ) - ).ok, - true, - ); - const answerAfterRecord = await workhub.handlers['workhub.coordination.answer']( - { turnId: 'recorded-turn', text: 'What is left on payments?' }, - CONTEXT, - ); - assert.deepEqual(answerAfterRecord, { - ok: false, - error: { - code: 'operation_conflict', - message: 'WorkHub Coordination Turn identity belongs to a different operation', - }, - }); - assert.deepEqual( - (await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID)).map( - ({ type, turnId }) => ({ type, turnId }), - ), - [ - { type: 'user', turnId: 'recorded-turn' }, - { type: 'assistant', turnId: 'recorded-turn' }, - { type: 'turn_state', turnId: 'recorded-turn' }, - ], - ); - assert.deepEqual( - (await store.listTurnsSnapshot(WORKHUB_COORDINATION_SESSION_ID)).map( - ({ turnId }) => turnId, - ), - ['recorded-turn'], - ); - } finally { - await store.close?.(); - await rm(root, { recursive: true, force: true }); - } - }); }); type CoordinationExecutions = Pick< RootTurnCoordinator, - 'startWorkHubCoordinationMessage' | 'hasRootTurnAdmission' + 'startWorkHubCoordinationMessage' | 'runWorkHubCoordinationOperation' >; /** @@ -2013,7 +1866,6 @@ type CoordinationExecutions = Pick< * reproduce the ordering the real ledger enforces. */ function coordinationExecutions(admission: SessionAdmissionGate) { - const admitted = new Set(); const starts: Parameters[0][] = []; const prepared: MessageContent[] = []; const executions: CoordinationExecutions = { @@ -2023,7 +1875,6 @@ function coordinationExecutions(admission: SessionAdmissionGate) { const content = await request.prepareFreshContent(lease); if (content.kind === 'rejected') return content.outcome; prepared.push(content.content); - admitted.add(request.turnId); return { ok: true, result: { @@ -2035,7 +1886,10 @@ function coordinationExecutions(admission: SessionAdmissionGate) { }; }); }, - hasRootTurnAdmission: async (_sessionId, turnId) => admitted.has(turnId), + runWorkHubCoordinationOperation: async (request) => { + if (!request.operation) throw new Error('Missing operation'); + return { ok: true, result: await request.operation(request.turnId) }; + }, }; return { executions, starts, prepared }; } @@ -2053,7 +1907,10 @@ function coordinator( message: 'WorkHub test execution is not configured', }, }), - hasRootTurnAdmission: async () => false, + runWorkHubCoordinationOperation: async (request) => { + if (!request.operation) throw new Error('Missing operation'); + return { ok: true, result: await request.operation(request.turnId) }; + }, }, admission: SessionAdmissionGate = new SessionAdmissionGate(), sessionActions: Partial = {}, diff --git a/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts index 04a51c1043..370ff26ccd 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts @@ -25,7 +25,6 @@ import { decodeWorkHubCoordinationActResult, decodeWorkHubCoordinationAnswerInput, decodeWorkHubCoordinationCandidatesResult, - decodeWorkHubCoordinationRecordInput, decodeWorkHubCoordinationResolveInput, decodeWorkHubCoordinationResolveResult, HOST_OPERATION_SPECS, @@ -55,18 +54,6 @@ test('WorkHub Coordination answer and summary inputs are closed and bounded', () decodeWorkHubCoordinationAnswerInput({ turnId: 'answer-turn', text: 'What changed?' }), { turnId: 'answer-turn', text: 'What changed?' }, ); - assert.deepEqual( - decodeWorkHubCoordinationRecordInput({ - turnId: 'summary-turn', - userText: 'Continue payment work', - assistantText: 'Submitted to Payment', - }), - { - turnId: 'summary-turn', - userText: 'Continue payment work', - assistantText: 'Submitted to Payment', - }, - ); assert.deepEqual( decodeWorkHubCoordinationActInput({ actionId: 'action-correction', @@ -171,22 +158,11 @@ test('WorkHub Coordination answer and summary inputs are closed and bounded', () (error) => error instanceof RuntimeHostProtocolError, ); assert.equal(HOST_OPERATION_SPECS['workhub.coordination.answer'].mode, 'command'); - assert.equal(HOST_OPERATION_SPECS['workhub.coordination.record'].mode, 'command'); assert.equal(REMOTE_OWNER_OPERATION_GRANTS.includes('workhub.coordination.answer'), true); - assert.equal(REMOTE_OWNER_OPERATION_GRANTS.includes('workhub.coordination.record'), true); assert.throws( () => decodeWorkHubCoordinationAnswerInput({ turnId: 'turn', text: 'answer', extra: true }), (error) => error instanceof RuntimeHostProtocolError, ); - assert.throws( - () => - decodeWorkHubCoordinationRecordInput({ - turnId: 'turn', - userText: 'user', - assistantText: 'x'.repeat(8 * 1024 + 1), - }), - (error) => error instanceof RuntimeHostProtocolError, - ); }); test('WorkHub Coordination resume has closed input and outcome shapes', () => { diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 536c7aa667..70a335284e 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -101,7 +101,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 = 125 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 126 as const; +// 126: Coordination actions own real Runtime Turns. Removes the synthetic record +// operation and projects typed action receipts; older peers cannot decode them. // 125: Live Turn snapshots carry an optional `rootExecutionKind:'context_compact'` // so a running context-compaction Turn can render a transcript row. Epoch-124 // peers reject the added optional field on the strict live snapshot shape. diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 1134fdb917..cbc856d960 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -356,7 +356,6 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'workhub.coordination.answer', 'workhub.coordination.act', 'workhub.coordination.candidates', - 'workhub.coordination.record', 'workhub.coordination.resolve', ] as const satisfies readonly OperationKey[]); diff --git a/packages/runtime-host/src/protocol/workhub-coordination.ts b/packages/runtime-host/src/protocol/workhub-coordination.ts index ebbd1092ef..ac366a79bb 100644 --- a/packages/runtime-host/src/protocol/workhub-coordination.ts +++ b/packages/runtime-host/src/protocol/workhub-coordination.ts @@ -82,12 +82,6 @@ export interface WorkHubCoordinationAnswerInput { readonly text: string; } -export interface WorkHubCoordinationRecordInput { - readonly turnId: string; - readonly userText: string; - readonly assistantText: string; -} - export interface WorkHubCoordinationTurnResult { readonly turnId: string; } @@ -183,40 +177,8 @@ export interface WorkHubCoordinationActInput { readonly confirmation?: WorkHubCoordinationDestructiveConfirmation; } -export type WorkHubCoordinationActResult = - | { readonly disposition: 'answer_here'; readonly coordinationTurnId: string } - | { readonly disposition: 'clarify'; readonly coordinationTurnId: string } - | { - readonly disposition: 'delegate_existing'; - readonly targetSessionId: string; - readonly targetTurnId: string; - readonly steered?: true; - } - | { - readonly disposition: 'create_new'; - readonly targetSessionId: string; - readonly targetTurnId: string; - readonly steered?: true; - } - | { - readonly disposition: 'replace'; - readonly replacementDisposition: 'delegate_existing' | 'create_new'; - readonly targetSessionId: string; - readonly targetTurnId: string; - readonly steered?: true; - } - | { - readonly disposition: 'stop_work'; - readonly outcome: 'cancelled_pending' | 'stop_delivered' | 'already_terminal' | 'not_owned'; - readonly targetSessionId: string; - readonly targetTurnId?: string; - } - | { - readonly disposition: 'resume_work'; - readonly outcome: 'resume_started' | 'already_running'; - readonly targetSessionId: string; - readonly targetTurnId?: string; - }; +export type { WorkHubActionResult as WorkHubCoordinationActResult } from '@maka/core/workhub-action-result'; +import type { WorkHubActionResult as WorkHubCoordinationActResult } from '@maka/core/workhub-action-result'; export const WORKHUB_COORDINATION_OPERATION_SPECS = { 'workhub.coordination.resolve': defineOperation< @@ -241,17 +203,6 @@ export const WORKHUB_COORDINATION_OPERATION_SPECS = { decodeInput: decodeWorkHubCoordinationAnswerInput, decodeOutput: decodeWorkHubCoordinationTurnResult, }), - 'workhub.coordination.record': defineOperation< - WorkHubCoordinationRecordInput, - WorkHubCoordinationTurnResult, - (typeof TURN_ERRORS)[number] - >({ - mode: 'command', - availability: 'ready', - errors: TURN_ERRORS, - decodeInput: decodeWorkHubCoordinationRecordInput, - decodeOutput: decodeWorkHubCoordinationTurnResult, - }), 'workhub.coordination.candidates': defineOperation< WorkHubCoordinationCandidatesInput, WorkHubCoordinationCandidatesResult, @@ -306,29 +257,6 @@ export function decodeWorkHubCoordinationAnswerInput( }; } -export function decodeWorkHubCoordinationRecordInput( - value: unknown, -): WorkHubCoordinationRecordInput { - const input = requireExactRecord(value, 'WorkHub Coordination record input', [ - 'turnId', - 'userText', - 'assistantText', - ]); - return { - turnId: requireEntityId(input.turnId, 'WorkHub Coordination Turn id'), - userText: requireUtf8String( - input.userText, - 'WorkHub Coordination user text', - WORKHUB_COORDINATION_TEXT_MAX_BYTES, - ), - assistantText: requireUtf8String( - input.assistantText, - 'WorkHub Coordination assistant text', - WORKHUB_COORDINATION_SUMMARY_MAX_BYTES, - ), - }; -} - export function decodeWorkHubCoordinationTurnResult(value: unknown): WorkHubCoordinationTurnResult { const result = requireExactRecord(value, 'WorkHub Coordination Turn result', ['turnId']); return { diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index c8f4af9ae9..23f38f5d66 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1563,13 +1563,13 @@ export async function createExecutionRuntimeHostComposition( .update(input.replacesDelegationId, 'utf8') .digest('hex') .slice(0, 48)}`, - turnId: input.actionId, + turnId: input.coordinationTurnId ?? input.actionId, ts: assignedAt, schemaVersion: WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, kind: 'delegation_superseded' as const, actionId: input.actionId, actionFingerprint: input.actionFingerprint, - coordinationTurnId: input.actionId, + coordinationTurnId: input.coordinationTurnId ?? input.actionId, supersededActionId: input.replacesActionId, supersededDelegationId: input.replacesDelegationId, replacementDelegationId: delegationId, @@ -1579,7 +1579,7 @@ export async function createExecutionRuntimeHostComposition( assignment: { type: 'workhub_coordination', id: `wha_${suffix}`, - turnId: input.actionId, + turnId: input.coordinationTurnId ?? input.actionId, ts: assignedAt, schemaVersion: supersession ? WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION @@ -1587,7 +1587,7 @@ export async function createExecutionRuntimeHostComposition( kind: 'delegation_assigned', actionId: input.actionId, actionFingerprint: input.actionFingerprint, - coordinationTurnId: input.actionId, + coordinationTurnId: input.coordinationTurnId ?? input.actionId, targetSessionId: input.targetSessionId, targetSessionName: input.targetSessionName, targetTurnId: turnId, diff --git a/packages/runtime-host/src/server/hosted-execution-recovery.ts b/packages/runtime-host/src/server/hosted-execution-recovery.ts index 4398157964..f9360d1cb5 100644 --- a/packages/runtime-host/src/server/hosted-execution-recovery.ts +++ b/packages/runtime-host/src/server/hosted-execution-recovery.ts @@ -399,7 +399,11 @@ function recoveryExecutionContract(execution: RootExecutionDescriptor): Recovery case 'external_message': return contract(true, true, 'root_replay'); case 'workhub_coordination': - return contract(false, true, 'root_replay'); + return contract( + false, + true, + execution.operation === 'action' ? 'host_recovery_closure' : 'root_replay', + ); case 'regenerate': return contract(false, true, 'root_replay'); case 'context_compact': @@ -436,6 +440,7 @@ function usesHostRecoveryClosure(execution: RootExecutionDescriptor): execution RootExecutionDescriptor, { kind: + | 'workhub_coordination' | 'goal' | 'legacy_automation' | 'agent_graph_supervisor_wake' @@ -446,6 +451,7 @@ function usesHostRecoveryClosure(execution: RootExecutionDescriptor): execution } > { return ( + (execution.kind === 'workhub_coordination' && execution.operation === 'action') || execution.kind === 'legacy_automation' || execution.kind === 'goal' || execution.kind === 'agent_graph_supervisor_wake' || diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index 76a6208b09..ae5bb91922 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -17,6 +17,7 @@ * under the License. */ +import type { WorkHubActionReceipt } from '@maka/core/workhub-action-result'; import { createHash, randomUUID } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import type { BackendStopMode } from '@maka/core/backend-types'; @@ -222,6 +223,7 @@ export type RootMessageStartRequest = }) | (RootMessageStartRequestBase & { readonly execution: Extract; + readonly operation?: (turnId: string) => Promise; readonly turnOrchestration?: undefined; prepareFreshContent(lease: SessionAdmissionLease): Promise; }); @@ -1678,16 +1680,93 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { operationUnavailable('WorkHub Coordination execution requires its reserved Session'), ); } + if ((request.execution.operation === 'action') !== (request.operation !== undefined)) { + return Promise.resolve( + operationUnavailable('Coordination execution mode does not match its runner'), + ); + } return this.startRootMessage(request, context); } - /** - * Whether a durable root Turn already owns this identity. WorkHub also writes - * Coordination Turns outside this coordinator, and must not append a second - * triplet into a Turn this admission ledger already owns. - */ - async hasRootTurnAdmission(sessionId: string, turnId: string): Promise { - return (await this.stores.agentRunStore.readRootTurnAdmission(sessionId, turnId)) !== undefined; + async runWorkHubCoordinationOperation( + request: Extract, + context: ConnectionContext, + ): Promise< + { ok: true; result: WorkHubActionReceipt } | Extract + > { + if (request.execution.operation !== 'action' || !request.operation) { + return operationUnavailable('Coordination action requires a Host operation'); + } + let turnId = request.turnId; + // A retry preserves the action identity, but never reopens a terminal Run. + // The existing admission chain and terminal facts identify prior attempts. + for (;;) { + const admission = await this.stores.agentRunStore.readRootTurnAdmission( + request.sessionId, + turnId, + ); + if (!admission) break; + if (!isDeepStrictEqual(admission.execution, request.execution)) + return operationConflict('Coordination action identity belongs to different content'); + const run = await this.readRunIfPresent(request.sessionId, admission.runId); + if (!run) break; + const snapshot = await this.readCanonicalSnapshot( + request.sessionId, + turnId, + admission.runId, + run, + ); + if (!isTerminalSnapshot(snapshot) || snapshot.status === 'completed') break; + turnId = `whretry_${createHash('sha256').update(admission.runId).digest('hex').slice(0, 48)}`; + } + const started = await this.startWorkHubCoordinationMessage({ ...request, turnId }, context); + if (!started.ok) return started; + const active = this.#executions.get(request.sessionId); + if (active?.turnId === turnId) await active.done; + const snapshot = await this.readCanonicalSnapshot( + request.sessionId, + turnId, + started.result.runId, + ); + if (snapshot.status !== 'completed') + return operationUnavailable('Coordination operation did not complete'); + const events = await this.stores.runtimeEventStore.readImmutableRuntimeEvents( + request.sessionId, + started.result.runId, + ); + const receipt = events.find((event) => event.actions?.coordination)?.actions?.coordination; + return receipt + ? { ok: true, result: receipt } + : operationUnavailable('Coordination receipt is unavailable'); + } + + private coordinationOperation( + request: RootMessageStartRequest, + admission: RootTurnAdmission, + ): HostedExecutionAdmission | undefined { + if ( + request.execution.kind !== 'workhub_coordination' || + !('operation' in request) || + !request.operation + ) + return undefined; + const execute = request.operation; + const content = requireHostedExecutionMessageContent(admission); + return { + sessionId: admission.sessionId, + turnId: admission.turnId, + runId: admission.runId, + userMessageId: admission.userMessageId, + execution: admission.execution, + content, + start: ({ runId, userMessageId, onRunStarted }) => + this.manager.runCoordinationOperation( + admission.sessionId, + { turnId: admission.turnId, ...content }, + { runId, userMessageId, onRunStarted }, + () => execute(admission.turnId), + ), + }; } private startRootMessage( @@ -1759,7 +1838,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { context.acquireResidency, lease, undefined, - undefined, + this.coordinationOperation(request, existing), reservation, ); } @@ -1872,7 +1951,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { context.acquireResidency, lease, undefined, - undefined, + this.coordinationOperation(request, admitted.admission), reservation, ); }); diff --git a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts index 8d701e46d9..1584aad240 100644 --- a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts +++ b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts @@ -174,7 +174,10 @@ export interface WorkHubRetirementResult { readonly targetTurnId?: string; } +type AdmittedWorkHubAction = WorkHubCoordinationActInput & { readonly coordinationTurnId?: string }; + export interface WorkHubDelegationAssignmentInput { + readonly coordinationTurnId?: string; readonly actionId: string; readonly actionFingerprint: `sha256:${string}`; readonly targetSessionId: string; @@ -199,6 +202,7 @@ export interface WorkHubDelegationReplacementAbortInput { } export interface WorkHubDelegationStopInput { + readonly coordinationTurnId?: string; readonly actionId: string; readonly actionFingerprint: `sha256:${string}`; readonly stopsActionId: string; @@ -279,9 +283,32 @@ export class WorkHubCoordinationActionGate { return candidateSet(await this.#effects.listSessions()); } + /** Bind admitted retries to the same stable replacement destination as the Gate. */ + async coordinationInputDigest(input: WorkHubCoordinationActInput): Promise<`sha256:${string}`> { + if (input.proposal.disposition !== 'replace') + return digest({ ...input, candidateSetId: undefined }); + const replaced = await this.#effects.readAssignment(input.proposal.replacesActionId); + if (!replaced) + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub replacement source is unavailable', + ); + const prepared = await this.#effects.readReplacement(replaced.delegationId); + const assigned = await this.#effects.readAssignment(input.actionId); + const destination = assigned?.targetSessionId ?? prepared?.targetSessionId; + if (destination) await this.#assertReplacementReplayTarget(input, destination); + const targetSessionId = + destination ?? (await this.#replacementAssignment(input, replaced)).targetSessionId; + return digest({ + fingerprint: replacementActionFingerprint(input, targetSessionId), + confirmation: input.confirmation, + }); + } + act( input: WorkHubCoordinationActInput, context: ConnectionContext, + admittedTurnId?: string, ): Promise { if (!input.userText.trim()) { return Promise.reject( @@ -308,7 +335,11 @@ export class WorkHubCoordinationActionGate { return replay.result; } - const result = this.#act(input, fingerprint, context); + const result = this.#act( + { ...input, ...(admittedTurnId ? { coordinationTurnId: admittedTurnId } : {}) }, + fingerprint, + context, + ); const action = { requestFingerprint, result }; this.#actions.set(input.actionId, action); // Successful actions remain a Host-lifetime fast path. Rejections release @@ -325,7 +356,7 @@ export class WorkHubCoordinationActionGate { } async #act( - input: WorkHubCoordinationActInput, + input: AdmittedWorkHubAction, fingerprint: `sha256:${string}`, context: ConnectionContext, ): Promise { @@ -358,14 +389,20 @@ export class WorkHubCoordinationActionGate { return this.#assign(assignmentInputFromRecord(durable), context); } if (proposal.disposition === 'answer_here') { - const turnId = coordinationTurnId(input.actionId, 'answer'); + const turnId = workHubCoordinationTurnId(input.actionId, 'answer'); await this.#claimAction(input.actionId, 'answer_here', fingerprint, turnId); await this.#effects.answer({ turnId, text: input.userText }, context); return { disposition: 'answer_here', coordinationTurnId: turnId }; } if (proposal.disposition === 'clarify') { - const turnId = coordinationTurnId(input.actionId, 'clarify'); - await this.#claimAction(input.actionId, 'clarify', fingerprint, turnId); + const turnId = + input.coordinationTurnId ?? workHubCoordinationTurnId(input.actionId, 'clarify'); + await this.#claimAction( + input.actionId, + 'clarify', + fingerprint, + workHubCoordinationTurnId(input.actionId, 'clarify'), + ); await this.#effects.clarify({ turnId, userText: input.userText, @@ -426,6 +463,7 @@ export class WorkHubCoordinationActionGate { } const requested = await this.#effects.prepareStop({ actionId: input.actionId, + ...(input.coordinationTurnId ? { coordinationTurnId: input.coordinationTurnId } : {}), actionFingerprint: stopFingerprint, stopsActionId: source.actionId, stopsDelegationId: source.delegationId, @@ -734,7 +772,7 @@ export class WorkHubCoordinationActionGate { } async #replacementAssignment( - input: WorkHubCoordinationActInput, + input: AdmittedWorkHubAction, replaced: WorkHubDelegationAssignedMessage, ): Promise { if (input.proposal.disposition !== 'replace') { @@ -757,6 +795,7 @@ export class WorkHubCoordinationActionGate { const targetSessionId = workHubCreatedSessionId(input.actionId); return { actionId: input.actionId, + ...(input.coordinationTurnId ? { coordinationTurnId: input.coordinationTurnId } : {}), actionFingerprint: replacementActionFingerprint(input, targetSessionId), targetSessionId, targetSessionName: target.title, @@ -802,6 +841,7 @@ export class WorkHubCoordinationActionGate { } return { actionId: input.actionId, + ...(input.coordinationTurnId ? { coordinationTurnId: input.coordinationTurnId } : {}), actionFingerprint: replacementActionFingerprint(input, destination.sessionId), targetSessionId: destination.sessionId, targetSessionName: destination.sessionName, @@ -1058,12 +1098,12 @@ function candidateRef(candidateSetId: string, sessionId: string): string { return `whc_${hash(`${candidateSetId}\0${sessionId}`).slice(0, 48)}`; } -function coordinationTurnId(actionId: string, kind: 'answer' | 'clarify'): string { +export function workHubCoordinationTurnId(actionId: string, kind: 'answer' | 'clarify'): string { return `wha_${hash(`${actionId}\0${kind}`).slice(0, 48)}`; } function delegationAssignment( - input: WorkHubCoordinationActInput, + input: AdmittedWorkHubAction, actionFingerprint: `sha256:${string}`, targetSessionId: string, targetSessionName: string, @@ -1080,6 +1120,7 @@ function delegationAssignment( } const base = { actionId: input.actionId, + coordinationTurnId: input.coordinationTurnId ?? input.actionId, actionFingerprint, targetSessionId, targetSessionName, @@ -1256,6 +1297,7 @@ function assignmentInputFromRecord( ): WorkHubDelegationAssignmentInput { return { actionId: assignment.actionId, + coordinationTurnId: assignment.coordinationTurnId, actionFingerprint: assignment.actionFingerprint, targetSessionId: assignment.targetSessionId, targetSessionName: assignment.targetSessionName, @@ -1276,6 +1318,7 @@ function assignmentInputFromReplacement( ): WorkHubDelegationAssignmentInput { return { actionId: replacement.actionId, + coordinationTurnId: replacement.coordinationTurnId, actionFingerprint: replacement.actionFingerprint, targetSessionId: replacement.targetSessionId, targetSessionName: replacement.targetSessionName, diff --git a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts index 1a33b3095c..dff2a45594 100644 --- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts +++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts @@ -44,7 +44,6 @@ import type { WorkHubCoordinationActResult, WorkHubCoordinationActInput, WorkHubCoordinationAnswerInput, - WorkHubCoordinationRecordInput, } from '../protocol/index.js'; import { WORKHUB_COORDINATION_SUMMARY_MAX_BYTES, @@ -63,6 +62,7 @@ import { WorkHubActionGateFailure, WorkHubCoordinationActionGate, type WorkHubActionGateEffects, + workHubCoordinationTurnId, } from './workhub-coordination-action-gate.js'; const CREATE_FINGERPRINT = `sha256:${createHash('sha256') @@ -70,7 +70,6 @@ const CREATE_FINGERPRINT = `sha256:${createHash('sha256') .digest('hex')}`; const COORDINATION_CWD_DIRECTORY = 'workhub-coordination'; const COORDINATION_TOOL_PROFILE = 'workhub-coordination-v1' as const; -const SYNTHETIC_COORDINATION_MODEL_ID = 'maka-workhub-coordination'; const COORDINATION_PERMISSION_MODE = 'explore' as const; const COORDINATION_COLLABORATION_MODE = 'agent' as const; const COORDINATION_ORCHESTRATION_MODE = 'default' as const; @@ -108,7 +107,7 @@ type CoordinationStores = Pick< type CoordinationExecutions = Pick< RootTurnCoordinator, - 'startWorkHubCoordinationMessage' | 'hasRootTurnAdmission' + 'startWorkHubCoordinationMessage' | 'runWorkHubCoordinationOperation' >; type WorkHubResumeResult = @@ -147,7 +146,6 @@ export class HostWorkHubCoordinationCoordinator { readonly handlers: WorkHubCoordinationOperationHandlerMap = { 'workhub.coordination.resolve': () => this.#resolve(), 'workhub.coordination.answer': (input, context) => this.#answer(input, context), - 'workhub.coordination.record': (input) => this.#record(input), 'workhub.coordination.candidates': () => this.#candidates(), 'workhub.coordination.act': (input, context) => this.#act(input, context), }; @@ -202,16 +200,8 @@ export class HostWorkHubCoordinationCoordinator { throw new WorkHubActionEffectFailure(outcome.error.code, outcome.error.message); } }, - clarify: async (input) => { - const outcome = await this.#record({ - turnId: input.turnId, - userText: input.userText, - assistantText: input.assistantText, - }); - if (!outcome.ok) { - throw new WorkHubActionEffectFailure(outcome.error.code, outcome.error.message); - } - }, + // Clarification is recorded by the admitted Run as a host receipt. + clarify: async () => undefined, assign: options.sessionActions.assign, prepareReplacement: (input) => this.#prepareReplacement(input), abortReplacement: (input) => this.#abortReplacement(input), @@ -237,13 +227,14 @@ export class HostWorkHubCoordinationCoordinator { build: (existing) => ({ type: 'workhub_coordination', id: `whp_${suffix}`, - turnId: input.actionId, + turnId: existing?.turnId ?? input.coordinationTurnId ?? input.actionId, ts: existing?.ts ?? Date.now(), schemaVersion: WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, kind: 'delegation_replacement_requested', actionId: input.actionId, actionFingerprint: input.actionFingerprint, - coordinationTurnId: input.actionId, + coordinationTurnId: + existing?.coordinationTurnId ?? input.coordinationTurnId ?? input.actionId, targetSessionId: input.targetSessionId, targetSessionName: input.targetSessionName, disposition: input.disposition, @@ -310,13 +301,14 @@ export class HostWorkHubCoordinationCoordinator { build: (existing) => ({ type: 'workhub_coordination', id: `whq_${suffix}`, - turnId: input.actionId, + turnId: existing?.turnId ?? input.coordinationTurnId ?? input.actionId, ts: existing?.ts ?? Date.now(), schemaVersion: WORKHUB_COORDINATION_STOP_SCHEMA_VERSION, kind: 'delegation_stop_requested', actionId: input.actionId, actionFingerprint: input.actionFingerprint, - coordinationTurnId: input.actionId, + coordinationTurnId: + existing?.coordinationTurnId ?? input.coordinationTurnId ?? input.actionId, stopsActionId: input.stopsActionId, stopsDelegationId: input.stopsDelegationId, targetSessionId: input.targetSessionId, @@ -521,7 +513,51 @@ export class HostWorkHubCoordinationCoordinator { context: ConnectionContext, ): Promise> { try { - return { ok: true, result: await this.#actionGate.act(input, context) }; + if (input.proposal.disposition === 'answer_here') { + return { ok: true, result: await this.#actionGate.act(input, context) }; + } + const coordinationTurnId = + input.proposal.disposition === 'clarify' + ? workHubCoordinationTurnId(input.actionId, 'clarify') + : input.actionId; + let result: WorkHubCoordinationActResult | undefined; + let failure: unknown; + const outcome = await this.#executions.runWorkHubCoordinationOperation( + { + sessionId: WORKHUB_COORDINATION_SESSION_ID, + turnId: coordinationTurnId, + execution: { + kind: 'workhub_coordination', + operation: 'action', + inputDigest: await this.#actionGate.coordinationInputDigest(input), + }, + archivedMessage: 'WorkHub Coordination Session is unavailable', + prepareFreshContent: async () => + (await this.#readSummaryMessages(coordinationTurnId)).length > 0 + ? { kind: 'rejected', outcome: turnIdentityConflict() } + : { kind: 'ready', content: normalizeMessageContent({ text: input.userText }) }, + operation: async (turnId) => { + try { + result = await this.#actionGate.act(input, context, turnId); + return { + actionId: input.actionId, + userText: input.userText, + result, + ...(input.proposal.disposition === 'clarify' + ? { clarification: input.proposal.assistantText } + : {}), + }; + } catch (error) { + failure = error; + throw error; + } + }, + }, + context, + ); + if (failure) throw failure; + if (!outcome.ok) return outcome; + return { ok: true, result: outcome.result.result }; } catch (error) { if (error instanceof WorkHubActionEffectFailure) { return { @@ -645,9 +681,8 @@ export class HostWorkHubCoordinationCoordinator { inputDigest: digest({ text: input.text }), }, archivedMessage: 'WorkHub Coordination Session is unavailable', - // A recorded summary owns its Turn identity durably but is admitted - // outside this ledger, so the probe runs under the admission lease: a - // concurrent `record` cannot slip a second triplet into this Turn. + // Released summaries have no admission row. Keep their Turn identities + // reserved when admitting a new Runtime-owned answer. prepareFreshContent: async () => { let recorded: readonly StoredMessage[]; try { @@ -670,76 +705,7 @@ export class HostWorkHubCoordinationCoordinator { return outcome.ok ? { ok: true, result: { turnId: input.turnId } } : outcome; } - #record( - input: WorkHubCoordinationRecordInput, - ): Promise> { - if (!input.userText.trim() || !input.assistantText.trim()) { - return Promise.resolve( - turnFailure( - 'workhub.coordination.record', - 'operation_conflict', - 'WorkHub Coordination summary text is empty', - ), - ); - } - return this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, async (lease) => { - let header: SessionHeader; - try { - header = await this.#stores.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID); - } catch { - return turnFailure( - 'workhub.coordination.record', - 'persistence_failed', - 'WorkHub Coordination Session state is unavailable', - ); - } - if (!validCoordinationHeader(header)) { - return turnFailure( - 'workhub.coordination.record', - 'operation_conflict', - 'WorkHub Coordination Session identity is unavailable', - ); - } - - const messages = coordinationSummaryMessages(input); - try { - const existing = await this.#readSummaryMessages(input.turnId); - if (existing.length > 0) { - return coordinationSummaryMatches(existing, input) - ? { ok: true, result: { turnId: input.turnId } } - : turnFailure( - 'workhub.coordination.record', - 'operation_conflict', - 'WorkHub Coordination Turn identity belongs to different content', - ); - } - // An answer owns its Turn identity in the root admission ledger. Both - // operations take the same Session admission, so this probe settles the - // race in one direction and the answer's own probe settles the other. - if ( - await this.#executions.hasRootTurnAdmission(WORKHUB_COORDINATION_SESSION_ID, input.turnId) - ) { - return turnFailure( - 'workhub.coordination.record', - 'operation_conflict', - TURN_IDENTITY_CONFLICT_MESSAGE, - ); - } - await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, messages); - await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); - return { ok: true, result: { turnId: input.turnId } }; - } catch { - this.#requestDrain(); - return turnFailure( - 'workhub.coordination.record', - 'commit_outcome_unknown', - 'WorkHub Coordination summary outcome is unknown', - ); - } - }); - } - - /** Reads the durable summary triplet a `record` would own for this Turn. */ + /** Reads released synthetic summaries solely to reject identity collisions. */ async #readSummaryMessages(turnId: string): Promise { const throughSequence = await this.#stores.readTranscriptHighWaterSnapshot( WORKHUB_COORDINATION_SESSION_ID, @@ -858,56 +824,6 @@ function coordinationSummaryMessageId( .slice(0, 48)}`; } -function coordinationSummaryMessages(input: WorkHubCoordinationRecordInput): StoredMessage[] { - const ts = Date.now(); - const messageId = (kind: (typeof COORDINATION_SUMMARY_MESSAGE_KINDS)[number]) => - coordinationSummaryMessageId(input.turnId, kind); - return [ - { - type: 'user', - id: messageId('user'), - turnId: input.turnId, - ts, - text: input.userText, - }, - { - type: 'assistant', - id: messageId('assistant'), - turnId: input.turnId, - ts: ts + 1, - text: input.assistantText, - modelId: SYNTHETIC_COORDINATION_MODEL_ID, - }, - { - type: 'turn_state', - id: messageId('state'), - turnId: input.turnId, - ts: ts + 2, - status: 'completed', - partialOutputRetained: false, - }, - ]; -} - -function coordinationSummaryMatches( - existing: readonly StoredMessage[], - input: WorkHubCoordinationRecordInput, -): boolean { - if (existing.length !== 3) return false; - const user = existing.find((message) => message.type === 'user'); - const assistant = existing.find((message) => message.type === 'assistant'); - const state = existing.find((message) => message.type === 'turn_state'); - return ( - user?.turnId === input.turnId && - user.text === input.userText && - assistant?.turnId === input.turnId && - assistant.text === input.assistantText && - assistant.modelId === SYNTHETIC_COORDINATION_MODEL_ID && - state?.turnId === input.turnId && - state.status === 'completed' - ); -} - function success(): OperationOutcome<'workhub.coordination.resolve'> { return { ok: true, @@ -941,7 +857,7 @@ function operationUnavailable(message: string) { return { ok: false, error: { code: 'operation_unavailable', message } } as const; } -function turnFailure( +function turnFailure( _operation: K, code: Extract, { readonly ok: false }>['error']['code'], message: string, diff --git a/packages/runtime/src/__tests__/agent-run-steering-recovery.test.ts b/packages/runtime/src/__tests__/agent-run-steering-recovery.test.ts index 0c1d48965b..e1ee43b5ba 100644 --- a/packages/runtime/src/__tests__/agent-run-steering-recovery.test.ts +++ b/packages/runtime/src/__tests__/agent-run-steering-recovery.test.ts @@ -361,6 +361,25 @@ test('recovers a steering transcript message from the committed RuntimeEvent led refs: { providerEventId: 'message-steering-crash-cut' }, }; await runtimeEventStore.appendRuntimeEvent(session.id, runId, runtimeEvent); + const receipt = { + actionId: 'coordination-action', + userText: 'Which task?', + clarification: 'Name the task.', + result: { disposition: 'clarify' as const, coordinationTurnId: turnId }, + }; + await runtimeEventStore.appendRuntimeEvent(session.id, runId, { + id: 'coordination-receipt', + invocationId: runtimeEvent.invocationId, + sessionId: session.id, + runId, + turnId, + ts: 3, + partial: false, + role: 'system', + author: 'host', + modelVisibility: 'hidden', + actions: { coordination: receipt }, + }); assert.deepEqual(await store.readMessages(session.id), []); const recoveredStore = createSessionStore(root); @@ -374,8 +393,8 @@ test('recovers a steering transcript message from the committed RuntimeEvent led now: () => 10, }); - assert.equal(await repair.repairSteeringMessagesOnce(session.id), 1); - assert.equal(await repair.repairSteeringMessagesOnce(session.id), 0); + assert.equal(await repair.repairRuntimeEventTranscriptOnce(session.id), 2); + assert.equal(await repair.repairRuntimeEventTranscriptOnce(session.id), 0); assert.deepEqual(await recoveredStore.readMessages(session.id), [ { type: 'user', @@ -389,6 +408,15 @@ test('recovers a steering transcript message from the committed RuntimeEvent led inlineReferences: steeringContent.inlineReferences, steeringEventId: runtimeEvent.id, }, + { + type: 'workhub_coordination', + kind: 'action_receipt', + schemaVersion: 1, + id: 'coordination-receipt', + turnId, + ts: 3, + receipt, + }, ]); } finally { await rm(root, { recursive: true, force: true }); diff --git a/packages/runtime/src/__tests__/runtime-event-read-model.test.ts b/packages/runtime/src/__tests__/runtime-event-read-model.test.ts index 97582e1db2..2cb8a9ee36 100644 --- a/packages/runtime/src/__tests__/runtime-event-read-model.test.ts +++ b/packages/runtime/src/__tests__/runtime-event-read-model.test.ts @@ -1849,6 +1849,14 @@ type ActionCoverageSamples = { }; const ACTION_COVERAGE_SAMPLES: ActionCoverageSamples = { + coordination: { + action: { + actionId: 'clarify', + userText: 'Which task?', + result: { disposition: 'clarify', coordinationTurnId: 'turn-1' }, + clarification: 'Please name a task.', + }, + }, handoffPause: { action: { protocol: 'runtime_handoff_pause_v1', @@ -2303,3 +2311,32 @@ function makeHeader(id: string): SessionHeader { schemaVersion: 1, }; } + +test('Coordination receipts materialize as host facts, never assistant output', () => { + const receipt = { + actionId: 'clarify', + userText: 'Which task?', + clarification: 'Please name a task.', + result: { disposition: 'clarify' as const, coordinationTurnId: turnId }, + }; + const out = projectRuntimeEventsToStoredMessages( + [ + ev({ + id: 'coordination', + author: 'host', + modelVisibility: 'hidden', + actions: { coordination: receipt }, + }), + ], + { invocations: [invocation] }, + ); + assert.ok( + out.messages.some( + (message) => message.type === 'workhub_coordination' && message.kind === 'action_receipt', + ), + ); + assert.equal( + out.messages.some((message) => message.type === 'assistant'), + false, + ); +}); diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 8ab0d367f0..6197a4f6b4 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -12011,6 +12011,15 @@ class GatedSteeringBackend implements AgentBackend { } class DelegatingRuntimeKernel implements RuntimeKernelLike { + async *runCoordinationOperation( + _sessionId: string, + _input: Parameters[1], + _options: unknown, + execute: Parameters[3], + ): AsyncIterable { + await execute(); + } + readonly starts: Array<{ sessionId: string; input: Parameters[1]; diff --git a/packages/runtime/src/runtime-event-read-model.ts b/packages/runtime/src/runtime-event-read-model.ts index f0478bf0ad..476846773f 100644 --- a/packages/runtime/src/runtime-event-read-model.ts +++ b/packages/runtime/src/runtime-event-read-model.ts @@ -18,7 +18,12 @@ */ import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation'; -import type { AssistantStepContentKind, StoredMessage, TurnStatus } from '@maka/core/session'; +import type { + AssistantStepContentKind, + StoredMessage, + TurnStatus, + WorkHubCoordinationActionMessage, +} from '@maka/core/session'; import type { RuntimeEvent, RuntimeEventStatus } from '@maka/core/runtime-event'; import type { ToolActivityKind, ToolResultContent } from '@maka/core/events'; import { markPersisted } from '@maka/core/persisted-value'; @@ -99,6 +104,21 @@ export function isContinuationStartRuntimeEvent(event: RuntimeEvent): boolean { ); } +export function projectRuntimeEventCoordinationReceipt( + event: RuntimeEvent, +): WorkHubCoordinationActionMessage | undefined { + if (!event.actions?.coordination) return undefined; + return { + type: 'workhub_coordination', + kind: 'action_receipt', + schemaVersion: 1, + id: event.id, + turnId: event.turnId, + ts: event.ts, + receipt: event.actions.coordination, + }; +} + /** * Whether the event can affect the StoredMessage projection or the state needed * to construct one. Pure control-plane facts are intentionally absent so a @@ -106,6 +126,7 @@ export function isContinuationStartRuntimeEvent(event: RuntimeEvent): boolean { */ export function affectsRuntimeEventStoredMessageProjection(event: RuntimeEvent): boolean { return ( + event.actions?.coordination !== undefined || event.content !== undefined || isTerminalRuntimeEvent(event) || event.actions?.permissionRequest !== undefined || @@ -269,6 +290,12 @@ export function projectRuntimeEventsToStoredMessages( } } + const coordinationReceipt = projectRuntimeEventCoordinationReceipt(event); + if (coordinationReceipt) { + messages.push(coordinationReceipt); + projected = true; + } + if (event.actions?.permissionRequest) { const request = event.actions.permissionRequest; state.permissionRequestById.set(request.requestId, { diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts index 05eddb7c8b..1206c8cb87 100644 --- a/packages/runtime/src/runtime-kernel.ts +++ b/packages/runtime/src/runtime-kernel.ts @@ -17,6 +17,8 @@ * under the License. */ +import { materializeRuntimeEventTranscriptProjection } from './runtime-ledger-repair.js'; +import type { WorkHubActionReceipt } from '@maka/core/workhub-action-result'; import type { AgentRunStore } from '@maka/core/agent-run'; import { agentRunCompositionFromEvents } from '@maka/core/agent-run'; import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation'; @@ -178,6 +180,12 @@ export interface RuntimeKernelLike { continuation: RuntimeContinuation, options?: ResumeContinuationOptions, ): AsyncIterable; + runCoordinationOperation( + sessionId: string, + input: UserMessageInput, + options: TurnStartOptions, + execute: () => Promise, + ): AsyncIterable; compactSession(sessionId: string, input?: CompactSessionInput): AsyncIterable; preflightContextCompaction(sessionId: string): Promise; stopSession(sessionId: string, input?: StopSessionInput): Promise; @@ -979,6 +987,117 @@ export class RuntimeKernel implements RuntimeKernelLike { ); } + /** Host coordination uses the same Run owner and terminal authority without a provider send. */ + async *runCoordinationOperation( + sessionId: string, + input: UserMessageInput, + options: TurnStartOptions, + execute: () => Promise, + ): AsyncIterable { + const execution = this.takeExecutionClaim(sessionId); + try { + await this.enterExecutionClaim(execution); + const header = await this.deps.store.readHeader(sessionId); + const run = new AgentRun({ + sessionId, + header, + userInput: input, + runId: options.runId, + userMessageId: options.userMessageId, + durability: 'required', + store: this.deps.store, + runStore: this.deps.runStore, + runtimeEventStore: this.deps.runtimeEventStore, + newId: this.deps.newId, + now: this.deps.now, + effectiveOrchestration: resolveEffectiveOrchestration('default', undefined), + hooks: { + reserveRun: async (id, nextHeader, activeRun) => { + const active = await this.reserveParentRun(id, nextHeader, activeRun, execution); + this.reserveExecutionClaim(execution, active, activeRun); + return active; + }, + unregisterRun: (active, activeRun) => this.unregisterParentRun(active, activeRun), + updateHeader: (id, patch) => this.updateHeader(id, patch), + updateStatus: (id, status, reason, ts) => this.updateStatus(id, status, reason, ts), + appendTurnState: (id, turnId, status, lineage, stateOptions) => + this.appendTurnState(id, turnId, status, lineage, stateOptions), + }, + }); + this.attachExecutionClaim(execution, run); + const owners = this.createRunOwnerScope(run, execution); + try { + owners.bindMessage(this.deps.messageAuthority, { + sessionId, + turnId: input.turnId, + runId: run.runId, + }); + await this.runBackendActivation(async () => { + run.bindProviderStateIdentity( + await this.prepareBackendForExecution(sessionId, header, execution), + ); + await run.begin(); + }); + await options.onRunStarted?.(run.runId, header); + this.settleReservedExecutionClaim(execution, run, { ok: true }); + } catch (error) { + await this.finalizeFailedRunStart(owners, run, execution, error); + return; + } + try { + if (run.isStopped()) return; + const receipt = await execute(); + const receiptEvent: RuntimeEvent = { + id: this.deps.newId(), + sessionId, + turnId: input.turnId, + runId: run.runId, + invocationId: run.runId, + ts: this.deps.now(), + partial: false, + role: 'system', + author: 'host', + modelVisibility: 'hidden', + actions: { coordination: receipt }, + }; + await run.recordRuntimeEvents([receiptEvent], { requireDurableWrite: true }); + await materializeRuntimeEventTranscriptProjection(this.deps.store, sessionId, receiptEvent); + if (run.isStopped()) return; + const complete: CompleteEvent = { + type: 'complete', + id: this.deps.newId(), + turnId: input.turnId, + ts: this.deps.now(), + stopReason: 'end_turn', + }; + await run.acceptMappedEvent( + complete, + mapSessionEventToRuntimeEvent( + complete, + this.runtimeEventMapContext({ + sessionId, + invocationId: run.runId, + runId: run.runId, + turnId: input.turnId, + }), + ), + { requireTerminalWrite: true }, + ); + yield complete; + } catch (error) { + await run.recordFailure(error); + throw error; + } finally { + const failures = new FailureCollector(); + await failures.capture(() => owners.finalize()); + await failures.capture(() => owners.releaseMessage()); + failures.throwIfAny(`Coordination cleanup failed for ${run.runId}`); + } + } finally { + this.releaseExecutionClaim(execution); + } + } + async *compactSession( sessionId: string, input: CompactSessionInput = {}, diff --git a/packages/runtime/src/runtime-ledger-repair.ts b/packages/runtime/src/runtime-ledger-repair.ts index 8ab8949c99..e9811d9c81 100644 --- a/packages/runtime/src/runtime-ledger-repair.ts +++ b/packages/runtime/src/runtime-ledger-repair.ts @@ -35,7 +35,10 @@ import type { SessionHeader } from '@maka/core/session'; import type { StoredMessage, TurnRecord } from '@maka/core/session'; import { backfillRuntimeEventsFromStoredMessages } from './runtime-event-backfill.js'; import type { RuntimeEventBackfillOutcome } from './runtime-event-backfill.js'; -import { projectRuntimeEventUserMessage } from './runtime-event-read-model.js'; +import { + projectRuntimeEventUserMessage, + projectRuntimeEventCoordinationReceipt, +} from './runtime-event-read-model.js'; export interface RuntimeLedgerRepairDeps { runtimeEventStore: RuntimeEventStore; @@ -56,7 +59,8 @@ export async function materializeRuntimeEventTranscriptProjection( event: RuntimeEvent, knownMessageIds?: Set, ): Promise { - const message = steeringMessageFromRuntimeEvent(event); + const message = + projectRuntimeEventCoordinationReceipt(event) ?? steeringMessageFromRuntimeEvent(event); if (!message) return false; const messageIds = knownMessageIds ?? new Set((await deps.readMessages(sessionId)).map((item) => item.id)); @@ -124,8 +128,8 @@ export class RuntimeLedgerRepair { }); } - async repairSteeringMessagesOnce(sessionId: string): Promise { - return this.withRepairQueue(sessionId, 'steering-transcript', async () => { + async repairRuntimeEventTranscriptOnce(sessionId: string): Promise { + return this.withRepairQueue(sessionId, 'runtime-event-transcript', async () => { const messages = await this.deps.readMessages(sessionId); const messageIds = new Set(messages.map((message) => message.id)); const inlineRunIds = new Set( diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index ddc4d9e3e4..9f4ae1ad07 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -29,6 +29,7 @@ * persistence and same-session serialization semantics. */ +import type { WorkHubActionReceipt } from '@maka/core/workhub-action-result'; import { createHash } from 'node:crypto'; import { isDeepStrictEqual } from 'node:util'; import { setTimeout as delay } from 'node:timers/promises'; @@ -1423,7 +1424,7 @@ export class SessionManager { if (this.runtimeLedgerRepair) { await recoverOr( policy, - () => this.runtimeLedgerRepair!.repairSteeringMessagesOnce(session.id), + () => this.runtimeLedgerRepair!.repairRuntimeEventTranscriptOnce(session.id), 0, ); } @@ -2405,6 +2406,15 @@ export class SessionManager { } } + runCoordinationOperation( + sessionId: string, + input: UserMessageInput, + options: TurnStartOptions, + execute: () => Promise, + ): AsyncIterable { + return this.runtimeKernel.runCoordinationOperation(sessionId, input, options, execute); + } + async *compactSession( sessionId: string, input: CompactSessionInput = {}, @@ -3762,6 +3772,12 @@ export class SessionManager { executionKind: input.execution.kind, goalId: input.execution.goalId, }; + } else if ( + input.execution.kind === 'workhub_coordination' && + input.execution.operation === 'action' + ) { + recoveryReason = 'coordination_action_admission_without_run'; + diagnostic = { executionKind: input.execution.kind, operation: input.execution.operation }; } else if (input.execution.kind === 'legacy_automation') { root = { kind: 'legacy_automation', legacyAutomationId: input.execution.automationId }; recoveryReason = 'legacy_automation_authority_removed'; diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index 58aa12a584..3a657e0a20 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -2029,11 +2029,21 @@ function normalizeRootExecutionDescriptor(value: unknown): RootExecutionDescript }); } if (value.kind === 'workhub_coordination') { - if (!hasExactKeys(value, ['kind', 'inputDigest']) || !isSha256Digest(value.inputDigest)) { + if ( + !hasExactKeys( + value, + value.operation === undefined + ? ['kind', 'inputDigest'] + : ['kind', 'inputDigest', 'operation'], + ) || + (value.operation !== undefined && value.operation !== 'action') || + !isSha256Digest(value.inputDigest) + ) { throw new Error('Invalid root execution descriptor'); } return Object.freeze({ kind: 'workhub_coordination', + ...(value.operation === 'action' ? { operation: 'action' as const } : {}), inputDigest: value.inputDigest, }); }