diff --git a/apps/desktop/e2e/workhub-reconstruction.spec.ts b/apps/desktop/e2e/workhub-reconstruction.spec.ts index bda08351cc..bc332ee7c2 100644 --- a/apps/desktop/e2e/workhub-reconstruction.spec.ts +++ b/apps/desktop/e2e/workhub-reconstruction.spec.ts @@ -19,6 +19,14 @@ import { expect, test, COMPOSER_INPUT } from './fixtures'; +type WorkHubEvidenceWindow = Window & { + makaE2eLatch?: { + arm(key: 'workHub.record', options?: { oneShot?: boolean }): void; + reject(key: 'workHub.record', message: string): void; + waitForCall(key: 'workHub.record'): Promise; + }; +}; + test('WorkHub rebuilds Session conversation after navigating away and back', async ({ window: page, }) => { @@ -65,6 +73,66 @@ test('WorkHub rebuilds Session conversation after navigating away and back', asy ).toBeVisible(); }); +test('WorkHub retries one accepted action after summary failure and renderer reload', async ({ + window: page, +}) => { + const composer = page.locator(COMPOSER_INPUT); + await composer.fill('检查支付回调重复投递时的幂等性'); + await composer.press('Enter'); + await expect(page.getByRole('button', { name: '重新生成' })).toHaveCount(1, { + timeout: 20_000, + }); + await page.evaluate(async () => { + await window.maka.settings.updateClient({ workHub: { enabled: true } }); + }); + await expect(page.getByRole('main', { name: 'WorkHub' })).toBeVisible(); + + const latchInstalled = await page.evaluate(() => { + const e2e = window as WorkHubEvidenceWindow; + if (!e2e.makaE2eLatch) return false; + e2e.makaE2eLatch.arm('workHub.record', { oneShot: true }); + return true; + }); + expect(latchInstalled, 'the isolated E2E summary latch is installed').toBe(true); + + const routedPrompt = '继续这个工作，补充重复投递测试点。'; + const workHubComposer = page.locator( + '.workhub-surface .maka-composer-editor [contenteditable="true"]', + ); + await workHubComposer.fill(routedPrompt); + const recordReached = page.evaluate(() => + (window as WorkHubEvidenceWindow).makaE2eLatch?.waitForCall('workHub.record'), + ); + await workHubComposer.press('Enter'); + await recordReached; + await page.evaluate(() => { + (window as WorkHubEvidenceWindow).makaE2eLatch?.reject( + 'workHub.record', + 'forced WorkHub summary failure', + ); + }); + + const failed = page.locator('.workhub-turn', { hasText: routedPrompt }); + await expect(failed.locator('.workhub-error')).toContainText('输入未能送达'); + await expect(workHubComposer).toHaveText(routedPrompt); + + await page.reload(); + + await expect(page.getByRole('main', { name: 'WorkHub' })).toBeVisible(); + const reloadedComposer = page.locator( + '.workhub-surface .maka-composer-editor [contenteditable="true"]', + ); + await expect(reloadedComposer).toHaveText(routedPrompt); + await reloadedComposer.press('Enter'); + + await expect(page.locator('.workhub-submitted').last()).toBeVisible(); + await expect( + page.locator('.workhub-user-bubble > p', { + hasText: routedPrompt, + }), + ).toHaveCount(1); +}); + test('WorkHub defers destructive correction until linked delegation exists', async ({ window: page, }) => { 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..f04847c419 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,33 @@ 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-commit-message', + turnId: 'coordination-turn', + ts: 2, + schemaVersion: 1, + kind: 'delegation_committed', + 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', + }, + ); + + 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 467fb567c6..a7812f1319 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -2469,6 +2469,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 af3d8d48d7..135a8c0e82 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -168,8 +168,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..105545caf4 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, @@ -41,8 +43,388 @@ import { createDesktopWorkHubSessionPort, type WorkHubDesktopSession, } from '../../renderer/workhub-session-port.js'; +import { WorkHubSendLease } from '../../renderer/workhub-send-lease.js'; +import { WorkHubCoordinationFailure } from '../../renderer/workhub-coordination-port.js'; + +test('production retry keeps one action identity across failure and renderer reload', () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const ids = ['action-1', 'action-2']; + const first = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => ids.shift()!, + }); + + assert.equal(first.acquire('Continue payment work'), 'action-1'); + + const restarted = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => ids.shift()!, + }); + assert.equal(restarted.acquire('Continue payment work'), 'action-1'); + restarted.complete('action-1'); + assert.equal(restarted.acquire('Continue payment work'), 'action-2'); +}); + +test('a failed storage retirement cannot resurrect a settled action in memory', () => { + const values = new Map(); + let rejectWrites = false; + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => { + if (rejectWrites) throw new Error('storage unavailable'); + values.set(key, value); + }, + removeItem: (key: string) => { + if (rejectWrites) throw new Error('storage unavailable'); + values.delete(key); + }, + }; + const ids = ['action-1', 'action-2']; + const lease = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => ids.shift()!, + }); + + const first = lease.acquire('Continue payment work'); + rejectWrites = true; + lease.complete(first); + + assert.equal(lease.acquire('Continue payment work'), 'action-2'); +}); + +test('typing the next draft while an action is in flight cannot revoke its summary identity', () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const ids = ['action-in-flight', 'action-next']; + const lease = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => ids.shift()!, + }); + const attempt = lease.acquireAttempt('Continue payment work'); + + lease.write('workhub', 'Start the next message'); + + assert.equal( + lease.summary(attempt.requestId, () => 'Accepted by Payments.'), + 'Accepted by Payments.', + ); + assert.deepEqual(lease.acquireAttempt('Start the next message'), { + requestId: 'action-in-flight', + text: 'Continue payment work', + retrying: true, + }); + assert.equal(lease.settle(attempt.requestId, true), false); + assert.equal(lease.read('workhub'), 'Start the next message'); + assert.deepEqual(lease.acquireAttempt('Start the next message'), { + requestId: 'action-next', + text: 'Start the next message', + retrying: false, + }); +}); + +test('a new send normalizes surrounding whitespace without retaining the sent draft', () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const lease = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'trimmed-action', + }); + lease.write('workhub', ' Continue payment work '); + + const attempt = lease.acquireAttempt('Continue payment work'); + + assert.equal(lease.read('workhub'), 'Continue payment work'); + assert.equal(lease.settle(attempt.requestId, true), true); +}); + +test('clarification choice retries through the same leased action identity', async () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const ids = ['choice-action-1', 'choice-action-2']; + const lease = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => ids.shift()!, + }); + const clarificationRequestId = lease.acquire('Continue the login work'); + assert.equal(lease.settle(clarificationRequestId, true), true); + lease.write('workhub', ''); + const submittedIds: string[] = []; + let failSummary = true; + const choose = () => submitLeasedWorkHubSurfaceInput({ + lease, + text: 'Continue the login work', + preserveDraft: true, + submit: async (attempt) => { + submittedIds.push(attempt.requestId); + if (failSummary) { + failSummary = false; + return undefined; + } + return { + kind: 'submitted', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: attempt.requestId, + target: { sessionId: 'login' }, + turnId: 'login-turn', + evidence: 'explicit_target', + }; + }, + }); + + assert.equal(await choose(), false); + assert.equal(lease.read('workhub'), 'Continue the login work'); + assert.equal(await choose(), false); + assert.equal(lease.read('workhub'), undefined); + assert.deepEqual(submittedIds, ['choice-action-2', 'choice-action-2']); +}); + +test('production retry identity is isolated by Runtime Host scope', () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const hostA = new WorkHubSendLease({ + scope: '["host-a","workhub_coordination"]', + storage, + createId: () => 'action-A', + }); + const hostB = new WorkHubSendLease({ + scope: '["host-b","workhub_coordination"]', + storage, + createId: () => 'action-B', + }); + + assert.equal(hostA.acquire('Continue payment work'), 'action-A'); + assert.equal(hostB.acquire('Continue payment work'), 'action-B'); + hostB.complete('action-B'); + assert.equal( + new WorkHubSendLease({ + scope: '["host-a","workhub_coordination"]', + storage, + createId: () => 'action-A-new', + }).acquire('Continue payment work'), + 'action-A', + ); +}); + +test('waiting keeps the action identity that may own an unrecorded summary', () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const first = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-1', + }); + const requestId = first.acquire('Continue payment work'); + + first.settle(requestId, workHubSubmissionClearsDraft({ + kind: 'waiting', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId, + text: 'Continue payment work', + target: { sessionId: 'payment' }, + })); + + assert.equal( + new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-2', + }).acquire('Continue payment work'), + 'action-1', + ); +}); + +test('waiting does not bind the final summary before the same action is accepted', async () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const recorded: string[] = []; + let attempts = 0; + const controller: WorkHubController = { + read: async () => ({ sessions: [], turns: [] }), + openConversation: async () => ({ close: async () => undefined }), + recordConversationTurn: async ({ turnId, assistantText }) => { + recorded.push(assistantText); + return { turnId }; + }, + resetVisitContext: () => {}, + subscribe: () => () => {}, + submit: async (input) => { + attempts += 1; + return attempts === 1 + ? { + kind: 'waiting' as const, + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + text: input.text, + target: { sessionId: 'payment' }, + } + : { + kind: 'submitted' as const, + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + target: { sessionId: 'payment' }, + turnId: 'payment-turn', + evidence: 'explicit_target' as const, + }; + }, + }; + const first = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-1', + }); + const requestId = first.acquire('Continue payment work'); + const send = (lease: WorkHubSendLease, retrying: boolean) => + submitAndRecordWorkHubSurfaceInput({ + controller, + request: { + requestId, + text: 'Continue payment work', + ...(retrying ? { retryAction: true as const } : {}), + }, + recordedUserText: 'Continue payment work', + summary: (result) => lease.summary( + requestId, + () => result.kind === 'waiting' ? 'Request not sent.' : 'Accepted by Payments.', + ), + onSummaryError: () => undefined, + }); + + const waiting = await send(first, false); + first.settle(requestId, workHubSubmissionClearsDraft(waiting)); + const restarted = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-2', + }); + assert.equal(restarted.acquire('Continue payment work'), requestId); + await send(restarted, true); + + assert.deepEqual(recorded, ['Accepted by Payments.']); +}); + +test('summary retry reuses the text first bound to the action identity', () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const first = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-1', + }); + const requestId = first.acquire('Continue payment work'); + assert.equal(first.summary(requestId, () => 'Sent to Payments · running'), 'Sent to Payments · running'); + + const restarted = new WorkHubSendLease({ + scope: 'host-a', + storage, + createId: () => 'action-2', + }); + assert.equal( + restarted.summary(requestId, () => 'Sent to Payment archive · completed'), + 'Sent to Payments · running', + ); +}); + +test('summary failure keeps the target action retryable under the same production identity', async () => { + const values = new Map(); + const storage = { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }; + const actionIds: string[] = []; + let summaries = 0; + const controller: WorkHubController = { + read: async () => ({ sessions: [], turns: [] }), + openConversation: async () => ({ close: async () => undefined }), + recordConversationTurn: async ({ turnId }) => { + summaries += 1; + if (summaries === 1) throw new Error('summary outcome unknown'); + return { turnId }; + }, + resetVisitContext: () => {}, + subscribe: () => () => {}, + submit: async (input) => { + actionIds.push(input.requestId); + return { + kind: 'submitted', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + target: { sessionId: 'payment' }, + turnId: 'payment-turn', + evidence: 'explicit_target', + }; + }, + }; + const send = (requestId: string) => submitAndRecordWorkHubSurfaceInput({ + controller, + request: { requestId, text: 'Continue payment work' }, + recordedUserText: 'Continue payment work', + summary: () => 'Sent to Payments.', + onSummaryError: () => undefined, + }); + const first = new WorkHubSendLease({ scope: 'host-a', storage, createId: () => 'action-1' }); + const requestId = first.acquire('Continue payment work'); + + await assert.rejects(send(requestId), /summary outcome unknown/u); + + const restarted = new WorkHubSendLease({ scope: 'host-a', storage, createId: () => 'action-2' }); + const retriedId = restarted.acquire('Continue payment work'); + await send(retriedId); + restarted.complete(retriedId); + + assert.deepEqual(actionIds, ['action-1', 'action-1']); + assert.equal(summaries, 2); +}); test('surface turns Action Gate rejections into safe actionable failures', () => { + assert.equal( + workHubSurfaceFailure( + new WorkHubCoordinationFailure( + 'operation_conflict', + 'WorkHub action is permanently abandoned', + ), + ), + 'action_changed', + ); assert.equal( workHubSurfaceFailure( new Error('WorkHub Session candidates changed; refresh before delegating'), 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 94129cd1f7..c44a49ae9d 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'; @@ -816,7 +817,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 217bcd3281..66de255e2c 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'; @@ -1614,7 +1615,7 @@ const makaBridge = { async act( coordinationSessionId: string, input: Omit, 'create'>, - ): Promise> { + ): Promise> { const scope = await resolveDesktopWorkHubCoordinationCreateScope( coordinationSessionId, runtimeHostSessionRef, @@ -1623,14 +1624,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( @@ -3289,14 +3297,22 @@ const makaBridge = { // exposeInMainWorld: the bridge is cloned into the main world at expose time, // and the exposed clone is sealed against later patching. if (process.env.MAKA_E2E === '1' && process.env.MAKA_E2E_USER_DATA_DIR) { - type LatchKey = 'newTasks.listInvocableSkills' | 'sessions.list' | 'settings.chunk'; + type LatchKey = + | 'newTasks.listInvocableSkills' + | 'sessions.list' + | 'settings.chunk' + | 'workHub.record'; const gates = new Map; oneShot: boolean }>(); const releases = new Map void; reject: (error: Error) => void }>(); + const callWaiters = new Map void>>(); const invocableSkillsWaiters = new Map void>>(); const waitForLatch = async (key: LatchKey): Promise => { const gate = gates.get(key); if (!gate) return; if (gate.oneShot) gates.delete(key); + const waiter = callWaiters.get(key)?.shift(); + if (callWaiters.get(key)?.length === 0) callWaiters.delete(key); + waiter?.(); await gate.promise; }; const wrapLatched = ( @@ -3314,6 +3330,10 @@ if (process.env.MAKA_E2E === '1' && process.env.MAKA_E2E_USER_DATA_DIR) { makaBridge.sessions.list.bind(makaBridge.sessions), 'sessions.list', ); + makaBridge.workHub.record = wrapLatched( + makaBridge.workHub.record.bind(makaBridge.workHub), + 'workHub.record', + ); const listInvocableSkills = makaBridge.skills.listInvocable.bind(makaBridge.skills); makaBridge.skills.listInvocable = async (...args) => { try { @@ -3346,6 +3366,13 @@ if (process.env.MAKA_E2E === '1' && process.env.MAKA_E2E_USER_DATA_DIR) { wait(key: 'settings.chunk') { return waitForLatch(key); }, + waitForCall(key: LatchKey) { + return new Promise((resolve) => { + const waiters = callWaiters.get(key) ?? []; + waiters.push(resolve); + callWaiters.set(key, waiters); + }); + }, waitForInvocableSkillsCall(sessionId: string) { return new Promise((resolve) => { const waiters = invocableSkillsWaiters.get(sessionId) ?? []; diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index 7a8fc2e9db..aabbca1a66 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -2819,6 +2819,7 @@ function AppShellContent({ ['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; 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..8f50d21aec --- /dev/null +++ b/apps/desktop/src/renderer/workhub-send-lease.ts @@ -0,0 +1,255 @@ +/* + * 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_SEND_LEASE_KEY = 'maka-workhub-send-lease-v2'; +const WORKHUB_DRAFT_KEY = 'workhub'; +const MAX_DRAFT_CHARS = 120_000; +const MAX_SUMMARY_CHARS = 4_000; +const MAX_SCOPE_CHARS = 1_024; +const SAFE_REQUEST_ID = /^[A-Za-z0-9_-]{1,128}$/u; + +type WorkHubSendLeaseStorage = Pick; + +interface WorkHubSendLeaseState { + readonly version: 2; + readonly draft: string; + readonly action?: { + readonly requestId: string; + readonly text: string; + readonly state: 'active' | 'settled'; + readonly summary?: string; + }; +} + +export interface WorkHubSendLeaseOptions { + readonly scope: string; + readonly storage?: WorkHubSendLeaseStorage; + readonly createId?: () => string; +} + +export interface WorkHubSendAttempt { + readonly requestId: string; + readonly text: string; + readonly retrying: boolean; +} + +/** + * Persists a Composer draft beside, but independently from, the Action Gate + * identity that owns an in-flight delivery. This lets a user type the next + * draft without revoking or overwriting recovery for the previous action. + */ +export class WorkHubSendLease { + #memory: WorkHubSendLeaseState | undefined; + readonly #scope: string; + readonly #storage: WorkHubSendLeaseStorage | undefined; + readonly #createId: () => string; + readonly #storageKey: string; + #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'); + } + this.#scope = options.scope; + this.#storage = options.storage ?? rendererPersistentStorage(); + this.#createId = options.createId ?? (() => crypto.randomUUID()); + this.#storageKey = `${WORKHUB_SEND_LEASE_KEY}:${encodeURIComponent(this.#scope)}`; + } + + acquire(text: string): string { + return this.acquireAttempt(text).requestId; + } + + acquireAttempt( + text: string, + options: { readonly preserveDraft?: boolean } = {}, + ): WorkHubSendAttempt { + const existing = this.#read(); + if (existing?.action?.state === 'active') { + return { + requestId: existing.action.requestId, + text: existing.action.text, + retrying: true, + }; + } + if ( + !options.preserveDraft && + existing?.action?.state === 'settled' && + existing.draft === text && + existing.action.text === text + ) { + return { + requestId: existing.action.requestId, + text: existing.action.text, + retrying: true, + }; + } + const requestId = this.#createId(); + this.#write({ + version: 2, + draft: options.preserveDraft ? existing?.draft ?? text : text, + action: { requestId, text, state: 'active' }, + }); + return { requestId, text, retrying: false }; + } + + complete(requestId: string): void { + const existing = this.#read(); + if (existing?.action?.requestId !== requestId) return; + this.#write({ version: 2, draft: existing.draft }); + } + + settle(requestId: string, clearsDraft: boolean): boolean { + const existing = this.#read(); + if (!clearsDraft || existing?.action?.requestId !== requestId) return false; + const draftUnchanged = existing.draft === existing.action.text; + if (!existing.draft) { + this.#write({ version: 2, draft: '' }); + } else { + this.#write({ + ...existing, + action: { ...existing.action, state: 'settled' }, + }); + } + return draftUnchanged; + } + + abandon(requestId: string): void { + this.complete(requestId); + } + + summary(requestId: string, create: () => string): string { + const existing = this.#read(); + if (existing?.action?.requestId !== requestId) { + throw new Error('WorkHub summary identity does not own the active send lease'); + } + if (existing.action.summary) return existing.action.summary; + const summary = create(); + if (!summary || summary.length > MAX_SUMMARY_CHARS) { + throw new Error('WorkHub coordination summary is invalid'); + } + this.#write({ + ...existing, + action: { ...existing.action, summary }, + }); + return summary; + } + + read(key: string | undefined): string | undefined { + return key === WORKHUB_DRAFT_KEY ? this.#read()?.draft : undefined; + } + + write(key: string | undefined, draft: string): void { + if (key !== WORKHUB_DRAFT_KEY) return; + if (!draft) { + const existing = this.#read(); + if (existing?.action?.state === 'active') { + this.#write({ ...existing, draft: '' }); + } else { + this.#remove(); + } + return; + } + const existing = this.#read(); + // Composer permits the user to type the next draft while the current send + // is still settling. Draft edits therefore cannot revoke the identity that + // owns an already-admitted target effect or its Coordination summary. + this.#write({ + version: 2, + draft, + ...(existing?.action ? { action: existing.action } : {}), + }); + } + + #read(): WorkHubSendLeaseState | undefined { + if (!this.#storageHealthy) return this.#memory; + try { + const raw = this.#storage?.getItem(this.#storageKey); + if (!raw) return this.#memory; + const value = JSON.parse(raw) as Partial; + if ( + value.version !== 2 || + typeof value.draft !== 'string' || + value.draft.length > MAX_DRAFT_CHARS || + (value.action !== undefined && !isWorkHubSendAction(value.action)) + ) { + return undefined; + } + const decoded = { + version: 2, + draft: value.draft, + ...(value.action ? { action: value.action } : {}), + } satisfies WorkHubSendLeaseState; + this.#memory = decoded; + return decoded; + } catch { + this.#storageHealthy = false; + return this.#memory; + } + } + + #write(value: WorkHubSendLeaseState): void { + this.#memory = value; + if (!this.#storageHealthy) return; + try { + this.#storage?.setItem(this.#storageKey, JSON.stringify(value)); + } catch { + this.#storageHealthy = false; + } + } + + #remove(): void { + this.#memory = undefined; + if (!this.#storageHealthy) return; + try { + this.#storage?.removeItem(this.#storageKey); + } catch { + this.#storageHealthy = false; + } + } +} + +function isWorkHubSendAction( + value: unknown, +): value is NonNullable { + if (!value || typeof value !== 'object') return false; + const candidate = value as Partial>; + return ( + typeof candidate.requestId === 'string' && + SAFE_REQUEST_ID.test(candidate.requestId) && + typeof candidate.text === 'string' && + candidate.text.length > 0 && + candidate.text.length <= MAX_DRAFT_CHARS && + (candidate.state === 'active' || candidate.state === 'settled') && + (candidate.summary === undefined || + (typeof candidate.summary === 'string' && + candidate.summary.length > 0 && + candidate.summary.length <= MAX_SUMMARY_CHARS)) + ); +} + +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 71f0179a6d..55b88ed21c 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,65 @@ 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. + if (result.kind === 'discussion' || result.kind === 'waiting') 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, { + preserveDraft: input.preserveDraft, + }); + if (input.preserveDraft && attempt.text !== input.text) return false; + const result = await input.submit(attempt); + if (!result) return false; + const clearsDraft = input.lease.settle( + attempt.requestId, + workHubSubmissionClearsDraft(result), + ); + if (input.preserveDraft && clearsDraft) { + input.lease.write('workhub', ''); + 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 +221,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 +291,19 @@ export function WorkHubSurface(props: { : turn, )); try { - const result = await submitWorkHubSurfaceInput({ + const result = await submitAndRecordWorkHubSurfaceInput({ controller: props.controller, - input, + request: input, + recordedUserText, + summary: (result) => sendLease.summary( + input.requestId, + () => workHubCoordinationSummary(result, projection, copy), + ), + // The ordinary Session admission may already have settled. A failed + // Coordination summary keeps this send incomplete so the retry + // reuses its durable Action Gate identity before filling the gap. + 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 +312,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 +335,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 +366,7 @@ export function WorkHubSurface(props: { composer={( {}} sendBlocked={pending || !surfaceReady} @@ -345,14 +422,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 +451,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; 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..1a7e4a3b8d 100644 --- a/docs/architecture/workhub-coordination-session-adr.md +++ b/docs/architecture/workhub-coordination-session-adr.md @@ -114,6 +114,46 @@ 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 closed, typed `workhub_coordination` records in the +existing Coordination Session transcript. An immutable `delegation_intent` is +appended before the target Session effect so an opaque candidate remains +recoverable after the candidate set changes or the Runtime Host restarts. A +`delegation_committed` record then binds that intent to the accepted target Turn +and acts as the durable action-replay result. A mutually exclusive +`delegation_abandoned` record spends an identity whose target effect was +definitively rejected; for `create_new`, it also closes a retired deterministic +Session id. The records carry an action fingerprint to reject conflicting reuse +of an action identity. They do not form a general workflow state machine and do +not persist target execution lifecycle. + +The renderer persists the Composer draft and its in-flight action as separate +fields in local storage until both target admission and its Coordination summary +settle. Draft edits cannot revoke an admitted action, and the identity survives a +renderer reload or full application relaunch. The lease is scoped by Coordination +Session, so switching Runtime Hosts cannot move or retire another Host's action +identity. Retry therefore reuses the same identity instead of treating retained +text as new work; `waiting_for_user` does not retire that identity, and the first +generated Coordination summary remains bound to the action across draft changes. +The durable fingerprint covers stable user intent, not snapshot-scoped candidate +ids; once prepared, the intent owns the resolved target, exact user text, and any +`create_new` title/workspace context. + +Recovery accepts the target Session's existing root receipt, pending admission, or +immutable steering proof as durable evidence, and checks that evidence before a +retry submits again. A waiting result is local and retryable: it neither consumes +the action's immutable Coordination summary nor records a false acceptance. A +definitive `create_new` submit rejection compensates through the ordinary +Session-retirement authority. The exact stable create may expose its revision again +only while the Session remains at the initial revision, so an uncertain retirement +can be retried without granting cleanup authority over a subsequently mutated +Session. `not_found` is successful cleanup, while revision, busy, and other +definitive cleanup failures stay local to the action; only an already-uncertain +commit path may request Runtime Host drain. An unknown submit outcome never removes +a possibly admitted Session. +Recovery is deliberately driven by explicit caller retry rather than an autonomous +startup scan: the latter would execute user work without a live request context and +turn this journal into a background workflow engine. + ## Consequences, costs, and reevaluation - WorkHub gains persistent conversational continuity without adding another @@ -125,13 +165,18 @@ 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 two invisible journal messages (intent + plus committed or abandoned) beside its three visible transcript messages. + Message-count page limits therefore retain up to roughly 40% fewer visible + delegated turns than an answer-only Coordination history. - 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..8159dbd8b9 --- /dev/null +++ b/packages/core/src/__tests__/workhub-coordination-record.test.ts @@ -0,0 +1,122 @@ +/* + * 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 exact delegation intent, commit, and abandonment records', () => { + const intent = { + type: 'workhub_coordination', + id: 'intent-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_intent', + actionId: 'action-id', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + userText: 'Continue payment work', + } as const; + const committed = { + ...intent, + id: 'commit-id', + ts: 2, + kind: 'delegation_committed', + delegationId: 'delegation-id', + targetTurnId: 'target-turn', + steered: true, + } as const; + const abandoned = { + ...intent, + id: 'abandoned-id', + ts: 3, + kind: 'delegation_abandoned', + reason: 'target_rejected', + } as const; + + assert.deepEqual(decodeCanonicalMessage(intent), intent); + assert.deepEqual(decodeCanonicalMessage(committed), committed); + assert.deepEqual(decodeCanonicalMessage(abandoned), abandoned); + }); + + test('rejects malformed or widened coordination records', () => { + const base = { + type: 'workhub_coordination', + id: 'intent-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_intent', + actionId: 'action-id', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + } as const; + + for (const invalid of [ + { ...base, coordinationTurnId: 'different-turn' }, + { ...base, actionFingerprint: 'not-a-digest' }, + { ...base, disposition: 'replace' }, + { ...base, userText: undefined }, + { ...base, sourceSessionId: 'injected' }, + { ...base, kind: 'delegation_committed' }, + { ...base, schemaVersion: 2 }, + ]) { + assert.throws(() => decodeCanonicalMessage(invalid), /Invalid stored message schema/u); + } + }); + + test('requires a complete create payload only for create_new intents', () => { + const create = { + type: 'workhub_coordination', + id: 'create-intent-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_intent', + 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' }, + }, + } 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..2724652180 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,61 @@ 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; +} + +/** Durable target choice written before a delegated Session effect is attempted. */ +export interface WorkHubDelegationIntentMessage extends WorkHubCoordinationMessageEnvelope { + kind: 'delegation_intent'; +} + +/** Durable proof that one Coordination action owns one accepted target Turn. */ +export interface WorkHubDelegationCommittedMessage extends WorkHubCoordinationMessageEnvelope { + kind: 'delegation_committed'; + delegationId: string; + targetTurnId: string; + steered?: true; +} + +/** Durable terminal proof that an action cannot be executed or retried. */ +export interface WorkHubDelegationAbandonedMessage extends WorkHubCoordinationMessageEnvelope { + kind: 'delegation_abandoned'; + reason: 'target_rejected' | 'created_session_retired'; +} + +export type WorkHubCoordinationMessage = + | WorkHubDelegationIntentMessage + | WorkHubDelegationCommittedMessage + | WorkHubDelegationAbandonedMessage; + export interface TurnRecord { turnId: string; firstSequence?: number; @@ -1030,6 +1086,72 @@ const TURN_STATE_MESSAGE_SHAPE = defineObjectShape()( 'errorClass', ], ); +const WORKHUB_DELEGATION_INTENT_MESSAGE_SHAPE = defineObjectShape()( + [ + 'type', + 'id', + 'turnId', + 'ts', + 'schemaVersion', + 'kind', + 'actionId', + 'actionFingerprint', + 'coordinationTurnId', + 'targetSessionId', + 'disposition', + 'userText', + ], + ['create'], +); +const WORKHUB_DELEGATION_COMMITTED_MESSAGE_SHAPE = + defineObjectShape()( + [ + 'type', + 'id', + 'turnId', + 'ts', + 'schemaVersion', + 'kind', + 'actionId', + 'actionFingerprint', + 'coordinationTurnId', + 'targetSessionId', + 'disposition', + 'userText', + 'delegationId', + 'targetTurnId', + ], + ['create', 'steered'], + ); +const WORKHUB_DELEGATION_ABANDONED_MESSAGE_SHAPE = + defineObjectShape()( + [ + 'type', + 'id', + 'turnId', + 'ts', + 'schemaVersion', + 'kind', + 'actionId', + 'actionFingerprint', + 'coordinationTurnId', + 'targetSessionId', + 'disposition', + 'userText', + 'reason', + ], + ['create'], + ); +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 +1299,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 +1317,65 @@ 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; + if (message.kind === 'delegation_intent') { + return hasExactShape(message, WORKHUB_DELEGATION_INTENT_MESSAGE_SHAPE); + } + if (message.kind === 'delegation_abandoned') { + return ( + hasExactShape(message, WORKHUB_DELEGATION_ABANDONED_MESSAGE_SHAPE) && + (message.reason === 'target_rejected' || message.reason === 'created_session_retired') + ); + } + return ( + message.kind === 'delegation_committed' && + hasExactShape(message, WORKHUB_DELEGATION_COMMITTED_MESSAGE_SHAPE) && + typeof message.delegationId === 'string' && + typeof message.targetTurnId === 'string' && + (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..f09df85a39 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,6 +546,7 @@ 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. return undefined; diff --git a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts index f0c3d9d372..3e3da205bd 100644 --- a/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts @@ -478,6 +478,57 @@ test('ordinary creation rejects the reserved WorkHub Coordination Session identi assert.equal(fixture.drainRequests(), 0); }); +test('WorkHub creation reports a discard revision for creation and a pristine replay', async () => { + let creates = 0; + const header = sessionHeader('session-1', []); + const fixture = createFixture({ + stores: { + createStableSession: async () => { + creates += 1; + return creates === 1 + ? { kind: 'created', record: headerSnapshot(header, 1) } + : { kind: 'existing', record: headerSnapshot(header, creates === 2 ? 1 : 2) }; + }, + readCatalogRecord: async () => catalogRecord(header, 1), + }, + }); + const input = { + sessionId: fixture.sessionId, + workspace: { kind: 'host_path' as const, path: process.cwd() }, + modelTarget: { kind: 'default' as const }, + }; + + const created = await fixture.coordinator.createForWorkHub(input); + const replayed = await fixture.coordinator.createForWorkHub(input); + const mutated = await fixture.coordinator.createForWorkHub(input); + + assert.equal(created.outcome.ok, true); + assert.equal(created.discardRevision, 1); + assert.equal(replayed.outcome.ok, true); + assert.equal(replayed.discardRevision, 1); + assert.equal(mutated.outcome.ok, true); + assert.equal(mutated.discardRevision, undefined); +}); + +test('WorkHub creation distinguishes a retired deterministic Session identity', async () => { + const fixture = createFixture({ + stores: { + probeStableSessionCreate: async () => ({ kind: 'conflict', reason: 'removed' }), + }, + }); + + const retired = await fixture.coordinator.createForWorkHub({ + sessionId: fixture.sessionId, + workspace: { kind: 'host_path', path: process.cwd() }, + modelTarget: { kind: 'default' }, + }); + + assert.equal(retired.outcome.ok, false); + assert.equal(retired.retired, true); + assert.equal(retired.discardRevision, undefined); + assert.equal(fixture.drainRequests(), 0); +}); + test('ordinary configuration rejects the WorkHub Coordination Session identity', async () => { let reads = 0; const fixture = createFixture({ 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..d5b87ba6ba 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,10 @@ import { WorkHubCoordinationActionGate, type WorkHubActionGateEffects, type WorkHubActionGateSession, + type WorkHubDelegationAbandoned, + type WorkHubDelegationCommit, + type WorkHubDelegationIntent, + type WorkHubDelegationRecord, } from '../server/workhub-coordination-action-gate.js'; import type { ConnectionContext } from '../server/operation-dispatcher.js'; @@ -202,8 +206,10 @@ describe('WorkHub Coordination Action Gate', () => { const first = await gate.act(input, CONTEXT); const replay = await gate.act(input, CONTEXT); + const restartedReplay = await new WorkHubCoordinationActionGate(effects).act(input, CONTEXT); assert.deepEqual(replay, first); + assert.deepEqual(restartedReplay, 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); @@ -216,18 +222,28 @@ describe('WorkHub Coordination Action Gate', () => { 'title', 'workspace', ]); + assert.deepEqual( + await new WorkHubCoordinationActionGate(effects).act( + { + ...input, + proposal: { disposition: 'create_new', title: 'Recomputed title' }, + create: { workspace: { kind: 'project', projectId: 'new-current-project' } }, + }, + CONTEXT, + ), + first, + ); await assert.rejects( - gate.act({ ...input, proposal: { disposition: 'create_new', title: 'Different' } }, CONTEXT), + gate.act({ ...input, userText: 'Create different work' }, CONTEXT), (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', ); assert.equal(effects.creations.length, 1); }); - test('effect rejection grants no root ownership and releases the action identity', async () => { + test('definitive effect rejection durably abandons the action identity', async () => { const effects = fakeEffects([session('ordinary')]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); - const submit = effects.submit; effects.submit = async () => { throw new WorkHubActionEffectFailure('unauthorized', 'Target permission denied'); }; @@ -245,11 +261,39 @@ describe('WorkHub Coordination Action Gate', () => { gate.act(input, CONTEXT), (error) => error instanceof WorkHubActionEffectFailure && error.code === 'unauthorized', ); - effects.submit = submit; - assert.equal((await gate.act(input, CONTEXT)).disposition, 'delegate_existing'); + assert.equal(effects.delegations.get(input.actionId)?.kind as string, 'delegation_abandoned'); + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act(input, CONTEXT), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + assert.deepEqual(effects.submissions, []); + }); + + test('commits a delegation after recovering an unknown submit outcome', async () => { + const effects = fakeEffects([session('ordinary')]); + const gate = new WorkHubCoordinationActionGate(effects); + const snapshot = await gate.candidates(); + effects.submitUnknownAfterAdmission = true; + + const result = await gate.act( + { + actionId: 'unknown-submit-action', + userText: 'Continue ordinary work', + candidateSetId: snapshot.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: snapshot.candidates[0]!.candidateRef, + }, + }, + CONTEXT, + ); + + assert.equal(result.disposition, 'delegate_existing'); + assert.equal(effects.submissions.length, 1); + assert.equal(effects.delegations.get('unknown-submit-action')?.kind, 'delegation_committed'); }); - test('replays an ordinary delegation without submitting twice', async () => { + test('replays an ordinary delegation durably across Action Gate restart', async () => { const effects = fakeEffects([session('ordinary')]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); @@ -264,11 +308,241 @@ describe('WorkHub Coordination Action Gate', () => { }; const first = await gate.act(input, CONTEXT); - const replay = await gate.act(input, CONTEXT); + const replay = await new WorkHubCoordinationActionGate(effects).act(input, CONTEXT); assert.deepEqual(replay, first); assert.equal(effects.submissions.length, 1); assert.equal(effects.submissions[0]?.sessionId, 'ordinary'); + assert.equal(effects.delegations.get(input.actionId)?.kind, 'delegation_committed'); + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act( + { ...input, userText: 'Different work' }, + CONTEXT, + ), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + assert.equal(effects.submissions.length, 1); + }); + + test('resumes a durable intent after restart without re-admitting stale candidates', async () => { + const effects = fakeEffects([session('ordinary')]); + const gate = new WorkHubCoordinationActionGate(effects); + const snapshot = await gate.candidates(); + const input = { + actionId: 'interrupted-delegate-action', + userText: 'Continue ordinary work', + candidateSetId: snapshot.candidateSetId, + proposal: { + disposition: 'delegate_existing' as const, + candidateRef: snapshot.candidates[0]!.candidateRef, + }, + }; + effects.commitFailuresRemaining = 1; + + await assert.rejects( + gate.act(input, CONTEXT), + (error) => + error instanceof WorkHubActionEffectFailure && error.code === 'commit_outcome_unknown', + ); + assert.equal(effects.delegations.get(input.actionId)?.kind, 'delegation_intent'); + assert.equal(effects.submissions.length, 1); + + effects.sessions[0] = session('ordinary', { statusUpdatedAt: 99 }); + const restarted = new WorkHubCoordinationActionGate(effects); + const refreshed = await restarted.candidates(); + await assert.rejects( + restarted.act( + { + actionId: input.actionId, + userText: input.userText, + proposal: { disposition: 'answer_here' }, + }, + CONTEXT, + ), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + const recovered = await restarted.act( + { + ...input, + candidateSetId: refreshed.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: refreshed.candidates[0]!.candidateRef, + }, + }, + CONTEXT, + ); + + assert.equal(recovered.disposition, 'delegate_existing'); + assert.equal(effects.submissions.length, 1); + assert.equal(effects.delegations.get(input.actionId)?.kind, 'delegation_committed'); + }); + + test('resumes create_new from the durable payload instead of recomputed caller context', async () => { + const effects = fakeEffects([session('ordinary')]); + const input = { + actionId: 'interrupted-create-action', + userText: 'Create a login audit', + proposal: { disposition: 'create_new' as const, title: 'Login audit' }, + create: { workspace: { kind: 'project' as const, projectId: 'original-project' } }, + }; + effects.commitFailuresRemaining = 1; + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act(input, CONTEXT), + (error) => + error instanceof WorkHubActionEffectFailure && error.code === 'commit_outcome_unknown', + ); + const recovered = await new WorkHubCoordinationActionGate(effects).act( + { + ...input, + proposal: { disposition: 'create_new', title: 'Recomputed title' }, + create: { workspace: { kind: 'project', projectId: 'new-current-project' } }, + }, + CONTEXT, + ); + + assert.equal(recovered.disposition, 'create_new'); + assert.deepEqual(effects.creations, [ + { + sessionId: effects.creations[0]!.sessionId, + workspace: { kind: 'project', projectId: 'original-project' }, + title: 'Login audit', + }, + ]); + assert.equal(effects.submissions.length, 1); + }); + + test('definitive create_new submit rejection retires the empty created Session', async () => { + const effects = fakeEffects([session('ordinary')]); + effects.submitFailure = new WorkHubActionEffectFailure( + 'operation_conflict', + 'Target submit was definitively rejected', + ); + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act( + { + actionId: 'rejected-create-action', + userText: 'Create a login audit', + proposal: { disposition: 'create_new', title: 'Login audit' }, + create: { workspace: { kind: 'project', projectId: 'project-1' } }, + }, + CONTEXT, + ), + /definitively rejected/u, + ); + + assert.equal(effects.discardedCreatedSessionIds.length, 1); + assert.match(effects.discardedCreatedSessionIds[0] ?? '', /^whs_[a-f0-9]{48}$/u); + assert.deepEqual(effects.creations, []); + assert.equal( + effects.delegations.get('rejected-create-action')?.kind as string, + 'delegation_abandoned', + ); + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act( + { + actionId: 'rejected-create-action', + userText: 'Create a login audit', + proposal: { disposition: 'create_new', title: 'Login audit' }, + create: { workspace: { kind: 'project', projectId: 'project-1' } }, + }, + CONTEXT, + ), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + assert.equal(effects.createAttempts, 1); + assert.equal(effects.submissions.length, 0); + }); + + test('unknown create_new submit outcome never retires a possibly admitted Session', async () => { + const effects = fakeEffects([session('ordinary')]); + effects.submitUnknownAfterAdmission = true; + effects.recoverSubmissionMiss = true; + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act( + { + actionId: 'unknown-create-action', + userText: 'Create a login audit', + proposal: { disposition: 'create_new', title: 'Login audit' }, + create: { workspace: { kind: 'project', projectId: 'project-1' } }, + }, + CONTEXT, + ), + (error) => + error instanceof WorkHubActionEffectFailure && error.code === 'commit_outcome_unknown', + ); + + assert.equal(effects.creations.length, 1); + assert.deepEqual(effects.discardedCreatedSessionIds, []); + }); + + test('retries definitive create_new cleanup after the first discard outcome is unknown', async () => { + const effects = fakeEffects([session('ordinary')]); + const input = { + actionId: 'retry-create-cleanup-action', + userText: 'Create a login audit', + proposal: { disposition: 'create_new' as const, title: 'Login audit' }, + create: { workspace: { kind: 'project' as const, projectId: 'project-1' } }, + }; + effects.submitFailure = new WorkHubActionEffectFailure( + 'operation_conflict', + 'Target submit was definitively rejected', + ); + effects.discardFailuresRemaining = 1; + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act(input, CONTEXT), + (error) => + error instanceof WorkHubActionEffectFailure && error.code === 'commit_outcome_unknown', + ); + assert.equal(effects.creations.length, 1); + + effects.submitFailure = new WorkHubActionEffectFailure( + 'operation_conflict', + 'Target submit was definitively rejected again', + ); + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act(input, CONTEXT), + /definitively rejected again/u, + ); + + assert.equal(effects.discardAttempts, 2); + assert.deepEqual(effects.creations, []); + }); + + test('retires an action when an unknown cleanup actually left a Session tombstone', async () => { + const effects = fakeEffects([session('ordinary')]); + const input = { + actionId: 'tombstoned-cleanup-action', + userText: 'Create a login audit', + proposal: { disposition: 'create_new' as const, title: 'Login audit' }, + create: { workspace: { kind: 'project' as const, projectId: 'project-1' } }, + }; + effects.submitFailure = new WorkHubActionEffectFailure( + 'operation_conflict', + 'Target submit was definitively rejected', + ); + effects.discardUnknownAfterRemoval = true; + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act(input, CONTEXT), + (error) => + error instanceof WorkHubActionEffectFailure && error.code === 'commit_outcome_unknown', + ); + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act(input, CONTEXT), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + + assert.equal(effects.delegations.get(input.actionId)?.kind, 'delegation_abandoned'); + assert.equal(effects.createAttempts, 2); + assert.equal(effects.discardAttempts, 1); + assert.deepEqual(effects.creations, []); }); }); @@ -290,6 +564,13 @@ function session( } function fakeEffects(initialSessions: WorkHubActionGateSession[]) { + const submitted = new Map< + string, + { + readonly input: { sessionId: string; messageId: string; text: string }; + readonly turnId: string; + } + >(); const state = { sessions: [...initialSessions], answers: [] as Array<{ turnId: string; text: string }>, @@ -304,6 +585,17 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { title: string; }>, submissions: [] as Array<{ sessionId: string; messageId: string; text: string }>, + delegations: new Map(), + commitFailuresRemaining: 0, + submitUnknownAfterAdmission: false as boolean, + submitFailure: undefined as WorkHubActionEffectFailure | undefined, + recoverSubmissionMiss: false as boolean, + discardAttempts: 0, + discardFailuresRemaining: 0, + discardUnknownAfterRemoval: false as boolean, + discardedCreatedSessionIds: [] as string[], + retiredCreatedSessionIds: new Set(), + createAttempts: 0, async listSessions() { return this.sessions; }, @@ -318,11 +610,98 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { workspace: { kind: 'project'; projectId: string } | { kind: 'host_path'; path: string }; title: string; }) { - this.creations.push(input); + this.createAttempts += 1; + if (this.retiredCreatedSessionIds.has(input.sessionId)) { + return { kind: 'retired' as const }; + } + const existing = this.creations.find(({ sessionId }) => sessionId === input.sessionId); + if (existing) assert.deepEqual(existing, input); + else this.creations.push(input); + return { kind: 'available' as const, discardRevision: 1 }; }, async submit(input: { sessionId: string; messageId: string; text: string }) { + if (this.submitFailure) { + const error = this.submitFailure; + this.submitFailure = undefined; + throw error; + } + const existing = submitted.get(input.messageId); + if (existing) { + assert.deepEqual(existing.input, input); + return { turnId: existing.turnId }; + } this.submissions.push(input); - return { turnId: `turn-${input.sessionId}` }; + const turnId = `turn-${input.sessionId}`; + submitted.set(input.messageId, { input, turnId }); + if (this.submitUnknownAfterAdmission) { + this.submitUnknownAfterAdmission = false; + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'Target submit outcome is unknown', + ); + } + return { turnId }; + }, + async discardCreated(input: { sessionId: string; expectedRevision: number }) { + assert.equal(input.expectedRevision, 1); + this.discardAttempts += 1; + if (this.discardUnknownAfterRemoval) { + this.discardUnknownAfterRemoval = false; + this.discardedCreatedSessionIds.push(input.sessionId); + this.retiredCreatedSessionIds.add(input.sessionId); + const index = this.creations.findIndex(({ sessionId }) => sessionId === input.sessionId); + if (index >= 0) this.creations.splice(index, 1); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'Created Session retirement outcome is unknown', + ); + } + if (this.discardFailuresRemaining > 0) { + this.discardFailuresRemaining -= 1; + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'Created Session retirement outcome is unknown', + ); + } + this.discardedCreatedSessionIds.push(input.sessionId); + this.retiredCreatedSessionIds.add(input.sessionId); + const index = this.creations.findIndex(({ sessionId }) => sessionId === input.sessionId); + if (index >= 0) this.creations.splice(index, 1); + }, + async recoverSubmission(input: { sessionId: string; messageId: string; text: string }) { + if (this.recoverSubmissionMiss) return undefined; + const existing = submitted.get(input.messageId); + if (!existing) return undefined; + assert.deepEqual(existing.input, input); + return { turnId: existing.turnId }; + }, + async readDelegation(actionId: string) { + return this.delegations.get(actionId); + }, + async prepareDelegation(intent: WorkHubDelegationIntent) { + const existing = this.delegations.get(intent.actionId); + if (existing) { + assert.deepEqual(existing, intent); + return; + } + this.delegations.set(intent.actionId, intent); + }, + async commitDelegation(commit: WorkHubDelegationCommit) { + const existing = this.delegations.get(commit.actionId); + assert.equal(existing?.kind, 'delegation_intent'); + if (this.commitFailuresRemaining > 0) { + this.commitFailuresRemaining -= 1; + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'Delegation commit outcome is unknown', + ); + } + this.delegations.set(commit.actionId, commit); + }, + async abandonDelegation(abandoned: WorkHubDelegationAbandoned) { + const existing = this.delegations.get(abandoned.actionId); + assert.equal(existing?.kind, 'delegation_intent'); + this.delegations.set(abandoned.actionId, abandoned); }, } satisfies WorkHubActionGateEffects & { sessions: WorkHubActionGateSession[]; @@ -334,6 +713,17 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { title: string; }>; submissions: Array<{ sessionId: string; messageId: string; text: string }>; + delegations: Map; + commitFailuresRemaining: number; + submitUnknownAfterAdmission: boolean; + submitFailure: WorkHubActionEffectFailure | undefined; + recoverSubmissionMiss: boolean; + discardAttempts: number; + discardFailuresRemaining: number; + discardUnknownAfterRemoval: boolean; + discardedCreatedSessionIds: string[]; + retiredCreatedSessionIds: Set; + createAttempts: number; }; 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..c609ae9d18 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts @@ -38,6 +38,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 +481,109 @@ 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 submissions: Array<{ sessionId: string; messageId: string; text: string }> = []; + const first = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + create: async () => ({ kind: 'available' }), + discardCreated: async () => undefined, + submit: async (input) => { + submissions.push(input); + return { turnId: 'payments-turn' }; + }, + recoverSubmission: async () => undefined, + }); + 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_intent', + actionId: 'payments-action', + targetSessionId: candidates.result.candidates[0]!.sessionId, + }, + { + kind: 'delegation_committed', + actionId: 'payments-action', + targetSessionId: candidates.result.candidates[0]!.sessionId, + }, + ], + ); + assert.equal(submissions.length, 1); + } finally { + await store.close?.(); + } + + store = createSessionStore(root); + try { + const restarted = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + create: async () => assert.fail('durable replay must not create a Session'), + discardCreated: async () => assert.fail('durable replay must not discard a Session'), + submit: async () => assert.fail('durable replay must not submit another Turn'), + recoverSubmission: async () => + assert.fail('durable replay must not recover an already committed 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 +718,15 @@ function coordinator( hasRootTurnAdmission: async () => false, }, admission: SessionAdmissionGate = new SessionAdmissionGate(), + sessionActions: Pick< + WorkHubActionGateEffects, + 'create' | 'discardCreated' | 'submit' | 'recoverSubmission' + > = { + create: async () => ({ kind: 'available' }), + discardCreated: async () => undefined, + submit: async ({ sessionId }) => ({ turnId: `turn-${sessionId}` }), + recoverSubmission: async () => undefined, + }, ) { return new HostWorkHubCoordinationCoordinator({ stateRoot: root, @@ -621,10 +734,7 @@ function coordinator( admission, continuity: { refreshCanonical: async () => undefined }, executions, - sessionActions: { - create: async () => undefined, - submit: async ({ sessionId }) => ({ turnId: `turn-${sessionId}` }), - }, + sessionActions, resolveCreateTarget: resolveCreateTarget ?? (async () => ({ diff --git a/packages/runtime-host/src/__tests__/workhub-target-submission-recovery.test.ts b/packages/runtime-host/src/__tests__/workhub-target-submission-recovery.test.ts new file mode 100644 index 0000000000..5cfe00820a --- /dev/null +++ b/packages/runtime-host/src/__tests__/workhub-target-submission-recovery.test.ts @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { messageContentDigest, normalizeMessageContent } from '@maka/core/events'; +import type { RuntimeEvent } from '@maka/core/runtime-event'; +import { ROOT_TURN_ADMISSION_SCHEMA_VERSION } from '@maka/storage/agent-run-store'; +import type { WorkHubSubmissionRecoveryStores } from '../server/workhub-target-submission-recovery.js'; +import { recoverWorkHubTargetSubmission } from '../server/workhub-target-submission-recovery.js'; + +test('recovers a handed-off steering submission from its immutable proof', async () => { + const text = 'Continue payment work'; + const event = { + id: 'steering-event', + invocationId: 'invocation-1', + runId: 'run-1', + sessionId: 'payment', + turnId: 'turn-1', + ts: 1, + partial: false, + role: 'user', + author: 'user', + content: { kind: 'text', text, steering: true }, + refs: { providerEventId: 'message-1' }, + } satisfies RuntimeEvent; + const stores = { + readRootTurnSourceMessageReceipt: async () => undefined, + readMessageAdmission: async () => undefined, + readRootTurnAdmission: async () => undefined, + readImmutableSteeringMessageProof: async () => ({ event }), + } satisfies WorkHubSubmissionRecoveryStores; + + assert.deepEqual( + await recoverWorkHubTargetSubmission(stores, { + sessionId: 'payment', + messageId: 'message-1', + text, + }), + { turnId: 'turn-1', steered: true }, + ); +}); + +test('recovers only a matching pending steering admission', async () => { + const text = 'Continue payment work'; + const content = normalizeMessageContent({ text }); + const admission = { + sessionId: 'payment', + turnId: 'turn-1', + runId: 'run-1', + messageId: 'message-1', + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn' as const, + placement: 'current_turn' as const, + disposition: 'steering' as const, + admittedAt: 1, + }; + const stores = { + readRootTurnSourceMessageReceipt: async () => undefined, + readMessageAdmission: async () => admission, + readRootTurnAdmission: async () => ({ + schemaVersion: ROOT_TURN_ADMISSION_SCHEMA_VERSION, + sessionId: 'payment', + turnId: 'turn-1', + runId: 'run-1', + userMessageId: 'message-1', + execution: { kind: 'external_message' }, + previousRootTurnId: null, + normalizedInput: content, + sourceMessages: [], + admittedAt: 1, + }), + readImmutableSteeringMessageProof: async () => undefined, + } satisfies WorkHubSubmissionRecoveryStores; + + assert.deepEqual( + await recoverWorkHubTargetSubmission(stores, { + sessionId: 'payment', + messageId: 'message-1', + text, + }), + { turnId: 'turn-1', steered: true }, + ); + assert.equal( + await recoverWorkHubTargetSubmission( + { ...stores, readMessageAdmission: async () => ({ ...admission, disposition: 'followup' }) }, + { sessionId: 'payment', messageId: 'message-1', text }, + ), + undefined, + ); +}); diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 51b9dfccbf..b260405db6 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -92,7 +92,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 53 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 54 as const; +// 54: WorkHub stores strict durable delegation intent, commit, and abandonment +// records. Older peers cannot decode these messages during transcript recovery. // 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 cf46c22b9c..8e7d512b9d 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -88,6 +88,7 @@ import { createHostChildAgentToolComposition, } from './child-agent-composition.js'; import { HostCanonicalPermissionOutcomeReader } from './canonical-permission-outcome-reader.js'; +import { recoverWorkHubTargetSubmission } from './workhub-target-submission-recovery.js'; import { HostArtifactCoordinator } from './artifact-coordinator.js'; import { HostAgentGraphCoordinator } from './agent-graph-coordinator.js'; import { HostAgentGraphExecutionCoordinator } from './agent-graph-execution-coordinator.js'; @@ -468,6 +469,7 @@ export async function createExecutionRuntimeHostComposition( let goal: HostGoalCoordinator | undefined; let deepResearch: HostDeepResearchCoordinator | undefined; let dailyReview: HostDailyReviewCoordinator | undefined; + let sessionRetirement: HostSessionRetirementCoordinator | undefined; const rootPort: HostMessageRootPort = { readSessionHeader: (sessionId) => requireRootCoordinator(rootCoordinator).readSessionHeader(sessionId), @@ -1230,7 +1232,7 @@ export async function createExecutionRuntimeHostComposition( executions: coordinator, sessionActions: { create: async (input) => { - const outcome = await sessionCatalog.createForWorkHub({ + const created = await sessionCatalog.createForWorkHub({ sessionId: input.sessionId, workspace: input.workspace, name: input.title, @@ -1238,12 +1240,38 @@ export async function createExecutionRuntimeHostComposition( collaborationMode: 'agent', orchestrationMode: 'default', }); + const { outcome } = created; + if (created.retired) return { kind: 'retired' as const }; if (!outcome.ok) { throw new WorkHubActionEffectFailure( outcome.error.code === 'invalid_request' ? 'operation_conflict' : outcome.error.code, outcome.error.message, ); } + return created.discardRevision === undefined + ? { kind: 'available' as const } + : { kind: 'available' as const, discardRevision: created.discardRevision }; + }, + discardCreated: async (input, connection) => { + const outcome = await requireSessionRetirement(sessionRetirement).handlers[ + 'session.remove' + ]( + { + sessionId: input.sessionId, + expectedRevision: input.expectedRevision, + }, + connection, + ); + if (!outcome.ok) { + if (outcome.error.code === 'not_found') return; + throw new WorkHubActionEffectFailure(outcome.error.code, outcome.error.message); + } + if (outcome.result.kind !== 'removed') { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub empty created Session changed before retirement', + ); + } }, submit: async (input, connection) => { const outcome = await messages.handlers['turn.message.submit']( @@ -1264,9 +1292,39 @@ export async function createExecutionRuntimeHostComposition( outcome.error.message, ); } - return outcome.result.disposition === 'turn_started' - ? { turnId: outcome.result.turnId } - : { turnId: input.messageId, steered: true as const }; + if (outcome.result.disposition === 'turn_started') { + return { turnId: outcome.result.turnId }; + } + try { + const admission = await stores.sessionStore.readMessageAdmission( + input.sessionId, + input.messageId, + ); + if (admission) return { turnId: admission.turnId, steered: true as const }; + } catch { + // The submit already settled; losing its exact Turn identity makes + // the WorkHub linkage outcome uncertain rather than retryable. + } + context.requestDrain(); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'WorkHub target Turn identity could not be proven', + ); + }, + recoverSubmission: async (input) => { + return recoverWorkHubTargetSubmission( + { + readRootTurnSourceMessageReceipt: (sessionId, messageId) => + stores.agentRunStore.readRootTurnSourceMessageReceipt(sessionId, messageId), + readMessageAdmission: (sessionId, messageId) => + stores.sessionStore.readMessageAdmission(sessionId, messageId), + readRootTurnAdmission: (sessionId, turnId) => + stores.agentRunStore.readRootTurnAdmission(sessionId, turnId), + readImmutableSteeringMessageProof: (sessionId, messageId) => + stores.runtimeEventStore.readImmutableSteeringMessageProof(sessionId, messageId), + }, + input, + ); }, }, resolveCreateTarget: async () => { @@ -1339,7 +1397,7 @@ export async function createExecutionRuntimeHostComposition( isSessionActive: (sessionId) => coordinator.readRootState(sessionId).kind !== 'idle', requestDrain: context.requestDrain, }); - const sessionRetirement = new HostSessionRetirementCoordinator({ + sessionRetirement = new HostSessionRetirementCoordinator({ stores: stores.sessionStore, admission: sessionAdmission, root: coordinator, @@ -1722,6 +1780,13 @@ function requireRootCoordinator(coordinator: RootTurnCoordinator | undefined): R return coordinator; } +function requireSessionRetirement( + coordinator: HostSessionRetirementCoordinator | undefined, +): HostSessionRetirementCoordinator { + if (!coordinator) throw new Error('Session retirement authority is unavailable'); + return coordinator; +} + function requireWorkspaceExecution( composition: RuntimeHostWorkspaceExecutionComposition | undefined, ): RuntimeHostWorkspaceExecutionComposition { diff --git a/packages/runtime-host/src/server/session-catalog-coordinator.ts b/packages/runtime-host/src/server/session-catalog-coordinator.ts index 01f7556259..88fea1b3c4 100644 --- a/packages/runtime-host/src/server/session-catalog-coordinator.ts +++ b/packages/runtime-host/src/server/session-catalog-coordinator.ts @@ -148,6 +148,10 @@ interface ResolvedSessionModel { readonly model: string; } +// Stable Session creation owns revision 1; any later metadata or execution +// mutation advances it and therefore revokes WorkHub's empty-Session cleanup. +const STABLE_SESSION_INITIAL_REVISION = 1; + /** Host-owned Session catalog, creation, and configuration authority. */ export class HostSessionCatalogCoordinator { readonly handlers: SessionCatalogOperationHandlerMap = { @@ -200,8 +204,29 @@ export class HostSessionCatalogCoordinator { } /** WorkHub Action Gate path; callers cannot bypass the typed operation outcome. */ - createForWorkHub(input: SessionCreateInput): Promise> { - return this.#create(input); + async createForWorkHub(input: SessionCreateInput): Promise<{ + readonly outcome: OperationOutcome<'session.create'>; + readonly discardRevision?: number; + readonly retired?: true; + }> { + let discardRevision: number | undefined; + let retired: true | undefined; + const rememberDiscardRevision = (revision: number) => { + discardRevision = revision; + }; + const outcome = await this.#create( + input, + rememberDiscardRevision, + rememberDiscardRevision, + () => { + retired = true; + }, + ); + return { + outcome, + ...(discardRevision === undefined ? {} : { discardRevision }), + ...(retired ? { retired } : {}), + }; } async #query( @@ -336,7 +361,12 @@ export class HostSessionCatalogCoordinator { } } - async #create(input: SessionCreateInput): Promise> { + async #create( + input: SessionCreateInput, + onCreated?: (revision: number) => void, + onPristineReplay?: (revision: number) => void, + onRetiredReplay?: () => void, + ): Promise> { if (isWorkHubCoordinationSessionId(input.sessionId)) { return createFailure( 'operation_conflict', @@ -359,11 +389,15 @@ export class HostSessionCatalogCoordinator { requestFingerprint, ); if (probe.kind === 'existing') { + if (probe.record.revision === STABLE_SESSION_INITIAL_REVISION) { + onPristineReplay?.(probe.record.revision); + } return createSuccess( projectSessionCatalogRecord(await this.#stores.readCatalogRecord(input.sessionId)), ); } if (probe.kind === 'conflict') { + if (probe.reason === 'removed') onRetiredReplay?.(); return createFailure( 'operation_conflict', 'Session identity belongs to a different create request', @@ -396,11 +430,19 @@ export class HostSessionCatalogCoordinator { input: createInput, }); if (result.kind === 'conflict') { + if (result.reason === 'removed') onRetiredReplay?.(); return createFailure( 'operation_conflict', 'Session identity belongs to a different create request', ); } + if (result.kind === 'created') onCreated?.(result.record.revision); + else if ( + result.kind === 'existing' && + result.record.revision === STABLE_SESSION_INITIAL_REVISION + ) { + onPristineReplay?.(result.record.revision); + } await this.#continuity.refreshCanonical(input.sessionId, lease); return createSuccess( projectSessionCatalogRecord(await this.#stores.readCatalogRecord(input.sessionId)), 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..8f8446e53d 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, + WorkHubDelegationAbandonedMessage, + WorkHubDelegationCommittedMessage, + WorkHubDelegationIntentMessage, +} from '@maka/core/session'; import { WORKHUB_COORDINATION_SESSION_ID, isWorkHubCoordinationSessionTarget, @@ -68,7 +74,16 @@ export interface WorkHubActionGateEffects { readonly sessionId: string; readonly workspace: WorkspaceTarget; readonly title: string; - }): Promise; + }): Promise< + { readonly kind: 'available'; readonly discardRevision?: number } | { readonly kind: 'retired' } + >; + discardCreated( + input: { + readonly sessionId: string; + readonly expectedRevision: number; + }, + context: ConnectionContext, + ): Promise; submit( input: { readonly sessionId: string; @@ -77,8 +92,39 @@ export interface WorkHubActionGateEffects { }, context: ConnectionContext, ): Promise<{ readonly turnId: string; readonly steered?: true }>; + recoverSubmission(input: { + readonly sessionId: string; + readonly messageId: string; + readonly text: string; + }): Promise<{ readonly turnId: string; readonly steered?: true } | undefined>; + readDelegation(actionId: string): Promise; + prepareDelegation(intent: WorkHubDelegationIntent): Promise; + commitDelegation(commit: WorkHubDelegationCommit): Promise; + abandonDelegation(abandoned: WorkHubDelegationAbandoned): Promise; } +type StoredDelegationEnvelopeKeys = 'type' | 'id' | 'turnId' | 'ts' | 'schemaVersion'; + +export type WorkHubDelegationIntent = Omit< + WorkHubDelegationIntentMessage, + StoredDelegationEnvelopeKeys +>; + +export type WorkHubDelegationCommit = Omit< + WorkHubDelegationCommittedMessage, + StoredDelegationEnvelopeKeys +>; + +export type WorkHubDelegationAbandoned = Omit< + WorkHubDelegationAbandonedMessage, + StoredDelegationEnvelopeKeys +>; + +export type WorkHubDelegationRecord = + | WorkHubDelegationIntent + | WorkHubDelegationCommit + | WorkHubDelegationAbandoned; + export type WorkHubActionEffectFailureCode = | 'host_not_ready' | 'host_draining' @@ -147,7 +193,17 @@ 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 replay = this.#actions.get(input.actionId); if (replay) { if (replay.fingerprint !== fingerprint) { @@ -161,12 +217,12 @@ export class WorkHubCoordinationActionGate { return replay.result; } - const result = this.#act(input, context); + const result = this.#act(input, fingerprint, context); const action = { fingerprint, 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 leave the + // in-memory slot so a pre-intent admission can retry; once an intent is + // durable, the journal independently keeps that action identity owned. void result.catch(() => { if (this.#actions.get(input.actionId) === action) { this.#actions.delete(input.actionId); @@ -178,9 +234,26 @@ export class WorkHubCoordinationActionGate { async #act( input: WorkHubCoordinationActInput, + fingerprint: `sha256:${string}`, context: ConnectionContext, ): Promise { const proposal = input.proposal; + const durable = await this.#effects.readDelegation(input.actionId); + if (durable) { + if (durable.actionFingerprint !== fingerprint) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub action identity belongs to a different proposal', + ); + } + if (durable.kind === 'delegation_committed') { + return committedResult(durable); + } + if (durable.kind === 'delegation_abandoned') { + throw abandonedAction(); + } + return this.#executeDelegation(durable, context); + } if (proposal.disposition === 'answer_here') { const turnId = coordinationTurnId(input.actionId, 'answer'); await this.#effects.answer({ turnId, text: input.userText }, context); @@ -203,20 +276,9 @@ 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, - }, - context, - ); - return executionResult('create_new', sessionId, submitted); + const intent = delegationIntent(input, fingerprint, sessionId); + await this.#effects.prepareDelegation(intent); + return this.#executeDelegation(intent, context); } const candidates = await this.candidates(); @@ -237,23 +299,102 @@ export class WorkHubCoordinationActionGate { } this.#assertTarget(target); - return this.#submitExisting(input, target, context); + const intent = delegationIntent(input, fingerprint, target.sessionId); + await this.#effects.prepareDelegation(intent); + return this.#executeDelegation(intent, context); } - async #submitExisting( - input: WorkHubCoordinationActInput, - target: WorkHubCoordinationCandidate, + async #executeDelegation( + intent: WorkHubDelegationIntent, 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); + let discardRevision: number | undefined; + if (intent.disposition === 'create_new') { + if (!intent.create) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub durable creation intent is incomplete', + ); + } + const created = await this.#effects.create({ + sessionId: intent.targetSessionId, + workspace: intent.create.workspace, + title: intent.create.title, + }); + if (created.kind === 'retired') { + await this.#abandonDelegation(intent, 'created_session_retired'); + throw abandonedAction(); + } + discardRevision = created.discardRevision; + } else if (intent.create) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub durable delegation intent contains creation context', + ); + } + + const message = { + sessionId: intent.targetSessionId, + messageId: actionMessageId(intent.actionId), + text: intent.userText, + }; + let submitted = await this.#effects.recoverSubmission(message); + if (!submitted) { + try { + submitted = await this.#effects.submit(message, context); + } catch (error) { + if (!(error instanceof WorkHubActionEffectFailure)) throw error; + if (isDefinitiveSubmissionFailure(error.code)) { + if (discardRevision !== undefined) { + try { + await this.#effects.discardCreated( + { + sessionId: intent.targetSessionId, + expectedRevision: discardRevision, + }, + context, + ); + } catch (cleanupError) { + if ( + !(cleanupError instanceof WorkHubActionEffectFailure) || + cleanupError.code === 'commit_outcome_unknown' + ) { + throw cleanupError; + } + // The target effect was definitively rejected. Cleanup may be + // unnecessary or conflict with later user changes, but that + // must not leave the action executable again. + } + } + await this.#abandonDelegation(intent, 'target_rejected'); + throw error; + } + if (error.code !== 'commit_outcome_unknown') throw error; + const recovered = await this.#effects.recoverSubmission(message); + if (!recovered) throw error; + submitted = recovered; + } + } + const commit: WorkHubDelegationCommit = { + ...intent, + kind: 'delegation_committed', + delegationId: delegationId(intent.actionId), + targetTurnId: submitted.turnId, + ...(submitted.steered ? { steered: true as const } : {}), + }; + await this.#effects.commitDelegation(commit); + return committedResult(commit); + } + + async #abandonDelegation( + intent: WorkHubDelegationIntent, + reason: WorkHubDelegationAbandoned['reason'], + ): Promise { + await this.#effects.abandonDelegation({ + ...intent, + kind: 'delegation_abandoned', + reason, + }); } #assertTarget(target: WorkHubCoordinationCandidate): void { @@ -277,6 +418,19 @@ export class WorkHubCoordinationActionGate { } } +function abandonedAction(): WorkHubActionGateFailure { + return new WorkHubActionGateFailure('action_conflict', 'WorkHub action is permanently abandoned'); +} + +function isDefinitiveSubmissionFailure(code: WorkHubActionEffectFailureCode): boolean { + return ( + code === 'not_found' || + code === 'session_archived' || + code === 'operation_conflict' || + code === 'unauthorized' + ); +} + export function candidateSet( sessions: readonly WorkHubActionGateSession[], ): WorkHubCoordinationCandidatesResult { @@ -328,6 +482,50 @@ function actionMessageId(actionId: string): string { return `whm_${hash(actionId).slice(0, 48)}`; } +function delegationId(actionId: string): string { + return `whd_${hash(`delegation\0${actionId}`).slice(0, 48)}`; +} + +function delegationIntent( + input: WorkHubCoordinationActInput, + actionFingerprint: `sha256:${string}`, + targetSessionId: string, +): WorkHubDelegationIntent { + 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 = { + kind: 'delegation_intent', + actionId: input.actionId, + actionFingerprint, + coordinationTurnId: input.actionId, + targetSessionId, + 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 { return `whs_${hash(`create\0${actionId}`).slice(0, 48)}`; } @@ -350,16 +548,12 @@ 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 { +function committedResult(commit: WorkHubDelegationCommit): WorkHubCoordinationActResult { return { - disposition, - targetSessionId: sessionId, - targetTurnId: submitted.turnId, - ...(submitted.steered ? { steered: true as const } : {}), + disposition: commit.disposition, + targetSessionId: commit.targetSessionId, + targetTurnId: commit.targetTurnId, + ...(commit.steered ? { steered: true as const } : {}), } as WorkHubCoordinationActResult; } @@ -367,6 +561,16 @@ 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 === 'clarify' + ? { assistantText: input.proposal.assistantText } + : {}), + }); +} + 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..72886effa9 100644 --- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts +++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts @@ -55,6 +55,7 @@ import { WorkHubCoordinationActionGate, type WorkHubActionGateEffects, } from './workhub-coordination-action-gate.js'; +import { WorkHubDelegationJournal } from './workhub-delegation-journal.js'; const CREATE_FINGERPRINT = `sha256:${createHash('sha256') .update('maka:workhub-coordination-session:v1', 'utf8') @@ -100,7 +101,10 @@ export interface HostWorkHubCoordinationCoordinatorOptions { readonly admission: SessionAdmissionGate; readonly continuity: Pick; readonly executions: CoordinationExecutions; - readonly sessionActions: Pick; + readonly sessionActions: Pick< + WorkHubActionGateEffects, + 'create' | 'discardCreated' | 'submit' | 'recoverSubmission' + >; readonly resolveCreateTarget: () => Promise; readonly requestDrain: () => void; } @@ -123,6 +127,7 @@ export class HostWorkHubCoordinationCoordinator { readonly #resolveCreateTarget: () => Promise; readonly #requestDrain: () => void; readonly #actionGate: WorkHubCoordinationActionGate; + readonly #delegations: WorkHubDelegationJournal; constructor(options: HostWorkHubCoordinationCoordinatorOptions) { this.#coordinationCwd = join(options.stateRoot, COORDINATION_CWD_DIRECTORY); @@ -132,6 +137,12 @@ export class HostWorkHubCoordinationCoordinator { this.#executions = options.executions; this.#resolveCreateTarget = options.resolveCreateTarget; this.#requestDrain = options.requestDrain; + this.#delegations = new WorkHubDelegationJournal({ + stores: options.stores, + admission: options.admission, + continuity: options.continuity, + requestDrain: options.requestDrain, + }); this.#actionGate = new WorkHubCoordinationActionGate({ listSessions: () => this.#stores.listHeaders(), answer: async (input, context) => { @@ -151,7 +162,13 @@ export class HostWorkHubCoordinationCoordinator { } }, create: options.sessionActions.create, + discardCreated: options.sessionActions.discardCreated, submit: options.sessionActions.submit, + recoverSubmission: options.sessionActions.recoverSubmission, + readDelegation: (actionId) => this.#delegations.read(actionId), + prepareDelegation: (intent) => this.#delegations.prepare(intent), + commitDelegation: (commit) => this.#delegations.commit(commit), + abandonDelegation: (abandoned) => this.#delegations.abandon(abandoned), }); } diff --git a/packages/runtime-host/src/server/workhub-delegation-journal.ts b/packages/runtime-host/src/server/workhub-delegation-journal.ts new file mode 100644 index 0000000000..7db3985648 --- /dev/null +++ b/packages/runtime-host/src/server/workhub-delegation-journal.ts @@ -0,0 +1,389 @@ +/* + * 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 { createHash } from 'node:crypto'; +import { isDeepStrictEqual } from 'node:util'; +import { + WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION, + WORKHUB_COORDINATION_SESSION_ID, + isWorkHubCoordinationSession, + type StoredMessage, + type WorkHubDelegationAbandonedMessage, + type WorkHubDelegationCommittedMessage, + type WorkHubDelegationIntentMessage, +} from '@maka/core/session'; +import type { SessionAuthorityStore } from '@maka/storage/session-store'; +import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js'; +import { + WorkHubActionEffectFailure, + type WorkHubDelegationAbandoned, + type WorkHubDelegationCommit, + type WorkHubDelegationIntent, + type WorkHubDelegationRecord, +} from './workhub-coordination-action-gate.js'; +import type { SessionAdmissionGate } from './session-admission-gate.js'; + +const RECORD_KINDS = ['delegation_intent', 'delegation_committed', 'delegation_abandoned'] as const; +// Two records may each repeat the bounded 48 KiB request plus create context; +// JSON escaping can expand one input byte to six encoded bytes. +const RECORD_READ_MAX_BYTES = 768 * 1024; + +type JournalStores = Pick< + SessionAuthorityStore, + | 'appendMessages' + | 'readHeaderSnapshot' + | 'readTranscriptHighWaterSnapshot' + | 'readTranscriptMessagesSnapshot' +>; + +export interface WorkHubDelegationJournalOptions { + readonly stores: JournalStores; + readonly admission: SessionAdmissionGate; + readonly continuity: Pick; + readonly requestDrain: () => void; +} + +/** + * Append-only authority for WorkHub action intent and committed delegation links. + * + * The interface exposes domain records only. Message identities, transcript + * snapshots, exact replay checks, and commit-outcome handling stay local here. + */ +export class WorkHubDelegationJournal { + readonly #stores: JournalStores; + readonly #admission: SessionAdmissionGate; + readonly #continuity: Pick; + readonly #requestDrain: () => void; + + constructor(options: WorkHubDelegationJournalOptions) { + this.#stores = options.stores; + this.#admission = options.admission; + this.#continuity = options.continuity; + this.#requestDrain = options.requestDrain; + } + + async read(actionId: string): Promise { + await this.#assertCoordinationSession(); + const messages = await this.#readMessages(actionId); + return this.#projectRecord(actionId, messages); + } + + prepare(intent: WorkHubDelegationIntent): Promise { + return this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, async (lease) => { + await this.#assertCoordinationSession(); + const existing = this.#projectRecord( + intent.actionId, + await this.#readMessages(intent.actionId), + ); + if (existing) { + if (!sameIntent(existing, intent)) throw actionConflict(); + return; + } + try { + await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [intentMessage(intent)]); + await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + this.#requestDrain(); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'WorkHub delegation intent outcome is unknown', + ); + } + }); + } + + commit(commit: WorkHubDelegationCommit): Promise { + return this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, async (lease) => { + await this.#assertCoordinationSession(); + const existing = this.#projectRecord( + commit.actionId, + await this.#readMessages(commit.actionId), + ); + if (existing?.kind === 'delegation_committed') { + if (!sameCommit(existing, commit)) throw actionConflict(); + return; + } + if (existing?.kind === 'delegation_abandoned') throw actionConflict(); + if (!existing || !sameIntent(existing, commit)) throw actionConflict(); + try { + await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ + committedMessage(commit), + ]); + await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + this.#requestDrain(); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'WorkHub delegation commit outcome is unknown', + ); + } + }); + } + + abandon(abandoned: WorkHubDelegationAbandoned): Promise { + return this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, async (lease) => { + await this.#assertCoordinationSession(); + const existing = this.#projectRecord( + abandoned.actionId, + await this.#readMessages(abandoned.actionId), + ); + if (existing?.kind === 'delegation_abandoned') { + if (!sameAbandoned(existing, abandoned)) throw actionConflict(); + return; + } + if (!existing || existing.kind !== 'delegation_intent' || !sameIntent(existing, abandoned)) { + throw actionConflict(); + } + try { + await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ + abandonedMessage(abandoned), + ]); + await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + this.#requestDrain(); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'WorkHub delegation abandonment outcome is unknown', + ); + } + }); + } + + async #assertCoordinationSession(): Promise { + try { + const header = await this.#stores.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID); + if (isWorkHubCoordinationSession(header) && !header.isArchived) return; + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub Coordination Session identity is unavailable', + ); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + throw new WorkHubActionEffectFailure( + 'persistence_failed', + 'WorkHub Coordination Session state is unavailable', + ); + } + } + + async #readMessages(actionId: string): Promise { + try { + const throughSequence = await this.#stores.readTranscriptHighWaterSnapshot( + WORKHUB_COORDINATION_SESSION_ID, + ); + if (throughSequence === null) return []; + return await this.#stores.readTranscriptMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID, { + messageIds: RECORD_KINDS.map((kind) => recordMessageId(actionId, kind)), + throughSequence, + maxBytes: RECORD_READ_MAX_BYTES, + maxMessages: RECORD_KINDS.length, + }); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + throw new WorkHubActionEffectFailure( + 'persistence_failed', + 'WorkHub delegation records are unavailable', + ); + } + } + + #projectRecord( + actionId: string, + messages: readonly StoredMessage[], + ): WorkHubDelegationRecord | undefined { + return projectRecord(actionId, messages); + } +} + +function projectRecord( + actionId: string, + messages: readonly StoredMessage[], +): WorkHubDelegationRecord | undefined { + if (messages.length === 0) return undefined; + const intent = messages.find( + (message): message is WorkHubDelegationIntentMessage => + message.type === 'workhub_coordination' && message.kind === 'delegation_intent', + ); + const committed = messages.find( + (message): message is WorkHubDelegationCommittedMessage => + message.type === 'workhub_coordination' && message.kind === 'delegation_committed', + ); + const abandoned = messages.find( + (message): message is WorkHubDelegationAbandonedMessage => + message.type === 'workhub_coordination' && message.kind === 'delegation_abandoned', + ); + if ( + messages.length !== + Number(intent !== undefined) + + Number(committed !== undefined) + + Number(abandoned !== undefined) || + intent?.actionId !== actionId || + (committed !== undefined && (!intent || !sameMessageIntent(intent, committed))) || + (abandoned !== undefined && (!intent || !sameMessageIntent(intent, abandoned))) || + (committed !== undefined && abandoned !== undefined) + ) { + throw new WorkHubActionEffectFailure( + 'persistence_failed', + 'WorkHub delegation record chain is invalid', + ); + } + return committed + ? commitRecord(committed) + : abandoned + ? abandonedRecord(abandoned) + : intent + ? intentRecord(intent) + : undefined; +} + +function intentMessage(intent: WorkHubDelegationIntent): WorkHubDelegationIntentMessage { + return { + type: 'workhub_coordination', + id: recordMessageId(intent.actionId, 'delegation_intent'), + turnId: intent.coordinationTurnId, + ts: Date.now(), + schemaVersion: WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION, + ...intent, + }; +} + +function committedMessage(commit: WorkHubDelegationCommit): WorkHubDelegationCommittedMessage { + return { + type: 'workhub_coordination', + id: recordMessageId(commit.actionId, 'delegation_committed'), + turnId: commit.coordinationTurnId, + ts: Date.now(), + schemaVersion: WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION, + ...commit, + }; +} + +function abandonedMessage( + abandoned: WorkHubDelegationAbandoned, +): WorkHubDelegationAbandonedMessage { + return { + type: 'workhub_coordination', + id: recordMessageId(abandoned.actionId, 'delegation_abandoned'), + turnId: abandoned.coordinationTurnId, + ts: Date.now(), + schemaVersion: WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION, + ...abandoned, + }; +} + +function intentRecord(message: WorkHubDelegationIntentMessage): WorkHubDelegationIntent { + return { + kind: message.kind, + actionId: message.actionId, + actionFingerprint: message.actionFingerprint, + coordinationTurnId: message.coordinationTurnId, + targetSessionId: message.targetSessionId, + disposition: message.disposition, + userText: message.userText, + ...(message.create ? { create: message.create } : {}), + }; +} + +function commitRecord(message: WorkHubDelegationCommittedMessage): WorkHubDelegationCommit { + return { + kind: 'delegation_committed', + actionId: message.actionId, + actionFingerprint: message.actionFingerprint, + coordinationTurnId: message.coordinationTurnId, + targetSessionId: message.targetSessionId, + disposition: message.disposition, + userText: message.userText, + ...(message.create ? { create: message.create } : {}), + delegationId: message.delegationId, + targetTurnId: message.targetTurnId, + ...(message.steered ? { steered: true as const } : {}), + }; +} + +function abandonedRecord(message: WorkHubDelegationAbandonedMessage): WorkHubDelegationAbandoned { + return { + kind: 'delegation_abandoned', + actionId: message.actionId, + actionFingerprint: message.actionFingerprint, + coordinationTurnId: message.coordinationTurnId, + targetSessionId: message.targetSessionId, + disposition: message.disposition, + userText: message.userText, + ...(message.create ? { create: message.create } : {}), + reason: message.reason, + }; +} + +function sameMessageIntent( + intent: WorkHubDelegationIntentMessage, + terminal: WorkHubDelegationCommittedMessage | WorkHubDelegationAbandonedMessage, +): boolean { + return sameIntent( + intentRecord(intent), + terminal.kind === 'delegation_committed' ? commitRecord(terminal) : abandonedRecord(terminal), + ); +} + +function sameIntent( + left: WorkHubDelegationRecord, + right: Omit, +): boolean { + return ( + left.actionId === right.actionId && + left.actionFingerprint === right.actionFingerprint && + left.coordinationTurnId === right.coordinationTurnId && + left.targetSessionId === right.targetSessionId && + left.disposition === right.disposition && + left.userText === right.userText && + isDeepStrictEqual(left.create, right.create) + ); +} + +function sameCommit(left: WorkHubDelegationCommit, right: WorkHubDelegationCommit): boolean { + return ( + sameIntent(left, right) && + left.delegationId === right.delegationId && + left.targetTurnId === right.targetTurnId && + left.steered === right.steered + ); +} + +function sameAbandoned( + left: WorkHubDelegationAbandoned, + right: WorkHubDelegationAbandoned, +): boolean { + return left.reason === right.reason && sameIntent(left, right); +} + +function recordMessageId(actionId: string, kind: (typeof RECORD_KINDS)[number]): string { + return `whj_${createHash('sha256') + .update(`${actionId}\0${kind}`, 'utf8') + .digest('hex') + .slice(0, 48)}`; +} + +function actionConflict(): WorkHubActionEffectFailure { + return new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub action identity belongs to different durable delegation content', + ); +} diff --git a/packages/runtime-host/src/server/workhub-target-submission-recovery.ts b/packages/runtime-host/src/server/workhub-target-submission-recovery.ts new file mode 100644 index 0000000000..14fe692897 --- /dev/null +++ b/packages/runtime-host/src/server/workhub-target-submission-recovery.ts @@ -0,0 +1,121 @@ +/* + * 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 { + messageContentDigest, + normalizeMessageContent, + type MessageContent, +} from '@maka/core/events'; +import type { + ImmutableSteeringMessageProof, + RootTurnAdmission, + RootTurnSourceMessageReceipt, +} from '@maka/storage/agent-run-store'; +import type { PendingMessageAdmission } from '@maka/storage/execution-stores'; +import { WorkHubActionEffectFailure } from './workhub-coordination-action-gate.js'; + +export interface WorkHubSubmissionRecoveryStores { + readRootTurnSourceMessageReceipt( + sessionId: string, + messageId: string, + ): Promise; + readMessageAdmission( + sessionId: string, + messageId: string, + ): Promise; + readRootTurnAdmission(sessionId: string, turnId: string): Promise; + readImmutableSteeringMessageProof( + sessionId: string, + messageId: string, + ): Promise; +} + +export async function recoverWorkHubTargetSubmission( + stores: WorkHubSubmissionRecoveryStores, + input: { readonly sessionId: string; readonly messageId: string; readonly text: string }, +): Promise<{ readonly turnId: string; readonly steered?: true } | undefined> { + const content = normalizeMessageContent({ text: input.text }); + const expectedDigest = messageContentDigest(content); + const receipt = await stores.readRootTurnSourceMessageReceipt(input.sessionId, input.messageId); + if (receipt) { + const source = receipt.sourceMessage; + const actualDigest = source.submittedContentDigest ?? messageContentDigest(source.content); + assertMatchingSubmission(source.placement, actualDigest, expectedDigest); + return source.disposition === 'turn_started' + ? { turnId: receipt.admission.turnId } + : source.disposition === 'steering' + ? { turnId: receipt.admission.turnId, steered: true } + : undefined; + } + + const steeringProof = await stores.readImmutableSteeringMessageProof( + input.sessionId, + input.messageId, + ); + if (steeringProof) { + const proofContent = workHubSteeringProofContent(steeringProof); + const proofDigest = + steeringProof.event.refs?.sourceMessageDigest ?? + (proofContent ? messageContentDigest(proofContent) : undefined); + if ( + steeringProof.event.content?.kind !== 'text' || + steeringProof.event.content.steering !== true || + proofDigest !== expectedDigest + ) { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub target steering identity belongs to different content', + ); + } + return { turnId: steeringProof.event.turnId, steered: true }; + } + + const admission = await stores.readMessageAdmission(input.sessionId, input.messageId); + if (!admission) return undefined; + assertMatchingSubmission( + admission.submittedPlacement, + admission.submittedContentDigest, + expectedDigest, + ); + const root = await stores.readRootTurnAdmission(input.sessionId, admission.turnId); + if (admission.disposition !== 'steering' || !root || root.runId !== admission.runId) + return undefined; + return { turnId: admission.turnId, steered: true }; +} + +function assertMatchingSubmission( + placement: 'current_turn' | 'next_turn', + actualDigest: string, + expectedDigest: string, +): void { + if (placement !== 'current_turn' || actualDigest !== expectedDigest) { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub target Message identity belongs to different content', + ); + } +} + +export function workHubSteeringProofContent( + proof: ImmutableSteeringMessageProof, +): MessageContent | undefined { + return proof.event.content?.kind === 'text' + ? normalizeMessageContent(proof.event.content) + : undefined; +}