diff --git a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts index 7a20e9ca77..955f77da94 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts @@ -76,7 +76,14 @@ test('drives Desktop Session operations through a real Runtime Host connection', }, 'turn.message.submit': async (input) => { assert.equal(input.originHostEpoch, hostEpoch); - return { ok: true, result: { disposition: 'steering', queueRevision: 1 } }; + return { + ok: true, + result: { + disposition: 'steering', + queueRevision: 1, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }, + }; }, }), beginDrain() {}, @@ -107,7 +114,11 @@ test('drives Desktop Session operations through a real Runtime Host connection', content: { text: 'Continue with the new constraints.' }, placement: 'current_turn', }), - { disposition: 'steering', queueRevision: 1 }, + { + disposition: 'steering', + queueRevision: 1, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }, ); await client.close(); diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts index 78c7c0ca20..b6f2797e1d 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts @@ -361,7 +361,11 @@ test("sends canonical content and uploads owned Attachment bytes through the Hos }, submitMessage: async (input) => { starts.push(input); - return { disposition: "turn_started", turnId: "turn-1" }; + return { + disposition: "turn_started", + turnId: "turn-1", + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; }, }); const ipc = ipcHarness(); @@ -466,7 +470,11 @@ test("uploads a selected workspace file as a Host-owned Session Artifact", async }, submitMessage: async (input) => { starts.push(input); - return { disposition: "turn_started", turnId: "turn-1" }; + return { + disposition: "turn_started", + turnId: "turn-1", + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; }, }), observer: unusedObserver(), @@ -566,7 +574,11 @@ test("submits an ordinary composer message once under its stable message identit getSession: async () => session(), submitMessage: async (input) => { submits.push(input); - return { disposition: "turn_started", turnId: "host-turn" }; + return { + disposition: "turn_started", + turnId: "host-turn", + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; }, }), observer: unusedObserver(), @@ -679,6 +691,11 @@ test('submits a slash Skill message and reports the Host Skill outcome', async ( test("queues a mid-turn send as steering when the Host reports the session busy", async () => { const submits: unknown[] = []; const changes: unknown[] = []; + const skillInvocation = { + loaded: [{ id: 'review', name: 'Review' }], + failed: [{ request: 'typo', reason: 'not_found' as const }], + receipts: [], + }; const ipc = ipcHarness(); registerExecutionIpc( { @@ -686,7 +703,7 @@ test("queues a mid-turn send as steering when the Host reports the session busy" getSession: async () => session(), submitMessage: async (input) => { submits.push(input); - return { disposition: "steering", queueRevision: 1 }; + return { disposition: "steering", queueRevision: 1, skillInvocation }; }, }), observer: unusedObserver(), @@ -721,7 +738,7 @@ test("queues a mid-turn send as steering when the Host reports the session busy" turnId: "turn-1", attachments: [], inlineReferences: [], - skillInvocation: { loaded: [], failed: [], receipts: [] }, + skillInvocation, }); assert.deepEqual(changes, [ { reason: "status-change", sessionId: "session-1" }, @@ -805,7 +822,11 @@ test("retries a dispatched send with its original message identity", async () => "connection_lost", ); } - return { disposition: "steering", queueRevision: 1 }; + return { + disposition: "steering", + queueRevision: 1, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; }, }), newId: () => "id-1", @@ -884,6 +905,7 @@ test("answers a send with the Turn the Host started for it", async () => { return { disposition: "turn_started", turnId: "turn-9", + skillInvocation: { loaded: [], failed: [], receipts: [] }, }; }, }), @@ -981,7 +1003,11 @@ test("lets the Host queue a textual Skill token as steering", async () => { getSession: async () => session(), submitMessage: async (input) => { submits.push(input); - return { disposition: "steering", queueRevision: 1 }; + return { + disposition: "steering", + queueRevision: 1, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; }, }), newId: () => "id-1", @@ -1052,6 +1078,11 @@ test("reports a Host-blocked Skill send as a Skill failure", async () => { test("queues explicit Desktop follow-ups", async () => { const submits: unknown[] = []; let sequence = 0; + const skillInvocation = { + loaded: [{ id: 'writer', name: 'Writer' }], + failed: [{ request: 'missing', reason: 'not_found' as const }], + receipts: [], + }; const ipc = ipcHarness(); registerExecutionIpc( { @@ -1059,7 +1090,7 @@ test("queues explicit Desktop follow-ups", async () => { getSession: async () => session(), submitMessage: async (input) => { submits.push(input); - return { disposition: "followup", queueRevision: 4 }; + return { disposition: "followup", queueRevision: 4, skillInvocation }; }, }), observer: unusedObserver(), @@ -1109,7 +1140,7 @@ test("queues explicit Desktop follow-ups", async () => { }, ], inlineReferences: [], - skillInvocation: { loaded: [], failed: [], receipts: [] }, + skillInvocation, }, ); assert.deepEqual(submits, [ @@ -1278,7 +1309,11 @@ test("binds steer and stop to Host-owned queue and active Turn identities", asyn 'Message disposition cannot be proven in this Host Epoch', ); } - return { disposition: "steering", queueRevision: 2 }; + return { + disposition: "steering", + queueRevision: 2, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; }, interruptTurn: async (input) => { stopLifecycle.push("interrupt"); diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts index d281173ac0..dfc07e366b 100644 --- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts @@ -372,7 +372,7 @@ export function registerRuntimeHostSessionExecutionIpc( turnId: submitted.turnId, attachments, inlineReferences, - skillInvocation: submitted.skillInvocation ?? EMPTY_SKILL_INVOCATION, + skillInvocation: submitted.skillInvocation, }; } // The sending surface believed this Session idle; nudge it to refresh so @@ -385,7 +385,7 @@ export function registerRuntimeHostSessionExecutionIpc( ...(sideConversation ? { messageId } : {}), attachments, inlineReferences, - skillInvocation: EMPTY_SKILL_INVOCATION, + skillInvocation: submitted.skillInvocation, }; }, ); @@ -487,7 +487,7 @@ export function registerRuntimeHostSessionExecutionIpc( turnId: result.turnId, attachments, inlineReferences, - skillInvocation: result.skillInvocation ?? EMPTY_SKILL_INVOCATION, + skillInvocation: result.skillInvocation, }; } // The submitting surface believed this Session idle when it steered; @@ -498,7 +498,7 @@ export function registerRuntimeHostSessionExecutionIpc( disposition: result.disposition, attachments, inlineReferences, - skillInvocation: EMPTY_SKILL_INVOCATION, + skillInvocation: result.skillInvocation, }; }, ); diff --git a/docs/desktop-message-queue.md b/docs/desktop-message-queue.md index 62954b30b3..95ed9d5a5d 100644 --- a/docs/desktop-message-queue.md +++ b/docs/desktop-message-queue.md @@ -28,7 +28,9 @@ Runtime Host already owns the durable message semantics: - `current_turn` queues steering for the next provider boundary. - `next_turn` queues a successor turn. - queue projections are authoritative. -- queue projections carry the canonical queued message content; mutation results return only queue state. +- queue projections carry the canonical queued message content; queue mutation results return only + queue state, while `turn.message.submit` also returns the Skill admission outcome for every + disposition. ## Desktop Behavior diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts index a3b5239894..0622490d72 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -5049,6 +5049,44 @@ describe('Maka Pi TUI runner', () => { } }); + test('shows partial Skill feedback when the Host queues the Message', async () => { + const terminal = new FakeTerminal(); + const driver = new HostSkillDriver( + { + loaded: [{ id: 'alpha', name: 'Alpha' }], + failed: [{ request: 'typo', reason: 'not_found' }], + receipts: [], + }, + 'steering', + ); + const run = runMakaPiTui({ + title: 'Maka', + driver, + cwd: '/repo', + model: 'claude-sonnet-4-5', + connectionSlug: 'claude-subscription', + permissionMode: 'ask', + terminal, + listSkills: async () => [], + }); + + terminal.input('/skill:alpha /skill:typo 帮我整理'); + terminal.input('\r'); + await waitFor(() => driver.prompts.length === 1); + await waitFor(() => { + const output = plainTerminalOutput(terminal.output()); + return output.includes('已加载技能：Alpha') && output.includes('/skill:typo（未找到）'); + }); + + exitMaka(terminal); + await Promise.race([ + run, + delay(CLOSE_BUDGET_MS).then(() => { + throw new Error('TUI did not close during test cleanup'); + }), + ]); + }); + test('does not create a turn when every skill token fails to resolve', async () => { { const terminal = new FakeTerminal(); @@ -7561,7 +7599,10 @@ class SlashCommandDriver extends FakeSessionDriver { } class HostSkillDriver extends SlashCommandDriver { - constructor(private readonly skillInvocation: SkillInvocationResult) { + constructor( + private readonly skillInvocation: SkillInvocationResult, + private readonly admittedDisposition: 'turn_started' | 'steering' = 'turn_started', + ) { super(); } @@ -7582,6 +7623,9 @@ class HostSkillDriver extends SlashCommandDriver { // Admitted: the receipt for what was resolved rides the answer, which is // the client's only sight of it. const admitted = await super.submitMessage(text, options); + if (this.admittedDisposition === 'steering') { + return { disposition: 'steering', queueRevision: 1, skillInvocation: this.skillInvocation }; + } return admitted?.disposition === 'turn_started' ? { ...admitted, skillInvocation: this.skillInvocation } : admitted; @@ -8318,7 +8362,11 @@ async function admitMessageAsTurn( summary: { ...fakeSessionSummary(turn.sessionId), ...driver.hostSummary }, }), ); - return { disposition: 'turn_started', turnId: turn.turnId }; + return { + disposition: 'turn_started', + turnId: turn.turnId, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; } function fakeSessionSummary( diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index 51f2ff0630..d15eeabcba 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -983,7 +983,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { // was dropped, and the submit answer is the only place it appears: the // Turn arrives through the started-Turn subscription, which carries // Session state rather than this Message's admission. - if (result?.disposition === 'turn_started' && result.skillInvocation) { + if (result) { const { loaded, failed } = result.skillInvocation; if (loaded.length > 0 || failed.length > 0) showSkillInvocation(result.skillInvocation); } diff --git a/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts b/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts index ab9fe82cba..f530cc72ba 100644 --- a/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts +++ b/packages/runtime-host/src/__tests__/canonical-session-projection.test.ts @@ -597,7 +597,11 @@ function createMessages( startFromMessage: async () => { throw new Error('unexpected root start'); }, - prepareMessage: async (input) => ({ kind: 'ready', content: input.content }), + prepareMessage: async (input) => ({ + kind: 'ready', + content: input.content, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }), claimStop: async () => { throw new Error('unexpected root stop'); }, diff --git a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts index 986e4328d4..8bb5a566e3 100644 --- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts +++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts @@ -701,6 +701,7 @@ export class ExecutionFixture { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt, }); const result = await stores.agentRunStore.admitRootTurn({ diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 6ff54ecd0d..d166b1e2c1 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -21,6 +21,7 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; import { messageContentDigest, type MessageContent } from '@maka/core/events'; import type { RuntimeEvent } from '@maka/core/runtime-event'; +import type { SkillInvocationResult } from '@maka/core/skill-invocation'; import type { MessageAdmissionStore, PendingMessageAdmission, @@ -28,6 +29,7 @@ import type { } from '@maka/storage/execution-stores'; import { MESSAGE_OPERATION_RESULT_MAX_BYTES, + MESSAGE_QUEUE_MAX_ENTRIES, MESSAGE_QUEUE_PROJECTION_MAX_BYTES, decodeSessionMessageQueueProjection, type SessionMessageQueueProjection, @@ -43,6 +45,7 @@ import { import { SessionAdmissionGate } from '../server/session-admission-gate.js'; const ROOT = { sessionId: 'session-1', turnId: 'turn-1', runId: 'run-1' } as const; +const EMPTY_SKILL_INVOCATION = { loaded: [], failed: [], receipts: [] } as const; test('idle submit starts exactly one root Turn and retry identity is connection-independent', async () => { const fixture = createFixture(); @@ -66,7 +69,11 @@ test('idle submit starts exactly one root Turn and retry identity is connection- assert.deepEqual(first, { ok: true, - result: { disposition: 'turn_started', turnId: 'idle-turn' }, + result: { + disposition: 'turn_started', + turnId: 'idle-turn', + skillInvocation: EMPTY_SKILL_INVOCATION, + }, }); assert.deepEqual(retry, first); assert.equal(fixture.startCalls(), 1); @@ -172,6 +179,15 @@ test('submit re-runs admission when the queue revision moves during preflight', operationContext(), ); assert.equal(steering.ok, true); + let preparationCalls = 0; + fixture.setMessagePreparation(async (message) => { + preparationCalls += 1; + return { + kind: 'ready', + content: message.content, + skillInvocation: EMPTY_SKILL_INVOCATION, + }; + }); const followup = await fixture.coordinator.handlers['turn.message.submit']( input('followup-1', 'queued task', 'next_turn'), @@ -183,24 +199,45 @@ test('submit re-runs admission when the queue revision moves during preflight', preflightCalls >= 2, `expected admission retry, preflight ran ${preflightCalls} time(s)`, ); + assert.equal(preparationCalls, 1, 'one admission must prepare Skills only once'); owner.release(); }); test('persists prepared Skill content while projecting the submitted text', async () => { const fixture = createFixture(); + const skillInvocation = { + loaded: [{ id: 'writer', name: 'Writer' }], + failed: [{ request: 'typo', reason: 'not_found' as const }], + receipts: [], + }; fixture.setMessagePreparation(async (input) => ({ kind: 'ready', content: { text: `Prepared\n\n${input.content.text}`, displayText: input.content.text, }, + skillInvocation, })); fixture.coordinator.reserveRootTurn(ROOT); const owner = fixture.coordinator.bindRun(ROOT); - assert.equal( - (await submit(fixture, 'skill-steering', '/skill:writer steer', 'current_turn')).ok, - true, + const steeringResult = await submit( + fixture, + 'skill-steering', + '/skill:writer steer', + 'current_turn', + ); + assert.deepEqual(steeringResult, { + ok: true, + result: { disposition: 'steering', queueRevision: 1, skillInvocation }, + }); + assert.deepEqual( + fixture.readMessageAdmission('skill-steering')?.skillInvocation, + skillInvocation, + ); + assert.deepEqual( + await submit(fixture, 'skill-steering', '/skill:writer steer', 'current_turn'), + steeringResult, ); assert.deepEqual(fixture.coordinator.projection(ROOT.sessionId).steering[0]?.content, { text: '/skill:writer steer', @@ -216,10 +253,16 @@ test('persists prepared Skill content while projecting the submitted text', asyn ); if (steering) owner.ack([steering.id]); - assert.equal( - (await submit(fixture, 'skill-followup', '/skill:writer follow', 'next_turn')).ok, - true, + const followupResult = await submit( + fixture, + 'skill-followup', + '/skill:writer follow', + 'next_turn', ); + assert.deepEqual(followupResult, { + ok: true, + result: { disposition: 'followup', queueRevision: 4, skillInvocation }, + }); owner.release(); const batch = fixture.coordinator.beginTerminalTransition(ROOT); assert.deepEqual(batch.content, { @@ -230,11 +273,169 @@ test('persists prepared Skill content while projecting the submitted text', asyn text: 'Prepared\n\n/skill:writer follow', displayText: '/skill:writer follow', }); + assert.deepEqual(batch.sources[0]?.skillInvocation, skillInvocation); const nextRoot = { sessionId: ROOT.sessionId, turnId: 'turn-2', runId: 'run-2' }; fixture.coordinator.commitNextRoot(batch, nextRoot); fixture.coordinator.abandonRootReservation(nextRoot); }); +test('blocks a queued Message when every Skill fails without mutating the queue', async () => { + const fixture = createFixture(); + const skillInvocation = { + loaded: [], + failed: [{ request: 'missing', reason: 'not_found' as const }], + receipts: [], + }; + fixture.setMessagePreparation(async () => ({ + kind: 'rejected', + error: 'Explicit Skill invocation could not be resolved', + skillInvocation, + })); + fixture.coordinator.reserveRootTurn(ROOT); + + assert.deepEqual( + await submit(fixture, 'skill-blocked', '/skill:missing inspect this', 'current_turn'), + { + ok: true, + result: { disposition: 'blocked', skillInvocation }, + }, + ); + assert.deepEqual(fixture.coordinator.projection(ROOT.sessionId), { + hostEpoch: 'epoch-1', + queueRevision: 0, + steering: [], + followup: [], + }); + assert.equal(fixture.readMessageAdmission('skill-blocked'), undefined); +}); + +test('an all-failed Skill invocation stays blocked when the queue is full', async () => { + const fixture = createFixture(); + fixture.coordinator.reserveRootTurn(ROOT); + for (let index = 0; index < MESSAGE_QUEUE_MAX_ENTRIES; index += 1) { + const admitted = await submit(fixture, `queued-${index}`, 'x', 'next_turn'); + assert.equal(admitted.ok, true, JSON.stringify(admitted)); + } + const skillInvocation = { + loaded: [], + failed: [{ request: 'missing', reason: 'not_found' as const }], + receipts: [], + }; + fixture.setMessagePreparation(async () => ({ + kind: 'rejected', + error: 'Explicit Skill invocation could not be resolved', + skillInvocation, + })); + + assert.deepEqual(await submit(fixture, 'blocked-at-capacity', '/skill:missing', 'current_turn'), { + ok: true, + result: { disposition: 'blocked', skillInvocation }, + }); + assert.equal( + fixture.coordinator.projection(ROOT.sessionId).followup.length, + MESSAGE_QUEUE_MAX_ENTRIES, + ); + + await fixture.coordinator.handlers['queue.retract']( + { originHostEpoch: 'epoch-1', sessionId: ROOT.sessionId, retractId: 'cleanup-full-queue' }, + operationContext(), + ); + fixture.coordinator.abandonRootReservation(ROOT); + await fixture.coordinator.close(); +}); + +test('queue admission budgets per-source Skill outcomes into the durable root record', async () => { + const fixture = createFixture(); + fixture.coordinator.reserveRootTurn(ROOT); + const skillInvocation = largeSkillInvocation(); + fixture.setMessagePreparation(async (message) => ({ + kind: 'ready', + content: message.content, + skillInvocation, + })); + + let admittedCount = 0; + let rejectedMessageId = ''; + for (let index = 0; index < MESSAGE_QUEUE_MAX_ENTRIES; index += 1) { + const messageId = `large-outcome-${index}`; + const outcome = await submit(fixture, messageId, 'x', 'next_turn'); + if (!outcome.ok) { + assert.equal(outcome.error.code, 'session_busy'); + rejectedMessageId = messageId; + break; + } + admittedCount += 1; + } + + assert.ok(admittedCount > 0 && admittedCount < MESSAGE_QUEUE_MAX_ENTRIES); + assert.equal(fixture.coordinator.projection(ROOT.sessionId).followup.length, admittedCount); + assert.equal(fixture.readMessageAdmission(rejectedMessageId), undefined); + + await fixture.coordinator.handlers['queue.retract']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + retractId: 'cleanup-large-outcome-queue', + }, + operationContext(), + ); + fixture.coordinator.abandonRootReservation(ROOT); + await fixture.coordinator.close(); +}); + +test('queue update budgets its new Skill outcome into the durable root record', async () => { + const fixture = createFixture(); + fixture.coordinator.reserveRootTurn(ROOT); + assert.equal((await submit(fixture, 'update-target', 'small', 'next_turn')).ok, true); + const skillInvocation = largeSkillInvocation(); + fixture.setMessagePreparation(async (message) => ({ + kind: 'ready', + content: message.content, + skillInvocation, + })); + for (let index = 0; index < MESSAGE_QUEUE_MAX_ENTRIES; index += 1) { + const outcome = await submit(fixture, `large-before-update-${index}`, 'x', 'next_turn'); + if (!outcome.ok) { + assert.equal(outcome.error.code, 'session_busy'); + break; + } + } + const projection = fixture.coordinator.projection(ROOT.sessionId); + const target = projection.followup[0]; + assert.ok(target); + + const updated = await fixture.coordinator.handlers['queue.entry.update']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + entryId: target.entryId, + updateId: 'large-outcome-update', + expectedQueueRevision: projection.queueRevision, + text: 'edited', + }, + operationContext(), + ); + + assert.equal(updated.ok, false); + if (!updated.ok) assert.equal(updated.error.code, 'session_busy'); + assert.deepEqual(fixture.readMessageAdmission('update-target')?.content, { text: 'small' }); + assert.deepEqual( + fixture.readMessageAdmission('update-target')?.skillInvocation, + EMPTY_SKILL_INVOCATION, + ); + + await fixture.coordinator.handlers['queue.retract']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + retractId: 'cleanup-large-outcome-update', + }, + operationContext(), + ); + fixture.coordinator.abandonRootReservation(ROOT); + await fixture.coordinator.close(); +}); + test('invalidates the canonical projection after each observable queue mutation', async () => { const changedSessions: string[] = []; const fixture = createFixture((sessionId) => changedSessions.push(sessionId)); @@ -321,6 +522,7 @@ test('recovered followups without a connection owner still form one successor ba submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 1, }); @@ -371,6 +573,7 @@ async function recoverExactTurnAcrossHostStop(): Promise { + const fixture = createFixture(); + const skillInvocation = { + loaded: [{ id: 'writer', name: 'Writer' }], + failed: [{ request: 'typo', reason: 'not_found' as const }], + receipts: [], + }; + await fixture.admissions.commitMessageAdmission({ + sessionId: ROOT.sessionId, + turnId: ROOT.turnId, + runId: ROOT.runId, + messageId: 'recovered-skill', + content: { + text: 'Writer', + displayText: '/skill:writer /skill:typo draft', + }, + submittedContentDigest: messageContentDigest({ + text: '/skill:writer /skill:typo draft', + }), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + skillInvocation, + admittedAt: 1, + }); + await fixture.coordinator.recoverPendingAfterHostRestart([ROOT.sessionId]); + fixture.setMessagePreparation(async () => { + throw new Error('recovered retries must not prepare Skills again'); + }); + + const retried = await submit( + fixture, + 'recovered-skill', + '/skill:writer /skill:typo draft', + 'current_turn', + ); + + assert.deepEqual(retried, { + ok: true, + result: { disposition: 'steering', queueRevision: 1, skillInvocation }, + }); + assert.equal(fixture.coordinator.projection(ROOT.sessionId).steering.length, 1); +}); + +test('an idle retry reuses the Skill outcome from its pending admission', async () => { + const fixture = createFixture(); + fixture.setRootState({ kind: 'idle' }); + const skillInvocation = { + loaded: [{ id: 'writer', name: 'Writer' }], + failed: [{ request: 'typo', reason: 'not_found' as const }], + receipts: [], + }; + await fixture.admissions.commitMessageAdmission({ + sessionId: ROOT.sessionId, + turnId: 'pending-turn', + runId: 'pending-run', + messageId: 'pending-skill', + content: { + text: 'Writer', + displayText: '/skill:writer /skill:typo draft', + }, + submittedContentDigest: messageContentDigest({ + text: '/skill:writer /skill:typo draft', + }), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + skillInvocation, + admittedAt: 1, + }); + + assert.deepEqual( + await submit(fixture, 'pending-skill', '/skill:writer /skill:typo draft', 'current_turn'), + { + ok: true, + result: { disposition: 'turn_started', turnId: 'idle-turn', skillInvocation }, + }, + ); + assert.deepEqual( + fixture.receipts.get('pending-skill')?.sourceMessage.skillInvocation, + skillInvocation, + ); +}); + test('binds the exact reserved Run after a pre-bind stop fence', async () => { const fixture = createFixture(); fixture.coordinator.reserveRootTurn(ROOT); @@ -767,7 +1056,11 @@ test('entry update preserves queue identity, order, and placement and replays it let preparedUpdateContent: MessageContent | undefined; fixture.setMessagePreparation(async (input) => { preparedUpdateContent = input.content; - return { kind: 'ready', content: input.content }; + return { + kind: 'ready', + content: input.content, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; }); const updated = await fixture.coordinator.handlers['queue.entry.update']( @@ -1218,7 +1511,11 @@ test('concurrent and completed submit retries share one Host-Epoch outcome', asy const outcome = await submitted; assert.deepEqual(outcome, { ok: true, - result: { disposition: 'steering', queueRevision: 1 }, + result: { + disposition: 'steering', + queueRevision: 1, + skillInvocation: EMPTY_SKILL_INVOCATION, + }, }); assert.deepEqual(await submit(fixture, 'delayed-submit', 'steer now', 'current_turn'), outcome); assert.deepEqual(fixture.coordinator.projection(ROOT.sessionId), { @@ -1579,6 +1876,7 @@ test('release folds unpulled steering ahead of follow-up without changing source attachments: [firstAttachment], quotes: firstQuotes, }), + skillInvocation: EMPTY_SKILL_INVOCATION, placement: 'current_turn', disposition: 'steering', }, @@ -1590,6 +1888,7 @@ test('release folds unpulled steering ahead of follow-up without changing source attachments: [secondAttachment], quotes: secondQuotes, }), + skillInvocation: EMPTY_SKILL_INVOCATION, placement: 'current_turn', disposition: 'steering', }, @@ -1607,6 +1906,7 @@ test('release folds unpulled steering ahead of follow-up without changing source attachments: [thirdAttachment], quotes: thirdQuotes, }), + skillInvocation: EMPTY_SKILL_INVOCATION, placement: 'next_turn', disposition: 'followup', }, @@ -1648,6 +1948,7 @@ test('terminal transition atomically folds messages submitted after run release' messageId: 'late-steer', content: { text: 'next intent' }, submittedContentDigest: messageContentDigest({ text: 'next intent' }), + skillInvocation: EMPTY_SKILL_INVOCATION, placement: 'current_turn', disposition: 'steering', }, @@ -1708,6 +2009,7 @@ test('a failed terminal root leaves no handed-off payload for restart recovery', submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 1, }); fixture.events.push( @@ -1925,9 +2227,34 @@ test('old-Epoch durable proof ignores structured content key order', async () => attachments: [attachment('ordered-content', 'proof.png')], inlineReferences: [{ kind: 'skill', value: '/skill:vision', label: 'Vision', start: 0 }], }; + const skillInvocation = { + loaded: [{ id: 'vision', name: 'Vision' }], + failed: [], + receipts: [ + { + invocation: 'explicit' as const, + request: 'vision', + success: true as const, + ref: '/skill:vision', + id: 'vision', + name: 'Vision', + scope: 'project' as const, + source: 'maka' as const, + truncated: false, + }, + ], + }; fixture.receipts.set( messageId, - sourceReceipt(messageId, content, 'next_turn', 'turn_started', 'durable-turn', content), + sourceReceipt( + messageId, + content, + 'next_turn', + 'turn_started', + 'durable-turn', + content, + skillInvocation, + ), ); const reordered: MessageContent = { @@ -1947,7 +2274,7 @@ test('old-Epoch durable proof ignores structured content key order', async () => assert.equal(messageContentDigest(reordered), messageContentDigest(content)); assert.deepEqual(await submitContent(fixture, messageId, reordered, 'next_turn', 'old-epoch'), { ok: true, - result: { disposition: 'turn_started', turnId: 'durable-turn' }, + result: { disposition: 'turn_started', turnId: 'durable-turn', skillInvocation }, }); }); @@ -2259,6 +2586,7 @@ function createFixture( let prepareMessage: NonNullable = async (input) => ({ kind: 'ready', content: input.content, + skillInvocation: { loaded: [], failed: [], receipts: [] }, }); let rootState: HostMessageRootState = { kind: 'active', ...ROOT }; let rootStateDelay: @@ -2307,6 +2635,7 @@ function createFixture( startFromMessage: async (input) => { startCalls += 1; const turnId = 'idle-turn'; + const skillInvocation = input.preparedSkillInvocation ?? EMPTY_SKILL_INVOCATION; // Store the source message the coordinator actually produced. Rebuilding // one from parts drops whatever the coordinator recorded about the // submit, which is the very thing a retry is compared against. @@ -2320,14 +2649,25 @@ function createFixture( receipts.set(input.sourceMessage.messageId, { admission: { ...receipt.admission, - sourceMessages: [input.sourceMessage], + skillInvocation, + sourceMessages: [ + { + ...input.sourceMessage, + content: input.content, + skillInvocation, + }, + ], ...(input.turnOrchestration ? { turnOrchestration: input.turnOrchestration } : {}), }, - sourceMessage: input.sourceMessage, + sourceMessage: { + ...input.sourceMessage, + content: input.content, + skillInvocation, + }, }); rootState = { kind: 'active', sessionId: input.sessionId, turnId, runId: 'idle-run' }; coordinator.reserveRootTurn(rootState); - return { turnId }; + return { turnId, skillInvocation }; }, startRecoveredMessages: async (input) => { recoveredBatches.push(input); @@ -2512,6 +2852,7 @@ function sourceReceipt( disposition: 'steering' | 'followup' | 'turn_started', turnId = 'durable-turn', submittedContent?: MessageContent, + skillInvocation?: SkillInvocationResult, ): RootTurnSourceMessageReceipt { const normalizedContent = typeof content === 'string' ? { text: content } : content; const sourceMessage = { @@ -2534,6 +2875,7 @@ function sourceReceipt( }, previousRootTurnId: ROOT.turnId, normalizedInput: normalizedContent, + ...(skillInvocation ? { skillInvocation } : {}), sourceMessages: [sourceMessage], admittedAt: 1, }, @@ -2565,6 +2907,28 @@ function steeringEvent( }; } +function largeSkillInvocation() { + const loaded = Array.from({ length: 40 }, (_, index) => ({ + id: `skill-${index}-${'i'.repeat(60)}`, + name: `Skill ${index} ${'n'.repeat(120)}`, + })); + return { + loaded, + failed: [], + receipts: loaded.map((skill, index) => ({ + invocation: 'explicit' as const, + request: `request-${index}-${'q'.repeat(280)}`, + success: true as const, + ref: `project:maka:${index}:${'r'.repeat(280)}`, + id: skill.id, + name: skill.name, + scope: 'project' as const, + source: 'maka' as const, + truncated: false, + })), + }; +} + function attachment(id: string, name: string) { return { kind: 'image' as const, diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 32d0daa7fa..79a75b025c 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -132,6 +132,10 @@ describe('Runtime Host bootstrap protocol', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 22); }); + test('publishes a new compatibility epoch for mandatory submit Skill outcomes', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 56); + }); + test('rejects the legacy connection update result in the current compatibility epoch', () => { assert.throws( () => @@ -1491,10 +1495,19 @@ describe('Runtime Host bootstrap protocol', () => { }); test('decodes exact submit dispositions and bounded retract and interrupt results', () => { + const skillInvocation = { loaded: [], failed: [], receipts: [] }; for (const result of [ - { disposition: 'steering', queueRevision: 2 }, - { disposition: 'followup', queueRevision: 3 }, - { disposition: 'turn_started', turnId: 'turn-2' }, + { disposition: 'steering', queueRevision: 2, skillInvocation }, + { disposition: 'followup', queueRevision: 3, skillInvocation }, + { disposition: 'turn_started', turnId: 'turn-2', skillInvocation }, + { + disposition: 'blocked', + skillInvocation: { + loaded: [], + failed: [{ request: 'missing', reason: 'not_found' }], + receipts: [], + }, + }, ]) { assert.doesNotThrow(() => decodeHostFrame({ @@ -1505,16 +1518,54 @@ describe('Runtime Host bootstrap protocol', () => { }), ); } + for (const result of [ + { disposition: 'steering', queueRevision: 2 }, + { disposition: 'followup', queueRevision: 3 }, + { disposition: 'turn_started', turnId: 'turn-2' }, + { disposition: 'blocked' }, + ]) { + assert.throws( + () => + decodeHostFrame({ + requestId: 'submit-response', + operation: 'turn.message.submit', + ok: true, + result, + }), + isInvalidFrame, + ); + } assert.throws( () => decodeHostFrame({ requestId: 'submit-response', operation: 'turn.message.submit', ok: true, - result: { disposition: 'turn_started', turnId: 'turn-2', queueRevision: 4 }, + result: { + disposition: 'turn_started', + turnId: 'turn-2', + queueRevision: 4, + skillInvocation, + }, }), isInvalidFrame, ); + for (const skillInvocation of [ + { loaded: 'invalid', failed: [], receipts: [] }, + { loaded: [{ id: 'writer', name: 'Writer' }], failed: [], receipts: [] }, + { loaded: [], failed: [], receipts: [] }, + ]) { + assert.throws( + () => + decodeHostFrame({ + requestId: 'submit-response', + operation: 'turn.message.submit', + ok: true, + result: { disposition: 'blocked', skillInvocation }, + }), + isInvalidFrame, + ); + } for (const [operation, requestId] of [ ['queue.entry.retract', 'entry-retract-response'], ['queue.entry.promote', 'entry-promote-response'], diff --git a/packages/runtime-host/src/__tests__/root-admission-owner.test.ts b/packages/runtime-host/src/__tests__/root-admission-owner.test.ts index 2502955ae0..9dc466a290 100644 --- a/packages/runtime-host/src/__tests__/root-admission-owner.test.ts +++ b/packages/runtime-host/src/__tests__/root-admission-owner.test.ts @@ -182,6 +182,30 @@ test('fails closed when a known durable admission identity drifts', async () => ...first.admission.sourceMessages.slice(1), ], }, + { + ...first.admission, + sourceMessages: [ + { + ...firstSource, + submittedIntent: { skillIds: ['writer'] }, + }, + ...first.admission.sourceMessages.slice(1), + ], + }, + { + ...first.admission, + sourceMessages: [ + { + ...firstSource, + skillInvocation: { + loaded: [{ id: 'writer', name: 'Writer' }], + failed: [], + receipts: [], + }, + }, + ...first.admission.sourceMessages.slice(1), + ], + }, ]; for (const drifted of sourceDrifts) { assert.throws(() => owner.assertKnownAdmission(drifted), /identity changed/); 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 5e041ebecd..9d8746d90b 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -830,6 +830,88 @@ test('turn.start resolves explicit Skills once before durable admission and repl } }); +test('queued Message preparation preserves partial and blocked Skill outcomes', async () => { + let blocked = false; + const readySkillInvocation = { + loaded: [{ id: 'writer', name: 'Writer' }], + failed: [{ request: 'typo', reason: 'not_found' as const }], + receipts: [ + { + invocation: 'explicit' as const, + request: 'writer', + success: true as const, + ref: 'project:maka:writer', + id: 'writer', + name: 'Writer', + scope: 'project' as const, + source: 'maka' as const, + truncated: false, + }, + { + invocation: 'explicit' as const, + request: 'typo', + success: false as const, + reason: 'not_found' as const, + }, + ], + }; + const blockedSkillInvocation = { + loaded: [], + failed: [{ request: 'missing', reason: 'not_found' as const }], + receipts: [], + }; + const fixture = await createFailureFixture({ + registerBackend: (backends) => + backends.register('ai-sdk', (context) => new FakeBackend(context)), + prepareSkillInvocation: async () => + blocked + ? { disposition: 'blocked', skillInvocation: blockedSkillInvocation } + : { + disposition: 'ready', + sendText: 'Write clearly.\n\nDraft this.', + skillInvocation: readySkillInvocation, + }, + }); + try { + assert.deepEqual( + await fixture.coordinator.prepareMessage({ + sessionId: fixture.sessionId, + turnId: 'turn-running', + content: { text: '/skill:writer /skill:typo Draft this.' }, + placement: 'current_turn', + }), + { + kind: 'ready', + content: { + text: 'Write clearly.\n\nDraft this.', + displayText: '/skill:writer /skill:typo Draft this.', + inlineReferences: [{ kind: 'skill', value: '/skill:writer', label: 'Writer', start: 0 }], + }, + skillInvocation: readySkillInvocation, + }, + ); + + blocked = true; + assert.deepEqual( + await fixture.coordinator.prepareMessage({ + sessionId: fixture.sessionId, + turnId: 'turn-running', + content: { text: '/skill:missing Draft this.' }, + placement: 'current_turn', + }), + { + kind: 'rejected', + error: 'Explicit Skill invocation could not be resolved', + skillInvocation: blockedSkillInvocation, + }, + ); + } finally { + await fixture.coordinator.close(); + await fixture.messages.close(); + await fixture.dispose(); + } +}); + test('turn.start durably replays an all-failed invocation without creating a Turn', async () => { let preparationCount = 0; const skillInvocation = { diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 45adfa103a..5974dbaab4 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -93,7 +93,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 56 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 57 as const; +// 57: Every `turn.message.submit` disposition carries the exact Skill +// invocation outcome. Older strict peers either omit or reject the new field. // 56: Failed Turn snapshots preserve the structured context-budget exhaustion // detail. Epoch-55 peers reject the optional field on the closed snapshot shape. // 55: Local owners can atomically revoke every credential for one access diff --git a/packages/runtime-host/src/protocol/message.ts b/packages/runtime-host/src/protocol/message.ts index ea4df35d27..920a4ed0b3 100644 --- a/packages/runtime-host/src/protocol/message.ts +++ b/packages/runtime-host/src/protocol/message.ts @@ -99,15 +99,13 @@ export interface TurnMessageSubmitInput { readonly turnOrchestration?: TurnOrchestration; } -export type TurnMessageSubmitResult = - | { readonly disposition: 'steering'; readonly queueRevision: number } - | { readonly disposition: 'followup'; readonly queueRevision: number } - | { - readonly disposition: 'turn_started'; - readonly turnId: string; - readonly skillInvocation?: SkillInvocationResult; - } - | { readonly disposition: 'blocked'; readonly skillInvocation: SkillInvocationResult }; +export type TurnMessageSubmitResult = { + readonly skillInvocation: SkillInvocationResult; +} & ( + | { readonly disposition: 'steering' | 'followup'; readonly queueRevision: number } + | { readonly disposition: 'turn_started'; readonly turnId: string } + | { readonly disposition: 'blocked' } +); export interface TurnMessageQueryInput { readonly sessionId: string; @@ -344,18 +342,15 @@ function decodeTurnMessageQueryResult(value: unknown): TurnMessageQueryResult { function decodeTurnMessageSubmitResult(value: unknown): TurnMessageSubmitResult { const record = requireRecord(value, 'turn.message.submit result'); if (record.disposition === 'turn_started') { - const shaped = requireShapedRecord( - record, - 'turn.message.submit turn_started result', - ['disposition', 'turnId'], - ['skillInvocation'], - ); + assertExactKeys(record, 'turn.message.submit turn_started result', [ + 'disposition', + 'turnId', + 'skillInvocation', + ]); return { disposition: 'turn_started', - turnId: requireEntityId(shaped.turnId, 'turnId'), - ...(shaped.skillInvocation !== undefined - ? { skillInvocation: decodeSubmitSkillInvocation(shaped.skillInvocation) } - : {}), + turnId: requireEntityId(record.turnId, 'turnId'), + skillInvocation: decodeSubmitSkillInvocation(record.skillInvocation), }; } if (record.disposition === 'blocked') { @@ -370,10 +365,15 @@ function decodeTurnMessageSubmitResult(value: unknown): TurnMessageSubmitResult return { disposition: 'blocked', skillInvocation }; } if (record.disposition === 'steering' || record.disposition === 'followup') { - assertExactKeys(record, 'turn.message.submit queued result', ['disposition', 'queueRevision']); + assertExactKeys(record, 'turn.message.submit queued result', [ + 'disposition', + 'queueRevision', + 'skillInvocation', + ]); return { disposition: record.disposition, queueRevision: requireCount(record.queueRevision, 'queueRevision'), + skillInvocation: decodeSubmitSkillInvocation(record.skillInvocation), }; } throw invalidProtocolFrame('Invalid turn.message.submit disposition'); diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 0ba526392f..286ad77ce5 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -38,6 +38,7 @@ import { } from '@maka/runtime/message-authority'; import { normalizeRootTurnAdmissionPayload, + rootTurnAdmissionRecordFits, submittedTurnIntentsEqual, type ImmutableSteeringMessageProof, type MessageAdmissionStore, @@ -93,6 +94,12 @@ type MessageOutcome = readonly error: { readonly code: MessageOperationErrorCode; readonly message: string }; }; +const EMPTY_SKILL_INVOCATION: SkillInvocationResult = { + loaded: [], + failed: [], + receipts: [], +}; + export interface HostMessageSessionHeader { readonly isArchived: boolean; readonly unavailableReason?: string; @@ -111,6 +118,8 @@ export interface HostMessageStartInput { readonly turnId?: string; readonly runId?: string; readonly skillIds?: readonly string[]; + /** A durable preparation recovered before root admission committed. */ + readonly preparedSkillInvocation?: SkillInvocationResult; readonly turnOrchestration?: TurnOrchestration; } @@ -119,7 +128,7 @@ export interface HostMessageStartInput { * the client can act on, or fails with an opaque reason. */ export type HostMessageStartOutcome = - | { readonly turnId: string; readonly skillInvocation?: SkillInvocationResult } + | { readonly turnId: string; readonly skillInvocation: SkillInvocationResult } | { readonly blocked: SkillInvocationResult } | { readonly error: string }; @@ -144,6 +153,18 @@ export interface HostMessagePreparationInput { readonly placement: MessagePlacement; } +export type HostMessagePreparationOutcome = + | { + readonly kind: 'ready'; + readonly content: MessageContent; + readonly skillInvocation: SkillInvocationResult; + } + | { + readonly kind: 'rejected'; + readonly error: string; + readonly skillInvocation?: SkillInvocationResult; + }; + export interface HostMessageStopClaim { readonly deliverStop: () => Promise; readonly terminal: Promise; @@ -166,18 +187,16 @@ export interface HostMessageRootPort { startFromMessage( input: HostMessageStartInput, admission: SessionAdmissionLease, - commitAdmission: (canonicalContent: MessageContent) => Promise, + commitAdmission: ( + canonicalContent: MessageContent, + skillInvocation: SkillInvocationResult, + ) => Promise, ): Promise; startRecoveredMessages?( input: HostMessageRecoveryBatch, admission: SessionAdmissionLease, ): Promise<{ readonly turnId: string } | { readonly error: string }>; - prepareMessage( - input: HostMessagePreparationInput, - ): Promise< - | { readonly kind: 'ready'; readonly content: MessageContent } - | { readonly kind: 'rejected'; readonly error: string } - >; + prepareMessage(input: HostMessagePreparationInput): Promise; claimStop( input: Omit, commitQueueFence: () => QueueFenceResult, @@ -227,6 +246,7 @@ interface LiveEntry { content: MessageContent; modelContent: MessageContent; submittedContentDigest: `sha256:${string}`; + skillInvocation: SkillInvocationResult; readonly placement: MessagePlacement; readonly disposition: 'steering' | 'followup'; readonly generation: number; @@ -737,6 +757,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { content: submittedProjectionContent(admission.content), modelContent: admission.content, submittedContentDigest: admission.submittedContentDigest, + skillInvocation: admission.skillInvocation, placement: admission.placement, disposition: admission.disposition, generation: state.generation, @@ -837,6 +858,12 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { // (#pull/#ack/#nack), so a submit's preflight snapshot can go stale while // it awaits. That is transient: re-read the queue and re-run admission // instead of surfacing a spurious session_busy to the client. + let preparedForRoot: + | { + readonly identity: RuntimeMessageRunIdentity; + readonly outcome: HostMessagePreparationOutcome; + } + | undefined; for (let attempt = 0; ; attempt++) { const header = await this.#root.readSessionHeader(input.sessionId); if (this.#failStopped) { @@ -884,18 +911,22 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const started = await this.#root.startFromMessage( { sessionId: input.sessionId, - content: payload.content, + content: pendingAdmission?.content ?? payload.content, sourceMessage, initiatingConnectionId, turnId, runId, - ...(payload.skillIds.length > 0 ? { skillIds: payload.skillIds } : {}), + ...(pendingAdmission + ? { preparedSkillInvocation: pendingAdmission.skillInvocation } + : payload.skillIds.length > 0 + ? { skillIds: payload.skillIds } + : {}), ...(payload.turnOrchestration ? { turnOrchestration: payload.turnOrchestration } : {}), }, admission, - async (canonicalContent) => { + async (canonicalContent, skillInvocation) => { await this.#admissions.commitMessageAdmission({ sessionId: input.sessionId, turnId, @@ -907,6 +938,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { placement: 'current_turn', disposition: 'steering', ...(intent ? { submittedIntent: intent } : {}), + skillInvocation, admittedAt: pendingAdmission?.admittedAt ?? Date.now(), }); }, @@ -931,7 +963,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const result = { disposition: 'turn_started', turnId: started.turnId, - ...(started.skillInvocation ? { skillInvocation: started.skillInvocation } : {}), + skillInvocation: started.skillInvocation ?? EMPTY_SKILL_INVOCATION, } as const; return success(result); } @@ -953,19 +985,58 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { 'Root state does not match message reservation', ); } - if (allLiveEntries(state).length >= MESSAGE_QUEUE_MAX_ENTRIES) { - return failure('session_busy', 'Message queue capacity is full'); + const existingEntry = allLiveEntries(state).find( + (entry) => entry.messageId === input.messageId, + ); + if (existingEntry) { + const existingAdmission = await this.#admissions.readMessageAdmission( + input.sessionId, + input.messageId, + ); + if ( + !existingAdmission || + existingAdmission.submittedContentDigest !== messageContentDigest(payload.content) || + existingAdmission.submittedPlacement !== input.placement + ) { + return failure('operation_conflict', 'Message admission has a different payload'); + } + const result = { + disposition: existingEntry.disposition, + queueRevision: state.revision, + skillInvocation: existingEntry.skillInvocation, + } as const; + this.#rememberCompletedOperation( + 'submit', + input.sessionId, + input.messageId, + payload, + result, + ); + return success(result); } const disposition = input.placement === 'current_turn' ? 'steering' : 'followup'; - const prepared = await this.#root.prepareMessage({ - sessionId: input.sessionId, - turnId: rootState.turnId, - content: payload.content, - placement: input.placement, - }); + const prepared = + preparedForRoot && sameRun(preparedForRoot.identity, rootState) + ? preparedForRoot.outcome + : await this.#root.prepareMessage({ + sessionId: input.sessionId, + turnId: rootState.turnId, + content: payload.content, + placement: input.placement, + }); + preparedForRoot = { identity: rootState, outcome: prepared }; if (prepared.kind === 'rejected') { + if (prepared.skillInvocation) { + return success({ + disposition: 'blocked', + skillInvocation: prepared.skillInvocation, + } as const); + } return failure('operation_conflict', prepared.error); } + if (allLiveEntries(state).length >= MESSAGE_QUEUE_MAX_ENTRIES) { + return failure('session_busy', 'Message queue capacity is full'); + } const candidateRevision = state.revision; const candidateGeneration = state.generation; const entryId = this.#createId(); @@ -1013,11 +1084,12 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { messageId: input.messageId, content: prepared.content, submittedContentDigest: messageContentDigest(payload.content), + skillInvocation: prepared.skillInvocation, placement: input.placement, disposition, }, ] satisfies RootTurnSourceMessage[]; - if (!rootAdmissionPayloadFits(prospectiveSources)) { + if (!rootAdmissionPayloadFits(input.sessionId, rootState.turnId, prospectiveSources)) { return failure('session_busy', 'Message queue cannot form a durable follow-up Turn'); } if ( @@ -1032,7 +1104,11 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { } continue; } - const result = { disposition, queueRevision: candidateRevision + 1 } as const; + const result = { + disposition, + queueRevision: candidateRevision + 1, + skillInvocation: prepared.skillInvocation, + } as const; const messageAdmission: PendingMessageAdmission = { sessionId: input.sessionId, turnId: rootState.turnId, @@ -1043,6 +1119,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { submittedPlacement: input.placement, placement: input.placement, disposition, + skillInvocation: prepared.skillInvocation, admittedAt: Date.now(), }; await this.#admissions.commitMessageAdmission(messageAdmission); @@ -1056,6 +1133,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { content: payload.content, modelContent: prepared.content, submittedContentDigest: messageAdmission.submittedContentDigest, + skillInvocation: messageAdmission.skillInvocation, placement: input.placement, disposition, generation: state.generation, @@ -1344,6 +1422,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { submittedPlacement: 'next_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: entry.skillInvocation, admittedAt: entry.admittedAt, }); state.followup.splice(index, 1); @@ -1415,10 +1494,11 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { ...sourceFromEntry(entry), content: modelContent, submittedContentDigest: messageContentDigest(content), + skillInvocation: prepared.skillInvocation, } : sourceFromEntry(entry), ) satisfies RootTurnSourceMessage[]; - if (!rootAdmissionPayloadFits(sources)) { + if (!rootAdmissionPayloadFits(input.sessionId, state.reservedRoot.turnId, sources)) { return failure('session_busy', 'Message queue mutation exceeds root admission capacity'); } if (!(await this.#preflightSessionSnapshot(input.sessionId, { queue: updatedProjection }))) { @@ -1444,11 +1524,13 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { submittedPlacement: admission?.submittedPlacement ?? queued.entry.placement, placement: queued.entry.placement, disposition: queued.entry.disposition, + skillInvocation: prepared.skillInvocation, admittedAt: queued.entry.admittedAt, }); queued.entry.content = content; queued.entry.modelContent = modelContent; queued.entry.submittedContentDigest = messageContentDigest(content); + queued.entry.skillInvocation = prepared.skillInvocation; this.#mutated(state); const result = { queueRevision: state.revision }; this.#rememberCompletedOperation( @@ -1685,7 +1767,12 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { return failure('operation_conflict', 'Durable message receipt has a different payload'); } if (source.disposition === 'turn_started') { - return success({ disposition: 'turn_started', turnId: receipt.admission.turnId }); + return success({ + disposition: 'turn_started', + turnId: receipt.admission.turnId, + skillInvocation: + source.skillInvocation ?? receipt.admission.skillInvocation ?? EMPTY_SKILL_INVOCATION, + }); } return failure( 'outcome_unknown', @@ -2178,6 +2265,7 @@ function sourceFromEntry(entry: LiveEntry): RootFollowupSource { messageId: entry.messageId, content: normalizeMessageContent(entry.modelContent), submittedContentDigest: entry.submittedContentDigest, + skillInvocation: entry.skillInvocation, placement: entry.placement, disposition: entry.disposition, }; @@ -2189,6 +2277,7 @@ function pendingMessageSource(admission: PendingMessageAdmission): RootTurnSourc content: normalizeMessageContent(admission.content), submittedContentDigest: admission.submittedContentDigest, ...(admission.submittedIntent ? { submittedIntent: admission.submittedIntent } : {}), + skillInvocation: admission.skillInvocation, placement: admission.placement, disposition: admission.disposition, }; @@ -2367,11 +2456,28 @@ function canonicalFollowupBatch(entries: readonly LiveEntry[]): { } } -function rootAdmissionPayloadFits(sources: readonly RootTurnSourceMessage[]): boolean { +function rootAdmissionPayloadFits( + sessionId: string, + previousTurnId: string, + sources: readonly RootTurnSourceMessage[], +): boolean { try { const content = aggregateMessageContent(sources.map((source) => source.content)); - normalizeRootTurnAdmissionPayload(content, sources); - return true; + const worstCaseId = 'i'.repeat(128); + return rootTurnAdmissionRecordFits({ + sessionId, + turnId: worstCaseId, + proposedRunId: worstCaseId, + proposedUserMessageId: sources.length === 1 ? worstCaseId : null, + execution: { + kind: 'external_message', + inputDigest: `sha256:${'f'.repeat(64)}`, + }, + previousRootTurnId: previousTurnId, + normalizedInput: content, + sourceMessages: sources, + admittedAt: Number.MAX_SAFE_INTEGER, + }); } catch { return false; } diff --git a/packages/runtime-host/src/server/root-admission-owner.ts b/packages/runtime-host/src/server/root-admission-owner.ts index b199fb930b..b3cd5cc6f3 100644 --- a/packages/runtime-host/src/server/root-admission-owner.ts +++ b/packages/runtime-host/src/server/root-admission-owner.ts @@ -30,6 +30,7 @@ import type { RootTurnAdmissionStore, RootTurnSourceMessage, } from '@maka/storage/execution-stores'; +import { submittedTurnIntentsEqual } from '@maka/storage/execution-stores'; type OwnedAdmitRootTurnInput = Omit; type Immutable = T extends (...args: never[]) => unknown @@ -133,6 +134,8 @@ function sameRootAdmission(left: RootTurnAdmission, right: RootTurnAdmission): b source.placement === other.placement && source.disposition === other.disposition && source.submittedContentDigest === other.submittedContentDigest && + submittedTurnIntentsEqual(source.submittedIntent, other.submittedIntent) && + isDeepStrictEqual(source.skillInvocation, other.skillInvocation) && messageContentsEqual(source.content, other.content) ); }) && diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index adbdc14136..7aa2b787cb 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -1025,7 +1025,10 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { startFromMessage( input: HostMessageStartInput, admissionLease: SessionAdmissionLease, - commitAdmission: (canonicalContent: MessageContent) => Promise, + commitAdmission: ( + canonicalContent: MessageContent, + skillInvocation: SkillInvocationResult, + ) => Promise, ): Promise { if (isWorkHubCoordinationSessionId(input.sessionId)) { return Promise.resolve({ error: WORKHUB_COORDINATION_EXECUTION_UNAVAILABLE_REASON }); @@ -1034,7 +1037,8 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { const content = normalizeMessageContent(input.content); if ( input.sourceMessage.disposition !== 'turn_started' || - !messageContentsEqual(input.sourceMessage.content, content) + (!input.preparedSkillInvocation && + !messageContentsEqual(input.sourceMessage.content, content)) ) { throw new RuntimeMessageAuthorityInvariantError( 'Idle Message start lost its canonical turn_started source', @@ -1056,15 +1060,25 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { const skillIds = input.skillIds ?? []; const hasSkillInvocation = skillIds.length > 0 || parseSkillInvocationTokens(content.text).length > 0; - const prepared = hasSkillInvocation - ? await this.prepareHostedSkillInvocationContent( - input.sessionId, - turnId, + const prepared = input.preparedSkillInvocation + ? ({ + kind: 'ready', content, - skillIds, - input.initiatingConnectionId, - ) - : ({ kind: 'ready', content } as const); + skillInvocation: input.preparedSkillInvocation, + } as const) + : hasSkillInvocation + ? await this.prepareHostedSkillInvocationContent( + input.sessionId, + turnId, + content, + skillIds, + input.initiatingConnectionId, + ) + : ({ + kind: 'ready', + content, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + } as const); if (prepared.kind === 'rejected') { // Skill resolution is the only rejection a client can act on, so it // travels back as structured feedback instead of an opaque error. @@ -1075,6 +1089,11 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { : prepared.outcome.error.message, }; } + const skillInvocation = prepared.skillInvocation ?? { + loaded: [], + failed: [], + receipts: [], + }; const canonicalContent = preflightRootMessageContent(prepared.content); if (!canonicalContent.ok) return { error: 'Prepared message content exceeds durable limits' }; @@ -1090,7 +1109,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { } await this.prepareFreshAgentGraphEpoch(header, input.turnOrchestration); - await commitAdmission(canonicalContent.content); + await commitAdmission(canonicalContent.content, skillInvocation); const admitted = await this.rootAdmissionOwner.admitRootTurn({ sessionId: input.sessionId, @@ -1099,15 +1118,17 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { proposedUserMessageId: input.sourceMessage.messageId, execution: { kind: 'external_message', - inputDigest: messageContentDigest(content), + inputDigest: + input.sourceMessage.submittedContentDigest ?? messageContentDigest(content), }, normalizedInput: canonicalContent.content, ...(input.turnOrchestration ? { turnOrchestration: input.turnOrchestration } : {}), - ...(prepared.skillInvocation ? { skillInvocation: prepared.skillInvocation } : {}), + skillInvocation, sourceMessages: [ { ...input.sourceMessage, content: normalizeMessageContent(canonicalContent.content), + skillInvocation, }, ], admittedAt: Date.now(), @@ -1144,7 +1165,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { } return { turnId, - ...(prepared.skillInvocation ? { skillInvocation: prepared.skillInvocation } : {}), + skillInvocation, }; } finally { this.releaseRootReservation(reservation); @@ -1214,16 +1235,26 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { }); } - prepareMessage( - input: HostMessagePreparationInput, - ): Promise< - | { readonly kind: 'ready'; readonly content: MessageContent } - | { readonly kind: 'rejected'; readonly error: string } + prepareMessage(input: HostMessagePreparationInput): Promise< + | { + readonly kind: 'ready'; + readonly content: MessageContent; + readonly skillInvocation: SkillInvocationResult; + } + | { + readonly kind: 'rejected'; + readonly error: string; + readonly skillInvocation?: SkillInvocationResult; + } > { return this.runCommand(async () => { const content = normalizeMessageContent(input.content); if (parseSkillInvocationTokens(content.text).length === 0) { - return { kind: 'ready', content }; + return { + kind: 'ready', + content, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; } const prepare = () => this.prepareSkillInvocationContent(input.sessionId, input.turnId, content, []); diff --git a/packages/storage/src/__tests__/root-turn-admission-normalization.test.ts b/packages/storage/src/__tests__/root-turn-admission-normalization.test.ts index 431133f91c..f402f24dfe 100644 --- a/packages/storage/src/__tests__/root-turn-admission-normalization.test.ts +++ b/packages/storage/src/__tests__/root-turn-admission-normalization.test.ts @@ -54,3 +54,29 @@ test('root admission preserves and validates each source submission digest', () ]), ); }); + +test('root admission preserves and validates each source Skill outcome', () => { + const content = { text: 'prepared', displayText: '/skill:writer draft' } as const; + const skillInvocation = { + loaded: [{ id: 'writer', name: 'Writer' }], + failed: [{ request: 'typo', reason: 'not_found' as const }], + receipts: [], + }; + const source = { + messageId: 'message-skill', + content, + skillInvocation, + placement: 'next_turn' as const, + disposition: 'followup' as const, + }; + + assert.deepEqual( + normalizeRootTurnAdmissionPayload(content, [source]).sourceMessages[0]?.skillInvocation, + skillInvocation, + ); + assert.throws(() => + normalizeRootTurnAdmissionPayload(content, [ + { ...source, skillInvocation: { loaded: [], failed: [], receipts: 'invalid' } }, + ]), + ); +}); 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 acbfce376a..a816797897 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -365,7 +365,12 @@ describe('SqliteSessionMetadataStore', () => { const store = createSqliteSessionMetadataStore(':memory:'); try { await store.create(fullHeader({ id: 'session-1', connectionLocked: false })); - const admission: PendingMessageAdmission = { + const skillInvocation = { + loaded: [{ id: 'review', name: 'Review' }], + failed: [{ request: 'typo', reason: 'not_found' as const }], + receipts: [], + }; + const admission = { sessionId: 'session-1', turnId: 'turn-1', runId: 'run-1', @@ -382,8 +387,9 @@ describe('SqliteSessionMetadataStore', () => { skillIds: ['review'], turnOrchestration: { mode: 'graph', source: 'slash_command' }, }, + skillInvocation, admittedAt: 10, - }; + } satisfies PendingMessageAdmission & { readonly skillInvocation: typeof skillInvocation }; const normalizedAdmission = { ...admission, @@ -402,6 +408,13 @@ describe('SqliteSessionMetadataStore', () => { (await store.listMessageAdmissions('session-1')).map((entry) => entry.messageId), ['message-1'], ); + await assert.rejects( + store.commitMessageAdmission({ + ...admission, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }), + /Message admission identity conflict/, + ); await store.markMessagesHandedOff({ sessionId: 'session-1', messageIds: ['message-1'], @@ -434,6 +447,56 @@ describe('SqliteSessionMetadataStore', () => { } }); + test('migrates v33 message admissions with an empty Skill invocation outcome', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-message-admission-v33-')); + const path = join(root, 'state.sqlite'); + try { + const setup = createSqliteSessionMetadataStore(path); + try { + await setup.create(fullHeader({ id: 'session-v33-admission' })); + await setup.commitMessageAdmission({ + sessionId: 'session-v33-admission', + turnId: 'turn-1', + runId: 'run-1', + messageId: 'message-1', + content: { text: 'queued before the migration' }, + submittedContentDigest: messageContentDigest({ text: 'queued before the migration' }), + submittedPlacement: 'next_turn', + placement: 'next_turn', + disposition: 'followup', + skillInvocation: { loaded: [], failed: [], receipts: [] }, + admittedAt: 10, + }); + } finally { + setup.close(); + } + + const legacy = new DatabaseSync(path); + try { + legacy.exec(` + ALTER TABLE message_admissions DROP COLUMN skill_invocation_json; + UPDATE session_metadata_schema SET version = 33 WHERE scope = 'session_metadata'; + `); + } finally { + legacy.close(); + } + + const migrated = createSqliteSessionMetadataStore(path); + try { + assert.equal(migrated.schemaVersion(), SQLITE_SESSION_METADATA_SCHEMA_VERSION); + assert.deepEqual( + (await migrated.readMessageAdmission('session-v33-admission', 'message-1')) + ?.skillInvocation, + { loaded: [], failed: [], receipts: [] }, + ); + } finally { + migrated.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + test('removes the accepted payload after transcript handoff', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-message-handoff-')); const path = join(root, 'state.sqlite'); @@ -450,6 +513,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }); await store.markMessagesHandedOff({ @@ -501,6 +565,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }; await store.commitMessageAdmission(admission); @@ -565,6 +630,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 11, }); assert.equal(admission.disposition, 'followup'); @@ -609,6 +675,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 20 + index, }); } diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index 6a87cfb0a5..5fb8f732d1 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -100,6 +100,8 @@ export interface RootTurnSourceMessage { messageId: string; content: MessageContent; submittedContentDigest?: `sha256:${string}`; + /** The admission-time Skill outcome for this exact source Message. */ + skillInvocation?: SkillInvocationResult; /** * The exact-Turn intent this Message was submitted with — the Skill ids and * the orchestration override. Content and placement do not describe it, so @@ -1259,6 +1261,16 @@ function normalizeAdmitRootTurnInput(input: AdmitRootTurnInput): RootTurnAdmissi return deepFreezeRootTurnAdmission(admission); } +/** Whether a proposed admission satisfies the complete durable record contract and size bound. */ +export function rootTurnAdmissionRecordFits(input: AdmitRootTurnInput): boolean { + try { + normalizeAdmitRootTurnInput(input); + return true; + } catch { + return false; + } +} + const MUTABLE_AGENT_RUN_HEADER_FIELDS = new Set([ 'status', 'updatedAt', @@ -1667,12 +1679,20 @@ function normalizeRootTurnSourceMessages(value: unknown): readonly RootTurnSourc 'disposition', ...(Object.hasOwn(item, 'submittedContentDigest') ? ['submittedContentDigest'] : []), ...(Object.hasOwn(item, 'submittedIntent') ? ['submittedIntent'] : []), + ...(Object.hasOwn(item, 'skillInvocation') ? ['skillInvocation'] : []), ]) ) { throw new Error(`Invalid root turn source message at index ${index}`); } - const { messageId, content, submittedContentDigest, submittedIntent, placement, disposition } = - item; + const { + messageId, + content, + submittedContentDigest, + submittedIntent, + skillInvocation, + placement, + disposition, + } = item; if ( typeof messageId !== 'string' || !isSafeId(messageId) || @@ -1701,6 +1721,9 @@ function normalizeRootTurnSourceMessages(value: unknown): readonly RootTurnSourc ...(submittedIntent !== undefined ? { submittedIntent: normalizeSubmittedTurnIntent(submittedIntent) } : {}), + ...(skillInvocation !== undefined + ? { skillInvocation: decodeSkillInvocationResult(skillInvocation) } + : {}), placement, disposition, }); @@ -1729,6 +1752,7 @@ function rootTurnAdmissionPayloadsEqual( source.disposition === other.disposition && source.submittedContentDigest === other.submittedContentDigest && submittedTurnIntentsEqual(source.submittedIntent, other.submittedIntent) && + isDeepStrictEqual(source.skillInvocation, other.skillInvocation) && messageContentsEqual(source.content, other.content) ); }) diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 14462e5363..7275011099 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -80,7 +80,10 @@ const executionStoresReaderKinds = new WeakMap(); const executionStoresWritersByLease = new WeakMap(); const executionStoresWritersOpeningByLease = new WeakMap>(); -export { normalizeRootTurnAdmissionPayload } from './agent-run-store.js'; +export { + normalizeRootTurnAdmissionPayload, + rootTurnAdmissionRecordFits, +} from './agent-run-store.js'; export { isSessionNotFoundError, SessionReadMarkerMessageNotFoundError, diff --git a/packages/storage/src/message-admission-store.ts b/packages/storage/src/message-admission-store.ts index f5512adcc4..0df58b29d5 100644 --- a/packages/storage/src/message-admission-store.ts +++ b/packages/storage/src/message-admission-store.ts @@ -19,6 +19,10 @@ import { isDeepStrictEqual } from 'node:util'; import { normalizeMessageContent, type MessageContent } from '@maka/core/events'; +import { + decodeSkillInvocationResult, + type SkillInvocationResult, +} from '@maka/core/skill-invocation'; import { normalizeSubmittedTurnIntent, submittedTurnIntentsEqual, @@ -46,6 +50,8 @@ export interface PendingMessageAdmission { * same submit reads as a different one. */ readonly submittedIntent?: SubmittedTurnIntent; + /** The Skill resolution answer returned for this admitted Message. */ + readonly skillInvocation: SkillInvocationResult; readonly admittedAt: number; } @@ -101,6 +107,7 @@ export function normalizePendingMessageAdmission( ...(admission.submittedIntent ? { submittedIntent: normalizeSubmittedTurnIntent(admission.submittedIntent) } : {}), + skillInvocation: decodeSkillInvocationResult(admission.skillInvocation), }); if (!/^sha256:[a-f0-9]{64}$/u.test(normalized.submittedContentDigest)) { throw new Error('Invalid pending Message submitted content digest'); @@ -125,6 +132,7 @@ export function samePendingMessageAdmission( a.disposition === b.disposition && a.admittedAt === b.admittedAt && submittedTurnIntentsEqual(a.submittedIntent, b.submittedIntent) && + isDeepStrictEqual(a.skillInvocation, b.skillInvocation) && isDeepStrictEqual(a.content, b.content) ); } diff --git a/packages/storage/src/sqlite-session-metadata-schema.ts b/packages/storage/src/sqlite-session-metadata-schema.ts index 9a39e05a65..665affd63a 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 = 33; +export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 34; export const SQLITE_SESSION_MESSAGE_CHUNK_BYTES = 64 * 1024; export const SQLITE_SESSION_MESSAGE_CHUNK_MARKER = '{"$maka":"session-message-chunks-v1"}'; @@ -1216,6 +1216,14 @@ const MIGRATIONS: ReadonlyMap = new Map([ AND json_extract(payload_json, '$.subagentParent') IS NOT NULL; `, ], + [ + 34, + ` + ALTER TABLE message_admissions + ADD COLUMN skill_invocation_json TEXT NOT NULL + DEFAULT '{"loaded":[],"failed":[],"receipts":[]}'; + `, + ], ]); if (MIGRATIONS.size !== SQLITE_SESSION_METADATA_SCHEMA_VERSION) { @@ -1273,10 +1281,14 @@ export function migrateSqliteSessionMetadataDatabase( ) { const sql = MIGRATIONS.get(version); if (!sql) throw new Error(`Missing SQLite session metadata migration ${version}`); - // Version 32 adds one column, and the post-merge convergence path replays - // it onto a database that may already carry it. SQLite has no - // `ADD COLUMN IF NOT EXISTS`, so the guard lives here. - if (version !== 32 || !hasColumn(db, 'message_admissions', 'submitted_intent_json')) { + // Versions 32 and 34 each add one column, and the post-merge convergence + // path can replay them onto a database that already carries the current + // table shape. SQLite has no `ADD COLUMN IF NOT EXISTS`, so the guards + // live here. + const columnAlreadyPresent = + (version === 32 && hasColumn(db, 'message_admissions', 'submitted_intent_json')) || + (version === 34 && hasColumn(db, 'message_admissions', 'skill_invocation_json')); + if (!columnAlreadyPresent) { db.exec(sql); } if (version === 29 && hasColumn(db, 'session_metadata', 'last_used_at')) { diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index 59b34eae88..a4ddf3bf69 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -249,6 +249,7 @@ interface MessageAdmissionRow { readonly queue_order?: unknown; readonly admitted_at?: unknown; readonly submitted_intent_json?: unknown; + readonly skill_invocation_json?: unknown; } function decodeMessageAdmissionRow( @@ -260,6 +261,7 @@ function decodeMessageAdmissionRow( typeof row.run_id !== 'string' || typeof row.message_id !== 'string' || typeof row.content_json !== 'string' || + typeof row.skill_invocation_json !== 'string' || typeof row.submitted_content_digest !== 'string' || (row.submitted_placement !== 'current_turn' && row.submitted_placement !== 'next_turn') || (row.placement !== 'current_turn' && row.placement !== 'next_turn') || @@ -285,6 +287,9 @@ function decodeMessageAdmissionRow( ...(typeof row.submitted_intent_json === 'string' ? { submittedIntent: normalizeSubmittedTurnIntent(JSON.parse(row.submitted_intent_json)) } : {}), + skillInvocation: JSON.parse( + row.skill_invocation_json, + ) as PendingMessageAdmission['skillInvocation'], admittedAt: row.admitted_at, }); } @@ -1580,7 +1585,7 @@ export class SqliteSessionMetadataStore { ` SELECT turn_id, run_id, message_id, content_json, submitted_content_digest, submitted_placement, placement, disposition, queue_order, admitted_at, - submitted_intent_json + submitted_intent_json, skill_invocation_json FROM message_admissions WHERE session_id = ? AND message_id = ? `, @@ -1619,8 +1624,8 @@ export class SqliteSessionMetadataStore { INSERT INTO message_admissions( session_id, turn_id, run_id, message_id, content_json, submitted_content_digest, submitted_placement, placement, disposition, queue_order, admitted_at, - submitted_intent_json - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + submitted_intent_json, skill_invocation_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, ) .run( @@ -1636,6 +1641,7 @@ export class SqliteSessionMetadataStore { orderRow.next_order, stored.admittedAt, stored.submittedIntent ? JSON.stringify(stored.submittedIntent) : null, + JSON.stringify(stored.skillInvocation), ); return stored; @@ -1655,7 +1661,7 @@ export class SqliteSessionMetadataStore { ` SELECT turn_id, run_id, message_id, content_json, submitted_content_digest, submitted_placement, placement, disposition, queue_order, admitted_at, - submitted_intent_json + submitted_intent_json, skill_invocation_json FROM message_admissions WHERE session_id = ? AND message_id = ? `, @@ -1688,7 +1694,7 @@ export class SqliteSessionMetadataStore { ` SELECT turn_id, run_id, message_id, content_json, submitted_content_digest, submitted_placement, placement, disposition, queue_order, admitted_at, - submitted_intent_json + submitted_intent_json, skill_invocation_json FROM message_admissions WHERE session_id = ? ORDER BY queue_order, sequence @@ -1729,7 +1735,7 @@ export class SqliteSessionMetadataStore { ` SELECT turn_id, run_id, message_id, content_json, submitted_content_digest, submitted_placement, placement, disposition, queue_order, admitted_at, - submitted_intent_json + submitted_intent_json, skill_invocation_json FROM message_admissions WHERE session_id = ? AND message_id = ? `, @@ -1839,7 +1845,7 @@ export class SqliteSessionMetadataStore { ` SELECT turn_id, run_id, message_id, content_json, submitted_content_digest, submitted_placement, placement, disposition, queue_order, admitted_at, - submitted_intent_json + submitted_intent_json, skill_invocation_json FROM message_admissions WHERE session_id = ? AND message_id = ? `, @@ -1859,7 +1865,8 @@ export class SqliteSessionMetadataStore { .prepare( ` UPDATE message_admissions - SET content_json = ?, submitted_content_digest = ?, placement = ?, disposition = ? + SET content_json = ?, submitted_content_digest = ?, placement = ?, disposition = ?, + skill_invocation_json = ? WHERE session_id = ? AND message_id = ? `, ) @@ -1868,6 +1875,7 @@ export class SqliteSessionMetadataStore { stored.submittedContentDigest, stored.placement, stored.disposition, + JSON.stringify(stored.skillInvocation), stored.sessionId, stored.messageId, );