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 f9e3819a7a..ed11340b31 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 711c22fb2f..65c8e81cd1 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 @@ -369,7 +369,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(); @@ -474,7 +478,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(), @@ -574,7 +582,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(), @@ -726,6 +738,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( { @@ -733,7 +750,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(), @@ -768,7 +785,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" }, @@ -852,7 +869,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", @@ -931,6 +952,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: [] }, }; }, }), @@ -1028,7 +1050,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", @@ -1099,6 +1125,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( { @@ -1106,7 +1137,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(), @@ -1156,7 +1187,7 @@ test("queues explicit Desktop follow-ups", async () => { }, ], inlineReferences: [], - skillInvocation: { loaded: [], failed: [], receipts: [] }, + skillInvocation, }, ); assert.deepEqual(submits, [ @@ -1325,7 +1356,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 9e7139044b..4e9405c6b3 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 @@ -385,7 +385,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 @@ -398,7 +398,7 @@ export function registerRuntimeHostSessionExecutionIpc( ...(sideConversation ? { messageId } : {}), attachments, inlineReferences, - skillInvocation: EMPTY_SKILL_INVOCATION, + skillInvocation: submitted.skillInvocation, }; }, ); @@ -500,7 +500,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; @@ -511,7 +511,7 @@ export function registerRuntimeHostSessionExecutionIpc( disposition: result.disposition, attachments, inlineReferences, - skillInvocation: EMPTY_SKILL_INVOCATION, + skillInvocation: result.skillInvocation, }; }, ); diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts index b791c793bf..bc52be65e4 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -6045,6 +6045,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(); @@ -9085,7 +9123,10 @@ class RejectingUserCommandStopDriver extends RunningUserCommandDriver { } class HostSkillDriver extends SlashCommandDriver { - constructor(private readonly skillInvocation: SkillInvocationResult) { + constructor( + private readonly skillInvocation: SkillInvocationResult, + private readonly admittedDisposition: 'turn_started' | 'steering' = 'turn_started', + ) { super(); } @@ -9106,6 +9147,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; @@ -9904,7 +9948,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 48dfa5908d..167f1016da 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -1095,7 +1095,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 385b042a0e..7482e9c50b 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 32f8ea3ffb..e81052602e 100644 --- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts +++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts @@ -711,6 +711,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 ef33c0f6c0..9a2b499818 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -23,8 +23,13 @@ import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test } from 'node:test'; -import { messageContentDigest, type MessageContent } from '@maka/core/events'; +import { + aggregateMessageContents, + messageContentDigest, + type MessageContent, +} from '@maka/core/events'; import type { RuntimeEvent } from '@maka/core/runtime-event'; +import type { SkillInvocationResult } from '@maka/core/skill-invocation'; import { WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_SESSION_ROLE, @@ -34,11 +39,14 @@ import type { MarkMessagesHandedOffInput, MessageAdmissionStore, PendingMessageAdmission, + RootTurnSourceMessage, RootTurnSourceMessageReceipt, } from '@maka/storage/execution-stores'; +import { rootTurnAdmissionRecordFits } from '@maka/storage/execution-stores'; import { createSessionStore } from '@maka/storage/session-store'; import { MESSAGE_OPERATION_RESULT_MAX_BYTES, + MESSAGE_QUEUE_MAX_ENTRIES, MESSAGE_QUEUE_PROJECTION_MAX_BYTES, decodeSessionMessageQueueProjection, type SessionMessageQueueProjection, @@ -54,6 +62,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('consumes an active-target admission before the terminal transition can make it idle', async (t) => { const root = await mkdtemp(join(tmpdir(), 'maka-workhub-active-consume-')); @@ -121,6 +130,7 @@ test('consumes an active-target admission before the terminal transition can mak submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 10, }, }); @@ -215,6 +225,7 @@ test('idle recovery starts one real preassigned WorkHub root and restores the re submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 10, }, }); @@ -300,6 +311,7 @@ test('idle recovery keeps promoted steering ahead of distinct real WorkHub roots submittedPlacement: 'next_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 9, }); const workHubMessageIds: string[] = []; @@ -337,6 +349,7 @@ test('idle recovery keeps promoted steering ahead of distinct real WorkHub roots submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 10, }, }); @@ -366,6 +379,7 @@ test('idle recovery preserves the exact root identity of durable steering', asyn submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 10, }); @@ -399,7 +413,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); @@ -474,6 +492,7 @@ test('message execution query reports the Turn that durably owns each Message', submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }); fixture.receipts.set( @@ -553,6 +572,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'), @@ -564,24 +592,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', @@ -597,10 +646,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, { @@ -611,11 +666,284 @@ 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('queued steering admission budgets per-source Skill outcomes into its 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', 'current_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).steering.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('queued steering capacity preflight includes the original submitted placement', async () => { + const skillInvocation = largeSkillInvocation(); + const source = ( + messageId: string, + sourceSkillInvocation: SkillInvocationResult, + includeSubmittedPlacement = true, + text = 'x', + ): RootTurnSourceMessage => ({ + messageId, + content: { text }, + submittedContentDigest: messageContentDigest({ text }), + ...(includeSubmittedPlacement ? { submittedPlacement: 'current_turn' as const } : {}), + skillInvocation: sourceSkillInvocation, + placement: 'current_turn', + disposition: 'steering', + }); + const fits = (sources: readonly RootTurnSourceMessage[]) => + rootTurnAdmissionRecordFits({ + sessionId: ROOT.sessionId, + turnId: 'i'.repeat(128), + proposedRunId: 'i'.repeat(128), + proposedUserMessageId: sources.length === 1 ? 'i'.repeat(128) : null, + execution: { + kind: 'external_message', + inputDigest: `sha256:${'f'.repeat(64)}`, + }, + previousRootTurnId: ROOT.turnId, + normalizedInput: aggregateMessageContents(sources.map((candidate) => candidate.content)), + sourceMessages: sources, + admittedAt: Number.MAX_SAFE_INTEGER, + }); + const tunableSkillInvocation = (bytes: number): SkillInvocationResult => { + assert.ok(bytes >= 100 && bytes <= 50 * 1024); + let remaining = bytes - 100; + const loaded = Array.from({ length: 50 }, (_, index) => ({ + id: `skill-${index}`, + name: `Skill ${index}`, + })); + const receipts = loaded.map((skill) => { + const requestExtra = Math.min(511, remaining); + remaining -= requestExtra; + const refExtra = Math.min(511, remaining); + remaining -= refExtra; + return { + invocation: 'explicit' as const, + request: 'q'.repeat(1 + requestExtra), + success: true as const, + ref: 'r'.repeat(1 + refExtra), + id: skill.id, + name: skill.name, + scope: 'project' as const, + source: 'maka' as const, + truncated: false, + }; + }); + assert.equal(remaining, 0); + return { loaded, failed: [], receipts }; + }; + + const existingSourceCountAtBoundary = 22; + const candidateSkillBytesAtBoundary = 33_424; + const existing = Array.from({ length: existingSourceCountAtBoundary }, (_, index) => + source(`capacity-source-${index}`, skillInvocation), + ); + const candidateSkillInvocation = tunableSkillInvocation(candidateSkillBytesAtBoundary); + assert.equal( + fits([...existing, source('capacity-boundary', candidateSkillInvocation, false, 'boundary')]), + true, + ); + assert.equal( + fits([...existing, source('capacity-boundary', candidateSkillInvocation, true, 'boundary')]), + false, + ); + + const fixture = createFixture(); + fixture.coordinator.reserveRootTurn(ROOT); + fixture.setMessagePreparation(async (message) => ({ + kind: 'ready', + content: message.content, + skillInvocation: + message.content.text === 'boundary' ? candidateSkillInvocation : skillInvocation, + })); + for (const existingSource of existing) { + const outcome = await submit(fixture, existingSource.messageId, 'x', 'current_turn'); + assert.equal(outcome.ok, true, JSON.stringify(outcome)); + } + const revisionBeforeCandidate = fixture.coordinator.projection(ROOT.sessionId).queueRevision; + + const outcome = await submit(fixture, 'capacity-boundary', 'boundary', 'current_turn'); + + assert.deepEqual(outcome, { + ok: false, + error: { + code: 'session_busy', + message: 'Message queue cannot form a durable follow-up Turn', + }, + }); + assert.equal( + fixture.coordinator.projection(ROOT.sessionId).queueRevision, + revisionBeforeCandidate, + ); + assert.equal(fixture.readMessageAdmission('capacity-boundary'), undefined); + + await fixture.coordinator.handlers['queue.retract']( + { + originHostEpoch: 'epoch-1', + sessionId: ROOT.sessionId, + retractId: 'cleanup-placement-capacity-boundary', + }, + 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', 'current_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', 'current_turn'); + if (!outcome.ok) { + assert.equal(outcome.error.code, 'session_busy'); + break; + } + } + const projection = fixture.coordinator.projection(ROOT.sessionId); + const target = projection.steering[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)); @@ -716,6 +1044,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, }); @@ -761,6 +1090,7 @@ test('recovery starts one explicit follow-up and keeps later messages queued', a submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: index + 1, }); } @@ -802,6 +1132,7 @@ test('recovery starts one explicit follow-up and keeps later messages queued', a submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 2, }); }); @@ -832,6 +1163,7 @@ test('recovery folds later steering ahead of an earlier explicit follow-up', asy runId: ROOT.runId, ...admission, submittedContentDigest: messageContentDigest(admission.content), + skillInvocation: EMPTY_SKILL_INVOCATION, }); } @@ -876,6 +1208,7 @@ test('recovery folds promoted steering ahead of an earlier explicit follow-up', sessionId: ROOT.sessionId, ...admission, submittedContentDigest: messageContentDigest(admission.content), + skillInvocation: EMPTY_SKILL_INVOCATION, }); } @@ -912,6 +1245,7 @@ async function recoverExactTurnAcrossHostStop(): Promise const retried = await resubmitRecoveredExact(fixture, 'graph'); // The intent survived the crash cut whole, so the unchanged retry reads as - // the same submit. The recovered source keeps the queued disposition the - // pending record can express, so the Host answers `outcome_unknown` — the - // client keeps its row and reconciles from canonical transcript — instead of - // rejecting a request it is already executing. - assert.equal(retried.ok, false); - if (!retried.ok) assert.equal(retried.error.code, 'outcome_unknown'); + // the same submit. The durable source retains the queued disposition and + // Skill outcome even though its previous Host Epoch's revision is gone. + assert.deepEqual(retried, { + ok: true, + result: { disposition: 'steering', skillInvocation: EMPTY_SKILL_INVOCATION }, + }); assert.equal(fixture.startCalls(), 0); }); @@ -981,6 +1315,7 @@ test('recovery treats a durable steering event as the handoff proof', async () = submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 1, }); fixture.events.push(steeringEvent('recovered-steering', 'recover this steering event')); @@ -1013,6 +1348,7 @@ test('active recovery rebuilds only admissions without a durable proof', async ( submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: EMPTY_SKILL_INVOCATION, admittedAt: 1, }); } @@ -1027,6 +1363,90 @@ test('active recovery rebuilds only admissions without a durable proof', async ( ); }); +test('a retry of a recovered queued Message reuses its durable Skill outcome', async () => { + 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); @@ -1320,7 +1740,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']( @@ -1865,7 +2289,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), { @@ -2226,6 +2654,8 @@ test('release folds unpulled steering ahead of follow-up without changing source attachments: [firstAttachment], quotes: firstQuotes, }), + submittedPlacement: 'current_turn', + skillInvocation: EMPTY_SKILL_INVOCATION, placement: 'current_turn', disposition: 'steering', }, @@ -2237,6 +2667,8 @@ test('release folds unpulled steering ahead of follow-up without changing source attachments: [secondAttachment], quotes: secondQuotes, }), + submittedPlacement: 'current_turn', + skillInvocation: EMPTY_SKILL_INVOCATION, placement: 'current_turn', disposition: 'steering', }, @@ -2295,6 +2727,8 @@ test('terminal transition atomically folds messages submitted after run release' messageId: 'late-steer', content: { text: 'next intent' }, submittedContentDigest: messageContentDigest({ text: 'next intent' }), + submittedPlacement: 'current_turn', + skillInvocation: EMPTY_SKILL_INVOCATION, placement: 'current_turn', disposition: 'steering', }, @@ -2450,6 +2884,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( @@ -2557,8 +2992,10 @@ test('submit retries use keyed Host-Epoch outcomes and durable proof while old-E 'next_turn', 'old-epoch', ); - assert.equal(oldFollow.ok, false); - if (!oldFollow.ok) assert.equal(oldFollow.error.code, 'outcome_unknown'); + assert.deepEqual(oldFollow, { + ok: true, + result: { disposition: 'followup', skillInvocation: EMPTY_SKILL_INVOCATION }, + }); fixture.events.push( steeringEvent('old-steer', { @@ -2592,8 +3029,10 @@ test('submit retries use keyed Host-Epoch outcomes and durable proof while old-E 'durable current follow-up', 'next_turn', ); - assert.equal(currentFollow.ok, false); - if (!currentFollow.ok) assert.equal(currentFollow.error.code, 'outcome_unknown'); + assert.deepEqual(currentFollow, { + ok: true, + result: { disposition: 'followup', skillInvocation: EMPTY_SKILL_INVOCATION }, + }); const displayConflict = await submitContent( fixture, 'old-follow', @@ -2659,6 +3098,60 @@ test('submit retries use keyed Host-Epoch outcomes and durable proof while old-E if (!reclaimedConflict.ok) assert.equal(reclaimedConflict.error.code, 'operation_conflict'); }); +test('old-Epoch durable receipts replay queued Skill outcomes without a queue revision', 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: `Writer\n\n${input.content.text}`, + displayText: input.content.text, + }, + skillInvocation, + })); + fixture.coordinator.reserveRootTurn(ROOT); + fixture.coordinator.bindRun(ROOT); + + for (const [messageId, placement, disposition, turnId] of [ + ['durable-skill-steering', 'current_turn', 'steering', ROOT.turnId], + ['durable-skill-followup', 'next_turn', 'followup', 'successor-turn'], + ] as const) { + const submittedContent = { text: `/skill:writer /skill:typo ${disposition}` }; + const submitted = await submitContent(fixture, messageId, submittedContent, placement); + assert.equal(submitted.ok, true); + const admission = fixture.readMessageAdmission(messageId); + assert.ok(admission); + const receipt = sourceReceipt( + messageId, + admission.content, + placement, + disposition, + turnId, + submittedContent, + skillInvocation, + ); + fixture.receipts.set(messageId, receipt); + await fixture.coordinator.handoffRootSources({ + sessionId: ROOT.sessionId, + turnId: receipt.admission.turnId, + runId: receipt.admission.runId, + messageIds: [messageId], + }); + + assert.deepEqual( + await submitContent(fixture, messageId, submittedContent, placement, 'old-epoch'), + { + ok: true, + result: { disposition, skillInvocation }, + }, + ); + } +}); + test('old-Epoch durable proof ignores structured content key order', async () => { const fixture = createFixture(); const messageId = 'ordered-content'; @@ -2667,9 +3160,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 = { @@ -2689,7 +3207,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 }, }); }); @@ -2755,8 +3273,10 @@ test('old-Epoch prepared Skill proofs retain the exact submitted message identit 'next_turn', 'old-epoch', ); - assert.equal(exactFollowup.ok, false); - if (!exactFollowup.ok) assert.equal(exactFollowup.error.code, 'outcome_unknown'); + assert.deepEqual(exactFollowup, { + ok: true, + result: { disposition: 'followup', skillInvocation: EMPTY_SKILL_INVOCATION }, + }); const conflictingFollowup = await submitContent( fixture, 'prepared-followup', @@ -2846,8 +3366,10 @@ test('old-Epoch retries prove each submitted message in a prepared follow-up bat 'next_turn', 'old-epoch', ); - assert.equal(exact.ok, false); - if (!exact.ok) assert.equal(exact.error.code, 'outcome_unknown'); + assert.deepEqual(exact, { + ok: true, + result: { disposition: 'followup', skillInvocation: EMPTY_SKILL_INVOCATION }, + }); } const conflict = await submitContent( fixture, @@ -3002,6 +3524,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: @@ -3056,6 +3579,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. @@ -3069,14 +3593,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); @@ -3274,6 +3809,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 = { @@ -3296,6 +3832,7 @@ function sourceReceipt( }, previousRootTurnId: ROOT.turnId, normalizedInput: normalizedContent, + ...(skillInvocation ? { skillInvocation } : {}), sourceMessages: [sourceMessage], admittedAt: 1, }, @@ -3350,6 +3887,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 86af74820d..2503c28655 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -134,6 +134,12 @@ describe('Runtime Host bootstrap protocol', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 22); }); + test('publishes a new compatibility epoch for mandatory submit Skill outcomes', () => { + // Submit Skill outcomes and explicit OAuth Connection targets independently + // claimed epoch 78, so their merge requires a distinct compatibility boundary. + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 78); + }); + test('rejects the legacy connection update result in the current compatibility epoch', () => { assert.throws( () => @@ -1689,10 +1695,21 @@ 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: 'steering', skillInvocation }, + { disposition: 'followup', skillInvocation }, + { disposition: 'turn_started', turnId: 'turn-2', skillInvocation }, + { + disposition: 'blocked', + skillInvocation: { + loaded: [], + failed: [{ request: 'missing', reason: 'not_found' }], + receipts: [], + }, + }, ]) { assert.doesNotThrow(() => decodeHostFrame({ @@ -1703,16 +1720,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..bb0c443909 100644 --- a/packages/runtime-host/src/__tests__/root-admission-owner.test.ts +++ b/packages/runtime-host/src/__tests__/root-admission-owner.test.ts @@ -98,6 +98,55 @@ test('recovery installs the validated tip and the successor extends it', async ( }); }); +test('recovers the original submitted placement for a promoted source after SQLite reopen', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-root-admission-placement-')); + try { + const store = createSqliteAgentRunStore(root); + let admitted: RootTurnAdmission; + try { + const owner = new RootAdmissionOwner(store); + await owner.recoverSession('session'); + const content = { text: 'promoted follow-up' }; + admitted = ( + await owner.admitRootTurn({ + sessionId: 'session', + turnId: 'turn-promoted', + proposedRunId: 'run-promoted', + proposedUserMessageId: 'message-promoted', + execution: { kind: 'external_message' }, + normalizedInput: content, + sourceMessages: [ + { + messageId: 'message-promoted', + content, + submittedPlacement: 'next_turn', + placement: 'current_turn', + disposition: 'steering', + }, + ], + admittedAt: 10, + }) + ).admission; + } finally { + store.close?.(); + } + + const reopenedStore = createSqliteAgentRunStore(root); + try { + const reopenedOwner = new RootAdmissionOwner(reopenedStore); + const [recovered] = await reopenedOwner.recoverSession('session'); + assert.ok(recovered); + assert.equal(recovered.sourceMessages[0]?.submittedPlacement, 'next_turn'); + assert.equal(recovered.sourceMessages[0]?.placement, 'current_turn'); + assert.doesNotThrow(() => reopenedOwner.assertKnownAdmission(admitted)); + } finally { + reopenedStore.close?.(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test('fails closed when a known durable admission identity drifts', async () => { await withStore(async (store) => { const first = await store.admitRootTurn({ @@ -149,6 +198,15 @@ test('fails closed when a known durable admission identity drifts', async () => ); const [firstSource] = first.admission.sourceMessages; assert.ok(firstSource); + assert.doesNotThrow(() => + owner.assertKnownAdmission({ + ...first.admission, + sourceMessages: [ + { ...firstSource, submittedPlacement: firstSource.placement }, + ...first.admission.sourceMessages.slice(1), + ], + }), + ); const sourceDrifts: RootTurnAdmission[] = [ { ...first.admission, @@ -182,6 +240,37 @@ test('fails closed when a known durable admission identity drifts', async () => ...first.admission.sourceMessages.slice(1), ], }, + { + ...first.admission, + sourceMessages: [ + { ...firstSource, submittedPlacement: 'next_turn' }, + ...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 f4bd6ec3b5..add418d708 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -984,6 +984,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/__tests__/workhub-coordination-coordinator.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts index bde8ce31fe..baa8e2d292 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts @@ -772,6 +772,7 @@ async function persistTestAssignment( submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: Date.now(), }, }); diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index c02211477c..a90738397c 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -95,7 +95,10 @@ 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 = 78 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 79 as const; +// 79: Every `turn.message.submit` disposition carries the exact Skill +// invocation outcome. Durable queued replays may omit the previous Host +// Epoch's transient queue revision; older strict peers reject either shape. // 78: OAuth login targets explicit create/existing Connection entities and // returns their canonical identity. Older peers reject both closed wire shapes. // 77: LLM and tool usage-log projections carry an optional `sessionTitle` (the diff --git a/packages/runtime-host/src/protocol/message.ts b/packages/runtime-host/src/protocol/message.ts index 3ca77b6f7a..4bb02151fe 100644 --- a/packages/runtime-host/src/protocol/message.ts +++ b/packages/runtime-host/src/protocol/message.ts @@ -100,15 +100,17 @@ export interface TurnMessageSubmitInput { readonly turnOrchestration?: TurnOrchestration; } -export type TurnMessageSubmitResult = - | { readonly disposition: 'steering'; readonly queueRevision: number } - | { readonly disposition: 'followup'; readonly queueRevision: number } +export type TurnMessageSubmitResult = { + readonly skillInvocation: SkillInvocationResult; +} & ( | { - readonly disposition: 'turn_started'; - readonly turnId: string; - readonly skillInvocation?: SkillInvocationResult; + readonly disposition: 'steering' | 'followup'; + /** Absent when an older Host Epoch can prove admission but not its transient revision. */ + readonly queueRevision?: number; } - | { readonly disposition: 'blocked'; readonly skillInvocation: SkillInvocationResult }; + | { readonly disposition: 'turn_started'; readonly turnId: string } + | { readonly disposition: 'blocked' } +); export interface TurnMessageQueryInput { readonly sessionId: string; @@ -424,18 +426,15 @@ function decodeTurnMessageExecutionQueryResult(value: unknown): TurnMessageExecu 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') { @@ -450,10 +449,18 @@ 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']); + const shaped = requireShapedRecord( + record, + 'turn.message.submit queued result', + ['disposition', 'skillInvocation'], + ['queueRevision'], + ); return { disposition: record.disposition, - queueRevision: requireCount(record.queueRevision, 'queueRevision'), + ...(shaped.queueRevision !== undefined + ? { queueRevision: requireCount(shaped.queueRevision, 'queueRevision') } + : {}), + skillInvocation: decodeSubmitSkillInvocation(shaped.skillInvocation), }; } throw invalidProtocolFrame('Invalid turn.message.submit disposition'); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index d98a8de3be..c9f4cddd4c 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1365,6 +1365,7 @@ export async function createExecutionRuntimeHostComposition( submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: assignedAt, }, ...(create ? { create } : {}), diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index fee7af749b..02dc46e7a2 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 MarkMessagesHandedOffInput, @@ -94,6 +95,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; @@ -112,6 +119,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; } @@ -120,7 +129,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 }; @@ -147,6 +156,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; @@ -169,18 +190,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, @@ -230,6 +249,8 @@ interface LiveEntry { content: MessageContent; modelContent: MessageContent; submittedContentDigest: `sha256:${string}`; + readonly submittedPlacement: MessagePlacement; + skillInvocation: SkillInvocationResult; readonly placement: MessagePlacement; readonly disposition: 'steering' | 'followup'; generation: number; @@ -869,6 +890,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { content: submittedProjectionContent(admission.content), modelContent: admission.content, submittedContentDigest: admission.submittedContentDigest, + submittedPlacement: admission.submittedPlacement, + skillInvocation: admission.skillInvocation, placement: admission.placement, disposition: admission.disposition, generation: state.generation, @@ -974,6 +997,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) { @@ -1000,6 +1029,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { messageId: input.messageId, content: payload.content, submittedContentDigest: messageContentDigest(payload.content), + submittedPlacement: input.placement, ...(intent ? { submittedIntent: intent } : {}), placement: input.placement, disposition: 'turn_started', @@ -1021,18 +1051,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, @@ -1044,6 +1078,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { placement: 'current_turn', disposition: 'steering', ...(intent ? { submittedIntent: intent } : {}), + skillInvocation, admittedAt: pendingAdmission?.admittedAt ?? Date.now(), }); }, @@ -1068,7 +1103,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); } @@ -1090,19 +1125,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(); @@ -1146,6 +1220,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { messageId: input.messageId, content: prepared.content, submittedContentDigest: messageContentDigest(payload.content), + submittedPlacement: input.placement, + skillInvocation: prepared.skillInvocation, placement: input.placement, disposition, } satisfies RootTurnSourceMessage; @@ -1155,7 +1231,14 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const prospectiveFollowup = state.followup.map(sourceFromEntry); if (disposition === 'steering') prospectiveSteering.push(candidateSource); else prospectiveFollowup.push(candidateSource); - if (!successorAdmissionsFit(prospectiveSteering, prospectiveFollowup)) { + if ( + !successorAdmissionsFit( + input.sessionId, + rootState.turnId, + prospectiveSteering, + prospectiveFollowup, + ) + ) { return failure('session_busy', 'Message queue cannot form a durable follow-up Turn'); } if ( @@ -1170,7 +1253,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, @@ -1181,6 +1268,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { submittedPlacement: input.placement, placement: input.placement, disposition, + skillInvocation: prepared.skillInvocation, admittedAt: Date.now(), }; await this.#admissions.commitMessageAdmission(messageAdmission); @@ -1194,6 +1282,8 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { content: payload.content, modelContent: prepared.content, submittedContentDigest: messageAdmission.submittedContentDigest, + submittedPlacement: messageAdmission.submittedPlacement, + skillInvocation: messageAdmission.skillInvocation, placement: input.placement, disposition, generation: state.generation, @@ -1489,7 +1579,14 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { const prospectiveFollowup = state.followup .filter((queued) => queued !== entry) .map(sourceFromEntry); - if (!successorAdmissionsFit(prospectiveSteering, prospectiveFollowup)) { + if ( + !successorAdmissionsFit( + input.sessionId, + state.reservedRoot.turnId, + prospectiveSteering, + prospectiveFollowup, + ) + ) { return failure('session_busy', 'Promoted Message exceeds steering admission capacity'); } await this.#admissions.updateMessageAdmission({ @@ -1499,9 +1596,10 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { messageId: entry.messageId, content: entry.modelContent, submittedContentDigest: entry.submittedContentDigest, - submittedPlacement: 'next_turn', + submittedPlacement: entry.submittedPlacement, placement: 'current_turn', disposition: 'steering', + skillInvocation: entry.skillInvocation, admittedAt: entry.admittedAt, }); state.followup.splice(index, 1); @@ -1573,11 +1671,19 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { ...sourceFromEntry(entry), content: modelContent, submittedContentDigest: messageContentDigest(content), + skillInvocation: prepared.skillInvocation, } : sourceFromEntry(entry); const steeringSources = [...state.inFlight.values(), ...state.steering].map(updatedSource); const followupSources = state.followup.map(updatedSource); - if (!successorAdmissionsFit(steeringSources, followupSources)) { + if ( + !successorAdmissionsFit( + input.sessionId, + state.reservedRoot.turnId, + steeringSources, + followupSources, + ) + ) { return failure('session_busy', 'Message queue mutation exceeds root admission capacity'); } if (!(await this.#preflightSessionSnapshot(input.sessionId, { queue: updatedProjection }))) { @@ -1603,11 +1709,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( @@ -1843,13 +1951,19 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { if (!sameSourcePayload(receipt, payload)) { return failure('operation_conflict', 'Durable message receipt has a different payload'); } + const skillInvocation = + source.skillInvocation ?? receipt.admission.skillInvocation ?? EMPTY_SKILL_INVOCATION; if (source.disposition === 'turn_started') { - return success({ disposition: 'turn_started', turnId: receipt.admission.turnId }); + return success({ + disposition: 'turn_started', + turnId: receipt.admission.turnId, + skillInvocation, + }); } - return failure( - 'outcome_unknown', - 'Durable message proof does not include the original queue revision', - ); + return success({ + disposition: source.disposition, + skillInvocation, + }); } const steeringProof = await this.#durableProof.readImmutableSteeringMessageProof( input.sessionId, @@ -2327,7 +2441,7 @@ function sameSourcePayload( (durableDigest ? durableDigest === messageContentDigest(input.content) : messageContentsEqual(source.content, input.content)) && - source.placement === input.placement && + (source.submittedPlacement ?? source.placement) === input.placement && submittedTurnIntentsEqual(source.submittedIntent, submittedTurnIntent(input)) ); } @@ -2337,6 +2451,8 @@ function sourceFromEntry(entry: LiveEntry): RootFollowupSource { messageId: entry.messageId, content: normalizeMessageContent(entry.modelContent), submittedContentDigest: entry.submittedContentDigest, + submittedPlacement: entry.submittedPlacement, + skillInvocation: entry.skillInvocation, placement: entry.placement, disposition: entry.disposition, }; @@ -2347,7 +2463,9 @@ function pendingMessageSource(admission: PendingMessageAdmission): RootTurnSourc messageId: admission.messageId, content: normalizeMessageContent(admission.content), submittedContentDigest: admission.submittedContentDigest, + submittedPlacement: admission.submittedPlacement, ...(admission.submittedIntent ? { submittedIntent: admission.submittedIntent } : {}), + skillInvocation: admission.skillInvocation, placement: admission.placement, disposition: admission.disposition, }; @@ -2587,23 +2705,42 @@ function hasNativeSteeringRootIdentity(admission: PendingMessageAdmission): bool return admission.disposition === 'steering' && admission.submittedPlacement === 'current_turn'; } -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; } } function successorAdmissionsFit( + sessionId: string, + previousTurnId: string, steering: readonly RootTurnSourceMessage[], followup: readonly RootTurnSourceMessage[], ): boolean { return ( - (steering.length === 0 || rootAdmissionPayloadFits(steering)) && - followup.every((source) => rootAdmissionPayloadFits([source])) + (steering.length === 0 || rootAdmissionPayloadFits(sessionId, previousTurnId, steering)) && + followup.every((source) => rootAdmissionPayloadFits(sessionId, previousTurnId, [source])) ); } diff --git a/packages/runtime-host/src/server/root-admission-owner.ts b/packages/runtime-host/src/server/root-admission-owner.ts index 759fa3ee2c..885b566f22 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 @@ -134,6 +135,10 @@ function sameRootAdmission(left: RootTurnAdmission, right: RootTurnAdmission): b source.placement === other.placement && source.disposition === other.disposition && source.submittedContentDigest === other.submittedContentDigest && + (source.submittedPlacement ?? source.placement) === + (other.submittedPlacement ?? other.placement) && + 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 86d7ceed9c..dc4b7ebb4e 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -1029,7 +1029,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 }); @@ -1038,7 +1041,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', @@ -1060,15 +1064,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. @@ -1079,6 +1093,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' }; @@ -1094,7 +1113,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, @@ -1103,15 +1122,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(), @@ -1148,7 +1169,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { } return { turnId, - ...(prepared.skillInvocation ? { skillInvocation: prepared.skillInvocation } : {}), + skillInvocation, }; } finally { this.releaseRootReservation(reservation); @@ -1218,16 +1239,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..6ec943fd9b 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,50 @@ test('root admission preserves and validates each source submission digest', () ]), ); }); + +test('root admission preserves and validates each source submitted placement', () => { + const content = { text: 'promoted follow-up' } as const; + const source = { + messageId: 'promoted-message', + content, + submittedPlacement: 'next_turn' as const, + placement: 'current_turn' as const, + disposition: 'steering' as const, + }; + + assert.equal( + normalizeRootTurnAdmissionPayload(content, [source]).sourceMessages[0]?.submittedPlacement, + 'next_turn', + ); + assert.throws(() => + normalizeRootTurnAdmissionPayload(content, [ + { ...source, submittedPlacement: 'invalid-placement' }, + ]), + ); +}); + +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 ee0f51612c..8aea6a22be 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -369,7 +369,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', @@ -386,8 +391,9 @@ describe('SqliteSessionMetadataStore', () => { skillIds: ['review'], turnOrchestration: { mode: 'graph', source: 'slash_command' }, }, + skillInvocation, admittedAt: 10, - }; + } satisfies PendingMessageAdmission & { readonly skillInvocation: typeof skillInvocation }; const normalizedAdmission = { ...admission, @@ -406,6 +412,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'], @@ -438,6 +451,56 @@ describe('SqliteSessionMetadataStore', () => { } }); + test('migrates v34 message admissions with an empty Skill invocation outcome', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-message-admission-v34-')); + const path = join(root, 'state.sqlite'); + try { + const setup = createSqliteSessionMetadataStore(path); + try { + await setup.create(fullHeader({ id: 'session-v34-admission' })); + await setup.commitMessageAdmission({ + sessionId: 'session-v34-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 = 34 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-v34-admission', 'message-1')) + ?.skillInvocation, + { loaded: [], failed: [], receipts: [] }, + ); + } finally { + migrated.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + test('materializes a proven Root message when its admission is absent', async () => { const store = createSqliteSessionMetadataStore(':memory:'); try { @@ -635,6 +698,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }); @@ -674,6 +738,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 24, }); @@ -931,6 +996,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 19, }); await store.cancelMessageAdmissions('session-legacy-cancelled', ['message-legacy-cancelled']); @@ -1021,6 +1087,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 21, }); @@ -1064,6 +1131,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 22, }); @@ -1177,6 +1245,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'current_turn', placement: 'current_turn', disposition: 'steering', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }); await store.markMessagesHandedOff({ @@ -1229,6 +1298,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'next_turn' as const, placement: 'next_turn' as const, disposition: 'followup' as const, + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }; await store.commitMessageAdmission(admission); @@ -1326,6 +1396,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }; await store.commitMessageAdmission(admission); @@ -1390,6 +1461,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 11, }); assert.equal(admission.disposition, 'followup'); @@ -1434,6 +1506,7 @@ describe('SqliteSessionMetadataStore', () => { submittedPlacement: 'next_turn', placement: 'next_turn', disposition: 'followup', + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 20 + index, }); } diff --git a/packages/storage/src/__tests__/workhub-message-assignment.test.ts b/packages/storage/src/__tests__/workhub-message-assignment.test.ts index a6fef6e413..be317b800e 100644 --- a/packages/storage/src/__tests__/workhub-message-assignment.test.ts +++ b/packages/storage/src/__tests__/workhub-message-assignment.test.ts @@ -192,6 +192,7 @@ function assignmentRequest( submittedPlacement: 'current_turn' as const, placement: 'current_turn' as const, disposition: 'steering' as const, + skillInvocation: { loaded: [], failed: [], receipts: [] }, admittedAt: 10, }, }; diff --git a/packages/storage/src/agent-run-store.ts b/packages/storage/src/agent-run-store.ts index 27481f3fec..6ab33b96cf 100644 --- a/packages/storage/src/agent-run-store.ts +++ b/packages/storage/src/agent-run-store.ts @@ -100,6 +100,10 @@ export interface RootTurnSourceMessage { messageId: string; content: MessageContent; submittedContentDigest?: `sha256:${string}`; + /** The original placement before queue promotion; absent legacy records use `placement`. */ + submittedPlacement?: 'current_turn' | 'next_turn'; + /** 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 @@ -1249,6 +1253,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', @@ -1658,13 +1672,23 @@ function normalizeRootTurnSourceMessages(value: unknown): readonly RootTurnSourc 'placement', 'disposition', ...(Object.hasOwn(item, 'submittedContentDigest') ? ['submittedContentDigest'] : []), + ...(Object.hasOwn(item, 'submittedPlacement') ? ['submittedPlacement'] : []), ...(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, + submittedPlacement, + submittedIntent, + skillInvocation, + placement, + disposition, + } = item; if ( typeof messageId !== 'string' || !isSafeId(messageId) || @@ -1674,6 +1698,9 @@ function normalizeRootTurnSourceMessages(value: unknown): readonly RootTurnSourc disposition !== 'turn_started') || (disposition === 'steering' && placement !== 'current_turn') || (disposition === 'followup' && placement !== 'next_turn') || + (submittedPlacement !== undefined && + submittedPlacement !== 'current_turn' && + submittedPlacement !== 'next_turn') || (submittedContentDigest !== undefined && !isSha256Digest(submittedContentDigest)) ) { throw new Error(`Invalid root turn source message at index ${index}`); @@ -1690,9 +1717,13 @@ function normalizeRootTurnSourceMessages(value: unknown): readonly RootTurnSourc MAX_ATTACHMENT_COUNT, ), ...(submittedContentDigest !== undefined ? { submittedContentDigest } : {}), + ...(submittedPlacement !== undefined ? { submittedPlacement } : {}), ...(submittedIntent !== undefined ? { submittedIntent: normalizeSubmittedTurnIntent(submittedIntent) } : {}), + ...(skillInvocation !== undefined + ? { skillInvocation: decodeSkillInvocationResult(skillInvocation) } + : {}), placement, disposition, }); @@ -1721,7 +1752,10 @@ function rootTurnAdmissionPayloadsEqual( source.placement === other.placement && source.disposition === other.disposition && source.submittedContentDigest === other.submittedContentDigest && + (source.submittedPlacement ?? source.placement) === + (other.submittedPlacement ?? other.placement) && 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 33bdf087e7..27507b5f9b 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -81,7 +81,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 920da782e0..c759c3703a 100644 --- a/packages/storage/src/message-admission-store.ts +++ b/packages/storage/src/message-admission-store.ts @@ -23,6 +23,10 @@ import { normalizeMessageContent, type MessageContent, } from '@maka/core/events'; +import { + decodeSkillInvocationResult, + type SkillInvocationResult, +} from '@maka/core/skill-invocation'; import { normalizeSubmittedTurnIntent, submittedTurnIntentsEqual, @@ -50,6 +54,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; } @@ -127,6 +133,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'); @@ -181,6 +188,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 da49f651a7..5c42ae80c0 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 = 34; +export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 35; export const SQLITE_SESSION_MESSAGE_CHUNK_BYTES = 64 * 1024; export const SQLITE_SESSION_MESSAGE_CHUNK_MARKER = '{"$maka":"session-message-chunks-v1"}'; @@ -1225,6 +1225,14 @@ const MIGRATIONS: ReadonlyMap = new Map([ SELECT 1; `, ], + [ + 35, + ` + ALTER TABLE message_admissions + ADD COLUMN skill_invocation_json TEXT NOT NULL + DEFAULT '{"loaded":[],"failed":[],"receipts":[]}'; + `, + ], ]); if (MIGRATIONS.size !== SQLITE_SESSION_METADATA_SCHEMA_VERSION) { @@ -1282,10 +1290,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 35 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 === 35 && 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 aeaa008467..b403ae3461 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -279,6 +279,7 @@ interface MessageAdmissionRow { readonly queue_order?: unknown; readonly admitted_at?: unknown; readonly submitted_intent_json?: unknown; + readonly skill_invocation_json?: unknown; } function decodeMessageAdmissionRow( @@ -290,6 +291,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') || @@ -315,6 +317,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, }); } @@ -1627,7 +1632,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 = ? `, @@ -1663,8 +1668,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( @@ -1680,6 +1685,7 @@ export class SqliteSessionMetadataStore { orderRow.next_order, stored.admittedAt, stored.submittedIntent ? JSON.stringify(stored.submittedIntent) : null, + JSON.stringify(stored.skillInvocation), ); } @@ -1841,7 +1847,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 = ? `, @@ -1874,7 +1880,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 @@ -1955,7 +1961,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 = ? `, @@ -2207,7 +2213,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 = ? `, @@ -2227,7 +2233,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 = ? `, ) @@ -2236,6 +2243,7 @@ export class SqliteSessionMetadataStore { stored.submittedContentDigest, stored.placement, stored.disposition, + JSON.stringify(stored.skillInvocation), stored.sessionId, stored.messageId, );