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

fix: enforce message snapshot invariants

Browse Source
This commit is contained in:
_Kerman
2026-07-28 15:33:00 +08:00
parent 0a3d38bb08
commit b1af35145b
34 changed files with 417 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/session-persistence/session-persistence/README.i18n.yaml
View File

@@ -1,6 +1,6 @@
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write
README.md: c99905e8aca0bdaf810de34841ea277b105a9d0f
README.zh.md: 106c28c5f9cd4330cf69b1648b669a04392e4e8a
# pnpm run verify-translation-pairing --write packages/session-persistence/session-persistence/README.md
README.md: 08d8adac8040747a6dac01dbc41525073f17060c
README.zh.md: 7676f27a1aa934eb3472e1b32b9ecd55d460fb63

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

@@ -13,8 +13,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. 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, signal?): 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; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. |
| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log whose events are detached and validated and whose identified messages are deeply frozen. 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, malformed messages, and unknown `version` reject. |
| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix with validated, deeply frozen identified messages, without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. |
| `list(signal?): Promise<SessionHeader[]>` | Lightweight listing from metadata, no full-log parse. The optional signal cancels backend listing work. A zero-event lazily-materialized session is absent from `list`. |
| `listSnapshots(signal?): 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. The optional signal requests cancellation of backend discovery work; first-party backends settle any started listing work before rejecting so an awaited call is quiescent. |

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

@@ -13,8 +13,8 @@
| `locate(meta): SessionLocation \| undefined` | 在不执行 I/O 或实体化的情况下解析绝对的每会话产物目标。没有独立本地产物的后端返回 `undefined`。 |
| `create(meta): Promise<void>` | 注册新会话元数据。可以将物理写入延迟到第一次 `append`（延迟实体化）。 |
| `append(id, events): Promise<void>` | 持久保存一个批次。仅追加；任何修复后，第一个事件 `seq` == 已存储 next-seq；非 JSON 可序列化数据会被拒绝，并命名违规类型。 |
| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续日志。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏和未知 `version` 会被拒绝。 |
| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 |
| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续的日志，其中事件已脱离并验证，带标识的消息已深度冻结。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏、格式错误的消息和未知 `version` 会被拒绝。 |
| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，其中带标识的消息已经验证并深度冻结；不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 |
| `list(signal?): Promise<SessionHeader[]>` | 从元数据轻量列出，不解析完整日志。可选信号取消后端列表工作。零事件延迟实体化会话不在 `list` 中。 |
| `listSnapshots(signal?): Promise<SessionPersistenceSnapshot[]>` | 返回轻量元数据和不透明品牌化每日志修订，不加载事件日志。日志及其后端存储不变时，修订保持相等；append 或变更性 load 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端在拒绝前结算已启动列表工作，使已等待调用完全停稳。 |

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

@@ -6,7 +6,12 @@
*/
import { Context } from 'cordis'
import { interruptedTurnClosers, SESSION_FORMAT_VERSION, snapshotJsonValue } from '@deepseek-ai/dsh-session'
import {
interruptedTurnClosers,
SESSION_FORMAT_VERSION,
snapshotJsonValue,
snapshotSessionEvent,
} from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
/**
@@ -141,6 +146,12 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId):
}
}
/** Materialize stored events as validated snapshots with immutable messages. */
function snapshotStoredEvents(events: readonly SessionEvent[], id: SessionId): SessionEvent[] {
assertSupportedEvents(events, id)
return events.map(snapshotSessionEvent)
}
/**
* Owns the backend-agnostic session write-path orchestration. A backend
* constructs one (`new PersistenceCoordinator(ctx, this)`), implements
@@ -307,10 +318,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
if (stored === undefined) throw new Error(`session "${id}" not found`)
this.assertStoredId(id, stored.meta)
this.assertVersion(stored.meta)
assertSupportedEvents(stored.events, id)
const events = snapshotStoredEvents(stored.events, id)
return {
meta: structuredClone(stored.meta),
events: structuredClone(stored.events),
events,
}
}
@@ -320,11 +331,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
const { meta, events, tornMarker } = stored
this.assertStoredId(id, meta)
this.assertVersion(meta)
assertSupportedEvents(events, id)
const storedEvents = snapshotStoredEvents(events, id)
// Preserve complete interrupted events and synthesize only missing closers.
const closers = interruptedTurnClosers(events)
const balanced = [...events, ...closers]
const closers = interruptedTurnClosers(storedEvents).map(snapshotSessionEvent)
const balanced = [...storedEvents, ...closers]
// Repair storage before publishing coordinator state.
if (tornMarker !== undefined || closers.length > 0) {
@@ -332,12 +343,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
}
// Keep coordinator metadata detached from the returned record.
this.states.set(id, { meta: { ...meta }, cursor: balanced.length, materialized: true })
return { meta, events: balanced }
return { meta: structuredClone(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))
const events = session.events.map(snapshotSessionEvent)
await this.flush(session)
const state = this.states.get(session.id)
/* v8 ignore next -- successful flush always publishes this live session's durable state */

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

@@ -92,6 +92,8 @@ export abstract class SessionPersistence extends Service {
* 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.
* Returned events are detached, and every identified message is deeply
* frozen; malformed identified messages reject before any stored event is returned.
* @param id - the persisted session to reload.
* @returns the header and a log ending on a balanced `turn/end`.
*/
@@ -101,7 +103,8 @@ export abstract class SessionPersistence extends Service {
* Inspect a header and its valid contiguous stored prefix without repairing
* a torn tail, closing an interrupted turn, or publishing coordinator state.
* This read is serialized with writes for the same id and returns detached
* values, so observers cannot mutate backend-owned state.
* values with deeply frozen identified messages, so observers cannot mutate message
* identity/content or backend-owned state. Malformed identified messages reject.
* @param id - the persisted session to inspect.
* @param signal - optional cancellation for queued and backend read work.
* @returns the header and valid stored event prefix exactly as observed.

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

@@ -239,6 +239,65 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
}
})
it('load and inspect return immutable identified-message snapshots', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
const id = SessionId('immutable-read')
const session = ctx.sessions.create(id, { meta: { cwd: WORK } })
send(session, oneTurnLog())
await ctx.sessions.flush(session)
for (const snapshot of [
await ctx.sessionPersistence.load(id),
await ctx.sessionPersistence.inspect(id),
]) {
const event = snapshot.events.find(candidate => candidate.type === 'user/message')
if (event?.type !== 'user/message') throw new Error('fixture lacks user/message')
expect(Object.isFrozen(event.data)).toBe(true)
expect(Object.isFrozen(event.data.content)).toBe(true)
expect(() => {
;(event.data as { id: string }).id = 'rewritten'
}).toThrow(TypeError)
expect(() => {
;(event.data.content[0] as { type: 'text'; text: string }).text = 'rewritten'
}).toThrow(TypeError)
}
} finally {
await fiber.dispose()
await fix.cleanup()
}
})
it('rejects malformed persisted message events before returning them', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
const id = SessionId('invalid-message-read')
await ctx.sessionPersistence.create(meta(id, WORK))
await ctx.sessionPersistence.append(id, [{
type: 'user/message',
seq: 0,
time: 1,
surfaceOp: 'append',
data: {
id: 'wrong-role',
role: 'assistant',
content: [{ type: 'text', text: 'wrong' }],
source: { kind: 'user' },
},
} as unknown as SessionEvent])
await expect(ctx.sessionPersistence.inspect(id))
.rejects.toThrow('message must have role "user"')
await expect(ctx.sessionPersistence.load(id))
.rejects.toThrow('message must have role "user"')
} finally {
await fiber.dispose()
await fix.cleanup()
}
})
it('append snapshots the batch: mutating the caller array/events after the call is ignored', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)