Merge latest master into subagent policy inheritance
Retarget the feature branch to the current master tip without rewriting its existing review history. Keeping this as a dedicated merge checkpoint makes the later simplification diff attributable to the stacked child rather than mixing base movement with design changes. Resolve the identified-message API drift in the feature tests by constructing complete user messages, reading the nested tool-result message shape, and adapting the prompt-submit listener signature. Preserve both sides of the user-approval conflict: master’s createUserMessage wrapper and the feature’s inherited-policy attribution. Regenerate the Cordis and persistence catalogs, re-record the session README pair, and refresh the affected ACP/headless fixtures so derived artifacts describe the merged source rather than either parent in isolation. Validated with the focused policy/session/persistence/query suites (430 tests), focused ACP/headless snapshots (3 tests), build, doc-sync (25 gates), lint, hygiene, and git diff checks.
This commit is contained in:
Tianyi Cui
@@ -102,9 +102,9 @@ describe.skipIf(process.platform === 'win32')('semantic checkpoint hard-crash re
|
||||
expect(result?.type === 'tool/result' && result.data.error).toEqual({
|
||||
name: 'ToolOutcomeUnknownError', code: TOOL_OUTCOME_UNKNOWN,
|
||||
})
|
||||
if (result?.type !== 'tool/result' || result.data.content[0]?.type !== 'text') {
|
||||
if (result?.type !== 'tool/result' || result.data.message.content[0].content[0]?.type !== 'text') {
|
||||
throw new Error('expected a text tool result')
|
||||
}
|
||||
expect(result.data.content[0].text).toContain('Do not retry blindly.')
|
||||
expect(result.data.message.content[0].content[0].text).toContain('Do not retry blindly.')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -2,7 +2,7 @@ import { writeFile } from 'node:fs/promises'
|
||||
import { Context } from 'cordis'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
|
||||
import { CallId, type GenerateOptions, LlmAdapter, type StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { createUserMessage, CallId, type GenerateOptions, LlmAdapter, type StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import * as checkpointPolicy from '../../src/index.ts'
|
||||
@@ -56,5 +56,5 @@ const handle = await ctx.agents.create({
|
||||
sessionId: SessionId('semantic-checkpoint-crash'),
|
||||
agentOptions: { provider: 'crash', model: 'crash' },
|
||||
})
|
||||
handle.agent.followup({ content: [{ type: 'text', text: 'exercise the crash boundary' }], source: { kind: 'user' } })
|
||||
handle.agent.followup(createUserMessage({ content: [{ type: 'text', text: 'exercise the crash boundary' }], source: { kind: 'user' } }))
|
||||
await waitForCrash()
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { MessageId, createUserMessage, createMessage } from '@deepseek-ai/dsh-llm'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat, symlink } from 'node:fs/promises'
|
||||
@@ -59,10 +60,10 @@ afterEach(async () => {
|
||||
|
||||
function appendClosedTurn(session: Session): void {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', {
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'hello' }],
|
||||
source: { kind: 'user' },
|
||||
}, { surfaceOp: 'append' })
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
}
|
||||
|
||||
@@ -211,7 +212,17 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
{ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/chunk', seq: 2, time: 3, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'he' } } },
|
||||
{ type: 'assistant/chunk', seq: 3, time: 4, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'llo' } } },
|
||||
{ type: 'assistant/message', seq: 4, time: 5, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append', sourceEventSeqs: [2, 3] },
|
||||
{ type: 'assistant/message', seq: 4, time: 5, data: {
|
||||
turn: 1, step: 1,
|
||||
message: createMessage({
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: 'hello' }],
|
||||
source: {
|
||||
kind: 'model',
|
||||
...{ provider: 'mock', model: 'mock' },
|
||||
},
|
||||
}),
|
||||
}, surfaceOp: 'append', sourceEventSeqs: [2, 3] },
|
||||
{ type: 'step/end', seq: 5, time: 6, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 6, time: 7, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
@@ -587,8 +598,12 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () =>
|
||||
const b = ctx.sessions.create(SessionId('sb'))
|
||||
a.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
b.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
a.append('user/message', { content: [{ type: 'text', text: 'A' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
b.append('user/message', { content: [{ type: 'text', text: 'B' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
a.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'A' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
b.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'B' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
a.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
b.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.sessions.flush(a)
|
||||
@@ -748,7 +763,17 @@ describe('SessionPersistenceJsonl: default packed chunk rows', () => {
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
|
||||
...deltas,
|
||||
{ type: 'assistant/message', seq: 7, time: 8, data: { turn: 1, step: 1, content: [{ type: 'text', text: 't0t1t2t3t4' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append', sourceEventSeqs: [2, 3, 4, 5, 6] },
|
||||
{ type: 'assistant/message', seq: 7, time: 8, data: {
|
||||
turn: 1, step: 1,
|
||||
message: createMessage({
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: 't0t1t2t3t4' }],
|
||||
source: {
|
||||
kind: 'model',
|
||||
...{ provider: 'mock', model: 'mock' },
|
||||
},
|
||||
}),
|
||||
}, surfaceOp: 'append', sourceEventSeqs: [2, 3, 4, 5, 6] },
|
||||
{ type: 'step/end', seq: 8, time: 9, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 9, time: 10, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
@@ -1104,9 +1129,11 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
// A seed that keeps every seq/type/time but mutates a payload must NOT be
|
||||
// accepted as "the same session" — otherwise drain filters those seqs as
|
||||
// already persisted and the divergent payload is silently lost.
|
||||
const tampered = oneTurnLog()
|
||||
const tampered = structuredClone(oneTurnLog())
|
||||
const userMsg = tampered[1]
|
||||
if (userMsg?.type === 'user/message') userMsg.data.content = [{ type: 'text', text: 'DIFFERENT' }]
|
||||
if (userMsg?.type === 'user/message') {
|
||||
(userMsg.data as { content: unknown[] }).content = [{ type: 'text', text: 'DIFFERENT' }]
|
||||
}
|
||||
let bad!: Session
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
bad = inner.sessions.create(SessionId('divergent'), { seed: tampered, meta: { cwd: '/a' } })
|
||||
@@ -1244,7 +1271,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
const session = ctx2.sessions.create(SessionId('flush-fail'))
|
||||
// A full turn lands in the write-behind buffer.
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
// Make the durable materialize fail on the next flush.
|
||||
const backend = ctx2.sessionPersistence as unknown as { materialize: (...args: unknown[]) => Promise<void> }
|
||||
@@ -1262,7 +1291,18 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
it('rejects non-JSON event data: BigInt, function, circular, Map, undefined property', async () => {
|
||||
const m = meta('serial')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
const bad = (extra: unknown) => [{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra } }] as unknown as SessionEvent[]
|
||||
const bad = (extra: unknown) => [{
|
||||
type: 'user/message',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: {
|
||||
id: MessageId('invalid-json'),
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: 'x' }],
|
||||
source: { kind: 'user' },
|
||||
extra,
|
||||
},
|
||||
}] as unknown as SessionEvent[]
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(1n))).rejects.toThrow(/non-JSON-serializable/)
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(() => 0))).rejects.toThrow(/non-JSON-serializable/)
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(Symbol('s')))).rejects.toThrow(/non-JSON-serializable/)
|
||||
@@ -1280,7 +1320,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
it('accepts well-formed JSON values (null, booleans, nested arrays/objects)', async () => {
|
||||
const m = meta('json-ok')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
const ev = [{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: { a: null, b: true, c: [1, 2, { d: 'nested' }] } } }] as unknown as SessionEvent[]
|
||||
const ev = [{ type: 'user/message', seq: 0, time: 1, data: createUserMessage({
|
||||
content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: { a: null, b: true, c: [1, 2, { d: 'nested' }] },
|
||||
}) }] as unknown as SessionEvent[]
|
||||
await ctx.sessionPersistence.append(m.id, ev)
|
||||
expect((await ctx.sessionPersistence.list()).map(h => h.id)).toContain(m.id)
|
||||
})
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { createUserMessage, createMessage } from '@deepseek-ai/dsh-llm'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { existsSync } from 'node:fs'
|
||||
@@ -296,7 +297,9 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
// A first turn that NEVER completed: turn/start + user/message, no turn/end.
|
||||
await b1.ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } } },
|
||||
{ type: 'user/message', seq: 1, time: 2, data: createUserMessage({
|
||||
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
|
||||
}) },
|
||||
])
|
||||
await b1.dispose()
|
||||
|
||||
@@ -815,8 +818,20 @@ describe('surface field round-trip', () => {
|
||||
const session = ctx.sessions.create(SessionId('roundtrip-surface'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('step/start', { turn: 1, step: 1 })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: 'append', sourceEventSeqs: [2] })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('assistant/message', {
|
||||
turn: 1, step: 1,
|
||||
message: createMessage({
|
||||
role: 'assistant',
|
||||
content: [],
|
||||
source: {
|
||||
kind: 'model',
|
||||
...{ provider: 'mock', model: 'mock' },
|
||||
},
|
||||
}),
|
||||
}, { surfaceOp: 'append', sourceEventSeqs: [2] })
|
||||
session.append('step/end', { turn: 1, step: 1 })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.sessions.flush(session)
|
||||
@@ -837,7 +852,13 @@ describe('surface field round-trip', () => {
|
||||
const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' })
|
||||
const session = ctx.sessions.create(SessionId('surface-noseq'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('steering/message', { turn: 1, content: [], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('steering/message', {
|
||||
turn: 1,
|
||||
message: createUserMessage({
|
||||
content: [],
|
||||
source: { kind: 'user' },
|
||||
}),
|
||||
}, { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.sessions.flush(session)
|
||||
const loaded = await ctx.sessionPersistence.load(SessionId('surface-noseq'))
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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. |
|
||||
|
||||
|
||||
@@ -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 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端在拒绝前结算已启动列表工作，使已等待调用完全停稳。 |
|
||||
|
||||
|
||||
@@ -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'
|
||||
|
||||
/**
|
||||
@@ -167,6 +172,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
|
||||
@@ -333,10 +344,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,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -346,11 +357,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) {
|
||||
@@ -358,12 +369,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 */
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { SESSION_FORMAT_VERSION, Session, SessionId, TOOL_NOT_STARTED, TOOL_OUTCOME_UNKNOWN } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent, SessionHeader, SurfaceEventType, SurfaceIntent } from '@deepseek-ai/dsh-session'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId, MessageId, createMessage, freezeMessage } from '@deepseek-ai/dsh-llm'
|
||||
import type { SessionPersistence } from '../src/index.ts'
|
||||
|
||||
/** A backend under test plus its teardown. */
|
||||
@@ -34,9 +34,24 @@ export function meta(id: string, cwd?: string): SessionHeader {
|
||||
export function oneTurnLog(): SessionEvent[] {
|
||||
return [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, surfaceOp: 'append' },
|
||||
{ type: 'user/message', seq: 1, time: 2, data: freezeMessage({
|
||||
id: MessageId('one-turn-user'),
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
|
||||
}), surfaceOp: 'append' },
|
||||
{ type: 'step/start', seq: 2, time: 3, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 3, time: 4, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append' },
|
||||
{ type: 'assistant/message', seq: 3, time: 4, data: {
|
||||
turn: 1, step: 1,
|
||||
message: freezeMessage({
|
||||
id: MessageId('one-turn-assistant'),
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: 'hello' }],
|
||||
source: {
|
||||
kind: 'model',
|
||||
...{ provider: 'mock', model: 'mock' },
|
||||
},
|
||||
}),
|
||||
}, surfaceOp: 'append' },
|
||||
{ type: 'step/end', seq: 4, time: 5, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 5, time: 6, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
@@ -194,9 +209,19 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
await persistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 8, time: 9, data: { turn: 2, step: 1, content: [
|
||||
{ type: 'tool-call', id: CallId('call-x'), name: 'bash', arguments: '{}' },
|
||||
], provenance: { provider: 'mock', model: 'mock' } } },
|
||||
{ type: 'assistant/message', seq: 8, time: 9, data: {
|
||||
turn: 2, step: 1,
|
||||
message: createMessage({
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{ type: 'tool-call', id: CallId('call-x'), name: 'bash', arguments: '{}' },
|
||||
],
|
||||
source: {
|
||||
kind: 'model',
|
||||
...{ provider: 'mock', model: 'mock' },
|
||||
},
|
||||
}),
|
||||
} },
|
||||
])
|
||||
|
||||
const loaded = await persistence.load(m.id)
|
||||
@@ -209,13 +234,17 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
])
|
||||
const synthetic = loaded.events.find(e => e.type === 'tool/result')
|
||||
expect(synthetic?.type === 'tool/result' && synthetic.data).toMatchObject({
|
||||
callId: CallId('call-x'), isError: true, error: { code: TOOL_NOT_STARTED },
|
||||
message: {
|
||||
source: { kind: 'tool', callId: CallId('call-x') },
|
||||
content: [{ type: 'tool-result', toolCallId: CallId('call-x'), isError: true }],
|
||||
},
|
||||
error: { code: TOOL_NOT_STARTED },
|
||||
})
|
||||
// The synthetic result carries the SAME callId as the orphaned tool-call,
|
||||
// so deriveMessages() pairs them — no provider-invalid dangling call.
|
||||
const call = loaded.events.findLast(e => e.type === 'assistant/message')
|
||||
const callId = call?.type === 'assistant/message'
|
||||
&& call.data.content.find(b => b.type === 'tool-call')
|
||||
&& call.data.message.content.find(b => b.type === 'tool-call')
|
||||
expect(callId && callId.type === 'tool-call' && callId.id).toBe(CallId('call-x'))
|
||||
} finally {
|
||||
await dispose()
|
||||
@@ -230,9 +259,19 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
await persistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/message', seq: 2, time: 3, data: { turn: 1, step: 1, content: [
|
||||
{ type: 'tool-call', id: CallId('call-risk'), name: 'write', arguments: '{}' },
|
||||
], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append' },
|
||||
{ type: 'assistant/message', seq: 2, time: 3, data: {
|
||||
turn: 1, step: 1,
|
||||
message: createMessage({
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{ type: 'tool-call', id: CallId('call-risk'), name: 'write', arguments: '{}' },
|
||||
],
|
||||
source: {
|
||||
kind: 'model',
|
||||
...{ provider: 'mock', model: 'mock' },
|
||||
},
|
||||
}),
|
||||
}, surfaceOp: 'append' },
|
||||
{ type: 'tool/call', seq: 3, time: 4, data: { turn: 1, step: 1, callId: CallId('call-risk'), name: 'write', arguments: '{}' } },
|
||||
])
|
||||
|
||||
@@ -241,11 +280,11 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
expect(synthetic?.type === 'tool/result' && synthetic.data.error).toEqual({
|
||||
name: 'ToolOutcomeUnknownError', code: TOOL_OUTCOME_UNKNOWN,
|
||||
})
|
||||
if (synthetic?.type !== 'tool/result' || synthetic.data.content[0]?.type !== 'text') {
|
||||
if (synthetic?.type !== 'tool/result' || synthetic.data.message.content[0].content[0]?.type !== 'text') {
|
||||
throw new Error('expected a text tool result')
|
||||
}
|
||||
expect(synthetic.data.content[0].text).toContain('retry only if the operation is read-only or idempotent')
|
||||
expect(synthetic.data.content[0].text).toContain('if it may have side effects, first verify external state or ask the user')
|
||||
expect(synthetic.data.message.content[0].content[0].text).toContain('retry only if the operation is read-only or idempotent')
|
||||
expect(synthetic.data.message.content[0].content[0].text).toContain('if it may have side effects, first verify external state or ask the user')
|
||||
const resumed = new Session(m.id, loaded.events, loaded.meta)
|
||||
const resumedResult = resumed.deriveMessages().find(message => message.content.some(block => block.type === 'tool-result'))
|
||||
expect(resumedResult?.content[0]).toMatchObject({
|
||||
@@ -379,7 +418,18 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
const mi = meta(`s5-${i}`)
|
||||
await persistence.create(mi)
|
||||
const events = [
|
||||
{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: bad } },
|
||||
{
|
||||
type: 'user/message',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: {
|
||||
id: MessageId(`invalid-json-${i}`),
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: 'x' }],
|
||||
source: { kind: 'user' },
|
||||
extra: bad,
|
||||
},
|
||||
},
|
||||
] as unknown as SessionEvent[]
|
||||
await expect(persistence.append(mi.id, events)).rejects.toThrow(/losslessly JSON-serializable/)
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
||||
/**
|
||||
* Shared write-path orchestration contract for backends using {@link PersistenceCoordinator}.
|
||||
* Unlike the public storage-semantics suite in `contract.ts`, it covers SessionStore event wiring,
|
||||
@@ -220,7 +221,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
try {
|
||||
const session = ctx.sessions.create(SessionId('mutate'), { meta: { cwd: WORK } })
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
const ev = session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const ev = session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'original' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
expect(() => {
|
||||
;(ev.data as { content: { type: 'text'; text: string }[] }).content[0]!.text = 'HACKED'
|
||||
}).toThrow(TypeError)
|
||||
@@ -236,19 +239,80 @@ 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)
|
||||
try {
|
||||
const m = meta('snapshot', WORK)
|
||||
await ctx.sessionPersistence.create(m)
|
||||
const events = oneTurnLog() // seqs 0..5
|
||||
const events = structuredClone(oneTurnLog()) // seqs 0..5
|
||||
const userMsg = events[1] // the user/message event
|
||||
const p = ctx.sessionPersistence.append(m.id, events)
|
||||
// Mutate the caller's array AND an event object after the call but before
|
||||
// the queued op runs: the snapshot taken at call time must shield the copy.
|
||||
events.push({ type: 'turn/start', seq: 6, time: 99, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } })
|
||||
if (userMsg?.type === 'user/message') userMsg.data.content = [{ type: 'text', text: 'MUTATED' }]
|
||||
if (userMsg?.type === 'user/message') {
|
||||
(userMsg.data as { content: unknown[] }).content = [{ type: 'text', text: 'MUTATED' }]
|
||||
}
|
||||
await p
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5]) // not 0..6
|
||||
@@ -321,7 +385,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
// A session exists BEFORE the persistence plugin is applied.
|
||||
const session = ctx.sessions.create(SessionId('pre-existing'), { meta: { cwd: WORK } })
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
|
||||
const fiber = await fix.mount(ctx)
|
||||
@@ -343,7 +409,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
const fiber = await fix.mount(ctx)
|
||||
const session = await liveSessionInFiber(ctx, 'drain', WORK)
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'buffered' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'buffered' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
// No explicit flush — dispose must drain.
|
||||
await fiber.dispose()
|
||||
@@ -369,7 +437,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
// Backend instance 1 materializes the session.
|
||||
const backend1 = await fix.mount(ctx)
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.sessions.flush(session)
|
||||
|
||||
@@ -379,7 +449,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
await backend1.dispose()
|
||||
await fix.mount(ctx)
|
||||
session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'again' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'again' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
|
||||
await expect(ctx.sessions.flush(session)).resolves.not.toThrow()
|
||||
|
||||
@@ -549,7 +621,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
try {
|
||||
const session = ctx.sessions.create(SessionId('idem'), { meta: { cwd: WORK } })
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'x' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.sessions.flush(session)
|
||||
// Re-emit session/created for the SAME live session (idempotent initFor).
|
||||
@@ -883,7 +957,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
|
||||
// state-undefined cursor path).
|
||||
const session = ctx.sessions.create(SessionId('flush-nostate'), { meta: { cwd: WORK } })
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('user/message', createUserMessage({
|
||||
content: [{ type: 'text', text: 'q' }], source: { kind: 'user' },
|
||||
}), { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.sessions.flush(session)
|
||||
const loaded = await ctx.sessionPersistence.load(SessionId('flush-nostate'))
|
||||
|
||||
Reference in New Issue
Block a user