fix(senpi,processing): revalidate trust lease before state deletion; restore tail dual gate and schedule coverage

Author: darko-mijic

src/processing/tail.ts

@@ -265,6 +265,8 @@ export interface RawTranscriptSessionWatchOptions extends RawTranscriptSessionTa
 export interface RawTranscriptSessionCommitOptions {
   readonly markerDir?: string;
   readonly allowedMarkerRoots?: readonly string[];
+  readonly lock?: RawTranscriptSessionLockOptions;
+  readonly onLocked?: () => void | Promise<void>;
 }
 
 export interface RawTranscriptReadOptions {
@@ -612,7 +614,11 @@ export async function commitRawTranscriptSessionCheckpoint(
     markerPath,
     sessionId,
     mainPathDigest,
-    checkpoint
+    checkpoint,
+    {
+      ...(options.lock === undefined ? {} : { lock: options.lock }),
+      ...(options.onLocked === undefined ? {} : { onLocked: options.onLocked }),
+    }
   );
 }
 
@@ -887,7 +893,10 @@ async function mutateRawTranscriptSessionMarker(
   sessionId: string,
   mainPathDigest: string,
   checkpoint: RawTranscriptSessionCheckpoint,
-  hooks?: RawTranscriptSessionLockHooks
+  options: {
+    readonly lock?: RawTranscriptSessionLockOptions;
+    readonly onLocked?: () => void | Promise<void>;
+  } = {}
 ): Promise<void> {
   await withRawTranscriptSessionMarkerLock(
     markerPath,
@@ -914,17 +923,19 @@ async function mutateRawTranscriptSessionMarker(
         currentRevision + 1,
         nextMarkers
       );
+      await options.onLocked?.();
     },
-    hooks
+    options.lock
   );
 }
 
 /**
  * Hold the raw-transcript session marker lock around `action`.
  *
  * Uses the shared token-lease core on `<markerPath>.lock`. Age-expired
- * tokens are reclaimable only when the recorded pid is dead and
- * `createdAt` is within clock-skew. `LeaseLockBusyError` is retried
+ * tokens are reclaimable only when the recorded pid is dead,
+ * `createdAt` is within clock-skew, and
+ * `now - createdAt >= SESSION_MARKER_LOCK_STALE_MS`. `LeaseLockBusyError` is retried
  * until `acquireTimeoutMs`. The canonical lock directory is never renamed or recursively removed; live tokens are never unlinked by another owner.
  *
  * @param markerPath - Session marker file whose sibling `.lock` is held.
@@ -1017,11 +1028,14 @@ function canReclaimRawTranscriptExpiredToken(
 ): boolean {
   const { token } = captured;
   if (!('pid' in token)) return false;
-  if (
-    'createdAt' in token &&
-    !isValidRawTranscriptLockCreatedAt(token.createdAt, now())
-  ) {
-    return false;
+  if ('createdAt' in token) {
+    const createdAt = token.createdAt;
+    if (!isValidRawTranscriptLockCreatedAt(createdAt, now())) {
+      return false;
+    }
+    if (now() - createdAt < SESSION_MARKER_LOCK_STALE_MS) {
+      return false;
+    }
   }
   return !probeAlive(token.pid);
 }

src/senpi/trust-writer.ts

@@ -346,6 +346,7 @@ export async function removeSenpiHookTrustEntry(
         key => key !== 'version' && key !== 'hooks'
       );
       if (leftoverIds.length === 0 && leftoverRootKeys.length === 0) {
+        await lease.assertHeld();
         rmSync(statePath, { force: true });
         return { path: statePath, id, removed: existed };
       }

tests/raw-transcript-lock-schedules.test.ts

@@ -1,4 +1,4 @@
-import { randomUUID } from 'node:crypto';
+import { createHash, randomUUID } from 'node:crypto';
 import {
   mkdir,
   mkdtemp,
@@ -9,14 +9,17 @@ import {
   writeFile,
 } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
-import { join } from 'node:path';
+import { join, resolve } from 'node:path';
 
 import { afterEach, describe, expect, it } from 'vitest';
 
 import {
-  withRawTranscriptSessionMarkerLock,
+  commitRawTranscriptSessionCheckpoint,
+  getRawTranscriptSessionMarkerPath,
+  type RawTranscriptSessionCommitOptions,
   type RawTranscriptSessionLockOptions,
 } from '../src/processing/tail.js';
+import type { RawTranscriptSessionCheckpoint } from '../src/processing/types.js';
 
 const tempRoots: string[] = [];
 const STALE_MS = 30_000;
@@ -42,13 +45,72 @@ afterEach(async () => {
 
 async function fixture(label: string): Promise<{
   root: string;
+  mainPath: string;
   markerPath: string;
   lockPath: string;
 }> {
   const root = await mkdtemp(join(tmpdir(), `raw-transcript-lock-${label}-`));
   tempRoots.push(root);
-  const markerPath = join(root, 'session.marker.json');
-  return { root, markerPath, lockPath: `${markerPath}.lock` };
+  const mainPath = join(root, 'session.jsonl');
+  const markerDir = join(root, 'markers');
+  await mkdir(markerDir, { recursive: true, mode: 0o700 });
+  await writeFile(mainPath, '');
+  const markerPath = getRawTranscriptSessionMarkerPath(mainPath, markerDir, [
+    root,
+  ]);
+  return { root, mainPath, markerPath, lockPath: `${markerPath}.lock` };
+}
+
+function commitOptions(
+  root: string,
+  extras: Omit<
+    RawTranscriptSessionCommitOptions,
+    'markerDir' | 'allowedMarkerRoots'
+  > = {}
+): RawTranscriptSessionCommitOptions {
+  return {
+    markerDir: join(root, 'markers'),
+    allowedMarkerRoots: [root],
+    ...extras,
+  };
+}
+
+function checkpoint(
+  mainPath: string,
+  baseRevision: number
+): RawTranscriptSessionCheckpoint {
+  return {
+    sessionId: 'session',
+    mainPathDigest: createHash('sha256')
+      .update(resolve(mainPath))
+      .digest('hex'),
+    baseRevision,
+    sources: [
+      {
+        sourceKind: 'main',
+        sourceId: 'main',
+        generation: 0,
+        byteOffset: 0,
+        fileSize: 0,
+      },
+    ],
+  };
+}
+
+function commitCheckpoint(
+  mainPath: string,
+  root: string,
+  baseRevision: number,
+  extras: Omit<
+    RawTranscriptSessionCommitOptions,
+    'markerDir' | 'allowedMarkerRoots'
+  > = {}
+): Promise<void> {
+  return commitRawTranscriptSessionCheckpoint(
+    mainPath,
+    checkpoint(mainPath, baseRevision),
+    commitOptions(root, extras)
+  );
 }
 
 async function tokenSnapshot(
@@ -91,40 +153,38 @@ function errorMessage(error: unknown): string {
 
 describe('raw transcript session lock three-party schedules', () => {
   it('T1 keeps fresh owner B canonical while displaced owner A releases', async () => {
-    const { markerPath, lockPath } = await fixture('t1-displaced-owner');
+    const { root, mainPath, lockPath } = await fixture('t1-displaced-owner');
     const holdA = deferred();
     const aEntered = deferred();
     const releaseBarrier = deferred();
     const aReleasePaused = deferred();
     const holdB = deferred();
     const bEntered = deferred();
 
-    const ownerA = withRawTranscriptSessionMarkerLock(
-      markerPath,
-      async () => {
-        aEntered.resolve();
-        await holdA.promise;
-      },
-      {
+    const ownerA = commitCheckpoint(mainPath, root, 0, {
+      lock: {
         onAfterReleaseTokensUnlinkedBeforeRmdir: async () => {
           aReleasePaused.resolve();
           await releaseBarrier.promise;
         },
-      }
-    );
+      },
+      onLocked: async () => {
+        aEntered.resolve();
+        await holdA.promise;
+      },
+    });
     await aEntered.promise;
 
-    const ownerB = withRawTranscriptSessionMarkerLock(
-      markerPath,
-      async () => {
+    const ownerB = commitCheckpoint(mainPath, root, 1, {
+      lock: {
+        now: () => Date.now() + STALE_MS + 5_000,
+        isProcessAlive: () => false,
+      },
+      onLocked: async () => {
         bEntered.resolve();
         await holdB.promise;
       },
-      {
-        now: () => Date.now() + STALE_MS + 5_000,
-        isProcessAlive: () => false,
-      }
-    );
+    });
     await bEntered.promise;
     const ownerBBeforeRelease = await tokenSnapshot(lockPath);
 
@@ -149,51 +209,48 @@ describe('raw transcript session lock three-party schedules', () => {
   });
 
   it('T2 rejects interloper C while displaced owner A release is in flight', async () => {
-    const { markerPath } = await fixture('t2-release-interloper');
+    const { root, mainPath } = await fixture('t2-release-interloper');
     const holdA = deferred();
     const aEntered = deferred();
     const releaseBarrier = deferred();
     const aReleasePaused = deferred();
     const holdB = deferred();
     const bEntered = deferred();
 
-    const ownerA = withRawTranscriptSessionMarkerLock(
-      markerPath,
-      async () => {
-        aEntered.resolve();
-        await holdA.promise;
-      },
-      {
+    const ownerA = commitCheckpoint(mainPath, root, 0, {
+      lock: {
         onAfterReleaseTokensUnlinkedBeforeRmdir: async () => {
           aReleasePaused.resolve();
           await releaseBarrier.promise;
         },
-      }
-    );
+      },
+      onLocked: async () => {
+        aEntered.resolve();
+        await holdA.promise;
+      },
+    });
     await aEntered.promise;
-    const ownerB = withRawTranscriptSessionMarkerLock(
-      markerPath,
-      async () => {
+    const ownerB = commitCheckpoint(mainPath, root, 1, {
+      lock: {
+        now: () => Date.now() + STALE_MS + 5_000,
+        isProcessAlive: () => false,
+      },
+      onLocked: async () => {
         bEntered.resolve();
         await holdB.promise;
       },
-      {
-        now: () => Date.now() + STALE_MS + 5_000,
-        isProcessAlive: () => false,
-      }
-    );
+    });
     await bEntered.promise;
 
     holdA.resolve();
     await aReleasePaused.promise;
     let interloperEntered = false;
-    const interloperError = await withRawTranscriptSessionMarkerLock(
-      markerPath,
-      async () => {
+    const interloperError = await commitCheckpoint(mainPath, root, 2, {
+      lock: FAST_ACQUIRE,
+      onLocked: () => {
         interloperEntered = true;
       },
-      FAST_ACQUIRE
-    ).then(
+    }).then(
       () => null,
       (error: unknown) => error
     );
@@ -213,32 +270,29 @@ describe('raw transcript session lock three-party schedules', () => {
   });
 
   it('T3 rejects interloper C while stale owner A reclamation is in flight', async () => {
-    const { markerPath, lockPath } = await fixture('t3-reclaim-interloper');
+    const { root, mainPath, lockPath } = await fixture('t3-reclaim-interloper');
     await plantStaleToken(lockPath);
     const reclaimBarrier = deferred();
     const reclaimerPaused = deferred();
 
-    const reclaimer = withRawTranscriptSessionMarkerLock(
-      markerPath,
-      async () => undefined,
-      {
+    const reclaimer = commitCheckpoint(mainPath, root, 0, {
+      lock: {
         isProcessAlive: () => false,
         onAfterExpiredTokensUnlinkedBeforeRmdir: async () => {
           reclaimerPaused.resolve();
           await reclaimBarrier.promise;
         },
-      }
-    );
+      },
+    });
     await reclaimerPaused.promise;
 
     let interloperEntered = false;
-    const interloperError = await withRawTranscriptSessionMarkerLock(
-      markerPath,
-      async () => {
+    const interloperError = await commitCheckpoint(mainPath, root, 0, {
+      lock: FAST_ACQUIRE,
+      onLocked: () => {
         interloperEntered = true;
       },
-      FAST_ACQUIRE
-    ).then(
+    }).then(
       () => null,
       (error: unknown) => error
     );

tests/raw-transcript-lock.test.ts

@@ -202,6 +202,32 @@ describe('raw transcript session lock adapter', () => {
     await expectMissing(lockPath);
   });
 
+  it('does not reclaim a stale-mtime token whose createdAt is still fresh', async () => {
+    const { markerPath, lockPath } = await fixture(
+      'stale-mtime-fresh-created-at'
+    );
+    const planted = await plantToken(lockPath, {
+      stale: true,
+      pid: DEAD_PID,
+      createdAt: Date.now(),
+    });
+    const before = await readFile(planted, 'utf8');
+    let actionRan = false;
+
+    await expect(
+      withRawTranscriptSessionMarkerLock(
+        markerPath,
+        async () => {
+          actionRan = true;
+        },
+        FAST_ACQUIRE
+      )
+    ).rejects.toThrow(`Timed out acquiring session marker lock '${lockPath}'`);
+
+    expect(actionRan).toBe(false);
+    expect(await readFile(planted, 'utf8')).toBe(before);
+  });
+
   it('does not reclaim a stale lock whose pid is still alive', async () => {
     const { markerPath, lockPath } = await fixture('stale-alive');
     const planted = await plantToken(lockPath, {