diff --git a/apps/desktop/e2e/workhub-reconstruction.spec.ts b/apps/desktop/e2e/workhub-reconstruction.spec.ts index e7a1396110..4cf098c93d 100644 --- a/apps/desktop/e2e/workhub-reconstruction.spec.ts +++ b/apps/desktop/e2e/workhub-reconstruction.spec.ts @@ -38,11 +38,11 @@ test('WorkHub rebuilds Session conversation after navigating away and back', asy await window.maka.settings.updateClient({ workHub: { enabled: true } }); }); await expect(page.getByRole('main', { name: 'WorkHub' })).toBeVisible(); - await expect( - page.locator('.workhub-projected-turn .workhub-user-bubble > p', { - hasText: initialPrompt, - }), - ).toBeVisible(); + // The conversation is the Coordination Session transcript. An ordinary + // Session is a routing target and a status row, never a turn in WorkHub. + await expect(page.getByText('1 项工作', { exact: true })).toBeVisible(); + await expect(page.locator('.workhub-turn')).toHaveCount(0); + await expect(page.locator('.workhub-empty h2')).toHaveText('从这里继续所有工作'); const routedPrompt = '继续这个工作,补充重复投递测试点。'; const workHubComposer = page.locator( 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 9eac36a268..40dfc58ecc 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 @@ -98,13 +98,39 @@ test('restarts a paginated catalog read instead of mixing revisions', async () = test('resolves WorkHub coordination through the dedicated Host operation', async () => { const { client, requests } = clientWithResponses([ { sessionId: 'maka_workhub_coordination' }, + { turnId: 'answer-turn' }, + { turnId: 'summary-turn' }, ]); assert.deepEqual(await client.resolveWorkHubCoordinationSession(), { sessionId: 'maka_workhub_coordination', }); + assert.deepEqual( + await client.answerWorkHubCoordination({ turnId: 'answer-turn', text: 'Question' }), + { turnId: 'answer-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.answer', + input: { turnId: 'answer-turn', text: 'Question' }, + }, + { + 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 2fe2eee394..0a47cb7915 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 @@ -24,12 +24,26 @@ 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 answers: unknown[] = []; + const records: unknown[] = []; registerRuntimeHostWorkHubIpc( { resolveWorkHubCoordinationSession: async () => { resolveCalls += 1; return { sessionId: 'maka_workhub_coordination' }; }, + answerWorkHubCoordination: async (input: { turnId: string; text: string }) => { + answers.push(input); + return { turnId: input.turnId }; + }, + recordWorkHubCoordination: async (input: { + turnId: string; + userText: string; + assistantText: string; + }) => { + records.push(input); + return { turnId: input.turnId }; + }, } as never, { handle: (channel: string, handler: (...args: unknown[]) => unknown) => { @@ -42,4 +56,22 @@ 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:answer')?.({}, { turnId: 'answer', text: 'Question' }), + { turnId: 'answer' }, + ); + assert.deepEqual( + await handlers.get('workhub:record')?.({}, { + turnId: 'record', + userText: 'Request', + assistantText: 'Summary', + }), + { turnId: 'record' }, + ); + assert.deepEqual(answers, [{ turnId: 'answer', text: 'Question' }]); + assert.deepEqual(records, [{ + turnId: 'record', + userText: 'Request', + assistantText: 'Summary', + }]); }); diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index 52e46af83d..a2921ccee5 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -135,7 +135,7 @@ test('read exposes existing ordinary Sessions as factual Work summaries', async assert.deepEqual(projection.turns, []); }); -test('read rebuilds a bounded conversation projection from ordinary Session turns', async () => { +test('read does not rebuild WorkHub conversation from ordinary Session turns', async () => { const sessions = port([ session('login', { sessionName: '登录刷新令牌', updatedAt: 30 }), session('internal', { kind: 'internal', updatedAt: 40 }), @@ -156,16 +156,8 @@ test('read rebuilds a bounded conversation projection from ordinary Session turn const projection = await createWorkHubController({ sessions }).read(); - assert.deepEqual(requestedTargets, [['login']]); - assert.deepEqual(projection.turns, [{ - messageId: 'user-1', - target: { sessionId: 'login' }, - turnId: 'turn-login', - text: '检查刷新令牌竞争条件', - state: 'completed', - result: '已定位到并发刷新窗口', - updatedAt: 20, - }]); + assert.deepEqual(requestedTargets, []); + assert.deepEqual(projection.turns, []); }); test('archived Sessions stay inspectable but are excluded from routing targets', async () => { @@ -2370,12 +2362,23 @@ test('submit lets strong foreign core evidence override a vague focus word', asy test('submit keeps unmatched non-executable conversation in WorkHub', async () => { let created = false; + const answered: Array<{ turnId: string; text: string }> = []; const sessions = port([]); sessions.create = async () => { created = true; return session('unexpected'); }; - const controller = createWorkHubController({ sessions }); + const controller = createWorkHubController({ + sessions, + coordination: { + open: async () => ({ close: async () => undefined }), + answer: async (input) => { + answered.push(input); + return { turnId: input.turnId }; + }, + record: async (input) => ({ turnId: input.turnId }), + }, + }); const result = await controller.submit({ requestId: 'request-discussion', @@ -2389,6 +2392,9 @@ test('submit keeps unmatched non-executable conversation in WorkHub', async () = text: '你觉得统一入口最重要的价值是什么?', }); assert.equal(created, false); + assert.deepEqual(answered, [ + { turnId: 'request-discussion', text: '你觉得统一入口最重要的价值是什么?' }, + ]); }); test('submit treats a design question containing an action word as discussion', async () => { 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 719d9c81ca..662bcf4f98 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -27,6 +27,10 @@ import { projectWorkHubSessionTurns, type WorkHubDesktopSession, } from '../../renderer/workhub-session-port.js'; +import { + createDesktopWorkHubCoordinationPort, + projectWorkHubCoordinationTurns, +} from '../../renderer/workhub-coordination-port.js'; import { WorkHubSessionSubmitError } from '../../renderer/workhub-controller.js'; function desktopSession( @@ -94,6 +98,79 @@ function transcriptsWith(messages: readonly StoredMessage[]) { }; } +test('projects the durable Coordination transcript into the WorkHub conversation', () => { + assert.deepEqual(projectWorkHubCoordinationTurns([ + { type: 'user', id: 'user-1', turnId: 'turn-1', ts: 10, text: 'What is next?' }, + { + type: 'assistant', + id: 'assistant-1', + turnId: 'turn-1', + ts: 11, + text: 'Slice 3 is next.', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'state-1', + turnId: 'turn-1', + ts: 12, + status: 'completed', + partialOutputRetained: true, + }, + ]), [{ + messageId: 'user-1', + turnId: 'turn-1', + text: 'What is next?', + result: 'Slice 3 is next.', + state: 'completed', + updatedAt: 11, + }]); +}); + +test('Coordination transcript adapter emits an initial empty ready snapshot and closes cleanly', async () => { + const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'coordination' }); + const snapshots: unknown[] = []; + let closes = 0; + const adapter = createDesktopWorkHubCoordinationPort({ + sessionId, + transcripts: { + open: async (requestedSessionId, handler) => { + assert.equal(requestedSessionId, sessionId); + handler({ + sessionId: 'coordination', + deliverySequence: 1, + generation: 'generation-1', + hostEpoch: 'epoch-1', + durableThrough: null, + fragments: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: false, + hasNewer: false, + reset: true, + ready: true, + }); + return { + sessionId, + generation: 'generation-1', + hostEpoch: 'epoch-1', + readThroughMessageId: null, + loadBefore: async () => {}, + loadAround: async () => {}, + close: async () => { closes += 1; }, + }; + }, + }, + answer: async (input) => ({ turnId: input.turnId }), + record: async (input) => ({ turnId: input.turnId }), + }); + + const handle = await adapter.open((turns) => snapshots.push(turns), () => {}); + assert.deepEqual(snapshots, [[]]); + await handle.close(); + assert.equal(closes, 1); +}); + test('projects durable Session messages into an ordered WorkHub conversation', () => { const turns = projectWorkHubSessionTurns({ target: { sessionId: 'payment' }, 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 cdf6febe41..0b7812c0ad 100644 --- a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts @@ -26,14 +26,12 @@ import { WorkHubCoordinationStatus, WorkHubProjectionRefreshGate, WorkHubSurfaceRouteGate, - projectedWorkHubTurnPresentation, submitWorkHubSurfaceInput, - visibleWorkHubProjectedTurns, + visibleWorkHubConversation, workHubSubmissionCanCorrect, workHubSubmissionClearsDraft, } from '../../renderer/workhub-surface.js'; import { - boundedWorkHubTimelineText, createWorkHubController, WORKHUB_ROUTING_STRATEGY_ID, type WorkHubController, @@ -96,13 +94,6 @@ test('surface projection refresh gate rejects older reads after a newer refresh assert.equal(second(), false); }); -test('projected archived Session keeps the actual turn state visible', () => { - assert.deepEqual(projectedWorkHubTurnPresentation('failed', true, 'zh'), { - heading: '来自已归档 Session:', - state: '失败', - }); -}); - test('surface keeps the Composer draft when routing fails or the target is waiting', () => { assert.equal(workHubSubmissionClearsDraft(undefined), false); assert.equal(workHubSubmissionClearsDraft({ @@ -134,108 +125,42 @@ test('surface disables correction after a request was steered into existing work assert.equal(workHubSubmissionCanCorrect({ ...submission, steered: true }), false); }); -test('surface hides a rebuilt Session turn while the matching local turn is still mounted', () => { - assert.deepEqual( - visibleWorkHubProjectedTurns( - [{ - messageId: 'user-0', - target: { sessionId: 'payment' }, - turnId: 'turn-payment', - text: '检查支付回调风险', - state: 'running', - updatedAt: 9, - }, { - messageId: 'user-1', - target: { sessionId: 'payment' }, - turnId: 'turn-payment', - text: '补充重复投递测试', - state: 'completed', - updatedAt: 10, - }], - [{ - requestId: 'request-payment', - text: '补充重复投递测试', - state: 'settled', - outcome: { - kind: 'submitted', - strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: 'request-payment', - target: { sessionId: 'payment' }, - turnId: 'turn-payment', - evidence: 'explicit_target', - }, - }], - ), - [{ - messageId: 'user-0', - target: { sessionId: 'payment' }, - turnId: 'turn-payment', - text: '检查支付回调风险', - state: 'running', - updatedAt: 9, - }], - ); -}); - -test('surface canonicalizes bounded text before suppressing a local duplicate', () => { - const text = '长'.repeat(700); - assert.deepEqual(visibleWorkHubProjectedTurns([{ - messageId: 'user-long', - target: { sessionId: 'payment' }, - turnId: 'turn-long', - text: boundedWorkHubTimelineText(text), - state: 'running', - updatedAt: 10, - }], [{ - requestId: 'request-long', - text, - state: 'settled', +test('surface replaces a local discussion placeholder with its durable model answer', () => { + const local = [{ + requestId: 'discussion-turn', + text: 'What is next?', + state: 'settled' as const, outcome: { - kind: 'submitted', + kind: 'discussion' as const, strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: 'request-long', - target: { sessionId: 'payment' }, - turnId: 'turn-long', - evidence: 'explicit_target', + requestId: 'discussion-turn', + text: 'What is next?', }, - }]), []); -}); - -test('surface suppresses the newest matching projected steering turn', () => { - const projected = [{ - messageId: 'user-old', - target: { sessionId: 'payment' }, - turnId: 'turn-payment', - text: '继续检查', - state: 'running' as const, - updatedAt: 9, - }, { - messageId: 'user-new', - target: { sessionId: 'payment' }, - turnId: 'turn-payment', - text: '继续检查', - state: 'running' as const, + }]; + const durable = [{ + messageId: 'user-message', + turnId: 'discussion-turn', + text: 'What is next?', + result: 'Slice 3 is next.', + state: 'completed' as const, updatedAt: 10, }]; - assert.deepEqual(visibleWorkHubProjectedTurns(projected, [{ - requestId: 'request-new', - text: '继续检查', - state: 'settled', - outcome: { - kind: 'submitted', - strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: 'request-new', - target: { sessionId: 'payment' }, - turnId: 'turn-payment', - evidence: 'explicit_target', - }, - }]), [projected[0]]); + + assert.deepEqual(visibleWorkHubConversation(durable, local), { + coordination: durable, + local: [], + }); }); test('surface keeps clarification and successful routing in WorkHub', async () => { const submissions: WorkHubSubmitInput[] = []; const controller: WorkHubController = { read: async () => ({ sessions: [], turns: [] }), + openConversation: async (handler) => { + handler([]); + return { close: async () => undefined }; + }, + recordConversationTurn: async ({ turnId }) => ({ turnId }), resetVisitContext: () => {}, subscribe: () => () => {}, submit: async (input) => { @@ -285,6 +210,11 @@ test('surface keeps clarification and successful routing in WorkHub', async () = test('surface leaves discussion in WorkHub instead of creating a task view', async () => { const controller: WorkHubController = { read: async () => ({ sessions: [], turns: [] }), + openConversation: async (handler) => { + handler([]); + return { close: async () => undefined }; + }, + recordConversationTurn: async ({ turnId }) => ({ turnId }), resetVisitContext: () => {}, subscribe: () => () => {}, submit: async (input) => ({ diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index af0a90bdad..ff9e34e11b 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -834,6 +834,18 @@ export class DesktopRuntimeHostClient { return this.request("workhub.coordination.resolve", {}); } + answerWorkHubCoordination( + input: OperationInput<"workhub.coordination.answer">, + ): Promise> { + return this.request("workhub.coordination.answer", input); + } + + 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 3afe04d7e4..d0ad7e94e1 100644 --- a/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts @@ -22,7 +22,9 @@ import type { ReconnectableReadIpcMain } from './ipc-reconnect-policy.js'; type RuntimeHostWorkHubClient = Pick< DesktopRuntimeHostClient, - 'resolveWorkHubCoordinationSession' + | 'answerWorkHubCoordination' + | 'recordWorkHubCoordination' + | 'resolveWorkHubCoordinationSession' >; /** Projects the Runtime Host WorkHub domain onto renderer IPC. */ @@ -33,4 +35,10 @@ export function registerRuntimeHostWorkHubIpc( ipcMain.handle('workhub:resolveCoordinationSession', () => client.resolveWorkHubCoordinationSession(), ); + ipcMain.handle('workhub:answer', (_event, input) => + client.answerWorkHubCoordination(input), + ); + ipcMain.handle('workhub:record', (_event, input) => + client.recordWorkHubCoordination(input), + ); } diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 877736d21f..f02ba643e9 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -759,6 +759,16 @@ export interface MakaBridge { workHub: { /** Resolve the active Runtime Host's stable coordination conversation. */ resolveCoordinationSession(): Promise; + /** Answer an ordinary question inside the persistent Coordination Session. */ + answer( + coordinationSessionId: string, + input: { turnId: string; text: string }, + ): Promise<{ turnId: string }>; + /** Persist one deterministic clarification or routing summary. */ + record( + coordinationSessionId: string, + input: { turnId: string; userText: string; assistantText: string }, + ): Promise<{ turnId: string }>; /** Create an ordinary Session on the exact Host owning the resolved conversation. */ createSession( coordinationSessionId: string, diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 8d39ee7f2f..f63a6ec63b 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -1535,6 +1535,26 @@ const makaBridge = { (scope) => ipcRenderer.invoke('workhub:resolveCoordinationSession', scope), ); }, + async answer( + coordinationSessionId: string, + input: { turnId: string; text: string }, + ): Promise<{ turnId: string }> { + const scope = await resolveDesktopWorkHubCoordinationCreateScope( + coordinationSessionId, + runtimeHostSessionRef, + ); + return ipcRenderer.invoke('workhub:answer', scope, input) as Promise<{ turnId: string }>; + }, + 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 createSession( coordinationSessionId: string, input: { name: string }, diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index e98b20e939..013ed52498 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -126,6 +126,7 @@ import { createWorkHubController } from './workhub-controller.js'; import { startWorkHubCoordinationLifecycle } from './workhub-coordination-lifecycle.js'; import { scopeWorkHubSessionsToCoordinationHost } from './workhub-coordination-host-scope.js'; import { createDesktopWorkHubSessionPort } from './workhub-session-port.js'; +import { createDesktopWorkHubCoordinationPort } from './workhub-coordination-port.js'; import { WorkHubCoordinationStatus, WorkHubSurface } from './workhub-surface.js'; import { getShellCopy, localizedShellErrorMessage } from './locales/shell-copy'; import { getDesktopConversationCopy } from './locales/conversation-copy'; @@ -1524,6 +1525,14 @@ function AppShellContent({ const workHubCoordinationGeneration = workHubCoordinationGenerationRef.current; const workHubController = useMemo( () => createWorkHubController({ + coordination: createDesktopWorkHubCoordinationPort({ + sessionId: workHubCoordinationSessionId ?? 'workhub-coordination-unresolved', + transcripts: window.maka.transcripts, + answer: (input) => + window.maka.workHub.answer(workHubCoordinationSessionId!, input), + record: (input) => + window.maka.workHub.record(workHubCoordinationSessionId!, input), + }), sessions: createDesktopWorkHubSessionPort({ sessions: scopeWorkHubSessionsToCoordinationHost( window.maka.sessions, diff --git a/apps/desktop/src/renderer/workhub-controller.ts b/apps/desktop/src/renderer/workhub-controller.ts index e382ee7fd5..cf2b288a3c 100644 --- a/apps/desktop/src/renderer/workhub-controller.ts +++ b/apps/desktop/src/renderer/workhub-controller.ts @@ -67,6 +67,15 @@ export interface WorkHubProjectedTurn { updatedAt: number; } +export interface WorkHubCoordinationTurn { + messageId: string; + turnId: string; + text: string; + state: WorkHubProjectedTurnState; + result?: string; + updatedAt: number; +} + const WORKHUB_TIMELINE_TEXT_LIMIT = 600; export function boundedWorkHubTimelineText(value: string): string { @@ -177,6 +186,19 @@ export interface WorkHubSessionPort { subscribe(handler: () => void): () => void; } +export interface WorkHubCoordinationPort { + open( + handler: (turns: readonly WorkHubCoordinationTurn[]) => void, + onError: (error: unknown) => void, + ): Promise<{ close(): Promise }>; + answer(input: { turnId: string; text: string }): Promise<{ turnId: string }>; + record(input: { + turnId: string; + userText: string; + assistantText: string; + }): Promise<{ turnId: string }>; +} + export class WorkHubSessionSubmitError extends Error { constructor( message: string, @@ -191,6 +213,15 @@ export class WorkHubSessionSubmitError extends Error { export interface WorkHubController { read(input?: WorkHubReadInput): Promise; submit(input: WorkHubSubmitInput): Promise; + openConversation( + handler: (turns: readonly WorkHubCoordinationTurn[]) => void, + onError: (error: unknown) => void, + ): Promise<{ close(): Promise }>; + recordConversationTurn(input: { + turnId: string; + userText: string; + assistantText: string; + }): Promise<{ turnId: string }>; subscribe(handler: () => void): () => void; resetVisitContext(): void; } @@ -213,7 +244,9 @@ interface WorkHubOwnershipTombstone { export function createWorkHubController(deps: { sessions: WorkHubSessionPort; + coordination?: WorkHubCoordinationPort; }): WorkHubController { + const coordination = deps.coordination ?? legacyTestCoordinationPort(); let routePolicy = createWorkHubRoutePolicy(); let focusReadVersion = 0; let pendingFocusReadVersion: number | undefined; @@ -544,6 +577,12 @@ export function createWorkHubController(deps: { if (failures.length > 0) throw failures[0]; }; return { + openConversation(handler, onError) { + return coordination.open(handler, onError); + }, + recordConversationTurn(input) { + return coordination.record(input); + }, subscribe(handler) { return deps.sessions.subscribe(handler); }, @@ -577,7 +616,10 @@ export function createWorkHubController(deps: { return { sessions: ordinary .map(({ kind: _kind, runningTurnIds: _runningTurnIds, ...session }) => session), - turns: await deps.sessions.recentTurns(ordinary.map((session) => session.target)), + // Slice 3 renders conversation only from the Coordination Session. + // Ordinary Session transcripts remain routing evidence, never a + // second WorkHub conversation source. + turns: [], }; } finally { if (input?.focus && pendingFocusReadVersion === readFocusVersion) { @@ -634,6 +676,10 @@ export function createWorkHubController(deps: { }; } if (decision.kind === 'discussion') { + await coordination.answer({ + turnId: input.requestId, + text: input.text, + }); return { kind: 'discussion', strategyId: WORKHUB_ROUTING_STRATEGY_ID, @@ -721,3 +767,18 @@ export function createWorkHubController(deps: { }, }; } + +function legacyTestCoordinationPort(): WorkHubCoordinationPort { + return { + async open(handler) { + handler([]); + return { close: async () => undefined }; + }, + async answer(input) { + return { turnId: input.turnId }; + }, + async record(input) { + return { turnId: input.turnId }; + }, + }; +} diff --git a/apps/desktop/src/renderer/workhub-coordination-port.ts b/apps/desktop/src/renderer/workhub-coordination-port.ts new file mode 100644 index 0000000000..38bcbf1bbc --- /dev/null +++ b/apps/desktop/src/renderer/workhub-coordination-port.ts @@ -0,0 +1,137 @@ +/* + * 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 { + deriveTurnRecords, + userFacingText, + type StoredMessage, + type TurnStatus, +} from '@maka/core/session'; +import { DesktopTranscriptRangeStore } from './desktop-transcript-range-store.js'; +import type { + WorkHubCoordinationPort, + WorkHubCoordinationTurn, + WorkHubProjectedTurnState, +} from './workhub-controller.js'; +import { boundedWorkHubTimelineText } from './workhub-controller.js'; +import type { WorkHubDesktopTranscriptBridge } from './workhub-session-port.js'; + +const WORKHUB_COORDINATION_TURN_LIMIT = 40; + +export function createDesktopWorkHubCoordinationPort(deps: { + sessionId: string; + transcripts: WorkHubDesktopTranscriptBridge; + answer(input: { turnId: string; text: string }): Promise<{ turnId: string }>; + record(input: { + turnId: string; + userText: string; + assistantText: string; + }): Promise<{ turnId: string }>; +}): WorkHubCoordinationPort { + return { + answer: deps.answer, + record: deps.record, + async open(handler, onError) { + const store = new DesktopTranscriptRangeStore(deps.sessionId); + let disposed = false; + let ready = false; + const handle = await deps.transcripts.open( + deps.sessionId, + (batch) => { + if (disposed) return; + try { + const changed = store.accept(batch); + ready ||= batch.ready; + if (ready && (changed || batch.ready)) { + handler(projectWorkHubCoordinationTurns(store.snapshot().messages)); + } + } catch (error) { + onError(error); + } + }, + (cancel) => { + if (disposed) cancel(); + }, + ).catch((error) => { + onError(error); + throw error; + }); + return { + async close() { + disposed = true; + await handle.close(); + }, + }; + }, + }; +} + +export function projectWorkHubCoordinationTurns( + messages: readonly StoredMessage[], +): WorkHubCoordinationTurn[] { + const stateByTurnId = new Map( + deriveTurnRecords(messages).map((turn) => [turn.turnId, projectState(turn.status)]), + ); + const turns: WorkHubCoordinationTurn[] = []; + const latestUserIndexByTurnId = new Map(); + + for (const message of messages) { + if (message.type === 'user') { + const text = boundedWorkHubTimelineText(userFacingText(message)); + if (!text) continue; + turns.push({ + messageId: message.id, + turnId: message.turnId, + text, + state: stateByTurnId.get(message.turnId) ?? 'running', + updatedAt: message.ts, + }); + latestUserIndexByTurnId.set(message.turnId, turns.length - 1); + continue; + } + if (message.type !== 'assistant') continue; + const userIndex = latestUserIndexByTurnId.get(message.turnId); + if (userIndex === undefined) continue; + const result = boundedWorkHubTimelineText(message.text); + turns[userIndex] = { + ...turns[userIndex]!, + ...(result ? { result } : {}), + updatedAt: Math.max(turns[userIndex]!.updatedAt, message.ts), + }; + } + + return turns + .sort((left, right) => + left.updatedAt - right.updatedAt || left.messageId.localeCompare(right.messageId), + ) + .slice(-WORKHUB_COORDINATION_TURN_LIMIT); +} + +function projectState(status: TurnStatus): WorkHubProjectedTurnState { + switch (status) { + case 'running': + return 'running'; + case 'aborted': + return 'aborted'; + case 'failed': + return 'failed'; + case 'completed': + return 'completed'; + } +} diff --git a/apps/desktop/src/renderer/workhub-surface.tsx b/apps/desktop/src/renderer/workhub-surface.tsx index c06228b2ce..c03b15ba53 100644 --- a/apps/desktop/src/renderer/workhub-surface.tsx +++ b/apps/desktop/src/renderer/workhub-surface.tsx @@ -29,13 +29,12 @@ import type { UiLocale } from '@maka/core/ui-locale'; import { ChatSurfaceLayout, Composer } from '@maka/ui'; import type { WorkHubController, + WorkHubCoordinationTurn, WorkHubProjection, - WorkHubProjectedTurn, WorkHubSessionSummary, WorkHubSubmission, WorkHubSubmitInput, } from './workhub-controller.js'; -import { boundedWorkHubTimelineText } from './workhub-controller.js'; export interface WorkHubConversationTurn { requestId: string; @@ -87,6 +86,26 @@ export function workHubSubmissionCanCorrect( return result.kind === 'submitted' && !result.steered; } +export function visibleWorkHubConversation( + coordination: readonly WorkHubCoordinationTurn[], + local: readonly WorkHubConversationTurn[], +): { + coordination: readonly WorkHubCoordinationTurn[]; + local: readonly WorkHubConversationTurn[]; +} { + const localByRequestId = new Map(local.map((turn) => [turn.requestId, turn])); + const visibleCoordination = coordination.filter( + (turn) => + localByRequestId.get(turn.turnId)?.outcome?.kind === 'discussion' || + !localByRequestId.has(turn.turnId), + ); + const coordinationTurnIds = new Set(coordination.map(({ turnId }) => turnId)); + const visibleLocal = local.filter( + (turn) => turn.outcome?.kind !== 'discussion' || !coordinationTurnIds.has(turn.requestId), + ); + return { coordination: visibleCoordination, local: visibleLocal }; +} + export async function submitWorkHubSurfaceInput(input: { controller: WorkHubController; input: WorkHubSubmitInput; @@ -94,60 +113,9 @@ export async function submitWorkHubSurfaceInput(input: { return input.controller.submit(input.input); } -export function visibleWorkHubProjectedTurns( - projected: readonly WorkHubProjectedTurn[], - local: readonly WorkHubConversationTurn[], -): WorkHubProjectedTurn[] { - const localTurnCounts = new Map(); - for (const turn of local) { - if (turn.outcome?.kind !== 'submitted') continue; - const key = projectedTurnKey(turn.outcome.target, turn.outcome.turnId, turn.text); - localTurnCounts.set(key, (localTurnCounts.get(key) ?? 0) + 1); - } - const visible: WorkHubProjectedTurn[] = []; - for (let index = projected.length - 1; index >= 0; index -= 1) { - const turn = projected[index]!; - const key = projectedTurnKey(turn.target, turn.turnId, turn.text); - const remaining = localTurnCounts.get(key) ?? 0; - if (remaining > 0) { - localTurnCounts.set(key, remaining - 1); - } else { - visible.push(turn); - } - } - return visible.reverse(); -} - -function projectedTurnKey( - target: { sessionId: string }, - turnId: string, - text: string, -): string { - return JSON.stringify([ - target.sessionId, - turnId, - boundedWorkHubTimelineText(text), - ]); -} - -export function projectedWorkHubTurnPresentation( - state: WorkHubProjectedTurn['state'], - archived: boolean, - locale: UiLocale, -): { heading: string; state: string } { - const copy = workHubCopy(locale); - return { - heading: archived ? copy.archivedSessionRecord : copy.sessionRecord, - state: copy.turnStates[state], - }; -} - /** - * A transient conversation projection over ordinary Sessions. - * - * This surface deliberately does not persist its own transcript. Session and - * Runtime remain authoritative; local turns only keep WorkHub conversational - * while the surface is mounted. + * The persistent Coordination Session transcript is the primary conversation. + * Ordinary Sessions remain a read-only status/routing projection. */ export function WorkHubSurface(props: { controller: WorkHubController; @@ -157,14 +125,17 @@ export function WorkHubSurface(props: { }) { const copy = workHubCopy(props.locale); const [projection, setProjection] = useState({ sessions: [], turns: [] }); + const [coordinationTurns, setCoordinationTurns] = useState([]); const [turns, setTurns] = useState([]); const [pending, setPending] = useState(false); const [initialLoadSettled, setInitialLoadSettled] = useState(false); + const [conversationReady, setConversationReady] = useState(false); // React state paints the lock; the gate closes the same-frame window before // a rerender can disable Composer and clarification controls. const routeGate = useRef(new WorkHubSurfaceRouteGate()).current; const refreshGate = useRef(new WorkHubProjectionRefreshGate()).current; const [loadError, setLoadError] = useState(false); + const [conversationError, setConversationError] = useState(false); const refresh = useCallback(async (focusSessionId?: string) => { const isLatest = refreshGate.begin(); try { @@ -192,13 +163,42 @@ export function WorkHubSurface(props: { }; }, [props.controller, props.initialFocusSessionId, refresh, refreshGate]); + useEffect(() => { + let disposed = false; + let handle: { close(): Promise } | undefined; + setConversationReady(false); + setConversationError(false); + void props.controller.openConversation( + (next) => { + if (disposed) return; + setCoordinationTurns(next); + setConversationReady(true); + setConversationError(false); + }, + () => { + if (disposed) return; + setConversationReady(true); + setConversationError(true); + }, + ).then((opened) => { + if (disposed) void opened.close().catch(() => undefined); + else handle = opened; + }).catch(() => undefined); + return () => { + disposed = true; + void handle?.close().catch(() => undefined); + }; + }, [props.controller]); + const route = useCallback(async ( input: WorkHubSubmitInput, + localRequestId: string = input.requestId, + recordedUserText: string = input.text, ): Promise => { return routeGate.run(async () => { setPending(true); setTurns((current) => current.map((turn) => - turn.requestId === input.requestId + turn.requestId === localRequestId ? { ...turn, state: 'routing', outcome: undefined } : turn, )); @@ -207,8 +207,21 @@ export function WorkHubSurface(props: { controller: props.controller, input, }); + if (result.kind !== 'discussion') { + try { + await props.controller.recordConversationTurn({ + turnId: input.requestId, + userText: recordedUserText, + assistantText: workHubCoordinationSummary(result, projection, copy), + }); + } catch { + // The ordinary Session admission has already settled. A failed + // Coordination summary must not make retry duplicate that work. + setConversationError(true); + } + } setTurns((current) => current.map((turn) => - turn.requestId === input.requestId + turn.requestId === localRequestId ? { ...turn, state: 'settled', outcome: result } : turn, )); @@ -216,7 +229,7 @@ export function WorkHubSurface(props: { return result; } catch { setTurns((current) => current.map((turn) => - turn.requestId === input.requestId + turn.requestId === localRequestId ? { ...turn, state: 'failed', outcome: undefined } : turn, )); @@ -225,20 +238,23 @@ export function WorkHubSurface(props: { setPending(false); } }); - }, [props.controller, refresh, routeGate]); + }, [copy, projection, props.controller, refresh, routeGate]); const send = useCallback(async (value: string) => { const text = value.trim(); - if (!text || !initialLoadSettled || routeGate.pending) return false; + if (!text || !initialLoadSettled || !conversationReady || routeGate.pending) return false; const requestId = crypto.randomUUID(); setTurns((current) => [...current, { requestId, text, state: 'routing' }]); const result = await route({ requestId, text }); // Composer clears only accepted drafts. Waiting, delivery failures, and a // ref-blocked duplicate keep the exact text available for retry. return workHubSubmissionClearsDraft(result); - }, [initialLoadSettled, route, routeGate]); - const projectedTurns = visibleWorkHubProjectedTurns(projection.turns, turns); - const conversationEmpty = projectedTurns.length === 0 && turns.length === 0; + }, [conversationReady, initialLoadSettled, route, routeGate]); + const visible = visibleWorkHubConversation(coordinationTurns, turns); + const visibleCoordinationTurns = visible.coordination; + const visibleLocalTurns = visible.local; + const conversationEmpty = visibleCoordinationTurns.length === 0 && visibleLocalTurns.length === 0; + const surfaceReady = initialLoadSettled && conversationReady; return ( {}} - sendBlocked={pending || !initialLoadSettled} + sendBlocked={pending || !surfaceReady} modelLabel="WorkHub" /> )} @@ -260,7 +276,7 @@ export function WorkHubSurface(props: {

WorkHub

{copy.subtitle}

- {initialLoadSettled + {surfaceReady ? copy.workCount(projection.sessions.length) : copy.loading} @@ -272,51 +288,60 @@ export function WorkHubSurface(props: { gap={4} isStreaming={pending} > - {!initialLoadSettled ? ( + {!surfaceReady ? ( - ) : loadError ? ( -
{copy.loadFailed}
- ) : conversationEmpty ? ( + ) : conversationEmpty && !loadError && !conversationError ? (

{copy.emptyTitle}

{copy.emptyBody(projection.sessions.length)}

) : (
- {projectedTurns.map((turn) => ( - {copy.loadFailed}
+ ) : null} + {visibleCoordinationTurns.map((turn) => ( + ))} - {turns.map((turn) => ( + {visibleLocalTurns.map((turn) => ( void route({ - requestId: turn.requestId, - text: turn.text, - explicitTarget: target, - ...(turn.outcome?.kind === 'clarification' && turn.outcome.correction - ? { correction: turn.outcome.correction } - : {}), - })} - onCorrect={(from, target) => void route({ - requestId: turn.requestId, - text: turn.text, - explicitTarget: target, - correction: { - from: from.target, - turnId: from.turnId, - ...(from.steered ? { steered: true } : {}), - }, - })} + onChoose={(target) => { + const selected = projection.sessions.find( + (session) => session.target.sessionId === target.sessionId, + ); + void route({ + requestId: crypto.randomUUID(), + text: turn.text, + explicitTarget: target, + ...(turn.outcome?.kind === 'clarification' && turn.outcome.correction + ? { correction: turn.outcome.correction } + : {}), + }, turn.requestId, copy.choseWork(selected?.sessionName ?? copy.sessionFallback)); + }} + onCorrect={(from, target) => { + const selected = projection.sessions.find( + (session) => session.target.sessionId === target.sessionId, + ); + void route({ + requestId: crypto.randomUUID(), + text: turn.text, + explicitTarget: target, + correction: { + from: from.target, + turnId: from.turnId, + ...(from.steered ? { steered: true } : {}), + }, + }, turn.requestId, copy.correctedWork(selected?.sessionName ?? copy.sessionFallback)); + }} onOpenSession={props.onOpenSession} /> ))} @@ -411,38 +436,48 @@ function WorkHubLoadingState(props: { label: string }) { ); } -function ProjectedWorkHubTurnView(props: { - turn: WorkHubProjectedTurn; - projection: WorkHubProjection; +function CoordinationTurnView(props: { + turn: WorkHubCoordinationTurn; copy: ReturnType; - onOpenSession(sessionId: string): void; }) { - const session = props.projection.sessions.find( - (candidate) => candidate.target.sessionId === props.turn.target.sessionId, - ); - const presentation = projectedWorkHubTurnPresentation( - props.turn.state, - session?.archived ?? false, - props.copy.locale, - ); return ( - - + + {props.turn.result ? ( +

{props.turn.result}

+ ) : props.turn.state === 'running' ? ( +

{props.copy.answering}

+ ) : ( +

+ {props.copy.turnStates[props.turn.state]} +

+ )}
); } +export function workHubCoordinationSummary( + result: Exclude, + projection: WorkHubProjection, + copy: ReturnType, +): string { + if (result.kind === 'clarification') { + return `${copy.chooseWork} ${result.options.map(({ sessionName }) => sessionName).join('、')}`; + } + if (result.kind === 'waiting') { + return `${copy.waitingForDecision} ${copy.requestNotSent}`; + } + const target = projection.sessions.find( + (session) => session.target.sessionId === result.target.sessionId, + ); + const name = target?.sessionName ?? copy.sessionFallback; + const state = target + ? target.archived + ? copy.archived + : copy.states[target.state] + : copy.accepted; + return `${copy.sentTo} ${name} · ${state}`; +} + function WorkHubTurnView(props: { turn: WorkHubConversationTurn; projection: WorkHubProjection; @@ -626,9 +661,10 @@ function workHubCopy(locale: UiLocale) { chooseWork: '这条输入可能与多项工作有关,请选择目标:', discussionStayed: '这条内容暂时保留在 WorkHub,没有创建或改动 Session。', discussionHint: '提出明确的执行目标后,我会把它交给对应的 Session。', + answering: '正在回答…', + choseWork: (name: string) => `选择“${name}”`, + correctedWork: (name: string) => `更正目标为“${name}”`, sentTo: '已交给:', accepted: '已接收', sessionFallback: '普通 Session', - sessionRecord: '来自 Session:', - archivedSessionRecord: '来自已归档 Session:', correctTarget: '更正目标', correctedFrom: (name: string) => `已从“${name}”更正`, waitingForDecision: '这项工作正在等待你的决定。', @@ -655,9 +691,10 @@ function workHubCopy(locale: UiLocale) { chooseWork: 'This input may relate to more than one task. Choose a target:', discussionStayed: 'This stayed in WorkHub without creating or changing a Session.', discussionHint: 'State an executable goal and I will hand it to the owning Session.', + answering: 'Answering…', + choseWork: (name: string) => `Choose “${name}”`, + correctedWork: (name: string) => `Correct the target to “${name}”`, sentTo: 'Sent to:', accepted: 'Accepted', sessionFallback: 'Ordinary Session', - sessionRecord: 'From Session:', - archivedSessionRecord: 'From archived Session:', correctTarget: 'Correct target', correctedFrom: (name: string) => `Corrected from “${name}”`, waitingForDecision: 'This work is waiting for your decision.', diff --git a/packages/core/src/agent-run.ts b/packages/core/src/agent-run.ts index 574a353962..8c3dbd0275 100644 --- a/packages/core/src/agent-run.ts +++ b/packages/core/src/agent-run.ts @@ -80,6 +80,11 @@ export type RootExecutionDescriptor = inputDigest?: `sha256:${string}`; maxSteps?: number; } + | { + /** Tool-free conversational execution admitted only by WorkHub authority. */ + kind: 'workhub_coordination'; + inputDigest: `sha256:${string}`; + } | { kind: 'regenerate'; sourceTurnId: string } | { kind: 'context_compact' } | { kind: 'scheduled_task'; scheduledTaskId: string } diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index aeee5b0ae1..e9896dc2e2 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -205,7 +205,7 @@ export function isTurnStatus(value: unknown): value is TurnStatus { // Header (JSONL line 1) // ============================================================================ -export const SESSION_TOOL_PROFILES = ['headless-coding-v1'] as const; +export const SESSION_TOOL_PROFILES = ['headless-coding-v1', 'workhub-coordination-v1'] as const; export type SessionToolProfile = (typeof SESSION_TOOL_PROFILES)[number]; export function isSessionToolProfile(value: unknown): value is SessionToolProfile { diff --git a/packages/runtime-host/src/__tests__/host-session-availability.test.ts b/packages/runtime-host/src/__tests__/host-session-availability.test.ts new file mode 100644 index 0000000000..82d601cb35 --- /dev/null +++ b/packages/runtime-host/src/__tests__/host-session-availability.test.ts @@ -0,0 +1,103 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { + WORKHUB_COORDINATION_SESSION_ID, + WORKHUB_COORDINATION_SESSION_ROLE, +} from '@maka/core/session'; +import { + runtimeHostExecutionUnavailableReason, + WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON, + WORKHUB_COORDINATION_TARGET_UNAVAILABLE_REASON, +} from '../server/host-session-availability.js'; + +const execution = { + kind: 'workhub_coordination', + inputDigest: `sha256:${'a'.repeat(64)}`, +} as const; +const base = { + collaborationMode: 'agent' as const, + orchestrationMode: 'default' as const, + permissionMode: 'explore' as const, + subagentWorkspace: undefined, + transcriptLedgerVersion: 1 as const, + toolProfile: 'workhub-coordination-v1' as const, +}; + +test('WorkHub execution requires the exact reserved id, role, and zero-tool profile', () => { + assert.equal( + runtimeHostExecutionUnavailableReason( + { + ...base, + id: WORKHUB_COORDINATION_SESSION_ID, + role: WORKHUB_COORDINATION_SESSION_ROLE, + }, + execution, + ), + undefined, + ); + assert.equal( + runtimeHostExecutionUnavailableReason( + { + ...base, + id: WORKHUB_COORDINATION_SESSION_ID, + role: undefined, + }, + execution, + ), + WORKHUB_COORDINATION_TARGET_UNAVAILABLE_REASON, + ); + assert.equal( + runtimeHostExecutionUnavailableReason( + { + ...base, + id: 'ordinary-session', + role: WORKHUB_COORDINATION_SESSION_ROLE, + }, + execution, + ), + WORKHUB_COORDINATION_TARGET_UNAVAILABLE_REASON, + ); + assert.equal( + runtimeHostExecutionUnavailableReason( + { + ...base, + id: WORKHUB_COORDINATION_SESSION_ID, + role: WORKHUB_COORDINATION_SESSION_ROLE, + toolProfile: undefined, + }, + execution, + ), + WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON, + ); + assert.equal( + runtimeHostExecutionUnavailableReason( + { + ...base, + id: WORKHUB_COORDINATION_SESSION_ID, + role: WORKHUB_COORDINATION_SESSION_ROLE, + orchestrationMode: 'graph', + }, + execution, + ), + WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON, + ); +}); diff --git a/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts b/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts index 3e3bf291bc..00100de786 100644 --- a/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts +++ b/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts @@ -117,3 +117,20 @@ test('the headless coding profile freezes prompt, tools, memory, and foreground ); assert.equal((await schema.safeParseAsync({ command: 'true', pty: true })).success, false); }); + +test('the WorkHub coordination profile has conversational authority but zero tools', () => { + const profile = hostedExecutionRunProfile('workhub-coordination-v1'); + assert.ok(profile); + assert.deepEqual(profile.toolNames, []); + assert.equal(profile.memoryExtraction, false); + assert.match(profile.systemPrompt, /conversational coordinator for WorkHub/u); + assert.match(profile.systemPrompt, /no tools, filesystem authority/u); + + const productTool: MakaTool = { + name: 'Read', + description: 'Read files', + parameters: z.object({}), + impl: async () => 'not reachable', + }; + assert.deepEqual(projectHostedExecutionTools([productTool], 'workhub-coordination-v1'), []); +}); 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 7d6e7b4d8c..18bdd5a793 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts @@ -23,6 +23,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { DatabaseSync } from 'node:sqlite'; import { describe, test } from 'node:test'; +import type { MessageContent } from '@maka/core/events'; import { WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_SESSION_ROLE, @@ -30,6 +31,7 @@ import { import { createSessionStore, type SessionAuthorityStore } from '@maka/storage/session-store'; import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store'; 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'; import { SessionOperationFailure } from '../server/session-catalog-coordinator.js'; import { @@ -65,6 +67,7 @@ describe('Host WorkHub Coordination coordinator', () => { ); const header = await store.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID); assert.equal(header.role, WORKHUB_COORDINATION_SESSION_ROLE); + assert.equal(header.toolProfile, 'workhub-coordination-v1'); assert.equal(header.projectId, null); assert.equal(header.cwd, join(root, 'workhub-coordination')); assert.equal((await store.listHeaders()).length, 1); @@ -72,6 +75,24 @@ describe('Host WorkHub Coordination coordinator', () => { await store.close?.(); } + const database = new DatabaseSync(join(root, OPERATIONAL_STATE_DATABASE_NAME)); + try { + database + .prepare( + `UPDATE session_metadata + SET payload_json = json_set( + json_remove(payload_json, '$.toolProfile'), + '$.permissionMode', 'ask', + '$.collaborationMode', 'plan', + '$.orchestrationMode', 'graph' + ) + WHERE session_id = ?`, + ) + .run(WORKHUB_COORDINATION_SESSION_ID); + } finally { + database.close(); + } + store = createSessionStore(root); try { const restarted = await coordinator(root, store).handlers['workhub.coordination.resolve']( @@ -82,6 +103,14 @@ describe('Host WorkHub Coordination coordinator', () => { ok: true, result: { sessionId: WORKHUB_COORDINATION_SESSION_ID }, }); + assert.equal( + (await store.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID)).toolProfile, + 'workhub-coordination-v1', + ); + const migrated = await store.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID); + assert.equal(migrated.permissionMode, 'explore'); + assert.equal(migrated.collaborationMode, 'agent'); + assert.equal(migrated.orchestrationMode, 'default'); assert.equal((await store.listHeaders()).length, 1); } finally { await store.close?.(); @@ -353,26 +382,234 @@ describe('Host WorkHub Coordination coordinator', () => { await rm(root, { recursive: true, force: true }); } }); + + test('answers through the dedicated Coordination root without creating an ordinary Session', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-answer-')); + const store = createSessionStore(root); + const admission = new SessionAdmissionGate(); + const { executions, starts, prepared } = coordinationExecutions(admission); + try { + const workhub = coordinator(root, store, () => undefined, undefined, executions, admission); + assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); + assert.deepEqual( + await workhub.handlers['workhub.coordination.answer']( + { turnId: 'answer-turn', text: 'What should we do next?' }, + CONTEXT, + ), + { ok: true, result: { turnId: 'answer-turn' } }, + ); + assert.equal(starts.length, 1); + assert.equal(starts[0]?.sessionId, WORKHUB_COORDINATION_SESSION_ID); + assert.equal(starts[0]?.execution.kind, 'workhub_coordination'); + assert.deepEqual(prepared, [{ text: 'What should we do next?' }]); + assert.deepEqual( + (await store.listHeaders()).map(({ id, role }) => ({ id, role })), + [{ id: WORKHUB_COORDINATION_SESSION_ID, role: WORKHUB_COORDINATION_SESSION_ROLE }], + ); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + 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 messages = await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID); + assert.equal(messages.length, 3); + assert.deepEqual( + messages.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, 3); + 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('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' +>; + +/** + * Stands in for the root admission ledger: answers claim their Turn identity + * under the same Session admission the coordinator uses, so the fake can + * 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 = { + startWorkHubCoordinationMessage: async (request) => { + starts.push(request); + return admission.run(WORKHUB_COORDINATION_SESSION_ID, async (lease) => { + 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: { + sessionId: request.sessionId, + turnId: request.turnId, + runId: `workhub-run-${request.turnId}`, + status: 'running', + }, + }; + }); + }, + hasRootTurnAdmission: async (_sessionId, turnId) => admitted.has(turnId), + }; + return { executions, starts, prepared }; +} + function coordinator( root: string, store: SessionAuthorityStore, requestDrain: () => void = () => undefined, - resolveCreateTarget: () => Promise = async () => ({ - llmConnectionSlug: 'test-connection', - model: 'test-model', - permissionMode: 'explore', - collaborationMode: 'agent', - orchestrationMode: 'default', - }), + resolveCreateTarget: (() => Promise) | undefined = undefined, + executions: CoordinationExecutions = { + startWorkHubCoordinationMessage: async () => ({ + ok: false, + error: { + code: 'operation_unavailable', + message: 'WorkHub test execution is not configured', + }, + }), + hasRootTurnAdmission: async () => false, + }, + admission: SessionAdmissionGate = new SessionAdmissionGate(), ) { return new HostWorkHubCoordinationCoordinator({ stateRoot: root, stores: store, - admission: new SessionAdmissionGate(), + admission, continuity: { refreshCanonical: async () => undefined }, - resolveCreateTarget, + executions, + resolveCreateTarget: + resolveCreateTarget ?? + (async () => ({ + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'explore', + collaborationMode: 'agent', + orchestrationMode: 'default', + })), requestDrain, }); } 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 ccd1b93dc9..db90fb8d0c 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts @@ -21,9 +21,12 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { RuntimeHostProtocolError } from '../protocol/errors.js'; import { + decodeWorkHubCoordinationAnswerInput, + decodeWorkHubCoordinationRecordInput, decodeWorkHubCoordinationResolveInput, decodeWorkHubCoordinationResolveResult, HOST_OPERATION_SPECS, + REMOTE_OWNER_OPERATION_GRANTS, RUNTIME_HOST_COMPATIBILITY_EPOCH, } from '../protocol/index.js'; @@ -33,7 +36,7 @@ test('WorkHub Coordination resolve has a closed empty input and bounded identity sessionId: 'coordination', }); assert.equal(HOST_OPERATION_SPECS['workhub.coordination.resolve'].mode, 'command'); - assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 48); + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 49); assert.throws( () => decodeWorkHubCoordinationResolveInput({ sessionId: 'caller-selected' }), (error) => error instanceof RuntimeHostProtocolError, @@ -43,3 +46,39 @@ test('WorkHub Coordination resolve has a closed empty input and bounded identity (error) => error instanceof RuntimeHostProtocolError, ); }); + +test('WorkHub Coordination answer and summary inputs are closed and bounded', () => { + assert.deepEqual( + 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.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, + ); +}); diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index a42993bfec..30f01b2cb7 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -92,7 +92,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 = 49 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 50 as const; +// 50: WorkHub can append durable coordination summaries and admit tool-free +// answers through its reserved Coordination Session authority. // 49: WorkHub resolves one durable Coordination Session per Runtime Host. // Older peers do not know the operation or the hidden Session role. // 48: Session branch creation accepts an explicit Side Conversation intent. diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 9de8bb5771..9d95dec0bf 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -325,6 +325,8 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'turn.stop', 'usage.query', 'web-search.execute', + 'workhub.coordination.answer', + '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 fab9e24ced..66866c0945 100644 --- a/packages/runtime-host/src/protocol/workhub-coordination.ts +++ b/packages/runtime-host/src/protocol/workhub-coordination.ts @@ -17,9 +17,12 @@ * under the License. */ -import { requireEntityId, requireExactRecord } from './codec.js'; +import { requireEntityId, requireExactRecord, requireUtf8String } from './codec.js'; import { defineOperation } from './operation-spec.js'; +const COORDINATION_TEXT_MAX_BYTES = 48 * 1024; +const COORDINATION_SUMMARY_MAX_BYTES = 8 * 1024; + const RESOLVE_ERRORS = [ 'host_not_ready', 'host_draining', @@ -30,12 +33,40 @@ const RESOLVE_ERRORS = [ 'internal_failure', ] as const; +const TURN_ERRORS = [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'not_found', + 'session_archived', + 'session_busy', + 'operation_conflict', + 'persistence_failed', + 'commit_outcome_unknown', + 'internal_failure', +] as const; + export type WorkHubCoordinationResolveInput = Record; export interface WorkHubCoordinationResolveResult { readonly sessionId: string; } +export interface WorkHubCoordinationAnswerInput { + readonly turnId: string; + readonly text: string; +} + +export interface WorkHubCoordinationRecordInput { + readonly turnId: string; + readonly userText: string; + readonly assistantText: string; +} + +export interface WorkHubCoordinationTurnResult { + readonly turnId: string; +} + export const WORKHUB_COORDINATION_OPERATION_SPECS = { 'workhub.coordination.resolve': defineOperation< WorkHubCoordinationResolveInput, @@ -48,6 +79,28 @@ export const WORKHUB_COORDINATION_OPERATION_SPECS = { decodeInput: decodeWorkHubCoordinationResolveInput, decodeOutput: decodeWorkHubCoordinationResolveResult, }), + 'workhub.coordination.answer': defineOperation< + WorkHubCoordinationAnswerInput, + WorkHubCoordinationTurnResult, + (typeof TURN_ERRORS)[number] + >({ + mode: 'command', + availability: 'ready', + errors: TURN_ERRORS, + 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, + }), } as const; export function decodeWorkHubCoordinationResolveInput( @@ -65,3 +118,47 @@ export function decodeWorkHubCoordinationResolveResult( sessionId: requireEntityId(result.sessionId, 'WorkHub Coordination Session id'), }; } + +export function decodeWorkHubCoordinationAnswerInput( + value: unknown, +): WorkHubCoordinationAnswerInput { + const input = requireExactRecord(value, 'WorkHub Coordination answer input', ['turnId', 'text']); + return { + turnId: requireEntityId(input.turnId, 'WorkHub Coordination Turn id'), + text: requireUtf8String( + input.text, + 'WorkHub Coordination answer text', + COORDINATION_TEXT_MAX_BYTES, + ), + }; +} + +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', + COORDINATION_TEXT_MAX_BYTES, + ), + assistantText: requireUtf8String( + input.assistantText, + 'WorkHub Coordination assistant text', + COORDINATION_SUMMARY_MAX_BYTES, + ), + }; +} + +export function decodeWorkHubCoordinationTurnResult(value: unknown): WorkHubCoordinationTurnResult { + const result = requireExactRecord(value, 'WorkHub Coordination Turn result', ['turnId']); + return { + turnId: requireEntityId(result.turnId, 'WorkHub Coordination Turn id'), + }; +} diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 575d7b1060..ef0f62ba3d 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1225,6 +1225,7 @@ export async function createExecutionRuntimeHostComposition( stores: stores.sessionStore, admission: sessionAdmission, continuity: continuityCoordinator, + executions: coordinator, resolveCreateTarget: async () => { const { projectId: _projectId, ...target } = await sessionCatalog.resolveExternalSessionImportTarget(); diff --git a/packages/runtime-host/src/server/host-session-availability.ts b/packages/runtime-host/src/server/host-session-availability.ts index cb3573ced1..88c7b2e1fa 100644 --- a/packages/runtime-host/src/server/host-session-availability.ts +++ b/packages/runtime-host/src/server/host-session-availability.ts @@ -18,7 +18,13 @@ */ import type { RootExecutionDescriptor } from '@maka/core/agent-run'; -import { isWorkHubCoordinationSessionTarget, type SessionHeader } from '@maka/core/session'; +import { + isWorkHubCoordinationSession, + isWorkHubCoordinationSessionId, + isWorkHubCoordinationSessionTarget, + type SessionHeader, + type SessionToolProfile, +} from '@maka/core/session'; const WORKTREE_CHILD_UNAVAILABLE_REASON = 'Worktree child Sessions must be continued through their parent agent.'; @@ -27,6 +33,8 @@ const CHILD_CONTINUATION_UNAVAILABLE_REASON = const IMPORT_STAGING_UNAVAILABLE_REASON = 'Imported Session history is still being prepared.'; export const WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON = 'WorkHub Coordination Session execution requires WorkHub authority'; +export const WORKHUB_COORDINATION_TARGET_UNAVAILABLE_REASON = + 'WorkHub Coordination execution requires the reserved Coordination Session'; export function runtimeHostExternalTurnUnavailableReason( header: Pick< @@ -53,11 +61,30 @@ export function runtimeHostExecutionUnavailableReason( header: Pick< SessionHeader, 'id' | 'role' | 'collaborationMode' | 'subagentWorkspace' | 'transcriptLedgerVersion' - >, + > & { + readonly toolProfile?: SessionToolProfile; + readonly permissionMode?: SessionHeader['permissionMode']; + readonly orchestrationMode?: SessionHeader['orchestrationMode']; + }, execution: RootExecutionDescriptor, ): string | undefined { + const coordinationTarget = isWorkHubCoordinationSessionTarget(header); + const coordinationIdentity = + isWorkHubCoordinationSessionId(header.id) && isWorkHubCoordinationSession(header); return ( - (isWorkHubCoordinationSessionTarget(header) + (coordinationTarget && execution.kind !== 'workhub_coordination' + ? WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON + : undefined) ?? + (execution.kind === 'workhub_coordination' && !coordinationIdentity + ? WORKHUB_COORDINATION_TARGET_UNAVAILABLE_REASON + : undefined) ?? + (execution.kind === 'workhub_coordination' && header.toolProfile !== 'workhub-coordination-v1' + ? WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON + : undefined) ?? + (execution.kind === 'workhub_coordination' && + (header.permissionMode !== 'explore' || + (header.collaborationMode ?? 'agent') !== 'agent' || + (header.orchestrationMode ?? 'default') !== 'default') ? WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON : undefined) ?? (header.transcriptLedgerVersion === 0 ? IMPORT_STAGING_UNAVAILABLE_REASON : undefined) ?? diff --git a/packages/runtime-host/src/server/hosted-execution-projection.ts b/packages/runtime-host/src/server/hosted-execution-projection.ts index f5f50f82cd..b5ab57eb2d 100644 --- a/packages/runtime-host/src/server/hosted-execution-projection.ts +++ b/packages/runtime-host/src/server/hosted-execution-projection.ts @@ -100,6 +100,7 @@ function assertRunMatchesExecution( } switch (execution.kind) { case 'external_message': + case 'workhub_coordination': return; case 'regenerate': case 'context_compact': @@ -143,6 +144,7 @@ function assertTrustedAgentIdentity( { kind: | 'external_message' + | 'workhub_coordination' | 'regenerate' | 'context_compact' | 'scheduled_task' diff --git a/packages/runtime-host/src/server/hosted-execution-recovery.ts b/packages/runtime-host/src/server/hosted-execution-recovery.ts index 6de7a9a061..d720c185d5 100644 --- a/packages/runtime-host/src/server/hosted-execution-recovery.ts +++ b/packages/runtime-host/src/server/hosted-execution-recovery.ts @@ -353,6 +353,8 @@ function recoveryExecutionContract(execution: RootExecutionDescriptor): Recovery switch (execution.kind) { case 'external_message': return contract(true, true, 'root_replay'); + case 'workhub_coordination': + return contract(false, true, 'root_replay'); case 'regenerate': return contract(false, true, 'root_replay'); case 'context_compact': diff --git a/packages/runtime-host/src/server/hosted-execution-tool-profile.ts b/packages/runtime-host/src/server/hosted-execution-tool-profile.ts index 35e28d0aec..5c497f2e40 100644 --- a/packages/runtime-host/src/server/hosted-execution-tool-profile.ts +++ b/packages/runtime-host/src/server/hosted-execution-tool-profile.ts @@ -48,6 +48,14 @@ const HEADLESS_CODING_V1_BASH_PARAMETERS = z }) .strict(); +const WORKHUB_COORDINATION_V1_SYSTEM_PROMPT = [ + 'You are the conversational coordinator for WorkHub.', + 'Answer ordinary questions directly and help the user clarify intent.', + 'Reply in the language used by the user unless they ask for another language.', + 'This conversation has no tools, filesystem authority, or authority over ordinary Sessions.', + 'Never claim to have inspected files, run commands, changed a Session, or completed concrete work.', +].join(' '); + export interface HostedExecutionRunProfile { readonly toolNames: readonly string[]; readonly systemPrompt: string; @@ -65,6 +73,13 @@ export function hostedExecutionRunProfile( memoryExtraction: false, }; } + if (profile === 'workhub-coordination-v1') { + return { + toolNames: [], + systemPrompt: WORKHUB_COORDINATION_V1_SYSTEM_PROMPT, + memoryExtraction: false, + }; + } profile satisfies never; throw new Error('Unknown Session tool profile'); } diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index e38b1df87d..9e36149c13 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -167,7 +167,7 @@ type RootMessageStartOutcome = export type RootMessageExecution = Extract< RootExecutionDescriptor, - { kind: 'external_message' | 'regenerate' } + { kind: 'external_message' | 'workhub_coordination' | 'regenerate' } >; interface RootMessageStartRequestBase { @@ -194,6 +194,11 @@ export type RootMessageStartRequest = readonly execution: Extract; readonly turnOrchestration?: undefined; prepareContent(): Promise; + }) + | (RootMessageStartRequestBase & { + readonly execution: Extract; + readonly turnOrchestration?: undefined; + prepareFreshContent(lease: SessionAdmissionLease): Promise; }); export type RootMessageContentPreparation = @@ -1339,6 +1344,40 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { operationUnavailable(WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON), ); } + if (request.execution.kind === 'workhub_coordination') { + return Promise.resolve( + operationUnavailable(WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON), + ); + } + return this.startRootMessage(request, context); + } + + /** Dedicated WorkHub authority; the ordinary interactive entry stays closed. */ + startWorkHubCoordinationMessage( + request: Extract, + context: ConnectionContext, + ): Promise { + if (!isWorkHubCoordinationSessionId(request.sessionId)) { + return Promise.resolve( + operationUnavailable('WorkHub Coordination execution requires its reserved Session'), + ); + } + 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; + } + + private startRootMessage( + request: RootMessageStartRequest, + context: ConnectionContext, + ): Promise { return this.runCommand(async () => { await this.awaitTerminalRootCleanup(request.sessionId); const activeAtEntry = this.#executions.has(request.sessionId); @@ -1418,7 +1457,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { if (header.isArchived) { return completedStart(sessionArchived(request.archivedMessage)); } - const unavailableReason = runtimeHostExternalTurnUnavailableReason(header); + const unavailableReason = runtimeHostExecutionUnavailableReason(header, request.execution); if (unavailableReason) return completedStart(operationUnavailable(unavailableReason)); if (this.#executions.has(request.sessionId)) { return completedStart(sessionBusy('Session already has an active root Turn')); @@ -1449,9 +1488,12 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { attachments, ); if (attachmentError) return completedStart(operationConflict(attachmentError)); - const binding = prepared.commitCapabilityBinding - ? await prepared.commitCapabilityBinding() - : await this.clientCapabilities?.bindSession(request.sessionId, context.connectionId); + const binding = + request.execution.kind === 'workhub_coordination' + ? undefined + : prepared.commitCapabilityBinding + ? await prepared.commitCapabilityBinding() + : await this.clientCapabilities?.bindSession(request.sessionId, context.connectionId); if (binding && !binding.ok) { return completedStart(operationConflict(binding.message)); } diff --git a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts index f6e766b0f4..af08e6c419 100644 --- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts +++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts @@ -20,6 +20,7 @@ import { createHash } from 'node:crypto'; import { mkdir } from 'node:fs/promises'; import { join } from 'node:path'; +import { normalizeMessageContent } from '@maka/core/events'; import type { CreateSessionInput } from '@maka/core/runtime-inputs'; import { WORKHUB_COORDINATION_SESSION_ID, @@ -27,10 +28,19 @@ import { isWorkHubCoordinationSession, isWorkHubCoordinationSessionId, type SessionHeader, + type StoredMessage, } from '@maka/core/session'; import type { SessionAuthorityStore, SessionHeaderSnapshot } from '@maka/storage/session-store'; -import type { OperationOutcome } from '../protocol/index.js'; -import type { WorkHubCoordinationOperationHandlerMap } from './operation-dispatcher.js'; +import type { + OperationOutcome, + WorkHubCoordinationAnswerInput, + WorkHubCoordinationRecordInput, +} from '../protocol/index.js'; +import type { + ConnectionContext, + WorkHubCoordinationOperationHandlerMap, +} from './operation-dispatcher.js'; +import type { RootTurnCoordinator } from './root-turn-coordinator.js'; import { SessionAdmissionGate } from './session-admission-gate.js'; import { SessionOperationFailure } from './session-catalog-coordinator.js'; import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js'; @@ -39,10 +49,29 @@ const CREATE_FINGERPRINT = `sha256:${createHash('sha256') .update('maka:workhub-coordination-session:v1', 'utf8') .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; +const COORDINATION_SUMMARY_MESSAGE_KINDS = ['user', 'assistant', 'state'] as const; +const TURN_IDENTITY_CONFLICT_MESSAGE = + 'WorkHub Coordination Turn identity belongs to a different operation'; type CoordinationStores = Pick< SessionAuthorityStore, - 'createStableSession' | 'probeStableSessionCreate' | 'updateHeaderVersioned' + | 'appendMessages' + | 'createStableSession' + | 'probeStableSessionCreate' + | 'readHeaderSnapshot' + | 'readTranscriptHighWaterSnapshot' + | 'readTranscriptMessagesSnapshot' + | 'updateHeaderVersioned' +>; + +type CoordinationExecutions = Pick< + RootTurnCoordinator, + 'startWorkHubCoordinationMessage' | 'hasRootTurnAdmission' >; export type CoordinationCreateTarget = Omit; @@ -52,6 +81,7 @@ export interface HostWorkHubCoordinationCoordinatorOptions { readonly stores: CoordinationStores; readonly admission: SessionAdmissionGate; readonly continuity: Pick; + readonly executions: CoordinationExecutions; readonly resolveCreateTarget: () => Promise; readonly requestDrain: () => void; } @@ -60,12 +90,15 @@ export interface HostWorkHubCoordinationCoordinatorOptions { 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), }; readonly #coordinationCwd: string; readonly #stores: CoordinationStores; readonly #admission: SessionAdmissionGate; readonly #continuity: Pick; + readonly #executions: CoordinationExecutions; readonly #resolveCreateTarget: () => Promise; readonly #requestDrain: () => void; @@ -74,6 +107,7 @@ export class HostWorkHubCoordinationCoordinator { this.#stores = options.stores; this.#admission = options.admission; this.#continuity = options.continuity; + this.#executions = options.executions; this.#resolveCreateTarget = options.resolveCreateTarget; this.#requestDrain = options.requestDrain; } @@ -104,8 +138,8 @@ export class HostWorkHubCoordinationCoordinator { } if (probe.kind === 'existing') { - return validCoordinationHeader(probe.record.header) - ? await this.#alignWorkspace(probe.record) + return validCoordinationIdentityHeader(probe.record.header) + ? await this.#alignSession(probe.record) : identityConflict(); } if (probe.kind === 'conflict') return identityConflict(); @@ -123,10 +157,14 @@ export class HostWorkHubCoordinationCoordinator { requestFingerprint: CREATE_FINGERPRINT, input: { ...target, + permissionMode: COORDINATION_PERMISSION_MODE, + collaborationMode: COORDINATION_COLLABORATION_MODE, + orchestrationMode: COORDINATION_ORCHESTRATION_MODE, cwd: this.#coordinationCwd, projectId: null, name: 'WorkHub', role: WORKHUB_COORDINATION_SESSION_ROLE, + toolProfile: COORDINATION_TOOL_PROFILE, }, }); if (result.kind === 'conflict') return identityConflict(); @@ -148,21 +186,173 @@ export class HostWorkHubCoordinationCoordinator { }); } + async #answer( + input: WorkHubCoordinationAnswerInput, + context: ConnectionContext, + ): Promise> { + if (!input.text.trim()) { + return turnFailure( + 'workhub.coordination.answer', + 'operation_conflict', + 'WorkHub answer text is empty', + ); + } + const outcome = await this.#executions.startWorkHubCoordinationMessage( + { + sessionId: WORKHUB_COORDINATION_SESSION_ID, + turnId: input.turnId, + execution: { + kind: 'workhub_coordination', + 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. + prepareFreshContent: async () => { + let recorded: readonly StoredMessage[]; + try { + recorded = await this.#readSummaryMessages(input.turnId); + } catch { + return { + kind: 'rejected', + outcome: operationUnavailable( + 'WorkHub Coordination Turn identity could not be verified', + ), + }; + } + return recorded.length > 0 + ? { kind: 'rejected', outcome: turnIdentityConflict() } + : { kind: 'ready', content: normalizeMessageContent({ text: input.text }) }; + }, + }, + context, + ); + 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. */ + async #readSummaryMessages(turnId: string): Promise { + const throughSequence = await this.#stores.readTranscriptHighWaterSnapshot( + WORKHUB_COORDINATION_SESSION_ID, + ); + if (throughSequence === null) return []; + return this.#stores.readTranscriptMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID, { + messageIds: COORDINATION_SUMMARY_MESSAGE_KINDS.map((kind) => + coordinationSummaryMessageId(turnId, kind), + ), + throughSequence, + maxBytes: 32 * 1024, + maxMessages: COORDINATION_SUMMARY_MESSAGE_KINDS.length, + }); + } + /** * The workspace path is derived from the Host state root, so it moves with the * installation. Identity stays in the id/role pair and the durable path is * repaired in place — rejecting the drift would strand the one Session no * ordinary lifecycle operation is allowed to relocate or retire. */ - async #alignWorkspace( + async #alignSession( record: SessionHeaderSnapshot, ): Promise> { - if (record.header.cwd === this.#coordinationCwd) return success(); + if ( + record.header.toolProfile !== undefined && + record.header.toolProfile !== COORDINATION_TOOL_PROFILE + ) { + return identityConflict(); + } + if (record.header.cwd === this.#coordinationCwd && validCoordinationHeader(record.header)) { + return success(); + } let repaired: SessionHeaderSnapshot; try { repaired = await this.#stores.updateHeaderVersioned( WORKHUB_COORDINATION_SESSION_ID, - { cwd: this.#coordinationCwd }, + { + ...(record.header.cwd === this.#coordinationCwd ? {} : { cwd: this.#coordinationCwd }), + ...(record.header.toolProfile === COORDINATION_TOOL_PROFILE + ? {} + : { toolProfile: COORDINATION_TOOL_PROFILE }), + ...(record.header.permissionMode === COORDINATION_PERMISSION_MODE + ? {} + : { permissionMode: COORDINATION_PERMISSION_MODE }), + ...((record.header.collaborationMode ?? 'agent') === COORDINATION_COLLABORATION_MODE + ? {} + : { collaborationMode: COORDINATION_COLLABORATION_MODE }), + ...((record.header.orchestrationMode ?? 'default') === COORDINATION_ORCHESTRATION_MODE + ? {} + : { orchestrationMode: COORDINATION_ORCHESTRATION_MODE }), + }, record.revision, ); } catch { @@ -188,7 +378,7 @@ function createTargetFailure(error: unknown): OperationOutcome<'workhub.coordina ); } -function validCoordinationHeader(header: SessionHeader): boolean { +function validCoordinationIdentityHeader(header: SessionHeader): boolean { return ( isWorkHubCoordinationSessionId(header.id) && isWorkHubCoordinationSession(header) && @@ -201,6 +391,80 @@ function validCoordinationHeader(header: SessionHeader): boolean { ); } +function validCoordinationHeader(header: SessionHeader): boolean { + return ( + validCoordinationIdentityHeader(header) && + header.toolProfile === COORDINATION_TOOL_PROFILE && + header.permissionMode === COORDINATION_PERMISSION_MODE && + (header.collaborationMode ?? 'agent') === COORDINATION_COLLABORATION_MODE && + (header.orchestrationMode ?? 'default') === COORDINATION_ORCHESTRATION_MODE + ); +} + +function digest(value: unknown): `sha256:${string}` { + return `sha256:${createHash('sha256').update(JSON.stringify(value)).digest('hex')}`; +} + +function coordinationSummaryMessageId( + turnId: string, + kind: (typeof COORDINATION_SUMMARY_MESSAGE_KINDS)[number], +): string { + return `workhub_${createHash('sha256') + .update(`${turnId}\0${kind}`, 'utf8') + .digest('hex') + .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, @@ -221,3 +485,23 @@ function failure( ): OperationOutcome<'workhub.coordination.resolve'> { return { ok: false, error: { code, message } }; } + +/** Fresh-admission rejections for the answer's lease-scoped identity probe. */ +function turnIdentityConflict() { + return { + ok: false, + error: { code: 'operation_conflict', message: TURN_IDENTITY_CONFLICT_MESSAGE }, + } as const; +} + +function operationUnavailable(message: string) { + return { ok: false, error: { code: 'operation_unavailable', message } } as const; +} + +function turnFailure( + _operation: K, + code: Extract, { readonly ok: false }>['error']['code'], + message: string, +): OperationOutcome { + return { ok: false, error: { code, message } } as OperationOutcome; +} diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 8a5975de84..a2cb3dec0e 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -4623,7 +4623,10 @@ export class SessionManager { wakeId: input.execution.wakeId, attemptId: input.execution.attemptId, }; - } else if (input.execution.kind !== 'external_message') { + } else if ( + input.execution.kind !== 'external_message' && + input.execution.kind !== 'workhub_coordination' + ) { if ( session.subagentParent?.kind !== 'subagent' || session.subagentRuntime?.agentId !== input.execution.agentId || diff --git a/packages/storage/src/__tests__/workhub-coordination-root-admission.test.ts b/packages/storage/src/__tests__/workhub-coordination-root-admission.test.ts new file mode 100644 index 0000000000..8da5ea98f3 --- /dev/null +++ b/packages/storage/src/__tests__/workhub-coordination-root-admission.test.ts @@ -0,0 +1,74 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import type { RootExecutionDescriptor } from '@maka/core/agent-run'; +import { createSqliteAgentRunStore } from '../agent-run-store.js'; + +test('WorkHub Coordination admission preserves its bounded content identity across restart', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-admission-')); + const inputDigest = `sha256:${'a'.repeat(64)}` as const; + try { + const store = createSqliteAgentRunStore(root); + const admitted = await store.admitRootTurn({ + sessionId: 'coordination-session', + turnId: 'coordination-turn', + proposedRunId: 'coordination-run', + proposedUserMessageId: 'coordination-message', + execution: { kind: 'workhub_coordination', inputDigest }, + previousRootTurnId: null, + normalizedInput: { text: 'What should happen next?' }, + sourceMessages: [], + admittedAt: 50, + }); + assert.equal(admitted.kind, 'admitted'); + store.close?.(); + + const reopened = createSqliteAgentRunStore(root); + assert.deepEqual( + await reopened.readRootTurnAdmission('coordination-session', 'coordination-turn'), + admitted.admission, + ); + await assert.rejects( + () => + reopened.admitRootTurn({ + sessionId: 'coordination-session', + turnId: 'invalid-coordination-turn', + proposedRunId: 'invalid-coordination-run', + proposedUserMessageId: 'invalid-coordination-message', + execution: { + kind: 'workhub_coordination', + inputDigest: 'sha256:not-a-digest', + } as RootExecutionDescriptor, + previousRootTurnId: 'coordination-turn', + normalizedInput: { text: 'Invalid identity' }, + sourceMessages: [], + admittedAt: 60, + }), + /Invalid root execution descriptor/u, + ); + reopened.close?.(); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index f11839265a..9b1af4fe73 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -1895,6 +1895,15 @@ function normalizeRootExecutionDescriptor(value: unknown): RootExecutionDescript ...(value.maxSteps !== undefined ? { maxSteps: value.maxSteps } : {}), }); } + if (value.kind === 'workhub_coordination') { + if (!hasExactKeys(value, ['kind', 'inputDigest']) || !isSha256Digest(value.inputDigest)) { + throw new Error('Invalid root execution descriptor'); + } + return Object.freeze({ + kind: 'workhub_coordination', + inputDigest: value.inputDigest, + }); + } if (value.kind === 'regenerate') { if ( !hasExactKeys(value, ['kind', 'sourceTurnId']) ||