diff --git a/docs/session-task-ledger-lifecycle.md b/docs/session-task-ledger-lifecycle.md index 1908f0a63e..4e9568a662 100644 --- a/docs/session-task-ledger-lifecycle.md +++ b/docs/session-task-ledger-lifecycle.md @@ -170,6 +170,23 @@ across Host epochs. The authority preserves the existing `task-events.jsonl`, `tasks.json`, legacy-read, and backfill behavior. Runtime Host is the sole interactive writer after production activation. +Historical tool presentation uses the separate `task.mutation.query` operation. +The caller supplies exact `{ turnId, toolCallId }` correlations taken from the +durable Session transcript. Runtime Host derives one immutable, sanitized +`create` or `update` presentation for each correlation directly from the +sequenced Task Ledger events; it never joins an old tool row to the current +Task snapshot and does not persist a second presentation authority. Legacy, +missing, or non-projectable correlations return explicit unresolved results. + +Mutation traversal is stateless and append-stable. The first page fixes a +high-water pair consisting of the latest included SQLite sequence and its +event id. Continuations repeat the same bounded correlation list and carry an +opaque cursor bound to the Session, correlation digest, high-water pair, and +next result position. Events appended above that watermark are excluded from +the traversal. If purge or replacement causes the high-water sequence to name +a different event, the Host returns `history_changed`, and the Client discards +the partial traversal instead of mixing two ledger incarnations. + ## Child Agent Ownership `agent_spawn(task_id=...)` resolves the task in the current session and claims diff --git a/package-lock.json b/package-lock.json index bb1e604af8..f901cb6ffd 100644 --- a/package-lock.json +++ b/package-lock.json @@ -14360,7 +14360,9 @@ "zod": "^4.4.3" }, "devDependencies": { + "@ai-sdk/provider": "4.0.7", "@types/ws": "^8.18.1", + "ai": "7.0.70", "electron": "^43.4.1" } }, diff --git a/packages/cli/src/__tests__/pi-task-mutation-hydration.test.ts b/packages/cli/src/__tests__/pi-task-mutation-hydration.test.ts new file mode 100644 index 0000000000..df003ebd88 --- /dev/null +++ b/packages/cli/src/__tests__/pi-task-mutation-hydration.test.ts @@ -0,0 +1,398 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import type { StoredMessage } from '@maka/core/session'; +import type { TaskMutationLookup } from '@maka/runtime-host/protocol'; +import { createTaskMutationHydrationController } from '../pi-task-mutation-hydration.js'; +import { + createMakaPiTranscriptState, + replaceTranscriptWithStoredMessages, +} from '../pi-transcript.js'; + +test('hydrates exact Task mutations atomically and de-duplicates frozen notices', async () => { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages()); + const queries: string[][] = []; + let changed = 0; + const controller = createTaskMutationHydrationController({ + state, + driver: { + queryTaskMutations: async (_sessionId, correlations) => { + queries.push(correlations.map(({ toolCallId }) => toolCallId)); + return correlations.map((correlation) => found(correlation.turnId, correlation.toolCallId)); + }, + }, + onChanged: () => { + changed += 1; + }, + }); + controller.replace('session-1'); + const firstTool = taskTool(state); + state.renderGeometry.entryFirstLine = new Map([[firstTool, 0]]); + state.renderGeometry.entryLineCount = new Map([[firstTool, 2]]); + state.renderGeometry.viewportTop = 5; + controller.schedule('session-1'); + await flush(); + + assert.deepEqual(queries, [['call-1']]); + assert.equal(firstTool.taskMutation?.kind, 'found'); + assert.equal(firstTool.taskMutationVersion, 1); + assert.equal(state.entries.filter((entry) => entry.kind === 'notice').length, 1); + assert.equal(changed, 1); + + controller.replace('session-1'); + replaceTranscriptWithStoredMessages(state, taskMessages()); + const replacementTool = taskTool(state); + state.renderGeometry.entryFirstLine = new Map([[replacementTool, 0]]); + state.renderGeometry.entryLineCount = new Map([[replacementTool, 2]]); + controller.schedule('session-1'); + await flush(); + assert.equal(replacementTool.taskMutation?.kind, 'found'); + assert.equal(state.entries.filter((entry) => entry.kind === 'notice').length, 0); + controller.dispose(); +}); + +test('invokes a driver method with its receiver intact', async () => { + class Driver { + #queries = 0; + + async queryTaskMutations( + _sessionId: string, + correlations: readonly { readonly turnId: string; readonly toolCallId: string }[], + ): Promise { + this.#queries += 1; + return correlations.map((correlation) => found(correlation.turnId, correlation.toolCallId)); + } + + queryCount(): number { + return this.#queries; + } + } + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages()); + const driver = new Driver(); + const controller = createTaskMutationHydrationController({ + state, + driver, + onChanged: () => undefined, + }); + controller.replace('session-1'); + controller.schedule('session-1'); + await flush(); + + assert.equal(driver.queryCount(), 1); + assert.equal(taskTool(state).taskMutation?.kind, 'found'); + controller.dispose(); +}); + +test('rejects a found presentation whose operation conflicts with the tool call', async () => { + for (const [toolName, operation] of [ + ['task_create', 'update'], + ['task_update', 'create'], + ] as const) { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages(toolName)); + let changed = 0; + const controller = createTaskMutationHydrationController({ + state, + driver: { + queryTaskMutations: async (_sessionId, correlations) => + correlations.map((correlation) => mismatchedFound(correlation, operation)), + }, + onChanged: () => { + changed += 1; + }, + }); + controller.replace('session-1'); + controller.schedule('session-1'); + await flush(); + + assert.equal(taskTool(state).taskMutation, undefined); + assert.equal(changed, 0); + controller.dispose(); + } +}); + +test('drops a late hydration after same-session transcript replacement', async () => { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages()); + const pending = deferred(); + const controller = createTaskMutationHydrationController({ + state, + driver: { queryTaskMutations: async () => pending.promise }, + onChanged: () => assert.fail('stale hydration must not publish'), + }); + controller.replace('session-1'); + controller.schedule('session-1'); + controller.replace('session-1'); + replaceTranscriptWithStoredMessages(state, taskMessages()); + pending.resolve([found('turn-1', 'call-1')]); + await flush(); + assert.equal(taskTool(state).taskMutation, undefined); + controller.dispose(); +}); + +test('does not announce hydration while the rendered tool still crosses the viewport', async () => { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages()); + const tool = taskTool(state); + state.renderGeometry.entryFirstLine = new Map([[tool, 3]]); + state.renderGeometry.entryLineCount = new Map([[tool, 4]]); + state.renderGeometry.viewportTop = 5; + let changed = 0; + const controller = createTaskMutationHydrationController({ + state, + driver: { + queryTaskMutations: async (_sessionId, correlations) => + correlations.map((correlation) => found(correlation.turnId, correlation.toolCallId)), + }, + onChanged: () => { + changed += 1; + }, + }); + controller.replace('session-1'); + controller.schedule('session-1'); + await flush(); + + assert.equal(tool.taskMutation?.kind, 'found'); + assert.equal( + state.entries.some((entry) => entry.kind === 'notice'), + false, + ); + assert.equal(changed, 1); + controller.dispose(); +}); + +test('does not re-query found immutable mutations when later entries settle', async () => { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessagesFor('turn-a', 'call-a')); + const secondState = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(secondState, taskMessagesFor('turn-b', 'call-b')); + state.entries.push(...secondState.entries); + const queries: string[][] = []; + const controller = createTaskMutationHydrationController({ + state, + driver: { + queryTaskMutations: async (_sessionId, correlations) => { + queries.push(correlations.map(({ toolCallId }) => toolCallId)); + return correlations.map((correlation) => + correlation.toolCallId === 'call-a' + ? found(correlation.turnId, correlation.toolCallId) + : { kind: 'not_found' as const, correlation }, + ); + }, + }, + onChanged: () => undefined, + }); + controller.replace('session-1'); + controller.schedule('session-1'); + await flush(); + + const thirdState = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(thirdState, taskMessagesFor('turn-c', 'call-c')); + state.entries.push(...thirdState.entries); + controller.schedule('session-1'); + await flush(); + + assert.deepEqual(queries, [ + ['call-a', 'call-b'], + ['call-b', 'call-c'], + ]); + controller.dispose(); +}); + +test('never announces unavailable frozen Task details as ready', async () => { + for (const reason of ['not_found', 'incompatible'] as const) { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages()); + const tool = taskTool(state); + state.renderGeometry.entryFirstLine = new Map([[tool, 0]]); + state.renderGeometry.entryLineCount = new Map([[tool, 2]]); + state.renderGeometry.viewportTop = 5; + let changed = 0; + const controller = createTaskMutationHydrationController({ + state, + driver: { + queryTaskMutations: async (_sessionId, correlations) => + correlations.map((correlation) => ({ kind: reason, correlation })), + }, + onChanged: () => { + changed += 1; + }, + }); + controller.replace('session-1'); + controller.schedule('session-1'); + await flush(); + + assert.equal(tool.taskMutation?.kind, 'unresolved'); + assert.equal( + state.entries.some((entry) => entry.kind === 'notice'), + false, + ); + assert.equal(changed, 1); + controller.dispose(); + } +}); + +test('keeps running unresolved history hidden until a settled projection is queried', async () => { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages()); + const tool = taskTool(state); + tool.callStatus = 'running'; + const controller = createTaskMutationHydrationController({ + state, + driver: { + queryTaskMutations: async (_sessionId, correlations) => + correlations.map((correlation) => ({ kind: 'not_found', correlation })), + }, + onChanged: () => undefined, + }); + controller.replace('session-1'); + controller.schedule('session-1'); + await flush(); + assert.deepEqual(tool.taskMutation, { + kind: 'unresolved', + reason: 'not_found', + observedSettled: false, + fingerprint: tool.taskMutation?.fingerprint, + }); + + tool.callStatus = 'completed'; + assert.equal( + tool.taskMutation?.kind === 'unresolved' && tool.taskMutation.observedSettled, + false, + ); + controller.schedule('session-1'); + await flush(); + assert.equal(tool.taskMutation?.kind === 'unresolved' && tool.taskMutation.observedSettled, true); + controller.dispose(); +}); + +function taskMessages(toolName: 'task_create' | 'task_update' = 'task_create'): StoredMessage[] { + return taskMessagesFor('turn-1', 'call-1', toolName); +} + +function taskMessagesFor( + turnId: string, + toolCallId: string, + toolName: 'task_create' | 'task_update' = 'task_create', +): StoredMessage[] { + return [ + { + type: 'tool_call', + id: toolCallId, + turnId, + ts: 1, + toolName, + args: { tasks: [{ subject: 'First task' }] }, + }, + { + type: 'tool_result', + id: 'result-1', + turnId, + ts: 2, + toolUseId: toolCallId, + isError: false, + content: { kind: 'text', text: 'Created' }, + }, + ]; +} + +function mismatchedFound( + correlation: { readonly turnId: string; readonly toolCallId: string }, + operation: 'create' | 'update', +): TaskMutationLookup { + return operation === 'create' + ? { + kind: 'found', + correlation, + presentation: { + operation, + correlation, + changes: [ + { + taskId: 'task-1', + key: 'T1', + subject: 'First task', + nextStatus: 'pending', + }, + ], + }, + } + : { + kind: 'found', + correlation, + presentation: { + operation, + correlation, + changes: [ + { + taskId: 'task-1', + key: 'T1', + subject: 'First task', + previousStatus: 'pending', + nextStatus: 'completed', + }, + ], + }, + }; +} + +function taskTool(state: ReturnType) { + const tool = state.entries.find( + (entry): entry is Extract<(typeof state.entries)[number], { kind: 'tool' }> => + entry.kind === 'tool' && entry.toolUseId === 'call-1', + ); + if (!tool) throw new Error('Expected Task tool entry'); + return tool; +} + +function found(turnId: string, toolCallId: string): TaskMutationLookup { + const correlation = { turnId, toolCallId }; + return { + kind: 'found', + correlation, + presentation: { + operation: 'create', + correlation, + changes: [ + { + taskId: 'task-1', + key: 'T1', + subject: 'First task', + nextStatus: 'pending', + }, + ], + }, + }; +} + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + +async function flush(): Promise { + await new Promise((resolve) => setImmediate(resolve)); +} diff --git a/packages/cli/src/__tests__/pi-task-mutation-render.test.ts b/packages/cli/src/__tests__/pi-task-mutation-render.test.ts new file mode 100644 index 0000000000..e191ac7466 --- /dev/null +++ b/packages/cli/src/__tests__/pi-task-mutation-render.test.ts @@ -0,0 +1,223 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import type { StoredMessage } from '@maka/core/session'; +import { MakaTranscriptComponent } from '../pi-tui-layout.js'; +import { + createMakaPiTranscriptState, + renderMakaPiTranscript, + replaceTranscriptWithStoredMessages, +} from '../pi-transcript.js'; + +test('renders 20 creates inline and summarizes 21 only on the live surface', () => { + for (const count of [20, 21]) { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages('task_create')); + const tool = onlyTool(state); + tool.taskMutation = { + kind: 'found', + fingerprint: `create-${count}`, + presentation: { + operation: 'create', + correlation: { turnId: 'turn-1', toolCallId: 'call-1' }, + changes: Array.from({ length: count }, (_, index) => ({ + taskId: `task-${index + 1}`, + key: `T${index + 1}`, + subject: `Task ${index + 1}`, + nextStatus: 'pending' as const, + })), + }, + }; + tool.taskMutationVersion = 1; + const live = plain(renderMakaPiTranscript(state, metadata(), 100)); + const document = plain( + new MakaTranscriptComponent(state, metadata).createDocumentRenderer()(100), + ); + if (count === 20) { + assert.match(live, /T20\s+pending\s+Task 20/); + } else { + assert.match(live, /Added 21 tasks · \/transcript to view full list/); + assert.doesNotMatch(live, /T21\s+pending/); + assert.match(document, /T21\s+pending\s+Task 21/); + } + } +}); + +test('leaves task_list and task_get on the ordinary tool renderer', () => { + for (const toolName of ['task_list', 'task_get'] as const) { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages(toolName)); + const rendered = plain(renderMakaPiTranscript(state, metadata(), 100)); + assert.doesNotMatch(rendered, /Added \d+ tasks|Task details unavailable/); + assert.match(rendered, new RegExp(toolName)); + } +}); + +test('renders update transitions and only settled unresolved history', () => { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, taskMessages('task_update')); + const tool = onlyTool(state); + tool.taskMutation = { + kind: 'found', + fingerprint: 'update-found', + presentation: { + operation: 'update', + correlation: { turnId: 'turn-1', toolCallId: 'call-1' }, + changes: [ + { + taskId: 'task-1', + key: 'T1', + subject: 'Ship it', + previousStatus: 'in_progress', + nextStatus: 'completed', + evidence: 'Tests passed', + }, + ], + }, + }; + tool.taskMutationVersion = 1; + assert.match( + plain(renderMakaPiTranscript(state, metadata(), 100)), + /T1\s+in_progress → completed\s+Ship it — Tests passed/, + ); + + tool.taskMutation = { + kind: 'unresolved', + reason: 'not_found', + observedSettled: false, + fingerprint: 'unresolved-running', + }; + tool.taskMutationVersion += 1; + assert.doesNotMatch( + plain(renderMakaPiTranscript(state, metadata(), 100)), + /Task details unavailable/, + ); + tool.taskMutation = { ...tool.taskMutation, observedSettled: true }; + tool.taskMutationVersion += 1; + assert.match(plain(renderMakaPiTranscript(state, metadata(), 100)), /Task details unavailable/); +}); + +test('neutralizes Task text controls on live and transcript document surfaces', () => { + const cases = [ + { + subject: '\x1b[31mVisible subject\x1b[0m', + nextStatus: 'pending' as const, + }, + { + subject: '\x1b\x07', + previousStatus: 'pending' as const, + nextStatus: 'blocked' as const, + reason: '\x1b]52;c;Y2xpcGJvYXJk\x07\x1b]8;;https://example.invalid\x1b\\Visible reason', + }, + { + subject: 'Completed task', + previousStatus: 'in_progress' as const, + nextStatus: 'completed' as const, + evidence: '\x1b\x9b', + }, + ]; + for (const [index, change] of cases.entries()) { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages( + state, + taskMessages(index === 0 ? 'task_create' : 'task_update'), + ); + const documentRenderer = new MakaTranscriptComponent(state, metadata).createDocumentRenderer(); + documentRenderer(100); + const tool = onlyTool(state); + tool.taskMutation = { + kind: 'found', + fingerprint: `unsafe-${index}`, + presentation: { + operation: index === 0 ? 'create' : 'update', + correlation: { turnId: 'turn-1', toolCallId: 'call-1' }, + changes: [{ taskId: 'task-1', key: 'T1', ...change }], + }, + }; + tool.taskMutationVersion = 1; + for (const rendered of [ + renderMakaPiTranscript(state, metadata(), 100), + documentRenderer(100), + ]) { + const raw = rendered.join('\n'); + assert.equal(raw.includes('\x1b]52;'), false); + assert.equal(raw.includes('\x1b]8;'), false); + assert.equal(raw.includes('\x1b\\'), false); + assert.equal(raw.includes('\x07'), false); + assert.equal(raw.includes('\x9b'), false); + if (index === 0) assert.match(plain(rendered), /\[31mVisible subject \[0m/); + if (index === 1) { + assert.match( + plain(rendered), + /\[unsafe text removed\][\s\S]*\]52;c;Y2xpcGJvYXJk \]8;;https:\/\/example.invalid[\s\S]*\\Visible reason/, + ); + } + if (index === 2) assert.match(plain(rendered), /Completed task — \[unsafe text removed\]/); + } + } +}); + +function taskMessages( + toolName: 'task_create' | 'task_update' | 'task_list' | 'task_get', +): StoredMessage[] { + return [ + { + type: 'tool_call', + id: 'call-1', + turnId: 'turn-1', + ts: 1, + toolName, + args: {}, + }, + { + type: 'tool_result', + id: 'result-1', + turnId: 'turn-1', + ts: 2, + toolUseId: 'call-1', + isError: false, + content: { kind: 'text', text: 'done' }, + }, + ]; +} + +function onlyTool(state: ReturnType) { + const tool = state.entries.find( + (entry): entry is Extract<(typeof state.entries)[number], { kind: 'tool' }> => + entry.kind === 'tool', + ); + if (!tool) throw new Error('Expected tool entry'); + return tool; +} + +function metadata() { + return { + title: 'maka', + cwd: '/repo', + model: 'model', + connectionSlug: 'openai', + permissionMode: 'ask', + }; +} + +function plain(lines: readonly string[]): string { + return lines.join('\n').replace(/\x1b\[[0-?]*[ -/]*[@-~]/g, ''); +} diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 571afe359e..6486441fb7 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -36,6 +36,8 @@ import { } from '@maka/runtime-host/client'; import { SESSION_CONTINUITY_SCHEMA_VERSION, + TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES, + TASK_MUTATION_QUERY_INPUT_MAX_ENCODED_BYTES, type GoalProjection, type InteractionPendingSnapshot, type OperationInput, @@ -44,6 +46,8 @@ import { type SessionContinuitySnapshot, type SessionUpdateResult, type SubscriptionFrame, + type TaskMutationCorrelation, + type TaskMutationQueryResult, } from '@maka/runtime-host/protocol'; import { projectSessionCatalogSummary } from '@maka/runtime-host/client'; import { @@ -57,6 +61,208 @@ import type { import { WAIT_BUDGET_MS } from './tui-terminal-mock.js'; describe('Runtime Host Maka Session driver', () => { + test('collects Task mutation pages and stable correlation batches before returning', async () => { + const connection = new FakeConnection([]); + const correlations = Array.from({ length: 129 }, (_, index) => ({ + turnId: `turn-${index}`, + toolCallId: `call-${index}`, + })); + const firstBatch = correlations.slice(0, 128); + const secondBatch = correlations.slice(128); + connection.taskMutationOutcomes.push( + mutationPage(firstBatch.slice(0, 64), 'cursor-1'), + { kind: 'history_changed', expected: mutationRevision('a'), actual: mutationRevision('b') }, + mutationPage(firstBatch, null), + mutationPage(secondBatch, null), + ); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/repo', + llmConnectionSlug: 'openai', + model: 'model', + }); + const lookups = await driver.queryTaskMutations!('session-1', [ + ...correlations, + correlations[0]!, + ]); + + assert.deepEqual( + lookups.map(({ correlation }) => correlation), + correlations, + ); + const requests = connection.requests.filter( + ({ operation }) => operation === 'task.mutation.query', + ); + assert.equal(requests.length, 4); + assert.equal( + (requests[0]!.input as { correlations: readonly unknown[] }).correlations.length, + 128, + ); + assert.equal((requests[1]!.input as { kind: string }).kind, 'continue'); + assert.equal((requests[2]!.input as { kind: string }).kind, 'start'); + assert.equal( + (requests[3]!.input as { correlations: readonly unknown[] }).correlations.length, + 1, + ); + }); + + test('batches Task mutation correlations by encoded bytes and keeps continuation inputs bounded', async () => { + const connection = new FakeConnection([]); + const correlations = Array.from({ length: 128 }, (_, index) => ({ + turnId: `turn-${index}`, + toolCallId: `legacy:nested:${String(index).padStart(3, '0')}${'x'.repeat(2_029)}`, + })); + let firstBatch = true; + connection.taskMutationResponder = (input) => { + if (input.kind === 'continue') return mutationPage(input.correlations.slice(1), null); + if (firstBatch) { + firstBatch = false; + return mutationPage(input.correlations.slice(0, 1), 'cursor-1'); + } + return mutationPage(input.correlations, null); + }; + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/repo', + llmConnectionSlug: 'openai', + model: 'model', + }); + + const lookups = await driver.queryTaskMutations!('session-1', correlations); + + assert.deepEqual( + lookups.map(({ correlation }) => correlation), + correlations, + ); + const requests = connection.requests.filter( + ({ operation }) => operation === 'task.mutation.query', + ); + assert.ok(requests.length > 2); + assert.ok(requests.some(({ input }) => (input as { kind: string }).kind === 'continue')); + for (const { input } of requests) { + const query = input as OperationInput<'task.mutation.query'>; + assert.ok( + Buffer.byteLength(JSON.stringify(query.correlations), 'utf8') <= + TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES, + ); + assert.ok( + Buffer.byteLength(JSON.stringify(query), 'utf8') <= + TASK_MUTATION_QUERY_INPUT_MAX_ENCODED_BYTES, + ); + } + }); + + test('rejects the whole Task mutation query when a later byte batch fails', async () => { + const connection = new FakeConnection([]); + const correlations = Array.from({ length: 128 }, (_, index) => ({ + turnId: `turn-${index}`, + toolCallId: `legacy:nested:${String(index).padStart(3, '0')}${'x'.repeat(2_029)}`, + })); + let batches = 0; + connection.taskMutationResponder = (input) => { + batches += 1; + if (batches === 2) throw new Error('later batch unavailable'); + return mutationPage(input.correlations, null); + }; + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/repo', + llmConnectionSlug: 'openai', + model: 'model', + }); + + await assert.rejects( + driver.queryTaskMutations!('session-1', correlations), + /later batch unavailable/, + ); + assert.equal(batches, 2); + }); + + test('rejects a repeated Task mutation cursor without returning a partial batch', async () => { + const connection = new FakeConnection([]); + const correlations = [ + { turnId: 'turn-1', toolCallId: 'call-1' }, + { turnId: 'turn-2', toolCallId: 'call-2' }, + ]; + connection.taskMutationOutcomes.push( + mutationPage(correlations.slice(0, 1), 'cursor-1'), + mutationPage(correlations.slice(1), 'cursor-1'), + ); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/repo', + llmConnectionSlug: 'openai', + model: 'model', + }); + await assert.rejects( + driver.queryTaskMutations!('session-1', correlations), + /repeated a continuation cursor/, + ); + }); + + test('rejects Task mutation pages from a different Session', async () => { + const connection = new FakeConnection([]); + const correlations = [{ turnId: 'turn-1', toolCallId: 'call-1' }]; + connection.taskMutationOutcomes.push( + mutationPage(correlations, null, { sessionId: 'session-2' }), + ); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/repo', + llmConnectionSlug: 'openai', + model: 'model', + }); + await assert.rejects( + driver.queryTaskMutations!('session-1', correlations), + /different Session/, + ); + }); + + test('rejects a Task mutation continuation from a different Session', async () => { + const connection = new FakeConnection([]); + const correlations = [ + { turnId: 'turn-1', toolCallId: 'call-1' }, + { turnId: 'turn-2', toolCallId: 'call-2' }, + ]; + connection.taskMutationOutcomes.push( + mutationPage(correlations.slice(0, 1), 'cursor-1'), + mutationPage(correlations.slice(1), null, { sessionId: 'session-2' }), + ); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/repo', + llmConnectionSlug: 'openai', + model: 'model', + }); + await assert.rejects( + driver.queryTaskMutations!('session-1', correlations), + /different Session/, + ); + }); + + test('rejects Task mutation revision drift and keeps the frozen continuation revision', async () => { + const connection = new FakeConnection([]); + const correlations = [ + { turnId: 'turn-1', toolCallId: 'call-1' }, + { turnId: 'turn-2', toolCallId: 'call-2' }, + ]; + connection.taskMutationOutcomes.push( + mutationPage(correlations.slice(0, 1), 'cursor-1', { revision: mutationRevision('a') }), + mutationPage(correlations.slice(1), null, { revision: mutationRevision('b') }), + ); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/repo', + llmConnectionSlug: 'openai', + model: 'model', + }); + await assert.rejects(driver.queryTaskMutations!('session-1', correlations), /changed revision/); + const continuation = connection.requests[1]!.input as { + revision: `sha256:${string}`; + }; + assert.equal(continuation.revision, mutationRevision('a')); + }); + test('maps authoritative Catalog activity into Session summaries', () => { assert.equal( projectSessionCatalogSummary(sessionProjection({ activityAt: 42 })).activityAt, @@ -2512,6 +2718,42 @@ describe('Runtime Host Maka Session driver', () => { }); }); +function mutationPage( + correlations: readonly TaskMutationCorrelation[], + nextCursor: string | null, + options: { + readonly sessionId?: string; + readonly revision?: `sha256:${string}`; + } = {}, +): TaskMutationQueryResult { + return { + kind: 'page', + sessionId: options.sessionId ?? 'session-1', + revision: options.revision ?? mutationRevision('a'), + lookups: correlations.map((correlation, index) => ({ + kind: 'found', + correlation, + presentation: { + operation: 'create', + correlation, + changes: [ + { + taskId: `task-${index + 1}`, + key: `T${index + 1}`, + subject: `Task ${index + 1}`, + nextStatus: 'pending', + }, + ], + }, + })), + nextCursor, + }; +} + +function mutationRevision(character: string): `sha256:${string}` { + return `sha256:${character.repeat(64)}`; +} + class FakeConnection { readonly requests: Array<{ operation: string; input: unknown }> = []; readonly sessionQueries: Array> = []; @@ -2528,6 +2770,10 @@ class FakeConnection { /** Scripted goal.query results, shifted per call; defaults to null (no goal). */ readonly goalQueryResults: Array = []; readonly messageSubmitOutcomes: Array | Error> = []; + readonly taskMutationOutcomes: Array = []; + taskMutationResponder: + | ((input: OperationInput<'task.mutation.query'>) => TaskMutationQueryResult) + | undefined; /** * Operations held open by a test. The request is recorded on entry and then * waits, so a test can hold one round trip and observe what the driver does @@ -2636,6 +2882,17 @@ class FakeConnection { goal: this.goalQueryResults.shift() ?? null, } as OperationOutput; } + if (operation === 'task.mutation.query') { + if (this.taskMutationResponder) { + return this.taskMutationResponder( + input as OperationInput<'task.mutation.query'>, + ) as OperationOutput; + } + const outcome = this.taskMutationOutcomes.shift(); + if (!outcome) throw new Error('Unexpected task.mutation.query request'); + if (outcome instanceof Error) throw outcome; + return outcome as OperationOutput; + } if (operation === 'session.configuration.update') { const update = input as OperationInput<'session.configuration.update'>; const outcome = this.configurationOutcomes.shift(); diff --git a/packages/cli/src/pi-task-mutation-hydration.ts b/packages/cli/src/pi-task-mutation-hydration.ts new file mode 100644 index 0000000000..85b7c974fb --- /dev/null +++ b/packages/cli/src/pi-task-mutation-hydration.ts @@ -0,0 +1,209 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { createHash } from 'node:crypto'; +import type { TaskMutationCorrelation, TaskMutationLookup } from '@maka/runtime-host/protocol'; +import type { MakaSessionDriver } from './session-driver.js'; +import type { + MakaPiTaskMutationState, + MakaPiToolEntry, + MakaPiTranscriptState, +} from './pi-transcript.js'; + +interface TaskMutationTarget { + readonly entry: MakaPiToolEntry; + readonly correlation: TaskMutationCorrelation; + readonly operation: 'create' | 'update'; + readonly observedSettled: boolean; +} + +export interface TaskMutationHydrationController { + /** Invalidates old requests before the transcript entry identities are replaced. */ + replace(sessionId: string | null): void; + /** Hydrates every current Task mutation entry through one atomic UI commit. */ + schedule(sessionId: string | null): void; + dispose(): void; +} + +export function createTaskMutationHydrationController(input: { + readonly state: MakaPiTranscriptState; + readonly driver: Pick; + readonly onChanged: () => void; +}): TaskMutationHydrationController { + let sessionId: string | null = null; + let transcriptGeneration = 0; + let requestGeneration = 0; + let disposed = false; + const announced = new Set(); + + const replace = (nextSessionId: string | null): void => { + transcriptGeneration += 1; + requestGeneration += 1; + if (nextSessionId !== sessionId) announced.clear(); + sessionId = nextSessionId; + }; + + const schedule = (requestedSessionId: string | null): void => { + if (disposed || !requestedSessionId || requestedSessionId !== sessionId) return; + if (!input.driver.queryTaskMutations) return; + const targets = taskMutationTargets(input.state); + if (targets.length === 0) return; + const capturedTranscriptGeneration = transcriptGeneration; + const capturedRequestGeneration = ++requestGeneration; + void input.driver + .queryTaskMutations( + requestedSessionId, + targets.map(({ correlation }) => correlation), + ) + .then((lookups) => { + if ( + disposed || + sessionId !== requestedSessionId || + transcriptGeneration !== capturedTranscriptGeneration || + requestGeneration !== capturedRequestGeneration || + lookups.length !== targets.length + ) { + return; + } + const entries = new Set(input.state.entries); + for (let index = 0; index < targets.length; index += 1) { + const target = targets[index]!; + const lookup = lookups[index]!; + if ( + !entries.has(target.entry) || + correlationKey(target.correlation) !== correlationKey(lookup.correlation) || + (lookup.kind === 'found' && lookup.presentation.operation !== target.operation) + ) { + return; + } + } + + let changed = false; + let frozenFoundChanged = false; + for (let index = 0; index < targets.length; index += 1) { + const target = targets[index]!; + const lookup = lookups[index]!; + const next = mutationState(lookup, target.observedSettled); + const previous = target.entry.taskMutation; + if (previous?.fingerprint === next.fingerprint) continue; + target.entry.taskMutation = next; + target.entry.taskMutationVersion = (target.entry.taskMutationVersion ?? 0) + 1; + changed = true; + if ( + next.kind !== 'found' || + previous?.kind === 'found' || + !isFrozenRenderedEntry(input.state, target.entry) + ) { + continue; + } + const noticeKey = `${requestedSessionId}\0${correlationKey(target.correlation)}\0${next.fingerprint}`; + if (announced.has(noticeKey)) continue; + announced.add(noticeKey); + frozenFoundChanged = true; + } + if (frozenFoundChanged) { + input.state.entries.push({ + kind: 'notice', + level: 'info', + text: 'Task details are ready · /transcript to view the full history.', + }); + } + if (changed || frozenFoundChanged) input.onChanged(); + }) + .catch(() => undefined); + }; + + return { + replace, + schedule, + dispose: () => { + disposed = true; + transcriptGeneration += 1; + requestGeneration += 1; + announced.clear(); + sessionId = null; + }, + }; +} + +function taskMutationTargets(state: MakaPiTranscriptState): TaskMutationTarget[] { + const targets: TaskMutationTarget[] = []; + const seen = new Set(); + for (const entry of state.entries) { + if ( + entry.kind !== 'tool' || + !entry.turnId || + (entry.toolName !== 'task_create' && entry.toolName !== 'task_update') || + entry.userOwned === true || + entry.suppressed === true || + entry.taskMutation?.kind === 'found' + ) { + continue; + } + const correlation = { turnId: entry.turnId, toolCallId: entry.toolUseId }; + const key = correlationKey(correlation); + if (seen.has(key)) continue; + seen.add(key); + targets.push({ + entry, + correlation, + operation: entry.toolName === 'task_create' ? 'create' : 'update', + observedSettled: entry.callStatus !== 'running', + }); + } + return targets; +} + +function mutationState( + lookup: TaskMutationLookup, + observedSettled: boolean, +): MakaPiTaskMutationState { + if (lookup.kind === 'found') { + return { + kind: 'found', + presentation: structuredClone(lookup.presentation), + fingerprint: fingerprint(lookup), + }; + } + return { + kind: 'unresolved', + reason: lookup.kind, + observedSettled, + fingerprint: fingerprint({ ...lookup, observedSettled }), + }; +} + +function isFrozenRenderedEntry(state: MakaPiTranscriptState, entry: MakaPiToolEntry): boolean { + const firstLine = state.renderGeometry.entryFirstLine?.get(entry); + const lineCount = state.renderGeometry.entryLineCount?.get(entry); + return ( + firstLine !== undefined && + lineCount !== undefined && + firstLine < state.renderGeometry.viewportTop && + (lineCount === 0 || firstLine + lineCount <= state.renderGeometry.viewportTop) + ); +} + +function fingerprint(value: unknown): string { + return createHash('sha256').update(JSON.stringify(value)).digest('hex'); +} + +function correlationKey(correlation: TaskMutationCorrelation): string { + return JSON.stringify([correlation.turnId, correlation.toolCallId]); +} diff --git a/packages/cli/src/pi-transcript-tools.ts b/packages/cli/src/pi-transcript-tools.ts index 6b9bdb5ad3..b38d418f5b 100644 --- a/packages/cli/src/pi-transcript-tools.ts +++ b/packages/cli/src/pi-transcript-tools.ts @@ -20,6 +20,8 @@ import type { ToolOutputStream, ToolResultContent } from '@maka/core/events'; import { formatQuietJsonValue, formatToolInvocationLine } from '@maka/core/tool-quiet-preview'; import { redactSecrets } from '@maka/core/display-redaction'; +import { TASK_EVIDENCE_MAX_CHARS, TASK_SUBJECT_MAX_CHARS } from '@maka/core/task-ledger'; +import { sanitizeUnicodeText } from '@maka/core/text-sanitize'; import { isActiveShellRunStatus, type PtyShellOutput, @@ -49,10 +51,76 @@ export function renderToolBlock( entry: MakaPiToolEntry, width: number, expanded: boolean, + surface: 'live' | 'document' = 'live', ): string[] { + const taskMutation = renderTaskMutationBlock(entry, width, surface); + if (taskMutation) return taskMutation; return expanded ? renderExpandedToolBlock(entry, width) : renderCompactToolBlock(entry, width); } +const TASK_MUTATION_LIVE_INLINE_MAX_CHANGES = 20; +const UNSAFE_TASK_TEXT_REMOVED = '[unsafe text removed]'; + +function renderTaskMutationBlock( + entry: MakaPiToolEntry, + width: number, + surface: 'live' | 'document', +): string[] | undefined { + const mutation = entry.taskMutation; + if (!mutation) return undefined; + const title = entry.title ?? (entry.toolName === 'task_create' ? 'Task Create' : 'Task Update'); + if (mutation.kind === 'unresolved') { + if (!mutation.observedSettled || makaPiToolPresentationStatus(entry) === 'running') { + return undefined; + } + return [fitLine(`${toolDisc(entry)} ${title} ${ansi.dim('Task details unavailable')}`, width)]; + } + + const { presentation } = mutation; + if ( + presentation.operation === 'create' && + surface === 'live' && + presentation.changes.length > TASK_MUTATION_LIVE_INLINE_MAX_CHANGES + ) { + return [ + fitLine( + `${toolDisc(entry)} ${title} Added ${presentation.changes.length} tasks · /transcript to view full list`, + width, + ), + ]; + } + + const summary = + presentation.operation === 'create' + ? `Added ${presentation.changes.length} ${presentation.changes.length === 1 ? 'task' : 'tasks'}` + : 'Updated task'; + const lines = [fitLine(`${toolDisc(entry)} ${title} ${summary}`, width)]; + for (const change of presentation.changes) { + const transition = + change.previousStatus === undefined + ? change.nextStatus + : `${change.previousStatus} → ${change.nextStatus}`; + const subject = safeTaskMutationText(change.subject, TASK_SUBJECT_MAX_CHARS); + const rawDetail = change.reason ?? change.evidence; + const detail = + rawDetail === undefined + ? undefined + : safeTaskMutationText(rawDetail, TASK_EVIDENCE_MAX_CHARS); + lines.push( + ...renderIndented( + `${change.key} ${transition} ${subject}${detail ? ` — ${detail}` : ''}`, + width, + 2, + ), + ); + } + return lines; +} + +function safeTaskMutationText(value: string, maxCodePoints: number): string { + return sanitizeUnicodeText(value, { maxCodePoints }) || UNSAFE_TASK_TEXT_REMOVED; +} + /** Status disc for a tool row: green = done, accent = running, danger = error/aborted/failed, muted = detached/unavailable. */ function toolDisc(entry: MakaPiToolEntry): string { const status = makaPiToolPresentationStatus(entry); diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts index bbbec37545..3369c4ac01 100644 --- a/packages/cli/src/pi-transcript.ts +++ b/packages/cli/src/pi-transcript.ts @@ -65,7 +65,7 @@ import { goalStatusLineText, isLiveGoalStatus } from './pi-goal.js'; import { renderToolBlock } from './pi-transcript-tools.js'; import { getTuiPrimaryGuidance } from './tui-primary-guidance.js'; import { renderTuiShortcutCopy } from './tui-shortcut-copy.js'; -import type { GoalProjection } from '@maka/runtime-host/protocol'; +import type { GoalProjection, TaskMutationPresentation } from '@maka/runtime-host/protocol'; export interface MakaPiUsageSummary { /** Cumulative cost in USD across the session. */ @@ -139,6 +139,8 @@ export interface MakaPiRenderGeometry { * treat as "nothing safely reachable" while the viewport has scrolled. */ entryFirstLine: Map | undefined; + /** Rendered line count per entry from the same render as `entryFirstLine`. */ + entryLineCount?: Map; /** * pi-tui's live-viewport top in transcript-line coordinates (the transcript * is the first layout child, so transcript line i is composed line i). Held @@ -157,6 +159,19 @@ export interface MakaPiToolOutputDelta { redacted: boolean; } +export type MakaPiTaskMutationState = + | { + kind: 'found'; + presentation: TaskMutationPresentation; + fingerprint: string; + } + | { + kind: 'unresolved'; + reason: 'not_found' | 'incompatible'; + observedSettled: boolean; + fingerprint: string; + }; + const LIVE_TOOL_BUFFER_MAX_CHARS = 64 * 1024; const LIVE_TOOL_BUFFER_MAX_CHUNKS = 512; @@ -180,6 +195,10 @@ export type MakaPiTranscriptEntry = result?: ToolResultContent; /** In-memory revision for render-cache invalidation when a result is replaced. */ resultVersion: number; + /** Host-owned immutable Task history projection, separate from ToolResult authority. */ + taskMutation?: MakaPiTaskMutationState; + /** Independent render-cache revision for Task mutation hydration. */ + taskMutationVersion?: number; progress: BoundedChunkBuffer; outputDeltas: BoundedChunkBuffer; durationMs?: number; @@ -233,7 +252,7 @@ export function createMakaPiTranscriptState(): MakaPiTranscriptState { queuedInteractions: [], expandAllTools: false, expandAllThinking: false, - renderGeometry: { entryFirstLine: undefined, viewportTop: 0 }, + renderGeometry: { entryFirstLine: undefined, entryLineCount: undefined, viewportTop: 0 }, usage: { costUsd: 0, cacheHitInput: 0, cacheMissInput: 0 }, steering: [], followup: [], @@ -391,6 +410,7 @@ export function appendUserCommandToTranscript( input: { command: input.command }, result: input.result, resultVersion: 1, + taskMutationVersion: 0, progress: createProgressBuffer(), outputDeltas: createOutputBuffer(), callStatus: toolResultActivityStatus( @@ -459,6 +479,7 @@ export function replaceTranscriptWithStoredMessages( // when the replacement is a pure truncation or identical content, pi-tui // keeps its viewport and so does the estimate. state.renderGeometry.entryFirstLine = undefined; + state.renderGeometry.entryLineCount = undefined; state.usage = { costUsd: 0, cacheHitInput: 0, cacheMissInput: 0 }; // Queues are per-active-run; a switched/reset session has none pending. state.steering = []; @@ -829,6 +850,7 @@ export function applyMakaSessionEventToTranscript( // replaces it with the durable full args. input: projectToolActivityArgs(event.toolName, event.args ?? event.argsPreview), resultVersion: 0, + taskMutationVersion: 0, progress: createProgressBuffer(), outputDeltas: createOutputBuffer(), callStatus: 'running', @@ -886,6 +908,7 @@ export function applyMakaSessionEventToTranscript( outputDeltas: createOutputBuffer(), ...(!event.contentOmitted ? { result: event.content } : {}), resultVersion: event.contentOmitted ? 0 : 1, + taskMutationVersion: 0, durationMs: event.durationMs, callStatus: toolResultActivityStatus(event.isError, event.content), expanded: state.expandAllTools, @@ -1121,6 +1144,7 @@ function storedToolToTranscriptEntry( ): MakaPiToolEntry { const entry: MakaPiToolEntry = { kind: 'tool', + turnId: call.turnId, toolUseId: call.id, toolName: call.toolName, ...(call.displayName ? { title: call.displayName } : {}), @@ -1129,6 +1153,7 @@ function storedToolToTranscriptEntry( outputDeltas: createOutputBuffer(), ...(result ? { result: result.content } : {}), resultVersion: result ? 1 : 0, + taskMutationVersion: 0, ...(result?.durationMs !== undefined ? { durationMs: result.durationMs } : {}), callStatus: result ? toolResultActivityStatus(result.isError, result.content) @@ -1353,7 +1378,9 @@ export function renderMakaPiTranscript( state: MakaPiTranscriptState, metadata: MakaPiTranscriptMetadata, width: number, + options: { surface?: 'live' | 'document' } = {}, ): string[] { + const surface = options.surface ?? 'live'; const safeWidth = Math.max(1, width); const lines: string[] = []; @@ -1365,12 +1392,14 @@ export function renderMakaPiTranscript( } const entryFirstLine = new Map(); + const entryLineCount = new Map(); const viewportTop = state.renderGeometry.viewportTop; let previousVisibleEntry: MakaPiTranscriptEntry | undefined; for (let i = 0; i < state.entries.length; i += 1) { const entry = state.entries[i]!; if (entry.kind === 'tool' && entry.suppressed) { entryFirstLine.set(entry, lines.length); + entryLineCount.set(entry, 0); continue; } // A blank gap separates human-facing boundaries (user/assistant/thinking/ @@ -1394,10 +1423,13 @@ export function renderMakaPiTranscript( const fullyOffScreen = lines.length < viewportTop && (entryHeight === 0 || lines.length + entryHeight <= viewportTop); - lines.push(...renderTranscriptEntryMemoized(entry, safeWidth, fullyOffScreen)); + const renderedEntry = renderTranscriptEntryMemoized(entry, safeWidth, fullyOffScreen, surface); + entryLineCount.set(entry, renderedEntry.length); + lines.push(...renderedEntry); previousVisibleEntry = entry; } state.renderGeometry.entryFirstLine = entryFirstLine; + state.renderGeometry.entryLineCount = entryLineCount; if (state.pendingInteraction?.type === 'sandbox_boundary_request') { lines.push(''); @@ -1488,6 +1520,7 @@ function renderTranscriptEntryMemoized( entry: MakaPiTranscriptEntry, width: number, offScreen: boolean, + surface: 'live' | 'document', ): string[] { // Off-screen entries live in terminal scrollback, which is immutable: any // change to their rendered lines forces pi-tui's differential renderer into a @@ -1500,15 +1533,19 @@ function renderTranscriptEntryMemoized( const cached = transcriptEntryRenderCache.get(entry); if (cached && cached.width === width) return cached.lines; } - const signature = transcriptEntrySignature(entry, width); + const signature = transcriptEntrySignature(entry, width, surface); const cached = transcriptEntryRenderCache.get(entry); if (cached && cached.signature === signature) return cached.lines; - const lines = renderTranscriptEntryBlock(entry, width); + const lines = renderTranscriptEntryBlock(entry, width, surface); transcriptEntryRenderCache.set(entry, { signature, lines, width }); return lines; } -function renderTranscriptEntryBlock(entry: MakaPiTranscriptEntry, width: number): string[] { +function renderTranscriptEntryBlock( + entry: MakaPiTranscriptEntry, + width: number, + surface: 'live' | 'document', +): string[] { // Keep the conversation stream inside a one-cell gutter. The editor owns // the full terminal width, so this makes the two surfaces align without // changing any of the individual block renderers' internal prefixes. @@ -1526,7 +1563,7 @@ function renderTranscriptEntryBlock(entry: MakaPiTranscriptEntry, width: number) case 'thinking': return renderThinkingBlock(entry, contentWidth, entry.expanded); case 'tool': - return renderToolBlock(entry, contentWidth, entry.expanded); + return renderToolBlock(entry, contentWidth, entry.expanded, surface); case 'notice': return renderNotice(entry, contentWidth); } @@ -1547,7 +1584,11 @@ function isBlankTranscriptLine(line: string): boolean { return line.replace(/\x1b\[[0-?]*[ -/]*[@-~]/g, '').trim().length === 0; } -function transcriptEntrySignature(entry: MakaPiTranscriptEntry, width: number): string { +function transcriptEntrySignature( + entry: MakaPiTranscriptEntry, + width: number, + surface: 'live' | 'document', +): string { switch (entry.kind) { // User text is immutable, so length is a safe change key. case 'user': @@ -1575,6 +1616,7 @@ function transcriptEntrySignature(entry: MakaPiTranscriptEntry, width: number): return [ 'tool', width, + surface, entry.expanded ? 1 : 0, makaPiToolPresentationStatus(entry), entry.durationMs ?? '', @@ -1582,6 +1624,7 @@ function transcriptEntrySignature(entry: MakaPiTranscriptEntry, width: number): entry.progress.version, entry.outputDeltas.version, entry.resultVersion, + entry.taskMutationVersion ?? 0, ].join('|'); } } diff --git a/packages/cli/src/pi-tui-layout.ts b/packages/cli/src/pi-tui-layout.ts index 3d31fbe689..d708c16313 100644 --- a/packages/cli/src/pi-tui-layout.ts +++ b/packages/cli/src/pi-tui-layout.ts @@ -81,10 +81,15 @@ export class MakaTranscriptComponent implements Component { { ...this.state, entries: this.state.entries.map(documentEntry), - renderGeometry: { entryFirstLine: undefined, viewportTop: 0 }, + renderGeometry: { + entryFirstLine: undefined, + entryLineCount: undefined, + viewportTop: 0, + }, }, this.metadata(), width, + { surface: 'document' }, ); } } diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index 48dfa5908d..47a4dc06e3 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -123,6 +123,7 @@ import { runMakaPiTuiTurn, type MakaPiTuiTurnRequest } from './pi-tui-turn.js'; import { editorTheme, selectListTheme } from './tui-ansi.js'; import { MakaAutocompleteAboveEditorComponent } from './tui-autocomplete-layout.js'; import { TranscriptViewerOverlay } from './pi-tui-transcript-viewer.js'; +import { createTaskMutationHydrationController } from './pi-task-mutation-hydration.js'; import { McpManagementOverlay } from './pi-tui-mcp-status.js'; import type { TuiMcpManagement } from './tui-mcp-control.js'; import { createShellRunElapsedTicker } from './shell-run-elapsed-ticker.js'; @@ -368,6 +369,11 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { }; const tui = new TuiMainScreen(terminal); const state = createMakaPiTranscriptState(); + const taskMutationHydration = createTaskMutationHydrationController({ + state, + driver: input.driver, + onChanged: () => requestRender(), + }); // A pending confirmation is meaningful only for the exact transcript whose // geometry produced it; reconnect/session replacement starts fresh. let expansionCollapseConfirm: { kind: ExpansionEntryKind; at: number } | undefined; @@ -379,9 +385,12 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { messages: readonly StoredMessage[], options: { preserveClientLocalEntries?: boolean } = {}, ): void => { + const sessionId = input.driver.getSessionId(); + taskMutationHydration.replace(sessionId); expansionCollapseConfirm = undefined; rememberTranscriptModel(messages); replaceTranscriptWithStoredMessages(state, messages, options); + taskMutationHydration.schedule(sessionId); }; let cwd = input.cwd; let model = input.model; @@ -686,6 +695,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { shellRunElapsedTicker.sync(); requestRender(); } + taskMutationHydration.schedule(sessionId); }) ?? (() => {}); const shellRunElapsedTicker = createShellRunElapsedTicker({ state, @@ -863,6 +873,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { unsubscribeStartedTurns(); unsubscribeResolvedInteractions(); unsubscribeTranscriptReplacements(); + taskMutationHydration.dispose(); shellRunHydration.dispose(); shellRunElapsedTicker.dispose(); stopTurnElapsedTicker(); diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts index 36ba9707c8..3658a337a0 100644 --- a/packages/cli/src/runtime-host-session-driver.ts +++ b/packages/cli/src/runtime-host-session-driver.ts @@ -70,11 +70,18 @@ import { SessionCatalogProjection, SessionUpdateResult, SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES, + TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES, + TASK_MUTATION_QUERY_MAX_CORRELATIONS, + taskMutationCorrelationsEncodedByteLength, WorkspaceTarget, type GoalControlAction, type GoalProjection, type SessionContinuitySnapshot, type TurnResumeParkReason, + type TaskMutationCorrelation, + type TaskMutationLookup, + type TaskMutationQueryResult, + type TaskMutationRevision, } from '@maka/runtime-host/protocol'; import { RuntimeHostSessionChannel } from './runtime-host-session-channel.js'; import type { RuntimeHostSessionChannelOpenResult } from './runtime-host-session-channel.js'; @@ -1023,6 +1030,98 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { return () => this.#transcriptListeners.delete(listener); } + async queryTaskMutations( + sessionId: string, + correlations: readonly TaskMutationCorrelation[], + ): Promise { + const unique = uniqueTaskMutationCorrelations(correlations); + const deadline = Date.now() + TASK_MUTATION_QUERY_DEADLINE_MS; + const lookups: TaskMutationLookup[] = []; + for (const batch of taskMutationCorrelationBatches(unique)) { + lookups.push(...(await this.#queryTaskMutationBatch(sessionId, batch, deadline))); + } + return lookups; + } + + async #queryTaskMutationBatch( + sessionId: string, + correlations: readonly TaskMutationCorrelation[], + deadline: number, + ): Promise { + for (let attempt = 0; attempt <= TASK_MUTATION_HISTORY_RESTARTS; attempt += 1) { + const lookups: TaskMutationLookup[] = []; + const cursors = new Set(); + let revision: TaskMutationRevision | undefined; + let result = await this.#requestTaskMutationPage( + { kind: 'start', sessionId, correlations }, + deadline, + ); + let pages = 0; + while (result.kind === 'page') { + if (result.sessionId !== sessionId) { + throw new Error('Task mutation query returned a different Session'); + } + if (revision !== undefined && result.revision !== revision) { + throw new Error('Task mutation query changed revision during pagination'); + } + revision ??= result.revision; + pages += 1; + if (pages > TASK_MUTATION_MAX_PAGES) { + throw new Error('Task mutation query exceeded its page budget'); + } + lookups.push(...result.lookups); + if (!result.nextCursor) { + assertExactTaskMutationLookups(correlations, lookups); + return lookups; + } + if (cursors.has(result.nextCursor)) { + throw new Error('Task mutation query repeated a continuation cursor'); + } + cursors.add(result.nextCursor); + result = await this.#requestTaskMutationPage( + { + kind: 'continue', + sessionId, + correlations, + revision, + cursor: result.nextCursor, + }, + deadline, + ); + } + if (attempt === TASK_MUTATION_HISTORY_RESTARTS) { + throw new Error('Task mutation history changed too often to read consistently'); + } + } + throw new Error('Task mutation query exhausted its restart budget'); + } + + async #requestTaskMutationPage( + input: OperationInput<'task.mutation.query'>, + deadline: number, + ): Promise { + let transientAttempts = 0; + while (true) { + const remaining = deadline - Date.now(); + if (remaining <= 0) throw new Error('Task mutation query timed out'); + try { + return await this.#connection.request('task.mutation.query', input, remaining); + } catch (error) { + if ( + error instanceof RuntimeHostOperationError && + (error.code === 'host_not_ready' || + error.code === 'host_draining' || + error.code === 'internal_failure') && + transientAttempts < TASK_MUTATION_TRANSIENT_RETRIES + ) { + transientAttempts += 1; + continue; + } + throw error; + } + } + } + listShellRunUpdates(sessionId: string): Promise { return readRuntimeHostResources(this.#connection, sessionId); } @@ -1645,6 +1744,68 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { } } +const TASK_MUTATION_QUERY_DEADLINE_MS = 15_000; +const TASK_MUTATION_HISTORY_RESTARTS = 2; +const TASK_MUTATION_TRANSIENT_RETRIES = 2; +const TASK_MUTATION_MAX_PAGES = 256; + +function uniqueTaskMutationCorrelations( + correlations: readonly TaskMutationCorrelation[], +): TaskMutationCorrelation[] { + const unique: TaskMutationCorrelation[] = []; + const seen = new Set(); + for (const correlation of correlations) { + const key = taskMutationCorrelationKey(correlation); + if (seen.has(key)) continue; + seen.add(key); + unique.push(correlation); + } + return unique; +} + +function taskMutationCorrelationBatches( + correlations: readonly TaskMutationCorrelation[], +): TaskMutationCorrelation[][] { + const batches: TaskMutationCorrelation[][] = []; + let batch: TaskMutationCorrelation[] = []; + for (const correlation of correlations) { + const candidate = [...batch, correlation]; + if ( + batch.length > 0 && + (candidate.length > TASK_MUTATION_QUERY_MAX_CORRELATIONS || + taskMutationCorrelationsEncodedByteLength(candidate) > + TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES) + ) { + batches.push(batch); + batch = [correlation]; + continue; + } + batch = candidate; + } + if (batch.length > 0) batches.push(batch); + return batches; +} + +function assertExactTaskMutationLookups( + correlations: readonly TaskMutationCorrelation[], + lookups: readonly TaskMutationLookup[], +): void { + if ( + lookups.length !== correlations.length || + lookups.some( + (lookup, index) => + taskMutationCorrelationKey(lookup.correlation) !== + taskMutationCorrelationKey(correlations[index]!), + ) + ) { + throw new Error('Task mutation query returned an incomplete or reordered result'); + } +} + +function taskMutationCorrelationKey(correlation: TaskMutationCorrelation): string { + return JSON.stringify([correlation.turnId, correlation.toolCallId]); +} + function workspaceTargetForCreate( current: { readonly target?: WorkspaceTarget; readonly hostCwd: string }, input: Pick, diff --git a/packages/cli/src/session-driver.ts b/packages/cli/src/session-driver.ts index 4dac719577..2dd14ef46f 100644 --- a/packages/cli/src/session-driver.ts +++ b/packages/cli/src/session-driver.ts @@ -31,6 +31,8 @@ import type { SkillInvocationResult } from '@maka/core/skill-invocation'; import type { GoalControlAction, GoalProjection, + TaskMutationCorrelation, + TaskMutationLookup, TurnMessageQueryResult, TurnMessageSubmitResult, } from '@maka/runtime-host/protocol'; @@ -199,6 +201,11 @@ export interface MakaSessionDriver { reason: MakaTranscriptReplacementReason, ) => void, ): () => void; + /** Reads immutable Task mutation presentations for exact durable tool calls. */ + queryTaskMutations?( + sessionId: string, + correlations: readonly TaskMutationCorrelation[], + ): Promise; /** * Prepares a fresh Session: stops every live user-owned command first so * their cards and Ctrl+C affordance never outlive the identity swap. diff --git a/packages/core/package.json b/packages/core/package.json index 7c0dfeec11..90af8a76b5 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -68,6 +68,7 @@ "./agent-graph-client-projection": "./dist/agent-graph-client-projection.js", "./agent-graph-supervisor-wake": "./dist/agent-graph-supervisor-wake.js", "./task-ledger": "./dist/task-ledger.js", + "./text-sanitize": "./dist/text-sanitize.js", "./foreign-session": "./dist/foreign-session.js", "./external-session": "./dist/external-session.js", "./deep-research-run": "./dist/deep-research-run.js", diff --git a/packages/runtime-host/package.json b/packages/runtime-host/package.json index 2a830bef1b..7b79acb341 100644 --- a/packages/runtime-host/package.json +++ b/packages/runtime-host/package.json @@ -34,7 +34,9 @@ "zod": "^4.4.3" }, "devDependencies": { + "@ai-sdk/provider": "4.0.7", "@types/ws": "^8.18.1", + "ai": "7.0.70", "electron": "^43.4.1" } } diff --git a/packages/runtime-host/src/__tests__/execution-host.test.ts b/packages/runtime-host/src/__tests__/execution-host.test.ts index de2c0c7ac6..0b57e98b3e 100644 --- a/packages/runtime-host/src/__tests__/execution-host.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host.test.ts @@ -35,13 +35,17 @@ import { connect, type Socket } from 'node:net'; import { tmpdir } from 'node:os'; import { dirname, join } from 'node:path'; import { test } from 'node:test'; +import type { LanguageModelV4StreamPart } from '@ai-sdk/provider'; import { TOOL_BOUNDARY_PROTOCOL_V1 } from '@maka/core/runtime-event'; +import { createExternalExecutionBoundary } from '@maka/core/sandbox-boundary'; import { canonicalToolArgsHash } from '@maka/core/tool-args-identity'; import type { AgentRunHeader } from '@maka/core/agent-run'; -import type { MessageContent } from '@maka/core/events'; +import type { MessageContent, SessionEvent } from '@maka/core/events'; +import type { LlmConnection } from '@maka/core/llm-connections'; import type { ConnectionCatalogEntry } from '@maka/core/runtime-policy'; import { decodeStoredMessage as decodePersistedStoredMessage, + type SessionHeader, type StoredMessage, } from '@maka/core/session'; import { markPersisted } from '@maka/core/persisted-value'; @@ -49,6 +53,7 @@ import type { Task } from '@maka/core/task-ledger'; import type { ScheduledTask } from '@maka/core/scheduled-task'; import { isTerminalRuntimeEvent } from '@maka/core/runtime-event'; import type { RuntimeEvent } from '@maka/core/runtime-event'; +import { AiSdkBackend } from '@maka/runtime/ai-sdk-backend'; import { buildTaskLedgerTools } from '@maka/runtime/task-ledger-tools'; import { buildRecoveredTerminalRuntimeEvent, @@ -61,6 +66,7 @@ import { FAKE_WAIT_FOR_STEERING_PROMPT, } from '@maka/runtime/test-only/fake-backend'; import { type MakaTool, type MakaToolContext } from '@maka/runtime/tool-runtime'; +import { MockLanguageModelV4, convertArrayToReadableStream } from 'ai/test'; import { openInteractiveExecutionStoresForRead, openInteractiveExecutionStoresForWrite, @@ -90,6 +96,8 @@ import { type SubscriptionFrame, type TaskLedgerQueryResult, type TaskLedgerRevision, + type TaskMutationCorrelation, + type TaskMutationQueryResult, type TurnMessageSubmitInput, type TurnSnapshot, } from '../protocol/index.js'; @@ -123,6 +131,11 @@ import { const decodeStoredMessage = (value: unknown): StoredMessage => decodePersistedStoredMessage(markPersisted(value)); +const ZERO_MODEL_USAGE = { + inputTokens: { total: 0, noCache: 0, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 0, text: 0, reasoning: 0 }, +}; + test('production Host resumes a Session through the ScheduledTask authority', { timeout: 30_000, }, async () => { @@ -244,6 +257,14 @@ test('dual UDS Clients query persisted Task Ledger tool-port mutations across Ho await withExecutionRoot(async (fixture) => { const initialRunId = randomUUID(); const initialTurnId = randomUUID(); + const createCorrelation = { + turnId: initialTurnId, + toolCallId: randomUUID(), + } satisfies TaskMutationCorrelation; + const updateCorrelation = { + turnId: initialTurnId, + toolCallId: randomUUID(), + } satisfies TaskMutationCorrelation; // Exercise the Runtime-facing port before Host startup; Hosted tool composition is separate. const toolPortProjection = await withOwnedTaskLedgerToolPort( fixture, @@ -251,7 +272,7 @@ test('dual UDS Clients query persisted Task Ledger tool-port mutations across Ho const context = taskLedgerToolContext(fixture, { runId: initialRunId, turnId: initialTurnId, - toolCallId: randomUUID(), + toolCallId: createCorrelation.toolCallId, }); const create = requireTaskLedgerTool(tools, 'task_create'); const createInput = create.parameters.parse({ @@ -265,7 +286,7 @@ test('dual UDS Clients query persisted Task Ledger tool-port mutations across Ho const updateInput = update.parameters.parse({ id: 'T1', status: 'in_progress' }); await update.impl(updateInput, { ...context, - toolCallId: randomUUID(), + toolCallId: updateCorrelation.toolCallId, }); return coordinator.list(fixture.sessionId, { includeTerminal: true, @@ -313,6 +334,42 @@ test('dual UDS Clients query persisted Task Ledger tool-port mutations across Ho assert.equal(byKey.task?.owner?.runId, initialRunId); assert.equal(byKey.task?.owner?.turnId, initialTurnId); + const desktopMutations = await collectTaskMutationProjection(desktop, fixture.sessionId, [ + createCorrelation, + updateCorrelation, + ]); + const tuiMutations = await collectTaskMutationProjection(tui, fixture.sessionId, [ + createCorrelation, + updateCorrelation, + ]); + assert.deepEqual(tuiMutations, desktopMutations); + assert.deepEqual( + desktopMutations.lookups.map((lookup) => + lookup.kind === 'found' + ? { + kind: lookup.kind, + operation: lookup.presentation.operation, + changes: lookup.presentation.changes.length, + subject: lookup.presentation.changes[0]?.subject, + } + : { kind: lookup.kind }, + ), + [ + { + kind: 'found', + operation: 'create', + changes: TASK_LEDGER_PAGE_MAX_ITEMS + 1, + subject: 'Authority acceptance task 1', + }, + { + kind: 'found', + operation: 'update', + changes: 1, + subject: 'Authority acceptance task 1', + }, + ], + ); + const firstPage = desktopProjection.pages[0]; assert.ok(firstPage?.nextCursor); staleContinuation = { @@ -378,6 +435,287 @@ test('dual UDS Clients query persisted Task Ledger tool-port mutations across Ho }); }); +test('Host queries the exact nested Task identities produced by Code Mode', async () => { + await withExecutionRoot(async (fixture) => { + const runId = randomUUID(); + const turnId = randomUUID(); + const parentToolCallId = `provider-${'x'.repeat(120)}`; + const events: SessionEvent[] = []; + + await withOwnedTaskLedgerToolPort(fixture, async (_coordinator, tools) => { + let step = 0; + let nextId = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + step += 1; + return { + stream: convertArrayToReadableStream( + step === 1 + ? [ + { type: 'stream-start' as const, warnings: [] }, + { + type: 'tool-call' as const, + toolCallId: parentToolCallId, + toolName: 'exec', + input: JSON.stringify({ + code: [ + "await tools.task_create({ tasks: [{ subject: 'Code Mode task' }] });", + "return await tools.task_update({ id: 'T1', status: 'in_progress' });", + ].join('\n'), + }), + }, + { + type: 'finish' as const, + finishReason: { unified: 'tool-calls' as const, raw: 'tool_calls' }, + usage: ZERO_MODEL_USAGE, + }, + ] + : [ + { type: 'stream-start' as const, warnings: [] }, + { + type: 'finish' as const, + finishReason: { unified: 'stop' as const, raw: 'stop' }, + usage: ZERO_MODEL_USAGE, + }, + ], + ), + }; + }, + }); + const backend = new AiSdkBackend({ + sessionId: fixture.sessionId, + header: taskMutationSessionHeader(fixture), + appendMessage: async () => undefined, + readExecutionBoundary: async () => createExternalExecutionBoundary(), + connection: taskMutationConnection(), + apiKey: 'sk-test', + modelId: 'mock-model', + modelFactory: () => model, + tools, + maxSteps: 1, + newId: () => `runtime-id-${++nextId}`, + now: () => 1, + }); + for await (const event of backend.send({ + invocationId: randomUUID(), + runId, + turnId, + text: 'Create and update a task', + context: [], + toolMode: 'code_mode', + })) { + events.push(event); + } + }); + + const correlations = events + .filter( + (event): event is Extract => + event.type === 'tool_start' && + (event.toolName === 'task_create' || event.toolName === 'task_update'), + ) + .map((event) => ({ turnId, toolCallId: event.toolUseId })); + assert.equal(correlations.length, 2); + assert.ok( + correlations.every(({ toolCallId }) => /^code_nested_v1_[a-f0-9]{64}$/.test(toolCallId)), + ); + + const host = await fixture.startHost(); + const client = await connectClient(fixture.root); + try { + const projection = await collectTaskMutationProjection( + client, + fixture.sessionId, + correlations, + ); + assert.deepEqual( + projection.lookups.map((lookup) => + lookup.kind === 'found' + ? { + correlation: lookup.correlation, + operation: lookup.presentation.operation, + } + : { correlation: lookup.correlation, kind: lookup.kind }, + ), + [ + { correlation: correlations[0], operation: 'create' }, + { correlation: correlations[1], operation: 'update' }, + ], + ); + } finally { + await client.close(); + await fixture.stopHost(host); + } + }); +}); + +test('Task mutation cursor freezes appends and detects purged history incarnation', async () => { + await withExecutionRoot(async (fixture) => { + const runId = randomUUID(); + const turnId = randomUUID(); + const createCorrelation = { + turnId, + toolCallId: `legacy:nested:${'x'.repeat(2_032)}`, + }; + assert.equal(Buffer.byteLength(JSON.stringify(createCorrelation.toolCallId), 'utf8'), 2_048); + const updateCorrelations: TaskMutationCorrelation[] = []; + await withOwnedTaskLedgerToolPort(fixture, async (_coordinator, tools) => { + const create = requireTaskLedgerTool(tools, 'task_create'); + await create.impl( + create.parameters.parse({ + tasks: Array.from({ length: 200 }, () => ({ subject: '\0'.repeat(200) })), + }), + taskLedgerToolContext(fixture, { runId, ...createCorrelation }), + ); + const update = requireTaskLedgerTool(tools, 'task_update'); + for (let index = 1; index < 128; index += 1) { + await update.impl( + update.parameters.parse({ id: `T${index}`, status: 'in_progress' }), + taskLedgerToolContext(fixture, { + runId, + turnId, + toolCallId: randomUUID(), + }), + ); + const correlation = { turnId, toolCallId: `update-${index}:nested:${randomUUID()}` }; + updateCorrelations.push(correlation); + await update.impl( + update.parameters.parse({ + id: `T${index}`, + status: 'completed', + completionEvidence: '证'.repeat(1000), + }), + taskLedgerToolContext(fixture, { runId, ...correlation }), + ); + } + }); + const correlations = [createCorrelation, ...updateCorrelations]; + assert.equal(correlations.length, 128); + + const firstHost = await fixture.startHost(); + const firstClient = await connectClient(fixture.root); + let firstPage: TaskMutationPage; + try { + const result = await firstClient.request('task.mutation.query', { + kind: 'start', + sessionId: fixture.sessionId, + correlations, + }); + assert.equal(result.kind, 'page'); + if (result.kind !== 'page') throw new Error('Expected initial Task mutation page'); + assert.equal(result.lookups[0]?.kind, 'found'); + assert.equal( + result.lookups[0]?.kind === 'found' ? result.lookups[0].presentation.changes.length : 0, + 200, + ); + assert.ok(Buffer.byteLength(JSON.stringify(result), 'utf8') < 320 * 1024); + assert.ok(result.nextCursor); + firstPage = result; + const tamperedCursor = JSON.parse( + Buffer.from(result.nextCursor, 'base64url').toString('utf8'), + ) as Record; + tamperedCursor.offset = Number(tamperedCursor.offset) + 1; + await assert.rejects( + firstClient.request('task.mutation.query', { + kind: 'continue', + sessionId: fixture.sessionId, + correlations, + revision: result.revision, + cursor: Buffer.from(JSON.stringify(tamperedCursor), 'utf8').toString('base64url'), + }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'invalid_request', + ); + } finally { + await firstClient.close(); + await fixture.stopHost(firstHost); + } + + await withOwnedTaskLedgerToolPort(fixture, async (_coordinator, tools) => { + const update = requireTaskLedgerTool(tools, 'task_update'); + const appendedCorrelation = { turnId: randomUUID(), toolCallId: randomUUID() }; + await update.impl( + update.parameters.parse({ id: 'T128', status: 'in_progress' }), + taskLedgerToolContext(fixture, { + runId: randomUUID(), + ...appendedCorrelation, + }), + ); + }); + + const successorHost = await fixture.startHost(); + const successor = await connectClient(fixture.root); + try { + const pages = [firstPage]; + let cursor = firstPage.nextCursor; + while (cursor) { + const result = await successor.request('task.mutation.query', { + kind: 'continue', + sessionId: fixture.sessionId, + correlations, + revision: firstPage.revision, + cursor, + }); + assert.equal(result.kind, 'page'); + if (result.kind !== 'page') throw new Error('Expected frozen Task mutation continuation'); + assert.equal(result.revision, firstPage.revision); + pages.push(result); + cursor = result.nextCursor; + } + const lookups = pages.flatMap((page) => page.lookups); + assert.equal(lookups.length, correlations.length); + assert.deepEqual( + lookups.map((lookup) => lookup.correlation), + correlations, + ); + assert.ok(lookups.every((lookup) => lookup.kind === 'found')); + } finally { + await successor.close(); + await fixture.stopHost(successorHost); + } + + const owner = await tryAcquireInteractiveRootOwner(fixture.capability); + assert.ok(owner); + if (!owner) throw new Error('Unable to acquire Task Ledger purge authority'); + const writer = await openInteractiveTaskLedgerStoreForWrite(owner.lease); + try { + await writer.purgeConversationTaskLedger(fixture.sessionId); + await writer.create(fixture.sessionId, [{ subject: 'Replacement ledger' }], { + runId: randomUUID(), + turnId: randomUUID(), + toolCallId: randomUUID(), + source: 'tool', + actor: 'main_agent', + }); + } finally { + writer.close(); + await owner.close(); + } + + const replacementHost = await fixture.startHost(); + const replacementClient = await connectClient(fixture.root); + try { + assert.ok(firstPage.nextCursor); + const changed = await replacementClient.request('task.mutation.query', { + kind: 'continue', + sessionId: fixture.sessionId, + correlations, + revision: firstPage.revision, + cursor: firstPage.nextCursor, + }); + assert.equal(changed.kind, 'history_changed'); + if (changed.kind !== 'history_changed') { + throw new Error('Expected purged Task mutation history to invalidate the cursor'); + } + assert.equal(changed.expected, firstPage.revision); + assert.notEqual(changed.actual, firstPage.revision); + } finally { + await replacementClient.close(); + await fixture.stopHost(replacementHost); + } + }); +}); + async function seedDispatchedClientCapability( fixture: ExecutionFixture, ): Promise<{ runId: string; toolName: string }> { @@ -1274,6 +1612,40 @@ function taskLedgerToolContext( }; } +function taskMutationSessionHeader(fixture: ExecutionFixture): SessionHeader { + return { + id: fixture.sessionId, + workspaceRoot: fixture.root, + cwd: fixture.root, + createdAt: 1, + name: 'Task mutation Code Mode integration', + titleIsManual: false, + isFlagged: false, + labels: [], + isArchived: false, + status: 'active', + hasUnread: false, + backend: 'ai-sdk', + llmConnectionSlug: 'openai', + connectionLocked: true, + model: 'mock-model', + permissionMode: 'bypass', + schemaVersion: 1, + }; +} + +function taskMutationConnection(): LlmConnection { + return { + slug: 'openai', + providerType: 'openai', + defaultModel: 'mock-model', + name: 'OpenAI', + enabled: true, + createdAt: 1, + updatedAt: 1, + }; +} + async function collectTaskLedgerProjection( client: RuntimeHostConnection, sessionId: string, @@ -1315,6 +1687,52 @@ async function collectTaskLedgerProjection( }; } +type TaskMutationPage = Extract; + +async function collectTaskMutationProjection( + client: RuntimeHostConnection, + sessionId: string, + correlations: readonly TaskMutationCorrelation[], +): Promise<{ + revision: TaskMutationPage['revision']; + pages: TaskMutationPage[]; + lookups: TaskMutationPage['lookups']; +}> { + const pages: TaskMutationPage[] = []; + let result = await client.request('task.mutation.query', { + kind: 'start', + sessionId, + correlations, + }); + assert.equal(result.kind, 'page'); + if (result.kind !== 'page') throw new Error('Expected initial Task mutation page'); + const revision = result.revision; + + while (true) { + assert.equal(result.sessionId, sessionId); + assert.equal(result.revision, revision); + pages.push(result); + if (result.nextCursor === null) break; + result = await client.request('task.mutation.query', { + kind: 'continue', + sessionId, + correlations, + revision, + cursor: result.nextCursor, + }); + assert.equal(result.kind, 'page'); + if (result.kind !== 'page') { + throw new Error('Task mutation history changed while collecting a stable projection'); + } + } + + return { + revision, + pages, + lookups: pages.flatMap((page) => page.lookups), + }; +} + async function waitForScheduledTaskCompletion( client: RuntimeHostConnection, taskId: string, diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 81bcfd7543..fecac05481 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -248,6 +248,11 @@ describe('Runtime Host bootstrap protocol', () => { assert.equal(Object.hasOwn(HOST_OPERATION_SPECS, 'execution.inspect.query'), true); }); + test('publishes a new compatibility epoch for durable Task mutation queries', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 76); + assert.equal(Object.hasOwn(HOST_OPERATION_SPECS, 'task.mutation.query'), true); + }); + test('adds credential rotation without changing existing credential inputs', () => { const issueInput = { principalKind: 'remote_owner', diff --git a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts index 7ba9c351de..d58244cf77 100644 --- a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts @@ -375,6 +375,22 @@ async function verifyConcurrentRevisionAuthority( 'Legacy child task', 'Retained task', ]); + const copiedMutation = await tui.request('task.mutation.query', { + kind: 'start', + sessionId: branch.id, + correlations: [{ turnId: 'turn-1', toolCallId: 'task-create-turn-1' }], + }); + assert.equal(copiedMutation.kind, 'page'); + if (copiedMutation.kind !== 'page') { + assert.fail('Branch Task mutation query must return a page'); + } + assert.equal(copiedMutation.lookups[0]?.kind, 'found'); + assert.equal( + copiedMutation.lookups[0]?.kind === 'found' + ? copiedMutation.lookups[0].presentation.changes[0]?.subject + : undefined, + 'Retained task', + ); const renamed = await desktop.request('session.metadata.update', { sessionId: sourceSessionId, @@ -1473,6 +1489,8 @@ async function seedSource( } await tasks.create(source.id, [{ subject: 'Retained task' }], { turnId: 'turn-1', + runId: 'run-turn-1', + toolCallId: 'task-create-turn-1', source: 'tool', actor: 'main_agent', }); @@ -1488,6 +1506,8 @@ async function seedSource( { status: 'in_progress' }, { turnId: 'turn-2', + runId: 'run-turn-2', + toolCallId: 'task-update-turn-2', source: 'tool', actor: 'main_agent', }, diff --git a/packages/runtime-host/src/__tests__/task-mutation-projection.test.ts b/packages/runtime-host/src/__tests__/task-mutation-projection.test.ts new file mode 100644 index 0000000000..3ce2b06592 --- /dev/null +++ b/packages/runtime-host/src/__tests__/task-mutation-projection.test.ts @@ -0,0 +1,240 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import type { TaskLedgerEvent } from '@maka/core/task-ledger'; +import type { SequencedTaskLedgerEvent } from '@maka/storage/task-ledger-authority'; +import { projectTaskMutationLookups } from '../server/task-mutation-projection.js'; + +const createCorrelation = { turnId: 'turn-1', toolCallId: 'call-create' } as const; +const updateCorrelation = { turnId: 'turn-2', toolCallId: 'call-update' } as const; + +describe('Task mutation projection', () => { + test('keeps create history immutable after a later rename', () => { + const rows = [ + row(0, event('event-1', 'task_created', createCorrelation, 'Original subject')), + row( + 1, + event('event-2', 'task_updated', updateCorrelation, 'Renamed subject', { + previousStatus: 'pending', + }), + ), + ]; + const lookups = projectTaskMutationLookups(rows, [createCorrelation, updateCorrelation]); + assert.equal(lookups[0]?.kind, 'found'); + assert.equal( + lookups[0]?.kind === 'found' ? lookups[0].presentation.changes[0]?.subject : undefined, + 'Original subject', + ); + assert.equal(lookups[1]?.kind, 'found'); + assert.equal( + lookups[1]?.kind === 'found' ? lookups[1].presentation.changes[0]?.subject : undefined, + 'Renamed subject', + ); + }); + + test('projects a contiguous create batch and redacts exact mutation detail', () => { + const startCorrelation = { turnId: 'turn-start', toolCallId: 'call-start' }; + const blocked = event('event-3', 'task_blocked', updateCorrelation, 'Blocked task', { + previousStatus: 'in_progress', + nextStatus: 'blocked', + reason: 'Waiting for ghp_abcdefghijklmnopqrstuvwxyz123456', + }); + const lookups = projectTaskMutationLookups( + [ + row(0, event('event-1', 'task_created', createCorrelation, 'First')), + row(1, event('event-2', 'task_created', createCorrelation, 'Second', { taskIndex: 2 })), + row( + 2, + event('event-start', 'task_started', startCorrelation, 'Blocked task', { + previousStatus: 'pending', + nextStatus: 'in_progress', + }), + ), + row(3, blocked), + ], + [createCorrelation, updateCorrelation], + ); + assert.equal(lookups[0]?.kind, 'found'); + assert.deepEqual( + lookups[0]?.kind === 'found' + ? lookups[0].presentation.changes.map(({ key, subject }) => ({ key, subject })) + : [], + [ + { key: 'T1', subject: 'First' }, + { key: 'T2', subject: 'Second' }, + ], + ); + assert.equal( + lookups[1]?.kind === 'found' ? lookups[1].presentation.changes[0]?.reason : undefined, + 'Waiting for [redacted]', + ); + }); + + test('returns typed unresolved results without guessing from another event', () => { + const mismatch = event('event-1', 'task_blocked', updateCorrelation, 'Blocked task', { + previousStatus: 'in_progress', + nextStatus: 'blocked', + reason: 'top-level reason', + }); + mismatch.task.blockedReason = 'different snapshot reason'; + const missing = { turnId: 'turn-missing', toolCallId: 'call-missing' }; + assert.deepEqual(projectTaskMutationLookups([row(0, mismatch)], [updateCorrelation, missing]), [ + { kind: 'incompatible', correlation: updateCorrelation }, + { kind: 'not_found', correlation: missing }, + ]); + }); + + test('rejects non-contiguous reuse of one correlation', () => { + const unrelated = { turnId: 'turn-other', toolCallId: 'call-other' }; + const rows = [ + row(0, event('event-1', 'task_created', createCorrelation, 'First')), + row(1, event('event-2', 'task_created', unrelated, 'Other')), + row(2, event('event-3', 'task_created', createCorrelation, 'Second', { taskIndex: 2 })), + ]; + assert.deepEqual(projectTaskMutationLookups(rows, [createCorrelation]), [ + { kind: 'incompatible', correlation: createCorrelation }, + ]); + }); + + test('rejects non-canonical create keys and update transitions', () => { + const duplicateKeyRows = [ + row(0, event('event-1', 'task_created', createCorrelation, 'First')), + row( + 1, + event('event-2', 'task_created', createCorrelation, 'Second', { + taskIndex: 2, + taskKey: 'T1', + }), + ), + ]; + assert.deepEqual(projectTaskMutationLookups(duplicateKeyRows, [createCorrelation]), [ + { kind: 'incompatible', correlation: createCorrelation }, + ]); + + for (const updateEvent of [ + event('event-3', 'task_completed', updateCorrelation, 'Skipped', { + previousStatus: 'pending', + nextStatus: 'completed', + evidence: 'Done', + }), + event('event-4', 'task_blocked', updateCorrelation, 'Mismatched type', { + previousStatus: 'pending', + nextStatus: 'in_progress', + }), + ]) { + assert.deepEqual(projectTaskMutationLookups([row(0, updateEvent)], [updateCorrelation]), [ + { kind: 'incompatible', correlation: updateCorrelation }, + ]); + } + }); + + test('rejects mutations that conflict with the canonical global replay', () => { + const firstCreate = { turnId: 'turn-first', toolCallId: 'call-first' }; + const secondCreate = { turnId: 'turn-second', toolCallId: 'call-second' }; + const falsePrevious = { turnId: 'turn-false', toolCallId: 'call-false' }; + const conflictingStatusRows = [ + row(0, event('event-1', 'task_created', firstCreate, 'Pending task')), + row( + 1, + event('event-2', 'task_completed', falsePrevious, 'False previous status', { + previousStatus: 'in_progress', + nextStatus: 'completed', + evidence: 'Done', + }), + ), + ]; + assert.deepEqual(projectTaskMutationLookups(conflictingStatusRows, [falsePrevious]), [ + { kind: 'incompatible', correlation: falsePrevious }, + ]); + + const duplicateGlobalKeyRows = [ + row(0, event('event-3', 'task_created', firstCreate, 'First')), + row( + 1, + event('event-4', 'task_created', secondCreate, 'Second', { + taskIndex: 2, + taskKey: 'T1', + }), + ), + ]; + assert.deepEqual(projectTaskMutationLookups(duplicateGlobalKeyRows, [secondCreate]), [ + { kind: 'incompatible', correlation: secondCreate }, + ]); + + const unknownUpdate = event('event-5', 'task_started', updateCorrelation, 'Unknown task', { + previousStatus: 'pending', + nextStatus: 'in_progress', + }); + assert.deepEqual(projectTaskMutationLookups([row(0, unknownUpdate)], [updateCorrelation]), [ + { kind: 'incompatible', correlation: updateCorrelation }, + ]); + }); +}); + +function row(sequence: number, ledgerEvent: TaskLedgerEvent): SequencedTaskLedgerEvent { + return { sequence, event: ledgerEvent }; +} + +function event( + eventId: string, + type: TaskLedgerEvent['type'], + correlation: { turnId: string; toolCallId: string }, + subject: string, + options: { + previousStatus?: TaskLedgerEvent['previousStatus']; + nextStatus?: TaskLedgerEvent['nextStatus']; + reason?: string; + evidence?: string; + taskIndex?: number; + taskKey?: string; + } = {}, +): TaskLedgerEvent { + const taskIndex = options.taskIndex ?? 1; + const nextStatus = options.nextStatus ?? 'pending'; + const task = { + id: `task-${taskIndex}`, + key: options.taskKey ?? `T${taskIndex}`, + subject, + status: nextStatus, + createdAt: 1, + updatedAt: 2, + ...(nextStatus === 'blocked' && options.reason ? { blockedReason: options.reason } : {}), + ...(nextStatus === 'failed' && options.reason ? { failureReason: options.reason } : {}), + ...(nextStatus === 'completed' && options.evidence + ? { completionEvidence: options.evidence } + : {}), + }; + return { + eventId, + type, + ts: 2, + sessionId: 'session-1', + taskId: task.id, + ...(options.previousStatus ? { previousStatus: options.previousStatus } : {}), + nextStatus, + task, + ...(options.reason ? { reason: options.reason } : {}), + ...(options.evidence ? { evidence: options.evidence } : {}), + refs: { runId: 'run-1', ...correlation }, + source: 'tool', + actor: 'main_agent', + }; +} diff --git a/packages/runtime-host/src/__tests__/task-mutation-protocol.test.ts b/packages/runtime-host/src/__tests__/task-mutation-protocol.test.ts new file mode 100644 index 0000000000..1c57066c33 --- /dev/null +++ b/packages/runtime-host/src/__tests__/task-mutation-protocol.test.ts @@ -0,0 +1,380 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { RuntimeHostProtocolError } from '../protocol/errors.js'; +import { + decodeTaskMutationQueryInput, + decodeTaskMutationQueryResult, + encodeTaskMutationQueryResult, + TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES, + TASK_MUTATION_PAGE_MAX_BYTES, + TASK_MUTATION_TOOL_CALL_ID_MAX_ENCODED_BYTES, + taskMutationCorrelationsEncodedByteLength, + type TaskMutationChange, + type TaskMutationCorrelation, + type TaskMutationQueryResult, +} from '../protocol/task-mutation.js'; + +const revision = `sha256:${'a'.repeat(64)}` as const; +const correlation = { turnId: 'turn-1', toolCallId: 'call-1' } as const; + +describe('Task mutation protocol', () => { + test('requires exact, unique correlations on start and continuation', () => { + assert.deepEqual( + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations: [correlation], + }), + { + kind: 'start', + sessionId: 'session-1', + correlations: [correlation], + }, + ); + assertInvalid(() => + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations: [correlation, correlation], + }), + ); + assertInvalid(() => + decodeTaskMutationQueryInput({ + kind: 'continue', + sessionId: 'session-1', + correlations: [correlation], + revision, + cursor: 'opaque', + offset: 1, + }), + ); + }); + + test('round-trips opaque nested Code Mode tool-call identities', () => { + const nestedCreate = { + turnId: 'turn-create', + toolCallId: 'provider-call:nested:123e4567-e89b-12d3-a456-426614174000', + } as const; + const nestedUpdate = { + turnId: 'turn-update', + toolCallId: 'provider-call:nested:223e4567-e89b-12d3-a456-426614174000', + } as const; + assert.deepEqual( + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations: [nestedCreate, nestedUpdate], + }), + { + kind: 'start', + sessionId: 'session-1', + correlations: [nestedCreate, nestedUpdate], + }, + ); + assert.deepEqual( + decodeTaskMutationQueryInput({ + kind: 'continue', + sessionId: 'session-1', + correlations: [nestedCreate, nestedUpdate], + revision, + cursor: 'opaque', + }), + { + kind: 'continue', + sessionId: 'session-1', + correlations: [nestedCreate, nestedUpdate], + revision, + cursor: 'opaque', + }, + ); + const result = encodeTaskMutationQueryResult({ + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [found(nestedCreate, [change(1)]), { kind: 'not_found', correlation: nestedUpdate }], + nextCursor: null, + }); + assert.deepEqual(decodeTaskMutationQueryResult(result), result); + + assert.equal( + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations: [{ turnId: 'turn-1', toolCallId: 'x'.repeat(129) }], + }).correlations[0]?.toolCallId, + 'x'.repeat(129), + ); + for (const toolCallId of ['', '\0'.repeat(342)]) { + assertInvalid(() => + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations: [{ turnId: 'turn-1', toolCallId }], + }), + ); + } + assertInvalid(() => + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations: [{ turnId: 'turn:still-strict', toolCallId: nestedCreate.toolCallId }], + }), + ); + }); + + test('bounds legacy opaque ids and aggregate correlations by encoded bytes', () => { + const escapedBoundary = '\0'.repeat(341); + assert.equal(Buffer.byteLength(JSON.stringify(escapedBoundary), 'utf8'), 2048); + assert.equal(TASK_MUTATION_TOOL_CALL_ID_MAX_ENCODED_BYTES, 2048); + assert.equal( + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations: [{ turnId: 'turn-1', toolCallId: escapedBoundary }], + }).correlations[0]?.toolCallId, + escapedBoundary, + ); + + const correlations = Array.from({ length: 128 }, (_, index) => ({ + turnId: `turn-${index}`, + toolCallId: `${index}-${'x'.repeat(1_800)}`, + })); + assert.ok( + taskMutationCorrelationsEncodedByteLength(correlations) > + TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES, + ); + assertInvalid(() => + decodeTaskMutationQueryInput({ + kind: 'start', + sessionId: 'session-1', + correlations, + }), + ); + }); + + test('round-trips found, unresolved, and history-changed results', () => { + const result: TaskMutationQueryResult = { + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [ + found(correlation, [change(1)]), + { + kind: 'not_found', + correlation: { turnId: 'turn-2', toolCallId: 'call-2' }, + }, + { + kind: 'incompatible', + correlation: { turnId: 'turn-3', toolCallId: 'call-3' }, + }, + ], + nextCursor: 'opaque', + }; + const encoded = encodeTaskMutationQueryResult(result); + assert.deepEqual(decodeTaskMutationQueryResult(encoded), encoded); + assert.deepEqual( + decodeTaskMutationQueryResult({ + kind: 'history_changed', + expected: revision, + actual: `sha256:${'b'.repeat(64)}`, + }), + { + kind: 'history_changed', + expected: revision, + actual: `sha256:${'b'.repeat(64)}`, + }, + ); + }); + + test('sanitizes producer text once and rejects non-canonical wire text', () => { + const result = encodeTaskMutationQueryResult({ + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [ + found( + correlation, + [ + change(1, { + subject: 'Inspect ghp_abcdefghijklmnopqrstuvwxyz123456', + previousStatus: 'in_progress', + nextStatus: 'completed', + evidence: 'Verified ghp_abcdefghijklmnopqrstuvwxyz123456', + }), + ], + 'update', + ), + ], + nextCursor: null, + }); + assert.equal(result.kind, 'page'); + if (result.kind !== 'page' || result.lookups[0]?.kind !== 'found') { + throw new Error('Expected encoded Task mutation'); + } + const projected = result.lookups[0].presentation.changes[0]; + assert.equal(projected?.subject, 'Inspect [redacted]'); + assert.equal(projected?.evidence, 'Verified [redacted]'); + assert.deepEqual(decodeTaskMutationQueryResult(result), result); + + assertInvalid(() => + decodeTaskMutationQueryResult({ + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [ + found(correlation, [ + change(1, { + subject: 'Inspect ghp_abcdefghijklmnopqrstuvwxyz123456', + }), + ]), + ], + nextCursor: null, + }), + ); + }); + + test('rejects operation-incompatible and duplicate forged changes', () => { + for (const changes of [ + [change(1, { previousStatus: 'in_progress' })], + [change(1, { nextStatus: 'completed', evidence: 'Done' })], + [change(1), change(1)], + [change(1), change(2, { key: 'T1' })], + ]) { + assertInvalid(() => + decodeTaskMutationQueryResult({ + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [found(correlation, changes)], + nextCursor: null, + }), + ); + } + assertInvalid(() => + decodeTaskMutationQueryResult({ + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [found(correlation, [change(1)], 'update')], + nextCursor: null, + }), + ); + assertInvalid(() => + decodeTaskMutationQueryResult({ + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [ + found( + correlation, + [ + change(1, { + previousStatus: 'pending', + nextStatus: 'completed', + evidence: 'Done', + }), + ], + 'update', + ), + ], + nextCursor: null, + }), + ); + }); + + test('fits one maximum legal create without splitting its presentation', () => { + const changes = Array.from({ length: 200 }, (_, index) => { + const taskId = `t-${index}-${'x.'.repeat(64)}`.slice(0, 64); + const taskNumber = String(index + 1); + const key = `T${taskNumber}.${'1'.repeat(62 - taskNumber.length)}`; + return change(index + 1, { + taskId, + key, + subject: '😀'.repeat(200), + }); + }); + const encoded = encodeTaskMutationQueryResult({ + kind: 'page', + sessionId: 'session-1', + revision, + lookups: [found(correlation, changes)], + nextCursor: null, + }); + assert.ok(Buffer.byteLength(JSON.stringify(encoded), 'utf8') < TASK_MUTATION_PAGE_MAX_BYTES); + assert.deepEqual(decodeTaskMutationQueryResult(encoded), encoded); + }); + + test('rejects a page whose complete presentations exceed the byte budget', () => { + const lookups = Array.from({ length: 128 }, (_, index) => { + const itemCorrelation = { turnId: `turn-${index}`, toolCallId: `call-${index}` }; + return found( + itemCorrelation, + [ + change(index + 1, { + previousStatus: 'in_progress', + nextStatus: 'completed', + evidence: '😀'.repeat(1000), + }), + ], + 'update', + ); + }); + const oversized = { + kind: 'page', + sessionId: 'session-1', + revision, + lookups, + nextCursor: null, + }; + assert.ok(Buffer.byteLength(JSON.stringify(oversized), 'utf8') > TASK_MUTATION_PAGE_MAX_BYTES); + assertInvalid(() => encodeTaskMutationQueryResult(oversized)); + }); +}); + +function found( + itemCorrelation: TaskMutationCorrelation, + changes: readonly TaskMutationChange[], + operation: 'create' | 'update' = 'create', +) { + return { + kind: 'found' as const, + correlation: itemCorrelation, + presentation: { operation, correlation: itemCorrelation, changes }, + }; +} + +function change(index: number, overrides: Partial = {}): TaskMutationChange { + return { + taskId: `task-${index}`, + key: `T${index}`, + subject: `Task ${index}`, + nextStatus: 'pending', + ...overrides, + }; +} + +function assertInvalid(action: () => unknown): void { + assert.throws( + action, + (error: unknown) => error instanceof RuntimeHostProtocolError && error.code === 'invalid_frame', + ); +} diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 0092796aee..50def9c63c 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -87,6 +87,7 @@ export * from './session-retirement.js'; export * from './session-transcript.js'; export * from './session-turns.js'; export * from './task-ledger.js'; +export * from './task-mutation.js'; export * from './workspace.js'; export * from './workhub-coordination.js'; export * from './websocket-path.js'; @@ -95,7 +96,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 76 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 77 as const; +// 77: Clients may query the durable, tool-call-correlated Task mutation +// projection. Older peers do not know the closed operation/result vocabulary. // 76: Peer Mesh endpoint and Mesh display names are signed, persisted facts // managed through Host operations rather than local-only Client labels. // 75: Peer Mesh routes identify whether a peer is a Client or Runtime Host so diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 8476927962..8c05b77e1a 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -60,6 +60,7 @@ import { SESSION_RETIREMENT_OPERATION_SPECS } from './session-retirement.js'; import { SESSION_EFFECT_OPERATION_SPECS } from './session-effects.js'; import { SKILL_CATALOG_OPERATION_SPECS } from './skill-catalog.js'; import { TASK_LEDGER_OPERATION_SPECS } from './task-ledger.js'; +import { TASK_MUTATION_OPERATION_SPECS } from './task-mutation.js'; import { TURN_OPERATION_SPECS } from './turn.js'; import { USAGE_PRICING_OPERATION_SPECS } from './usage-pricing.js'; import { WEB_SEARCH_OPERATION_SPECS } from './web-search.js'; @@ -200,6 +201,7 @@ export const HOST_OPERATION_SPECS = composeOperationSpecMaps( PROJECT_CATALOG_OPERATION_SPECS, MESSAGE_OPERATION_SPECS, TASK_LEDGER_OPERATION_SPECS, + TASK_MUTATION_OPERATION_SPECS, INTERACTION_OPERATION_SPECS, SESSION_CONTINUITY_OPERATION_SPECS, SESSION_TRANSCRIPT_OPERATION_SPECS, @@ -326,6 +328,7 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'subscription.close', 'subscription.open', 'task.ledger.query', + 'task.mutation.query', 'turn.interrupt', 'turn.message.execution.query', 'turn.message.query', diff --git a/packages/runtime-host/src/protocol/task-mutation.ts b/packages/runtime-host/src/protocol/task-mutation.ts new file mode 100644 index 0000000000..072c8cfea4 --- /dev/null +++ b/packages/runtime-host/src/protocol/task-mutation.ts @@ -0,0 +1,492 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { + TASK_EVIDENCE_MAX_CHARS, + TASK_SUBJECT_MAX_CHARS, + canTransitionTaskStatus, + isSafeTaskId, + isTaskKey, + isTaskStatus, + normalizeTaskEvidenceText, + normalizeTaskSubject, + sanitizeTaskLedgerTask, + type Task, + type TaskStatus, +} from '@maka/core/task-ledger'; +import { requireEntityId, requireExactRecord, requireRecord } from './codec.js'; +import { invalidProtocolFrame } from './errors.js'; +import { defineOperation } from './operation-spec.js'; + +export const TASK_MUTATION_QUERY_MAX_CORRELATIONS = 128; +export const TASK_MUTATION_PAGE_MAX_ITEMS = 128; +export const TASK_MUTATION_PAGE_MAX_BYTES = 320 * 1024; +export const TASK_MUTATION_CURSOR_MAX_BYTES = 1024; +export const TASK_MUTATION_TOOL_CALL_ID_MAX_ENCODED_BYTES = 2 * 1024; +export const TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES = 192 * 1024; +export const TASK_MUTATION_QUERY_INPUT_MAX_ENCODED_BYTES = 224 * 1024; + +const QUERY_ERRORS = [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'invalid_request', + 'not_found', + 'internal_failure', +] as const; + +export type TaskMutationRevision = `sha256:${string}`; + +export interface TaskMutationCorrelation { + readonly turnId: string; + readonly toolCallId: string; +} + +export interface TaskMutationChange { + readonly taskId: string; + readonly key: string; + readonly subject: string; + readonly previousStatus?: TaskStatus; + readonly nextStatus: TaskStatus; + readonly reason?: string; + readonly evidence?: string; +} + +export interface TaskMutationPresentation { + readonly operation: 'create' | 'update'; + readonly correlation: TaskMutationCorrelation; + readonly changes: readonly TaskMutationChange[]; +} + +export type TaskMutationLookup = + | { + readonly kind: 'found'; + readonly correlation: TaskMutationCorrelation; + readonly presentation: TaskMutationPresentation; + } + | { + readonly kind: 'not_found' | 'incompatible'; + readonly correlation: TaskMutationCorrelation; + }; + +export type TaskMutationQueryInput = + | { + readonly kind: 'start'; + readonly sessionId: string; + readonly correlations: readonly TaskMutationCorrelation[]; + } + | { + readonly kind: 'continue'; + readonly sessionId: string; + readonly correlations: readonly TaskMutationCorrelation[]; + readonly revision: TaskMutationRevision; + readonly cursor: string; + }; + +export type TaskMutationQueryResult = + | { + readonly kind: 'page'; + readonly sessionId: string; + readonly revision: TaskMutationRevision; + readonly lookups: readonly TaskMutationLookup[]; + readonly nextCursor: string | null; + } + | { + readonly kind: 'history_changed'; + readonly expected: TaskMutationRevision; + readonly actual: TaskMutationRevision; + }; + +export const TASK_MUTATION_OPERATION_SPECS = { + 'task.mutation.query': defineOperation< + TaskMutationQueryInput, + TaskMutationQueryResult, + (typeof QUERY_ERRORS)[number] + >({ + mode: 'query', + availability: 'ready', + errors: QUERY_ERRORS, + decodeInput: decodeTaskMutationQueryInput, + decodeOutput: decodeTaskMutationQueryResult, + }), +} as const; + +export function decodeTaskMutationQueryInput(value: unknown): TaskMutationQueryInput { + const record = requireRecord(value, 'task mutation query input'); + if (record.kind === 'start') { + const input = requireExactRecord(record, 'task mutation query start input', [ + 'kind', + 'sessionId', + 'correlations', + ]); + return boundedTaskMutationQueryInput({ + kind: 'start', + sessionId: requireEntityId(input.sessionId, 'sessionId'), + correlations: taskMutationCorrelations(input.correlations), + }); + } + if (record.kind === 'continue') { + const input = requireExactRecord(record, 'task mutation query continuation input', [ + 'kind', + 'sessionId', + 'correlations', + 'revision', + 'cursor', + ]); + return boundedTaskMutationQueryInput({ + kind: 'continue', + sessionId: requireEntityId(input.sessionId, 'sessionId'), + correlations: taskMutationCorrelations(input.correlations), + revision: taskMutationRevision(input.revision, 'task mutation revision'), + cursor: taskMutationCursor(input.cursor), + }); + } + throw invalidProtocolFrame('Invalid task mutation query kind'); +} + +export function decodeTaskMutationQueryResult(value: unknown): TaskMutationQueryResult { + return taskMutationQueryResult(value, 'decode'); +} + +export function encodeTaskMutationQueryResult(value: unknown): TaskMutationQueryResult { + return taskMutationQueryResult(value, 'encode'); +} + +function taskMutationQueryResult( + value: unknown, + direction: 'encode' | 'decode', +): TaskMutationQueryResult { + const record = requireRecord(value, 'task mutation query result'); + if (record.kind === 'history_changed') { + const changed = requireExactRecord(record, 'task mutation history changed result', [ + 'kind', + 'expected', + 'actual', + ]); + return { + kind: 'history_changed', + expected: taskMutationRevision(changed.expected, 'expected task mutation revision'), + actual: taskMutationRevision(changed.actual, 'actual task mutation revision'), + }; + } + if (record.kind !== 'page') throw invalidProtocolFrame('Invalid task mutation query result kind'); + const page = requireExactRecord(record, 'task mutation query page', [ + 'kind', + 'sessionId', + 'revision', + 'lookups', + 'nextCursor', + ]); + if (!Array.isArray(page.lookups) || page.lookups.length > TASK_MUTATION_PAGE_MAX_ITEMS) { + throw invalidProtocolFrame('Task mutation page exceeds item limit'); + } + const decoded: TaskMutationQueryResult = { + kind: 'page', + sessionId: requireEntityId(page.sessionId, 'sessionId'), + revision: taskMutationRevision(page.revision, 'task mutation revision'), + lookups: page.lookups.map((lookup) => taskMutationLookup(lookup, direction)), + nextCursor: page.nextCursor === null ? null : taskMutationCursor(page.nextCursor), + }; + if (jsonByteLength(decoded) > TASK_MUTATION_PAGE_MAX_BYTES) { + throw invalidProtocolFrame('Task mutation page exceeds byte limit'); + } + return decoded; +} + +function taskMutationCorrelations(value: unknown): readonly TaskMutationCorrelation[] { + if ( + !Array.isArray(value) || + value.length === 0 || + value.length > TASK_MUTATION_QUERY_MAX_CORRELATIONS + ) { + throw invalidProtocolFrame('Invalid task mutation correlations'); + } + const correlations = value.map(taskMutationCorrelation); + if ( + taskMutationCorrelationsEncodedByteLength(correlations) > + TASK_MUTATION_CORRELATIONS_MAX_ENCODED_BYTES + ) { + throw invalidProtocolFrame('Task mutation correlations exceed byte limit'); + } + const unique = new Set(correlations.map(correlationKey)); + if (unique.size !== correlations.length) { + throw invalidProtocolFrame('Duplicate task mutation correlation'); + } + return correlations; +} + +function taskMutationCorrelation(value: unknown): TaskMutationCorrelation { + const correlation = requireExactRecord(value, 'task mutation correlation', [ + 'turnId', + 'toolCallId', + ]); + if ( + typeof correlation.toolCallId !== 'string' || + correlation.toolCallId.length === 0 || + jsonByteLength(correlation.toolCallId) > TASK_MUTATION_TOOL_CALL_ID_MAX_ENCODED_BYTES + ) { + throw invalidProtocolFrame('Invalid toolCallId'); + } + return { + turnId: requireEntityId(correlation.turnId, 'turnId'), + toolCallId: correlation.toolCallId, + }; +} + +export function taskMutationCorrelationsEncodedByteLength( + correlations: readonly TaskMutationCorrelation[], +): number { + return jsonByteLength(correlations); +} + +function boundedTaskMutationQueryInput(input: TaskMutationQueryInput): TaskMutationQueryInput { + if (jsonByteLength(input) > TASK_MUTATION_QUERY_INPUT_MAX_ENCODED_BYTES) { + throw invalidProtocolFrame('Task mutation query input exceeds byte limit'); + } + return input; +} + +function taskMutationLookup(value: unknown, direction: 'encode' | 'decode'): TaskMutationLookup { + const record = requireRecord(value, 'task mutation lookup'); + if (record.kind === 'not_found' || record.kind === 'incompatible') { + const lookup = requireExactRecord(record, 'task mutation unresolved lookup', [ + 'kind', + 'correlation', + ]); + return { kind: record.kind, correlation: taskMutationCorrelation(lookup.correlation) }; + } + if (record.kind !== 'found') throw invalidProtocolFrame('Invalid task mutation lookup kind'); + const lookup = requireExactRecord(record, 'task mutation found lookup', [ + 'kind', + 'correlation', + 'presentation', + ]); + const correlation = taskMutationCorrelation(lookup.correlation); + const presentation = taskMutationPresentation(lookup.presentation, direction); + if (correlationKey(correlation) !== correlationKey(presentation.correlation)) { + throw invalidProtocolFrame('Task mutation lookup correlation mismatch'); + } + return { kind: 'found', correlation, presentation }; +} + +function taskMutationPresentation( + value: unknown, + direction: 'encode' | 'decode', +): TaskMutationPresentation { + const record = requireExactRecord(value, 'task mutation presentation', [ + 'operation', + 'correlation', + 'changes', + ]); + if (record.operation !== 'create' && record.operation !== 'update') { + throw invalidProtocolFrame('Invalid task mutation operation'); + } + if ( + !Array.isArray(record.changes) || + record.changes.length === 0 || + record.changes.length > 200 || + (record.operation === 'update' && record.changes.length !== 1) + ) { + throw invalidProtocolFrame('Invalid task mutation changes'); + } + const changes = record.changes.map((change) => taskMutationChange(change, direction)); + if (record.operation === 'create') { + const taskIds = new Set(); + const taskKeys = new Set(); + for (const change of changes) { + if ( + change.previousStatus !== undefined || + change.nextStatus !== 'pending' || + change.reason !== undefined || + change.evidence !== undefined || + taskIds.has(change.taskId) || + taskKeys.has(change.key) + ) { + throw invalidProtocolFrame('Invalid create task mutation changes'); + } + taskIds.add(change.taskId); + taskKeys.add(change.key); + } + } else { + const change = changes[0]; + if ( + change?.previousStatus === undefined || + !canTransitionTaskStatus(change.previousStatus, change.nextStatus, { explicitReopen: true }) + ) { + throw invalidProtocolFrame('Invalid update task mutation change'); + } + } + return { + operation: record.operation, + correlation: taskMutationCorrelation(record.correlation), + changes, + }; +} + +function taskMutationChange(value: unknown, direction: 'encode' | 'decode'): TaskMutationChange { + const record = requireRecord(value, 'task mutation change'); + assertAllowedKeys(record, 'task mutation change', [ + 'taskId', + 'key', + 'subject', + 'previousStatus', + 'nextStatus', + 'reason', + 'evidence', + ]); + for (const field of ['taskId', 'key', 'subject', 'nextStatus'] as const) { + if (!Object.hasOwn(record, field)) throw invalidProtocolFrame('Invalid task mutation fields'); + } + if (!isSafeTaskId(record.taskId)) throw invalidProtocolFrame('Invalid task mutation taskId'); + if (!isTaskKey(record.key)) throw invalidProtocolFrame('Invalid task mutation task key'); + if (!isTaskStatus(record.nextStatus)) throw invalidProtocolFrame('Invalid task mutation status'); + if (record.previousStatus !== undefined && !isTaskStatus(record.previousStatus)) { + throw invalidProtocolFrame('Invalid previous task mutation status'); + } + const subject = canonicalSubject(record.subject, direction); + const reason = optionalDetail(record.reason, record.nextStatus, 'reason', direction); + const evidence = optionalDetail(record.evidence, record.nextStatus, 'evidence', direction); + if ( + ((record.nextStatus === 'blocked' || record.nextStatus === 'failed') && !reason) || + (record.nextStatus === 'completed' && !evidence) + ) { + throw invalidProtocolFrame('Task mutation status requires exact detail'); + } + return { + taskId: record.taskId, + key: record.key, + subject, + ...(record.previousStatus !== undefined ? { previousStatus: record.previousStatus } : {}), + nextStatus: record.nextStatus, + ...(reason !== undefined ? { reason } : {}), + ...(evidence !== undefined ? { evidence } : {}), + }; +} + +function canonicalSubject(value: unknown, direction: 'encode' | 'decode'): string { + if (typeof value !== 'string' || Array.from(value).length > TASK_SUBJECT_MAX_CHARS) { + throw invalidProtocolFrame('Invalid task mutation subject'); + } + const sanitized = sanitizeTaskLedgerText(value, 'subject'); + const normalized = normalizeTaskSubject(sanitized); + const canonical = normalized.ok + ? normalized.value + : sanitized.trim().length === 0 + ? '[redacted]' + : null; + if (canonical === null || (direction === 'decode' && canonical !== value)) { + throw invalidProtocolFrame('Task mutation subject is not sanitized'); + } + return canonical; +} + +function optionalDetail( + value: unknown, + status: TaskStatus, + kind: 'reason' | 'evidence', + direction: 'encode' | 'decode', +): string | undefined { + if (value === undefined) return undefined; + if (typeof value !== 'string' || Array.from(value).length > TASK_EVIDENCE_MAX_CHARS) { + throw invalidProtocolFrame(`Invalid task mutation ${kind}`); + } + if ( + (kind === 'reason' && status !== 'blocked' && status !== 'failed') || + (kind === 'evidence' && status !== 'completed') + ) { + throw invalidProtocolFrame(`Task mutation ${kind} is incompatible with status`); + } + const field = + kind === 'evidence' + ? 'completionEvidence' + : status === 'blocked' + ? 'blockedReason' + : 'failureReason'; + const sanitized = sanitizeTaskLedgerText(value, field); + const normalized = normalizeTaskEvidenceText(sanitized, field); + const canonical = normalized.ok + ? normalized.value + : sanitized.trim().length === 0 + ? undefined + : null; + if (canonical === null || (direction === 'decode' && canonical !== value)) { + throw invalidProtocolFrame(`Task mutation ${kind} is not sanitized`); + } + return canonical; +} + +function sanitizeTaskLedgerText( + value: string, + field: 'subject' | 'blockedReason' | 'failureReason' | 'completionEvidence', +): string { + const task: Task = { + id: 'task-mutation-wire-sanitizer', + key: 'T1', + subject: field === 'subject' ? value : 'Task mutation', + status: + field === 'blockedReason' + ? 'blocked' + : field === 'failureReason' + ? 'failed' + : field === 'completionEvidence' + ? 'completed' + : 'pending', + createdAt: 0, + updatedAt: 0, + ...(field !== 'subject' ? { [field]: value } : {}), + }; + return sanitizeTaskLedgerTask(task)[field] ?? ''; +} + +function taskMutationRevision(value: unknown, label: string): TaskMutationRevision { + if (typeof value !== 'string' || !/^sha256:[a-f0-9]{64}$/.test(value)) { + throw invalidProtocolFrame(`Invalid ${label}`); + } + return value as TaskMutationRevision; +} + +function taskMutationCursor(value: unknown): string { + if ( + typeof value !== 'string' || + value.length === 0 || + Buffer.byteLength(value, 'utf8') > TASK_MUTATION_CURSOR_MAX_BYTES + ) { + throw invalidProtocolFrame('Invalid task mutation cursor'); + } + return value; +} + +function correlationKey(correlation: TaskMutationCorrelation): string { + return JSON.stringify([correlation.turnId, correlation.toolCallId]); +} + +function assertAllowedKeys( + record: Record, + label: string, + keys: readonly string[], +): void { + const allowed = new Set(keys); + if (Object.keys(record).some((key) => !allowed.has(key))) { + throw invalidProtocolFrame(`Unknown ${label} field`); + } +} + +function jsonByteLength(value: unknown): number { + return Buffer.byteLength(JSON.stringify(value), 'utf8'); +} diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index 18bf74b3c1..36736957e8 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -135,7 +135,10 @@ export type SessionCatalogOperationKey = Exclude< | SessionRetirementOperationKey | SessionEffectOperationKey >; -export type TaskLedgerOperationKey = Extract; +export type TaskLedgerOperationKey = Extract< + OperationKey, + 'task.ledger.query' | 'task.mutation.query' +>; export type ArtifactOperationKey = Extract; export type SkillCatalogOperationKey = Extract; export type UsagePricingOperationKey = Extract; diff --git a/packages/runtime-host/src/server/task-ledger-coordinator.ts b/packages/runtime-host/src/server/task-ledger-coordinator.ts index e447fd995a..4da8b1a31d 100644 --- a/packages/runtime-host/src/server/task-ledger-coordinator.ts +++ b/packages/runtime-host/src/server/task-ledger-coordinator.ts @@ -36,17 +36,26 @@ import { import { encodeTaskLedgerTask, encodeTaskLedgerQueryResult, + encodeTaskMutationQueryResult, TASK_LEDGER_PAGE_MAX_BYTES, TASK_LEDGER_PAGE_MAX_ITEMS, + TASK_MUTATION_PAGE_MAX_BYTES, + TASK_MUTATION_PAGE_MAX_ITEMS, type OperationOutcome, type TaskLedgerQueryInput, type TaskLedgerQueryResult, type TaskLedgerRevision, type TaskLedgerTask, + type TaskMutationCorrelation, + type TaskMutationLookup, + type TaskMutationQueryInput, + type TaskMutationQueryResult, + type TaskMutationRevision, } from '../protocol/index.js'; import type { TaskLedgerOperationHandlerMap } from './operation-dispatcher.js'; import { SessionAdmissionGate } from './session-admission-gate.js'; import type { SessionPresenceReader } from './session-presence.js'; +import { projectTaskMutationLookups } from './task-mutation-projection.js'; const CANONICAL_LIST_OPTIONS = Object.freeze({ includeTerminal: true, @@ -58,6 +67,7 @@ const CANONICAL_LIST_OPTIONS = Object.freeze({ export class HostTaskLedgerCoordinator implements TaskLedgerStore { readonly handlers: TaskLedgerOperationHandlerMap = { 'task.ledger.query': (input) => this.#query(input), + 'task.mutation.query': (input) => this.#queryMutations(input), }; readonly #writer: InteractiveTaskLedgerWriter; @@ -177,6 +187,61 @@ export class HostTaskLedgerCoordinator implements TaskLedgerStore { return success(createPage(input.sessionId, revision, tasks, offset)); }); } + + #queryMutations(input: TaskMutationQueryInput): Promise> { + return this.sessionAdmission.run(input.sessionId, async () => { + if ((await this.sessions.probeSessionRemoval(input.sessionId)).kind !== 'present') { + return mutationNotFound('Session was not found'); + } + const correlationDigest = taskMutationCorrelationDigest(input.correlations); + const cursor = input.kind === 'continue' ? decodeTaskMutationCursor(input.cursor) : undefined; + if ( + input.kind === 'continue' && + (!cursor || + cursor.sessionId !== input.sessionId || + cursor.correlationDigest !== correlationDigest || + input.revision !== taskMutationRevisionFromCursor(cursor)) + ) { + return mutationInvalidRequest('Task mutation cursor is invalid'); + } + + const currentRows = await this.#writer.readSequencedEvents(input.sessionId); + const currentWatermark = taskMutationWatermark(currentRows); + if (input.kind === 'continue' && cursor) { + const frozenRow = currentRows[cursor.throughSequence]; + if (!frozenRow || frozenRow.event.eventId !== cursor.throughEventId) { + return mutationSuccess({ + kind: 'history_changed', + expected: input.revision, + actual: taskMutationRevision(input.sessionId, correlationDigest, currentWatermark), + }); + } + } + + const watermark = cursor + ? { sequence: cursor.throughSequence, eventId: cursor.throughEventId } + : currentWatermark; + const rows = watermark + ? currentRows.filter(({ sequence }) => sequence <= watermark.sequence) + : []; + const revision = taskMutationRevision(input.sessionId, correlationDigest, watermark); + const lookups = projectTaskMutationLookups(rows, input.correlations); + const offset = cursor?.offset ?? 0; + if (offset > lookups.length || (offset === lookups.length && offset !== 0)) { + return mutationInvalidRequest('Task mutation cursor is invalid'); + } + return mutationSuccess( + createTaskMutationPage( + input.sessionId, + revision, + correlationDigest, + watermark, + lookups, + offset, + ), + ); + }); + } } function taskLedgerRevision(tasks: readonly TaskLedgerTask[]): TaskLedgerRevision { @@ -247,3 +312,182 @@ function notFound(message: string): OperationOutcome<'task.ledger.query'> { function invariantFailure(message: string): Error { return new Error(`Task ledger coordinator invariant failed: ${message}`); } + +interface TaskMutationWatermark { + readonly sequence: number; + readonly eventId: string; +} + +interface TaskMutationCursorPayload { + readonly version: 1; + readonly sessionId: string; + readonly correlationDigest: string; + readonly throughSequence: number; + readonly throughEventId: string; + readonly offset: number; + readonly checksum: string; +} + +type TaskMutationCursorContent = Omit; + +function taskMutationWatermark( + rows: readonly { sequence: number; event: { eventId: string } }[], +): TaskMutationWatermark | undefined { + const row = rows.at(-1); + return row ? { sequence: row.sequence, eventId: row.event.eventId } : undefined; +} + +function taskMutationCorrelationDigest(correlations: readonly TaskMutationCorrelation[]): string { + return createHash('sha256').update(JSON.stringify(correlations)).digest('hex'); +} + +function taskMutationRevision( + sessionId: string, + correlationDigest: string, + watermark: TaskMutationWatermark | undefined, +): TaskMutationRevision { + return `sha256:${createHash('sha256') + .update(JSON.stringify([sessionId, correlationDigest, watermark ?? null])) + .digest('hex')}`; +} + +function taskMutationRevisionFromCursor(cursor: TaskMutationCursorPayload): TaskMutationRevision { + return taskMutationRevision(cursor.sessionId, cursor.correlationDigest, { + sequence: cursor.throughSequence, + eventId: cursor.throughEventId, + }); +} + +function createTaskMutationPage( + sessionId: string, + revision: TaskMutationRevision, + correlationDigest: string, + watermark: TaskMutationWatermark | undefined, + lookups: readonly TaskMutationLookup[], + offset: number, +): TaskMutationQueryResult { + const pageLookups: TaskMutationLookup[] = []; + for (let index = offset; index < lookups.length; index += 1) { + if (pageLookups.length >= TASK_MUTATION_PAGE_MAX_ITEMS) break; + const lookup = lookups[index]; + if (!lookup) throw invariantFailure('Task mutation lookup index was out of bounds'); + const candidateLookups = [...pageLookups, lookup]; + const nextOffset = index + 1; + const nextCursor = + nextOffset < lookups.length && watermark + ? encodeTaskMutationCursor({ + version: 1, + sessionId, + correlationDigest, + throughSequence: watermark.sequence, + throughEventId: watermark.eventId, + offset: nextOffset, + }) + : null; + const candidate = { + kind: 'page' as const, + sessionId, + revision, + lookups: candidateLookups, + nextCursor, + }; + if (Buffer.byteLength(JSON.stringify(candidate), 'utf8') > TASK_MUTATION_PAGE_MAX_BYTES) break; + pageLookups.push(lookup); + } + + if (pageLookups.length === 0 && offset < lookups.length) { + throw invariantFailure('A canonical Task mutation exceeded the page result byte limit'); + } + const nextOffset = offset + pageLookups.length; + const nextCursor = + nextOffset < lookups.length && watermark + ? encodeTaskMutationCursor({ + version: 1, + sessionId, + correlationDigest, + throughSequence: watermark.sequence, + throughEventId: watermark.eventId, + offset: nextOffset, + }) + : null; + return encodeTaskMutationQueryResult({ + kind: 'page', + sessionId, + revision, + lookups: pageLookups, + nextCursor, + }); +} + +function encodeTaskMutationCursor(cursor: TaskMutationCursorContent): string { + const content: TaskMutationCursorContent = { + version: cursor.version, + sessionId: cursor.sessionId, + correlationDigest: cursor.correlationDigest, + throughSequence: cursor.throughSequence, + throughEventId: cursor.throughEventId, + offset: cursor.offset, + }; + return Buffer.from( + JSON.stringify({ ...content, checksum: taskMutationCursorChecksum(content) }), + 'utf8', + ).toString('base64url'); +} + +function decodeTaskMutationCursor(cursor: string): TaskMutationCursorPayload | undefined { + try { + const parsed = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')) as Record< + string, + unknown + >; + if ( + Object.keys(parsed).length !== 7 || + parsed.version !== 1 || + typeof parsed.sessionId !== 'string' || + typeof parsed.correlationDigest !== 'string' || + !/^[a-f0-9]{64}$/.test(parsed.correlationDigest) || + typeof parsed.throughSequence !== 'number' || + !Number.isSafeInteger(parsed.throughSequence) || + parsed.throughSequence < 0 || + typeof parsed.throughEventId !== 'string' || + typeof parsed.offset !== 'number' || + !Number.isSafeInteger(parsed.offset) || + parsed.offset <= 0 || + typeof parsed.checksum !== 'string' || + !/^[a-f0-9]{64}$/.test(parsed.checksum) + ) { + return undefined; + } + const payload = parsed as unknown as TaskMutationCursorPayload; + const content: TaskMutationCursorContent = { + version: payload.version, + sessionId: payload.sessionId, + correlationDigest: payload.correlationDigest, + throughSequence: payload.throughSequence, + throughEventId: payload.throughEventId, + offset: payload.offset, + }; + return payload.checksum === taskMutationCursorChecksum(content) ? payload : undefined; + } catch { + return undefined; + } +} + +function taskMutationCursorChecksum(content: TaskMutationCursorContent): string { + return createHash('sha256') + .update('maka.task-mutation-cursor.v1\0') + .update(JSON.stringify(content)) + .digest('hex'); +} + +function mutationSuccess(result: TaskMutationQueryResult): OperationOutcome<'task.mutation.query'> { + return { ok: true, result }; +} + +function mutationInvalidRequest(message: string): OperationOutcome<'task.mutation.query'> { + return { ok: false, error: { code: 'invalid_request', message } }; +} + +function mutationNotFound(message: string): OperationOutcome<'task.mutation.query'> { + return { ok: false, error: { code: 'not_found', message } }; +} diff --git a/packages/runtime-host/src/server/task-mutation-projection.ts b/packages/runtime-host/src/server/task-mutation-projection.ts new file mode 100644 index 0000000000..054370ff72 --- /dev/null +++ b/packages/runtime-host/src/server/task-mutation-projection.ts @@ -0,0 +1,222 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { + TASK_LEDGER_MAX_TASKS, + canTransitionTaskStatus, + isSafeTaskId, + isTaskKey, + projectTaskLedgerEvents, + sanitizeTaskLedgerTask, + type Task, + type TaskLedgerEvent, +} from '@maka/core/task-ledger'; +import type { SequencedTaskLedgerEvent } from '@maka/storage/task-ledger-authority'; +import type { + TaskMutationChange, + TaskMutationCorrelation, + TaskMutationLookup, + TaskMutationPresentation, +} from '../protocol/index.js'; + +const UPDATE_EVENT_TYPES = new Set([ + 'task_updated', + 'task_started', + 'task_blocked', + 'task_completed', + 'task_failed', + 'task_cancelled', + 'task_reopened', +]); + +/** + * Derive immutable, display-safe mutation facts from the canonical Task event log. + * Request order is preserved so every correlation occupies one deterministic slot. + */ +export function projectTaskMutationLookups( + rows: readonly SequencedTaskLedgerEvent[], + correlations: readonly TaskMutationCorrelation[], +): readonly TaskMutationLookup[] { + const canonicalHistory = projectTaskLedgerEvents(rows.map(({ event }) => event)); + const historyIsCanonical = canonicalHistory.diagnostics.length === 0; + const rowsByCorrelation = new Map(); + for (const row of rows) { + const refs = row.event.refs; + if (!refs?.turnId || !refs.toolCallId) continue; + const key = correlationKey({ turnId: refs.turnId, toolCallId: refs.toolCallId }); + const matched = rowsByCorrelation.get(key) ?? []; + matched.push(row); + rowsByCorrelation.set(key, matched); + } + + return correlations.map((correlation) => { + const matched = rowsByCorrelation.get(correlationKey(correlation)); + if (!matched || matched.length === 0) return { kind: 'not_found', correlation }; + if (!historyIsCanonical) return { kind: 'incompatible', correlation }; + const presentation = projectPresentation(correlation, matched); + return presentation + ? { kind: 'found', correlation, presentation } + : { kind: 'incompatible', correlation }; + }); +} + +function projectPresentation( + correlation: TaskMutationCorrelation, + rows: readonly SequencedTaskLedgerEvent[], +): TaskMutationPresentation | undefined { + if (!isCompatibleCorrelationGroup(rows)) return undefined; + const operation = rows[0]?.event.type === 'task_created' ? 'create' : 'update'; + if ( + (operation === 'create' && !isCompatibleCreate(rows)) || + (operation === 'update' && !isCompatibleUpdate(rows)) + ) { + return undefined; + } + const changes: TaskMutationChange[] = []; + for (const { event } of rows) { + const change = projectChange(event); + if (!change) return undefined; + changes.push(change); + } + return { operation, correlation, changes }; +} + +function isCompatibleCorrelationGroup(rows: readonly SequencedTaskLedgerEvent[]): boolean { + const firstRunId = rows[0]?.event.refs?.runId; + if (!firstRunId) return false; + for (let index = 0; index < rows.length; index += 1) { + const row = rows[index]; + if (!row) return false; + if (index > 0 && row.sequence !== rows[index - 1]!.sequence + 1) return false; + if ( + row.event.source !== 'tool' || + row.event.actor !== 'main_agent' || + row.event.refs?.runId !== firstRunId + ) { + return false; + } + } + return true; +} + +function isCompatibleCreate(rows: readonly SequencedTaskLedgerEvent[]): boolean { + if (rows.length === 0 || rows.length > TASK_LEDGER_MAX_TASKS) return false; + const taskIds = new Set(); + const taskKeys = new Set(); + for (const { event } of rows) { + const key = event.task.key; + if ( + event.type !== 'task_created' || + event.previousStatus !== undefined || + event.nextStatus !== 'pending' || + !key || + taskIds.has(event.taskId) || + taskKeys.has(key) + ) { + return false; + } + taskIds.add(event.taskId); + taskKeys.add(key); + } + return true; +} + +function isCompatibleUpdate(rows: readonly SequencedTaskLedgerEvent[]): boolean { + const event = rows.length === 1 ? rows[0]?.event : undefined; + return ( + event !== undefined && + UPDATE_EVENT_TYPES.has(event.type) && + event.previousStatus !== undefined && + canTransitionTaskStatus(event.previousStatus, event.nextStatus, { + explicitReopen: event.type === 'task_reopened', + }) && + isCompatibleUpdateEventType(event) + ); +} + +function isCompatibleUpdateEventType(event: TaskLedgerEvent): boolean { + switch (event.type) { + case 'task_updated': + return event.previousStatus === event.nextStatus; + case 'task_started': + return event.nextStatus === 'in_progress'; + case 'task_blocked': + return event.nextStatus === 'blocked'; + case 'task_completed': + return event.nextStatus === 'completed'; + case 'task_failed': + return event.nextStatus === 'failed'; + case 'task_cancelled': + return event.nextStatus === 'cancelled'; + case 'task_reopened': + return ( + (event.previousStatus === 'completed' && event.nextStatus === 'in_progress') || + (event.previousStatus === 'cancelled' && event.nextStatus === 'pending') || + (event.previousStatus === 'failed' && event.nextStatus === 'pending') + ); + default: + return false; + } +} + +function projectChange(event: TaskLedgerEvent): TaskMutationChange | undefined { + const key = event.task.key; + if (!key || !isTaskKey(key) || !isSafeTaskId(event.taskId)) return undefined; + const expectedReason = + event.nextStatus === 'blocked' + ? event.task.blockedReason + : event.nextStatus === 'failed' + ? event.task.failureReason + : undefined; + const expectedEvidence = + event.nextStatus === 'completed' ? event.task.completionEvidence : undefined; + if ( + ((event.nextStatus === 'blocked' || event.nextStatus === 'failed') && !expectedReason) || + (event.nextStatus === 'completed' && !expectedEvidence) + ) { + return undefined; + } + if (event.reason !== expectedReason || event.evidence !== expectedEvidence) return undefined; + + const task = sanitizeTaskLedgerTask({ ...event.task, key } as Task); + const reason = + event.nextStatus === 'blocked' + ? task.blockedReason + : event.nextStatus === 'failed' + ? task.failureReason + : undefined; + const evidence = event.nextStatus === 'completed' ? task.completionEvidence : undefined; + return { + taskId: event.taskId, + key, + subject: task.subject, + ...(event.previousStatus !== undefined ? { previousStatus: event.previousStatus } : {}), + nextStatus: event.nextStatus, + ...(reason !== undefined ? { reason } : {}), + ...(evidence !== undefined ? { evidence } : {}), + }; +} + +export function taskMutationCorrelationKey(correlation: TaskMutationCorrelation): string { + return correlationKey(correlation); +} + +function correlationKey(correlation: TaskMutationCorrelation): string { + return JSON.stringify([correlation.turnId, correlation.toolCallId]); +} diff --git a/packages/runtime/src/__tests__/code-mode-backend.test.ts b/packages/runtime/src/__tests__/code-mode-backend.test.ts index 61b37f8c83..d981e6e5e3 100644 --- a/packages/runtime/src/__tests__/code-mode-backend.test.ts +++ b/packages/runtime/src/__tests__/code-mode-backend.test.ts @@ -366,6 +366,54 @@ test('links nested activity to the durable outer exec operation', async () => { assert.equal(nestedDurable?.modelVisibility, 'hidden'); }); +test('bounds long-parent nested Task identities while preserving the durable parent ref', async () => { + const parentToolCallId = `provider-${'x'.repeat(120)}`; + const prepared: ToolPreparedCommit[] = []; + let seq = 0; + const sink: RuntimeCommitSink = { + commitToolPrepared: async (input) => { + prepared.push(input); + return { created: true, runtimeEventSeq: ++seq }; + }, + commitToolOutcome: async () => ({ created: true, runtimeEventSeq: ++seq }), + }; + const taskTools: MakaTool[] = ['task_create', 'task_update'].map((name) => ({ + name, + description: name, + parameters: z.object({}), + impl: () => ({ ok: true }), + })); + await collect( + backend( + execThenStopModel( + 'return await Promise.all([tools.task_create({}), tools.task_update({})])', + parentToolCallId, + ), + [], + sink, + { tools: taskTools }, + ).send({ + invocationId: 'inv-long-parent', + runId: 'run-long-parent', + turnId: 'turn-long-parent', + text: 'mutate tasks', + context: [], + toolMode: 'code_mode', + }), + ); + + const nested = prepared.filter( + (commit) => commit.toolName === 'task_create' || commit.toolName === 'task_update', + ); + assert.deepEqual(nested.map(({ toolName }) => toolName).sort(), ['task_create', 'task_update']); + assert.equal(new Set(nested.map(({ providerToolCallId }) => providerToolCallId)).size, 2); + for (const commit of nested) { + assert.match(commit.providerToolCallId, /^code_nested_v1_[a-f0-9]{64}$/); + assert.ok(commit.providerToolCallId.length <= 128); + assert.equal(commit.runtimeEvent.refs?.parentToolCallId, parentToolCallId); + } +}); + test('propagates nested durable commit failures out of the outer exec', async () => { const outcomes: number[] = []; const sink: RuntimeCommitSink = { @@ -990,6 +1038,7 @@ function backend( function execThenStopModel( code = 'return await tools.lookup({ id: "nested" })', + toolCallId = 'exec-1', ): MockLanguageModelV4 { let step = 0; return new MockLanguageModelV4({ @@ -1001,7 +1050,7 @@ function execThenStopModel( { type: 'stream-start', warnings: [] }, { type: 'tool-call', - toolCallId: 'exec-1', + toolCallId, toolName: 'exec', input: JSON.stringify({ code }), }, diff --git a/packages/runtime/src/__tests__/code-mode-nested-tool-call-id.test.ts b/packages/runtime/src/__tests__/code-mode-nested-tool-call-id.test.ts new file mode 100644 index 0000000000..653da60a32 --- /dev/null +++ b/packages/runtime/src/__tests__/code-mode-nested-tool-call-id.test.ts @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + CODE_MODE_NESTED_TOOL_CALL_ID_MAX_CHARS, + codeModeNestedToolCallId, +} from '../code-mode-nested-tool-call-id.js'; + +test('preserves fitting Code Mode nested tool-call identities exactly', () => { + const child = 'c'.repeat(36); + const parent = 'p'.repeat( + CODE_MODE_NESTED_TOOL_CALL_ID_MAX_CHARS - ':nested:'.length - child.length, + ); + assert.equal(codeModeNestedToolCallId(parent, child), `${parent}:nested:${child}`); +}); + +test('hashes oversized identities with framed, domain-separated tuple input', () => { + const child = 'c'.repeat(36); + const parent = 'p'.repeat( + CODE_MODE_NESTED_TOOL_CALL_ID_MAX_CHARS - ':nested:'.length - child.length + 1, + ); + const identity = codeModeNestedToolCallId(parent, child); + assert.match(identity, /^code_nested_v1_[a-f0-9]{64}$/); + assert.ok(identity.length <= CODE_MODE_NESTED_TOOL_CALL_ID_MAX_CHARS); + assert.equal(codeModeNestedToolCallId(parent, child), identity); + + const prefix = 'x'.repeat(120); + const firstParent = `${prefix}:nested:y`; + const firstChild = 'z'; + const secondParent = prefix; + const secondChild = 'y:nested:z'; + assert.equal(`${firstParent}:nested:${firstChild}`, `${secondParent}:nested:${secondChild}`); + assert.notEqual( + codeModeNestedToolCallId(firstParent, firstChild), + codeModeNestedToolCallId(secondParent, secondChild), + ); +}); + +test('uses the same UTF-16 length boundary as the Host id decoder', () => { + const child = 'child'; + const fittingParent = '😀'.repeat( + Math.floor((CODE_MODE_NESTED_TOOL_CALL_ID_MAX_CHARS - ':nested:'.length - child.length) / 2), + ); + assert.equal(codeModeNestedToolCallId(fittingParent, child), `${fittingParent}:nested:${child}`); + assert.match(codeModeNestedToolCallId(`${fittingParent}😀`, child), /^code_nested_v1_/); +}); diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index f298b8c3b7..f39bea8f80 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -134,6 +134,7 @@ import { DEFAULT_CODE_MODE_EXECUTION_POLICY, executeCodeCell, } from './code-mode.js'; +import { codeModeNestedToolCallId } from './code-mode-nested-tool-call-id.js'; import { StreamWatchdog, formatStreamWatchdogError, @@ -3195,10 +3196,11 @@ export class AiSdkBackend implements AgentBackend { const tool = snapshot.get(name); if (!tool) throw new Error(`Tool "${name}" is not active or nestable in this cell`); const parsedInput = await validateCodeModeToolInput(tool, input); + const childToolCallId = this.newId(); const settlement = await scope.toolRuntime.settleToolCallRaw({ tool, turnId: context.turnId, - toolCallId: `${context.toolCallId}:nested:${this.newId()}`, + toolCallId: codeModeNestedToolCallId(context.toolCallId, childToolCallId), input: parsedInput, abortSignal: signal, eventSink: nestedEventSink, diff --git a/packages/runtime/src/code-mode-nested-tool-call-id.ts b/packages/runtime/src/code-mode-nested-tool-call-id.ts new file mode 100644 index 0000000000..d3bcc1e5bc --- /dev/null +++ b/packages/runtime/src/code-mode-nested-tool-call-id.ts @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { createHash } from 'node:crypto'; + +export const CODE_MODE_NESTED_TOOL_CALL_ID_MAX_CHARS = 128; + +const HASH_DOMAIN = 'maka.code-mode.nested-tool-call-id.v1'; +const HASHED_PREFIX = 'code_nested_v1_'; + +/** + * Keeps the historical readable identity while it fits the Host opaque-id + * contract, then falls back to a domain-separated digest. The complete parent + * identity remains available separately as `parentToolCallId`. + */ +export function codeModeNestedToolCallId( + parentToolCallId: string, + childToolCallId: string, +): string { + const candidate = `${parentToolCallId}:nested:${childToolCallId}`; + if (candidate.length <= CODE_MODE_NESTED_TOOL_CALL_ID_MAX_CHARS) return candidate; + const digest = createHash('sha256') + .update(JSON.stringify([HASH_DOMAIN, parentToolCallId, childToolCallId]), 'utf8') + .digest('hex'); + return `${HASHED_PREFIX}${digest}`; +} diff --git a/packages/storage/src/__tests__/task-ledger-authority.test.ts b/packages/storage/src/__tests__/task-ledger-authority.test.ts index d0ca59dc87..efd1435e1f 100644 --- a/packages/storage/src/__tests__/task-ledger-authority.test.ts +++ b/packages/storage/src/__tests__/task-ledger-authority.test.ts @@ -204,6 +204,47 @@ describe('interactive task ledger authority', () => { assert.equal(copied?.owner?.runId, 'copied-root-run'); }); }); + + test('exposes sequenced canonical events and preserves tool correlation across copy', async () => { + await withInteractiveOwner(async ({ writer }) => { + const sourceSessionId = 'sequenced-source'; + const targetSessionId = 'sequenced-target'; + await writer.create(sourceSessionId, [{ subject: 'Correlated task' }], { + turnId: 'turn-1', + runId: 'source-run', + toolCallId: 'tool-call-1', + source: 'tool', + actor: 'main_agent', + }); + const sourceEvents = await writer.readSequencedEvents(sourceSessionId); + assert.equal(sourceEvents.length, 1); + assert.equal(sourceEvents[0]?.sequence, 0); + + await writer.copyConversationTaskLedger({ + sourceSessionId, + targetSessionId, + turnIds: ['turn-1'], + runIdMap: [{ sourceRunId: 'source-run', targetRunId: 'target-run' }], + }); + const targetEvents = await writer.readSequencedEvents(targetSessionId); + assert.equal(targetEvents.length, 1); + assert.equal(targetEvents[0]?.sequence, 0); + assert.notEqual(targetEvents[0]?.event.eventId, sourceEvents[0]?.event.eventId); + assert.equal(targetEvents[0]?.event.sessionId, targetSessionId); + assert.deepEqual(targetEvents[0]?.event.refs, { + turnId: 'turn-1', + toolCallId: 'tool-call-1', + runId: 'target-run', + }); + + const returned = targetEvents[0]; + if (returned) returned.event.task.subject = 'Caller mutation'; + assert.equal( + (await writer.readSequencedEvents(targetSessionId))[0]?.event.task.subject, + 'Correlated task', + ); + }); + }); }); async function withInteractiveOwner( diff --git a/packages/storage/src/task-ledger-authority.ts b/packages/storage/src/task-ledger-authority.ts index b63328b706..750e22b72f 100644 --- a/packages/storage/src/task-ledger-authority.ts +++ b/packages/storage/src/task-ledger-authority.ts @@ -35,6 +35,7 @@ import { } from './task-ledger-store-internal.js'; export type { TaskLedgerCanonicalReader } from './task-ledger-store-internal.js'; +export type { SequencedTaskLedgerEvent } from './task-ledger-store-internal.js'; export type { ConversationTaskLedgerCopyInput } from './task-ledger-store.js'; const writerBrand: unique symbol = Symbol('InteractiveTaskLedgerWriter'); @@ -132,6 +133,15 @@ function createInteractiveWriterFacade( [writerBrand]: true, list: (sessionId, options) => run(() => canonicalReader.list(sessionId, options)), get: (sessionId, id, options) => run(() => canonicalReader.get(sessionId, id, options)), + readSequencedEvents: (sessionId) => + run(async () => { + const events = await canonicalReader.readSequencedEvents(sessionId); + return Object.freeze( + events.map(({ sequence, event }) => + Object.freeze({ sequence, event: structuredClone(event) }), + ), + ); + }), create: (sessionId, drafts, context) => run(() => store.create(sessionId, drafts, context)), update: (sessionId, id, patch, context) => run(() => store.update(sessionId, id, patch, context)), diff --git a/packages/storage/src/task-ledger-store-internal.ts b/packages/storage/src/task-ledger-store-internal.ts index 03340d10c3..cec63b229b 100644 --- a/packages/storage/src/task-ledger-store-internal.ts +++ b/packages/storage/src/task-ledger-store-internal.ts @@ -17,14 +17,26 @@ * under the License. */ -import type { Task, TaskLedgerListOptions, TaskLedgerStore } from '@maka/core/task-ledger'; +import type { + Task, + TaskLedgerEvent, + TaskLedgerListOptions, + TaskLedgerStore, +} from '@maka/core/task-ledger'; + +export interface SequencedTaskLedgerEvent { + readonly sequence: number; + readonly event: TaskLedgerEvent; +} export interface TaskLedgerCanonicalReader { list(sessionId: string, options?: TaskLedgerListOptions): Promise; get(sessionId: string, id: string, options?: TaskLedgerListOptions): Promise; + readSequencedEvents(sessionId: string): Promise; } -// Package-private bridge: this module must stay outside both the root barrel and package exports. +// Package-private registration bridge. The authenticated authority facade may +// deliberately re-export a minimal read capability to Runtime Host. const canonicalReaderByStore = new WeakMap(); export function registerTaskLedgerCanonicalReader( diff --git a/packages/storage/src/task-ledger-store.ts b/packages/storage/src/task-ledger-store.ts index 0b2c6a485b..0bf8fc7438 100644 --- a/packages/storage/src/task-ledger-store.ts +++ b/packages/storage/src/task-ledger-store.ts @@ -48,7 +48,10 @@ import { } from '@maka/core/task-ledger'; import { chainWrite } from './write-queue.js'; import { assertSafeSessionId } from './session-store.js'; -import { registerTaskLedgerCanonicalReader } from './task-ledger-store-internal.js'; +import { + registerTaskLedgerCanonicalReader, + type SequencedTaskLedgerEvent, +} from './task-ledger-store-internal.js'; import { acquireOperationalStateDatabase, type OperationalStateDatabaseLease, @@ -92,6 +95,7 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore { registerTaskLedgerCanonicalReader(this, { list: (sessionId, options) => this.#listCanonical(sessionId, options), get: (sessionId, id, options) => this.#getCanonical(sessionId, id, options), + readSequencedEvents: (sessionId) => this.#readSequencedEvents(sessionId), }); } @@ -574,6 +578,10 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore { } private async readTaskEvents(sessionId: string): Promise { + return (await this.#readSequencedEvents(sessionId)).map(({ event }) => event); + } + + async #readSequencedEvents(sessionId: string): Promise { return readSqliteTaskLedgerEvents(this.#lease.database, sessionId); } @@ -653,25 +661,39 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore { } } -function readSqliteTaskLedgerEvents(database: DatabaseSync, sessionId: string): TaskLedgerEvent[] { +function readSqliteTaskLedgerEvents( + database: DatabaseSync, + sessionId: string, +): readonly SequencedTaskLedgerEvent[] { assertSafeSessionId(sessionId); const rows = database .prepare(` - SELECT record_json + SELECT sequence, event_id, record_json FROM workflow_task_ledger_events WHERE session_id = ? ORDER BY sequence `) - .all(sessionId) as Array<{ record_json?: unknown }>; + .all(sessionId) as Array<{ sequence?: unknown; event_id?: unknown; record_json?: unknown }>; return rows.map((row, index) => { + if ( + typeof row.sequence !== 'number' || + !Number.isSafeInteger(row.sequence) || + row.sequence !== index + ) { + throw new Error(`Invalid SQLite task event sequence at row ${index}`); + } if (typeof row.record_json !== 'string') { throw new Error(`Invalid SQLite task event at sequence ${index}`); } const parsed = JSON.parse(row.record_json); - if (!isTaskLedgerEvent(parsed) || parsed.sessionId !== sessionId) { + if ( + !isTaskLedgerEvent(parsed) || + parsed.sessionId !== sessionId || + row.event_id !== parsed.eventId + ) { throw new Error(`Invalid SQLite task event at sequence ${index}`); } - return parsed; + return Object.freeze({ sequence: row.sequence, event: parsed }); }); }