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

feat(session): persist optional time zones

Browse Source
This commit is contained in:
pku-xht
2026-08-06 05:06:57 +08:00
committed by Tianyi Cui
parent d61059364e
commit a667ec55d6
25 changed files with 516 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,7 +4,7 @@ English | [中文](README.zh.md)
The durable session-persistence Service Definition (`ctx.sessionPersistence`). Defines WHAT a persistence backend does — durably store, reload, and list sessions — without saying HOW. Mirrors the `dsh-bash` capability-seam template ([capability seams](../../../.agents/notes/implemented/architecture/2026-06-13-capability-seams.md)): an abstract service here, a Service provider in a sibling package, and Consumers that inject the service.
The persisted unit IS the existing `SessionEvent` (event-sourced model — the log is the single source of truth), so there is no parallel "persisted message" type. Metadata that is NOT replayable conversation state (format version, cwd, lineage, seed boundary, origin, delegation depth) travels separately as `SessionHeader`, owned by `dsh-session` and re-exported here.
The persisted unit IS the existing `SessionEvent` (event-sourced model — the log is the single source of truth), so there is no parallel "persisted message" type. Metadata that is NOT replayable conversation state (format version, cwd, optional time zone, lineage, seed boundary, origin, delegation depth) travels separately as `SessionHeader`, owned by `dsh-session` and re-exported here.
## Service API (`ctx.sessionPersistence`)
@@ -33,7 +33,9 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
Each `session/event` copies its event into the session controller. The first pending event starts a fixed batching window; later events join without resetting its deadline. The configured `writeBatchMaxDelayMs` bounds this intentional wait, not event-loop, initialization, serialized-operation, or backend latency. Events admitted during a write form a new bounded batch. `session/flush` cancels the wait and is a shared quiescence barrier that drains events admitted while it runs. A background failure is logged once, retains the ordered batch, and pauses automatic retry; a new event starts a fresh window, while explicit flush or backend teardown retries immediately and surfaces a repeated failure.
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. For a cold id, inspection reads, validates, freezes, and constructs one unpublished Session; repeated inspection reuses that object graph only while its source revision remains current. `prepare(id)` performs the same check before repair, reserves the exact Session, commits any pending torn-tail/interrupted-turn repair, and returns it for publication. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn.
A live controller retains no seed copy. If first initialization rejects, the next flush borrows the current append-only Session log, rechecks the backend's actual cursor, and appends only the missing suffix before draining retained events. Concurrent retries share one initialization attempt; a committed-but-rejected write therefore neither duplicates the prefix nor permanently poisons the Session.
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. For a cold id, inspection reads, validates, freezes, and constructs one unpublished Session; repeated inspection reuses that object graph only while its source revision remains current. `prepare(id)` performs the same check before repair, reserves the exact Session, commits any pending torn-tail/interrupted-turn repair, and returns it for publication. HMR adoption reads through `loadStored`, compares cwd and any stored `timeZone`, and never closes the active turn. A stored header without `timeZone` is the compatibility exception: a zoned live object may adopt it, but the stored header remains headerless and is never backfilled.
Backend reads normalize the exact supported same-version shapes before current-shape validation. Pre-identity messages receive the deterministic id `legacy-message:<session-id>:<event-seq>`; a tool-result content replacement inherits its target's imported id. A pre-react-loop `turn/start` loses its obsolete trigger, a removed `steering/message` becomes the same identified `user/message`, and an older `turn/end` maps its terminal reason without inventing a caller that the old record did not name. The coordinator uses the same normalized view for `load`, `inspect`, `readFrom`, ownerless-state claims, and HMR prefix adoption. Storage remains append-only: reads do not rewrite old records, and later appends use the current shape. These are narrow import exceptions from the [pre-identity message](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md) and [pre-react-loop session](../../../.agents/notes/implemented/bug-fix/2026-08-04-load-pre-react-loop-sessions.md) decisions, not a general v0 migration promise.
@@ -54,11 +56,11 @@ The `PersistenceBackend<TornMarker>` hooks (the only contract between the coordi
| `list(signal?)` | List all stored metadata, observing optional cancellation. |
| `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 takes ownership of fresh backend values, validates and freezes them once, and retains at most the configured number of unpublished Sessions without calling `commitRepair`. A retained source is reused or repaired only when its revision still equals `readStoredRevision`; otherwise the coordinator reloads it. This freshness check does not add cross-process writer exclusion. Revision retries converge when the durable log remains unchanged for one read/check round trip; continuous external writers can delay `load`, `inspect`, or `prepare`. 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).
The coordinator asserts the stored id and validates the optional stored `timeZone` as a string before repair or publication. Live adoption compares stored/live cwd and requires an exact live match when the stored header has a zone; an absent stored zone remains absent. Its `inspect()` path takes ownership of fresh backend values, validates and freezes them once, and retains at most the configured number of unpublished Sessions without calling `commitRepair`. A retained source is reused or repaired only when its revision still equals `readStoredRevision`; otherwise the coordinator reloads it. This freshness check does not add cross-process writer exclusion. Revision retries converge when the durable log remains unchanged for one read/check round trip; continuous external writers can delay `load`, `inspect`, or `prepare`. 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).
## Metadata and location types
Re-exported from `dsh-session`: `SessionHeader` (immutable session metadata: `version`, `id`, `createdAt`, `cwd?`, `parentSession?`, `seedLength?`, `origin?`, `delegationDepth?`). `SessionLocation` is `{ readonly kind: string; readonly path: string }`; its path is an absolute backend target, not proof that the artifact exists or contains an unflushed turn.
Re-exported from `dsh-session`: `SessionHeader` (immutable session metadata: `version`, `id`, `createdAt`, `cwd?`, `timeZone?`, `parentSession?`, `seedLength?`, `origin?`, `delegationDepth?`). `SessionLocation` is `{ readonly kind: string; readonly path: string }`; its path is an absolute backend target, not proof that the artifact exists or contains an unflushed turn.
## Model Experience

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

@@ -589,6 +589,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
if (!Number.isSafeInteger(snapshot.createdAt) || snapshot.createdAt < 0) {
return Promise.reject(new TypeError('session metadata createdAt must be a non-negative safe integer'))
}
if (snapshot.timeZone !== undefined && typeof snapshot.timeZone !== 'string') {
return Promise.reject(new TypeError('session metadata timeZone must be a string'))
}
return this.serialize(snapshot.id, () => this.createCore(snapshot))
}
@@ -801,7 +804,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
signal?.throwIfAborted()
if (suffix === undefined) throw new Error(`session "${id}" not found`)
this.assertStoredId(id, suffix.meta)
this.assertVersion(suffix.meta)
this.assertStoredHeader(suffix.meta)
if (suffix.events.some(needsLegacyPrefix)) {
const whole = await this.readStoredPrefix(id, signal)
return { meta: whole.meta, events: whole.events.filter(event => event.seq >= fromSeq) }
@@ -823,7 +826,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
signal?.throwIfAborted()
if (stored === undefined) throw new Error(`session "${id}" not found`)
this.assertStoredId(id, stored.meta)
this.assertVersion(stored.meta)
this.assertStoredHeader(stored.meta)
return {
meta: structuredClone(stored.meta),
events: snapshotStoredEvents(stored.events, id),
@@ -837,7 +840,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
try {
const { meta, events, revision, tornMarker } = stored
this.assertStoredId(id, meta)
this.assertVersion(meta)
this.assertStoredHeader(meta)
const storedEvents = adoptStoredEvents(events, id)
// Preserve complete interrupted events and synthesize only missing closers.
@@ -981,10 +984,14 @@ export class PersistenceCoordinator<TornMarker = unknown> {
}
}
private assertVersion(meta: SessionHeader): void {
/** Validate fixed fields decoded from backend-owned storage. */
private assertStoredHeader(meta: SessionHeader): void {
if (meta.version !== SESSION_FORMAT_VERSION) {
throw new Error(`unsupported session format version ${meta.version} for "${meta.id}" (only v${SESSION_FORMAT_VERSION} is supported)`)
}
if (meta.timeZone !== undefined && typeof meta.timeZone !== 'string') {
throw new Error(`stored session "${meta.id}" timeZone must be a string`)
}
}
/** Reject backend metadata that is not bound to the requested session id. */
@@ -994,6 +1001,19 @@ export class PersistenceCoordinator<TornMarker = unknown> {
}
}
/** Compare the immutable metadata fields that participate in live adoption identity. */
private assertAdoptableIdentity(meta: SessionHeader, session: Session): void {
this.assertStoredHeader(meta)
if (meta.cwd !== session.header.cwd) {
throw new Error(`session "${session.header.id}" is already persisted at a different cwd (persisted: ${String(meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`)
}
// A stored headerless session is the one compatibility case: it remains
// headerless even if a current caller supplied a zone for the live object.
if (meta.timeZone !== undefined && meta.timeZone !== session.header.timeZone) {
throw new Error(`session "${session.header.id}" is already persisted with a different timeZone (persisted: ${meta.timeZone}, live: ${String(session.header.timeZone)}) (id collision)`)
}
}
// --- write path (session/event → flush drain) ---
private installWritePath(): void {
@@ -1161,9 +1181,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
// the stored header's cwd. The seed guard then ensures the live events
// reproduce the persisted prefix; otherwise a fresh session reusing the
// id could have its leading events filtered as already written.
if (tracked.meta.cwd !== session.header.cwd) {
throw new Error(`session "${id}" is already persisted at a different cwd (persisted: ${String(tracked.meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`)
}
this.assertAdoptableIdentity(tracked.meta, session)
if (!await this.seedMatchesPersisted(id, seed, tracked.cursor)) {
throw new Error(`session "${id}" is already persisted with ${tracked.cursor} event(s) that do not match this live session (id collision)`)
}
@@ -1214,10 +1232,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
private async adoptLivePrefix(session: Session, seed: readonly SessionEvent[], stored: StoredPrefix<TornMarker>): Promise<void> {
const { meta, events, tornMarker } = stored
this.assertStoredId(session.header.id, meta)
if (meta.cwd !== session.header.cwd) {
throw new Error(`session "${session.header.id}" is already persisted at a different cwd (persisted: ${String(meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`)
}
this.assertVersion(meta)
this.assertAdoptableIdentity(meta, session)
const storedEvents = snapshotStoredEvents(events, session.header.id)
if (!seedCoversPrefix(seed, storedEvents)) {
throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`)

37
packages/session/session-persistence/tests/contract.ts
View File

@@ -21,12 +21,13 @@ export interface ContractBackend {
}
/** Build a minimal {@link SessionHeader} for a session id. */
export function meta(id: string, cwd?: string): SessionHeader {
export function meta(id: string, cwd?: string, timeZone?: string): SessionHeader {
return {
version: SESSION_FORMAT_VERSION,
id: SessionId(id),
createdAt: 1000,
...cwd !== undefined ? { cwd } : {},
...timeZone !== undefined ? { timeZone } : {},
}
}
@@ -86,19 +87,49 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
it('round-trips a session: create + append → load returns identical meta and byte-identical events', async () => {
const { persistence, dispose } = await make()
try {
const m = meta('s1', '/work')
const m = meta('s1', '/work', 'Asia/Shanghai')
const log = oneTurnLog()
await persistence.create(m)
await persistence.append(m.id, log)
const loaded = await persistence.load(m.id)
expect(loaded.meta).toMatchObject({ version: SESSION_FORMAT_VERSION, id: m.id, cwd: '/work' })
expect(loaded.meta).toMatchObject(m)
expect(loaded.events).toEqual(log)
} finally {
await dispose()
}
})
it('keeps a headerless session headerless across storage reads', async () => {
const { persistence, dispose } = await make()
try {
const m = meta('headerless', '/work')
await persistence.create(m)
await persistence.append(m.id, oneTurnLog())
expect((await persistence.inspect(m.id)).meta.timeZone).toBeUndefined()
expect((await persistence.load(m.id)).meta.timeZone).toBeUndefined()
expect((await persistence.list()).find(header => header.id === m.id)?.timeZone).toBeUndefined()
} finally {
await dispose()
}
})
it('rejects non-string timeZone metadata without reserving its session id', async () => {
const { persistence, dispose } = await make()
try {
const invalid = { ...meta('invalid-time-zone'), timeZone: 1 as unknown as string }
await expect(persistence.create(invalid)).rejects.toThrow('session metadata timeZone must be a string')
const valid = meta('invalid-time-zone', undefined, 'UTC')
await persistence.create(valid)
await persistence.append(valid.id, oneTurnLog())
expect((await persistence.load(valid.id)).meta.timeZone).toBe('UTC')
} finally {
await dispose()
}
})
it('rejects a fractional creation timestamp without reserving its session id', async () => {
const { persistence, dispose } = await make()
try {

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

@@ -908,6 +908,69 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
}
})
it('stored-prefix adoption rejects a different present timeZone', async () => {
const fix = await makeFixture()
const first = await freshCtx(fix)
try {
const stored = first.ctx.sessions.create(SessionId('zone-adoption'), {
meta: { cwd: WORK, timeZone: 'Asia/Shanghai' },
})
send(stored, oneTurnLog())
await first.ctx.sessions.flush(stored)
} finally {
await first.fiber.dispose()
}
const ctx = new Context()
await ctx.plugin(SessionStore)
const live = ctx.sessions.create(SessionId('zone-adoption'), {
seed: oneTurnLog(),
meta: { cwd: WORK, timeZone: 'America/New_York' },
})
const second = await fix.mount(ctx)
try {
await expect(ctx.sessions.flush(live)).rejects.toThrow(/different timeZone|id collision/)
} finally {
await second.dispose()
await ctx.fiber.dispose()
await fix.cleanup()
}
})
it('stored-prefix adoption keeps a headerless record headerless for a zoned live session', async () => {
const fix = await makeFixture()
const log = [
...oneTurnLog(),
{ type: 'session/end-seed', seq: 6, time: 7, data: {} },
] as SessionEvent[]
const first = await freshCtx(fix)
try {
const stored = first.ctx.sessions.create(SessionId('headerless-zone-adoption'), {
seed: log,
meta: { cwd: WORK },
})
await first.ctx.sessions.flush(stored)
} finally {
await first.fiber.dispose()
}
const ctx = new Context()
await ctx.plugin(SessionStore)
const live = ctx.sessions.create(SessionId('headerless-zone-adoption'), {
seed: log,
meta: { cwd: WORK, timeZone: 'Asia/Shanghai' },
})
const second = await fix.mount(ctx)
try {
await expect(ctx.sessions.flush(live)).resolves.toBe(true)
expect((await ctx.sessionPersistence.load(live.id)).meta.timeZone).toBeUndefined()
} finally {
await second.dispose()
await ctx.fiber.dispose()
await fix.cleanup()
}
})
it('HMR: adoption persists the live SUFFIX that was ahead of the stored prefix', async () => {
const fix = await makeFixture()
const ctx = new Context()
@@ -1104,6 +1167,55 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
}
})
it('a zoned live session claims headerless ownerless state without backfilling it', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
await ctx.sessionPersistence.create(meta('headerless-zone-claim', WORK))
const live = ctx.sessions.create(SessionId('headerless-zone-claim'), {
seed: oneTurnLog(),
meta: { cwd: WORK, timeZone: 'Asia/Shanghai' },
})
await expect(ctx.sessions.flush(live)).resolves.toBe(true)
expect((await ctx.sessionPersistence.load(live.id)).meta.timeZone).toBeUndefined()
} finally {
await fiber.dispose()
await fix.cleanup()
}
})
it('ownerless state with a timeZone only accepts the same live identity', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
await ctx.sessionPersistence.create(meta('same-zone-claim', WORK, 'Asia/Shanghai'))
const matching = ctx.sessions.create(SessionId('same-zone-claim'), {
seed: oneTurnLog(),
meta: { cwd: WORK, timeZone: 'Asia/Shanghai' },
})
await expect(ctx.sessions.flush(matching)).resolves.toBe(true)
expect((await ctx.sessionPersistence.load(matching.id)).meta.timeZone).toBe('Asia/Shanghai')
await ctx.sessionPersistence.create(meta('different-zone-claim', WORK, 'Asia/Shanghai'))
const conflicting = ctx.sessions.create(SessionId('different-zone-claim'), {
seed: oneTurnLog(),
meta: { cwd: WORK, timeZone: 'America/New_York' },
})
await expect(ctx.sessions.flush(conflicting)).rejects.toThrow(/different timeZone|id collision/)
await ctx.sessionPersistence.create(meta('missing-zone-claim', WORK, 'Asia/Shanghai'))
const missing = ctx.sessions.create(SessionId('missing-zone-claim'), {
seed: oneTurnLog(),
meta: { cwd: WORK },
})
await expect(ctx.sessions.flush(missing)).rejects.toThrow(/different timeZone|id collision/)
} finally {
await fiber.dispose()
await fix.cleanup()
}
})
it('a fresh session reusing a previously-loaded id is rejected (ownerless guard)', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)

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

@@ -374,6 +374,28 @@ describe('PersistenceCoordinator bounded writes', () => {
})
describe('PersistenceCoordinator stored identity', () => {
it('rejects a non-string timeZone decoded by a backend', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const backend = new ControlledBackend()
const id = SessionId('invalid-stored-zone')
backend.store.set(id, {
meta: { ...meta(id), timeZone: 1 as unknown as string },
events: [],
})
let coordinator!: PersistenceCoordinator<never>
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
coordinator = new PersistenceCoordinator(inner, backend)
}, { inject: ['sessions'] }))
try {
await expect(coordinator.inspect(id)).rejects.toThrow(/stored session .* timeZone must be a string/)
expect((coordinator as unknown as CoordinatorInternals).states.size).toBe(0)
} finally {
await fiber.dispose()
await ctx.fiber.dispose()
}
})
it('rejects a mismatched backend header before repair or state publication', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)