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..b67ba2db4a --- /dev/null +++ b/packages/storage/src/__tests__/file-session-repository.test.ts @@ -0,0 +1,436 @@ +/* + * 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 { + 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'); + }); +}); + +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__/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..9851f93ce0 --- /dev/null +++ b/packages/storage/src/file-session-repository.ts @@ -0,0 +1,1293 @@ +/* + * 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 { withFileUpdateLock } from './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 { + return await withFileUpdateLock(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) => `${entry.sessionId}\u0000${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..16309dba14 --- /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 { readFile, rm, writeFile } 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 source = await this.resolveForkSource(admitted.source); + + // 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); + } + + 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 resolveForkSource(source: SessionRevisionRef): Promise { + const initial = this.sessions.get(source.sessionId); + if (!initial || initial.head.ref.revision !== source.revision) { + throw repositoryError( + 'source_revision_not_available', + 'Fork source Session revision is not available', + ); + } + + const committed = copyCommittedSessionRevision(initial.head); + await this.assertCheckpointReadable(committed.checkpoint); + + // A source head advance while asynchronous object verification ran makes + // this request ineligible. Later retries of an admitted fork use the + // durable claim above and intentionally do not revalidate this condition. + const afterVerification = this.sessions.get(source.sessionId); + if ( + !afterVerification || + afterVerification.head.ref.revision !== source.revision || + afterVerification.agentId !== committed.agentId + ) { + throw repositoryError( + 'source_revision_not_available', + 'Fork source Session revision is no longer current', + ); + } + return committed; + } + + 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', + ); + } + try { + await writeFile(request.destination, stored.bytes, { flag: 'wx', mode: 0o600 }); + } catch (error) { + 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 }); +}