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

fix: return stored metadata from live loads

Browse Source
This commit is contained in:
_Kerman
2026-07-23 21:28:52 +08:00
parent 4e5166828b
commit c596dddfe1
10 changed files with 36 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -11,7 +11,7 @@ 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. 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 flushed balanced snapshot for a live session, rejecting while its turn is open; 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. |
| `load(id): Promise<{ meta; events }>` | Return the stored header plus a flushed balanced event snapshot for a live session, rejecting while its turn is open; 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. |
| `list(): Promise<SessionHeader[]>` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. |
## Invariants every backend must honor
@@ -27,7 +27,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
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.
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 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.
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 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.

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

@@ -286,9 +286,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
/** Return a durable balanced live snapshot without applying cold crash repair. */
private async loadLiveSnapshot(session: Session): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
const meta = structuredClone(session.header)
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`)

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

@@ -78,7 +78,8 @@ export abstract class SessionPersistence extends Service {
* 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 as a durable snapshot, while an open live turn rejects.
* 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.

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

@@ -612,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()