fix(storage): publish imported context payloads atomically

packages/runtime/src/__tests__/session-export.test.ts

@@ -1059,3 +1059,60 @@ test(
     await assert.rejects(stat(destination));
   }),
 );
+
+test(
+  'exports under authority the caller already holds',
+  withRoot('maka-export-lease', async (root, workspaceRoot) => {
+    const sessionId = await createSession(workspaceRoot, { name: 'Exported' });
+    const destination = join(root, 'bundle.maka-session');
+
+    const { resolveStorageRoot, tryAcquireInteractiveRootOwner } = await import(
+      '@maka/storage/root-authority'
+    );
+    // What a Runtime Host is: it took this authority at startup and holds it
+    // until it exits. The lock is an election taken with `tryLock`, so a second
+    // exclusive hold is refused even inside the process that already has it --
+    // which is why a Host cannot reach the export by calling it, only by
+    // lending what it holds.
+    const capability = await resolveStorageRoot({ path: workspaceRoot, kind: 'interactive' });
+    const owner = await tryAcquireInteractiveRootOwner(capability);
+    assert.ok(owner, 'the probe must hold the authority for this test to mean anything');
+    try {
+      // A workspace that has actually been used: the context store exists and
+      // holds a payload, which is what makes the authority necessary at all.
+      const { openInteractiveContextOffloadStoreForWrite } = await import(
+        '@maka/storage/context-offload-store'
+      );
+      const store = await openInteractiveContextOffloadStoreForWrite(owner.lease, {
+        limits: {
+          ownerMaxBytes: { read_image_snapshot: 4096, tool_result_archive: 4096 },
+          sessionLogicalBytes: 1_000_000,
+          workspacePhysicalBytes: 10_000_000,
+        },
+      });
+      const put = await store.put({
+        sessionId,
+        owner: { kind: 'read_image_snapshot', ownerId: 'shot-1' },
+        bytes: new TextEncoder().encode('PAYLOAD'),
+        mediaType: 'image/png',
+      });
+      assert.equal(put.ok, true);
+      await store.close();
+
+      const refused = await exportSessionBundle({ workspaceRoot, sessionId, destination });
+      assert.equal(refused.ok, false, 'electing the authority cannot work while it is held');
+
+      const exported = await exportSessionBundle({
+        workspaceRoot,
+        sessionId,
+        destination,
+        lease: owner.lease,
+      });
+      if (!exported.ok) assert.fail(`export failed: ${JSON.stringify(exported.reason)}`);
+      assert.deepEqual(exported.export.sessionIds, [sessionId]);
+      assert.ok((await stat(destination)).size > 0);
+    } finally {
+      await owner?.close();
+    }
+  }),
+);

packages/runtime/src/__tests__/session-import.test.ts

@@ -18,9 +18,9 @@
  */
 
 import assert from 'node:assert/strict';
-import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
+import { mkdir, mkdtemp, readdir, rm, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
-import { join } from 'node:path';
+import { basename, dirname, join } from 'node:path';
 import { DatabaseSync } from 'node:sqlite';
 import { test } from 'node:test';
 import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store';
@@ -835,3 +835,170 @@ test('accepts an identical artifact left by a crashed attempt, and only an ident
     await rm(target.root, { recursive: true, force: true });
   }
 });
+
+test('refuses a context payload whose path already holds different bytes', async () => {
+  const source = await makeWorkspace('maka-import-payload-source');
+  const target = await makeWorkspace('maka-import-payload-target');
+  try {
+    const sessionId = await createSession(source.workspaceRoot);
+    await seedHistory(source.workspaceRoot, sessionId);
+    const seeded = await seedContext(source.workspaceRoot, sessionId, 'ABC');
+
+    const targetSession = await createSession(target.workspaceRoot, 'Unrelated');
+    await seedContext(target.workspaceRoot, targetSession, 'ZZ');
+
+    // Payloads are addressed by the hash of their content, so the same path in
+    // two workspaces is supposed to mean the same bytes. When it does not, the
+    // target holds something this bundle cannot explain -- and the import used
+    // to swallow `EEXIST` and report success over it.
+    const occupied = join(target.workspaceRoot, 'context-offload-values', seeded.relativePath);
+    await mkdir(dirname(occupied), { recursive: true });
+    await writeFile(occupied, 'NOT-THE-SAME');
+
+    await assert.rejects(
+      () => importState(target.workspaceRoot, source.workspaceRoot),
+      /Context payload already names different content/,
+    );
+
+    const { readFile } = await import('node:fs/promises');
+    assert.equal(await readFile(occupied, 'utf8'), 'NOT-THE-SAME', 'the other payload survives');
+  } finally {
+    await rm(source.root, { recursive: true, force: true });
+    await rm(target.root, { recursive: true, force: true });
+  }
+});
+
+test('publishes a context payload past a staging file a crashed attempt left', async () => {
+  const source = await makeWorkspace('maka-import-staging-source');
+  const target = await makeWorkspace('maka-import-staging-target');
+  try {
+    const sessionId = await createSession(source.workspaceRoot);
+    await seedHistory(source.workspaceRoot, sessionId);
+    const seeded = await seedContext(source.workspaceRoot, sessionId, 'ABC');
+
+    const targetSession = await createSession(target.workspaceRoot, 'Unrelated');
+    await seedContext(target.workspaceRoot, targetSession, 'ZZ');
+
+    // A payload now becomes visible by `link`, so an interrupted copy leaves its
+    // half-written bytes under the staging name rather than at the path other
+    // Sessions read. The retry has to get past its own leftover.
+    const values = join(target.workspaceRoot, 'context-offload-values');
+    const destination = join(values, seeded.relativePath);
+    const staging = join(dirname(destination), `.${basename(destination)}.import.tmp`);
+    await mkdir(dirname(destination), { recursive: true });
+    await writeFile(staging, 'HALF');
+
+    const imported = await importState(target.workspaceRoot, source.workspaceRoot);
+    assert.equal(imported.contextRefs, 1);
+
+    const { readFile } = await import('node:fs/promises');
+    assert.equal(await readFile(destination, 'utf8'), 'ABC', 'the leftover was not published');
+    assert.deepEqual(
+      (await readdir(dirname(destination))).filter((name) => name.endsWith('.import.tmp')),
+      [],
+      'staging files do not outlive the import',
+    );
+  } finally {
+    await rm(source.root, { recursive: true, force: true });
+    await rm(target.root, { recursive: true, force: true });
+  }
+});
+
+test('imports under authority the caller already holds', async () => {
+  const source = await makeWorkspace('maka-import-lease-source');
+  const target = await makeWorkspace('maka-import-lease-target');
+  try {
+    const sessionId = await createSession(source.workspaceRoot);
+    await seedHistory(source.workspaceRoot, sessionId);
+    const seeded = await seedContext(source.workspaceRoot, sessionId, 'ABC');
+    await createSession(target.workspaceRoot, 'Unrelated');
+
+    const { resolveStorageRoot, tryAcquireInteractiveRootOwner } = await import(
+      '@maka/storage/root-authority'
+    );
+    const { importSessionBundleState } = await import('@maka/storage/session-bundle-policy');
+
+    // What a Runtime Host is: it took this authority at startup and holds it
+    // for its lifetime.
+    const capability = await resolveStorageRoot({
+      path: target.workspaceRoot,
+      kind: 'interactive',
+    });
+    const owner = await tryAcquireInteractiveRootOwner(capability);
+    assert.ok(owner, 'the probe must hold the authority for this test to mean anything');
+    try {
+      // The lock is an election, not a mutex: it refuses a second hold even
+      // from the process already holding it, so electing here cannot work.
+      await assert.rejects(
+        () =>
+          importSessionBundleState({
+            stateRoot: target.workspaceRoot,
+            bundleStateRoot: source.workspaceRoot,
+          }),
+        /offline Storage Root/,
+      );
+
+      const imported = await importSessionBundleState({
+        stateRoot: target.workspaceRoot,
+        bundleStateRoot: source.workspaceRoot,
+        lease: owner.lease,
+      });
+      assert.deepEqual([...imported.sessionIds], [sessionId]);
+      assert.equal(imported.contextRefs, 1);
+      const { readFile } = await import('node:fs/promises');
+      assert.equal(
+        await readFile(
+          join(target.workspaceRoot, 'context-offload-values', seeded.relativePath),
+          'utf8',
+        ),
+        'ABC',
+      );
+    } finally {
+      await owner?.close();
+    }
+  } finally {
+    await rm(source.root, { recursive: true, force: true });
+    await rm(target.root, { recursive: true, force: true });
+  }
+});
+
+test('refuses a lease that names a different Storage Root', async () => {
+  const source = await makeWorkspace('maka-import-wrong-lease-source');
+  const target = await makeWorkspace('maka-import-wrong-lease-target');
+  const other = await makeWorkspace('maka-import-wrong-lease-other');
+  try {
+    const sessionId = await createSession(source.workspaceRoot);
+    await seedHistory(source.workspaceRoot, sessionId);
+    await seedContext(source.workspaceRoot, sessionId, 'ABC');
+    await createSession(target.workspaceRoot, 'Unrelated');
+    await createSession(other.workspaceRoot, 'Elsewhere');
+
+    const { resolveStorageRoot, tryAcquireInteractiveRootOwner } = await import(
+      '@maka/storage/root-authority'
+    );
+    const { importSessionBundleState } = await import('@maka/storage/session-bundle-policy');
+
+    // A valid lease is still only authority over the root it names. Accepting
+    // one for a different root would write to a directory nobody holds.
+    const capability = await resolveStorageRoot({ path: other.workspaceRoot, kind: 'interactive' });
+    const owner = await tryAcquireInteractiveRootOwner(capability);
+    assert.ok(owner);
+    try {
+      await assert.rejects(
+        () =>
+          importSessionBundleState({
+            stateRoot: target.workspaceRoot,
+            bundleStateRoot: source.workspaceRoot,
+            lease: owner.lease,
+          }),
+        /does not name this Storage Root/,
+      );
+    } finally {
+      await owner?.close();
+    }
+  } finally {
+    await rm(source.root, { recursive: true, force: true });
+    await rm(target.root, { recursive: true, force: true });
+    await rm(other.root, { recursive: true, force: true });
+  }
+});

packages/runtime/src/session-export.ts

@@ -44,6 +44,7 @@ import {
   type SessionBundleFileErrorDetails,
   type SessionBundleLimits,
 } from '@maka/storage/session-bundle-contract';
+import type { StorageRootLease } from '@maka/storage/root-authority';
 import { createSessionBundleFileService } from '@maka/storage/session-bundle-file-service';
 import {
   exportSessionBundleState,
@@ -93,6 +94,15 @@ export interface ExportSessionBundleInput {
   destination: string;
   limits?: SessionBundleLimits;
   now?: () => number;
+  /**
+   * Storage Root authority the caller already holds.
+   *
+   * The Runtime Host takes it at startup and holds it for its lifetime, and
+   * the lock is an election that refuses a second hold -- so the Host can only
+   * do this by lending what it has. A caller with no authority of its own, the
+   * CLI included, omits it and the authority is elected as before.
+   */
+  lease?: StorageRootLease<'interactive', 'write'>;
 }
 
 export type ExportSessionBundleFailure =
@@ -148,6 +158,7 @@ export async function exportSessionBundle(
         allowShared: true,
         destinationRoot: stateRoot,
         sessionId: input.sessionId,
+        ...(input.lease ? { lease: input.lease } : {}),
         requireQuiescent: true,
         includeSubtree: true,
         omitDiagnostics: true,

packages/runtime/src/session-import.ts

@@ -36,6 +36,7 @@ import {
   SessionBundleFileError,
   type SessionBundleLimits,
 } from '@maka/storage/session-bundle-contract';
+import type { StorageRootLease } from '@maka/storage/root-authority';
 import { createSessionBundleFileService } from '@maka/storage/session-bundle-file-service';
 import {
   importSessionBundleState,
@@ -48,6 +49,15 @@ export interface ImportSessionBundleInput {
   /** The `.maka-session` file to read. */
   source: string;
   limits?: SessionBundleLimits;
+  /**
+   * Storage Root authority the caller already holds.
+   *
+   * The Runtime Host takes it at startup and holds it for its lifetime, and
+   * the lock is an election that refuses a second hold -- so the Host can only
+   * do this by lending what it has. A caller with no authority of its own, the
+   * CLI included, omits it and the authority is elected as before.
+   */
+  lease?: StorageRootLease<'interactive', 'write'>;
 }
 
 export type ImportSessionBundleFailure =
@@ -97,6 +107,7 @@ export async function importSessionBundle(
     const merged = await importSessionBundleState({
       stateRoot: input.workspaceRoot,
       bundleStateRoot: hydration.stateRoot,
+      ...(input.lease ? { lease: input.lease } : {}),
     });
     return { ok: true, ...merged, manifest };
   } catch (error) {

packages/storage/src/context-offload-snapshot.ts

@@ -28,6 +28,7 @@ import {
   resolveStorageRoot,
   runWithStorageRootLease,
   tryAcquireInteractiveRootOwner,
+  type StorageRootLease,
 } from './root-authority.js';
 import {
   migrateSqliteContextOffloadDatabase,
@@ -58,6 +59,19 @@ export async function withOfflineContextSnapshot<T>(
      * runs unprotected.
      */
     requireAuthority?: boolean;
+    /**
+     * Run under authority the caller already holds instead of electing it.
+     *
+     * The owner lock is an election, not a mutex: it is taken with `tryLock`
+     * and refuses a second exclusive hold on the same file even from the same
+     * process. So a Runtime Host cannot reach this path by calling it -- it
+     * would be refused by its own lock -- and the only way it can prepare or
+     * accept a bundle is to lend the authority it took at startup.
+     *
+     * The lease must name this same root. A valid lease for a DIFFERENT root
+     * would otherwise authorise writing to a directory nobody holds.
+     */
+    lease?: StorageRootLease<'interactive', 'write'>;
   } = {},
 ): Promise<T> {
   if (
@@ -66,6 +80,13 @@ export async function withOfflineContextSnapshot<T>(
   ) {
     return operation(false);
   }
+  const lease = options.lease;
+  if (lease) {
+    if ((await realpath(root).catch(() => resolve(root))) !== lease.canonicalPath) {
+      throw new Error('Context snapshot lease does not name this Storage Root');
+    }
+    return runWithStorageRootLease(lease, 'interactive', 'write', () => operation(true));
+  }
   // Discovery finds a marked root; it does not make one. A workspace that has
   // never been opened is exactly the target an import writes to first, so when
   // the caller says it is about to write, the root is resolved -- which