Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into feat/send-unify

Browse Source
This commit is contained in:
Turtle
2026-07-24 16:31:00 +08:00
committed by GitHub
parent 4a0d8bb0d5 65d29da8a1
commit bb124c2869
162 changed files with 5245 additions and 2329 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/session-persistence/session-persistence-jsonl/README.md
View File

@@ -43,7 +43,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the
## Write path
The plugin buffers frozen session events and drains them on flush or disposal. 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 waits for initialization and the final drain so no write lands after teardown.
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

2
packages/session-persistence/session-persistence-sqlite/README.md
View File

@@ -33,7 +33,7 @@ interface Config {
## Write path
Like the JSONL backend, the plugin also installs the `session/event` → buffer → `session/flush` drain: it copies each already-frozen event into a persistence-owned buffer, persists a fork's seed once on `session/created`, keeps a per-session write cursor so a resumed session never re-appends stored events, and seeds existing live sessions on apply (HMR does not replay `session/created`). Dispose awaits every in-flight init + final drain and then closes the database, so no write lands after teardown.
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

16
packages/session-persistence/session-persistence/README.md
View File

@@ -10,8 +10,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
|---|---|
| `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. |
| `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
| `append(id, events): Promise<void>` | Durably persist a batch (from the `session/flush` drain). Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic closers — an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. |
| `append(id, events): Promise<void>` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and unknown `version` reject. |
| `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. |
@@ -25,13 +25,15 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
## The write coordinator
`PersistenceCoordinator` owns per-id state, write-behind buffers and serialization, the `session/event` → `session/flush` drain, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its stateful methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives. Side-effect-free location queries and lightweight snapshot listing remain backend-owned because they describe storage topology and revision identity; see the [coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
`PersistenceCoordinator` owns per-id state and serialization, one eager write controller per live session, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its stateful methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md) and [flush-controller simplification](../../../.agents/notes/implemented/simplification/2026-07-23-collapse-persistence-flush-state.md).
The coordinator implements durability at checkpoints but does not select their schedule. Persisted deployments explicitly compose [`dsh-session-checkpoint-policy`](../session-checkpoint-policy) when they want request-, tool-dispatch-, and completed-step recovery boundaries; omitting it leaves the loop's coarser checkpoints intact.
Each `session/event` copies its event into the session controller and starts an eager drain without blocking the producer. Concurrent notifications share the current drain; events admitted during a write remain pending and trigger the next batch. `session/flush` is an observation barrier that waits until the controller has no current or pending batch. An eager failure is logged and retains the batch; the next explicit flush or backend teardown retries it and surfaces failure to its caller.
When a live session emits `session/disposed`, the coordinator waits for its initialization, serializes a final buffer drain, then releases every map entry owned by that exact `Session` object. A failed final drain keeps the pending buffer for backend teardown to retry. Backend teardown stops event admission first, awaits all in-flight session retirements and remaining per-id operations, drains any retained buffers, and only then closes the storage handle.
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.
The side-effect-free `locate` query remains backend-owned because it describes storage topology rather than write orchestration.
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.
The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinator and storage):
@@ -74,6 +76,6 @@ Persistence does not mutate live request prefixes. A resumed loop can reuse prov
## Known Limitations and Deferred Work
- **No deletion or retention surface** — the seam is `create`/`append`/`load`/`list` only; pruning stored sessions is out-of-band backend maintenance.
- **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.

210
packages/session-persistence/session-persistence/src/coordinator.ts
View File

@@ -91,6 +91,13 @@ interface SessionState {
owner?: Session
}
/** One live session's initialization and eager write-behind controller. */
interface LiveSessionState {
pending: SessionEvent[]
init: Promise<void>
flush: Promise<void> | undefined
}
/** Collect the rejection reasons from a set of promises (none-throwing). */
async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unknown[]> {
const settled = await Promise.allSettled([...promises])
@@ -145,21 +152,15 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId):
export class PersistenceCoordinator<TornMarker = unknown> {
/** Backend bookkeeping keyed by session id (NOT the live Session object). */
private states = new Map<SessionId, SessionState>()
/** Write-behind buffers keyed by the live Session (write path). */
private buffers = new Map<Session, SessionEvent[]>()
/** Lifecycle and write-behind state keyed by the exact live Session. */
private live = new Map<Session, LiveSessionState>()
/** Cold loads currently reserving an id across backend reads and repair writes. */
private coldLoads = new Set<SessionId>()
/**
* Per-session serialization: every operation chains onto the prior one for the
* same id, so writes for one session never interleave. Keyed by session id.
*/
private chains = new Map<SessionId, Promise<unknown>>()
/**
* Init promises keyed by live session object, preventing an id-reusing
* replacement from inheriting stale initialization. Flush is the public
* observation boundary; callers do not inspect this bookkeeping directly.
*/
private inits = new Map<Session, Promise<void>>()
/** Final drains started by fire-and-forget session disposal notifications. */
private retirements = new Set<Promise<void>>()
constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
this.installWritePath()
@@ -248,8 +249,18 @@ export class PersistenceCoordinator<TornMarker = unknown> {
* @param id - the persisted session to reload.
* @returns the header plus the event log, ending on a balanced `turn/end`.
*/
load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
return this.serialize(id, () => this.loadCore(id))
async load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
const selected = await this.serialize(id, async () => {
const live = this.ctx.sessions.get(id)
if (live !== undefined) return { live }
this.coldLoads.add(id)
try {
return { loaded: await this.loadCore(id) }
} finally {
this.coldLoads.delete(id)
}
})
return 'loaded' in selected ? selected.loaded : this.loadLiveSnapshot(selected.live)
}
/**
@@ -295,6 +306,21 @@ export class PersistenceCoordinator<TornMarker = unknown> {
return { meta, events: balanced }
}
/** Return a durable balanced live snapshot without applying cold crash repair. */
private async loadLiveSnapshot(session: Session): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
const events = session.events.map(event => structuredClone(event))
await this.flush(session)
const state = this.states.get(session.id)
/* v8 ignore next -- successful flush always publishes this live session's durable state */
if (state === undefined) throw new Error(`session "${session.id}" lost persistence state during load`)
const meta = structuredClone(state.meta)
if (events.length === 0) throw new Error(`session "${session.id}" not found`)
if (interruptedTurnClosers(events).length > 0) {
throw new Error(`cannot load session "${session.id}" while its live turn is open; use the live Session or wait for the turn to close`)
}
return { meta, events }
}
// Listing is a direct backend read and needs no coordinator state.
// --- per-id serialization + adoption helpers ---
@@ -305,7 +331,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
* public methods must NOT call each other (deadlock); they call the unserialized
* `*Core` helpers instead.
*/
private serialize<T>(id: SessionId, op: () => Promise<T>): Promise<T> {
private serialize<T>(id: SessionId, op: () => Promise<T> | T): Promise<T> {
const prior = this.chains.get(id) ?? Promise.resolve()
const next = prior.then(op, op)
// Keep the chain alive but swallow this op's rejection for the NEXT waiter
@@ -353,15 +379,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
// reverse registration order, so event admission closes before this final
// drain reaches quiescence and closes the backend.
ctx.effect(() => async () => {
await this.awaitRetirements()
let disposeError: unknown
try {
const errors = [
...await settledErrors(this.inits.values()),
...await settledErrors([...this.buffers.keys()].map(s => this.flush(s))),
...await settledErrors(this.chains.values()),
]
const errors = await settledErrors([...this.live.keys()].map(session => this.flush(session)))
while (this.chains.size > 0) await Promise.allSettled([...this.chains.values()])
if (errors.length > 0) {
throw new AggregateError(errors, `${this.backend.name} dispose failed`)
}
@@ -382,25 +403,25 @@ export class PersistenceCoordinator<TornMarker = unknown> {
}
}, `${this.backend.name} write path`)
// Capture the header on creation; persist a fork's seed once. Record the init
// promise so flush/dispose can await it (onCreated is async).
ctx.on('session/created', (session) => { void this.initFor(session) })
// Session emits an owned frozen event. Keep a persistence-owned copy anyway
// so the write-behind queue owns exactly the record it will flush rather than
// retaining a product-layer record by identity. Serializability is guaranteed
// at the source, so structuredClone is safe.
ctx.on('session/event', (session, event) => {
let buffer = this.buffers.get(session)
if (!buffer) this.buffers.set(session, buffer = [])
buffer.push(structuredClone(event))
// Capture the header on creation and persist a fork's seed once.
ctx.on('session/created', (session) => {
if (this.coldLoads.has(session.id)) {
throw new Error(`cannot publish session "${session.id}" while its persisted history is loading`)
}
void this.initFor(session)
})
// Drain to the backend at the durability checkpoint.
// Keep a persistence-owned copy of each frozen event and start an eager drain.
ctx.on('session/event', (session, event) => {
const live = this.initFor(session)
live.pending.push(structuredClone(event))
if (live.flush === undefined) this.scheduleDrain(session, live)
})
// Callers use flush as the observation barrier for the eager write path.
ctx.on('session/flush', session => this.flush(session))
// Session disposal is observe-only, so the coordinator observes the
// detached task itself and backend teardown awaits quiescence.
// Session disposal is observe-only, so retirement contains its own failure.
ctx.on('session/disposed', (session) => { this.retire(session) })
// HMR: a hot reload does not replay session/created, so seed existing live
@@ -408,52 +429,34 @@ export class PersistenceCoordinator<TornMarker = unknown> {
for (const session of ctx.sessions.list()) void this.initFor(session)
}
/** Start, observe, and track one disposed session's final drain. */
/** Start and observe one disposed session's final drain. */
private retire(session: Session): void {
const task = this.retireCore(session)
this.retirements.add(task)
const settled = (): void => { this.retirements.delete(task) }
void task.then(settled, (error: unknown) => {
settled()
if (!this.live.has(session)) return
void this.retireCore(session).catch((error: unknown) => {
this.ctx.logger.warn(`${this.backend.name}: session "${session.id}" retirement failed: ${String(error)}`)
})
}
/** Drain and release state owned by one exact disposed Session lifecycle. */
private async retireCore(session: Session): Promise<void> {
await this.inits.get(session)
await this.flush(session)
const id = session.header.id
await this.serialize(id, async () => {
await this.drain(session)
this.buffers.delete(session)
this.inits.delete(session)
await this.serialize(id, () => {
this.live.delete(session)
if (this.states.get(id)?.owner === session) this.states.delete(id)
})
}
/** Await every retirement admitted before listener teardown. */
private async awaitRetirements(): Promise<void> {
while (this.retirements.size > 0) {
await Promise.allSettled([...this.retirements])
}
}
/** Start (once) the async init for a session and remember its promise. */
private initFor(session: Session): Promise<void> {
const existing = this.inits.get(session)
/** Return the one lifecycle controller for a live session, creating it if needed. */
private initFor(session: Session): LiveSessionState {
const existing = this.live.get(session)
if (existing) return existing
// Snapshot the seed SYNCHRONOUSLY — initFor runs inside the `session/created`
// emit, before any later append invalidates the public array snapshot. Events
// are already frozen; cloning gives persistence independent ownership.
const seed = session.events.map(e => structuredClone(e))
const p = this.onCreated(session, seed)
// Attach a no-op rejection handler so a failing init does not surface as an
// unhandled rejection if no flush observes `p` before it rejects. The REAL
// error is still delivered: flush/dispose await the same `p` from the map.
p.catch(() => { /* observed by flush/dispose via the stored promise */ })
this.inits.set(session, p)
return p
const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined }
this.live.set(session, live)
live.init = this.serialize(session.header.id, () => this.onCreated(session, seed))
live.init.catch(() => { /* observed by flush/dispose through the controller */ })
return live
}
/**
@@ -508,13 +511,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
// Persist the seed SUFFIX beyond the persisted prefix. Constructor seed
// events never emit session/event, so the buffer never sees them.
const suffix = seed.slice(tracked.cursor)
if (suffix.length > 0) await this.append(id, suffix)
if (suffix.length > 0) await this.appendCore(id, suffix)
return
}
// Owned by a DIFFERENT live session. Reclaim ONLY a truly-abandoned id
// (never materialized, no pending buffer); else it is a real collision.
const ownerBuffer = this.buffers.get(tracked.owner)
if (!tracked.materialized && !ownerBuffer?.length) {
const owner = this.live.get(tracked.owner)
if (!tracked.materialized && !owner?.pending.length) {
this.states.delete(id)
} else {
throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
@@ -528,20 +529,20 @@ export class PersistenceCoordinator<TornMarker = unknown> {
// Do NOT route through loadCore(): that crash-repairs open turns as
// interrupted, which is wrong for HMR while the live Session is still the
// authority and may append the real step/turn end later.
await this.serialize(id, () => this.adoptLivePrefix(session, seed, live))
await this.adoptLivePrefix(session, seed, live)
return
}
// case 4: a genuinely new session. Register its meta (lazy), then persist its
// seed (events present at creation time) once.
const meta: SessionHeader = { ...session.header }
await this.create(meta)
await this.createCore(meta)
// Bind this state to the live session so a later DIFFERENT session reusing
// the id is detected as a collision (case 1) rather than silently no-opped.
const created = this.states.get(id)
/* v8 ignore next -- create() always sets the state for the id */
if (created !== undefined) created.owner = session
if (seed.length > 0) await this.append(id, seed)
if (seed.length > 0) await this.appendCore(id, seed)
}
/**
@@ -574,36 +575,43 @@ export class PersistenceCoordinator<TornMarker = unknown> {
}
private async flush(session: Session): Promise<void> {
// Wait for the session's init (onCreated) so the state/cursor and any
// fork-seed persistence are in place before draining. Awaiting the same
// promise initFor stored also surfaces an init failure (e.g. a collision)
// here, where the caller of session/flush observes it.
await this.inits.get(session)
// Serialize the WHOLE drain (read cursor → append → splice) on the per-session
// chain so two concurrent flushes cannot both read the same cursor and
// seq-mismatch on the second append.
await this.serialize(session.header.id, () => this.drain(session))
const live = this.initFor(session)
await live.init
const overlapping = live.flush
if (overlapping !== undefined) await Promise.allSettled([overlapping])
while (live.flush !== undefined || live.pending.length > 0) {
if (live.flush !== undefined) await live.flush
else await this.ensureFlush(session, live)
}
}
/** Drain a session's write buffer to the backend. Caller serializes this per id. */
private async drain(session: Session): Promise<void> {
const buffer = this.buffers.get(session)
if (!buffer?.length) return
// Copy WITHOUT removing: the buffer is the only durable-pending copy of these
// events. Drain it only AFTER the append commits; events pushed during the
// await sit past batch.length and survive the prefix splice, so a
// retry/dispose re-drains the rest.
const batch = buffer.slice()
const state = this.states.get(session.header.id)
// Only append events at or beyond the write cursor (a resumed session's seed
// is already stored). flush awaits the init above, which always sets state,
// so the `?? 0` fallback is a defensive guard that never fires in practice.
/** Start an eager drain without exposing its failure to the synchronous append. */
private scheduleDrain(session: Session, live: LiveSessionState): void {
void this.ensureFlush(session, live).catch((error: unknown) => {
this.ctx.logger.warn(`${this.backend.name}: eager drain for session "${session.id}" failed (buffered events retained): ${String(error)}`)
})
}
/** Start one drain for the complete pending batch. */
private ensureFlush(session: Session, live: LiveSessionState): Promise<void> {
const flush = live.init
.then(() => this.serialize(session.header.id, () => this.drain(session.header.id, live)))
.finally(() => { live.flush = undefined })
live.flush = flush
void flush.then(() => {
if (live.pending.length > 0) this.scheduleDrain(session, live)
}, () => {})
return flush
}
/** Drain one stable prefix; events admitted during the write remain pending. */
private async drain(id: SessionId, live: LiveSessionState): Promise<void> {
const batch = live.pending.slice()
const state = this.states.get(id)
/* v8 ignore next -- state is always set by the awaited init before flush */
const cursor = state?.cursor ?? 0
const fresh = batch.filter(e => e.seq >= cursor)
// appendCore (NOT the serialized append) — drain already runs inside the
// per-session chain, so re-entering via append() would deadlock.
if (fresh.length > 0) await this.appendCore(session.header.id, fresh)
buffer.splice(0, batch.length)
await this.appendCore(id, fresh)
live.pending.splice(0, batch.length)
}
}

18
packages/session-persistence/session-persistence/src/index.ts
View File

@@ -73,10 +73,9 @@ export abstract class SessionPersistence extends Service {
abstract create(meta: SessionHeader): Promise<void>
/**
* Durably persist a batch of events (called from the write-behind drain at
* the `session/flush` checkpoint). Honors the append-only and contiguous-seq
* contracts: the first event's `seq` MUST equal the stored next-seq (after
* `load` has durably closed any interrupted turn). Rejects non-JSON-
* Durably persist a batch of events. Honors the append-only and contiguous-
* seq contracts: the first event's `seq` MUST equal the stored next-seq
* (after `load` has durably closed any interrupted turn). Rejects non-JSON-
* serializable `event.data` with an error naming the offending event type.
* @param id - the session the batch belongs to.
* @param events - the contiguous batch to persist, in seq order.
@@ -85,9 +84,14 @@ export abstract class SessionPersistence extends Service {
/**
* Load a header and balanced contiguous log. A complete interrupted final
* turn is preserved and durably closed with missing tool errors plus any open
* step and turn boundaries; only a torn final record is discarded. Unknown
* versions and corruption in the committed prefix reject.
* turn is preserved and durably closed with missing tool errors plus any open
* step and turn boundaries; only a torn final record is discarded. Unknown
* versions and corruption in the committed prefix reject. Implementations
* MUST NOT crash-repair an identity still bound to a live Session: a balanced
* live log may return with its stored header as a durable snapshot, while an
* open live turn rejects.
* A coordinator-backed cold load reserves the identity across storage awaits,
* so concurrent publication of a same-id live Session rejects.
* @param id - the persisted session to reload.
* @returns the header and a log ending on a balanced `turn/end`.
*/

119
packages/session-persistence/session-persistence/tests/coordinator-contract.ts
View File

@@ -88,6 +88,84 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
}
})
it('rejects crash-repair load while a live session owns the persisted prefix', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
let session!: Session
const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
session = inner.sessions.create(SessionId('live-load'), { meta: { cwd: WORK } })
}, { inject: ['sessions'] }))
try {
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
await ctx.sessions.flush(session)
await expect(ctx.sessionPersistence.load(session.id))
.rejects.toThrow(`cannot load session "${session.id}" while its live turn is open`)
send(session, oneTurnLog().slice(1))
await ctx.sessions.flush(session)
await sessionFiber.dispose()
await vi.waitFor(async () => {
const loaded = await ctx.sessionPersistence.load(session.id)
expect(loaded.events.map(event => event.type)).toEqual(oneTurnLog().map(event => event.type))
expect(loaded.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'completed' } },
})
})
} finally {
await sessionFiber.dispose()
await fiber.dispose()
await fix.cleanup()
}
})
it('rechecks live ownership after a cold load enters the per-id chain', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
const id = SessionId('queued-load-live-race')
const header = meta(id, WORK)
const start: SessionEvent = {
type: 'turn/start',
seq: 0,
time: 1,
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
}
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(id, [start])
const loading = ctx.sessionPersistence.load(id)
const live = ctx.sessions.create(id, { seed: [start], meta: header })
await expect(loading).rejects.toThrow(/live turn is open/)
live.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.sessions.flush(live)
const loaded = await ctx.sessionPersistence.load(id)
expect(loaded.events.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
expect(loaded.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'completed' } },
})
} finally {
await fiber.dispose()
await fix.cleanup()
}
})
it('does not load an unmaterialized empty live session', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
const session = ctx.sessions.create(SessionId('empty-live'), { meta: { cwd: WORK } })
await expect(ctx.sessionPersistence.load(session.id)).rejects.toThrow(/not found/)
} finally {
await fiber.dispose()
await fix.cleanup()
}
})
it('round-trips the seed boundary (seedLength) through persistence', async () => {
// A forked child records how many leading events were inherited via the seed; the
// boundary must survive a reload (so a resume/replay can tell the inherited prefix from
@@ -95,9 +173,13 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
const session = ctx.sessions.create(SessionId('forked-child'), { meta: { cwd: WORK, seedLength: 3 } })
let session!: Session
const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
session = inner.sessions.create(SessionId('forked-child'), { meta: { cwd: WORK, seedLength: 3 } })
}, { inject: ['sessions'] }))
send(session, oneTurnLog())
await ctx.sessions.flush(session)
await sessionFiber.dispose()
const loaded = await ctx.sessionPersistence.load(SessionId('forked-child'))
expect(loaded.meta.seedLength).toBe(3)
@@ -114,11 +196,15 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
const session = ctx.sessions.create(SessionId('delegated-child'), {
meta: { cwd: WORK, parentSession: SessionId('root'), delegationDepth: 2 },
})
let session!: Session
const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
session = inner.sessions.create(SessionId('delegated-child'), {
meta: { cwd: WORK, parentSession: SessionId('root'), delegationDepth: 2 },
})
}, { inject: ['sessions'] }))
send(session, oneTurnLog())
await ctx.parallel('session/flush', session)
await sessionFiber.dispose()
const loaded = await ctx.sessionPersistence.load(SessionId('delegated-child'))
expect(loaded.meta.delegationDepth).toBe(2)
@@ -526,20 +612,31 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const { ctx, fiber } = await freshCtx(fix)
try {
// Materialize and load (ownerless, cursor = 6).
await ctx.sessionPersistence.create(meta('claim', WORK))
const storedMeta = meta('claim', WORK)
await ctx.sessionPersistence.create(storedMeta)
await ctx.sessionPersistence.append(SessionId('claim'), oneTurnLog())
const { events } = await ctx.sessionPersistence.load(SessionId('claim'))
const { events, meta: durableMeta } = await ctx.sessionPersistence.load(SessionId('claim'))
// A live session SEEDED with the loaded log PLUS a new turn claims the
// ownerless state and persists only the suffix.
const cont = ctx.sessions.create(SessionId('claim'), { seed: [
...events,
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
], meta: { cwd: WORK } })
let cont!: Session
const contFiber = await ctx.plugin(Object.assign((inner: Context) => {
cont = inner.sessions.create(SessionId('claim'), { seed: [
...events,
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
], meta: { cwd: WORK, createdAt: 2000 } })
}, { inject: ['sessions'] }))
await ctx.sessions.flush(cont)
const loaded = await ctx.sessionPersistence.load(SessionId('claim'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
expect(loaded.meta).toEqual(durableMeta)
expect(loaded.meta.createdAt).toBe(1000)
await contFiber.dispose()
await vi.waitFor(async () => {
expect((await ctx.sessionPersistence.load(SessionId('claim'))).meta).toEqual(durableMeta)
})
} finally {
await fiber.dispose()
await fix.cleanup()

271
packages/session-persistence/session-persistence/tests/persistence.spec.ts
View File

@@ -48,10 +48,8 @@ interface MemoryConfig { store?: MemoryStore }
/** Test-only view of the coordinator containers whose retirement is the contract under test. */
interface CoordinatorInternals {
states: Map<unknown, unknown>
buffers: Map<unknown, unknown>
live: Map<unknown, { pending: unknown[]; flush: Promise<void> | undefined }>
chains: Map<unknown, unknown>
inits: Map<unknown, unknown>
retirements: Set<Promise<void>>
}
/**
@@ -209,6 +207,75 @@ runCoordinatorContract('memory', async (): Promise<CoordinatorFixture> => {
}
})
describe('PersistenceCoordinator eager writes', () => {
it('starts a follow-up batch for events admitted during an in-flight write', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const appendGate = Promise.withResolvers<boolean>()
backend.beforeAppend = async (attempt) => {
if (attempt === 1) await appendGate.promise
}
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
const session = ctx.sessions.create(SessionId('eager-follow-up'))
await ctx.sessions.flush(session)
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
appendGate.resolve(true)
await vi.waitFor(() => {
expect(backend.appendAttempts).toBe(2)
expect(backend.store.get(session.id)?.events.map(event => event.seq)).toEqual([0, 1])
})
} finally {
appendGate.resolve(true)
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('retries a failed overlapping eager write at the explicit flush barrier', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const appendGate = Promise.withResolvers<boolean>()
backend.beforeAppend = async (attempt) => {
if (attempt === 1) {
await appendGate.promise
throw new Error('transient eager failure')
}
}
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
const session = ctx.sessions.create(SessionId('eager-flush-retry'))
await ctx.sessions.flush(session)
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
const barriers = [ctx.sessions.flush(session), ctx.sessions.flush(session)]
appendGate.resolve(true)
await expect(Promise.all(barriers)).resolves.toEqual([undefined, undefined])
expect(backend.appendAttempts).toBe(2)
expect(backend.store.get(session.id)?.events.map(event => event.seq)).toEqual([0, 1])
} finally {
appendGate.resolve(true)
await fiber.dispose()
await ctx.fiber.dispose()
}
})
})
describe('PersistenceCoordinator stored identity', () => {
it('rejects a mismatched backend header before repair or state publication', async () => {
const ctx = new Context()
@@ -237,6 +304,48 @@ describe('PersistenceCoordinator stored identity', () => {
await ctx.fiber.dispose()
}
})
it('reserves a cold id across asynchronous storage repair', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('cold-load-reservation')
const header = meta(id)
const start: SessionEvent = {
type: 'turn/start',
seq: 0,
time: 1,
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
}
backend.store.set(id, { meta: header, events: [start] })
const loadGate = Promise.withResolvers<boolean>()
backend.beforeLoadStored = async () => { await loadGate.promise }
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
const loading = coordinator.load(id)
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) })
await expect(ctx.plugin(Object.assign((inner: Context) => {
inner.sessions.create(id, { seed: [start], meta: header })
}, { inject: ['sessions'] }))).rejects.toThrow(/persisted history is loading/)
expect(ctx.sessions.get(id)).toBeUndefined()
loadGate.resolve(true)
const loaded = await loading
expect(loaded.events.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
const resumed = ctx.sessions.create(id, { seed: loaded.events, meta: loaded.meta })
await expect(ctx.sessions.flush(resumed)).resolves.toBeUndefined()
} finally {
loadGate.resolve(true)
await fiber.dispose()
await ctx.fiber.dispose()
}
})
})
describe('PersistenceCoordinator retirement', () => {
@@ -244,35 +353,30 @@ describe('PersistenceCoordinator retirement', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
let coordinator!: PersistenceCoordinator<never>
const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
const loadGate = Promise.withResolvers<boolean>()
backend.beforeLoadStored = async (attempt) => {
if (attempt === 1) await loadGate.promise
}
try {
const id = SessionId('retiring-lazy-owner')
let first!: Session
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
first = inner.sessions.create(id)
inner.sessions.create(id)
}, { inject: ['sessions'] }))
await ctx.sessions.flush(first)
const baselineLoads = backend.loadAttempts
backend.beforeLoadStored = async () => { await loadGate.promise }
const blockingLoad = coordinator.load(id)
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) })
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) })
await firstFiber.dispose()
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create(id)
}, { inject: ['sessions'] }))
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 2) })
const reuseFlush = ctx.sessions.flush(reuse)
loadGate.resolve(true)
await expect(blockingLoad).rejects.toThrow(/not found/)
await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
await expect(reuseFlush).resolves.toBeUndefined()
} finally {
loadGate.resolve(true)
await backendFiber.dispose()
@@ -280,7 +384,45 @@ describe('PersistenceCoordinator retirement', () => {
}
})
it('a retiring owner with buffered events still rejects same-id reuse', async () => {
it('a replacement queued before retirement cleanup still collides with the live owner', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
const appendGate = Promise.withResolvers<boolean>()
try {
const id = SessionId('retiring-live-owner')
let first!: Session
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
first = inner.sessions.create(id)
}, { inject: ['sessions'] }))
await ctx.sessions.flush(first)
backend.beforeAppend = async () => { await appendGate.promise }
first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
await firstFiber.dispose()
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create(id)
}, { inject: ['sessions'] }))
const reuseFlush = ctx.sessions.flush(reuse)
appendGate.resolve(true)
await expect(reuseFlush).rejects.toThrow(/bound to a different live session/)
expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
} finally {
appendGate.resolve(true)
await backendFiber.dispose()
await ctx.fiber.dispose()
}
})
it('a racing cold load survives retirement cleanup and rejects same-id reuse', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
@@ -288,6 +430,7 @@ describe('PersistenceCoordinator retirement', () => {
const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
const appendGate = Promise.withResolvers<boolean>()
const loadGate = Promise.withResolvers<boolean>()
try {
@@ -297,27 +440,37 @@ describe('PersistenceCoordinator retirement', () => {
first = inner.sessions.create(id)
}, { inject: ['sessions'] }))
await ctx.sessions.flush(first)
backend.beforeAppend = async () => { await appendGate.promise }
first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
await firstFiber.dispose()
const baselineLoads = backend.loadAttempts
backend.beforeLoadStored = async () => { await loadGate.promise }
const blockingLoad = coordinator.load(id)
const coldLoad = coordinator.load(id)
appendGate.resolve(true)
await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) })
await firstFiber.dispose()
await expect(ctx.plugin(Object.assign((inner: Context) => {
inner.sessions.create(id)
}, { inject: ['sessions'] }))).rejects.toThrow(/persisted history is loading/)
loadGate.resolve(true)
await expect(coldLoad).resolves.toMatchObject({
events: [{ seq: 0 }, { seq: 1 }],
})
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create(id)
}, { inject: ['sessions'] }))
await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/bound to a different live session/)
loadGate.resolve(true)
await expect(blockingLoad).rejects.toThrow(/not found/)
await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/id collision/)
await vi.waitFor(() => {
expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
})
} finally {
appendGate.resolve(true)
loadGate.resolve(true)
await backendFiber.dispose()
await ctx.fiber.dispose()
@@ -381,8 +534,9 @@ describe('PersistenceCoordinator retirement', () => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
const internals = coordinator as unknown as CoordinatorInternals
backend.beforeAppend = async (attempt) => {
if (attempt === 1) {
let retryEnabled = false
backend.beforeAppend = async () => {
if (!retryEnabled) {
backend.lifecycle.push('append-failed')
throw new Error('transient append failure')
}
@@ -399,17 +553,18 @@ describe('PersistenceCoordinator retirement', () => {
await sessionFiber.dispose()
await vi.waitFor(() => {
expect(backend.appendAttempts).toBe(1)
expect(internals.retirements.size).toBe(0)
expect(backend.appendAttempts).toBeGreaterThanOrEqual(1)
expect([...internals.live.values()][0]?.pending).toEqual(expect.arrayContaining([
expect.objectContaining({ seq: 0 }),
expect.objectContaining({ seq: 1 }),
]))
})
expect([...internals.buffers.values()]).toEqual([expect.arrayContaining([
expect.objectContaining({ seq: 0 }),
expect.objectContaining({ seq: 1 }),
])])
retryEnabled = true
await backendFiber.dispose()
expect(backend.store.get(SessionId('retry-retirement'))?.events.map(event => event.seq)).toEqual([0, 1])
expect(backend.lifecycle).toEqual(['append-failed', 'append-committed', 'close'])
expect(backend.lifecycle.at(-2)).toBe('append-committed')
expect(backend.lifecycle.at(-1)).toBe('close')
} finally {
await backendFiber.dispose()
await ctx.fiber.dispose()
@@ -442,7 +597,8 @@ describe('PersistenceCoordinator retirement', () => {
await sessionFiber.dispose()
await vi.waitFor(() => {
expect(backend.appendAttempts).toBe(1)
expect(internals.retirements.size).toBe(1)
expect(internals.live.size).toBe(1)
expect([...internals.live.values()][0]?.flush).toBeInstanceOf(Promise)
})
let disposed = false
@@ -461,6 +617,47 @@ describe('PersistenceCoordinator retirement', () => {
await ctx.fiber.dispose()
}
})
it('backend teardown waits for a detached public append before close', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
const appendGate = Promise.withResolvers<boolean>()
backend.beforeAppend = async () => {
backend.lifecycle.push('append-started')
await appendGate.promise
backend.lifecycle.push('append-committed')
}
try {
const id = SessionId('inflight-public-append')
await coordinator.create(meta(id))
const append = coordinator.append(id, [{
type: 'turn/start',
seq: 0,
time: 1,
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
}])
await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
let disposed = false
const teardown = fiber.dispose().then(() => { disposed = true })
await Promise.resolve()
expect(disposed).toBe(false)
appendGate.resolve(true)
await Promise.all([append, teardown])
expect(backend.lifecycle).toEqual(['append-started', 'append-committed', 'close'])
} finally {
appendGate.resolve(true)
await fiber.dispose()
await ctx.fiber.dispose()
}
})
})
describe('SessionPersistence service registration', () => {
@@ -590,11 +787,9 @@ describe('SessionPersistence service registration', () => {
expect(ctx.sessions.list()).toHaveLength(0)
expect({
states: coordinator.states.size,
buffers: coordinator.buffers.size,
live: coordinator.live.size,
chains: coordinator.chains.size,
inits: coordinator.inits.size,
retirements: coordinator.retirements.size,
}).toEqual({ states: 0, buffers: 0, chains: 0, inits: 0, retirements: 0 })
}).toEqual({ states: 0, live: 0, chains: 0 })
})
} finally {
await fiber.dispose()