diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a34d0ea26d..0177374e5b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -54,16 +54,37 @@ jobs: fetch-depth: 0 persist-credentials: false + - id: comparison + name: Resolve checked-out comparison + env: + CI_EVENT_NAME: ${{ github.event_name }} + PUSH_BASE_SHA: ${{ github.event.before }} + PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + # Every diff-based gate must compare the same checked-out tree. + # PR event bases can predate the main parent of the actual merge. + BASE_SHA="" + HEAD_SHA="$(git rev-parse --verify HEAD)" + if [[ "$CI_EVENT_NAME" == "pull_request" ]]; then + MERGED_HEAD_SHA="$(git rev-parse --verify 'HEAD^2')" + if [[ -z "$PR_HEAD_SHA" || "$MERGED_HEAD_SHA" != "$PR_HEAD_SHA" ]]; then + echo "CI comparison requires the merge of the requested PR head" >&2 + exit 1 + fi + BASE_SHA="$(git rev-parse --verify 'HEAD^1')" + elif [[ "$CI_EVENT_NAME" == "push" && ! "$PUSH_BASE_SHA" =~ ^0+$ ]]; then + BASE_SHA="$PUSH_BASE_SHA" + fi + printf 'base=%s\nhead=%s\n' "$BASE_SHA" "$HEAD_SHA" >> "$GITHUB_OUTPUT" + - id: plan name: Select affected test surfaces env: - BASE_SHA: ${{ github.event_name == 'push' && github.event.before || github.event.pull_request.base.sha }} - HEAD_SHA: ${{ github.event_name == 'push' && github.sha || github.event.pull_request.head.sha }} + BASE_SHA: ${{ steps.comparison.outputs.base }} + HEAD_SHA: ${{ steps.comparison.outputs.head }} run: | - # PR checks may predate later main changes, so the planning lane also - # validates the exact merged delta. Dispatches and unavailable history - # fail safe to every surface. - if [[ "${{ github.event_name }}" == "workflow_dispatch" ]] || [[ "$BASE_SHA" =~ ^0+$ ]] || ! git cat-file -e "${BASE_SHA}^{commit}"; then + # Dispatches, initial pushes and unavailable history select every surface. + if [[ -z "$BASE_SHA" ]] || ! git cat-file -e "${BASE_SHA}^{commit}"; then node scripts/ci-test-plan.mjs --full >> "$GITHUB_OUTPUT" else node scripts/ci-test-plan.mjs --base "$BASE_SHA" --head "$HEAD_SHA" >> "$GITHUB_OUTPUT" @@ -97,7 +118,9 @@ jobs: # incompatible protocols under one number (#3313). - name: Guard the protocol compatibility epoch if: github.event_name == 'pull_request' - run: node scripts/protocol-epoch-check.mjs --base 'HEAD^1' + env: + BASE_SHA: ${{ steps.comparison.outputs.base }} + run: node scripts/protocol-epoch-check.mjs --base "$BASE_SHA" - name: Test the epoch guard run: node --test --test-concurrency=1 scripts/protocol-epoch-check.test.mjs @@ -140,10 +163,10 @@ jobs: - name: Check locale hygiene if: steps.plan.outputs.code == 'true' env: - BASE_SHA: ${{ github.event_name == 'push' && github.event.before || github.event.pull_request.base.sha }} + BASE_SHA: ${{ steps.comparison.outputs.base }} run: | node --test scripts/check-locale-hygiene.test.mjs - if [[ -n "$BASE_SHA" && ! "$BASE_SHA" =~ ^0+$ ]]; then + if [[ -n "$BASE_SHA" ]]; then npm run check:locale-hygiene -- --base "$BASE_SHA" else npm run check:locale-hygiene @@ -214,9 +237,9 @@ jobs: - name: Check renderer architecture if: steps.plan.outputs.code == 'true' env: - BASE_SHA: ${{ github.event_name == 'push' && github.event.before || github.event.pull_request.base.sha }} + BASE_SHA: ${{ steps.comparison.outputs.base }} run: | - if [[ -n "$BASE_SHA" && ! "$BASE_SHA" =~ ^0+$ ]]; then + if [[ -n "$BASE_SHA" ]]; then npm run check:renderer-architecture -- --base "$BASE_SHA" else npm run check:renderer-architecture diff --git a/packages/storage/src/__tests__/file-session-repository.test.ts b/packages/storage/src/__tests__/file-session-repository.test.ts new file mode 100644 index 0000000000..62dee5035c --- /dev/null +++ b/packages/storage/src/__tests__/file-session-repository.test.ts @@ -0,0 +1,519 @@ +/* + * 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 { fork } from 'node:child_process'; +import { createHash } from 'node:crypto'; +import { once } from 'node:events'; +import { lstat, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { + openFileSessionRepository, + type FileSessionRepository, +} from '../file-session-repository.js'; +import { + materializeSessionCheckpointV1, + publishSessionCheckpointV1, + SessionRepositoryError, +} from '../session-repository.js'; +import type { SessionBundleArtifact, Sha256Digest } from '../session-bundle-contract.js'; + +test('persists a current Manifest head and first immutable object prefix across reopened adapters', async () => { + await withTemporaryDirectory(async (root) => { + const first = await openFileSessionRepository({ storageRoot: root }); + const initial = await publishCheckpoint( + first, + await writeArtifact(root, 'initial.tar.zst', 'initial bytes'), + ); + const created = await first.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const next = await publishCheckpoint( + first, + await writeArtifact(root, 'next.tar.zst', 'next bytes'), + ); + const committed = await first.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + commitId: 'commit-a', + }); + + const reopened = await openFileSessionRepository({ storageRoot: root }); + assert.deepEqual(await reopened.checkoutCurrent('session-a'), committed); + await assert.rejects( + reopened.checkoutExact(created.ref), + hasRepositoryCode('revision_not_available'), + ); + await reopened.objectStore.assertReadable(committed.checkpoint.manifest); + await reopened.objectStore.assertReadable(committed.checkpoint.value.compatibilityBundle); + }); +}); + +test('serializes concurrent local CAS writers across adapter instances', async () => { + await withTemporaryDirectory(async (root) => { + const left = await openFileSessionRepository({ storageRoot: root }); + const right = await openFileSessionRepository({ storageRoot: root }); + const initial = await publishCheckpoint( + left, + await writeArtifact(root, 'initial.tar.zst', 'initial bytes'), + ); + const created = await left.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const leftCheckpoint = await publishCheckpoint( + left, + await writeArtifact(root, 'left.tar.zst', 'left bytes'), + ); + const rightCheckpoint = await publishCheckpoint( + right, + await writeArtifact(root, 'right.tar.zst', 'right bytes'), + ); + + const results = await Promise.allSettled([ + left.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: leftCheckpoint, + }), + right.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: rightCheckpoint, + }), + ]); + assert.equal(results.filter((result) => result.status === 'fulfilled').length, 1); + const rejected = results.find((result) => result.status === 'rejected'); + assert.ok(rejected); + if (!rejected || rejected.status !== 'rejected') return; + assert.ok(rejected.reason instanceof SessionRepositoryError); + assert.equal(rejected.reason.code, 'revision_conflict'); + }); +}); + +for (const crashPoint of ['before-rename', 'after-rename'] as const) { + test(`recovers a repository writer killed ${crashPoint} without removing its lock`, async () => { + await withTemporaryDirectory(async (root) => { + const repository = await openFileSessionRepository({ storageRoot: root }); + const initial = await publishCheckpoint( + repository, + await writeArtifact(root, 'initial.tar.zst', 'initial bytes'), + ); + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const next = await publishCheckpoint( + repository, + await writeArtifact(root, 'next.tar.zst', 'next bytes'), + ); + const input = { + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + commitId: 'interrupted-commit', + }; + const inputPath = join(root, 'commit-input.json'); + await writeFile(inputPath, JSON.stringify(input), 'utf8'); + const writer = fork( + new URL('./fixtures/file-session-repository-crash-writer.js', import.meta.url), + [root, inputPath, crashPoint], + { stdio: ['ignore', 'ignore', 'inherit', 'ipc'] }, + ); + const closed = new Promise((resolve) => writer.once('close', () => resolve())); + try { + const ready = new AbortController(); + const timer = setTimeout(() => ready.abort(), 10_000); + try { + const [message] = await Promise.race([ + once(writer, 'message', { signal: ready.signal }), + closed.then(() => { + throw new Error('Repository writer exited before its crash point'); + }), + ]); + assert.equal(message, crashPoint); + } finally { + clearTimeout(timer); + ready.abort(); + } + assert.equal(writer.kill('SIGKILL'), true); + await closed; + // No finally ran in the child. Recovery must tolerate the marker that + // is still on disk; this test never removes repository-internal state. + await lstat(join(root, 'session-repository-v1.json.lock')); + + const reopened = await openFileSessionRepository({ storageRoot: root }); + assert.equal( + (await reopened.checkoutCurrent('session-a')).ref.revision, + crashPoint === 'before-rename' ? 'r1' : 'r2', + ); + const recovered = await reopened.commit(input); + assert.equal(recovered.ref.revision, 'r2'); + assert.deepEqual(recovered.checkpoint, next); + assert.deepEqual(await reopened.commit(input), recovered); + // An after-rename retry can return its existing receipt without taking + // a write lock. A new commit proves the repository is actually writable. + const advanced = await reopened.commit({ + sessionId: 'session-a', + expectedRevision: recovered.ref.revision, + checkpoint: initial, + commitId: 'after-recovery', + }); + assert.equal(advanced.ref.revision, 'r3'); + const final = await openFileSessionRepository({ storageRoot: root }); + assert.deepEqual(await final.checkoutCurrent('session-a'), advanced); + assert.deepEqual(await final.commit(input), recovered); + } finally { + if (writer.exitCode === null && writer.signalCode === null) writer.kill('SIGKILL'); + await closed; + } + }); + }); +} + +test('persists Fork source binding and crash recovery across reopened adapters', async () => { + await withTemporaryDirectory(async (root) => { + const first = await openFileSessionRepository({ storageRoot: root }); + const sourceCheckpoint = await publishCheckpoint( + first, + await writeArtifact(root, 'source.tar.zst', 'source bytes'), + ); + const source = await first.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: sourceCheckpoint, + }); + const pending = await first.claimFork({ + forkId: 'fork-a', + source: source.ref, + targetSessionId: 'session-b', + }); + assert.equal(pending.state, 'pending'); + assert.equal(pending.sourceAgentId, 'agent-a'); + assert.deepEqual(pending.sourceCheckpoint, sourceCheckpoint); + + const advancedCheckpoint = await publishCheckpoint( + first, + await writeArtifact(root, 'source-advanced.tar.zst', 'source advanced bytes'), + ); + await first.commit({ + sessionId: source.ref.sessionId, + expectedRevision: source.ref.revision, + checkpoint: advancedCheckpoint, + }); + + const afterCrash = await openFileSessionRepository({ storageRoot: root }); + const recovered = await afterCrash.claimFork({ + forkId: 'fork-a', + source: source.ref, + targetSessionId: 'session-b', + }); + assert.deepEqual(recovered.sourceCheckpoint, sourceCheckpoint); + await afterCrash.objectStore.assertReadable(recovered.sourceCheckpoint.manifest); + await afterCrash.objectStore.assertReadable( + recovered.sourceCheckpoint.value.compatibilityBundle, + ); + const recoveredBundle = join(root, 'recovered-source.tar.zst'); + const bundleSource = await materializeSessionCheckpointV1({ + objectStore: afterCrash.objectStore, + checkpoint: recovered.sourceCheckpoint, + destination: recoveredBundle, + maxBytes: 1024, + }); + assert.equal(bundleSource.path, recoveredBundle); + assert.equal( + bundleSource.expectedArchiveDigest, + sourceCheckpoint.value.compatibilityBundle.digest, + ); + assert.equal((await readFile(recoveredBundle)).toString(), 'source bytes'); + const targetCheckpoint = await publishCheckpoint( + afterCrash, + await writeArtifact(root, 'target.tar.zst', 'target bytes'), + ); + await assert.rejects( + afterCrash.createSession({ + sessionId: 'session-b', + agentId: 'agent-a', + checkpoint: targetCheckpoint, + lastCommittedActivationId: 'source-activation', + forkedFrom: source.ref, + createdByForkId: 'fork-a', + }), + /Fork-created Session must not carry an Activation identity/, + ); + const target = await afterCrash.createSession({ + sessionId: 'session-b', + agentId: 'agent-a', + checkpoint: targetCheckpoint, + forkedFrom: source.ref, + createdByForkId: 'fork-a', + }); + + const afterTargetCrash = await openFileSessionRepository({ storageRoot: root }); + const completed = await afterTargetCrash.completeFork({ forkId: 'fork-a' }); + assert.deepEqual(completed.target, target.ref); + assert.deepEqual(await afterTargetCrash.completeFork({ forkId: 'fork-a' }), completed); + }); +}); + +test('fails closed when durable local control-plane state is corrupt', async () => { + await withTemporaryDirectory(async (root) => { + const repository = await openFileSessionRepository({ storageRoot: root }); + const checkpoint = await publishCheckpoint( + repository, + await writeArtifact(root, 'initial.tar.zst', 'initial bytes'), + ); + await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint, + }); + await writeFile(join(root, 'session-repository-v1.json'), '{broken', 'utf8'); + const reopened = await openFileSessionRepository({ storageRoot: root }); + await assert.rejects( + reopened.checkoutCurrent('session-a'), + hasRepositoryCode('integrity_mismatch'), + ); + }); +}); + +test('fails closed when persisted revision allocation contradicts its current head', async () => { + await withTemporaryDirectory(async (root) => { + const repository = await openFileSessionRepository({ storageRoot: root }); + const initial = await publishCheckpoint( + repository, + await writeArtifact(root, 'initial.tar.zst', 'initial bytes'), + ); + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const next = await publishCheckpoint( + repository, + await writeArtifact(root, 'next.tar.zst', 'next bytes'), + ); + await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + commitId: 'commit-a', + }); + + const statePath = join(root, 'session-repository-v1.json'); + const state = JSON.parse(await readFile(statePath, 'utf8')) as { + sessions: Array<{ nextRevisionNumber: number }>; + }; + state.sessions[0].nextRevisionNumber = 2; + await writeFile(statePath, `${JSON.stringify(state)}\n`, 'utf8'); + + const reopened = await openFileSessionRepository({ storageRoot: root }); + await assert.rejects( + reopened.checkoutCurrent('session-a'), + hasRepositoryCode('integrity_mismatch'), + ); + }); +}); + +test('fails closed when a persisted commit receipt contradicts the current head', async () => { + await withTemporaryDirectory(async (root) => { + const repository = await openFileSessionRepository({ storageRoot: root }); + const initial = await publishCheckpoint( + repository, + await writeArtifact(root, 'initial.tar.zst', 'initial bytes'), + ); + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const next = await publishCheckpoint( + repository, + await writeArtifact(root, 'next.tar.zst', 'next bytes'), + ); + await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + commitId: 'commit-a', + }); + + const statePath = join(root, 'session-repository-v1.json'); + const state = JSON.parse(await readFile(statePath, 'utf8')) as { + commits: Array<{ result: { checkpoint: unknown } }>; + sessions: Array<{ createdRevision: { checkpoint: unknown } }>; + }; + state.commits[0].result.checkpoint = state.sessions[0].createdRevision.checkpoint; + await writeFile(statePath, `${JSON.stringify(state)}\n`, 'utf8'); + + const reopened = await openFileSessionRepository({ storageRoot: root }); + await assert.rejects( + reopened.checkoutCurrent('session-a'), + hasRepositoryCode('integrity_mismatch'), + ); + }); +}); + +test('uses Session and CAS preflight errors before checking an unrelated candidate object', async () => { + await withTemporaryDirectory(async (root) => { + const repository = await openFileSessionRepository({ storageRoot: root }); + const initial = await publishCheckpoint( + repository, + await writeArtifact(root, 'initial.tar.zst', 'initial bytes'), + ); + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const unreadableCreateCandidate = await publishCheckpoint( + repository, + await writeArtifact(root, 'create-candidate.tar.zst', 'create candidate bytes'), + ); + await removeLocalObject(root, unreadableCreateCandidate.manifest); + await assert.rejects( + repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-other', + checkpoint: unreadableCreateCandidate, + }), + hasRepositoryCode('session_already_exists'), + ); + + const next = await publishCheckpoint( + repository, + await writeArtifact(root, 'next.tar.zst', 'next bytes'), + ); + await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + }); + const unreadableCommitCandidate = await publishCheckpoint( + repository, + await writeArtifact(root, 'commit-candidate.tar.zst', 'commit candidate bytes'), + ); + await removeLocalObject(root, unreadableCommitCandidate.manifest); + await assert.rejects( + repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: unreadableCommitCandidate, + }), + hasRepositoryCode('revision_conflict'), + ); + }); +}); + +test('streams immutable object publication and materialization across concurrent adapters', async () => { + await withTemporaryDirectory(async (root) => { + const left = await openFileSessionRepository({ storageRoot: root }); + const right = await openFileSessionRepository({ storageRoot: root }); + const payload = Buffer.alloc(1024 * 1024 + 17, 0x61); + const artifactPath = join(root, 'large.tar.zst'); + await writeFile(artifactPath, payload); + const artifact: SessionBundleArtifact = { + path: artifactPath, + archiveDigest: digest(payload), + compressedBytes: payload.byteLength, + decompressedTarBytes: payload.byteLength, + payloadBytes: payload.byteLength, + entryCount: 1, + }; + + const [leftCheckpoint, rightCheckpoint] = await Promise.all([ + publishCheckpoint(left, artifact), + publishCheckpoint(right, artifact), + ]); + assert.deepEqual(leftCheckpoint, rightCheckpoint); + + const destination = join(root, 'materialized-large.tar.zst'); + await left.objectStore.materialize({ + ref: leftCheckpoint.value.compatibilityBundle, + destination, + maxBytes: payload.byteLength, + }); + assert.deepEqual(await readFile(destination), payload); + await assert.rejects( + left.objectStore.materialize({ + ref: leftCheckpoint.value.compatibilityBundle, + destination: join(root, 'too-small.tar.zst'), + maxBytes: payload.byteLength - 1, + }), + hasRepositoryCode('quota_exceeded'), + ); + await (await openFileSessionRepository({ storageRoot: root })).objectStore.assertReadable( + leftCheckpoint.value.compatibilityBundle, + ); + }); +}); + +function publishCheckpoint(repository: FileSessionRepository, artifact: SessionBundleArtifact) { + return publishSessionCheckpointV1({ + objectStore: repository.objectStore, + compatibilityBundle: artifact, + }); +} + +async function withTemporaryDirectory(operation: (root: string) => Promise): Promise { + const root = await mkdtemp(join(tmpdir(), 'maka-file-session-repository-')); + try { + await operation(root); + } finally { + await rm(root, { recursive: true, force: true }); + } +} + +async function writeArtifact( + directory: string, + name: string, + contents: string, +): Promise { + const bytes = Buffer.from(contents); + const path = join(directory, name); + await writeFile(path, bytes); + return { + path, + archiveDigest: digest(bytes), + compressedBytes: bytes.byteLength, + decompressedTarBytes: bytes.byteLength, + payloadBytes: bytes.byteLength, + entryCount: 1, + }; +} + +function digest(value: Uint8Array): Sha256Digest { + return `sha256:${createHash('sha256').update(value).digest('hex')}` as Sha256Digest; +} + +async function removeLocalObject(root: string, ref: { readonly objectRef: string }): Promise { + const id = ref.objectRef.slice('maka-local-object://v1/'.length); + await rm(join(root, 'objects', id.slice(0, 2), id)); +} + +function hasRepositoryCode(code: SessionRepositoryError['code']): (error: unknown) => boolean { + return (error): boolean => error instanceof SessionRepositoryError && error.code === code; +} diff --git a/packages/storage/src/__tests__/fixtures/file-session-repository-crash-writer.ts b/packages/storage/src/__tests__/fixtures/file-session-repository-crash-writer.ts new file mode 100644 index 0000000000..e93874c05d --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/file-session-repository-crash-writer.ts @@ -0,0 +1,46 @@ +/* + * 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 fs from 'node:fs'; +import { syncBuiltinESMExports } from 'node:module'; +import { join } from 'node:path'; +import type { CommitSessionRevisionInput } from '../../session-repository.js'; + +const [root, inputPath, crashPoint] = process.argv.slice(2); +if (!root || !inputPath || !['before-rename', 'after-rename'].includes(crashPoint)) { + throw new Error('Expected repository root, commit input and crash point'); +} +const statePath = join(root, 'session-repository-v1.json'); +const originalRename = fs.promises.rename.bind(fs.promises); +fs.promises.rename = async (...args) => { + if (args[1].toString() !== statePath) return originalRename(...args); + if (crashPoint === 'after-rename') await originalRename(...args); + // Pause a real public commit while its repository lock is held, without + // adding a production fault-injection API or running the holder's finally. + process.send?.(crashPoint); + await new Promise(() => setInterval(() => undefined, 1_000)); +}; +syncBuiltinESMExports(); + +const { openFileSessionRepository } = await import('../../file-session-repository.js'); +const input = JSON.parse( + await fs.promises.readFile(inputPath, 'utf8'), +) as CommitSessionRevisionInput; +await (await openFileSessionRepository({ storageRoot: root })).commit(input); +throw new Error('Repository commit unexpectedly passed its crash point'); diff --git a/packages/storage/src/__tests__/session-repository-conformance.test.ts b/packages/storage/src/__tests__/session-repository-conformance.test.ts new file mode 100644 index 0000000000..38740d15fc --- /dev/null +++ b/packages/storage/src/__tests__/session-repository-conformance.test.ts @@ -0,0 +1,330 @@ +/* + * 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 { mkdtemp, open, readFile, rm, writeFile, type FileHandle } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { openFileSessionRepository } from '../file-session-repository.js'; +import { + createInMemoryImmutableObjectStore, + createInMemorySessionRepository, + publishSessionCheckpointV1, + SESSION_BUNDLE_OBJECT_MEDIA_TYPE, + SessionRepositoryError, + type ImmutableObjectRef, + type ImmutableObjectStore, + type SessionRepository, + type StoredSessionCheckpoint, +} from '../session-repository.js'; +import type { Sha256Digest } from '../session-bundle-contract.js'; + +for (const backend of ['memory', 'file'] as const) { + test(`${backend}: preserves structured commit identities across reads and reopen`, async () => { + await withRepository(backend, async ({ directory, repository, objectStore, reopen }) => { + const checkpoint = await publishCheckpoint(objectStore, directory, 'initial'); + const next = await publishCheckpoint(objectStore, directory, 'next'); + const identities = [ + { sessionId: 'a', commitId: 'b\u0000c' }, + { sessionId: 'a\u0000b', commitId: 'c' }, + ]; + const unrelated = await repository.createSession({ + sessionId: 'unrelated', + agentId: 'agent', + checkpoint, + }); + for (const { sessionId } of identities) { + await repository.createSession({ sessionId, agentId: 'agent', checkpoint }); + } + const inputs = identities.map((identity) => ({ + ...identity, + expectedRevision: 'r1', + checkpoint: next, + })); + const results = []; + for (const input of inputs) results.push(await repository.commit(input)); + assert.deepEqual( + results.map((result) => result.ref.revision), + ['r2', 'r2'], + ); + + for (const reader of [repository, await reopen()]) { + assert.deepEqual(await reader.checkoutCurrent('unrelated'), unrelated); + for (const [index, input] of inputs.entries()) { + assert.deepEqual(await reader.checkoutCurrent(input.sessionId), results[index]); + assert.deepEqual(await reader.checkoutExact(results[index].ref), results[index]); + assert.deepEqual(await reader.commit(input), results[index]); + await assert.rejects( + reader.commit({ ...input, checkpoint }), + hasRepositoryCode('idempotency_conflict'), + ); + } + } + const writer = await reopen(); + for (const input of inputs) { + const advanced = await writer.commit({ + ...input, + commitId: 'after-reopen', + expectedRevision: 'r2', + checkpoint, + }); + assert.equal(advanced.ref.revision, 'r3'); + assert.deepEqual(await (await reopen()).checkoutCurrent(input.sessionId), advanced); + } + }); + }); + + for (const existing of ['caller data', 'previous materialization'] as const) { + test(`${backend}: failed materialization preserves ${existing}`, async () => { + await withRepository(backend, async ({ directory, objectStore }) => { + const checkpoint = await publishCheckpoint(objectStore, directory, 'initial'); + const ref = checkpoint.value.compatibilityBundle; + const destination = join(directory, 'materialized.tar.zst'); + const input = { ref, destination, maxBytes: ref.bytes }; + const expected = existing === 'caller data' ? 'unrelated caller contents' : 'initial'; + if (existing === 'caller data') await writeFile(destination, expected); + else await objectStore.materialize(input); + + await assert.rejects(objectStore.materialize(input), hasRepositoryCode('io_failure')); + assert.equal(await readFile(destination, 'utf8'), expected); + }); + }); + } + + test(`${backend}: concurrent materialization preserves the exclusive winner`, async () => { + await withRepository(backend, async ({ directory, objectStore }) => { + const checkpoint = await publishCheckpoint(objectStore, directory, 'initial'); + const ref = checkpoint.value.compatibilityBundle; + const destination = join(directory, 'materialized.tar.zst'); + const input = { ref, destination, maxBytes: ref.bytes }; + const results = await Promise.allSettled([ + objectStore.materialize(input), + objectStore.materialize(input), + ]); + assert.equal(results.filter((result) => result.status === 'fulfilled').length, 1); + const failure = results.find((result) => result.status === 'rejected'); + assert.ok(failure?.status === 'rejected'); + assert.ok(hasRepositoryCode('io_failure')(failure.reason)); + assert.equal(await readFile(destination, 'utf8'), 'initial'); + }); + }); + + for (const concurrent of ['identical', 'conflicting', 'absent'] as const) { + test(`${backend}: reconciles ${concurrent} Fork claim before rejecting an advanced source`, { + timeout: 10_000, + }, async (t) => { + await withRepository(backend, async ({ directory, repository, objectStore, reopen }) => { + const checkpoint = await publishCheckpoint(objectStore, directory, 'initial'); + const next = await publishCheckpoint(objectStore, directory, 'next'); + const source = await repository.createSession({ + sessionId: 'source', + agentId: 'agent', + checkpoint, + }); + const input = { forkId: 'fork', source: source.ref, targetSessionId: 'target' }; + const reading = deferred(); + const released = deferred(); + const assertReadable = objectStore.assertReadable.bind(objectStore); + let blockNextBundleRead = true; + t.mock.method(objectStore, 'assertReadable', async (ref: ImmutableObjectRef) => { + await assertReadable(ref); + if (blockNextBundleRead && ref.mediaType === SESSION_BUNDLE_OBJECT_MEDIA_TYPE) { + blockNextBundleRead = false; + reading.resolve(); + await released.promise; + } + }); + + const claim = repository.claimFork(input); + try { + await Promise.race([ + reading.promise, + claim.then(() => { + throw new Error('Claim completed before verification barrier'); + }), + ]); + const winningInput = { + ...input, + targetSessionId: concurrent === 'conflicting' ? 'other-target' : 'target', + }; + const winner = + concurrent === 'absent' ? undefined : await (await reopen()).claimFork(winningInput); + const advanced = await (await reopen()).commit({ + sessionId: source.ref.sessionId, + expectedRevision: source.ref.revision, + checkpoint: next, + }); + released.resolve(); + + if (concurrent === 'identical') { + assert.deepEqual(await claim, winner); + } else { + await assert.rejects( + claim, + hasRepositoryCode( + concurrent === 'conflicting' + ? 'idempotency_conflict' + : 'source_revision_not_available', + ), + ); + } + const reader = await reopen(); + assert.deepEqual(await reader.checkoutCurrent('source'), advanced); + if (winner) { + assert.deepEqual(winner.sourceCheckpoint, checkpoint); + assert.deepEqual(await reader.claimFork(winningInput), winner); + } else { + // The rejected attempt must not leave a claim behind. + const admitted = await reader.claimFork({ ...input, source: advanced.ref }); + assert.deepEqual(admitted.source, advanced.ref); + assert.deepEqual(admitted.sourceCheckpoint, next); + } + } finally { + released.resolve(); + await Promise.allSettled([claim]); + } + }); + }); + } +} + +test('memory: cleans up a partial file only after owning its exclusive creation', async (t) => { + await withRepository('memory', async ({ directory, objectStore }) => { + const checkpoint = await publishCheckpoint(objectStore, directory, 'initial'); + const ref = checkpoint.value.compatibilityBundle; + const destination = join(directory, 'materialized.tar.zst'); + const input = { ref, destination, maxBytes: ref.bytes }; + const probe = await open(join(directory, 'handle-probe'), 'wx'); + const prototype = Object.getPrototypeOf(probe) as FileHandle; + await probe.close(); + const failure = new Error('Injected failure after partial write'); + const write = prototype.writeFile; + const mocked = t.mock.method(prototype, 'writeFile', async function (this: FileHandle) { + await write.call(this, Buffer.from('partial')); + throw failure; + }); + try { + await assert.rejects( + objectStore.materialize(input), + (error: unknown) => + error instanceof SessionRepositoryError && + error.code === 'io_failure' && + error.cause === failure, + ); + await assert.rejects(readFile(destination), { code: 'ENOENT' }); + } finally { + mocked.mock.restore(); + } + await objectStore.materialize(input); + assert.equal(await readFile(destination, 'utf8'), 'initial'); + }); +}); + +test('file: still rejects a genuinely duplicated commit identity on reopen', async () => { + await withRepository('file', async ({ directory, repository, objectStore, reopen }) => { + const checkpoint = await publishCheckpoint(objectStore, directory, 'initial'); + await repository.createSession({ sessionId: 'a\u0000b', agentId: 'agent', checkpoint }); + await repository.commit({ + sessionId: 'a\u0000b', + commitId: 'c', + expectedRevision: 'r1', + checkpoint, + }); + const path = join(directory, 'repository', 'session-repository-v1.json'); + const state = JSON.parse(await readFile(path, 'utf8')) as { commits: unknown[] }; + state.commits.push(state.commits[0]); + await writeFile(path, JSON.stringify(state)); + await assert.rejects( + (await reopen()).checkoutCurrent('a\u0000b'), + (error: unknown) => + error instanceof SessionRepositoryError && + error.code === 'integrity_mismatch' && + error.cause instanceof Error && + error.cause.message === 'Commit identity is duplicated', + ); + }); +}); + +interface RepositoryContext { + directory: string; + repository: SessionRepository; + objectStore: ImmutableObjectStore; + /** The memory implementation shares its state; the file implementation reopens from disk. */ + reopen(): Promise; +} + +async function withRepository( + backend: 'memory' | 'file', + operation: (context: RepositoryContext) => Promise, +): Promise { + const directory = await mkdtemp(join(tmpdir(), 'maka-repository-conformance-')); + try { + if (backend === 'file') { + const storageRoot = join(directory, 'repository'); + const repository = await openFileSessionRepository({ storageRoot }); + await operation({ + directory, + repository, + objectStore: repository.objectStore, + reopen: () => openFileSessionRepository({ storageRoot }), + }); + } else { + const objectStore = createInMemoryImmutableObjectStore(); + const repository = createInMemorySessionRepository({ objectStore }); + await operation({ directory, repository, objectStore, reopen: async () => repository }); + } + } finally { + await rm(directory, { recursive: true, force: true }); + } +} + +async function publishCheckpoint( + objectStore: ImmutableObjectStore, + directory: string, + contents: string, +): Promise { + const bytes = Buffer.from(contents); + const path = join(directory, `${contents}.tar.zst`); + await writeFile(path, bytes); + return publishSessionCheckpointV1({ + objectStore, + compatibilityBundle: { + path, + archiveDigest: `sha256:${createHash('sha256').update(bytes).digest('hex')}` as Sha256Digest, + compressedBytes: bytes.byteLength, + decompressedTarBytes: bytes.byteLength, + payloadBytes: bytes.byteLength, + entryCount: 1, + }, + }); +} + +function hasRepositoryCode(code: SessionRepositoryError['code']): (error: unknown) => boolean { + return (error) => error instanceof SessionRepositoryError && error.code === code; +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} diff --git a/packages/storage/src/__tests__/session-repository.test.ts b/packages/storage/src/__tests__/session-repository.test.ts new file mode 100644 index 0000000000..5fd3218de4 --- /dev/null +++ b/packages/storage/src/__tests__/session-repository.test.ts @@ -0,0 +1,869 @@ +/* + * 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 { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { + createInMemoryImmutableObjectStore, + createInMemorySessionRepository, + createSessionCheckpointManifestV1, + encodeSessionCheckpointManifestV1, + materializeSessionCheckpointV1, + publishSessionCheckpointV1, + SESSION_BUNDLE_OBJECT_MEDIA_TYPE, + SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE, + SessionRepositoryError, + type ImmutableObjectInput, + type ImmutableObjectRef, + type ImmutableObjectStore, + type SessionCheckpointManifestV1, + type SessionRepository, + type StoredSessionCheckpoint, +} from '../session-repository.js'; +import type { SessionBundleArtifact, Sha256Digest } from '../session-bundle-contract.js'; + +test('publishes and verifies Bundle then Manifest before creating an exact head', async () => { + await withTemporaryDirectory(async (directory) => { + const base = createInMemoryImmutableObjectStore(); + const events: string[] = []; + const objectStore: ImmutableObjectStore = { + publish: async (input) => { + events.push(`publish:${input.mediaType}`); + return base.publish(input); + }, + assertReadable: async (ref) => { + events.push(`assert:${ref.mediaType}`); + await base.assertReadable(ref); + }, + materialize: (input) => base.materialize(input), + }; + const repository = createInMemorySessionRepository({ objectStore }); + const artifact = await writeArtifact(directory, 'initial.tar.zst', 'initial Bundle bytes'); + const checkpoint = await publishCheckpoint(objectStore, artifact); + + assert.deepEqual(events, [ + `publish:${SESSION_BUNDLE_OBJECT_MEDIA_TYPE}`, + `assert:${SESSION_BUNDLE_OBJECT_MEDIA_TYPE}`, + `publish:${SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE}`, + `assert:${SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE}`, + ]); + assert.equal(checkpoint.value.schemaVersion, 1); + assert.equal(checkpoint.value.compatibilityBundle.digest, artifact.archiveDigest); + + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint, + lastCommittedActivationId: 'activation-a', + }); + + assert.equal(repository.forkIdempotencyRetention, 'indefinite'); + assert.equal(created.ref.revision, 'r1'); + assert.deepEqual(created.checkpoint, checkpoint); + assert.deepEqual(await repository.checkoutCurrent('session-a'), created); + assert.deepEqual(await repository.checkoutExact(created.ref), created); + }); +}); + +test('retains only the current Manifest revision and never falls forward', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + const checkpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'next.tar.zst', 'next Bundle bytes'), + ); + const committed = await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint, + }); + + await assert.rejects( + repository.checkoutExact(created.ref), + hasRepositoryCode('revision_not_available'), + ); + assert.deepEqual(await repository.checkoutCurrent(created.ref.sessionId), committed); + assert.deepEqual(await repository.checkoutExact(committed.ref), committed); + }); +}); + +test('a source-head race returns the requested Manifest and Bundle rather than a newer head', async () => { + await withTemporaryDirectory(async (directory) => { + const base = createInMemoryImmutableObjectStore(); + let blockNextBundleRead = false; + let reading: (() => void) | undefined; + let releaseRead: (() => void) | undefined; + const readStarted = new Promise((resolve) => { + reading = resolve; + }); + const readReleased = new Promise((resolve) => { + releaseRead = resolve; + }); + const objectStore: ImmutableObjectStore = { + publish: (input) => base.publish(input), + assertReadable: async (ref) => { + await base.assertReadable(ref); + if (!blockNextBundleRead || ref.mediaType !== SESSION_BUNDLE_OBJECT_MEDIA_TYPE) return; + blockNextBundleRead = false; + reading?.(); + await readReleased; + }, + materialize: (input) => base.materialize(input), + }; + const repository = createInMemorySessionRepository({ objectStore }); + const initial = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'initial.tar.zst', 'initial'), + ); + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const next = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'next.tar.zst', 'next'), + ); + + blockNextBundleRead = true; + const exactRead = repository.checkoutExact(created.ref); + await readStarted; + const committed = await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + }); + releaseRead?.(); + + assert.deepEqual(await exactRead, created); + assert.deepEqual(await repository.checkoutExact(committed.ref), committed); + }); +}); + +test('rejects stale concurrent writers without overwriting the winning head', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + const left = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'left.tar.zst', 'left'), + ); + const right = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'right.tar.zst', 'right'), + ); + const results = await Promise.allSettled([ + repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: left, + }), + repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: right, + }), + ]); + + assert.equal(results.filter((result) => result.status === 'fulfilled').length, 1); + const rejected = results.find((result) => result.status === 'rejected'); + assert.ok(rejected); + if (!rejected || rejected.status !== 'rejected') return; + assert.ok(rejected.reason instanceof SessionRepositoryError); + assert.equal(rejected.reason.code, 'revision_conflict'); + + const winner = results.find( + ( + result, + ): result is PromiseFulfilledResult>> => + result.status === 'fulfilled', + ); + assert.ok(winner); + if (!winner) return; + assert.deepEqual(await repository.checkoutExact(winner.value.ref), winner.value); + }); +}); + +test('reconciles concurrent and later retries of one commit identity', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + const checkpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'first.tar.zst', 'first'), + ); + const input = { + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint, + commitId: 'commit-a', + }; + const [first, concurrentRetry] = await Promise.all([ + repository.commit(input), + repository.commit(input), + ]); + + assert.deepEqual(concurrentRetry, first); + assert.deepEqual(await repository.commit(input), first); + const nextCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'second.tar.zst', 'second'), + ); + const second = await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: first.ref.revision, + checkpoint: nextCheckpoint, + }); + assert.equal(second.ref.revision, 'r3'); + await assert.rejects( + repository.commit({ ...input, checkpoint: nextCheckpoint }), + hasRepositoryCode('idempotency_conflict'), + ); + }); +}); + +test('never makes a head visible for an unpublished Manifest', async () => { + await withReadySession(async ({ repository, created, checkpoint }) => { + const manifestBytes = encodeSessionCheckpointManifestV1(checkpoint.value); + const unpublished: StoredSessionCheckpoint = { + manifest: { + objectRef: 'memory://immutable-objects/not-published', + digest: digest(manifestBytes), + bytes: manifestBytes.byteLength, + mediaType: SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE, + }, + value: checkpoint.value, + }; + await assert.rejects( + repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: unpublished, + }), + hasRepositoryCode('object_not_found'), + ); + assert.deepEqual(await repository.checkoutExact(created.ref), created); + }); +}); + +test('never makes a head visible when a Manifest names an unreadable Bundle', async () => { + await withReadySession(async ({ repository, objectStore, created }) => { + const missingBundle: ImmutableObjectRef = { + objectRef: 'memory://immutable-objects/missing-bundle', + digest: digest('missing Bundle bytes'), + bytes: Buffer.byteLength('missing Bundle bytes'), + mediaType: SESSION_BUNDLE_OBJECT_MEDIA_TYPE, + }; + const value = createSessionCheckpointManifestV1(missingBundle); + const checkpoint = await publishManifest(objectStore, value); + + await assert.rejects( + repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint, + }), + hasRepositoryCode('object_not_found'), + ); + assert.deepEqual(await repository.checkoutExact(created.ref), created); + }); +}); + +test('fails closed when Bundle bytes do not match their trusted archive digest', async () => { + await withTemporaryDirectory(async (directory) => { + const objectStore = createInMemoryImmutableObjectStore(); + const artifact = await writeArtifact(directory, 'corrupt.tar.zst', 'real bytes'); + await assert.rejects( + publishSessionCheckpointV1({ + objectStore, + compatibilityBundle: { ...artifact, archiveDigest: digest('different bytes') }, + }), + hasRepositoryCode('integrity_mismatch'), + ); + }); +}); + +test('rejects a Manifest value that does not match its immutable reference', async () => { + await withReadySession(async ({ repository, objectStore, directory, created, checkpoint }) => { + const other = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'other.tar.zst', 'other'), + ); + const mismatched: StoredSessionCheckpoint = { + manifest: checkpoint.manifest, + value: other.value, + }; + + await assert.rejects( + repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: mismatched, + }), + hasRepositoryCode('integrity_mismatch'), + ); + }); +}); + +test('claims only a retained readable source and captures its Agent binding', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + await assert.rejects( + repository.claimFork({ + forkId: 'missing-source', + source: { sessionId: 'missing-session', revision: 'r1' }, + targetSessionId: 'session-b', + }), + hasRepositoryCode('source_revision_not_available'), + ); + + const next = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'source-next.tar.zst', 'source next'), + ); + const advanced = await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + }); + + await assert.rejects( + repository.claimFork({ + forkId: 'retired-source', + source: created.ref, + targetSessionId: 'session-b', + }), + hasRepositoryCode('source_revision_not_available'), + ); + const pending = await repository.claimFork({ + forkId: 'retired-source', + source: advanced.ref, + targetSessionId: 'session-b', + }); + assert.equal(pending.state, 'pending'); + assert.equal(pending.sourceAgentId, created.agentId); + }); +}); + +test('does not claim a Fork from an unreadable source checkpoint', async () => { + await withTemporaryDirectory(async (directory) => { + const base = createInMemoryImmutableObjectStore(); + let failedObjectRef: string | undefined; + const objectStore: ImmutableObjectStore = { + publish: (input) => base.publish(input), + assertReadable: async (ref) => { + if (ref.objectRef === failedObjectRef) { + throw new SessionRepositoryError('integrity_mismatch', 'Fork source is damaged'); + } + await base.assertReadable(ref); + }, + materialize: (input) => base.materialize(input), + }; + const repository = createInMemorySessionRepository({ objectStore }); + const checkpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'source.tar.zst', 'source'), + ); + const source = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint, + }); + + failedObjectRef = checkpoint.manifest.objectRef; + await assert.rejects( + repository.claimFork({ + forkId: 'fork-a', + source: source.ref, + targetSessionId: 'session-b', + }), + hasRepositoryCode('integrity_mismatch'), + ); + + failedObjectRef = undefined; + assert.equal( + ( + await repository.claimFork({ + forkId: 'fork-a', + source: source.ref, + targetSessionId: 'session-b', + }) + ).state, + 'pending', + ); + }); +}); + +test('does not claim a Fork when its source head moves during verification', async () => { + await withTemporaryDirectory(async (directory) => { + const base = createInMemoryImmutableObjectStore(); + let blockNextBundleRead = false; + let reading: (() => void) | undefined; + let releaseRead: (() => void) | undefined; + const readStarted = new Promise((resolve) => { + reading = resolve; + }); + const readReleased = new Promise((resolve) => { + releaseRead = resolve; + }); + const objectStore: ImmutableObjectStore = { + publish: (input) => base.publish(input), + assertReadable: async (ref) => { + await base.assertReadable(ref); + if (!blockNextBundleRead || ref.mediaType !== SESSION_BUNDLE_OBJECT_MEDIA_TYPE) return; + blockNextBundleRead = false; + reading?.(); + await readReleased; + }, + materialize: (input) => base.materialize(input), + }; + const repository = createInMemorySessionRepository({ objectStore }); + const initial = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'initial.tar.zst', 'initial'), + ); + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: initial, + }); + const next = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'next.tar.zst', 'next'), + ); + + blockNextBundleRead = true; + const claim = repository.claimFork({ + forkId: 'fork-a', + source: created.ref, + targetSessionId: 'session-b', + }); + await readStarted; + const advanced = await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: next, + }); + releaseRead?.(); + + await assert.rejects(claim, hasRepositoryCode('source_revision_not_available')); + assert.equal( + ( + await repository.claimFork({ + forkId: 'fork-a', + source: advanced.ref, + targetSessionId: 'session-b', + }) + ).state, + 'pending', + ); + }); +}); + +test('binds a Fork target to its verified source Agent and a distinct Session identity', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + await assert.rejects( + repository.claimFork({ + forkId: 'same-session', + source: created.ref, + targetSessionId: created.ref.sessionId, + }), + hasRepositoryCode('invalid_fork_target'), + ); + + await repository.claimFork({ + forkId: 'fork-a', + source: created.ref, + targetSessionId: 'session-b', + }); + const targetCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'fork-target.tar.zst', 'fork target'), + ); + await assert.rejects( + repository.createSession({ + sessionId: 'session-b', + agentId: 'agent-b', + checkpoint: targetCheckpoint, + forkedFrom: created.ref, + createdByForkId: 'fork-a', + }), + hasRepositoryCode('fork_agent_mismatch'), + ); + await assert.rejects( + repository.createSession({ + sessionId: 'session-b', + agentId: created.agentId, + checkpoint: targetCheckpoint, + lastCommittedActivationId: 'source-activation', + forkedFrom: created.ref, + createdByForkId: 'fork-a', + }), + /Fork-created Session must not carry an Activation identity/, + ); + + const target = await repository.createSession({ + sessionId: 'session-b', + agentId: created.agentId, + checkpoint: targetCheckpoint, + forkedFrom: created.ref, + createdByForkId: 'fork-a', + }); + assert.deepEqual((await repository.completeFork({ forkId: 'fork-a' })).target, target.ref); + }); +}); + +test('claims Fork identity before target creation and resumes both crash windows', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + const request = { + forkId: 'fork-a', + source: created.ref, + targetSessionId: 'session-b', + }; + const targetCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'fork-target.tar.zst', 'fork target'), + ); + + const pending = await repository.claimFork(request); + assert.equal(pending.state, 'pending'); + assert.equal(pending.sourceAgentId, created.agentId); + assert.deepEqual(pending.sourceCheckpoint, created.checkpoint); + assert.deepEqual(await repository.claimFork(request), pending); + + // V1 no longer retains the source revision as a Session head after it + // advances. A pending Fork must retain the exact checkpoint it admitted so + // recovery can still materialize and repack that source. + const advancedCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'source-advanced.tar.zst', 'source advanced'), + ); + await repository.commit({ + sessionId: created.ref.sessionId, + expectedRevision: created.ref.revision, + checkpoint: advancedCheckpoint, + }); + await assert.rejects( + repository.checkoutExact(created.ref), + hasRepositoryCode('revision_not_available'), + ); + await objectStore.assertReadable(pending.sourceCheckpoint.manifest); + await objectStore.assertReadable(pending.sourceCheckpoint.value.compatibilityBundle); + assert.deepEqual(await repository.claimFork(request), pending); + + const materialized = await materializeSessionCheckpointV1({ + objectStore, + checkpoint: pending.sourceCheckpoint, + destination: join(directory, 'recovered-source.tar.zst'), + maxBytes: pending.sourceCheckpoint.value.compatibilityBundle.bytes, + }); + assert.equal( + materialized.expectedArchiveDigest, + created.checkpoint.value.compatibilityBundle.digest, + ); + assert.deepEqual(await readFile(materialized.path), Buffer.from('initial Bundle bytes')); + await assert.rejects( + materializeSessionCheckpointV1({ + objectStore, + checkpoint: pending.sourceCheckpoint, + destination: join(directory, 'over-budget-source.tar.zst'), + maxBytes: pending.sourceCheckpoint.value.compatibilityBundle.bytes - 1, + }), + hasRepositoryCode('quota_exceeded'), + ); + + // Simulates a retry after a crash before target creation. + const target = await repository.createSession({ + sessionId: request.targetSessionId, + agentId: created.agentId, + checkpoint: targetCheckpoint, + forkedFrom: created.ref, + createdByForkId: request.forkId, + }); + assert.equal(target.ref.revision, 'r1'); + assert.notDeepEqual(target.ref, created.ref); + + // Simulates a retry after target creation but before operation completion. + assert.deepEqual( + await repository.createSession({ + sessionId: request.targetSessionId, + agentId: created.agentId, + checkpoint: targetCheckpoint, + forkedFrom: created.ref, + createdByForkId: request.forkId, + }), + target, + ); + const completed = await repository.completeFork({ forkId: request.forkId }); + assert.equal(completed.state, 'completed'); + assert.deepEqual(completed.target, target.ref); + assert.deepEqual(await repository.completeFork({ forkId: request.forkId }), completed); + + await assert.rejects( + repository.claimFork({ ...request, targetSessionId: 'other-session' }), + hasRepositoryCode('idempotency_conflict'), + ); + }); +}); + +test('never adopts a target created by a different Fork operation', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + await repository.claimFork({ + forkId: 'fork-owner', + source: created.ref, + targetSessionId: 'session-b', + }); + await repository.claimFork({ + forkId: 'fork-contender', + source: created.ref, + targetSessionId: 'session-b', + }); + const targetCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'fork-target.tar.zst', 'fork target'), + ); + await repository.createSession({ + sessionId: 'session-b', + agentId: created.agentId, + checkpoint: targetCheckpoint, + forkedFrom: created.ref, + createdByForkId: 'fork-owner', + }); + + await assert.rejects( + repository.createSession({ + sessionId: 'session-b', + agentId: created.agentId, + checkpoint: targetCheckpoint, + forkedFrom: created.ref, + createdByForkId: 'fork-contender', + }), + hasRepositoryCode('session_already_exists'), + ); + await assert.rejects( + repository.completeFork({ forkId: 'fork-contender' }), + hasRepositoryCode('idempotency_conflict'), + ); + }); +}); + +test('keeps a Fork pending until its target checkpoint is readable', async () => { + await withTemporaryDirectory(async (directory) => { + const base = createInMemoryImmutableObjectStore(); + let failedObjectRef: string | undefined; + const objectStore: ImmutableObjectStore = { + publish: (input) => base.publish(input), + assertReadable: async (ref) => { + if (ref.objectRef === failedObjectRef) { + throw new SessionRepositoryError('integrity_mismatch', 'Fork target is damaged'); + } + await base.assertReadable(ref); + }, + materialize: (input) => base.materialize(input), + }; + const repository = createInMemorySessionRepository({ objectStore }); + const sourceCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'fork-source.tar.zst', 'fork source'), + ); + const source = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint: sourceCheckpoint, + }); + await repository.claimFork({ + forkId: 'fork-a', + source: source.ref, + targetSessionId: 'session-b', + }); + const targetCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'fork-target.tar.zst', 'fork target'), + ); + await repository.createSession({ + sessionId: 'session-b', + agentId: source.agentId, + checkpoint: targetCheckpoint, + forkedFrom: source.ref, + createdByForkId: 'fork-a', + }); + + failedObjectRef = targetCheckpoint.manifest.objectRef; + await assert.rejects( + repository.completeFork({ forkId: 'fork-a' }), + hasRepositoryCode('integrity_mismatch'), + ); + failedObjectRef = undefined; + assert.equal((await repository.completeFork({ forkId: 'fork-a' })).state, 'completed'); + }); +}); + +test('uses independent CAS sequences for source and Fork target Sessions', async () => { + await withReadySession(async ({ repository, objectStore, directory, created }) => { + await repository.claimFork({ + forkId: 'fork-a', + source: created.ref, + targetSessionId: 'session-b', + }); + const forkCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'fork-target.tar.zst', 'fork target'), + ); + const target = await repository.createSession({ + sessionId: 'session-b', + agentId: created.agentId, + checkpoint: forkCheckpoint, + forkedFrom: created.ref, + createdByForkId: 'fork-a', + }); + const targetCheckpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'target-next.tar.zst', 'target next'), + ); + const advancedTarget = await repository.commit({ + sessionId: target.ref.sessionId, + expectedRevision: target.ref.revision, + checkpoint: targetCheckpoint, + }); + + assert.equal(advancedTarget.ref.revision, 'r2'); + assert.deepEqual(await repository.checkoutExact(created.ref), created); + }); +}); + +test('fails closed when a published Manifest or Bundle disappears or changes', async () => { + await withTemporaryDirectory(async (directory) => { + for (const target of ['manifest', 'bundle'] as const) { + for (const code of ['object_not_found', 'integrity_mismatch'] as const) { + const base = createInMemoryImmutableObjectStore(); + let failedObjectRef: string | undefined; + const objectStore: ImmutableObjectStore = { + publish: (input) => base.publish(input), + assertReadable: async (ref) => { + if (ref.objectRef === failedObjectRef) { + throw new SessionRepositoryError(code, 'Object changed after publication'); + } + await base.assertReadable(ref); + }, + materialize: (input) => base.materialize(input), + }; + const repository = createInMemorySessionRepository({ objectStore }); + const checkpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, `${target}-${code}.tar.zst`, `${target}-${code}`), + ); + const created = await repository.createSession({ + sessionId: `session-${target}-${code}`, + agentId: 'agent-a', + checkpoint, + }); + failedObjectRef = + target === 'manifest' + ? checkpoint.manifest.objectRef + : checkpoint.value.compatibilityBundle.objectRef; + + await assert.rejects(repository.checkoutExact(created.ref), hasRepositoryCode(code)); + } + } + }); +}); + +async function withReadySession( + operation: (context: { + repository: SessionRepository; + objectStore: ImmutableObjectStore; + directory: string; + checkpoint: StoredSessionCheckpoint; + created: Awaited>; + }) => Promise, +): Promise { + await withTemporaryDirectory(async (directory) => { + const objectStore = createInMemoryImmutableObjectStore(); + const repository = createInMemorySessionRepository({ objectStore }); + const checkpoint = await publishCheckpoint( + objectStore, + await writeArtifact(directory, 'initial.tar.zst', 'initial Bundle bytes'), + ); + const created = await repository.createSession({ + sessionId: 'session-a', + agentId: 'agent-a', + checkpoint, + }); + await operation({ repository, objectStore, directory, checkpoint, created }); + }); +} + +function publishCheckpoint( + objectStore: ImmutableObjectStore, + compatibilityBundle: SessionBundleArtifact, +): Promise { + return publishSessionCheckpointV1({ objectStore, compatibilityBundle }); +} + +async function publishManifest( + objectStore: ImmutableObjectStore, + value: SessionCheckpointManifestV1, +): Promise { + const bytes = encodeSessionCheckpointManifestV1(value); + const input: ImmutableObjectInput = { + digest: digest(bytes), + bytes: bytes.byteLength, + mediaType: SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE, + source: { kind: 'bytes', value: bytes }, + }; + const manifest = await objectStore.publish(input); + await objectStore.assertReadable(manifest); + return { manifest, value }; +} + +async function withTemporaryDirectory( + operation: (directory: string) => Promise, +): Promise { + const directory = await mkdtemp(join(tmpdir(), 'maka-session-repository-')); + try { + await operation(directory); + } finally { + await rm(directory, { recursive: true, force: true }); + } +} + +async function writeArtifact( + directory: string, + name: string, + contents: string, +): Promise { + const bytes = Buffer.from(contents); + const path = join(directory, name); + await writeFile(path, bytes); + return { + path, + archiveDigest: digest(bytes), + compressedBytes: bytes.byteLength, + decompressedTarBytes: bytes.byteLength, + payloadBytes: bytes.byteLength, + entryCount: 1, + }; +} + +function digest(value: Uint8Array | string): Sha256Digest { + return `sha256:${createHash('sha256').update(value).digest('hex')}` as Sha256Digest; +} + +function hasRepositoryCode(code: SessionRepositoryError['code']): (error: unknown) => boolean { + return (error: unknown): boolean => + error instanceof SessionRepositoryError && error.code === code; +} diff --git a/packages/storage/src/file-session-repository.ts b/packages/storage/src/file-session-repository.ts new file mode 100644 index 0000000000..b92105102d --- /dev/null +++ b/packages/storage/src/file-session-repository.ts @@ -0,0 +1,1295 @@ +/* + * 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, randomUUID } from 'node:crypto'; +import { constants as fsConstants } from 'node:fs'; +import { createReadStream } from 'node:fs'; +import { mkdir, open, readFile, rename, rm, type FileHandle } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import { + isNonEmptyUnicodeString, + isSha256Digest, + type Sha256Digest, +} from './session-bundle-contract.js'; +import { syncDirectory, syncDirectoryChain } from './stable-storage.js'; +import { + createSessionCheckpointManifestV1, + encodeSessionCheckpointManifestV1, + SESSION_BUNDLE_OBJECT_MEDIA_TYPE, + SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE, + SessionRepositoryError, + type ClaimForkInput, + type CommitSessionRevisionInput, + type CommittedSessionRevision, + type CompleteForkInput, + type CompletedForkOperation, + type CreateSessionInput, + type ForkOperation, + type ImmutableObjectInput, + type ImmutableObjectMaterializationInput, + type ImmutableObjectRef, + type ImmutableObjectStore, + type PendingForkOperation, + type SessionCheckpointManifestV1, + type SessionRepository, + type SessionRepositoryErrorCode, + type SessionRepositoryRevision, + type SessionRevisionRef, + type StoredSessionCheckpoint, +} from './session-repository.js'; +import { withProcessLifetimeFileUpdateLock } from './process-lifetime-file-update-lock.js'; + +const MAX_IDENTIFIER_LENGTH = 512; +const MAX_OBJECT_REF_LENGTH = 2_048; +const STATE_FILE_NAME = 'session-repository-v1.json'; + +export interface OpenFileSessionRepositoryInput { + /** Directory owned by this local adapter. It must not be a live Session root. */ + readonly storageRoot: string; +} + +export interface FileSessionRepository extends SessionRepository { + readonly objectStore: ImmutableObjectStore; +} + +/** + * Opens the durable local adapter. Object bytes are immutable files while the + * small Session control plane is one atomically replaced state document. + */ +export async function openFileSessionRepository( + input: OpenFileSessionRepositoryInput, +): Promise { + if (!isRecord(input)) throw new TypeError('File Session Repository options must be an object'); + const storageRoot = requireIdentifier(input.storageRoot, 'Storage root', MAX_OBJECT_REF_LENGTH); + await mkdir(storageRoot, { recursive: true, mode: 0o700 }); + const objectStore = new FileImmutableObjectStore(join(storageRoot, 'objects'), storageRoot); + return new FileSessionRepositoryAdapter(join(storageRoot, STATE_FILE_NAME), objectStore); +} + +class FileSessionRepositoryAdapter implements FileSessionRepository { + readonly forkIdempotencyRetention = 'indefinite' as const; + + constructor( + private readonly statePath: string, + readonly objectStore: ImmutableObjectStore, + ) {} + + async checkoutCurrent(sessionId: string): Promise { + const admittedSessionId = requireIdentifier(sessionId, 'Session identity'); + const state = await this.readState(); + const session = findSession(state, admittedSessionId); + if (!session) throw repositoryError('session_not_found', 'Cloud Session was not found'); + const result = copyCommittedSessionRevision(session.head); + await assertCheckpointReadable(this.objectStore, result.checkpoint); + return result; + } + + async checkoutExact(ref: SessionRevisionRef): Promise { + const requested = admitRevisionRef(ref, 'Session revision reference'); + const state = await this.readState(); + const session = findSession(state, requested.sessionId); + if (!session) throw repositoryError('session_not_found', 'Cloud Session was not found'); + if (session.head.ref.revision !== requested.revision) { + throw repositoryError( + 'revision_not_available', + 'Requested Session revision is not available', + ); + } + const result = copyCommittedSessionRevision(session.head); + await assertCheckpointReadable(this.objectStore, result.checkpoint); + return result; + } + + async createSession(input: CreateSessionInput): Promise { + const admitted = admitCreateSessionInput(input); + // Match the contract's linearization/error precedence: an existing Session + // is authoritative before an unrelated candidate object is consulted. + const beforeVerification = await this.readState(); + const existing = findSession(beforeVerification, admitted.sessionId); + if (existing) { + const reconciled = reconcileExistingSessionCreate(existing, admitted); + await assertCheckpointReadable(this.objectStore, reconciled.checkpoint); + return copyCommittedSessionRevision(reconciled); + } + await assertCheckpointReadable(this.objectStore, admitted.checkpoint); + const result = await this.mutate((state) => { + const concurrent = findSession(state, admitted.sessionId); + if (concurrent) return reconcileExistingSessionCreate(concurrent, admitted); + if (admitted.createdByForkId !== undefined) assertPendingForkCreate(state, admitted); + const initial = committedRevision({ + sessionId: admitted.sessionId, + revision: 'r1', + agentId: admitted.agentId, + checkpoint: admitted.checkpoint, + lastCommittedActivationId: admitted.lastCommittedActivationId, + forkedFrom: admitted.forkedFrom, + }); + state.sessions.push({ + sessionId: admitted.sessionId, + agentId: admitted.agentId, + head: initial, + nextRevisionNumber: 2, + ...(admitted.forkedFrom === undefined ? {} : { forkedFrom: admitted.forkedFrom }), + ...(admitted.createdByForkId === undefined + ? {} + : { createdByForkId: admitted.createdByForkId }), + createdRevision: initial, + }); + return initial; + }); + await assertCheckpointReadable(this.objectStore, result.checkpoint); + return copyCommittedSessionRevision(result); + } + + async commit(input: CommitSessionRevisionInput): Promise { + const admitted = admitCommitSessionRevisionInput(input); + const existing = await this.readState(); + const prior = findCommit(existing, admitted.sessionId, admitted.commitId); + if (prior) { + assertSameCommitInput(prior.input, admitted); + await assertCheckpointReadable(this.objectStore, prior.result.checkpoint); + return copyCommittedSessionRevision(prior.result); + } + const session = findSession(existing, admitted.sessionId); + if (!session) throw repositoryError('session_not_found', 'Cloud Session was not found'); + if (session.head.ref.revision !== admitted.expectedRevision) { + throw repositoryError('revision_conflict', 'Cloud Session head changed before commit'); + } + await assertCheckpointReadable(this.objectStore, admitted.checkpoint); + const result = await this.mutate((state) => { + const repeated = findCommit(state, admitted.sessionId, admitted.commitId); + if (repeated) { + assertSameCommitInput(repeated.input, admitted); + return repeated.result; + } + const session = findSession(state, admitted.sessionId); + if (!session) throw repositoryError('session_not_found', 'Cloud Session was not found'); + if (session.head.ref.revision !== admitted.expectedRevision) { + throw repositoryError('revision_conflict', 'Cloud Session head changed before commit'); + } + const result = committedRevision({ + sessionId: session.sessionId, + revision: `r${session.nextRevisionNumber}`, + agentId: session.agentId, + checkpoint: admitted.checkpoint, + lastCommittedActivationId: admitted.lastCommittedActivationId, + forkedFrom: session.forkedFrom, + }); + session.nextRevisionNumber += 1; + session.head = result; + if (admitted.commitId !== undefined) { + state.commits.push({ + sessionId: admitted.sessionId, + commitId: admitted.commitId, + input: admitted, + result, + }); + } + return result; + }); + await assertCheckpointReadable(this.objectStore, result.checkpoint); + return copyCommittedSessionRevision(result); + } + + async claimFork(input: ClaimForkInput): Promise { + const admitted = admitClaimForkInput(input); + const state = await this.readState(); + const existing = findFork(state, admitted.forkId); + if (existing) return reconcileForkClaim(existing, admitted); + if (admitted.targetSessionId === admitted.source.sessionId) { + throw repositoryError( + 'invalid_fork_target', + 'Fork target Session must differ from its source Session', + ); + } + const source = requireCurrentForkSource(state, admitted.source); + const sourceCheckpoint = admitStoredCheckpoint(source.head.checkpoint); + await assertCheckpointReadable(this.objectStore, sourceCheckpoint); + return this.mutate((latest) => { + const raced = findFork(latest, admitted.forkId); + if (raced) return reconcileForkClaim(raced, admitted); + const stillCurrent = requireCurrentForkSource(latest, admitted.source); + const pending: PersistentPendingFork = { + state: 'pending', + forkId: admitted.forkId, + source: admitted.source, + sourceAgentId: stillCurrent.agentId, + sourceCheckpoint, + targetSessionId: admitted.targetSessionId, + }; + latest.forks.push(pending); + return copyForkOperation(pending); + }); + } + + async completeFork(input: CompleteForkInput): Promise { + const forkId = requireIdentifier(input?.forkId, 'Fork identity'); + const state = await this.readState(); + const operation = findFork(state, forkId); + if (!operation) throw repositoryError('idempotency_conflict', 'Fork identity was not claimed'); + if (operation.state === 'completed') return copyCompletedForkOperation(operation); + const target = requireValidForkTarget(state, operation); + await assertCheckpointReadable(this.objectStore, target.createdRevision.checkpoint); + return this.mutate((latest) => { + const current = findFork(latest, forkId); + if (!current) throw repositoryError('idempotency_conflict', 'Fork identity was not claimed'); + if (current.state === 'completed') return copyCompletedForkOperation(current); + const target = requireValidForkTarget(latest, current); + const completed: PersistentCompletedFork = { + ...current, + state: 'completed', + target: copyRevisionRef(target.createdRevision.ref), + }; + replaceFork(latest, completed); + return copyCompletedForkOperation(completed); + }); + } + + private async readState(): Promise { + try { + const bytes = await readFile(this.statePath); + return decodeState(bytes); + } catch (error) { + if (isNodeError(error, 'ENOENT')) return emptyState(); + if (error instanceof SessionRepositoryError) throw error; + throw repositoryError('io_failure', 'Local Session Repository could not read state', error); + } + } + + private async mutate(operation: (state: PersistentState) => T): Promise { + await mkdir(dirname(this.statePath), { recursive: true, mode: 0o700 }); + try { + // The native lease is released on process death, allowing the next + // writer to reclaim an ownerless marker without stealing a live lock. + return await withProcessLifetimeFileUpdateLock(this.statePath, async () => { + const state = await this.readState(); + const result = operation(state); + await writeStateAtomically(this.statePath, state); + return result; + }); + } catch (error) { + if (error instanceof SessionRepositoryError) throw error; + throw repositoryError('io_failure', 'Local Session Repository could not update state', error); + } + } +} + +class FileImmutableObjectStore implements ImmutableObjectStore { + constructor( + private readonly objectsRoot: string, + private readonly storageRoot: string, + ) {} + + async publish(input: ImmutableObjectInput): Promise { + const admitted = admitImmutableObjectInput(input); + const ref = immutableObjectRef({ + objectRef: localObjectRef(admitted.digest, admitted.mediaType), + digest: admitted.digest, + bytes: admitted.bytes, + mediaType: admitted.mediaType, + }); + const destination = objectPath(this.objectsRoot, ref); + await mkdir(dirname(destination), { recursive: true, mode: 0o700 }); + const temporary = `${destination}.${randomUUID()}.tmp`; + try { + await writePublishedObject(admitted, temporary); + try { + await linkNoReplace(temporary, destination); + } catch (error) { + if (!isNodeError(error, 'EEXIST')) throw error; + } + // An EEXIST result means another writer published the same immutable + // name. It is not readable until that writer's directory chain has a + // durability barrier too, so both paths wait for one before returning. + await syncDirectoryChain(dirname(destination), this.storageRoot); + } catch (error) { + throw normalizeFileError(error, 'Immutable object publication failed'); + } finally { + await rm(temporary, { force: true }).catch(() => {}); + } + await this.assertReadable(ref); + return ref; + } + + async assertReadable(input: ImmutableObjectRef): Promise { + const ref = admitImmutableObjectRef(input); + if (ref.objectRef !== localObjectRef(ref.digest, ref.mediaType)) { + throw repositoryError('integrity_mismatch', 'Immutable object reference is not canonical'); + } + try { + await assertFileMatchesImmutableRef(objectPath(this.objectsRoot, ref), ref); + } catch (error) { + if (error instanceof SessionRepositoryError) throw error; + if (isNodeError(error, 'ENOENT')) { + throw repositoryError('object_not_found', 'Immutable object was not found'); + } + throw repositoryError('io_failure', 'Immutable object could not be read', error); + } + } + + async materialize(input: ImmutableObjectMaterializationInput): Promise { + const request = admitImmutableObjectMaterializationInput(input); + if (request.ref.objectRef !== localObjectRef(request.ref.digest, request.ref.mediaType)) { + throw repositoryError('integrity_mismatch', 'Immutable object reference is not canonical'); + } + if (request.ref.bytes > request.maxBytes) { + throw repositoryError( + 'quota_exceeded', + 'Immutable object exceeds materialization byte limit', + ); + } + try { + // Verify the retained object before creating the caller-owned file. The + // copy below verifies it again while streaming, so a corrupt object can + // never be returned merely because it changed between the two reads. + await this.assertReadable(request.ref); + await copyImmutableFile( + objectPath(this.objectsRoot, request.ref), + request.destination, + request.ref, + ); + } catch (error) { + if (error instanceof SessionRepositoryError) throw error; + throw repositoryError('io_failure', 'Immutable object could not be materialized', error); + } + } +} + +interface PersistentState { + readonly schemaVersion: 1; + readonly sessions: PersistentSession[]; + readonly commits: PersistentCommit[]; + readonly forks: PersistentFork[]; +} + +interface PersistentSession { + readonly sessionId: string; + readonly agentId: string; + head: CommittedSessionRevision; + nextRevisionNumber: number; + readonly forkedFrom?: SessionRevisionRef; + readonly createdByForkId?: string; + readonly createdRevision: CommittedSessionRevision; +} + +interface PersistentCommit { + readonly sessionId: string; + readonly commitId: string; + readonly input: InternalCommitInput; + readonly result: CommittedSessionRevision; +} + +interface PersistentPendingFork extends PendingForkOperation {} +interface PersistentCompletedFork extends CompletedForkOperation {} +type PersistentFork = PersistentPendingFork | PersistentCompletedFork; + +interface InternalCommitInput { + readonly sessionId: string; + readonly expectedRevision: SessionRepositoryRevision; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly commitId?: string; +} + +interface InternalCreateInput { + readonly sessionId: string; + readonly agentId: string; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly forkedFrom?: SessionRevisionRef; + readonly createdByForkId?: string; +} + +interface InternalClaimForkInput { + readonly forkId: string; + readonly source: SessionRevisionRef; + readonly targetSessionId: string; +} + +function emptyState(): PersistentState { + return { schemaVersion: 1, sessions: [], commits: [], forks: [] }; +} + +function findSession(state: PersistentState, sessionId: string): PersistentSession | undefined { + return state.sessions.find((entry) => entry.sessionId === sessionId); +} + +function findCommit( + state: PersistentState, + sessionId: string, + commitId: string | undefined, +): PersistentCommit | undefined { + return commitId === undefined + ? undefined + : state.commits.find((entry) => entry.sessionId === sessionId && entry.commitId === commitId); +} + +function findFork(state: PersistentState, forkId: string): PersistentFork | undefined { + return state.forks.find((entry) => entry.forkId === forkId); +} + +function replaceFork(state: PersistentState, replacement: PersistentFork): void { + const index = state.forks.findIndex((entry) => entry.forkId === replacement.forkId); + if (index < 0) throw repositoryError('idempotency_conflict', 'Fork identity was not claimed'); + state.forks[index] = replacement; +} + +function requireCurrentForkSource( + state: PersistentState, + source: SessionRevisionRef, +): PersistentSession { + const session = findSession(state, source.sessionId); + if (!session || session.head.ref.revision !== source.revision) { + throw repositoryError( + 'source_revision_not_available', + 'Fork source Session revision is not available', + ); + } + return session; +} + +function requireValidForkTarget( + state: PersistentState, + operation: PersistentPendingFork, +): PersistentSession { + const target = findSession(state, operation.targetSessionId); + if (!target) throw repositoryError('session_not_found', 'Fork target Session was not found'); + if (target.agentId !== operation.sourceAgentId) { + throw repositoryError( + 'fork_agent_mismatch', + 'Fork target Agent does not match its source Agent', + ); + } + if ( + target.createdByForkId !== operation.forkId || + !target.forkedFrom || + !sameRevisionRef(target.forkedFrom, operation.source) + ) { + throw repositoryError('idempotency_conflict', 'Fork target was created by another operation'); + } + return target; +} + +function assertPendingForkCreate(state: PersistentState, input: InternalCreateInput): void { + const operation = + input.createdByForkId === undefined ? undefined : findFork(state, input.createdByForkId); + if ( + !operation || + operation.state !== 'pending' || + operation.targetSessionId !== input.sessionId || + !input.forkedFrom || + !sameRevisionRef(operation.source, input.forkedFrom) + ) { + throw repositoryError( + 'idempotency_conflict', + 'Fork target does not match its claimed operation', + ); + } + if (operation.sourceAgentId !== input.agentId) { + throw repositoryError('fork_agent_mismatch', 'Fork target Agent must match its source Agent'); + } +} + +function reconcileExistingSessionCreate( + existing: PersistentSession, + input: InternalCreateInput, +): CommittedSessionRevision { + if ( + input.createdByForkId !== undefined && + existing.createdByForkId === input.createdByForkId && + existing.agentId === input.agentId && + sameStoredCheckpoint(existing.createdRevision.checkpoint, input.checkpoint) && + existing.createdRevision.lastCommittedActivationId === input.lastCommittedActivationId && + sameOptionalRevisionRef(existing.forkedFrom, input.forkedFrom) + ) { + return existing.createdRevision; + } + throw repositoryError('session_already_exists', 'Cloud Session already exists'); +} + +function reconcileForkClaim( + operation: PersistentFork, + input: InternalClaimForkInput, +): ForkOperation { + if ( + operation.targetSessionId !== input.targetSessionId || + !sameRevisionRef(operation.source, input.source) + ) { + throw repositoryError('idempotency_conflict', 'Fork identity was reused with different input'); + } + return copyForkOperation(operation); +} + +function assertSameCommitInput(left: InternalCommitInput, right: InternalCommitInput): void { + if ( + left.sessionId !== right.sessionId || + left.expectedRevision !== right.expectedRevision || + !sameStoredCheckpoint(left.checkpoint, right.checkpoint) || + left.lastCommittedActivationId !== right.lastCommittedActivationId || + left.commitId !== right.commitId + ) { + throw repositoryError( + 'idempotency_conflict', + 'Commit identity was reused with different input', + ); + } +} + +async function assertCheckpointReadable( + objectStore: ImmutableObjectStore, + checkpoint: StoredSessionCheckpoint, +): Promise { + const admitted = admitStoredCheckpoint(checkpoint); + try { + await objectStore.assertReadable(admitted.manifest); + await objectStore.assertReadable(admitted.value.compatibilityBundle); + } catch (error) { + if (error instanceof SessionRepositoryError) throw error; + throw repositoryError('io_failure', 'Immutable Object Store operation failed', error); + } +} + +function admitCommitSessionRevisionInput(input: CommitSessionRevisionInput): InternalCommitInput { + if (!isRecord(input)) throw new TypeError('Session commit input must be an object'); + return { + sessionId: requireIdentifier(input.sessionId, 'Session identity'), + expectedRevision: requireIdentifier(input.expectedRevision, 'Expected revision'), + checkpoint: admitStoredCheckpoint(input.checkpoint), + ...(input.lastCommittedActivationId === undefined + ? {} + : { + lastCommittedActivationId: requireIdentifier( + input.lastCommittedActivationId, + 'Activation identity', + ), + }), + ...(input.commitId === undefined + ? {} + : { commitId: requireIdentifier(input.commitId, 'Commit identity') }), + }; +} + +function admitCreateSessionInput(input: CreateSessionInput): InternalCreateInput { + if (!isRecord(input)) throw new TypeError('Session creation input must be an object'); + const forkedFrom = + input.forkedFrom === undefined ? undefined : admitRevisionRef(input.forkedFrom, 'Fork source'); + const createdByForkId = + input.createdByForkId === undefined + ? undefined + : requireIdentifier(input.createdByForkId, 'Fork identity'); + if ((forkedFrom === undefined) !== (createdByForkId === undefined)) { + throw new TypeError('Fork lineage and Fork identity must be supplied together'); + } + if (createdByForkId !== undefined && input.lastCommittedActivationId !== undefined) { + throw new TypeError('Fork-created Session must not carry an Activation identity'); + } + return { + sessionId: requireIdentifier(input.sessionId, 'Session identity'), + agentId: requireIdentifier(input.agentId, 'Agent identity'), + checkpoint: admitStoredCheckpoint(input.checkpoint), + ...(input.lastCommittedActivationId === undefined + ? {} + : { + lastCommittedActivationId: requireIdentifier( + input.lastCommittedActivationId, + 'Activation identity', + ), + }), + ...(forkedFrom === undefined ? {} : { forkedFrom }), + ...(createdByForkId === undefined ? {} : { createdByForkId }), + }; +} + +function admitClaimForkInput(input: ClaimForkInput): InternalClaimForkInput { + if (!isRecord(input)) throw new TypeError('Fork claim input must be an object'); + return { + forkId: requireIdentifier(input.forkId, 'Fork identity'), + source: admitRevisionRef(input.source, 'Fork source'), + targetSessionId: requireIdentifier(input.targetSessionId, 'Fork target Session identity'), + }; +} + +function admitStoredCheckpoint(input: StoredSessionCheckpoint): StoredSessionCheckpoint { + if (!isRecord(input)) throw new TypeError('Stored Session checkpoint must be an object'); + const manifest = admitImmutableObjectRef(input.manifest); + if (!isRecord(input.value) || input.value.schemaVersion !== 1) { + throw new TypeError('Session checkpoint Manifest schema version must be 1'); + } + const value = createSessionCheckpointManifestV1( + admitImmutableObjectRef(input.value.compatibilityBundle), + ); + if (manifest.mediaType !== SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE) { + throw new TypeError('Session checkpoint Manifest has an unsupported media type'); + } + const bytes = encodeSessionCheckpointManifestV1(value); + if (manifest.digest !== digestBytes(bytes) || manifest.bytes !== bytes.byteLength) { + throw repositoryError('integrity_mismatch', 'Manifest value does not match its reference'); + } + return { manifest, value }; +} + +function admitImmutableObjectInput(input: ImmutableObjectInput): ImmutableObjectInput { + if (!isRecord(input) || !isSha256Digest(input.digest) || !isByteCount(input.bytes)) { + throw new TypeError('Immutable object input is invalid'); + } + const mediaType = requireIdentifier(input.mediaType, 'Immutable object media type'); + if (!isRecord(input.source)) throw new TypeError('Immutable object source is invalid'); + if (input.source.kind === 'file') { + return { + ...input, + mediaType, + source: { + kind: 'file', + path: requireIdentifier(input.source.path, 'Object source path', MAX_OBJECT_REF_LENGTH), + }, + }; + } + if (input.source.kind === 'bytes' && input.source.value instanceof Uint8Array) { + return { + ...input, + mediaType, + source: { kind: 'bytes', value: Uint8Array.from(input.source.value) }, + }; + } + throw new TypeError('Immutable object source is invalid'); +} + +function admitImmutableObjectMaterializationInput( + input: ImmutableObjectMaterializationInput, +): ImmutableObjectMaterializationInput { + if (!isRecord(input)) { + throw new TypeError('Immutable object materialization input must be an object'); + } + if (!isByteCount(input.maxBytes)) { + throw new TypeError( + 'Immutable object materialization byte limit must be a non-negative safe integer', + ); + } + return { + ref: admitImmutableObjectRef(input.ref), + destination: requireIdentifier( + input.destination, + 'Immutable object materialization destination', + MAX_OBJECT_REF_LENGTH, + ), + maxBytes: input.maxBytes, + }; +} + +function admitImmutableObjectRef(input: ImmutableObjectRef): ImmutableObjectRef { + if (!isRecord(input) || !isSha256Digest(input.digest) || !isByteCount(input.bytes)) { + throw new TypeError('Immutable object reference is invalid'); + } + return immutableObjectRef({ + objectRef: requireIdentifier( + input.objectRef, + 'Immutable object reference', + MAX_OBJECT_REF_LENGTH, + ), + digest: input.digest, + bytes: input.bytes, + mediaType: requireIdentifier(input.mediaType, 'Immutable object media type'), + }); +} + +function admitRevisionRef(input: SessionRevisionRef, label: string): SessionRevisionRef { + if (!isRecord(input)) throw new TypeError(`${label} must be an object`); + return copyRevisionRef({ + sessionId: requireIdentifier(input.sessionId, `${label} Session identity`), + revision: requireIdentifier(input.revision, `${label} revision`), + }); +} + +function committedRevision(input: { + readonly sessionId: string; + readonly revision: string; + readonly agentId: string; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly forkedFrom?: SessionRevisionRef; +}): CommittedSessionRevision { + return { + ref: copyRevisionRef({ sessionId: input.sessionId, revision: input.revision }), + agentId: input.agentId, + checkpoint: admitStoredCheckpoint(input.checkpoint), + ...(input.lastCommittedActivationId === undefined + ? {} + : { lastCommittedActivationId: input.lastCommittedActivationId }), + ...(input.forkedFrom === undefined ? {} : { forkedFrom: copyRevisionRef(input.forkedFrom) }), + }; +} + +function copyCommittedSessionRevision(input: CommittedSessionRevision): CommittedSessionRevision { + return committedRevision({ + sessionId: input.ref.sessionId, + revision: input.ref.revision, + agentId: input.agentId, + checkpoint: input.checkpoint, + ...(input.lastCommittedActivationId === undefined + ? {} + : { lastCommittedActivationId: input.lastCommittedActivationId }), + ...(input.forkedFrom === undefined ? {} : { forkedFrom: input.forkedFrom }), + }); +} + +function immutableObjectRef(input: ImmutableObjectRef): ImmutableObjectRef { + return { + objectRef: input.objectRef, + digest: input.digest, + bytes: input.bytes, + mediaType: input.mediaType, + }; +} + +function copyRevisionRef(input: SessionRevisionRef): SessionRevisionRef { + return { sessionId: input.sessionId, revision: input.revision }; +} + +function copyForkOperation(input: PersistentFork): ForkOperation { + return input.state === 'pending' + ? { + state: 'pending', + forkId: input.forkId, + source: copyRevisionRef(input.source), + sourceAgentId: input.sourceAgentId, + sourceCheckpoint: admitStoredCheckpoint(input.sourceCheckpoint), + targetSessionId: input.targetSessionId, + } + : copyCompletedForkOperation(input); +} + +function copyCompletedForkOperation(input: PersistentCompletedFork): CompletedForkOperation { + return { + state: 'completed', + forkId: input.forkId, + source: copyRevisionRef(input.source), + sourceAgentId: input.sourceAgentId, + sourceCheckpoint: admitStoredCheckpoint(input.sourceCheckpoint), + targetSessionId: input.targetSessionId, + target: copyRevisionRef(input.target), + }; +} + +function sameStoredCheckpoint( + left: StoredSessionCheckpoint, + right: StoredSessionCheckpoint, +): boolean { + return ( + sameImmutableObjectRef(left.manifest, right.manifest) && + sameImmutableObjectRef(left.value.compatibilityBundle, right.value.compatibilityBundle) + ); +} + +function sameImmutableObjectRef(left: ImmutableObjectRef, right: ImmutableObjectRef): boolean { + return ( + left.objectRef === right.objectRef && + left.digest === right.digest && + left.bytes === right.bytes && + left.mediaType === right.mediaType + ); +} + +function sameRevisionRef(left: SessionRevisionRef, right: SessionRevisionRef): boolean { + return left.sessionId === right.sessionId && left.revision === right.revision; +} + +function sameOptionalRevisionRef( + left: SessionRevisionRef | undefined, + right: SessionRevisionRef | undefined, +): boolean { + return left === undefined || right === undefined ? left === right : sameRevisionRef(left, right); +} + +async function writePublishedObject( + input: ImmutableObjectInput, + destination: string, +): Promise { + if (input.source.kind === 'bytes') { + const bytes = Uint8Array.from(input.source.value); + assertImmutableBytesMatch(bytes, input.digest, input.bytes); + await writeNewFile(destination, async (handle) => writeAll(handle, bytes)); + return; + } + await copyImmutableFile(input.source.path, destination, { + digest: input.digest, + bytes: input.bytes, + }); +} + +/** + * Copies and verifies in chunks. The supplied reference is deliberately the + * bound: a malicious or changing source can never make this adapter retain an + * unbounded in-memory buffer or publish more bytes than it declared. + */ +async function copyImmutableFile( + source: string, + destination: string, + ref: Pick, +): Promise { + await writeNewFile(destination, async (handle) => { + const digest = createHash('sha256'); + let bytes = 0; + for await (const rawChunk of createReadStream(source)) { + const chunk = Buffer.from(rawChunk); + bytes += chunk.byteLength; + if (bytes > ref.bytes) { + throw repositoryError( + 'integrity_mismatch', + 'Immutable object bytes exceed declared metadata', + ); + } + digest.update(chunk); + await writeAll(handle, chunk); + } + assertImmutableDigestMatch(bytes, digest.digest('hex'), ref.bytes, ref.digest); + }); +} + +async function assertFileMatchesImmutableRef(path: string, ref: ImmutableObjectRef): Promise { + const digest = createHash('sha256'); + let bytes = 0; + for await (const rawChunk of createReadStream(path)) { + const chunk = Buffer.from(rawChunk); + bytes += chunk.byteLength; + if (bytes > ref.bytes) { + throw repositoryError( + 'integrity_mismatch', + 'Immutable object bytes no longer match metadata', + ); + } + digest.update(chunk); + } + assertImmutableDigestMatch(bytes, digest.digest('hex'), ref.bytes, ref.digest); +} + +async function writeNewFile( + path: string, + writer: (handle: FileHandle) => Promise, +): Promise { + let handle: FileHandle | undefined; + try { + handle = await open(path, 'wx', 0o600); + await writer(handle); + await handle.sync(); + } catch (error) { + if (handle) { + await handle.close().catch(() => {}); + handle = undefined; + await rm(path, { force: true }).catch(() => {}); + } + throw error; + } finally { + if (handle) await handle.close(); + } +} + +async function writeAll(handle: FileHandle, value: Uint8Array): Promise { + const bytes = Buffer.from(value); + let offset = 0; + while (offset < bytes.byteLength) { + const result = await handle.write(bytes, offset, bytes.byteLength - offset, null); + if (result.bytesWritten <= 0) { + throw repositoryError('io_failure', 'Immutable object write made no progress'); + } + offset += result.bytesWritten; + } +} + +function assertImmutableBytesMatch( + bytes: Uint8Array, + expectedDigest: Sha256Digest, + expectedBytes: number, +): void { + if (bytes.byteLength !== expectedBytes || digestBytes(bytes) !== expectedDigest) { + throw repositoryError( + 'integrity_mismatch', + 'Immutable object bytes do not match declared metadata', + ); + } +} + +function assertImmutableDigestMatch( + actualBytes: number, + digestHex: string, + expectedBytes: number, + expectedDigest: Sha256Digest, +): void { + const actualDigest = `sha256:${digestHex}`; + if (actualBytes !== expectedBytes || actualDigest !== expectedDigest) { + throw repositoryError('integrity_mismatch', 'Immutable object bytes no longer match metadata'); + } +} + +function localObjectRef(digest: Sha256Digest, mediaType: string): string { + return `maka-local-object://v1/${createHash('sha256').update(`${digest}\u0000${mediaType}`).digest('hex')}`; +} + +function objectPath(objectsRoot: string, ref: ImmutableObjectRef): string { + const id = ref.objectRef.slice('maka-local-object://v1/'.length); + return join(objectsRoot, id.slice(0, 2), id); +} + +async function linkNoReplace(source: string, destination: string): Promise { + const { link } = await import('node:fs/promises'); + await link(source, destination); +} + +async function writeStateAtomically(path: string, state: PersistentState): Promise { + const temporary = `${path}.${randomUUID()}.tmp`; + const bytes = Buffer.from(`${JSON.stringify(state)}\n`, 'utf8'); + try { + const handle = await open( + temporary, + fsConstants.O_CREAT | fsConstants.O_EXCL | fsConstants.O_WRONLY, + 0o600, + ); + try { + await handle.writeFile(bytes); + await handle.sync(); + } finally { + await handle.close(); + } + await rename(temporary, path); + await syncDirectory(dirname(path)); + } finally { + await rm(temporary, { force: true }).catch(() => {}); + } +} + +function decodeState(bytes: Uint8Array): PersistentState { + let parsed: unknown; + try { + parsed = JSON.parse(new TextDecoder().decode(bytes)); + } catch (error) { + throw repositoryError( + 'integrity_mismatch', + 'Local Session Repository state is invalid JSON', + error, + ); + } + if ( + !isRecord(parsed) || + parsed.schemaVersion !== 1 || + !Array.isArray(parsed.sessions) || + !Array.isArray(parsed.commits) || + !Array.isArray(parsed.forks) + ) { + throw repositoryError('integrity_mismatch', 'Local Session Repository state schema is invalid'); + } + try { + const state: PersistentState = { + schemaVersion: 1, + sessions: parsed.sessions.map(decodeSession), + commits: parsed.commits.map(decodeCommit), + forks: parsed.forks.map(decodeFork), + }; + assertUnique(state.sessions, (entry) => entry.sessionId, 'Session identity'); + assertUnique( + state.commits, + (entry) => JSON.stringify([entry.sessionId, entry.commitId]), + 'Commit identity', + ); + assertUnique(state.forks, (entry) => entry.forkId, 'Fork identity'); + assertPersistentStateConsistency(state); + return state; + } catch (error) { + if (error instanceof SessionRepositoryError) throw error; + throw repositoryError('integrity_mismatch', 'Local Session Repository state is invalid', error); + } +} + +function decodeSession(value: unknown): PersistentSession { + if (!isRecord(value)) throw new TypeError('Session state is invalid'); + const sessionId = requireIdentifier(value.sessionId, 'Session identity'); + const agentId = requireIdentifier(value.agentId, 'Agent identity'); + const forkedFrom = + value.forkedFrom === undefined + ? undefined + : admitRevisionRef(value.forkedFrom as SessionRevisionRef, 'Fork source'); + const createdByForkId = + value.createdByForkId === undefined + ? undefined + : requireIdentifier(value.createdByForkId, 'Fork identity'); + if ((forkedFrom === undefined) !== (createdByForkId === undefined)) { + throw new TypeError('Session Fork lineage and Fork identity must be supplied together'); + } + const head = decodeCommitted(value.head); + const createdRevision = decodeCommitted(value.createdRevision); + if ( + head.ref.sessionId !== sessionId || + head.agentId !== agentId || + createdRevision.ref.sessionId !== sessionId || + createdRevision.agentId !== agentId + ) { + throw new TypeError('Session state identity binding is invalid'); + } + return { + sessionId, + agentId, + head, + nextRevisionNumber: requireRevisionNumber(value.nextRevisionNumber), + ...(forkedFrom === undefined ? {} : { forkedFrom }), + ...(createdByForkId === undefined ? {} : { createdByForkId }), + createdRevision, + }; +} + +function decodeCommit(value: unknown): PersistentCommit { + if (!isRecord(value)) throw new TypeError('Commit state is invalid'); + const input = admitCommitSessionRevisionInput(value.input as CommitSessionRevisionInput); + if (input.commitId === undefined) throw new TypeError('Commit state lacks identity'); + const sessionId = requireIdentifier(value.sessionId, 'Session identity'); + const commitId = requireIdentifier(value.commitId, 'Commit identity'); + const result = decodeCommitted(value.result); + if ( + input.sessionId !== sessionId || + input.commitId !== commitId || + result.ref.sessionId !== sessionId + ) { + throw new TypeError('Commit state identity binding is invalid'); + } + return { + sessionId, + commitId, + input, + result, + }; +} + +function decodeFork(value: unknown): PersistentFork { + if (!isRecord(value)) throw new TypeError('Fork state is invalid'); + const base = { + forkId: requireIdentifier(value.forkId, 'Fork identity'), + source: admitRevisionRef(value.source as SessionRevisionRef, 'Fork source'), + sourceAgentId: requireIdentifier(value.sourceAgentId, 'Fork source Agent identity'), + sourceCheckpoint: admitStoredCheckpoint(value.sourceCheckpoint as StoredSessionCheckpoint), + targetSessionId: requireIdentifier(value.targetSessionId, 'Fork target Session identity'), + }; + if (value.state === 'pending') return { state: 'pending', ...base }; + if (value.state === 'completed') { + return { + state: 'completed', + ...base, + target: admitRevisionRef(value.target as SessionRevisionRef, 'Fork target'), + }; + } + throw new TypeError('Fork state is invalid'); +} + +function decodeCommitted(value: unknown): CommittedSessionRevision { + if (!isRecord(value)) throw new TypeError('Committed revision is invalid'); + return committedRevision({ + sessionId: admitRevisionRef(value.ref as SessionRevisionRef, 'Session revision reference') + .sessionId, + revision: admitRevisionRef(value.ref as SessionRevisionRef, 'Session revision reference') + .revision, + agentId: requireIdentifier(value.agentId, 'Agent identity'), + checkpoint: admitStoredCheckpoint(value.checkpoint as StoredSessionCheckpoint), + ...(value.lastCommittedActivationId === undefined + ? {} + : { + lastCommittedActivationId: requireIdentifier( + value.lastCommittedActivationId, + 'Activation identity', + ), + }), + ...(value.forkedFrom === undefined + ? {} + : { forkedFrom: admitRevisionRef(value.forkedFrom as SessionRevisionRef, 'Fork source') }), + }); +} + +function requireRevisionNumber(value: unknown): number { + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 2) + throw new TypeError('Next revision number is invalid'); + return value; +} + +/** + * JSON syntax and individual field validation are not sufficient for the + * control document: the records must describe one non-contradictory Session + * history before any writer derives a new revision from it. + */ +function assertPersistentStateConsistency(state: PersistentState): void { + for (const session of state.sessions) { + assertPersistentSessionConsistency(session); + assertPersistentCreatedForkTarget(state, session); + } + + const committedRevisions = new Set(); + for (const commit of state.commits) { + const session = findSession(state, commit.sessionId); + if (!session) throw new TypeError('Commit receipt references an unknown Session'); + const expectedRevision = revisionNumber( + commit.input.expectedRevision, + 'Commit expected revision', + ); + const resultRevision = revisionNumber(commit.result.ref.revision, 'Commit result revision'); + const headRevision = revisionNumber(session.head.ref.revision, 'Session head revision'); + if ( + resultRevision < 2 || + resultRevision > headRevision || + expectedRevision + 1 !== resultRevision || + commit.result.agentId !== session.agentId || + !sameStoredCheckpoint(commit.input.checkpoint, commit.result.checkpoint) || + commit.input.lastCommittedActivationId !== commit.result.lastCommittedActivationId || + !sameOptionalRevisionRef(commit.result.forkedFrom, session.forkedFrom) || + (resultRevision === headRevision && + !sameCommittedSessionRevision(commit.result, session.head)) + ) { + throw new TypeError('Commit receipt contradicts Session state'); + } + const key = `${commit.sessionId}\u0000${commit.result.ref.revision}`; + if (committedRevisions.has(key)) throw new TypeError('Commit result revision is duplicated'); + committedRevisions.add(key); + } + + for (const fork of state.forks) assertPersistentForkConsistency(state, fork); +} + +function assertPersistentSessionConsistency(session: PersistentSession): void { + const createdRevision = revisionNumber( + session.createdRevision.ref.revision, + 'Session creation revision', + ); + const headRevision = revisionNumber(session.head.ref.revision, 'Session head revision'); + if ( + createdRevision !== 1 || + headRevision >= Number.MAX_SAFE_INTEGER || + session.nextRevisionNumber !== headRevision + 1 || + !sameOptionalRevisionRef(session.createdRevision.forkedFrom, session.forkedFrom) || + !sameOptionalRevisionRef(session.head.forkedFrom, session.forkedFrom) || + (session.createdByForkId !== undefined && + session.createdRevision.lastCommittedActivationId !== undefined) + ) { + throw new TypeError('Session revision sequence is inconsistent'); + } + if (headRevision === 1 && !sameCommittedSessionRevision(session.head, session.createdRevision)) { + throw new TypeError('Session initial head contradicts creation revision'); + } +} + +function assertPersistentForkConsistency(state: PersistentState, fork: PersistentFork): void { + if (fork.source.sessionId === fork.targetSessionId) { + throw new TypeError('Fork target Session must differ from its source Session'); + } + const source = findSession(state, fork.source.sessionId); + if (!source || source.agentId !== fork.sourceAgentId) { + throw new TypeError('Fork source Agent binding is inconsistent'); + } + const target = findSession(state, fork.targetSessionId); + if (fork.state === 'completed') { + if ( + !target || + target.agentId !== fork.sourceAgentId || + target.createdByForkId !== fork.forkId || + !target.forkedFrom || + !sameRevisionRef(target.forkedFrom, fork.source) || + !sameRevisionRef(fork.target, target.createdRevision.ref) + ) { + throw new TypeError('Completed Fork target is inconsistent'); + } + } +} + +function assertPersistentCreatedForkTarget( + state: PersistentState, + session: PersistentSession, +): void { + if (session.createdByForkId === undefined || session.forkedFrom === undefined) return; + const fork = findFork(state, session.createdByForkId); + if ( + !fork || + fork.targetSessionId !== session.sessionId || + fork.sourceAgentId !== session.agentId || + !sameRevisionRef(fork.source, session.forkedFrom) + ) { + throw new TypeError('Fork-created Session does not match its claimed operation'); + } +} + +function revisionNumber(revision: string, label: string): number { + const match = /^r([1-9][0-9]*)$/u.exec(revision); + if (!match) throw new TypeError(`${label} is not a canonical revision`); + const value = Number(match[1]); + if (!Number.isSafeInteger(value)) throw new TypeError(`${label} is outside the safe range`); + return value; +} + +function sameCommittedSessionRevision( + left: CommittedSessionRevision, + right: CommittedSessionRevision, +): boolean { + return ( + sameRevisionRef(left.ref, right.ref) && + left.agentId === right.agentId && + sameStoredCheckpoint(left.checkpoint, right.checkpoint) && + left.lastCommittedActivationId === right.lastCommittedActivationId && + sameOptionalRevisionRef(left.forkedFrom, right.forkedFrom) + ); +} + +function assertUnique(values: readonly T[], key: (value: T) => string, label: string): void { + const seen = new Set(); + for (const value of values) { + const current = key(value); + if (seen.has(current)) throw new TypeError(`${label} is duplicated`); + seen.add(current); + } +} + +function requireIdentifier( + value: unknown, + label: string, + maximumLength = MAX_IDENTIFIER_LENGTH, +): string { + if (!isNonEmptyUnicodeString(value) || value.length > maximumLength) { + throw new TypeError(`${label} must be a bounded non-empty Unicode string`); + } + return value; +} + +function isByteCount(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function digestBytes(value: Uint8Array): Sha256Digest { + return `sha256:${createHash('sha256').update(value).digest('hex')}` as Sha256Digest; +} + +function repositoryError( + code: SessionRepositoryErrorCode, + message: string, + cause?: unknown, +): SessionRepositoryError { + return new SessionRepositoryError(code, message, cause === undefined ? {} : { cause }); +} + +function normalizeFileError(error: unknown, message: string): SessionRepositoryError { + if (error instanceof SessionRepositoryError) return error; + return repositoryError('io_failure', message, error); +} + +function isNodeError(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && 'code' in error && error.code === code; +} diff --git a/packages/storage/src/session-repository.ts b/packages/storage/src/session-repository.ts new file mode 100644 index 0000000000..bff1517a2d --- /dev/null +++ b/packages/storage/src/session-repository.ts @@ -0,0 +1,1229 @@ +/* + * 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 { open, readFile, rm, type FileHandle } from 'node:fs/promises'; +import { + isNonEmptyUnicodeString, + isSha256Digest, + type SessionBundleArtifact, + type SessionBundleSource, + type Sha256Digest, +} from './session-bundle-contract.js'; + +const MAX_IDENTIFIER_LENGTH = 512; +const MAX_OBJECT_REF_LENGTH = 2_048; + +export const SESSION_BUNDLE_OBJECT_MEDIA_TYPE = + 'application/vnd.maka.session-bundle+tar;version=1;compression=zstd' as const; +export const SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE = + 'application/vnd.maka.session-checkpoint-manifest+json;version=1' as const; + +/** + * A Repository revision is opaque to callers. Revisions are scoped to one + * Cloud Session and must never be reused as object digests or across Sessions. + */ +export type SessionRepositoryRevision = string; + +export interface SessionRevisionRef { + readonly sessionId: string; + readonly revision: SessionRepositoryRevision; +} + +/** Trusted metadata for one immutable object. */ +export interface ImmutableObjectRef { + readonly objectRef: string; + readonly digest: Sha256Digest; + readonly bytes: number; + readonly mediaType: string; +} + +export type ImmutableObjectSource = + | { + readonly kind: 'file'; + readonly path: string; + } + | { + readonly kind: 'bytes'; + readonly value: Uint8Array; + }; + +export interface ImmutableObjectInput { + readonly digest: Sha256Digest; + readonly bytes: number; + readonly mediaType: string; + readonly source: ImmutableObjectSource; +} + +/** Caller-owned, bounded materialization target for one immutable object. */ +export interface ImmutableObjectMaterializationInput { + readonly ref: ImmutableObjectRef; + /** A new private file path. Materialization must never overwrite it. */ + readonly destination: string; + /** The caller's maximum acceptable byte count for this materialization. */ + readonly maxBytes: number; +} + +/** + * Large immutable bytes live behind this port. Its publication semantics are + * deliberately distinct from the Repository's head-CAS semantics. + */ +export interface ImmutableObjectStore { + /** + * Publishes bytes under a non-overwritable reference. Returning an existing + * reference for identical input is allowed, but the exact bytes and declared + * metadata must already be durably readable before this method returns. + */ + publish(input: ImmutableObjectInput): Promise; + /** Verifies the exact reference, byte count, media type, and digest. */ + assertReadable(ref: ImmutableObjectRef): Promise; + /** + * Materializes verified bytes at a caller-owned new file path without + * overwriting it. The operation must reject objects over maxBytes and verify + * the exact reference, byte count, and digest before it returns. + */ + materialize(input: ImmutableObjectMaterializationInput): Promise; +} + +export interface SessionCheckpointManifestV1 { + readonly schemaVersion: 1; + readonly compatibilityBundle: ImmutableObjectRef; +} + +/** + * The value is carried with its immutable reference so a Repository can + * validate the canonical Manifest digest without acquiring a general read API. + */ +export interface StoredSessionCheckpoint { + readonly manifest: ImmutableObjectRef; + readonly value: SessionCheckpointManifestV1; +} + +export interface CommittedSessionRevision { + readonly ref: SessionRevisionRef; + readonly agentId: string; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly forkedFrom?: SessionRevisionRef; +} + +export interface CommitSessionRevisionInput { + readonly sessionId: string; + readonly expectedRevision: SessionRepositoryRevision; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + /** + * Optional caller operation identity. Retrying the same identity and input + * returns the original committed revision rather than allocating another. + */ + readonly commitId?: string; +} + +export interface CreateSessionInput { + readonly sessionId: string; + readonly agentId: string; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly forkedFrom?: SessionRevisionRef; + /** Required for a Fork-created target Session. */ + readonly createdByForkId?: string; +} + +export interface ClaimForkInput { + readonly forkId: string; + readonly source: SessionRevisionRef; + readonly targetSessionId: string; +} + +export interface CompleteForkInput { + readonly forkId: string; +} + +export interface PendingForkOperation { + readonly state: 'pending'; + readonly forkId: string; + readonly source: SessionRevisionRef; + /** Captured from the verified source Session when the Fork is first claimed. */ + readonly sourceAgentId: string; + /** + * The exact source checkpoint admitted by the claim. V1 retains only the + * current head, so a later source advance must not make a pending Fork + * unable to recover its source bytes and metadata. + */ + readonly sourceCheckpoint: StoredSessionCheckpoint; + readonly targetSessionId: string; +} + +export interface CompletedForkOperation extends Omit { + readonly state: 'completed'; + readonly target: SessionRevisionRef; +} + +export type ForkOperation = PendingForkOperation | CompletedForkOperation; + +/** + * V1 never expires completed Fork identities. A production Repository must + * retain this mapping durably; the in-memory conformance implementation does + * so for its lifetime. + */ +export type ForkIdempotencyRetention = 'indefinite'; + +/** + * Strongly consistent Session metadata and operation records live behind this + * port. Immutable object publication remains a separate prerequisite. + */ +export interface SessionRepository { + readonly forkIdempotencyRetention: ForkIdempotencyRetention; + /** + * Resolves the current head and verifies its checkpoint as one Repository + * read. Callers that do not already hold a revision must not maintain an + * independent current-head record. + */ + checkoutCurrent(sessionId: string): Promise; + checkoutExact(ref: SessionRevisionRef): Promise; + createSession(input: CreateSessionInput): Promise; + commit(input: CommitSessionRevisionInput): Promise; + claimFork(input: ClaimForkInput): Promise; + completeFork(input: CompleteForkInput): Promise; +} + +export type SessionRepositoryErrorCode = + | 'session_not_found' + | 'source_revision_not_available' + | 'revision_not_available' + | 'revision_conflict' + | 'session_already_exists' + | 'idempotency_conflict' + | 'invalid_fork_target' + | 'fork_agent_mismatch' + | 'object_not_found' + | 'integrity_mismatch' + | 'quota_exceeded' + | 'io_failure'; + +export class SessionRepositoryError extends Error { + constructor( + readonly code: SessionRepositoryErrorCode, + message: string, + options: ErrorOptions = {}, + ) { + super(message, options.cause === undefined ? undefined : { cause: options.cause }); + this.name = 'SessionRepositoryError'; + } +} + +export interface PublishSessionCheckpointV1Input { + readonly objectStore: ImmutableObjectStore; + readonly compatibilityBundle: SessionBundleArtifact; +} + +/** + * Publishes and verifies the compatibility Bundle before publishing and + * verifying the immutable Manifest that names it. The returned checkpoint is + * suitable for a later Repository create or head-CAS operation. + */ +export async function publishSessionCheckpointV1( + input: PublishSessionCheckpointV1Input, +): Promise { + if (!isRecord(input)) + throw new TypeError('Session checkpoint publication input must be an object'); + const objectStore = requireImmutableObjectStore(input.objectStore); + const artifact = admitSessionBundleArtifact(input.compatibilityBundle); + const compatibilityBundle = await publishVerifiedObject(objectStore, { + digest: artifact.archiveDigest, + bytes: artifact.compressedBytes, + mediaType: SESSION_BUNDLE_OBJECT_MEDIA_TYPE, + source: { kind: 'file', path: artifact.path }, + }); + const value = createSessionCheckpointManifestV1(compatibilityBundle); + const manifestBytes = encodeSessionCheckpointManifestV1(value); + const manifest = await publishVerifiedObject(objectStore, { + digest: digestBytes(manifestBytes), + bytes: manifestBytes.byteLength, + mediaType: SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE, + source: { kind: 'bytes', value: manifestBytes }, + }); + return storedSessionCheckpoint({ manifest, value }); +} + +/** + * Converts a retained V1 compatibility Bundle reference into the exact + * filesystem source expected by the Bundle inspect/hydrate boundary. + * + * This is deliberately a materialization seam, not a general Repository read: + * callers own the bounded destination and clean it up after hydration/repack. + */ +export async function materializeSessionCheckpointV1(input: { + readonly objectStore: ImmutableObjectStore; + readonly checkpoint: StoredSessionCheckpoint; + readonly destination: string; + readonly maxBytes: number; +}): Promise { + if (!isRecord(input)) + throw new TypeError('Session checkpoint materialization input must be an object'); + const objectStore = requireImmutableObjectStore(input.objectStore); + const checkpoint = admitStoredSessionCheckpoint(input.checkpoint); + const request = admitImmutableObjectMaterializationInput({ + ref: checkpoint.value.compatibilityBundle, + destination: input.destination, + maxBytes: input.maxBytes, + }); + try { + await objectStore.assertReadable(checkpoint.manifest); + await objectStore.assertReadable(request.ref); + await objectStore.materialize(request); + return Object.freeze({ + path: request.destination, + expectedArchiveDigest: request.ref.digest, + }); + } catch (error) { + throw normalizeObjectStoreError(error); + } +} + +export function createSessionCheckpointManifestV1( + compatibilityBundle: ImmutableObjectRef, +): SessionCheckpointManifestV1 { + const admitted = admitImmutableObjectRef(compatibilityBundle); + if (admitted.mediaType !== SESSION_BUNDLE_OBJECT_MEDIA_TYPE) { + throw new TypeError('V1 compatibility Bundle has an unsupported media type'); + } + return Object.freeze({ + schemaVersion: 1, + compatibilityBundle: copyImmutableObjectRef(admitted), + }); +} + +/** RFC 8785/JCS V1 encoding used to bind a Manifest value to its object digest. */ +export function encodeSessionCheckpointManifestV1(input: SessionCheckpointManifestV1): Uint8Array { + const value = admitSessionCheckpointManifestV1(input); + return new TextEncoder().encode( + JSON.stringify({ + compatibilityBundle: { + bytes: value.compatibilityBundle.bytes, + digest: value.compatibilityBundle.digest, + mediaType: value.compatibilityBundle.mediaType, + objectRef: value.compatibilityBundle.objectRef, + }, + schemaVersion: value.schemaVersion, + }), + ); +} + +export function createInMemoryImmutableObjectStore(): ImmutableObjectStore { + return new InMemoryImmutableObjectStore(); +} + +export interface CreateInMemorySessionRepositoryOptions { + readonly objectStore: ImmutableObjectStore; +} + +/** + * Deterministic conformance implementation for coordinators and Fork tests. + * It models the V1 single-current-head retention policy, but it is not a + * durable control-plane backend. + */ +export function createInMemorySessionRepository( + options: CreateInMemorySessionRepositoryOptions, +): SessionRepository { + if (!isRecord(options)) throw new TypeError('In-memory Repository options must be an object'); + return new InMemorySessionRepository(requireImmutableObjectStore(options.objectStore)); +} + +class InMemorySessionRepository implements SessionRepository { + readonly forkIdempotencyRetention = 'indefinite' as const; + + private readonly sessions = new Map(); + private readonly commitsBySession = new Map>(); + private readonly forks = new Map(); + + constructor(private readonly objectStore: ImmutableObjectStore) {} + + async checkoutCurrent(sessionId: string): Promise { + const admittedSessionId = requireIdentifier(sessionId, 'Session identity'); + const session = this.sessions.get(admittedSessionId); + if (!session) throw repositoryError('session_not_found', 'Cloud Session was not found'); + + // Capture exactly the current head selected by this read before awaiting + // object verification. A subsequent writer cannot substitute its head. + const committed = copyCommittedSessionRevision(session.head); + await this.assertCheckpointReadable(committed.checkpoint); + return committed; + } + + async checkoutExact(ref: SessionRevisionRef): Promise { + const requested = admitRevisionRef(ref, 'Session revision reference'); + const session = this.sessions.get(requested.sessionId); + if (!session) throw repositoryError('session_not_found', 'Cloud Session was not found'); + if (session.head.ref.revision !== requested.revision) { + throw repositoryError( + 'revision_not_available', + 'Requested Session revision is not available', + ); + } + + // Capture the exact record before asynchronous object reads. A later head + // advance may make this revision non-current, but cannot substitute bytes. + const committed = copyCommittedSessionRevision(session.head); + await this.assertCheckpointReadable(committed.checkpoint); + return committed; + } + + async commit(input: CommitSessionRevisionInput): Promise { + const admitted = admitCommitSessionRevisionInput(input); + const prior = this.reconcilePriorCommit(admitted); + if (prior) { + await this.assertCheckpointReadable(prior.checkpoint); + return prior; + } + + const initial = this.sessions.get(admitted.sessionId); + if (!initial) throw repositoryError('session_not_found', 'Cloud Session was not found'); + if (initial.head.ref.revision !== admitted.expectedRevision) { + throw repositoryError('revision_conflict', 'Cloud Session head changed before commit'); + } + + await this.assertCheckpointReadable(admitted.checkpoint); + + // Object verification may yield. Reconcile an identical concurrent retry, + // then recheck CAS so another commit cannot be overwritten. + const admittedWhileReading = this.reconcilePriorCommit(admitted); + if (admittedWhileReading) return admittedWhileReading; + const session = this.sessions.get(admitted.sessionId); + if (!session) throw repositoryError('session_not_found', 'Cloud Session was not found'); + if (session.head.ref.revision !== admitted.expectedRevision) { + throw repositoryError('revision_conflict', 'Cloud Session head changed before commit'); + } + + const result = committedRevision({ + sessionId: admitted.sessionId, + revision: nextRevision(session), + agentId: session.agentId, + checkpoint: admitted.checkpoint, + lastCommittedActivationId: admitted.lastCommittedActivationId, + forkedFrom: session.forkedFrom, + }); + session.head = result; + if (admitted.commitId !== undefined) { + let records = this.commitsBySession.get(admitted.sessionId); + if (!records) { + records = new Map(); + this.commitsBySession.set(admitted.sessionId, records); + } + records.set(admitted.commitId, { input: admitted, result }); + } + return copyCommittedSessionRevision(result); + } + + async createSession(input: CreateSessionInput): Promise { + const admitted = admitCreateSessionInput(input); + const existing = this.sessions.get(admitted.sessionId); + if (existing) { + const reconciled = this.reconcileExistingSessionCreate(existing, admitted); + await this.assertCheckpointReadable(reconciled.checkpoint); + return reconciled; + } + + await this.assertCheckpointReadable(admitted.checkpoint); + + // Object verification may yield. Create-if-absent is decided only after it + // returns, at this method's synchronous linearization point. + const afterVerification = this.sessions.get(admitted.sessionId); + if (afterVerification) { + const reconciled = this.reconcileExistingSessionCreate(afterVerification, admitted); + await this.assertCheckpointReadable(reconciled.checkpoint); + return reconciled; + } + + if (admitted.createdByForkId !== undefined) this.assertPendingForkCreate(admitted); + const initial = committedRevision({ + sessionId: admitted.sessionId, + revision: 'r1', + agentId: admitted.agentId, + checkpoint: admitted.checkpoint, + lastCommittedActivationId: admitted.lastCommittedActivationId, + forkedFrom: admitted.forkedFrom, + }); + this.sessions.set(admitted.sessionId, { + agentId: admitted.agentId, + head: initial, + nextRevisionNumber: 2, + forkedFrom: admitted.forkedFrom, + createdByForkId: admitted.createdByForkId, + createdRevision: initial, + }); + return copyCommittedSessionRevision(initial); + } + + async claimFork(input: ClaimForkInput): Promise { + const admitted = admitClaimForkInput(input); + const existing = this.forks.get(admitted.forkId); + if (existing) { + if (!sameForkClaim(existing, admitted)) { + throw repositoryError( + 'idempotency_conflict', + 'Fork identity was reused with different input', + ); + } + return copyForkOperation(existing); + } + if (admitted.targetSessionId === admitted.source.sessionId) { + throw repositoryError( + 'invalid_fork_target', + 'Fork target Session must differ from its source Session', + ); + } + + // The first claim is the linearization point for the source binding. It + // must prove the named revision is still current and readable before a + // retry can rely on this durable Fork record. + const initial = this.sessions.get(admitted.source.sessionId); + if (!initial || initial.head.ref.revision !== admitted.source.revision) { + throw repositoryError( + 'source_revision_not_available', + 'Fork source Session revision is not available', + ); + } + const source = copyCommittedSessionRevision(initial.head); + await this.assertCheckpointReadable(source.checkpoint); + + // Source verification may yield. Reconcile another claimant that won while + // it was in progress rather than overwriting its durable idempotency fact. + const claimedWhileReading = this.forks.get(admitted.forkId); + if (claimedWhileReading) { + if (!sameForkClaim(claimedWhileReading, admitted)) { + throw repositoryError( + 'idempotency_conflict', + 'Fork identity was reused with different input', + ); + } + return copyForkOperation(claimedWhileReading); + } + + // Only a new claim needs a current source. Keep this final check and the + // insertion synchronous, after reconciling any already admitted operation. + const afterVerification = this.sessions.get(admitted.source.sessionId); + if ( + !afterVerification || + afterVerification.head.ref.revision !== admitted.source.revision || + afterVerification.agentId !== source.agentId + ) { + throw repositoryError( + 'source_revision_not_available', + 'Fork source Session revision is no longer current', + ); + } + const pending: InternalPendingForkOperation = { + state: 'pending', + forkId: admitted.forkId, + source: admitted.source, + sourceAgentId: source.agentId, + sourceCheckpoint: source.checkpoint, + targetSessionId: admitted.targetSessionId, + }; + this.forks.set(admitted.forkId, pending); + return copyForkOperation(pending); + } + + async completeFork(input: CompleteForkInput): Promise { + const forkId = requireIdentifier(input?.forkId, 'Fork identity'); + const operation = this.forks.get(forkId); + if (!operation) throw repositoryError('idempotency_conflict', 'Fork identity was not claimed'); + if (operation.state === 'completed') return copyCompletedForkOperation(operation); + + const target = this.sessions.get(operation.targetSessionId); + if (!target) throw repositoryError('session_not_found', 'Fork target Session was not found'); + if (target.agentId !== operation.sourceAgentId) { + throw repositoryError( + 'fork_agent_mismatch', + 'Fork target Agent does not match its verified source Agent', + ); + } + if ( + target.createdByForkId !== operation.forkId || + !target.forkedFrom || + !sameRevisionRef(target.forkedFrom, operation.source) + ) { + throw repositoryError('idempotency_conflict', 'Fork target was created by another operation'); + } + await this.assertCheckpointReadable(target.createdRevision.checkpoint); + + const completed: InternalCompletedForkOperation = { + ...operation, + state: 'completed', + target: copyRevisionRef(target.createdRevision.ref), + }; + this.forks.set(forkId, completed); + return copyCompletedForkOperation(completed); + } + + private reconcilePriorCommit( + input: InternalCommitSessionRevisionInput, + ): CommittedSessionRevision | undefined { + if (input.commitId === undefined) return undefined; + const prior = this.commitsBySession.get(input.sessionId)?.get(input.commitId); + if (!prior) return undefined; + if (!sameCommitInput(prior.input, input)) { + throw repositoryError( + 'idempotency_conflict', + 'Commit identity was reused with different input', + ); + } + return copyCommittedSessionRevision(prior.result); + } + + private reconcileExistingSessionCreate( + existing: SessionState, + input: InternalCreateSessionInput, + ): CommittedSessionRevision { + if ( + input.createdByForkId !== undefined && + existing.createdByForkId === input.createdByForkId && + sameCreateInput(existing.createdRevision, existing.agentId, existing.forkedFrom, input) + ) { + return copyCommittedSessionRevision(existing.createdRevision); + } + throw repositoryError('session_already_exists', 'Cloud Session already exists'); + } + + private assertPendingForkCreate(input: InternalCreateSessionInput): void { + const operation = this.forks.get(input.createdByForkId!); + if ( + !operation || + operation.state !== 'pending' || + operation.targetSessionId !== input.sessionId || + !input.forkedFrom || + !sameRevisionRef(operation.source, input.forkedFrom) + ) { + throw repositoryError( + 'idempotency_conflict', + 'Fork target does not match its claimed operation', + ); + } + if (operation.sourceAgentId !== input.agentId) { + throw repositoryError( + 'fork_agent_mismatch', + 'Fork target Agent must match its verified source Agent', + ); + } + } + + private async assertCheckpointReadable( + input: StoredSessionCheckpoint, + ): Promise { + const checkpoint = admitStoredSessionCheckpoint(input); + try { + await this.objectStore.assertReadable(checkpoint.manifest); + await this.objectStore.assertReadable(checkpoint.value.compatibilityBundle); + return checkpoint; + } catch (error) { + throw normalizeObjectStoreError(error); + } + } +} + +class InMemoryImmutableObjectStore implements ImmutableObjectStore { + private readonly objects = new Map(); + + async publish(input: ImmutableObjectInput): Promise { + const admitted = admitImmutableObjectInput(input); + let bytes: Uint8Array; + try { + bytes = + admitted.source.kind === 'file' + ? await readFile(admitted.source.path) + : Uint8Array.from(admitted.source.value); + } catch (error) { + throw repositoryError( + 'io_failure', + 'Immutable object publication could not read bytes', + error, + ); + } + if (bytes.byteLength !== admitted.bytes || digestBytes(bytes) !== admitted.digest) { + throw repositoryError( + 'integrity_mismatch', + 'Immutable object bytes do not match declared metadata', + ); + } + + const mediaTypeKey = digestBytes(new TextEncoder().encode(admitted.mediaType)).slice( + 'sha256:'.length, + 'sha256:'.length + 16, + ); + const ref = immutableObjectRef({ + objectRef: `memory://immutable-objects/${admitted.digest.slice('sha256:'.length)}/${mediaTypeKey}`, + digest: admitted.digest, + bytes: admitted.bytes, + mediaType: admitted.mediaType, + }); + const existing = this.objects.get(ref.objectRef); + if (existing) { + if (!sameImmutableObjectRef(existing.ref, ref) || !sameBytes(existing.bytes, bytes)) { + throw repositoryError( + 'integrity_mismatch', + 'Immutable object reference already contains different bytes or metadata', + ); + } + return copyImmutableObjectRef(existing.ref); + } + this.objects.set(ref.objectRef, { ref, bytes: Uint8Array.from(bytes) }); + return copyImmutableObjectRef(ref); + } + + async assertReadable(input: ImmutableObjectRef): Promise { + const ref = admitImmutableObjectRef(input); + const stored = this.objects.get(ref.objectRef); + if (!stored) throw repositoryError('object_not_found', 'Immutable object was not found'); + if ( + !sameImmutableObjectRef(stored.ref, ref) || + stored.bytes.byteLength !== ref.bytes || + digestBytes(stored.bytes) !== ref.digest + ) { + throw repositoryError( + 'integrity_mismatch', + 'Immutable object bytes no longer match their trusted metadata', + ); + } + } + + async materialize(input: ImmutableObjectMaterializationInput): Promise { + const request = admitImmutableObjectMaterializationInput(input); + const stored = this.objects.get(request.ref.objectRef); + if (!stored) throw repositoryError('object_not_found', 'Immutable object was not found'); + if ( + !sameImmutableObjectRef(stored.ref, request.ref) || + stored.bytes.byteLength !== request.ref.bytes || + digestBytes(stored.bytes) !== request.ref.digest + ) { + throw repositoryError( + 'integrity_mismatch', + 'Immutable object bytes no longer match their trusted metadata', + ); + } + if (request.ref.bytes > request.maxBytes) { + throw repositoryError( + 'quota_exceeded', + 'Immutable object exceeds materialization byte limit', + ); + } + let handle: FileHandle | undefined; + try { + handle = await open(request.destination, 'wx', 0o600); + await handle.writeFile(stored.bytes); + await handle.close(); + handle = undefined; + } catch (error) { + // An exclusive-open failure gives us no ownership of the existing path. + if (handle) { + await handle.close().catch(() => {}); + await rm(request.destination, { force: true }).catch(() => {}); + } + throw repositoryError('io_failure', 'Immutable object could not be materialized', error); + } + } +} + +interface SessionState { + readonly agentId: string; + head: CommittedSessionRevision; + nextRevisionNumber: number; + readonly forkedFrom?: SessionRevisionRef; + readonly createdByForkId?: string; + readonly createdRevision: CommittedSessionRevision; +} + +interface InMemoryObject { + readonly ref: ImmutableObjectRef; + readonly bytes: Uint8Array; +} + +interface InternalCommitSessionRevisionInput { + readonly sessionId: string; + readonly expectedRevision: SessionRepositoryRevision; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly commitId?: string; +} + +interface InternalCreateSessionInput { + readonly sessionId: string; + readonly agentId: string; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly forkedFrom?: SessionRevisionRef; + readonly createdByForkId?: string; +} + +interface InternalClaimForkInput { + readonly forkId: string; + readonly source: SessionRevisionRef; + readonly targetSessionId: string; +} + +interface CommitRecord { + readonly input: InternalCommitSessionRevisionInput; + readonly result: CommittedSessionRevision; +} + +type InternalForkOperation = InternalPendingForkOperation | InternalCompletedForkOperation; + +interface InternalPendingForkOperation extends PendingForkOperation {} + +interface InternalCompletedForkOperation extends CompletedForkOperation {} + +async function publishVerifiedObject( + objectStore: ImmutableObjectStore, + input: ImmutableObjectInput, +): Promise { + const expected = admitImmutableObjectInput(input); + try { + const published = admitImmutableObjectRef(await objectStore.publish(expected)); + if ( + published.digest !== expected.digest || + published.bytes !== expected.bytes || + published.mediaType !== expected.mediaType + ) { + throw repositoryError( + 'integrity_mismatch', + 'Published immutable object metadata does not match its input', + ); + } + await objectStore.assertReadable(published); + return copyImmutableObjectRef(published); + } catch (error) { + throw normalizeObjectStoreError(error); + } +} + +function admitSessionBundleArtifact(input: SessionBundleArtifact): SessionBundleArtifact { + if (!isRecord(input)) throw new TypeError('Session Bundle artifact must be an object'); + const path = requireIdentifier(input.path, 'Bundle archive path', MAX_OBJECT_REF_LENGTH); + if (!isSha256Digest(input.archiveDigest)) { + throw new TypeError('Bundle archive digest must be SHA-256'); + } + if (!isByteCount(input.compressedBytes)) { + throw new TypeError('Bundle compressed byte count must be a non-negative safe integer'); + } + return { + ...input, + path, + archiveDigest: input.archiveDigest, + compressedBytes: input.compressedBytes, + }; +} + +function admitImmutableObjectInput(input: ImmutableObjectInput): ImmutableObjectInput { + if (!isRecord(input)) throw new TypeError('Immutable object input must be an object'); + if (!isSha256Digest(input.digest)) throw new TypeError('Immutable object digest must be SHA-256'); + if (!isByteCount(input.bytes)) { + throw new TypeError('Immutable object byte count must be a non-negative safe integer'); + } + const mediaType = requireIdentifier(input.mediaType, 'Immutable object media type'); + if (!isRecord(input.source)) throw new TypeError('Immutable object source must be an object'); + if (input.source.kind === 'file') { + return Object.freeze({ + digest: input.digest, + bytes: input.bytes, + mediaType, + source: Object.freeze({ + kind: 'file' as const, + path: requireIdentifier( + input.source.path, + 'Immutable object source path', + MAX_OBJECT_REF_LENGTH, + ), + }), + }); + } + if (input.source.kind === 'bytes' && input.source.value instanceof Uint8Array) { + return Object.freeze({ + digest: input.digest, + bytes: input.bytes, + mediaType, + source: Object.freeze({ kind: 'bytes' as const, value: Uint8Array.from(input.source.value) }), + }); + } + throw new TypeError('Immutable object source must contain file or byte content'); +} + +function admitImmutableObjectMaterializationInput( + input: ImmutableObjectMaterializationInput, +): ImmutableObjectMaterializationInput { + if (!isRecord(input)) + throw new TypeError('Immutable object materialization input must be an object'); + if (!isByteCount(input.maxBytes)) { + throw new TypeError( + 'Immutable object materialization byte limit must be a non-negative safe integer', + ); + } + return Object.freeze({ + ref: admitImmutableObjectRef(input.ref), + destination: requireIdentifier( + input.destination, + 'Immutable object materialization destination', + MAX_OBJECT_REF_LENGTH, + ), + maxBytes: input.maxBytes, + }); +} + +function admitImmutableObjectRef(input: ImmutableObjectRef): ImmutableObjectRef { + if (!isRecord(input)) throw new TypeError('Immutable object reference must be an object'); + const objectRef = requireIdentifier( + input.objectRef, + 'Immutable object reference', + MAX_OBJECT_REF_LENGTH, + ); + if (!isSha256Digest(input.digest)) throw new TypeError('Immutable object digest must be SHA-256'); + if (!isByteCount(input.bytes)) { + throw new TypeError('Immutable object byte count must be a non-negative safe integer'); + } + return immutableObjectRef({ + objectRef, + digest: input.digest, + bytes: input.bytes, + mediaType: requireIdentifier(input.mediaType, 'Immutable object media type'), + }); +} + +function admitSessionCheckpointManifestV1( + input: SessionCheckpointManifestV1, +): SessionCheckpointManifestV1 { + if (!isRecord(input)) throw new TypeError('Session checkpoint Manifest must be an object'); + if (input.schemaVersion !== 1) { + throw new TypeError('Session checkpoint Manifest schema version must be 1'); + } + return createSessionCheckpointManifestV1(input.compatibilityBundle); +} + +function admitStoredSessionCheckpoint(input: StoredSessionCheckpoint): StoredSessionCheckpoint { + if (!isRecord(input)) throw new TypeError('Stored Session checkpoint must be an object'); + const manifest = admitImmutableObjectRef(input.manifest); + const value = admitSessionCheckpointManifestV1(input.value); + if (manifest.mediaType !== SESSION_CHECKPOINT_MANIFEST_MEDIA_TYPE) { + throw new TypeError('Session checkpoint Manifest has an unsupported media type'); + } + const encoded = encodeSessionCheckpointManifestV1(value); + if (manifest.digest !== digestBytes(encoded) || manifest.bytes !== encoded.byteLength) { + throw repositoryError( + 'integrity_mismatch', + 'Session checkpoint Manifest value does not match its immutable reference', + ); + } + return storedSessionCheckpoint({ manifest, value }); +} + +function admitCommitSessionRevisionInput( + input: CommitSessionRevisionInput, +): InternalCommitSessionRevisionInput { + if (!isRecord(input)) throw new TypeError('Session commit input must be an object'); + return Object.freeze({ + sessionId: requireIdentifier(input.sessionId, 'Session identity'), + expectedRevision: requireIdentifier(input.expectedRevision, 'Expected revision'), + checkpoint: admitStoredSessionCheckpoint(input.checkpoint), + ...(input.lastCommittedActivationId === undefined + ? {} + : { + lastCommittedActivationId: requireIdentifier( + input.lastCommittedActivationId, + 'Activation identity', + ), + }), + ...(input.commitId === undefined + ? {} + : { commitId: requireIdentifier(input.commitId, 'Commit identity') }), + }); +} + +function admitCreateSessionInput(input: CreateSessionInput): InternalCreateSessionInput { + if (!isRecord(input)) throw new TypeError('Session creation input must be an object'); + const forkedFrom = + input.forkedFrom === undefined ? undefined : admitRevisionRef(input.forkedFrom, 'Fork source'); + const createdByForkId = + input.createdByForkId === undefined + ? undefined + : requireIdentifier(input.createdByForkId, 'Fork identity'); + if ((forkedFrom === undefined) !== (createdByForkId === undefined)) { + throw new TypeError('Fork lineage and Fork identity must be supplied together'); + } + if (createdByForkId !== undefined && input.lastCommittedActivationId !== undefined) { + throw new TypeError('Fork-created Session must not carry an Activation identity'); + } + return Object.freeze({ + sessionId: requireIdentifier(input.sessionId, 'Session identity'), + agentId: requireIdentifier(input.agentId, 'Agent identity'), + checkpoint: admitStoredSessionCheckpoint(input.checkpoint), + ...(input.lastCommittedActivationId === undefined + ? {} + : { + lastCommittedActivationId: requireIdentifier( + input.lastCommittedActivationId, + 'Activation identity', + ), + }), + ...(forkedFrom === undefined ? {} : { forkedFrom }), + ...(createdByForkId === undefined ? {} : { createdByForkId }), + }); +} + +function admitClaimForkInput(input: ClaimForkInput): InternalClaimForkInput { + if (!isRecord(input)) throw new TypeError('Fork claim input must be an object'); + return Object.freeze({ + forkId: requireIdentifier(input.forkId, 'Fork identity'), + source: admitRevisionRef(input.source, 'Fork source'), + targetSessionId: requireIdentifier(input.targetSessionId, 'Fork target Session identity'), + }); +} + +function admitRevisionRef(input: SessionRevisionRef, label: string): SessionRevisionRef { + if (!isRecord(input)) throw new TypeError(`${label} must be an object`); + return copyRevisionRef({ + sessionId: requireIdentifier(input.sessionId, `${label} Session identity`), + revision: requireIdentifier(input.revision, `${label} revision`), + }); +} + +function committedRevision(input: { + readonly sessionId: string; + readonly revision: SessionRepositoryRevision; + readonly agentId: string; + readonly checkpoint: StoredSessionCheckpoint; + readonly lastCommittedActivationId?: string; + readonly forkedFrom?: SessionRevisionRef; +}): CommittedSessionRevision { + return Object.freeze({ + ref: copyRevisionRef({ sessionId: input.sessionId, revision: input.revision }), + agentId: input.agentId, + checkpoint: copyStoredSessionCheckpoint(input.checkpoint), + ...(input.lastCommittedActivationId === undefined + ? {} + : { lastCommittedActivationId: input.lastCommittedActivationId }), + ...(input.forkedFrom === undefined ? {} : { forkedFrom: copyRevisionRef(input.forkedFrom) }), + }); +} + +function immutableObjectRef(input: ImmutableObjectRef): ImmutableObjectRef { + return Object.freeze({ + objectRef: input.objectRef, + digest: input.digest, + bytes: input.bytes, + mediaType: input.mediaType, + }); +} + +function copyImmutableObjectRef(input: ImmutableObjectRef): ImmutableObjectRef { + return immutableObjectRef(input); +} + +function storedSessionCheckpoint(input: StoredSessionCheckpoint): StoredSessionCheckpoint { + return Object.freeze({ + manifest: copyImmutableObjectRef(input.manifest), + value: Object.freeze({ + schemaVersion: 1, + compatibilityBundle: copyImmutableObjectRef(input.value.compatibilityBundle), + }), + }); +} + +function copyStoredSessionCheckpoint(input: StoredSessionCheckpoint): StoredSessionCheckpoint { + return storedSessionCheckpoint(input); +} + +function copyRevisionRef(input: SessionRevisionRef): SessionRevisionRef { + return Object.freeze({ sessionId: input.sessionId, revision: input.revision }); +} + +function copyCommittedSessionRevision(input: CommittedSessionRevision): CommittedSessionRevision { + return committedRevision({ + sessionId: input.ref.sessionId, + revision: input.ref.revision, + agentId: input.agentId, + checkpoint: input.checkpoint, + ...(input.lastCommittedActivationId === undefined + ? {} + : { lastCommittedActivationId: input.lastCommittedActivationId }), + ...(input.forkedFrom === undefined ? {} : { forkedFrom: input.forkedFrom }), + }); +} + +function copyForkOperation(input: InternalForkOperation): ForkOperation { + if (input.state === 'pending') { + return Object.freeze({ + state: 'pending', + forkId: input.forkId, + source: copyRevisionRef(input.source), + sourceAgentId: input.sourceAgentId, + sourceCheckpoint: copyStoredSessionCheckpoint(input.sourceCheckpoint), + targetSessionId: input.targetSessionId, + }); + } + return copyCompletedForkOperation(input); +} + +function copyCompletedForkOperation(input: InternalCompletedForkOperation): CompletedForkOperation { + return Object.freeze({ + state: 'completed', + forkId: input.forkId, + source: copyRevisionRef(input.source), + sourceAgentId: input.sourceAgentId, + sourceCheckpoint: copyStoredSessionCheckpoint(input.sourceCheckpoint), + targetSessionId: input.targetSessionId, + target: copyRevisionRef(input.target), + }); +} + +function sameImmutableObjectRef(left: ImmutableObjectRef, right: ImmutableObjectRef): boolean { + return ( + left.objectRef === right.objectRef && + left.digest === right.digest && + left.bytes === right.bytes && + left.mediaType === right.mediaType + ); +} + +function sameSessionCheckpointManifestV1( + left: SessionCheckpointManifestV1, + right: SessionCheckpointManifestV1, +): boolean { + return ( + left.schemaVersion === right.schemaVersion && + sameImmutableObjectRef(left.compatibilityBundle, right.compatibilityBundle) + ); +} + +function sameStoredSessionCheckpoint( + left: StoredSessionCheckpoint, + right: StoredSessionCheckpoint, +): boolean { + return ( + sameImmutableObjectRef(left.manifest, right.manifest) && + sameSessionCheckpointManifestV1(left.value, right.value) + ); +} + +function sameRevisionRef(left: SessionRevisionRef, right: SessionRevisionRef): boolean { + return left.sessionId === right.sessionId && left.revision === right.revision; +} + +function sameCommitInput( + left: InternalCommitSessionRevisionInput, + right: InternalCommitSessionRevisionInput, +): boolean { + return ( + left.sessionId === right.sessionId && + left.expectedRevision === right.expectedRevision && + sameStoredSessionCheckpoint(left.checkpoint, right.checkpoint) && + left.lastCommittedActivationId === right.lastCommittedActivationId && + left.commitId === right.commitId + ); +} + +function sameCreateInput( + created: CommittedSessionRevision, + agentId: string, + forkedFrom: SessionRevisionRef | undefined, + input: InternalCreateSessionInput, +): boolean { + return ( + created.agentId === agentId && + created.agentId === input.agentId && + sameStoredSessionCheckpoint(created.checkpoint, input.checkpoint) && + created.lastCommittedActivationId === input.lastCommittedActivationId && + sameOptionalRevisionRef(forkedFrom, input.forkedFrom) + ); +} + +function sameOptionalRevisionRef( + left: SessionRevisionRef | undefined, + right: SessionRevisionRef | undefined, +): boolean { + return left === undefined || right === undefined ? left === right : sameRevisionRef(left, right); +} + +function sameForkClaim(operation: InternalForkOperation, input: InternalClaimForkInput): boolean { + return ( + operation.targetSessionId === input.targetSessionId && + sameRevisionRef(operation.source, input.source) + ); +} + +function nextRevision(session: SessionState): SessionRepositoryRevision { + const revision = `r${session.nextRevisionNumber}`; + session.nextRevisionNumber += 1; + return revision; +} + +function requireIdentifier( + value: unknown, + label: string, + maximumLength = MAX_IDENTIFIER_LENGTH, +): string { + if (!isNonEmptyUnicodeString(value) || value.length > maximumLength) { + throw new TypeError(`${label} must be a bounded non-empty Unicode string`); + } + return value; +} + +function requireImmutableObjectStore(value: unknown): ImmutableObjectStore { + if ( + !isRecord(value) || + typeof value.publish !== 'function' || + typeof value.assertReadable !== 'function' || + typeof value.materialize !== 'function' + ) { + throw new TypeError( + 'Immutable Object Store must implement publish, assertReadable, and materialize', + ); + } + return value as unknown as ImmutableObjectStore; +} + +function isByteCount(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function digestBytes(value: Uint8Array): Sha256Digest { + return `sha256:${createHash('sha256').update(value).digest('hex')}` as Sha256Digest; +} + +function sameBytes(left: Uint8Array, right: Uint8Array): boolean { + if (left.byteLength !== right.byteLength) return false; + for (let index = 0; index < left.byteLength; index += 1) { + if (left[index] !== right[index]) return false; + } + return true; +} + +function normalizeObjectStoreError(error: unknown): SessionRepositoryError { + if (error instanceof SessionRepositoryError) return error; + return repositoryError('io_failure', 'Immutable Object Store operation failed', error); +} + +function repositoryError( + code: SessionRepositoryErrorCode, + message: string, + cause?: unknown, +): SessionRepositoryError { + return new SessionRepositoryError(code, message, cause === undefined ? {} : { cause }); +} diff --git a/scripts/ci-workflow-policy.test.mjs b/scripts/ci-workflow-policy.test.mjs index 62c03b362d..e2ded11c0c 100644 --- a/scripts/ci-workflow-policy.test.mjs +++ b/scripts/ci-workflow-policy.test.mjs @@ -28,7 +28,10 @@ * this file asserts exactly that, for every suite the install-free steps run. */ import assert from 'node:assert/strict'; -import { readdirSync, readFileSync } from 'node:fs'; +import { execFileSync, spawnSync } from 'node:child_process'; +import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import test from 'node:test'; import { formatGitHubOutputs, loadWorkspaceGraph, planTests } from './ci-test-plan.mjs'; @@ -68,7 +71,7 @@ test('one unconditional job carries the required context on every pull request', assert.doesNotMatch(jobsBlock, /^ {4}if:/mu); }); -test('planning runs first and every later step gates on its outputs', () => { +test('comparison precedes planning and every later gate uses plan outputs', () => { const workflow = readWorkflow('ci.yml'); // With the job split gone there is no `needs` context to read. GitHub @@ -77,7 +80,10 @@ test('planning runs first and every later step gates on its outputs', () => { assert.doesNotMatch(workflow, /needs\.plan\.outputs/u); const planStep = workflow.indexOf(' - id: plan\n'); + const comparisonStep = workflow.indexOf(' - id: comparison\n'); assert.ok(planStep >= 0, 'no planning step'); + assert.ok(comparisonStep >= 0 && comparisonStep < planStep, 'resolve comparison before planning'); + assert.doesNotMatch(workflow.slice(comparisonStep, planStep), /\n\s+if:|continue-on-error/u); assert.ok(planStep < workflow.indexOf('steps.plan.outputs')); assert.match( workflow, @@ -90,15 +96,24 @@ test('core CI validates pull requests and the resulting main branch state', () = assert.match(workflow, /pull_request:\n\s+branches: \[main\]/u); assert.match(workflow, /push:\n\s+branches: \[main\]/u); - assert.match( - workflow, - /BASE_SHA: \$\{\{ github\.event_name == 'push' && github\.event\.before \|\| github\.event\.pull_request\.base\.sha \}\}/u, - ); - assert.match( - workflow, - /HEAD_SHA: \$\{\{ github\.event_name == 'push' && github\.sha \|\| github\.event\.pull_request\.head\.sha \}\}/u, - ); - assert.match(workflow, /\[\[ "\$BASE_SHA" =~ \^0\+\$ \]\]/u); + assert.match(workflow, /PUSH_BASE_SHA: \$\{\{ github\.event\.before \}\}/u); + assert.match(workflow, /PR_HEAD_SHA: \$\{\{ github\.event\.pull_request\.head\.sha \}\}/u); + assert.doesNotMatch(workflow, /github\.event\.pull_request\.base\.sha/u); +}); + +test('every core diff gate consumes the shared comparison without resolving another base', () => { + const workflow = readWorkflow('ci.yml'); + const gates = workflow.split('\n - ').filter((step) => step.includes('--base ')); + assert.equal(gates.length, 4, 'planner, epoch, locale and renderer must all be covered'); + for (const gate of gates) { + assert.match(gate, /BASE_SHA: \$\{\{ steps\.comparison\.outputs\.base \}\}/u); + assert.match(gate, /--base "\$BASE_SHA"/u); + assert.doesNotMatch( + gate, + /github\.event\.(?:before|pull_request)|git (?:rev-parse|merge-base)/u, + ); + } + assert.match(workflow, /HEAD_SHA: \$\{\{ steps\.comparison\.outputs\.head \}\}/u); }); test('core CI uses the Windows inventory package-script authority', () => { @@ -108,6 +123,175 @@ test('core CI uses the Windows inventory package-script authority', () => { assert.doesNotMatch(workflow, /run: node scripts\/windows-test-inventory\.mjs --check/u); }); +test('shared comparison drives every diff gate on refreshed merges, pushes and dispatches', () => { + const workflow = readWorkflow('ci.yml'); + const scriptFor = (header) => { + const start = workflow.indexOf(` - ${header}\n`); + assert.ok(start >= 0, header); + const step = workflow.slice(start, workflow.indexOf('\n - ', start + 1)); + const script = step.split(' run: |\n')[1] ?? step.match(/ run: ([^\n]+)/u)?.[1]; + assert.ok(script, header); + return script; + }; + const comparisonScript = scriptFor('id: comparison'); + const root = mkdtempSync(join(tmpdir(), 'ci-comparison-')); + const git = (...args) => + execFileSync('git', ['-c', 'core.hooksPath=/dev/null', ...args], { + cwd: root, + encoding: 'utf8', + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); + const commit = (message) => { + git( + '-c', + 'user.name=CI fixture', + '-c', + 'user.email=ci@example.invalid', + '-c', + 'commit.gpgsign=false', + 'commit', + '--allow-empty', + '-m', + message, + ); + return git('rev-parse', 'HEAD'); + }; + let invocation = 0; + const readLines = (path) => + existsSync(path) ? readFileSync(path, 'utf8').trim().split('\n') : []; + const run = (script, env) => { + const outputPath = join(root, `output-${invocation}`); + const invocationPath = join(root, `invocations-${invocation++}`); + const result = spawnSync('bash', ['-e', '-c', script], { + cwd: root, + encoding: 'utf8', + env: { + ...process.env, + BASE_SHA: '', + CI_EVENT_NAME: '', + PUSH_BASE_SHA: '', + PR_HEAD_SHA: '', + ...env, + GITHUB_OUTPUT: outputPath, + INVOCATION_LOG: invocationPath, + }, + }); + return { + ...result, + outputs: Object.fromEntries(readLines(outputPath).map((line) => line.split('='))), + invocations: readLines(invocationPath), + }; + }; + const assertConsumers = (env, expectedBase, expectedHead, fullPlan = !expectedBase) => { + const result = run(comparisonScript, env); + assert.equal(result.status, 0, result.stderr); + assert.deepEqual(result.outputs, { base: expectedBase, head: expectedHead }); + const consumers = [ + [ + 'id: plan', + [ + 'node', + 'scripts/ci-test-plan.mjs', + ...(fullPlan ? ['--full'] : ['--base', expectedBase, '--head', expectedHead]), + ], + ], + [ + 'name: Check locale hygiene', + [ + 'node', + '--test', + 'scripts/check-locale-hygiene.test.mjs', + 'npm', + 'run', + 'check:locale-hygiene', + ...(expectedBase ? ['--', '--base', expectedBase] : []), + ], + ], + [ + 'name: Check renderer architecture', + [ + 'npm', + 'run', + 'check:renderer-architecture', + ...(expectedBase ? ['--', '--base', expectedBase] : []), + ], + ], + ]; + if (env.CI_EVENT_NAME === 'pull_request') { + consumers.push([ + 'name: Guard the protocol compatibility epoch', + ['node', 'scripts/protocol-epoch-check.mjs', '--base', expectedBase], + ]); + } + for (const [header, expected] of consumers) { + const consumer = run( + `node() { printf '%s\\n' node "$@" >> "$INVOCATION_LOG"; }\nnpm() { printf '%s\\n' npm "$@" >> "$INVOCATION_LOG"; }\n${scriptFor(header)}`, + { BASE_SHA: result.outputs.base, HEAD_SHA: result.outputs.head }, + ); + assert.equal(consumer.status, 0, `${header}: ${consumer.stderr}`); + assert.deepEqual(consumer.invocations, expected, header); + } + }; + try { + git('init', '--initial-branch=main'); + const oldBase = commit('original main'); + git('checkout', '-b', 'feature'); + const prHead = commit('storage change'); + git('checkout', 'main'); + const newBase = commit('main advanced after the PR event'); + git( + '-c', + 'user.name=CI fixture', + '-c', + 'user.email=ci@example.invalid', + '-c', + 'commit.gpgsign=false', + 'merge', + '--no-ff', + 'feature', + '-m', + 'PR merge', + ); + const mergedHead = git('rev-parse', 'HEAD'); + + // Ignore any stale base: compare the actual main parent to the checked-out merge. + assertConsumers( + { CI_EVENT_NAME: 'pull_request', BASE_SHA: oldBase, PR_HEAD_SHA: prHead }, + newBase, + mergedHead, + ); + // Pushes (including merge commits) must still compare with event.before. + assertConsumers({ CI_EVENT_NAME: 'push', PUSH_BASE_SHA: oldBase }, oldBase, mergedHead); + assertConsumers({ CI_EVENT_NAME: 'push', PUSH_BASE_SHA: '0'.repeat(40) }, '', mergedHead); + assertConsumers({ CI_EVENT_NAME: 'push' }, '', mergedHead); + // Missing push history selects all tests, but must not silently replace + // the requested base for the gates (which still reject an unavailable ref). + assertConsumers( + { CI_EVENT_NAME: 'push', PUSH_BASE_SHA: 'f'.repeat(40) }, + 'f'.repeat(40), + mergedHead, + true, + ); + assertConsumers({ CI_EVENT_NAME: 'workflow_dispatch', PUSH_BASE_SHA: oldBase }, '', mergedHead); + + for (const invalidHead of [oldBase, '']) { + const mismatch = run(comparisonScript, { + CI_EVENT_NAME: 'pull_request', + PR_HEAD_SHA: invalidHead, + }); + assert.notEqual(mismatch.status, 0); + assert.deepEqual(mismatch.outputs, {}, 'wrong-head checkouts must not publish a comparison'); + } + git('checkout', '--detach', prHead); + assertConsumers({ CI_EVENT_NAME: 'push', PUSH_BASE_SHA: oldBase }, oldBase, prHead); + const unmerged = run(comparisonScript, { CI_EVENT_NAME: 'pull_request', PR_HEAD_SHA: prHead }); + assert.notEqual(unmerged.status, 0); + assert.deepEqual(unmerged.outputs, {}, 'a linear PR checkout cannot supply the merge base'); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + test('contract checks run before dependency setup and can fail the job', () => { const workflow = readWorkflow('ci.yml'); const setupNodeStart = workflow.indexOf(' - uses: actions/setup-node@');