Merge commit 'a4a34b2a71f6bdb92c723840dca4d7fbd71c3148' into worktree/session-format-05-v1-v2-chunk-migration

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

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.
 

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 事实都不得改变。
 

packages/session/session-format-v0-to-v1/src/dispositions.ts

@@ -4,8 +4,6 @@ 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[]
 }
 
 /**
@@ -20,13 +18,11 @@ export function defineReleasedPayloadDisposition(
   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]),
   })
 }
 
@@ -35,32 +31,23 @@ const disposition = defineReleasedPayloadDisposition
 /**
  * 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<Record<string, ReleasedV0PayloadDisposition>> = 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']),
@@ -70,8 +57,6 @@ export const RELEASED_V0_EVENT_DISPOSITIONS: Readonly<Record<string, ReleasedV0P
   'compaction/summary': disposition(
     ['compactionId', 'summary', 'shadowedRange', 'shadowedSeqs', 'shadowedTokenCount', 'provider', 'model'],
     ['sourceCommandId', 'maxTokens', 'usage', 'rawOutput', 'llmStreamCall'],
-    [],
-    ['summary[].type', 'rawOutput[].type'],
   ),
   'feedback/record': disposition(['text']),
   'goal/change': disposition(
@@ -102,9 +87,6 @@ export const RELEASED_V0_EVENT_DISPOSITIONS: Readonly<Record<string, ReleasedV0P
   'session/title': disposition(['title', 'messageSeqs', 'source']),
   'session/title-llm-request': disposition(
     ['titleProvider', 'messageSeqs', 'route', 'system', 'messages', 'maxTokens'],
-    [],
-    [],
-    ['messages[].content[].type', 'messages[].source.kind'],
   ),
   'step/end': disposition(['turn', 'step']),
   'step/start': disposition(['turn', 'step']),
@@ -115,12 +97,7 @@ export const RELEASED_V0_EVENT_DISPOSITIONS: Readonly<Record<string, ReleasedV0P
   'subagent/model-selection-policy': disposition(['allowedModels']),
   'team/member': disposition(['version', 'teamId', 'member']),
   'team/message/delivered': disposition(['version', 'teamId', 'messageId', 'targetId']),
-  'team/message/queued': disposition(
-    ['version', 'teamId', 'message'],
-    [],
-    [],
-    ['message.content[].type'],
-  ),
+  'team/message/queued': disposition(['version', 'teamId', 'message']),
   'team/task': disposition(['version', 'teamId', 'task']),
   'todo/write': disposition(['todos']),
   'tool-workflow/agent-end': disposition(['runId', 'seq', 'outcome']),
@@ -132,7 +109,6 @@ export const RELEASED_V0_EVENT_DISPOSITIONS: Readonly<Record<string, ReleasedV0P
     ['rootCallId', 'parentCallId', 'subCallId', 'name', 'arguments', 'isError', 'content'],
     [],
     ['arguments'],
-    ['content[].type'],
   ),
   'tool/code-dispatch-start': disposition(
     ['rootCallId', 'parentCallId', 'subCallId', 'name', 'arguments'],
@@ -143,16 +119,10 @@ export const RELEASED_V0_EVENT_DISPOSITIONS: Readonly<Record<string, ReleasedV0P
     ['turn', 'step', 'message'],
     ['error', 'meta'],
     ['meta'],
-    ['message.content[].content[].type'],
   ),
-  'turn/end': disposition(['turn', 'reason'], [], [], ['reason.kind']),
+  'turn/end': disposition(['turn', 'reason']),
   'turn/start': disposition(['turn']),
-  'user/message': disposition(
-    ['role', 'id', 'content', 'source'],
-    [],
-    [],
-    ['content[].type', 'source.kind'],
-  ),
+  'user/message': disposition(['role', 'id', 'content', 'source']),
   'web/deepseek-search-llm-request': disposition(['endpoint', 'apiVersion', 'body']),
 })
 

packages/session/session-format-v0-to-v1/src/payload-validation.ts

@@ -128,7 +128,9 @@ export function assertReleasedPayloadSemantics(event: SessionFormatEvent, versio
       } else if (data['maxRetries'] !== undefined) {
         throw new SessionFormatError(`${label} always mode must omit maxRetries`)
       }
-      if (countValue(data['delayMs'], `${label} delayMs`) > 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`)

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)
   })

packages/session/session-format-v0-to-v1/tests/validation.spec.ts

@@ -290,13 +290,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', () => {

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<StablePhysicalFile> {
-  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))
   }
 }

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 () => {