You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by ku...@apache.org on 2026/09/07 11:34:55 UTC
(maka) branch main updated: feat(storage): add bounded ledger archive reconstruction reader (#4972)
This is an automated email from the ASF dual-hosted git repository.
likun666661 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new f08f12abb8 feat(storage): add bounded ledger archive reconstruction reader (#4972)
f08f12abb8 is described below
commit f08f12abb8eecaeb70a3e32122f3a65a0daa6ec4
Author: KunLi <90...@users.noreply.github.com>
AuthorDate: Mon Sep 7 19:34:50 2026 +0800
feat(storage): add bounded ledger archive reconstruction reader (#4972)
* feat(storage): add bounded ledger archive reconstruction reader
Read complete target-scoped evidence through a lease-bound indexed query, then reconstruct only applied v1 archive sources using the existing projection reducer and frozen encoding. Keep production writers and Artifact fallback unchanged.
Generated-by: Codex
* fix(storage): distinguish unavailable archive evidence from corruption
Keep lease and database availability failures separate from JSON and record decoding failures. Assert exact revocation reasons and runtime error mapping.
Generated-by: Codex
---------
Co-authored-by: likun <ku...@alva.xyz>
---
packages/core/package.json | 1 +
packages/core/src/tool-result-archive-evidence.ts | 41 +++
packages/runtime/package.json | 1 +
.../ledger-tool-result-archive-reader.test.ts | 308 +++++++++++++++++++++
.../src/ledger-tool-result-archive-reader.ts | 128 +++++++++
.../runtime/src/tool-result-archive-encoding.ts | 45 +++
packages/storage/package.json | 1 +
.../__tests__/tool-result-archive-evidence.test.ts | 278 +++++++++++++++++++
.../storage/src/sqlite-core-execution-schema.ts | 9 +-
.../storage/src/tool-result-archive-evidence.ts | 126 +++++++++
10 files changed, 937 insertions(+), 1 deletion(-)
diff --git a/packages/core/package.json b/packages/core/package.json
index fb9da71f9d..45d46600f6 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -55,6 +55,7 @@
"./user-question": "./dist/user-question.js",
"./connections": "./dist/connections.js",
"./context-offload": "./dist/context-offload.js",
+ "./tool-result-archive-evidence": "./dist/tool-result-archive-evidence.js",
"./attachments": "./dist/attachments.js",
"./artifacts": "./dist/artifacts.js",
"./pet": "./dist/pet.js",
diff --git a/packages/core/src/tool-result-archive-evidence.ts b/packages/core/src/tool-result-archive-evidence.ts
new file mode 100644
index 0000000000..d3d6a8a95e
--- /dev/null
+++ b/packages/core/src/tool-result-archive-evidence.ts
@@ -0,0 +1,41 @@
+/*
+ * 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 type { RuntimeEvent } from './runtime-event.js';
+import type { AgentRunEvent } from './agent-run.js';
+
+/** Complete, bounded evidence for one target, never a partial transition history. */
+export type ToolResultArchiveEvidence =
+ | {
+ readonly ok: true;
+ readonly event: RuntimeEvent;
+ readonly transitions: readonly AgentRunEvent[];
+ }
+ | { readonly ok: false; readonly reason: 'not_found' | 'too_large' | 'corrupt' | 'unavailable' };
+
+export interface ToolResultArchiveEvidenceReader {
+ /** The caller's authenticated Session, not an authority inferred from a ref/hash. */
+ read(input: {
+ readonly sessionId: string;
+ readonly runtimeEventId: string;
+ }): Promise<ToolResultArchiveEvidence>;
+}
+
+export const TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_TRANSITIONS = 64;
+export const TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_BYTES = 2 * 1024 * 1024;
diff --git a/packages/runtime/package.json b/packages/runtime/package.json
index 6d44e393bc..a26b18008e 100644
--- a/packages/runtime/package.json
+++ b/packages/runtime/package.json
@@ -110,6 +110,7 @@
"./tool-availability": "./dist/tool-availability.js",
"./tool-free-model-call": "./dist/tool-free-model-call.js",
"./tool-result-archive": "./dist/tool-result-archive.js",
+ "./ledger-tool-result-archive-reader": "./dist/ledger-tool-result-archive-reader.js",
"./tool-result-archive-capability": "./dist/tool-result-archive-capability.js",
"./tool-result-archive-resource": "./dist/tool-result-archive-resource.js",
"./tool-runtime": "./dist/tool-runtime.js",
diff --git a/packages/runtime/src/__tests__/ledger-tool-result-archive-reader.test.ts b/packages/runtime/src/__tests__/ledger-tool-result-archive-reader.test.ts
new file mode 100644
index 0000000000..e5532c6d40
--- /dev/null
+++ b/packages/runtime/src/__tests__/ledger-tool-result-archive-reader.test.ts
@@ -0,0 +1,308 @@
+/*
+ * 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 { createHash } from 'node:crypto';
+import { test } from 'node:test';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { join } from 'node:path';
+import { tmpdir } from 'node:os';
+import { DatabaseSync } from 'node:sqlite';
+import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority';
+import { openToolResultArchiveEvidenceReader } from '@maka/storage/tool-result-archive-evidence';
+import type { RuntimeEvent } from '@maka/core/runtime-event';
+import type { AgentRunEvent } from '@maka/core/agent-run';
+import type { DurableToolResultProjection } from '@maka/core/durable-tool-result-projection';
+import { buildModelProjectionTransition } from '@maka/core/model-projection-transition';
+import { createLedgerToolResultArchiveReader } from '../ledger-tool-result-archive-reader.js';
+import { serializedToolResultProjection } from '../tool-result-archive-transition.js';
+import { serializeToolResultProjectionV1 } from '../tool-result-archive-encoding.js';
+import {
+ buildArchivedToolResultPlaceholder,
+ type ToolResultArchiveReaderInput,
+} from '../tool-result-archive.js';
+
+function fixture(
+ source: DurableToolResultProjection = { version: 1, kind: 'text', text: 'bounded model output' },
+) {
+ const event: RuntimeEvent = {
+ id: 'response',
+ sessionId: 'session',
+ runId: 'run',
+ invocationId: 'invocation',
+ turnId: 'turn',
+ ts: 1,
+ partial: false,
+ author: 'tool',
+ role: 'tool',
+ content: {
+ kind: 'function_response',
+ id: 'call',
+ name: 'Read',
+ result: 'RAW SECRET MUST NOT BE READ',
+ modelProjection: source,
+ },
+ };
+ const body = serializeToolResultProjectionV1(source);
+ const placeholder = buildArchivedToolResultPlaceholder({
+ artifactId: 'archive',
+ runtimeEventId: event.id,
+ toolCallId: 'call',
+ toolName: 'Read',
+ bodySha256: createHash('sha256').update(body).digest('hex'),
+ originalBytes: Buffer.byteLength(body),
+ originalEstimatedTokens: 100,
+ reason: 'stale_tool_result_pruned_before_compact',
+ });
+ const transition = buildModelProjectionTransition({
+ sessionId: 'session',
+ target: {
+ runtimeEventId: 'response',
+ part: 'tool_result',
+ toolCallId: 'call',
+ toolName: 'Read',
+ },
+ sourceProjection: source,
+ replacement: { version: 1, kind: 'json', value: placeholder as never },
+ now: 2,
+ });
+ const records = [envelope(transition)];
+ const reader = createLedgerToolResultArchiveReader({
+ read: async () => ({ ok: true, event, transitions: records }),
+ });
+ return { event, placeholder, transition, records, reader, body, source };
+}
+function envelope(transition: ReturnType<typeof buildModelProjectionTransition>): AgentRunEvent {
+ return {
+ id: transition.transitionId,
+ type: 'model_projection_transition_recorded',
+ sessionId: 'session',
+ runId: 'run',
+ turnId: 'turn',
+ ts: 2,
+ data: { runtimeEventId: 'response', part: 'tool_result', transition },
+ };
+}
+
+test('reads committed SQLite evidence after reopen without any Artifact payload', async () => {
+ const root = await mkdtemp(join(tmpdir(), 'maka-ledger-archive-'));
+ const capability = await resolveStorageRoot({ path: root, kind: 'interactive' });
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ try {
+ const initial = await openToolResultArchiveEvidenceReader(owner.lease);
+ initial.close();
+ const f = fixture();
+ const db = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ db.prepare(
+ 'INSERT INTO runtime_events(event_id, session_id, invocation_id, run_id, turn_id, event_seq, event_kind, payload_json, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
+ ).run(
+ 'response',
+ 'session',
+ 'invocation',
+ 'run',
+ 'turn',
+ 1,
+ 'function_response',
+ JSON.stringify(f.event),
+ 1,
+ );
+ db.prepare(
+ 'INSERT INTO core_agent_runs(session_id, run_id, created_at) VALUES (?, ?, ?)',
+ ).run('session', 'run', 1);
+ db.prepare('INSERT INTO core_agent_run_events VALUES (?, ?, ?, ?, ?, ?, ?)').run(
+ 'session',
+ 'run',
+ 1,
+ f.transition.transitionId,
+ 'model_projection_transition_recorded',
+ 2,
+ JSON.stringify(f.records[0]),
+ );
+ } finally {
+ db.close();
+ }
+ const evidence = await openToolResultArchiveEvidenceReader(owner.lease);
+ try {
+ const reader = createLedgerToolResultArchiveReader(evidence);
+ assert.deepEqual(await reader({ ...f.placeholder, sessionId: 'session' }), {
+ ok: true,
+ serializedResult: f.body,
+ });
+ assert.equal((await reader({ ...f.placeholder, sessionId: 'foreign' })).ok, false);
+ } finally {
+ evidence.close();
+ }
+ } finally {
+ await owner.close();
+ await rm(root, { recursive: true, force: true });
+ await rm(owner.controlDirectory, { recursive: true, force: true });
+ }
+});
+
+test('v1 archive encoding is byte-identical for text, JSON, denial and media', () => {
+ const sources: DurableToolResultProjection[] = [
+ { version: 1, kind: 'text', text: '中文\n"quoted"\\' },
+ { version: 1, kind: 'json', value: { z: ['中文', 1, null], a: 'line\nend' } },
+ { version: 1, kind: 'execution_denied', reason: 'no' },
+ {
+ version: 1,
+ kind: 'content',
+ parts: [
+ { kind: 'text', text: 'caption' },
+ {
+ kind: 'artifact',
+ mediaType: 'image/png',
+ ref: { kind: 'session_context', sessionId: 'session', refId: 'image' },
+ },
+ ],
+ },
+ ];
+ for (const source of sources)
+ assert.equal(serializeToolResultProjectionV1(source), serializedToolResultProjection(source));
+});
+
+test('reads the source of an applied archive transition, never the raw result', async () => {
+ const f = fixture();
+ assert.deepEqual(await f.reader({ ...f.placeholder, sessionId: 'session' }), {
+ ok: true,
+ serializedResult: f.body,
+ });
+});
+
+test('refuses missing projections instead of applying the legacy tool projector', async () => {
+ const f = fixture();
+ if (f.event.content?.kind === 'function_response') delete f.event.content.modelProjection;
+ assert.deepEqual(await f.reader({ ...f.placeholder, sessionId: 'session' }), {
+ ok: false,
+ reason: 'source_mismatch',
+ });
+});
+
+test('fails closed on Session, identity, encoding, size and hash mismatch', async () => {
+ const f = fixture();
+ for (const override of [
+ { sessionId: 'foreign' },
+ { toolCallId: 'foreign' },
+ { rewriteVersion: 2 },
+ { originalBytes: 999 },
+ { bodySha256: 'f'.repeat(64) },
+ { maxBytes: 1 },
+ ]) {
+ assert.equal(
+ (
+ await f.reader({
+ ...f.placeholder,
+ sessionId: 'session',
+ ...override,
+ } as ToolResultArchiveReaderInput)
+ ).ok,
+ false,
+ );
+ }
+});
+
+test('cannot read an archive without a committed, applicable transition', async () => {
+ const f = fixture();
+ f.records.length = 0;
+ assert.equal((await f.reader({ ...f.placeholder, sessionId: 'session' })).ok, false);
+ f.records.push(envelope({ ...f.transition, sourceProjectionDigest: `sha256:${'a'.repeat(64)}` }));
+ assert.equal((await f.reader({ ...f.placeholder, sessionId: 'session' })).ok, false);
+});
+
+test('rejects losing siblings using the existing reducer and accepts duplicate appends', async () => {
+ const f = fixture();
+ const sibling = buildModelProjectionTransition({
+ sessionId: 'session',
+ target: f.transition.target,
+ sourceProjection: f.source,
+ replacement: { version: 1, kind: 'text', text: 'sibling' },
+ now: 3,
+ });
+ // IDs are the reducer's tie break; do not invent a different winner here.
+ const archiveWins = f.transition.transitionId < sibling.transitionId;
+ f.records.push(envelope(sibling), envelope(f.transition));
+ assert.equal((await f.reader({ ...f.placeholder, sessionId: 'session' })).ok, archiveWins);
+ f.records.reverse();
+ assert.equal((await f.reader({ ...f.placeholder, sessionId: 'session' })).ok, archiveWins);
+});
+
+test('reconstructs a predecessor replacement rather than the original or final projection', async () => {
+ const f = fixture();
+ const before = buildModelProjectionTransition({
+ sessionId: 'session',
+ target: f.transition.target,
+ sourceProjection: { version: 1, kind: 'text', text: 'earlier' },
+ replacement: f.source,
+ now: 1,
+ });
+ if (f.event.content?.kind === 'function_response')
+ f.event.content.modelProjection = { version: 1, kind: 'text', text: 'earlier' };
+ const archive = buildModelProjectionTransition({
+ sessionId: 'session',
+ target: f.transition.target,
+ sourceProjection: f.source,
+ replacement: f.transition.replacement,
+ previousTransitionId: before.transitionId,
+ now: 2,
+ });
+ const final = buildModelProjectionTransition({
+ sessionId: 'session',
+ target: f.transition.target,
+ sourceProjection: archive.replacement,
+ replacement: { version: 1, kind: 'text', text: 'final' },
+ previousTransitionId: archive.transitionId,
+ now: 3,
+ });
+ f.records.splice(0, f.records.length, envelope(final), envelope(archive), envelope(before));
+ assert.deepEqual(await f.reader({ ...f.placeholder, sessionId: 'session' }), {
+ ok: true,
+ serializedResult: f.body,
+ });
+});
+
+test('unknown transition versions and incomplete evidence do not expose a source', async () => {
+ const f = fixture();
+ f.records[0]!.data = {
+ runtimeEventId: 'response',
+ transition: { ...f.transition, version: 999 },
+ };
+ assert.equal((await f.reader({ ...f.placeholder, sessionId: 'session' })).ok, false);
+ const reader = createLedgerToolResultArchiveReader({
+ read: async () => ({ ok: false, reason: 'too_large' }),
+ });
+ assert.deepEqual(await reader({ ...f.placeholder, sessionId: 'session' }), {
+ ok: false,
+ reason: 'too_large',
+ });
+});
+
+test('availability maps to read_failed while corrupt evidence remains corrupt', async () => {
+ const f = fixture();
+ for (const reason of ['unavailable', 'corrupt'] as const) {
+ const reader = createLedgerToolResultArchiveReader({
+ read: async () => ({ ok: false, reason }),
+ });
+ assert.deepEqual(await reader({ ...f.placeholder, sessionId: 'session' }), {
+ ok: false,
+ reason: reason === 'unavailable' ? 'read_failed' : 'corrupt',
+ });
+ }
+});
diff --git a/packages/runtime/src/ledger-tool-result-archive-reader.ts b/packages/runtime/src/ledger-tool-result-archive-reader.ts
new file mode 100644
index 0000000000..398ccd53a2
--- /dev/null
+++ b/packages/runtime/src/ledger-tool-result-archive-reader.ts
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { createHash } from 'node:crypto';
+import { decodeDurableToolResultProjection } from '@maka/core/durable-tool-result-projection';
+import type { ModelProjectionTransition } from '@maka/core/model-projection-transition';
+import type { ToolResultArchiveEvidenceReader } from '@maka/core/tool-result-archive-evidence';
+import {
+ decodeLedgerTransition,
+ reduceEffectiveModelProjections,
+} from './model-projection-transition-ledger.js';
+import {
+ isArchivedToolResultPlaceholder,
+ type ToolResultArchiveReader,
+} from './tool-result-archive.js';
+import { serializeToolResultProjectionV1 } from './tool-result-archive-encoding.js';
+
+/**
+ * Read-only v1 reconstruction. It never calls a live tool projector, reads an
+ * Artifact, changes history, or authorizes access from a hash alone.
+ * The evidence port is trusted to return the complete bounded target history.
+ */
+export function createLedgerToolResultArchiveReader(
+ evidence: ToolResultArchiveEvidenceReader,
+): ToolResultArchiveReader {
+ return async (input) => {
+ const request = { ...input };
+ if (
+ !isArchivedToolResultPlaceholder(request) ||
+ !/^[a-f0-9]{64}$/.test(request.bodySha256) ||
+ !Number.isSafeInteger(request.originalBytes) ||
+ request.originalBytes < 1
+ ) {
+ return { ok: false, reason: 'corrupt' };
+ }
+ const maxBytes = request.maxBytes ?? request.originalBytes;
+ if (!Number.isSafeInteger(maxBytes) || maxBytes < request.originalBytes)
+ return { ok: false, reason: 'too_large' };
+ try {
+ const loaded = await evidence.read({
+ sessionId: request.sessionId,
+ runtimeEventId: request.runtimeEventId,
+ });
+ if (!loaded.ok)
+ return {
+ ok: false,
+ reason: loaded.reason === 'unavailable' ? 'read_failed' : loaded.reason,
+ };
+ const { event } = loaded;
+ const content = event.content;
+ if (event.sessionId !== request.sessionId) return { ok: false, reason: 'session_mismatch' };
+ if (
+ event.id !== request.runtimeEventId ||
+ content?.kind !== 'function_response' ||
+ content.id !== request.toolCallId ||
+ content.name !== request.toolName ||
+ content.providerExecuted ||
+ !content.modelProjection
+ )
+ return { ok: false, reason: 'source_mismatch' };
+ let source = decodeDurableToolResultProjection(content.modelProjection);
+ const byId = new Map<string, ModelProjectionTransition>();
+ for (const record of loaded.transitions) {
+ const transition = decodeLedgerTransition(record, request.sessionId);
+ if (
+ !transition ||
+ record.sessionId !== request.sessionId ||
+ transition.target.runtimeEventId !== event.id
+ )
+ return { ok: false, reason: 'corrupt' };
+ const previous = byId.get(transition.transitionId);
+ if (
+ previous &&
+ JSON.stringify({ ...previous, createdAt: 0 }) !==
+ JSON.stringify({ ...transition, createdAt: 0 })
+ ) {
+ return { ok: false, reason: 'corrupt' };
+ }
+ byId.set(transition.transitionId, transition);
+ }
+ const reduction = reduceEffectiveModelProjections([event], [...byId.values()]);
+ for (const transition of reduction.applied) {
+ const replacement = transition.replacement;
+ const placeholder = replacement.kind === 'json' ? replacement.value : undefined;
+ if (
+ isArchivedToolResultPlaceholder(placeholder) &&
+ placeholder.artifactId === request.artifactId
+ ) {
+ if (
+ placeholder.runtimeEventId !== request.runtimeEventId ||
+ placeholder.toolCallId !== request.toolCallId ||
+ placeholder.toolName !== request.toolName ||
+ placeholder.bodySha256 !== request.bodySha256 ||
+ placeholder.originalBytes !== request.originalBytes ||
+ placeholder.rewriteVersion !== request.rewriteVersion
+ )
+ return { ok: false, reason: 'source_mismatch' };
+ const serializedResult = serializeToolResultProjectionV1(source);
+ if (Buffer.byteLength(serializedResult, 'utf8') !== request.originalBytes)
+ return { ok: false, reason: 'size_mismatch' };
+ if (createHash('sha256').update(serializedResult).digest('hex') !== request.bodySha256)
+ return { ok: false, reason: 'corrupt' };
+ return { ok: true, serializedResult };
+ }
+ source = transition.replacement;
+ }
+ return { ok: false, reason: 'not_found' };
+ } catch {
+ return { ok: false, reason: 'corrupt' };
+ }
+ };
+}
diff --git a/packages/runtime/src/tool-result-archive-encoding.ts b/packages/runtime/src/tool-result-archive-encoding.ts
new file mode 100644
index 0000000000..d2ecf90f92
--- /dev/null
+++ b/packages/runtime/src/tool-result-archive-encoding.ts
@@ -0,0 +1,45 @@
+/*
+ * 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 type { DurableToolResultProjection } from '@maka/core/durable-tool-result-projection';
+
+/** Frozen bytes for archive rewriteVersion 1. Do not inherit future UI/provider formatting changes. */
+export function serializeToolResultProjectionV1(projection: DurableToolResultProjection): string {
+ switch (projection.kind) {
+ case 'text':
+ return JSON.stringify(projection.text);
+ case 'json':
+ return JSON.stringify(projection.value);
+ case 'execution_denied':
+ return JSON.stringify({ kind: 'text', text: projection.reason ?? '' });
+ case 'failure':
+ return JSON.stringify(projection.message);
+ case 'content':
+ return JSON.stringify(
+ projection.parts.map((part) =>
+ part.kind === 'text'
+ ? { type: 'text', text: part.text }
+ : {
+ type: 'text',
+ text: `[Artifact ${JSON.stringify(part.ref.kind === 'session_context' ? part.ref.refId : part.ref.relativePath)} (${part.mediaType}) is stored in this Session.]`,
+ },
+ ),
+ );
+ }
+}
diff --git a/packages/storage/package.json b/packages/storage/package.json
index dc28e6f7aa..af97e8ebbf 100644
--- a/packages/storage/package.json
+++ b/packages/storage/package.json
@@ -11,6 +11,7 @@
"./artifact-stores": "./dist/artifact-stores.js",
"./config-transfer": "./dist/config-transfer.js",
"./context-offload-store": "./dist/context-offload-store.js",
+ "./tool-result-archive-evidence": "./dist/tool-result-archive-evidence.js",
"./credential-store": "./dist/credential-store.js",
"./daily-review-authority": "./dist/daily-review-authority.js",
"./deep-research-authority": "./dist/deep-research-authority.js",
diff --git a/packages/storage/src/__tests__/tool-result-archive-evidence.test.ts b/packages/storage/src/__tests__/tool-result-archive-evidence.test.ts
new file mode 100644
index 0000000000..7d83b7850c
--- /dev/null
+++ b/packages/storage/src/__tests__/tool-result-archive-evidence.test.ts
@@ -0,0 +1,278 @@
+/*
+ * 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 { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
+import { test, after, type TestContext } from 'node:test';
+import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '../root-authority.js';
+import { openToolResultArchiveEvidenceReader } from '../tool-result-archive-evidence.js';
+import { MODEL_PROJECTION_TARGET_SQL } from '../sqlite-core-execution-schema.js';
+import {
+ trackControlDirectory,
+ removeTrackedControlDirectories,
+} from './fixtures/control-directory-hygiene.js';
+
+after(removeTrackedControlDirectories);
+async function fixture(t: TestContext) {
+ const root = await mkdtemp(join(tmpdir(), 'maka-archive-evidence-'));
+ t.after(() => rm(root, { recursive: true, force: true }));
+ const capability = trackControlDirectory(
+ await resolveStorageRoot({ path: root, kind: 'interactive' }),
+ );
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ const reader = await openToolResultArchiveEvidenceReader(owner.lease);
+ const db = new DatabaseSync(join(root, 'runtime.sqlite'));
+ t.after(async () => {
+ db.close();
+ reader.close();
+ await owner.close();
+ });
+ const event = {
+ id: 'response',
+ sessionId: 'session',
+ runId: 'run',
+ invocationId: 'invocation',
+ turnId: 'turn',
+ ts: 1,
+ partial: false,
+ author: 'tool',
+ role: 'tool',
+ content: {
+ kind: 'function_response',
+ id: 'call',
+ name: 'Read',
+ result: 'raw',
+ modelProjection: { version: 1, kind: 'text', text: 'model' },
+ },
+ };
+ db.prepare(
+ 'INSERT INTO runtime_events(event_id, session_id, invocation_id, run_id, turn_id, event_seq, event_kind, payload_json, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
+ ).run(
+ 'response',
+ 'session',
+ 'invocation',
+ 'run',
+ 'turn',
+ 1,
+ 'function_response',
+ JSON.stringify(event),
+ 1,
+ );
+ db.prepare('INSERT INTO core_agent_runs(session_id, run_id, created_at) VALUES (?, ?, ?)').run(
+ 'session',
+ 'run',
+ 1,
+ );
+ const insert = (sequence: number, target: string | null, padding = '') => {
+ const record = {
+ id: 'transition-' + sequence,
+ type: 'model_projection_transition_recorded',
+ sessionId: 'session',
+ runId: 'run',
+ turnId: 'turn',
+ ts: 2,
+ data: { ...(target === null ? {} : { runtimeEventId: target }), padding },
+ };
+ db.prepare('INSERT INTO core_agent_run_events VALUES (?, ?, ?, ?, ?, ?, ?)').run(
+ 'session',
+ 'run',
+ sequence,
+ record.id,
+ record.type,
+ 2,
+ JSON.stringify(record),
+ );
+ };
+ return { root, owner, reader, db, insert };
+}
+
+test('target evidence ignores 12k unrelated records, survives reopen and checks Session scope', async (t) => {
+ const f = await fixture(t);
+ f.db.exec('BEGIN');
+ for (let i = 0; i < 12000; i += 1) f.insert(i, 'unrelated-' + i);
+ f.insert(12000, 'response');
+ f.db.exec('COMMIT');
+ const result = await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' });
+ assert.equal(result.ok, true);
+ if (result.ok) assert.equal(result.transitions.length, 1);
+ assert.deepEqual(await f.reader.read({ sessionId: 'foreign', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'not_found',
+ });
+ f.reader.close();
+ const reopened = await openToolResultArchiveEvidenceReader(f.owner.lease);
+ try {
+ assert.deepEqual(
+ await reopened.read({ sessionId: 'session', runtimeEventId: 'response' }),
+ result,
+ );
+ } finally {
+ reopened.close();
+ }
+ const plan = f.db
+ .prepare(`EXPLAIN QUERY PLAN SELECT run_id, sequence FROM core_agent_run_events INDEXED BY core_model_projection_target
+ WHERE event_type = 'model_projection_transition_recorded' AND session_id = ? AND ${MODEL_PROJECTION_TARGET_SQL} = ? LIMIT ?`)
+ .all('session', 'response', 65);
+ assert.match(JSON.stringify(plan), /SEARCH.*core_model_projection_target/);
+});
+
+test('checks byte and record budgets before fetching any ledger JSON', async (t) => {
+ const f = await fixture(t);
+ f.insert(1, 'response', 'x'.repeat(3 * 1024 * 1024));
+ let materialized = 0;
+ const prepare = DatabaseSync.prototype.prepare;
+ t.mock.method(DatabaseSync.prototype, 'prepare', function (this: DatabaseSync, sql: string) {
+ const statement = prepare.call(this, sql);
+ const get = statement.get.bind(statement);
+ const all = statement.all.bind(statement);
+ const count = (row: Record<string, unknown> | undefined) => {
+ for (const key of ['payload_json', 'record_json'])
+ if (typeof row?.[key] === 'string') materialized += Buffer.byteLength(row[key]);
+ };
+ t.mock.method(statement, 'get', (...args: Parameters<typeof get>) => {
+ const row = get(...args);
+ count(row);
+ return row;
+ });
+ t.mock.method(statement, 'all', (...args: Parameters<typeof all>) => {
+ const rows = all(...args);
+ rows.forEach(count);
+ return rows;
+ });
+ return statement;
+ });
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'too_large',
+ });
+ assert.equal(materialized, 0);
+ f.db.exec('DELETE FROM core_agent_run_events');
+ for (let i = 0; i < 65; i += 1) f.insert(i, 'response');
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'too_large',
+ });
+ assert.equal(materialized, 0);
+});
+
+test('unscoped malformed transitions prevent a false complete history', async (t) => {
+ const f = await fixture(t);
+ f.insert(1, null);
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'corrupt',
+ });
+ f.db.exec('DELETE FROM core_agent_run_events');
+ f.insert(1, 'response');
+ f.db.exec("UPDATE core_agent_run_events SET record_json = '{'");
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'corrupt',
+ });
+});
+
+test('upgrades the target index without rewriting immutable transition records', async (t) => {
+ const f = await fixture(t);
+ f.insert(1, 'response');
+ const before = f.db.prepare('SELECT record_json FROM core_agent_run_events').all();
+ f.reader.close();
+ f.db.exec(
+ "DROP INDEX core_model_projection_target; UPDATE operational_schema_migrations SET version = 8 WHERE scope = 'core_execution'",
+ );
+ const reader = await openToolResultArchiveEvidenceReader(f.owner.lease);
+ try {
+ assert.equal(
+ (await reader.read({ sessionId: 'session', runtimeEventId: 'response' })).ok,
+ true,
+ );
+ assert.deepEqual(f.db.prepare('SELECT record_json FROM core_agent_run_events').all(), before);
+ assert.equal(
+ f.db
+ .prepare("SELECT version FROM operational_schema_migrations WHERE scope = 'core_execution'")
+ .get()?.version,
+ 9,
+ );
+ } finally {
+ reader.close();
+ }
+});
+
+test('reader close and root revocation never return evidence', async (t) => {
+ const f = await fixture(t);
+ await f.owner.close();
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'unavailable',
+ });
+ f.reader.close();
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'unavailable',
+ });
+});
+
+test('database read failures are unavailable, not corrupt evidence', async (t) => {
+ const f = await fixture(t);
+ const prepare = DatabaseSync.prototype.prepare;
+ const failure = t.mock.method(
+ DatabaseSync.prototype,
+ 'prepare',
+ function (this: DatabaseSync, sql: string) {
+ if (sql.includes('length(CAST(payload_json AS BLOB))'))
+ throw new Error('injected database unavailable');
+ return prepare.call(this, sql);
+ },
+ );
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'unavailable',
+ });
+ failure.mock.restore();
+ assert.equal(
+ (await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' })).ok,
+ true,
+ );
+});
+
+test('invalid event JSON and transition envelopes remain corrupt', async (t) => {
+ const f = await fixture(t);
+ f.insert(1, 'response');
+ const saved = f.db
+ .prepare("SELECT payload_json FROM runtime_events WHERE event_id = 'response'")
+ .get()!;
+ f.db.exec("UPDATE runtime_events SET payload_json = '{' WHERE event_id = 'response'");
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'corrupt',
+ });
+ f.db
+ .prepare("UPDATE runtime_events SET payload_json = ? WHERE event_id = 'response'")
+ .run(saved.payload_json!);
+ f.db.exec(
+ "UPDATE core_agent_run_events SET record_json = json_set(record_json, '$.ts', 'invalid-timestamp')",
+ );
+ assert.deepEqual(await f.reader.read({ sessionId: 'session', runtimeEventId: 'response' }), {
+ ok: false,
+ reason: 'corrupt',
+ });
+});
diff --git a/packages/storage/src/sqlite-core-execution-schema.ts b/packages/storage/src/sqlite-core-execution-schema.ts
index eff59d38f7..efc3252f13 100644
--- a/packages/storage/src/sqlite-core-execution-schema.ts
+++ b/packages/storage/src/sqlite-core-execution-schema.ts
@@ -19,7 +19,9 @@
import type { DatabaseSync } from 'node:sqlite';
-export const SQLITE_CORE_EXECUTION_SCHEMA_VERSION = 8;
+export const SQLITE_CORE_EXECUTION_SCHEMA_VERSION = 9;
+export const MODEL_PROJECTION_TARGET_SQL =
+ "CASE WHEN json_valid(record_json) THEN CASE WHEN json_type(record_json, '$.data.transition.target.runtimeEventId') = 'text' THEN nullif(json_extract(record_json, '$.data.transition.target.runtimeEventId'), '') WHEN json_type(record_json, '$.data.runtimeEventId') = 'text' THEN nullif(json_extract(record_json, '$.data.runtimeEventId'), '') END END";
export function migrateSqliteCoreExecutionDatabase(db: DatabaseSync): void {
db.exec(`
@@ -54,6 +56,11 @@ export function migrateSqliteCoreExecutionDatabase(db: DatabaseSync): void {
CREATE INDEX IF NOT EXISTS core_agent_run_events_type_sequence
ON core_agent_run_events(event_type, session_id, run_id, sequence);
+ CREATE INDEX IF NOT EXISTS core_model_projection_target
+ ON core_agent_run_events(session_id,
+ ${MODEL_PROJECTION_TARGET_SQL}
+ ) WHERE event_type = 'model_projection_transition_recorded';
+
CREATE TABLE IF NOT EXISTS core_agent_run_projections (
session_id TEXT NOT NULL,
event_type TEXT NOT NULL,
diff --git a/packages/storage/src/tool-result-archive-evidence.ts b/packages/storage/src/tool-result-archive-evidence.ts
new file mode 100644
index 0000000000..17ac5c6d77
--- /dev/null
+++ b/packages/storage/src/tool-result-archive-evidence.ts
@@ -0,0 +1,126 @@
+/*
+ * 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 {
+ TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_BYTES as MAX_BYTES,
+ TOOL_RESULT_ARCHIVE_EVIDENCE_MAX_TRANSITIONS as MAX_TRANSITIONS,
+ type ToolResultArchiveEvidenceReader,
+ type ToolResultArchiveEvidence,
+} from '@maka/core/tool-result-archive-evidence';
+import {
+ assertStorageRootLease,
+ runWithStorageRootLease,
+ type StorageRootLease,
+} from './root-authority.js';
+import { decodeRuntimeEvent } from '@maka/core/runtime-event';
+import { decodeAgentRunEvent } from '@maka/core/agent-run';
+
+import { MODEL_PROJECTION_TARGET_SQL as TARGET } from './sqlite-core-execution-schema.js';
+const TRANSITIONS = 'core_agent_run_events INDEXED BY core_model_projection_target';
+const KIND = "event_type = 'model_projection_transition_recorded'";
+
+/** Reader-first foundation; no archive writer or fallback is activated by opening it. */
+export async function openToolResultArchiveEvidenceReader(
+ lease: StorageRootLease<'interactive', 'read'> | StorageRootLease<'interactive', 'write'>,
+): Promise<ToolResultArchiveEvidenceReader & { close(): void }> {
+ await assertStorageRootLease(lease, 'interactive', lease.access);
+ const { acquireOperationalStateDatabase } = await import('./operational-state-store.js');
+ await assertStorageRootLease(lease, 'interactive', lease.access);
+ const database = acquireOperationalStateDatabase(lease.canonicalPath, {
+ schemaMigration: lease.access === 'read' ? 'require_current' : 'migrate',
+ });
+ let closed = false;
+ return {
+ close() {
+ if (!closed) {
+ closed = true;
+ database.close();
+ }
+ },
+ async read(input): Promise<ToolResultArchiveEvidence> {
+ const accepted = { ...input };
+ if (closed) return { ok: false, reason: 'unavailable' };
+ try {
+ return await runWithStorageRootLease(lease, 'interactive', lease.access, async () => {
+ if (closed) return { ok: false, reason: 'unavailable' };
+ return database.transaction('read', () => {
+ const db = database.database;
+ const { sessionId, runtimeEventId } = accepted;
+ const eventSize = db
+ .prepare(
+ 'SELECT length(CAST(payload_json AS BLOB)) AS bytes FROM runtime_events WHERE event_id = ? AND session_id = ?',
+ )
+ .get(runtimeEventId, sessionId);
+ if (!eventSize) return { ok: false, reason: 'not_found' };
+ // An unscoped unreadable transition prevents proving completeness.
+ if (
+ db
+ .prepare(
+ `SELECT 1 FROM ${TRANSITIONS} WHERE ${KIND} AND session_id = ? AND ${TARGET} IS NULL LIMIT 1`,
+ )
+ .get(sessionId)
+ )
+ return { ok: false, reason: 'corrupt' };
+ const sizes = db
+ .prepare(`SELECT run_id, sequence, length(CAST(record_json AS BLOB)) AS bytes FROM ${TRANSITIONS}
+ WHERE ${KIND} AND session_id = ? AND ${TARGET} = ? LIMIT ?`)
+ .all(sessionId, runtimeEventId, MAX_TRANSITIONS + 1);
+ if (sizes.length > MAX_TRANSITIONS) return { ok: false, reason: 'too_large' };
+ let bytes = Number(eventSize.bytes);
+ for (const row of sizes) bytes += Number(row.bytes);
+ if (!Number.isSafeInteger(bytes) || bytes < 0 || bytes > MAX_BYTES)
+ return { ok: false, reason: 'too_large' };
+ const raw = db
+ .prepare(
+ 'SELECT payload_json FROM runtime_events WHERE event_id = ? AND session_id = ?',
+ )
+ .get(runtimeEventId, sessionId);
+ let event: ReturnType<typeof decodeRuntimeEvent>;
+ try {
+ event = decodeRuntimeEvent(JSON.parse(String(raw?.payload_json)));
+ } catch {
+ return { ok: false, reason: 'corrupt' };
+ }
+ if (event.sessionId !== sessionId || event.id !== runtimeEventId)
+ return { ok: false, reason: 'corrupt' };
+ const read = db.prepare(
+ 'SELECT record_json FROM core_agent_run_events WHERE session_id = ? AND run_id = ? AND sequence = ?',
+ );
+ const transitions: ReturnType<typeof decodeAgentRunEvent>[] = [];
+ for (const row of sizes) {
+ const stored = read.get(sessionId, row.run_id!, row.sequence!);
+ let transition: ReturnType<typeof decodeAgentRunEvent>;
+ try {
+ transition = decodeAgentRunEvent(JSON.parse(String(stored?.record_json)));
+ } catch {
+ return { ok: false, reason: 'corrupt' };
+ }
+ if (transition.sessionId !== sessionId) return { ok: false, reason: 'corrupt' };
+ transitions.push(transition);
+ }
+ return { ok: true, event, transitions };
+ });
+ });
+ } catch {
+ // Failure to acquire/read the store is not evidence that its records are invalid.
+ return { ok: false, reason: 'unavailable' };
+ }
+ },
+ };
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org