From 618a7a46e83629f8dbe5e2bd2453610bf619f042 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Thu, 3 Sep 2026 13:07:26 +0800 Subject: [PATCH] fix(session): accept released migration data --- .../session-format-v0-to-v1/README.i18n.yaml | 4 +- .../session/session-format-v0-to-v1/README.md | 2 +- .../session-format-v0-to-v1/README.zh.md | 2 +- .../src/dispositions.ts | 44 ++---------- .../src/payload-validation.ts | 4 +- .../tests/relationships.spec.ts | 2 + .../tests/validation.spec.ts | 25 +++++-- .../src/generation.ts | 27 ++++++-- .../tests/generation.spec.ts | 69 ++++++++++++++++++- 9 files changed, 123 insertions(+), 56 deletions(-) diff --git a/packages/session/session-format-v0-to-v1/README.i18n.yaml b/packages/session/session-format-v0-to-v1/README.i18n.yaml index 6283201a79..aa1ed91da2 100644 --- a/packages/session/session-format-v0-to-v1/README.i18n.yaml +++ b/packages/session/session-format-v0-to-v1/README.i18n.yaml @@ -2,5 +2,5 @@ # 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 packages/session/session-format-v0-to-v1/README.md -README.md: 2ae4e4a20d07009766df917aca9cd46070fc3639 -README.zh.md: 64137fccae4ee144b6a0a22f95ae4aed18044643 +README.md: 1685831b49814b80979a0e54cb1a6a6d6b618a1d +README.zh.md: 4a511920f4b364d730e9db2bb59e7b43ac47c63a diff --git a/packages/session/session-format-v0-to-v1/README.md b/packages/session/session-format-v0-to-v1/README.md index 2ae4e4a20d..1685831b49 100644 --- a/packages/session/session-format-v0-to-v1/README.md +++ b/packages/session/session-format-v0-to-v1/README.md @@ -38,7 +38,7 @@ const migratedV1 = sessionFormatV0ToV1.migrate(decodedV0) `releasedV0SessionFormatCodec` reads the exact v0 header and physical rows, including packed assistant deltas and range-encoded provenance. `sessionFormatV0ToV1` normalizes and strictly validates a complete detached artifact. `releasedV1SessionFormatCodec` preserves the v1 physical layout without freezing the ordinary event vocabulary; the catalog restores current events against the installed Session package. -The alpha edge refuses every event type outside its frozen inventory, including an unknown event marked `ignorable: true`. It also refuses unexpected payload members. `tool/result.meta` and nested PTC `arguments` remain explicit opaque JSON fields and are preserved without Session-sequence interpretation. The disposition inventory separately names merge-extensible nested discriminants: unknown content-block `type`, message-source `kind`, assistant finish-reason `kind`, and `turn/end` reason `kind` arms remain owner-opaque JSON while their known arms receive structural validation. +The alpha edge refuses every event type outside its frozen inventory, including an unknown event marked `ignorable: true`. It also refuses unexpected payload members. `tool/result.meta` and nested PTC `arguments` remain explicit opaque JSON fields and are preserved without Session-sequence interpretation. Unknown content-block `type`, message-source `kind`, assistant finish-reason `kind`, and `turn/end` reason `kind` arms remain owner-opaque JSON while their known arms receive structural validation. The bounded historical normalizers convert `steering/message` to `user/message`, remove `turn/start.trigger`, convert retired `turn/end` reasons, add the current message wrappers and deterministic legacy message ids, and remove the obsolete `request/header.header.messagePrefix` duplicate. Retired `request/header-delta`, `mode/set`, and the `request/header` fallback reason refuse migration. No other event, reference, source, or payload fact may change. diff --git a/packages/session/session-format-v0-to-v1/README.zh.md b/packages/session/session-format-v0-to-v1/README.zh.md index 64137fccae..4a511920f4 100644 --- a/packages/session/session-format-v0-to-v1/README.zh.md +++ b/packages/session/session-format-v0-to-v1/README.zh.md @@ -38,7 +38,7 @@ const migratedV1 = sessionFormatV0ToV1.migrate(decodedV0) `releasedV0SessionFormatCodec` 读取精确的 v0 标头与物理行，包括打包的 Assistant 增量和范围编码的来源序号。`sessionFormatV0ToV1` 规范化并严格校验一个完整且分离的产物。`releasedV1SessionFormatCodec` 在不冻结普通事件词表的前提下保留 v1 物理布局；目录会根据已安装的 Session 包还原当前事件。 -Alpha 迁移边会拒绝冻结清单之外的所有事件类型，包括带有 `ignorable: true` 标记的未知事件。它也会拒绝意外的 payload 成员。`tool/result.meta` 与嵌套 PTC `arguments` 是显式的不透明 JSON 字段；迁移会原样保留它们，不把其中的数字解释为 Session 序号。处置清单会单独指明可合并扩展的嵌套判别字段：未知 content-block `type`、message-source `kind`、assistant finish-reason `kind` 与 `turn/end` reason `kind` 分支保持 owner-opaque JSON，已知分支则接受结构校验。 +Alpha 迁移边会拒绝冻结清单之外的所有事件类型，包括带有 `ignorable: true` 标记的未知事件。它也会拒绝意外的 payload 成员。`tool/result.meta` 与嵌套 PTC `arguments` 是显式的不透明 JSON 字段；迁移会原样保留它们，不把其中的数字解释为 Session 序号。未知 content-block `type`、message-source `kind`、assistant finish-reason `kind` 与 `turn/end` reason `kind` 分支保持 owner-opaque JSON，已知分支则接受结构校验。 有限的历史规范化会把 `steering/message` 转换为 `user/message`、移除 `turn/start.trigger`、转换已停用的 `turn/end` reason、添加当前消息包装层与确定性的旧消息 id，并移除已停用且重复的 `request/header.header.messagePrefix`。已停用的 `request/header-delta`、`mode/set` 和 `request/header` fallback reason 会使迁移失败。除此之外，任何事件、引用、来源或 payload 事实都不得改变。 diff --git a/packages/session/session-format-v0-to-v1/src/dispositions.ts b/packages/session/session-format-v0-to-v1/src/dispositions.ts index a64bdd81ca..827745da31 100644 --- a/packages/session/session-format-v0-to-v1/src/dispositions.ts +++ b/packages/session/session-format-v0-to-v1/src/dispositions.ts @@ -4,53 +4,40 @@ export interface ReleasedV0PayloadDisposition { readonly optional: readonly string[] /** JSON members whose nested representation is intentionally owner-opaque. */ readonly opaque: readonly string[] - /** Nested discriminant paths whose unknown arms remain owner-opaque JSON. */ - readonly extensionArms: readonly string[] } function disposition( required: readonly string[], optional: readonly string[] = [], opaque: readonly string[] = [], - extensionArms: readonly string[] = [], ): ReleasedV0PayloadDisposition { return Object.freeze({ required: Object.freeze([...required]), optional: Object.freeze([...optional]), opaque: Object.freeze([...opaque]), - extensionArms: Object.freeze([...extensionArms]), }) } /** * Frozen released-v0 event and payload-member inventory. * Every listed member is preserved by the identity edge. Members in `opaque` - * remain lossless JSON without nested Session-sequence interpretation. Paths - * in `extensionArms` validate known variants and preserve unknown variants as - * owner-opaque JSON, matching the merge-extensible runtime vocabulary. + * remain lossless JSON without nested Session-sequence interpretation. Nested + * merge-extensible discriminants validate known variants and preserve + * unknown variants as owner-opaque JSON. */ export const RELEASED_V0_EVENT_DISPOSITIONS: Readonly> = Object.freeze({ 'agent-preset/selected': disposition(['agentPreset']), 'agent/inbox/spliced': disposition( ['target', 'start', 'inserted'], ['removedCount', 'outcome'], - [], - ['inserted[].content[].type', 'inserted[].source.kind'], ), 'approval/asked': disposition(['id', 'toolName'], ['callId', 'reason']), 'approval/decided': disposition(['id', 'outcome']), 'approval/policy': disposition(['policy'], ['source']), - 'assistant/chunk': disposition( - ['turn', 'step', 'chunk'], - [], - [], - ['chunk.blockType', 'chunk.block.type', 'chunk.reason.kind'], - ), + 'assistant/chunk': disposition(['turn', 'step', 'chunk']), 'assistant/message': disposition( ['turn', 'step', 'message'], ['usage', 'interrupted'], - [], - ['message.content[].type'], ), 'command/done': disposition(['commandId', 'kind'], ['text', 'sourceEventSeq']), 'command/run': disposition(['commandId', 'name', 'source'], ['args']), @@ -60,8 +47,6 @@ export const RELEASED_V0_EVENT_DISPOSITIONS: Readonly 2_147_483_647) { + const delayMs = finiteNumberValue(data['delayMs'], `${label} delayMs`) + if (delayMs < 0) throw new SessionFormatError(`${label} delayMs must be non-negative`) + if (delayMs > 2_147_483_647) { throw new SessionFormatError(`${label} delayMs exceeds the timer range`) } llmFailureValue(data['failure'], `${label} failure`) diff --git a/packages/session/session-format-v0-to-v1/tests/relationships.spec.ts b/packages/session/session-format-v0-to-v1/tests/relationships.spec.ts index 235acd1caf..7f7ef30821 100644 --- a/packages/session/session-format-v0-to-v1/tests/relationships.spec.ts +++ b/packages/session/session-format-v0-to-v1/tests/relationships.spec.ts @@ -270,8 +270,10 @@ describe('released v1 whole-artifact relationships', () => { expect(() => decode([...prefix, retry({ retry: 2, maxRetries: 1 })])).toThrow() expect(() => decode([...prefix, retry({ provider: 'q' })])).toThrow(/provider/) expect(() => decode([...prefix, retry({ failure: { message: 'x', code: 'X', status: 99 } })])).toThrow(/status/) + expect(() => decode([...prefix, retry({ delayMs: -0.5 })])).toThrow(/non-negative/) expect(() => decode([...prefix, retry({ delayMs: 2_147_483_648 })])).toThrow(/timer/) expect(() => decode([...prefix, retry({ failure: { message: 'x', code: 'X', providerRetryAfterMs: 0 } })])).toThrow(/positive/) + expect(decode([...prefix, retry({ delayMs: 1.5 })]).events).toHaveLength(4) expect(decode([...prefix, retry({ failure: { message: 'x', code: 'X', providerRetryAfterMs: 1.5 } })]).events) .toHaveLength(4) }) diff --git a/packages/session/session-format-v0-to-v1/tests/validation.spec.ts b/packages/session/session-format-v0-to-v1/tests/validation.spec.ts index 3da7ff2878..307a9634f2 100644 --- a/packages/session/session-format-v0-to-v1/tests/validation.spec.ts +++ b/packages/session/session-format-v0-to-v1/tests/validation.spec.ts @@ -287,13 +287,24 @@ describe('released event and payload inventory', () => { } }) - it('publishes every merge-extensible nested arm as owner-opaque policy', () => { - expect(RELEASED_V0_EVENT_DISPOSITIONS['user/message']?.extensionArms) - .toEqual(['content[].type', 'source.kind']) - expect(RELEASED_V0_EVENT_DISPOSITIONS['assistant/chunk']?.extensionArms) - .toContain('chunk.reason.kind') - expect(RELEASED_V0_EVENT_DISPOSITIONS['turn/end']?.extensionArms) - .toEqual(['reason.kind']) + it.each([ + ['user/message content block', 'user/message', { + ...userMessage, + content: [{ type: 'future-block', private: { preserved: true } }], + }], + ['user/message source', 'user/message', { + ...userMessage, + source: { kind: 'future-source', private: { preserved: true } }, + }], + ['assistant finish reason', 'assistant/chunk', { + turn: 1, step: 0, + chunk: { type: 'finish', reason: { kind: 'future-reason', private: { preserved: true } } }, + }], + ['turn/end reason', 'turn/end', { + turn: 1, reason: { kind: 'future-reason', private: { preserved: true } }, + }], + ] as const)('preserves an unknown merge-extensible %s arm', (_name, type, data) => { + expect(() => { assertPayload(type, data) }).not.toThrow() }) it('refuses unknown v0 events even when the envelope marks them ignorable', () => { diff --git a/packages/session/session-persistence-jsonl/src/generation.ts b/packages/session/session-persistence-jsonl/src/generation.ts index 07eb7bf380..255272e9d0 100644 --- a/packages/session/session-persistence-jsonl/src/generation.ts +++ b/packages/session/session-persistence-jsonl/src/generation.ts @@ -239,6 +239,11 @@ function isEEXIST(error: unknown): boolean { return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST' } +/** Whether a filesystem-owned failure should retain its original errno and path. */ +function isErrnoException(error: unknown): error is NodeJS.ErrnoException { + return typeof (error as NodeJS.ErrnoException | null)?.code === 'string' +} + function identity(value: JsonlPhysicalIdentity): string { return [value.dev, value.ino, value.size, value.mtimeNs, value.ctimeNs].join(':') } @@ -247,15 +252,19 @@ function fingerprint(value: JsonlPhysicalIdentity, bytes: Buffer): string { return `${identity(value)}:${createHash('sha256').update(bytes).digest('hex')}` } -/** Read exact physical bytes only when the stat identity brackets one stable read. */ +/** + * Read one stable revision with a single retry. If an append overlaps both + * reads, return the second read's committed pre-read prefix instead of + * starving behind a continuous writer. + */ async function readStableSnapshot( path: string, signal: AbortSignal | undefined, fs: GenerationFileSystem, ): Promise { - for (;;) { - signal?.throwIfAborted() - const before = await fs.stat(path) + signal?.throwIfAborted() + let before = await fs.stat(path) + for (let attempt = 0; ; attempt += 1) { const bytes = await fs.readFile(path, signal) signal?.throwIfAborted() const after = await fs.stat(path) @@ -263,6 +272,10 @@ async function readStableSnapshot( signal?.throwIfAborted() return { bytes, identity: after } } + if (attempt === 1) { + return { bytes: bytes.subarray(0, Number(before.size)), identity: before } + } + before = after } } @@ -647,10 +660,14 @@ async function reopenExpectedCurrent( throw new Error(`target is a ${kind}`) } const snapshot = await validatePhysicalCurrent(currentPath, compression, format, signal, internals) - if (!snapshot.bytes.equals(expectedBytes)) throw new Error('target bytes differ from the migrated generation') + if (snapshot.bytes.length < expectedBytes.length + || !snapshot.bytes.subarray(0, expectedBytes.length).equals(expectedBytes)) { + throw new Error('target bytes do not begin with the migrated generation') + } return snapshot } catch (error: unknown) { if (signal?.aborted) signal.throwIfAborted() + if (isErrnoException(error)) throw error throw new JsonlGenerationTargetConflictError(currentPath, asError(error)) } } diff --git a/packages/session/session-persistence-jsonl/tests/generation.spec.ts b/packages/session/session-persistence-jsonl/tests/generation.spec.ts index dbf598bd3d..721dab44af 100644 --- a/packages/session/session-persistence-jsonl/tests/generation.spec.ts +++ b/packages/session/session-persistence-jsonl/tests/generation.spec.ts @@ -169,6 +169,28 @@ describe('JSONL immutable generation publication', () => { expect(await readFile(request.sourcePath, 'utf8')).toBe(contents) }) + it('bounds current snapshot retries under continuous revision churn', async () => { + const root = await tempRoot() + const request = options(root, 'none', adapter(), 1) + const contents = line(header(1)) + line(event0) + await writeFile(request.sourcePath, contents) + let revision = 0n + const statFile = vi.fn(async (path: string) => { + const value = await stat(path, { bigint: true }) + revision += 1n + return { ...value, mtimeNs: value.mtimeNs + revision } + }) + const readChangingFile = vi.fn(async () => Buffer.from(contents + line(event1))) + + const result = await __jsonlGenerationTest.ensure(request, { + fs: { stat: statFile, readFile: readChangingFile }, + }) + + expect(result.snapshot.bytes.toString('utf8')).toBe(contents) + expect(readChangingFile).toHaveBeenCalledTimes(2) + expect(statFile).toHaveBeenCalledTimes(3) + }) + it('returns a disposable Zstandard body owner on the current fast path', async () => { const root = await tempRoot() const request = options(root, 'zstd', adapter(), 1) @@ -603,6 +625,32 @@ describe('JSONL immutable generation publication', () => { expect((await readdir(root)).sort()).toEqual(['session.jsonl', 'session.v1.jsonl']) }) + it.each(['none', 'zstd'] as const)( + 'accepts a valid append on a %s target created by another migration', + async (compression) => { + const root = await tempRoot() + const request = options(root, compression) + const source = compression === 'zstd' + ? await encodeZstd(0, [event0]) + : Buffer.from(line(header(0)) + line(event0)) + const expected = compression === 'zstd' + ? await encodeZstd(1, [event0]) + : Buffer.from(line(header(1)) + line(event0)) + const appended = compression === 'zstd' + ? await compressZstdFrame(line(event1)) + : Buffer.from(line(event1)) + const winner = Buffer.concat([expected, appended]) + await writeFile(request.sourcePath, source) + await writeFile(request.currentPath, winner) + + const result = await ensureJsonlGenerationCurrent(request) + + expect(result).toMatchObject({ status: 'migrated', path: request.currentPath }) + expect(result.snapshot.bytes).toEqual(winner) + expect(await readFile(request.currentPath)).toEqual(winner) + }, + ) + it('accepts an identical regular hardlink target', async () => { const root = await tempRoot() const request = options(root) @@ -755,7 +803,7 @@ describe('JSONL immutable generation publication', () => { expect((await readdir(root)).every(name => !name.includes('.tmp'))).toBe(true) }) - it('reports an absent exclusive-publication winner as a target conflict', async () => { + it('preserves ENOENT when an exclusive-publication winner disappears', async () => { const root = await tempRoot() const request = options(root) await writeFile(request.sourcePath, line(header(0)) + line(event0)) @@ -763,7 +811,24 @@ describe('JSONL immutable generation publication', () => { await expect(__jsonlGenerationTest.ensure(request, { platform: 'darwin', fs: posixSimulationFs({ link: async () => { throw fsError('EEXIST') } }), - })).rejects.toBeInstanceOf(JsonlGenerationTargetConflictError) + })).rejects.toMatchObject({ code: 'ENOENT', path: request.currentPath }) + }) + + it('preserves a filesystem error while reopening a committed target', async () => { + const root = await tempRoot() + const request = options(root) + const failure = fsError('EACCES', 'current target is unreadable') + failure.path = request.currentPath + await writeFile(request.sourcePath, line(header(0)) + line(event0)) + + await expect(__jsonlGenerationTest.ensure(request, { + fs: { + readFile: async (path, signal) => { + if (path === request.currentPath) throw failure + return readFile(path, signal === undefined ? undefined : { signal }) + }, + }, + })).rejects.toBe(failure) }) it('rethrows the exact abort reason during committed reopen and leaves the target', async () => {