refactor(tui): defer cross-process resume locking
This commit is contained in:
Turtle
@@ -480,14 +480,6 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
||||
signature: 'abstract listSnapshots(): Promise<SessionPersistenceSnapshot[]>',
|
||||
jsDoc: '/**\n * List materialized sessions with cheap per-log change tokens.\n *\n * Repeated observations of an unchanged log return the same revision. A\n * successful mutating {@link load} repair changes the next listed revision.\n * Revisions also distinguish independently backed stores so backend-local\n * counters cannot compare equal across different persistence sources.\n * @returns one header and opaque revision per materialized session without loading full logs.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'claimLive(id: SessionId): Promise<SessionLiveLease>',
|
||||
jsDoc: '/**\n * Atomically acquire this process\'s live ownership of a session id.\n * Reentrant claims share one backend lease. First-party backends override\n * this process-local fallback to reject another live process and reclaim a\n * dead owner.\n * @param id - session identity that is about to become live.\n * @returns a single-release reference owned by the caller.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'isLive(id: SessionId): Promise<boolean>',
|
||||
jsDoc: '/**\n * Check whether any process currently owns a live lease for this session.\n * The base implementation reports only claims on this service instance.\n * @param id - persisted or prospective session identity.\n * @returns true while a non-stale lease exists, including this process\'s lease.\n */',
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
@@ -1817,10 +1809,6 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'SessionLineageTrace',
|
||||
declaration: 'export type SessionLineageTrace = {\n target: SessionRecord;\n ancestors: SessionRecord[];\n descendants: SessionLineageNode[];\n} & ({\n complete: true;\n root: SessionRecord;\n} | {\n complete: false;\n unresolvedParentId: SessionId;\n});',
|
||||
},
|
||||
{
|
||||
name: 'SessionLiveLease',
|
||||
declaration: 'export interface SessionLiveLease {\n release(): Promise<void>;\n}',
|
||||
},
|
||||
{
|
||||
name: 'SessionLocation',
|
||||
declaration: 'export interface SessionLocation {\n readonly kind: string;\n readonly path: string;\n}',
|
||||
|
||||
@@ -25,7 +25,7 @@ import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import type { SessionLiveLease, SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
||||
import {
|
||||
bindReactLoopAgentContext,
|
||||
prepareReactLoopAgent,
|
||||
@@ -114,7 +114,6 @@ class AgentCreationTransaction {
|
||||
private scope: Scope | undefined
|
||||
private session: Session | undefined
|
||||
private lifecycleDispose: (() => Promise<void> | void) | undefined
|
||||
private liveLease: SessionLiveLease | undefined
|
||||
private detachSession: (() => void) | undefined
|
||||
private detachAgent: (() => void) | undefined
|
||||
private publishing = false
|
||||
@@ -187,12 +186,6 @@ class AgentCreationTransaction {
|
||||
])
|
||||
}
|
||||
|
||||
/** Retain a pre-load persistence lease until this transaction fully tears down. */
|
||||
holdLiveLease(lease: SessionLiveLease): void {
|
||||
this.assertActive()
|
||||
this.liveLease = lease
|
||||
}
|
||||
|
||||
/** Construct the driver and scope, then install their complete ordered lifecycle. */
|
||||
prepare(options: AgentOptions, session: Session, maxParallelToolCalls: number): ReactLoopAgent {
|
||||
this.assertActive()
|
||||
@@ -226,11 +219,6 @@ class AgentCreationTransaction {
|
||||
// First yielded, disposed last.
|
||||
yield () => { this.finish() }
|
||||
yield scope.rawDispose
|
||||
yield async () => {
|
||||
const lease = this.liveLease
|
||||
this.liveLease = undefined
|
||||
await lease?.release()
|
||||
}
|
||||
yield () => {
|
||||
this.detachSession?.()
|
||||
this.detachSession = undefined
|
||||
@@ -327,13 +315,7 @@ class AgentCreationTransaction {
|
||||
try {
|
||||
await this.scope?.dispose()
|
||||
} finally {
|
||||
try {
|
||||
const lease = this.liveLease
|
||||
this.liveLease = undefined
|
||||
await lease?.release()
|
||||
} finally {
|
||||
this.finish()
|
||||
}
|
||||
this.finish()
|
||||
}
|
||||
}
|
||||
})())
|
||||
@@ -625,15 +607,6 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
options.signal,
|
||||
)
|
||||
try {
|
||||
const claiming = persistence.claimLive(options.resumeSessionId)
|
||||
let lease: SessionLiveLease
|
||||
try {
|
||||
lease = await transaction.waitFor(claiming)
|
||||
} catch (error) {
|
||||
void claiming.then(claim => claim.release(), () => {})
|
||||
throw error
|
||||
}
|
||||
transaction.holdLiveLease(lease)
|
||||
const loaded = await transaction.waitFor(persistence.load(options.resumeSessionId))
|
||||
transaction.assertActive()
|
||||
const session = this.runtime.ctx.sessions.prepare(options.resumeSessionId, {
|
||||
|
||||
@@ -391,51 +391,6 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('owner unload during live-lease acquisition releases a late claim', async () => {
|
||||
const sessionId = SessionId('resume-claim-owner-unload')
|
||||
const root = await persistSession(sessionId)
|
||||
const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
|
||||
const claiming = Promise.withResolvers<Awaited<ReturnType<typeof ctx.sessionPersistence.claimLive>>>()
|
||||
const claimStarted = Promise.withResolvers<undefined>()
|
||||
const originalClaim = ctx.sessionPersistence.claimLive.bind(ctx.sessionPersistence)
|
||||
ctx.sessionPersistence.claimLive = (id) => {
|
||||
expect(id).toBe(sessionId)
|
||||
claimStarted.resolve(undefined)
|
||||
return claiming.promise
|
||||
}
|
||||
|
||||
let resuming!: ReturnType<typeof ctx.agents.resume>
|
||||
const owner = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
resuming = inner.agents.resume({ resumeSessionId: sessionId })
|
||||
}, { inject: ['agents'] }))
|
||||
await claimStarted.promise
|
||||
const rejection = expect(promptly(resuming)).rejects.toThrow(/owner disposed during setup/)
|
||||
await promptly(owner.dispose())
|
||||
await rejection
|
||||
|
||||
let releases = 0
|
||||
claiming.resolve({ release: () => { releases += 1; return Promise.resolve() } })
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
expect(releases).toBe(1)
|
||||
ctx.sessionPersistence.claimLive = originalClaim
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('propagates a rejected live-lease claim without loading or publishing', async () => {
|
||||
const sessionId = SessionId('resume-claim-rejected')
|
||||
const root = await persistSession(sessionId)
|
||||
const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
|
||||
let loads = 0
|
||||
ctx.sessionPersistence.claimLive = () => Promise.reject(new Error('occupied elsewhere'))
|
||||
ctx.sessionPersistence.load = () => { loads += 1; return Promise.reject(new Error('must not load')) }
|
||||
await expect(ctx.agents.resume({ resumeSessionId: sessionId }))
|
||||
.rejects.toThrow('occupied elsewhere')
|
||||
expect(loads).toBe(0)
|
||||
expect(ctx.agents.get(sessionId)).toBeUndefined()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('AgentLoop unload aborts persistence load and awaits wrapper settlement', async () => {
|
||||
const sessionId = SessionId('resume-load-factory-unload')
|
||||
const root = await persistSession(sessionId)
|
||||
|
||||
@@ -45,7 +45,7 @@ Swappable LLM, bash, filesystem, and other capability providers remain in the le
|
||||
| `ui` | owner defaults | TUI presentation settings such as reasoning, color, and card height |
|
||||
| `resumeSessionId` | — | Exact persisted session to resume |
|
||||
|
||||
Fresh runs mint a `main-session-<uuid>` session id and pass it to both the TUI and configured agent. Resumed runs bind both components to `resumeSessionId`. The TUI mounts before the spine so it can render a matching config-start failure instead of leaving a blank terminal. The app composes persistence and session query for `/resume`; an embedding host may additionally provide `tuiResumeHost` for safe in-place process handoff.
|
||||
Fresh runs mint a `main-session-<uuid>` session id and pass it to both the TUI and configured agent. Resumed runs bind both components to `resumeSessionId`. The TUI mounts before the spine so it can render a matching config-start failure instead of leaving a blank terminal. The app composes persistence and session query for `/resume`; an embedding host may additionally provide `tuiResumeHost` for in-place process handoff.
|
||||
|
||||
## The bin
|
||||
|
||||
|
||||
@@ -6,9 +6,6 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
|
||||
|
||||
```
|
||||
<root>/
|
||||
.live/
|
||||
<encoded-id>.lock # PID + nonce cross-process live lease
|
||||
<encoded-id>.lock.reclaim # ephemeral stale-owner takeover guard
|
||||
cwd-<sha256(cwd)[:12]>/ # per-project bucket (or _no-cwd/ when no cwd)
|
||||
<encoded-id>.jsonl.zstd # default: checksummed header frame + append frames
|
||||
<encoded-id>.jsonl # only with compression: 'none'
|
||||
@@ -46,7 +43,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the
|
||||
|
||||
## Write path
|
||||
|
||||
The plugin copies frozen session events into one controller per live session and starts an eager drain. Before a session can flush or resume, the coordinator claims an exclusive `.live/<encoded-id>.lock` containing the process PID and an exec-stable nonce; another live process is rejected, while a dead owner is reclaimed under the separate `.reclaim` guard. Concurrent events share the current write; events admitted during it form a follow-up batch, while `session/flush` waits until both current and pending batches are durable. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. Disposal drains every retained controller before releasing its lease.
|
||||
The plugin copies frozen session events into one controller per live session and starts an eager drain. Concurrent events share the current write; events admitted during it form a follow-up batch, while `session/flush` waits until both current and pending batches are durable. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. The owning backend instance serializes operations for one session; disposal drains every retained controller before teardown.
|
||||
|
||||
## Model Experience
|
||||
|
||||
@@ -69,6 +66,5 @@ 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 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.
|
||||
- **One live writer per session** — append and repair are coordinated only inside the owning backend instance. Another backend instance or process must not write the same session until that owner reaches quiescent disposal; initial same-id publication remains collision-safe through the POSIX no-overwrite hard link or Windows write-through rename without replacement.
|
||||
- **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.
|
||||
|
||||
@@ -14,9 +14,8 @@ import { dirname, join, resolve } from 'node:path'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import {
|
||||
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
|
||||
sessionLeaseOwnerIsLive, shareSessionLiveLease,
|
||||
type PersistenceBackend, type SessionLiveLease, type SessionLiveOwner,
|
||||
type SessionLocation, type SessionPersistenceSnapshot, type StoredPrefix,
|
||||
type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
|
||||
type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
@@ -65,11 +64,6 @@ interface JsonlTornMarker {
|
||||
recoveredEvents: SessionEvent[]
|
||||
}
|
||||
|
||||
interface JsonlLiveLeaseRecord {
|
||||
pid: number
|
||||
nonce: string
|
||||
}
|
||||
|
||||
/** Whether a filesystem error means absence; every non-ENOENT failure must surface. */
|
||||
function isENOENT(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
|
||||
@@ -141,14 +135,6 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
return this.coordinator.inspect(id)
|
||||
}
|
||||
|
||||
override claimLive(id: SessionId): Promise<SessionLiveLease> {
|
||||
return this.coordinator.claimLive(id)
|
||||
}
|
||||
|
||||
override isLive(id: SessionId): Promise<boolean> {
|
||||
return this.coordinator.isLive(id)
|
||||
}
|
||||
|
||||
// One method serves both public `list` and the backend hook; delegating it to
|
||||
// the coordinator would call this hook recursively.
|
||||
|
||||
@@ -288,110 +274,6 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
return snapshots
|
||||
}
|
||||
|
||||
/** Atomically publish one process lease, reclaiming a crashed owner's record. */
|
||||
async acquireLive(id: SessionId, owner: SessionLiveOwner): Promise<() => Promise<void>> {
|
||||
const path = this.liveLeasePath(id)
|
||||
return shareSessionLiveLease(`jsonl:${path}`, () => this.acquireLiveFile(path, id, owner))
|
||||
}
|
||||
|
||||
private async acquireLiveFile(
|
||||
path: string,
|
||||
id: SessionId,
|
||||
owner: SessionLiveOwner,
|
||||
): Promise<() => Promise<void>> {
|
||||
await mkdir(dirname(path), { recursive: true, mode: 0o700 })
|
||||
for (;;) {
|
||||
try {
|
||||
const handle = await open(path, 'wx', 0o600)
|
||||
try {
|
||||
await handle.writeFile(`${JSON.stringify(owner)}\n`, 'utf8')
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
break
|
||||
} catch (error) {
|
||||
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 || sessionLeaseOwnerIsLive(current, owner)) {
|
||||
throw new Error(`session "${id}" is occupied by another live process`)
|
||||
}
|
||||
const reclaimPath = `${path}.reclaim`
|
||||
let reclaim: Awaited<ReturnType<typeof open>>
|
||||
try {
|
||||
reclaim = await open(reclaimPath, 'wx', 0o600)
|
||||
} catch (reclaimError) {
|
||||
/* v8 ignore else -- non-contention filesystem failures are propagated verbatim and are not portable to induce */
|
||||
if ((reclaimError as NodeJS.ErrnoException).code === 'EEXIST') {
|
||||
throw new Error(`session "${id}" live-lease reclamation is already in progress`)
|
||||
}
|
||||
/* v8 ignore next -- non-contention filesystem failures are propagated verbatim and are not portable to induce */
|
||||
throw reclaimError
|
||||
}
|
||||
try {
|
||||
/* v8 ignore start -- cross-process revalidation is covered by the two-process race test */
|
||||
const latest = await this.readLiveLease(path)
|
||||
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 (sessionLeaseOwnerIsLive(latest, owner)) {
|
||||
throw new Error(`session "${id}" is occupied by another live process`)
|
||||
}
|
||||
await rm(path, { force: true })
|
||||
}
|
||||
/* v8 ignore stop */
|
||||
} finally {
|
||||
try {
|
||||
await reclaim.close()
|
||||
} finally {
|
||||
await rm(reclaimPath, { force: true })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return async () => {
|
||||
const current = await this.readLiveLease(path)
|
||||
if (current?.pid === owner.pid && current.nonce === owner.nonce) await rm(path, { force: true })
|
||||
}
|
||||
}
|
||||
|
||||
/** 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 (sessionLeaseOwnerIsLive(current, owner)) return true
|
||||
return false
|
||||
}
|
||||
|
||||
private liveLeasePath(id: SessionId): string {
|
||||
return join(this.root, '.live', `${encodeSegment(id)}.lock`)
|
||||
}
|
||||
|
||||
private async readLiveLease(path: string): Promise<JsonlLiveLeaseRecord | undefined> {
|
||||
let text: string
|
||||
try {
|
||||
text = await readFile(path, 'utf8')
|
||||
} catch (error) {
|
||||
if (isENOENT(error)) return undefined
|
||||
throw error
|
||||
}
|
||||
let value: unknown
|
||||
try {
|
||||
value = JSON.parse(text)
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
if (typeof value !== 'object' || value === null
|
||||
|| !Number.isSafeInteger((value as { pid?: unknown }).pid)
|
||||
|| (value as { pid: number }).pid <= 0
|
||||
|| typeof (value as { nonce?: unknown }).nonce !== 'string'
|
||||
|| (value as { nonce: string }).nonce.length === 0) return undefined
|
||||
return value as JsonlLiveLeaseRecord
|
||||
}
|
||||
|
||||
private async listArtifacts(): Promise<Array<{ header: SessionHeader; path: string }>> {
|
||||
await this.ensureRootEncoding()
|
||||
const artifacts: Array<{ header: SessionHeader; path: string }> = []
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
/** Child process that holds one JSONL live-session lease until it is killed. */
|
||||
|
||||
import { writeFile } from 'node:fs/promises'
|
||||
import { Context } from 'cordis'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
|
||||
const [root, marker] = process.argv.slice(2)
|
||||
if (root === undefined || marker === undefined) throw new Error('usage: live-lease-child.ts <root> <marker>')
|
||||
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
await ctx.sessionPersistence.claimLive(SessionId('leased-session'))
|
||||
await writeFile(marker, 'held')
|
||||
await new Promise<never>(() => { setInterval(() => {}, 60_000) })
|
||||
@@ -1,33 +0,0 @@
|
||||
/** Child process competing to reclaim one stale JSONL live-session lease. */
|
||||
|
||||
import { access, writeFile } from 'node:fs/promises'
|
||||
import { Context } from 'cordis'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
|
||||
const [root, gate, marker, rawId] = process.argv.slice(2)
|
||||
if (root === undefined || gate === undefined || marker === undefined || rawId === undefined) {
|
||||
throw new Error('usage: live-lease-race-child.ts <root> <gate> <marker> <session-id>')
|
||||
}
|
||||
|
||||
for (;;) {
|
||||
try {
|
||||
await access(gate)
|
||||
break
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
|
||||
await new Promise(resolve => setTimeout(resolve, 5))
|
||||
}
|
||||
}
|
||||
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
|
||||
try {
|
||||
await ctx.sessionPersistence.claimLive(SessionId(rawId))
|
||||
await writeFile(marker, 'claimed')
|
||||
await new Promise<never>(() => { setInterval(() => {}, 60_000) })
|
||||
} catch (error) {
|
||||
await writeFile(marker, `rejected:${error instanceof Error ? error.message : String(error)}`)
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
@@ -1,24 +1,17 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { spawn } from 'node:child_process'
|
||||
import { Context } from 'cordis'
|
||||
import { access, appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises'
|
||||
import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { isAbsolute, join, relative, resolve } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import { sessionLiveOwner } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { encodeSegment, eventLines, logPath, scanLog, sessionDir, toHeaderLine } from '../src/format.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
|
||||
|
||||
let root: string
|
||||
const dirs: string[] = []
|
||||
const repoRoot = fileURLToPath(new URL('../../../../', import.meta.url))
|
||||
const leaseChild = fileURLToPath(new URL('./fixtures/live-lease-child.ts', import.meta.url))
|
||||
const leaseRaceChild = fileURLToPath(new URL('./fixtures/live-lease-race-child.ts', import.meta.url))
|
||||
const tsxLoader = fileURLToPath(import.meta.resolve('tsx'))
|
||||
|
||||
type MutableSessionHeader = { -readonly [K in keyof SessionHeader]: SessionHeader[K] }
|
||||
|
||||
@@ -149,186 +142,6 @@ describe('SessionPersistenceJsonl: format helpers', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: cross-process live leases', () => {
|
||||
it('reference-counts one physical lease across backend instances in the process', async () => {
|
||||
const dir = await freshRoot()
|
||||
const contexts = [new Context(), new Context()]
|
||||
for (const ctx of contexts) {
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
|
||||
}
|
||||
try {
|
||||
const first = await contexts[0]!.sessionPersistence.claimLive(SessionId('shared-live'))
|
||||
const second = await contexts[1]!.sessionPersistence.claimLive(SessionId('shared-live'))
|
||||
await first.release()
|
||||
await expect(contexts[1]!.sessionPersistence.isLive(SessionId('shared-live'))).resolves.toBe(true)
|
||||
await second.release()
|
||||
await expect(contexts[1]!.sessionPersistence.isLive(SessionId('shared-live'))).resolves.toBe(false)
|
||||
} finally {
|
||||
await Promise.all(contexts.map(ctx => ctx.fiber.dispose()))
|
||||
}
|
||||
})
|
||||
|
||||
it('disables another live owner and reclaims its lease after the process exits', async () => {
|
||||
const dir = await freshRoot()
|
||||
const marker = join(dir, 'lease-held')
|
||||
const child = spawn(process.execPath, ['--import', tsxLoader, leaseChild, dir, marker], {
|
||||
cwd: repoRoot,
|
||||
env: { ...process.env, TSX_TSCONFIG_PATH: join(repoRoot, 'tsconfig.json') },
|
||||
stdio: ['ignore', 'ignore', 'pipe'],
|
||||
})
|
||||
let stderr = ''
|
||||
child.stderr.setEncoding('utf8')
|
||||
child.stderr.on('data', (chunk: string) => { stderr += chunk })
|
||||
try {
|
||||
await vi.waitFor(() => access(marker), { timeout: 30_000 })
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
|
||||
try {
|
||||
await expect(ctx.sessionPersistence.isLive(SessionId('leased-session'))).resolves.toBe(true)
|
||||
await expect(ctx.sessionPersistence.claimLive(SessionId('leased-session')))
|
||||
.rejects.toThrow('occupied by another live process')
|
||||
const closed = new Promise<void>(resolve => child.once('close', () => { resolve() }))
|
||||
child.kill()
|
||||
await closed
|
||||
await expect(ctx.sessionPersistence.isLive(SessionId('leased-session'))).resolves.toBe(false)
|
||||
const leasePath = join(dir, '.live', `${encodeSegment('leased-session')}.lock`)
|
||||
await writeFile(leasePath, `${JSON.stringify({ pid: child.pid, nonce: 'dead-owner' })}\n`)
|
||||
const claim = await ctx.sessionPersistence.claimLive(SessionId('leased-session'))
|
||||
await claim.release()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
} catch (error) {
|
||||
throw new Error(`live-lease child failed: ${stderr}`, { cause: error })
|
||||
} finally {
|
||||
if (child.exitCode === null && child.signalCode === null) child.kill()
|
||||
}
|
||||
}, 40_000)
|
||||
|
||||
it('fails closed on malformed lease records and surfaces lease read errors', async () => {
|
||||
const dir = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
|
||||
const liveDir = join(dir, '.live')
|
||||
await mkdir(liveDir, { recursive: true })
|
||||
try {
|
||||
const malformed = [
|
||||
'not json',
|
||||
JSON.stringify(null),
|
||||
JSON.stringify({ pid: 1.5, nonce: 'x' }),
|
||||
JSON.stringify({ pid: 0, nonce: 'x' }),
|
||||
JSON.stringify({ pid: process.pid, nonce: 1 }),
|
||||
JSON.stringify({ pid: process.pid, nonce: '' }),
|
||||
]
|
||||
for (const [index, content] of malformed.entries()) {
|
||||
const id = SessionId(`malformed-${index}`)
|
||||
const path = join(liveDir, `${encodeSegment(id)}.lock`)
|
||||
await writeFile(path, content)
|
||||
await expect(ctx.sessionPersistence.isLive(id)).resolves.toBe(true)
|
||||
await expect(ctx.sessionPersistence.claimLive(id)).rejects.toThrow('occupied by another live process')
|
||||
}
|
||||
|
||||
const unreadable = SessionId('unreadable-lease')
|
||||
await mkdir(join(liveDir, `${encodeSegment(unreadable)}.lock`))
|
||||
await expect(ctx.sessionPersistence.isLive(unreadable)).rejects.toThrow()
|
||||
|
||||
const replaced = SessionId('replaced-release')
|
||||
const claim = await ctx.sessionPersistence.claimLive(replaced)
|
||||
const replacedPath = join(liveDir, `${encodeSegment(replaced)}.lock`)
|
||||
await writeFile(replacedPath, JSON.stringify({ pid: process.pid, nonce: 'replacement' }))
|
||||
await claim.release()
|
||||
expect(await readFile(replacedPath, 'utf8')).toContain('replacement')
|
||||
|
||||
const inherited = SessionId('inherited-owner')
|
||||
const inheritedPath = join(liveDir, `${encodeSegment(inherited)}.lock`)
|
||||
await writeFile(inheritedPath, JSON.stringify(sessionLiveOwner()))
|
||||
await expect(ctx.sessionPersistence.isLive(inherited)).resolves.toBe(true)
|
||||
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()
|
||||
|
||||
const guarded = SessionId('guarded-reclaim')
|
||||
const guardedPath = join(liveDir, `${encodeSegment(guarded)}.lock`)
|
||||
await writeFile(guardedPath, JSON.stringify({ pid: 2_147_483_647, nonce: 'dead-owner' }))
|
||||
await writeFile(`${guardedPath}.reclaim`, 'busy')
|
||||
await expect(ctx.sessionPersistence.claimLive(guarded))
|
||||
.rejects.toThrow('reclamation is already in progress')
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('allows exactly one process to reclaim a stale lease', async () => {
|
||||
const dir = await freshRoot()
|
||||
const liveDir = join(dir, '.live')
|
||||
await mkdir(liveDir, { recursive: true })
|
||||
const sessionId = SessionId('reclaim-race')
|
||||
await writeFile(
|
||||
join(liveDir, `${encodeSegment(sessionId)}.lock`),
|
||||
JSON.stringify({ pid: 2_147_483_647, nonce: 'dead-owner' }),
|
||||
)
|
||||
const gate = join(dir, 'race-start')
|
||||
const markers = [join(dir, 'race-a'), join(dir, 'race-b')]
|
||||
const children = markers.map(marker => spawn(
|
||||
process.execPath,
|
||||
['--import', tsxLoader, leaseRaceChild, dir, gate, marker, sessionId],
|
||||
{
|
||||
cwd: repoRoot,
|
||||
env: { ...process.env, TSX_TSCONFIG_PATH: join(repoRoot, 'tsconfig.json') },
|
||||
stdio: ['ignore', 'ignore', 'pipe'],
|
||||
},
|
||||
))
|
||||
const errors = ['', '']
|
||||
children.forEach((child, index) => {
|
||||
child.stderr.setEncoding('utf8')
|
||||
child.stderr.on('data', (chunk: string) => { errors[index] = (errors[index] ?? '') + chunk })
|
||||
})
|
||||
try {
|
||||
await writeFile(gate, 'go')
|
||||
await vi.waitFor(() => Promise.all(markers.map(marker => access(marker))), { timeout: 30_000 })
|
||||
const outcomes = await Promise.all(markers.map(marker => readFile(marker, 'utf8')))
|
||||
expect(outcomes.filter(outcome => outcome === 'claimed')).toHaveLength(1)
|
||||
expect(outcomes.filter(outcome => outcome.startsWith('rejected:'))).toHaveLength(1)
|
||||
|
||||
const winner = children[outcomes.findIndex(outcome => outcome === 'claimed')]!
|
||||
const loser = children[outcomes.findIndex(outcome => outcome.startsWith('rejected:'))]!
|
||||
if (loser.exitCode === null && loser.signalCode === null) {
|
||||
await new Promise<void>(resolve => loser.once('close', () => { resolve() }))
|
||||
}
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
|
||||
try {
|
||||
await expect(ctx.sessionPersistence.claimLive(sessionId))
|
||||
.rejects.toThrow('occupied by another live process')
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
const closed = new Promise<void>(resolve => winner.once('close', () => { resolve() }))
|
||||
winner.kill()
|
||||
await closed
|
||||
} catch (error) {
|
||||
throw new Error(`live-lease race children failed: ${errors.join('\n')}`, { cause: error })
|
||||
} finally {
|
||||
for (const child of children) {
|
||||
if (child.exitCode === null && child.signalCode === null) child.kill()
|
||||
}
|
||||
}
|
||||
}, 40_000)
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
let ctx: Context
|
||||
beforeEach(async () => {
|
||||
|
||||
@@ -8,7 +8,7 @@ A SQLite durable session-persistence backend — a second `SessionPersistence` i
|
||||
|
||||
## Storage model
|
||||
|
||||
Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data, source_event_seqs, surface_op)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). The two `TEXT` columns `source_event_seqs` and `surface_op` are nullable; they store the event's optional surface-metadata fields (see [session surface](../../../.agents/notes/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`), a per-materialization incarnation id, and a monotonic per-log revision live in a `sessions` row; a singleton state row carries the immutable store id, and `live_session_leases` stores one PID and exec-stable nonce per live session. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`list` reports exactly the sessions that have a row).
|
||||
Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data, source_event_seqs, surface_op)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). The two `TEXT` columns `source_event_seqs` and `surface_op` are nullable; they store the event's optional surface-metadata fields (see [session surface](../../../.agents/notes/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`), a per-materialization incarnation id, and a monotonic per-log revision live in a `sessions` row; a singleton state row carries the immutable store id. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`list` reports exactly the sessions that have a row).
|
||||
|
||||
The repository's Node range supports unflagged `node:sqlite`. The database enables foreign keys and uses the configured journal mode (`wal` by default; use a rollback mode where WAL shared-memory files are unsuitable). `PRAGMA user_version` stores the table-layout version; databases with any other version are rejected because this unreleased format has no migrations.
|
||||
|
||||
@@ -33,7 +33,7 @@ interface Config {
|
||||
|
||||
## Write path
|
||||
|
||||
Like the JSONL backend, the plugin copies each frozen `session/event` into one controller per live session and starts an eager drain. A live lease is acquired in a `BEGIN IMMEDIATE` transaction before flush or resume and released after the exact lifecycle retires. Concurrent events share the current transaction; events admitted during it form a follow-up batch, while `session/flush` waits until both current and pending batches are durable. The controller persists a fork's seed once, keeps a write cursor so resume never re-appends stored events, and seeds live sessions on apply because HMR does not replay `session/created`. Dispose drains every retained controller before closing the database.
|
||||
Like the JSONL backend, the plugin copies each frozen `session/event` into one controller per live session and starts an eager drain. Concurrent events share the current transaction; events admitted during it form a follow-up batch, while `session/flush` waits until both current and pending batches are durable. The controller persists a fork's seed once, keeps a write cursor so resume never re-appends stored events, and seeds live sessions on apply because HMR does not replay `session/created`. Dispose drains every retained controller before closing the database.
|
||||
|
||||
## Model Experience
|
||||
|
||||
@@ -57,4 +57,3 @@ 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.
|
||||
|
||||
@@ -15,9 +15,8 @@ import { mkdir, open } from 'node:fs/promises'
|
||||
import { dirname, resolve } from 'node:path'
|
||||
import {
|
||||
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
|
||||
sessionLeaseOwnerIsLive, shareSessionLiveLease,
|
||||
type PersistenceBackend, type SessionLiveLease, type SessionLiveOwner,
|
||||
type SessionLocation, type SessionPersistenceSnapshot, type StoredPrefix,
|
||||
type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
|
||||
type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
@@ -162,14 +161,6 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
return this.coordinator.inspect(id)
|
||||
}
|
||||
|
||||
override claimLive(id: SessionId): Promise<SessionLiveLease> {
|
||||
return this.coordinator.claimLive(id)
|
||||
}
|
||||
|
||||
override isLive(id: SessionId): Promise<boolean> {
|
||||
return this.coordinator.isLive(id)
|
||||
}
|
||||
|
||||
// One method serves both public `list` and the backend hook; delegating it to
|
||||
// the coordinator would call this hook recursively.
|
||||
|
||||
@@ -280,55 +271,6 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
}))
|
||||
}
|
||||
|
||||
/** Atomically acquire one SQLite-backed process lease. */
|
||||
async acquireLive(id: SessionId, owner: SessionLiveOwner): Promise<() => Promise<void>> {
|
||||
await this.ready
|
||||
return shareSessionLiveLease(
|
||||
`sqlite:${this.storeIdentity}:${id}`,
|
||||
() => Promise.resolve().then(() => this.acquireLiveRow(id, owner)),
|
||||
)
|
||||
}
|
||||
|
||||
private acquireLiveRow(id: SessionId, owner: SessionLiveOwner): () => Promise<void> {
|
||||
this.db.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
const current = this.liveLeaseFor(id)
|
||||
if (current !== undefined
|
||||
&& (current.pid !== owner.pid || current.nonce !== owner.nonce)) {
|
||||
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)
|
||||
}
|
||||
this.db.prepare(`
|
||||
INSERT INTO live_session_leases (session_id, pid, nonce) VALUES (?, ?, ?)
|
||||
ON CONFLICT(session_id) DO UPDATE SET pid = excluded.pid, nonce = excluded.nonce
|
||||
`).run(id, owner.pid, owner.nonce)
|
||||
this.db.exec('COMMIT')
|
||||
} catch (error) {
|
||||
this.db.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
return async () => {
|
||||
await this.ready
|
||||
this.db.prepare(
|
||||
'DELETE FROM live_session_leases WHERE session_id = ? AND pid = ? AND nonce = ?',
|
||||
).run(id, owner.pid, owner.nonce)
|
||||
}
|
||||
}
|
||||
|
||||
/** Report a non-stale SQLite lease and remove a crashed owner's row. */
|
||||
async inspectLive(id: SessionId, owner: SessionLiveOwner): Promise<boolean> {
|
||||
await this.ready
|
||||
const current = this.liveLeaseFor(id)
|
||||
if (current === undefined) return false
|
||||
if ((current.pid === owner.pid && current.nonce === owner.nonce)
|
||||
|| 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
|
||||
}
|
||||
|
||||
/** Close the database handle (awaited by the coordinator's dispose, post-drain). */
|
||||
async close(): Promise<void> {
|
||||
await this.ready
|
||||
@@ -342,11 +284,6 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
return this.db.prepare('SELECT * FROM sessions WHERE id = ?').get(id) as unknown as SessionRow | undefined
|
||||
}
|
||||
|
||||
private liveLeaseFor(id: SessionId): { pid: number; nonce: string } | undefined {
|
||||
return this.db.prepare('SELECT pid, nonce FROM live_session_leases WHERE session_id = ?')
|
||||
.get(id) as { pid: number; nonce: string } | undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert-or-replace a session's metadata row. The only caller is the first
|
||||
* materializing `appendBatch`, so writing the row IS the materialization (its
|
||||
|
||||
@@ -17,7 +17,7 @@ import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepsee
|
||||
* layout; orthogonal to a session's own `version` (which versions the EVENT
|
||||
* vocabulary, stored per session in the `sessions` row).
|
||||
*/
|
||||
export const SCHEMA_VERSION = 9
|
||||
export const SCHEMA_VERSION = 8
|
||||
|
||||
/**
|
||||
* A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
|
||||
@@ -68,7 +68,7 @@ export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
|
||||
* rather than being migrated in place.
|
||||
* @param path - the SQLite database file to open (created when absent).
|
||||
* @param journalMode - validated journal pragma.
|
||||
* @returns the open handle with pragmas applied and all tables ensured.
|
||||
* @returns the open handle with pragmas applied and all three tables ensured.
|
||||
*/
|
||||
export function openDatabase(path: string, journalMode: JournalMode): DatabaseSync {
|
||||
const db = new DatabaseSync(path)
|
||||
@@ -128,13 +128,6 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
|
||||
PRIMARY KEY (session_id, seq)
|
||||
) STRICT
|
||||
`)
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS live_session_leases (
|
||||
session_id TEXT PRIMARY KEY,
|
||||
pid INTEGER NOT NULL,
|
||||
nonce TEXT NOT NULL
|
||||
) STRICT
|
||||
`)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { chmod, mkdtemp, rm, stat, symlink, writeFile } from 'node:fs/promises'
|
||||
@@ -7,16 +7,12 @@ import { dirname, join } from 'node:path'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SurfaceEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceSqlite, { SCHEMA_VERSION } from '@deepseek-ai/dsh-session-persistence-sqlite'
|
||||
import { sessionLiveOwner } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { openDatabase, rowToEvent, scanRows, type EventRow } from '../src/schema.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
|
||||
|
||||
const dirs: string[] = []
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks()
|
||||
for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
|
||||
})
|
||||
afterEach(async () => { 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 {
|
||||
@@ -446,7 +442,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
})
|
||||
|
||||
it('exposes the schema version constant', () => {
|
||||
expect(SCHEMA_VERSION).toBe(9)
|
||||
expect(SCHEMA_VERSION).toBe(8)
|
||||
})
|
||||
|
||||
it('keeps the revision stable for an empty repair hook', async () => {
|
||||
@@ -462,47 +458,6 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
})
|
||||
|
||||
describe('SessionPersistenceSqlite: edge cases', () => {
|
||||
it('claims, rejects, reclaims, inspects, and releases SQLite live leases', async () => {
|
||||
const path = await freshDbPath()
|
||||
const b = await backend(path)
|
||||
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', 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)
|
||||
db.close()
|
||||
|
||||
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()
|
||||
await memory.plugin(SessionStore)
|
||||
await memory.plugin(SessionPersistenceSqlite, { path: ':memory:' })
|
||||
const memoryClaim = await memory.sessionPersistence.claimLive(SessionId('memory-live'))
|
||||
expect(await memory.sessionPersistence.isLive(SessionId('memory-live'))).toBe(true)
|
||||
await memoryClaim.release()
|
||||
await memory.fiber.dispose()
|
||||
})
|
||||
|
||||
it('rejects and closes a current-schema database with an invalid store identity', async () => {
|
||||
const path = await freshDbPath()
|
||||
const db = openDatabase(path, 'wal')
|
||||
|
||||
@@ -15,10 +15,6 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
|
||||
| `inspect(id): Promise<{ meta; events }>` | Return a detached valid stored prefix without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; intended for read models and other observers that must never recover a log. |
|
||||
| `list(): Promise<SessionHeader[]>` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. |
|
||||
| `listSnapshots(): Promise<SessionPersistenceSnapshot[]>` | Lightweight metadata plus an opaque branded per-log revision, without loading event logs. A revision stays equal while that log and its backing store are unchanged, changes after append or mutating load repair, and cannot collide solely because two stores use the same local counter. |
|
||||
| `claimLive(id): Promise<SessionLiveLease>` | Atomically claim live ownership. First-party backends reject another live process and reclaim a dead owner; release follows quiescence. |
|
||||
| `isLive(id): Promise<boolean>` | Report a current non-stale live lease, including one owned by this process. |
|
||||
|
||||
The abstract base supplies a process-local fallback for lightweight third-party implementations. A backend that needs multi-process safety overrides both live-lease methods.
|
||||
|
||||
## Invariants every backend must honor
|
||||
|
||||
@@ -35,7 +31,7 @@ Each `session/event` copies its event into the session controller and starts an
|
||||
|
||||
Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it with the coordinator's stored header only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. A cold load reserves its id across backend reads and repair writes, so concurrent publication of a same-id live `Session` rejects and rolls back. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn.
|
||||
|
||||
When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state and the backend-owned live lease for that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, releases their leases, awaits per-id operations, and only then closes the storage handle.
|
||||
When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle.
|
||||
|
||||
The side-effect-free `locate` and lightweight `listSnapshots` queries remain backend-owned because they describe storage topology and revision identity rather than write orchestration.
|
||||
|
||||
@@ -48,8 +44,6 @@ The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinato
|
||||
| `appendBatch(meta, events, isMaterialized)` | Durably append a contiguous batch, lazily materializing ATOMICALLY when not yet materialized. |
|
||||
| `commitRepair(meta, tornMarker, closers)` | Make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined` — a marker may be falsy, e.g. seq/offset `0`) and append `closers`. NOT required to be atomic. Used by load (truncate + closers) and live-adoption (truncate only). |
|
||||
| `list()` | List all stored metadata. |
|
||||
| `acquireLive?(id, owner)` | Atomically acquire a backend-owned cross-process lease and return its physical release. |
|
||||
| `inspectLive?(id, owner)` | Report or reclaim a backend-owned lease without acquiring it. |
|
||||
| `close?()` | Optional lifecycle teardown (e.g. close a db handle), awaited after the dispose drain. |
|
||||
|
||||
The coordinator asserts the stored id and compares stored/live cwd before repair or live adoption. Its `inspect()` path validates and clones the prefix without calling `commitRepair` or publishing write state. The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). A third-party backend MAY implement the abstract service directly without the coordinator, but it must provide the same non-mutating inspection and trustworthy lightweight snapshot revisions. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
|
||||
@@ -85,4 +79,3 @@ 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.
|
||||
|
||||
@@ -8,8 +8,6 @@
|
||||
import { Context } from 'cordis'
|
||||
import { interruptedTurnClosers, SESSION_FORMAT_VERSION, snapshotJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import { sessionLiveOwner } from './lease.ts'
|
||||
import type { SessionLiveLease, SessionLiveOwner } from './lease.ts'
|
||||
|
||||
/**
|
||||
* A stored session's header, valid contiguous event prefix, and optional opaque
|
||||
@@ -65,12 +63,6 @@ export interface PersistenceBackend<TornMarker = unknown> {
|
||||
/** List all stored (materialized) sessions' metadata. */
|
||||
list(): Promise<SessionHeader[]>
|
||||
|
||||
/** Optionally acquire a backend-owned cross-process live-session lease. */
|
||||
acquireLive?(id: SessionId, owner: SessionLiveOwner): Promise<() => Promise<void>>
|
||||
|
||||
/** Optionally inspect and reclaim a backend-owned live-session lease. */
|
||||
inspectLive?(id: SessionId, owner: SessionLiveOwner): Promise<boolean>
|
||||
|
||||
/**
|
||||
* Optional lifecycle teardown (e.g. close a database handle). Awaited by the
|
||||
* coordinator's dispose effect AFTER the quiescence drain. A stateless file
|
||||
@@ -104,7 +96,6 @@ interface LiveSessionState {
|
||||
pending: SessionEvent[]
|
||||
init: Promise<void>
|
||||
flush: Promise<void> | undefined
|
||||
lease?: SessionLiveLease
|
||||
}
|
||||
|
||||
/** Collect the rejection reasons from a set of promises (none-throwing). */
|
||||
@@ -170,12 +161,6 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
* same id, so writes for one session never interleave. Keyed by session id.
|
||||
*/
|
||||
private chains = new Map<SessionId, Promise<unknown>>()
|
||||
/** One backend lease with process-local reference counting per session id. */
|
||||
private liveClaims = new Map<SessionId, {
|
||||
refs: number
|
||||
releaseBackend: () => Promise<void>
|
||||
}>()
|
||||
private readonly liveOwner = sessionLiveOwner()
|
||||
|
||||
constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
|
||||
this.installWritePath()
|
||||
@@ -288,61 +273,6 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
return this.serialize(id, () => this.inspectCore(id))
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquire one process-local reference to the backend's cross-process lease.
|
||||
* @param id - session identity about to become live.
|
||||
* @returns one idempotent release capability.
|
||||
*/
|
||||
async claimLive(id: SessionId): Promise<SessionLiveLease> {
|
||||
const acquireLive = this.backend.acquireLive?.bind(this.backend)
|
||||
if (acquireLive === undefined) return { release: () => Promise.resolve() }
|
||||
await this.serialize(id, async () => {
|
||||
const existing = this.liveClaims.get(id)
|
||||
if (existing !== undefined) {
|
||||
existing.refs += 1
|
||||
return
|
||||
}
|
||||
const releaseBackend = await acquireLive(id, this.liveOwner)
|
||||
this.liveClaims.set(id, { refs: 1, releaseBackend })
|
||||
})
|
||||
let releaseTask: Promise<void> | undefined
|
||||
return {
|
||||
release: () => {
|
||||
if (releaseTask !== undefined) return releaseTask
|
||||
const task = this.serialize(id, async () => {
|
||||
const claim = this.liveClaims.get(id)
|
||||
/* v8 ignore next -- this capability is returned only after its claim enters the serialized map */
|
||||
if (claim === undefined) return
|
||||
claim.refs -= 1
|
||||
if (claim.refs > 0) return
|
||||
try {
|
||||
await claim.releaseBackend()
|
||||
} catch (error) {
|
||||
claim.refs += 1
|
||||
throw error
|
||||
}
|
||||
this.liveClaims.delete(id)
|
||||
})
|
||||
const wrapped = task.catch((error: unknown) => {
|
||||
releaseTask = undefined
|
||||
throw error
|
||||
})
|
||||
releaseTask = wrapped
|
||||
return wrapped
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check the backend's current cross-process lease state.
|
||||
* @param id - session identity to inspect.
|
||||
* @returns whether this or another live process owns the session.
|
||||
*/
|
||||
isLive(id: SessionId): Promise<boolean> {
|
||||
if (this.liveClaims.has(id)) return Promise.resolve(true)
|
||||
return this.backend.inspectLive?.(id, this.liveOwner) ?? Promise.resolve(false)
|
||||
}
|
||||
|
||||
private async inspectCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
|
||||
const stored = await this.backend.loadStored(id)
|
||||
if (stored === undefined) throw new Error(`session "${id}" not found`)
|
||||
@@ -452,9 +382,6 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
let disposeError: unknown
|
||||
try {
|
||||
const errors = await settledErrors([...this.live.keys()].map(session => this.flush(session)))
|
||||
errors.push(...await settledErrors(
|
||||
[...this.live.values()].flatMap(live => live.lease === undefined ? [] : [live.lease.release()]),
|
||||
))
|
||||
while (this.chains.size > 0) await Promise.allSettled([...this.chains.values()])
|
||||
if (errors.length > 0) {
|
||||
throw new AggregateError(errors, `${this.backend.name} dispose failed`)
|
||||
@@ -514,8 +441,6 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
private async retireCore(session: Session): Promise<void> {
|
||||
await this.flush(session)
|
||||
const id = session.header.id
|
||||
const live = this.live.get(session)
|
||||
await live?.lease?.release()
|
||||
await this.serialize(id, () => {
|
||||
this.live.delete(session)
|
||||
if (this.states.get(id)?.owner === session) this.states.delete(id)
|
||||
@@ -529,16 +454,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
const seed = session.events.map(e => structuredClone(e))
|
||||
const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined }
|
||||
this.live.set(session, live)
|
||||
live.init = this.claimLive(session.id).then(async (lease) => {
|
||||
live.lease = lease
|
||||
try {
|
||||
await this.serialize(session.header.id, () => this.onCreated(session, seed))
|
||||
} catch (error) {
|
||||
delete live.lease
|
||||
await lease.release()
|
||||
throw error
|
||||
}
|
||||
})
|
||||
live.init = this.serialize(session.header.id, () => this.onCreated(session, seed))
|
||||
live.init.catch(() => { /* observed by flush/dispose through the controller */ })
|
||||
return live
|
||||
}
|
||||
|
||||
@@ -8,18 +8,10 @@
|
||||
import { Context, Service } from 'cordis'
|
||||
import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionPersistenceRevision } from './revision.ts'
|
||||
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 {
|
||||
sessionLeaseOwnerIsLive,
|
||||
sessionLeaseProcessIsLive,
|
||||
sessionLiveOwner,
|
||||
shareSessionLiveLease,
|
||||
} from './lease.ts'
|
||||
export type { SessionLiveLease, SessionLiveOwner } from './lease.ts'
|
||||
|
||||
/** Lightweight immutable source identity returned without loading a full log. */
|
||||
export interface SessionPersistenceSnapshot {
|
||||
@@ -58,8 +50,6 @@ export interface SessionLocation {
|
||||
* rewriting committed events.
|
||||
*/
|
||||
export abstract class SessionPersistence extends Service {
|
||||
private readonly localLiveClaims = new Map<SessionId, number>()
|
||||
|
||||
constructor(ctx: Context) {
|
||||
super(ctx, 'sessionPersistence')
|
||||
}
|
||||
@@ -133,39 +123,6 @@ export abstract class SessionPersistence extends Service {
|
||||
* @returns one header and opaque revision per materialized session without loading full logs.
|
||||
*/
|
||||
abstract listSnapshots(): Promise<SessionPersistenceSnapshot[]>
|
||||
|
||||
/**
|
||||
* Atomically acquire this process's live ownership of a session id.
|
||||
* Reentrant claims share one backend lease. First-party backends override
|
||||
* this process-local fallback to reject another live process and reclaim a
|
||||
* dead owner.
|
||||
* @param id - session identity that is about to become live.
|
||||
* @returns a single-release reference owned by the caller.
|
||||
*/
|
||||
claimLive(id: SessionId): Promise<SessionLiveLease> {
|
||||
this.localLiveClaims.set(id, (this.localLiveClaims.get(id) ?? 0) + 1)
|
||||
let released = false
|
||||
return Promise.resolve({
|
||||
release: () => {
|
||||
if (released) return Promise.resolve()
|
||||
released = true
|
||||
const refs = this.localLiveClaims.get(id) as number
|
||||
if (refs <= 1) this.localLiveClaims.delete(id)
|
||||
else this.localLiveClaims.set(id, refs - 1)
|
||||
return Promise.resolve()
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Check whether any process currently owns a live lease for this session.
|
||||
* The base implementation reports only claims on this service instance.
|
||||
* @param id - persisted or prospective session identity.
|
||||
* @returns true while a non-stale lease exists, including this process's lease.
|
||||
*/
|
||||
isLive(id: SessionId): Promise<boolean> {
|
||||
return Promise.resolve(this.localLiveClaims.has(id))
|
||||
}
|
||||
}
|
||||
|
||||
export default SessionPersistence
|
||||
|
||||
@@ -1,123 +0,0 @@
|
||||
/** Process-backed identity helpers for cross-process live-session leases. */
|
||||
|
||||
import { randomUUID } from 'node:crypto'
|
||||
|
||||
const LIVE_OWNER_ENV = 'DSH_SESSION_LIVE_OWNER'
|
||||
|
||||
/** Process identity stored in backend-owned cross-process live-session leases. */
|
||||
export interface SessionLiveOwner {
|
||||
/** Operating-system process id; retained across an `execve` handoff. */
|
||||
readonly pid: number
|
||||
/** Exec-stable process-start nonce used when the observer has the same PID. */
|
||||
readonly nonce: string
|
||||
}
|
||||
|
||||
/** Idempotent capability releasing one acquired live-session lease reference. */
|
||||
export interface SessionLiveLease {
|
||||
/** Release this caller's lease reference after its live session reaches quiescence. */
|
||||
release(): Promise<void>
|
||||
}
|
||||
|
||||
/**
|
||||
* Stable owner inherited only by an exec-replaced process, not inferred from a session id.
|
||||
* @returns this process's PID and exec-stable nonce.
|
||||
*/
|
||||
export function sessionLiveOwner(): SessionLiveOwner {
|
||||
const nonce = process.env[LIVE_OWNER_ENV] ?? randomUUID()
|
||||
process.env[LIVE_OWNER_ENV] = nonce
|
||||
return { pid: process.pid, nonce }
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a lease pid still names a process; permission denial counts as live.
|
||||
* @param pid - positive operating-system process id from a lease record.
|
||||
* @returns true unless the operating system reports that the process is absent.
|
||||
*/
|
||||
export function sessionLeaseProcessIsLive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch (error) {
|
||||
return (error as NodeJS.ErrnoException).code !== 'ESRCH'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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>()
|
||||
|
||||
/**
|
||||
* Reference-count one physical lease across backend instances in this process.
|
||||
* @param key - backend-kind plus canonical storage location and session id.
|
||||
* @param acquire - single physical acquisition performed for the first reference.
|
||||
* @returns an idempotent release for this caller's reference.
|
||||
*/
|
||||
export async function shareSessionLiveLease(
|
||||
key: string,
|
||||
acquire: () => Promise<() => Promise<void>>,
|
||||
): Promise<() => Promise<void>> {
|
||||
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
|
||||
throw error
|
||||
}
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,92 +0,0 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import {
|
||||
sessionLeaseOwnerIsLive,
|
||||
sessionLeaseProcessIsLive,
|
||||
sessionLiveOwner,
|
||||
shareSessionLiveLease,
|
||||
} from '../src/lease.ts'
|
||||
|
||||
const originalOwner = process.env.DSH_SESSION_LIVE_OWNER
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
if (originalOwner === undefined) delete process.env.DSH_SESSION_LIVE_OWNER
|
||||
else process.env.DSH_SESSION_LIVE_OWNER = originalOwner
|
||||
})
|
||||
|
||||
describe('process live-session lease helpers', () => {
|
||||
it('creates one exec-stable owner identity and classifies process liveness', () => {
|
||||
delete process.env.DSH_SESSION_LIVE_OWNER
|
||||
const first = sessionLiveOwner()
|
||||
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 })
|
||||
expect(sessionLeaseProcessIsLive(999_998)).toBe(true)
|
||||
})
|
||||
|
||||
it('shares one physical lease until every process-local reference releases', async () => {
|
||||
const releasePhysical = vi.fn<() => Promise<void>>(() => Promise.resolve())
|
||||
const acquire = vi.fn<() => Promise<() => Promise<void>>>(() => Promise.resolve(releasePhysical))
|
||||
const key = `shared-${randomUUID()}`
|
||||
const first = await shareSessionLiveLease(key, acquire)
|
||||
const second = await shareSessionLiveLease(key, acquire)
|
||||
expect(acquire).toHaveBeenCalledTimes(1)
|
||||
await first()
|
||||
expect(releasePhysical).not.toHaveBeenCalled()
|
||||
await second()
|
||||
await second()
|
||||
expect(releasePhysical).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('removes failed acquisitions and retries a failed physical release', async () => {
|
||||
const key = `retry-${randomUUID()}`
|
||||
await expect(shareSessionLiveLease(key, () => Promise.reject(new Error('claim failed'))))
|
||||
.rejects.toThrow('claim failed')
|
||||
|
||||
let releases = 0
|
||||
const release = await shareSessionLiveLease(key, () => Promise.resolve(async () => {
|
||||
releases += 1
|
||||
if (releases === 1) throw new Error('release failed')
|
||||
}))
|
||||
await expect(release()).rejects.toThrow('release failed')
|
||||
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)
|
||||
})
|
||||
})
|
||||
@@ -4,7 +4,7 @@ import SessionStore, { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
|
||||
type PersistenceBackend, type SessionLiveOwner, type SessionPersistenceSnapshot, type StoredPrefix,
|
||||
type PersistenceBackend, type SessionPersistenceSnapshot, type StoredPrefix,
|
||||
} from '../src/index.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from './coordinator-contract.ts'
|
||||
@@ -348,46 +348,6 @@ describe('PersistenceCoordinator stored identity', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('PersistenceCoordinator live leases', () => {
|
||||
it('degrades without backend hooks and retries a failed final release', async () => {
|
||||
const fallbackCtx = new Context()
|
||||
await fallbackCtx.plugin(SessionStore)
|
||||
const fallback = new PersistenceCoordinator(fallbackCtx, new ControlledBackend())
|
||||
const fallbackClaim = await fallback.claimLive(SessionId('fallback-live'))
|
||||
expect(await fallback.isLive(SessionId('fallback-live'))).toBe(false)
|
||||
await fallbackClaim.release()
|
||||
await fallbackCtx.fiber.dispose()
|
||||
|
||||
class LeaseBackend extends ControlledBackend {
|
||||
releaseAttempts = 0
|
||||
async acquireLive(_id: SessionId, _owner: SessionLiveOwner): Promise<() => Promise<void>> {
|
||||
return async () => {
|
||||
this.releaseAttempts += 1
|
||||
if (this.releaseAttempts === 1) throw new Error('lease release failed')
|
||||
}
|
||||
}
|
||||
inspectLive(): Promise<boolean> {
|
||||
return Promise.resolve(true)
|
||||
}
|
||||
}
|
||||
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const backend = new LeaseBackend()
|
||||
const coordinator = new PersistenceCoordinator(ctx, backend)
|
||||
const first = await coordinator.claimLive(SessionId('leased'))
|
||||
const second = await coordinator.claimLive(SessionId('leased'))
|
||||
expect(await coordinator.isLive(SessionId('leased'))).toBe(true)
|
||||
await first.release()
|
||||
await expect(second.release()).rejects.toThrow('lease release failed')
|
||||
await expect(second.release()).resolves.toBeUndefined()
|
||||
await expect(second.release()).resolves.toBeUndefined()
|
||||
expect(backend.releaseAttempts).toBe(2)
|
||||
expect(await coordinator.isLive(SessionId('leased'))).toBe(true)
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
describe('PersistenceCoordinator retirement', () => {
|
||||
it('a retiring unmaterialized owner without buffered events releases its id', async () => {
|
||||
const ctx = new Context()
|
||||
@@ -835,20 +795,4 @@ describe('SessionPersistence service registration', () => {
|
||||
await fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('provides a reference-counted process-local lease fallback', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(MemoryPersistence)
|
||||
const id = SessionId('local-live')
|
||||
const first = await ctx.sessionPersistence.claimLive(id)
|
||||
const second = await ctx.sessionPersistence.claimLive(id)
|
||||
expect(await ctx.sessionPersistence.isLive(id)).toBe(true)
|
||||
await first.release()
|
||||
await first.release()
|
||||
expect(await ctx.sessionPersistence.isLive(id)).toBe(true)
|
||||
await second.release()
|
||||
expect(await ctx.sessionPersistence.isLive(id)).toBe(false)
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -30,7 +30,7 @@ The footer sums the session's reported usage as `↑<uncached input> ↓<output>
|
||||
|
||||
`/status` adds a point-in-time diagnostics card to the transcript and remains available while the agent runs. It reports the session id, title, working directory, selected provider/model, reasoning-block visibility, agent state, event/turn/step/tool-call counts, exact input/output/cache token buckets, KV-cache hit rate, token-meter context use and capacity, creation time, and latest event time. Missing titles, models, cache input, or context capacity are labeled instead of inferred. The card is terminal-only and does not duplicate the compact footer.
|
||||
|
||||
`/resume` opens a keyboard selector over the current workspace. Candidates are sorted by last logged activity and searchable by log-backed title or session id; each row reports current/live/persisted state, last turn outcome, recent provider/model, and durable goal phase when present. The current session, another live owner's session, an unreadable log, a mismatched cwd, or a session whose logged provider has no current adapter remains visible but disabled. Selection repeats those checks, requires the current agent to be idle, and claims the target live lease before flushing the current session; a lost claim race or later recoverable failure leaves the current TUI running and releases any acquired reservation. The TUI then stops the terminal UI and calls the optional host-owned `TuiRuntime.handoffResume`; where `process.execve` is available, the shipped `dsh` host disposes the app and atomically replaces its process while retaining the reservation, so two runtimes never own the terminal together. Resume restores the same `SessionId`, transcript, title, todos, and durable goal; goal activation remains disarmed and the TUI asks for human confirmation or `/goal resume`.
|
||||
`/resume` opens a keyboard selector over the current workspace. Candidates are sorted by last logged activity and searchable by log-backed title or session id; each row reports current/live/persisted state, last turn outcome, recent provider/model, and durable goal phase when present. The current session, a session already live in this runtime, an unreadable log, a mismatched cwd, or a session whose logged provider has no current adapter remains visible but disabled. Selection repeats those checks and requires the current agent to be idle before flushing the current session. The TUI then stops the terminal UI and calls the optional host-owned `TuiRuntime.handoffResume`; where `process.execve` is available, the shipped `dsh` host disposes the app and replaces its process. Resume restores the same `SessionId`, transcript, title, todos, and durable goal; goal activation remains disarmed and the TUI asks for human confirmation or `/goal resume`.
|
||||
|
||||
`resumeCommand` remains the deployment-owned fallback: exiting prints it only after the current session is durable, and a host without in-place handoff shows the selected session's command. `{session}` expands to the session id. TUI code never executes the template or arbitrary shell text.
|
||||
|
||||
@@ -156,6 +156,7 @@ Append-only; newly visible content follows the reusable request prefix and does
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Resume has no cross-process session lock** — the selector rejects sessions known to be live in its own runtime, but another process can resume the same persisted id before or during handoff. Deployments that can run concurrent hosts must coordinate ownership outside the TUI.
|
||||
- **One configured session owns the transcript and editor** — questions from other agents can still use the shared overlay provider, but session rendering and prompt input remain bound to `sessionId`.
|
||||
- **Tool cards are text terminal presentations** — terminal, diff, and generic cards use tool-owned titles/content, but session content currently has no image block for inline image rendering.
|
||||
- **Non-TTY operation is intentionally unsupported** — app bundles that need automation must compose a one-shot or server front door (`dsh-cli-demo`, `dsh-acp`) rather than expecting an internal fallback.
|
||||
|
||||
@@ -79,7 +79,7 @@ import type {
|
||||
} from '@deepseek-ai/dsh-session-query'
|
||||
// Type import also declaration-merges the optional `sessionPersistence`
|
||||
// service onto `Context` so `ctx.get('sessionPersistence')` is typed.
|
||||
import type { SessionLiveLease } from '@deepseek-ai/dsh-session-persistence'
|
||||
import type {} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SkillDefinition, SkillResourceBase, SkillService } from '@deepseek-ai/dsh-skill'
|
||||
import type {
|
||||
FileDiff,
|
||||
@@ -349,7 +349,7 @@ export interface TuiRuntime {
|
||||
formatCwd?: (cwd: string | undefined) => string
|
||||
/** Monotonic-enough wall clock for elapsed status rendering. Defaults to `Date.now`. */
|
||||
now?(): number
|
||||
/** Host-owned safe process handoff; absent leaves `resumeCommand` as the fallback. */
|
||||
/** Host-owned process handoff; absent leaves `resumeCommand` as the fallback. */
|
||||
handoffResume?: TuiResumeHost['handoff']
|
||||
}
|
||||
|
||||
@@ -1283,7 +1283,6 @@ interface ResumeRoute {
|
||||
|
||||
interface ResumeCandidate {
|
||||
record: SessionRecord
|
||||
occupied: boolean
|
||||
title: string
|
||||
lastActivityAt: number
|
||||
lastTurn: string
|
||||
@@ -1324,7 +1323,6 @@ function summarizeResumeCandidate(
|
||||
snapshot: SessionLogSnapshot,
|
||||
currentId: SessionId,
|
||||
cwd: string | undefined,
|
||||
occupied: boolean,
|
||||
availableProviders: ReadonlySet<string>,
|
||||
): ResumeCandidate {
|
||||
const title = foldSessionTitle(snapshot.events)?.title ?? 'Untitled session'
|
||||
@@ -1332,14 +1330,13 @@ function summarizeResumeCandidate(
|
||||
const foldedGoal = foldGoal(snapshot.events).goal
|
||||
let disabledReason: string | undefined
|
||||
if (record.header.id === currentId) disabledReason = 'current session'
|
||||
else if (record.live || occupied) disabledReason = 'occupied by another live agent'
|
||||
else if (record.live) disabledReason = 'session is already live in this runtime'
|
||||
else if (record.header.cwd !== cwd) disabledReason = 'different workspace'
|
||||
else if (route !== undefined && !availableProviders.has(route.provider)) {
|
||||
disabledReason = `session is complete, but route is currently unavailable (${route.provider}/${route.model})`
|
||||
}
|
||||
return {
|
||||
record,
|
||||
occupied,
|
||||
title,
|
||||
lastActivityAt: snapshot.events.at(-1)?.time ?? snapshot.session.createdAt,
|
||||
lastTurn: resumeTurnLabel(snapshot),
|
||||
@@ -1422,7 +1419,7 @@ class ResumeDialog implements Component, Focusable {
|
||||
const selected = index === this.selectedIndex
|
||||
const status = [
|
||||
candidate.disabledReason === 'current session' ? 'current' : undefined,
|
||||
candidate.record.live || candidate.occupied ? 'live' : undefined,
|
||||
candidate.record.live ? 'live' : undefined,
|
||||
candidate.record.persisted ? 'persisted' : undefined,
|
||||
].filter((value): value is string => value !== undefined).join(' · ')
|
||||
const lead = `${selected ? '›' : ' '} ${displayText(candidate.title)}`
|
||||
@@ -1841,8 +1838,6 @@ export function createTuiChat(
|
||||
let modelOverlay: TuiOverlaySession | undefined
|
||||
let resumeOverlay: TuiOverlaySession | undefined
|
||||
let resumeInFlight = false
|
||||
let resumeReservation: SessionLiveLease | undefined
|
||||
let resumeReservationCommitted = false
|
||||
let resumeScan = 0
|
||||
let tuiServiceFiber: Fiber | undefined
|
||||
const target: AgentLlmTargetRef = { current: initialTarget(agent), assembled: undefined }
|
||||
@@ -1855,12 +1850,6 @@ export function createTuiChat(
|
||||
const now = (): number => runtime.now?.() ?? Date.now()
|
||||
const agentStatus = (): AgentStatus => agent.status
|
||||
const isDisposed = (): boolean => disposed
|
||||
const releaseResumeReservation = async (): Promise<void> => {
|
||||
const reservation = resumeReservation
|
||||
if (reservation === undefined) return
|
||||
await reservation.release()
|
||||
resumeReservation = undefined
|
||||
}
|
||||
|
||||
// A configured subtitle renders as a banner line; when absent, the banner has
|
||||
// no subtitle. The banner itself sweeps in on start (see startBannerReveal).
|
||||
@@ -2444,8 +2433,6 @@ export function createTuiChat(
|
||||
shuttingDown ??= (async () => {
|
||||
disposed = true
|
||||
overlayManager.beginShutdown()
|
||||
/* v8 ignore else -- the committed branch is the non-returning exec handoff covered by the keyless PTY test */
|
||||
if (!resumeReservationCommitted) await releaseResumeReservation()
|
||||
contextResolution = undefined
|
||||
clearStatus()
|
||||
for (const controller of commandControllers) controller.abort(new Error('TUI disposed'))
|
||||
@@ -2825,9 +2812,6 @@ export function createTuiChat(
|
||||
providers: ReadonlySet<string>,
|
||||
): Promise<ResumeCandidate> => {
|
||||
try {
|
||||
const occupied = record.live || (record.persisted && persistence !== undefined
|
||||
? await persistence.isLive(record.header.id)
|
||||
: false)
|
||||
let snapshot: SessionLogSnapshot
|
||||
const live = ctx.sessions.get(record.header.id)
|
||||
if (live !== undefined) {
|
||||
@@ -2845,13 +2829,11 @@ export function createTuiChat(
|
||||
snapshot,
|
||||
agent.session.id,
|
||||
agent.session.header.cwd,
|
||||
occupied,
|
||||
providers,
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
return {
|
||||
record,
|
||||
occupied: record.live,
|
||||
title: 'Unreadable session',
|
||||
lastActivityAt: record.header.createdAt,
|
||||
lastTurn: 'log unavailable',
|
||||
@@ -2895,14 +2877,8 @@ export function createTuiChat(
|
||||
: `This host cannot hand off in place. Exit and run: ${fallback}`, 'warning')
|
||||
return
|
||||
}
|
||||
if (persistence === undefined) {
|
||||
throw new Error('Resume is unavailable: session persistence is not mounted.')
|
||||
}
|
||||
resumeReservation = await persistence.claimLive(checked.record.header.id)
|
||||
if (disposed) {
|
||||
await releaseResumeReservation()
|
||||
return
|
||||
}
|
||||
/* v8 ignore next -- shutdown during preflight invalidates an awaited service read or reaches this guard */
|
||||
if (disposed) return
|
||||
await ctx.sessions.flush(agent.session)
|
||||
// Disposal can run while the flush promise is pending; TypeScript does not model that reentry.
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
|
||||
@@ -2916,29 +2892,18 @@ export function createTuiChat(
|
||||
if (disposed) return
|
||||
ui.stop()
|
||||
terminalReleased = true
|
||||
resumeReservationCommitted = true
|
||||
await hostHandoff(checked.record.header.id)
|
||||
throw new Error('resume host returned without replacing the process')
|
||||
} catch (error: unknown) {
|
||||
/* v8 ignore next -- a committed host disposes this TUI and never returns; recoverable rejection keeps it live */
|
||||
if (!disposed) {
|
||||
resumeReservationCommitted = false
|
||||
let reported = error
|
||||
try {
|
||||
await releaseResumeReservation()
|
||||
} catch (releaseError: unknown) {
|
||||
reported = new Error(
|
||||
`${errorChain(error)}; target reservation release failed: ${errorChain(releaseError)}`,
|
||||
)
|
||||
}
|
||||
if (terminalReleased) {
|
||||
ui.start()
|
||||
ui.setFocus(editor)
|
||||
appendNotice(`Resume handoff failed: ${errorChain(reported)}`, 'error')
|
||||
appendNotice(`Resume handoff failed: ${errorChain(error)}`, 'error')
|
||||
} else {
|
||||
await overlay.close()
|
||||
resumeOverlay = undefined
|
||||
appendNotice(`Resume failed: ${errorChain(reported)}`, 'error')
|
||||
appendNotice(`Resume failed: ${errorChain(error)}`, 'error')
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
|
||||
@@ -10,7 +10,6 @@ import AgentRegistry, {
|
||||
import type { ContentBlock, LlmModelContext, LlmModelInfo, LlmProviderInfo } from '@deepseek-ai/dsh-llm'
|
||||
import CommandService from '@deepseek-ai/dsh-commands'
|
||||
import SessionStore, { SessionId, type Session, type SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionLiveLease } from '@deepseek-ai/dsh-session-persistence'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import type { ToolDefinition } from '@deepseek-ai/dsh-tools'
|
||||
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
|
||||
@@ -53,8 +52,6 @@ export interface TuiHarnessOptions {
|
||||
sessionPersistence?: {
|
||||
list(): Promise<SessionHeader[]>
|
||||
load?(id: ReturnType<typeof SessionId>): Promise<{ meta: SessionHeader; events: Session['events'] }>
|
||||
isLive?(id: ReturnType<typeof SessionId>): Promise<boolean>
|
||||
claimLive?(id: ReturnType<typeof SessionId>): Promise<SessionLiveLease>
|
||||
}
|
||||
handoffResume?: TuiRuntime['handoffResume']
|
||||
/** Set false to exercise the optional session-query degradation path. */
|
||||
@@ -140,12 +137,6 @@ export async function createTuiTestHarness<TerminalType extends Terminal, Exit e
|
||||
inspect: persistence.load === undefined
|
||||
? (id: ReturnType<typeof SessionId>) => Promise.reject(new Error(`session "${id}" not found`))
|
||||
: (id: ReturnType<typeof SessionId>) => persistence.load!(id),
|
||||
claimLive: persistence.claimLive === undefined
|
||||
? () => Promise.resolve({ release: () => Promise.resolve() })
|
||||
: (id: ReturnType<typeof SessionId>) => persistence.claimLive!(id),
|
||||
isLive: persistence.isLive === undefined
|
||||
? () => Promise.resolve(false)
|
||||
: (id: ReturnType<typeof SessionId>) => persistence.isLive!(id),
|
||||
} as never)
|
||||
}
|
||||
if (options.mountSessionQuery !== false && ctx.get('sessionQuery') === undefined) {
|
||||
|
||||
@@ -396,7 +396,7 @@ describe('resume command and /resume', () => {
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
it('keeps persisted query records readable when live-lease inspection is unavailable', async () => {
|
||||
it('keeps persisted query records readable without a persistence service', async () => {
|
||||
const target = header('query-only-persisted', 10, '/workspace')
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
@@ -503,21 +503,19 @@ describe('resume command and /resume', () => {
|
||||
expect(result.terminal.stopped).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it('preflights route availability and occupied or corrupt sessions without losing the current TUI', async () => {
|
||||
it('preflights route availability and corrupt sessions without losing the current TUI', async () => {
|
||||
const missing = header('missing-route', 10, '/workspace')
|
||||
const occupied = header('occupied', 20, '/workspace')
|
||||
const corrupt = header('corrupt', 30, '/workspace')
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
config: { resumeCommand: RESUME },
|
||||
sessionPersistence: {
|
||||
list: async () => [missing, occupied, corrupt],
|
||||
isLive: async id => id === occupied.id,
|
||||
list: async () => [missing, corrupt],
|
||||
load: async (id) => {
|
||||
if (id === corrupt.id) throw new Error('checksum mismatch')
|
||||
return {
|
||||
meta: id === missing.id ? missing : occupied,
|
||||
events: resumeEvents(id === missing.id ? 'Missing adapter' : 'Busy session', id === missing.id ? 'absent-provider' : 'deepseek'),
|
||||
meta: missing,
|
||||
events: resumeEvents('Missing adapter', 'absent-provider'),
|
||||
}
|
||||
},
|
||||
},
|
||||
@@ -527,7 +525,6 @@ describe('resume command and /resume', () => {
|
||||
await tick(); await tick()
|
||||
expect(result.terminal.output).toContain('Missing adapter')
|
||||
expect(result.terminal.output).toContain('absent-provider/model-1')
|
||||
expect(result.terminal.output).toContain('Busy session')
|
||||
expect(result.terminal.output).toContain('Unreadable session')
|
||||
result.terminal.send('Missing adapter')
|
||||
result.terminal.send('\r')
|
||||
@@ -537,6 +534,38 @@ describe('resume command and /resume', () => {
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
it('keeps a session already live in this runtime visible but disabled', async () => {
|
||||
const target = header('live-target', 10, '/workspace')
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>()
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
handoffResume: handoff,
|
||||
async configureContext(ctx) {
|
||||
ctx.provide('tools', { get: () => undefined } as never)
|
||||
ctx.provide('sessionQuery', {
|
||||
listSessions: () => Promise.resolve([{
|
||||
header: target,
|
||||
live: true,
|
||||
persisted: true,
|
||||
}]),
|
||||
readSession: () => Promise.resolve({
|
||||
session: target,
|
||||
events: resumeEvents('Live target'),
|
||||
}),
|
||||
} as never)
|
||||
},
|
||||
})
|
||||
result.terminal.send('/resume')
|
||||
result.terminal.send('\r')
|
||||
await tick(); await tick()
|
||||
result.terminal.send('Live target')
|
||||
result.terminal.send('\r')
|
||||
await tick()
|
||||
expect(result.terminal.output).toContain('session is already live in this runtime')
|
||||
expect(handoff).not.toHaveBeenCalled()
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
it('falls back to assistant provenance and header creation time for sparse logs', async () => {
|
||||
const assistantOnly = header('assistant-route', 20, '/workspace')
|
||||
const empty = header('empty-log', 10, '/workspace')
|
||||
@@ -562,8 +591,6 @@ describe('resume command and /resume', () => {
|
||||
|
||||
it('flushes, releases the terminal, and invokes one host handoff for the same SessionId', async () => {
|
||||
const target = header('target-session', 10, '/workspace')
|
||||
const releaseReservation = vi.fn(() => Promise.resolve())
|
||||
const claimLive = vi.fn(async () => ({ release: releaseReservation }))
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>(() => Promise.reject(new Error('test host retained process')))
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
@@ -571,7 +598,6 @@ describe('resume command and /resume', () => {
|
||||
sessionPersistence: {
|
||||
list: async () => [target],
|
||||
load: async () => ({ meta: target, events: resumeEvents('Target session') }),
|
||||
claimLive,
|
||||
},
|
||||
})
|
||||
result.terminal.send('/resume')
|
||||
@@ -582,8 +608,6 @@ describe('resume command and /resume', () => {
|
||||
await tick(); await tick()
|
||||
expect(handoff).toHaveBeenCalledTimes(1)
|
||||
expect(handoff).toHaveBeenCalledWith(target.id)
|
||||
expect(claimLive).toHaveBeenCalledWith(target.id)
|
||||
expect(releaseReservation).toHaveBeenCalledTimes(1)
|
||||
expect(result.terminal.stopped).toBeGreaterThan(0)
|
||||
expect(result.terminal.output).toContain('Resume handoff failed: test host retained process')
|
||||
await dispose(result)
|
||||
@@ -635,39 +659,46 @@ describe('resume command and /resume', () => {
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
it('keeps the current TUI when the target reservation loses the preflight race', async () => {
|
||||
const target = header('reservation-race', 10, '/workspace')
|
||||
it('does not flush or hand off when disposal begins during selected-session preflight', async () => {
|
||||
const target = header('dispose-during-preflight', 10, '/workspace')
|
||||
const secondListing = Promise.withResolvers<SessionRecord[]>()
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>()
|
||||
const flush = vi.fn()
|
||||
let listings = 0
|
||||
const record: SessionRecord = { header: target, live: false, persisted: true }
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
handoffResume: handoff,
|
||||
async configureContext(ctx) {
|
||||
ctx.provide('tools', { get: () => undefined } as never)
|
||||
ctx.on('session/flush', flush)
|
||||
},
|
||||
sessionPersistence: {
|
||||
list: async () => [target],
|
||||
load: async () => ({ meta: target, events: resumeEvents('Reservation race') }),
|
||||
claimLive: () => Promise.reject(new Error('occupied after preflight')),
|
||||
ctx.provide('sessionQuery', {
|
||||
listSessions: () => ++listings === 1 ? Promise.resolve([record]) : secondListing.promise,
|
||||
readSession: () => Promise.resolve({
|
||||
session: target,
|
||||
events: resumeEvents('Dispose during preflight'),
|
||||
}),
|
||||
} as never)
|
||||
},
|
||||
})
|
||||
result.terminal.send('/resume')
|
||||
result.terminal.send('\r')
|
||||
await tick(); await tick()
|
||||
result.terminal.send('Reservation race')
|
||||
await tick()
|
||||
result.terminal.send('Dispose during preflight')
|
||||
result.terminal.send('\r')
|
||||
await tick(); await tick()
|
||||
expect(result.terminal.output).toContain('Resume failed: occupied after preflight')
|
||||
await vi.waitFor(() => { expect(listings).toBe(2) })
|
||||
await dispose(result)
|
||||
secondListing.resolve([record])
|
||||
await tick()
|
||||
expect(flush).not.toHaveBeenCalled()
|
||||
expect(handoff).not.toHaveBeenCalled()
|
||||
expect(result.terminal.stopped).toBe(0)
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
it('refuses host handoff when a query backend has no persistence lease service', async () => {
|
||||
it('hands off a validated session exposed by a query backend without a persistence service', async () => {
|
||||
const target = header('query-without-persistence', 10, '/workspace')
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>()
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>(
|
||||
() => Promise.reject(new Error('test host retained process')),
|
||||
)
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
handoffResume: handoff,
|
||||
@@ -692,42 +723,14 @@ describe('resume command and /resume', () => {
|
||||
result.terminal.send('Query without persistence')
|
||||
result.terminal.send('\r')
|
||||
await tick(); await tick()
|
||||
expect(result.terminal.output).toContain('session persistence is not mounted')
|
||||
expect(handoff).not.toHaveBeenCalled()
|
||||
expect(handoff).toHaveBeenCalledWith(target.id)
|
||||
expect(result.terminal.output).toContain('Resume handoff failed: test host retained process')
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
it('releases a reservation that resolves after TUI disposal', async () => {
|
||||
const target = header('late-reservation', 10, '/workspace')
|
||||
const claiming = Promise.withResolvers<{ release(): Promise<void> }>()
|
||||
const release = vi.fn(() => Promise.resolve())
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>()
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
handoffResume: handoff,
|
||||
sessionPersistence: {
|
||||
list: async () => [target],
|
||||
load: async () => ({ meta: target, events: resumeEvents('Late reservation') }),
|
||||
claimLive: () => claiming.promise,
|
||||
},
|
||||
})
|
||||
result.terminal.send('/resume')
|
||||
result.terminal.send('\r')
|
||||
await tick(); await tick()
|
||||
result.terminal.send('Late reservation')
|
||||
result.terminal.send('\r')
|
||||
await tick()
|
||||
await dispose(result)
|
||||
claiming.resolve({ release })
|
||||
await tick()
|
||||
expect(release).toHaveBeenCalledTimes(1)
|
||||
expect(handoff).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not hand off after disposal begins during the current-session flush', async () => {
|
||||
const target = header('dispose-during-flush', 10, '/workspace')
|
||||
const flushing = Promise.withResolvers<undefined>()
|
||||
const release = vi.fn(() => Promise.resolve())
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>()
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
@@ -739,7 +742,6 @@ describe('resume command and /resume', () => {
|
||||
sessionPersistence: {
|
||||
list: async () => [target],
|
||||
load: async () => ({ meta: target, events: resumeEvents('Dispose during flush') }),
|
||||
claimLive: async () => ({ release }),
|
||||
},
|
||||
})
|
||||
result.terminal.send('/resume')
|
||||
@@ -752,14 +754,12 @@ describe('resume command and /resume', () => {
|
||||
await tick()
|
||||
flushing.resolve(undefined)
|
||||
await disposing
|
||||
expect(release).toHaveBeenCalledTimes(1)
|
||||
expect(handoff).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not hand off after disposal begins while terminal input drains', async () => {
|
||||
const target = header('dispose-during-drain', 10, '/workspace')
|
||||
const draining = Promise.withResolvers<undefined>()
|
||||
const release = vi.fn(() => Promise.resolve())
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>()
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
@@ -767,7 +767,6 @@ describe('resume command and /resume', () => {
|
||||
sessionPersistence: {
|
||||
list: async () => [target],
|
||||
load: async () => ({ meta: target, events: resumeEvents('Dispose during drain') }),
|
||||
claimLive: async () => ({ release }),
|
||||
},
|
||||
})
|
||||
result.terminal.drainInput.mockImplementationOnce(() => draining.promise)
|
||||
@@ -780,36 +779,33 @@ describe('resume command and /resume', () => {
|
||||
await dispose(result)
|
||||
draining.resolve(undefined)
|
||||
await tick()
|
||||
expect(release).toHaveBeenCalledTimes(1)
|
||||
expect(handoff).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('reports a target reservation release failure after a recoverable host rejection', async () => {
|
||||
const target = header('release-failure', 10, '/workspace')
|
||||
let releases = 0
|
||||
it('does not restart the terminal when a pending host rejects during disposal', async () => {
|
||||
const target = header('host-rejects-during-disposal', 10, '/workspace')
|
||||
const host = Promise.withResolvers<never>()
|
||||
const handoff = vi.fn<NonNullable<TuiRuntime['handoffResume']>>(() => host.promise)
|
||||
const result = await setup({
|
||||
cwd: '/workspace',
|
||||
handoffResume: () => Promise.reject(new Error('host rejected')),
|
||||
handoffResume: handoff,
|
||||
sessionPersistence: {
|
||||
list: async () => [target],
|
||||
load: async () => ({ meta: target, events: resumeEvents('Release failure') }),
|
||||
claimLive: async () => ({
|
||||
release: () => ++releases === 1
|
||||
? Promise.reject(new Error('lock unavailable'))
|
||||
: Promise.resolve(),
|
||||
}),
|
||||
load: async () => ({ meta: target, events: resumeEvents('Host disposal') }),
|
||||
},
|
||||
})
|
||||
result.terminal.send('/resume')
|
||||
result.terminal.send('\r')
|
||||
await tick(); await tick()
|
||||
result.terminal.send('Release failure')
|
||||
result.terminal.send('Host disposal')
|
||||
result.terminal.send('\r')
|
||||
await tick(); await tick()
|
||||
expect(result.terminal.output).toContain('target reservation release failed')
|
||||
expect(result.terminal.output).toContain('release failed: lock')
|
||||
await vi.waitFor(() => { expect(handoff).toHaveBeenCalled() })
|
||||
const startsBeforeDispose = result.terminal.started
|
||||
await dispose(result)
|
||||
expect(releases).toBe(2)
|
||||
host.reject(new Error('host rejected after disposal'))
|
||||
await tick()
|
||||
expect(result.terminal.started).toBe(startsBeforeDispose)
|
||||
expect(result.terminal.output).not.toContain('host rejected after disposal')
|
||||
})
|
||||
|
||||
it('rejects a candidate whose cwd changes between listing and preflight', async () => {
|
||||
|
||||
Reference in New Issue
Block a user