diff --git a/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts index 70a84dedf9..0755f7ce12 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 @@ -46,6 +46,7 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain' listWorkHubCoordinationCandidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [], + delegations: [{ actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }], }), actWorkHubCoordination: async (input: unknown) => { actions.push(input); @@ -90,6 +91,7 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain' assert.deepEqual(await handlers.get('workhub:candidates')?.({}), { candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [], + delegations: [{ actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }], }); assert.deepEqual( await handlers.get('workhub:act')?.({}, { diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index 13b46ce35c..c69eb8ce9e 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -24,6 +24,7 @@ import type { WorkHubCoordinationActInput } from '@maka/runtime-host/protocol'; import { createWorkHubController as createGatedWorkHubController, WORKHUB_ROUTING_STRATEGY_ID, + WorkHubCoordinationFailure, type WorkHubSessionFacts, type WorkHubSessionPort, type WorkHubCoordinationTurn, @@ -32,7 +33,6 @@ import { createWorkHubRoutePolicy, workHubNewSessionName, } from '../../renderer/workhub-route-policy.js'; -import { WorkHubCoordinationFailure } from '../../renderer/workhub-coordination-port.js'; const appShellUrl = [ new URL('../../renderer/app-shell.tsx', import.meta.url), @@ -193,6 +193,14 @@ function createWorkHubController({ sessions }: { sessions: TestSessionPort }) { targetSessionId: input.proposal.expects.targetSessionId, }; } + if (input.proposal.disposition === 'resume_work') { + return { + disposition: 'resume_work', + outcome: 'resume_started', + targetSessionId: input.proposal.expects.targetSessionId, + targetTurnId: 'resumed-turn', + }; + } const target = candidateByRef.get(input.proposal.candidateRef); if (!target) throw new Error('unknown test candidate'); const admitted = await sessions.submit(target.target, input.userText, input.actionId); @@ -420,6 +428,163 @@ test('an anaphoric stop asks for a fresh named imperative without offering a rou await handle.close(); }); +test('a named resume submits and reports what the Host did', async () => { + const sessions = port([session('payments', { sessionName: 'Payments' })]); + const actions: WorkHubCoordinationActInput[] = []; + const controller = createGatedWorkHubController({ + sessions, + coordination: { + open: async (handler) => { + handler([], []); + return { close: async () => undefined }; + }, + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('a resume must not read route candidates'), + act: async (input) => { + actions.push(input); + return { + disposition: 'resume_work', + outcome: 'resume_started', + targetSessionId: 'payments', + targetTurnId: 'resumed-turn', + }; + }, + }, + }); + const handle = await controller.openConversation(() => undefined, () => undefined); + + const result = await controller.submit({ requestId: 'resume-1', text: 'Resume Payments' }); + + assert.deepEqual(result, { + kind: 'resume', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: 'resume-1', + target: { sessionId: 'payments' }, + outcome: 'resume_started', + targetTurnId: 'resumed-turn', + }); + // The proposal names the Session and carries no confirmation: resume ends + // nothing, so it needs no authority a delegation did not already grant. + assert.deepEqual(actions, [{ + actionId: 'resume-1', + userText: 'Resume Payments', + proposal: { disposition: 'resume_work', expects: { targetSessionId: 'payments' } }, + }]); + await handle.close(); +}); + +test('an anaphoric resume asks for a named work item', async () => { + const controller = createGatedWorkHubController({ + sessions: port([session('payments', { sessionName: 'Payments' })]), + coordination: { + open: async (handler) => { + handler([], []); + return { close: async () => undefined }; + }, + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('resume clarification must not read route candidates'), + act: async () => assert.fail('anaphoric resume must not reach the Action Gate'), + }, + }); + const handle = await controller.openConversation(() => undefined, () => undefined); + + assert.deepEqual(await controller.submit({ requestId: 'resume-it', text: 'Resume it' }), { + kind: 'clarification', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: 'resume-it', + text: 'Resume it', + options: [], + reason: 'resume_target_required', + }); + await handle.close(); +}); + +test('a resume the Host will not admit becomes its clarification', async () => { + const sessions = port([session('payments', { sessionName: 'Payments' })]); + const controller = createGatedWorkHubController({ + sessions, + coordination: { + open: async (handler) => { + handler([], []); + return { close: async () => undefined }; + }, + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('a resume must not read route candidates'), + act: async () => { + throw new WorkHubCoordinationFailure( + 'operation_conflict', + 'WorkHub has no active durable delegation to resume on that Session', + ); + }, + }, + }); + const handle = await controller.openConversation(() => undefined, () => undefined); + + assert.deepEqual(await controller.submit({ requestId: 'resume-2', text: 'Resume Payments' }), { + kind: 'clarification', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: 'resume-2', + text: 'Resume Payments', + options: [], + reason: 'resume_target_unavailable', + }); + await handle.close(); +}); + +test('a Runtime Host without safe-boundary resume explains why it cannot resume', async () => { + const controller = createGatedWorkHubController({ + sessions: port([session('payments', { sessionName: 'Payments' })]), + coordination: { + open: async (handler) => { + handler([], []); + return { close: async () => undefined }; + }, + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('a resume must not read route candidates'), + act: async () => { + throw new WorkHubCoordinationFailure( + 'operation_unavailable', + 'Safe-boundary resume is disabled for this Runtime Host', + ); + }, + }, + }); + const handle = await controller.openConversation(() => undefined, () => undefined); + + assert.deepEqual(await controller.submit({ requestId: 'resume-disabled', text: 'Resume Payments' }), { + kind: 'clarification', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: 'resume-disabled', + text: 'Resume Payments', + options: [], + reason: 'resume_operation_unavailable', + }); + await handle.close(); +}); + +test('a recovering Runtime Host tells the user to retry resume', async () => { + const controller = createGatedWorkHubController({ + sessions: port([session('payments', { sessionName: 'Payments' })]), + coordination: { + open: async (handler) => { + handler([], []); + return { close: async () => undefined }; + }, + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('a resume must not read route candidates'), + act: async () => { + throw new WorkHubCoordinationFailure('host_not_ready', 'Runtime Host is recovering'); + }, + }, + }); + const handle = await controller.openConversation(() => undefined, () => undefined); + + const result = await controller.submit({ requestId: 'resume-recovering', text: 'Resume Payments' }); + assert.equal(result.kind, 'clarification'); + if (result.kind === 'clarification') assert.equal(result.reason, 'resume_host_recovering'); + await handle.close(); +}); + test('a named stop reports the Gate refusal instead of judging the target itself', async () => { // The renderer no longer decides whether a Session can be stopped, so it // submits and lets the Gate answer. Its refusal is the clarification, which 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 00ba4f9303..108b1bb594 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -29,7 +29,6 @@ import { } from '../../renderer/workhub-session-port.js'; import { createDesktopWorkHubCoordinationPort, - projectWorkHubActiveDelegations, projectWorkHubCoordinationTurns, } from '../../renderer/workhub-coordination-port.js'; @@ -168,16 +167,9 @@ test('projects the durable Coordination transcript into the WorkHub conversation }, updatedAt: 20, }]); - assert.deepEqual(projectWorkHubActiveDelegations( - messages.map((message, sequence) => ({ message, sequence })), - ), [{ - actionId: 'action-1', - targetSessionId: 'payments', - sequence: 3, - }]); }); -test('rebuilds active linkage outside the bounded visible timeline in transcript order', () => { +test('bounds the visible timeline independently of old delegation linkage', () => { const assignment: StoredMessage = { type: 'workhub_coordination', id: 'assignment-old', @@ -211,13 +203,6 @@ test('rebuilds active linkage outside the bounded visible timeline in transcript projectWorkHubCoordinationTurns(messages).some((turn) => turn.messageId === assignment.id), false, ); - assert.deepEqual(projectWorkHubActiveDelegations( - messages.map((message, sequence) => ({ message, sequence })), - ), [{ - actionId: 'action-old', - targetSessionId: 'payments', - sequence: 0, - }]); }); test('projects durable create_new disposition as an explicit new-work announcement', () => { @@ -285,10 +270,6 @@ test('a durable replacement abort terminalizes the retired source linkage', () = reason: 'target_unavailable', }; - assert.deepEqual(projectWorkHubActiveDelegations([ - { sequence: 0, message: assignment }, - { sequence: 1, message: aborted }, - ]), []); assert.equal( projectWorkHubCoordinationTurns([assignment, aborted])[0]?.assignment?.linkState, 'aborted', @@ -328,22 +309,33 @@ test('direct-stop projection is retryable until resolved and preserves not_owned outcome: 'not_owned', }); assert.equal(projected[0]?.assignment?.linkState, 'active'); - assert.deepEqual(projectWorkHubActiveDelegations([ - { sequence: 0, message: assignment }, - { sequence: 1, message: requested }, - { sequence: 2, message: notOwned }, - ]), [{ actionId: 'source-action', targetSessionId: 'payments', sequence: 0 }]); const stopped = { ...notOwned, outcome: 'stop_delivered' as const }; assert.equal( projectWorkHubCoordinationTurns([assignment, requested, stopped])[0]?.assignment?.linkState, 'stopped', ); - assert.deepEqual(projectWorkHubActiveDelegations([ - { sequence: 0, message: assignment }, - { sequence: 1, message: requested }, - { sequence: 2, message: stopped }, - ]), []); +}); + +test('resume projection survives reload and exposes its durable outcome', () => { + const resumed: StoredMessage = { + type: 'workhub_coordination', id: 'resume', turnId: 'resume-action', ts: 3, + schemaVersion: 4, kind: 'delegation_resume', actionId: 'resume-action', + actionFingerprint: `sha256:${'b'.repeat(64)}`, coordinationTurnId: 'resume-action', + resumesActionId: 'source-action', resumesDelegationId: 'payments-delegation', + targetSessionId: 'payments', + targetSessionName: 'Payments', userText: 'Resume Payments', outcome: 'resume_started', + targetTurnId: 'resumed-turn', + }; + + assert.deepEqual(projectWorkHubCoordinationTurns([resumed]), [{ + messageId: 'resume', turnId: 'resume-action', text: 'Resume Payments', + state: 'completed', + resume: { + targetSessionId: 'payments', targetSessionName: 'Payments', outcome: 'resume_started', + }, + updatedAt: 3, + }]); }); test('durable supersession terminalizes only the replaced linkage', () => { @@ -403,9 +395,6 @@ test('durable supersession terminalizes only the replaced linkage', () => { projectWorkHubCoordinationTurns(messages).map((turn) => turn.assignment?.linkState), ['superseded', 'active'], ); - assert.deepEqual(projectWorkHubActiveDelegations( - messages.map((message, sequence) => ({ message, sequence })), - ), [{ actionId: 'action-new', targetSessionId: 'login', sequence: 1 }]); }); test('Coordination transcript adapter emits an initial empty ready snapshot and closes cleanly', async () => { @@ -446,6 +435,7 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and candidates: async () => ({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [], + delegations: [], }), act: async () => ({ ok: true, @@ -465,7 +455,7 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and assert.equal(closes, 1); }); -test('Coordination transcript reset rebuilds active linkage outside the resident window', async () => { +test('Coordination transcript reads old active linkage without replaying older history', async () => { const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'coordination' }); const assignment: StoredMessage = { type: 'workhub_coordination', @@ -485,13 +475,13 @@ test('Coordination transcript reset rebuilds active linkage outside the resident disposition: 'delegate_existing', userText: 'Continue payments', }; - const recent: StoredMessage = { + const recent: StoredMessage[] = Array.from({ length: 1 }, (_, index) => ({ type: 'user', - id: 'recent-user', - turnId: 'recent-turn', - ts: 2, - text: 'Recent coordination', - }; + id: `recent-user-${index}`, + turnId: `recent-turn-${index}`, + ts: index + 2, + text: `Recent coordination ${index}`, + })); const fragment = (message: StoredMessage, sequence: number) => { const data = new TextEncoder().encode(JSON.stringify(message)); return { @@ -506,9 +496,9 @@ test('Coordination transcript reset rebuilds active linkage outside the resident let deliver: ((batch: DesktopTranscriptBatch) => void) | undefined; let generation = 'generation-1'; let historyLoads = 0; - let historyBatchReady = true; + let snapshotReads = 0; const snapshots: unknown[] = []; - const adapter = createDesktopWorkHubCoordinationPort({ + const portDeps: Parameters[0] = { sessionId, transcripts: { open: async (_requestedSessionId, handler) => { @@ -519,7 +509,7 @@ test('Coordination transcript reset rebuilds active linkage outside the resident generation, hostEpoch: 'epoch-1', durableThrough: 1, - fragments: [fragment(recent, 1)], + fragments: recent.map((message, index) => fragment(message, index + 1)), evictedDurableSequences: [], completedOverlayMessageIds: [], hasOlder: true, @@ -534,20 +524,6 @@ test('Coordination transcript reset rebuilds active linkage outside the resident readThroughMessageId: null, loadBefore: async () => { historyLoads += 1; - handler({ - sessionId: 'coordination', - deliverySequence: historyLoads + 1, - generation, - hostEpoch: 'epoch-1', - durableThrough: 1, - fragments: [fragment(assignment, 0)], - evictedDurableSequences: [], - completedOverlayMessageIds: [], - hasOlder: false, - hasNewer: false, - reset: false, - ready: historyBatchReady, - }); }, loadAround: async () => {}, close: async () => {}, @@ -555,15 +531,24 @@ test('Coordination transcript reset rebuilds active linkage outside the resident }, }, record: async (input) => ({ turnId: input.turnId }), - candidates: async () => ({ - candidateSetId: `sha256:${'b'.repeat(64)}`, - candidates: [], - }), + candidates: async () => { + snapshotReads += 1; + return { + candidateSetId: `sha256:${'b'.repeat(64)}`, + candidates: [], + delegations: [{ + actionId: assignment.actionId, + targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }), + sequence: 0, + }], + }; + }, act: async () => ({ ok: true, result: { disposition: 'answer_here', coordinationTurnId: 'coordination-turn' }, }), - }); + }; + const adapter = createDesktopWorkHubCoordinationPort(portDeps); const handle = await adapter.open( (_turns, activeDelegations) => snapshots.push(activeDelegations), @@ -574,17 +559,17 @@ test('Coordination transcript reset rebuilds active linkage outside the resident targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }), sequence: 0, }]); + assert.equal(historyLoads, 0); + assert.equal(snapshotReads, 1); generation = 'generation-2'; - historyBatchReady = false; - const snapshotsBeforeReset = snapshots.length; deliver?.({ sessionId: 'coordination', deliverySequence: 3, generation, hostEpoch: 'epoch-1', durableThrough: 1, - fragments: [fragment(recent, 1)], + fragments: recent.map((message, index) => fragment(message, index + 1)), evictedDurableSequences: [], completedOverlayMessageIds: [], hasOlder: true, @@ -592,25 +577,10 @@ test('Coordination transcript reset rebuilds active linkage outside the resident reset: true, ready: false, }); - await Promise.resolve(); - await Promise.resolve(); + await new Promise((resolve) => setImmediate(resolve)); - assert.equal(historyLoads, 2); - assert.equal(snapshots.length, snapshotsBeforeReset); - deliver?.({ - sessionId: 'coordination', - deliverySequence: 5, - generation, - hostEpoch: 'epoch-1', - durableThrough: 1, - fragments: [fragment(recent, 1)], - evictedDurableSequences: [], - completedOverlayMessageIds: [], - hasOlder: false, - hasNewer: false, - reset: false, - ready: true, - }); + assert.equal(historyLoads, 0); + assert.equal(snapshotReads, 2); assert.deepEqual(snapshots.at(-1), [{ actionId: 'action-old', targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }), @@ -619,6 +589,80 @@ test('Coordination transcript reset rebuilds active linkage outside the resident await handle.close(); }); +test('Coordination transcript coalesces a burst into one Host snapshot refresh', async () => { + const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'coordination' }); + let deliver: ((batch: DesktopTranscriptBatch) => void) | undefined; + let snapshotReads = 0; + const adapter = createDesktopWorkHubCoordinationPort({ + sessionId, + transcripts: { + open: async (_requestedSessionId, handler) => { + deliver = handler; + handler({ + sessionId: 'coordination', + deliverySequence: 1, + generation: 'generation-1', + hostEpoch: 'epoch-1', + durableThrough: null, + fragments: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: false, + hasNewer: false, + reset: true, + ready: true, + }); + return { + sessionId, + generation: 'generation-1', + hostEpoch: 'epoch-1', + readThroughMessageId: null, + loadBefore: async () => {}, + loadAround: async () => {}, + close: async () => {}, + }; + }, + }, + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => { + snapshotReads += 1; + return { + candidateSetId: `sha256:${'c'.repeat(64)}`, + candidates: [], + delegations: [], + }; + }, + act: async () => ({ + ok: true, + result: { disposition: 'answer_here', coordinationTurnId: 'coordination-turn' }, + }), + }); + const handle = await adapter.open(() => {}, (error) => assert.fail(String(error))); + assert.equal(snapshotReads, 1); + + for (let deliverySequence = 2; deliverySequence <= 4; deliverySequence += 1) { + deliver?.({ + sessionId: 'coordination', + deliverySequence, + generation: 'generation-1', + hostEpoch: 'epoch-1', + durableThrough: null, + fragments: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: false, + hasNewer: false, + reset: false, + ready: true, + }); + } + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(snapshotReads, 2); + await handle.close(); +}); + test('projects durable Session messages into an ordered WorkHub conversation', () => { const turns = projectWorkHubSessionTurns({ target: { sessionId: 'payment' }, diff --git a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts index 8d9166794e..e88af5b222 100644 --- a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts @@ -665,6 +665,14 @@ test('real Session projection creates new guide topics and preserves origin ambi targetSessionId: input.proposal.expects.targetSessionId, }; } + if (input.proposal.disposition === 'resume_work') { + return { + disposition: 'resume_work', + outcome: 'resume_started', + targetSessionId: input.proposal.expects.targetSessionId, + targetTurnId: 'resumed-turn', + }; + } const targetSessionId = input.proposal.candidateRef.replace(/^candidate-/u, ''); const admitted = await send(targetSessionId, { type: 'send', diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index bcb2babbf9..a17565b16c 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -2028,6 +2028,10 @@ const makaBridge = { ...candidate, sessionId: recordRuntimeHostSessionScope(scope, candidate.sessionId), })), + delegations: result.delegations.map((delegation) => ({ + ...delegation, + targetSessionId: recordRuntimeHostSessionScope(scope, delegation.targetSessionId), + })), }; }, async act( diff --git a/apps/desktop/src/renderer/workhub-controller.ts b/apps/desktop/src/renderer/workhub-controller.ts index f900300483..2795705a26 100644 --- a/apps/desktop/src/renderer/workhub-controller.ts +++ b/apps/desktop/src/renderer/workhub-controller.ts @@ -27,6 +27,7 @@ import { createWorkHubRoutePolicy, type WorkHubRouteEvidence, type WorkHubStopClarificationReason, + type WorkHubNamedActionRouteDecision, } from './workhub-route-policy.js'; import type { OperationError, @@ -132,6 +133,11 @@ export interface WorkHubCoordinationTurn { readonly targetSessionName: string; readonly outcome?: Extract['outcome']; }; + resume?: { + readonly targetSessionId: string; + readonly targetSessionName: string; + readonly outcome: Extract['outcome']; + }; updatedAt: number; } @@ -215,6 +221,13 @@ export type WorkHubSubmission = ( outcome: Extract['outcome']; targetTurnId?: string; } + | { + kind: 'resume'; + requestId: string; + target: WorkHubSessionTarget; + outcome: Extract['outcome']; + targetTurnId?: string; + } ) & { strategyId: WorkHubRoutingStrategyId }; /** @@ -358,6 +371,85 @@ export function createWorkHubController(deps: { ...(correction ? { correctedFrom: correction.from } : {}), }; }; + const submitNamedDelegationAction = async ( + input: WorkHubSubmitInput, + decision: WorkHubNamedActionRouteDecision, + kind: 'resume' | 'stop', + ): Promise | undefined> => { + if (decision.kind === 'not_requested') return undefined; + if (decision.kind === 'clarification') { + return { + kind: 'clarification', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + text: input.text, + options: [], + reason: decision.reason, + }; + } + const { target } = decision; + try { + const admitted = await coordination.act({ + actionId: input.requestId, + userText: input.text, + proposal: { + disposition: kind === 'resume' ? 'resume_work' : 'stop_work', + expects: { targetSessionId: target.sessionId }, + }, + ...(kind === 'stop' ? { confirmation: { kind: 'user_stop' as const } } : {}), + }); + const result = { + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + target, + }; + if (kind === 'resume' && admitted.disposition === 'resume_work') { + return { + ...result, + kind: 'resume', + outcome: admitted.outcome, + ...(admitted.targetTurnId ? { targetTurnId: admitted.targetTurnId } : {}), + }; + } + if (kind === 'stop' && admitted.disposition === 'stop_work') { + return { + ...result, + kind: 'stop', + outcome: admitted.outcome, + ...(admitted.targetTurnId ? { targetTurnId: admitted.targetTurnId } : {}), + }; + } + throw new Error('WorkHub Action Gate returned an unexpected disposition'); + } catch (error) { + if ( + kind === 'resume' && + error instanceof WorkHubCoordinationFailure && + (error.code === 'operation_unavailable' || error.code === 'host_not_ready') + ) { + return { + kind: 'clarification', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + text: input.text, + options: [], + reason: error.code === 'host_not_ready' + ? 'resume_host_recovering' + : 'resume_operation_unavailable', + }; + } + if (error instanceof WorkHubCoordinationFailure && error.code === 'operation_conflict') { + return { + kind: 'clarification', + strategyId: WORKHUB_ROUTING_STRATEGY_ID, + requestId: input.requestId, + text: input.text, + options: [], + reason: kind === 'resume' ? 'resume_target_unavailable' : 'stop_target_unavailable', + }; + } + throw error; + } + }; return { async openConversation(handler, onError) { let disposed = false; @@ -490,67 +582,18 @@ export function createWorkHubController(deps: { const sessions = await deps.sessions.list(); reconcileFocus(submissionPolicy, sessions); const ordinary = sessions.filter((session) => session.kind === 'ordinary'); + const resumeDecision = submissionPolicy.resolveResume({ + text: input.text, + sessions: ordinary, + }); + const resume = await submitNamedDelegationAction(input, resumeDecision, 'resume'); + if (resume) return resume; const stopDecision = submissionPolicy.resolveStop({ text: input.text, sessions: ordinary, }); - if (stopDecision.kind !== 'not_requested') { - if (stopDecision.kind === 'clarification') { - return { - kind: 'clarification', - strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: input.requestId, - text: input.text, - options: [], - reason: stopDecision.reason, - }; - } - const { target } = stopDecision; - let admitted; - try { - admitted = await coordination.act({ - actionId: input.requestId, - userText: input.text, - proposal: { - disposition: 'stop_work', - // Only the Session the reference resolved to. Which delegation - // that Session still owns is the Host's to decide, under the - // lease that ends it. - expects: { targetSessionId: target.sessionId }, - }, - confirmation: { kind: 'user_stop' }, - }); - } catch (error) { - // The Gate refusing the stop is an answer, not a fault: it is the - // only party that can say the Session owns no single stoppable - // delegation. Anything else is a real failure and still throws. - if ( - error instanceof WorkHubCoordinationFailure && - error.code === 'operation_conflict' - ) { - return { - kind: 'clarification', - strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: input.requestId, - text: input.text, - options: [], - reason: 'stop_target_unavailable', - }; - } - throw error; - } - if (admitted.disposition !== 'stop_work') { - throw new Error('WorkHub Action Gate returned an unexpected disposition'); - } - return { - kind: 'stop', - strategyId: WORKHUB_ROUTING_STRATEGY_ID, - requestId: input.requestId, - target, - outcome: admitted.outcome, - ...(admitted.targetTurnId ? { targetTurnId: admitted.targetTurnId } : {}), - }; - } + const stop = await submitNamedDelegationAction(input, stopDecision, 'stop'); + if (stop) return stop; const candidateSet = await coordination.candidates(); const candidateBySessionId = new Map( candidateSet.candidates.map((candidate) => [candidate.sessionId, candidate]), diff --git a/apps/desktop/src/renderer/workhub-coordination-port.ts b/apps/desktop/src/renderer/workhub-coordination-port.ts index 114fed8532..9858ddb3e7 100644 --- a/apps/desktop/src/renderer/workhub-coordination-port.ts +++ b/apps/desktop/src/renderer/workhub-coordination-port.ts @@ -34,12 +34,14 @@ import type { WorkHubCoordinationActInput, WorkHubCoordinationActResult, WorkHubCoordinationCandidatesResult, + WorkHubCoordinationSnapshotResult, OperationOutcome, OperationError, } from '@maka/runtime-host/protocol'; import { boundedWorkHubTimelineText, WorkHubCoordinationFailure } from './workhub-controller.js'; export { WorkHubCoordinationFailure }; + import type { WorkHubDesktopTranscriptBridge } from './workhub-session-port.js'; const WORKHUB_COORDINATION_TURN_LIMIT = 40; @@ -52,7 +54,7 @@ export function createDesktopWorkHubCoordinationPort(deps: { userText: string; assistantText: string; }): Promise<{ turnId: string }>; - candidates(): Promise; + candidates(): Promise; act( input: Omit, ): Promise>; @@ -71,49 +73,44 @@ export function createDesktopWorkHubCoordinationPort(deps: { const store = new DesktopTranscriptRangeStore(deps.sessionId); let disposed = false; let ready = false; - let historyReady = false; - let historyGeneration = 0; + let refreshGeneration = 0; let handle: Awaited> | undefined; - let historyLane = Promise.resolve(); - const coordinationMessagesBySequence = new Map(); + let refreshLane = Promise.resolve(); + let refreshScheduled = false; + let scheduledRefreshGeneration = 0; + let activeDelegations: readonly WorkHubActiveDelegation[] = []; const emit = () => { - const messages = store.snapshot().messages; - handler( - projectWorkHubCoordinationTurns(messages), - projectWorkHubActiveDelegations( - [...coordinationMessagesBySequence.entries()] - .sort(([left], [right]) => left - right) - .map(([sequence, message]) => ({ sequence, message })), - ), - ); + handler(projectWorkHubCoordinationTurns(store.snapshot().messages), activeDelegations); + }; + const scheduleRefresh = (generation: number) => { + scheduledRefreshGeneration = generation; + if (refreshScheduled) return; + refreshScheduled = true; + refreshLane = refreshLane + .then(async () => { + const scheduledGeneration = scheduledRefreshGeneration; + refreshScheduled = false; + if (disposed || scheduledGeneration !== refreshGeneration) return; + const result = await deps.candidates(); + if (disposed || scheduledGeneration !== refreshGeneration) return; + activeDelegations = result.delegations; + if (ready) emit(); + }) + .catch(onError); }; const opened = await deps.transcripts.open( deps.sessionId, (batch) => { if (disposed) return; try { + const reset = batch.reset === true; if (batch.reset) { - coordinationMessagesBySequence.clear(); ready = false; - historyReady = false; - const generation = ++historyGeneration; - if (handle) { - historyLane = historyLane.then(async () => { - if (disposed || generation !== historyGeneration) return; - await rebuildCompleteHistory(generation); - }).catch((error) => { - onError(error); - }); - } + refreshGeneration += 1; } const changed = store.accept(batch); - for (const { sequence, message } of store.durableEntries()) { - if (message.type === 'workhub_coordination') { - coordinationMessagesBySequence.set(sequence, message); - } - } ready ||= batch.ready; - if (historyReady && ready && (changed || batch.ready)) emit(); + if (handle && (reset || changed || batch.ready)) scheduleRefresh(refreshGeneration); } catch (error) { onError(error); } @@ -127,28 +124,10 @@ export function createDesktopWorkHubCoordinationPort(deps: { }); handle = opened; - async function rebuildCompleteHistory(generation: number): Promise { - if (!handle) return; - while (!disposed && generation === historyGeneration && store.range().hasOlder) { - const before = store.range().oldestSequence; - await handle.loadBefore(before); - const after = store.range(); - if (after.hasOlder && after.oldestSequence === before) { - throw new Error('WorkHub Coordination transcript history did not advance'); - } - } - if (disposed || generation !== historyGeneration) return; - const range = store.range(); - if (range.hasNewer && range.durableThrough !== null) { - await handle.loadAround(range.durableThrough); - } - if (disposed || generation !== historyGeneration) return; - historyReady = true; - if (ready) emit(); - } - try { - await rebuildCompleteHistory(historyGeneration); + const result = await deps.candidates(); + activeDelegations = result.delegations; + if (ready) emit(); } catch (error) { disposed = true; await handle.close().catch(() => undefined); @@ -164,27 +143,6 @@ export function createDesktopWorkHubCoordinationPort(deps: { }; } -export function projectWorkHubActiveDelegations( - entries: ReadonlyArray<{ readonly sequence: number; readonly message: StoredMessage }>, -): WorkHubActiveDelegation[] { - const terminalDelegationIds = new Set( - entries.flatMap(({ message }) => { - const terminal = terminalDelegationLink(message); - return terminal ? [terminal.delegationId] : []; - }), - ); - return entries.flatMap(({ message, sequence }) => - message.type === 'workhub_coordination' && - message.kind === 'delegation_assigned' && - !terminalDelegationIds.has(message.delegationId) - ? [{ - actionId: message.actionId, - targetSessionId: message.targetSessionId, - sequence, - }] - : []); -} - export function projectWorkHubCoordinationTurns( messages: readonly StoredMessage[], ): WorkHubCoordinationTurn[] { @@ -207,6 +165,21 @@ export function projectWorkHubCoordinationTurns( } for (const message of messages) { + if (message.type === 'workhub_coordination' && message.kind === 'delegation_resume') { + turns.push({ + messageId: message.id, + turnId: message.coordinationTurnId, + text: boundedWorkHubTimelineText(message.userText), + state: 'completed', + resume: { + targetSessionId: message.targetSessionId, + targetSessionName: message.targetSessionName, + outcome: message.outcome, + }, + updatedAt: message.ts, + }); + continue; + } if (message.type === 'workhub_coordination' && message.kind === 'delegation_stop_requested') { const resolution = stopResolutionByDelegationId.get(message.stopsDelegationId); turns.push({ diff --git a/apps/desktop/src/renderer/workhub-route-policy.ts b/apps/desktop/src/renderer/workhub-route-policy.ts index af2960c94f..90c855b7ef 100644 --- a/apps/desktop/src/renderer/workhub-route-policy.ts +++ b/apps/desktop/src/renderer/workhub-route-policy.ts @@ -79,14 +79,24 @@ export type WorkHubStopClarificationReason = /** The stop names more than one existing Session. */ | 'stop_target_ambiguous' /** The Host refused the stop; its conflict is the whole answer. */ - | 'stop_target_unavailable'; + | 'stop_target_unavailable' + /** The resume names more than one existing Session. */ + | 'resume_target_ambiguous' + /** The resume names no safe target of its own. */ + | 'resume_target_required' + /** The Host refused the resume; its conflict is the whole answer. */ + | 'resume_target_unavailable' + /** This Host does not expose safe-boundary resume. */ + | 'resume_operation_unavailable' + /** The Host is still recovering; retry may succeed. */ + | 'resume_host_recovering'; /** * A stop clarification never offers route options. Choosing one re-sends the * original text as work, and stop-shaped text is exactly what must not be * delivered to a Session that way, so the reason carries the whole answer. */ -export type WorkHubStopRouteDecision = +export type WorkHubNamedActionRouteDecision = | { kind: 'not_requested' } | { kind: 'clarification'; reason: WorkHubStopClarificationReason } | { kind: 'target'; target: WorkHubRouteTarget }; @@ -95,7 +105,11 @@ export interface WorkHubRoutePolicy { resolveStop(input: { text: string; sessions: WorkHubRoutableSession[]; - }): WorkHubStopRouteDecision; + }): WorkHubNamedActionRouteDecision; + resolveResume(input: { + text: string; + sessions: WorkHubRoutableSession[]; + }): WorkHubNamedActionRouteDecision; resolve(input: { text: string; sessions: WorkHubRoutableSession[]; @@ -132,6 +146,38 @@ const MIN_STRONG_SINGLE_LATIN_LENGTH = 8; const MAX_UNCERTAINTY_OPTIONS = 5; const MAX_RELATED_CLARIFICATION_OPTIONS = 4; +function resolveNamedDelegationAction( + sessionResolver: WorkHubSessionResolver, + reference: string, + sessions: WorkHubRoutableSession[], + ambiguousReason: WorkHubStopClarificationReason, +): WorkHubNamedActionRouteDecision { + const sessionByRef = new Map(sessions.map((session) => [session.target.sessionId, session])); + const resolution = sessionResolver.resolve({ + reference: { text: reference }, + sessions: sessions.map(resolverSession), + }); + if (resolution.kind === 'none') return { kind: 'not_requested' }; + // The tail rule. The Resolver reports what the reference said after the name; + // one of these commands may add punctuation and nothing else, so + // `Stop Payments and Login` names no target here even though `Payments` + // matched. + const admissible = resolution.candidates.filter( + ({ evidence }) => + evidence.kind === 'elided_name_punctuation' || + /^[.!?。!?]*$/u.test(evidence.remainder), + ); + if (admissible.length === 0) return { kind: 'not_requested' }; + // One candidate only. A ranked resolver may return several; neither action + // picks a winner from a ranking it cannot justify. + if (resolution.kind === 'ambiguous' || admissible.length > 1) { + return { kind: 'clarification', reason: ambiguousReason }; + } + const resolved = sessionByRef.get(admissible[0]!.ref); + if (!resolved) return { kind: 'not_requested' }; + return { kind: 'target', target: resolved.target }; +} + /** * Deep routing module for R2.4. * @@ -163,41 +209,30 @@ function createWorkHubRoutePolicyVisit( // reference still fails closed, and a resolved Session that is not uniquely // stoppable says why. resolveStop({ text, sessions }) { - const intent = readWorkHubRequestIntent(text); - if (!intent.stop.cue) return { kind: 'not_requested' }; - const reference = intent.stop.imperative ? intent.stop.target : undefined; - if (!reference) { + const action = readWorkHubRequestIntent(text).stop; + if (!action.cue) return { kind: 'not_requested' }; + if (!action.imperative || !action.target) { return { kind: 'clarification', reason: 'stop_target_required' }; } - const sessionByRef = new Map( - sessions.map((session) => [session.target.sessionId, session]), + return resolveNamedDelegationAction( + sessionResolver, + action.target, + sessions, + 'stop_target_ambiguous', ); - const resolution = sessionResolver.resolve({ - reference: { text: reference }, - sessions: sessions.map(resolverSession), - }); - if (resolution.kind === 'none') return { kind: 'not_requested' }; - // Stop's own tail rule. The Resolver reports what the reference said - // after the name; a destructive command may add punctuation and nothing - // else, so `Stop Payments and Login` names no stoppable target here even - // though `Payments` matched. - const admissible = resolution.candidates.filter( - ({ evidence }) => - evidence.kind === 'elided_name_punctuation' || - /^[.!?。!?]*$/u.test(evidence.remainder), - ); - if (admissible.length === 0) return { kind: 'not_requested' }; - // Stop admits one candidate only. A ranked resolver may return several; - // this action never picks a winner from a ranking it cannot justify. - if (resolution.kind === 'ambiguous' || admissible.length > 1) { - return { kind: 'clarification', reason: 'stop_target_ambiguous' }; + }, + resolveResume({ text, sessions }) { + const action = readWorkHubRequestIntent(text).resume; + if (!action.cue) return { kind: 'not_requested' }; + if (!action.imperative || !action.target) { + return { kind: 'clarification', reason: 'resume_target_required' }; } - const resolved = sessionByRef.get(admissible[0]!.ref); - if (!resolved) return { kind: 'not_requested' }; - // The reference resolved, which is everything this policy can prove. - // Which delegation to end, and whether there is one at all, is the - // Host's answer and is made under the lease that performs the stop. - return { kind: 'target', target: resolved.target }; + return resolveNamedDelegationAction( + sessionResolver, + action.target, + sessions, + 'resume_target_ambiguous', + ); }, resolve({ text, sessions, originPromptBySessionId, explicitTarget }) { const intent = readWorkHubRequestIntent(text); diff --git a/apps/desktop/src/renderer/workhub-surface.tsx b/apps/desktop/src/renderer/workhub-surface.tsx index 8a7cf93ad2..b944edd7ff 100644 --- a/apps/desktop/src/renderer/workhub-surface.tsx +++ b/apps/desktop/src/renderer/workhub-surface.tsx @@ -27,20 +27,20 @@ import { import { Button } from '@astryxdesign/core/Button'; import type { UiLocale } from '@maka/core/ui-locale'; import { ChatSurfaceLayout, Composer } from '@maka/ui'; -import type { - WorkHubController, - WorkHubCoordinationTurn, - WorkHubDelegationLinkState, - WorkHubProjection, - WorkHubSessionSummary, - WorkHubSubmission, - WorkHubSubmitInput, +import { + type WorkHubController, + type WorkHubCoordinationTurn, + type WorkHubDelegationLinkState, + type WorkHubProjection, + type WorkHubSessionSummary, + type WorkHubSubmission, + type WorkHubSubmitInput, } from './workhub-controller.js'; +import { WorkHubCoordinationFailure } from './workhub-coordination-port.js'; import { WorkHubSendLease, type WorkHubSendAttempt, } from './workhub-send-lease.js'; -import { WorkHubCoordinationFailure } from './workhub-coordination-port.js'; export interface WorkHubConversationTurn { requestId: string; @@ -176,7 +176,8 @@ export async function submitAndRecordWorkHubSurfaceInput(input: { result.kind === 'discussion' || result.kind === 'waiting' || result.kind === 'submitted' || - result.kind === 'stop' + result.kind === 'stop' || + result.kind === 'resume' ) { return result; } @@ -320,7 +321,8 @@ export function WorkHubSurface(props: { ? { ...turn, state: 'settled', outcome: result } : turn, )); - if (result.kind === 'submitted' || result.kind === 'stop') await refresh(); + if (result.kind === 'submitted' || result.kind === 'stop' || result.kind === 'resume') + await refresh(); return result; } catch (error) { if (isTerminalWorkHubSurfaceFailure(error)) { @@ -573,10 +575,15 @@ export function WorkHubCoordinationTurnView(props: { (candidate) => candidate.target.sessionId === props.turn.stop!.targetSessionId, ) : undefined; + const resumedSession = props.turn.resume + ? props.projection.sessions.find( + (candidate) => candidate.target.sessionId === props.turn.resume!.targetSessionId, + ) + : undefined; return ( + ) : props.turn.resume ? ( + ) : assignment ? ( session.target.sessionId === result.target.sessionId, ); @@ -683,6 +707,7 @@ function WorkHubTurnView(props: { const { turn, copy } = props; const submitted = turn.outcome?.kind === 'submitted' ? turn.outcome : undefined; const stopped = turn.outcome?.kind === 'stop' ? turn.outcome : undefined; + const resumed = turn.outcome?.kind === 'resume' ? turn.outcome : undefined; const target = submitted ? props.projection.sessions.find((session) => session.target.sessionId === submitted.target.sessionId) : undefined; @@ -739,6 +764,18 @@ function WorkHubTurnView(props: { copy={copy} onOpenSession={props.onOpenSession} /> + ) : resumed ? ( + session.target.sessionId === resumed.target.sessionId, + )} + targetSessionId={resumed.target.sessionId} + heading={copy.resumeOutcomes[resumed.outcome]} + state={copy.resumeRecorded} + result={undefined} + copy={copy} + onOpenSession={props.onOpenSession} + /> ) : submitted ? ( { }, ); }); + + test('decodes one exact observed resume record', () => { + const resumed = { + type: 'workhub_coordination', + id: 'resume-id', + turnId: 'resume-action', + ts: 6, + schemaVersion: 4, + kind: 'delegation_resume', + actionId: 'resume-action', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'resume-action', + resumesActionId: 'original-action', + resumesDelegationId: 'original-delegation', + targetSessionId: 'payments', + targetSessionName: 'Payments', + userText: 'Resume Payments', + outcome: 'resume_started', + targetTurnId: 'resumed-turn', + } as const; + + assert.deepEqual(decodeCanonicalMessage(resumed), resumed); + for (const invalid of [ + { ...resumed, targetTurnId: undefined }, + { ...resumed, outcome: 'parked' }, + { ...resumed, sourceRunId: 'injected' }, + ]) { + assert.throws(() => decodeCanonicalMessage(invalid), /Invalid stored message schema/u); + } + const alreadyRunning = { + ...resumed, + outcome: 'already_running' as const, + targetTurnId: undefined, + }; + assert.deepEqual(decodeCanonicalMessage(alreadyRunning), alreadyRunning); + }); }); diff --git a/packages/core/src/__tests__/workhub-creation-intent.test.ts b/packages/core/src/__tests__/workhub-creation-intent.test.ts index 04cc0f8bb6..4a1107f119 100644 --- a/packages/core/src/__tests__/workhub-creation-intent.test.ts +++ b/packages/core/src/__tests__/workhub-creation-intent.test.ts @@ -1126,3 +1126,45 @@ test('a spoken Chinese stop is a stop, in the same range English already covers' assert.equal(readWorkHubRequestIntent(text).stop.cue, false, text); } }); + +test('a resume names one Session, and reads like a stop everywhere else', () => { + // Resume asks the Host to carry on work an interruption left unfinished, so + // it is admitted on the same terms as a stop: a direct speech act naming one + // existing Session, in either language. + for (const [text, target] of [ + ['Resume Payments', 'Payments'], + ['恢复支付任务', '支付任务'], + ['接着跑支付任务', '支付任务'], + ] as const) { + assert.deepEqual( + readWorkHubRequestIntent(text).resume, + { cue: true, imperative: true, target }, + text, + ); + } + + for (const text of ['Resume it', '恢复它']) { + assert.deepEqual(readWorkHubRequestIntent(text).resume, { cue: true, imperative: false }, text); + } + + // Ambiguous verbs remain ordinary Session instructions rather than being + // consumed as WorkHub resume commands. + for (const text of [ + 'Continue Payments', + 'Restart Payments', + '继续支付任务', + '请继续支付任务', + '重新开始支付任务', + 'Should I resume Payments?', + 'Do not resume Payments', + 'Resume "Payments', + ]) { + assert.equal(readWorkHubRequestIntent(text).resume.imperative, false, text); + } + + // Stop and resume are separate speech acts; neither reads as the other, and + // ordinary work is neither. + assert.equal(readWorkHubRequestIntent('Stop Payments').resume.imperative, false); + assert.equal(readWorkHubRequestIntent('Resume Payments').stop.cue, false); + assert.equal(readWorkHubRequestIntent('Fix the login bug').resume.imperative, false); +}); diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index dc03aea7be..4e4e7044bd 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -941,6 +941,7 @@ export interface TurnStateMessage { export const WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION = 1 as const; export const WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION = 2 as const; export const WORKHUB_COORDINATION_STOP_SCHEMA_VERSION = 3 as const; +export const WORKHUB_COORDINATION_RESUME_SCHEMA_VERSION = 4 as const; export type WorkHubDelegationDisposition = 'delegate_existing' | 'create_new'; @@ -1077,6 +1078,26 @@ export interface WorkHubDelegationStopResolvedMessage { targetTurnId?: string; } +/** Durable observed result of a resume attempt. */ +export interface WorkHubDelegationResumeMessage { + type: 'workhub_coordination'; + id: string; + turnId: string; + ts: number; + schemaVersion: typeof WORKHUB_COORDINATION_RESUME_SCHEMA_VERSION; + kind: 'delegation_resume'; + actionId: string; + actionFingerprint: `sha256:${string}`; + coordinationTurnId: string; + resumesActionId: string; + resumesDelegationId: string; + targetSessionId: string; + targetSessionName: string; + userText: string; + outcome: 'resume_started' | 'already_running'; + targetTurnId?: string; +} + /** * The exact durable operation one WorkHub action identity is allowed to own. * @@ -1091,7 +1112,12 @@ export type WorkHubActionOperation = | 'delegate_existing' | 'create_new' | 'replace' - | 'stop'; + | 'stop' + // Resume claims like every other disposition, so one action identity still + // means one operation. A Host that predates this value refuses to read a + // claim carrying it, which is a downgrade hazard and not a wire one: the + // table is local, and the row only exists once a resume has been admitted. + | 'resume'; /** Durable global binding from one action identity to one exact operation. */ export interface WorkHubActionClaim { @@ -1110,7 +1136,8 @@ export type WorkHubCoordinationMessage = | WorkHubDelegationReplacementAbortedMessage | WorkHubDelegationSupersededMessage | WorkHubDelegationStopRequestedMessage - | WorkHubDelegationStopResolvedMessage; + | WorkHubDelegationStopResolvedMessage + | WorkHubDelegationResumeMessage; function isWorkHubDelegationStopResolution( outcome: unknown, @@ -1380,6 +1407,26 @@ const WORKHUB_DELEGATION_STOP_RESOLVED_MESSAGE_SHAPE = ], ['targetTurnId'], ); +const WORKHUB_DELEGATION_RESUME_MESSAGE_SHAPE = defineObjectShape()( + [ + 'type', + 'id', + 'turnId', + 'ts', + 'schemaVersion', + 'kind', + 'actionId', + 'actionFingerprint', + 'coordinationTurnId', + 'resumesActionId', + 'resumesDelegationId', + 'targetSessionId', + 'targetSessionName', + 'userText', + 'outcome', + ], + ['targetTurnId'], +); const WORKHUB_DELEGATION_CREATE_SHAPE = defineObjectShape()( ['title', 'workspace'], [], @@ -1570,6 +1617,29 @@ function decodeMessage( } function isWorkHubCoordinationMessage(message: Record): boolean { + if (message.kind === 'delegation_resume') { + const started = message.outcome === 'resume_started'; + return ( + hasMessageEnvelope(message, true) && + hasExactShape(message, WORKHUB_DELEGATION_RESUME_MESSAGE_SHAPE) && + message.schemaVersion === WORKHUB_COORDINATION_RESUME_SCHEMA_VERSION && + isWorkHubActionIdentity(message) && + typeof message.resumesActionId === 'string' && + message.resumesActionId.length > 0 && + typeof message.resumesDelegationId === 'string' && + message.resumesDelegationId.length > 0 && + typeof message.targetSessionId === 'string' && + message.targetSessionId.length > 0 && + typeof message.targetSessionName === 'string' && + message.targetSessionName.trim().length > 0 && + typeof message.userText === 'string' && + message.userText.trim().length > 0 && + (started || message.outcome === 'already_running') && + (started + ? typeof message.targetTurnId === 'string' && message.targetTurnId.length > 0 + : message.targetTurnId === undefined) + ); + } if (message.kind === 'delegation_stop_requested') { return ( hasMessageEnvelope(message, true) && diff --git a/packages/core/src/workhub-creation-intent.ts b/packages/core/src/workhub-creation-intent.ts index f55679b311..640bcee31d 100644 --- a/packages/core/src/workhub-creation-intent.ts +++ b/packages/core/src/workhub-creation-intent.ts @@ -110,6 +110,10 @@ const DIRECT_STOP_REQUEST = // equivalent either. const DIRECT_CHINESE_STOP_REQUEST = /^\s*(?:(?:请|请帮我|帮我|麻烦你?)\s*)?(?:停止|停掉|停下|取消|终止|中止)\s*(?:(?:这个|该)?(?:会话|工作|任务)\s*)?(.+?)\s*[。!]?\s*$/iu; +const DIRECT_RESUME_REQUEST = + /^\s*(?:(?:please|kindly)\s+)?resume\s+(?:(?:the|this)\s+)?(?:(?:session|work|task|job)\s+)?(.+?)\s*[.!。!]?\s*$/iu; +const DIRECT_CHINESE_RESUME_REQUEST = + /^\s*(?:(?:请|请帮我|帮我|麻烦你?)\s*)?(?:恢复|接着跑)\s*(?:(?:这个|该)?(?:会话|工作|任务)\s*)?(.+?)\s*[。!]?\s*$/iu; const UNSAFE_STOP_TARGET = /^(?:it|this|that|one|everything|all|current|session|work|task|job|(?:this|that|current)\s+(?:session|work|task|job)|它|这个|那个|全部|当前|会话|工作|任务|(?:这个|那个|当前)(?:会话|工作|任务))$/iu; @@ -151,6 +155,13 @@ export interface WorkHubRequestIntent { readonly imperative: boolean; readonly target?: string; }; + readonly resume: { + /** A direct resume speech act was present, but its target may still be unsafe. */ + readonly cue: boolean; + /** True only for a direct, explicitly named resume command. */ + readonly imperative: boolean; + readonly target?: string; + }; } /** @@ -316,8 +327,14 @@ export function readWorkHubRequestIntent(value: string): WorkHubRequestIntent { : { kind: 'unusable' }; const correctionCue = hasWorkHubCorrectionCue(source); const existingTarget = affirmativeWorkHubExistingCorrectionTarget(source); - const stopCue = directWorkHubStopCue(source, literalMask.malformed); - const stopTarget = stopCue ? directWorkHubStopTarget(source, false) : undefined; + const stop = directWorkHubNamedAction(source, literalMask.malformed, [ + DIRECT_STOP_REQUEST, + DIRECT_CHINESE_STOP_REQUEST, + ]); + const resume = directWorkHubNamedAction(source, literalMask.malformed, [ + DIRECT_RESUME_REQUEST, + DIRECT_CHINESE_RESUME_REQUEST, + ]); const actions = allMatches(masked, EXECUTION_ACTION); const execution: WorkHubExecutionIntent = literalMask.malformed || naming.kind === 'unusable' || hasDominatingDeliberation(masked) @@ -335,9 +352,14 @@ export function readWorkHubRequestIntent(value: string): WorkHubRequestIntent { ...(existingTarget ? { existingTarget } : {}), }, stop: { - cue: stopCue, - imperative: Boolean(stopTarget), - ...(stopTarget ? { target: stopTarget } : {}), + cue: stop.cue, + imperative: Boolean(stop.target), + ...(stop.target ? { target: stop.target } : {}), + }, + resume: { + cue: resume.cue, + imperative: Boolean(resume.target), + ...(resume.target ? { target: resume.target } : {}), }, }; } @@ -450,6 +472,19 @@ export function matchWorkHubSessionName( return { kind: 'named', remainder: normalizedTarget.slice(matchedName.length).trim() }; } +/** Whether a direct stop/resume reference names exactly this Session. */ +export function workHubNamedDelegationActionTargetsSession( + action: { readonly imperative: boolean; readonly target?: string }, + sessionName: string, +): boolean { + if (!action.imperative || !action.target) return false; + const match = matchWorkHubSessionName(action.target, sessionName); + return ( + match.kind === 'elided_name_punctuation' || + (match.kind === 'named' && /^[.!?。!?]*$/u.test(match.remainder)) + ); +} + /** * The correction policy's tail rule. A correction may name its target and then * say what to do with it, but a withdrawal anywhere in the reference retracts @@ -481,22 +516,20 @@ function correctionTargetMatchesSession(target: string, sessionName: string): bo return workHubCorrectionAdmitsReference(target, matchWorkHubSessionName(target, sessionName)); } -function directWorkHubStopTarget(value: string, malformedLiteral: boolean): string | undefined { - if (malformedLiteral || /[??]\s*$/u.test(value)) return undefined; - const match = DIRECT_STOP_REQUEST.exec(value) ?? DIRECT_CHINESE_STOP_REQUEST.exec(value); +function directWorkHubNamedAction( + value: string, + malformedLiteral: boolean, + patterns: readonly [RegExp, RegExp], +): { readonly cue: boolean; readonly target?: string } { + if (malformedLiteral || /[??]\s*$/u.test(value)) return { cue: false }; + const match = patterns[0].exec(value) ?? patterns[1].exec(value); const rawTarget = match?.[1]?.trim(); - if (!rawTarget) return undefined; - const target = stripMatchingStopQuotes(rawTarget.replace(/[.!。!]+\s*$/u, '').trim()); - if (!target || UNSAFE_STOP_TARGET.test(target)) return undefined; - return target; -} - -function directWorkHubStopCue(value: string, malformedLiteral: boolean): boolean { - if (malformedLiteral || /[??]\s*$/u.test(value)) return false; - return Boolean(DIRECT_STOP_REQUEST.test(value) || DIRECT_CHINESE_STOP_REQUEST.test(value)); + if (!rawTarget) return { cue: Boolean(match) }; + const target = stripMatchingActionQuotes(rawTarget.replace(/[.!。!]+\s*$/u, '').trim()); + return !target || UNSAFE_STOP_TARGET.test(target) ? { cue: true } : { cue: true, target }; } -function stripMatchingStopQuotes(value: string): string { +function stripMatchingActionQuotes(value: string): string { const pairs = new Map([ ['"', '"'], ["'", "'"], diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index f9a68677ec..b2492054a3 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -686,6 +686,214 @@ test('WorkHub creates new work through the production assignment composition', a }); }); +test('WorkHub Stop retires the running continuation after Resume', async () => { + await withCompositionRoot(async ({ root, owner }) => { + const connectionId = await configureFakeDefaultTarget(owner); + const { composition, manager } = await createCapturedExecutionComposition(owner, { + safeBoundaryResume: true, + }); + const context = { + hostEpoch: 'execution-composition-test', + connectionId: 'workhub-resume-stop-client', + principal: 'local_os_user' as const, + acquireResidency: () => ({ release() {} }), + }; + let continuation: { turnId: string; runId: string } | undefined; + let targetSessionId: string | undefined; + try { + const target = await manager.createSession({ + cwd: root, + llmConnectionId: connectionId, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + name: 'Payments', + }); + targetSessionId = target.id; + await composition.handlers['workhub.coordination.resolve']({}, context); + const candidates = await composition.handlers['workhub.coordination.candidates']({}, context); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const candidate = candidates.result.candidates.find( + ({ sessionId }) => sessionId === target.id, + ); + assert.ok(candidate); + if (!candidate) return; + + const delegated = await composition.handlers['workhub.coordination.act']( + { + actionId: 'workhub-resume-stop-delegation', + userText: FAKE_HOLD_OPEN_PROMPT, + candidateSetId: candidates.result.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: candidate.candidateRef, + }, + }, + context, + ); + assert.equal(delegated.ok, true, JSON.stringify(delegated)); + if (!delegated.ok || delegated.result.disposition !== 'delegate_existing') return; + const original = await composition.handlers['turn.query']( + { sessionId: target.id, turnId: delegated.result.targetTurnId }, + context, + ); + assert.equal(original.ok, true); + if (!original.ok) return; + await composition.handlers['turn.stop']( + { sessionId: target.id, turnId: original.result.turnId, runId: original.result.runId }, + context, + ); + + const resumed = await composition.handlers['workhub.coordination.act']( + { + actionId: 'workhub-resume-stop-resume', + userText: 'Resume Payments', + proposal: { + disposition: 'resume_work', + expects: { targetSessionId: target.id }, + }, + }, + context, + ); + assert.equal(resumed.ok, true, JSON.stringify(resumed)); + if ( + !resumed.ok || + resumed.result.disposition !== 'resume_work' || + !resumed.result.targetTurnId + ) + return; + const resumedTurn = await composition.handlers['turn.query']( + { sessionId: target.id, turnId: resumed.result.targetTurnId }, + context, + ); + assert.equal(resumedTurn.ok, true); + if (!resumedTurn.ok) return; + continuation = { turnId: resumedTurn.result.turnId, runId: resumedTurn.result.runId }; + assert.equal(resumedTurn.result.status, 'running'); + + const stopped = await composition.handlers['workhub.coordination.act']( + { + actionId: 'workhub-resume-stop-stop', + userText: 'Stop Payments', + confirmation: { kind: 'user_stop' }, + proposal: { + disposition: 'stop_work', + expects: { targetSessionId: target.id }, + }, + }, + context, + ); + assert.deepEqual(stopped, { + ok: true, + result: { + disposition: 'stop_work', + outcome: 'stop_delivered', + targetSessionId: target.id, + targetTurnId: continuation.turnId, + }, + }); + const terminal = await composition.handlers['turn.query']( + { sessionId: target.id, turnId: continuation.turnId }, + context, + ); + assert.equal(terminal.ok, true); + if (terminal.ok) assert.equal(terminal.result.status, 'cancelled'); + } finally { + if (continuation && targetSessionId) { + await composition.handlers['turn.stop']( + { sessionId: targetSessionId, ...continuation }, + context, + ); + } + await composition.close(); + } + }); +}); + +test('WorkHub does not record resume while safe-boundary resume is disabled', async () => { + await withCompositionRoot(async ({ root, owner }) => { + const connectionId = await configureFakeDefaultTarget(owner); + const { composition, manager } = await createCapturedExecutionComposition(owner, { + safeBoundaryResume: false, + }); + const context = { + hostEpoch: 'execution-composition-test', + connectionId: 'workhub-disabled-resume-client', + principal: 'local_os_user' as const, + acquireResidency: () => ({ release() {} }), + }; + try { + const target = await manager.createSession({ + cwd: root, + llmConnectionId: connectionId, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + name: 'Payments', + }); + await composition.handlers['workhub.coordination.resolve']({}, context); + const candidates = await composition.handlers['workhub.coordination.candidates']({}, context); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const candidate = candidates.result.candidates.find( + ({ sessionId }) => sessionId === target.id, + ); + assert.ok(candidate); + if (!candidate) return; + const delegated = await composition.handlers['workhub.coordination.act']( + { + actionId: 'workhub-disabled-resume-delegation', + userText: FAKE_HOLD_OPEN_PROMPT, + candidateSetId: candidates.result.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: candidate.candidateRef, + }, + }, + context, + ); + assert.equal(delegated.ok, true, JSON.stringify(delegated)); + if (!delegated.ok || delegated.result.disposition !== 'delegate_existing') return; + const original = await composition.handlers['turn.query']( + { sessionId: target.id, turnId: delegated.result.targetTurnId }, + context, + ); + assert.equal(original.ok, true); + if (!original.ok) return; + await composition.handlers['turn.stop']( + { sessionId: target.id, turnId: original.result.turnId, runId: original.result.runId }, + context, + ); + + const actionId = 'workhub-disabled-resume'; + const resumed = await composition.handlers['workhub.coordination.act']( + { + actionId, + userText: 'Resume Payments', + proposal: { + disposition: 'resume_work', + expects: { targetSessionId: target.id }, + }, + }, + context, + ); + assert.deepEqual(resumed, { + ok: false, + error: { + code: 'operation_unavailable', + message: 'Safe-boundary resume is disabled for this Runtime Host', + }, + }); + await composition.close(); + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + assert.equal(await stores.sessionStore.readWorkHubResume(actionId), undefined); + } finally { + await composition.close(); + } + }); +}); + test('WorkHub correction replaces its link without stopping a shared manual Turn', async () => { await withCompositionRoot(async ({ root, owner }) => { const connectionId = await configureFakeDefaultTarget(owner); @@ -1551,17 +1759,23 @@ async function seedLegacyFakeBackendSession( return sessionId; } -async function createCapturedExecutionComposition(owner: InteractiveRootOwner): Promise<{ +async function createCapturedExecutionComposition( + owner: InteractiveRootOwner, + options: { readonly safeBoundaryResume?: boolean } = {}, +): Promise<{ composition: Awaited>; manager: SessionManager; }> { const originalRecover = SessionManager.prototype.recoverInterruptedSessionsStrict; + const originalSafeBoundaryResume = process.env.MAKA_RUNTIME_SAFE_BOUNDARY_RESUME; let manager: SessionManager | undefined; SessionManager.prototype.recoverInterruptedSessionsStrict = async function (stores) { manager = this; return originalRecover.call(this, stores); }; try { + if (options.safeBoundaryResume === true) process.env.MAKA_RUNTIME_SAFE_BOUNDARY_RESUME = '1'; + if (options.safeBoundaryResume === false) delete process.env.MAKA_RUNTIME_SAFE_BOUNDARY_RESUME; // The production composition no longer registers a test backend of its // own; the deterministic one arrives through the same `primaryBackendFactory` // seam the Desktop E2E run uses. @@ -1574,6 +1788,11 @@ async function createCapturedExecutionComposition(owner: InteractiveRootOwner): if (!manager) throw new Error('Production execution composition did not construct Runtime'); return { composition, manager }; } finally { + if (originalSafeBoundaryResume === undefined) { + delete process.env.MAKA_RUNTIME_SAFE_BOUNDARY_RESUME; + } else { + process.env.MAKA_RUNTIME_SAFE_BOUNDARY_RESUME = originalSafeBoundaryResume; + } SessionManager.prototype.recoverInterruptedSessionsStrict = originalRecover; } } diff --git a/packages/runtime-host/src/__tests__/root-admission-owner.test.ts b/packages/runtime-host/src/__tests__/root-admission-owner.test.ts index bb0c443909..8fd25edd13 100644 --- a/packages/runtime-host/src/__tests__/root-admission-owner.test.ts +++ b/packages/runtime-host/src/__tests__/root-admission-owner.test.ts @@ -45,6 +45,8 @@ test('poisons a Session after an ambiguous durable admission failure', async () }, readRootTurnAdmission: (sessionId, turnId) => durableStore.readRootTurnAdmission(sessionId, turnId), + readRootTurnContinuationAdmission: (sessionId, sourceTurnId, sourceRunId) => + durableStore.readRootTurnContinuationAdmission(sessionId, sourceTurnId, sourceRunId), readRootTurnSourceMessageReceipt: (sessionId, sourceMessageId) => durableStore.readRootTurnSourceMessageReceipt(sessionId, sourceMessageId), listRootTurnAdmissionsForRecovery: (sessionId) => @@ -284,6 +286,7 @@ test('snapshots recovered admissions without retaining mutable caller references const store: RootTurnAdmissionStore = { admitRootTurn: async () => ({ kind: 'admitted', admission }), readRootTurnAdmission: async () => admission, + readRootTurnContinuationAdmission: async () => undefined, readRootTurnSourceMessageReceipt: async () => undefined, listRootTurnAdmissionsForRecovery: async () => [admission], }; @@ -366,6 +369,7 @@ test('returns an owned admission instead of retaining the mutable store result', const store: RootTurnAdmissionStore = { admitRootTurn: async () => ({ kind: 'admitted', admission: durableAdmission }), readRootTurnAdmission: async () => durableAdmission, + readRootTurnContinuationAdmission: async () => undefined, readRootTurnSourceMessageReceipt: async () => undefined, listRootTurnAdmissionsForRecovery: async () => [], }; diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index 553d5d1d4b..e95b794c60 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -1268,6 +1268,8 @@ test('idle Skill admission persists a canonical draft without history before roo throw new Error('injected root admission failure'); }, readRootTurnAdmission: (sessionId, turnId) => store.readRootTurnAdmission(sessionId, turnId), + readRootTurnContinuationAdmission: (sessionId, sourceTurnId, sourceRunId) => + store.readRootTurnContinuationAdmission(sessionId, sourceTurnId, sourceRunId), readRootTurnSourceMessageReceipt: (sessionId, messageId) => store.readRootTurnSourceMessageReceipt(sessionId, messageId), listRootTurnAdmissionsForRecovery: (sessionId) => @@ -3195,6 +3197,8 @@ test('successor admission failure retains the terminal transition and its confir return store.admitRootTurn(input); }, readRootTurnAdmission: (sessionId, turnId) => store.readRootTurnAdmission(sessionId, turnId), + readRootTurnContinuationAdmission: (sessionId, sourceTurnId, sourceRunId) => + store.readRootTurnContinuationAdmission(sessionId, sourceTurnId, sourceRunId), readRootTurnSourceMessageReceipt: (sessionId, messageId) => store.readRootTurnSourceMessageReceipt(sessionId, messageId), listRootTurnAdmissionsForRecovery: (sessionId) => @@ -3303,6 +3307,8 @@ test('shutdown contains a successor backend start rejected by Interaction drain' return store.admitRootTurn(input); }, readRootTurnAdmission: (sessionId, turnId) => store.readRootTurnAdmission(sessionId, turnId), + readRootTurnContinuationAdmission: (sessionId, sourceTurnId, sourceRunId) => + store.readRootTurnContinuationAdmission(sessionId, sourceTurnId, sourceRunId), readRootTurnSourceMessageReceipt: (sessionId, messageId) => store.readRootTurnSourceMessageReceipt(sessionId, messageId), listRootTurnAdmissionsForRecovery: (sessionId) => 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 5d79ca924e..7be94fe3c8 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 @@ -39,6 +39,7 @@ import { type WorkHubDelegationAssignmentInput, type WorkHubDelegationReplacementAbortInput, type WorkHubDelegationReplacementInput, + type WorkHubDelegationResumeInput, type WorkHubDelegationRetirementClaim, type WorkHubDelegationStopInput, type WorkHubDelegationStopResolutionInput, @@ -329,6 +330,213 @@ describe('WorkHub Coordination Action Gate', () => { expects: { targetSessionId }, }); + const resumeProposal = (targetSessionId: string) => ({ + disposition: 'resume_work' as const, + expects: { targetSessionId }, + }); + + const delegatedTo = (effects: ReturnType, sessionId: string) => { + effects.assignmentRecords.set( + 'source-action', + assignmentRecord( + { + actionId: 'source-action', + actionFingerprint: `sha256:${'a'.repeat(64)}`, + targetSessionId: sessionId, + targetSessionName: 'Payments', + disposition: 'delegate_existing', + userText: 'Fix payment retry', + }, + 'source-turn', + ), + ); + }; + + test('resumes the one delegation the named Session owns', async () => { + const effects = fakeEffects([session('payments', { name: 'Payments' })]); + delegatedTo(effects, 'payments'); + + const result = await new WorkHubCoordinationActionGate(effects).act( + { + actionId: 'resume-action', + userText: 'Resume Payments', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ); + + assert.deepEqual(result, { + disposition: 'resume_work', + outcome: 'resume_started', + targetSessionId: 'payments', + targetTurnId: 'resumed-turn', + }); + assert.equal(effects.resumeCalls.length, 1); + const resumeCall = effects.resumeCalls[0]; + assert.ok(resumeCall); + assert.equal(resumeCall.source.actionId, 'source-action'); + // Resume claims like every other disposition, so the identity is spent. + assert.equal(effects.actionClaims.get('resume-action')?.operation, 'resume'); + }); + + test('resume binds the trusted named target to the proposed Session', async () => { + const effects = fakeEffects([ + session('payments', { name: 'Payments' }), + session('login', { name: 'Login' }), + ]); + delegatedTo(effects, 'payments'); + + await assert.rejects( + () => + new WorkHubCoordinationActionGate(effects).act( + { + actionId: 'resume-wrong-target', + userText: 'Resume Login', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + assert.equal(effects.resumeCalls.length, 0); + }); + + test('resume needs a named command and carries no destructive confirmation', async () => { + const effects = fakeEffects([session('payments', { name: 'Payments' })]); + delegatedTo(effects, 'payments'); + const gate = new WorkHubCoordinationActionGate(effects); + + // Anaphora names nothing, so it never reaches the delegation. + await assert.rejects( + () => + gate.act( + { + actionId: 'resume-anaphora', + userText: 'Resume it', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ), + /explicit named command/u, + ); + // A question is not a command. + await assert.rejects( + () => + gate.act( + { + actionId: 'resume-question', + userText: 'Should I resume Payments?', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ), + /explicit named command/u, + ); + assert.equal(effects.resumeCalls.length, 0); + }); + + test('resume refuses a Session that does not own exactly one delegation', async () => { + const none = fakeEffects([session('payments', { name: 'Payments' })]); + await assert.rejects( + () => + new WorkHubCoordinationActionGate(none).act( + { + actionId: 'resume-none', + userText: 'Resume Payments', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ), + /no active durable delegation to resume/u, + ); + + const several = fakeEffects([session('payments', { name: 'Payments' })]); + delegatedTo(several, 'payments'); + several.assignmentRecords.set( + 'second-action', + assignmentRecord( + { + actionId: 'second-action', + actionFingerprint: `sha256:${'b'.repeat(64)}`, + targetSessionId: 'payments', + targetSessionName: 'Payments', + disposition: 'delegate_existing', + userText: 'Also fix the receipts', + }, + 'second-turn', + ), + ); + await assert.rejects( + () => + new WorkHubCoordinationActionGate(several).act( + { + actionId: 'resume-many', + userText: 'Resume Payments', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ), + /does not identify one active durable delegation/u, + ); + assert.equal(several.resumeCalls.length, 0); + }); + + test('resume ignores a retired link when one delegation still holds work', async () => { + const effects = fakeEffects([session('payments', { name: 'Payments' })]); + delegatedTo(effects, 'payments'); + const retired = effects.assignmentRecords.get('source-action')!; + effects.assignmentRecords.set( + 'second-action', + assignmentRecord( + { + actionId: 'second-action', + actionFingerprint: `sha256:${'b'.repeat(64)}`, + targetSessionId: 'payments', + targetSessionName: 'Payments', + disposition: 'delegate_existing', + userText: 'Fix the interrupted receipt retry', + }, + 'second-turn', + ), + ); + effects.retirements.push(retired); + + const result = await new WorkHubCoordinationActionGate(effects).act( + { + actionId: 'resume-one-live', + userText: 'Resume Payments', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ); + + assert.equal(result.disposition, 'resume_work'); + const call = effects.resumeCalls[0]; + assert.ok(call); + assert.equal(call.source.actionId, 'second-action'); + }); + + test('resume reports when the delegated work is already running', async () => { + const effects = fakeEffects([session('payments', { name: 'Payments' })]); + delegatedTo(effects, 'payments'); + effects.resumeOutcome = { outcome: 'already_running' }; + + const result = await new WorkHubCoordinationActionGate(effects).act( + { + actionId: 'resume-already-running', + userText: 'Resume Payments', + proposal: resumeProposal('payments'), + }, + CONTEXT, + ); + + assert.deepEqual(result, { + disposition: 'resume_work', + outcome: 'already_running', + targetSessionId: 'payments', + }); + }); + test('stops exactly one named durable delegation and replays its observed outcome', async () => { const effects = fakeEffects([session('payments', { name: 'Payments' })]); effects.assignmentRecords.set( @@ -2563,6 +2771,25 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { ? 'same_claim' : 'conflict'; }, + resumeCalls: [] as WorkHubDelegationResumeInput[], + resumeOutcome: { + outcome: 'resume_started' as const, + targetTurnId: 'resumed-turn', + } as { + outcome: 'resume_started' | 'already_running'; + targetTurnId?: string; + }, + async resume(input: WorkHubDelegationResumeInput) { + this.resumeCalls.push(input); + return { + disposition: 'resume_work' as const, + outcome: this.resumeOutcome.outcome, + targetSessionId: input.source.targetSessionId, + ...(this.resumeOutcome.targetTurnId + ? { targetTurnId: this.resumeOutcome.targetTurnId } + : {}), + }; + }, async readActionClaim(actionId: string) { return actionClaims.get(actionId); }, @@ -2757,6 +2984,11 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { stopRequests: Map; stopResolutions: Map; retirements: WorkHubDelegationAssignedMessage[]; + resumeCalls: WorkHubDelegationResumeInput[]; + resumeOutcome: { + outcome: 'resume_started' | 'already_running'; + targetTurnId?: string; + }; }; } 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 cb70f8d3b9..cb81e62c57 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts @@ -29,11 +29,13 @@ import { normalizeMessageContent, type MessageContent, } from '@maka/core/events'; +import { deferred } from '@maka/core/test-only/async-primitives'; import { WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_SESSION_ROLE, WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, type StoredMessage, + type WorkHubDelegationAssignedMessage, } from '@maka/core/session'; import { createSessionStore, type SessionAuthorityStore } from '@maka/storage/session-store'; import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store'; @@ -45,10 +47,14 @@ 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 { + WorkHubActionEffectFailure, + type WorkHubActionGateEffects, +} from '../server/workhub-coordination-action-gate.js'; import { HostWorkHubCoordinationCoordinator, type CoordinationCreateTarget, + type HostWorkHubCoordinationCoordinatorOptions, } from '../server/workhub-coordination-coordinator.js'; const CONTEXT: ConnectionContext = { @@ -502,9 +508,13 @@ describe('Host WorkHub Coordination coordinator', () => { }); const assignments: string[] = []; const first = coordinator(root, store, () => undefined, undefined, undefined, undefined, { - assign: async (input) => { + assign: async (input, context, publishCommittedAssignment) => { assignments.push(input.actionId); - return persistTestAssignment(store, input, 'payments-turn'); + return persistTestAssignmentAction(store, 'payments-turn')( + input, + context, + publishCommittedAssignment, + ); }, }); assert.equal((await first.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); @@ -545,6 +555,17 @@ describe('Host WorkHub Coordination coordinator', () => { ], ); assert.equal(assignments.length, 1); + assert.deepEqual(candidates.result.delegations, []); + const current = await first.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(current.ok, true); + if (!current.ok) return; + assert.deepEqual(current.result.delegations, [ + { + actionId: 'payments-action', + targetSessionId: candidates.result.candidates[0]!.sessionId, + sequence: 0, + }, + ]); } finally { await store.close?.(); } @@ -552,7 +573,7 @@ describe('Host WorkHub Coordination coordinator', () => { store = createSessionStore(root); try { const restarted = coordinator(root, store, () => undefined, undefined, undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, 'payments-turn'), + assign: persistTestAssignmentAction(store, 'payments-turn'), }); const candidates = await restarted.handlers['workhub.coordination.candidates']({}, CONTEXT); assert.equal(candidates.ok, true); @@ -582,6 +603,86 @@ describe('Host WorkHub Coordination coordinator', () => { } }); + test('rebuilds chunked active linkage once per Host lifetime from the Coordination ledger', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-active-ledger-')); + const store = createSessionStore(root); + try { + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + assert.equal( + (await coordinator(root, store).handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, + true, + ); + await store.appendMessages( + WORKHUB_COORDINATION_SESSION_ID, + Array.from({ length: 256 }, (_, index) => ({ + type: 'user' as const, + id: `historical-message-${index}`, + turnId: `historical-turn-${index}`, + ts: index, + text: 'historical coordination message', + })), + ); + await persistTestAssignment( + store, + { + actionId: 'chunked-action', + actionFingerprint: `sha256:${'7'.repeat(64)}`, + targetSessionId: target.id, + targetSessionName: target.name, + disposition: 'delegate_existing', + userText: '\\'.repeat(40 * 1024), + }, + 'payments-turn', + ); + + let ledgerScans = 0; + const stores = new Proxy(store, { + get(authority, property, receiver) { + if (property === 'readTranscriptRecordsSnapshot') { + return async ( + ...args: Parameters + ) => { + ledgerScans += 1; + return authority.readTranscriptRecordsSnapshot(...args); + }; + } + const value = Reflect.get(authority, property, receiver) as unknown; + return typeof value === 'function' ? value.bind(authority) : value; + }, + }) as SessionAuthorityStore; + + for (const host of [coordinator(root, stores), coordinator(root, stores)]) { + const first = await host.handlers['workhub.coordination.candidates']({}, CONTEXT); + const second = await host.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(first.ok, true); + assert.equal(second.ok, true); + if (!first.ok || !second.ok) continue; + assert.deepEqual(first.result.delegations, second.result.delegations); + assert.deepEqual(first.result.delegations, [ + { + actionId: 'chunked-action', + targetSessionId: target.id, + sequence: 256, + }, + ]); + } + assert.equal( + ledgerScans, + 4, + 'each Host rebuilds two ledger pages once, then serves snapshots from memory', + ); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + test('persists direct-stop request and resolution before replaying after restart', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-workhub-stop-')); let store = createSessionStore(root); @@ -597,7 +698,7 @@ describe('Host WorkHub Coordination coordinator', () => { targetId = target.id; let retireCalls = 0; const workhub = coordinator(root, store, () => undefined, undefined, undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, 'payments-turn'), + assign: persistTestAssignmentAction(store, 'payments-turn'), retireDelegation: async () => { retireCalls += 1; return { outcome: 'stop_delivered', targetTurnId: 'payments-turn' }; @@ -687,6 +788,231 @@ describe('Host WorkHub Coordination coordinator', () => { } }); + test('a stop observes an assignment committed by a concurrent admitted action', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-concurrent-stop-')); + const store = createSessionStore(root); + const admission = new SessionAdmissionGate(); + const committed = deferred(); + const releaseAssignment = deferred(); + try { + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + const workhub = coordinator(root, store, () => undefined, undefined, undefined, admission, { + assign: (input, _context, publishCommittedAssignment) => + admission.runMany([WORKHUB_COORDINATION_SESSION_ID, input.targetSessionId], async () => { + const result = await persistTestAssignment(store, input, 'payments-turn'); + committed.resolve(); + await releaseAssignment.promise; + if (result.newAssignment) { + await publishCommittedAssignment(result.newAssignment); + } + return { turnId: result.turnId }; + }), + retireDelegation: async () => ({ outcome: 'cancelled_pending' }), + }); + assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); + const candidates = await workhub.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const candidate = candidates.result.candidates.find( + ({ sessionId }) => sessionId === target.id, + ); + assert.ok(candidate); + if (!candidate) return; + + const assignment = workhub.handlers['workhub.coordination.act']( + { + actionId: 'source-action', + userText: 'Fix payment retry', + candidateSetId: candidates.result.candidateSetId, + proposal: { disposition: 'delegate_existing', candidateRef: candidate.candidateRef }, + }, + CONTEXT, + ); + await committed.promise; + const stop = workhub.handlers['workhub.coordination.act']( + { + actionId: 'stop-action', + userText: 'Stop Payments', + proposal: { + disposition: 'stop_work', + expects: { targetSessionId: target.id }, + }, + confirmation: { kind: 'user_stop' }, + }, + CONTEXT, + ); + await new Promise((resolve) => setImmediate(resolve)); + releaseAssignment.resolve(); + + assert.equal((await assignment).ok, true); + const stopped = await stop; + assert.equal(stopped.ok, true, JSON.stringify(stopped)); + if (stopped.ok) assert.equal(stopped.result.disposition, 'stop_work'); + } finally { + releaseAssignment.resolve(); + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('persists one resume result after the Host starts the continuation', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-resume-')); + let store = createSessionStore(root); + let targetId = ''; + const resumeInput = () => ({ + actionId: 'resume-action', + userText: 'Resume Payments', + proposal: { + disposition: 'resume_work' as const, + expects: { targetSessionId: targetId }, + }, + }); + try { + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + targetId = target.id; + let resumeCalls = 0; + const workhub = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + assign: persistTestAssignmentAction(store, 'payments-turn'), + resumeDelegation: async () => { + resumeCalls += 1; + return { + outcome: 'resume_started', + targetTurnId: 'resumed-turn', + }; + }, + }); + assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); + const candidates = await workhub.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const candidate = candidates.result.candidates.find( + ({ sessionId }) => sessionId === target.id, + )!; + assert.equal( + ( + await workhub.handlers['workhub.coordination.act']( + { + actionId: 'source-action', + userText: 'Fix payment retry', + candidateSetId: candidates.result.candidateSetId, + proposal: { disposition: 'delegate_existing', candidateRef: candidate.candidateRef }, + }, + CONTEXT, + ) + ).ok, + true, + ); + + const resumed = await workhub.handlers['workhub.coordination.act'](resumeInput(), CONTEXT); + assert.deepEqual(resumed, { + ok: true, + result: { + disposition: 'resume_work', + outcome: 'resume_started', + targetSessionId: target.id, + targetTurnId: 'resumed-turn', + }, + }); + assert.equal(resumeCalls, 1); + const durable = await store.readWorkHubResume('resume-action'); + assert.equal(durable?.kind, 'delegation_resume'); + assert.equal(durable?.resumesActionId, 'source-action'); + assert.equal(durable?.targetSessionId, target.id); + assert.equal(durable?.outcome, 'resume_started'); + assert.equal(durable?.targetTurnId, 'resumed-turn'); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('reports recovery without durably parking a resume action', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-resume-recovering-')); + const store = createSessionStore(root); + try { + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + let recovering = true; + const workhub = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + assign: persistTestAssignmentAction(store, 'payments-turn'), + resumeDelegation: async () => { + if (recovering) { + throw new WorkHubActionEffectFailure( + 'host_not_ready', + 'WorkHub is still recovering the delegated execution', + ); + } + return { outcome: 'resume_started', targetTurnId: 'resumed-turn' }; + }, + }); + assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); + const candidates = await workhub.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const candidate = candidates.result.candidates.find( + ({ sessionId }) => sessionId === target.id, + )!; + assert.equal( + ( + await workhub.handlers['workhub.coordination.act']( + { + actionId: 'source-action', + userText: 'Fix payment retry', + candidateSetId: candidates.result.candidateSetId, + proposal: { disposition: 'delegate_existing', candidateRef: candidate.candidateRef }, + }, + CONTEXT, + ) + ).ok, + true, + ); + + const input = { + actionId: 'resume-after-recovery', + userText: 'Resume Payments', + proposal: { + disposition: 'resume_work' as const, + expects: { targetSessionId: target.id }, + }, + }; + assert.deepEqual(await workhub.handlers['workhub.coordination.act'](input, CONTEXT), { + ok: false, + error: { + code: 'host_not_ready', + message: 'WorkHub is still recovering the delegated execution', + }, + }); + assert.equal(await store.readWorkHubResume(input.actionId), undefined); + + recovering = false; + const retried = await workhub.handlers['workhub.coordination.act'](input, CONTEXT); + assert.equal(retried.ok, true); + if (retried.ok && retried.result.disposition === 'resume_work') { + assert.equal(retried.result.outcome, 'resume_started'); + } + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + test('rechecks sole-delegation stop preconditions after the advisory active-link read', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-workhub-stop-race-')); const store = createSessionStore(root); @@ -698,51 +1024,43 @@ describe('Host WorkHub Coordination coordinator', () => { model: 'test-model', permissionMode: 'ask', }); + const admission = new SessionAdmissionGate(); let injected = false; - const stores = new Proxy(store, { + let race: (() => Promise) | undefined; + const racingAdmission = new Proxy(admission, { get(authority, property, receiver) { - if (property === 'readMessagesSnapshot') { - return async (sessionId: string) => { - const messages = await authority.readMessagesSnapshot(sessionId); - if ( - !injected && - sessionId === WORKHUB_COORDINATION_SESSION_ID && - messages.some( - (message) => - message.type === 'workhub_coordination' && - message.kind === 'delegation_assigned' && - message.actionId === 'source-action', - ) - ) { + if (property === 'runMany') { + return async ( + sessionIds: readonly string[], + operation: Parameters[1], + ): Promise => { + if (!injected && race) { injected = true; - await persistTestAssignment( - authority, - { - actionId: 'racing-action', - actionFingerprint: `sha256:${'8'.repeat(64)}`, - targetSessionId: target.id, - targetSessionName: 'Payments', - disposition: 'delegate_existing', - userText: 'A second payment delegation', - }, - 'racing-turn', - ); + await race(); } - return messages; + return authority.runMany(sessionIds, operation) as Promise; }; } const value = Reflect.get(authority, property, receiver) as unknown; return typeof value === 'function' ? value.bind(authority) : value; }, - }) as SessionAuthorityStore; + }) as SessionAdmissionGate; let retireCalls = 0; - const workhub = coordinator(root, stores, () => undefined, undefined, undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, 'source-turn'), - retireDelegation: async () => { - retireCalls += 1; - return { outcome: 'cancelled_pending' }; + const workhub = coordinator( + root, + store, + () => undefined, + undefined, + undefined, + racingAdmission, + { + assign: persistTestAssignmentAction(store, (input) => `${input.actionId}-turn`), + retireDelegation: async () => { + retireCalls += 1; + return { outcome: 'cancelled_pending' }; + }, }, - }); + ); assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); const candidates = await workhub.handlers['workhub.coordination.candidates']({}, CONTEXT); assert.equal(candidates.ok, true); @@ -764,6 +1082,18 @@ describe('Host WorkHub Coordination coordinator', () => { ).ok, true, ); + race = async () => { + const raced = await workhub.handlers['workhub.coordination.act']( + { + actionId: 'racing-action', + userText: 'A second payment delegation', + candidateSetId: candidates.result.candidateSetId, + proposal: { disposition: 'delegate_existing', candidateRef: candidate.candidateRef }, + }, + CONTEXT, + ); + assert.equal(raced.ok, true); + }; const stopped = await workhub.handlers['workhub.coordination.act']( { @@ -838,7 +1168,7 @@ describe('Host WorkHub Coordination coordinator', () => { undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, 'payments-turn'), + assign: persistTestAssignmentAction(store, 'payments-turn'), retireDelegation: async () => ({ outcome: 'cancelled_pending' }), }, ); @@ -946,7 +1276,7 @@ describe('Host WorkHub Coordination coordinator', () => { }, }) as SessionAuthorityStore; const workhub = coordinator(root, stores, () => undefined, undefined, undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, 'payments-turn'), + assign: persistTestAssignmentAction(store, 'payments-turn'), retireDelegation: async () => ({ outcome: 'stop_delivered' as const, targetTurnId: 'payments-turn', @@ -1009,7 +1339,7 @@ describe('Host WorkHub Coordination coordinator', () => { } }); - test('a claimed stop refuses by name when its delegation was replaced', async () => { + test('a claimed stop refuses by name after restart when its delegation was replaced', async () => { // The claim survived a crash before its request. By the retry the link it // bound itself to is gone and another has taken its place on the same // Session, so re-deriving would silently bind this action to a delegation @@ -1027,7 +1357,7 @@ describe('Host WorkHub Coordination coordinator', () => { permissionMode: 'ask', }); const workhub = coordinator(root, store, () => undefined, undefined, undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, `${input.actionId}-turn`), + assign: persistTestAssignmentAction(store, (input) => `${input.actionId}-turn`), retireDelegation: async () => assert.fail('a spent stop identity must not retire work'), }); assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); @@ -1096,7 +1426,11 @@ describe('Host WorkHub Coordination coordinator', () => { 'successor-turn', ); - const refused = await workhub.handlers['workhub.coordination.act']( + const restarted = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + assign: persistTestAssignmentAction(store, (input) => `${input.actionId}-turn`), + retireDelegation: async () => assert.fail('a spent stop identity must not retire work'), + }); + const refused = await restarted.handlers['workhub.coordination.act']( { actionId: 'stop-action', userText: 'Stop Payments', @@ -1136,7 +1470,7 @@ describe('Host WorkHub Coordination coordinator', () => { permissionMode: 'ask', }); const workhub = coordinator(root, store, () => undefined, undefined, undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, 'payments-turn'), + assign: persistTestAssignmentAction(store, 'payments-turn'), retireDelegation: async () => assert.fail('a terminal delegation must not be retired'), }); assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); @@ -1245,7 +1579,7 @@ describe('Host WorkHub Coordination coordinator', () => { }, }) as SessionAdmissionGate; const workhub = coordinator(root, store, () => undefined, undefined, undefined, observed, { - assign: (input) => persistTestAssignment(store, input, `${input.actionId}-turn`), + assign: persistTestAssignmentAction(store, (input) => `${input.actionId}-turn`), retireDelegation: async () => ({ outcome: 'stop_delivered' as const, targetTurnId: 'source-action-turn', @@ -1305,85 +1639,6 @@ describe('Host WorkHub Coordination coordinator', () => { } }); - test('one stop reads the Coordination transcript twice, not once per proof', async () => { - // The Gate derives the delegation from the active links, then admission - // reproves it under the lease. Those are the two reads that decide. Any - // further pass re-derives an answer the stop already holds, on a transcript - // that only grows. - const root = await mkdtemp(join(tmpdir(), 'maka-workhub-scan-count-')); - const store = createSessionStore(root); - try { - const target = await store.create({ - cwd: root, - name: 'Payments', - llmConnectionSlug: 'test-connection', - model: 'test-model', - permissionMode: 'ask', - }); - let coordinationReads = 0; - let counting = false; - const stores = new Proxy(store, { - get(authority, property, receiver) { - if (property === 'readMessagesSnapshot') { - return async (sessionId: string) => { - if (counting && sessionId === WORKHUB_COORDINATION_SESSION_ID) coordinationReads += 1; - return authority.readMessagesSnapshot(sessionId); - }; - } - const value = Reflect.get(authority, property, receiver) as unknown; - return typeof value === 'function' ? value.bind(authority) : value; - }, - }) as SessionAuthorityStore; - const workhub = coordinator(root, stores, () => undefined, undefined, undefined, undefined, { - assign: (input) => persistTestAssignment(store, input, 'payments-turn'), - retireDelegation: async () => ({ - outcome: 'stop_delivered' as const, - targetTurnId: 'payments-turn', - }), - }); - assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); - const candidates = await workhub.handlers['workhub.coordination.candidates']({}, CONTEXT); - assert.equal(candidates.ok, true); - if (!candidates.ok) return; - assert.equal( - ( - await workhub.handlers['workhub.coordination.act']( - { - actionId: 'source-action', - userText: 'Fix payment retry', - candidateSetId: candidates.result.candidateSetId, - proposal: { - disposition: 'delegate_existing', - candidateRef: candidates.result.candidates.find( - ({ sessionId }) => sessionId === target.id, - )!.candidateRef, - }, - }, - CONTEXT, - ) - ).ok, - true, - ); - - counting = true; - const stopped = await workhub.handlers['workhub.coordination.act']( - { - actionId: 'stop-action', - userText: 'Stop Payments', - proposal: { disposition: 'stop_work', expects: { targetSessionId: target.id } }, - confirmation: { kind: 'user_stop' }, - }, - CONTEXT, - ); - - assert.equal(stopped.ok, true); - assert.equal(coordinationReads, 2, 'a stop derives once and reproves once'); - } 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); @@ -1518,10 +1773,13 @@ function coordinator( hasRootTurnAdmission: async () => false, }, admission: SessionAdmissionGate = new SessionAdmissionGate(), - sessionActions: Partial< - Pick - > = {}, + sessionActions: Partial = {}, ) { + const assign = + sessionActions.assign ?? + (async ({ targetSessionId }: Parameters[0]) => ({ + turnId: `turn-${targetSessionId}`, + })); return new HostWorkHubCoordinationCoordinator({ stateRoot: root, stores: store, @@ -1529,10 +1787,14 @@ function coordinator( continuity: { refreshCanonical: async () => undefined }, executions, sessionActions: { - assign: async ({ targetSessionId }) => ({ turnId: `turn-${targetSessionId}` }), readDelegationRetirement: async () => 'not_retired', + resumeDelegation: async () => ({ + outcome: 'resume_started' as const, + targetTurnId: 'resumed-turn', + }), retireDelegation: async () => ({ outcome: 'cancelled_pending' }), ...sessionActions, + assign, }, resolveCreateTarget: resolveCreateTarget ?? @@ -1551,7 +1813,14 @@ async function persistTestAssignment( store: SessionAuthorityStore, input: Parameters[0], targetTurnId: string, -): Promise<{ turnId: string }> { +): Promise< + { readonly turnId: string } & { + readonly newAssignment?: { + readonly sequence: number; + readonly assignment: WorkHubDelegationAssignedMessage; + }; + } +> { const suffix = createHash('sha256').update(input.actionId, 'utf8').digest('hex').slice(0, 48); const content = normalizeMessageContent({ text: input.userText }); const result = await store.assignWorkHubMessage({ @@ -1588,5 +1857,27 @@ async function persistTestAssignment( admittedAt: Date.now(), }, }); - return { turnId: result.assignment.targetTurnId }; + return { + turnId: result.assignment.targetTurnId, + ...(result.kind === 'assigned' + ? { newAssignment: { sequence: result.sequence, assignment: result.assignment } } + : {}), + }; +} + +function persistTestAssignmentAction( + store: SessionAuthorityStore, + targetTurnId: string | ((input: Parameters[0]) => string), +): HostWorkHubCoordinationCoordinatorOptions['sessionActions']['assign'] { + return async (input, _context, publishCommittedAssignment) => { + const result = await persistTestAssignment( + store, + input, + typeof targetTurnId === 'string' ? targetTurnId : targetTurnId(input), + ); + if (result.newAssignment) { + await publishCommittedAssignment(result.newAssignment); + } + return { turnId: result.turnId }; + }; } diff --git a/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts index 25cd512c4e..7601be7af1 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts @@ -189,7 +189,97 @@ test('WorkHub Coordination answer and summary inputs are closed and bounded', () ); }); +test('WorkHub Coordination resume has closed input and outcome shapes', () => { + assert.deepEqual( + decodeWorkHubCoordinationActInput({ + actionId: 'action-resume', + userText: 'Resume Payments', + proposal: { + disposition: 'resume_work', + expects: { targetSessionId: 'payments' }, + }, + }), + { + actionId: 'action-resume', + userText: 'Resume Payments', + proposal: { + disposition: 'resume_work', + expects: { targetSessionId: 'payments' }, + }, + }, + ); + + for (const invalid of [ + { + actionId: 'action-resume-confirmed', + userText: 'Resume Payments', + proposal: { disposition: 'resume_work', expects: { targetSessionId: 'payments' } }, + confirmation: { kind: 'user_stop' }, + }, + { + actionId: 'action-resume-missing-target', + userText: 'Resume Payments', + proposal: { disposition: 'resume_work' }, + }, + { + actionId: 'action-resume-injected', + userText: 'Resume Payments', + proposal: { + disposition: 'resume_work', + expects: { targetSessionId: 'payments' }, + targetSessionId: 'injected', + }, + }, + ]) { + assert.throws( + () => decodeWorkHubCoordinationActInput(invalid), + (error) => error instanceof RuntimeHostProtocolError, + ); + } + + for (const result of [ + { + disposition: 'resume_work', + outcome: 'resume_started', + targetSessionId: 'payments', + targetTurnId: 'turn-2', + }, + { + disposition: 'resume_work', + outcome: 'already_running', + targetSessionId: 'payments', + }, + ]) { + assert.deepEqual(decodeWorkHubCoordinationActResult(result), result); + } + + for (const invalid of [ + { + disposition: 'resume_work', + outcome: 'parked', + targetSessionId: 'payments', + }, + { + disposition: 'resume_work', + outcome: 'already_running', + targetSessionId: 'payments', + parkReason: 'safety_check_failed', + }, + { + disposition: 'resume_work', + outcome: 'resume_started', + targetSessionId: 'payments', + }, + ]) { + assert.throws( + () => decodeWorkHubCoordinationActResult(invalid), + (error) => error instanceof RuntimeHostProtocolError, + ); + } +}); + test('WorkHub Coordination candidates are bounded and carry opaque proposal identities', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 110); const result = decodeWorkHubCoordinationCandidatesResult({ candidateSetId: `sha256:${'a'.repeat(64)}`, candidates: [ @@ -205,8 +295,12 @@ test('WorkHub Coordination candidates are bounded and carry opaque proposal iden updatedAt: 7, }, ], + delegations: [{ actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }], }); assert.equal(result.candidates[0]?.candidateRef, 'candidate_a'); + assert.deepEqual(result.delegations, [ + { actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }, + ]); assert.equal(HOST_OPERATION_SPECS['workhub.coordination.candidates'].mode, 'query'); assert.equal(REMOTE_OWNER_OPERATION_GRANTS.includes('workhub.coordination.candidates'), true); assert.throws( @@ -214,6 +308,7 @@ test('WorkHub Coordination candidates are bounded and carry opaque proposal iden decodeWorkHubCoordinationCandidatesResult({ candidateSetId: 'caller-invented', candidates: [], + delegations: [], }), (error) => error instanceof RuntimeHostProtocolError, ); diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index b58467c6ee..efd451fd35 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -101,7 +101,12 @@ 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 = 112 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 114 as const; +// 114: WorkHub Coordination admits a `resume_work` proposal and answers with a +// `resume_work` outcome. An older peer's closed decoder rejects both the +// disposition it does not know and the result kind it cannot read. +// 113: WorkHub exposes current active delegation linkage in the existing Host +// snapshot. Older peers disagree on the closed result shape. // 112: Owners can query the Host execution environment through an extensible, // bounded resource-envelope contract. Older Hosts do not implement the query. // 111: Client Capability tool schemas may use draft-07 tuple additionalItems. diff --git a/packages/runtime-host/src/protocol/workhub-coordination.ts b/packages/runtime-host/src/protocol/workhub-coordination.ts index 735c6a840e..06f372ad6f 100644 --- a/packages/runtime-host/src/protocol/workhub-coordination.ts +++ b/packages/runtime-host/src/protocol/workhub-coordination.ts @@ -112,11 +112,21 @@ export interface WorkHubCoordinationCandidate { export type WorkHubCoordinationCandidatesInput = Record; +export interface WorkHubCoordinationActiveDelegation { + readonly actionId: string; + readonly targetSessionId: string; + readonly sequence: number; +} + export interface WorkHubCoordinationCandidatesResult { readonly candidateSetId: string; readonly candidates: readonly WorkHubCoordinationCandidate[]; } +export interface WorkHubCoordinationSnapshotResult extends WorkHubCoordinationCandidatesResult { + readonly delegations: readonly WorkHubCoordinationActiveDelegation[]; +} + export type WorkHubCoordinationProposal = | { readonly disposition: 'answer_here' } | { readonly disposition: 'clarify'; readonly assistantText: string } @@ -146,6 +156,16 @@ export type WorkHubCoordinationProposal = * links, and on replay from the durable claim this action already owns. */ readonly expects: WorkHubCoordinationStopPreconditions; + } + | { + readonly disposition: 'resume_work'; + /** + * Resume names the Session it resolved and nothing else, for the same + * reason a stop does: which delegation is live is the Host's to know. + * The Gate revalidates it, so a resolution that has gone stale fails + * closed instead of restarting work the user never named. + */ + readonly expects: WorkHubCoordinationStopPreconditions; }; export interface WorkHubCoordinationStopPreconditions { @@ -202,6 +222,12 @@ export type WorkHubCoordinationActResult = readonly outcome: 'cancelled_pending' | 'stop_delivered' | 'already_terminal' | 'not_owned'; readonly targetSessionId: string; readonly targetTurnId?: string; + } + | { + readonly disposition: 'resume_work'; + readonly outcome: 'resume_started' | 'already_running'; + readonly targetSessionId: string; + readonly targetTurnId?: string; }; export const WORKHUB_COORDINATION_OPERATION_SPECS = { @@ -240,7 +266,7 @@ export const WORKHUB_COORDINATION_OPERATION_SPECS = { }), 'workhub.coordination.candidates': defineOperation< WorkHubCoordinationCandidatesInput, - WorkHubCoordinationCandidatesResult, + WorkHubCoordinationSnapshotResult, (typeof CANDIDATE_ERRORS)[number] >({ mode: 'query', @@ -331,12 +357,13 @@ export function decodeWorkHubCoordinationCandidatesInput( export function decodeWorkHubCoordinationCandidatesResult( value: unknown, -): WorkHubCoordinationCandidatesResult { +): WorkHubCoordinationSnapshotResult { const result = requireExactRecord(value, 'WorkHub Coordination candidates result', [ 'candidateSetId', 'candidates', + 'delegations', ]); - if (!Array.isArray(result.candidates)) { + if (!Array.isArray(result.candidates) || !Array.isArray(result.delegations)) { throw invalidProtocolFrame('Invalid WorkHub Coordination candidates'); } if (result.candidates.length > WORKHUB_COORDINATION_CANDIDATE_MAX_ITEMS) { @@ -345,6 +372,18 @@ export function decodeWorkHubCoordinationCandidatesResult( return { candidateSetId: candidateSetId(result.candidateSetId), candidates: result.candidates.map(decodeWorkHubCoordinationCandidate), + delegations: result.delegations.map((value) => { + const delegation = requireExactRecord(value, 'Active WorkHub delegation', [ + 'actionId', + 'targetSessionId', + 'sequence', + ]); + return { + actionId: requireEntityId(delegation.actionId, 'WorkHub action id'), + targetSessionId: requireEntityId(delegation.targetSessionId, 'WorkHub target Session id'), + sequence: requireCount(delegation.sequence, 'WorkHub transcript sequence'), + }; + }), }; } @@ -516,6 +555,32 @@ export function decodeWorkHubCoordinationActResult(value: unknown): WorkHubCoord }), }; } + if (result.disposition === 'resume_work') { + const exact = requireShapedRecord( + result, + 'WorkHub Coordination resume result', + ['disposition', 'outcome', 'targetSessionId'], + ['targetTurnId'], + ); + if (exact.outcome !== 'resume_started' && exact.outcome !== 'already_running') { + throw invalidProtocolFrame('Invalid WorkHub resume outcome'); + } + // Only a started continuation names a Turn: the Host has one to name, and + // the other two outcomes changed nothing that could carry an identity. + if ((exact.outcome === 'resume_started') !== (exact.targetTurnId !== undefined)) { + throw invalidProtocolFrame('Invalid WorkHub resume target Turn'); + } + return { + disposition: 'resume_work', + outcome: exact.outcome, + targetSessionId: requireEntityId(exact.targetSessionId, 'WorkHub target Session id'), + ...(exact.targetTurnId === undefined + ? {} + : { + targetTurnId: requireEntityId(exact.targetTurnId, 'WorkHub target Turn id'), + }), + }; + } throw invalidProtocolFrame('Invalid WorkHub Coordination action disposition'); } @@ -626,6 +691,16 @@ function decodeWorkHubCoordinationProposal(value: unknown): WorkHubCoordinationP expects: decodeWorkHubCoordinationStopPreconditions(exact.expects), }; } + if (proposal.disposition === 'resume_work') { + const exact = requireExactRecord(proposal, 'WorkHub resume proposal', [ + 'disposition', + 'expects', + ]); + return { + disposition: 'resume_work', + expects: decodeWorkHubCoordinationStopPreconditions(exact.expects), + }; + } throw invalidProtocolFrame('Invalid WorkHub Coordination proposal disposition'); } diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 2e38c38f8d..c2e7211dc0 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -33,6 +33,7 @@ import { import { isDeepResearchSession, type SessionHeader, + type WorkHubDelegationAssignedMessage, WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, } from '@maka/core/session'; @@ -184,7 +185,10 @@ import type { TurnOperationHandlerMap } from './operation-dispatcher.js'; import { HostUsagePricingCoordinator } from './usage-pricing-coordinator.js'; import { HostWebSearchCoordinator } from './web-search-coordinator.js'; import { HostWorkHubCoordinationCoordinator } from './workhub-coordination-coordinator.js'; -import { WorkHubActionEffectFailure } from './workhub-coordination-action-gate.js'; +import { + WorkHubActionEffectFailure, + workHubResumedTurnId, +} from './workhub-coordination-action-gate.js'; type ExecutionConnectionRef = Parameters< RuntimePolicyStoresWriter['operations']['resolveExecutionConnection'] @@ -1387,12 +1391,109 @@ export async function createExecutionRuntimeHostComposition( turnId: disposition.turnId, runId: disposition.runId, }; - if (isActiveWorkHubRoot(coordinator, identity)) return 'not_retired'; + const latest = await coordinator.readLatestRootTurnLineage(identity); + if (isActiveWorkHubRoot(coordinator, latest)) return 'not_retired'; // The same restart window as `stopOwnedWorkHubRoot`: an unregistered // root is not evidence that its work ended. - const snapshot = await coordinator.read(identity); + const snapshot = await coordinator.read(latest); return isHostedExecutionTerminal(snapshot) ? 'retired' : 'recovering'; }, + // Resolve and resume only the execution lineage owned by this + // delegation. A Session-wide latest-failure query could otherwise + // continue unrelated work started directly in the same Session. + resumeDelegation: async (assignment, context) => { + const disposition = await messages.readMessageExecutionDisposition( + assignment.targetSessionId, + assignment.targetMessageId, + ); + if (disposition.kind === 'recovering') { + throw new WorkHubActionEffectFailure( + 'host_not_ready', + 'WorkHub is still recovering the delegated execution', + ); + } + if (disposition.kind !== 'owned_root') { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub delegated execution is not resumable', + ); + } + const source = await coordinator.readLatestRootTurnLineage({ + sessionId: assignment.targetSessionId, + turnId: disposition.turnId, + runId: disposition.runId, + }); + if (isActiveWorkHubRoot(coordinator, source)) { + return { outcome: 'already_running' as const }; + } + const snapshot = await coordinator.read(source); + if (!isHostedExecutionTerminal(snapshot)) { + throw new WorkHubActionEffectFailure( + 'host_not_ready', + 'WorkHub is still recovering the delegated execution', + ); + } + if (snapshot.status !== 'failed' && snapshot.status !== 'cancelled') { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub delegated execution is not resumable', + ); + } + const plan = await coordinator.handlers['turn.resume.query']( + { sessionId: assignment.targetSessionId, sourceRunId: source.runId }, + context, + ); + if (!plan.ok) throw new WorkHubActionEffectFailure(plan.error.code, plan.error.message); + if (plan.result.disposition === 'parked') { + throw new WorkHubActionEffectFailure( + plan.result.reason === 'resume_feature_disabled' + ? 'operation_unavailable' + : 'operation_conflict', + plan.result.reason === 'resume_feature_disabled' + ? 'Safe-boundary resume is disabled for this Runtime Host' + : 'WorkHub delegated execution is not resumable', + ); + } + if ( + plan.result.sourceRunId !== source.runId || + plan.result.sourceTurnId !== source.turnId + ) { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub resume source lineage changed during planning', + ); + } + const targetTurnId = workHubResumedTurnId( + assignment.delegationId, + plan.result.sourceRunId, + ); + const started = await coordinator.handlers['turn.resume.start']( + { + sessionId: assignment.targetSessionId, + turnId: targetTurnId, + sourceRunId: plan.result.sourceRunId, + sourceRuntimeEventHighWater: plan.result.sourceRuntimeEventHighWater, + }, + context, + ); + if (!started.ok) { + throw new WorkHubActionEffectFailure(started.error.code, started.error.message); + } + if (started.result.kind === 'parked') { + throw new WorkHubActionEffectFailure( + started.result.plan.reason === 'resume_feature_disabled' + ? 'operation_unavailable' + : 'operation_conflict', + started.result.plan.reason === 'resume_feature_disabled' + ? 'Safe-boundary resume is disabled for this Runtime Host' + : 'WorkHub delegated execution is not resumable', + ); + } + return { + outcome: 'resume_started' as const, + targetTurnId: started.result.turn.turnId, + }; + }, retireDelegation: async (assignment, retirement) => { const disposition = await messages.cancelMessageIfPending( assignment.targetSessionId, @@ -1412,11 +1513,11 @@ export async function createExecutionRuntimeHostComposition( return { outcome: 'not_owned' as const, targetTurnId: disposition.turnId }; } if (disposition.kind === 'owned_root') { - const identity = { + const identity = await coordinator.readLatestRootTurnLineage({ sessionId: assignment.targetSessionId, turnId: disposition.turnId, runId: disposition.runId, - }; + }); return retirement.cause === 'direct_stop' ? stopOwnedWorkHubRoot(coordinator, identity, retirement.cancellationClaimId) : stopReplacedWorkHubRoot(coordinator, identity); @@ -1424,7 +1525,7 @@ export async function createExecutionRuntimeHostComposition( disposition satisfies never; throw new Error('Unhandled WorkHub Message retirement disposition'); }, - assign: async (input) => { + assign: async (input, _context, publishCommittedAssignment) => { const durable = await stores.sessionStore.readWorkHubAssignment(input.actionId); if (durable) { await messages.consumePendingAdmissions([durable.targetSessionId]); @@ -1528,6 +1629,15 @@ export async function createExecutionRuntimeHostComposition( ...(create ? { create } : {}), ...(supersession ? { supersession } : {}), }); + if (result.kind === 'assigned') { + // Publish the rebuildable Host index before this shared + // admission is released, so readers cannot observe the + // durable assignment without its active linkage. + await publishCommittedAssignment({ + sequence: result.sequence, + assignment: result.assignment, + }); + } // Keep the durable steering identity and its live queue owner // under one Session admission. A terminal transition must not // observe the committed Message before the queue does. diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index f5db2de586..6d2a53ef8b 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -519,6 +519,30 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { return this.#admissions.has(sessionId) ? { kind: 'reserved' } : { kind: 'idle' }; } + /** Returns the newest Host-admitted continuation descended from one root execution. */ + async readLatestRootTurnLineage(identity: HostedExecutionRef): Promise { + const origin = await this.stores.agentRunStore.readRootTurnAdmission( + identity.sessionId, + identity.turnId, + ); + if (!origin || origin.runId !== identity.runId) { + throw new RuntimeMessageAuthorityInvariantError( + `Root execution ${identity.turnId}/${identity.runId} has no durable admission`, + ); + } + let latest = origin; + while (true) { + const continuation = await this.stores.agentRunStore.readRootTurnContinuationAdmission( + identity.sessionId, + latest.turnId, + latest.runId, + ); + if (!continuation) break; + latest = continuation; + } + return { sessionId: latest.sessionId, turnId: latest.turnId, runId: latest.runId }; + } + startHostedExternalTransition( input: HostedExternalTurnTransitionInput, context: ConnectionContext, 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 bea4056209..01575074c7 100644 --- a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts +++ b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts @@ -42,6 +42,7 @@ import { readWorkHubRequestIntent, workHubCorrectionTargetsSession, workHubCreationAuthorizesTitle, + workHubNamedDelegationActionTargetsSession, } from '@maka/core/workhub-creation-intent'; import type { WorkHubCoordinationActInput, @@ -141,6 +142,10 @@ export interface WorkHubActionGateEffects { assignment: WorkHubDelegationAssignedMessage, retirement: WorkHubDelegationRetirementClaim, ): Promise; + resume( + input: WorkHubDelegationResumeInput, + context: ConnectionContext, + ): Promise>; } /** @@ -155,6 +160,14 @@ export interface WorkHubDelegationRetirementClaim { readonly cause: 'direct_stop' | 'replacement'; } +export interface WorkHubDelegationResumeInput { + readonly actionId: string; + readonly actionFingerprint: `sha256:${string}`; + readonly source: WorkHubDelegationAssignedMessage; + readonly targetSessionName: string; + readonly userText: string; +} + export interface WorkHubRetirementResult { readonly outcome: WorkHubDelegationStopOutcome | 'recovering'; readonly targetTurnId?: string; @@ -422,6 +435,39 @@ export class WorkHubCoordinationActionGate { }); return this.#stop(requested, source); } + if (proposal.disposition === 'resume_work') { + if (!requestIntent.resume.imperative) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub resume requires an explicit named command in trusted user text', + ); + } + const source = await this.#resumeSource(proposal.expects.targetSessionId); + const sessions = await this.#effects.listSessions(); + const currentTargetName = sessions.find(({ id }) => id === source.targetSessionId)?.name; + if ( + !currentTargetName || + !workHubNamedDelegationActionTargetsSession(requestIntent.resume, currentTargetName) + ) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub resume target is not affirmed in trusted user text', + ); + } + const resumeFingerprint = resumeActionFingerprint(input, source); + await this.#claimAction(input.actionId, 'resume', resumeFingerprint, source.delegationId); + return this.#effects.resume( + { + actionId: input.actionId, + actionFingerprint: resumeFingerprint, + source, + targetSessionName: currentTargetName, + userText: input.userText, + }, + context, + ); + } + if (proposal.disposition === 'create_new') { if (!input.create || !workHubCreationAuthorizesTitle(requestIntent, proposal.title)) { throw new WorkHubActionGateFailure( @@ -524,6 +570,10 @@ export class WorkHubCoordinationActionGate { ); } + async #resumeSource(targetSessionId: string): Promise { + return this.#soleWorkingDelegation(targetSessionId, 'resume'); + } + /** * The delegation a stop names, by the only two keys that can name it. * @@ -564,12 +614,32 @@ export class WorkHubCoordinationActionGate { return claimed; } } + const resolved = await this.#soleWorkingDelegation(targetSessionId, 'stop'); + // A claim with no request behind it resolves from the active links like a + // first attempt, but only while those links still name the delegation it + // bound itself to. If that one left and another took its place, the + // fingerprint derived here would no longer match the claim, and since + // claims are never deleted the refusal would be permanent and unexplained. + // Say why instead: the identity is spent, and the retry needs a new one. + if (claim?.operation === 'stop' && resolved.delegationId !== claim.subject) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub stop identity is already bound to a different delegation', + ); + } + return resolved; + } + + async #soleWorkingDelegation( + targetSessionId: string, + operation: 'resume' | 'stop', + ): Promise { const active = await this.#effects.listActiveAssignments(); const onTarget = active.filter((assignment) => assignment.targetSessionId === targetSessionId); if (onTarget.length === 0) { throw new WorkHubActionGateFailure( 'action_conflict', - 'WorkHub has no active durable delegation to stop on that Session', + `WorkHub has no active durable delegation to ${operation} on that Session`, ); } // One link is the answer whatever state its work is in. Whether that work @@ -592,23 +662,11 @@ export class WorkHubCoordinationActionGate { if (holdingWork.length !== 1) { throw new WorkHubActionGateFailure( 'action_conflict', - 'WorkHub stop target does not identify one active durable delegation', + `WorkHub ${operation} target does not identify one active durable delegation`, ); } resolved = holdingWork[0]!; } - // A claim with no request behind it resolves from the active links like a - // first attempt, but only while those links still name the delegation it - // bound itself to. If that one left and another took its place, the - // fingerprint derived here would no longer match the claim, and since - // claims are never deleted the refusal would be permanent and unexplained. - // Say why instead: the identity is spent, and the retry needs a new one. - if (claim?.operation === 'stop' && resolved.delegationId !== claim.subject) { - throw new WorkHubActionGateFailure( - 'action_conflict', - 'WorkHub stop identity is already bound to a different delegation', - ); - } return resolved; } @@ -1048,6 +1106,10 @@ function workHubCreatedSessionId(actionId: string): string { return `whs_${hash(`create\0${actionId}`).slice(0, 48)}`; } +export function workHubResumedTurnId(delegationId: string, sourceRunId: string): string { + return `wht_${hash(`resume\0${delegationId}\0${sourceRunId}`).slice(0, 48)}`; +} + function workspaceProjection(session: WorkHubActionGateSession): WorkspaceProjection { return { target: @@ -1119,6 +1181,23 @@ function replacementActionFingerprint( }); } +function resumeActionFingerprint( + input: WorkHubCoordinationActInput, + source: WorkHubDelegationAssignedMessage, +): `sha256:${string}` { + if (input.proposal.disposition !== 'resume_work') { + throw new WorkHubActionGateFailure('action_conflict', 'Invalid WorkHub resume replay'); + } + return digest({ + userText: input.userText, + disposition: 'resume_work', + resumesActionId: source.actionId, + resumesDelegationId: source.delegationId, + targetSessionId: source.targetSessionId, + targetMessageId: source.targetMessageId, + }); +} + function stopActionFingerprint( input: WorkHubCoordinationActInput, source: WorkHubDelegationAssignedMessage, diff --git a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts index 80eaf04b2a..1daa38e546 100644 --- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts +++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts @@ -28,6 +28,7 @@ import { WORKHUB_COORDINATION_SESSION_ROLE, WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, WORKHUB_COORDINATION_STOP_SCHEMA_VERSION, + WORKHUB_COORDINATION_RESUME_SCHEMA_VERSION, isWorkHubCoordinationSession, isWorkHubCoordinationSessionId, type SessionHeader, @@ -37,10 +38,16 @@ import { type WorkHubDelegationReplacementRequestedMessage, type WorkHubDelegationStopRequestedMessage, type WorkHubDelegationStopResolvedMessage, + type WorkHubDelegationResumeMessage, } from '@maka/core/session'; -import type { SessionAuthorityStore, SessionHeaderSnapshot } from '@maka/storage/session-store'; +import { + isSessionNotFoundError, + type SessionAuthorityStore, + type SessionHeaderSnapshot, +} from '@maka/storage/session-store'; import type { OperationOutcome, + WorkHubCoordinationActResult, WorkHubCoordinationActInput, WorkHubCoordinationAnswerInput, WorkHubCoordinationRecordInput, @@ -62,6 +69,7 @@ import { WorkHubActionGateFailure, WorkHubCoordinationActionGate, type WorkHubActionGateEffects, + type WorkHubDelegationResumeInput, } from './workhub-coordination-action-gate.js'; const CREATE_FINGERPRINT = `sha256:${createHash('sha256') @@ -82,6 +90,13 @@ const COORDINATION_SUMMARY_READ_MAX_BYTES = JSON_ESCAPE_MAX_BYTES_PER_INPUT_BYTE * (WORKHUB_COORDINATION_TEXT_MAX_BYTES + WORKHUB_COORDINATION_SUMMARY_MAX_BYTES) + 16 * 1024; +const ACTIVE_DELEGATION_SCAN_MAX_BYTES = 4 * 1024 * 1024; +const ACTIVE_DELEGATION_SCAN_MAX_MESSAGES = 256; + +interface ActiveWorkHubAssignment { + readonly sequence: number; + readonly assignment: WorkHubDelegationAssignedMessage; +} type CoordinationStores = Pick< SessionAuthorityStore, @@ -93,15 +108,16 @@ type CoordinationStores = Pick< | 'probeSessionRemoval' | 'probeStableSessionCreate' | 'readHeaderSnapshot' - | 'readMessagesSnapshot' | 'readWorkHubAssignment' | 'readWorkHubReplacement' | 'readWorkHubReplacementAbort' | 'readWorkHubSupersession' | 'readWorkHubStopRequest' | 'readWorkHubStopResolution' + | 'readWorkHubResume' | 'readTranscriptHighWaterSnapshot' | 'readTranscriptMessagesSnapshot' + | 'readTranscriptRecordsSnapshot' | 'updateHeaderVersioned' >; @@ -110,6 +126,29 @@ type CoordinationExecutions = Pick< 'startWorkHubCoordinationMessage' | 'hasRootTurnAdmission' >; +type WorkHubResumeResult = + | { + readonly outcome: 'resume_started'; + readonly targetTurnId: string; + } + | { readonly outcome: 'already_running' }; + +type CoordinationSessionActions = Pick< + WorkHubActionGateEffects, + 'readDelegationRetirement' | 'retireDelegation' +> & { + assign( + ...args: [ + ...Parameters, + publishCommittedAssignment: (assignment: ActiveWorkHubAssignment) => Promise, + ] + ): ReturnType; + resumeDelegation( + assignment: WorkHubDelegationAssignedMessage, + context: ConnectionContext, + ): Promise; +}; + export type CoordinationCreateTarget = Omit; export interface HostWorkHubCoordinationCoordinatorOptions { @@ -118,10 +157,7 @@ export interface HostWorkHubCoordinationCoordinatorOptions { readonly admission: SessionAdmissionGate; readonly continuity: Pick; readonly executions: CoordinationExecutions; - readonly sessionActions: Pick< - WorkHubActionGateEffects, - 'assign' | 'readDelegationRetirement' | 'retireDelegation' - >; + readonly sessionActions: CoordinationSessionActions; readonly resolveCreateTarget: () => Promise; readonly requestDrain: () => void; } @@ -145,6 +181,7 @@ export class HostWorkHubCoordinationCoordinator { readonly #requestDrain: () => void; readonly #actionGate: WorkHubCoordinationActionGate; readonly #readDelegationRetirement: HostWorkHubCoordinationCoordinatorOptions['sessionActions']['readDelegationRetirement']; + #activeAssignments: Promise> | undefined; constructor(options: HostWorkHubCoordinationCoordinatorOptions) { this.#coordinationCwd = join(options.stateRoot, COORDINATION_CWD_DIRECTORY); @@ -170,7 +207,8 @@ export class HostWorkHubCoordinationCoordinator { probeTargetRemoval: async (sessionId) => (await this.#stores.probeSessionRemoval(sessionId)).kind, readAssignment: (actionId) => this.#stores.readWorkHubAssignment(actionId), - listActiveAssignments: () => this.#listActiveAssignments(), + listActiveAssignments: () => + this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, () => this.#listActiveAssignments()), readReplacement: (delegationId) => this.#stores.readWorkHubReplacement(delegationId), readReplacementAbort: (delegationId) => this.#stores.readWorkHubReplacementAbort(delegationId), @@ -193,13 +231,18 @@ export class HostWorkHubCoordinationCoordinator { throw new WorkHubActionEffectFailure(outcome.error.code, outcome.error.message); } }, - assign: options.sessionActions.assign, + assign: async (input, context) => { + return options.sessionActions.assign(input, context, (assignment) => + this.#applyActiveMessage(assignment.sequence, assignment.assignment), + ); + }, prepareReplacement: (input) => this.#prepareReplacement(input), abortReplacement: (input) => this.#abortReplacement(input), prepareStop: (input) => this.#prepareStop(input), resolveStop: (input) => this.#resolveStop(input), readDelegationRetirement: options.sessionActions.readDelegationRetirement, retireDelegation: options.sessionActions.retireDelegation, + resume: (input, context) => this.#resume(input, context, options.sessionActions), }); } @@ -286,10 +329,10 @@ export class HostWorkHubCoordinationCoordinator { }), conflictMessage: 'WorkHub delegation already has a different stop claim', beforeAppend: async () => { - const [replacement, supersession, messages] = await Promise.all([ + const [replacement, supersession, activeAssignments] = await Promise.all([ this.#stores.readWorkHubReplacement(input.stopsDelegationId), this.#stores.readWorkHubSupersession(input.stopsDelegationId), - this.#stores.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID), + this.#listActiveAssignments(), ]); if (replacement || supersession) { throw new WorkHubActionGateFailure( @@ -297,7 +340,6 @@ export class HostWorkHubCoordinationCoordinator { 'WorkHub delegation is already being replaced', ); } - const activeAssignments = activeWorkHubAssignments(messages); // Held lanes make this the last moment the one-target proof can change. // It is proved from opaque delegation identity, so a concurrent rename // is harmless while a concurrent delegation to the same Session is not. @@ -333,9 +375,7 @@ export class HostWorkHubCoordinationCoordinator { } async #listActiveAssignments(): Promise { - return activeWorkHubAssignments( - await this.#stores.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID), - ); + return (await this.#listActiveAssignmentRecords()).map(({ assignment }) => assignment); } #resolveStop( @@ -375,6 +415,54 @@ export class HostWorkHubCoordinationCoordinator { }); } + async #resume( + input: WorkHubDelegationResumeInput, + context: ConnectionContext, + actions: CoordinationSessionActions, + ): Promise> { + const existing = await this.#stores.readWorkHubResume(input.actionId); + if (existing) return coordinationResumeResult(existing); + const resumed = await actions.resumeDelegation(input.source, context); + const suffix = createHash('sha256').update(input.actionId, 'utf8').digest('hex').slice(0, 48); + const resolution = await this.#commitCoordinationFact({ + admissionSessionIds: [WORKHUB_COORDINATION_SESSION_ID, input.source.targetSessionId], + read: () => this.#stores.readWorkHubResume(input.actionId), + build: (durable) => ({ + type: 'workhub_coordination', + id: `whn_${suffix}`, + turnId: input.actionId, + ts: durable?.ts ?? Date.now(), + schemaVersion: WORKHUB_COORDINATION_RESUME_SCHEMA_VERSION, + kind: 'delegation_resume', + actionId: input.actionId, + actionFingerprint: input.actionFingerprint, + coordinationTurnId: input.actionId, + resumesActionId: input.source.actionId, + resumesDelegationId: input.source.delegationId, + targetSessionId: input.source.targetSessionId, + targetSessionName: input.targetSessionName, + userText: input.userText, + outcome: resumed.outcome, + ...(resumed.outcome === 'resume_started' ? { targetTurnId: resumed.targetTurnId } : {}), + }), + conflictMessage: 'WorkHub resume already has a different resolution', + beforeAppend: async () => { + const source = (await this.#listActiveAssignments()).find( + ({ actionId, delegationId }) => + actionId === input.source.actionId && delegationId === input.source.delegationId, + ); + if (!source || source.targetSessionId !== input.source.targetSessionId) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub resume source is no longer active', + ); + } + }, + unknownOutcomeMessage: 'WorkHub resume resolution outcome is unknown', + }); + return coordinationResumeResult(resolution); + } + #abortReplacement( input: Parameters[0], ): Promise { @@ -435,11 +523,15 @@ export class HostWorkHubCoordinationCoordinator { await options.beforeAppend(); try { await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [requested]); + await this.#applyLatestActiveMessage(requested); await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); return requested; } catch { const replay = await options.read().catch(() => undefined); - if (replay && isDeepStrictEqual(replay, requested)) return replay; + if (replay && isDeepStrictEqual(replay, requested)) { + await this.#applyLatestActiveMessage(replay); + return replay; + } this.#requestDrain(); throw new WorkHubActionEffectFailure( 'commit_outcome_unknown', @@ -452,7 +544,23 @@ export class HostWorkHubCoordinationCoordinator { async #candidates(): Promise> { try { - return { ok: true, result: await this.#actionGate.candidates() }; + const [result, activeAssignments] = await Promise.all([ + this.#actionGate.candidates(), + this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, () => + this.#listActiveAssignmentRecords(), + ), + ]); + return { + ok: true, + result: { + ...result, + delegations: activeAssignments.map(({ sequence, assignment }) => ({ + actionId: assignment.actionId, + targetSessionId: assignment.targetSessionId, + sequence, + })), + }, + }; } catch { return { ok: false, @@ -464,6 +572,37 @@ export class HostWorkHubCoordinationCoordinator { } } + async #listActiveAssignmentRecords(): Promise { + const active = await this.#activeAssignmentIndex(); + return [...active.values()].sort( + (left, right) => + left.sequence - right.sequence || + left.assignment.delegationId.localeCompare(right.assignment.delegationId), + ); + } + + #activeAssignmentIndex(): Promise> { + return (this.#activeAssignments ??= readActiveWorkHubAssignments(this.#stores)); + } + + async #applyLatestActiveMessage(message: StoredMessage): Promise { + if (!activeDelegationTransition(message)) return; + const sequence = await this.#stores.readTranscriptHighWaterSnapshot( + WORKHUB_COORDINATION_SESSION_ID, + ); + if (sequence === null) { + throw new WorkHubActionEffectFailure( + 'persistence_failed', + 'WorkHub Coordination transcript is unavailable', + ); + } + await this.#applyActiveMessage(sequence, message); + } + + async #applyActiveMessage(sequence: number, message: StoredMessage): Promise { + projectActiveWorkHubMessage(await this.#activeAssignmentIndex(), sequence, message); + } + async #act( input: WorkHubCoordinationActInput, context: ConnectionContext, @@ -788,28 +927,78 @@ function validCoordinationHeader(header: SessionHeader): boolean { ); } +async function readActiveWorkHubAssignments( + stores: Pick, +): Promise> { + const active = new Map(); + let position = 0; + let throughSequence: number | null | undefined; + try { + while (true) { + const page = await stores.readTranscriptRecordsSnapshot(WORKHUB_COORDINATION_SESSION_ID, { + direction: 'newer', + position, + maxStoredBytes: ACTIVE_DELEGATION_SCAN_MAX_BYTES, + maxMessages: ACTIVE_DELEGATION_SCAN_MAX_MESSAGES, + ...(throughSequence === undefined ? {} : { throughSequence }), + }); + throughSequence = page.throughSequence; + for (const { sequence, message } of page.records) { + projectActiveWorkHubMessage(active, sequence, message); + } + if (page.nextPosition === null) return active; + if (page.nextPosition <= position) { + throw new Error('WorkHub Coordination transcript scan did not advance'); + } + position = page.nextPosition; + } + } catch (error) { + if (isSessionNotFoundError(error)) return active; + throw error; + } +} + +function activeDelegationTransition(message: StoredMessage): boolean { + return ( + message.type === 'workhub_coordination' && + (message.kind === 'delegation_assigned' || + message.kind === 'delegation_superseded' || + message.kind === 'delegation_replacement_aborted' || + (message.kind === 'delegation_stop_resolved' && message.outcome !== 'not_owned')) + ); +} + +function projectActiveWorkHubMessage( + active: Map, + sequence: number, + message: StoredMessage, +): void { + if (message.type !== 'workhub_coordination') return; + if (message.kind === 'delegation_assigned') { + if (message.replacesDelegationId) active.delete(message.replacesDelegationId); + active.set(message.delegationId, { sequence, assignment: message }); + } else if (message.kind === 'delegation_superseded') { + active.delete(message.supersededDelegationId); + } else if (message.kind === 'delegation_replacement_aborted') { + active.delete(message.abortedDelegationId); + } else if (message.kind === 'delegation_stop_resolved' && message.outcome !== 'not_owned') { + active.delete(message.stopsDelegationId); + } +} + function digest(value: unknown): `sha256:${string}` { return `sha256:${createHash('sha256').update(JSON.stringify(value)).digest('hex')}`; } -function activeWorkHubAssignments( - messages: readonly StoredMessage[], -): WorkHubDelegationAssignedMessage[] { - const terminalDelegationIds = new Set(); - const assignments: WorkHubDelegationAssignedMessage[] = []; - for (const message of messages) { - if (message.type !== 'workhub_coordination') continue; - if (message.kind === 'delegation_assigned') { - assignments.push(message); - } else if (message.kind === 'delegation_superseded') { - terminalDelegationIds.add(message.supersededDelegationId); - } else if (message.kind === 'delegation_replacement_aborted') { - terminalDelegationIds.add(message.abortedDelegationId); - } else if (message.kind === 'delegation_stop_resolved' && message.outcome !== 'not_owned') { - terminalDelegationIds.add(message.stopsDelegationId); - } - } - return assignments.filter(({ delegationId }) => !terminalDelegationIds.has(delegationId)); +function coordinationResumeResult( + resolution: WorkHubDelegationResumeMessage, +): Extract { + return { + disposition: 'resume_work', + outcome: resolution.outcome, + targetSessionId: resolution.targetSessionId, + ...(resolution.targetTurnId ? { targetTurnId: resolution.targetTurnId } : {}), + }; } function workHubDestructiveClaimIdentitySuffix(delegationId: string): string { diff --git a/packages/storage/src/__tests__/safe-boundary-continuation-admission.test.ts b/packages/storage/src/__tests__/safe-boundary-continuation-admission.test.ts new file mode 100644 index 0000000000..cac49621f3 --- /dev/null +++ b/packages/storage/src/__tests__/safe-boundary-continuation-admission.test.ts @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; +import { test } from 'node:test'; +import { createSqliteAgentRunStore } from '../agent-run-store.js'; +import { migrateSqliteCoreExecutionDatabase } from '../sqlite-core-execution-schema.js'; + +test('core execution migration preserves databases with historical continuation forks', () => { + const database = new DatabaseSync(':memory:'); + try { + database.exec(` + CREATE TABLE core_root_turn_admissions ( + session_id TEXT NOT NULL, + turn_id TEXT NOT NULL, + admitted_at INTEGER NOT NULL, + record_json TEXT NOT NULL, + PRIMARY KEY (session_id, turn_id) + ); + `); + const insert = database.prepare(` + INSERT INTO core_root_turn_admissions(session_id, turn_id, admitted_at, record_json) + VALUES (?, ?, ?, ?) + `); + for (const [turnId, admittedAt] of [ + ['continuation-a', 20], + ['continuation-b', 30], + ] as const) { + insert.run( + 'session', + turnId, + admittedAt, + JSON.stringify({ + sessionId: 'session', + turnId, + execution: { + kind: 'safe_boundary_continuation', + sourceTurnId: 'source-turn', + sourceRunId: 'source-run', + }, + }), + ); + } + + assert.doesNotThrow(() => migrateSqliteCoreExecutionDatabase(database)); + assert.equal( + database.prepare('SELECT COUNT(*) AS count FROM core_root_turn_admissions').get()?.count, + 2, + ); + } finally { + database.close(); + } +}); + +test('safe-boundary continuation admission is indexed by its source execution', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-continuation-admission-')); + try { + const store = createSqliteAgentRunStore(root); + const origin = await store.admitRootTurn({ + sessionId: 'session', + turnId: 'source-turn', + proposedRunId: 'source-run', + proposedUserMessageId: 'source-message', + execution: { kind: 'external_message' }, + previousRootTurnId: null, + normalizedInput: { text: 'Start work' }, + sourceMessages: [], + admittedAt: 10, + }); + assert.equal(origin.kind, 'admitted'); + const continuation = await store.admitRootTurn({ + sessionId: 'session', + turnId: 'continuation-turn', + proposedRunId: 'continuation-run', + proposedUserMessageId: null, + execution: { + kind: 'safe_boundary_continuation', + sourceInvocationId: 'source-invocation', + sourceRunId: 'source-run', + sourceTurnId: 'source-turn', + sourceRuntimeEventHighWater: 7, + claimId: 'continuation-claim', + boundaryDigest: `sha256:${'a'.repeat(64)}`, + providerReplayDigest: `sha256:${'b'.repeat(64)}`, + safetyDigest: `sha256:${'c'.repeat(64)}`, + targetInvocationId: 'continuation-invocation', + }, + previousRootTurnId: 'source-turn', + normalizedInput: null, + sourceMessages: [], + admittedAt: 20, + }); + assert.equal(continuation.kind, 'admitted'); + await assert.rejects( + store.admitRootTurn({ + sessionId: 'session', + turnId: 'competing-continuation-turn', + proposedRunId: 'competing-continuation-run', + proposedUserMessageId: null, + execution: { + kind: 'safe_boundary_continuation', + sourceInvocationId: 'source-invocation', + sourceRunId: 'source-run', + sourceTurnId: 'source-turn', + sourceRuntimeEventHighWater: 7, + claimId: 'competing-continuation-claim', + boundaryDigest: `sha256:${'d'.repeat(64)}`, + providerReplayDigest: `sha256:${'e'.repeat(64)}`, + safetyDigest: `sha256:${'f'.repeat(64)}`, + targetInvocationId: 'competing-continuation-invocation', + }, + previousRootTurnId: 'source-turn', + normalizedInput: null, + sourceMessages: [], + admittedAt: 30, + }), + /already has continuation continuation-turn/, + ); + + assert.deepEqual( + await store.readRootTurnContinuationAdmission('session', 'source-turn', 'source-run'), + continuation.admission, + ); + assert.equal( + await store.readRootTurnContinuationAdmission('session', 'source-turn', 'other-run'), + undefined, + ); + store.close?.(); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts index 906c5d6ace..4a809c22d2 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -1513,10 +1513,17 @@ describe('SqliteSessionMetadataStore', () => { actionFingerprint: `sha256:${'a'.repeat(64)}` as const, subject: 'whd_payments', }; + const resumeClaim = { + actionId: 'resume-action', + operation: 'resume' as const, + actionFingerprint: `sha256:${'b'.repeat(64)}` as const, + subject: 'whd_payments', + }; let store = createSqliteSessionMetadataStore(path); try { assert.equal(await store.claimWorkHubAction(stopClaim), 'claimed'); assert.equal(await store.claimWorkHubAction(stopClaim), 'same_claim'); + assert.equal(await store.claimWorkHubAction(resumeClaim), 'claimed'); } finally { store.close(); } @@ -1524,6 +1531,7 @@ describe('SqliteSessionMetadataStore', () => { store = createSqliteSessionMetadataStore(path); try { assert.deepEqual(await store.readWorkHubActionClaim('stop-action'), stopClaim); + assert.deepEqual(await store.readWorkHubActionClaim('resume-action'), resumeClaim); assert.equal(await store.claimWorkHubAction(stopClaim), 'same_claim'); // A second delegation, a second disposition, and a changed payload are // each a different operation for the same identity. diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index dd855c9b3f..283778133b 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -202,6 +202,11 @@ export type AdmitRootTurnResult = export interface RootTurnAdmissionStore { admitRootTurn(input: AdmitRootTurnInput): Promise; readRootTurnAdmission(sessionId: string, turnId: string): Promise; + readRootTurnContinuationAdmission( + sessionId: string, + sourceTurnId: string, + sourceRunId: string, + ): Promise; readRootTurnSourceMessageReceipt( sessionId: string, sourceMessageId: string, @@ -588,6 +593,26 @@ class SqliteAgentRunStore implements DurableAgentRunStore { ) { throw new Error('Root Turn identity is already rejected'); } + if (admission.execution.kind === 'safe_boundary_continuation') { + const sourceOwner = this.#lease.database + .prepare(` + SELECT turn_id + FROM core_root_turn_admissions + WHERE session_id = ? + AND json_extract(record_json, '$.execution.sourceTurnId') = ? + AND json_extract(record_json, '$.execution.sourceRunId') = ? + AND json_extract(record_json, '$.execution.kind') = 'safe_boundary_continuation' + LIMIT 1 + `) + .get( + admission.sessionId, + admission.execution.sourceTurnId, + admission.execution.sourceRunId, + ) as { turn_id?: unknown } | undefined; + if (sourceOwner) { + throw new Error(`Root execution already has continuation ${String(sourceOwner.turn_id)}`); + } + } for (const source of admission.sourceMessages) { const proof = this.#lease.database .prepare(` @@ -635,6 +660,52 @@ class SqliteAgentRunStore implements DurableAgentRunStore { return readSqliteRootTurnAdmission(this.#lease.database, sessionId, turnId); } + async readRootTurnContinuationAdmission( + sessionId: string, + sourceTurnId: string, + sourceRunId: string, + ): Promise { + assertSafeId(sessionId, 'Invalid session id'); + assertSafeId(sourceTurnId, 'Invalid source turn id'); + assertSafeId(sourceRunId, 'Invalid source run id'); + const rows = this.#lease.database + .prepare(` + SELECT turn_id, record_json + FROM core_root_turn_admissions + WHERE session_id = ? + AND json_extract(record_json, '$.execution.sourceTurnId') = ? + AND json_extract(record_json, '$.execution.sourceRunId') = ? + AND json_extract(record_json, '$.execution.kind') = 'safe_boundary_continuation' + ORDER BY admitted_at, turn_id + LIMIT 2 + `) + .all(sessionId, sourceTurnId, sourceRunId) as Array<{ + turn_id?: unknown; + record_json?: unknown; + }>; + if (rows.length === 0) return undefined; + if (rows.length > 1) { + throw new Error('Root execution has multiple durable continuation admissions'); + } + const row = rows[0]!; + if (typeof row.turn_id !== 'string' || typeof row.record_json !== 'string') { + throw new Error('Invalid SQLite root turn continuation admission row'); + } + const admission = normalizeRootTurnAdmission( + JSON.parse(row.record_json), + sessionId, + row.turn_id, + ); + if ( + admission.execution.kind !== 'safe_boundary_continuation' || + admission.execution.sourceTurnId !== sourceTurnId || + admission.execution.sourceRunId !== sourceRunId + ) { + throw new Error('Root turn continuation index disagrees with its durable admission'); + } + return admission; + } + async readRootTurnStartRejection( sessionId: string, turnId: string, diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 29e0679225..8216a86406 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -193,6 +193,11 @@ export interface ExecutionAgentRunReader { type: AgentRunProjectionKey, ): Promise; readRootTurnAdmission(sessionId: string, turnId: string): Promise; + readRootTurnContinuationAdmission( + sessionId: string, + sourceTurnId: string, + sourceRunId: string, + ): Promise; readRootTurnSourceMessageReceipt( sessionId: string, sourceMessageId: string, @@ -380,6 +385,7 @@ async function createExecutionStoresForWrite sessionStore.readWorkHubStopRequest(delegationId)), readWorkHubStopResolution: (delegationId) => run(() => sessionStore.readWorkHubStopResolution(delegationId)), + readWorkHubResume: (actionId) => run(() => sessionStore.readWorkHubResume(actionId)), claimWorkHubAction: (claim) => run(() => sessionStore.claimWorkHubAction(claim)), readWorkHubActionClaim: (actionId) => run(() => sessionStore.readWorkHubActionClaim(actionId)), @@ -520,6 +526,10 @@ async function createExecutionStoresForWrite agentRunStore.admitRootTurn(input)), readRootTurnAdmission: (sessionId, turnId) => run(() => agentRunStore.readRootTurnAdmission(sessionId, turnId)), + readRootTurnContinuationAdmission: (sessionId, sourceTurnId, sourceRunId) => + run(() => + agentRunStore.readRootTurnContinuationAdmission(sessionId, sourceTurnId, sourceRunId), + ), readRootTurnStartRejection: (sessionId, turnId) => run(() => agentRunStore.readRootTurnStartRejection(sessionId, turnId)), commitRootTurnStartRejection: (input: CommitRootTurnStartRejectionInput) => @@ -656,6 +666,10 @@ async function openExecutionStoresForRead agentRunStore.readEventProjection(sessionId, type)), readRootTurnAdmission: (sessionId, turnId) => run(() => agentRunStore.readRootTurnAdmission(sessionId, turnId)), + readRootTurnContinuationAdmission: (sessionId, sourceTurnId, sourceRunId) => + run(() => + agentRunStore.readRootTurnContinuationAdmission(sessionId, sourceTurnId, sourceRunId), + ), readRootTurnSourceMessageReceipt: (sessionId, sourceMessageId) => run(() => agentRunStore.readRootTurnSourceMessageReceipt(sessionId, sourceMessageId)), }, diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index c745b72675..cb12cb6e43 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -90,6 +90,7 @@ import { type WorkHubActionClaimOutcome, type WorkHubDelegationStopRequestedMessage, type WorkHubDelegationStopResolvedMessage, + type WorkHubDelegationResumeMessage, type WorkHubDelegationSupersededMessage, } from '@maka/core/session'; import type { @@ -200,6 +201,7 @@ export interface WorkHubMessageAssignmentRequest { export interface WorkHubMessageAssignmentResult { readonly kind: 'assigned' | 'existing'; readonly targetCreated: boolean; + readonly sequence: number; readonly assignment: WorkHubDelegationAssignedMessage; } @@ -440,6 +442,7 @@ export interface SessionAuthorityStore extends SessionStore, MessageAdmissionSto readWorkHubStopResolution( delegationId: string, ): Promise; + readWorkHubResume(actionId: string): Promise; /** * Durably binds one action identity to one exact WorkHub operation before its * effect. Survives removal of the target Session so a committed destructive @@ -756,6 +759,15 @@ class SqliteSessionStore implements SessionAuthorityStore { : undefined; } + async readWorkHubResume(actionId: string): Promise { + const message = await this.readWorkHubCoordinationMessage( + `whn_${workHubIdentitySuffix(actionId)}`, + ); + return message?.type === 'workhub_coordination' && message.kind === 'delegation_resume' + ? message + : undefined; + } + async claimWorkHubAction(claim: WorkHubActionClaim): Promise { await this.ensureReady(); return this.metadata.claimWorkHubAction(claim); diff --git a/packages/storage/src/sqlite-core-execution-schema.ts b/packages/storage/src/sqlite-core-execution-schema.ts index caefc73fca..d094dfe63a 100644 --- a/packages/storage/src/sqlite-core-execution-schema.ts +++ b/packages/storage/src/sqlite-core-execution-schema.ts @@ -19,7 +19,7 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_CORE_EXECUTION_SCHEMA_VERSION = 7; +export const SQLITE_CORE_EXECUTION_SCHEMA_VERSION = 8; export function migrateSqliteCoreExecutionDatabase(db: DatabaseSync): void { db.exec(` @@ -172,6 +172,16 @@ export function migrateSqliteCoreExecutionDatabase(db: DatabaseSync): void { ON core_agent_runs(session_id, latest_model_call_sequence, run_id) WHERE latest_model_call_sequence IS NOT NULL; + DROP INDEX IF EXISTS core_root_turn_continuation_source; + + CREATE INDEX core_root_turn_continuation_source + ON core_root_turn_admissions( + session_id, + json_extract(record_json, '$.execution.sourceTurnId'), + json_extract(record_json, '$.execution.sourceRunId') + ) + WHERE json_extract(record_json, '$.execution.kind') = 'safe_boundary_continuation'; + DROP INDEX IF EXISTS core_agent_runs_identity; DROP TABLE IF EXISTS core_message_receipts; diff --git a/packages/storage/src/sqlite-session-metadata-schema.ts b/packages/storage/src/sqlite-session-metadata-schema.ts index 525934c7de..af375aaa44 100644 --- a/packages/storage/src/sqlite-session-metadata-schema.ts +++ b/packages/storage/src/sqlite-session-metadata-schema.ts @@ -19,7 +19,7 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 38; +export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 39; export const SQLITE_SESSION_MESSAGE_CHUNK_BYTES = 64 * 1024; export const SQLITE_SESSION_MESSAGE_CHUNK_MARKER = '{"$maka":"session-message-chunks-v1"}'; @@ -1268,6 +1268,44 @@ const MIGRATIONS: ReadonlyMap = new Map([ ); `, ], + [ + 39, + ` + -- SQLite cannot widen a CHECK constraint in place. Rebuild the action + -- claim table so resume receives the same durable one-action/one-operation + -- ownership as every other WorkHub action. + ALTER TABLE workhub_action_claims RENAME TO workhub_action_claims_v38; + + CREATE TABLE workhub_action_claims ( + action_id TEXT PRIMARY KEY, + operation TEXT NOT NULL CHECK ( + operation IN ( + 'answer_here', 'clarify', 'delegate_existing', 'create_new', 'replace', 'stop', 'resume' + ) + ), + action_fingerprint TEXT NOT NULL, + subject TEXT NOT NULL, + claimed_at INTEGER NOT NULL CHECK (claimed_at >= 0) + ); + + INSERT INTO workhub_action_claims( + action_id, + operation, + action_fingerprint, + subject, + claimed_at + ) + SELECT + action_id, + operation, + action_fingerprint, + subject, + claimed_at + FROM workhub_action_claims_v38; + + DROP TABLE workhub_action_claims_v38; + `, + ], ]); if (MIGRATIONS.size !== SQLITE_SESSION_METADATA_SCHEMA_VERSION) { diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index c1a13de8d5..94240f46a8 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -241,6 +241,7 @@ export interface SqliteWorkHubMessageAssignmentRequest { export interface SqliteWorkHubMessageAssignmentResult { readonly kind: 'assigned' | 'existing'; readonly targetCreated: boolean; + readonly sequence: number; readonly assignment: WorkHubDelegationAssignedMessage; } @@ -1782,10 +1783,11 @@ export class SqliteSessionMetadataStore { throw new SessionMetadataConflictError('WorkHub Coordination Session is unavailable'); } - const existingAssignment = this.readMessageByIdSync( + const existingAssignmentRecord = this.readMessageRecordByIdSync( WORKHUB_COORDINATION_SESSION_ID, assignment.id, ); + const existingAssignment = existingAssignmentRecord?.message; if (existingAssignment) { if ( existingAssignment.type !== 'workhub_coordination' || @@ -1799,6 +1801,7 @@ export class SqliteSessionMetadataStore { return { kind: 'existing' as const, targetCreated: false, + sequence: existingAssignmentRecord.sequence, assignment: existingAssignment, }; } @@ -1924,18 +1927,21 @@ export class SqliteSessionMetadataStore { ) { throw new SessionMetadataConflictError('Invalid WorkHub transcript sequence'); } - this.insertSessionMessagesSync( - WORKHUB_COORDINATION_SESSION_ID, - sequenceRow.last_sequence + 1, - [ - { message: committedAssignment, json: committedAssignmentJson }, - ...(supersession && supersessionJson - ? [{ message: supersession, json: supersessionJson }] - : []), - ], - ); + const firstSequence = sequenceRow.last_sequence + 1; + const entries = [ + { message: committedAssignment, json: committedAssignmentJson }, + ...(supersession && supersessionJson + ? [{ message: supersession, json: supersessionJson }] + : []), + ]; + this.insertSessionMessagesSync(WORKHUB_COORDINATION_SESSION_ID, firstSequence, entries); this.updateCatalogProjectionSync(WORKHUB_COORDINATION_SESSION_ID, request.projection, false); - return { kind: 'assigned' as const, targetCreated, assignment: committedAssignment }; + return { + kind: 'assigned' as const, + targetCreated, + sequence: firstSequence, + assignment: committedAssignment, + }; }); } @@ -5326,7 +5332,10 @@ export class SqliteSessionMetadataStore { return row ? decodeRecord(row) : undefined; } - private readMessageByIdSync(sessionId: string, messageId: string): StoredMessage | undefined { + private readMessageRecordByIdSync( + sessionId: string, + messageId: string, + ): { readonly sequence: number; readonly message: StoredMessage } | undefined { const row = this.db .prepare( ` @@ -5338,7 +5347,15 @@ export class SqliteSessionMetadataStore { `, ) .get(sessionId, messageId) as StoredSessionMessagePayloadRow | undefined; - return row ? decodeStoredMessageRecordRow(this.db, sessionId, row) : undefined; + if (!row) return undefined; + return { + sequence: requireStoredMessageSequence(row.sequence, sessionId), + message: decodeStoredMessageRecordRow(this.db, sessionId, row), + }; + } + + private readMessageByIdSync(sessionId: string, messageId: string): StoredMessage | undefined { + return this.readMessageRecordByIdSync(sessionId, messageId)?.message; } private readSessionMessageOrderingSync( @@ -7059,7 +7076,8 @@ function isWorkHubActionOperation(value: unknown): value is WorkHubActionOperati value === 'delegate_existing' || value === 'create_new' || value === 'replace' || - value === 'stop' + value === 'stop' || + value === 'resume' ); }