From 04825ce7cf5717051f21ab7dae7ea397e684ea4b Mon Sep 17 00:00:00 2001 From: AstroHan Date: Sat, 5 Sep 2026 22:34:09 +0800 Subject: [PATCH] refactor(workhub): resolve delegation linkage on demand Generated-by: Codex Generated-by: Claude Code --- .../main/__tests__/workhub-controller.test.ts | 45 +- .../__tests__/workhub-session-port.test.ts | 324 +++++---- .../src/renderer/workhub-controller.ts | 59 +- .../src/renderer/workhub-coordination-port.ts | 124 ++-- .../__tests__/execution-composition.test.ts | 27 + .../src/__tests__/message-coordinator.test.ts | 27 + .../workhub-coordination-action-gate.test.ts | 3 +- .../workhub-coordination-coordinator.test.ts | 635 ++++++++++++++---- .../workhub-coordination-protocol.test.ts | 3 + packages/runtime-host/src/protocol/index.ts | 4 +- .../src/protocol/workhub-coordination.ts | 24 +- .../src/server/execution-composition.ts | 16 +- .../src/server/message-coordinator.ts | 10 + .../workhub-coordination-action-gate.ts | 9 +- .../workhub-coordination-coordinator.ts | 95 +-- .../workhub-message-assignment.test.ts | 269 ++++++++ packages/storage/src/execution-stores.ts | 7 + packages/storage/src/session-store.ts | 28 +- .../src/sqlite-session-metadata-store.ts | 140 ++++ 19 files changed, 1349 insertions(+), 500 deletions(-) diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index 13b46ce35c..3b524129e7 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -245,11 +245,7 @@ test('conversation acknowledges a durable assignment before projecting target ex sessions, coordination: { open: async (handler) => { - handler([assignment], [{ - actionId: assignment.assignment!.actionId, - targetSessionId: assignment.assignment!.targetSessionId, - sequence: 0, - }]); + handler([assignment]); return { close: async () => undefined }; }, record: async (input) => ({ turnId: input.turnId }), @@ -294,11 +290,7 @@ test('conversation feedback never lets an older refresh overwrite newer target s coordination: { open: async (handler) => { const assignment = coordinationAssignmentTurn(); - handler([assignment], [{ - actionId: assignment.assignment!.actionId, - targetSessionId: assignment.assignment!.targetSessionId, - sequence: 0, - }]); + handler([assignment]); return { close: async () => undefined }; }, record: async (input) => ({ turnId: input.turnId }), @@ -340,11 +332,7 @@ test('direct stop bypasses routing candidates and preserves a not_owned delegati sessions, coordination: { open: async (handler) => { - handler([coordinationAssignmentTurn()], [{ - actionId: 'action-1', - targetSessionId: 'payments', - sequence: 0, - }]); + handler([coordinationAssignmentTurn()]); return { close: async () => undefined }; }, record: async (input) => ({ turnId: input.turnId }), @@ -400,7 +388,7 @@ test('an anaphoric stop asks for a fresh named imperative without offering a rou sessions, coordination: { open: async (handler) => { - handler([], [{ actionId: 'action-1', targetSessionId: 'payments', sequence: 0 }]); + handler([]); return { close: async () => undefined }; }, record: async (input) => ({ turnId: input.turnId }), @@ -430,7 +418,7 @@ test('a named stop reports the Gate refusal instead of judging the target itself sessions, coordination: { open: async (handler) => { - handler([], []); + handler([]); return { close: async () => undefined }; }, record: async (input) => ({ turnId: input.turnId }), @@ -464,7 +452,7 @@ test('a stop that fails for any other reason is a fault, not a clarification', a sessions, coordination: { open: async (handler) => { - handler([], []); + handler([]); return { close: async () => undefined }; }, record: async (input) => ({ turnId: input.turnId }), @@ -494,7 +482,7 @@ test('stop-shaped ordinary work routes normally instead of looping on clarificat sessions, coordination: { open: async (handler) => { - handler([], [{ actionId: 'action-1', targetSessionId: 'payments', sequence: 0 }]); + handler([]); return { close: async () => undefined }; }, record: async (input) => ({ turnId: input.turnId }), @@ -1785,6 +1773,7 @@ test('production natural-language corrections retain the prior delegation link', }), ]); const candidateSetId = `sha256:${'e'.repeat(64)}`; + const latestActionIdBySessionId = new Map(); const candidates = [ { candidateRef: 'candidate-login', @@ -1814,7 +1803,15 @@ test('production natural-language corrections retain the prior delegation link', coordination: { open: async () => ({ close: async () => undefined }), record: async (input) => ({ turnId: input.turnId }), - candidates: async () => ({ candidateSetId, candidates }), + candidates: async () => ({ + candidateSetId, + candidates: candidates.map((candidate) => { + const latestDelegationActionId = latestActionIdBySessionId.get(candidate.sessionId); + return latestDelegationActionId + ? { ...candidate, latestDelegationActionId } + : candidate; + }), + }), act: async (input) => { actions.push(input); if (input.proposal.disposition === 'replace') { @@ -1826,6 +1823,10 @@ test('production natural-language corrections retain the prior delegation link', targetTurnId: `turn-${input.actionId}`, }; } + latestActionIdBySessionId.set( + input.proposal.target.candidateRef === 'candidate-login' ? 'login' : 'payment', + input.actionId, + ); return { disposition: 'replace', replacementDisposition: 'delegate_existing', @@ -1838,6 +1839,10 @@ test('production natural-language corrections retain the prior delegation link', if (input.proposal.disposition !== 'delegate_existing') { throw new Error('unexpected test disposition'); } + latestActionIdBySessionId.set( + input.proposal.candidateRef === 'candidate-login' ? 'login' : 'payment', + input.actionId, + ); return { disposition: 'delegate_existing', targetSessionId: input.proposal.candidateRef === 'candidate-login' 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..ed02dcdf78 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,12 @@ 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('durable supersession terminalizes only the replaced linkage', () => { @@ -403,20 +374,20 @@ 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 () => { +test('Coordination transcript adapter never replays history and completes only the latest record', async () => { const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'coordination' }); const snapshots: unknown[] = []; let closes = 0; + const latestLoads: Array<{ sequence: number; maxBytes: number | undefined }> = []; + let deliver: ((batch: DesktopTranscriptBatch) => void) | undefined; const adapter = createDesktopWorkHubCoordinationPort({ sessionId, transcripts: { open: async (requestedSessionId, handler) => { assert.equal(requestedSessionId, sessionId); + deliver = handler; handler({ sessionId: 'coordination', deliverySequence: 1, @@ -426,7 +397,7 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and fragments: [], evictedDurableSequences: [], completedOverlayMessageIds: [], - hasOlder: false, + hasOlder: true, hasNewer: false, reset: true, ready: true, @@ -436,17 +407,47 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and generation: 'generation-1', hostEpoch: 'epoch-1', readThroughMessageId: null, - loadBefore: async () => {}, - loadAround: async () => {}, + loadBefore: async () => assert.fail('conversation open must not replay older history'), + loadAround: async (sequence, maxBytes) => { + latestLoads.push({ sequence, maxBytes }); + const message: StoredMessage = { + type: 'user', + id: 'latest-message', + turnId: 'latest-turn', + ts: 7, + text: 'Latest bounded WorkHub record', + }; + const data = new TextEncoder().encode(JSON.stringify(message)); + handler({ + sessionId: 'coordination', + deliverySequence: 3, + generation: 'generation-2', + hostEpoch: 'epoch-1', + durableThrough: 7, + fragments: [ + { + source: 'durable', + identity: 7, + order: null, + byteOffset: 0, + totalBytes: data.byteLength, + data, + }, + ], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: true, + hasNewer: false, + reset: false, + ready: true, + }); + }, close: async () => { closes += 1; }, }; }, }, record: async (input) => ({ turnId: input.turnId }), - candidates: async () => ({ - candidateSetId: `sha256:${'a'.repeat(64)}`, - candidates: [], - }), + candidates: async () => assert.fail('conversation open must not read route candidates'), act: async () => ({ ok: true, result: { @@ -457,57 +458,44 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and }); const handle = await adapter.open( - (turns, activeDelegations) => snapshots.push([turns, activeDelegations]), - () => {}, + (turns) => snapshots.push(turns), + (error) => assert.fail(String(error)), ); - assert.deepEqual(snapshots, [[[], []]]); + assert.deepEqual(snapshots, [[]]); + deliver?.({ + sessionId: 'coordination', + deliverySequence: 2, + generation: 'generation-2', + hostEpoch: 'epoch-1', + durableThrough: 7, + fragments: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: true, + hasNewer: false, + reset: true, + ready: true, + }); + await Promise.resolve(); + assert.deepEqual(latestLoads, [{ sequence: 7, maxBytes: 512 * 1024 }]); + assert.deepEqual(snapshots, [[], [ + { + messageId: 'latest-message', + turnId: 'latest-turn', + text: 'Latest bounded WorkHub record', + state: 'completed', + updatedAt: 7, + }, + ]]); await handle.close(); assert.equal(closes, 1); }); -test('Coordination transcript reset rebuilds active linkage outside the resident window', async () => { +test('Coordination transcript adapter retries latest-record completion in the same generation', async () => { const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'coordination' }); - const assignment: StoredMessage = { - type: 'workhub_coordination', - id: 'assignment-old', - turnId: 'action-old', - ts: 1, - schemaVersion: 1, - kind: 'delegation_assigned', - actionId: 'action-old', - actionFingerprint: `sha256:${'a'.repeat(64)}`, - coordinationTurnId: 'action-old', - targetSessionId: 'payments', - targetSessionName: 'Payments', - targetTurnId: 'payments-turn', - targetMessageId: 'payments-message', - delegationId: 'payments-delegation', - disposition: 'delegate_existing', - userText: 'Continue payments', - }; - const recent: StoredMessage = { - type: 'user', - id: 'recent-user', - turnId: 'recent-turn', - ts: 2, - text: 'Recent coordination', - }; - const fragment = (message: StoredMessage, sequence: number) => { - const data = new TextEncoder().encode(JSON.stringify(message)); - return { - source: 'durable' as const, - identity: sequence, - order: null, - byteOffset: 0, - totalBytes: data.byteLength, - data, - }; - }; + const errors: unknown[] = []; + let latestLoads = 0; let deliver: ((batch: DesktopTranscriptBatch) => void) | undefined; - let generation = 'generation-1'; - let historyLoads = 0; - let historyBatchReady = true; - const snapshots: unknown[] = []; const adapter = createDesktopWorkHubCoordinationPort({ sessionId, transcripts: { @@ -516,10 +504,10 @@ test('Coordination transcript reset rebuilds active linkage outside the resident handler({ sessionId: 'coordination', deliverySequence: 1, - generation, + generation: 'generation-1', hostEpoch: 'epoch-1', - durableThrough: 1, - fragments: [fragment(recent, 1)], + durableThrough: 7, + fragments: [], evictedDurableSequences: [], completedOverlayMessageIds: [], hasOlder: true, @@ -529,93 +517,147 @@ test('Coordination transcript reset rebuilds active linkage outside the resident }); return { sessionId, - generation, + generation: 'generation-1', hostEpoch: 'epoch-1', 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, - }); + loadBefore: async () => assert.fail('conversation open must not replay older history'), + loadAround: async () => { + latestLoads += 1; + if (latestLoads === 1) throw new Error('transient latest-record read failure'); }, - loadAround: async () => {}, close: async () => {}, }; }, }, record: async (input) => ({ turnId: input.turnId }), - candidates: async () => ({ - candidateSetId: `sha256:${'b'.repeat(64)}`, - candidates: [], - }), + candidates: async () => assert.fail('conversation open must not read route candidates'), act: async () => ({ ok: true, - result: { disposition: 'answer_here', coordinationTurnId: 'coordination-turn' }, + result: { + disposition: 'answer_here', + coordinationTurnId: 'coordination-turn', + }, }), }); - const handle = await adapter.open( - (_turns, activeDelegations) => snapshots.push(activeDelegations), - (error) => assert.fail(String(error)), - ); - assert.deepEqual(snapshots.at(-1), [{ - actionId: 'action-old', - targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }), - sequence: 0, - }]); + const handle = await adapter.open(() => {}, (error) => errors.push(error)); + await Promise.resolve(); + await Promise.resolve(); + assert.equal(latestLoads, 1); + assert.equal(errors.length, 1); + + deliver?.({ + sessionId: 'coordination', + deliverySequence: 2, + generation: 'generation-1', + hostEpoch: 'epoch-1', + durableThrough: 7, + fragments: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: true, + hasNewer: false, + reset: false, + ready: true, + }); + await Promise.resolve(); + assert.equal(latestLoads, 2); - generation = 'generation-2'; - historyBatchReady = false; - const snapshotsBeforeReset = snapshots.length; deliver?.({ sessionId: 'coordination', deliverySequence: 3, - generation, + generation: 'generation-1', hostEpoch: 'epoch-1', - durableThrough: 1, - fragments: [fragment(recent, 1)], + durableThrough: 7, + fragments: [], evictedDurableSequences: [], completedOverlayMessageIds: [], hasOlder: true, hasNewer: false, reset: true, - ready: false, + ready: true, }); await Promise.resolve(); - await Promise.resolve(); + assert.equal(latestLoads, 3); + await handle.close(); +}); - assert.equal(historyLoads, 2); - assert.equal(snapshots.length, snapshotsBeforeReset); +test('Coordination transcript adapter ignores a stale latest-record failure after reset', async () => { + const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'coordination' }); + const errors: unknown[] = []; + let latestLoads = 0; + let rejectStaleLoad: ((error: Error) => void) | undefined; + let deliver: ((batch: DesktopTranscriptBatch) => void) | undefined; + const adapter = createDesktopWorkHubCoordinationPort({ + sessionId, + transcripts: { + open: async (_requestedSessionId, handler) => { + deliver = handler; + handler({ + sessionId: 'coordination', + deliverySequence: 1, + generation: 'generation-1', + hostEpoch: 'epoch-1', + durableThrough: 7, + fragments: [], + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: true, + hasNewer: false, + reset: true, + ready: true, + }); + return { + sessionId, + generation: 'generation-1', + hostEpoch: 'epoch-1', + readThroughMessageId: null, + loadBefore: async () => assert.fail('conversation open must not replay older history'), + loadAround: async () => { + latestLoads += 1; + if (latestLoads === 1) { + await new Promise((_resolve, reject) => { + rejectStaleLoad = reject; + }); + } + }, + close: async () => {}, + }; + }, + }, + record: async (input) => ({ turnId: input.turnId }), + candidates: async () => assert.fail('conversation open must not read route candidates'), + act: async () => ({ + ok: true, + result: { + disposition: 'answer_here', + coordinationTurnId: 'coordination-turn', + }, + }), + }); + + const handle = await adapter.open(() => {}, (error) => errors.push(error)); + assert.equal(latestLoads, 1); deliver?.({ sessionId: 'coordination', - deliverySequence: 5, - generation, + deliverySequence: 2, + generation: 'generation-1', hostEpoch: 'epoch-1', - durableThrough: 1, - fragments: [fragment(recent, 1)], + durableThrough: 7, + fragments: [], evictedDurableSequences: [], completedOverlayMessageIds: [], - hasOlder: false, + hasOlder: true, hasNewer: false, - reset: false, + reset: true, ready: true, }); - assert.deepEqual(snapshots.at(-1), [{ - actionId: 'action-old', - targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }), - sequence: 0, - }]); + await Promise.resolve(); + assert.equal(latestLoads, 2); + rejectStaleLoad?.(new Error('stale latest-record failure')); + await Promise.resolve(); + await Promise.resolve(); + assert.deepEqual(errors, []); await handle.close(); }); diff --git a/apps/desktop/src/renderer/workhub-controller.ts b/apps/desktop/src/renderer/workhub-controller.ts index f900300483..959381ae30 100644 --- a/apps/desktop/src/renderer/workhub-controller.ts +++ b/apps/desktop/src/renderer/workhub-controller.ts @@ -137,13 +137,6 @@ export interface WorkHubCoordinationTurn { export type WorkHubDelegationLinkState = 'active' | 'superseded' | 'aborted' | 'stopped'; -/** Unbounded, rebuildable linkage state kept separate from the bounded timeline. */ -export interface WorkHubActiveDelegation { - readonly actionId: string; - readonly targetSessionId: string; - readonly sequence: number; -} - const WORKHUB_TIMELINE_TEXT_LIMIT = 600; export function boundedWorkHubTimelineText(value: string): string { @@ -248,10 +241,7 @@ export interface WorkHubSessionPort { export interface WorkHubCoordinationPort { open( - handler: ( - turns: readonly WorkHubCoordinationTurn[], - activeDelegations: readonly WorkHubActiveDelegation[], - ) => void, + handler: (turns: readonly WorkHubCoordinationTurn[]) => void, onError: (error: unknown) => void, ): Promise<{ close(): Promise }>; record(input: { @@ -288,40 +278,16 @@ export function createWorkHubController(deps: { let routePolicy = createWorkHubRoutePolicy(); let focusReadVersion = 0; let pendingFocusReadVersion: number | undefined; - const activeActionIdsBySessionId = new Map(); - const removeActiveAction = (sessionId: string, actionId: string) => { - const remaining = (activeActionIdsBySessionId.get(sessionId) ?? []).filter( - (candidate) => candidate !== actionId, - ); - if (remaining.length === 0) { - activeActionIdsBySessionId.delete(sessionId); - return; - } - activeActionIdsBySessionId.set(sessionId, remaining); - }; - const addActiveAction = (sessionId: string, actionId: string) => { - const active = activeActionIdsBySessionId.get(sessionId) ?? []; - if (!active.includes(actionId)) { - activeActionIdsBySessionId.set(sessionId, [...active, actionId]); - } - }; - const correctionFor = (from: WorkHubSessionTarget): WorkHubCorrectionContext => { - const sourceActionId = activeActionIdsBySessionId.get(from.sessionId)?.at(-1); + const correctionFor = ( + from: WorkHubSessionTarget, + candidateBySessionId: ReadonlyMap, + ): WorkHubCorrectionContext => { + const sourceActionId = candidateBySessionId.get(from.sessionId)?.latestDelegationActionId; if (!sourceActionId) { throw new Error('WorkHub linked correction requires an active durable delegation'); } return { from, sourceActionId }; }; - const reconcileActiveDelegations = ( - activeDelegations: readonly WorkHubActiveDelegation[], - ) => { - activeActionIdsBySessionId.clear(); - for (const delegation of [...activeDelegations].sort( - (left, right) => left.sequence - right.sequence, - )) { - addActiveAction(delegation.targetSessionId, delegation.actionId); - } - }; const reconcileFocus = ( policy: ReturnType, sessions: readonly WorkHubSessionFacts[], @@ -342,10 +308,6 @@ export function createWorkHubController(deps: { correction: WorkHubCorrectionContext | undefined, ): Extract => { const target = { sessionId: admitted.targetSessionId }; - if (correction) { - removeActiveAction(correction.from.sessionId, correction.sourceActionId); - } - addActiveAction(target.sessionId, input.requestId); policy.rememberTarget(target); return { kind: 'submitted', @@ -405,9 +367,8 @@ export function createWorkHubController(deps: { }); let handle: { close(): Promise } | undefined; try { - handle = await coordination.open((turns, activeDelegations) => { + handle = await coordination.open((turns) => { if (disposed) return; - reconcileActiveDelegations(activeDelegations); latestTurns = turns; generation += 1; // The atomic assignment is already durable acknowledgement, so emit @@ -576,7 +537,7 @@ export function createWorkHubController(deps: { }); if (decision.kind === 'clarification') { const correction = decision.correctedFrom - ? correctionFor(decision.correctedFrom) + ? correctionFor(decision.correctedFrom, candidateBySessionId) : undefined; return { kind: 'clarification', @@ -606,7 +567,9 @@ export function createWorkHubController(deps: { }; } const correction = input.correction ?? - (decision.correctedFrom ? correctionFor(decision.correctedFrom) : undefined); + (decision.correctedFrom + ? correctionFor(decision.correctedFrom, candidateBySessionId) + : undefined); if (decision.kind === 'new_session') { const { title } = decision; const admitted = await coordination.act(correction diff --git a/apps/desktop/src/renderer/workhub-coordination-port.ts b/apps/desktop/src/renderer/workhub-coordination-port.ts index 114fed8532..954e68286b 100644 --- a/apps/desktop/src/renderer/workhub-coordination-port.ts +++ b/apps/desktop/src/renderer/workhub-coordination-port.ts @@ -27,7 +27,6 @@ import { DesktopTranscriptRangeStore } from './desktop-transcript-range-store.js import type { WorkHubCoordinationPort, WorkHubCoordinationTurn, - WorkHubActiveDelegation, WorkHubProjectedTurnState, } from './workhub-controller.js'; import type { @@ -43,6 +42,7 @@ export { WorkHubCoordinationFailure }; import type { WorkHubDesktopTranscriptBridge } from './workhub-session-port.js'; const WORKHUB_COORDINATION_TURN_LIMIT = 40; +const WORKHUB_COORDINATION_LATEST_RECORD_MAX_BYTES = 512 * 1024; export function createDesktopWorkHubCoordinationPort(deps: { sessionId: string; @@ -71,21 +71,47 @@ export function createDesktopWorkHubCoordinationPort(deps: { const store = new DesktopTranscriptRangeStore(deps.sessionId); let disposed = false; let ready = false; - let historyReady = false; - let historyGeneration = 0; + let completedLatestGeneration: string | undefined; + let loadingLatestGeneration: string | undefined; + let latestLoadRevision = 0; let handle: Awaited> | undefined; - let historyLane = Promise.resolve(); - const coordinationMessagesBySequence = new Map(); 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)); + }; + const emitOrCompleteLatest = () => { + if (!handle || !ready) return; + const snapshot = store.snapshot(); + const latestRecordIsIncomplete = + snapshot.durableThrough !== null && + (snapshot.newestSequence === null || snapshot.newestSequence < snapshot.durableThrough); + if ( + latestRecordIsIncomplete && + completedLatestGeneration !== snapshot.generation && + loadingLatestGeneration !== snapshot.generation + ) { + const generation = snapshot.generation; + const revision = latestLoadRevision; + loadingLatestGeneration = generation; + void handle + .loadAround( + snapshot.durableThrough, + WORKHUB_COORDINATION_LATEST_RECORD_MAX_BYTES, + ) + .then(() => { + if (latestLoadRevision !== revision) return; + completedLatestGeneration = generation; + if (loadingLatestGeneration === generation) loadingLatestGeneration = undefined; + }) + .catch((error) => { + if (disposed || latestLoadRevision !== revision) return; + if (loadingLatestGeneration === generation) { + loadingLatestGeneration = undefined; + } + onError(error); + }); + return; + } + emit(); }; const opened = await deps.transcripts.open( deps.sessionId, @@ -93,27 +119,14 @@ export function createDesktopWorkHubCoordinationPort(deps: { if (disposed) return; try { 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); - }); - } + latestLoadRevision += 1; + completedLatestGeneration = undefined; + loadingLatestGeneration = undefined; } 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 (changed || batch.ready) emitOrCompleteLatest(); } catch (error) { onError(error); } @@ -127,33 +140,7 @@ 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); - } catch (error) { - disposed = true; - await handle.close().catch(() => undefined); - throw error; - } + emitOrCompleteLatest(); return { async close() { disposed = true; @@ -164,27 +151,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[] { diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index f9a68677ec..3e5de76f15 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -680,6 +680,29 @@ test('WorkHub creates new work through the production assignment composition', a const session = (await manager.listSessions()).find(({ id }) => id === targetSessionId); assert.equal(session?.name, 'Login stability'); assert.equal(session?.llmConnectionId, connectionId); + + const current = await composition.handlers['workhub.coordination.candidates']({}, context); + assert.equal(current.ok, true); + if (!current.ok) return; + assert.equal( + current.result.candidates.find(({ sessionId }) => sessionId === targetSessionId) + ?.latestDelegationActionId, + 'workhub-create-action', + ); + const stopped = await composition.handlers['workhub.coordination.act']( + { + actionId: 'workhub-create-stop-action', + userText: 'Stop Login stability', + confirmation: { kind: 'user_stop' }, + proposal: { + disposition: 'stop_work', + expects: { targetSessionId }, + }, + }, + context, + ); + assert.equal(stopped.ok, true, JSON.stringify(stopped)); + if (stopped.ok) assert.equal(stopped.result.disposition, 'stop_work'); } finally { await composition.close(); } @@ -770,6 +793,10 @@ test('WorkHub correction replaces its link without stopping a shared manual Turn ); return proof.ok && proof.result.resolutions[0]?.state === 'owned'; }); + assert.deepEqual( + await stores.sessionStore.readActiveWorkHubAssignmentsByTarget([source.id]), + [assignment], + ); const stopped = await composition.handlers['workhub.coordination.act']( { diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 56e31728ff..a16d721c79 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -665,6 +665,33 @@ test('message execution query reports the Turn that durably owns each Message', }); }); +test('message execution disposition reuses a held Session admission', async () => { + const fixture = createFixture(); + const content = { text: 'pending delegation' }; + await fixture.admissions.commitMessageAdmission({ + ...ROOT, + messageId: 'pending-delegation', + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, + admittedAt: 10, + }); + + assert.deepEqual( + await fixture.sessionAdmission.run(ROOT.sessionId, (lease) => + fixture.coordinator.readMessageExecutionDispositionAdmitted( + ROOT.sessionId, + 'pending-delegation', + lease, + ), + ), + { kind: 'pending' }, + ); +}); + test('submit re-runs admission when the queue revision moves during preflight', async () => { let preflightCalls = 0; const fixture = createFixture(undefined, async () => { 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..b7c9d598b2 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 @@ -2573,10 +2573,11 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { async readAssignment(actionId: string) { return assignmentRecords.get(actionId); }, - async listActiveAssignments() { + async listActiveAssignments(targetSessionId) { return [...assignmentRecords.values()].filter((assignment) => { const stopOutcome = stopResolutions.get(assignment.delegationId)?.outcome; return ( + assignment.targetSessionId === targetSessionId && !supersessions.has(assignment.delegationId) && !replacementAborts.has(assignment.delegationId) && (stopOutcome === undefined || stopOutcome === 'not_owned') 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..f18b8c1556 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'; @@ -49,6 +51,7 @@ import type { WorkHubActionGateEffects } from '../server/workhub-coordination-ac import { HostWorkHubCoordinationCoordinator, type CoordinationCreateTarget, + type HostWorkHubCoordinationCoordinatorOptions, } from '../server/workhub-coordination-coordinator.js'; const CONTEXT: ConnectionContext = { @@ -502,9 +505,9 @@ describe('Host WorkHub Coordination coordinator', () => { }); const assignments: string[] = []; const first = coordinator(root, store, () => undefined, undefined, undefined, undefined, { - assign: async (input) => { + assign: async (input, context) => { assignments.push(input.actionId); - return persistTestAssignment(store, input, 'payments-turn'); + return persistTestAssignmentAction(store, 'payments-turn')(input, context); }, }); assert.equal((await first.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); @@ -545,6 +548,11 @@ describe('Host WorkHub Coordination coordinator', () => { ], ); assert.equal(assignments.length, 1); + assert.equal(candidates.result.candidates[0]?.latestDelegationActionId, undefined); + const current = await first.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(current.ok, true); + if (!current.ok) return; + assert.equal(current.result.candidates[0]?.latestDelegationActionId, 'payments-action'); } finally { await store.close?.(); } @@ -552,7 +560,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 +590,316 @@ describe('Host WorkHub Coordination coordinator', () => { } }); + test('reads current linkage only through the bounded candidate target', 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', + ); + await persistTestAssignment( + store, + { + actionId: 'terminated-newer-action', + actionFingerprint: `sha256:${'8'.repeat(64)}`, + targetSessionId: target.id, + targetSessionName: target.name, + disposition: 'delegate_existing', + userText: 'A newer delegation that has since ended', + }, + 'terminated-newer-turn', + ); + const terminated = await store.readWorkHubAssignment('terminated-newer-action'); + assert.ok(terminated); + await store.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ + { + type: 'workhub_coordination', + id: `whx_${createHash('sha256') + .update(terminated.delegationId) + .digest('hex') + .slice(0, 48)}`, + turnId: 'terminated-newer-action', + ts: Date.now(), + schemaVersion: WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, + kind: 'delegation_superseded', + actionId: 'superseding-action', + actionFingerprint: `sha256:${'9'.repeat(64)}`, + coordinationTurnId: 'terminated-newer-action', + supersededActionId: terminated.actionId, + supersededDelegationId: terminated.delegationId, + replacementDelegationId: 'whd_terminal_probe', + }, + ]); + const latestActiveActionId = 'latest-active-action'; + await persistTestAssignment( + store, + { + actionId: latestActiveActionId, + actionFingerprint: `sha256:${'a'.repeat(64)}`, + targetSessionId: target.id, + targetSessionName: target.name, + disposition: 'delegate_existing', + userText: 'The latest active delegation', + }, + 'latest-active-turn', + ); + const pagedTerminalAssignments: WorkHubDelegationAssignedMessage[] = []; + for (const actionId of Array.from( + { length: 32 }, + (_, index) => `paged-terminal-action-${index}`, + )) { + await persistTestAssignment( + store, + { + actionId, + actionFingerprint: `sha256:${createHash('sha256').update(actionId).digest('hex')}`, + targetSessionId: target.id, + targetSessionName: target.name, + disposition: 'delegate_existing', + userText: 'A newer delegation that has since ended', + }, + `${actionId}-turn`, + ); + const assignment = await store.readWorkHubAssignment(actionId); + assert.ok(assignment); + pagedTerminalAssignments.push(assignment); + } + await store.appendMessages( + WORKHUB_COORDINATION_SESSION_ID, + pagedTerminalAssignments.map((assignment, index) => ({ + type: 'workhub_coordination' as const, + id: `whx_${createHash('sha256') + .update(assignment.delegationId) + .digest('hex') + .slice(0, 48)}`, + turnId: `paged-supersession-${index}`, + ts: Date.now() + index, + schemaVersion: WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, + kind: 'delegation_superseded' as const, + actionId: `paged-supersession-${index}`, + actionFingerprint: `sha256:${createHash('sha256') + .update(`paged-supersession-${index}`) + .digest('hex')}` as const, + coordinationTurnId: `paged-supersession-${index}`, + supersededActionId: assignment.actionId, + supersededDelegationId: assignment.delegationId, + replacementDelegationId: `whd_paged_terminal_${index}`, + })), + ); + + const terminalOnlyTarget = await store.create({ + cwd: root, + name: 'Terminal only', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + await persistTestAssignment( + store, + { + actionId: 'terminal-only-action', + actionFingerprint: `sha256:${'b'.repeat(64)}`, + targetSessionId: terminalOnlyTarget.id, + targetSessionName: terminalOnlyTarget.name, + disposition: 'delegate_existing', + userText: 'This delegation is terminal', + }, + 'terminal-only-turn', + ); + const terminalOnlyAssignment = await store.readWorkHubAssignment('terminal-only-action'); + assert.ok(terminalOnlyAssignment); + await store.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ + { + type: 'workhub_coordination', + id: `whx_${createHash('sha256') + .update(terminalOnlyAssignment.delegationId) + .digest('hex') + .slice(0, 48)}`, + turnId: 'terminal-only-supersession', + ts: Date.now(), + schemaVersion: WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, + kind: 'delegation_superseded', + actionId: 'terminal-only-supersession', + actionFingerprint: `sha256:${'c'.repeat(64)}`, + coordinationTurnId: 'terminal-only-supersession', + supersededActionId: terminalOnlyAssignment.actionId, + supersededDelegationId: terminalOnlyAssignment.delegationId, + replacementDelegationId: 'whd_terminal_only_probe', + }, + ]); + + const targetReads: Array = []; + const stores = new Proxy(store, { + get(authority, property, receiver) { + if (property === 'readMessagesSnapshot' || property === 'readTranscriptRecordsSnapshot') { + return async () => assert.fail('candidate lookup must not scan Coordination history'); + } + if (property === 'readActiveWorkHubAssignmentsByTarget') { + return async ( + ...args: Parameters + ) => { + targetReads.push([args[0], args[1]]); + return [...(await authority.readActiveWorkHubAssignmentsByTarget(...args))].reverse(); + }; + } + 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.candidates, second.result.candidates); + assert.equal( + first.result.candidates.find(({ sessionId }) => sessionId === target.id) + ?.latestDelegationActionId, + latestActiveActionId, + ); + assert.equal( + first.result.candidates.find(({ sessionId }) => sessionId === terminalOnlyTarget.id) + ?.latestDelegationActionId, + undefined, + ); + } + // One bounded read per page, not one per candidate. + assert.equal(targetReads.length, 4); + assert.equal( + targetReads.every( + ([sessionIds, maxAssignmentsPerTarget]) => + maxAssignmentsPerTarget === 1 && + sessionIds.includes(target.id) && + sessionIds.includes(terminalOnlyTarget.id), + ), + true, + ); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('rejects a correction whose source is no longer the latest active linkage', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-stale-correction-')); + const store = createSessionStore(root); + try { + const source = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + const destination = await store.create({ + cwd: root, + name: 'Login', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + const workhub = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + retireDelegation: async () => assert.fail('a stale correction must not retire work'), + }); + assert.equal((await workhub.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); + await persistTestAssignment( + store, + { + actionId: 'stale-source-action', + actionFingerprint: `sha256:${'a'.repeat(64)}`, + targetSessionId: source.id, + targetSessionName: source.name, + disposition: 'delegate_existing', + userText: 'First payment delegation', + }, + 'stale-source-turn', + ); + const staleSource = await store.readWorkHubAssignment('stale-source-action'); + assert.ok(staleSource); + const staleCandidates = await workhub.handlers['workhub.coordination.candidates']( + {}, + CONTEXT, + ); + assert.equal(staleCandidates.ok, true); + if (!staleCandidates.ok) return; + const destinationCandidate = staleCandidates.result.candidates.find( + ({ sessionId }) => sessionId === destination.id, + ); + assert.ok(destinationCandidate); + if (!destinationCandidate) return; + await persistTestAssignment( + store, + { + actionId: 'newer-source-action', + actionFingerprint: `sha256:${'b'.repeat(64)}`, + targetSessionId: source.id, + targetSessionName: source.name, + disposition: 'delegate_existing', + userText: 'Second payment delegation', + }, + 'newer-source-turn', + ); + + const correction = await workhub.handlers['workhub.coordination.act']( + { + actionId: 'stale-correction-action', + userText: 'No, move this to Login instead', + candidateSetId: staleCandidates.result.candidateSetId, + confirmation: { kind: 'user_correction' }, + proposal: { + disposition: 'replace', + replacesActionId: staleSource.actionId, + target: { + disposition: 'delegate_existing', + candidateRef: destinationCandidate.candidateRef, + }, + }, + }, + CONTEXT, + ); + + assert.equal(correction.ok, false); + if (!correction.ok) assert.equal(correction.error.code, 'operation_conflict'); + assert.equal(await store.readWorkHubReplacement(staleSource.delegationId), undefined); + assert.equal(await store.readWorkHubSupersession(staleSource.delegationId), undefined); + } 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 +915,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 +1005,76 @@ 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) => + admission.runMany([WORKHUB_COORDINATION_SESSION_ID, input.targetSessionId], async () => { + const result = await persistTestAssignment(store, input, 'payments-turn'); + committed.resolve(); + await releaseAssignment.promise; + 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('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 +1086,47 @@ 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`), + readDelegationRetirement: async (_assignment, lease) => { + assert.ok(lease, 'the held target admission must be reused for retirement reads'); + return 'not_retired'; + }, + 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 +1148,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 +1234,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 +1342,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 +1405,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 +1423,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); @@ -1070,7 +1466,10 @@ describe('Host WorkHub Coordination coordinator', () => { await store.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ { type: 'workhub_coordination', - id: 'whs_replaced_probe', + id: `whx_${createHash('sha256') + .update(assignment.delegationId) + .digest('hex') + .slice(0, 48)}`, turnId: 'replaced-probe-turn', ts: Date.now(), schemaVersion: WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, @@ -1096,7 +1495,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 +1539,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); @@ -1164,11 +1567,28 @@ describe('Host WorkHub Coordination coordinator', () => { ); const assignment = await store.readWorkHubAssignment('source-action'); assert.ok(assignment); + const stopInput = { + actionId: 'stop-action', + userText: 'Stop Payments', + proposal: { disposition: 'stop_work' as const, expects: { targetSessionId: target.id } }, + confirmation: { kind: 'user_stop' as const }, + }; assert.equal( await store.claimWorkHubAction({ actionId: 'stop-action', operation: 'stop', - actionFingerprint: `sha256:${'b'.repeat(64)}`, + actionFingerprint: `sha256:${createHash('sha256') + .update( + JSON.stringify({ + userText: stopInput.userText, + disposition: 'stop_work', + stopsActionId: assignment.actionId, + stopsDelegationId: assignment.delegationId, + targetSessionId: assignment.targetSessionId, + targetMessageId: assignment.targetMessageId, + }), + ) + .digest('hex')}`, subject: assignment.delegationId, }), 'claimed', @@ -1176,7 +1596,10 @@ describe('Host WorkHub Coordination coordinator', () => { await store.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ { type: 'workhub_coordination', - id: 'whs_terminal_probe', + id: `whx_${createHash('sha256') + .update(assignment.delegationId) + .digest('hex') + .slice(0, 48)}`, turnId: 'terminal-probe-turn', ts: Date.now(), schemaVersion: WORKHUB_COORDINATION_REPLACEMENT_SCHEMA_VERSION, @@ -1190,15 +1613,7 @@ describe('Host WorkHub Coordination coordinator', () => { }, ]); - const conflicted = 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, - ); + const conflicted = await workhub.handlers['workhub.coordination.act'](stopInput, CONTEXT); assert.equal(conflicted.ok, false); if (!conflicted.ok) assert.equal(conflicted.error.code, 'operation_conflict'); assert.equal(await store.readWorkHubStopResolution(assignment.delegationId), undefined); @@ -1245,7 +1660,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 +1720,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 +1854,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 +1868,10 @@ function coordinator( continuity: { refreshCanonical: async () => undefined }, executions, sessionActions: { - assign: async ({ targetSessionId }) => ({ turnId: `turn-${targetSessionId}` }), readDelegationRetirement: async () => 'not_retired', retireDelegation: async () => ({ outcome: 'cancelled_pending' }), ...sessionActions, + assign, }, resolveCreateTarget: resolveCreateTarget ?? @@ -1551,7 +1890,7 @@ async function persistTestAssignment( store: SessionAuthorityStore, input: Parameters[0], targetTurnId: string, -): Promise<{ turnId: string }> { +): Promise<{ readonly turnId: string }> { const suffix = createHash('sha256').update(input.actionId, 'utf8').digest('hex').slice(0, 48); const content = normalizeMessageContent({ text: input.userText }); const result = await store.assignWorkHubMessage({ @@ -1590,3 +1929,17 @@ async function persistTestAssignment( }); return { turnId: result.assignment.targetTurnId }; } + +function persistTestAssignmentAction( + store: SessionAuthorityStore, + targetTurnId: string | ((input: Parameters[0]) => string), +): HostWorkHubCoordinationCoordinatorOptions['sessionActions']['assign'] { + return async (input) => { + const result = await persistTestAssignment( + store, + input, + typeof targetTurnId === 'string' ? targetTurnId : targetTurnId(input), + ); + 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..5a957aae82 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts @@ -190,6 +190,7 @@ test('WorkHub Coordination answer and summary inputs are closed and bounded', () }); 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: [ @@ -203,10 +204,12 @@ test('WorkHub Coordination candidates are bounded and carry opaque proposal iden }, state: 'active', updatedAt: 7, + latestDelegationActionId: 'action-a', }, ], }); assert.equal(result.candidates[0]?.candidateRef, 'candidate_a'); + assert.equal(result.candidates[0]?.latestDelegationActionId, 'action-a'); assert.equal(HOST_OPERATION_SPECS['workhub.coordination.candidates'].mode, 'query'); assert.equal(REMOTE_OWNER_OPERATION_GRANTS.includes('workhub.coordination.candidates'), true); assert.throws( diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index b4223846eb..c3d9ed2d8a 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -101,7 +101,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 116 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 117 as const; +// 117: WorkHub exposes only one correction linkage per bounded candidate and +// no longer returns the Host's complete active-link set. // 116: User deletion rejects workflow-owned Artifacts with operation_conflict. // 115: Artifact creation requires explicit source ownership. // 114: Artifacts are physically deleted and no longer expose tombstone status. diff --git a/packages/runtime-host/src/protocol/workhub-coordination.ts b/packages/runtime-host/src/protocol/workhub-coordination.ts index 735c6a840e..1f38312344 100644 --- a/packages/runtime-host/src/protocol/workhub-coordination.ts +++ b/packages/runtime-host/src/protocol/workhub-coordination.ts @@ -108,6 +108,8 @@ export interface WorkHubCoordinationCandidate { readonly workspace: WorkspaceProjection; readonly state: WorkHubCoordinationCandidateState; readonly updatedAt: number; + /** Latest durable linkage for compare-and-swap correction; never model-facing. */ + readonly latestDelegationActionId?: string; } export type WorkHubCoordinationCandidatesInput = Record; @@ -520,14 +522,12 @@ export function decodeWorkHubCoordinationActResult(value: unknown): WorkHubCoord } function decodeWorkHubCoordinationCandidate(value: unknown): WorkHubCoordinationCandidate { - const candidate = requireExactRecord(value, 'WorkHub Coordination candidate', [ - 'candidateRef', - 'sessionId', - 'sessionName', - 'workspace', - 'state', - 'updatedAt', - ]); + const candidate = requireShapedRecord( + value, + 'WorkHub Coordination candidate', + ['candidateRef', 'sessionId', 'sessionName', 'workspace', 'state', 'updatedAt'], + ['latestDelegationActionId'], + ); return { candidateRef: requireEntityId(candidate.candidateRef, 'WorkHub candidate ref'), sessionId: requireEntityId(candidate.sessionId, 'WorkHub candidate Session id'), @@ -535,6 +535,14 @@ function decodeWorkHubCoordinationCandidate(value: unknown): WorkHubCoordination workspace: decodeWorkspaceProjection(candidate.workspace), state: candidateState(candidate.state), updatedAt: requireCount(candidate.updatedAt, 'WorkHub candidate update time'), + ...(candidate.latestDelegationActionId === undefined + ? {} + : { + latestDelegationActionId: requireEntityId( + candidate.latestDelegationActionId, + 'WorkHub latest delegation action id', + ), + }), }; } diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index d6fad15d56..27f8c4bc59 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1359,11 +1359,17 @@ export async function createExecutionRuntimeHostComposition( continuity: continuityCoordinator, executions: coordinator, sessionActions: { - readDelegationRetirement: async (assignment) => { - const disposition = await messages.readMessageExecutionDisposition( - assignment.targetSessionId, - assignment.targetMessageId, - ); + readDelegationRetirement: async (assignment, admission) => { + const disposition = admission + ? await messages.readMessageExecutionDispositionAdmitted( + assignment.targetSessionId, + assignment.targetMessageId, + admission, + ) + : await messages.readMessageExecutionDisposition( + assignment.targetSessionId, + assignment.targetMessageId, + ); if (disposition.kind === 'recovering') return 'recovering'; if (disposition.kind === 'pending') return 'not_retired'; if (disposition.kind === 'cancelled' || disposition.kind === 'shared_turn') { diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 0d5a6eec87..4694dc7919 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -566,6 +566,16 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { ); } + readMessageExecutionDispositionAdmitted( + sessionId: string, + messageId: string, + admission: SessionAdmissionLease, + ): Promise { + return this.#sessionAdmission.runAdmitted(sessionId, admission, () => + this.#resolveMessageExecution(sessionId, messageId), + ); + } + async #resolveMessageExecution( sessionId: string, messageId: string, 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..a270e319a2 100644 --- a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts +++ b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts @@ -53,6 +53,7 @@ import type { } from '../protocol/index.js'; import { WORKHUB_COORDINATION_CANDIDATE_MAX_ITEMS } from '../protocol/index.js'; import type { ConnectionContext } from './operation-dispatcher.js'; +import type { SessionAdmissionLease } from './session-admission-gate.js'; const SIDE_CONVERSATION_LABEL = 'mode:side_conversation'; const ACTION_REPLAY_MAX_ITEMS = 256; @@ -99,7 +100,9 @@ export interface WorkHubActionGateEffects { */ probeTargetRemoval(sessionId: string): Promise<'present' | 'removed' | 'absent'>; readAssignment(actionId: string): Promise; - listActiveAssignments(): Promise; + listActiveAssignments( + targetSessionId: string, + ): Promise; readReplacement( delegationId: string, ): Promise; @@ -136,6 +139,7 @@ export interface WorkHubActionGateEffects { ): Promise; readDelegationRetirement( assignment: WorkHubDelegationAssignedMessage, + admission?: SessionAdmissionLease, ): Promise<'not_retired' | 'retired' | 'recovering'>; retireDelegation( assignment: WorkHubDelegationAssignedMessage, @@ -564,8 +568,7 @@ export class WorkHubCoordinationActionGate { return claimed; } } - const active = await this.#effects.listActiveAssignments(); - const onTarget = active.filter((assignment) => assignment.targetSessionId === targetSessionId); + const onTarget = await this.#effects.listActiveAssignments(targetSessionId); if (onTarget.length === 0) { throw new WorkHubActionGateFailure( 'action_conflict', diff --git a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts index 80eaf04b2a..5f14a3ffc7 100644 --- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts +++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts @@ -54,7 +54,7 @@ import type { WorkHubCoordinationOperationHandlerMap, } from './operation-dispatcher.js'; import type { RootTurnCoordinator } from './root-turn-coordinator.js'; -import { SessionAdmissionGate } from './session-admission-gate.js'; +import { SessionAdmissionGate, type SessionAdmissionLease } from './session-admission-gate.js'; import { SessionOperationFailure } from './session-catalog-coordinator.js'; import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js'; import { @@ -93,8 +93,8 @@ type CoordinationStores = Pick< | 'probeSessionRemoval' | 'probeStableSessionCreate' | 'readHeaderSnapshot' - | 'readMessagesSnapshot' | 'readWorkHubAssignment' + | 'readActiveWorkHubAssignmentsByTarget' | 'readWorkHubReplacement' | 'readWorkHubReplacementAbort' | 'readWorkHubSupersession' @@ -170,7 +170,10 @@ export class HostWorkHubCoordinationCoordinator { probeTargetRemoval: async (sessionId) => (await this.#stores.probeSessionRemoval(sessionId)).kind, readAssignment: (actionId) => this.#stores.readWorkHubAssignment(actionId), - listActiveAssignments: () => this.#listActiveAssignments(), + // This lookup is advisory. Stop and replacement both repeat their exact + // proof under the Coordination and target admissions before writing. + listActiveAssignments: (targetSessionId) => + this.#stores.readActiveWorkHubAssignmentsByTarget([targetSessionId]), readReplacement: (delegationId) => this.#stores.readWorkHubReplacement(delegationId), readReplacementAbort: (delegationId) => this.#stores.readWorkHubReplacementAbort(delegationId), @@ -208,6 +211,7 @@ export class HostWorkHubCoordinationCoordinator { ): Promise { const suffix = workHubDestructiveClaimIdentitySuffix(input.replacesDelegationId); return this.#commitCoordinationFact({ + admissionSessionIds: [WORKHUB_COORDINATION_SESSION_ID, input.replacedTargetSessionId], read: () => this.#stores.readWorkHubReplacement(input.replacesDelegationId), build: (existing) => ({ type: 'workhub_coordination', @@ -231,6 +235,21 @@ export class HostWorkHubCoordinationCoordinator { }), conflictMessage: 'WorkHub action identity belongs to a different replacement', beforeAppend: async () => { + const latest = ( + await this.#stores.readActiveWorkHubAssignmentsByTarget( + [input.replacedTargetSessionId], + 1, + ) + )[0]; + if ( + latest?.actionId !== input.replacesActionId || + latest.delegationId !== input.replacesDelegationId + ) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub correction source is no longer the latest active delegation', + ); + } const stopRequest = await this.#stores.readWorkHubStopRequest(input.replacesDelegationId); if (stopRequest) { const resolution = await this.#stores.readWorkHubStopResolution( @@ -285,11 +304,11 @@ export class HostWorkHubCoordinationCoordinator { userText: input.userText, }), conflictMessage: 'WorkHub delegation already has a different stop claim', - beforeAppend: async () => { - const [replacement, supersession, messages] = await Promise.all([ + beforeAppend: async (lease) => { + 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.#stores.readActiveWorkHubAssignmentsByTarget([input.targetSessionId]), ]); if (replacement || supersession) { throw new WorkHubActionGateFailure( @@ -297,14 +316,10 @@ 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. - const targetActive = activeAssignments.filter( - (assignment) => assignment.targetSessionId === input.targetSessionId, - ); - const source = targetActive.find( + const source = activeAssignments.find( (assignment) => assignment.actionId === input.stopsActionId && assignment.delegationId === input.stopsDelegationId, @@ -318,9 +333,9 @@ export class HostWorkHubCoordinationCoordinator { // A delegation whose work already finished stays linked but competes // for nothing; only work that could still be stopped makes the target // ambiguous. - for (const competitor of targetActive) { + for (const competitor of activeAssignments) { if (competitor.delegationId === source.delegationId) continue; - if ((await this.#readDelegationRetirement(competitor)) !== 'retired') { + if ((await this.#readDelegationRetirement(competitor, lease)) !== 'retired') { throw new WorkHubActionGateFailure( 'action_conflict', 'WorkHub stop target does not identify one active durable delegation', @@ -332,12 +347,6 @@ export class HostWorkHubCoordinationCoordinator { }); } - async #listActiveAssignments(): Promise { - return activeWorkHubAssignments( - await this.#stores.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID), - ); - } - #resolveStop( input: Parameters[0], ): Promise { @@ -418,7 +427,7 @@ export class HostWorkHubCoordinationCoordinator { readonly read: () => Promise; readonly build: (existing: T | undefined) => T; readonly conflictMessage: string; - readonly beforeAppend: () => Promise; + readonly beforeAppend: (lease: SessionAdmissionLease) => Promise; readonly unknownOutcomeMessage: string; }): Promise { return this.#admission.runMany( @@ -432,7 +441,7 @@ export class HostWorkHubCoordinationCoordinator { } return existing; } - await options.beforeAppend(); + await options.beforeAppend(lease); try { await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [requested]); await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); @@ -452,7 +461,29 @@ export class HostWorkHubCoordinationCoordinator { async #candidates(): Promise> { try { - return { ok: true, result: await this.#actionGate.candidates() }; + const result = await this.#actionGate.candidates(); + // One bounded read for the whole page. The candidate set is already + // capped, and a per-candidate lookup would rescan each target's history. + const latestByTarget = new Map( + ( + await this.#stores.readActiveWorkHubAssignmentsByTarget( + result.candidates.map(({ sessionId }) => sessionId), + 1, + ) + ).map((assignment) => [assignment.targetSessionId, assignment.actionId]), + ); + return { + ok: true, + result: { + candidateSetId: result.candidateSetId, + candidates: result.candidates.map((candidate) => { + const latestDelegationActionId = latestByTarget.get(candidate.sessionId); + return latestDelegationActionId + ? { ...candidate, latestDelegationActionId } + : candidate; + }), + }, + }; } catch { return { ok: false, @@ -792,26 +823,6 @@ 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 workHubDestructiveClaimIdentitySuffix(delegationId: string): string { return createHash('sha256').update(delegationId, 'utf8').digest('hex').slice(0, 48); } diff --git a/packages/storage/src/__tests__/workhub-message-assignment.test.ts b/packages/storage/src/__tests__/workhub-message-assignment.test.ts index 983cfacefc..a3e69d8160 100644 --- a/packages/storage/src/__tests__/workhub-message-assignment.test.ts +++ b/packages/storage/src/__tests__/workhub-message-assignment.test.ts @@ -77,6 +77,9 @@ test('atomically commits one WorkHub assignment and target admission', async () ), [request.assignment], ); + assert.deepEqual(await store.readActiveWorkHubAssignmentsByTarget([target.id]), [ + request.assignment, + ]); const coordination = await store.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID); assert.equal(coordination.lastMessageAt, request.assignment.ts); await store.markMessagesHandedOff({ @@ -87,6 +90,266 @@ test('atomically commits one WorkHub assignment and target admission', async () const replayAfterConsumption = await store.assignWorkHubMessage(request); assert.equal(replayAfterConsumption.kind, 'existing'); assert.deepEqual(replayAfterConsumption.assignment, request.assignment); + assert.deepEqual(await store.readActiveWorkHubAssignmentsByTarget([target.id]), [ + request.assignment, + ]); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('scans every target Message lifecycle once and preserves Coordination order', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-target-linkage-')); + const store = createSessionStore(root); + try { + await createCoordinationSession(store, root); + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }); + const unrelated = await store.create({ + cwd: root, + name: 'Login', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }); + const oldest = assignmentRequest('target-oldest', target.id, 'Payments', 'oldest-turn'); + const middle = assignmentRequest('target-middle', target.id, 'Payments', 'middle-turn'); + const newest = assignmentRequest('target-newest', target.id, 'Payments', 'newest-turn'); + await store.assignWorkHubMessage(oldest); + await store.assignWorkHubMessage( + assignmentRequest('unrelated-action', unrelated.id, 'Login', 'unrelated-turn'), + ); + await store.assignWorkHubMessage(middle); + await store.markMessagesHandedOff({ + sessionId: target.id, + messageIds: [middle.admission.messageId], + turnId: middle.admission.turnId, + }); + await store.assignWorkHubMessage(newest); + assert.equal( + await store.claimMessageAdmissionCancellation( + target.id, + newest.admission.messageId, + 'newest-cancellation-claim', + ), + 'cancelled_by_claim', + ); + + assert.deepEqual(await store.readActiveWorkHubAssignmentsByTarget([target.id]), [ + newest.assignment, + middle.assignment, + oldest.assignment, + ]); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('keeps target assignments reachable when their Message lifecycle changes', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-target-linkage-transition-')); + const store = createSessionStore(root); + try { + await createCoordinationSession(store, root); + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }); + const requests = ['transition-first', 'transition-second', 'transition-third'] + .map((actionId) => assignmentRequest(actionId, target.id, target.name, `${actionId}-turn`)) + .sort((left, right) => left.admission.messageId.localeCompare(right.admission.messageId)); + for (const request of requests) await store.assignWorkHubMessage(request); + + await store.markMessagesHandedOff({ + sessionId: target.id, + messageIds: [requests[1]!.admission.messageId], + turnId: requests[1]!.admission.turnId, + }); + assert.equal( + await store.claimMessageAdmissionCancellation( + target.id, + requests[0]!.admission.messageId, + 'transition-cancellation-claim', + ), + 'cancelled_by_claim', + ); + + assert.deepEqual( + (await store.readActiveWorkHubAssignmentsByTarget([target.id])).map( + ({ actionId }) => actionId, + ), + [...requests].reverse().map(({ assignment }) => assignment.actionId), + ); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('ignores ordinary WorkHub-shaped Message ids without hiding a real linkage', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-target-linkage-namespace-')); + const store = createSessionStore(root); + try { + await createCoordinationSession(store, root); + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }); + const unrelated = await store.create({ + cwd: root, + name: 'Login', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }); + const real = assignmentRequest('real-target-action', target.id, target.name, 'real-turn'); + const other = assignmentRequest( + 'other-target-action', + unrelated.id, + unrelated.name, + 'other-turn', + ); + await store.assignWorkHubMessage(real); + await store.assignWorkHubMessage(other); + const ordinaryContent = normalizeMessageContent({ text: 'An ordinary pending Message' }); + const ordinaryIds = [ + ...Array.from( + { length: 33 }, + (_, index) => `whm_${(index + 1).toString(16).padStart(48, '0')}`, + ), + other.admission.messageId, + ]; + assert.equal( + ordinaryIds.slice(0, -1).every((id) => id < real.admission.messageId), + true, + ); + for (const [index, messageId] of ordinaryIds.entries()) { + await store.commitMessageAdmission({ + sessionId: target.id, + turnId: `ordinary-turn-${index}`, + runId: `ordinary-run-${index}`, + messageId, + content: ordinaryContent, + submittedContentDigest: messageContentDigest(ordinaryContent), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, + admittedAt: index, + }); + } + + assert.deepEqual( + (await store.readActiveWorkHubAssignmentsByTarget([target.id])).map( + ({ actionId }) => actionId, + ), + [real.assignment.actionId], + ); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('retires a link on every terminal record and on no other outcome', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-terminal-matrix-')); + const store = createSessionStore(root); + try { + await createCoordinationSession(store, root); + const target = await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }); + const requests = Object.fromEntries( + ['plain', 'superseded', 'aborted', 'stopped', 'not-owned'].map((actionId) => [ + actionId, + assignmentRequest(actionId, target.id, 'Payments', `${actionId}-turn`), + ]), + ) as Record<'plain' | 'superseded' | 'aborted' | 'stopped' | 'not-owned', AssignmentRequest>; + for (const request of Object.values(requests)) await store.assignWorkHubMessage(request); + + const superseded = requests.superseded.assignment; + const aborted = requests.aborted.assignment; + const stopped = requests.stopped.assignment; + const notOwned = requests['not-owned'].assignment; + const supersession: WorkHubDelegationSupersededMessage = { + type: 'workhub_coordination', + id: `whx_${terminalSuffix(superseded.delegationId)}`, + turnId: 'terminal-matrix', + ts: 20, + schemaVersion: 2, + kind: 'delegation_superseded', + actionId: 'terminal-matrix-supersede', + actionFingerprint: `sha256:${'d'.repeat(64)}`, + coordinationTurnId: 'terminal-matrix', + supersededActionId: superseded.actionId, + supersededDelegationId: superseded.delegationId, + replacementDelegationId: 'whd_terminal_matrix_replacement', + }; + const replacementAbort: WorkHubDelegationReplacementAbortedMessage = { + type: 'workhub_coordination', + id: `whb_${terminalSuffix(aborted.delegationId)}`, + turnId: 'terminal-matrix', + ts: 21, + schemaVersion: 2, + kind: 'delegation_replacement_aborted', + actionId: 'terminal-matrix-abort', + actionFingerprint: `sha256:${'e'.repeat(64)}`, + coordinationTurnId: 'terminal-matrix', + abortedActionId: aborted.actionId, + abortedDelegationId: aborted.delegationId, + targetSessionId: target.id, + reason: 'target_unavailable', + }; + const stopResolution = ( + assignment: WorkHubDelegationAssignedMessage, + outcome: WorkHubDelegationStopResolvedMessage['outcome'], + ts: number, + ): WorkHubDelegationStopResolvedMessage => ({ + type: 'workhub_coordination', + id: `whz_${terminalSuffix(assignment.delegationId)}`, + turnId: 'terminal-matrix', + ts, + schemaVersion: 3, + kind: 'delegation_stop_resolved', + actionId: `terminal-matrix-stop-${outcome}`, + actionFingerprint: `sha256:${'f'.repeat(64)}`, + coordinationTurnId: 'terminal-matrix', + stopsActionId: assignment.actionId, + stopsDelegationId: assignment.delegationId, + targetSessionId: target.id, + targetTurnId: assignment.targetTurnId, + outcome, + }); + await store.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ + supersession, + replacementAbort, + stopResolution(stopped, 'stop_delivered', 22), + // `not_owned` means WorkHub never held the work, so the link survives. + stopResolution(notOwned, 'not_owned', 23), + ]); + + assert.deepEqual( + (await store.readActiveWorkHubAssignmentsByTarget([target.id])).map( + ({ actionId }) => actionId, + ), + ['not-owned', 'plain'], + ); } finally { await store.close?.(); await rm(root, { recursive: true, force: true }); @@ -455,6 +718,12 @@ async function createCoordinationSession( }); } +function terminalSuffix(delegationId: string): string { + return createHash('sha256').update(delegationId, 'utf8').digest('hex').slice(0, 48); +} + +type AssignmentRequest = ReturnType; + function assignmentRequest( actionId: string, targetSessionId: string, diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 29e0679225..27c8e3541f 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -370,6 +370,13 @@ async function createExecutionStoresForWrite sessionStore.createStableSession(request, initialBoundary)), assignWorkHubMessage: (request) => run(() => sessionStore.assignWorkHubMessage(request)), readWorkHubAssignment: (actionId) => run(() => sessionStore.readWorkHubAssignment(actionId)), + readActiveWorkHubAssignmentsByTarget: (targetSessionIds, maxAssignmentsPerTarget) => + run(() => + sessionStore.readActiveWorkHubAssignmentsByTarget( + targetSessionIds, + maxAssignmentsPerTarget, + ), + ), readWorkHubReplacement: (delegationId) => run(() => sessionStore.readWorkHubReplacement(delegationId)), readWorkHubReplacementAbort: (delegationId) => diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index c745b72675..eb05f3fb1f 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -425,6 +425,11 @@ export interface SessionAuthorityStore extends SessionStore, MessageAdmissionSto request: WorkHubMessageAssignmentRequest, ): Promise; readWorkHubAssignment(actionId: string): Promise; + /** Newest active assignment first, across every requested target. */ + readActiveWorkHubAssignmentsByTarget( + targetSessionIds: readonly string[], + maxAssignmentsPerTarget?: number, + ): Promise; readWorkHubReplacement( delegationId: string, ): Promise; @@ -699,6 +704,17 @@ class SqliteSessionStore implements SessionAuthorityStore { : undefined; } + async readActiveWorkHubAssignmentsByTarget( + targetSessionIds: readonly string[], + maxAssignmentsPerTarget?: number, + ): Promise { + await this.ensureReady(); + return this.metadata.readActiveWorkHubAssignmentsByTarget( + targetSessionIds, + maxAssignmentsPerTarget, + ); + } + async readWorkHubReplacement( delegationId: string, ): Promise { @@ -770,17 +786,7 @@ class SqliteSessionStore implements SessionAuthorityStore { messageId: string, ): Promise { await this.ensureReady(); - const throughSequence = await this.metadata.readTranscriptHighWater( - WORKHUB_COORDINATION_SESSION_ID, - ); - if (throughSequence === null) return undefined; - const messages = await this.metadata.readTranscriptMessages(WORKHUB_COORDINATION_SESSION_ID, { - messageIds: [messageId], - throughSequence, - maxMessages: 1, - maxBytes: 768 * 1024, - }); - return messages[0]; + return this.metadata.readMessageById(WORKHUB_COORDINATION_SESSION_ID, messageId); } async discardStableConversationCopy( diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index c1a13de8d5..a9bbe0e089 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -178,6 +178,9 @@ const SQLITE_TRANSCRIPT_MESSAGE_LOOKUP_BATCH_SIZE = 256; const SQLITE_TURN_CONTRIBUTION_MAX_SOURCE_MESSAGES = 1_024; const SQLITE_TURN_CONTRIBUTION_MAX_SOURCE_BYTES = 4 * 1024 * 1024; const SQLITE_TURN_LANDMARK_LEGACY_NEIGHBOR_MESSAGES = 32; +// Each target Session binds three parameters in the linkage query. Stay well +// inside SQLite's bound-parameter limit. +const WORKHUB_TARGET_LINKAGE_MAX_SESSIONS = 256; function decodeStoredMessage(value: unknown): StoredMessage { return decodePersistedStoredMessage(markPersisted(value)); @@ -2127,6 +2130,143 @@ export class SqliteSessionMetadataStore { }); } + async readActiveWorkHubAssignmentsByTarget( + targetSessionIds: readonly string[], + maxAssignmentsPerTarget?: number, + ): Promise { + this.assertOpen(); + for (const sessionId of targetSessionIds) assertSafeSessionId(sessionId); + if (targetSessionIds.length > WORKHUB_TARGET_LINKAGE_MAX_SESSIONS) { + throw new Error('Invalid WorkHub target Session count'); + } + if ( + maxAssignmentsPerTarget !== undefined && + (!Number.isSafeInteger(maxAssignmentsPerTarget) || + maxAssignmentsPerTarget < 1 || + maxAssignmentsPerTarget > 256) + ) { + throw new Error('Invalid WorkHub target Message limit'); + } + const targets = [...new Set(targetSessionIds)]; + if (targets.length === 0) return []; + return this.readTransaction(() => { + type Row = { session_id?: unknown; message_id?: unknown }; + const list = targets.map(() => '?').join(', '); + // One Message moves between these lifecycle tables. Combine every target's + // identities once, then resolve activity from the canonical Coordination + // ledger in this same read transaction. That avoids rebuilding the target + // set once per page or once per candidate, without introducing another + // durable representation. + const rows = this.db + .prepare( + ` + WITH target_messages(session_id, message_id) AS ( + SELECT session_id, message_id + FROM message_admissions + WHERE session_id IN (${list}) + AND message_id GLOB 'whm_*' + AND length(message_id) = 52 + UNION + SELECT session_id, message_id + FROM session_messages + WHERE session_id IN (${list}) + AND message_id GLOB 'whm_*' + AND length(message_id) = 52 + UNION + SELECT session_id, message_id + FROM cancelled_message_admissions + WHERE session_id IN (${list}) + AND message_id GLOB 'whm_*' + AND length(message_id) = 52 + ) + SELECT target.session_id, target.message_id + FROM target_messages AS target + CROSS JOIN session_messages AS assignment INDEXED BY session_messages_by_identity + WHERE assignment.session_id = ? + AND assignment.message_id = 'wha_' || substr(target.message_id, 5) + ORDER BY assignment.sequence DESC + `, + ) + .iterate( + ...targets, + ...targets, + ...targets, + WORKHUB_COORDINATION_SESSION_ID, + ) as Iterable; + const assignments: WorkHubDelegationAssignedMessage[] = []; + const acceptedPerTarget = new Map(); + for (const row of rows) { + if (typeof row.message_id !== 'string' || typeof row.session_id !== 'string') { + throw new SessionMetadataConflictError('Invalid WorkHub target Message identity'); + } + const targetSessionId = row.session_id; + if ( + maxAssignmentsPerTarget !== undefined && + (acceptedPerTarget.get(targetSessionId) ?? 0) >= maxAssignmentsPerTarget + ) { + continue; + } + const assignment = this.readMessageByIdSync( + WORKHUB_COORDINATION_SESSION_ID, + `wha_${row.message_id.slice('whm_'.length)}`, + ); + if ( + assignment?.type !== 'workhub_coordination' || + assignment.kind !== 'delegation_assigned' || + assignment.targetSessionId !== targetSessionId || + assignment.targetMessageId !== row.message_id + ) { + continue; + } + const terminalSuffix = createHash('sha256') + .update(assignment.delegationId, 'utf8') + .digest('hex') + .slice(0, 48); + const supersession = this.readMessageByIdSync( + WORKHUB_COORDINATION_SESSION_ID, + `whx_${terminalSuffix}`, + ); + if ( + supersession?.type === 'workhub_coordination' && + supersession.kind === 'delegation_superseded' + ) { + continue; + } + const replacementAbort = this.readMessageByIdSync( + WORKHUB_COORDINATION_SESSION_ID, + `whb_${terminalSuffix}`, + ); + if ( + replacementAbort?.type === 'workhub_coordination' && + replacementAbort.kind === 'delegation_replacement_aborted' + ) { + continue; + } + const stopResolution = this.readMessageByIdSync( + WORKHUB_COORDINATION_SESSION_ID, + `whz_${terminalSuffix}`, + ); + if ( + stopResolution?.type === 'workhub_coordination' && + stopResolution.kind === 'delegation_stop_resolved' && + stopResolution.outcome !== 'not_owned' + ) { + continue; + } + assignments.push(assignment); + acceptedPerTarget.set(targetSessionId, (acceptedPerTarget.get(targetSessionId) ?? 0) + 1); + } + return assignments; + }); + } + + async readMessageById(sessionId: string, messageId: string): Promise { + this.assertOpen(); + assertSafeSessionId(sessionId); + assertSafeSessionId(messageId); + return this.readTransaction(() => this.readMessageByIdSync(sessionId, messageId)); + } + async markMessagesHandedOff(input: MarkMessagesHandedOffInput): Promise { this.assertOpen(); assertSafeSessionId(input.sessionId);