You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/09/04 01:08:45 UTC
(maka) 01/01: fix(workhub): bound coordination history replay
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch fix/4647-bounded-coordination-replay
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 5308791feceec8e1006887c35ca4f70c1ef2f55a
Author: AstroHan <le...@outlook.com>
AuthorDate: Fri Sep 4 09:08:24 2026 +0800
fix(workhub): bound coordination history replay
Generated-by: Codex
---
.../runtime-host-workhub-ipc-main.test.ts | 6 ++
.../main/__tests__/workhub-session-port.test.ts | 113 ++++++---------------
apps/desktop/src/main/runtime-host-client.ts | 4 +
.../src/main/runtime-host-workhub-ipc-main.ts | 2 +
apps/desktop/src/preload/bridge-contract.d.ts | 4 +
apps/desktop/src/preload/preload.ts | 18 ++++
apps/desktop/src/renderer/app-shell.tsx | 2 +
.../src/renderer/workhub-coordination-port.ts | 85 ++++++----------
.../workhub-coordination-coordinator.test.ts | 45 ++++----
.../workhub-coordination-protocol.test.ts | 14 +++
packages/runtime-host/src/protocol/index.ts | 3 +-
packages/runtime-host/src/protocol/operations.ts | 1 +
.../src/protocol/workhub-coordination.ts | 53 ++++++++++
.../src/server/workhub-coordination-coordinator.ts | 53 +++++-----
.../__tests__/workhub-message-assignment.test.ts | 71 +++++++++++++
packages/storage/src/execution-stores.ts | 1 +
packages/storage/src/session-store.ts | 8 ++
.../storage/src/sqlite-session-metadata-schema.ts | 53 +++++++++-
.../storage/src/sqlite-session-metadata-store.ts | 89 ++++++++++++++--
19 files changed, 433 insertions(+), 192 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts
index 2aa190fb92..21a09d6efb 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-workhub-ipc-main.test.ts
@@ -52,6 +52,9 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain'
candidateSetId: `sha256:${'a'.repeat(64)}`,
candidates: [],
}),
+ listActiveWorkHubDelegations: async () => ({
+ delegations: [{ actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }],
+ }),
actWorkHubCoordination: async (input: unknown) => {
actions.push(input);
return {
@@ -101,6 +104,9 @@ test('projects WorkHub coordination resolution through its dedicated IPC domain'
candidateSetId: `sha256:${'a'.repeat(64)}`,
candidates: [],
});
+ assert.deepEqual(await handlers.get('workhub:active')?.({}), {
+ delegations: [{ actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }],
+ });
assert.deepEqual(
await handlers.get('workhub:act')?.({}, {
actionId: 'create-action',
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..b8f0f420bc 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,9 +374,6 @@ test('durable supersession terminalizes only the replaced linkage', () => {
projectWorkHubCoordinationTurns(messages).map((turn) => turn.assignment?.linkState),
['superseded', 'active'],
);
- assert.deepEqual(projectWorkHubActiveDelegations(
- messages.map((message, sequence) => ({ message, sequence })),
- ), [{ actionId: 'action-new', targetSessionId: 'login', sequence: 1 }]);
});
test('Coordination transcript adapter emits an initial empty ready snapshot and closes cleanly', async () => {
@@ -447,6 +415,7 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and
candidateSetId: `sha256:${'a'.repeat(64)}`,
candidates: [],
}),
+ active: async () => ({ delegations: [] }),
act: async () => ({
ok: true,
result: {
@@ -465,7 +434,7 @@ test('Coordination transcript adapter emits an initial empty ready snapshot and
assert.equal(closes, 1);
});
-test('Coordination transcript reset rebuilds active linkage outside the resident window', async () => {
+test('Coordination transcript reads old active linkage without replaying older history', async () => {
const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'coordination' });
const assignment: StoredMessage = {
type: 'workhub_coordination',
@@ -485,13 +454,13 @@ test('Coordination transcript reset rebuilds active linkage outside the resident
disposition: 'delegate_existing',
userText: 'Continue payments',
};
- const recent: StoredMessage = {
+ const recent: StoredMessage[] = Array.from({ length: 40 }, (_, index) => ({
type: 'user',
- id: 'recent-user',
- turnId: 'recent-turn',
- ts: 2,
- text: 'Recent coordination',
- };
+ id: `recent-user-${index}`,
+ turnId: `recent-turn-${index}`,
+ ts: index + 2,
+ text: `Recent coordination ${index}`,
+ }));
const fragment = (message: StoredMessage, sequence: number) => {
const data = new TextEncoder().encode(JSON.stringify(message));
return {
@@ -506,9 +475,9 @@ test('Coordination transcript reset rebuilds active linkage outside the resident
let deliver: ((batch: DesktopTranscriptBatch) => void) | undefined;
let generation = 'generation-1';
let historyLoads = 0;
- let historyBatchReady = true;
+ let activeReads = 0;
const snapshots: unknown[] = [];
- const adapter = createDesktopWorkHubCoordinationPort({
+ const portDeps: Parameters<typeof createDesktopWorkHubCoordinationPort>[0] = {
sessionId,
transcripts: {
open: async (_requestedSessionId, handler) => {
@@ -518,8 +487,8 @@ test('Coordination transcript reset rebuilds active linkage outside the resident
deliverySequence: 1,
generation,
hostEpoch: 'epoch-1',
- durableThrough: 1,
- fragments: [fragment(recent, 1)],
+ durableThrough: 40,
+ fragments: recent.map((message, index) => fragment(message, index + 1)),
evictedDurableSequences: [],
completedOverlayMessageIds: [],
hasOlder: true,
@@ -534,26 +503,22 @@ test('Coordination transcript reset rebuilds active linkage outside the resident
readThroughMessageId: null,
loadBefore: async () => {
historyLoads += 1;
- handler({
- sessionId: 'coordination',
- deliverySequence: historyLoads + 1,
- generation,
- hostEpoch: 'epoch-1',
- durableThrough: 1,
- fragments: [fragment(assignment, 0)],
- evictedDurableSequences: [],
- completedOverlayMessageIds: [],
- hasOlder: false,
- hasNewer: false,
- reset: false,
- ready: historyBatchReady,
- });
},
loadAround: async () => {},
close: async () => {},
};
},
},
+ active: async () => {
+ activeReads += 1;
+ return {
+ delegations: [{
+ actionId: assignment.actionId,
+ targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }),
+ sequence: 0,
+ }],
+ };
+ },
record: async (input) => ({ turnId: input.turnId }),
candidates: async () => ({
candidateSetId: `sha256:${'b'.repeat(64)}`,
@@ -563,7 +528,8 @@ test('Coordination transcript reset rebuilds active linkage outside the resident
ok: true,
result: { disposition: 'answer_here', coordinationTurnId: 'coordination-turn' },
}),
- });
+ };
+ const adapter = createDesktopWorkHubCoordinationPort(portDeps);
const handle = await adapter.open(
(_turns, activeDelegations) => snapshots.push(activeDelegations),
@@ -574,17 +540,17 @@ test('Coordination transcript reset rebuilds active linkage outside the resident
targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }),
sequence: 0,
}]);
+ assert.equal(historyLoads, 0);
+ assert.equal(activeReads, 1);
generation = 'generation-2';
- historyBatchReady = false;
- const snapshotsBeforeReset = snapshots.length;
deliver?.({
sessionId: 'coordination',
deliverySequence: 3,
generation,
hostEpoch: 'epoch-1',
- durableThrough: 1,
- fragments: [fragment(recent, 1)],
+ durableThrough: 40,
+ fragments: recent.map((message, index) => fragment(message, index + 1)),
evictedDurableSequences: [],
completedOverlayMessageIds: [],
hasOlder: true,
@@ -592,25 +558,10 @@ test('Coordination transcript reset rebuilds active linkage outside the resident
reset: true,
ready: false,
});
- await Promise.resolve();
- await Promise.resolve();
+ await new Promise<void>((resolve) => setImmediate(resolve));
- assert.equal(historyLoads, 2);
- assert.equal(snapshots.length, snapshotsBeforeReset);
- deliver?.({
- sessionId: 'coordination',
- deliverySequence: 5,
- generation,
- hostEpoch: 'epoch-1',
- durableThrough: 1,
- fragments: [fragment(recent, 1)],
- evictedDurableSequences: [],
- completedOverlayMessageIds: [],
- hasOlder: false,
- hasNewer: false,
- reset: false,
- ready: true,
- });
+ assert.equal(historyLoads, 0);
+ assert.equal(activeReads, 2);
assert.deepEqual(snapshots.at(-1), [{
actionId: 'action-old',
targetSessionId: desktopSessionKey({ hostId: 'local-host', sessionId: 'payments' }),
diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts
index 5a39dba44b..0627818e02 100644
--- a/apps/desktop/src/main/runtime-host-client.ts
+++ b/apps/desktop/src/main/runtime-host-client.ts
@@ -971,6 +971,10 @@ export class DesktopRuntimeHostClient {
return this.request("workhub.coordination.candidates", {});
}
+ listActiveWorkHubDelegations() {
+ return this.request("workhub.coordination.active", {});
+ }
+
actWorkHubCoordination(
input: OperationInput<"workhub.coordination.act">,
): Promise<OperationOutput<"workhub.coordination.act">> {
diff --git a/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts b/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts
index 4210a0c38c..b09acd7618 100644
--- a/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts
+++ b/apps/desktop/src/main/runtime-host-workhub-ipc-main.ts
@@ -32,6 +32,7 @@ type RuntimeHostWorkHubClient = Pick<
DesktopRuntimeHostClient,
| 'actWorkHubCoordination'
| 'answerWorkHubCoordination'
+ | 'listActiveWorkHubDelegations'
| 'listWorkHubCoordinationCandidates'
| 'recordWorkHubCoordination'
| 'resolveWorkHubCoordinationSession'
@@ -60,6 +61,7 @@ export function registerRuntimeHostWorkHubIpc(
client.recordWorkHubCoordination(input),
);
ipcMain.handle('workhub:candidates', () => client.listWorkHubCoordinationCandidates());
+ ipcMain.handle('workhub:active', () => client.listActiveWorkHubDelegations());
ipcMain.handle('workhub:act', async (_event, rawInput: RendererWorkHubActionInput) => {
try {
const proposal = rawInput?.proposal;
diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts
index fe84b755f4..128d2fa191 100644
--- a/apps/desktop/src/preload/bridge-contract.d.ts
+++ b/apps/desktop/src/preload/bridge-contract.d.ts
@@ -1049,6 +1049,10 @@ export interface MakaBridge {
candidates(
coordinationSessionId: string,
): Promise<OperationOutput<'workhub.coordination.candidates'>>;
+ /** Read current active delegation linkage without replaying Coordination history. */
+ active(
+ coordinationSessionId: string,
+ ): Promise<OperationOutput<'workhub.coordination.active'>>;
/** Submit a typed proposal; trusted creation context is added outside the renderer. */
act(
coordinationSessionId: string,
diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts
index ce26162376..1660f8e951 100644
--- a/apps/desktop/src/preload/preload.ts
+++ b/apps/desktop/src/preload/preload.ts
@@ -1936,6 +1936,24 @@ const makaBridge = {
})),
};
},
+ async active(
+ coordinationSessionId: string,
+ ): Promise<OperationOutput<'workhub.coordination.active'>> {
+ const scope = await resolveDesktopWorkHubCoordinationCreateScope(
+ coordinationSessionId,
+ runtimeHostSessionRef,
+ );
+ const result = await ipcRenderer.invoke(
+ 'workhub:active',
+ scope,
+ ) as OperationOutput<'workhub.coordination.active'>;
+ return {
+ delegations: result.delegations.map((delegation) => ({
+ ...delegation,
+ targetSessionId: recordRuntimeHostSessionScope(scope, delegation.targetSessionId),
+ })),
+ };
+ },
async act(
coordinationSessionId: string,
input: Omit<OperationInput<'workhub.coordination.act'>, 'create'>,
diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx
index 3f3a212dd7..38eed65b20 100644
--- a/apps/desktop/src/renderer/app-shell.tsx
+++ b/apps/desktop/src/renderer/app-shell.tsx
@@ -1465,6 +1465,8 @@ function AppShellContent({
window.maka.workHub.record(workHubCoordinationSessionId!, input),
candidates: () =>
window.maka.workHub.candidates(workHubCoordinationSessionId!),
+ active: () =>
+ window.maka.workHub.active(workHubCoordinationSessionId!),
act: (input) =>
window.maka.workHub.act(workHubCoordinationSessionId!, input),
}),
diff --git a/apps/desktop/src/renderer/workhub-coordination-port.ts b/apps/desktop/src/renderer/workhub-coordination-port.ts
index 114fed8532..d2100b7a55 100644
--- a/apps/desktop/src/renderer/workhub-coordination-port.ts
+++ b/apps/desktop/src/renderer/workhub-coordination-port.ts
@@ -34,6 +34,7 @@ import type {
WorkHubCoordinationActInput,
WorkHubCoordinationActResult,
WorkHubCoordinationCandidatesResult,
+ WorkHubCoordinationActiveResult,
OperationOutcome,
OperationError,
} from '@maka/runtime-host/protocol';
@@ -53,6 +54,7 @@ export function createDesktopWorkHubCoordinationPort(deps: {
assistantText: string;
}): Promise<{ turnId: string }>;
candidates(): Promise<WorkHubCoordinationCandidatesResult>;
+ active(): Promise<WorkHubCoordinationActiveResult>;
act(
input: Omit<WorkHubCoordinationActInput, 'create'>,
): Promise<OperationOutcome<'workhub.coordination.act'>>;
@@ -71,49 +73,38 @@ export function createDesktopWorkHubCoordinationPort(deps: {
const store = new DesktopTranscriptRangeStore(deps.sessionId);
let disposed = false;
let ready = false;
- let historyReady = false;
let historyGeneration = 0;
let handle: Awaited<ReturnType<typeof deps.transcripts.open>> | undefined;
let historyLane = Promise.resolve();
- const coordinationMessagesBySequence = new Map<number, StoredMessage>();
+ let activeDelegations: readonly WorkHubActiveDelegation[] = [];
const emit = () => {
- const messages = store.snapshot().messages;
- handler(
- projectWorkHubCoordinationTurns(messages),
- projectWorkHubActiveDelegations(
- [...coordinationMessagesBySequence.entries()]
- .sort(([left], [right]) => left - right)
- .map(([sequence, message]) => ({ sequence, message })),
- ),
- );
+ handler(projectWorkHubCoordinationTurns(store.snapshot().messages), activeDelegations);
+ };
+ const scheduleRefresh = (generation: number) => {
+ historyLane = historyLane
+ .then(async () => {
+ if (disposed || generation !== historyGeneration) return;
+ await rebuildVisibleHistory(generation);
+ const result = await deps.active();
+ if (disposed || generation !== historyGeneration) return;
+ activeDelegations = result.delegations;
+ if (ready) emit();
+ })
+ .catch(onError);
};
const opened = await deps.transcripts.open(
deps.sessionId,
(batch) => {
if (disposed) return;
try {
+ const reset = batch.reset === true;
if (batch.reset) {
- coordinationMessagesBySequence.clear();
ready = false;
- historyReady = false;
- const generation = ++historyGeneration;
- if (handle) {
- historyLane = historyLane.then(async () => {
- if (disposed || generation !== historyGeneration) return;
- await rebuildCompleteHistory(generation);
- }).catch((error) => {
- onError(error);
- });
- }
+ historyGeneration += 1;
}
const changed = store.accept(batch);
- for (const { sequence, message } of store.durableEntries()) {
- if (message.type === 'workhub_coordination') {
- coordinationMessagesBySequence.set(sequence, message);
- }
- }
ready ||= batch.ready;
- if (historyReady && ready && (changed || batch.ready)) emit();
+ if (handle && (reset || changed || batch.ready)) scheduleRefresh(historyGeneration);
} catch (error) {
onError(error);
}
@@ -127,9 +118,15 @@ export function createDesktopWorkHubCoordinationPort(deps: {
});
handle = opened;
- async function rebuildCompleteHistory(generation: number): Promise<void> {
+ async function rebuildVisibleHistory(generation: number): Promise<void> {
if (!handle) return;
- while (!disposed && generation === historyGeneration && store.range().hasOlder) {
+ while (
+ !disposed &&
+ generation === historyGeneration &&
+ store.range().hasOlder &&
+ projectWorkHubCoordinationTurns(store.snapshot().messages).length <
+ WORKHUB_COORDINATION_TURN_LIMIT
+ ) {
const before = store.range().oldestSequence;
await handle.loadBefore(before);
const after = store.range();
@@ -143,12 +140,13 @@ export function createDesktopWorkHubCoordinationPort(deps: {
await handle.loadAround(range.durableThrough);
}
if (disposed || generation !== historyGeneration) return;
- historyReady = true;
- if (ready) emit();
}
try {
- await rebuildCompleteHistory(historyGeneration);
+ await rebuildVisibleHistory(historyGeneration);
+ const result = await deps.active();
+ activeDelegations = result.delegations;
+ if (ready) emit();
} catch (error) {
disposed = true;
await handle.close().catch(() => undefined);
@@ -164,27 +162,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__/workhub-coordination-coordinator.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts
index cb70f8d3b9..6c6422511a 100644
--- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts
@@ -545,6 +545,18 @@ describe('Host WorkHub Coordination coordinator', () => {
],
);
assert.equal(assignments.length, 1);
+ assert.deepEqual(await first.handlers['workhub.coordination.active']({}, CONTEXT), {
+ ok: true,
+ result: {
+ delegations: [
+ {
+ actionId: 'payments-action',
+ targetSessionId: candidates.result.candidates[0]!.sessionId,
+ sequence: 0,
+ },
+ ],
+ },
+ });
} finally {
await store.close?.();
}
@@ -656,6 +668,7 @@ describe('Host WorkHub Coordination coordinator', () => {
(await store.readWorkHubStopResolution(assignment.delegationId))?.outcome,
'stop_delivered',
);
+ assert.deepEqual(await store.listActiveWorkHubAssignments(), []);
} finally {
await store.close?.();
}
@@ -701,18 +714,12 @@ describe('Host WorkHub Coordination coordinator', () => {
let injected = false;
const stores = new Proxy(store, {
get(authority, property, receiver) {
- if (property === 'readMessagesSnapshot') {
- return async (sessionId: string) => {
- const messages = await authority.readMessagesSnapshot(sessionId);
+ if (property === 'listActiveWorkHubAssignments') {
+ return async () => {
+ const assignments = await authority.listActiveWorkHubAssignments();
if (
!injected &&
- sessionId === WORKHUB_COORDINATION_SESSION_ID &&
- messages.some(
- (message) =>
- message.type === 'workhub_coordination' &&
- message.kind === 'delegation_assigned' &&
- message.actionId === 'source-action',
- )
+ assignments.some(({ assignment }) => assignment.actionId === 'source-action')
) {
injected = true;
await persistTestAssignment(
@@ -728,7 +735,7 @@ describe('Host WorkHub Coordination coordinator', () => {
'racing-turn',
);
}
- return messages;
+ return assignments;
};
}
const value = Reflect.get(authority, property, receiver) as unknown;
@@ -1305,11 +1312,9 @@ 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.
+ test('one stop reads the active delegation authority twice, not the transcript', async () => {
+ // The Gate derives the delegation, then admission reproves it under the
+ // lease. Those are the two current-state reads that decide.
const root = await mkdtemp(join(tmpdir(), 'maka-workhub-scan-count-'));
const store = createSessionStore(root);
try {
@@ -1324,10 +1329,10 @@ describe('Host WorkHub Coordination coordinator', () => {
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);
+ if (property === 'listActiveWorkHubAssignments') {
+ return async () => {
+ if (counting) coordinationReads += 1;
+ return authority.listActiveWorkHubAssignments();
};
}
const value = Reflect.get(authority, property, receiver) as unknown;
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..7228f6279b 100644
--- a/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts
+++ b/packages/runtime-host/src/__tests__/workhub-coordination-protocol.test.ts
@@ -24,6 +24,7 @@ import {
decodeWorkHubCoordinationActInput,
decodeWorkHubCoordinationActResult,
decodeWorkHubCoordinationAnswerInput,
+ decodeWorkHubCoordinationActiveResult,
decodeWorkHubCoordinationCandidatesResult,
decodeWorkHubCoordinationRecordInput,
decodeWorkHubCoordinationResolveInput,
@@ -50,6 +51,19 @@ test('WorkHub Coordination resolve has a closed empty input and bounded identity
);
});
+test('WorkHub Coordination active linkage is a current-state query', () => {
+ assert.deepEqual(
+ decodeWorkHubCoordinationActiveResult({
+ delegations: [{ actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }],
+ }),
+ {
+ delegations: [{ actionId: 'action-a', targetSessionId: 'session-a', sequence: 3 }],
+ },
+ );
+ assert.equal(HOST_OPERATION_SPECS['workhub.coordination.active'].mode, 'query');
+ assert.equal(REMOTE_OWNER_OPERATION_GRANTS.includes('workhub.coordination.active'), true);
+});
+
test('WorkHub Coordination answer and summary inputs are closed and bounded', () => {
assert.deepEqual(
decodeWorkHubCoordinationAnswerInput({ turnId: 'answer-turn', text: 'What changed?' }),
diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts
index d5a69a4d1e..9c19a630bc 100644
--- a/packages/runtime-host/src/protocol/index.ts
+++ b/packages/runtime-host/src/protocol/index.ts
@@ -100,7 +100,8 @@ 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 = 109 as const;
+export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 110 as const;
+// 110: WorkHub exposes current active delegation linkage as a Host-owned query.
// 109: accepted Client Capability invocations may carry one bounded nested form
// Interaction request/result round trip.
// 108: Session Interaction snapshots, forwarded Runtime events, and Agent Graph
diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts
index 0d52b1445e..69e46df638 100644
--- a/packages/runtime-host/src/protocol/operations.ts
+++ b/packages/runtime-host/src/protocol/operations.ts
@@ -349,6 +349,7 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([
'web-search.execute',
'workhub.coordination.answer',
'workhub.coordination.act',
+ 'workhub.coordination.active',
'workhub.coordination.candidates',
'workhub.coordination.record',
'workhub.coordination.resolve',
diff --git a/packages/runtime-host/src/protocol/workhub-coordination.ts b/packages/runtime-host/src/protocol/workhub-coordination.ts
index 735c6a840e..5af397e78d 100644
--- a/packages/runtime-host/src/protocol/workhub-coordination.ts
+++ b/packages/runtime-host/src/protocol/workhub-coordination.ts
@@ -117,6 +117,18 @@ export interface WorkHubCoordinationCandidatesResult {
readonly candidates: readonly WorkHubCoordinationCandidate[];
}
+export type WorkHubCoordinationActiveInput = Record<string, never>;
+
+export interface WorkHubCoordinationActiveDelegation {
+ readonly actionId: string;
+ readonly targetSessionId: string;
+ readonly sequence: number;
+}
+
+export interface WorkHubCoordinationActiveResult {
+ readonly delegations: readonly WorkHubCoordinationActiveDelegation[];
+}
+
export type WorkHubCoordinationProposal =
| { readonly disposition: 'answer_here' }
| { readonly disposition: 'clarify'; readonly assistantText: string }
@@ -249,6 +261,17 @@ export const WORKHUB_COORDINATION_OPERATION_SPECS = {
decodeInput: decodeWorkHubCoordinationCandidatesInput,
decodeOutput: decodeWorkHubCoordinationCandidatesResult,
}),
+ 'workhub.coordination.active': defineOperation<
+ WorkHubCoordinationActiveInput,
+ WorkHubCoordinationActiveResult,
+ (typeof CANDIDATE_ERRORS)[number]
+ >({
+ mode: 'query',
+ availability: 'ready',
+ errors: CANDIDATE_ERRORS,
+ decodeInput: decodeWorkHubCoordinationActiveInput,
+ decodeOutput: decodeWorkHubCoordinationActiveResult,
+ }),
'workhub.coordination.act': defineOperation<
WorkHubCoordinationActInput,
WorkHubCoordinationActResult,
@@ -329,6 +352,36 @@ export function decodeWorkHubCoordinationCandidatesInput(
return {};
}
+export function decodeWorkHubCoordinationActiveInput(
+ value: unknown,
+): WorkHubCoordinationActiveInput {
+ requireExactRecord(value, 'WorkHub Coordination active input', []);
+ return {};
+}
+
+export function decodeWorkHubCoordinationActiveResult(
+ value: unknown,
+): WorkHubCoordinationActiveResult {
+ const result = requireExactRecord(value, 'WorkHub Coordination active result', ['delegations']);
+ if (!Array.isArray(result.delegations)) {
+ throw invalidProtocolFrame('Invalid active WorkHub delegations');
+ }
+ return {
+ delegations: result.delegations.map((value) => {
+ const delegation = requireExactRecord(value, 'Active WorkHub delegation', [
+ 'actionId',
+ 'targetSessionId',
+ 'sequence',
+ ]);
+ return {
+ actionId: requireEntityId(delegation.actionId, 'WorkHub action id'),
+ targetSessionId: requireEntityId(delegation.targetSessionId, 'WorkHub target Session id'),
+ sequence: requireCount(delegation.sequence, 'WorkHub transcript sequence'),
+ };
+ }),
+ };
+}
+
export function decodeWorkHubCoordinationCandidatesResult(
value: unknown,
): WorkHubCoordinationCandidatesResult {
diff --git a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts
index 80eaf04b2a..8847621fb2 100644
--- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts
+++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts
@@ -88,12 +88,12 @@ type CoordinationStores = Pick<
| 'appendMessages'
| 'createStableSession'
| 'listHeaders'
+ | 'listActiveWorkHubAssignments'
| 'claimWorkHubAction'
| 'readWorkHubActionClaim'
| 'probeSessionRemoval'
| 'probeStableSessionCreate'
| 'readHeaderSnapshot'
- | 'readMessagesSnapshot'
| 'readWorkHubAssignment'
| 'readWorkHubReplacement'
| 'readWorkHubReplacementAbort'
@@ -132,6 +132,7 @@ export class HostWorkHubCoordinationCoordinator {
'workhub.coordination.resolve': () => this.#resolve(),
'workhub.coordination.answer': (input, context) => this.#answer(input, context),
'workhub.coordination.record': (input) => this.#record(input),
+ 'workhub.coordination.active': () => this.#active(),
'workhub.coordination.candidates': () => this.#candidates(),
'workhub.coordination.act': (input, context) => this.#act(input, context),
};
@@ -286,10 +287,10 @@ export class HostWorkHubCoordinationCoordinator {
}),
conflictMessage: 'WorkHub delegation already has a different stop claim',
beforeAppend: async () => {
- const [replacement, supersession, messages] = await Promise.all([
+ const [replacement, supersession, activeAssignments] = await Promise.all([
this.#stores.readWorkHubReplacement(input.stopsDelegationId),
this.#stores.readWorkHubSupersession(input.stopsDelegationId),
- this.#stores.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID),
+ this.#listActiveAssignments(),
]);
if (replacement || supersession) {
throw new WorkHubActionGateFailure(
@@ -297,7 +298,6 @@ export class HostWorkHubCoordinationCoordinator {
'WorkHub delegation is already being replaced',
);
}
- const activeAssignments = activeWorkHubAssignments(messages);
// Held lanes make this the last moment the one-target proof can change.
// It is proved from opaque delegation identity, so a concurrent rename
// is harmless while a concurrent delegation to the same Session is not.
@@ -333,9 +333,7 @@ export class HostWorkHubCoordinationCoordinator {
}
async #listActiveAssignments(): Promise<readonly WorkHubDelegationAssignedMessage[]> {
- return activeWorkHubAssignments(
- await this.#stores.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID),
- );
+ return (await this.#stores.listActiveWorkHubAssignments()).map(({ assignment }) => assignment);
}
#resolveStop(
@@ -464,6 +462,27 @@ export class HostWorkHubCoordinationCoordinator {
}
}
+ async #active(): Promise<OperationOutcome<'workhub.coordination.active'>> {
+ try {
+ const delegations = (await this.#stores.listActiveWorkHubAssignments()).map(
+ ({ sequence, assignment }) => ({
+ actionId: assignment.actionId,
+ targetSessionId: assignment.targetSessionId,
+ sequence,
+ }),
+ );
+ return { ok: true, result: { delegations } };
+ } catch {
+ return {
+ ok: false,
+ error: {
+ code: 'persistence_failed',
+ message: 'Active WorkHub delegations are unavailable',
+ },
+ };
+ }
+ }
+
async #act(
input: WorkHubCoordinationActInput,
context: ConnectionContext,
@@ -792,26 +811,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<string>();
- 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..8f1f8449db 100644
--- a/packages/storage/src/__tests__/workhub-message-assignment.test.ts
+++ b/packages/storage/src/__tests__/workhub-message-assignment.test.ts
@@ -23,6 +23,7 @@ import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';
+import { DatabaseSync } from 'node:sqlite';
import { messageContentDigest, normalizeMessageContent } from '@maka/core/events';
import {
WORKHUB_COORDINATION_SESSION_ID,
@@ -77,6 +78,12 @@ test('atomically commits one WorkHub assignment and target admission', async ()
),
[request.assignment],
);
+ assert.deepEqual(await store.listActiveWorkHubAssignments(), [
+ {
+ sequence: 0,
+ assignment: request.assignment,
+ },
+ ]);
const coordination = await store.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID);
assert.equal(coordination.lastMessageAt, request.assignment.ts);
await store.markMessagesHandedOff({
@@ -93,6 +100,51 @@ test('atomically commits one WorkHub assignment and target admission', async ()
}
});
+test('migration backfills active WorkHub assignments from the historical ledger', async () => {
+ const root = await mkdtemp(join(tmpdir(), 'maka-workhub-active-migration-'));
+ const store = createSessionStore(root);
+ try {
+ await createCoordinationSession(store, root);
+ const target = await store.create({
+ cwd: root,
+ name: 'Payments',
+ llmConnectionSlug: 'test',
+ model: 'test',
+ permissionMode: 'ask',
+ });
+ const request = assignmentRequest('legacy-action', target.id, 'Payments', 'legacy-turn');
+ await store.assignWorkHubMessage(request);
+ await store.close?.();
+
+ const legacy = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ legacy.exec(`
+ DROP TABLE workhub_active_delegations;
+ UPDATE session_metadata_schema
+ SET version = 38
+ WHERE scope = 'session_metadata';
+ `);
+ } finally {
+ legacy.close();
+ }
+
+ const migrated = createSessionStore(root);
+ try {
+ assert.deepEqual(await migrated.listActiveWorkHubAssignments(), [
+ {
+ sequence: 0,
+ assignment: request.assignment,
+ },
+ ]);
+ } finally {
+ await migrated.close?.();
+ }
+ } finally {
+ await store.close?.();
+ await rm(root, { recursive: true, force: true });
+ }
+});
+
test('rolls create_new Session back when assignment validation fails', async () => {
const root = await mkdtemp(join(tmpdir(), 'maka-workhub-create-assignment-'));
const store = createSessionStore(root);
@@ -242,6 +294,12 @@ test('atomically commits a replacement assignment with the old-link supersession
.map((message) => message.kind),
['delegation_assigned', 'delegation_assigned', 'delegation_superseded'],
);
+ assert.deepEqual(await store.listActiveWorkHubAssignments(), [
+ {
+ sequence: 1,
+ assignment: committedAssignment,
+ },
+ ]);
} finally {
await store.close?.();
await rm(root, { recursive: true, force: true });
@@ -317,6 +375,7 @@ test('an aborted replacement cannot later commit a supersession', async () => {
reason: 'target_unavailable',
};
await store.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [abort]);
+ assert.deepEqual(await store.listActiveWorkHubAssignments(), []);
await assert.rejects(
store.assignWorkHubMessage({ ...base, assignment, supersession }),
@@ -378,6 +437,10 @@ test('an unresolved stop claim blocks replacement while not_owned releases the l
};
await store.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [request]);
assert.deepEqual(await store.readWorkHubStopRequest(original.assignment.delegationId), request);
+ assert.deepEqual(
+ (await store.listActiveWorkHubAssignments()).map(({ assignment }) => assignment.actionId),
+ [original.assignment.actionId],
+ );
const base = assignmentRequest('after-stop', destination.id, 'Login', 'destination-turn');
const assignment: WorkHubDelegationAssignedMessage = {
@@ -426,10 +489,18 @@ test('an unresolved stop claim blocks replacement while not_owned releases the l
await store.readWorkHubStopResolution(original.assignment.delegationId),
resolution,
);
+ assert.deepEqual(
+ (await store.listActiveWorkHubAssignments()).map(({ assignment }) => assignment.actionId),
+ [original.assignment.actionId],
+ );
assert.equal(
(await store.assignWorkHubMessage({ ...base, assignment, supersession })).kind,
'assigned',
);
+ assert.deepEqual(
+ (await store.listActiveWorkHubAssignments()).map(({ assignment }) => assignment.actionId),
+ [assignment.actionId],
+ );
} finally {
await store.close?.();
await rm(root, { recursive: true, force: true });
diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts
index ed9714e275..154d20a10c 100644
--- a/packages/storage/src/execution-stores.ts
+++ b/packages/storage/src/execution-stores.ts
@@ -362,6 +362,7 @@ async function createExecutionStoresForWrite<K extends StorageRootKind, E extend
createStableSession: (request, initialBoundary) =>
run(() => sessionStore.createStableSession(request, initialBoundary)),
assignWorkHubMessage: (request) => run(() => sessionStore.assignWorkHubMessage(request)),
+ listActiveWorkHubAssignments: () => run(() => sessionStore.listActiveWorkHubAssignments()),
readWorkHubAssignment: (actionId) => run(() => sessionStore.readWorkHubAssignment(actionId)),
readWorkHubReplacement: (delegationId) =>
run(() => sessionStore.readWorkHubReplacement(delegationId)),
diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts
index c745b72675..663861ce97 100644
--- a/packages/storage/src/session-store.ts
+++ b/packages/storage/src/session-store.ts
@@ -29,6 +29,7 @@ import {
type SqliteSessionMetadataStore,
type StableSessionCreateProbe,
type VersionedSessionIdentity,
+ type ActiveWorkHubAssignment,
} from './sqlite-session-metadata-store.js';
import { isDiscardableConversationCopy } from './session-conversation-copy.js';
import {
@@ -424,6 +425,8 @@ export interface SessionAuthorityStore extends SessionStore, MessageAdmissionSto
assignWorkHubMessage(
request: WorkHubMessageAssignmentRequest,
): Promise<WorkHubMessageAssignmentResult>;
+ /** Current Host-owned delegation linkage; does not replay the transcript. */
+ listActiveWorkHubAssignments(): Promise<readonly ActiveWorkHubAssignment[]>;
readWorkHubAssignment(actionId: string): Promise<WorkHubDelegationAssignedMessage | undefined>;
readWorkHubReplacement(
delegationId: string,
@@ -688,6 +691,11 @@ class SqliteSessionStore implements SessionAuthorityStore {
return result;
}
+ async listActiveWorkHubAssignments(): Promise<readonly ActiveWorkHubAssignment[]> {
+ await this.ensureReady();
+ return this.metadata.listActiveWorkHubAssignments();
+ }
+
async readWorkHubAssignment(
actionId: string,
): Promise<WorkHubDelegationAssignedMessage | undefined> {
diff --git a/packages/storage/src/sqlite-session-metadata-schema.ts b/packages/storage/src/sqlite-session-metadata-schema.ts
index 525934c7de..b1e8ea9d66 100644
--- a/packages/storage/src/sqlite-session-metadata-schema.ts
+++ b/packages/storage/src/sqlite-session-metadata-schema.ts
@@ -19,7 +19,7 @@
import type { DatabaseSync } from 'node:sqlite';
-export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 38;
+export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 39;
export const SQLITE_SESSION_MESSAGE_CHUNK_BYTES = 64 * 1024;
export const SQLITE_SESSION_MESSAGE_CHUNK_MARKER = '{"$maka":"session-message-chunks-v1"}';
@@ -1268,6 +1268,57 @@ const MIGRATIONS: ReadonlyMap<number, string> = new Map([
);
`,
],
+ [
+ 39,
+ `
+ -- Current delegation linkage is a Host-owned projection. The transcript
+ -- remains the historical ledger, but readers no longer need to replay it
+ -- to answer which delegations are active now.
+ CREATE TABLE IF NOT EXISTS workhub_active_delegations (
+ delegation_id TEXT PRIMARY KEY,
+ action_id TEXT NOT NULL UNIQUE,
+ target_session_id TEXT NOT NULL,
+ transcript_sequence INTEGER NOT NULL CHECK (transcript_sequence >= 0),
+ assignment_json TEXT NOT NULL
+ );
+
+ INSERT OR IGNORE INTO workhub_active_delegations(
+ delegation_id, action_id, target_session_id, transcript_sequence, assignment_json
+ )
+ SELECT
+ json_extract(assignment.record_json, '$.delegationId'),
+ json_extract(assignment.record_json, '$.actionId'),
+ json_extract(assignment.record_json, '$.targetSessionId'),
+ assignment.sequence,
+ assignment.record_json
+ FROM session_messages AS assignment
+ WHERE assignment.session_id = 'maka_workhub_coordination'
+ AND json_extract(assignment.record_json, '$.type') = 'workhub_coordination'
+ AND json_extract(assignment.record_json, '$.kind') = 'delegation_assigned'
+ AND NOT EXISTS (
+ SELECT 1
+ FROM session_messages AS terminal
+ WHERE terminal.session_id = assignment.session_id
+ AND json_extract(terminal.record_json, '$.type') = 'workhub_coordination'
+ AND (
+ (
+ json_extract(terminal.record_json, '$.kind') = 'delegation_superseded'
+ AND json_extract(terminal.record_json, '$.supersededDelegationId') =
+ json_extract(assignment.record_json, '$.delegationId')
+ ) OR (
+ json_extract(terminal.record_json, '$.kind') = 'delegation_replacement_aborted'
+ AND json_extract(terminal.record_json, '$.abortedDelegationId') =
+ json_extract(assignment.record_json, '$.delegationId')
+ ) OR (
+ json_extract(terminal.record_json, '$.kind') = 'delegation_stop_resolved'
+ AND json_extract(terminal.record_json, '$.outcome') <> 'not_owned'
+ AND json_extract(terminal.record_json, '$.stopsDelegationId') =
+ json_extract(assignment.record_json, '$.delegationId')
+ )
+ )
+ );
+ `,
+ ],
]);
if (MIGRATIONS.size !== SQLITE_SESSION_METADATA_SCHEMA_VERSION) {
diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts
index c1a13de8d5..4635bfb8bc 100644
--- a/packages/storage/src/sqlite-session-metadata-store.ts
+++ b/packages/storage/src/sqlite-session-metadata-store.ts
@@ -244,6 +244,11 @@ export interface SqliteWorkHubMessageAssignmentResult {
readonly assignment: WorkHubDelegationAssignedMessage;
}
+export interface ActiveWorkHubAssignment {
+ readonly sequence: number;
+ readonly assignment: WorkHubDelegationAssignedMessage;
+}
+
export interface SessionMetadataRecord {
header: SessionHeader;
metadataVersion: number;
@@ -1606,6 +1611,7 @@ export class SqliteSessionMetadataStore {
}
const sequence = row.last_sequence + 1;
this.insertSessionMessagesSync(sessionId, sequence, encoded);
+ this.updateActiveWorkHubAssignmentsSync(sessionId, sequence, encoded);
this.updateCatalogProjectionSync(sessionId, projection, false, lockConnection);
});
}
@@ -1924,21 +1930,52 @@ export class SqliteSessionMetadataStore {
) {
throw new SessionMetadataConflictError('Invalid WorkHub transcript sequence');
}
- this.insertSessionMessagesSync(
+ const firstSequence = sequenceRow.last_sequence + 1;
+ const entries = [
+ { message: committedAssignment, json: committedAssignmentJson },
+ ...(supersession && supersessionJson
+ ? [{ message: supersession, json: supersessionJson }]
+ : []),
+ ];
+ this.insertSessionMessagesSync(WORKHUB_COORDINATION_SESSION_ID, firstSequence, entries);
+ this.updateActiveWorkHubAssignmentsSync(
WORKHUB_COORDINATION_SESSION_ID,
- sequenceRow.last_sequence + 1,
- [
- { message: committedAssignment, json: committedAssignmentJson },
- ...(supersession && supersessionJson
- ? [{ message: supersession, json: supersessionJson }]
- : []),
- ],
+ firstSequence,
+ entries,
);
this.updateCatalogProjectionSync(WORKHUB_COORDINATION_SESSION_ID, request.projection, false);
return { kind: 'assigned' as const, targetCreated, assignment: committedAssignment };
});
}
+ async listActiveWorkHubAssignments(): Promise<readonly ActiveWorkHubAssignment[]> {
+ this.assertOpen();
+ return this.readTransaction(() => {
+ const rows = this.db
+ .prepare(
+ `
+ SELECT transcript_sequence, assignment_json
+ FROM workhub_active_delegations
+ ORDER BY transcript_sequence, delegation_id
+ `,
+ )
+ .all() as unknown as Array<{
+ readonly transcript_sequence: number;
+ readonly assignment_json: string;
+ }>;
+ return rows.map((row) => {
+ const assignment = decodeStoredMessage(JSON.parse(row.assignment_json) as unknown);
+ if (
+ assignment.type !== 'workhub_coordination' ||
+ assignment.kind !== 'delegation_assigned'
+ ) {
+ throw new SessionMetadataConflictError('Invalid active WorkHub assignment');
+ }
+ return { sequence: row.transcript_sequence, assignment };
+ });
+ });
+ }
+
async readMessageAdmission(
sessionId: string,
messageId: string,
@@ -5484,6 +5521,42 @@ export class SqliteSessionMetadataStore {
}
}
+ private updateActiveWorkHubAssignmentsSync(
+ sessionId: string,
+ firstSequence: number,
+ entries: readonly { readonly message: StoredMessage; readonly json: string }[],
+ ): void {
+ if (sessionId !== WORKHUB_COORDINATION_SESSION_ID) return;
+ const insert = this.db.prepare(`
+ INSERT INTO workhub_active_delegations(
+ delegation_id, action_id, target_session_id, transcript_sequence, assignment_json
+ ) VALUES (?, ?, ?, ?, ?)
+ `);
+ const remove = this.db.prepare(
+ 'DELETE FROM workhub_active_delegations WHERE delegation_id = ?',
+ );
+ for (let index = 0; index < entries.length; index += 1) {
+ const entry = entries[index]!;
+ const message = entry.message;
+ if (message.type !== 'workhub_coordination') continue;
+ if (message.kind === 'delegation_assigned') {
+ insert.run(
+ message.delegationId,
+ message.actionId,
+ message.targetSessionId,
+ firstSequence + index,
+ entry.json,
+ );
+ } else if (message.kind === 'delegation_superseded') {
+ remove.run(message.supersededDelegationId);
+ } else if (message.kind === 'delegation_replacement_aborted') {
+ remove.run(message.abortedDelegationId);
+ } else if (message.kind === 'delegation_stop_resolved' && message.outcome !== 'not_owned') {
+ remove.run(message.stopsDelegationId);
+ }
+ }
+ }
+
private replaceSessionMessageSync(
sessionId: string,
sequence: number,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org