You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by mi...@apache.org on 2026/09/01 03:23:04 UTC
(maka) branch main updated: feat(storage): implement production session snapshots (#4361)
This is an automated email from the ASF dual-hosted git repository.
MicroGery pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new b9128e186e feat(storage): implement production session snapshots (#4361)
b9128e186e is described below
commit b9128e186e43333c287ec960d4a430661927f0fc
Author: MicroGrey <ch...@outlook.com>
AuthorDate: Tue Sep 1 11:23:00 2026 +0800
feat(storage): implement production session snapshots (#4361)
Generated-by: OpenAI Codex
---
packages/runtime/package.json | 1 +
.../__tests__/quiescent-session-snapshot.test.ts | 159 +++
packages/runtime/src/quiescent-session-snapshot.ts | 126 +++
packages/storage/package.json | 2 +
.../__tests__/production-session-snapshot.test.ts | 309 ++++++
.../src/__tests__/public-entrypoints.test.ts | 4 +
.../storage/src/production-session-snapshot.ts | 1050 ++++++++++++++++++++
7 files changed, 1651 insertions(+)
diff --git a/packages/runtime/package.json b/packages/runtime/package.json
index 5ed51233f8..854124140e 100644
--- a/packages/runtime/package.json
+++ b/packages/runtime/package.json
@@ -32,6 +32,7 @@
"./runtime-event-read-model": "./dist/runtime-event-read-model.js",
"./runtime-resume": "./dist/runtime-resume.js",
"./runtime-kernel": "./dist/runtime-kernel.js",
+ "./quiescent-session-snapshot": "./dist/quiescent-session-snapshot.js",
"./session-event-runtime-mapper": "./dist/session-event-runtime-mapper.js",
"./stream-graph-readiness": "./dist/stream-graph-readiness.js",
"./stream-graph-admission": "./dist/stream-graph-admission.js",
diff --git a/packages/runtime/src/__tests__/quiescent-session-snapshot.test.ts b/packages/runtime/src/__tests__/quiescent-session-snapshot.test.ts
new file mode 100644
index 0000000000..a944b4857f
--- /dev/null
+++ b/packages/runtime/src/__tests__/quiescent-session-snapshot.test.ts
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { SessionSnapshotError } from '@maka/storage/quiescent-session-snapshot';
+import { RuntimeKernel, SessionQuiescentMutationBusyError } from '../runtime-kernel.js';
+import { createRuntimeSessionSnapshotQuiescenceAuthority } from '../quiescent-session-snapshot.js';
+
+test('enters Runtime Kernel quiescence before preparing a Session snapshot', async () => {
+ const calls: string[][] = [];
+ const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+ {
+ async runSessionQuiescentMutation(sessionIds, operation) {
+ calls.push([...sessionIds]);
+ return await operation();
+ },
+ },
+ {
+ assertSnapshotEligible(makaSessionId) {
+ assert.equal(makaSessionId, 'session-1');
+ },
+ },
+ );
+ const value = await authority.runQuiescent(
+ { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+ async () => 'prepared',
+ );
+ assert.equal(value, 'prepared');
+ assert.deepEqual(calls, [['session-1']]);
+});
+
+test('maps an active Runtime execution claim to a stable snapshot_busy error', async () => {
+ const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+ {
+ async runSessionQuiescentMutation() {
+ throw new SessionQuiescentMutationBusyError(['session-1']);
+ },
+ },
+ {
+ assertSnapshotEligible() {},
+ },
+ );
+ await assert.rejects(
+ authority.runQuiescent(
+ { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+ async () => undefined,
+ ),
+ (error) => {
+ assert.ok(error instanceof SessionSnapshotError);
+ assert.equal(error.code, 'snapshot_busy');
+ assert.deepEqual(error.details, { phase: 'admission' });
+ return true;
+ },
+ );
+});
+
+test('does not invoke snapshot work when Host eligibility rejects the Session', async () => {
+ let invoked = false;
+ const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+ {
+ async runSessionQuiescentMutation(_sessionIds, operation) {
+ return await operation();
+ },
+ },
+ {
+ assertSnapshotEligible() {
+ throw new SessionSnapshotError('snapshot_busy', 'Session has a pending approval', {
+ details: { phase: 'admission' },
+ });
+ },
+ },
+ );
+ await assert.rejects(
+ authority.runQuiescent(
+ { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+ async () => {
+ invoked = true;
+ },
+ ),
+ (error) => error instanceof SessionSnapshotError && error.code === 'snapshot_busy',
+ );
+ assert.equal(invoked, false);
+});
+
+test('actual Runtime Kernel serializes an admitted mutation before snapshot work', async () => {
+ const kernel = new RuntimeKernel({} as never);
+ let releaseMutation!: () => void;
+ const mutationRelease = new Promise<void>((resolve) => {
+ releaseMutation = resolve;
+ });
+ let markMutationStarted!: () => void;
+ const mutationStarted = new Promise<void>((resolve) => {
+ markMutationStarted = resolve;
+ });
+ const mutation = kernel.runSessionAdmissionMutation(['session-1'], async () => {
+ markMutationStarted();
+ await mutationRelease;
+ });
+ await mutationStarted;
+
+ let snapshotInvoked = false;
+ const authority = createRuntimeSessionSnapshotQuiescenceAuthority(kernel, {
+ assertSnapshotEligible() {},
+ });
+ const snapshot = authority.runQuiescent(
+ { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+ async () => {
+ snapshotInvoked = true;
+ },
+ );
+ await Promise.resolve();
+ assert.equal(snapshotInvoked, false);
+ releaseMutation();
+ await Promise.all([mutation, snapshot]);
+ assert.equal(snapshotInvoked, true);
+});
+
+test('does not run a snapshot admitted after its cancellation', async () => {
+ const controller = new AbortController();
+ controller.abort();
+ let invoked = false;
+ const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+ {
+ async runSessionQuiescentMutation(_sessionIds, operation) {
+ return await operation();
+ },
+ },
+ {
+ assertSnapshotEligible() {},
+ },
+ );
+ await assert.rejects(
+ authority.runQuiescent(
+ { makaSessionId: 'session-1', cancellation: { signal: controller.signal } },
+ async () => {
+ invoked = true;
+ },
+ ),
+ (error) => error instanceof SessionSnapshotError && error.code === 'snapshot_cancelled',
+ );
+ assert.equal(invoked, false);
+});
diff --git a/packages/runtime/src/quiescent-session-snapshot.ts b/packages/runtime/src/quiescent-session-snapshot.ts
new file mode 100644
index 0000000000..437b8a12cb
--- /dev/null
+++ b/packages/runtime/src/quiescent-session-snapshot.ts
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import {
+ createFileProductionSessionSnapshotService,
+ type FileProductionSessionSnapshotService,
+ type FileProductionSessionSnapshotServiceOptions,
+} from '@maka/storage/production-session-snapshot';
+import {
+ SessionSnapshotError,
+ type SessionSnapshotCancellation,
+ type SessionSnapshotQuiescenceAuthority,
+} from '@maka/storage/quiescent-session-snapshot';
+import { SessionQuiescentMutationBusyError, type RuntimeKernelLike } from './runtime-kernel.js';
+
+export interface RuntimeSessionSnapshotQuiescenceOptions {
+ /**
+ * Host/control-plane check for mutable states that RuntimeKernel does not
+ * own directly, such as pending approvals, background processes, or
+ * externally resumable actions. It must reject any state that could mutate
+ * Session state or the workspace during the snapshot operation.
+ */
+ readonly assertSnapshotEligible: (
+ makaSessionId: string,
+ cancellation: SessionSnapshotCancellation,
+ ) => Promise<void> | void;
+}
+
+export type RuntimeProductionSessionSnapshotServiceOptions = Omit<
+ FileProductionSessionSnapshotServiceOptions,
+ 'quiescence'
+> & {
+ readonly kernel: Pick<RuntimeKernelLike, 'runSessionQuiescentMutation'>;
+ readonly quiescence: RuntimeSessionSnapshotQuiescenceOptions;
+};
+
+/**
+ * The Runtime-owned composition point for the #2369 production pipeline. A
+ * control plane supplies roots, limits, and lifetime ownership; this factory
+ * supplies the authoritative Runtime mutation boundary before any state or
+ * workspace bytes are copied.
+ */
+export function createRuntimeProductionSessionSnapshotService(
+ options: RuntimeProductionSessionSnapshotServiceOptions,
+): Promise<FileProductionSessionSnapshotService> {
+ const { kernel, quiescence, ...storageOptions } = options;
+
+ return createFileProductionSessionSnapshotService({
+ ...storageOptions,
+ quiescence: createRuntimeSessionSnapshotQuiescenceAuthority(kernel, quiescence),
+ });
+}
+
+/**
+ * Adapts RuntimeKernel's authoritative mutation lane to the storage snapshot
+ * coordinator. The callback rechecks cancellation after admission, so an
+ * aborted request which was queued behind an earlier mutation performs no copy.
+ */
+export function createRuntimeSessionSnapshotQuiescenceAuthority(
+ kernel: Pick<RuntimeKernelLike, 'runSessionQuiescentMutation'>,
+ options: RuntimeSessionSnapshotQuiescenceOptions,
+): SessionSnapshotQuiescenceAuthority {
+ if (!kernel.runSessionQuiescentMutation) {
+ throw new TypeError('Runtime Kernel does not expose Session quiescence authority');
+ }
+ return {
+ async runQuiescent(input, operation): Promise<unknown> {
+ assertActive(input.cancellation);
+ try {
+ return await kernel.runSessionQuiescentMutation!([input.makaSessionId], async () => {
+ assertActive(input.cancellation);
+ await options.assertSnapshotEligible(input.makaSessionId, input.cancellation);
+ assertActive(input.cancellation);
+ return operation();
+ });
+ } catch (error) {
+ if (error instanceof SessionSnapshotError) throw error;
+ if (error instanceof SessionQuiescentMutationBusyError) {
+ throw new SessionSnapshotError(
+ 'snapshot_busy',
+ 'Session is busy and cannot be snapshotted',
+ {
+ cause: error,
+ details: { phase: 'admission' },
+ },
+ );
+ }
+ throw new SessionSnapshotError(
+ 'io_failure',
+ 'Unable to enter the Session snapshot boundary',
+ {
+ cause: error,
+ details: { phase: 'admission' },
+ },
+ );
+ }
+ },
+ } as SessionSnapshotQuiescenceAuthority;
+}
+
+function assertActive(cancellation: SessionSnapshotCancellation): void {
+ if (!cancellation.signal.aborted) return;
+ throw new SessionSnapshotError(
+ 'snapshot_cancelled',
+ 'Session snapshot preparation was cancelled',
+ {
+ details: { phase: 'admission' },
+ },
+ );
+}
diff --git a/packages/storage/package.json b/packages/storage/package.json
index 771f3d9d8d..b36af8f4fa 100644
--- a/packages/storage/package.json
+++ b/packages/storage/package.json
@@ -41,6 +41,8 @@
"./runtime-event-persistence": "./dist/runtime-event-persistence.js",
"./runtime-policy-stores": "./dist/runtime-policy-stores.js",
"./scheduled-task-store": "./dist/scheduled-task-store.js",
+ "./quiescent-session-snapshot": "./dist/quiescent-session-snapshot.js",
+ "./production-session-snapshot": "./dist/production-session-snapshot.js",
"./session-bundle-policy": "./dist/session-bundle-policy.js",
"./session-copy-cleanup": "./dist/session-copy-cleanup.js",
"./session-todo-authority": "./dist/session-todo-authority.js",
diff --git a/packages/storage/src/__tests__/production-session-snapshot.test.ts b/packages/storage/src/__tests__/production-session-snapshot.test.ts
new file mode 100644
index 0000000000..18dcf51bca
--- /dev/null
+++ b/packages/storage/src/__tests__/production-session-snapshot.test.ts
@@ -0,0 +1,309 @@
+/*
+ * 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 { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { test } from 'node:test';
+import type { CreateSessionInput } from '@maka/core/runtime-inputs';
+import { acquireProcessLifetimeOwner } from '../process-lifetime-owner.js';
+import {
+ createFileProductionSessionSnapshotService,
+ SESSION_SNAPSHOT_STATE_IDENTITY_MEDIA_TYPE,
+} from '../production-session-snapshot.js';
+import {
+ SessionSnapshotError,
+ type SessionSnapshotQuiescenceAuthority,
+ type SessionSnapshotWorkspaceConfirmationAuthority,
+} from '../quiescent-session-snapshot.js';
+import type { PackQuiescentSessionBundleInput } from '../production-session-snapshot.js';
+import type { SessionBundleLimits } from '../session-bundle-contract.js';
+import { createSessionBundleFileService } from '../session-bundle-file-service.js';
+import { createSessionStore } from '../session-store.js';
+
+const limits: SessionBundleLimits = {
+ maxCompressedBytes: 4 * 1024 * 1024,
+ maxDecompressedTarBytes: 8 * 1024 * 1024,
+ maxPayloadBytes: 4 * 1024 * 1024,
+ maxFileBytes: 2 * 1024 * 1024,
+ maxEntryCount: 100,
+ maxManifestBytes: 256 * 1024,
+ maxStateIdentityBytes: 64 * 1024,
+ maxPathBytes: 255,
+ maxPathDepth: 16,
+};
+
+const immediateQuiescence: SessionSnapshotQuiescenceAuthority = {
+ async runQuiescent(_input, operation) {
+ return await operation();
+ },
+};
+
+test('prepares real Session state and a policy-filtered workspace, then packs and hydrates it', async () => {
+ const fixture = await createFixture();
+ try {
+ await writeFile(join(fixture.workspaceRoot, 'README.md'), 'portable workspace\n');
+ await writeFile(join(fixture.workspaceRoot, 'package-lock.json'), '{"lockfileVersion":3}\n');
+ await mkdir(join(fixture.workspaceRoot, 'node_modules', 'dep'), { recursive: true });
+ await writeFile(join(fixture.workspaceRoot, 'node_modules', 'dep', 'index.js'), 'excluded');
+ await mkdir(join(fixture.workspaceRoot, '.git'), { recursive: true });
+ await writeFile(join(fixture.workspaceRoot, '.git', 'config'), 'excluded');
+ await mkdir(join(fixture.workspaceRoot, '.cache'), { recursive: true });
+ await writeFile(join(fixture.workspaceRoot, '.cache', 'result'), 'excluded');
+ await mkdir(join(fixture.workspaceRoot, 'logs'), { recursive: true });
+ await writeFile(join(fixture.workspaceRoot, 'logs', 'runtime.log'), 'excluded');
+
+ const service = await fixture.createService();
+ const archivePath = join(fixture.root, 'bundle.tar.zst');
+ const artifact = await service.pack({ destination: archivePath });
+
+ const hydratedRoot = join(fixture.root, 'hydrated');
+ const hydrated = await createSessionBundleFileService().hydrate({
+ source: { path: archivePath, expectedArchiveDigest: artifact.archiveDigest },
+ expectedSessionId: fixture.cloudSessionId,
+ destinationRoot: hydratedRoot,
+ limits,
+ });
+ assert.equal(
+ await readFile(join(hydrated.workspaceRoot, 'README.md'), 'utf8'),
+ 'portable workspace\n',
+ );
+ assert.equal(
+ await readFile(join(hydrated.workspaceRoot, 'package-lock.json'), 'utf8'),
+ '{"lockfileVersion":3}\n',
+ );
+ await assert.rejects(readFile(join(hydrated.workspaceRoot, 'node_modules', 'dep', 'index.js')));
+ await assert.rejects(readFile(join(hydrated.workspaceRoot, '.git', 'config')));
+ await assert.rejects(readFile(join(hydrated.workspaceRoot, '.cache', 'result')));
+ await assert.rejects(readFile(join(hydrated.workspaceRoot, 'logs', 'runtime.log')));
+ assert.equal((await readFile(join(hydrated.stateRoot, 'runtime.sqlite'))).byteLength > 0, true);
+
+ const inspection = await createSessionBundleFileService().inspect({
+ source: { path: archivePath, expectedArchiveDigest: artifact.archiveDigest },
+ limits,
+ });
+ assert.equal(inspection.stateIdentity.mediaType, SESSION_SNAPSHOT_STATE_IDENTITY_MEDIA_TYPE);
+ assert.deepEqual(JSON.parse(Buffer.from(inspection.stateIdentity.bytes).toString('utf8')), {
+ schemaVersion: 1,
+ makaSessionId: fixture.sessionId,
+ });
+ assert.deepEqual(
+ (await readdir(fixture.stagingParent)).filter((name) => name.startsWith('.snapshot-')),
+ [],
+ );
+ assert.deepEqual(
+ (await readdir(fixture.stagingParent)).filter((name) => name.startsWith('snapshot-')),
+ [],
+ );
+ } finally {
+ await fixture.close();
+ }
+});
+
+test('fails closed for a known user-authored secret and removes private staging', async () => {
+ const fixture = await createFixture();
+ try {
+ await writeFile(join(fixture.workspaceRoot, '.env'), 'TOKEN=not-portable\n');
+ const service = await fixture.createService();
+ await assert.rejects(service.prepare({}), (error) => {
+ assert.ok(error instanceof SessionSnapshotError);
+ assert.equal(error.code, 'policy_rejected');
+ assert.deepEqual(error.details, {
+ phase: 'workspace',
+ policyCategory: 'known_secret_file',
+ });
+ return true;
+ });
+ assert.deepEqual(
+ (await readdir(fixture.stagingParent)).filter((name) => name.includes('snapshot')),
+ [],
+ );
+ } finally {
+ await fixture.close();
+ }
+});
+
+test('binds the configured Maka Session and Cloud envelope despite untrusted call fields', async () => {
+ const fixture = await createFixture();
+ try {
+ await writeFile(join(fixture.workspaceRoot, 'README.md'), 'bound workspace\n');
+ const service = await fixture.createService();
+ const archivePath = join(fixture.root, 'bound.tar.zst');
+ const artifact = await service.pack({
+ destination: archivePath,
+ makaSessionId: 'other-session',
+ sessionId: 'other-cloud-session',
+ } as unknown as PackQuiescentSessionBundleInput);
+ const inspection = await createSessionBundleFileService().inspect({
+ source: { path: archivePath, expectedArchiveDigest: artifact.archiveDigest },
+ limits,
+ });
+ assert.equal(inspection.manifest.envelope.sessionId, fixture.cloudSessionId);
+ assert.deepEqual(JSON.parse(Buffer.from(inspection.stateIdentity.bytes).toString('utf8')), {
+ schemaVersion: 1,
+ makaSessionId: fixture.sessionId,
+ });
+ } finally {
+ await fixture.close();
+ }
+});
+
+test('rejects a production root nested under the workspace before copying', async () => {
+ const fixture = await createFixture();
+ try {
+ await assert.rejects(
+ fixture.createService({ stagingParent: join(fixture.workspaceRoot, 'staging') }),
+ (error) => error instanceof SessionSnapshotError && error.code === 'unsafe_source',
+ );
+ } finally {
+ await fixture.close();
+ }
+});
+
+test('uses an authenticated confirmation authority for suspected secret directories', async () => {
+ const fixture = await createFixture();
+ try {
+ await mkdir(join(fixture.workspaceRoot, 'secrets'));
+ await writeFile(
+ join(fixture.workspaceRoot, 'secrets', 'reference.txt'),
+ 'not a secret value\n',
+ );
+ const confirmations: string[] = [];
+ const confirmationAuthority: SessionSnapshotWorkspaceConfirmationAuthority = {
+ async resolveConfirmation(input) {
+ confirmations.push(`${input.makaSessionId}:${input.confirmationPath}`);
+ assert.equal(input.confirmationGrantId, 'grant-1');
+ return { action: 'include' };
+ },
+ };
+ const service = await fixture.createService({ confirmationAuthority });
+ const prepared = await service.prepare({ confirmationGrantId: 'grant-1' });
+ try {
+ assert.equal(
+ await readFile(join(prepared.snapshot.workspaceRoot, 'secrets', 'reference.txt'), 'utf8'),
+ 'not a secret value\n',
+ );
+ assert.deepEqual(confirmations, [`${fixture.sessionId}:secrets`]);
+ } finally {
+ await prepared.release();
+ }
+ } finally {
+ await fixture.close();
+ }
+});
+
+test('reserves state entries before admitting workspace entries', async () => {
+ const fixture = await createFixture();
+ try {
+ await writeFile(join(fixture.workspaceRoot, 'README.md'), 'would exceed bundle entry quota\n');
+ const service = await fixture.createService({
+ limits: { ...limits, maxEntryCount: 4 },
+ });
+ await assert.rejects(service.prepare({}), (error) => {
+ assert.ok(error instanceof SessionSnapshotError);
+ assert.equal(error.code, 'quota_exceeded');
+ assert.deepEqual(error.details, { phase: 'workspace' });
+ return true;
+ });
+ } finally {
+ await fixture.close();
+ }
+});
+
+async function createFixture(): Promise<{
+ readonly root: string;
+ readonly stateRoot: string;
+ readonly configRoot: string;
+ readonly workspaceRoot: string;
+ readonly stagingParent: string;
+ readonly cleanupStateRoot: string;
+ readonly sessionId: string;
+ readonly cloudSessionId: string;
+ createService: (overrides?: {
+ readonly workspaceRoot?: string;
+ readonly stagingParent?: string;
+ readonly cleanupStateRoot?: string;
+ readonly limits?: SessionBundleLimits;
+ readonly confirmationAuthority?: SessionSnapshotWorkspaceConfirmationAuthority;
+ }) => ReturnType<typeof createFileProductionSessionSnapshotService>;
+ close(): Promise<void>;
+}> {
+ const root = await mkdtemp(join(tmpdir(), 'maka-production-snapshot-'));
+ const stateRoot = join(root, 'state');
+ const configRoot = join(root, 'config');
+ const workspaceRoot = join(root, 'workspace');
+ const stagingParent = join(root, 'staging');
+ const cleanupStateRoot = join(root, 'cleanup-state');
+ await Promise.all([
+ mkdir(configRoot),
+ mkdir(workspaceRoot),
+ mkdir(stagingParent, { mode: 0o700 }),
+ ]);
+ const sessions = createSessionStore(stateRoot);
+ const session = await sessions.create(sessionInput());
+ await sessions.appendMessage(session.id, {
+ type: 'user',
+ id: 'message-1',
+ turnId: 'turn-1',
+ ts: 1,
+ text: 'durable message',
+ });
+ await sessions.close?.();
+ const owner = await acquireProcessLifetimeOwner(cleanupStateRoot);
+ const cloudSessionId = 'cloud-session-1';
+ return {
+ root,
+ stateRoot,
+ configRoot,
+ workspaceRoot,
+ stagingParent,
+ cleanupStateRoot,
+ sessionId: session.id,
+ cloudSessionId,
+ createService: (overrides = {}) =>
+ createFileProductionSessionSnapshotService({
+ session: { makaSessionId: session.id, cloudSessionId },
+ stateRoot,
+ configRoot,
+ workspaceRoot: overrides.workspaceRoot ?? workspaceRoot,
+ stagingParent: overrides.stagingParent ?? stagingParent,
+ cleanupStateRoot: overrides.cleanupStateRoot ?? cleanupStateRoot,
+ processLifetimeOwner: owner,
+ quiescence: immediateQuiescence,
+ limits: overrides.limits ?? limits,
+ confirmationAuthority: overrides.confirmationAuthority,
+ }),
+ async close(): Promise<void> {
+ await owner.close();
+ await rm(root, { recursive: true, force: true });
+ },
+ };
+}
+
+function sessionInput(): CreateSessionInput {
+ return {
+ cwd: '/tmp/workspace',
+ llmConnectionSlug: 'fake',
+ model: 'fake-model',
+ permissionMode: 'ask',
+ name: 'Portable Session',
+ labels: [],
+ };
+}
diff --git a/packages/storage/src/__tests__/public-entrypoints.test.ts b/packages/storage/src/__tests__/public-entrypoints.test.ts
index 1e07c0c858..0144408fa2 100644
--- a/packages/storage/src/__tests__/public-entrypoints.test.ts
+++ b/packages/storage/src/__tests__/public-entrypoints.test.ts
@@ -58,8 +58,12 @@ const SQLITE_BACKED_ENTRYPOINTS = [
'./model-call-ledger',
'./operational-state-store',
'./plan-authority',
+ // Production snapshot composition exports one filtered runtime SQLite database.
+ './production-session-snapshot',
'./project-catalog',
'./project-catalog-authority',
+ // Snapshot staging cleanup persists a lease through Session-copy cleanup.
+ './quiescent-session-snapshot',
'./runtime-event-persistence',
'./scheduled-task-store',
'./session-bundle-policy',
diff --git a/packages/storage/src/production-session-snapshot.ts b/packages/storage/src/production-session-snapshot.ts
new file mode 100644
index 0000000000..dfedcf0273
--- /dev/null
+++ b/packages/storage/src/production-session-snapshot.ts
@@ -0,0 +1,1050 @@
+/*
+ * 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 { Buffer } from 'node:buffer';
+import { constants as fsConstants } from 'node:fs';
+import type { BigIntStats } from 'node:fs';
+import { lstat, mkdir, open, readdir, realpath } from 'node:fs/promises';
+import { dirname, isAbsolute, join, relative, resolve } from 'node:path';
+import {
+ assertSessionBundleLimits,
+ type OpaqueStateIdentityDescriptor,
+ type SessionBundleArtifact,
+ type SessionBundleFileService,
+ type SessionBundleLimits,
+} from './session-bundle-contract.js';
+import { createSessionBundleFileService } from './session-bundle-file-service.js';
+import { exportSessionBundleState } from './session-bundle-policy.js';
+import {
+ createFileQuiescentSessionSnapshotCoordinator,
+ createFileSessionSnapshotStagingCleanupAuthority,
+ type PreparedSessionBundleHandle,
+ type SessionSnapshotCancellation,
+ SessionSnapshotError,
+ type SessionSnapshotPrivateStagingRootAuthority,
+ type SessionSnapshotQuiescenceAuthority,
+ type SessionSnapshotStagingCleanupRecovery,
+ type SessionSnapshotWorkspaceConfirmationResolver,
+ type SessionSnapshotWorkspaceConfirmationAuthority,
+ type SessionSnapshotWorkspaceEntry,
+ type SessionSnapshotWorkspaceExclusionCategory,
+ type SessionSnapshotWorkspacePolicy,
+ type SessionSnapshotWorkspacePreparer,
+ type SessionSnapshotWorkspacePreparation,
+ type SessionSnapshotStatePreparer,
+} from './quiescent-session-snapshot.js';
+import { isSessionBundleUstarPathV1 } from './session-bundle-ustar.js';
+import type { ProcessLifetimeOwner } from './process-lifetime-owner.js';
+import { isSafeSessionId } from './session-store.js';
+
+/** State-layer-owned descriptor format; the Bundle codec only transports these bytes. */
+export const SESSION_SNAPSHOT_STATE_IDENTITY_MEDIA_TYPE =
+ 'application/vnd.maka.session-state-identity+json;version=1';
+
+const NO_FOLLOW_OPEN_FLAG = process.platform === 'win32' ? 0 : fsConstants.O_NOFOLLOW;
+const WORKSPACE_COPY_CHUNK_BYTES = 64 * 1024;
+
+export interface FileSessionSnapshotStatePreparerOptions {
+ /** Session-owned state root, never the host configuration root. */
+ readonly stateRoot: string;
+ /** Host configuration root; it is checked by the exporter but never copied. */
+ readonly configRoot: string;
+}
+
+/**
+ * Creates the production state adapter for #2369. The existing state exporter
+ * owns semantic filtering and SQLite backup; this adapter only binds it to the
+ * quiescent-snapshot contract and emits the opaque identity descriptor.
+ */
+export function createFileSessionSnapshotStatePreparer(
+ options: FileSessionSnapshotStatePreparerOptions,
+): SessionSnapshotStatePreparer {
+ const stateRoot = requireAbsolutePath(options.stateRoot, 'stateRoot');
+ const configRoot = requireAbsolutePath(options.configRoot, 'configRoot');
+ return {
+ async prepareState(input): Promise<OpaqueStateIdentityDescriptor> {
+ assertSnapshotActive(input.cancellation, 'state');
+ const destinationRoot = requireAbsolutePath(input.destinationRoot, 'destinationRoot');
+ try {
+ await exportSessionBundleState({
+ stateRoot,
+ configRoot,
+ destinationRoot,
+ sessionId: input.makaSessionId,
+ });
+ } catch (error) {
+ throw asSnapshotError(
+ 'io_failure',
+ 'Unable to export Session snapshot state',
+ 'state',
+ error,
+ );
+ }
+ assertSnapshotActive(input.cancellation, 'state');
+ return createSessionSnapshotStateIdentity(input.makaSessionId);
+ },
+ };
+}
+
+export function createSessionSnapshotStateIdentity(
+ makaSessionId: string,
+): OpaqueStateIdentityDescriptor {
+ if (typeof makaSessionId !== 'string' || makaSessionId.length === 0) {
+ throw new TypeError('Session snapshot Maka Session identity is required');
+ }
+ return Object.freeze({
+ mediaType: SESSION_SNAPSHOT_STATE_IDENTITY_MEDIA_TYPE,
+ bytes: Uint8Array.from(
+ Buffer.from(`${JSON.stringify({ schemaVersion: 1, makaSessionId })}\n`, 'utf8'),
+ ),
+ });
+}
+
+export interface FileSessionSnapshotWorkspacePreparerOptions {
+ /** The live workspace root for one Session. */
+ readonly workspaceRoot: string;
+ /** Required, bounded Bundle policy; snapshot copying never has unlimited quotas. */
+ readonly limits: SessionBundleLimits;
+ /** Internal production composition hook that reserves already-staged state budget. */
+ readonly remainingLimitsForDestinationRoot?: (destinationRoot: string) => SessionBundleLimits;
+}
+
+/**
+ * Copies a workspace under the coordinator's pinned policy. It deliberately
+ * does not accept a caller-supplied policy: the coordinator supplies that
+ * policy at invocation time and this preparer applies every decision.
+ */
+export function createFileSessionSnapshotWorkspacePreparer(
+ options: FileSessionSnapshotWorkspacePreparerOptions,
+): SessionSnapshotWorkspacePreparer {
+ const workspaceRoot = requireAbsolutePath(options.workspaceRoot, 'workspaceRoot');
+ assertSessionBundleLimits(options.limits);
+ const limits = Object.freeze({ ...options.limits });
+ return {
+ async prepareWorkspace(input): Promise<SessionSnapshotWorkspacePreparation> {
+ assertSnapshotActive(input.cancellation, 'workspace');
+ const destinationRoot = requireAbsolutePath(input.destinationRoot, 'destinationRoot');
+ const effectiveLimits =
+ options.remainingLimitsForDestinationRoot?.(destinationRoot) ?? limits;
+ assertSessionBundleLimits(effectiveLimits);
+ const sourceRoot = await canonicalWorkspaceRoot(workspaceRoot);
+ await assertMissing(destinationRoot, 'Workspace snapshot destination already exists');
+ await mkdir(destinationRoot, { mode: 0o700 });
+
+ const budget: WorkspaceCopyBudget = {
+ includedEntries: 0,
+ excludedEntries: 0,
+ payloadBytes: 0,
+ exclusions: emptyExclusionCounts(),
+ seenCaseFoldedPaths: new Map(),
+ };
+ try {
+ await copyWorkspaceDirectory({
+ sourceDirectory: sourceRoot.path,
+ destinationDirectory: destinationRoot,
+ relativeDirectory: '',
+ expectedDirectoryIdentity: sourceRoot.identity,
+ policy: input.policy,
+ confirmation: input.confirmation,
+ cancellation: input.cancellation,
+ limits: effectiveLimits,
+ budget,
+ });
+ } catch (error) {
+ if (error instanceof SessionSnapshotError) throw error;
+ throw asSnapshotError(
+ 'io_failure',
+ 'Unable to prepare Session snapshot workspace',
+ 'workspace',
+ error,
+ );
+ }
+ assertSnapshotActive(input.cancellation, 'workspace');
+ return Object.freeze({
+ includedEntries: budget.includedEntries,
+ excludedEntries: budget.excludedEntries,
+ excludedEntriesByCategory: Object.freeze({ ...budget.exclusions }),
+ payloadBytes: budget.payloadBytes,
+ });
+ },
+ };
+}
+
+export interface FileProductionSessionSnapshotServiceOptions {
+ /**
+ * Host-authorized binding for this one production snapshot service. Callers
+ * cannot select a Maka Session, workspace, and Cloud envelope independently.
+ */
+ readonly session: ProductionSessionSnapshotBinding;
+ readonly stateRoot: string;
+ readonly configRoot: string;
+ readonly workspaceRoot: string;
+ readonly stagingParent: string;
+ /** Separate durable root for the staging-cleanup lease database. */
+ readonly cleanupStateRoot: string;
+ readonly processLifetimeOwner: ProcessLifetimeOwner;
+ readonly quiescence: SessionSnapshotQuiescenceAuthority;
+ readonly limits: SessionBundleLimits;
+ readonly bundleFileService?: SessionBundleFileService;
+ readonly privateStagingRootAuthority?: SessionSnapshotPrivateStagingRootAuthority;
+ readonly confirmationAuthority?: SessionSnapshotWorkspaceConfirmationAuthority;
+}
+
+export interface ProductionSessionSnapshotBinding {
+ /** Runtime/storage identity whose state and workspace this service snapshots. */
+ readonly makaSessionId: string;
+ /** Control-plane identity embedded in and verified against the Bundle envelope. */
+ readonly cloudSessionId: string;
+}
+
+export interface PackQuiescentSessionBundleInput {
+ readonly destination: string;
+ readonly lastCommittedActivationId?: string;
+ readonly confirmationGrantId?: string;
+ readonly signal?: AbortSignal;
+ readonly deadlineAt?: number;
+}
+
+export interface FileProductionSessionSnapshotService {
+ recover(): Promise<SessionSnapshotStagingCleanupRecovery>;
+ prepare(input: {
+ readonly confirmationGrantId?: string;
+ readonly signal?: AbortSignal;
+ readonly deadlineAt?: number;
+ }): Promise<PreparedSessionBundleHandle>;
+ pack(input: PackQuiescentSessionBundleInput): Promise<SessionBundleArtifact>;
+}
+
+/**
+ * Production composition and first codec call site for #2369. Startup awaits
+ * persisted staging recovery before this function resolves, so callers cannot
+ * serve snapshot requests while an earlier process's private staging is still
+ * unreconciled.
+ */
+export async function createFileProductionSessionSnapshotService(
+ options: FileProductionSessionSnapshotServiceOptions,
+): Promise<FileProductionSessionSnapshotService> {
+ const session = requireProductionSessionSnapshotBinding(options.session);
+ assertSessionBundleLimits(options.limits);
+ const limits = Object.freeze({ ...options.limits });
+ const stagingCleanup = createFileSessionSnapshotStagingCleanupAuthority({
+ cleanupStateRoot: requireAbsolutePath(options.cleanupStateRoot, 'cleanupStateRoot'),
+ stagingParent: requireAbsolutePath(options.stagingParent, 'stagingParent'),
+ processLifetimeOwner: options.processLifetimeOwner,
+ privateStagingRootAuthority: options.privateStagingRootAuthority,
+ });
+ await stagingCleanup.recover();
+ await assertProductionRootsSeparate({
+ stateRoot: options.stateRoot,
+ configRoot: options.configRoot,
+ workspaceRoot: options.workspaceRoot,
+ stagingParent: stagingCleanup.stagingParent,
+ cleanupStateRoot: options.cleanupStateRoot,
+ });
+ const stateBudgetsByStagingRoot = new Map<string, SnapshotStagingBudget>();
+ const fileStatePreparer = createFileSessionSnapshotStatePreparer({
+ stateRoot: options.stateRoot,
+ configRoot: options.configRoot,
+ });
+ const state: SessionSnapshotStatePreparer = {
+ async prepareState(input): Promise<OpaqueStateIdentityDescriptor> {
+ const identity = await fileStatePreparer.prepareState(input);
+ const payload = await measurePreparedStatePayload(
+ input.destinationRoot,
+ limits,
+ input.cancellation,
+ );
+ const budget = reserveStateBundleBudget(payload, identity, limits);
+ stateBudgetsByStagingRoot.set(dirname(input.destinationRoot), budget);
+ return identity;
+ },
+ };
+ const workspace = createFileSessionSnapshotWorkspacePreparer({
+ workspaceRoot: options.workspaceRoot,
+ limits,
+ remainingLimitsForDestinationRoot(destinationRoot) {
+ const stagingRoot = dirname(destinationRoot);
+ const stateBudget = stateBudgetsByStagingRoot.get(stagingRoot);
+ if (!stateBudget) {
+ throw new SessionSnapshotError(
+ 'io_failure',
+ 'Session snapshot state budget is unavailable for workspace copying',
+ { details: { phase: 'workspace' } },
+ );
+ }
+ return deriveWorkspaceLimits(limits, stateBudget);
+ },
+ });
+ const coordinator = createFileQuiescentSessionSnapshotCoordinator({
+ stagingParent: stagingCleanup.stagingParent,
+ stagingCleanup,
+ quiescence: options.quiescence,
+ state,
+ workspace,
+ confirmationAuthority: options.confirmationAuthority,
+ privateStagingRootAuthority: options.privateStagingRootAuthority,
+ });
+ const bundleFileService = options.bundleFileService ?? createSessionBundleFileService();
+ const prepare = async (input: {
+ readonly confirmationGrantId?: string;
+ readonly signal?: AbortSignal;
+ readonly deadlineAt?: number;
+ }): Promise<PreparedSessionBundleHandle> => {
+ try {
+ return await coordinator.prepare({ ...input, makaSessionId: session.makaSessionId });
+ } finally {
+ stateBudgetsByStagingRoot.clear();
+ }
+ };
+ return Object.freeze({
+ recover: () => stagingCleanup.recover(),
+ prepare,
+ async pack(input: PackQuiescentSessionBundleInput): Promise<SessionBundleArtifact> {
+ const prepared = await prepare(input);
+ let artifact: SessionBundleArtifact | undefined;
+ let primaryFailure: unknown;
+ try {
+ artifact = await bundleFileService.pack({
+ snapshot: prepared.snapshot,
+ envelope: {
+ sessionId: session.cloudSessionId,
+ ...(input.lastCommittedActivationId === undefined
+ ? {}
+ : { lastCommittedActivationId: input.lastCommittedActivationId }),
+ },
+ destination: input.destination,
+ limits,
+ });
+ } catch (error) {
+ primaryFailure = error;
+ }
+ try {
+ await prepared.release();
+ } catch (cleanupFailure) {
+ if (primaryFailure !== undefined) {
+ throw new AggregateError(
+ [primaryFailure, cleanupFailure],
+ 'Session Bundle packing failed and snapshot cleanup also failed',
+ );
+ }
+ throw cleanupFailure;
+ }
+ if (primaryFailure !== undefined) throw primaryFailure;
+ if (!artifact) throw new Error('Session Bundle packing completed without an artifact');
+ return artifact;
+ },
+ });
+}
+
+type WorkspaceCopyBudget = {
+ includedEntries: number;
+ excludedEntries: number;
+ payloadBytes: number;
+ exclusions: Record<SessionSnapshotWorkspaceExclusionCategory, number>;
+ seenCaseFoldedPaths: Map<string, string>;
+};
+
+type SnapshotStagingBudget = {
+ /** Includes state-identity.json and the state/ + workspace/ root entries. */
+ readonly entryCount: number;
+ /** Includes state files and state-identity.json, but never directories. */
+ readonly payloadBytes: number;
+};
+
+async function measurePreparedStatePayload(
+ root: string,
+ limits: SessionBundleLimits,
+ cancellation: SessionSnapshotCancellation,
+): Promise<{ readonly entryCount: number; readonly payloadBytes: number }> {
+ const budget = { entryCount: 0, payloadBytes: 0 };
+ await measurePreparedStateDirectory(root, budget, limits, cancellation);
+ return budget;
+}
+
+async function measurePreparedStateDirectory(
+ directory: string,
+ budget: { entryCount: number; payloadBytes: number },
+ limits: SessionBundleLimits,
+ cancellation: SessionSnapshotCancellation,
+): Promise<void> {
+ assertSnapshotActive(cancellation, 'state');
+ const metadata = await lstatBigInt(directory);
+ if (!metadata.isDirectory() || metadata.isSymbolicLink()) {
+ throw new SessionSnapshotError('unsafe_source', 'Prepared Session state is unsafe', {
+ details: { phase: 'state' },
+ });
+ }
+ const children = await readdir(directory, { encoding: 'buffer', withFileTypes: true });
+ for (const child of children) {
+ assertSnapshotActive(cancellation, 'state');
+ const name = decodeFilesystemName(child.name);
+ const path = join(directory, name);
+ const childMetadata = await lstatBigInt(path);
+ if (childMetadata.isSymbolicLink()) {
+ throw new SessionSnapshotError('unsafe_source', 'Prepared Session state is unsafe', {
+ details: { phase: 'state' },
+ });
+ }
+ if (childMetadata.isDirectory()) {
+ budget.entryCount = reserveStateEntry(budget.entryCount, limits);
+ await measurePreparedStateDirectory(path, budget, limits, cancellation);
+ continue;
+ }
+ if (!childMetadata.isFile()) {
+ throw new SessionSnapshotError('unsafe_source', 'Prepared Session state is unsafe', {
+ details: { phase: 'state' },
+ });
+ }
+ if (childMetadata.nlink !== 1n) {
+ throw new SessionSnapshotError('unsafe_source', 'Prepared Session state is unsafe', {
+ details: { phase: 'state' },
+ });
+ }
+ const bytes = safeSize(childMetadata.size);
+ if (bytes > limits.maxFileBytes) throw stateQuotaExceeded();
+ budget.payloadBytes = reserveStatePayload(budget.payloadBytes, bytes, limits);
+ budget.entryCount = reserveStateEntry(budget.entryCount, limits);
+ }
+}
+
+function reserveStateBundleBudget(
+ state: { readonly entryCount: number; readonly payloadBytes: number },
+ identity: OpaqueStateIdentityDescriptor,
+ limits: SessionBundleLimits,
+): SnapshotStagingBudget {
+ const identityBytes = identity.bytes.byteLength;
+ if (identityBytes > limits.maxStateIdentityBytes || identityBytes > limits.maxFileBytes) {
+ throw stateQuotaExceeded();
+ }
+ const payloadBytes = reserveStatePayload(state.payloadBytes, identityBytes, limits);
+ // state-identity.json, state/, and workspace/ are always emitted by the codec.
+ const entryCount = reserveStateEntry(
+ reserveStateEntry(reserveStateEntry(state.entryCount, limits), limits),
+ limits,
+ );
+ return Object.freeze({ entryCount, payloadBytes });
+}
+
+function deriveWorkspaceLimits(
+ limits: SessionBundleLimits,
+ reserved: SnapshotStagingBudget,
+): SessionBundleLimits {
+ if (
+ reserved.payloadBytes > limits.maxPayloadBytes ||
+ reserved.entryCount > limits.maxEntryCount
+ ) {
+ throw stateQuotaExceeded();
+ }
+ return Object.freeze({
+ ...limits,
+ maxPayloadBytes: limits.maxPayloadBytes - reserved.payloadBytes,
+ maxEntryCount: limits.maxEntryCount - reserved.entryCount,
+ });
+}
+
+function reserveStatePayload(current: number, added: number, limits: SessionBundleLimits): number {
+ const next = safeAdd(current, added);
+ if (next > limits.maxPayloadBytes) throw stateQuotaExceeded();
+ return next;
+}
+
+function reserveStateEntry(current: number, limits: SessionBundleLimits): number {
+ const next = safeAdd(current, 1);
+ if (next > limits.maxEntryCount) throw stateQuotaExceeded();
+ return next;
+}
+
+function stateQuotaExceeded(): SessionSnapshotError {
+ return new SessionSnapshotError('quota_exceeded', 'Session snapshot state exceeds Bundle quota', {
+ details: { phase: 'state' },
+ });
+}
+
+async function copyWorkspaceDirectory(input: {
+ readonly sourceDirectory: string;
+ readonly destinationDirectory: string;
+ readonly relativeDirectory: string;
+ readonly expectedDirectoryIdentity?: FilesystemIdentity;
+ readonly policy: SessionSnapshotWorkspacePolicy;
+ readonly confirmation: SessionSnapshotWorkspaceConfirmationResolver;
+ readonly cancellation: SessionSnapshotCancellation;
+ readonly limits: SessionBundleLimits;
+ readonly budget: WorkspaceCopyBudget;
+}): Promise<void> {
+ assertSnapshotActive(input.cancellation, 'workspace');
+ const directoryBefore = await readDirectoryFingerprint(input.sourceDirectory);
+ if (
+ input.expectedDirectoryIdentity !== undefined &&
+ !sameFilesystemIdentity(directoryBefore, input.expectedDirectoryIdentity)
+ ) {
+ throw new SessionSnapshotError('source_changed', 'Workspace root changed before copying', {
+ details: { phase: 'workspace' },
+ });
+ }
+ const children = await readdir(input.sourceDirectory, {
+ encoding: 'buffer',
+ withFileTypes: true,
+ });
+ const normalized = children
+ .map((child) => decodeFilesystemName(child.name))
+ .sort((left, right) => Buffer.compare(Buffer.from(left), Buffer.from(right)));
+ for (const name of normalized) {
+ assertSnapshotActive(input.cancellation, 'workspace');
+ const relativePath =
+ input.relativeDirectory.length === 0 ? name : `${input.relativeDirectory}/${name}`;
+ const sourcePath = join(input.sourceDirectory, name);
+ const destinationPath = join(input.destinationDirectory, name);
+ const metadata = await lstatBigInt(sourcePath);
+ const entry = classifyFilesystemEntry(relativePath, metadata);
+ registerWorkspacePath(input.budget, relativePath);
+ assertWorkspacePathBudget(relativePath, entry.kind, input.limits);
+
+ const policyDecision = input.policy.classify(entry);
+ if (policyDecision.kind === 'reject') {
+ throw new SessionSnapshotError(
+ 'policy_rejected',
+ 'Workspace snapshot policy rejected an entry',
+ { details: { phase: 'workspace', policyCategory: policyDecision.category } },
+ );
+ }
+ if (policyDecision.kind === 'exclude') {
+ recordExclusion(input.budget, policyDecision.category);
+ continue;
+ }
+ if (policyDecision.kind === 'confirm') {
+ const resolution = await input.confirmation.resolve(entry);
+ assertSnapshotActive(input.cancellation, 'workspace');
+ if (resolution.kind === 'exclude') {
+ recordExclusion(input.budget, resolution.category);
+ continue;
+ }
+ }
+
+ if (entry.kind === 'directory') {
+ await mkdir(destinationPath, { mode: 0o700 });
+ recordIncludedEntry(input.budget, input.limits);
+ await copyWorkspaceDirectory({
+ ...input,
+ sourceDirectory: sourcePath,
+ destinationDirectory: destinationPath,
+ relativeDirectory: relativePath,
+ expectedDirectoryIdentity: undefined,
+ });
+ continue;
+ }
+
+ const before = fileFingerprint(metadata);
+ const bytes = safeSize(metadata.size);
+ assertWorkspaceFileQuota(bytes, input.budget, input.limits);
+ recordIncludedEntry(input.budget, input.limits);
+ await copyWorkspaceFile({
+ sourcePath,
+ destinationPath,
+ expected: before,
+ expectedSize: bytes,
+ sourceMetadata: metadata,
+ cancellation: input.cancellation,
+ });
+ const after = await lstatBigInt(sourcePath);
+ const destination = await lstatBigInt(destinationPath);
+ if (
+ !after.isFile() ||
+ after.nlink !== 1n ||
+ !sameFileFingerprint(before, fileFingerprint(after)) ||
+ !destination.isFile() ||
+ destination.isSymbolicLink() ||
+ destination.nlink !== 1n ||
+ destination.size !== metadata.size
+ ) {
+ throw new SessionSnapshotError('source_changed', 'Workspace changed while being copied', {
+ details: { phase: 'workspace' },
+ });
+ }
+ input.budget.payloadBytes += bytes;
+ }
+ const directoryAfter = await readDirectoryFingerprint(input.sourceDirectory);
+ if (!sameDirectoryFingerprint(directoryBefore, directoryAfter)) {
+ throw new SessionSnapshotError('source_changed', 'Workspace changed while being copied', {
+ details: { phase: 'workspace' },
+ });
+ }
+}
+
+function classifyFilesystemEntry(
+ relativePath: string,
+ metadata: BigIntStats,
+): SessionSnapshotWorkspaceEntry {
+ if (metadata.isSymbolicLink()) {
+ throw new SessionSnapshotError('unsafe_source', 'Workspace contains a symbolic link', {
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+ }
+ if (metadata.isDirectory()) return { relativePath, kind: 'directory' };
+ if (metadata.isFile()) {
+ if (metadata.nlink !== 1n) {
+ throw new SessionSnapshotError('unsafe_source', 'Workspace contains a hard-linked file', {
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+ }
+ return { relativePath, kind: 'file' };
+ }
+ throw new SessionSnapshotError('unsafe_source', 'Workspace contains an unsupported entry', {
+ details: { phase: 'workspace', policyCategory: 'unsupported_entry' },
+ });
+}
+
+function assertWorkspacePathBudget(
+ relativePath: string,
+ kind: SessionSnapshotWorkspaceEntry['kind'],
+ limits: SessionBundleLimits,
+): void {
+ const archivePath = `workspace/${relativePath}${kind === 'directory' ? '/' : ''}`;
+ if (!isSessionBundleUstarPathV1(archivePath)) {
+ throw new SessionSnapshotError('unsafe_source', 'Workspace contains an unsafe path', {
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+ }
+ if (
+ Buffer.byteLength(archivePath, 'utf8') > limits.maxPathBytes ||
+ archivePath.replace(/\/$/u, '').split('/').length > limits.maxPathDepth
+ ) {
+ throw new SessionSnapshotError('quota_exceeded', 'Workspace path exceeds Bundle quota', {
+ details: { phase: 'workspace' },
+ });
+ }
+}
+
+function assertWorkspaceFileQuota(
+ bytes: number,
+ budget: WorkspaceCopyBudget,
+ limits: SessionBundleLimits,
+): void {
+ if (bytes > limits.maxFileBytes || safeAdd(budget.payloadBytes, bytes) > limits.maxPayloadBytes) {
+ throw new SessionSnapshotError('quota_exceeded', 'Workspace payload exceeds Bundle quota', {
+ details: { phase: 'workspace' },
+ });
+ }
+}
+
+function recordIncludedEntry(budget: WorkspaceCopyBudget, limits: SessionBundleLimits): void {
+ if (safeAdd(budget.includedEntries, 1) > limits.maxEntryCount) {
+ throw new SessionSnapshotError('quota_exceeded', 'Workspace entry count exceeds Bundle quota', {
+ details: { phase: 'workspace' },
+ });
+ }
+ budget.includedEntries += 1;
+}
+
+function recordExclusion(
+ budget: WorkspaceCopyBudget,
+ category: SessionSnapshotWorkspaceExclusionCategory,
+): void {
+ budget.excludedEntries += 1;
+ budget.exclusions[category] += 1;
+}
+
+function emptyExclusionCounts(): Record<SessionSnapshotWorkspaceExclusionCategory, number> {
+ return {
+ dependency_tree: 0,
+ source_control: 0,
+ cache: 0,
+ log: 0,
+ runtime_scratch: 0,
+ confirmed_secret_path: 0,
+ };
+}
+
+function registerWorkspacePath(budget: WorkspaceCopyBudget, path: string): void {
+ const caseFolded = path.toLocaleLowerCase('en-US');
+ const previous = budget.seenCaseFoldedPaths.get(caseFolded);
+ if (previous !== undefined && previous !== path) {
+ throw new SessionSnapshotError('unsafe_source', 'Workspace contains case-conflicting paths', {
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+ }
+ budget.seenCaseFoldedPaths.set(caseFolded, path);
+}
+
+async function canonicalWorkspaceRoot(workspaceRoot: string): Promise<{
+ readonly path: string;
+ readonly identity: FilesystemIdentity;
+}> {
+ let root;
+ try {
+ root = await lstatBigInt(workspaceRoot);
+ } catch (error) {
+ throw asSnapshotError('unsafe_source', 'Workspace root is unavailable', 'workspace', error);
+ }
+ if (!root.isDirectory() || root.isSymbolicLink()) {
+ throw new SessionSnapshotError('unsafe_source', 'Workspace root is unsafe', {
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+ }
+ try {
+ const canonical = await realpath(workspaceRoot);
+ const configuredAfter = await lstatBigInt(workspaceRoot);
+ const canonicalMetadata = await lstatBigInt(canonical);
+ if (
+ !configuredAfter.isDirectory() ||
+ configuredAfter.isSymbolicLink() ||
+ !canonicalMetadata.isDirectory() ||
+ canonicalMetadata.isSymbolicLink() ||
+ !sameFilesystemIdentity(root, configuredAfter) ||
+ !sameFilesystemIdentity(root, canonicalMetadata)
+ ) {
+ throw new SessionSnapshotError('source_changed', 'Workspace root changed while resolving', {
+ details: { phase: 'workspace' },
+ });
+ }
+ return {
+ path: canonical,
+ identity: filesystemIdentity(root),
+ };
+ } catch (error) {
+ if (error instanceof SessionSnapshotError) throw error;
+ throw asSnapshotError('unsafe_source', 'Workspace root is unavailable', 'workspace', error);
+ }
+}
+
+async function assertMissing(path: string, message: string): Promise<void> {
+ try {
+ await lstat(path);
+ } catch (error) {
+ if (isNodeError(error, 'ENOENT')) return;
+ throw error;
+ }
+ throw new SessionSnapshotError('unsafe_source', message, {
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+}
+
+type FileFingerprint = {
+ readonly dev: bigint;
+ readonly ino: bigint;
+ readonly size: bigint;
+ readonly mtimeNs: bigint;
+ readonly ctimeNs: bigint;
+};
+
+type DirectoryFingerprint = Pick<FileFingerprint, 'dev' | 'ino' | 'mtimeNs' | 'ctimeNs'>;
+type FilesystemIdentity = Pick<FileFingerprint, 'dev' | 'ino'>;
+
+async function copyWorkspaceFile(input: {
+ readonly sourcePath: string;
+ readonly destinationPath: string;
+ readonly expected: FileFingerprint;
+ readonly expectedSize: number;
+ readonly sourceMetadata: BigIntStats;
+ readonly cancellation: SessionSnapshotCancellation;
+}): Promise<void> {
+ let source: Awaited<ReturnType<typeof open>> | undefined;
+ let destination: Awaited<ReturnType<typeof open>> | undefined;
+ try {
+ source = await open(input.sourcePath, fsConstants.O_RDONLY | NO_FOLLOW_OPEN_FLAG);
+ const openedSource = (await source.stat({ bigint: true })) as BigIntStats;
+ if (
+ !openedSource.isFile() ||
+ openedSource.nlink !== 1n ||
+ !sameFileFingerprint(input.expected, fileFingerprint(openedSource))
+ ) {
+ throw sourceChangedDuringWorkspaceCopy();
+ }
+ destination = await open(
+ input.destinationPath,
+ fsConstants.O_WRONLY | fsConstants.O_CREAT | fsConstants.O_EXCL | NO_FOLLOW_OPEN_FLAG,
+ destinationFileMode(input.sourceMetadata),
+ );
+ const buffer = Buffer.allocUnsafe(Math.min(WORKSPACE_COPY_CHUNK_BYTES, input.expectedSize));
+ let copied = 0;
+ while (copied < input.expectedSize) {
+ assertSnapshotActive(input.cancellation, 'workspace');
+ const { bytesRead } = await source.read(
+ buffer,
+ 0,
+ Math.min(buffer.byteLength, input.expectedSize - copied),
+ copied,
+ );
+ if (bytesRead === 0) throw sourceChangedDuringWorkspaceCopy();
+ let written = 0;
+ while (written < bytesRead) {
+ assertSnapshotActive(input.cancellation, 'workspace');
+ const result = await destination.write(
+ buffer,
+ written,
+ bytesRead - written,
+ copied + written,
+ );
+ if (result.bytesWritten === 0) {
+ throw new SessionSnapshotError('io_failure', 'Workspace snapshot copy made no progress', {
+ details: { phase: 'workspace' },
+ });
+ }
+ written += result.bytesWritten;
+ }
+ copied += bytesRead;
+ }
+ assertSnapshotActive(input.cancellation, 'workspace');
+ const sourceAfter = (await source.stat({ bigint: true })) as BigIntStats;
+ const destinationAfter = (await destination.stat({ bigint: true })) as BigIntStats;
+ if (
+ !sourceAfter.isFile() ||
+ sourceAfter.nlink !== 1n ||
+ !sameFileFingerprint(input.expected, fileFingerprint(sourceAfter)) ||
+ !destinationAfter.isFile() ||
+ destinationAfter.isSymbolicLink() ||
+ destinationAfter.nlink !== 1n ||
+ destinationAfter.size !== BigInt(input.expectedSize)
+ ) {
+ throw sourceChangedDuringWorkspaceCopy();
+ }
+ } finally {
+ try {
+ await destination?.close();
+ } finally {
+ await source?.close();
+ }
+ }
+}
+
+function destinationFileMode(metadata: BigIntStats): number {
+ return (metadata.mode & 0o111n) === 0n ? 0o600 : 0o700;
+}
+
+function sourceChangedDuringWorkspaceCopy(): SessionSnapshotError {
+ return new SessionSnapshotError('source_changed', 'Workspace changed while being copied', {
+ details: { phase: 'workspace' },
+ });
+}
+
+function fileFingerprint(metadata: BigIntStats): FileFingerprint {
+ return {
+ dev: metadata.dev,
+ ino: metadata.ino,
+ size: metadata.size,
+ mtimeNs: metadata.mtimeNs,
+ ctimeNs: metadata.ctimeNs,
+ };
+}
+
+function filesystemIdentity(metadata: Pick<BigIntStats, 'dev' | 'ino'>): FilesystemIdentity {
+ return { dev: metadata.dev, ino: metadata.ino };
+}
+
+function sameFilesystemIdentity(
+ left: Pick<BigIntStats, 'dev' | 'ino'>,
+ right: Pick<BigIntStats, 'dev' | 'ino'>,
+): boolean {
+ return left.dev === right.dev && left.ino === right.ino;
+}
+
+async function readDirectoryFingerprint(path: string): Promise<DirectoryFingerprint> {
+ const metadata = await lstatBigInt(path);
+ if (!metadata.isDirectory() || metadata.isSymbolicLink()) {
+ throw new SessionSnapshotError(
+ 'source_changed',
+ 'Workspace directory changed while being copied',
+ {
+ details: { phase: 'workspace' },
+ },
+ );
+ }
+ return fileFingerprint(metadata);
+}
+
+function sameFileFingerprint(left: FileFingerprint, right: FileFingerprint): boolean {
+ return (
+ left.dev === right.dev &&
+ left.ino === right.ino &&
+ left.size === right.size &&
+ left.mtimeNs === right.mtimeNs &&
+ left.ctimeNs === right.ctimeNs
+ );
+}
+
+function sameDirectoryFingerprint(
+ left: DirectoryFingerprint,
+ right: DirectoryFingerprint,
+): boolean {
+ return (
+ left.dev === right.dev &&
+ left.ino === right.ino &&
+ left.mtimeNs === right.mtimeNs &&
+ left.ctimeNs === right.ctimeNs
+ );
+}
+
+async function lstatBigInt(path: string): Promise<BigIntStats> {
+ return (await lstat(path, { bigint: true })) as BigIntStats;
+}
+
+function decodeFilesystemName(value: string | Buffer): string {
+ if (typeof value === 'string') {
+ if (value.length === 0 || value.includes('\0') || value.includes('/') || value.includes('\\')) {
+ throw new SessionSnapshotError('unsafe_source', 'Workspace entry name is unsafe', {
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+ }
+ return value;
+ }
+ try {
+ const decoded = new TextDecoder('utf-8', { fatal: true }).decode(value);
+ return decodeFilesystemName(decoded);
+ } catch (error) {
+ if (error instanceof SessionSnapshotError) throw error;
+ throw new SessionSnapshotError('unsafe_source', 'Workspace entry name is not valid UTF-8', {
+ cause: error,
+ details: { phase: 'workspace', policyCategory: 'unsafe_path' },
+ });
+ }
+}
+
+function safeSize(value: bigint): number {
+ if (value < 0n || value > BigInt(Number.MAX_SAFE_INTEGER)) {
+ throw new SessionSnapshotError('quota_exceeded', 'Workspace file size is unsupported', {
+ details: { phase: 'workspace' },
+ });
+ }
+ return Number(value);
+}
+
+function safeAdd(left: number, right: number): number {
+ if (!Number.isSafeInteger(left) || !Number.isSafeInteger(right) || left < 0 || right < 0) {
+ throw new SessionSnapshotError('quota_exceeded', 'Workspace quota accounting is invalid', {
+ details: { phase: 'workspace' },
+ });
+ }
+ const result = left + right;
+ if (!Number.isSafeInteger(result)) {
+ throw new SessionSnapshotError('quota_exceeded', 'Workspace quota accounting overflowed', {
+ details: { phase: 'workspace' },
+ });
+ }
+ return result;
+}
+
+function assertSnapshotActive(
+ cancellation: SessionSnapshotCancellation,
+ phase: 'state' | 'workspace',
+): void {
+ if (!cancellation.signal.aborted) return;
+ throw new SessionSnapshotError(
+ 'snapshot_cancelled',
+ 'Session snapshot preparation was cancelled',
+ {
+ details: { phase },
+ },
+ );
+}
+
+function requireAbsolutePath(value: string, label: string): string {
+ if (typeof value !== 'string' || !isAbsolute(value)) {
+ throw new TypeError(`Session snapshot ${label} must be an absolute path`);
+ }
+ return resolve(value);
+}
+
+function requireProductionSessionSnapshotBinding(
+ value: ProductionSessionSnapshotBinding,
+): Readonly<ProductionSessionSnapshotBinding> {
+ if (!value || typeof value !== 'object') {
+ throw new TypeError('Session snapshot production Session binding is required');
+ }
+ if (!isSafeSessionId(value.makaSessionId)) {
+ throw new TypeError('Session snapshot production Maka Session identity is invalid');
+ }
+ if (typeof value.cloudSessionId !== 'string' || value.cloudSessionId.length === 0) {
+ throw new TypeError('Session snapshot production Cloud Session identity is required');
+ }
+ return Object.freeze({
+ makaSessionId: value.makaSessionId,
+ cloudSessionId: value.cloudSessionId,
+ });
+}
+
+async function assertProductionRootsSeparate(input: {
+ readonly stateRoot: string;
+ readonly configRoot: string;
+ readonly workspaceRoot: string;
+ readonly stagingParent: string;
+ readonly cleanupStateRoot: string;
+}): Promise<void> {
+ const roots = await Promise.all(
+ Object.entries(input).map(async ([label, path]) => ({
+ label,
+ path: await canonicalDirectoryRoot(path, label),
+ })),
+ );
+ for (let leftIndex = 0; leftIndex < roots.length; leftIndex += 1) {
+ for (let rightIndex = leftIndex + 1; rightIndex < roots.length; rightIndex += 1) {
+ const left = roots[leftIndex]!;
+ const right = roots[rightIndex]!;
+ if (rootsOverlap(left.path, right.path)) {
+ throw new SessionSnapshotError(
+ 'unsafe_source',
+ 'Session snapshot production roots overlap',
+ { details: { phase: 'staging' } },
+ );
+ }
+ }
+ }
+}
+
+async function canonicalDirectoryRoot(path: string, label: string): Promise<string> {
+ const absolute = requireAbsolutePath(path, label);
+ try {
+ const canonical = await realpath(absolute);
+ const metadata = await lstatBigInt(canonical);
+ if (!metadata.isDirectory() || metadata.isSymbolicLink()) {
+ throw new Error('root is not a directory');
+ }
+ return canonical;
+ } catch (error) {
+ throw new SessionSnapshotError('unsafe_source', 'Session snapshot root is unsafe', {
+ cause: error,
+ details: { phase: 'staging' },
+ });
+ }
+}
+
+function rootsOverlap(left: string, right: string): boolean {
+ const leftToRight = relative(left, right);
+ const rightToLeft = relative(right, left);
+ return (
+ leftToRight === '' ||
+ rightToLeft === '' ||
+ (!leftToRight.startsWith('..') && !isAbsolute(leftToRight)) ||
+ (!rightToLeft.startsWith('..') && !isAbsolute(rightToLeft))
+ );
+}
+
+function asSnapshotError(
+ code: 'io_failure' | 'unsafe_source',
+ message: string,
+ phase: 'state' | 'workspace',
+ cause: unknown,
+): SessionSnapshotError {
+ if (cause instanceof SessionSnapshotError) return cause;
+ return new SessionSnapshotError(code, message, { cause, details: { phase } });
+}
+
+function isNodeError(error: unknown, code: string): boolean {
+ return (
+ error instanceof Error && 'code' in error && (error as NodeJS.ErrnoException).code === code
+ );
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org