feat(session-persistence): readFrom(seq) primitive for watermark tail reads
SessionPersistence grows readFrom(id, fromSeq, signal?): the non-mutating read-from-seq primitive for checkpoint consumers (the persisted projection cache folds only the tail past its watermark). Coordinator owns validation, per-id serialization, and the sequential fallback (loadStored + forward skip); SQLite implements the optional seek-capable loadStoredFrom hook (WHERE seq >= ?), JSONL stays sequential by contract. Contract suite covers suffix exactness, empty-tail, non-mutation, and cancellation; seam README (both languages) documents the method and the hook.
This commit is contained in:
@@ -289,6 +289,43 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
await expect(persistence.listSnapshots(controller.signal)).rejects.toBe(reason)
|
||||
await expect(persistence.inspect(SessionId('cancelled-inspect'), controller.signal))
|
||||
.rejects.toBe(reason)
|
||||
await expect(persistence.readFrom(SessionId('cancelled-read-from'), 0, controller.signal))
|
||||
.rejects.toBe(reason)
|
||||
} finally {
|
||||
await dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('readFrom returns exactly the stored suffix from the requested seq, without mutating the log', async () => {
|
||||
const { persistence, dispose } = await make()
|
||||
try {
|
||||
const m = meta('read-from', '/work')
|
||||
const log = oneTurnLog()
|
||||
await persistence.create(m)
|
||||
await persistence.append(m.id, log)
|
||||
|
||||
const whole = await persistence.readFrom(m.id, 0)
|
||||
expect(whole.meta).toMatchObject({ id: m.id, cwd: '/work' })
|
||||
expect(whole.events).toEqual(log)
|
||||
|
||||
const suffix = await persistence.readFrom(m.id, 3)
|
||||
expect(suffix.events).toEqual(log.slice(3))
|
||||
expect(suffix.events[0]?.seq).toBe(3)
|
||||
|
||||
// At/past the stored end: an empty tail, never an error.
|
||||
await expect(persistence.readFrom(m.id, log.length)).resolves.toMatchObject({ events: [] })
|
||||
await expect(persistence.readFrom(m.id, log.length + 100)).resolves.toMatchObject({ events: [] })
|
||||
|
||||
// Non-mutating: an interrupted-turn log is served as stored, no closers.
|
||||
await persistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
])
|
||||
const tail = await persistence.readFrom(m.id, 6)
|
||||
expect(tail.events.map(event => event.type)).toEqual(['turn/start'])
|
||||
|
||||
await expect(persistence.readFrom(SessionId('absent-read-from'), 0)).rejects.toThrow('not found')
|
||||
await expect(persistence.readFrom(m.id, -1)).rejects.toThrow('non-negative safe integer')
|
||||
await expect(persistence.readFrom(m.id, 1.5)).rejects.toThrow('non-negative safe integer')
|
||||
} finally {
|
||||
await dispose()
|
||||
}
|
||||
|
||||
@@ -99,6 +99,10 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
|
||||
return this.coordinator.inspect(id, signal)
|
||||
}
|
||||
|
||||
readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
|
||||
return this.coordinator.readFrom(id, fromSeq, signal)
|
||||
}
|
||||
|
||||
// --- PersistenceBackend hooks (the Map storage primitives) ---
|
||||
|
||||
// A Map-backed store has no torn tails, so `tornMarker` is never set.
|
||||
|
||||
Reference in New Issue
Block a user