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/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 }); }); }