From 201f8844238d7a70146c13827ef1180a2204f3ae Mon Sep 17 00:00:00 2001 From: me2seeks Date: Sat, 29 Aug 2026 13:07:09 +0800 Subject: [PATCH 1/6] feat(runtime-host): separate queued successor turns Generated-by: OpenAI Codex --- docs/desktop-message-queue.md | 2 +- .../__tests__/execution-host-message.test.ts | 65 +++--- .../__tests__/execution-host-queue.test.ts | 13 +- .../src/__tests__/message-coordinator.test.ts | 207 +++++++++++++----- .../__tests__/root-turn-coordinator.test.ts | 6 +- .../src/server/message-coordinator.ts | 123 ++++++++--- 6 files changed, 294 insertions(+), 122 deletions(-) diff --git a/docs/desktop-message-queue.md b/docs/desktop-message-queue.md index 62954b30b3..2fc30d8722 100644 --- a/docs/desktop-message-queue.md +++ b/docs/desktop-message-queue.md @@ -26,7 +26,7 @@ Desktop used a rendered `streaming` prop to decide whether a composer submit sta Runtime Host already owns the durable message semantics: - `current_turn` queues steering for the next provider boundary. -- `next_turn` queues a successor turn. +- `next_turn` queues one successor turn per accepted message. - queue projections are authoritative. - queue projections carry the canonical queued message content; mutation results return only queue state. diff --git a/packages/runtime-host/src/__tests__/execution-host-message.test.ts b/packages/runtime-host/src/__tests__/execution-host-message.test.ts index 51058a93e0..f11a478e0d 100644 --- a/packages/runtime-host/src/__tests__/execution-host-message.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-message.test.ts @@ -287,38 +287,41 @@ test('steering becomes durable and ordered followups automatically start the nex } const chain = await fixture.readAdmissionChain(); - assert.equal(chain.length, 2); + assert.equal(chain.length, 3); assert.equal(chain[1]?.previousRootTurnId, firstTurnId); - assert.equal(chain[1]?.userMessageId, null); - assert.deepEqual( - chain[1]?.sourceMessages.map(({ messageId, content, placement, disposition }) => ({ - messageId, - content, - placement, - disposition, - })), - orderedFollowupSources.map((source) => ({ - ...source, - placement: 'next_turn', - disposition: 'followup', - })), + assert.equal(chain[2]?.previousRootTurnId, chain[1]?.turnId); + for (const [index, source] of orderedFollowupSources.entries()) { + const admission = chain[index + 1]; + assert.equal(admission?.userMessageId, source.messageId); + assert.deepEqual( + admission?.sourceMessages.map(({ messageId, content, placement, disposition }) => ({ + messageId, + content, + placement, + disposition, + })), + [{ ...source, placement: 'next_turn', disposition: 'followup' }], + ); + } + const followupTurnIds = chain.slice(1).map((admission) => admission.turnId); + const followupLedgers = await Promise.all( + followupTurnIds.map((turnId) => fixture.readTurn(turnId)), ); - assert.deepEqual(chain[1]?.normalizedInput, { - text: `${orderedFollowupSources[0].content.text}\n\n${orderedFollowupSources[1].content.text}`, - displayText: `${orderedFollowupSources[0].content.text}\n\n${orderedFollowupSources[1].content.displayText}`, - attachments: orderedFollowupSources[1].content.attachments, - quotes: orderedFollowupSources.flatMap((source) => source.content.quotes ?? []), - }); - const followupTurnId = chain[1]?.turnId; - assert.ok(followupTurnId); - const followupLedger = await fixture.readTurn(followupTurnId); const expectedQuotes = orderedFollowupSources.flatMap((source) => source.content.quotes ?? []); - assert.equal(followupLedger.userMessages.length, followupSources.length); assert.deepEqual( - followupLedger.userMessages.flatMap((message) => message.quotes ?? []), + followupLedgers.map((ledger) => ledger.userMessages.length), + [1, 1], + ); + assert.deepEqual( + followupLedgers.flatMap((ledger) => + ledger.userMessages.flatMap((message) => message.quotes ?? []), + ), + expectedQuotes, + ); + assert.deepEqual( + followupLedgers.flatMap((ledger) => userRuntimeContent(ledger.runtimeEvents)?.quotes ?? []), expectedQuotes, ); - assert.deepEqual(userRuntimeContent(followupLedger.runtimeEvents)?.quotes, expectedQuotes); const sessionUserMessages = await fixture.readSessionUserMessages(); for (const source of orderedFollowupSources) { assert.equal( @@ -327,13 +330,15 @@ test('steering becomes durable and ordered followups automatically start the nex ); } assert.equal( - sessionUserMessages.filter((message) => message.turnId === followupTurnId).length, + sessionUserMessages.filter((message) => followupTurnIds.includes(message.turnId)).length, orderedFollowupSources.length, ); assert.deepEqual( - sessionUserMessages - .filter((message) => message.turnId === followupTurnId) - .map((message) => message.id), + followupTurnIds.flatMap((turnId) => + sessionUserMessages + .filter((message) => message.turnId === turnId) + .map((message) => message.id), + ), orderedFollowupSources.map((source) => source.messageId), ); }); diff --git a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts index 3fa1fd998c..46e8d4fc34 100644 --- a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts @@ -218,17 +218,18 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as await tui.close(); await fixture.stopHost(host); const chain = await fixture.readAdmissionChain(); + assert.equal(chain.length, 3); assert.deepEqual( - chain.map((admission) => admission.turnId), + chain.slice(0, 2).map((admission) => admission.turnId), [firstTurnId, successor.snapshot.rootTurn.turnId], ); + assert.equal(chain[2]?.previousRootTurnId, successor.snapshot.rootTurn.turnId); assert.deepEqual( - chain[1]?.sourceMessages.map((source) => source.messageId), - [desktopFollowupId, tuiFollowupId], + chain.slice(1).map((admission) => admission.sourceMessages.map((source) => source.messageId)), + [[desktopFollowupId], [tuiFollowupId]], ); - assert.deepEqual(chain[1]?.normalizedInput, { - text: `${desktopFollowupContent.text}\n\n${tuiFollowupContent.text}`, - }); + assert.deepEqual(chain[1]?.normalizedInput, desktopFollowupContent); + assert.deepEqual(chain[2]?.normalizedInput, tuiFollowupContent); }); }); diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 240adfd9d4..77051bdafb 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -479,7 +479,7 @@ test('invalidates the canonical projection after each observable queue mutation' await fixture.coordinator.close(); }); -test('hands a mixed-Client queue to one Session successor', async () => { +test('hands each explicit follow-up to its own Session successor', async () => { const fixture = createFixture(); fixture.coordinator.reserveRootTurn(ROOT); const owner = fixture.coordinator.bindRun(ROOT); @@ -491,41 +491,55 @@ test('hands a mixed-Client queue to one Session successor', async () => { placement, }); - const steering = await fixture.coordinator.handlers['turn.message.submit']( - input('steering-from-b', 'first aggregate source', 'current_turn'), + const first = await fixture.coordinator.handlers['turn.message.submit']( + input('followup-from-b', 'first successor', 'next_turn'), operationContext('connection-b'), ); - const followup = await fixture.coordinator.handlers['turn.message.submit']( - input('followup-from-c', 'second aggregate source', 'next_turn'), + const second = await fixture.coordinator.handlers['turn.message.submit']( + input('followup-from-c', 'second successor', 'next_turn'), operationContext('connection-c'), ); - assert.equal(steering.ok, true); - assert.equal(followup.ok, true); + assert.equal(first.ok, true); + assert.equal(second.ok, true); owner.release(); - const batch = fixture.coordinator.beginTerminalTransition(ROOT); + const firstBatch = fixture.coordinator.beginTerminalTransition(ROOT); assert.deepEqual( - batch.sources.map((source) => source.messageId), - ['steering-from-b', 'followup-from-c'], + firstBatch.sources.map((source) => source.messageId), + ['followup-from-b'], ); - fixture.coordinator.commitNextRoot(batch, { + const secondRoot = { sessionId: ROOT.sessionId, turnId: 'turn-2', runId: 'run-2', + }; + fixture.coordinator.commitNextRoot(firstBatch, secondRoot); + assert.equal(fixture.liveResidencies(), 1); + const nextOwner = fixture.coordinator.bindRun(secondRoot); + nextOwner.release(); + const secondBatch = fixture.coordinator.beginTerminalTransition(secondRoot); + assert.deepEqual( + secondBatch.sources.map((source) => source.messageId), + ['followup-from-c'], + ); + fixture.coordinator.commitNextRoot(secondBatch, { + sessionId: ROOT.sessionId, + turnId: 'turn-3', + runId: 'run-3', }); assert.equal(fixture.liveResidencies(), 0); - const nextOwner = fixture.coordinator.bindRun({ + const finalOwner = fixture.coordinator.bindRun({ sessionId: ROOT.sessionId, - turnId: 'turn-2', - runId: 'run-2', + turnId: 'turn-3', + runId: 'run-3', }); - nextOwner.release(); + finalOwner.release(); fixture.coordinator.completeIdle( fixture.coordinator.beginTerminalTransition({ sessionId: ROOT.sessionId, - turnId: 'turn-2', - runId: 'run-2', + turnId: 'turn-3', + runId: 'run-3', }), ); await fixture.coordinator.close(); @@ -576,6 +590,79 @@ test('recovered followups without a connection owner still form one successor ba ); }); +test('recovery starts one explicit follow-up and keeps later messages queued', async () => { + const fixture = createFixture(); + for (const [index, messageId] of ['recovered-first', 'recovered-second'].entries()) { + const content = { text: messageId }; + await fixture.admissions.commitMessageAdmission({ + sessionId: ROOT.sessionId, + turnId: ROOT.turnId, + runId: ROOT.runId, + messageId, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'next_turn', + placement: 'next_turn', + disposition: 'followup', + admittedAt: index + 1, + }); + } + + fixture.setRootState({ kind: 'idle' }); + await fixture.coordinator.recoverPendingAfterHostRestart([ROOT.sessionId]); + + assert.deepEqual( + fixture.recoveredBatches.map((batch) => batch.sources.map((source) => source.messageId)), + [['recovered-first']], + ); + assert.deepEqual( + fixture.coordinator.projection(ROOT.sessionId).followup.map((entry) => entry.messageId), + ['recovered-second'], + ); +}); + +test('recovery folds later steering ahead of an earlier explicit follow-up', async () => { + const fixture = createFixture(); + for (const admission of [ + { + messageId: 'recovered-followup', + content: { text: 'future work' }, + submittedPlacement: 'next_turn' as const, + placement: 'next_turn' as const, + disposition: 'followup' as const, + admittedAt: 1, + }, + { + messageId: 'recovered-steering', + content: { text: 'correct the current work' }, + submittedPlacement: 'current_turn' as const, + placement: 'current_turn' as const, + disposition: 'steering' as const, + admittedAt: 2, + }, + ]) { + await fixture.admissions.commitMessageAdmission({ + sessionId: ROOT.sessionId, + turnId: ROOT.turnId, + runId: ROOT.runId, + ...admission, + submittedContentDigest: messageContentDigest(admission.content), + }); + } + + fixture.setRootState({ kind: 'idle' }); + await fixture.coordinator.recoverPendingAfterHostRestart([ROOT.sessionId]); + + assert.deepEqual( + fixture.recoveredBatches.map((batch) => batch.sources.map((source) => source.messageId)), + [['recovered-steering']], + ); + assert.deepEqual( + fixture.coordinator.projection(ROOT.sessionId).followup.map((entry) => entry.messageId), + ['recovered-followup'], + ); +}); + // The Host stopped after the Message admission committed and before the root // admission that carries the exact-Turn intent was written. async function recoverExactTurnAcrossHostStop(): Promise> { @@ -779,24 +866,24 @@ test('full snapshot preflight rejection leaves queue, replay outcome, residency, await fixture.coordinator.close(); }); -test('queue admission rejects content that cannot form a durable follow-up Turn', async () => { +test('separate follow-ups do not share one root-admission capacity budget', async () => { const fixture = createFixture(); fixture.coordinator.reserveRootTurn(ROOT); const first = await submit(fixture, 'large-followup', 'x'.repeat(40 * 1024), 'next_turn'); assert.equal(first.ok && first.result.disposition, 'followup'); - const projectionBefore = structuredClone(fixture.coordinator.projection(ROOT.sessionId)); - - const rejected = await submitContent( + const second = await submitContent( fixture, 'display-followup', { text: 'model', displayText: 'human' }, 'next_turn', ); - assert.equal(rejected.ok, false); - if (!rejected.ok) assert.equal(rejected.error.code, 'session_busy'); - assert.deepEqual(fixture.coordinator.projection(ROOT.sessionId), projectionBefore); - assert.equal(fixture.liveResidencies(), 1); + assert.equal(second.ok && second.result.disposition, 'followup'); + assert.deepEqual( + fixture.coordinator.projection(ROOT.sessionId).followup.map((entry) => entry.messageId), + ['large-followup', 'display-followup'], + ); + assert.equal(fixture.liveResidencies(), 2); const retracted = await fixture.coordinator.handlers['queue.retract']( { originHostEpoch: 'epoch-1', sessionId: ROOT.sessionId, retractId: 'cleanup-large' }, @@ -1427,11 +1514,19 @@ test('queued mutations reject a queue that is draining into the next Turn', asyn originHostEpoch: 'epoch-1', sessionId: ROOT.sessionId, reorderId: 'reorder-after-commit', - entryIds: [], + entryIds: ['id-2'], }, operationContext(), ); assert.equal(after.ok, true); + await fixture.coordinator.handlers['queue.retract']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + retractId: 'cleanup-after-commit', + }, + operationContext(), + ); fixture.coordinator.abandonRootReservation({ sessionId: ROOT.sessionId, turnId: 'turn-2', @@ -1795,10 +1890,10 @@ test('release folds unpulled steering ahead of follow-up without changing source owner.release(); const batch = fixture.coordinator.beginTerminalTransition(ROOT); assert.deepEqual(batch.content, { - text: 'first\n\nsecond\n\nthird', - displayText: 'first\n\nsecond\n\nthird', - attachments: [firstAttachment, secondAttachment, thirdAttachment], - quotes: [...firstQuotes, ...secondQuotes, ...thirdQuotes], + text: 'first\n\nsecond', + displayText: 'first\n\nsecond', + attachments: [firstAttachment, secondAttachment], + quotes: [...firstQuotes, ...secondQuotes], }); assert.deepEqual(batch.sources, [ { @@ -1829,42 +1924,42 @@ test('release folds unpulled steering ahead of follow-up without changing source placement: 'current_turn', disposition: 'steering', }, - { - messageId: 'follow-1', - content: { - text: 'third', - displayText: 'third', - attachments: [thirdAttachment], - quotes: thirdQuotes, - }, - submittedContentDigest: messageContentDigest({ - text: 'third', - displayText: 'third', - attachments: [thirdAttachment], - quotes: thirdQuotes, - }), - placement: 'next_turn', - disposition: 'followup', - }, ]); assert.equal(fixture.liveResidencies(), 3); - fixture.coordinator.commitNextRoot(batch, { + const steeringSuccessor = { sessionId: ROOT.sessionId, turnId: 'turn-2', runId: 'run-2', + }; + fixture.coordinator.commitNextRoot(batch, steeringSuccessor); + assert.equal(fixture.liveResidencies(), 1); + const next = fixture.coordinator.bindRun(steeringSuccessor); + next.release(); + const followupBatch = fixture.coordinator.beginTerminalTransition(steeringSuccessor); + assert.deepEqual(followupBatch.content, { + text: 'third', + displayText: 'third', + attachments: [thirdAttachment], + quotes: thirdQuotes, }); - assert.equal(fixture.liveResidencies(), 0); - const next = fixture.coordinator.bindRun({ + assert.deepEqual( + followupBatch.sources.map((source) => source.messageId), + ['follow-1'], + ); + const followupSuccessor = { sessionId: ROOT.sessionId, - turnId: 'turn-2', - runId: 'run-2', - }); - next.release(); + turnId: 'turn-3', + runId: 'run-3', + }; + fixture.coordinator.commitNextRoot(followupBatch, followupSuccessor); + assert.equal(fixture.liveResidencies(), 0); + const final = fixture.coordinator.bindRun(followupSuccessor); + final.release(); const empty = fixture.coordinator.beginTerminalTransition({ sessionId: ROOT.sessionId, - turnId: 'turn-2', - runId: 'run-2', + turnId: 'turn-3', + runId: 'run-3', }); fixture.coordinator.completeIdle(empty); }); diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index 3abf3d6db7..f4bd6ec3b5 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -3400,7 +3400,7 @@ test('an exact active retry preserves the Client Capability admission binding', } }); -test('mixed-Client queued follow-ups use one Session successor without connection-local tools', { +test('mixed-Client queued follow-ups use separate Session successors without connection-local tools', { timeout: 20_000, }, async () => { const clientCapabilities = new HostClientCapabilityCoordinator({ @@ -3510,6 +3510,8 @@ test('mixed-Client queued follow-ups use one Session successor without connectio await waitUntil(() => backend?.sendCount === 2); backend?.release(); + await waitUntil(() => backend?.sendCount === 3); + backend?.release(); await waitUntil( () => fixture.coordinator.readRootState(fixture.sessionId).kind === 'idle', 5_000, @@ -3519,7 +3521,7 @@ test('mixed-Client queued follow-ups use one Session successor without connectio ); assert.deepEqual( admissions.map((admission) => admission.sourceMessages.map((source) => source.messageId)), - [[], ['followup-from-provider-b', 'followup-from-provider-a']], + [[], ['followup-from-provider-b'], ['followup-from-provider-a']], ); assert.deepEqual( (await fixture.stores.sessionStore.readMessages(fixture.sessionId)) diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 8d116cb494..8cbb345bb8 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -584,7 +584,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { this.#mutated(state); } state.run = undefined; - const entries = [...state.followup]; + const entries = nextSuccessorItems(state.followup); const followup = canonicalFollowupBatch(entries); const transition: TerminalTransition = { transitionId: this.#createId(), @@ -802,15 +802,20 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { 'Message recovery authority is unavailable', ); } + const recoveryOrder = [ + ...pending.filter((entry) => entry.disposition === 'steering'), + ...pending.filter((entry) => entry.disposition !== 'steering'), + ]; + const recoveryBatch = nextSuccessorItems(recoveryOrder); const started = await this.#root.startRecoveredMessages( { sessionId, - content: aggregateMessageContents(pending.map((entry) => entry.content)), - submittedContent: aggregateMessageContents(pending.map((entry) => entry.content)), - sources: pending.map(pendingMessageSource), - ...pendingSteeringRootIdentity(pending), - ...(pending.length === 1 && pending[0]!.submittedIntent - ? { submittedIntent: pending[0]!.submittedIntent } + content: aggregateMessageContents(recoveryBatch.map((entry) => entry.content)), + submittedContent: aggregateMessageContents(recoveryBatch.map((entry) => entry.content)), + sources: recoveryBatch.map(pendingMessageSource), + ...pendingSteeringRootIdentity(recoveryBatch), + ...(recoveryBatch.length === 1 && recoveryBatch[0]!.submittedIntent + ? { submittedIntent: recoveryBatch[0]!.submittedIntent } : {}), }, admissionLease, @@ -820,14 +825,32 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { `Durable Message recovery failed: ${started.error}`, ); } + const recoveredMessageIds = new Set(recoveryBatch.map((entry) => entry.messageId)); + const remaining = pending.filter((entry) => !recoveredMessageIds.has(entry.messageId)); + if (remaining.length > 0) { + const active = await this.#root.readRootState(sessionId); + if (active.kind !== 'active') { + throw new RuntimeMessageAuthorityInvariantError( + 'Recovered successor did not become the active root Turn', + ); + } + this.#restorePendingAdmissions(sessionId, active, remaining); + } return; } + this.#restorePendingAdmissions(sessionId, rootState, pending); + } + + #restorePendingAdmissions( + sessionId: string, + rootState: RuntimeMessageRunIdentity & { readonly kind: 'active' }, + pending: readonly PendingMessageAdmission[], + ): void { if (!this.#sessions.has(sessionId)) this.#state(sessionId); const state = this.#requireState(sessionId); if (!state.reservedRoot) this.reserveRootTurn(rootState); if (!sameRun(state.reservedRoot!, rootState)) return; for (const admission of pending) { - if (admission.turnId !== rootState.turnId || admission.runId !== rootState.runId) continue; const existing = allLiveEntries(state).find( (entry) => entry.messageId === admission.messageId, ); @@ -836,8 +859,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const entry: LiveEntry = { entryId: this.#createId(), messageId: admission.messageId, - turnId: admission.turnId, - runId: admission.runId, + turnId: rootState.turnId, + runId: rootState.runId, admittedAt: admission.admittedAt, content: submittedProjectionContent(admission.content), modelContent: admission.content, @@ -1115,19 +1138,20 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { if (!interruptResultFits(candidate, rootState)) { return failure('session_busy', 'Message queue interrupt result capacity is full'); } - const prospectiveSources = [ - ...[...state.inFlight.values(), ...state.steering, ...state.followup].map( - sourceFromEntry, - ), - { - messageId: input.messageId, - content: prepared.content, - submittedContentDigest: messageContentDigest(payload.content), - placement: input.placement, - disposition, - }, - ] satisfies RootTurnSourceMessage[]; - if (!rootAdmissionPayloadFits(prospectiveSources)) { + const candidateSource = { + messageId: input.messageId, + content: prepared.content, + submittedContentDigest: messageContentDigest(payload.content), + placement: input.placement, + disposition, + } satisfies RootTurnSourceMessage; + const prospectiveSteering = [...state.inFlight.values(), ...state.steering].map( + sourceFromEntry, + ); + const prospectiveFollowup = state.followup.map(sourceFromEntry); + if (disposition === 'steering') prospectiveSteering.push(candidateSource); + else prospectiveFollowup.push(candidateSource); + if (!successorAdmissionsFit(prospectiveSteering, prospectiveFollowup)) { return failure('session_busy', 'Message queue cannot form a durable follow-up Turn'); } if ( @@ -1449,6 +1473,21 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { } return failure('not_found', 'Message queue entry does not exist'); } + const promotedSource = { + ...sourceFromEntry(entry), + placement: 'current_turn', + disposition: 'steering', + } satisfies RootTurnSourceMessage; + const prospectiveSteering = [...state.inFlight.values(), ...state.steering].map( + sourceFromEntry, + ); + prospectiveSteering.push(promotedSource); + const prospectiveFollowup = state.followup + .filter((queued) => queued !== entry) + .map(sourceFromEntry); + if (!successorAdmissionsFit(prospectiveSteering, prospectiveFollowup)) { + return failure('session_busy', 'Promoted Message exceeds steering admission capacity'); + } await this.#admissions.updateMessageAdmission({ sessionId: input.sessionId, turnId: entry.turnId, @@ -1524,16 +1563,17 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { if (!projectionFitsEveryEntryState(updatedProjection)) { return failure('session_busy', 'Message queue projection capacity is full'); } - const sources = allLiveEntries(state).map((entry) => + const updatedSource = (entry: LiveEntry): RootTurnSourceMessage => entry === queued.entry ? { ...sourceFromEntry(entry), content: modelContent, submittedContentDigest: messageContentDigest(content), } - : sourceFromEntry(entry), - ) satisfies RootTurnSourceMessage[]; - if (!rootAdmissionPayloadFits(sources)) { + : sourceFromEntry(entry); + const steeringSources = [...state.inFlight.values(), ...state.steering].map(updatedSource); + const followupSources = state.followup.map(updatedSource); + if (!successorAdmissionsFit(steeringSources, followupSources)) { return failure('session_busy', 'Message queue mutation exceeds root admission capacity'); } if (!(await this.#preflightSessionSnapshot(input.sessionId, { queue: updatedProjection }))) { @@ -2496,6 +2536,25 @@ function canonicalFollowupBatch(entries: readonly LiveEntry[]): { } } +/** + * One explicit next-turn Message owns one successor root Turn. Steering that + * missed the final provider boundary is different: those entries all targeted + * the finishing Turn, so keep their correction context together in the first + * successor rather than turning each interjection into unrelated future work. + */ +function nextSuccessorItems< + T extends { readonly disposition: 'steering' | 'followup' | 'turn_started' }, +>(entries: readonly T[]): T[] { + if (entries.length === 0) return []; + if (entries[0]!.disposition !== 'steering') return [entries[0]!]; + const steering: T[] = []; + for (const entry of entries) { + if (entry.disposition !== 'steering') break; + steering.push(entry); + } + return steering; +} + function rootAdmissionPayloadFits(sources: readonly RootTurnSourceMessage[]): boolean { try { const content = aggregateMessageContent(sources.map((source) => source.content)); @@ -2506,6 +2565,16 @@ function rootAdmissionPayloadFits(sources: readonly RootTurnSourceMessage[]): bo } } +function successorAdmissionsFit( + steering: readonly RootTurnSourceMessage[], + followup: readonly RootTurnSourceMessage[], +): boolean { + return ( + (steering.length === 0 || rootAdmissionPayloadFits(steering)) && + followup.every((source) => rootAdmissionPayloadFits([source])) + ); +} + function interruptResultFits( projection: SessionMessageQueueProjection, identity: RuntimeMessageRunIdentity, From 3346b9f911dc9c5db4f7439157dc9c827bfc4074 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Sun, 30 Aug 2026 00:32:28 +0800 Subject: [PATCH 2/6] fix(runtime-host): preserve recovered queue identity Generated-by: OpenAI Codex --- .../src/__tests__/message-coordinator.test.ts | 36 +++++++++++++++++++ .../src/server/message-coordinator.ts | 4 +-- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 77051bdafb..e5554a0c7c 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -619,6 +619,34 @@ test('recovery starts one explicit follow-up and keeps later messages queued', a fixture.coordinator.projection(ROOT.sessionId).followup.map((entry) => entry.messageId), ['recovered-second'], ); + + const projection = fixture.coordinator.projection(ROOT.sessionId); + const remainingEntryId = projection.followup[0]?.entryId; + assert.ok(remainingEntryId); + const updated = await fixture.coordinator.handlers['queue.entry.update']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + entryId: remainingEntryId, + updateId: 'update-recovered-second', + expectedQueueRevision: projection.queueRevision, + text: 'edited after recovery', + }, + operationContext(), + ); + assert.equal(updated.ok, true); + assert.deepEqual(fixture.readMessageAdmission('recovered-second'), { + sessionId: ROOT.sessionId, + turnId: ROOT.turnId, + runId: ROOT.runId, + messageId: 'recovered-second', + content: { text: 'edited after recovery' }, + submittedContentDigest: messageContentDigest({ text: 'edited after recovery' }), + submittedPlacement: 'next_turn', + placement: 'next_turn', + disposition: 'followup', + admittedAt: 2, + }); }); test('recovery folds later steering ahead of an earlier explicit follow-up', async () => { @@ -2881,6 +2909,14 @@ function memoryMessageAdmissionStore( updateMessageAdmission: async (admission) => { const existing = admissions.get(admission.messageId); if (!existing) throw new Error(`Missing admission ${admission.messageId}`); + if ( + existing.admission.turnId !== admission.turnId || + existing.admission.runId !== admission.runId || + existing.admission.submittedPlacement !== admission.submittedPlacement || + existing.admission.admittedAt !== admission.admittedAt + ) { + throw new Error(`Message admission update identity conflict: ${admission.messageId}`); + } existing.admission = admission; }, reorderMessageAdmissions: async () => undefined, diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 8cbb345bb8..1bcb8daedc 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -859,8 +859,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const entry: LiveEntry = { entryId: this.#createId(), messageId: admission.messageId, - turnId: rootState.turnId, - runId: rootState.runId, + turnId: admission.turnId, + runId: admission.runId, admittedAt: admission.admittedAt, content: submittedProjectionContent(admission.content), modelContent: admission.content, From b0f9fd8980f0bf58bdab6ac452be0f8b52b9ec67 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Sun, 30 Aug 2026 00:51:47 +0800 Subject: [PATCH 3/6] fix(runtime-host): rebind carried queue ownership Generated-by: OpenAI Codex --- .../src/__tests__/message-coordinator.test.ts | 86 +++++++++++++++++++ .../src/server/message-coordinator.ts | 38 ++++---- 2 files changed, 102 insertions(+), 22 deletions(-) diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index e5554a0c7c..1137d86ff1 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -1369,6 +1369,92 @@ test('entry promote moves a follow-up into the steering queue', async () => { assert.equal(fixture.liveResidencies(), 0); }); +test('a carried follow-up promoted in its successor requeues after nack', async () => { + const fixture = createFixture(); + fixture.coordinator.reserveRootTurn(ROOT); + const firstOwner = fixture.coordinator.bindRun(ROOT); + await submit(fixture, 'first-successor', 'first', 'next_turn'); + await submit(fixture, 'carried-followup', 'second', 'next_turn'); + firstOwner.release(); + const firstBatch = fixture.coordinator.beginTerminalTransition(ROOT); + const successor = { sessionId: ROOT.sessionId, turnId: 'turn-2', runId: 'run-2' }; + fixture.coordinator.commitNextRoot(firstBatch, successor); + fixture.setRootState({ kind: 'active', ...successor }); + + const owner = fixture.coordinator.bindRun(successor); + const entryId = fixture.coordinator.projection(ROOT.sessionId).followup[0]?.entryId; + assert.ok(entryId); + const promoted = await fixture.coordinator.handlers['queue.entry.promote']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + entryId, + promoteId: 'promote-carried-for-nack', + }, + operationContext(), + ); + assert.equal(promoted.ok, true); + const leases = owner.pull(); + assert.deepEqual( + leases.map((lease) => lease.messageId), + ['carried-followup'], + ); + owner.nack(leases.map((lease) => lease.id)); + assert.deepEqual( + fixture.coordinator.projection(ROOT.sessionId).steering.map((entry) => entry.messageId), + ['carried-followup'], + ); +}); + +test('an acked carried follow-up is not redelivered after restart', async () => { + const fixture = createFixture(); + fixture.coordinator.reserveRootTurn(ROOT); + const firstOwner = fixture.coordinator.bindRun(ROOT); + await submit(fixture, 'first-successor', 'first', 'next_turn'); + await submit(fixture, 'carried-followup', 'second', 'next_turn'); + firstOwner.release(); + const firstBatch = fixture.coordinator.beginTerminalTransition(ROOT); + const successor = { sessionId: ROOT.sessionId, turnId: 'turn-2', runId: 'run-2' }; + fixture.coordinator.commitNextRoot(firstBatch, successor); + fixture.setRootState({ kind: 'active', ...successor }); + + const owner = fixture.coordinator.bindRun(successor); + const entryId = fixture.coordinator.projection(ROOT.sessionId).followup[0]?.entryId; + assert.ok(entryId); + const promoted = await fixture.coordinator.handlers['queue.entry.promote']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + entryId, + promoteId: 'promote-carried-for-ack', + }, + operationContext(), + ); + assert.equal(promoted.ok, true); + const leases = owner.pull(); + assert.equal(leases.length, 1); + owner.ack(leases.map((lease) => lease.id)); + fixture.events.push({ + ...steeringEvent('carried-followup', 'second'), + turnId: successor.turnId, + runId: successor.runId, + }); + + await fixture.coordinator.materializeMessageHandoffsForRun({ + ...successor, + messageIds: [], + }); + assert.equal(fixture.readMessageAdmission('carried-followup'), undefined); + await fixture.admissions.markMessagesHandedOff({ + sessionId: ROOT.sessionId, + messageIds: ['first-successor'], + turnId: successor.turnId, + }); + fixture.setRootState({ kind: 'idle' }); + await fixture.coordinator.recoverPendingAfterHostRestart([ROOT.sessionId]); + assert.deepEqual(fixture.recoveredBatches, []); +}); + test('editing a promoted entry preserves its original submitted placement', async () => { const fixture = createFixture(); fixture.coordinator.reserveRootTurn(ROOT); diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 1bcb8daedc..1d198cc954 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -224,15 +224,15 @@ export type CandidateSnapshotPreflight = ( interface LiveEntry { readonly entryId: string; readonly messageId: string; - readonly turnId: string; - readonly runId: string; + readonly admissionTurnId: string; + readonly admissionRunId: string; readonly admittedAt: number; content: MessageContent; modelContent: MessageContent; submittedContentDigest: `sha256:${string}`; readonly placement: MessagePlacement; readonly disposition: 'steering' | 'followup'; - readonly generation: number; + generation: number; readonly residency: RuntimeHostResidency; state: 'queued' | 'in_flight' | 'released'; leaseId?: string; @@ -609,6 +609,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { } this.#commitTransition(state); state.generation += 1; + for (const entry of allLiveEntries(state)) entry.generation = state.generation; state.reservedRoot = { ...identity }; state.phase = 'open'; this.#mutated(state); @@ -674,11 +675,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { provenRootMessages.push(await this.#readProvenRootMessage(input, messageId)); } for (const admission of admissions) { - if ( - admission.turnId !== input.turnId || - admission.runId !== input.runId || - admission.disposition !== 'steering' - ) { + if (admission.disposition !== 'steering') { continue; } const proof = await this.#durableProof.readImmutableSteeringMessageProof( @@ -778,14 +775,11 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { sessionId, admission.messageId, ); - if ( - steering?.event.turnId === admission.turnId && - steering.event.runId === admission.runId - ) { + if (steering) { await this.materializeMessageHandoffsForRun({ sessionId, - turnId: admission.turnId, - runId: admission.runId, + turnId: steering.event.turnId, + runId: steering.event.runId, messageIds: [], }); } else { @@ -859,8 +853,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const entry: LiveEntry = { entryId: this.#createId(), messageId: admission.messageId, - turnId: admission.turnId, - runId: admission.runId, + admissionTurnId: admission.turnId, + admissionRunId: admission.runId, admittedAt: admission.admittedAt, content: submittedProjectionContent(admission.content), modelContent: admission.content, @@ -1184,8 +1178,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const entry: LiveEntry = { entryId, messageId: input.messageId, - turnId: rootState.turnId, - runId: rootState.runId, + admissionTurnId: rootState.turnId, + admissionRunId: rootState.runId, admittedAt: messageAdmission.admittedAt, content: payload.content, modelContent: prepared.content, @@ -1490,8 +1484,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { } await this.#admissions.updateMessageAdmission({ sessionId: input.sessionId, - turnId: entry.turnId, - runId: entry.runId, + turnId: entry.admissionTurnId, + runId: entry.admissionRunId, messageId: entry.messageId, content: entry.modelContent, submittedContentDigest: entry.submittedContentDigest, @@ -1591,8 +1585,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { ); await this.#admissions.updateMessageAdmission({ sessionId: input.sessionId, - turnId: queued.entry.turnId, - runId: queued.entry.runId, + turnId: queued.entry.admissionTurnId, + runId: queued.entry.admissionRunId, messageId: queued.entry.messageId, content: modelContent, submittedContentDigest: messageContentDigest(content), From 2b1181f497dbd4b8fbbf215e67ab59fb6fa62641 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Sun, 30 Aug 2026 01:00:49 +0800 Subject: [PATCH 4/6] fix(storage): prove cross-turn steering handoff Generated-by: OpenAI Codex --- .../src/__tests__/message-coordinator.test.ts | 12 +++ .../src/server/message-coordinator.ts | 12 +++ .../sqlite-session-metadata-store.test.ts | 92 +++++++++++++++++++ packages/storage/src/execution-stores.ts | 1 + .../storage/src/message-admission-store.ts | 24 +++++ .../src/sqlite-session-metadata-store.ts | 50 ++++++++-- 6 files changed, 185 insertions(+), 6 deletions(-) diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 1137d86ff1..ed13073682 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -2121,6 +2121,8 @@ test('run settlement hands off only steering admissions with immutable proof', a fixture.coordinator.reserveRootTurn(ROOT); const owner = fixture.coordinator.bindRun(ROOT); await submit(fixture, 'steer-proved', 'provider must see this', 'current_turn'); + const admittedAt = fixture.readMessageAdmission('steer-proved')?.admittedAt; + assert.ok(admittedAt); const [lease] = owner.pull(); assert.ok(lease); owner.ack([lease.id]); @@ -2140,6 +2142,16 @@ test('run settlement hands off only steering admissions with immutable proof', a sessionId: ROOT.sessionId, messageIds: ['steer-proved'], turnId: ROOT.turnId, + provenSteeringMessages: [ + { + messageId: 'steer-proved', + admissionTurnId: ROOT.turnId, + admissionRunId: ROOT.runId, + executionTurnId: ROOT.turnId, + content: { text: 'provider must see this' }, + admittedAt, + }, + ], }, ]); const batch = fixture.coordinator.beginTerminalTransition(ROOT); diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 1d198cc954..e325c80488 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -669,6 +669,9 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const provenRootMessages: Array< NonNullable[number] > = []; + const provenSteeringMessages: Array< + NonNullable[number] + > = []; const admissions = await this.#admissions.listMessageAdmissions(input.sessionId); for (const messageId of new Set(input.messageIds)) { messageIds.add(messageId); @@ -684,6 +687,14 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { ); if (proof?.event.turnId === input.turnId && proof.event.runId === input.runId) { messageIds.add(admission.messageId); + provenSteeringMessages.push({ + messageId: admission.messageId, + admissionTurnId: admission.turnId, + admissionRunId: admission.runId, + executionTurnId: proof.event.turnId, + content: admission.content, + admittedAt: admission.admittedAt, + }); } } await this.#admissions.markMessagesHandedOff({ @@ -691,6 +702,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { messageIds: [...messageIds], turnId: input.turnId, ...(provenRootMessages.length > 0 ? { provenRootMessages } : {}), + ...(provenSteeringMessages.length > 0 ? { provenSteeringMessages } : {}), }); } diff --git a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts index bd8a6aac81..8bd46b8e48 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -1212,6 +1212,98 @@ describe('SqliteSessionMetadataStore', () => { } }); + test('accepts a proof-backed steering handoff from a later execution Turn', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-message-cross-turn-steering-')); + const path = join(root, 'state.sqlite'); + const store = createSqliteSessionMetadataStore(path); + try { + await store.create(fullHeader({ id: 'session-1' })); + const content = { text: 'carried into a later successor' }; + const admission = { + sessionId: 'session-1', + turnId: 'turn-1', + runId: 'run-1', + messageId: 'message-1', + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'next_turn' as const, + placement: 'next_turn' as const, + disposition: 'followup' as const, + admittedAt: 10, + }; + await store.commitMessageAdmission(admission); + await store.updateMessageAdmission({ + ...admission, + placement: 'current_turn', + disposition: 'steering', + }); + + await assert.rejects( + store.markMessagesHandedOff({ + sessionId: 'session-1', + messageIds: ['message-1'], + turnId: 'turn-2', + provenSteeringMessages: [ + { + messageId: 'message-1', + admissionTurnId: 'wrong-turn', + admissionRunId: 'run-1', + executionTurnId: 'turn-2', + content, + admittedAt: 10, + }, + ], + }), + /Proven steering admission identity conflict/, + ); + + await store.markMessagesHandedOff({ + sessionId: 'session-1', + messageIds: ['message-1'], + turnId: 'turn-2', + provenSteeringMessages: [ + { + messageId: 'message-1', + admissionTurnId: 'turn-1', + admissionRunId: 'run-1', + executionTurnId: 'turn-2', + content, + admittedAt: 10, + }, + ], + }); + + await store.markMessagesHandedOff({ + sessionId: 'session-1', + messageIds: ['message-1'], + turnId: 'turn-2', + provenSteeringMessages: [ + { + messageId: 'message-1', + admissionTurnId: 'turn-1', + admissionRunId: 'run-1', + executionTurnId: 'turn-2', + content, + admittedAt: 10, + }, + ], + }); + + assert.equal(await store.readMessageAdmission('session-1', 'message-1'), undefined); + assert.deepEqual( + (await store.readMessages('session-1')).map((message) => ({ + id: message.id, + turnId: message.turnId, + text: message.type === 'user' ? message.text : undefined, + })), + [{ id: 'message-1', turnId: 'turn-2', text: content.text }], + ); + } finally { + store.close(); + await rm(root, { recursive: true, force: true }); + } + }); + test('retract replaces an accepted payload with a minimal identity tombstone', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-message-retract-')); const path = join(root, 'state.sqlite'); diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 58b86e1346..33bdf087e7 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -116,6 +116,7 @@ export type { MarkMessagesHandedOffInput, MessageAdmissionStore, PendingMessageAdmission, + ProvenSteeringMessageHandoff, } from './message-admission-store.js'; export { submittedTurnIntentsEqual } from './submitted-turn-intent.js'; export type { SubmittedTurnIntent } from './submitted-turn-intent.js'; diff --git a/packages/storage/src/message-admission-store.ts b/packages/storage/src/message-admission-store.ts index 8bfacbb2e8..95753023b4 100644 --- a/packages/storage/src/message-admission-store.ts +++ b/packages/storage/src/message-admission-store.ts @@ -59,11 +59,22 @@ export interface ProvenRootMessageHandoff { readonly admittedAt: number; } +/** Immutable proof that an admission was delivered as steering by a later execution owner. */ +export interface ProvenSteeringMessageHandoff { + readonly messageId: string; + readonly admissionTurnId: string; + readonly admissionRunId: string; + readonly executionTurnId: string; + readonly content: MessageContent; + readonly admittedAt: number; +} + export interface MarkMessagesHandedOffInput { readonly sessionId: string; readonly messageIds: readonly string[]; readonly turnId: string; readonly provenRootMessages?: readonly ProvenRootMessageHandoff[]; + readonly provenSteeringMessages?: readonly ProvenSteeringMessageHandoff[]; } export interface MessageAdmissionStore { @@ -134,6 +145,19 @@ export function normalizeProvenRootMessageHandoff( }); } +export function normalizeProvenSteeringMessageHandoff( + handoff: ProvenSteeringMessageHandoff, +): ProvenSteeringMessageHandoff { + assertSafeId(handoff.messageId, 'Invalid proven steering Message identity'); + assertSafeId(handoff.admissionTurnId, 'Invalid proven steering admission Turn'); + assertSafeId(handoff.admissionRunId, 'Invalid proven steering admission Run'); + assertSafeId(handoff.executionTurnId, 'Invalid proven steering execution Turn'); + if (!Number.isSafeInteger(handoff.admittedAt) || handoff.admittedAt < 0) { + throw new Error('Invalid proven steering Message timestamp'); + } + return Object.freeze({ ...handoff, content: decodeMessageContent(handoff.content) }); +} + export function samePendingMessageAdmission( left: PendingMessageAdmission, right: PendingMessageAdmission, diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index bd707e0a7c..2609643817 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -102,10 +102,12 @@ import { markPersisted } from '@maka/core/persisted-value'; import { normalizePendingMessageAdmission, normalizeProvenRootMessageHandoff, + normalizeProvenSteeringMessageHandoff, samePendingMessageAdmission, type MarkMessagesHandedOffInput, type PendingMessageAdmission, type ProvenRootMessageHandoff, + type ProvenSteeringMessageHandoff, } from './message-admission-store.js'; import { normalizeSubmittedTurnIntent } from './submitted-turn-intent.js'; import { @@ -1903,6 +1905,24 @@ export class SqliteSessionMetadataStore { } provenRootMessages.set(normalized.messageId, normalized); } + const provenSteeringMessages = new Map(); + for (const proof of input.provenSteeringMessages ?? []) { + const normalized = normalizeProvenSteeringMessageHandoff(proof); + if (!requestedMessageIds.has(normalized.messageId)) { + throw new SessionMetadataConflictError( + 'Proven steering Message identity is not present in messageIds', + ); + } + if (provenSteeringMessages.has(normalized.messageId)) { + throw new SessionMetadataConflictError( + 'Proven steering Messages contain duplicate identities', + ); + } + if (normalized.executionTurnId !== input.turnId) { + throw new SessionMetadataConflictError('Proven steering execution Turn conflict'); + } + provenSteeringMessages.set(normalized.messageId, normalized); + } this.transaction(() => { const lastSequenceRow = this.db .prepare( @@ -1929,6 +1949,7 @@ export class SqliteSessionMetadataStore { const existingSequences = new Map(); for (const messageId of unique) { const fallback = provenRootMessages.get(messageId); + const steeringProof = provenSteeringMessages.get(messageId); const admissionRow = this.db .prepare( ` @@ -1943,10 +1964,22 @@ export class SqliteSessionMetadataStore { const admission = admissionRow ? decodeMessageAdmissionRow(input.sessionId, admissionRow) : undefined; + const provenCrossTurnSteering = + admission !== undefined && + steeringProof !== undefined && + admission.disposition === 'steering' && + admission.turnId === steeringProof.admissionTurnId && + admission.runId === steeringProof.admissionRunId && + admission.admittedAt === steeringProof.admittedAt && + messageContentsEqual(admission.content, steeringProof.content); + if (admission !== undefined && steeringProof !== undefined && !provenCrossTurnSteering) { + throw new SessionMetadataConflictError('Proven steering admission identity conflict'); + } if ( admission !== undefined && admission.turnId !== input.turnId && - admission.disposition !== 'followup' + admission.disposition !== 'followup' && + !provenCrossTurnSteering ) { throw new SessionMetadataConflictError('Message admission Turn conflict'); } @@ -2015,14 +2048,12 @@ export class SqliteSessionMetadataStore { const row = rows[0]!; const recordJson = readStoredMessageRecordJson(this.db, input.sessionId, sequence, row); const message = decodeStoredMessage(JSON.parse(recordJson) as unknown); + const expectedSource = admission ?? fallback ?? steeringProof; if ( message.type !== 'user' || message.id !== messageId || - ((admission !== undefined || fallback !== undefined) && - !messageContentsEqual( - normalizeMessageContent(message), - (admission ?? fallback)!.content, - )) + (expectedSource !== undefined && + !messageContentsEqual(normalizeMessageContent(message), expectedSource.content)) ) { throw new SessionMetadataConflictError( 'Message admission transcript identity conflict', @@ -2031,6 +2062,13 @@ export class SqliteSessionMetadataStore { if (message.turnId !== input.turnId) { throw new SessionMetadataConflictError('Message admission transcript Turn conflict'); } + if ( + steeringProof !== undefined && + (message.ts !== steeringProof.admittedAt || + message.turnId !== steeringProof.executionTurnId) + ) { + throw new SessionMetadataConflictError('Proven steering transcript identity conflict'); + } existingSequences.set(messageId, sequence); } if (admission) { From e3ffa1cf76e4ef21f15c5336344fc01e8f5533a3 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Sun, 30 Aug 2026 12:01:43 +0800 Subject: [PATCH 5/6] fix(runtime-host): preserve preassigned recovery batches Generated-by: OpenAI Codex --- .../src/__tests__/message-coordinator.test.ts | 45 +++++++++++++++++++ .../src/server/message-coordinator.ts | 20 ++++++--- 2 files changed, 59 insertions(+), 6 deletions(-) diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index ed13073682..f6bc1c222c 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -691,6 +691,51 @@ test('recovery folds later steering ahead of an earlier explicit follow-up', asy ); }); +test('recovery folds promoted steering ahead of an earlier explicit follow-up', async () => { + const fixture = createFixture(); + for (const admission of [ + { + messageId: 'recovered-followup', + content: { text: 'future work' }, + turnId: ROOT.turnId, + runId: ROOT.runId, + submittedPlacement: 'next_turn' as const, + placement: 'next_turn' as const, + disposition: 'followup' as const, + admittedAt: 1, + }, + { + messageId: 'recovered-promoted', + content: { text: 'promoted correction' }, + turnId: 'earlier-turn', + runId: 'earlier-run', + submittedPlacement: 'next_turn' as const, + placement: 'current_turn' as const, + disposition: 'steering' as const, + admittedAt: 2, + }, + ]) { + await fixture.admissions.commitMessageAdmission({ + sessionId: ROOT.sessionId, + ...admission, + submittedContentDigest: messageContentDigest(admission.content), + }); + } + + fixture.setRootState({ kind: 'idle' }); + await fixture.coordinator.recoverPendingAfterHostRestart([ROOT.sessionId]); + + assert.deepEqual( + fixture.recoveredBatches.map((batch) => batch.sources.map((source) => source.messageId)), + [['recovered-promoted']], + ); + assert.equal(fixture.recoveredBatches[0]?.rootIdentity, undefined); + assert.deepEqual( + fixture.coordinator.projection(ROOT.sessionId).followup.map((entry) => entry.messageId), + ['recovered-followup'], + ); +}); + // The Host stopped after the Message admission committed and before the root // admission that carries the exact-Turn intent was written. async function recoverExactTurnAcrossHostStop(): Promise> { diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index e325c80488..ad16ff7a37 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -808,11 +808,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { 'Message recovery authority is unavailable', ); } - const recoveryOrder = [ - ...pending.filter((entry) => entry.disposition === 'steering'), - ...pending.filter((entry) => entry.disposition !== 'steering'), - ]; - const recoveryBatch = nextSuccessorItems(recoveryOrder); + const recoveryBatch = nextRecoveredSuccessorItems(pending); const started = await this.#root.startRecoveredMessages( { sessionId, @@ -2358,7 +2354,9 @@ function pendingMessageSource(admission: PendingMessageAdmission): RootTurnSourc function pendingSteeringRootIdentity( pending: readonly PendingMessageAdmission[], ): Pick { - const steering = pending.filter((entry) => entry.disposition === 'steering'); + const steering = pending.filter( + (entry) => entry.disposition === 'steering' && entry.submittedPlacement === 'current_turn', + ); const first = steering[0]; if (!first) return {}; if (steering.some((entry) => entry.turnId !== first.turnId || entry.runId !== first.runId)) { @@ -2561,6 +2559,16 @@ function nextSuccessorItems< return steering; } +function nextRecoveredSuccessorItems( + pending: readonly PendingMessageAdmission[], +): PendingMessageAdmission[] { + const steeringIntent = pending.filter( + (entry) => entry.disposition === 'steering' || entry.submittedPlacement === 'current_turn', + ); + if (steeringIntent.length > 0) return steeringIntent; + return pending.length > 0 ? [pending[0]!] : []; +} + function rootAdmissionPayloadFits(sources: readonly RootTurnSourceMessage[]): boolean { try { const content = aggregateMessageContent(sources.map((source) => source.content)); From f4015b7f4c846966f59f9fe86ddd2c083232df22 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Sun, 30 Aug 2026 12:47:15 +0800 Subject: [PATCH 6/6] fix(runtime-host): validate steering recovery proofs Generated-by: OpenAI Codex --- .../src/__tests__/message-coordinator.test.ts | 205 ++++++++++++++++-- .../src/server/message-coordinator.ts | 20 +- .../sqlite-session-metadata-store.test.ts | 6 + .../storage/src/message-admission-store.ts | 6 + .../src/sqlite-session-metadata-store.ts | 9 +- 5 files changed, 218 insertions(+), 28 deletions(-) diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index f6bc1c222c..ef33c0f6c0 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -145,33 +145,94 @@ test('consumes an active-target admission before the terminal transition can mak assert.equal(fixture.drainRequests(), 0); }); -test('idle recovery resolves differently preassigned Messages to their shared successor Turn', async () => { - const fixture = createFixture(); +test('idle recovery starts one real preassigned WorkHub root and restores the remainder', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-idle-recovery-')); + const store = createSessionStore(root); + t.after(async () => { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + }); + await store.createStableSession({ + sessionId: WORKHUB_COORDINATION_SESSION_ID, + requestFingerprint: `sha256:${'a'.repeat(64)}`, + input: { + cwd: root, + name: 'WorkHub', + role: WORKHUB_COORDINATION_SESSION_ROLE, + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'explore', + toolProfile: 'workhub-coordination-v1', + }, + }); + await store.createStableSession({ + sessionId: ROOT.sessionId, + requestFingerprint: `sha256:${'b'.repeat(64)}`, + input: { + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }, + }); + const fixture = createFixture(undefined, () => true, store); fixture.setRootState({ kind: 'idle' }); - for (const [messageId, turnId, runId] of [ - ['workhub-message-a', 'preassigned-turn-a', 'preassigned-run-a'], - ['workhub-message-b', 'preassigned-turn-b', 'preassigned-run-b'], - ] as const) { + const messageIds: string[] = []; + for (const actionId of ['preassigned-action-a', 'preassigned-action-b']) { + const suffix = createHash('sha256').update(actionId, 'utf8').digest('hex').slice(0, 48); + const messageId = `whm_${suffix}`; + const turnId = `wht_${suffix}`; + const runId = `whr_${suffix}`; + messageIds.push(messageId); const content = { text: `recover ${messageId}` }; - await fixture.admissions.commitMessageAdmission({ - sessionId: ROOT.sessionId, - turnId, - runId, - messageId, - content, - submittedContentDigest: messageContentDigest(content), - submittedPlacement: 'current_turn', - placement: 'current_turn', - disposition: 'followup', - admittedAt: 10, + await store.assignWorkHubMessage({ + assignment: { + type: 'workhub_coordination', + id: `wha_${suffix}`, + turnId: actionId, + ts: 10, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId, + actionFingerprint: `sha256:${suffix.padEnd(64, '0')}`, + coordinationTurnId: actionId, + targetSessionId: ROOT.sessionId, + targetSessionName: 'Payments', + targetTurnId: turnId, + targetMessageId: messageId, + delegationId: `whd_${suffix}`, + disposition: 'delegate_existing', + userText: content.text, + }, + admission: { + sessionId: ROOT.sessionId, + turnId, + runId, + messageId, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 10, + }, }); } await fixture.coordinator.consumePendingAdmissions([ROOT.sessionId]); + assert.deepEqual( + fixture.recoveredBatches.map((batch) => batch.sources.map((s) => s.messageId)), + [[messageIds[0]]], + ); + assert.deepEqual( + fixture.coordinator.projection(ROOT.sessionId).steering.map((entry) => entry.messageId), + [messageIds[1]], + ); const resolved = await fixture.coordinator.handlers['turn.message.execution.query']( { sessionId: ROOT.sessionId, - messageIds: ['workhub-message-a', 'workhub-message-b'], + messageIds, }, operationContext(), ); @@ -181,22 +242,118 @@ test('idle recovery resolves differently preassigned Messages to their shared su result: { resolutions: [ { - messageId: 'workhub-message-a', + messageId: messageIds[0], state: 'owned', turnId: 'recovered-turn', runId: 'durable-run', }, { - messageId: 'workhub-message-b', - state: 'owned', - turnId: 'recovered-turn', - runId: 'durable-run', + messageId: messageIds[1], + state: 'pending', }, ], }, }); }); +test('idle recovery keeps promoted steering ahead of distinct real WorkHub roots', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-promoted-recovery-')); + const store = createSessionStore(root); + t.after(async () => { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + }); + await store.createStableSession({ + sessionId: WORKHUB_COORDINATION_SESSION_ID, + requestFingerprint: `sha256:${'a'.repeat(64)}`, + input: { + cwd: root, + name: 'WorkHub', + role: WORKHUB_COORDINATION_SESSION_ROLE, + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'explore', + toolProfile: 'workhub-coordination-v1', + }, + }); + await store.createStableSession({ + sessionId: ROOT.sessionId, + requestFingerprint: `sha256:${'b'.repeat(64)}`, + input: { + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test', + model: 'test', + permissionMode: 'ask', + }, + }); + const fixture = createFixture(undefined, () => true, store); + fixture.setRootState({ kind: 'idle' }); + const promotedContent = { text: 'promoted correction' }; + await fixture.admissions.commitMessageAdmission({ + sessionId: ROOT.sessionId, + turnId: 'earlier-turn', + runId: 'earlier-run', + messageId: 'promoted-message', + content: promotedContent, + submittedContentDigest: messageContentDigest(promotedContent), + submittedPlacement: 'next_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 9, + }); + const workHubMessageIds: string[] = []; + for (const actionId of ['later-action-a', 'later-action-b']) { + const suffix = createHash('sha256').update(actionId, 'utf8').digest('hex').slice(0, 48); + const messageId = `whm_${suffix}`; + const content = { text: `recover ${messageId}` }; + workHubMessageIds.push(messageId); + await store.assignWorkHubMessage({ + assignment: { + type: 'workhub_coordination', + id: `wha_${suffix}`, + turnId: actionId, + ts: 10, + schemaVersion: 1, + kind: 'delegation_assigned', + actionId, + actionFingerprint: `sha256:${suffix.padEnd(64, '0')}`, + coordinationTurnId: actionId, + targetSessionId: ROOT.sessionId, + targetSessionName: 'Payments', + targetTurnId: `wht_${suffix}`, + targetMessageId: messageId, + delegationId: `whd_${suffix}`, + disposition: 'delegate_existing', + userText: content.text, + }, + admission: { + sessionId: ROOT.sessionId, + turnId: `wht_${suffix}`, + runId: `whr_${suffix}`, + messageId, + content, + submittedContentDigest: messageContentDigest(content), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 10, + }, + }); + } + + await fixture.coordinator.consumePendingAdmissions([ROOT.sessionId]); + + assert.deepEqual( + fixture.recoveredBatches.map((batch) => batch.sources.map((s) => s.messageId)), + [['promoted-message']], + ); + assert.deepEqual( + fixture.coordinator.projection(ROOT.sessionId).steering.map((entry) => entry.messageId), + workHubMessageIds, + ); +}); + test('idle recovery preserves the exact root identity of durable steering', async () => { const fixture = createFixture(); fixture.setRootState({ kind: 'idle' }); @@ -2193,6 +2350,8 @@ test('run settlement hands off only steering admissions with immutable proof', a admissionTurnId: ROOT.turnId, admissionRunId: ROOT.runId, executionTurnId: ROOT.turnId, + eventId: 'event-steer-proved', + eventTs: 1, content: { text: 'provider must see this' }, admittedAt, }, diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index ad16ff7a37..fee7af749b 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -692,6 +692,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { admissionTurnId: admission.turnId, admissionRunId: admission.runId, executionTurnId: proof.event.turnId, + eventId: proof.event.id, + eventTs: proof.event.ts, content: admission.content, admittedAt: admission.admittedAt, }); @@ -2565,10 +2567,26 @@ function nextRecoveredSuccessorItems( const steeringIntent = pending.filter( (entry) => entry.disposition === 'steering' || entry.submittedPlacement === 'current_turn', ); - if (steeringIntent.length > 0) return steeringIntent; + const first = steeringIntent[0]; + if (first) { + const firstHasRootIdentity = hasNativeSteeringRootIdentity(first); + const compatible: PendingMessageAdmission[] = []; + for (const entry of steeringIntent) { + if (hasNativeSteeringRootIdentity(entry) !== firstHasRootIdentity) break; + if (firstHasRootIdentity && (entry.turnId !== first.turnId || entry.runId !== first.runId)) { + break; + } + compatible.push(entry); + } + return compatible; + } return pending.length > 0 ? [pending[0]!] : []; } +function hasNativeSteeringRootIdentity(admission: PendingMessageAdmission): boolean { + return admission.disposition === 'steering' && admission.submittedPlacement === 'current_turn'; +} + function rootAdmissionPayloadFits(sources: readonly RootTurnSourceMessage[]): boolean { try { const content = aggregateMessageContent(sources.map((source) => source.content)); diff --git a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts index 8bd46b8e48..ee0f51612c 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -1249,6 +1249,8 @@ describe('SqliteSessionMetadataStore', () => { admissionTurnId: 'wrong-turn', admissionRunId: 'run-1', executionTurnId: 'turn-2', + eventId: 'event-message-1', + eventTs: 20, content, admittedAt: 10, }, @@ -1267,6 +1269,8 @@ describe('SqliteSessionMetadataStore', () => { admissionTurnId: 'turn-1', admissionRunId: 'run-1', executionTurnId: 'turn-2', + eventId: 'event-message-1', + eventTs: 20, content, admittedAt: 10, }, @@ -1283,6 +1287,8 @@ describe('SqliteSessionMetadataStore', () => { admissionTurnId: 'turn-1', admissionRunId: 'run-1', executionTurnId: 'turn-2', + eventId: 'event-message-1', + eventTs: 20, content, admittedAt: 10, }, diff --git a/packages/storage/src/message-admission-store.ts b/packages/storage/src/message-admission-store.ts index 95753023b4..920da782e0 100644 --- a/packages/storage/src/message-admission-store.ts +++ b/packages/storage/src/message-admission-store.ts @@ -65,6 +65,8 @@ export interface ProvenSteeringMessageHandoff { readonly admissionTurnId: string; readonly admissionRunId: string; readonly executionTurnId: string; + readonly eventId: string; + readonly eventTs: number; readonly content: MessageContent; readonly admittedAt: number; } @@ -152,6 +154,10 @@ export function normalizeProvenSteeringMessageHandoff( assertSafeId(handoff.admissionTurnId, 'Invalid proven steering admission Turn'); assertSafeId(handoff.admissionRunId, 'Invalid proven steering admission Run'); assertSafeId(handoff.executionTurnId, 'Invalid proven steering execution Turn'); + assertSafeId(handoff.eventId, 'Invalid proven steering RuntimeEvent identity'); + if (!Number.isSafeInteger(handoff.eventTs) || handoff.eventTs < 0) { + throw new Error('Invalid proven steering RuntimeEvent timestamp'); + } if (!Number.isSafeInteger(handoff.admittedAt) || handoff.admittedAt < 0) { throw new Error('Invalid proven steering Message timestamp'); } diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index 2609643817..aeaa008467 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -2031,9 +2031,9 @@ export class SqliteSessionMetadataStore { type: 'user', id: messageId, turnId: input.turnId, - ts: source.admittedAt, + ts: steeringProof?.eventTs ?? source.admittedAt, ...source.content, - steeringEventId: messageId, + steeringEventId: steeringProof?.eventId ?? messageId, }); const json = JSON.stringify(message); (historicalMessageIdSet.has(messageId) @@ -2064,8 +2064,9 @@ export class SqliteSessionMetadataStore { } if ( steeringProof !== undefined && - (message.ts !== steeringProof.admittedAt || - message.turnId !== steeringProof.executionTurnId) + (message.ts !== steeringProof.eventTs || + message.turnId !== steeringProof.executionTurnId || + message.steeringEventId !== steeringProof.eventId) ) { throw new SessionMetadataConflictError('Proven steering transcript identity conflict'); }