diff --git a/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts b/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts index 2ab121f450..0a0f5e3a70 100644 --- a/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts @@ -26,6 +26,7 @@ import { EMPTY_USAGE_PROVENANCE } from '@maka/core/usage-ledger-merge'; import { projectDesktopSessionEvent, projectDesktopSessionSummary, + projectDesktopStoredMessage, projectDesktopTurnRecord, projectDesktopUsageStats, } from '../../shared/desktop-session-projection.js'; @@ -199,6 +200,35 @@ test('projects only present Usage Session ids into the Desktop host namespace', assert.equal(projected.logs[1]?.sessionId, undefined); }); +test('projects durable WorkHub delegation targets into the Desktop host namespace', () => { + const projected = projectDesktopStoredMessage( + { hostId: 'remote-root' }, + { + type: 'workhub_coordination', + id: 'delegation-assignment-message', + turnId: 'coordination-turn', + ts: 2, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId: 'action-id', + actionFingerprint: `sha256:${'a'.repeat(64)}`, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + userText: 'Continue payment work', + delegationId: 'delegation-id', + targetTurnId: 'payments-turn', + targetMessageId: 'payments-message', + targetSessionName: 'Payments', + }, + ); + + assert.equal(projected.type, 'workhub_coordination'); + if (projected.type === 'workhub_coordination') { + assert.equal(projected.targetSessionId, JSON.stringify(['remote-root', 'payments'])); + } +}); + function summary(id: string): SessionSummary { return { id, 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 95e6f40d21..2aa190fb92 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 @@ -19,6 +19,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; +import { RuntimeHostOperationError } from '@maka/runtime-host/client'; import { registerRuntimeHostWorkHubIpc } from '../runtime-host-workhub-ipc-main.js'; test('projects WorkHub coordination resolution through its dedicated IPC domain', async () => { @@ -111,9 +112,12 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain' }, }), { - disposition: 'create_new', - targetSessionId: createdSessionId, - targetTurnId: 'created-turn', + ok: true, + result: { + disposition: 'create_new', + targetSessionId: createdSessionId, + targetTurnId: 'created-turn', + }, }, ); assert.deepEqual(actions, [{ @@ -126,3 +130,43 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain' }]); assert.deepEqual(changes, [{ reason: 'created', sessionId: createdSessionId }]); }); + +test('serializes typed WorkHub action failures across Electron IPC', async () => { + const handlers = new Map unknown>(); + registerRuntimeHostWorkHubIpc( + { + actWorkHubCoordination: async () => { + throw new RuntimeHostOperationError( + 'workhub.coordination.act', + 'operation_conflict', + 'WorkHub action is permanently abandoned', + ); + }, + } as never, + { + handle: (channel: string, handler: (...args: unknown[]) => unknown) => { + handlers.set(channel, handler); + }, + } as never, + { + resolveCreateProject: async () => ({ kind: 'host_path', path: '/workspace' }), + emitSessionsChanged: () => undefined, + }, + ); + + assert.deepEqual( + await handlers.get('workhub:act')?.({}, { + actionId: 'abandoned-action', + userText: 'Continue payment work', + candidateSetId: `sha256:${'a'.repeat(64)}`, + proposal: { disposition: 'delegate_existing', candidateRef: 'candidate' }, + }), + { + ok: false, + error: { + code: 'operation_conflict', + message: 'WorkHub action is permanently abandoned', + }, + }, + ); +}); diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index 6ed05e7962..ff438b0540 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -2515,6 +2515,54 @@ test('production submission delegates only through the Runtime-owned candidate r }]); }); +test('production retry reaches durable Action Gate replay while target is waiting', async () => { + const actions: unknown[] = []; + const sessions = port([ + session('payment', { state: 'waiting_for_user' }), + ]); + const controller = createGatedWorkHubController({ + sessions, + coordination: { + open: async () => ({ close: async () => undefined }), + answer: async (input) => ({ turnId: input.turnId }), + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => ({ + candidateSetId: `sha256:${'c'.repeat(64)}`, + candidates: [{ + candidateRef: 'candidate-payment', + sessionId: 'payment', + sessionName: 'payment', + workspace: { + target: { kind: 'host_path', path: '/workspace/payment' }, + hostCwd: '/workspace/payment', + }, + state: 'waiting_for_user', + updatedAt: 2, + }], + }), + act: async (input) => { + actions.push(input); + return { + disposition: 'delegate_existing', + targetSessionId: 'payment', + targetTurnId: 'already-committed-turn', + }; + }, + }, + }); + + const result = await controller.submit({ + requestId: 'summary-recovery-action', + text: '继续支付工作', + explicitTarget: { sessionId: 'payment' }, + retryAction: true, + }); + + assert.equal(result.kind, 'submitted'); + assert.equal(result.kind === 'submitted' ? result.turnId : undefined, 'already-committed-turn'); + assert.equal(actions.length, 1); +}); + test('production defers destructive correction until persistent delegation exists', async () => { const actions: unknown[] = []; const sessions = port([session('source'), session('target')]); 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 50e4ec5afd..8f3eb7d0c2 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -117,6 +117,24 @@ test('projects the durable Coordination transcript into the WorkHub conversation status: 'completed', partialOutputRetained: true, }, + { + type: 'workhub_coordination', + id: 'assignment-1', + turnId: 'action-1', + ts: 20, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId: 'action-1', + actionFingerprint: `sha256:${'a'.repeat(64)}`, + coordinationTurnId: 'action-1', + targetSessionId: 'payments', + targetSessionName: 'Payments', + targetTurnId: 'payments-turn', + targetMessageId: 'payments-message', + delegationId: 'payments-delegation', + disposition: 'delegate_existing', + userText: 'Continue payments', + }, ]), [{ messageId: 'user-1', turnId: 'turn-1', @@ -124,6 +142,16 @@ test('projects the durable Coordination transcript into the WorkHub conversation result: 'Slice 3 is next.', state: 'completed', updatedAt: 11, + }, { + messageId: 'assignment-1', + turnId: 'action-1', + text: 'Continue payments', + state: 'completed', + assignment: { + targetSessionId: 'payments', + targetSessionName: 'Payments', + }, + updatedAt: 20, }]); }); @@ -168,8 +196,11 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and candidates: [], }), act: async () => ({ - disposition: 'answer_here', - coordinationTurnId: 'coordination-turn', + ok: true, + result: { + disposition: 'answer_here', + coordinationTurnId: 'coordination-turn', + }, }), }); diff --git a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts index 3440e0b854..b6f1eb03ec 100644 --- a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts @@ -26,6 +26,8 @@ import { WorkHubCoordinationStatus, WorkHubProjectionRefreshGate, WorkHubSurfaceRouteGate, + submitAndRecordWorkHubSurfaceInput, + submitLeasedWorkHubSurfaceInput, submitWorkHubSurfaceInput, visibleWorkHubConversation, workHubSurfaceFailure, @@ -37,6 +39,7 @@ import { type WorkHubController, type WorkHubSubmitInput, } from '../../renderer/workhub-controller.js'; +import { WorkHubSendLease } from '../../renderer/workhub-send-lease.js'; import { createDesktopWorkHubSessionPort, type WorkHubDesktopSession, @@ -81,18 +84,19 @@ test('surface route gate rejects same-frame duplicate operations and reopens aft }); test('Coordination lifecycle keeps a visible loading state and exposes failure recovery', () => { - const renderStatus = (state: 'resolving' | 'failed') => renderToStaticMarkup( - createElement(LocaleProvider, { - locale: 'en', - children: createElement(AstryxLocaleProvider, { - children: createElement(WorkHubCoordinationStatus, { - locale: 'en', - state, - onRetry: () => undefined, + const renderStatus = (state: 'resolving' | 'failed') => + renderToStaticMarkup( + createElement(LocaleProvider, { + locale: 'en', + children: createElement(AstryxLocaleProvider, { + children: createElement(WorkHubCoordinationStatus, { + locale: 'en', + state, + onRetry: () => undefined, + }), }), }), - }), - ); + ); const resolving = renderStatus('resolving'); const failed = renderStatus('failed'); @@ -116,41 +120,51 @@ test('surface projection refresh gate rejects older reads after a newer refresh test('surface keeps the Composer draft when routing fails or the target is waiting', () => { assert.equal(workHubSubmissionClearsDraft(undefined), false); - assert.equal(workHubSubmissionClearsDraft({ - kind: 'waiting', - strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: 'waiting', - text: '继续处理', - target: { sessionId: 'payment' }, - }), false); - assert.equal(workHubSubmissionClearsDraft({ - kind: 'discussion', - strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: 'discussion', - text: '先讨论方向', - }), true); + assert.equal( + workHubSubmissionClearsDraft({ + kind: 'waiting', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: 'waiting', + text: '继续处理', + target: { sessionId: 'payment' }, + }), + false, + ); + assert.equal( + workHubSubmissionClearsDraft({ + kind: 'discussion', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: 'discussion', + text: '先讨论方向', + }), + true, + ); }); 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: 'discussion' as const, - strategyId: WORKHUB_ROUTING_STRATEGY_ID, + const local = [ + { requestId: 'discussion-turn', text: 'What is next?', + state: 'settled' as const, + outcome: { + kind: 'discussion' as const, + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: 'discussion-turn', + text: 'What is next?', + }, + }, + ]; + const durable = [ + { + messageId: 'user-message', + turnId: 'discussion-turn', + text: 'What is next?', + result: 'Slice 3 is next.', + state: 'completed' as const, + updatedAt: 10, }, - }]; - 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(visibleWorkHubConversation(durable, local), { coordination: durable, @@ -177,11 +191,13 @@ test('surface keeps clarification and successful routing in WorkHub', async () = strategyId: WORKHUB_ROUTING_STRATEGY_ID, requestId: input.requestId, text: input.text, - options: [{ - target: { sessionId: 'payment' }, - projectName: 'billing', - sessionName: '支付回调幂等性', - }], + options: [ + { + target: { sessionId: 'payment' }, + projectName: 'billing', + sessionName: '支付回调幂等性', + }, + ], }; } return { @@ -241,20 +257,24 @@ test('surface leaves discussion in WorkHub instead of creating a task view', asy test('real Session projection creates new guide topics and preserves origin ambiguity', async () => { let clock = 10; - const sessions: WorkHubDesktopSession[] = [{ - id: 'login', - name: '刷新令牌过期致重复登录的排查计划', - labels: [], - isArchived: false, - status: 'active', - projectId: 'project-router', - lastMessageAt: clock, - lastMessagePreview: '已经整理为检查清单', - }]; - const prompts = new Map([[ - 'login', - ['排查登录刷新令牌过期导致重复登录的问题，先只分析并列出计划，不修改文件。'], - ]]); + const sessions: WorkHubDesktopSession[] = [ + { + id: 'login', + name: '刷新令牌过期致重复登录的排查计划', + labels: [], + isArchived: false, + status: 'active', + projectId: 'project-router', + lastMessageAt: clock, + lastMessagePreview: '已经整理为检查清单', + }, + ]; + const prompts = new Map([ + [ + 'login', + ['排查登录刷新令牌过期导致重复登录的问题，先只分析并列出计划，不修改文件。'], + ], + ]); const created: string[] = []; const port = createDesktopWorkHubSessionPort({ transcripts: { @@ -264,8 +284,8 @@ test('real Session projection creates new guide topics and preserves origin ambi }, sessions: { list: async () => sessions, - listTurns: async (sessionId) => (prompts.get(sessionId) ?? []) - .map((userPromptPreview) => ({ userPromptPreview })), + listTurns: async (sessionId) => + (prompts.get(sessionId) ?? []).map((userPromptPreview) => ({ userPromptPreview })), create: async ({ name }) => { const id = name.includes('支付回调') ? 'payment' : 'layout'; const session: WorkHubDesktopSession = { @@ -294,9 +314,8 @@ test('real Session projection creates new guide topics and preserves origin ambi stop: async () => {}, subscribeChanges: () => () => {}, }, - projectName: (projectId) => projectId === 'project-router' - ? 'maka-workhub-session-router' - : 'maka-agent', + projectName: (projectId) => + projectId === 'project-router' ? 'maka-workhub-session-router' : 'maka-agent', newTurnId: () => `turn-${clock + 1}`, }); const controller = createWorkHubController({ sessions: port }); @@ -322,7 +341,156 @@ test('real Session projection creates new guide topics and preserves origin ambi assert.equal(layout.kind === 'submitted' ? layout.evidence : undefined, 'new_session'); assert.deepEqual(created, ['payment', 'layout']); assert.equal(ambiguous.kind, 'clarification'); - assert.deepEqual(ambiguous.kind === 'clarification' - ? ambiguous.options.map((option) => option.target.sessionId) - : [], ['login', 'payment']); + assert.deepEqual( + ambiguous.kind === 'clarification' + ? ambiguous.options.map((option) => option.target.sessionId) + : [], + ['login', 'payment'], + ); }); + +test('action identity survives reload while edited text remains current', () => { + const { storage } = memoryStorage(); + const first = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-1', + }); + assert.deepEqual(first.acquireAttempt('Continue payments'), { + requestId: 'action-1', + text: 'Continue payments', + retrying: false, + }); + + const restarted = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-2', + }); + assert.deepEqual(restarted.acquireAttempt('Investigate login instead'), { + requestId: 'action-1', + text: 'Investigate login instead', + retrying: true, + }); +}); + +test('draft and action identity have independent Host-scoped lifecycles', () => { + const { storage } = memoryStorage(); + const hostA = new WorkHubSendLease({ scope: 'host-a', storage, createId: () => 'action-a' }); + const hostB = new WorkHubSendLease({ scope: 'host-b', storage, createId: () => 'action-b' }); + hostA.write('workhub', 'A draft'); + hostB.write('workhub', 'B draft'); + assert.equal(hostA.acquire('A send'), 'action-a'); + assert.equal(hostB.acquire('B send'), 'action-b'); + hostA.complete('action-a'); + assert.equal(hostA.read('workhub'), 'A draft'); + assert.equal(hostB.read('workhub'), 'B draft'); + assert.equal(hostB.acquire('B retry'), 'action-b'); +}); + +test('successful delegated submission needs no renderer summary write', async () => { + let records = 0; + const controller = fakeController({ + submit: async (input) => ({ + kind: 'submitted', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + target: { sessionId: 'payments' }, + turnId: 'payments-turn', + evidence: 'explicit_target', + }), + record: async ({ turnId }) => { + records += 1; + return { turnId }; + }, + }); + const result = await submitAndRecordWorkHubSurfaceInput({ + controller, + request: { requestId: 'action-1', text: 'Continue payments' }, + recordedUserText: 'Continue payments', + summary: () => 'Sent to Payments', + onSummaryError: () => assert.fail('no summary write is expected'), + }); + assert.equal(result.kind, 'submitted'); + assert.equal(records, 0); +}); + +test('lease retires only after an acknowledged submission', async () => { + const { storage } = memoryStorage(); + let sends = 0; + const lease = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => `action-${sends + 1}`, + }); + lease.write('workhub', 'Continue payments'); + const cleared = await submitLeasedWorkHubSurfaceInput({ + lease, + text: 'Continue payments', + submit: async (attempt) => { + sends += 1; + return { + kind: 'submitted', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: attempt.requestId, + target: { sessionId: 'payments' }, + turnId: 'payments-turn', + evidence: 'explicit_target', + }; + }, + }); + assert.equal(cleared, true); + assert.equal(lease.acquire('Next work'), 'action-2'); +}); + +test('clarification choice retires its action without clearing the Composer draft', async () => { + const { storage } = memoryStorage(); + const lease = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-choice', + }); + lease.write('workhub', 'Unrelated draft'); + + const clearsComposer = await submitLeasedWorkHubSurfaceInput({ + lease, + text: 'Unrelated draft', + preserveDraft: true, + submit: async (attempt) => ({ + kind: 'submitted', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: attempt.requestId, + target: { sessionId: 'payments' }, + turnId: 'payments-turn', + evidence: 'explicit_target', + }), + }); + + assert.equal(clearsComposer, false); + assert.equal(lease.read('workhub'), 'Unrelated draft'); +}); + +function memoryStorage() { + const values = new Map(); + return { + storage: { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }, + }; +} + +function fakeController(input: { + submit: WorkHubController['submit']; + record: WorkHubController['recordConversationTurn']; +}): WorkHubController { + return { + read: async () => ({ sessions: [], turns: [] }), + submit: input.submit, + openConversation: async () => ({ close: async () => undefined }), + recordConversationTurn: input.record, + subscribe: () => () => undefined, + resetVisitContext: () => undefined, + }; +} 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 621bc5a570..c16a816054 100644 --- a/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts @@ -18,10 +18,13 @@ */ import type { + OperationError, + OperationOutcome, WorkHubCoordinationActInput, WorkHubCoordinationActResult, WorkspaceTarget, } from '@maka/runtime-host/protocol'; +import { RuntimeHostOperationError } from '@maka/runtime-host/client'; import type { DesktopRuntimeHostClient } from './runtime-host-client.js'; import type { ReconnectableReadIpcMain } from './ipc-reconnect-policy.js'; @@ -58,33 +61,62 @@ export function registerRuntimeHostWorkHubIpc( ); ipcMain.handle('workhub:candidates', () => client.listWorkHubCoordinationCandidates()); ipcMain.handle('workhub:act', async (_event, rawInput: RendererWorkHubActionInput) => { - const proposal = rawInput?.proposal; - const base = { - actionId: rawInput?.actionId, - userText: rawInput?.userText, - proposal, - } as Pick; - let result: WorkHubCoordinationActResult; - if (proposal?.disposition === 'create_new') { - result = await client.actWorkHubCoordination({ - ...base, - create: { - workspace: await options.resolveCreateProject(), - }, - }); - } else { - result = await client.actWorkHubCoordination({ - ...base, - ...(rawInput?.candidateSetId === undefined - ? {} - : { candidateSetId: rawInput.candidateSetId }), - }); + try { + const proposal = rawInput?.proposal; + const base = { + actionId: rawInput?.actionId, + userText: rawInput?.userText, + proposal, + } as Pick; + let result: WorkHubCoordinationActResult; + if (proposal?.disposition === 'create_new') { + result = await client.actWorkHubCoordination({ + ...base, + create: { + workspace: await options.resolveCreateProject(), + }, + }); + } else { + result = await client.actWorkHubCoordination({ + ...base, + ...(rawInput?.candidateSetId === undefined + ? {} + : { candidateSetId: rawInput.candidateSetId }), + }); + } + if (result.disposition === 'create_new') { + options.emitSessionsChanged('created', result.targetSessionId); + } else if (result.disposition === 'delegate_existing') { + options.emitSessionsChanged('status-change', result.targetSessionId); + } + return { ok: true, result } satisfies OperationOutcome<'workhub.coordination.act'>; + } catch (error) { + if (!(error instanceof RuntimeHostOperationError)) throw error; + return { + ok: false, + error: workHubActError(error), + } satisfies OperationOutcome<'workhub.coordination.act'>; } - if (result.disposition === 'create_new') { - options.emitSessionsChanged('created', result.targetSessionId); - } else if (result.disposition === 'delegate_existing') { - options.emitSessionsChanged('status-change', result.targetSessionId); - } - return result; }); } + +function workHubActError( + error: RuntimeHostOperationError, +): OperationError<'workhub.coordination.act'> { + switch (error.code) { + case 'host_not_ready': + case 'host_draining': + case 'unauthorized': + case 'operation_unavailable': + case 'not_found': + case 'session_archived': + case 'session_busy': + case 'operation_conflict': + case 'persistence_failed': + case 'commit_outcome_unknown': + case 'internal_failure': + return { code: error.code, message: error.message }; + default: + return { code: 'internal_failure', message: 'WorkHub action failed' }; + } +} diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 877c9b65a5..c92e2d20ea 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -105,6 +105,7 @@ import type { WorkBoardItem, WorkBoardListQuery, WorkBoardPage } from '@maka/cor import type { WorkBoardMutationOptions } from '@maka/storage/work-board-store'; import type { OperationInput, + OperationOutcome, OperationOutput, } from '@maka/runtime-host/protocol'; import type { AgentGraphEpochDirectory } from '@maka/runtime-host/client'; @@ -858,7 +859,7 @@ export interface MakaBridge { act( coordinationSessionId: string, input: Omit, 'create'>, - ): Promise>; + ): Promise>; /** 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 e5550f015e..f1a8ce589b 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -216,6 +216,7 @@ import type { OnboardingMilestoneId } from '@maka/core/onboarding'; import { SCHEDULED_TASK_CATALOG_MAX_ITEMS, type OperationInput, + type OperationOutcome, type OperationOutput, } from '@maka/runtime-host/protocol'; import type { AgentGraphEpochDirectory } from '@maka/runtime-host/client'; @@ -1642,7 +1643,7 @@ const makaBridge = { async act( coordinationSessionId: string, input: Omit, 'create'>, - ): Promise> { + ): Promise> { const scope = await resolveDesktopWorkHubCoordinationCreateScope( coordinationSessionId, runtimeHostSessionRef, @@ -1651,14 +1652,21 @@ const makaBridge = { 'workhub:act', scope, input, - ) as OperationOutput<'workhub.coordination.act'>; - if (result.disposition === 'answer_here' || result.disposition === 'clarify') return result; + ) as OperationOutcome<'workhub.coordination.act'>; + if (!result.ok) return result; + if ( + result.result.disposition === 'answer_here' || + result.result.disposition === 'clarify' + ) return result; return { - ...result, - targetSessionId: desktopSessionKey({ + ok: true, + result: { + ...result.result, + targetSessionId: desktopSessionKey({ hostId: scope.hostId, - sessionId: result.targetSessionId, - }), + sessionId: result.result.targetSessionId, + }), + }, }; }, async createSession( diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index 2c61228a42..458df497d9 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -2810,6 +2810,7 @@ function AppShellContent({ `已自动添加 ${count} 个域名。代理仅作用于 AI 模型请求。`, testing: '测试中…', testCurrent: '测试当前配置', proxyReachable: '代理可达', proxyTestFailed: '代理测试失败', proxyTestError: '代理测试出错', @@ -386,7 +386,7 @@ const SETTINGS_PREFERENCES_COPY_BY_LOCALE = { removeErrors: { invalid_id: 'The pet ID is invalid.', remove_failed: 'The local pet pack could not be removed.' }, }, general: { - incognito: 'Incognito mode', incognitoHelp: 'Pause local memory, web search, and scheduled task triggers.', enableIncognito: 'Enable incognito mode', incognitoFailed: 'Could not change incognito mode', notifications: 'Send a system notification when finished', notificationsHelp: 'Notify when a response finishes or fails while the window is in the background.', notificationsFailed: 'Could not change notification settings', workspaceInstructions: 'Follow project instructions', workspaceInstructionsHelp: 'Automatically read existing AGENTS.md, CLAUDE.md, or GEMINI.md files in each project. Manage the files in their respective projects.', workspaceInstructionsFailed: 'Could not change project instruction settings', workHub: 'Enable WorkHub', workHubHelp: 'See existing work in one entry and conservatively route new input to ordinary tasks.', workHubFailed: 'Could not change WorkHub setting', updateFailed: 'The setting was not applied. Try again later.', defaultModel: 'Default model', defaultModelHelp: 'Model used by new tasks.', notSet: 'Not set', saveDefaultModelFailed: 'Could not save the default model', defaultPermission: 'Default permission mode', defaultPermissionHelp: 'Initial permission mode for new tasks; it can be changed at any time.', saveDefaultPermissionFailed: 'Could not save the default permission mode', defaultThinking: 'Default thinking level', defaultThinkingHelp: 'Thinking level for new tasks; models that do not offer the chosen level use their own default.', followModelDefault: 'Follow model default', saveDefaultThinkingFailed: 'Could not save the default thinking level', proxy: 'Proxy server', proxyHelp: 'Configure a network proxy for AI model requests', enableProxy: 'Enable proxy server', saveNetworkFailed: 'Could not save network settings', proxyProtocol: 'Proxy protocol', serverAddress: 'Server address', port: 'Port', proxyAuth: 'Proxy authentication', proxyAuthHelp: 'Enable this when a username and password are required.', enableProxyAuth: 'Enable proxy authentication', username: 'Username', password: 'Password', bypassList: 'Proxy bypass list', bypassHelp: 'These domains connect directly. Separate multiple domains with commas.', autoBypass: (count) => `${count} ${count === 1 ? 'domain was' : 'domains were'} added automatically. The proxy applies to AI model requests only.`, testing: 'Testing…', testCurrent: 'Test current configuration', proxyReachable: 'Proxy is reachable', proxyTestFailed: 'Proxy test failed', proxyTestError: 'Could not test proxy', + incognito: 'Incognito mode', incognitoHelp: 'Pause local memory, web search, and scheduled task triggers.', enableIncognito: 'Enable incognito mode', incognitoFailed: 'Could not change incognito mode', notifications: 'Send a system notification when finished', notificationsHelp: 'Notify when a response finishes or fails while the window is in the background.', notificationsFailed: 'Could not change notification settings', workspaceInstructions: 'Follow project instructions', workspaceInstructionsHelp: 'Automatically read existing AGENTS.md, CLAUDE.md, or GEMINI.md files in each project. Manage the files in their respective projects.', workspaceInstructionsFailed: 'Could not change project instruction settings', workHub: 'Enable WorkHub', workHubHelp: 'WorkHub is not available yet. This toggle is for development testing and does not enable a usable feature.', workHubFailed: 'Could not change WorkHub setting', updateFailed: 'The setting was not applied. Try again later.', defaultModel: 'Default model', defaultModelHelp: 'Model used by new tasks.', notSet: 'Not set', saveDefaultModelFailed: 'Could not save the default model', defaultPermission: 'Default permission mode', defaultPermissionHelp: 'Initial permission mode for new tasks; it can be changed at any time.', saveDefaultPermissionFailed: 'Could not save the default permission mode', defaultThinking: 'Default thinking level', defaultThinkingHelp: 'Thinking level for new tasks; models that do not offer the chosen level use their own default.', followModelDefault: 'Follow model default', saveDefaultThinkingFailed: 'Could not save the default thinking level', proxy: 'Proxy server', proxyHelp: 'Configure a network proxy for AI model requests', enableProxy: 'Enable proxy server', saveNetworkFailed: 'Could not save network settings', proxyProtocol: 'Proxy protocol', serverAddress: 'Server address', port: 'Port', proxyAuth: 'Proxy authentication', proxyAuthHelp: 'Enable this when a username and password are required.', enableProxyAuth: 'Enable proxy authentication', username: 'Username', password: 'Password', bypassList: 'Proxy bypass list', bypassHelp: 'These domains connect directly. Separate multiple domains with commas.', autoBypass: (count) => `${count} ${count === 1 ? 'domain was' : 'domains were'} added automatically. The proxy applies to AI model requests only.`, testing: 'Testing…', testCurrent: 'Test current configuration', proxyReachable: 'Proxy is reachable', proxyTestFailed: 'Proxy test failed', proxyTestError: 'Could not test proxy', shellPreference: 'Bash tool shell', shellPreferenceHelp: 'Automatic keeps the PowerShell-first Windows default. Git Bash is an explicit override for the current Runtime Host.', shellAuto: 'Automatic (recommended)', shellGitBash: 'Git Bash', shellExecutable: 'Git Bash executable', shellExecutableHelp: 'Enter the absolute path to bash.exe on the Windows machine running the Runtime Host. The legacy System32 WSL Bash shim is also recognized; Maka verifies GNU Bash before saving.', saveShell: 'Save shell setting', savingShell: 'Saving…', shellSaved: 'Saved', saveShellFailed: 'Could not save shell setting', shellExecutableRejected: 'The current Runtime Host could not run that path as GNU Bash. Check that the Host runs Windows, the path exists, and the file is named bash.exe.', }, about: { diff --git a/apps/desktop/src/renderer/workhub-controller.ts b/apps/desktop/src/renderer/workhub-controller.ts index 420e11a8af..a024b2df29 100644 --- a/apps/desktop/src/renderer/workhub-controller.ts +++ b/apps/desktop/src/renderer/workhub-controller.ts @@ -78,6 +78,10 @@ export interface WorkHubCoordinationTurn { text: string; state: WorkHubProjectedTurnState; result?: string; + assignment?: { + readonly targetSessionId: string; + readonly targetSessionName: string; + }; updatedAt: number; } @@ -99,6 +103,7 @@ export interface WorkHubProjection { export interface WorkHubSubmitInput { requestId: string; text: string; + retryAction?: true; explicitTarget?: WorkHubSessionTarget; correction?: WorkHubCorrectionContext; } @@ -807,7 +812,7 @@ function createWorkHubControllerImplementation(deps: { if (!targetSession && evidence !== 'new_session') { throw new Error('WorkHub target Session is unavailable'); } - if (targetSession?.state === 'waiting_for_user') { + if (targetSession?.state === 'waiting_for_user' && !input.retryAction) { return { kind: 'waiting', strategyId: WORKHUB_ROUTING_STRATEGY_ID, diff --git a/apps/desktop/src/renderer/workhub-coordination-port.ts b/apps/desktop/src/renderer/workhub-coordination-port.ts index e2581cd67f..05e4a1c421 100644 --- a/apps/desktop/src/renderer/workhub-coordination-port.ts +++ b/apps/desktop/src/renderer/workhub-coordination-port.ts @@ -33,12 +33,24 @@ import type { WorkHubCoordinationActInput, WorkHubCoordinationActResult, WorkHubCoordinationCandidatesResult, + OperationOutcome, + OperationError, } from '@maka/runtime-host/protocol'; import { boundedWorkHubTimelineText } from './workhub-controller.js'; import type { WorkHubDesktopTranscriptBridge } from './workhub-session-port.js'; const WORKHUB_COORDINATION_TURN_LIMIT = 40; +export class WorkHubCoordinationFailure extends Error { + constructor( + readonly code: OperationError<'workhub.coordination.act'>['code'], + message: string, + ) { + super(message); + this.name = 'WorkHubCoordinationFailure'; + } +} + export function createDesktopWorkHubCoordinationPort(deps: { sessionId: string; transcripts: WorkHubDesktopTranscriptBridge; @@ -49,13 +61,21 @@ export function createDesktopWorkHubCoordinationPort(deps: { assistantText: string; }): Promise<{ turnId: string }>; candidates(): Promise; - act(input: Omit): Promise; + act( + input: Omit, + ): Promise>; }): WorkHubCoordinationPort { return { answer: deps.answer, record: deps.record, candidates: deps.candidates, - act: deps.act, + async act(input) { + const outcome = await deps.act(input); + if (!outcome.ok) { + throw new WorkHubCoordinationFailure(outcome.error.code, outcome.error.message); + } + return outcome.result; + }, async open(handler, onError) { const store = new DesktopTranscriptRangeStore(deps.sessionId); let disposed = false; @@ -101,6 +121,20 @@ export function projectWorkHubCoordinationTurns( const latestUserIndexByTurnId = new Map(); for (const message of messages) { + if (message.type === 'workhub_coordination' && message.kind === 'delegation_assigned') { + turns.push({ + messageId: message.id, + turnId: message.coordinationTurnId, + text: boundedWorkHubTimelineText(message.userText), + state: 'completed', + assignment: { + targetSessionId: message.targetSessionId, + targetSessionName: message.targetSessionName, + }, + updatedAt: message.ts, + }); + continue; + } if (message.type === 'user') { const text = boundedWorkHubTimelineText(userFacingText(message)); if (!text) continue; diff --git a/apps/desktop/src/renderer/workhub-send-lease.ts b/apps/desktop/src/renderer/workhub-send-lease.ts new file mode 100644 index 0000000000..f8ce076624 --- /dev/null +++ b/apps/desktop/src/renderer/workhub-send-lease.ts @@ -0,0 +1,165 @@ +/* + * 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. + */ + +const WORKHUB_ACTION_KEY = 'maka-workhub-action-v1'; +const WORKHUB_DRAFT_STORAGE_KEY = 'maka-workhub-draft-v1'; +const WORKHUB_DRAFT_KEY = 'workhub'; +const MAX_DRAFT_CHARS = 120_000; +const MAX_SCOPE_CHARS = 1_024; +const SAFE_REQUEST_ID = /^[A-Za-z0-9_-]{1,128}$/u; + +type WorkHubSendLeaseStorage = Pick; + +export interface WorkHubSendLeaseOptions { + readonly scope: string; + readonly storage?: WorkHubSendLeaseStorage; + readonly createId?: () => string; +} + +export interface WorkHubSendAttempt { + readonly requestId: string; + readonly text: string; + readonly retrying: boolean; +} + +/** + * Keeps only the Host-scoped idempotency key until the Host acknowledges it. + * Composer draft text has its own storage key and lifecycle. + */ +export class WorkHubSendLease { + readonly #storage: WorkHubSendLeaseStorage | undefined; + readonly #createId: () => string; + readonly #actionKey: string; + readonly #draftKey: string; + #memoryRequestId: string | undefined; + #memoryDraft: string | undefined; + #storageHealthy = true; + + constructor(options: WorkHubSendLeaseOptions) { + if (!options.scope || options.scope.length > MAX_SCOPE_CHARS) { + throw new TypeError('WorkHub send lease requires a bounded Runtime Host scope'); + } + const scope = encodeURIComponent(options.scope); + this.#storage = options.storage ?? rendererPersistentStorage(); + this.#createId = options.createId ?? (() => crypto.randomUUID()); + this.#actionKey = `${WORKHUB_ACTION_KEY}:${scope}`; + this.#draftKey = `${WORKHUB_DRAFT_STORAGE_KEY}:${scope}`; + } + + acquire(text: string): string { + return this.acquireAttempt(text).requestId; + } + + acquireAttempt(text: string): WorkHubSendAttempt { + const existing = this.#readRequestId(); + if (existing) return { requestId: existing, text, retrying: true }; + const requestId = this.#createId(); + if (!SAFE_REQUEST_ID.test(requestId)) { + throw new Error('WorkHub action identity is invalid'); + } + this.#writeRequestId(requestId); + return { requestId, text, retrying: false }; + } + + complete(requestId: string): void { + if (this.#readRequestId() !== requestId) return; + this.#removeRequestId(); + } + + settle(requestId: string, text: string, clearsDraft: boolean): boolean { + if (!clearsDraft || this.#readRequestId() !== requestId) return false; + const draftUnchanged = this.read(WORKHUB_DRAFT_KEY) === text; + this.#removeRequestId(); + return draftUnchanged; + } + + abandon(requestId: string): void { + this.complete(requestId); + } + + read(key: string | undefined): string | undefined { + if (key !== WORKHUB_DRAFT_KEY) return undefined; + if (!this.#storageHealthy) return this.#memoryDraft; + try { + const draft = this.#storage?.getItem(this.#draftKey) ?? undefined; + if (draft === undefined || draft.length > MAX_DRAFT_CHARS) return this.#memoryDraft; + this.#memoryDraft = draft; + return draft; + } catch { + this.#storageHealthy = false; + return this.#memoryDraft; + } + } + + write(key: string | undefined, draft: string): void { + if (key !== WORKHUB_DRAFT_KEY) return; + if (draft.length > MAX_DRAFT_CHARS) return; + this.#memoryDraft = draft || undefined; + if (!this.#storageHealthy) return; + try { + if (draft) this.#storage?.setItem(this.#draftKey, draft); + else this.#storage?.removeItem(this.#draftKey); + } catch { + this.#storageHealthy = false; + } + } + + #readRequestId(): string | undefined { + if (!this.#storageHealthy) return this.#memoryRequestId; + try { + const requestId = this.#storage?.getItem(this.#actionKey) ?? undefined; + if (!requestId || !SAFE_REQUEST_ID.test(requestId)) return this.#memoryRequestId; + this.#memoryRequestId = requestId; + return requestId; + } catch { + this.#storageHealthy = false; + return this.#memoryRequestId; + } + } + + #writeRequestId(requestId: string): void { + this.#memoryRequestId = requestId; + if (!this.#storageHealthy) return; + try { + this.#storage?.setItem(this.#actionKey, requestId); + } catch { + this.#storageHealthy = false; + } + } + + #removeRequestId(): void { + this.#memoryRequestId = undefined; + if (!this.#storageHealthy) return; + try { + this.#storage?.removeItem(this.#actionKey); + } catch { + this.#storageHealthy = false; + } + } +} + +function rendererPersistentStorage(): WorkHubSendLeaseStorage | undefined { + try { + return typeof window === 'undefined' || typeof document === 'undefined' + ? undefined + : window.localStorage; + } catch { + return undefined; + } +} diff --git a/apps/desktop/src/renderer/workhub-surface.tsx b/apps/desktop/src/renderer/workhub-surface.tsx index 11cb4c028e..49ca281aea 100644 --- a/apps/desktop/src/renderer/workhub-surface.tsx +++ b/apps/desktop/src/renderer/workhub-surface.tsx @@ -35,6 +35,11 @@ import type { WorkHubSubmission, WorkHubSubmitInput, } from './workhub-controller.js'; +import { + WorkHubSendLease, + type WorkHubSendAttempt, +} from './workhub-send-lease.js'; +import { WorkHubCoordinationFailure } from './workhub-coordination-port.js'; export interface WorkHubConversationTurn { requestId: string; @@ -89,6 +94,14 @@ export function workHubSubmissionClearsDraft( } export function workHubSurfaceFailure(error: unknown): WorkHubSurfaceFailure { + if (error instanceof WorkHubCoordinationFailure) { + if (error.code === 'operation_conflict') return 'action_changed'; + if (error.code === 'not_found' || error.code === 'session_archived') { + return 'candidates_changed'; + } + if (error.code === 'session_busy') return 'target_waiting'; + return 'delivery_failed'; + } const message = error instanceof Error ? error.message : ''; if ( /candidates changed|not in the admitted candidate set|source or target is not in/iu.test( @@ -134,12 +147,70 @@ export async function submitWorkHubSurfaceInput(input: { return input.controller.submit(input.input); } +export async function submitAndRecordWorkHubSurfaceInput(input: { + controller: WorkHubController; + request: WorkHubSubmitInput; + recordedUserText: string; + summary(result: Exclude): string; + onSummaryError(): void; +}): Promise { + const result = await submitWorkHubSurfaceInput({ + controller: input.controller, + input: input.request, + }); + // Waiting is a local, retryable admission result: the request has not been + // accepted and must not consume the immutable Coordination summary owned by + // this action identity. A later same-identity retry may still be admitted. + // Delegations project directly from the Host's atomic delegation_assigned + // record. Only local clarification still needs the generic summary path. + if ( + result.kind === 'discussion' || + result.kind === 'waiting' || + result.kind === 'submitted' + ) { + return result; + } + try { + await input.controller.recordConversationTurn({ + turnId: input.request.requestId, + userText: input.recordedUserText, + assistantText: input.summary(result), + disposition: result.kind === 'clarification' ? 'clarify' : 'summary', + }); + } catch (error) { + input.onSummaryError(); + throw error; + } + return result; +} + +export async function submitLeasedWorkHubSurfaceInput(input: { + lease: WorkHubSendLease; + text: string; + preserveDraft?: boolean; + submit(attempt: WorkHubSendAttempt): Promise; +}): Promise { + const attempt = input.lease.acquireAttempt(input.text); + const result = await input.submit(attempt); + if (!result) return false; + const clearsDraft = input.lease.settle( + attempt.requestId, + attempt.text, + workHubSubmissionClearsDraft(result), + ); + if (input.preserveDraft && clearsDraft) { + return false; + } + return clearsDraft; +} + /** * The persistent Coordination Session transcript is the primary conversation. * Ordinary Sessions remain a read-only status/routing projection. */ export function WorkHubSurface(props: { controller: WorkHubController; + leaseScope: string; locale: UiLocale; initialFocusSessionId?: string; onOpenSession(sessionId: string): void; @@ -155,6 +226,7 @@ export function WorkHubSurface(props: { // a rerender can disable Composer and clarification controls. const routeGate = useRef(new WorkHubSurfaceRouteGate()).current; const refreshGate = useRef(new WorkHubProjectionRefreshGate()).current; + const sendLease = useRef(new WorkHubSendLease({ scope: props.leaseScope })).current; const [loadError, setLoadError] = useState(false); const [conversationError, setConversationError] = useState(false); const refresh = useCallback(async (focusSessionId?: string) => { @@ -224,24 +296,15 @@ export function WorkHubSurface(props: { : turn, )); try { - const result = await submitWorkHubSurfaceInput({ + const result = await submitAndRecordWorkHubSurfaceInput({ controller: props.controller, - input, + request: input, + recordedUserText, + summary: (result) => workHubCoordinationSummary(result, projection, copy), + // Clarification remains a local transcript write; delegated sends + // are projected directly from the Host-owned assignment record. + onSummaryError: () => setConversationError(true), }); - if (result.kind !== 'discussion') { - try { - await props.controller.recordConversationTurn({ - turnId: input.requestId, - userText: recordedUserText, - assistantText: workHubCoordinationSummary(result, projection, copy), - disposition: result.kind === 'clarification' ? 'clarify' : 'summary', - }); - } 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 === localRequestId ? { ...turn, state: 'settled', outcome: result } @@ -250,6 +313,9 @@ export function WorkHubSurface(props: { if (result.kind === 'submitted') await refresh(); return result; } catch (error) { + if (isTerminalWorkHubSurfaceFailure(error)) { + sendLease.abandon(input.requestId); + } setTurns((current) => current.map((turn) => turn.requestId === localRequestId ? { @@ -270,13 +336,24 @@ export function WorkHubSurface(props: { const send = useCallback(async (value: string) => { const text = value.trim(); 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); - }, [conversationReady, initialLoadSettled, route, routeGate]); + return submitLeasedWorkHubSurfaceInput({ + lease: sendLease, + text, + submit: async (attempt) => { + const { requestId } = attempt; + setTurns((current) => current.some((turn) => turn.requestId === requestId) + ? current.map((turn) => turn.requestId === requestId + ? { requestId, text: attempt.text, state: 'routing' } + : turn) + : [...current, { requestId, text: attempt.text, state: 'routing' }]); + return route({ + requestId, + text: attempt.text, + ...(attempt.retrying ? { retryAction: true as const } : {}), + }); + }, + }); + }, [conversationReady, initialLoadSettled, route, routeGate, sendLease]); const visible = visibleWorkHubConversation(coordinationTurns, turns); const visibleCoordinationTurns = visible.coordination; const visibleLocalTurns = visible.local; @@ -290,6 +367,7 @@ export function WorkHubSurface(props: { composer={( {}} sendBlocked={pending || !surfaceReady} @@ -345,14 +423,22 @@ export function WorkHubSurface(props: { const selected = projection.sessions.find( (session) => session.target.sessionId === target.sessionId, ); - void route({ - requestId: crypto.randomUUID(), + void submitLeasedWorkHubSurfaceInput({ + lease: sendLease, text: turn.text, - explicitTarget: target, - ...(turn.outcome?.kind === 'clarification' && turn.outcome.correction - ? { correction: turn.outcome.correction } - : {}), - }, turn.requestId, copy.choseWork(selected?.sessionName ?? copy.sessionFallback)); + preserveDraft: true, + submit: (attempt) => route({ + requestId: attempt.requestId, + text: attempt.text, + explicitTarget: target, + ...(attempt.retrying ? { retryAction: true as const } : {}), + ...(turn.outcome?.kind === 'clarification' && turn.outcome.correction + ? { correction: turn.outcome.correction } + : {}), + }, turn.requestId, copy.choseWork( + selected?.sessionName ?? copy.sessionFallback, + )), + }); }} onOpenSession={props.onOpenSession} /> @@ -366,6 +452,16 @@ export function WorkHubSurface(props: { ); } +function isTerminalWorkHubSurfaceFailure(error: unknown): boolean { + return ( + error instanceof WorkHubCoordinationFailure && + (error.code === 'operation_conflict' || + error.code === 'not_found' || + error.code === 'session_archived' || + error.code === 'unauthorized') + ); +} + /** Visible lifecycle state while the active Host's Coordination Session is unavailable. */ export function WorkHubCoordinationStatus(props: { locale: UiLocale; @@ -454,7 +550,11 @@ function CoordinationTurnView(props: { }) { return ( - {props.turn.result ? ( + {props.turn.assignment ? ( +

+ {props.copy.sentTo} {props.turn.assignment.targetSessionName} · {props.copy.accepted} +

+ ) : props.turn.result ? (

{props.turn.result}

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

{props.copy.answering}

diff --git a/apps/desktop/src/shared/desktop-session-projection.ts b/apps/desktop/src/shared/desktop-session-projection.ts index daf09d2b1e..f0ef38cddd 100644 --- a/apps/desktop/src/shared/desktop-session-projection.ts +++ b/apps/desktop/src/shared/desktop-session-projection.ts @@ -124,6 +124,11 @@ export function projectDesktopStoredMessage( return message.parentSessionId ? { ...message, parentSessionId: projectSessionId(host, message.parentSessionId) } : message; + case 'workhub_coordination': + return { + ...message, + targetSessionId: projectSessionId(host, message.targetSessionId), + }; default: return message; } diff --git a/docs/architecture/workhub-coordination-session-adr.md b/docs/architecture/workhub-coordination-session-adr.md index 18db78e671..eade3fbfb9 100644 --- a/docs/architecture/workhub-coordination-session-adr.md +++ b/docs/architecture/workhub-coordination-session-adr.md @@ -85,9 +85,9 @@ Every WorkHub input resolves to exactly one proposed **disposition**: All model and routing output is advisory. Before any write, a deterministic **Action Gate** admits or rejects the proposed disposition and operation. The gate enforces Runtime Host and target validity, archive and waiting state, self-route -exclusion, explicit `create_new`, expected-Turn ownership for Stop, confirmation -requirements, and existing tool and permission ceilings. Neither a model nor a -routing policy can directly authorize a write or expand execution authority. +exclusion, explicit `create_new`, and existing tool and permission ceilings. +Replacement, supersession, and Stop ownership remain deferred. Neither a model +nor a routing policy can directly authorize a write or expand execution authority. ## Delegation links rather than copies transcripts @@ -100,20 +100,43 @@ coordinationTurnId targetSessionId targetTurnId disposition -status ``` -`status` describes only whether the coordination-owned link is `active` or -`superseded`; it never mirrors the target Turn's execution lifecycle. Target -acceptance, running, waiting, completion, failure, abort, and recovery state remain -ordinary Session facts. WorkHub derives those states as read-only projections and -does not persist them as independent Coordination Session truth. +The initial assignment link does not mirror the target Turn's execution lifecycle. +Target acceptance, running, waiting, completion, failure, abort, and recovery state +remain ordinary Session facts. WorkHub derives those states as read-only +projections and does not persist them as independent Coordination Session truth. +Future replacement support may add coordination-owned `active` / `superseded` +linkage without turning target execution status into WorkHub-owned state. The ordinary Session records the delegated request, tools, side effects, and authoritative result. WorkHub may display a bounded projection or record a coordination summary, but it does not copy the ordinary Session's complete transcript into the Coordination Session. +Delegation linkage uses one closed, typed `delegation_assigned` record in the +existing Coordination Session transcript. Under the Coordination and target +Session admission authorities, one `runtime.sqlite` transaction commits that +record together with the target pending-message admission. For `create_new`, the +target Session metadata is created in the same transaction. The record carries the +exact user text, resolved target and target Message/Turn identities, creation +context, and stable display name. Its action fingerprint rejects conflicting reuse +of an action identity. + +The transaction is the user-visible assignment boundary. Before commit neither +Session observes the work; after commit both the WorkHub linkage and target input +exist. Waking or continuing the in-memory executor happens only after commit. A +Host crash between commit and wake is handled by ordinary pending-message recovery, +so WorkHub does not own a second recovery state machine or compensation chain. +The `delegation_assigned` record itself projects the visible WorkHub turn; the +renderer does not append a second summary. + +The renderer persists only a Host-scoped action id until acknowledgement. Composer +draft text uses a separate storage key and lifecycle. A reload therefore preserves +idempotency without freezing old text or coupling draft edits to Host authority. +`waiting_for_user` remains a local, retryable result because no assignment has yet +been committed. + ## Consequences, costs, and reevaluation - WorkHub gains persistent conversational continuity without adding another @@ -125,13 +148,16 @@ transcript into the Coordination Session. other Host's Sessions. - The special Session role adds provisioning, lookup, recovery, retention, and UI obligations even though it deliberately reuses the existing Session substrate. +- Every delegated Coordination turn adds one typed assignment record, which is + also its visible timeline source. - Whether Work is 1:1 with Session, 1:N over Sessions, or an independent durable entity remains unresolved. - Cross-Runtime-Host coordination remains deferred. - Coordination Session role representation, lazy creation, durable lookup, - recovery, and per-Host UI resolution are implemented. Coordination transcript, - disposition, delegation-link, and Action Gate behavior remain later work; this - ADR defines their authority boundaries without implementing them. + recovery, per-Host UI resolution, persistent transcript, closed dispositions, + and the Action Gate are implemented. Durable delegation linkage is encoded in + that transcript; target lifecycle projection, linked correction, and destructive + replacement/Stop recovery remain later work. Reevaluate the per-Host decision if supported workflows require one WorkHub conversation to coordinate ordinary Sessions on multiple Runtime Hosts, or if Host diff --git a/packages/core/src/__tests__/workhub-coordination-record.test.ts b/packages/core/src/__tests__/workhub-coordination-record.test.ts new file mode 100644 index 0000000000..9856a12d54 --- /dev/null +++ b/packages/core/src/__tests__/workhub-coordination-record.test.ts @@ -0,0 +1,118 @@ +/* + * 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 { describe, test } from 'node:test'; +import { decodeCanonicalMessage } from '../session.js'; + +const FINGERPRINT = `sha256:${'a'.repeat(64)}`; + +describe('WorkHub Coordination stored records', () => { + test('decodes one exact atomic delegation assignment', () => { + const assigned = { + type: 'workhub_coordination', + id: 'assignment-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId: 'action-id', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + userText: 'Continue payment work', + delegationId: 'delegation-id', + targetTurnId: 'target-turn', + targetMessageId: 'target-message', + targetSessionName: 'Payments', + steered: true, + } as const; + + assert.deepEqual(decodeCanonicalMessage(assigned), assigned); + }); + + test('rejects malformed or widened coordination records', () => { + const base = { + type: 'workhub_coordination', + id: 'assignment-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId: 'action-id', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + userText: 'Continue payment work', + delegationId: 'delegation-id', + targetTurnId: 'target-turn', + targetMessageId: 'target-message', + targetSessionName: 'Payments', + } as const; + + for (const invalid of [ + { ...base, coordinationTurnId: 'different-turn' }, + { ...base, actionFingerprint: 'not-a-digest' }, + { ...base, disposition: 'replace' }, + { ...base, targetMessageId: undefined }, + { ...base, sourceSessionId: 'injected' }, + { ...base, kind: 'delegation_intent' }, + { ...base, schemaVersion: 2 }, + ]) { + assert.throws(() => decodeCanonicalMessage(invalid), /Invalid stored message schema/u); + } + }); + + test('requires a complete create payload only for create_new assignments', () => { + const create = { + type: 'workhub_coordination', + id: 'create-assignment-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId: 'action-id', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'created-session', + disposition: 'create_new', + userText: 'Create a login audit', + create: { + title: 'Login audit', + workspace: { kind: 'project', projectId: 'project-maka' }, + }, + delegationId: 'delegation-id', + targetTurnId: 'target-turn', + targetMessageId: 'target-message', + targetSessionName: 'Login audit', + } as const; + + assert.deepEqual(decodeCanonicalMessage(create), create); + assert.throws( + () => decodeCanonicalMessage({ ...create, create: undefined }), + /Invalid stored message schema/u, + ); + assert.throws( + () => decodeCanonicalMessage({ ...create, disposition: 'delegate_existing' }), + /Invalid stored message schema/u, + ); + }); +}); diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index e9896dc2e2..61f1e4a6e3 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -756,6 +756,7 @@ export type StoredMessage = | PermissionDecisionMessage | TokenUsageMessage | TurnStateMessage + | WorkHubCoordinationMessage | SystemNoteMessage; export interface UserMessage extends MessageContent { @@ -908,6 +909,51 @@ export interface TurnStateMessage { partialOutputRetained: boolean; } +export const WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION = 1 as const; + +export type WorkHubDelegationDisposition = 'delegate_existing' | 'create_new'; + +export type WorkHubDelegationWorkspace = + | { readonly kind: 'project'; readonly projectId: string } + | { readonly kind: 'host_path'; readonly path: string }; + +export interface WorkHubDelegationCreateSpec { + readonly title: string; + readonly workspace: WorkHubDelegationWorkspace; +} + +interface WorkHubCoordinationMessageEnvelope { + type: 'workhub_coordination'; + id: string; + /** The Coordination Turn that owns this action. */ + turnId: string; + ts: number; + schemaVersion: typeof WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION; + actionId: string; + actionFingerprint: `sha256:${string}`; + coordinationTurnId: string; + targetSessionId: string; + disposition: WorkHubDelegationDisposition; + /** Exact target payload; retained so retry does not depend on renderer memory. */ + userText: string; + /** Present exactly for create_new. */ + create?: WorkHubDelegationCreateSpec; +} + +/** + * Atomic proof that one Coordination action and one target Message admission + * were committed together by the Runtime Host. + */ +export interface WorkHubDelegationAssignedMessage extends WorkHubCoordinationMessageEnvelope { + kind: 'delegation_assigned'; + delegationId: string; + targetTurnId: string; + targetMessageId: string; + targetSessionName: string; + steered?: true; +} +export type WorkHubCoordinationMessage = WorkHubDelegationAssignedMessage; + export interface TurnRecord { turnId: string; firstSequence?: number; @@ -1030,6 +1076,38 @@ const TURN_STATE_MESSAGE_SHAPE = defineObjectShape()( 'errorClass', ], ); +const WORKHUB_DELEGATION_ASSIGNED_MESSAGE_SHAPE = + defineObjectShape()( + [ + 'type', + 'id', + 'turnId', + 'ts', + 'schemaVersion', + 'kind', + 'actionId', + 'actionFingerprint', + 'coordinationTurnId', + 'targetSessionId', + 'disposition', + 'userText', + 'delegationId', + 'targetTurnId', + 'targetMessageId', + 'targetSessionName', + ], + ['create', 'steered'], + ); +const WORKHUB_DELEGATION_CREATE_SHAPE = defineObjectShape()( + ['title', 'workspace'], + [], +); +const WORKHUB_DELEGATION_PROJECT_WORKSPACE_SHAPE = defineObjectShape< + Extract +>()(['kind', 'projectId'], []); +const WORKHUB_DELEGATION_HOST_PATH_WORKSPACE_SHAPE = defineObjectShape< + Extract +>()(['kind', 'path'], []); const SYSTEM_NOTE_MESSAGE_SHAPE = defineObjectShape()( ['type', 'id', 'ts', 'kind'], ['turnId', 'data'], @@ -1177,6 +1255,11 @@ function decodeMessage( ) return message as unknown as TurnStateMessage; break; + case 'workhub_coordination': + if (isWorkHubCoordinationMessage(message)) { + return message as unknown as WorkHubCoordinationMessage; + } + break; case 'system_note': if ( hasExactShape(message, SYSTEM_NOTE_MESSAGE_SHAPE) && @@ -1190,6 +1273,59 @@ function decodeMessage( throw new Error('Invalid stored message schema'); } +function isWorkHubCoordinationMessage(message: Record): boolean { + const common = + hasMessageEnvelope(message, true) && + message.schemaVersion === WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION && + typeof message.actionId === 'string' && + typeof message.actionFingerprint === 'string' && + /^sha256:[a-f0-9]{64}$/u.test(message.actionFingerprint) && + typeof message.coordinationTurnId === 'string' && + message.turnId === message.coordinationTurnId && + typeof message.targetSessionId === 'string' && + typeof message.userText === 'string' && + message.userText.trim().length > 0 && + ((message.disposition === 'delegate_existing' && message.create === undefined) || + (message.disposition === 'create_new' && isWorkHubDelegationCreateSpec(message.create))) && + (message.disposition === 'delegate_existing' || message.disposition === 'create_new'); + if (!common) return false; + return ( + message.kind === 'delegation_assigned' && + hasExactShape(message, WORKHUB_DELEGATION_ASSIGNED_MESSAGE_SHAPE) && + typeof message.delegationId === 'string' && + typeof message.targetTurnId === 'string' && + typeof message.targetMessageId === 'string' && + typeof message.targetSessionName === 'string' && + message.targetSessionName.trim().length > 0 && + (message.steered === undefined || message.steered === true) + ); +} + +function isWorkHubDelegationCreateSpec(value: unknown): value is WorkHubDelegationCreateSpec { + if ( + !isRecord(value) || + !hasExactShape(value, WORKHUB_DELEGATION_CREATE_SHAPE) || + typeof value.title !== 'string' || + value.title.trim().length === 0 || + !isRecord(value.workspace) + ) { + return false; + } + if (value.workspace.kind === 'project') { + return ( + hasExactShape(value.workspace, WORKHUB_DELEGATION_PROJECT_WORKSPACE_SHAPE) && + typeof value.workspace.projectId === 'string' && + value.workspace.projectId.length > 0 + ); + } + return ( + value.workspace.kind === 'host_path' && + hasExactShape(value.workspace, WORKHUB_DELEGATION_HOST_PATH_WORKSPACE_SHAPE) && + typeof value.workspace.path === 'string' && + value.workspace.path.length > 0 + ); +} + function decodeStoredMessageContent( value: unknown, decodeToolResultContent: (content: unknown) => ToolResultContent, diff --git a/packages/core/src/thread-search.ts b/packages/core/src/thread-search.ts index 881124d3e5..7e6086e8d5 100644 --- a/packages/core/src/thread-search.ts +++ b/packages/core/src/thread-search.ts @@ -465,6 +465,7 @@ export function threadSearchMatchKind(message: StoredMessage): ThreadSearchMatch case 'permission_decision': case 'token_usage': case 'turn_state': + case 'workhub_coordination': case 'system_note': throw new Error(`Message type ${message.type} is not searchable`); } @@ -488,6 +489,8 @@ export function formatSearchResultSummary(message: StoredMessage): string { return '用量记录'; case 'turn_state': return '回合状态'; + case 'workhub_coordination': + return 'WorkHub 协调记录'; case 'system_note': return '系统记录'; } @@ -543,8 +546,11 @@ export function collectSearchableText(message: StoredMessage): string | undefine case 'permission_decision': case 'token_usage': case 'turn_state': + case 'workhub_coordination': case 'system_note': - // Excluded — not user-typed / not user-visible content. + // Coordination records are rendered by WorkHub, but the reserved + // Coordination Session is intentionally outside general thread search. + // The remaining cases are not user-typed / not user-visible content. return undefined; } } diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 6ff54ecd0d..48f24ad74a 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -18,14 +18,23 @@ */ import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { test } from 'node:test'; import { messageContentDigest, type MessageContent } from '@maka/core/events'; import type { RuntimeEvent } from '@maka/core/runtime-event'; +import { + WORKHUB_COORDINATION_SESSION_ID, + WORKHUB_COORDINATION_SESSION_ROLE, +} from '@maka/core/session'; import type { MessageAdmissionStore, PendingMessageAdmission, RootTurnSourceMessageReceipt, } from '@maka/storage/execution-stores'; +import { createSessionStore } from '@maka/storage/session-store'; import { MESSAGE_OPERATION_RESULT_MAX_BYTES, MESSAGE_QUEUE_PROJECTION_MAX_BYTES, @@ -44,6 +53,83 @@ import { SessionAdmissionGate } from '../server/session-admission-gate.js'; const ROOT = { sessionId: 'session-1', turnId: 'turn-1', runId: 'run-1' } as const; +test('consumes an atomically committed active-target admission exactly once', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-active-consume-')); + const store = createSessionStore(root); + t.after(async () => { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + }); + await store.createStableSession({ + sessionId: WORKHUB_COORDINATION_SESSION_ID, + requestFingerprint: `sha256:${'a'.repeat(64)}`, + input: { + cwd: root, + name: 'WorkHub', + role: WORKHUB_COORDINATION_SESSION_ROLE, + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'explore', + toolProfile: 'workhub-coordination-v1', + }, + }); + await store.createStableSession({ + sessionId: ROOT.sessionId, + requestFingerprint: `sha256:${'b'.repeat(64)}`, + input: { + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }, + }); + const fixture = createFixture(undefined, () => true, store); + fixture.coordinator.reserveRootTurn(ROOT); + fixture.coordinator.bindRun(ROOT); + const content = { text: 'atomic WorkHub assignment' }; + const actionId = 'action-active-target'; + const suffix = createHash('sha256').update(actionId, 'utf8').digest('hex').slice(0, 48); + const messageId = `whm_${suffix}`; + await store.assignWorkHubMessage({ + assignment: { + type: 'workhub_coordination', + id: `wha_${suffix}`, + turnId: actionId, + ts: 10, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId, + actionFingerprint: `sha256:${'c'.repeat(64)}`, + coordinationTurnId: actionId, + targetSessionId: ROOT.sessionId, + targetSessionName: 'Payments', + targetTurnId: ROOT.turnId, + targetMessageId: messageId, + delegationId: `whd_${suffix}`, + disposition: 'delegate_existing', + userText: content.text, + steered: true, + }, + admission: { + ...ROOT, + messageId, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 10, + }, + }); + + await fixture.coordinator.consumePendingAdmissions([ROOT.sessionId]); + await fixture.coordinator.consumePendingAdmissions([ROOT.sessionId]); + + assert.equal(fixture.coordinator.projection(ROOT.sessionId).steering.length, 1); + assert.equal(fixture.drainRequests(), 0); +}); + test('idle submit starts exactly one root Turn and retry identity is connection-independent', async () => { const fixture = createFixture(); fixture.setRootState({ kind: 'idle' }); @@ -2250,6 +2336,7 @@ test('canonical retry omits redundant display text and empty ordered refs', asyn function createFixture( onProjectionChanged?: (sessionId: string) => void, preflightSessionSnapshot: HostMessageCoordinatorOptions['preflightSessionSnapshot'] = () => true, + admissionsOverride?: MessageAdmissionStore, ) { let nextId = 1; let liveResidencies = 0; @@ -2277,7 +2364,8 @@ function createFixture( state: 'accepted' | 'handed_off' | 'executed' | 'cancelled'; } >(); - const admissions = memoryMessageAdmissionStore(messageAdmissions); + const admissions = admissionsOverride ?? memoryMessageAdmissionStore(messageAdmissions); + const sessionAdmission = new SessionAdmissionGate(); const stopClaimed = deferred(); const terminal = deferred(); let coordinator: HostMessageCoordinator; @@ -2384,7 +2472,7 @@ function createFixture( }, }, admissions, - sessionAdmission: new SessionAdmissionGate(), + sessionAdmission, acquireResidency: () => { liveResidencies += 1; let released = false; @@ -2407,6 +2495,7 @@ function createFixture( return { coordinator, admissions, + sessionAdmission, setRootState: (state: HostMessageRootState) => { rootState = state; }, diff --git a/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts index 9d36dae540..095bea5ffd 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts @@ -25,6 +25,7 @@ import { WorkHubCoordinationActionGate, type WorkHubActionGateEffects, type WorkHubActionGateSession, + type WorkHubDelegationAssignmentInput, } from '../server/workhub-coordination-action-gate.js'; import type { ConnectionContext } from '../server/operation-dispatcher.js'; @@ -36,7 +37,7 @@ const CONTEXT: ConnectionContext = { }; describe('WorkHub Coordination Action Gate', () => { - test('exposes only bounded ordinary candidates and proposals use opaque refs', async () => { + test('exposes only bounded ordinary candidates and opaque refs', async () => { const effects = fakeEffects([ session('ordinary'), session('archived', { isArchived: true }), @@ -53,7 +54,6 @@ describe('WorkHub Coordination Action Gate', () => { session('maka_workhub_coordination', { role: 'workhub_coordination' }), ]); const result = await new WorkHubCoordinationActionGate(effects).candidates(); - assert.deepEqual( result.candidates.map(({ sessionId }) => sessionId), ['ordinary', 'waiting'], @@ -67,16 +67,15 @@ describe('WorkHub Coordination Action Gate', () => { assert.equal(bounded.candidates.length, 32); }); - test('rejects stale and invented candidates before any Session effect', async () => { + test('rejects stale candidates before assignment', async () => { const effects = fakeEffects([session('payments')]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); - effects.sessions[0] = session('payments', { statusUpdatedAt: 9 }); - + effects.sessions[0] = session('payments', { lastMessageAt: 9 }); await assert.rejects( gate.act( { - actionId: 'stale-action', + actionId: 'stale', userText: 'Continue payments', candidateSetId: snapshot.candidateSetId, proposal: { @@ -88,12 +87,12 @@ describe('WorkHub Coordination Action Gate', () => { ), (error) => error instanceof WorkHubActionGateFailure && error.code === 'candidate_set_stale', ); - assert.deepEqual(effects.submissions, []); + assert.equal(effects.assignments.length, 0); const refreshed = await gate.candidates(); const retried = await gate.act( { - actionId: 'stale-action', + actionId: 'stale', userText: 'Continue payments', candidateSetId: refreshed.candidateSetId, proposal: { @@ -104,13 +103,12 @@ describe('WorkHub Coordination Action Gate', () => { CONTEXT, ); assert.equal(retried.disposition, 'delegate_existing'); - effects.submissions.length = 0; const current = await gate.candidates(); await assert.rejects( gate.act( { - actionId: 'invented-action', + actionId: 'invented', userText: 'Continue payments', candidateSetId: current.candidateSetId, proposal: { disposition: 'delegate_existing', candidateRef: 'invented_candidate' }, @@ -120,19 +118,18 @@ describe('WorkHub Coordination Action Gate', () => { (error) => error instanceof WorkHubActionGateFailure && error.code === 'candidate_unavailable', ); - assert.deepEqual(effects.submissions, []); + assert.equal(effects.assignments.length, 1); }); test('rejects waiting targets independently of strategy behavior', async () => { const effects = fakeEffects([session('waiting', { status: 'waiting_for_user' })]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); - await assert.rejects( gate.act( { - actionId: 'waiting-action', - userText: 'Do another thing', + actionId: 'waiting', + userText: 'Continue', candidateSetId: snapshot.candidateSetId, proposal: { disposition: 'delegate_existing', @@ -144,40 +141,55 @@ describe('WorkHub Coordination Action Gate', () => { (error) => error instanceof WorkHubActionGateFailure && error.code === 'target_waiting_for_user', ); - assert.deepEqual(effects.submissions, []); }); - test('answers and clarifies only through the Coordination transcript effects', async () => { + test('answers and clarifies only through Coordination effects', async () => { const effects = fakeEffects([session('ordinary')]); const gate = new WorkHubCoordinationActionGate(effects); - - const answered = await gate.act( - { - actionId: 'answer-action', - userText: 'What is useMemo?', - proposal: { disposition: 'answer_here' }, - }, + await gate.act( + { actionId: 'answer', userText: 'Summarize', proposal: { disposition: 'answer_here' } }, CONTEXT, ); - const clarified = await gate.act( + await gate.act( { - actionId: 'clarify-action', - userText: 'Continue that task', - proposal: { disposition: 'clarify', assistantText: 'Which task do you mean?' }, + actionId: 'clarify', + userText: 'Which one?', + proposal: { disposition: 'clarify', assistantText: 'Choose a Session' }, }, CONTEXT, ); - - assert.equal(answered.disposition, 'answer_here'); - assert.equal(clarified.disposition, 'clarify'); assert.equal(effects.answers.length, 1); assert.equal(effects.clarifications.length, 1); - assert.deepEqual(effects.submissions, []); - assert.equal(effects.creations.length, 0); + assert.equal(effects.assignments.length, 0); }); - test('only create_new creates and retries the exact action idempotently', async () => { - const effects = fakeEffects([session('ordinary')]); + test('delegates through one assignment effect', async () => { + const effects = fakeEffects([session('payments', { name: 'Payments' })]); + const gate = new WorkHubCoordinationActionGate(effects); + const snapshot = await gate.candidates(); + const result = await gate.act( + { + actionId: 'delegate', + userText: 'Continue payments', + candidateSetId: snapshot.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: snapshot.candidates[0]!.candidateRef, + }, + }, + CONTEXT, + ); + assert.deepEqual(result, { + disposition: 'delegate_existing', + targetSessionId: 'payments', + targetTurnId: 'turn-delegate', + }); + assert.equal(effects.assignments[0]!.targetSessionName, 'Payments'); + assert.equal(effects.assignments[0]!.userText, 'Continue payments'); + }); + + test('create_new carries creation context into the same assignment', async () => { + const effects = fakeEffects([]); const gate = new WorkHubCoordinationActionGate(effects); await assert.rejects( gate.act( @@ -190,72 +202,97 @@ describe('WorkHub Coordination Action Gate', () => { ), (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', ); - assert.equal(effects.creations.length, 0); const input = { - actionId: 'create-action', + actionId: 'create', userText: 'Create an accessibility audit', proposal: { disposition: 'create_new' as const, title: 'Accessibility audit' }, - create: { - workspace: { kind: 'host_path' as const, path: '/workspace' }, - }, + create: { workspace: { kind: 'host_path' as const, path: '/workspace' } }, }; - const first = await gate.act(input, CONTEXT); const replay = await gate.act(input, CONTEXT); - assert.deepEqual(replay, first); - assert.equal(effects.creations.length, 1); - assert.equal(effects.submissions.length, 1); - assert.match(effects.creations[0]?.sessionId ?? '', /^whs_[a-f0-9]{48}$/u); - assert.equal(first.disposition, 'create_new'); - if (first.disposition === 'create_new') { - assert.equal(first.targetSessionId, effects.creations[0]?.sessionId); - } - assert.deepEqual(Object.keys(effects.creations[0]!).sort(), [ - 'sessionId', - 'title', - 'workspace', - ]); + assert.equal(effects.assignments.length, 1); + const restartedReplay = await new WorkHubCoordinationActionGate(effects).act(input, CONTEXT); + assert.deepEqual(restartedReplay, first); + assert.equal(effects.assignments.length, 2); + assert.deepEqual(effects.assignments[0], effects.assignments[1]); + assert.match(effects.assignments[0]!.targetSessionId, /^whs_[a-f0-9]{48}$/u); + assert.deepEqual(effects.assignments[0]!.create, { + title: 'Accessibility audit', + workspace: input.create.workspace, + }); await assert.rejects( gate.act({ ...input, proposal: { disposition: 'create_new', title: 'Different' } }, CONTEXT), (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', ); - assert.equal(effects.creations.length, 1); + assert.equal(effects.assignments.length, 2); }); - test('effect rejection grants no root ownership and releases the action identity', async () => { - const effects = fakeEffects([session('ordinary')]); + test('one in-memory action identity cannot change payload', async () => { + const effects = fakeEffects([session('payments'), session('login', { lastMessageAt: 1 })]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); - const submit = effects.submit; - effects.submit = async () => { - throw new WorkHubActionEffectFailure('unauthorized', 'Target permission denied'); - }; const input = { - actionId: 'permission-rejected-action', - userText: 'Continue ordinary work', + actionId: 'same-action', + userText: 'Continue payments', candidateSetId: snapshot.candidateSetId, proposal: { disposition: 'delegate_existing' as const, candidateRef: snapshot.candidates[0]!.candidateRef, }, }; + await gate.act(input, CONTEXT); + await assert.rejects( + gate.act({ ...input, userText: 'Different work' }, CONTEXT), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + await assert.rejects( + gate.act( + { + ...input, + proposal: { + disposition: 'delegate_existing', + candidateRef: snapshot.candidates[1]!.candidateRef, + }, + }, + CONTEXT, + ), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + }); + test('an assignment rejection releases the action identity for retry', async () => { + const effects = fakeEffects([session('payments')]); + const gate = new WorkHubCoordinationActionGate(effects); + const snapshot = await gate.candidates(); + const input = { + actionId: 'permission-rejected', + userText: 'Continue payments', + candidateSetId: snapshot.candidateSetId, + proposal: { + disposition: 'delegate_existing' as const, + candidateRef: snapshot.candidates[0]!.candidateRef, + }, + }; + const assign = effects.assign; + effects.assign = async () => { + throw new WorkHubActionEffectFailure('unauthorized', 'Target permission denied'); + }; await assert.rejects( gate.act(input, CONTEXT), (error) => error instanceof WorkHubActionEffectFailure && error.code === 'unauthorized', ); - effects.submit = submit; + effects.assign = assign; assert.equal((await gate.act(input, CONTEXT)).disposition, 'delegate_existing'); }); - test('replays an ordinary delegation without submitting twice', async () => { - const effects = fakeEffects([session('ordinary')]); + test('replays an ordinary delegation without assigning twice', async () => { + const effects = fakeEffects([session('payments')]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); const input = { - actionId: 'delegate-action', - userText: 'Continue ordinary work', + actionId: 'delegate-replay', + userText: 'Continue payments', candidateSetId: snapshot.candidateSetId, proposal: { disposition: 'delegate_existing' as const, @@ -267,8 +304,7 @@ describe('WorkHub Coordination Action Gate', () => { const replay = await gate.act(input, CONTEXT); assert.deepEqual(replay, first); - assert.equal(effects.submissions.length, 1); - assert.equal(effects.submissions[0]?.sessionId, 'ordinary'); + assert.equal(effects.assignments.length, 1); }); }); @@ -278,19 +314,25 @@ function session( ): WorkHubActionGateSession { return { id, - cwd: `/workspace/${id}`, + cwd: '/workspace', projectId: null, createdAt: 1, + lastMessageAt: 2, name: id, labels: [], isArchived: false, status: 'active', + statusUpdatedAt: 2, ...patch, }; } function fakeEffects(initialSessions: WorkHubActionGateSession[]) { - const state = { + const durable = new Map< + string, + { input: WorkHubDelegationAssignmentInput; result: { turnId: string } } + >(); + return { sessions: [...initialSessions], answers: [] as Array<{ turnId: string; text: string }>, clarifications: [] as Array<{ @@ -298,42 +340,34 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { userText: string; assistantText: string; }>, - creations: [] as Array<{ - sessionId: string; - workspace: { kind: 'project'; projectId: string } | { kind: 'host_path'; path: string }; - title: string; - }>, - submissions: [] as Array<{ sessionId: string; messageId: string; text: string }>, + assignments: [] as WorkHubDelegationAssignmentInput[], async listSessions() { return this.sessions; }, + async readAssignment() { + return undefined; + }, async answer(input: { turnId: string; text: string }) { this.answers.push(input); }, async clarify(input: { turnId: string; userText: string; assistantText: string }) { this.clarifications.push(input); }, - async create(input: { - sessionId: string; - workspace: { kind: 'project'; projectId: string } | { kind: 'host_path'; path: string }; - title: string; - }) { - this.creations.push(input); - }, - async submit(input: { sessionId: string; messageId: string; text: string }) { - this.submissions.push(input); - return { turnId: `turn-${input.sessionId}` }; + async assign(input: WorkHubDelegationAssignmentInput) { + this.assignments.push(input); + const existing = durable.get(input.actionId); + if (existing) { + assert.deepEqual(existing.input, input); + return existing.result; + } + const result = { turnId: `turn-${input.actionId}` }; + durable.set(input.actionId, { input, result }); + return result; }, } satisfies WorkHubActionGateEffects & { sessions: WorkHubActionGateSession[]; answers: Array<{ turnId: string; text: string }>; clarifications: Array<{ turnId: string; userText: string; assistantText: string }>; - creations: Array<{ - sessionId: string; - workspace: { kind: 'project'; projectId: string } | { kind: 'host_path'; path: string }; - title: string; - }>; - submissions: Array<{ sessionId: string; messageId: string; text: string }>; + assignments: WorkHubDelegationAssignmentInput[]; }; - return state; } 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 d78d325d19..bde8ce31fe 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts @@ -18,12 +18,17 @@ */ import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; import { mkdtemp, rm, stat } from 'node:fs/promises'; 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 { + messageContentDigest, + normalizeMessageContent, + type MessageContent, +} from '@maka/core/events'; import { WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_SESSION_ROLE, @@ -38,6 +43,7 @@ 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 type { WorkHubActionGateEffects } from '../server/workhub-coordination-action-gate.js'; import { HostWorkHubCoordinationCoordinator, type CoordinationCreateTarget, @@ -480,6 +486,97 @@ describe('Host WorkHub Coordination coordinator', () => { } }); + test('persists delegated action ownership and replays it after Host restart', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-delegation-')); + const userText = 'Continue payment work. '.repeat(900); + let store = createSessionStore(root); + try { + await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + const assignments: string[] = []; + const first = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + assign: async (input) => { + assignments.push(input.actionId); + return persistTestAssignment(store, input, 'payments-turn'); + }, + }); + assert.equal((await first.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); + const candidates = await first.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const input = { + actionId: 'payments-action', + userText, + candidateSetId: candidates.result.candidateSetId, + proposal: { + disposition: 'delegate_existing' as const, + candidateRef: candidates.result.candidates[0]!.candidateRef, + }, + }; + const admitted = await first.handlers['workhub.coordination.act'](input, CONTEXT); + assert.deepEqual(admitted, { + ok: true, + result: { + disposition: 'delegate_existing', + targetSessionId: candidates.result.candidates[0]!.sessionId, + targetTurnId: 'payments-turn', + }, + }); + assert.deepEqual( + (await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID)) + .filter((message) => message.type === 'workhub_coordination') + .map(({ kind, actionId, targetSessionId }) => ({ kind, actionId, targetSessionId })), + [ + { + kind: 'delegation_assigned', + actionId: 'payments-action', + targetSessionId: candidates.result.candidates[0]!.sessionId, + }, + ], + ); + assert.equal(assignments.length, 1); + } finally { + await store.close?.(); + } + + store = createSessionStore(root); + try { + const restarted = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + assign: (input) => persistTestAssignment(store, input, 'payments-turn'), + }); + const candidates = await restarted.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const replayed = await restarted.handlers['workhub.coordination.act']( + { + actionId: 'payments-action', + userText, + candidateSetId: candidates.result.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: candidates.result.candidates[0]!.candidateRef, + }, + }, + CONTEXT, + ); + assert.equal(replayed.ok, true); + if (replayed.ok) { + assert.equal(replayed.result.disposition, 'delegate_existing'); + if (replayed.result.disposition === 'delegate_existing') { + assert.equal(replayed.result.targetTurnId, 'payments-turn'); + } + } + } 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); @@ -614,6 +711,9 @@ function coordinator( hasRootTurnAdmission: async () => false, }, admission: SessionAdmissionGate = new SessionAdmissionGate(), + sessionActions: Pick = { + assign: async ({ targetSessionId }) => ({ turnId: `turn-${targetSessionId}` }), + }, ) { return new HostWorkHubCoordinationCoordinator({ stateRoot: root, @@ -621,10 +721,7 @@ function coordinator( admission, continuity: { refreshCanonical: async () => undefined }, executions, - sessionActions: { - create: async () => undefined, - submit: async ({ sessionId }) => ({ turnId: `turn-${sessionId}` }), - }, + sessionActions, resolveCreateTarget: resolveCreateTarget ?? (async () => ({ @@ -637,3 +734,46 @@ function coordinator( requestDrain, }); } + +async function persistTestAssignment( + store: SessionAuthorityStore, + input: Parameters[0], + targetTurnId: string, +): Promise<{ turnId: string }> { + const suffix = createHash('sha256').update(input.actionId, 'utf8').digest('hex').slice(0, 48); + const content = normalizeMessageContent({ text: input.userText }); + const result = await store.assignWorkHubMessage({ + assignment: { + type: 'workhub_coordination', + id: `wha_${suffix}`, + turnId: input.actionId, + ts: Date.now(), + schemaVersion: 1, + kind: 'delegation_assigned', + actionId: input.actionId, + actionFingerprint: input.actionFingerprint, + coordinationTurnId: input.actionId, + targetSessionId: input.targetSessionId, + targetSessionName: input.targetSessionName, + targetTurnId, + targetMessageId: `whm_${suffix}`, + delegationId: `whd_${suffix}`, + disposition: input.disposition, + userText: input.userText, + ...(input.create ? { create: input.create } : {}), + }, + admission: { + sessionId: input.targetSessionId, + turnId: targetTurnId, + runId: `whr_${suffix}`, + messageId: `whm_${suffix}`, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: Date.now(), + }, + }); + return { turnId: result.assignment.targetTurnId }; +} diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index a02f619545..9999886345 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -93,7 +93,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 = 59 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 60 as const; +// 60: WorkHub stores a canonical delegation assignment record. Older peers +// cannot decode this message during transcript recovery. // 59: Scheduled Turn provider-retry frames may carry an optional host-clock // `ts`, letting a mid-wait re-projection recompute the authoritative // remaining duration. Older peers decode the frame with an exact key list @@ -111,7 +113,7 @@ export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 59 as const; // principal, closing pairing-finalize races that credential-by-ID revocation cannot. // 54: Client-bound pairing candidates restrict pre-claim authority and bind // their durable credential to the claiming Client identity; it is also reserved -// by concurrent protocol changes in #3390 and #3935. +// by concurrent protocol changes in #3390. // 53: Message admission answers `turn.message.submit` with an explicit // disposition, and queued Messages can be proven cancelled. Older peers read the // answer as a bare acknowledgement and cannot reconcile their own projection. diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 403e7af4c8..46619ff9d8 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -18,7 +18,7 @@ */ import { createHash, randomUUID } from 'node:crypto'; -import { normalizeMessageContent } from '@maka/core/events'; +import { messageContentDigest, normalizeMessageContent } from '@maka/core/events'; import { describeChatConfigurationReason, NO_REAL_CONNECTION_CODE, @@ -27,7 +27,7 @@ import type { RuntimeExecutionConnection } from '@maka/core/llm-connections'; import { generalizedErrorMessage } from '@maka/core/redaction'; import { emptyPlanSessionState } from '@maka/core/plan'; import type { PermissionMode } from '@maka/core/permission'; -import { isDeepResearchSession } from '@maka/core/session'; +import { isDeepResearchSession, WORKHUB_COORDINATION_SESSION_ID } from '@maka/core/session'; import { filterModelVisibleTaskLedgerTasks } from '@maka/core/task-ledger'; import { AgentGraphCoordinator } from '@maka/runtime/stream-graph-coordinator'; import { AgentGraphSupervisorWakeCoordinator } from '@maka/runtime/agent-graph-supervisor-wake'; @@ -1230,44 +1230,100 @@ export async function createExecutionRuntimeHostComposition( continuity: continuityCoordinator, executions: coordinator, sessionActions: { - create: async (input) => { - const outcome = await sessionCatalog.createForWorkHub({ - sessionId: input.sessionId, - workspace: input.workspace, - name: input.title, - modelTarget: { kind: 'default' }, - collaborationMode: 'agent', - orchestrationMode: 'default', - }); - if (!outcome.ok) { - throw new WorkHubActionEffectFailure( - outcome.error.code === 'invalid_request' ? 'operation_conflict' : outcome.error.code, - outcome.error.message, - ); - } - }, - submit: async (input, connection) => { - const outcome = await messages.handlers['turn.message.submit']( - { - originHostEpoch: connection.hostEpoch, - sessionId: input.sessionId, - messageId: input.messageId, - content: normalizeMessageContent({ text: input.text }), - placement: 'current_turn', - }, - connection, - ); - if (!outcome.ok) { - throw new WorkHubActionEffectFailure( - outcome.error.code === 'outcome_unknown' - ? 'commit_outcome_unknown' - : outcome.error.code, - outcome.error.message, - ); - } - return outcome.result.disposition === 'turn_started' - ? { turnId: outcome.result.turnId } - : { turnId: input.messageId, steered: true as const }; + assign: async (input) => { + const durable = await stores.sessionStore.readWorkHubAssignment(input.actionId); + const create = + !durable && input.create + ? await sessionCatalog.prepareWorkHubCreate({ + sessionId: input.targetSessionId, + workspace: input.create.workspace, + name: input.create.title, + modelTarget: { kind: 'default' }, + collaborationMode: 'agent', + orchestrationMode: 'default', + }) + : undefined; + const suffix = createHash('sha256') + .update(input.actionId, 'utf8') + .digest('hex') + .slice(0, 48); + const messageId = `whm_${suffix}`; + const content = normalizeMessageContent({ text: input.userText }); + const persisted = + durable ?? + (await sessionAdmission.runMany( + [WORKHUB_COORDINATION_SESSION_ID, input.targetSessionId], + async (lease) => { + const rootState = coordinator.readRootState(input.targetSessionId); + if (!create && rootState.kind === 'reserved') { + throw new WorkHubActionEffectFailure( + 'session_busy', + 'A target root Turn is being admitted', + ); + } + const steered = rootState.kind === 'active'; + const turnId = steered ? rootState.turnId : `wht_${suffix}`; + const runId = steered ? rootState.runId : `whr_${suffix}`; + const assignedAt = Date.now(); + const result = await stores.sessionStore.assignWorkHubMessage({ + assignment: { + type: 'workhub_coordination', + id: `wha_${suffix}`, + turnId: input.actionId, + ts: assignedAt, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId: input.actionId, + actionFingerprint: input.actionFingerprint, + coordinationTurnId: input.actionId, + targetSessionId: input.targetSessionId, + targetSessionName: input.targetSessionName, + targetTurnId: turnId, + targetMessageId: messageId, + delegationId: `whd_${suffix}`, + disposition: input.disposition, + userText: input.userText, + ...(steered ? { steered: true as const } : {}), + ...(input.create ? { create: input.create } : {}), + }, + admission: { + sessionId: input.targetSessionId, + turnId, + runId, + messageId, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: assignedAt, + }, + ...(create ? { create } : {}), + }); + try { + await continuityCoordinator.refreshCanonical( + WORKHUB_COORDINATION_SESSION_ID, + lease, + ); + await continuityCoordinator.refreshCanonical(input.targetSessionId, lease); + } catch { + // The atomic assignment is already committed. Projection + // refresh is rebuildable and must not reject the action. + } + return result.assignment; + }, + )); + + // Assignment is already the acknowledged durable outcome. Consume + // the exact stored admission after releasing the assignment leases; + // failure leaves it pending for the same normal recovery consumer. + void messages + .consumePendingAdmissions([persisted.targetSessionId]) + .catch(() => undefined); + return { + turnId: persisted.targetTurnId, + ...(persisted.steered ? { steered: true as const } : {}), + }; }, }, resolveCreateTarget: async () => { diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 0ba526392f..1db629328d 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -647,106 +647,118 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { } async recoverPendingAfterHostRestart(sessionIds: readonly string[]): Promise { - for (const sessionId of sessionIds) { - const admissions = await this.#admissions.listMessageAdmissions(sessionId); - if (admissions.length === 0) continue; - const pending = [] as PendingMessageAdmission[]; - for (const admission of admissions) { - const source = await this.#durableProof.readRootTurnSourceMessageReceipt( + await this.consumePendingAdmissions(sessionIds); + } + + /** Consume canonical pending admissions without creating a second admission. */ + async consumePendingAdmissions(sessionIds: readonly string[]): Promise { + for (const sessionId of new Set(sessionIds)) { + await this.#sessionAdmission.run(sessionId, (admission) => + this.#consumePendingAdmissions(sessionId, admission), + ); + } + } + + async #consumePendingAdmissions( + sessionId: string, + admissionLease: SessionAdmissionLease, + ): Promise { + const admissions = await this.#admissions.listMessageAdmissions(sessionId); + if (admissions.length === 0) return; + const pending = [] as PendingMessageAdmission[]; + for (const admission of admissions) { + const source = await this.#durableProof.readRootTurnSourceMessageReceipt( + sessionId, + admission.messageId, + ); + if ( + source?.admission.turnId === admission.turnId && + source.admission.runId === admission.runId && + source.sourceMessage.messageId === admission.messageId + ) { + await this.materializeMessageHandoffsForRun({ + sessionId, + turnId: source.admission.turnId, + runId: source.admission.runId, + messageIds: [admission.messageId], + }); + } else { + const steering = await this.#durableProof.readImmutableSteeringMessageProof( sessionId, admission.messageId, ); if ( - source?.admission.turnId === admission.turnId && - source.admission.runId === admission.runId && - source.sourceMessage.messageId === admission.messageId + steering?.event.turnId === admission.turnId && + steering.event.runId === admission.runId ) { await this.materializeMessageHandoffsForRun({ sessionId, - turnId: source.admission.turnId, - runId: source.admission.runId, + turnId: admission.turnId, + runId: admission.runId, messageIds: [admission.messageId], }); } else { - const steering = await this.#durableProof.readImmutableSteeringMessageProof( - sessionId, - admission.messageId, - ); - if ( - steering?.event.turnId === admission.turnId && - steering.event.runId === admission.runId - ) { - await this.materializeMessageHandoffsForRun({ - sessionId, - turnId: admission.turnId, - runId: admission.runId, - messageIds: [admission.messageId], - }); - } else { - pending.push(admission); - } + pending.push(admission); } } - if (pending.length === 0) continue; - const rootState = await this.#root.readRootState(sessionId); - if (rootState.kind !== 'active') { - if (rootState.kind !== 'idle') continue; - if (!this.#root.startRecoveredMessages) { - throw new RuntimeMessageAuthorityInvariantError( - 'Message recovery authority is unavailable', - ); - } - const started = await this.#sessionAdmission.run(sessionId, (admission) => - this.#root.startRecoveredMessages!( - { - sessionId, - content: aggregateMessageContents(pending.map((entry) => entry.content)), - submittedContent: aggregateMessageContents(pending.map((entry) => entry.content)), - sources: pending.map(pendingMessageSource), - ...(pending.length === 1 && pending[0]!.submittedIntent - ? { submittedIntent: pending[0]!.submittedIntent } - : {}), - }, - admission, - ), + } + if (pending.length === 0) return; + const rootState = await this.#root.readRootState(sessionId); + if (rootState.kind !== 'active') { + if (rootState.kind !== 'idle') return; + if (!this.#root.startRecoveredMessages) { + throw new RuntimeMessageAuthorityInvariantError( + 'Message recovery authority is unavailable', ); - if ('error' in started) { - throw new RuntimeMessageAuthorityInvariantError( - `Durable Message recovery failed: ${started.error}`, - ); - } - continue; } - if (!this.#sessions.has(sessionId)) this.#state(sessionId); - const state = this.#requireState(sessionId); - if (!state.reservedRoot) this.reserveRootTurn(rootState); - if (!sameRun(state.reservedRoot!, rootState)) continue; - for (const admission of pending) { - if (admission.turnId !== rootState.turnId || admission.runId !== rootState.runId) continue; - const existing = allLiveEntries(state).find( - (entry) => entry.messageId === admission.messageId, + const started = await this.#root.startRecoveredMessages( + { + sessionId, + content: aggregateMessageContents(pending.map((entry) => entry.content)), + submittedContent: aggregateMessageContents(pending.map((entry) => entry.content)), + sources: pending.map(pendingMessageSource), + ...(pending.length === 1 && pending[0]!.submittedIntent + ? { submittedIntent: pending[0]!.submittedIntent } + : {}), + }, + admissionLease, + ); + if ('error' in started) { + throw new RuntimeMessageAuthorityInvariantError( + `Durable Message recovery failed: ${started.error}`, ); - if (existing) continue; - const residency = this.#acquireResidency(); - const entry: LiveEntry = { - entryId: this.#createId(), - messageId: admission.messageId, - turnId: admission.turnId, - runId: admission.runId, - admittedAt: admission.admittedAt, - content: submittedProjectionContent(admission.content), - modelContent: admission.content, - submittedContentDigest: admission.submittedContentDigest, - placement: admission.placement, - disposition: admission.disposition, - generation: state.generation, - residency, - state: 'queued', - }; - if (entry.disposition === 'steering') state.steering.push(entry); - else state.followup.push(entry); - this.#mutated(state); } + return; + } + if (!this.#sessions.has(sessionId)) this.#state(sessionId); + const state = this.#requireState(sessionId); + if (!state.reservedRoot) this.reserveRootTurn(rootState); + if (!sameRun(state.reservedRoot!, rootState)) return; + for (const admission of pending) { + if (admission.turnId !== rootState.turnId || admission.runId !== rootState.runId) continue; + const existing = allLiveEntries(state).find( + (entry) => entry.messageId === admission.messageId, + ); + if (existing) continue; + const residency = this.#acquireResidency(); + const entry: LiveEntry = { + entryId: this.#createId(), + messageId: admission.messageId, + turnId: admission.turnId, + runId: admission.runId, + admittedAt: admission.admittedAt, + content: submittedProjectionContent(admission.content), + modelContent: admission.content, + submittedContentDigest: admission.submittedContentDigest, + placement: admission.placement, + disposition: admission.disposition, + generation: state.generation, + residency, + state: 'queued', + }; + if (entry.disposition === 'steering') state.steering.push(entry); + else state.followup.push(entry); + this.#mutated(state); } } @@ -774,6 +786,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { private submit( input: TurnMessageSubmitInput, context: ConnectionContext, + admission?: SessionAdmissionLease, ): Promise> { const payload = canonicalSubmitPayload(input); const isCurrentEpoch = input.originHostEpoch === this.#hostEpoch; @@ -790,9 +803,11 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { if (this.#failStopped) { return Promise.resolve(failure('host_draining', 'Runtime Host message authority has failed')); } - if (!isCurrentEpoch) return this.#submitAdmitted(input, payload, context.connectionId); + if (!isCurrentEpoch) { + return this.#submitAdmitted(input, payload, context.connectionId, admission); + } const key = operationKey(input.sessionId, input.messageId); - const result = this.#submitAdmitted(input, payload, context.connectionId); + const result = this.#submitAdmitted(input, payload, context.connectionId, admission); this.#pendingSubmits.set(key, { payload, result }); void result.then( () => this.#deletePendingSubmit(key, result), @@ -805,8 +820,11 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { input: TurnMessageSubmitInput, payload: CanonicalSubmitPayload, initiatingConnectionId: string, + admittedLease?: SessionAdmissionLease, ): Promise> { - return this.#sessionAdmission.run(input.sessionId, async (admission) => { + const execute = async ( + admission: SessionAdmissionLease, + ): Promise> => { if (this.#failStopped) { return failure('host_draining', 'Runtime Host message authority has failed'); } @@ -1074,7 +1092,12 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { ); return success(result); } - }); + }; + return admittedLease + ? this.#sessionAdmission.runAdmitted(input.sessionId, admittedLease, () => + execute(admittedLease), + ) + : this.#sessionAdmission.run(input.sessionId, execute); } private retract(input: QueueRetractInput): Promise> { diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts index 01f7556259..7264c27a88 100644 --- a/packages/runtime-host/src/server/session-catalog-coordinator.ts +++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts @@ -49,6 +49,7 @@ import { type SessionHeaderSnapshot, type ExecutionStoresWriter, } from '@maka/storage/execution-stores'; +import type { CreateStableSessionRequest } from '@maka/storage/session-store'; import type { RuntimePolicyStoresWriter } from '@maka/storage/runtime-policy-stores'; import { SessionConfigurationRevisionConflictError, @@ -204,6 +205,34 @@ export class HostSessionCatalogCoordinator { return this.#create(input); } + /** Prepare external facts before WorkHub commits create + assignment atomically. */ + async prepareWorkHubCreate(input: SessionCreateInput): Promise { + const prepared = await prepareCreate(input); + return this.#workspaceResolver.runWithUsageRecorded(input.workspace, async (workspace) => { + const [model, policy] = await Promise.all([ + this.#resolveModel(input.modelTarget, input.thinkingLevel), + this.#readRuntimePolicy(), + ]); + return { + sessionId: input.sessionId, + requestFingerprint: createRequestFingerprint(input, prepared), + input: { + cwd: workspace.cwd, + ...(workspace.projectId === null ? {} : { projectId: workspace.projectId }), + name: prepared.name, + labels: [...prepared.labels], + llmConnectionSlug: model.connectionSlug, + model: model.model, + ...(input.thinkingLevel === undefined ? {} : { thinkingLevel: input.thinkingLevel }), + ...(input.toolProfile === undefined ? {} : { toolProfile: input.toolProfile }), + permissionMode: prepared.permissionMode ?? policy.policy.chatDefaults.permissionMode, + collaborationMode: input.collaborationMode ?? 'agent', + orchestrationMode: input.orchestrationMode ?? 'default', + }, + }; + }); + } + async #query( input: SessionCatalogQueryInput, ): Promise> { diff --git a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts index bf95b82ae8..dfab6e260a 100644 --- a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts +++ b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts @@ -18,7 +18,13 @@ */ import { createHash } from 'node:crypto'; -import type { SessionHeader, SessionStatus } from '@maka/core/session'; +import type { + SessionHeader, + SessionStatus, + WorkHubDelegationAssignedMessage, + WorkHubDelegationCreateSpec, + WorkHubDelegationDisposition, +} from '@maka/core/session'; import { WORKHUB_COORDINATION_SESSION_ID, isWorkHubCoordinationSessionTarget, @@ -55,6 +61,7 @@ export type WorkHubActionGateSession = Pick< export interface WorkHubActionGateEffects { listSessions(): Promise; + readAssignment(actionId: string): Promise; answer( input: { readonly turnId: string; readonly text: string }, context: ConnectionContext, @@ -64,21 +71,22 @@ export interface WorkHubActionGateEffects { readonly userText: string; readonly assistantText: string; }): Promise; - create(input: { - readonly sessionId: string; - readonly workspace: WorkspaceTarget; - readonly title: string; - }): Promise; - submit( - input: { - readonly sessionId: string; - readonly messageId: string; - readonly text: string; - }, + assign( + input: WorkHubDelegationAssignmentInput, context: ConnectionContext, ): Promise<{ readonly turnId: string; readonly steered?: true }>; } +export interface WorkHubDelegationAssignmentInput { + readonly actionId: string; + readonly actionFingerprint: `sha256:${string}`; + readonly targetSessionId: string; + readonly targetSessionName: string; + readonly disposition: WorkHubDelegationDisposition; + readonly userText: string; + readonly create?: WorkHubDelegationCreateSpec; +} + export type WorkHubActionEffectFailureCode = | 'host_not_ready' | 'host_draining' @@ -120,7 +128,7 @@ export class WorkHubActionGateFailure extends Error { } interface ActionReplay { - readonly fingerprint: string; + readonly requestFingerprint: string; readonly result: Promise; } @@ -147,10 +155,21 @@ export class WorkHubCoordinationActionGate { input: WorkHubCoordinationActInput, context: ConnectionContext, ): Promise { - const fingerprint = digest(input); + if (!input.userText.trim()) { + return Promise.reject( + new WorkHubActionGateFailure('action_conflict', 'WorkHub action text is empty'), + ); + } + if (input.proposal.disposition === 'create_new' && !input.proposal.title.trim()) { + return Promise.reject( + new WorkHubActionGateFailure('action_conflict', 'WorkHub creation title is empty'), + ); + } + const fingerprint = actionFingerprint(input); + const requestFingerprint = digest(input); const replay = this.#actions.get(input.actionId); if (replay) { - if (replay.fingerprint !== fingerprint) { + if (replay.requestFingerprint !== requestFingerprint) { return Promise.reject( new WorkHubActionGateFailure( 'action_conflict', @@ -161,12 +180,12 @@ export class WorkHubCoordinationActionGate { return replay.result; } - const result = this.#act(input, context); - const action = { fingerprint, result }; + const result = this.#act(input, fingerprint, context); + const action = { requestFingerprint, result }; this.#actions.set(input.actionId, action); - // Successful actions remain replayable. A rejected admission does not own - // the action identity forever: callers must be able to refresh stale - // candidates or satisfy an actionable precondition and retry safely. + // Successful actions remain a Host-lifetime fast path. Rejections release + // the slot so a pre-assignment admission can retry; once assigned, SQLite + // independently owns the durable action identity. void result.catch(() => { if (this.#actions.get(input.actionId) === action) { this.#actions.delete(input.actionId); @@ -178,9 +197,20 @@ export class WorkHubCoordinationActionGate { async #act( input: WorkHubCoordinationActInput, + fingerprint: `sha256:${string}`, context: ConnectionContext, ): Promise { const proposal = input.proposal; + const durable = await this.#effects.readAssignment(input.actionId); + if (durable) { + if (durable.actionFingerprint !== fingerprint) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub action identity belongs to a different proposal', + ); + } + return this.#assign(assignmentInputFromRecord(durable), context); + } if (proposal.disposition === 'answer_here') { const turnId = coordinationTurnId(input.actionId, 'answer'); await this.#effects.answer({ turnId, text: input.userText }, context); @@ -203,20 +233,10 @@ export class WorkHubCoordinationActionGate { ); } const sessionId = workHubCreatedSessionId(input.actionId); - await this.#effects.create({ - sessionId, - workspace: input.create.workspace, - title: proposal.title, - }); - const submitted = await this.#effects.submit( - { - sessionId, - messageId: actionMessageId(input.actionId), - text: input.userText, - }, + return this.#assign( + delegationAssignment(input, fingerprint, sessionId, proposal.title), context, ); - return executionResult('create_new', sessionId, submitted); } const candidates = await this.candidates(); @@ -237,23 +257,23 @@ export class WorkHubCoordinationActionGate { } this.#assertTarget(target); - return this.#submitExisting(input, target, context); + return this.#assign( + delegationAssignment(input, fingerprint, target.sessionId, target.sessionName), + context, + ); } - async #submitExisting( - input: WorkHubCoordinationActInput, - target: WorkHubCoordinationCandidate, + async #assign( + assignment: WorkHubDelegationAssignmentInput, context: ConnectionContext, ): Promise { - const submitted = await this.#effects.submit( - { - sessionId: target.sessionId, - messageId: actionMessageId(input.actionId), - text: input.userText, - }, - context, - ); - return executionResult('delegate_existing', target.sessionId, submitted); + const admitted = await this.#effects.assign(assignment, context); + return { + disposition: assignment.disposition, + targetSessionId: assignment.targetSessionId, + targetTurnId: admitted.turnId, + ...(admitted.steered ? { steered: true as const } : {}), + } as WorkHubCoordinationActResult; } #assertTarget(target: WorkHubCoordinationCandidate): void { @@ -324,8 +344,44 @@ function coordinationTurnId(actionId: string, kind: 'answer' | 'clarify'): strin return `wha_${hash(`${actionId}\0${kind}`).slice(0, 48)}`; } -function actionMessageId(actionId: string): string { - return `whm_${hash(actionId).slice(0, 48)}`; +function delegationAssignment( + input: WorkHubCoordinationActInput, + actionFingerprint: `sha256:${string}`, + targetSessionId: string, + targetSessionName: string, +): WorkHubDelegationAssignmentInput { + const create = input.create; + if ( + input.proposal.disposition !== 'delegate_existing' && + input.proposal.disposition !== 'create_new' + ) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub local action cannot create a delegation intent', + ); + } + const base = { + actionId: input.actionId, + actionFingerprint, + targetSessionId, + targetSessionName, + disposition: input.proposal.disposition, + userText: input.userText, + } as const; + if (input.proposal.disposition === 'delegate_existing') return base; + if (!create) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub creation context is unavailable', + ); + } + return { + ...base, + create: { + title: input.proposal.title, + workspace: create.workspace, + }, + }; } function workHubCreatedSessionId(actionId: string): string { @@ -350,23 +406,40 @@ function updatedAt(session: WorkHubActionGateSession): number { return session.lastMessageAt ?? session.statusUpdatedAt ?? session.createdAt; } -function executionResult( - disposition: 'delegate_existing' | 'create_new', - sessionId: string, - submitted: { readonly turnId: string; readonly steered?: true }, -): WorkHubCoordinationActResult { - return { - disposition, - targetSessionId: sessionId, - targetTurnId: submitted.turnId, - ...(submitted.steered ? { steered: true as const } : {}), - } as WorkHubCoordinationActResult; -} - function digest(value: unknown): `sha256:${string}` { return `sha256:${hash(JSON.stringify(value))}`; } +function actionFingerprint(input: WorkHubCoordinationActInput): `sha256:${string}` { + return digest({ + userText: input.userText, + disposition: input.proposal.disposition, + ...(input.proposal.disposition === 'create_new' + ? { + title: input.proposal.title, + workspace: input.create?.workspace, + } + : {}), + ...(input.proposal.disposition === 'clarify' + ? { assistantText: input.proposal.assistantText } + : {}), + }); +} + +function assignmentInputFromRecord( + assignment: WorkHubDelegationAssignedMessage, +): WorkHubDelegationAssignmentInput { + return { + actionId: assignment.actionId, + actionFingerprint: assignment.actionFingerprint, + targetSessionId: assignment.targetSessionId, + targetSessionName: assignment.targetSessionName, + disposition: assignment.disposition, + userText: assignment.userText, + ...(assignment.create ? { create: assignment.create } : {}), + }; +} + function hash(value: string): string { return createHash('sha256').update(value, 'utf8').digest('hex'); } diff --git a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts index a90e8449b9..065b1806bb 100644 --- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts +++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts @@ -82,6 +82,7 @@ type CoordinationStores = Pick< | 'listHeaders' | 'probeStableSessionCreate' | 'readHeaderSnapshot' + | 'readWorkHubAssignment' | 'readTranscriptHighWaterSnapshot' | 'readTranscriptMessagesSnapshot' | 'updateHeaderVersioned' @@ -100,7 +101,7 @@ export interface HostWorkHubCoordinationCoordinatorOptions { readonly admission: SessionAdmissionGate; readonly continuity: Pick; readonly executions: CoordinationExecutions; - readonly sessionActions: Pick; + readonly sessionActions: Pick; readonly resolveCreateTarget: () => Promise; readonly requestDrain: () => void; } @@ -134,6 +135,7 @@ export class HostWorkHubCoordinationCoordinator { this.#requestDrain = options.requestDrain; this.#actionGate = new WorkHubCoordinationActionGate({ listSessions: () => this.#stores.listHeaders(), + readAssignment: (actionId) => this.#stores.readWorkHubAssignment(actionId), answer: async (input, context) => { const outcome = await this.#answer({ turnId: input.turnId, text: input.text }, context); if (!outcome.ok) { @@ -150,8 +152,7 @@ export class HostWorkHubCoordinationCoordinator { throw new WorkHubActionEffectFailure(outcome.error.code, outcome.error.message); } }, - create: options.sessionActions.create, - submit: options.sessionActions.submit, + assign: options.sessionActions.assign, }); } diff --git a/packages/storage/src/__tests__/workhub-message-assignment.test.ts b/packages/storage/src/__tests__/workhub-message-assignment.test.ts new file mode 100644 index 0000000000..a6fef6e413 --- /dev/null +++ b/packages/storage/src/__tests__/workhub-message-assignment.test.ts @@ -0,0 +1,198 @@ +/* + * 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 { createHash } from 'node:crypto'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { messageContentDigest, normalizeMessageContent } from '@maka/core/events'; +import { + WORKHUB_COORDINATION_SESSION_ID, + WORKHUB_COORDINATION_SESSION_ROLE, + type WorkHubDelegationAssignedMessage, +} from '@maka/core/session'; +import { createSessionStore, isSessionNotFoundError } from '../session-store.js'; + +test('atomically commits one WorkHub assignment and target admission', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-assignment-')); + const store = createSessionStore(root); + try { + await createCoordinationSession(store, root); + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }); + const request = assignmentRequest('action-one', target.id, 'Payments', 'target-turn'); + const first = await store.assignWorkHubMessage(request); + const replay = await store.assignWorkHubMessage({ + ...request, + assignment: { + ...request.assignment, + ts: request.assignment.ts + 10, + targetTurnId: 'recomputed-turn', + targetSessionName: 'Recomputed name', + }, + admission: { + ...request.admission, + turnId: 'recomputed-turn', + runId: 'recomputed-run', + }, + }); + + assert.equal(first.kind, 'assigned'); + assert.equal(replay.kind, 'existing'); + assert.equal(replay.assignment.targetTurnId, 'target-turn'); + assert.deepEqual( + await store.readMessageAdmission(target.id, request.admission.messageId), + request.admission, + ); + assert.deepEqual( + (await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID)).filter( + (message) => message.type === 'workhub_coordination', + ), + [request.assignment], + ); + const coordination = await store.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID); + assert.equal(coordination.lastMessageAt, request.assignment.ts); + await store.markMessagesHandedOff({ + sessionId: target.id, + messageIds: [request.admission.messageId], + turnId: request.admission.turnId, + }); + const replayAfterConsumption = await store.assignWorkHubMessage(request); + assert.equal(replayAfterConsumption.kind, 'existing'); + assert.deepEqual(replayAfterConsumption.assignment, request.assignment); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('rolls create_new Session back when assignment validation fails', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-create-assignment-')); + const store = createSessionStore(root); + try { + await createCoordinationSession(store, root); + const request = assignmentRequest( + 'create-action', + 'created-target', + 'Wrong name', + 'target-turn', + ); + await assert.rejects( + store.assignWorkHubMessage({ + ...request, + assignment: { + ...request.assignment, + disposition: 'create_new', + create: { + title: 'Actual name', + workspace: { kind: 'host_path', path: root }, + }, + }, + create: { + sessionId: 'created-target', + requestFingerprint: `sha256:${'b'.repeat(64)}`, + input: { + cwd: root, + name: 'Actual name', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }, + }, + }), + /display identity changed/u, + ); + await assert.rejects(store.readHeaderSnapshot('created-target'), (error) => + isSessionNotFoundError(error), + ); + assert.deepEqual(await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID), []); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +async function createCoordinationSession( + store: ReturnType, + root: string, +): Promise { + await store.createStableSession({ + sessionId: WORKHUB_COORDINATION_SESSION_ID, + requestFingerprint: `sha256:${'a'.repeat(64)}`, + input: { + cwd: root, + name: 'WorkHub', + role: WORKHUB_COORDINATION_SESSION_ROLE, + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'explore', + toolProfile: 'workhub-coordination-v1', + }, + }); +} + +function assignmentRequest( + actionId: string, + targetSessionId: string, + targetSessionName: string, + targetTurnId: string, +) { + const suffix = createHash('sha256').update(actionId, 'utf8').digest('hex').slice(0, 48); + const content = normalizeMessageContent({ text: 'Continue payment work' }); + const assignment: WorkHubDelegationAssignedMessage = { + type: 'workhub_coordination', + id: `wha_${suffix}`, + turnId: actionId, + ts: 10, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId, + actionFingerprint: `sha256:${'c'.repeat(64)}`, + coordinationTurnId: actionId, + targetSessionId, + targetSessionName, + targetTurnId, + targetMessageId: `whm_${suffix}`, + delegationId: `whd_${suffix}`, + disposition: 'delegate_existing', + userText: content.text, + }; + return { + assignment, + admission: { + sessionId: targetSessionId, + turnId: targetTurnId, + runId: `whr_${suffix}`, + messageId: assignment.targetMessageId, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn' as const, + placement: 'current_turn' as const, + disposition: 'steering' as const, + admittedAt: 10, + }, + }; +} diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 14462e5363..9dab9563d2 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -353,6 +353,8 @@ async function createExecutionStoresForWrite sessionStore.probeStableSessionCreate(sessionId, requestFingerprint)), createStableSession: (request, initialBoundary) => run(() => sessionStore.createStableSession(request, initialBoundary)), + assignWorkHubMessage: (request) => run(() => sessionStore.assignWorkHubMessage(request)), + readWorkHubAssignment: (actionId) => run(() => sessionStore.readWorkHubAssignment(actionId)), discardStableConversationCopy: (sessionId, requestFingerprint) => run(() => sessionStore.discardStableConversationCopy(sessionId, requestFingerprint)), createSubagent: (input, initialBoundary) => diff --git a/packages/storage/src/session-message-projection.ts b/packages/storage/src/session-message-projection.ts index 11ba43fde8..581ba6cc75 100644 --- a/packages/storage/src/session-message-projection.ts +++ b/packages/storage/src/session-message-projection.ts @@ -40,7 +40,13 @@ export function catalogPreviewForUserMessage(message: UserMessage): string | und export function latestVisibleMessageAt(messages: readonly StoredMessage[]): number | undefined { for (let index = messages.length - 1; index >= 0; index -= 1) { const message = messages[index]!; - if (message.type === 'user' || message.type === 'assistant') return message.ts; + if ( + message.type === 'user' || + message.type === 'assistant' || + message.type === 'workhub_coordination' + ) { + return message.ts; + } } return undefined; } @@ -64,6 +70,10 @@ export function lastMessagePreviewForMessages( const text = normalizePreviewText(message.text); if (text) return truncatePreview(text); } + if (message.type === 'workhub_coordination') { + const text = normalizePreviewText(message.userText); + if (text) return truncatePreview(text); + } } return undefined; } diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index e77d5dc093..f70f4504b0 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -47,6 +47,7 @@ import { isSessionStatus, isWorkHubCoordinationSessionId, subagentSessionRuntimeSummary, + WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_SESSION_ROLE, } from '@maka/core/session'; import { isCollaborationMode } from '@maka/core/collaboration'; @@ -82,6 +83,7 @@ import { type TurnRecord, type TurnStateMessage, type UserMessage, + type WorkHubDelegationAssignedMessage, } from '@maka/core/session'; import type { MessageAdmissionStore, PendingMessageAdmission } from './message-admission-store.js'; import { @@ -175,6 +177,19 @@ export interface CreateStableSessionRequest { readonly input: StableSessionCreateInput; } +export interface WorkHubMessageAssignmentRequest { + readonly assignment: WorkHubDelegationAssignedMessage; + readonly admission: PendingMessageAdmission; + /** Present exactly when the assignment creates its target Session. */ + readonly create?: CreateStableSessionRequest; +} + +export interface WorkHubMessageAssignmentResult { + readonly kind: 'assigned' | 'existing'; + readonly targetCreated: boolean; + readonly assignment: WorkHubDelegationAssignedMessage; +} + export type StableSessionCreateInput = CreateSessionInput & { readonly conversationCopy?: SessionConversationCopy; readonly role?: SessionRole; @@ -373,6 +388,11 @@ export interface SessionAuthorityStore extends SessionStore, MessageAdmissionSto request: CreateStableSessionRequest, initialBoundary?: ExecutionBoundary, ): Promise; + /** Atomically persist a WorkHub linkage and the target Message admission. */ + assignWorkHubMessage( + request: WorkHubMessageAssignmentRequest, + ): Promise; + readWorkHubAssignment(actionId: string): Promise; discardStableConversationCopy(sessionId: string, requestFingerprint: string): Promise; listCatalogPage( filter: SessionListFilter | undefined, @@ -576,6 +596,64 @@ class SqliteSessionStore implements SessionAuthorityStore { : result; } + async assignWorkHubMessage( + request: WorkHubMessageAssignmentRequest, + ): Promise { + await this.ensureReady(); + const create = request.create; + if (create) { + assertCoordinationIdentityPairing(create.sessionId, create.input.role); + if (create.sessionId !== request.assignment.targetSessionId) { + throw new Error('WorkHub assignment create identity does not match its target'); + } + } + const result = await this.metadata.assignWorkHubMessage({ + assignment: request.assignment, + admission: request.admission, + projection: projectSessionCatalogMessages([request.assignment]), + ...(create + ? { + create: { + header: buildSessionHeader( + this.workspaceRoot, + create.input, + create.sessionId, + create.input.conversationCopy, + ), + requestFingerprint: create.requestFingerprint, + }, + } + : {}), + }); + if (result.kind === 'assigned') { + for (const listener of this.transcriptChangeListeners) { + listener(WORKHUB_COORDINATION_SESSION_ID); + } + } + return result; + } + + async readWorkHubAssignment( + actionId: string, + ): Promise { + await this.ensureReady(); + const suffix = createHash('sha256').update(actionId, 'utf8').digest('hex').slice(0, 48); + const throughSequence = await this.metadata.readTranscriptHighWater( + WORKHUB_COORDINATION_SESSION_ID, + ); + if (throughSequence === null) return undefined; + const messages = await this.metadata.readTranscriptMessages(WORKHUB_COORDINATION_SESSION_ID, { + messageIds: [`wha_${suffix}`], + throughSequence, + maxMessages: 1, + maxBytes: 768 * 1024, + }); + const message = messages[0]; + return message?.type === 'workhub_coordination' && message.kind === 'delegation_assigned' + ? message + : undefined; + } + async discardStableConversationCopy( sessionId: string, requestFingerprint: string, diff --git a/packages/storage/src/sqlite-session-metadata-schema.ts b/packages/storage/src/sqlite-session-metadata-schema.ts index 9a39e05a65..da49f651a7 100644 --- a/packages/storage/src/sqlite-session-metadata-schema.ts +++ b/packages/storage/src/sqlite-session-metadata-schema.ts @@ -19,7 +19,7 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 33; +export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 34; export const SQLITE_SESSION_MESSAGE_CHUNK_BYTES = 64 * 1024; export const SQLITE_SESSION_MESSAGE_CHUNK_MARKER = '{"$maka":"session-message-chunks-v1"}'; @@ -1216,6 +1216,15 @@ const MIGRATIONS: ReadonlyMap = new Map([ AND json_extract(payload_json, '$.subagentParent') IS NOT NULL; `, ], + [ + 34, + ` + -- WorkHub delegation_assigned records are decoded by schema-aware builds. + -- Advancing the profile schema prevents an older build from opening a + -- transcript containing this new canonical message type. + SELECT 1; + `, + ], ]); if (MIGRATIONS.size !== SQLITE_SESSION_METADATA_SCHEMA_VERSION) { diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index 59b34eae88..dd053e6b99 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -92,6 +92,9 @@ import { type SessionHeaderPatch, type StoredMessage, type SubagentSessionParent, + type WorkHubDelegationAssignedMessage, + WORKHUB_COORDINATION_SESSION_ID, + WORKHUB_COORDINATION_SESSION_ROLE, decodeCanonicalMessage, decodeStoredMessage as decodePersistedStoredMessage, } from '@maka/core/session'; @@ -102,7 +105,11 @@ import { type PendingMessageAdmission, } from './message-admission-store.js'; import { normalizeSubmittedTurnIntent } from './submitted-turn-intent.js'; -import { messageContentsEqual, normalizeMessageContent } from '@maka/core/events'; +import { + messageContentDigest, + messageContentsEqual, + normalizeMessageContent, +} from '@maka/core/events'; import { type AgentGraphIntentAdmissionSnapshot, type AgentGraphTimelineMetadataSnapshot, @@ -208,6 +215,22 @@ export interface SqliteSessionMetadataStoreOptions { databaseLease?: OperationalStateDatabaseLease; } +export interface SqliteWorkHubMessageAssignmentRequest { + readonly assignment: WorkHubDelegationAssignedMessage; + readonly admission: PendingMessageAdmission; + readonly projection: SessionCatalogMessageProjection; + readonly create?: { + readonly header: SessionHeader; + readonly requestFingerprint: string; + }; +} + +export interface SqliteWorkHubMessageAssignmentResult { + readonly kind: 'assigned' | 'existing'; + readonly targetCreated: boolean; + readonly assignment: WorkHubDelegationAssignedMessage; +} + export interface SessionMetadataRecord { header: SessionHeader; metadataVersion: number; @@ -1575,70 +1598,225 @@ export class SqliteSessionMetadataStore { const stored = normalizePendingMessageAdmission(admission); return this.transaction(() => { if (!this.readRecordSync(stored.sessionId)) throw new SessionNotFoundError(stored.sessionId); - const existingRow = this.db - .prepare( - ` - SELECT turn_id, run_id, message_id, content_json, submitted_content_digest, - submitted_placement, placement, disposition, queue_order, admitted_at, - submitted_intent_json - FROM message_admissions - WHERE session_id = ? AND message_id = ? - `, - ) - .get(stored.sessionId, stored.messageId) as MessageAdmissionRow | undefined; - if (existingRow) { - const existing = decodeMessageAdmissionRow(stored.sessionId, existingRow); + const existing = this.readMessageAdmissionSync(stored.sessionId, stored.messageId); + if (existing) { if (!samePendingMessageAdmission(existing, stored)) { throw new SessionMetadataConflictError('Message admission identity conflict'); } return existing; } - const cancelled = this.db - .prepare( - 'SELECT 1 AS present FROM cancelled_message_admissions WHERE session_id = ? AND message_id = ?', - ) - .get(stored.sessionId, stored.messageId); - if (cancelled) { - throw new SessionMetadataConflictError('Message admission identity is already cancelled'); - } - const orderRow = this.db - .prepare( - ` + this.insertMessageAdmissionSync(stored); + return stored; + }); + } + + private readMessageAdmissionSync( + sessionId: string, + messageId: string, + ): PendingMessageAdmission | undefined { + const row = this.db + .prepare( + ` + SELECT turn_id, run_id, message_id, content_json, submitted_content_digest, + submitted_placement, placement, disposition, queue_order, admitted_at, + submitted_intent_json + FROM message_admissions + WHERE session_id = ? AND message_id = ? + `, + ) + .get(sessionId, messageId) as MessageAdmissionRow | undefined; + return row ? decodeMessageAdmissionRow(sessionId, row) : undefined; + } + + private insertMessageAdmissionSync(stored: PendingMessageAdmission): void { + const cancelled = this.db + .prepare( + 'SELECT 1 AS present FROM cancelled_message_admissions WHERE session_id = ? AND message_id = ?', + ) + .get(stored.sessionId, stored.messageId); + if (cancelled) { + throw new SessionMetadataConflictError('Message admission identity is already cancelled'); + } + const orderRow = this.db + .prepare( + ` SELECT COALESCE(MAX(queue_order), -1) + 1 AS next_order FROM message_admissions WHERE session_id = ? `, - ) - .get(stored.sessionId) as { next_order?: unknown }; - if (typeof orderRow.next_order !== 'number' || !Number.isSafeInteger(orderRow.next_order)) { - throw new SessionMetadataConflictError('Invalid message admission order'); - } - this.db - .prepare( - ` + ) + .get(stored.sessionId) as { next_order?: unknown }; + if (typeof orderRow.next_order !== 'number' || !Number.isSafeInteger(orderRow.next_order)) { + throw new SessionMetadataConflictError('Invalid message admission order'); + } + this.db + .prepare( + ` INSERT INTO message_admissions( session_id, turn_id, run_id, message_id, content_json, submitted_content_digest, submitted_placement, placement, disposition, queue_order, admitted_at, submitted_intent_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, - ) - .run( - stored.sessionId, - stored.turnId, - stored.runId, - stored.messageId, - JSON.stringify(stored.content), - stored.submittedContentDigest, - stored.submittedPlacement, - stored.placement, - stored.disposition, - orderRow.next_order, - stored.admittedAt, - stored.submittedIntent ? JSON.stringify(stored.submittedIntent) : null, + ) + .run( + stored.sessionId, + stored.turnId, + stored.runId, + stored.messageId, + JSON.stringify(stored.content), + stored.submittedContentDigest, + stored.submittedPlacement, + stored.placement, + stored.disposition, + orderRow.next_order, + stored.admittedAt, + stored.submittedIntent ? JSON.stringify(stored.submittedIntent) : null, + ); + } + + async assignWorkHubMessage( + request: SqliteWorkHubMessageAssignmentRequest, + ): Promise { + const assignmentJson = JSON.stringify(request.assignment); + const assignment = decodeCanonicalMessage(JSON.parse(assignmentJson) as unknown); + const admission = normalizePendingMessageAdmission(request.admission); + const suffix = createHash('sha256') + .update(request.assignment.actionId) + .digest('hex') + .slice(0, 48); + if ( + assignment.type !== 'workhub_coordination' || + assignment.kind !== 'delegation_assigned' || + assignment.targetSessionId !== admission.sessionId || + assignment.targetTurnId !== admission.turnId || + assignment.targetMessageId !== admission.messageId || + assignment.id !== `wha_${suffix}` || + assignment.targetMessageId !== `whm_${suffix}` || + assignment.delegationId !== `whd_${suffix}` || + !messageContentsEqual( + admission.content, + normalizeMessageContent({ text: assignment.userText }), + ) || + admission.submittedContentDigest !== messageContentDigest(admission.content) || + admission.submittedPlacement !== 'current_turn' || + admission.placement !== 'current_turn' || + admission.disposition !== 'steering' + ) { + throw new SessionMetadataConflictError('Invalid WorkHub assignment identity'); + } + const create = request.create + ? { + header: normalizeSessionHeader(request.create.header), + requestFingerprint: request.create.requestFingerprint, + } + : undefined; + if (create) { + assertSessionCreateFingerprint(create.requestFingerprint); + if (create.header.id !== assignment.targetSessionId) { + throw new SessionMetadataConflictError('WorkHub create identity does not match target'); + } + } + assertCatalogMessageProjection(request.projection); + if ((assignment.disposition === 'create_new') !== Boolean(create)) { + throw new SessionMetadataConflictError( + 'WorkHub create request does not match assignment disposition', + ); + } + + return this.transaction(() => { + const coordination = this.readRecordSync(WORKHUB_COORDINATION_SESSION_ID); + if ( + !coordination || + coordination.header.role !== WORKHUB_COORDINATION_SESSION_ROLE || + coordination.header.isArchived + ) { + throw new SessionMetadataConflictError('WorkHub Coordination Session is unavailable'); + } + + const existingAssignment = this.readMessageByIdSync( + WORKHUB_COORDINATION_SESSION_ID, + assignment.id, + ); + if (existingAssignment) { + if ( + existingAssignment.type !== 'workhub_coordination' || + existingAssignment.kind !== 'delegation_assigned' || + !sameWorkHubAssignmentRequest(existingAssignment, assignment) + ) { + throw new SessionMetadataConflictError( + 'WorkHub action identity belongs to a different assignment', + ); + } + return { + kind: 'existing' as const, + targetCreated: false, + assignment: existingAssignment, + }; + } + + let targetCreated = false; + if (create) { + const probe = this.probeStableSessionCreateSync( + create.header.id, + create.requestFingerprint, ); + if (probe.kind === 'conflict') { + throw new SessionMetadataConflictError( + 'WorkHub target Session identity belongs to a different create request', + ); + } + if (probe.kind === 'absent') { + const committedAt = this.now(); + this.db + .prepare( + ` + INSERT INTO session_create_claims(session_id, request_fingerprint, claimed_at) + VALUES (?, ?, ?) + `, + ) + .run(create.header.id, create.requestFingerprint, committedAt); + this.insertHeader(create.header, 1, committedAt); + targetCreated = true; + } + } - return stored; + const target = this.readRecordSync(assignment.targetSessionId); + if (!target || target.header.isArchived) { + throw new SessionMetadataConflictError('WorkHub target Session is unavailable'); + } + if (target.header.status === 'waiting_for_user') { + throw new SessionMetadataConflictError('WorkHub target Session is waiting for user input'); + } + if (target.header.name !== assignment.targetSessionName) { + throw new SessionMetadataConflictError('WorkHub target display identity changed'); + } + if (this.readMessageAdmissionSync(admission.sessionId, admission.messageId)) { + throw new SessionMetadataConflictError( + 'WorkHub target Message identity belongs to another admission', + ); + } + + this.insertMessageAdmissionSync(admission); + const sequenceRow = this.db + .prepare( + 'SELECT COALESCE(MAX(sequence), -1) AS last_sequence FROM session_messages WHERE session_id = ?', + ) + .get(WORKHUB_COORDINATION_SESSION_ID) as { last_sequence?: unknown }; + if ( + typeof sequenceRow.last_sequence !== 'number' || + !Number.isSafeInteger(sequenceRow.last_sequence) || + sequenceRow.last_sequence < -1 + ) { + throw new SessionMetadataConflictError('Invalid WorkHub transcript sequence'); + } + this.insertSessionMessagesSync( + WORKHUB_COORDINATION_SESSION_ID, + sequenceRow.last_sequence + 1, + [{ message: assignment, json: assignmentJson }], + ); + this.updateCatalogProjectionSync(WORKHUB_COORDINATION_SESSION_ID, request.projection, false); + return { kind: 'assigned' as const, targetCreated, assignment }; }); } @@ -4634,6 +4812,21 @@ export class SqliteSessionMetadataStore { return row ? decodeRecord(row) : undefined; } + private readMessageByIdSync(sessionId: string, messageId: string): StoredMessage | undefined { + const row = this.db + .prepare( + ` + SELECT message.sequence, message.record_json, payload.record_bytes, payload.sha256 + FROM session_messages AS message + LEFT JOIN session_message_payloads AS payload + ON payload.session_id = message.session_id AND payload.sequence = message.sequence + WHERE message.session_id = ? AND message.message_id = ? + `, + ) + .get(sessionId, messageId) as StoredSessionMessagePayloadRow | undefined; + return row ? decodeStoredMessageRecordRow(this.db, sessionId, row) : undefined; + } + private insertSessionMessagesSync( sessionId: string, firstSequence: number, @@ -6260,6 +6453,32 @@ function readStoredMessageRecordJson( return recordJson; } +function sameWorkHubAssignmentRequest( + existing: WorkHubDelegationAssignedMessage, + requested: WorkHubDelegationAssignedMessage, +): boolean { + return isDeepStrictEqual( + { + actionId: existing.actionId, + actionFingerprint: existing.actionFingerprint, + coordinationTurnId: existing.coordinationTurnId, + targetSessionId: existing.targetSessionId, + disposition: existing.disposition, + userText: existing.userText, + create: existing.create, + }, + { + actionId: requested.actionId, + actionFingerprint: requested.actionFingerprint, + coordinationTurnId: requested.coordinationTurnId, + targetSessionId: requested.targetSessionId, + disposition: requested.disposition, + userText: requested.userText, + create: requested.create, + }, + ); +} + function foldTurnContribution( current: SessionTurnContribution | undefined, turnId: string,