fix(tui): close resume handoff races

packages/session-persistence/session-persistence-jsonl/README.md

@@ -69,6 +69,6 @@ JSONL storage does not mutate live request prefixes. A resumed loop can reuse pr
 - **Only the configured encoding and current `SESSION_FORMAT_VERSION` (v0) load** — changing compression requires a separate/fresh root or selecting the legacy raw mode; the pre-release format has no migration.
 - **Compressed files are not directly line-readable** — use the backend to load them, or select `compression: 'none'` before writing a fresh root when text fixtures or external line readers are required.
 - **Nothing deletes session files** — logs accumulate under `root` until removed externally (the seam has no deletion surface).
-- **Lease scope is local-host advisory ownership** — PID liveness prevents two ordinary local Harness processes from resuming the same id, but it is not a distributed lease for shared network filesystems or hostile principals.
+- **Lease scope is local-host advisory ownership** — PID plus same-process nonce checks prevent two ordinary local Harness processes from resuming the same id, but foreign PID reuse remains fail-closed and this is not a distributed lease for shared network filesystems or hostile principals.
 - **A crash during stale-lease takeover fails closed** — if the reclaiming process itself crashes while holding the short-lived `.reclaim` guard, an operator must remove that guard after confirming no recovery is active.
 - **POSIX materialization requires hard-link support** — first append uses `link()` so same-id races fail instead of overwriting a committed log; Windows uses write-through rename without replacement.

packages/session-persistence/session-persistence-jsonl/src/index.ts

@@ -14,7 +14,7 @@ import { dirname, join, resolve } from 'node:path'
 import { randomBytes } from 'node:crypto'
 import {
   SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
-  sessionLeaseProcessIsLive, shareSessionLiveLease,
+  sessionLeaseOwnerIsLive, shareSessionLiveLease,
   type PersistenceBackend, type SessionLiveLease, type SessionLiveOwner,
   type SessionLocation, type SessionPersistenceSnapshot, type StoredPrefix,
 } from '@deepseek-ai/dsh-session-persistence'
@@ -314,7 +314,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
         if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error
         const current = await this.readLiveLease(path)
         if (current !== undefined && current.pid === owner.pid && current.nonce === owner.nonce) break
-        if (current === undefined || sessionLeaseProcessIsLive(current.pid)) {
+        if (current === undefined || sessionLeaseOwnerIsLive(current, owner)) {
           throw new Error(`session "${id}" is occupied by another live process`)
         }
         const reclaimPath = `${path}.reclaim`
@@ -335,7 +335,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
           if (latest === undefined) {
             if (await this.exists(path)) throw new Error(`session "${id}" has an unreadable live-process lease`)
           } else if (latest.pid !== owner.pid || latest.nonce !== owner.nonce) {
-            if (sessionLeaseProcessIsLive(latest.pid)) {
+            if (sessionLeaseOwnerIsLive(latest, owner)) {
               throw new Error(`session "${id}" is occupied by another live process`)
             }
             await rm(path, { force: true })
@@ -356,13 +356,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     }
   }
 
-  /** Report one non-stale process lease and clean up a crashed owner's record. */
+  /** Report one non-stale process lease; acquisition reclaims a crashed owner's record. */
   async inspectLive(id: SessionId, owner: SessionLiveOwner): Promise<boolean> {
     const path = this.liveLeasePath(id)
     const current = await this.readLiveLease(path)
     if (current === undefined) return await this.exists(path)
     if (current.pid === owner.pid && current.nonce === owner.nonce) return true
-    if (sessionLeaseProcessIsLive(current.pid)) return true
+    if (sessionLeaseOwnerIsLive(current, owner)) return true
     return false
   }
 

packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts

@@ -249,6 +249,13 @@ describe('SessionPersistenceJsonl: cross-process live leases', () => {
       const inheritedClaim = await ctx.sessionPersistence.claimLive(inherited)
       await inheritedClaim.release()
 
+      const reusedPid = SessionId('reused-pid')
+      const reusedPidPath = join(liveDir, `${encodeSegment(reusedPid)}.lock`)
+      await writeFile(reusedPidPath, JSON.stringify({ pid: process.pid, nonce: 'prior-incarnation' }))
+      await expect(ctx.sessionPersistence.isLive(reusedPid)).resolves.toBe(false)
+      const reusedPidClaim = await ctx.sessionPersistence.claimLive(reusedPid)
+      await reusedPidClaim.release()
+
       await expect(ctx.sessionPersistence.claimLive(SessionId('x'.repeat(300))))
         .rejects.toThrow()
 

packages/session-persistence/session-persistence-sqlite/README.md

@@ -57,3 +57,4 @@ SQLite storage does not mutate live request prefixes. A resumed loop can reuse p
 - **Write contention has no wait or retry policy** — the backend sets no busy timeout and retries no locked-database error, so another connection holding a write transaction makes the operation reject immediately.
 - **Only the current `SCHEMA_VERSION` opens** — a database with any other schema version is rejected rather than migrated (unreleased software; no persisted user data to preserve).
 - **Nothing deletes stored sessions** — rows accumulate until removed externally (the seam has no deletion surface; `ON DELETE CASCADE` is wired for such out-of-band cleanup).
+- **Foreign PID reuse is fail-closed** — same-PID claimants compare the exec-stable nonce, while other processes conservatively retain a stale row until the reused PID exits or an operator verifies and removes it.

packages/session-persistence/session-persistence-sqlite/src/index.ts

@@ -15,7 +15,7 @@ import { mkdir, open } from 'node:fs/promises'
 import { dirname, resolve } from 'node:path'
 import {
   SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
-  sessionLeaseProcessIsLive, shareSessionLiveLease,
+  sessionLeaseOwnerIsLive, shareSessionLiveLease,
   type PersistenceBackend, type SessionLiveLease, type SessionLiveOwner,
   type SessionLocation, type SessionPersistenceSnapshot, type StoredPrefix,
 } from '@deepseek-ai/dsh-session-persistence'
@@ -295,7 +295,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
       const current = this.liveLeaseFor(id)
       if (current !== undefined
         && (current.pid !== owner.pid || current.nonce !== owner.nonce)) {
-        if (sessionLeaseProcessIsLive(current.pid)) {
+        if (sessionLeaseOwnerIsLive(current, owner)) {
           throw new Error(`session "${id}" is occupied by another live process`)
         }
         this.db.prepare('DELETE FROM live_session_leases WHERE session_id = ?').run(id)
@@ -323,7 +323,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
     const current = this.liveLeaseFor(id)
     if (current === undefined) return false
     if ((current.pid === owner.pid && current.nonce === owner.nonce)
-      || sessionLeaseProcessIsLive(current.pid)) return true
+      || sessionLeaseOwnerIsLive(current, owner)) return true
     this.db.prepare('DELETE FROM live_session_leases WHERE session_id = ? AND pid = ? AND nonce = ?')
       .run(id, current.pid, current.nonce)
     return false

packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts

@@ -1,4 +1,4 @@
-import { afterEach, describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import { existsSync } from 'node:fs'
 import { chmod, mkdtemp, rm, stat, symlink, writeFile } from 'node:fs/promises'
@@ -13,7 +13,10 @@ import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../sessi
 import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
 
 const dirs: string[] = []
-afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
+afterEach(async () => {
+  vi.restoreAllMocks()
+  for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
+})
 
 async function expectFlushError(promise: Promise<unknown>, message: RegExp): Promise<void> {
   try {
@@ -465,9 +468,16 @@ describe('SessionPersistenceSqlite: edge cases', () => {
     await b.ctx.sessionPersistence.list()
     const concrete = b.ctx.sessionPersistence as SessionPersistenceSqlite
     const owner = sessionLiveOwner()
+    const occupiedPid = process.pid + 1
+    const originalKill = process.kill.bind(process)
+    vi.spyOn(process, 'kill').mockImplementation((pid, signal) => {
+      if (pid === occupiedPid) return true
+      return originalKill(pid, signal)
+    })
     const db = openDatabase(path, 'wal')
     const insert = db.prepare('INSERT INTO live_session_leases (session_id, pid, nonce) VALUES (?, ?, ?)')
-    insert.run('occupied-lease', process.pid, 'another-owner')
+    insert.run('occupied-lease', occupiedPid, 'another-owner')
+    insert.run('reused-pid', process.pid, 'prior-incarnation')
     insert.run('stale-claim', 2_147_483_647, 'dead-owner')
     insert.run('stale-inspect', 2_147_483_647, 'dead-owner')
     insert.run('owned-inspect', owner.pid, owner.nonce)
@@ -475,11 +485,13 @@ describe('SessionPersistenceSqlite: edge cases', () => {
 
     await expect(concrete.acquireLive(SessionId('occupied-lease'), owner))
       .rejects.toThrow('occupied by another live process')
+    const reused = await concrete.acquireLive(SessionId('reused-pid'), owner)
     const claim = await concrete.acquireLive(SessionId('stale-claim'), owner)
     expect(await concrete.inspectLive(SessionId('owned-inspect'), owner)).toBe(true)
     expect(await concrete.inspectLive(SessionId('stale-inspect'), owner)).toBe(false)
     expect(await concrete.inspectLive(SessionId('missing-inspect'), owner)).toBe(false)
     await claim()
+    await reused()
     await b.dispose()
 
     const memory = new Context()

packages/session-persistence/session-persistence/README.md

@@ -85,3 +85,4 @@ Persistence does not mutate live request prefixes. A resumed loop can reuse prov
 - **No deletion or retention surface** — pruning stored sessions is out-of-band backend maintenance.
 - **`list()` is unpaginated and unfiltered** — it returns every stored session's header; fine for local stores, unindexed at scale.
 - **Repair-time synthetic closers are the only crash story** — a backend must synthesize `tool/result`/`step/end`/`turn/end` closers on load; there is no partial-turn resume that continues an interrupted turn instead of closing it.
+- **Foreign PID reuse is fail-closed** — a claimant with the reused PID detects its different nonce and reclaims safely, but another process cannot observe that foreign process's private nonce and treats the PID as live until it exits or an operator verifies and removes the stale lease.

packages/session-persistence/session-persistence/src/index.ts

@@ -13,7 +13,12 @@ import type { SessionLiveLease } from './lease.ts'
 // Re-export the metadata vocabulary so consumers import it from the seam.
 export type { SessionHeader } from '@deepseek-ai/dsh-session'
 export { SessionPersistenceRevision } from './revision.ts'
-export { sessionLeaseProcessIsLive, sessionLiveOwner, shareSessionLiveLease } from './lease.ts'
+export {
+  sessionLeaseOwnerIsLive,
+  sessionLeaseProcessIsLive,
+  sessionLiveOwner,
+  shareSessionLiveLease,
+} from './lease.ts'
 export type { SessionLiveLease, SessionLiveOwner } from './lease.ts'
 
 /** Lightweight immutable source identity returned without loading a full log. */

packages/session-persistence/session-persistence/src/lease.ts

@@ -8,7 +8,7 @@ const LIVE_OWNER_ENV = 'DSH_SESSION_LIVE_OWNER'
 export interface SessionLiveOwner {
   /** Operating-system process id; retained across an `execve` handoff. */
   readonly pid: number
-  /** Per-process-start nonce that distinguishes PID reuse. */
+  /** Exec-stable process-start nonce used when the observer has the same PID. */
   readonly nonce: string
 }
 
@@ -42,9 +42,26 @@ export function sessionLeaseProcessIsLive(pid: number): boolean {
   }
 }
 
+/**
+ * Whether a recorded owner still names this process incarnation or another live PID.
+ * A same-PID nonce mismatch proves reuse and is stale; an unrelated live PID is
+ * fail-closed because its private nonce is not observable across processes.
+ * @param recorded - owner stored in the backend lease.
+ * @param observer - identity of the process inspecting or claiming the lease.
+ * @returns whether the recorded owner must still be treated as live.
+ */
+export function sessionLeaseOwnerIsLive(
+  recorded: SessionLiveOwner,
+  observer: SessionLiveOwner,
+): boolean {
+  if (recorded.pid === observer.pid) return recorded.nonce === observer.nonce
+  return sessionLeaseProcessIsLive(recorded.pid)
+}
+
 interface SharedLeaseEntry {
   refs: number
   readonly acquired: Promise<() => Promise<void>>
+  finalizing?: Promise<void>
 }
 
 const sharedLeases = new Map<string, SharedLeaseEntry>()
@@ -59,40 +76,48 @@ export async function shareSessionLiveLease(
   key: string,
   acquire: () => Promise<() => Promise<void>>,
 ): Promise<() => Promise<void>> {
-  let entry = sharedLeases.get(key)
-  if (entry === undefined) {
-    entry = { refs: 0, acquired: acquire() }
-    sharedLeases.set(key, entry)
-    void entry.acquired.catch(() => {
-      /* v8 ignore next -- no public operation can replace a still-acquiring module-private entry */
-      if (sharedLeases.get(key) === entry) sharedLeases.delete(key)
-    })
-  }
-  entry.refs += 1
-  try {
-    await entry.acquired
-  } catch (error) {
-    entry.refs -= 1
-    throw error
-  }
-  let releaseTask: Promise<void> | undefined
-  return () => {
-    if (releaseTask !== undefined) return releaseTask
-    const task = (async () => {
+  for (;;) {
+    let entry = sharedLeases.get(key)
+    if (entry?.finalizing !== undefined) {
+      await entry.finalizing
+      continue
+    }
+    if (entry === undefined) {
+      entry = { refs: 0, acquired: acquire() }
+      sharedLeases.set(key, entry)
+      void entry.acquired.catch(() => {
+        /* v8 ignore next -- no public operation can replace a still-acquiring module-private entry */
+        if (sharedLeases.get(key) === entry) sharedLeases.delete(key)
+      })
+    }
+    entry.refs += 1
+    try {
+      await entry.acquired
+    } catch (error) {
       entry.refs -= 1
-      if (entry.refs > 0 || sharedLeases.get(key) !== entry) return
-      const release = await entry.acquired
-      await release()
-      /* v8 ignore next -- the entry remains installed until this exact final release succeeds */
-      if (sharedLeases.get(key) === entry) sharedLeases.delete(key)
-    })()
-    const wrapped = task.catch((error: unknown) => {
-      entry.refs += 1
-      /* v8 ignore next -- this closure is the sole writer of its releaseTask until settlement */
-      if (releaseTask === wrapped) releaseTask = undefined
       throw error
-    })
-    releaseTask = wrapped
-    return wrapped
+    }
+    let releaseTask: Promise<void> | undefined
+    return () => {
+      if (releaseTask !== undefined) return releaseTask
+      const task = (async () => {
+        entry.refs -= 1
+        if (entry.refs > 0 || sharedLeases.get(key) !== entry) return
+        const release = await entry.acquired
+        await release()
+        /* v8 ignore next -- claims wait for finalization before they can replace this exact entry */
+        if (sharedLeases.get(key) === entry) sharedLeases.delete(key)
+      })()
+      const wrapped = task.catch((error: unknown) => {
+        entry.refs += 1
+        /* v8 ignore next -- this closure is the sole writer of its release state until settlement */
+        if (entry.finalizing === wrapped) delete entry.finalizing
+        releaseTask = undefined
+        throw error
+      })
+      if (entry.refs === 0 && sharedLeases.get(key) === entry) entry.finalizing = wrapped
+      releaseTask = wrapped
+      return wrapped
+    }
   }
 }

packages/session-persistence/session-persistence/tests/lease.spec.ts

@@ -1,6 +1,7 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import { randomUUID } from 'node:crypto'
 import {
+  sessionLeaseOwnerIsLive,
   sessionLeaseProcessIsLive,
   sessionLiveOwner,
   shareSessionLiveLease,
@@ -21,10 +22,14 @@ describe('process live-session lease helpers', () => {
     expect(first.pid).toBe(process.pid)
     expect(typeof first.nonce).toBe('string')
     expect(sessionLiveOwner()).toEqual(first)
+    expect(sessionLeaseOwnerIsLive(first, first)).toBe(true)
+    expect(sessionLeaseOwnerIsLive({ ...first, nonce: 'reused-pid' }, first)).toBe(false)
     expect(sessionLeaseProcessIsLive(process.pid)).toBe(true)
 
     const missing = Object.assign(new Error('missing'), { code: 'ESRCH' })
     vi.spyOn(process, 'kill').mockImplementationOnce(() => { throw missing })
+    expect(sessionLeaseOwnerIsLive({ pid: 999_999, nonce: 'gone' }, first)).toBe(false)
+    vi.spyOn(process, 'kill').mockImplementationOnce(() => { throw missing })
     expect(sessionLeaseProcessIsLive(999_999)).toBe(false)
     const denied = Object.assign(new Error('denied'), { code: 'EPERM' })
     vi.spyOn(process, 'kill').mockImplementationOnce(() => { throw denied })
@@ -59,4 +64,29 @@ describe('process live-session lease helpers', () => {
     await expect(release()).resolves.toBeUndefined()
     expect(releases).toBe(2)
   })
+
+  it('waits for a final physical release before reacquiring the same key', async () => {
+    const key = `finalizing-${randomUUID()}`
+    const releaseGate = Promise.withResolvers<undefined>()
+    const firstPhysicalRelease = vi.fn(() => releaseGate.promise)
+    const secondPhysicalRelease = vi.fn(() => Promise.resolve())
+    const releases: Array<() => Promise<void>> = [firstPhysicalRelease, secondPhysicalRelease]
+    let acquisitions = 0
+    const acquire = vi.fn<() => Promise<() => Promise<void>>>((): Promise<() => Promise<void>> => {
+      const release = releases[acquisitions++]
+      if (release === undefined) throw new Error('unexpected physical acquisition')
+      return Promise.resolve(release)
+    })
+    const first = await shareSessionLiveLease(key, acquire)
+    const finalizing = first()
+    const reacquiring = shareSessionLiveLease(key, acquire)
+    await Promise.resolve()
+    expect(acquire).toHaveBeenCalledTimes(1)
+    releaseGate.resolve(undefined)
+    await finalizing
+    const second = await reacquiring
+    expect(acquire).toHaveBeenCalledTimes(2)
+    await second()
+    expect(secondPhysicalRelease).toHaveBeenCalledTimes(1)
+  })
 })