Merge origin/master into codex/trim-redundant-comments

packages/client/hmr/README.md

@@ -2,7 +2,7 @@
 
 Hot reload for fetch-arrival client plugins. A static-arrival entry composed only into `--dev` graphs (`dsh web --dev`); production graphs omit the row, so the shell-bundled code stays inert.
 
-The plugin subscribes to the webserver's system SSE channel (`GET /plugins/events`) and reloads one plugin per `rebuilt` frame, serialized through a queue (the bundle handoff slot is single). The sequence per frame — `prefetch` (fetch the new bundle before touching anything), `invalidate`, `registry.delete` (before the fiber: a bare fiber dispose trips the vendored Loader's self-dispose branch, which would mark the entry disabled), drain the old fiber, delete `entry.fiber`, remove owned `<style data-plugin>` tags, `entry.refresh()` re-imports and remounts, `fiber.await()` rethrows startup failures loud. Dependents reload through cordis itself: a fiber's activation epoch strings its service providers' uids, so replacing a provider's fiber cascades every dependent with zero client-side graph analysis. Rebuild detection lives on the webserver: in dev mode it stat-polls each plugin's built `lib/client.js` (`fs.watchFile`) and broadcasts the `rebuilt` frame when the bundle's rev changes, so any tsdown watch process producing the bundle triggers HMR with no builder→host channel.
+The browser half subscribes to the system SSE channel (`GET /plugins/events`) and reloads one plugin per `rebuilt` frame, serialized through a queue (the bundle handoff slot is single). The sequence per frame — `prefetch` (fetch the new bundle before touching anything), `invalidate`, `registry.delete` (before the fiber: a bare fiber dispose trips the vendored Loader's self-dispose branch, which would mark the entry disabled), drain the old fiber, delete `entry.fiber`, remove owned `<style data-plugin>` tags, `entry.refresh()` re-imports and remounts, `fiber.await()` rethrows startup failures loud. Dependents reload through cordis itself: a fiber's activation epoch strings its service providers' uids, so replacing a provider's fiber cascades every dependent with zero client-side graph analysis. The node half detects rebuilds with one interval that stat-polls each graph bundle from a synchronous baseline, immediately re-hashes after adding a row, retains missing rows as dirty, and broadcasts only real rev changes; any tsdown watch process producing the bundle therefore triggers HMR with no builder→host channel.
 
 ## Model Experience
 

packages/client/hmr/src/index.ts

@@ -1,13 +1,12 @@
 /**
- * HMR plugin, node half: the host end of the dev reload chain. Stat-polls
- * every graph row's client bundle (fs.watchFile — polling by design: network
+ * HMR plugin, node half: the host end of the dev reload chain. One interval
+ * stat-polls every graph row's client bundle (polling by design: network
  * mounts deliver no inotify events), reports content changes through
  * `clientModuleHost.rebuilt(id)`, and serves the `/plugins/events` SSE channel
  * broadcasting graph/rebuilt frames to the browser half (src/client/).
  * Dev-only row: prod compositions never mount this plugin.
  */
-import type { Stats } from 'node:fs'
-import { unwatchFile, watchFile } from 'node:fs'
+import { statSync } from 'node:fs'
 import type { ServerResponse } from 'node:http'
 import type { Context } from 'cordis'
 import z from 'schemastery'
@@ -41,6 +40,13 @@ function sseData(frame: PluginsEventFrame): string {
   return `data: ${JSON.stringify(frame)}\n\n`
 }
 
+interface WatchedBundle {
+  path: string
+  mtimeMs: number
+  size: number
+  dirty: boolean
+}
+
 /**
  * Mount the dev chain: bundle watches, rebuilt reporting, and the SSE channel.
  * @param ctx - host plugin context carrying clientModuleHost and httpServer.
@@ -50,29 +56,59 @@ export function apply(ctx: Context, config: Config): void {
   // schemastery's .default() guarantees the field is set after validation.
   const pollIntervalMs = config.pollIntervalMs as number
 
-  // --- bundle watch: one fs.watchFile stat poll per graph row -------------
-  const watched = new Map<string, { path: string; listener: (curr: Stats, prev: Stats) => void }>()
+  // --- bundle watch: one HMR-owned stat poll ------------------------------
+  const watched = new Map<string, WatchedBundle>()
+
+  const rehash = (id: string, watch: WatchedBundle, current: { mtimeMs: number; size: number }): void => {
+    try {
+      // rebuilt() re-hashes; an unchanged hash stays silent (clientModuleHost
+      // fires onRebuilt only on a real rev change).
+      ctx.clientModuleHost.rebuilt(id)
+    } catch (error) {
+      const code = (error as NodeJS.ErrnoException).code
+      if (code === 'ENOENT') {
+        watch.dirty = true
+        return
+      }
+      ctx.logger.warn(error)
+    }
+    watch.mtimeMs = current.mtimeMs
+    watch.size = current.size
+    watch.dirty = false
+  }
 
   const watchRow = (id: string, path: string): void => {
-    const listener = (curr: Stats, prev: Stats): void => {
-      // fs.watchFile fires on any stat delta (atime included); only content
-      // signals count. An all-zero curr means the file vanished mid-rebuild
-      // — the completing write fires the next tick, so skipping is safe.
-      if (curr.mtimeMs === prev.mtimeMs && curr.size === prev.size) return
-      if (curr.mtimeMs === 0) return
-      try {
-        // rebuilt() re-hashes; an unchanged hash stays silent (clientModuleHost
-        // fires onRebuilt only on a real rev change). A torn read of a
-        // half-written bundle self-heals on the next poll tick.
-        ctx.clientModuleHost.rebuilt(id)
-      } catch (error) {
-        const code = (error as NodeJS.ErrnoException).code
-        if (code === 'ENOENT') return // mid-rename window; the completed write fires the next poll tick
-        ctx.logger.warn(error)
-      }
+    let baseline: { mtimeMs: number; size: number }
+    try {
+      baseline = statSync(path)
+    } catch (error) {
+      watched.set(id, { path, mtimeMs: 0, size: 0, dirty: true })
+      if ((error as NodeJS.ErrnoException).code !== 'ENOENT') ctx.logger.warn(error)
+      return
+    }
+    const watch = { path, mtimeMs: baseline.mtimeMs, size: baseline.size, dirty: false }
+    watched.set(id, watch)
+    // The module host hashed before publishing the graph. Re-hash immediately
+    // after capturing this baseline so a write in between cannot become an
+    // already-current baseline paired with a stale graph rev.
+    rehash(id, watch, baseline)
+  }
+
+  const pollWatches = (): void => {
+    for (const [id, watch] of watched) {
+      let current: { mtimeMs: number; size: number }
+      try {
+        current = statSync(watch.path)
+      } catch (error) {
+        watch.dirty = true
+        if ((error as NodeJS.ErrnoException).code !== 'ENOENT') ctx.logger.warn(error)
+        continue
+      }
+      if (!watch.dirty && current.mtimeMs === watch.mtimeMs && current.size === watch.size) continue
+      // Stat-before-hash preserves a detectable older baseline for writes that
+      // land during hashing. Repeated stat changes heal a torn read.
+      rehash(id, watch, current)
     }
-    watchFile(path, { interval: pollIntervalMs, persistent: false }, listener)
-    watched.set(id, { path, listener })
   }
 
   // Diff the watch set against the current graph: drop watches for removed
@@ -85,7 +121,6 @@ export function apply(ctx: Context, config: Config): void {
     }
     for (const [id, watch] of watched) {
       if (rows.get(id) === watch.path) continue
-      unwatchFile(watch.path, watch.listener)
       watched.delete(id)
     }
     for (const [id, path] of rows) {
@@ -99,9 +134,11 @@ export function apply(ctx: Context, config: Config): void {
     // own row — no self-exemption, a modules/hmr rebuild rides the same chain).
     syncWatches()
     const unsubscribe = ctx.clientModuleHost.onGraphChanged(syncWatches)
+    const timer = setInterval(pollWatches, pollIntervalMs)
+    timer.unref()
     return () => {
       unsubscribe()
-      for (const { path, listener } of watched.values()) unwatchFile(path, listener)
+      clearInterval(timer)
       watched.clear()
     }
   }, 'client-hmr: bundle watches')

packages/client/hmr/tests/node-half.spec.ts

@@ -2,7 +2,7 @@
  * Node half of the HMR plugin: bundle watches follow the graph, stat changes
  * report through clientModuleHost.rebuilt, and everything dies with the fiber.
  */
-import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
+import { mkdtempSync, rmSync, statSync, unlinkSync, utimesSync, writeFileSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { Context } from 'cordis'
@@ -24,18 +24,29 @@ afterEach(() => { rmSync(dir, { recursive: true, force: true }) })
  * the service class carries private scan state a literal need not reproduce.
  */
 type FakeHost = ClientModuleHostService & { rebuiltCalls: string[]; fireGraphChanged(): void }
-function fakeClientModuleHost(rows: Map<string, string>): FakeHost {
+interface FakeHostOptions {
+  beforeGraphRead?: () => void
+  rebuilt?: (id: string) => string | undefined
+}
+
+function fakeClientModuleHost(rows: Map<string, string>, options: FakeHostOptions = {}): FakeHost {
   const graphListeners = new Set<() => void>()
   const rebuiltCalls: string[] = []
   const fake: Pick<FakeHost, 'graph' | 'clientPath' | 'rebuilt' | 'onRebuilt' | 'onGraphChanged' | 'rebuiltCalls' | 'fireGraphChanged'> = {
     rebuiltCalls,
     fireGraphChanged: () => { for (const l of graphListeners) l() },
-    graph: (): WebBootGraph => ({
-      rev: 'r',
-      entries: [...rows.keys()].map(id => ({ id, url: `/plugins/${id}/client.js?rev=r`, rev: 'r' })),
-    }),
+    graph: (): WebBootGraph => {
+      options.beforeGraphRead?.()
+      return {
+        rev: 'r',
+        entries: [...rows.keys()].map(id => ({ id, url: `/plugins/${id}/client.js?rev=r`, rev: 'r' })),
+      }
+    },
     clientPath: id => rows.get(id),
-    rebuilt: (id) => { rebuiltCalls.push(id); return 'r2' },
+    rebuilt: (id) => {
+      rebuiltCalls.push(id)
+      return options.rebuilt?.(id) ?? 'r2'
+    },
     onRebuilt: () => () => {},
     onGraphChanged: (listener) => {
       graphListeners.add(listener)
@@ -81,6 +92,8 @@ describe('hmr node half', () => {
 
     expect(routes).toHaveLength(1)
     expect(routes[0]).toMatchObject({ kind: 'exact', path: EVENTS_ENDPOINT })
+    expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a'])
+    clientModuleHost.rebuiltCalls.length = 0
 
     // Nudge mtime past stat granularity so the poller sees a content signal.
     await new Promise(resolve => setTimeout(resolve, POLL_MS * 2))
@@ -103,14 +116,89 @@ describe('hmr node half', () => {
     const rows = new Map([['pkg-early', early]])
     const clientModuleHost = fakeClientModuleHost(rows)
     const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+    clientModuleHost.rebuiltCalls.length = 0
 
     writeFileSync(late, 'v1')
     rows.set('pkg-late', late)
     clientModuleHost.fireGraphChanged()
+    expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-late'])
+    clientModuleHost.rebuiltCalls.length = 0
 
     await new Promise(resolve => setTimeout(resolve, POLL_MS * 2))
     writeFileSync(late, 'v2-longer')
     await vi.waitFor(() => { expect(clientModuleHost.rebuiltCalls).toContain('pkg-late') }, { timeout: 3_000 })
+
+    rows.delete('pkg-late')
+    clientModuleHost.fireGraphChanged()
+    clientModuleHost.rebuiltCalls.length = 0
+    writeFileSync(late, 'v3-even-longer')
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 3))
+    expect(clientModuleHost.rebuiltCalls).toHaveLength(0)
+    await fiber.dispose()
+  })
+
+  it('rehashes after baseline capture so a construction-window write cannot become the baseline', async () => {
+    const bundle = join(dir, 'construction.js')
+    writeFileSync(bundle, 'v1')
+    let rewrite = true
+    const clientModuleHost = fakeClientModuleHost(new Map([['pkg-a', bundle]]), {
+      beforeGraphRead: () => {
+        if (!rewrite) return
+        rewrite = false
+        // The graph carries the hash from before this write. The old
+        // fs.watchFile registration asynchronously captured the new file as
+        // its first baseline and never requested a re-hash.
+        writeFileSync(bundle, 'v2-written-during-watch-construction')
+      },
+    })
+
+    const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+
+    expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a'])
+    clientModuleHost.rebuiltCalls.length = 0
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 3))
+    expect(clientModuleHost.rebuiltCalls).toHaveLength(0)
+    await fiber.dispose()
+  })
+
+  it('marks a vanished bundle dirty so identical metadata still re-hashes after it reappears', async () => {
+    const bundle = join(dir, 'replace.js')
+    writeFileSync(bundle, 'seed')
+    const fixedTime = new Date(1_600_000_000_000)
+    utimesSync(bundle, fixedTime, fixedTime)
+    const baseline = statSync(bundle)
+    const clientModuleHost = fakeClientModuleHost(new Map([['pkg-a', bundle]]))
+    const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+    clientModuleHost.rebuiltCalls.length = 0
+
+    unlinkSync(bundle)
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 2))
+    writeFileSync(bundle, 'x'.repeat(baseline.size))
+    utimesSync(bundle, fixedTime, fixedTime)
+    const restored = statSync(bundle)
+    expect({ mtimeMs: restored.mtimeMs, size: restored.size }).toEqual({
+      mtimeMs: baseline.mtimeMs,
+      size: baseline.size,
+    })
+    await vi.waitFor(() => { expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a']) }, { timeout: 3_000 })
+    await fiber.dispose()
+  })
+
+  it('retains a dirty baseline when the immediate re-hash races a rename', async () => {
+    const bundle = join(dir, 'rename.js')
+    writeFileSync(bundle, 'v1')
+    let first = true
+    const clientModuleHost = fakeClientModuleHost(new Map([['pkg-a', bundle]]), {
+      rebuilt: () => {
+        if (!first) return 'r2'
+        first = false
+        throw Object.assign(new Error('bundle renamed'), { code: 'ENOENT' })
+      },
+    })
+
+    const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+
+    await vi.waitFor(() => { expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a', 'pkg-a']) }, { timeout: 3_000 })
     await fiber.dispose()
   })
 })

packages/llm/llm-deepseek/README.md

@@ -49,7 +49,7 @@ Every request carries the shared attribution header from dsh-llm's `attributionH
 
 ## Errors
 
-Non-2xx responses throw `LlmError` with stable codes: `AUTH` (401/403), `QUOTA` (a response whose provider details identify exhausted quota, balance, or credits), `RATE_LIMIT` (other 429s), `CONTEXT_WINDOW_EXCEEDED` (a 400 whose provider code, type, or message identifies context overflow), `INVALID_REQUEST` (other 400s), `SERVER` (5xx), `HTTP_<status>` otherwise. Its serializable `failure` retains the HTTP status plus a valid positive `Retry-After` seconds/date delay and `x-request-id` / `x-deepseek-request-id` when present. A pre-response transport failure (DNS, refused connection, TLS, proxy) throws `TRANSPORT` naming the configured endpoint and chaining the original rejection as `cause`; caller aborts throw `ABORTED`, and the loop's cancellation signal remains authoritative. Protocol violations throw `STREAM_CLOSED` (no `[DONE]`) or `MALFORMED_RESPONSE` (bad JSON payload). Unknown wire `finish_reason`s (e.g. `content_filter`, `insufficient_system_resource`) become `finish {kind: 'error', failure}` chunks.
+Non-2xx responses throw `LlmError` with stable codes: `AUTH` (401/403), `QUOTA` (a response whose provider details identify exhausted quota, balance, or credits), `RATE_LIMIT` (other 429s), `CONTEXT_WINDOW_EXCEEDED` (a 400 whose provider code, type, or message identifies context overflow), `INVALID_REQUEST` (other 400s), `SERVER` (5xx), `HTTP_<status>` otherwise. Its serializable `failure` retains the HTTP status plus a valid positive `Retry-After` seconds/date delay and `x-request-id` / `x-deepseek-request-id` when present. A pre-response transport failure (DNS, refused connection, TLS, proxy) throws `TRANSPORT` naming the configured endpoint and chaining the original rejection as `cause`; caller aborts throw `ABORTED`, and the loop's cancellation signal remains authoritative. Protocol violations throw `STREAM_CLOSED` (no `[DONE]`) or `MALFORMED_RESPONSE` (bad JSON payload). Unknown wire `finish_reason`s (e.g. `content_filter`, `insufficient_system_resource`) become `finish {kind: 'error', failure}` chunks, and a completed stream whose `stop` (or absent) finish opened no content blocks becomes a `finish {kind: 'error'}` with code `EMPTY_RESPONSE` (retried by default policy).
 
 ## Testing
 

packages/llm/llm-deepseek/src/translate.ts

@@ -8,7 +8,7 @@
  * @module dsh-llm-deepseek/translate
  */
 
-import { CallId, LlmError } from '@deepseek-ai/dsh-llm'
+import { CallId, EMPTY_RESPONSE_CODE, LlmError } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock, FinishReason, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm'
 import { DONE } from './sse.ts'
 import type { WireChunk, WireUsage } from './types.ts'
@@ -80,6 +80,8 @@ function closeBlock(block: OpenBlock): ContentBlock {
  * Malformed JSON payloads abort the stream with `MALFORMED_RESPONSE`.
  * @param payloads - SSE data payloads from {@link parseSse}, `[DONE]`-terminated.
  * @returns deltas as they arrive; `block-end`s, `usage`, and `finish` are all deferred to the `[DONE]` sentinel.
+ *   A `stop` (or absent) finish with no opened blocks is a degenerate provider completion and maps to an
+ *   `EMPTY_RESPONSE` error finish instead of a successful empty message.
  */
 export async function* translate(payloads: AsyncIterable<string>): AsyncGenerator<StreamChunk> {
   let nextIndex = 0
@@ -102,7 +104,16 @@ export async function* translate(payloads: AsyncIterable<string>): AsyncGenerato
         yield { type: 'block-end', index: block.index, block: closeBlock(block) }
       }
       if (pendingUsage) yield { type: 'usage', usage: pendingUsage }
-      yield { type: 'finish', reason: pendingFinish ?? { kind: 'stop' } }
+      const reason = pendingFinish ?? { kind: 'stop' as const }
+      yield {
+        type: 'finish',
+        reason: reason.kind === 'stop' && order.length === 0
+          ? {
+            kind: 'error',
+            failure: { message: 'model returned a completed response with no content', code: EMPTY_RESPONSE_CODE },
+          }
+          : reason,
+      }
       return
     }
 

packages/llm/llm-deepseek/tests/translate.spec.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from 'vitest'
-import { BlockAssembler, LlmError } from '@deepseek-ai/dsh-llm'
+import { BlockAssembler, EMPTY_RESPONSE_CODE, LlmError } from '@deepseek-ai/dsh-llm'
 import type { StreamChunk } from '@deepseek-ai/dsh-llm'
 import { DONE } from '../src/sse.ts'
 import { mapFinishReason, mapUsage, translate } from '../src/translate.ts'
@@ -203,7 +203,50 @@ describe('translate: finish and usage handling', () => {
 
   it('handles chunks with no choices at all', async () => {
     const chunks = await collect(translate(feed({}, DONE)))
-    expect(chunks).toEqual([{ type: 'finish', reason: { kind: 'stop' } }])
+    expect(chunks).toEqual([{
+      type: 'finish',
+      reason: {
+        kind: 'error',
+        failure: { message: 'model returned a completed response with no content', code: EMPTY_RESPONSE_CODE },
+      },
+    }])
+  })
+
+  it('classifies an explicit stop with no opened blocks as EMPTY_RESPONSE, after usage', async () => {
+    const chunks = await collect(translate(feed(
+      firstChunk,
+      { choices: [{ delta: {}, finish_reason: 'stop' }], usage: { prompt_tokens: 7, completion_tokens: 0 } },
+      DONE,
+    )))
+    expect(chunks).toEqual([
+      { type: 'usage', usage: { inputTokens: 7, outputTokens: 0 } },
+      {
+        type: 'finish',
+        reason: {
+          kind: 'error',
+          failure: { message: 'model returned a completed response with no content', code: EMPTY_RESPONSE_CODE },
+        },
+      },
+    ])
+  })
+
+  it('keeps a reasoning-only stream a successful stop (any opened block counts)', async () => {
+    const chunks = await collect(translate(feed(
+      firstChunk,
+      { choices: [{ delta: { content: null, reasoning_content: 'mull' } }] },
+      { choices: [{ delta: {}, finish_reason: 'stop' }] },
+      DONE,
+    )))
+    expect(chunks.at(-1)).toEqual({ type: 'finish', reason: { kind: 'stop' } })
+  })
+
+  it('leaves non-stop finishes unclassified even with no opened blocks', async () => {
+    const chunks = await collect(translate(feed(
+      firstChunk,
+      { choices: [{ delta: {}, finish_reason: 'length' }] },
+      DONE,
+    )))
+    expect(chunks.at(-1)).toEqual({ type: 'finish', reason: { kind: 'max-tokens' } })
   })
 })
 

packages/llm/llm-pi-ai/README.md

@@ -45,7 +45,7 @@ If a listener rewrites assembled assistant content, the loop drops replay state
 ## Vocabulary differences
 
 - pi-ai tool-call arguments are parsed objects; the harness stores raw JSON strings. The adapter parses input and re-stringifies output.
-- pi-ai reports failures as in-stream error events; these map to `finish {kind:'error'|'aborted', failure}` chunks. Provider-specific error text distinguishes terminal `QUOTA` from transient `RATE_LIMIT`, while text and usage signals evaluated against the resolved model's context window normalize overflow to `CONTEXT_WINDOW_EXCEEDED`.
+- pi-ai reports failures as in-stream error events; these map to `finish {kind:'error'|'aborted', failure}` chunks. Provider-specific error text distinguishes terminal `QUOTA` from transient `RATE_LIMIT`, while text and usage signals evaluated against the resolved model's context window normalize overflow to `CONTEXT_WINDOW_EXCEEDED`. A terminal `stop` whose message carries no content blocks maps to a `finish {kind:'error'}` with code `EMPTY_RESPONSE` (retried by default policy) instead of a successful empty message.
 - pi-ai folds reasoning tokens into output usage; there is no separate reasoning count to map.
 - `GenerateOptions.stop` is rejected with `UNSUPPORTED_OPTION` because pi-ai's common streaming surface cannot guarantee it across providers.
 

packages/llm/llm-pi-ai/src/stream.ts

@@ -8,7 +8,7 @@
  * @module dsh-llm-pi-ai/stream
  */
 
-import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, isContextWindowExceededError, isQuotaExceededError, LlmError, QUOTA_EXCEEDED_CODE } from '@deepseek-ai/dsh-llm'
+import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, EMPTY_RESPONSE_CODE, isContextWindowExceededError, isQuotaExceededError, LlmError, QUOTA_EXCEEDED_CODE } from '@deepseek-ai/dsh-llm'
 import type { FinishReason, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm'
 import { isContextOverflow } from '@earendil-works/pi-ai'
 import type { AssistantMessage, AssistantMessageEvent, Usage as PiUsage } from '@earendil-works/pi-ai'
@@ -48,7 +48,8 @@ function classifyPiAiError(message: string): string {
  * @param contextWindow - resolved catalog capacity for usage-based overflow detection.
  * @returns the mapped harness reason. Recognized error text, `stop` usage above
  *   `contextWindow`, and zero-output `length` usage that fills the window map
- *   to `CONTEXT_WINDOW_EXCEEDED`.
+ *   to `CONTEXT_WINDOW_EXCEEDED`; a `stop` with no content blocks maps to an
+ *   `EMPTY_RESPONSE` error.
  */
 export function mapStopReason(message: AssistantMessage, contextWindow?: number): FinishReason {
   const piAiOverflow = isContextOverflow(message, contextWindow)
@@ -66,7 +67,19 @@ export function mapStopReason(message: AssistantMessage, contextWindow?: number)
   }
 
   switch (message.stopReason) {
-    case 'stop': return { kind: 'stop' }
+    case 'stop':
+      // A terminal stop that produced no content blocks is a degenerate
+      // provider completion, not a successful (empty) assistant message.
+      if (message.content.length === 0) {
+        return {
+          kind: 'error',
+          failure: {
+            message: `model "${message.model}" returned a completed response with no content`,
+            code: EMPTY_RESPONSE_CODE,
+          },
+        }
+      }
+      return { kind: 'stop' }
     case 'length': return { kind: 'max-tokens' }
     case 'toolUse': return { kind: 'tool-calls' }
     case 'aborted': return {

packages/llm/llm-pi-ai/tests/convert.spec.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from 'vitest'
-import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, LlmError } from '@deepseek-ai/dsh-llm'
+import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, EMPTY_RESPONSE_CODE, LlmError } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm'
 import type { AssistantMessage, AssistantMessageEvent, Usage } from '@earendil-works/pi-ai'
 import { toPiContext } from '../src/context.ts'
@@ -520,7 +520,22 @@ describe('mapStopReason / mapUsage', () => {
     ['toolUse', { kind: 'tool-calls' }],
     ['aborted', { kind: 'aborted', failure: { message: 'pi-ai stream aborted', code: 'ABORTED' } }],
   ] as const)('maps %s', (stopReason, expected) => {
-    expect(mapStopReason(assistant({ stopReason }))).toEqual(expected)
+    expect(mapStopReason(assistant({ stopReason, content: [{ type: 'text', text: 'ok' }] }))).toEqual(expected)
+  })
+
+  it('classifies a completed stop with no content as an EMPTY_RESPONSE error', () => {
+    expect(mapStopReason(assistant({ stopReason: 'stop' }))).toEqual({
+      kind: 'error',
+      failure: {
+        message: 'model "deepseek-v4-flash" returned a completed response with no content',
+        code: EMPTY_RESPONSE_CODE,
+      },
+    })
+  })
+
+  it('keeps a thinking-only stop successful (any block counts as content)', () => {
+    expect(mapStopReason(assistant({ stopReason: 'stop', content: [{ type: 'thinking', thinking: 'mull' }] })))
+      .toEqual({ kind: 'stop' })
   })
 
   it('defaults the error message when pi-ai omits it', () => {
@@ -580,7 +595,9 @@ describe('mapStopReason / mapUsage', () => {
   })
 
   it('uses the resolved context window for silent and length-stop overflows', () => {
-    const silent = assistant({ stopReason: 'stop', usage: usage(101, 0) })
+    // Non-empty content keeps the no-window branch on the successful stop path
+    // (an empty stop is EMPTY_RESPONSE, covered above); overflow wins over both.
+    const silent = assistant({ stopReason: 'stop', usage: usage(101, 0), content: [{ type: 'text', text: 'x' }] })
     expect(mapStopReason(silent)).toEqual({ kind: 'stop' })
     expect(mapStopReason(silent, 100)).toEqual({
       kind: 'error',

packages/llm/llm-retry/README.md

@@ -2,7 +2,7 @@
 
 Function plugin that retries selected transient model-request failures on the agent loop's closed-step recovery seam. It does not wrap `ctx.llm.stream()`: every adapter call remains one provider attempt, and every retry opens a fresh numbered step.
 
-The default policy permits two retries for `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`, using bounded exponential backoff from 500 ms to 10 seconds with 10 percent jitter. Delay bounds must fit Node's supported timer range. A valid `providerRetryAfterMs` replaces local backoff when it is within the configured cap; an over-cap instruction delegates to the next recovery policy instead.
+The default policy permits two retries for `EMPTY_RESPONSE`, `RATE_LIMIT`, `SERVER`, `TIMEOUT`, and `TRANSPORT`, using bounded exponential backoff from 500 ms to 10 seconds with 10 percent jitter. `EMPTY_RESPONSE` is the adapters' classification of a degenerate provider completion (a terminal stop with zero content blocks); the attempt produced nothing durable, so repeating it is safe. Delay bounds must fit Node's supported timer range. A valid `providerRetryAfterMs` replaces local backoff when it is within the configured cap; an over-cap instruction delegates to the next recovery policy instead.
 
 Before waiting, the plugin appends a non-surface `llm/retry` event with the failure and scheduled delay. Cancellation and plugin disposal abort the wait; disposal drains the plugin's active backoffs, and a callback captured before disposal fails closed if invoked afterward.
 
@@ -15,7 +15,7 @@ The separately published `./invariant` companion checks that every retry record
     initialDelayMs: 500
     maxDelayMs: 10000
     jitterRatio: 0.1
-    retryableCodes: [RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT]
+    retryableCodes: [EMPTY_RESPONSE, RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT]
 ```
 
 ## Model Experience

packages/llm/llm-retry/src/index.ts

@@ -33,7 +33,7 @@ const DEFAULT_MAX_TRANSIENT_RETRIES = 2
 const DEFAULT_INITIAL_DELAY_MS = 500
 const DEFAULT_MAX_DELAY_MS = 10_000
 const DEFAULT_JITTER_RATIO = 0.1
-const DEFAULT_RETRYABLE_CODES = Object.freeze(['RATE_LIMIT', 'SERVER', 'TIMEOUT', 'TRANSPORT'])
+const DEFAULT_RETRYABLE_CODES = Object.freeze(['EMPTY_RESPONSE', 'RATE_LIMIT', 'SERVER', 'TIMEOUT', 'TRANSPORT'])
 
 /** Deployment-owned limits and classification for transient request recovery. */
 export interface Config {

packages/llm/llm-retry/tests/retry.spec.ts

@@ -1,7 +1,7 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import type { Fiber } from 'cordis'
-import LlmService, { CallId, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
+import LlmService, { CallId, EMPTY_RESPONSE_CODE, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
 import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { SessionEvent } from '@deepseek-ai/dsh-session'
@@ -51,6 +51,25 @@ function textResponse(text: string): StreamChunk[] {
   ]
 }
 
+/**
+ * A degenerate empty provider completion as an error finish chunk. Both
+ * adapters emit this shape and the EMPTY_RESPONSE code (the field the policy
+ * routes on); the message text here is the deepseek adapter's phrasing (pi-ai
+ * qualifies it with the model name).
+ */
+function emptyCompletion(): StreamChunk[] {
+  return [
+    { type: 'usage', usage: { inputTokens: 0, outputTokens: 0 } },
+    {
+      type: 'finish',
+      reason: {
+        kind: 'error',
+        failure: { message: 'model returned a completed response with no content', code: EMPTY_RESPONSE_CODE },
+      },
+    },
+  ]
+}
+
 async function harness(
   adapter: LlmAdapter,
   config: retry.Config = {},
@@ -158,6 +177,39 @@ describe('bounded transient retry policy', () => {
     })
   })
 
+  it('retries an EMPTY_RESPONSE error finish under the default retryable codes', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      emptyCompletion(),
+      textResponse('recovered'),
+    ])
+    // No retryableCodes override: this proves the default policy covers the
+    // adapters' empty-completion classification end to end (finish-chunk error
+    // delivery, not a thrown stream error).
+    ;({ ctx: context } = await harness(adapter))
+    const agent = context.agentLoop.create(SessionId('retry-empty-response'), { provider: 'mock', model: 'mock' })
+    const scheduled = waitForRetry(context, agent, 1)
+
+    agent.followup([{ type: 'text', text: 'go' }])
+    const event = await scheduled
+    expect(event.data.failure).toEqual({
+      message: 'model returned a completed response with no content',
+      code: EMPTY_RESPONSE_CODE,
+    })
+
+    const idle = waitForIdle(context, agent)
+    await vi.advanceTimersByTimeAsync(500)
+    await idle
+
+    expect(adapter.requests).toHaveLength(2)
+    expect(agent.session.events.filter(event => event.type === 'assistant/message').map(event => event.data.step))
+      .toEqual([2])
+    expect(agent.session.deriveMessages().at(-1)).toMatchObject({
+      role: 'assistant',
+      content: [{ type: 'text', text: 'recovered' }],
+    })
+  })
+
   it('leaves partial failed chunks on their step without committing a message or tool side effect', async () => {
     vi.useFakeTimers()
     const adapter = new ScriptedAdapter([

packages/llm/llm/README.md

@@ -54,6 +54,7 @@ Every product adapter sends application identity on provider HTTP requests. `att
 - `errorChain(value)` — renders a thrown value with its full `cause` chain and AggregateError members for diagnostic surfaces (UI notices, logger lines, durable `turn/end` messages), so transport wrappers like undici's `TypeError: fetch failed` surface the underlying `ECONNREFUSED`/DNS/TLS detail instead of masking it. Rendering only — route on `code`, never by parsing the result.
 - `CONTEXT_WINDOW_EXCEEDED_CODE` — the provider-neutral code both DeepSeek adapters use when a request exceeds the model context window, regardless of thrown-HTTP versus in-band finish delivery. `isContextWindowExceededError(detail)` is their shared conservative classifier for OpenAI-compatible provider detail.
 - `QUOTA_EXCEEDED_CODE` — the non-transient provider-neutral code for exhausted account quota, balance, credits, budget, or usage limits. `isQuotaExceededError(detail)` keeps those failures distinct from request-rate limits.
+- `EMPTY_RESPONSE_CODE` — the provider-neutral code both adapters use for a degenerate provider completion: a terminal `stop` that carried no content blocks at all. Classified as an error finish (not a successful empty message) because the attempt produced nothing durable; `dsh-llm-retry` retries it by default.
 
 ### Real adapters
 

packages/llm/llm/src/error.ts

@@ -27,6 +27,17 @@ export const CONTEXT_WINDOW_EXCEEDED_CODE = 'CONTEXT_WINDOW_EXCEEDED'
 /** Canonical provider-neutral code for an exhausted account quota or balance. */
 export const QUOTA_EXCEEDED_CODE = 'QUOTA'
 
+/**
+ * Canonical provider-neutral code for a response that completed normally but
+ * carried no content blocks at all. Providers occasionally emit a degenerate
+ * completion (a terminal stop with zero output); adapters classify it as this
+ * failure instead of yielding an empty assistant message, because an empty
+ * message silently ends the turn with nothing for the user or the loop to act
+ * on. The attempt produced nothing durable, so retry policy treats it as safe
+ * to repeat.
+ */
+export const EMPTY_RESPONSE_CODE = 'EMPTY_RESPONSE'
+
 /** Structured codes and plain phrases that explicitly name a context bound being exceeded. */
 const STRUCTURED_CONTEXT_OVERFLOW = new RegExp(
   String.raw`(?:^|[^a-z0-9])context[\s_-](?:length|window)[\s_-]`

packages/session-persistence/session-persistence-jsonl/README.md

@@ -6,14 +6,16 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
 
 ```
 <root>/
-  cwd-<sha256(cwd)[:12]>/        # per-project bucket (or _no-cwd/ when no cwd)
-    <encoded-id>.jsonl.zstd      # default: checksummed header frame + append frames
-    <encoded-id>.jsonl           # only with compression: 'none'
+  --<normalized-cwd>--/          # readable project directory (or _no-cwd/)
+    <encoded-id>/                # session-owned directory
+      session.jsonl.zstd         # default: checksummed header frame + append frames
+      session.jsonl              # only with compression: 'none'
 ```
 
 - The first logical line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession?, seedLength?, delegationDepth }`. `delegationDepth` is required on disk and is `0` for a top-level session; a missing or invalid value rejects the log. Every subsequent logical line is one storage record; `assistant/chunk` events are never dropped, and `seq` stays contiguous across the decoded log (`events[i].seq === i`).
 - A storage record is a `SessionEvent` JSON verbatim, or — written only under `packChunks` — a **packed chunk row** (`text-chunks` / `reasoning-chunks` / `tool-call-chunks`; bare slash-less tags like the header's `session`, so row tags cannot be confused with event types): one line holding a run of ≥3 consecutive same-block `assistant/chunk` delta events, `seq0`/`time0` plus per-member `dt` gaps reconstructing every member's `seq`/`time` exactly. The lossless codec lives in `@deepseek-ai/dsh-session` (`packChunkRuns`/`decodeStorageRecord`) and whitelists exact shapes — anything unrecognized stores verbatim. Reading is layout-blind: `load` always decodes rows, so packed, unpacked, and mixed files load identically.
-- Session ids are unvalidated branded strings, so they are injectively escaped to a single safe path segment before use (no traversal, no collision).
+- The project directory keeps the normalized cwd readable for navigation and is bounded for filesystem component limits. Separator replacement and truncation are intentionally lossy, so cwd strings that normalize alike share a project directory; session ids still select distinct session directories. On a case-insensitive filesystem, identity validation accepts an alternate path spelling only when filesystem canonicalization resolves both spellings to the same transcript. The configured root remains deployment-controlled: it may be project-local, shared, temporary, or centralized. The [project-session directory decision](../../../.agents/notes/implemented/architecture/2026-07-24-project-session-directories.md) records this tradeoff.
+- Session ids are unvalidated branded strings, so they are injectively escaped to a single safe path segment before use (no traversal, no collision). The resulting directory is reserved for additional session-owned artifacts; discovery reads only the fixed transcript filename.
 
 ## Config
 
@@ -23,17 +25,17 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
 | `packChunks` | `boolean` (default `false`) | Write delta-chunk runs as packed rows (~60% smaller logical logs measured on a real coding session). Off, the written logical layout is byte-identical to the pre-packing format; reading packed rows works regardless of this switch. Off by default while the snapshot goldens stay one-event-per-line — recording with packing on rewrites every fixture `session.jsonl`. |
 | `compression` | `'zstd' \| 'none'` | Defaults to `'zstd'`; `'none'` retains newline-delimited UTF-8 text. |
 
-`locate(meta)` returns `{ kind: 'jsonl', path }` using the resolved absolute root and the same cwd-bucket/id encoding as materialization. It performs no filesystem I/O: the target can be returned before the file exists, and an existing file contains only the last flushed prefix.
+`locate(meta)` returns `{ kind: 'jsonl', path }` for the fixed transcript inside the resolved project/session directories. It performs no filesystem I/O: the target can be returned before the directory or file exists, and an existing file contains only the last flushed prefix.
 
 ## Physical encoding
 
 The default artifact is a standard concatenation of independent [Zstandard frames](../../../.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md): one checksummed frame containing only the header line, followed by one checksummed frame per durable append batch. The backend uses Node's built-in Zstandard API with its default compression level and exposes no level knob. Listing reads and validates only the header frame. `compression: 'none'` keeps the same logical lines in the original raw representation.
 
-A root belongs to one encoding. Startup discovery and targeted lookup reject the opposite suffix with an error naming the incompatible artifact and instructing the caller to select the matching mode or a separate root. There is no migration, mixed-root fallback, or dual write.
+A root belongs to one encoding. Startup discovery and targeted lookup reject the opposite suffix with an error naming the incompatible artifact and instructing the caller to select the matching mode or a separate root. Flat `<project>/<id>.jsonl*` artifacts are also rejected instead of ignored. There is no migration, mixed-root fallback, or dual write.
 
 ## Durability and crash semantics
 
-- **Bound storage identity.** Lookup requires one matching encoded filename across the cwd buckets, then verifies that the header id equals the requested id and that the header's id/cwd derive the selected path. Listing applies the same path check and rejects duplicate ids. Identity failures occur before repair or append.
+- **Bound storage identity.** Lookup requires one matching session directory across the readable project directories, then verifies that the header id equals the requested id and that the header's id/cwd derive the selected transcript path. Listing applies the same path check and rejects duplicate ids. Identity failures occur before repair or append.
 - **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s the encoded header and first batch in a temporary file. POSIX publishes it without overwrite via a hard link and `fsync`s the parent directory. Windows publishes it without overwrite via `MoveFileExW(..., MOVEFILE_WRITE_THROUGH)` and creates missing directories through the same write-through pattern. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
 - **Append-only.** Flushed events are never rewritten. Subsequent raw batches append lines; compressed batches append one frame. Both paths `fsync`, and a caught write or sync failure rolls the file back to its prior byte length.
 - **Crash recovery — preserve valid tail work.** `load` validates every complete compressed frame and scans their decompressed JSONL. If the last frame is structurally incomplete, the reader keeps its complete decoded records, truncates from that frame's start, and re-encodes those records with the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). Raw mode truncates from its first incomplete line. A checksum/decompression failure in a complete frame, or a defect at or before the last committed `turn/end`, is corruption and rejects.
@@ -64,6 +66,7 @@ JSONL storage does not mutate live request prefixes. A resumed loop can reuse pr
 ## Known Limitations and Deferred Work
 
 - **Only the configured encoding and current `SESSION_FORMAT_VERSION` (v0) load** — changing compression requires a separate/fresh root or selecting the legacy raw mode; the pre-release format has no migration.
+- **The flat-file storage layout does not load** — use a separate root or move pre-release artifacts into the project/session directory layout before loading.
 - **Compressed files are not directly line-readable** — use the backend to load them, or select `compression: 'none'` before writing a fresh root when text fixtures or external line readers are required.
 - **Nothing deletes session files** — logs accumulate under `root` until removed externally (the seam has no deletion surface).
 - **One live writer per session** — append and repair are coordinated only inside the owning backend instance. Another backend instance or process must not write the same session until that owner reaches quiescent disposal; initial same-id publication remains collision-safe through the POSIX no-overwrite hard link or Windows write-through rename without replacement.

packages/session-persistence/session-persistence-jsonl/src/format.ts

@@ -2,13 +2,12 @@
  * On-disk format helpers for the JSONL session-persistence backend: path
  * sanitization (a {@link SessionId} is an unvalidated branded string, so it
  * MUST be encoded before use in a path — no traversal, no collision), the
- * per-cwd directory layout, header-line (de)serialization, and the
+ * per-project/session directory layout, header-line (de)serialization, and the
  * truncation-repair offset computation.
  *
  * @module dsh-session-persistence-jsonl/format
  */
 
-import { createHash } from 'node:crypto'
 import { join } from 'node:path'
 import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader, SessionId, StorageRecord } from '@deepseek-ai/dsh-session'
@@ -123,24 +122,64 @@ export function encodeSegment(raw: string): string {
 }
 
 /**
- * The directory a session's files live in: the configured root, then a per-cwd
- * subdirectory so sessions group by project. The cwd subdir is a stable hash of
- * the cwd (short, collision-resistant, filesystem-safe); sessions without a
- * cwd go in a shared `_no-cwd` bucket.
- * @param root - the backend's session root directory.
- * @param cwd - the session's project directory; `undefined` selects the shared `_no-cwd` bucket.
- * @returns the per-cwd bucket directory path under `root`.
+ * Build the readable directory key for a project path.
+ * Filesystem separators and drive separators become `-`; unsafe code units use
+ * the same `~XXXX` escape as session ids. The key is bounded for filesystem
+ * component limits. Separator replacement and truncation are intentionally
+ * lossy, following the common human-navigable project-directory convention.
+ * @param cwd - the session's project directory.
+ * @returns a single filesystem-safe project directory name.
  */
-export function sessionDir(root: string, cwd: string | undefined): string {
+export function projectKey(cwd: string): string {
+  if (cwd.length === 0) throw new Error('cannot encode an empty project path')
+  let readable = ''
+  let separatorRun = false
+  for (let i = 0; i < cwd.length; i++) {
+    const code = cwd.charCodeAt(i)
+    const ch = String.fromCharCode(code)
+    if (ch === '/' || ch === '\\' || ch === ':') {
+      if (!separatorRun) readable += '-'
+      separatorRun = true
+    } else if (ch !== '~' && /^[A-Za-z0-9._-]$/.test(ch)) {
+      readable += ch
+      separatorRun = false
+    } else {
+      readable += '~' + code.toString(16).toUpperCase().padStart(4, '0')
+      separatorRun = false
+    }
+  }
+  const slug = readable.replace(/^-+/, '') || 'root'
+  return `--${slug.slice(0, 251)}--`
+}
+
+/**
+ * The configured root's human-navigable project directory. A configured root
+ * may be local or shared; this grouping does not prescribe its deployment.
+ * @param root - the backend's session root directory.
+ * @param cwd - the session's project directory; `undefined` selects `_no-cwd`.
+ * @returns the project directory path under `root`.
+ */
+export function projectDir(root: string, cwd: string | undefined): string {
   if (cwd === undefined) return join(root, '_no-cwd')
-  const hash = createHash('sha256').update(cwd).digest('hex').slice(0, 12)
-  return join(root, `cwd-${hash}`)
+  return join(root, projectKey(cwd))
+}
+
+/**
+ * The directory owned by one session and available for future session-local
+ * artifacts.
+ * @param root - the backend's session root directory.
+ * @param cwd - the session's project directory.
+ * @param id - the session id, encoded to one safe path segment.
+ * @returns the session directory beneath its project directory.
+ */
+export function sessionDir(root: string, cwd: string | undefined, id: SessionId): string {
+  return join(projectDir(root, cwd), encodeSegment(id))
 }
 
 /**
  * The append-only event-log file path for a session.
  * @param root - the backend's session root directory.
- * @param cwd - the session's project directory (picks the per-cwd bucket; `undefined` → `_no-cwd`).
+ * @param cwd - the session's project directory (`undefined` → `_no-cwd`).
  * @param id - the session id, path-encoded via {@link encodeSegment} before filesystem use.
  * @param compression - physical artifact encoding and filename suffix.
  * @returns the session's configured JSONL artifact path.
@@ -151,7 +190,7 @@ export function logPath(
   id: SessionId,
   compression: JsonlCompression,
 ): string {
-  return join(sessionDir(root, cwd), `${encodeSegment(id)}${logSuffix(compression)}`)
+  return join(sessionDir(root, cwd, id), `session${logSuffix(compression)}`)
 }
 
 /**

packages/session-persistence/session-persistence-jsonl/src/index.ts

@@ -9,7 +9,7 @@
 import { Context } from 'cordis'
 import z from 'schemastery'
 import { readdirSync } from 'node:fs'
-import { open, mkdir, readFile, readdir, link, rm, stat, truncate } from 'node:fs/promises'
+import { open, mkdir, readFile, readdir, realpath, link, rm, stat, truncate } from 'node:fs/promises'
 import { dirname, join, resolve } from 'node:path'
 import { randomBytes } from 'node:crypto'
 import {
@@ -19,7 +19,7 @@ import {
 } from '@deepseek-ai/dsh-session-persistence'
 import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
-  encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
+  encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir, toHeaderLine,
   type JsonlCompression,
 } from './format.ts'
 import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from './zstd.ts'
@@ -40,9 +40,9 @@ export interface Config {
   /**
    * Root directory for all session files. Required (no default): a default of
    * `process.cwd()` would scatter session files as the process's cwd changes
-   * (bash calls, subprocesses). Sessions group under per-cwd subdirectories. An
-   * existing root must be a readable directory; an absent root is created on
-   * first materialization.
+   * (bash calls, subprocesses). Sessions group under human-readable project
+   * directories, then per-session directories. An existing root must be a
+   * readable directory; an absent root is created on first materialization.
    */
   root: string
   /**
@@ -141,7 +141,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
   /* jscpd:ignore-end */
   // --- PersistenceBackend hooks (the file-bytes storage primitives) ---
 
-  /** Read a stored prefix by id across all cwd buckets when cwd is unknown. */
+  /** Read a stored prefix by id across all project directories when cwd is unknown. */
   async loadStored(id: SessionId): Promise<StoredPrefix<JsonlTornMarker> | undefined> {
     await this.ensureRootEncoding()
     const path = await this.findLog(id)
@@ -168,7 +168,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
           : {},
       }
     }
-    this.assertStoredIdentity(path, prefix.meta, expectedId)
+    await this.assertStoredIdentity(path, prefix.meta, expectedId)
     return prefix
   }
 
@@ -278,9 +278,12 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     await this.ensureRootEncoding()
     const artifacts: Array<{ header: SessionHeader; path: string }> = []
     const ids = new Set<SessionId>()
-    for (const dir of await this.listCwdDirs()) {
-      for (const name of await this.listArtifactNames(dir)) {
-        const path = join(dir, name)
+    for (const project of await this.listProjectDirs()) {
+      for (const dir of await this.listSessionDirs(project)) {
+        const opposite = join(dir, `session${logSuffix(this.oppositeCompression())}`)
+        if (await this.exists(opposite)) throw this.encodingMismatch(opposite)
+        const path = join(dir, `session${logSuffix(this.compression)}`)
+        if (!await this.exists(path)) continue
         // Read only headers so listing scales with session count, not log size.
         const first = this.compression === 'zstd'
           ? await this.readFirstZstdLine(path)
@@ -288,9 +291,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
         if (first === undefined) continue // empty/half-written file
         const meta = parseHeaderMeta(first)
         if (meta === undefined) continue // not a session header
-        this.assertStoredIdentity(path, meta)
+        await this.assertStoredIdentity(path, meta)
         if (ids.has(meta.id)) {
-          throw new Error(`duplicate JSONL session id "${meta.id}" appears in multiple cwd buckets`)
+          throw new Error(`duplicate JSONL session id "${meta.id}" appears in multiple project directories`)
         }
         ids.add(meta.id)
         artifacts.push({ header: meta, path })
@@ -303,20 +306,22 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
 
   /** Atomically write the header line + first batch (temp-write, fsync, publish). */
   private async materialize(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
-    const dir = sessionDir(this.root, meta.cwd)
+    const project = projectDir(this.root, meta.cwd)
+    const dir = sessionDir(this.root, meta.cwd, meta.id)
     const finalPath = logPath(this.root, meta.cwd, meta.id, this.compression)
     await this.rejectOppositeArtifact(meta.cwd, meta.id)
     const content = await this.encodeMaterialization(meta, events)
     /* v8 ignore next -- native Windows coverage exercises this platform dispatch; Linux covers the POSIX peer */
     if (process.platform === 'win32') {
-      await this.materializeWin32(dir, finalPath, meta.id, content)
+      await this.materializeWin32(project, dir, finalPath, meta.id, content)
     } else {
-      await this.materializePosix(dir, finalPath, meta.id, content)
+      await this.materializePosix(project, dir, finalPath, meta.id, content)
     }
   }
 
   /* v8 ignore start -- Windows uses the Win32 durable-publish path; POSIX coverage exercises this peer. */
   private async materializePosix(
+    project: string,
     dir: string,
     finalPath: string,
     id: SessionId,
@@ -324,8 +329,10 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
   ): Promise<void> {
     await mkdir(this.root, { recursive: true, mode: 0o700 })
     await this.syncDirPosix(dirname(this.root))
-    await mkdir(dir, { recursive: true, mode: 0o700 })
+    await mkdir(project, { recursive: true, mode: 0o700 })
     await this.syncDirPosix(this.root)
+    await mkdir(dir, { recursive: true, mode: 0o700 })
+    await this.syncDirPosix(project)
     await this.rejectExistingLog(finalPath, id)
     const tmp = await this.writeSyncedTempFile(finalPath, content)
     // Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
@@ -358,12 +365,14 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
 
   /* v8 ignore start -- native Windows coverage exercises this integration path */
   private async materializeWin32(
+    project: string,
     dir: string,
     finalPath: string,
     id: SessionId,
     content: Buffer | string,
   ): Promise<void> {
     await ensureDurableDirectoryWin32(this.root)
+    await ensureDurableDirectoryWin32(project)
     await ensureDurableDirectoryWin32(dir)
     await this.rejectExistingLog(finalPath, id)
     const tmp = await this.writeSyncedTempFile(finalPath, content)
@@ -541,19 +550,19 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     }
   }
 
-  /** Find the unique physical log for an id across every cwd bucket. */
+  /** Find the unique physical log for an id across every project directory. */
   private async findLog(id: SessionId): Promise<string | undefined> {
-    const target = encodeSegment(id) + logSuffix(this.compression)
-    const oppositeTarget = encodeSegment(id) + logSuffix(this.oppositeCompression())
     const matches: string[] = []
-    for (const dir of await this.listCwdDirs()) {
-      const path = join(dir, target)
-      const opposite = join(dir, oppositeTarget)
+    for (const project of await this.listProjectDirs()) {
+      await this.rejectLegacyFlatArtifact(project, id)
+      const dir = join(project, encodeSegment(id))
+      const path = join(dir, `session${logSuffix(this.compression)}`)
+      const opposite = join(dir, `session${logSuffix(this.oppositeCompression())}`)
       if (await this.exists(opposite)) throw this.encodingMismatch(opposite)
       if (await this.exists(path)) matches.push(path)
     }
     if (matches.length > 1) {
-      throw new Error(`duplicate JSONL session id "${id}" appears in multiple cwd buckets`)
+      throw new Error(`duplicate JSONL session id "${id}" appears in multiple project directories`)
     }
     return matches[0]
   }
@@ -569,7 +578,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
   }
 
   /** Reject metadata that does not identify the selected physical log. */
-  private assertStoredIdentity(path: string, meta: SessionHeader, expectedId?: SessionId): void {
+  private async assertStoredIdentity(path: string, meta: SessionHeader, expectedId?: SessionId): Promise<void> {
     if (expectedId !== undefined && meta.id !== expectedId) {
       throw new Error(`corrupt session log "${path}": requested id "${expectedId}" does not match header id "${meta.id}"`)
     }
@@ -579,13 +588,30 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     } catch (error) {
       throw new Error(`corrupt session log "${path}": header id cannot name a storage path`, { cause: error })
     }
-    if (path !== expectedPath) {
-      throw new Error(`corrupt session log "${path}": header id "${meta.id}" and cwd belong at "${expectedPath}"`)
+    if (path !== expectedPath && !await this.sameFile(path, expectedPath)) {
+      throw new Error(`corrupt session log "${path}": header id "${meta.id}" and cwd identify "${expectedPath}"`)
     }
   }
 
-  /** The cwd-bucket directories under the root (absolute paths). */
-  private async listCwdDirs(): Promise<string[]> {
+  /**
+   * Whether two path spellings resolve to the same physical file. This admits
+   * case aliases on case-insensitive filesystems without weakening identity
+   * checks on case-sensitive stores.
+   */
+  private async sameFile(path: string, expectedPath: string): Promise<boolean> {
+    try {
+      const [actual, expected] = await Promise.all([realpath(path), realpath(expectedPath)])
+      return actual === expected
+    } catch (error) {
+      /* v8 ignore else -- non-ENOENT realpath failures require an external permission or I/O fault */
+      if (isENOENT(error)) return false
+      /* v8 ignore next -- non-ENOENT realpath failures are external I/O faults, propagated unchanged */
+      throw error
+    }
+  }
+
+  /** The human-readable project directories under the configured root. */
+  private async listProjectDirs(): Promise<string[]> {
     try {
       const entries = await readdir(this.root, { withFileTypes: true })
       return entries.filter(e => e.isDirectory()).map(e => join(this.root, e.name))
@@ -596,13 +622,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     }
   }
 
-  private async listArtifactNames(dir: string): Promise<string[]> {
-    const entries = await readdir(dir)
-    const oppositeSuffix = logSuffix(this.oppositeCompression())
-    const incompatible = entries.find(name => name.endsWith(oppositeSuffix))
-    if (incompatible !== undefined) throw this.encodingMismatch(`${dir}/${incompatible}`)
-    const suffix = logSuffix(this.compression)
-    return entries.filter(name => name.endsWith(suffix))
+  /** List session-owned directories and reject the obsolete flat-file layout. */
+  private async listSessionDirs(project: string): Promise<string[]> {
+    const entries = await readdir(project, { withFileTypes: true })
+    const legacy = entries.find(entry =>
+      entry.isFile() && (entry.name.endsWith('.jsonl') || entry.name.endsWith('.jsonl.zstd')))
+    if (legacy !== undefined) throw this.legacyLayout(join(project, legacy.name))
+    return entries.filter(entry => entry.isDirectory()).map(entry => join(project, entry.name))
   }
 
   /** Reject a root that already belongs to the other physical encoding. */
@@ -612,11 +638,19 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
   }
 
   private async checkRootEncoding(): Promise<void> {
-    const oppositeSuffix = logSuffix(this.oppositeCompression())
-    for (const dir of await this.listCwdDirs()) {
-      const entries = await readdir(dir)
-      const incompatible = entries.find(name => name.endsWith(oppositeSuffix))
-      if (incompatible !== undefined) throw this.encodingMismatch(`${dir}/${incompatible}`)
+    for (const project of await this.listProjectDirs()) {
+      for (const dir of await this.listSessionDirs(project)) {
+        const incompatible = join(dir, `session${logSuffix(this.oppositeCompression())}`)
+        if (await this.exists(incompatible)) throw this.encodingMismatch(incompatible)
+      }
+    }
+  }
+
+  private async rejectLegacyFlatArtifact(project: string, id: SessionId): Promise<void> {
+    const encoded = encodeSegment(id)
+    for (const compression of ['zstd', 'none'] as const) {
+      const path = join(project, encoded + logSuffix(compression))
+      if (await this.exists(path)) throw this.legacyLayout(path)
     }
   }
 
@@ -637,6 +671,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     )
   }
 
+  private legacyLayout(path: string): Error {
+    return new Error(
+      `session artifact ${JSON.stringify(path)} uses the unsupported flat-file layout; `
+      + 'use a separate root or move it into a project/session directory before loading',
+    )
+  }
+
   private async exists(path: string): Promise<boolean> {
     try {
       const handle = await open(path, 'r')
@@ -646,7 +687,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
       // Only ENOENT means absent. A permission/I/O error must surface rather
       // than letting load or collision checks proceed under false absence.
       // Windows reports ENOENT, not ENOTDIR, for `regular-file/child`; verify
-      // the immediate parent so a blocked cwd bucket remains a storage fault.
+      // the immediate parent so a blocked session directory remains a storage fault.
       /* v8 ignore else -- Windows reports file-valued parents as ENOENT; POSIX covers direct ENOTDIR. */
       if (isENOENT(error)) {
         await this.assertLogParentAllowsAbsence(path)

packages/session-persistence/session-persistence-jsonl/src/win32.ts

@@ -12,7 +12,7 @@
  */
 
 import { mkdtemp, rm, stat } from 'node:fs/promises'
-import { basename, join, parse, resolve, toNamespacedPath } from 'node:path'
+import { join, parse, resolve, toNamespacedPath } from 'node:path'
 
 type MoveFileExW = (existing: string, replacement: string, flags: number) => number
 type GetLastError = () => number
@@ -139,7 +139,9 @@ export async function ensureDurableDirectoryWin32(target: string): Promise<void>
 }
 
 async function createLeafDirectoryWin32(parent: string, target: string): Promise<void> {
-  const staging = await mkdtemp(join(parent, `.dsh-mkdir-${basename(target)}-`))
+  // Keep the staging component independent of the target basename so a legal
+  // 255-byte target component does not make mkdtemp's sibling name too long.
+  const staging = await mkdtemp(join(parent, '.dsh-mkdir-'))
   try {
     await publishNewFileWin32(staging, target)
   } catch (error) {

packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts

@@ -1,12 +1,14 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
-import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises'
+import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat, symlink } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { isAbsolute, join, relative, resolve } from 'node:path'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
-import { encodeSegment, eventLines, logPath, scanLog, sessionDir, toHeaderLine } from '../src/format.ts'
+import {
+  encodeSegment, eventLines, logPath, projectDir, projectKey, scanLog, sessionDir, toHeaderLine,
+} from '../src/format.ts'
 import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
 import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
 
@@ -125,6 +127,16 @@ describe('SessionPersistenceJsonl: format helpers', () => {
     expect(() => encodeSegment('')).toThrow(/empty/)
   })
 
+  it('projectKey normalizes project paths into bounded readable names', () => {
+    expect(projectKey('/Users/qyj/work/deepseek-harness')).toBe('--Users-qyj-work-deepseek-harness--')
+    expect(projectKey('/a/b-c')).toBe(projectKey('/a-b/c'))
+    expect(projectKey('C:\\work\\agent')).toBe('--C-work-agent--')
+    expect(projectKey('/开发/~agent')).toBe('--~5F00~53D1-~007Eagent--')
+    expect(projectKey('/')).toBe('--root--')
+    expect(projectKey('/' + 'x'.repeat(1_000))).toHaveLength(255)
+    expect(() => projectKey('')).toThrow(/empty project path/)
+  })
+
   it('resolves a relative custom root before locating a session', async () => {
     const absoluteRoot = await freshRoot()
     const ctx = new Context()
@@ -161,15 +173,15 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     await ctx.sessionPersistence.create(m)
     // locate() is a pure target-path calculation: neither it nor create()
     // materializes a file before the first append.
-    const dir = sessionDir(root, '/work')
+    const dir = sessionDir(root, '/work', m.id)
     await expect(stat(rawLogPath(root, '/work', m.id))).rejects.toThrow()
     expect((await ctx.sessionPersistence.list()).map(h => h.id)).not.toContain(m.id)
 
     await ctx.sessionPersistence.append(m.id, oneTurnLog())
     // now materialized
+    expect((await stat(dir)).isDirectory()).toBe(true)
     expect((await stat(rawLogPath(root, '/work', m.id))).isFile()).toBe(true)
     expect((await ctx.sessionPersistence.list()).map(h => h.id)).toContain(m.id)
-    void dir
   })
 
   it('keeps the same location on resume and gives a fork its own location', async () => {
@@ -266,7 +278,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
   it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
     const m = meta('legacy-header-delta', '/legacy')
     const path = rawLogPath(root, m.cwd, m.id)
-    await mkdir(sessionDir(root, m.cwd), { recursive: true })
+    await mkdir(sessionDir(root, m.cwd, m.id), { recursive: true })
     await writeFile(path, [
       JSON.stringify(toHeaderLine(m)),
       JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
@@ -281,7 +293,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
   it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
     const m = meta('legacy-header-fallback', '/legacy')
     const path = rawLogPath(root, m.cwd, m.id)
-    await mkdir(sessionDir(root, m.cwd), { recursive: true })
+    await mkdir(sessionDir(root, m.cwd, m.id), { recursive: true })
     await writeFile(path, [
       JSON.stringify(toHeaderLine(m)),
       JSON.stringify({
@@ -711,7 +723,7 @@ describe('SessionPersistenceJsonl: packed chunk rows (packChunks: true)', () =>
     const log = chunkRunLog()
     // First turn written line-per-event by an unpacked-config writer (an old
     // file, hand-planted so this packed-config backend adopts it on load).
-    await mkdir(sessionDir(root, '/work'), { recursive: true })
+    await mkdir(sessionDir(root, '/work', m.id), { recursive: true })
     await writeFile(rawLogPath(root, '/work', m.id), [
       JSON.stringify({ type: 'session', version: 0, id: 'mixed', createdAt: 1000, cwd: '/work', delegationDepth: 0 }),
       ...log.map(e => JSON.stringify(e)),
@@ -807,34 +819,93 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await expect(stat(rawLogPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
   })
 
-  it('list discovers sessions across multiple cwd buckets', async () => {
+  it('list discovers sessions across multiple project directories', async () => {
     await ctx.sessionPersistence.create(meta('p1', '/projA'))
     await ctx.sessionPersistence.append(SessionId('p1'), oneTurnLog())
     await ctx.sessionPersistence.create(meta('p2', '/projB'))
     await ctx.sessionPersistence.append(SessionId('p2'), oneTurnLog())
-    await ctx.sessionPersistence.create(meta('p3')) // no cwd → _no-cwd bucket
+    await ctx.sessionPersistence.create(meta('p3')) // no cwd → _no-cwd project directory
     await ctx.sessionPersistence.append(SessionId('p3'), oneTurnLog())
 
     const ids = (await ctx.sessionPersistence.list()).map(x => x.id).sort()
     expect(ids).toEqual(['p1', 'p2', 'p3'])
   })
 
+  it('groups sessions whose cwd paths normalize to the same project directory', async () => {
+    const first = meta('normalized-first', '/a/b-c')
+    const second = meta('normalized-second', '/a-b/c')
+    await ctx.sessionPersistence.create(first)
+    await ctx.sessionPersistence.append(first.id, oneTurnLog())
+    await ctx.sessionPersistence.create(second)
+    await ctx.sessionPersistence.append(second.id, oneTurnLog())
+
+    expect(projectDir(root, first.cwd)).toBe(projectDir(root, second.cwd))
+    expect(await readdir(projectDir(root, first.cwd))).toEqual(expect.arrayContaining([
+      encodeSegment(first.id),
+      encodeSegment(second.id),
+    ]))
+    expect((await ctx.sessionPersistence.list()).map(header => header.id).sort())
+      .toEqual([first.id, second.id].sort())
+  })
+
   it('list on an empty root returns nothing', async () => {
     expect(await ctx.sessionPersistence.list()).toEqual([])
   })
 
-  it('list skips empty and non-header .jsonl files (metadata-only read)', async () => {
+  it('keeps the transcript in an extensible session-owned directory', async () => {
+    const m = meta('owned-directory', '/project')
+    await ctx.sessionPersistence.create(m)
+    await ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const dir = sessionDir(root, m.cwd, m.id)
+    await writeFile(join(dir, 'metadata.json'), '{}\n')
+    await writeFile(join(projectDir(root, m.cwd), 'README'), 'project metadata\n')
+    await mkdir(join(projectDir(root, m.cwd), 'reserved-session'), { recursive: true })
+
+    expect(await readdir(dir)).toEqual(expect.arrayContaining(['metadata.json', 'session.jsonl']))
+    expect((await ctx.sessionPersistence.list()).map(header => header.id)).toContain(m.id)
+    expect((await ctx.sessionPersistence.load(m.id)).events).toEqual(oneTurnLog())
+  })
+
+  it('rejects the obsolete flat-file layout instead of ignoring stored sessions', async () => {
+    const m = meta('legacy-flat', '/legacy')
+    const project = projectDir(root, m.cwd)
+    const path = join(project, `${encodeSegment(m.id)}.jsonl`)
+    await mkdir(project, { recursive: true })
+    await writeFile(path, [
+      JSON.stringify(toHeaderLine(m)),
+      ...oneTurnLog().map(event => JSON.stringify(event)),
+      '',
+    ].join('\n'))
+
+    await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported flat-file layout/)
+    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/unsupported flat-file layout/)
+  })
+
+  it('rejects a compressed obsolete flat-file artifact during targeted lookup', async () => {
+    const m = meta('legacy-compressed-flat', '/legacy')
+    const project = projectDir(root, m.cwd)
+    expect(await ctx.sessionPersistence.list()).toEqual([])
+    await mkdir(project, { recursive: true })
+    await writeFile(join(project, `${encodeSegment(m.id)}.jsonl.zstd`), 'legacy')
+
+    await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/unsupported flat-file layout/)
+  })
+
+  it('list skips empty and non-header session logs (metadata-only read)', async () => {
     // A real session…
     await ctx.sessionPersistence.create(meta('real', '/p'))
     await ctx.sessionPersistence.append(SessionId('real'), oneTurnLog())
-    // …alongside two junk files in the _no-cwd bucket: an EMPTY file (readFirstLine
-    // returns undefined) and a file whose first line is not a session header
-    // (parseHeaderMeta returns undefined). Both are skipped, not listed.
-    const bucket = join(root, '_no-cwd')
-    await mkdir(bucket, { recursive: true })
-    await writeFile(join(bucket, 'empty.jsonl'), '')
-    await writeFile(join(bucket, 'notheader.jsonl'), '{"type":"turn/start"}\n')
-    await writeFile(join(bucket, 'badjson.jsonl'), 'not json at all\n')
+    // …alongside junk session directories whose fixed transcript is empty or
+    // lacks a header. Both remain unmaterialized and are skipped.
+    for (const [id, content] of [
+      ['empty', ''],
+      ['notheader', '{"type":"turn/start"}\n'],
+      ['badjson', 'not json at all\n'],
+    ] as const) {
+      const path = rawLogPath(root, undefined, SessionId(id))
+      await mkdir(sessionDir(root, undefined, SessionId(id)), { recursive: true })
+      await writeFile(path, content)
+    }
 
     const ids = (await ctx.sessionPersistence.list()).map(x => x.id).sort()
     expect(ids).toEqual(['real'])
@@ -843,10 +914,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
   it('list reads a header line longer than the 8KB read chunk', async () => {
     // A tolerated extra field makes this valid header exceed the 8192-byte read buffer, proving
     // `readFirstLine` accumulates chunks before `list()` parses it.
-    const bucket = join(root, '_no-cwd')
-    await mkdir(bucket, { recursive: true })
+    const id = SessionId('big')
+    await mkdir(sessionDir(root, undefined, id), { recursive: true })
     const bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, delegationDepth: 0, pad: 'x'.repeat(9000) })
-    await writeFile(join(bucket, 'big.jsonl'), bigHeader + '\n')
+    await writeFile(rawLogPath(root, undefined, id), bigHeader + '\n')
     const ids = (await ctx.sessionPersistence.list()).map(x => x.id)
     expect(ids).toContain('big')
   })
@@ -857,30 +928,47 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await ctx.sessionPersistence.append(m.id, oneTurnLog())
     await rewriteHeader(rawLogPath(root, m.cwd, m.id), (header) => { header.cwd = '/elsewhere' })
 
-    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/and cwd belong at/)
+    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/and cwd identify/)
+  })
+
+  it('accepts an alternate project path only when it identifies the same physical log', async () => {
+    const m = meta('physical-alias', '/stored')
+    await ctx.sessionPersistence.create(m)
+    await ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const path = rawLogPath(root, m.cwd, m.id)
+    const aliasCwd = '/alias'
+    await symlink(
+      projectDir(root, m.cwd),
+      projectDir(root, aliasCwd),
+      process.platform === 'win32' ? 'junction' : 'dir',
+    )
+    await rewriteHeader(path, (header) => { header.cwd = aliasCwd })
+
+    expect((await ctx.sessionPersistence.load(m.id)).meta.cwd).toBe(aliasCwd)
+    expect((await ctx.sessionPersistence.list()).map(header => header.id)).toContain(m.id)
   })
 
   it('list rejects a session header whose id cannot name a storage path', async () => {
-    const bucket = sessionDir(root, undefined)
-    await mkdir(bucket, { recursive: true })
-    await writeFile(join(bucket, 'invalid-id.jsonl'), JSON.stringify({
+    const dir = join(projectDir(root, undefined), 'invalid-id')
+    await mkdir(dir, { recursive: true })
+    await writeFile(join(dir, 'session.jsonl'), JSON.stringify({
       type: 'session', version: 0, id: '', createdAt: 1, delegationDepth: 0,
     }) + '\n')
 
     await expect(ctx.sessionPersistence.list()).rejects.toThrow(/header id cannot name a storage path/)
   })
 
-  it('load and list reject one id materialized in multiple cwd buckets', async () => {
+  it('load and list reject one id materialized in multiple project directories', async () => {
     const id = SessionId('duplicate')
     for (const cwd of ['/a', '/b']) {
       const m = meta(id, cwd)
-      await mkdir(sessionDir(root, cwd), { recursive: true })
+      await mkdir(sessionDir(root, cwd, id), { recursive: true })
       const content = [JSON.stringify(toHeaderLine(m)), ...oneTurnLog().map(event => JSON.stringify(event))].join('\n') + '\n'
       await writeFile(rawLogPath(root, cwd, id), content)
     }
 
-    await expect(ctx.sessionPersistence.load(id)).rejects.toThrow(/appears in multiple cwd buckets/)
-    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/appears in multiple cwd buckets/)
+    await expect(ctx.sessionPersistence.load(id)).rejects.toThrow(/appears in multiple project directories/)
+    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/appears in multiple project directories/)
   })
 
   it('a DIFFERENT live session object reusing a disposed id gets its own init (no stale cache)', async () => {
@@ -1003,12 +1091,12 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await expect(backend.exists(join(blocker, 'child.jsonl'))).rejects.toThrow(/ENOTDIR/)
   })
 
-  it('materialization surfaces a cwd-bucket storage fault', async () => {
+  it('materialization surfaces a project-directory storage fault', async () => {
     const cwd = '/x'
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
     await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
-    await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
+    await writeFile(projectDir(root, cwd), 'x') // project path is now a file
     let s!: Session
     await ctx2.plugin(Object.assign((inner: Context) => {
       s = inner.sessions.create(SessionId('exists-fault'), { meta: { cwd } })
@@ -1056,14 +1144,14 @@ describe('SessionPersistenceJsonl: edge cases', () => {
   })
 
 
-  it('createCore rejects an id already on disk under a DIFFERENT cwd bucket', async () => {
+  it('createCore rejects an id already on disk under a different project directory', async () => {
     // Persist the id under cwd A.
     const a = meta('dup-id', '/projA')
     await ctx.sessionPersistence.create(a)
     await ctx.sessionPersistence.append(a.id, oneTurnLog())
     // A fresh backend creating the SAME id under cwd B must still refuse: load
-    // identifies by id across all buckets, so a second log would make resume
-    // nondeterministic. create scans every bucket, not just meta.cwd's.
+    // identifies by id across all projects, so a second log would make resume
+    // nondeterministic. create scans every project, not just meta.cwd's.
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
     await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })

packages/session-persistence/session-persistence-jsonl/tests/win32.spec.ts

@@ -151,6 +151,15 @@ describe('Windows durable namespace helpers', () => {
     expect(existsSync(raced)).toBe(true)
   })
 
+  it('keeps staging names valid for a maximum-length target component', async () => {
+    const { ensureDurableDirectoryWin32 } = await importWithFilesystemMove()
+    const root = await tempRoot()
+    const target = join(root, 'x'.repeat(255))
+
+    await ensureDurableDirectoryWin32(target)
+    expect(existsSync(target)).toBe(true)
+  })
+
   it('surfaces directory publication failures other than an existing-target race', async () => {
     const { ensureDurableDirectoryWin32 } = await importWithError(ERROR_ACCESS_DENIED)
     const root = await tempRoot()

packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts

@@ -391,15 +391,21 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
 
   it('skips empty, incomplete, and non-header compressed artifacts while rejecting malformed header frames', async () => {
     const root = await freshRoot()
-    const bucket = sessionDir(root, undefined)
-    await mkdir(bucket, { recursive: true })
-    await writeFile(join(bucket, 'empty.jsonl.zstd'), '')
-    await writeFile(join(bucket, 'partial.jsonl.zstd'), MAGIC)
-    await writeFile(join(bucket, 'not-header.jsonl.zstd'), await compressZstdFrame('{"type":"turn/start"}\n'))
+    for (const [id, content] of [
+      ['empty', Buffer.alloc(0)],
+      ['partial', MAGIC],
+      ['not-header', await compressZstdFrame('{"type":"turn/start"}\n')],
+    ] as const) {
+      const sessionId = SessionId(id)
+      await mkdir(sessionDir(root, undefined, sessionId), { recursive: true })
+      await writeFile(logPath(root, undefined, sessionId, 'zstd'), content)
+    }
     const ctx = await mount(root)
     expect(await ctx.sessionPersistence.list()).toEqual([])
 
-    await writeFile(join(bucket, 'two-lines.jsonl.zstd'), await compressZstdFrame([
+    const twoLinesId = SessionId('two-lines')
+    await mkdir(sessionDir(root, undefined, twoLinesId), { recursive: true })
+    await writeFile(logPath(root, undefined, twoLinesId, 'zstd'), await compressZstdFrame([
       JSON.stringify(toHeaderLine(meta('two-lines'))),
       JSON.stringify({ type: 'turn/start' }),
       '',
@@ -411,8 +417,9 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
 
   it('rejects missing, empty, and checksum-corrupt header frames on targeted reads', async () => {
     const root = await freshRoot()
-    const bucket = sessionDir(root, undefined)
-    await mkdir(bucket, { recursive: true })
+    for (const id of ['partial-only', 'empty-header', 'bad-checksum']) {
+      await mkdir(sessionDir(root, undefined, SessionId(id)), { recursive: true })
+    }
     await writeFile(logPath(root, undefined, SessionId('partial-only'), 'zstd'), MAGIC)
     await writeFile(logPath(root, undefined, SessionId('empty-header'), 'zstd'), await compressZstdFrame(''))
     const corruptHeader = Buffer.from(await compressZstdFrame(`${JSON.stringify(toHeaderLine(meta('bad-checksum')))}\n`))
@@ -453,7 +460,7 @@ describe('SessionPersistenceJsonl: encoding selection', () => {
     expect(await ctx.sessionPersistence.list()).toEqual([])
 
     const loadHeader = meta('late-raw-load', '/late')
-    await mkdir(sessionDir(root, loadHeader.cwd), { recursive: true })
+    await mkdir(sessionDir(root, loadHeader.cwd, loadHeader.id), { recursive: true })
     await writeFile(logPath(root, loadHeader.cwd, loadHeader.id, 'none'), [
       JSON.stringify(toHeaderLine(loadHeader)),
       ...oneTurnLog().map(e => JSON.stringify(e)),
@@ -471,13 +478,13 @@ describe('SessionPersistenceJsonl: encoding selection', () => {
     await ctx.sessionPersistence.list()
     const header = meta('late-raw-materialize', '/late')
     await ctx.sessionPersistence.create(header)
-    await mkdir(sessionDir(root, header.cwd), { recursive: true })
+    await mkdir(sessionDir(root, header.cwd, header.id), { recursive: true })
     await writeFile(logPath(root, header.cwd, header.id, 'none'), [
       JSON.stringify(toHeaderLine(header)),
       ...oneTurnLog().map(e => JSON.stringify(e)),
       '',
     ].join('\n'))
     await expect(ctx.sessionPersistence.append(header.id, oneTurnLog())).rejects.toThrow(/uses \.jsonl/)
-    expect((await readdir(sessionDir(root, header.cwd))).some(name => name.endsWith('.jsonl.zstd'))).toBe(false)
+    expect((await readdir(sessionDir(root, header.cwd, header.id))).some(name => name.endsWith('.jsonl.zstd'))).toBe(false)
   })
 })

packages/session-persistence/session-persistence/tests/coordinator-contract.ts

@@ -683,7 +683,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       const fix = await makeFixture()
       const { ctx, fiber } = await freshCtx(fix)
       try {
-        // Ownerless state created WITHOUT a cwd (the no-cwd bucket).
+        // Ownerless state created WITHOUT a cwd (the `_no-cwd` project directory).
         await ctx.sessionPersistence.create(meta('no-cwd-state'))
         // A live session reusing the id but WITH cwd WORK is a cwd mismatch
         // (undefined vs WORK) and must be rejected.

packages/support/acp-snapshot/src/harness.ts

@@ -535,39 +535,29 @@ function latestOpenTurn(content: string): number | undefined {
  * `parentSession`) leads, then each subagent child by ascending `createdAt`.
  *
  * Snapshot configs select the JSONL backend's raw mode, which lays sessions
- * out as `<root>/<cwd-bucket>/<encoded-id>.jsonl` (one bucket per cwd). A
- * parent and its same-cwd in-process child land in the SAME bucket, so
- * collecting all files across all buckets catches both. Returns `[]` if no log
- * was produced (a no-session scenario).
+ * out as `<root>/<project>/<session-id>/session.jsonl`. Recursive collection
+ * catches the primary and every child session. Returns `[]` if no log was
+ * produced (a no-session scenario).
  */
 async function harvestSessionLogs(root: string): Promise<HarvestedLog[]> {
-  let cwdDirs: string[]
+  let files: string[]
   try {
-    cwdDirs = await readdir(root)
+    files = await readdir(root, { recursive: true })
   } catch {
     return []
   }
   const logs: HarvestedLog[] = []
-  for (const dir of cwdDirs) {
-    const sub = join(root, dir)
-    let files: string[]
-    try {
-      files = await readdir(sub)
-    } catch {
-      continue
-    }
-    for (const f of files) {
-      if (!f.endsWith('.jsonl')) continue
-      const content = await readFile(join(sub, f), 'utf8')
-      const firstLine = content.split('\n').find(line => line.trim().length > 0) ?? '{}'
-      const header = JSON.parse(firstLine) as { id?: unknown; createdAt?: unknown; parentSession?: unknown }
-      logs.push({
-        id: typeof header.id === 'string' ? header.id : '',
-        createdAt: typeof header.createdAt === 'number' ? header.createdAt : 0,
-        ...typeof header.parentSession === 'string' ? { parentSession: header.parentSession } : {},
-        content,
-      })
-    }
+  for (const file of files) {
+    if (basename(file) !== 'session.jsonl') continue
+    const content = await readFile(join(root, file), 'utf8')
+    const firstLine = content.split('\n').find(line => line.trim().length > 0) ?? '{}'
+    const header = JSON.parse(firstLine) as { id?: unknown; createdAt?: unknown; parentSession?: unknown }
+    logs.push({
+      id: typeof header.id === 'string' ? header.id : '',
+      createdAt: typeof header.createdAt === 'number' ? header.createdAt : 0,
+      ...typeof header.parentSession === 'string' ? { parentSession: header.parentSession } : {},
+      content,
+    })
   }
   // Primary (no parentSession) first, then children by ascending createdAt. A
   // scenario has exactly one top-level session. In the synchronous cut sibling

packages/support/acp-snapshot/tests/fixtures/fake-acp-agent.ts

@@ -23,9 +23,9 @@ import { dirname, join } from 'node:path'
 import { randomUUID } from 'node:crypto'
 import { createInterface } from 'node:readline'
 
-/** One scripted session log: a file path under the sessions root plus its JSONL lines. */
+/** One scripted session log: a transcript path under the sessions root plus its JSONL lines. */
 interface ScriptedLog {
-  /** Path relative to `$DSH_SNAPSHOT_SESSIONS_ROOT`, e.g. `bucket/a.jsonl` (an empty dir segment is invalid). */
+  /** Path relative to `$DSH_SNAPSHOT_SESSIONS_ROOT`, e.g. `project/session/session.jsonl`. */
   file: string
   /**
    * The JSONL records. String templates `{{CWD}}` and `{{SID}}` are replaced
@@ -61,7 +61,7 @@ interface Behavior {
   logs?: ScriptedLog[]
   /** Leave a stray FILE directly under the sessions root (harvest must skip it). */
   strayRootFile?: boolean
-  /** Leave a stray non-`.jsonl` file inside a bucket (harvest must skip it). */
+  /** Leave a stray non-transcript file inside a project directory (harvest must skip it). */
   strayBucketFile?: boolean
   /** Delete the sessions root entirely (harvest must yield no logs). */
   deleteSessionsRoot?: boolean
@@ -125,7 +125,7 @@ function instantiate(value: unknown): unknown {
 
 /** Persist an open turn so cancellation tests wait on agent state, not presentation output. */
 function persistParkedTurnStart(): void {
-  parkedTurnLog = join(sessionsRoot, 'ready', 'open.jsonl')
+  parkedTurnLog = join(sessionsRoot, 'ready', sessionId, 'session.jsonl')
   mkdirSync(dirname(parkedTurnLog), { recursive: true })
   writeFileSync(parkedTurnLog, [
     JSON.stringify({ type: 'session', version: 0, id: sessionId, createdAt: 1, cwd: sessionCwd, delegationDepth: 0 }),

packages/support/acp-snapshot/tests/fixtures/record-suite/rec-child/behavior.json

@@ -1,11 +1,11 @@
 {
   "prompt": "respond",
   "logs": [
-    { "file": "b/parent.jsonl", "lines": [
+    { "file": "b/parent/session.jsonl", "lines": [
       { "type": "session", "id": "{{SID}}", "createdAt": 700, "cwd": "{{CWD}}", "delegationDepth": 0 },
       { "type": "request/header", "seq": 0, "time": 3, "data": { "header": { "config": { "model": "fake" }, "system": "SYS PROMPT", "tools": [{ "name": "t1", "description": "D1", "parameters": { "type": "object" } }] }, "reason": "initial" } }
     ]},
-    { "file": "b/child.jsonl", "lines": [
+    { "file": "b/child/session.jsonl", "lines": [
       { "type": "session", "id": "abababab-cdcd-4efe-8ada-badabadabada", "createdAt": 800, "cwd": "{{CWD}}", "parentSession": "{{SID}}", "delegationDepth": 1 },
       { "type": "request/header", "seq": 0, "time": 2, "data": { "header": { "config": { "model": "fake" }, "system": "SYS PROMPT", "tools": [{ "name": "t1", "description": "D1", "parameters": { "type": "object" } }] }, "reason": "initial" } }
     ]}

packages/support/acp-snapshot/tests/fixtures/record-suite/rec-pin/behavior.json

@@ -1,7 +1,7 @@
 {
   "prompt": "respond",
   "logs": [{
-    "file": "b/main.jsonl",
+    "file": "b/main/session.jsonl",
     "lines": [
       { "type": "session", "id": "{{SID}}", "createdAt": 600, "cwd": "{{CWD}}", "delegationDepth": 0 },
       { "type": "request/header", "seq": 0, "time": 4, "data": { "header": { "config": { "model": "fake" }, "system": "SYS PROMPT", "tools": [{ "name": "t1", "description": "D1", "parameters": { "type": "object" } }] }, "reason": "initial" } }

packages/support/acp-snapshot/tests/fixtures/suite/authored-error/behavior.json

@@ -1,7 +1,7 @@
 {
   "prompt": "error",
   "logs": [{
-    "file": "b/main.jsonl",
+    "file": "b/main/session.jsonl",
     "lines": [
       { "type": "session", "id": "{{SID}}", "createdAt": 500, "cwd": "{{CWD}}", "delegationDepth": 0 },
       { "type": "turn/end", "seq": 1, "time": 9, "data": { "error": "model exploded" } }

packages/support/acp-snapshot/tests/fixtures/suite/blocked-log/behavior.json

@@ -1,7 +1,7 @@
 {
   "prompt": "error",
   "logs": [{
-    "file": "b/main.jsonl",
+    "file": "b/main/session.jsonl",
     "lines": [
       { "type": "session", "id": "{{SID}}", "createdAt": 400, "cwd": "{{CWD}}", "delegationDepth": 0 },
       { "type": "hook/result", "seq": 1, "time": 8, "data": { "decision": "block", "durationMs": 37 } }

packages/support/acp-snapshot/tests/fixtures/suite/pin-turn/behavior.json

@@ -1,7 +1,7 @@
 {
   "prompt": "respond",
   "logs": [{
-    "file": "b/main.jsonl",
+    "file": "b/main/session.jsonl",
     "lines": [
       { "type": "session", "id": "{{SID}}", "createdAt": 100, "cwd": "{{CWD}}", "delegationDepth": 0 },
       { "type": "request/header", "seq": 0, "time": 100, "data": { "header": { "config": { "model": "fake" }, "system": "SYS PROMPT", "tools": [{ "name": "t1", "description": "D1", "parameters": { "type": "object" } }] }, "reason": "initial" } },

packages/support/acp-snapshot/tests/fixtures/suite/plain-turn/behavior.json

@@ -2,12 +2,12 @@
   "prompt": "respond",
   "echoWorkspace": true,
   "logs": [
-    { "file": "b/parent.jsonl", "lines": [
+    { "file": "b/parent/session.jsonl", "lines": [
       { "type": "session", "id": "{{SID}}", "createdAt": 200, "cwd": "{{CWD}}", "delegationDepth": 0 },
       { "type": "request/header", "seq": 0, "time": 5, "data": { "header": { "config": { "model": "fake" }, "system": "SYS PROMPT", "tools": [{ "name": "t1", "description": "D1", "parameters": { "type": "object" } }] }, "reason": "initial" } },
       { "type": "assistant/chunk", "seq": 1, "time": 5, "data": { "turn": 1, "step": 1, "chunk": { "type": "text-delta", "index": 0, "text": "hi" } } }
     ]},
-    { "file": "b/child.jsonl", "lines": [
+    { "file": "b/child/session.jsonl", "lines": [
       { "type": "session", "id": "eeeeeeee-1111-4222-8333-444444444444", "createdAt": 300, "cwd": "{{CWD}}", "parentSession": "{{SID}}", "delegationDepth": 1 },
       { "type": "request/header", "seq": 0, "time": 6, "data": { "header": { "config": { "model": "fake" }, "system": "SYS PROMPT", "tools": [{ "name": "t1", "description": "D1", "parameters": { "type": "object" } }] }, "reason": "initial" } }
     ]}

packages/support/acp-snapshot/tests/harness.spec.ts

@@ -380,7 +380,7 @@ describe('runScenario', () => {
     const { fixtureFile } = await scenario({
       permissionProbe: true,
       logs: [{
-        file: 'bucket/main.jsonl',
+        file: 'project/main/session.jsonl',
         lines: [
           { type: 'session', id: '{{SID}}', createdAt: 42, cwd: '{{CWD}}' },
           { type: 'turn/start', seq: 1, time: 9, data: { turn: 1 } },
@@ -546,7 +546,7 @@ describe('runScenario', () => {
       prompt: 'hang-until-cancel',
       persistLogsOnCancel: true,
       logs: [{
-        file: 'bucket/session.jsonl',
+        file: 'project/main/session.jsonl',
         lines: [
           { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
           { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'aborted' } } },
@@ -565,7 +565,7 @@ describe('runScenario', () => {
       prompt: 'hang-until-cancel',
       persistLogsOnCancel: true,
       logs: [{
-        file: 'bucket/session.jsonl',
+        file: 'project/main/session.jsonl',
         lines: [
           { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
           { type: 'turn/start', seq: 0, time: 1, data: { turn: 3 } },
@@ -596,7 +596,7 @@ describe('runScenario', () => {
       prompt: 'hang-until-cancel',
       persistLogsOnCancel: true,
       logs: [{
-        file: 'bucket/session.jsonl',
+        file: 'project/main/session.jsonl',
         lines: [
           { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
           { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
@@ -618,7 +618,7 @@ describe('runScenario', () => {
       prompt: 'hang-until-cancel',
       persistLogsOnCancel: true,
       logs: [{
-        file: 'bucket/session.jsonl',
+        file: 'project/main/session.jsonl',
         lines: [
           { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
           { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
@@ -642,7 +642,7 @@ describe('runScenario', () => {
         prompt: 'hang-until-cancel',
         persistLogsOnCancel: true,
         logs: [{
-          file: 'bucket/session.jsonl',
+          file: 'project/main/session.jsonl',
           lines: [
             { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
             { type: 'turn/start', seq: 0, time: 1, data: turn === undefined ? {} : { turn } },
@@ -673,7 +673,7 @@ describe('runScenario', () => {
       prompt: 'hang-until-cancel',
       persistLogsOnCancel: true,
       logs: [{
-        file: 'bucket/session.jsonl',
+        file: 'project/main/session.jsonl',
         lines: [
           { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
           { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
@@ -823,11 +823,11 @@ describe('runScenario', () => {
         // File names chosen so readdir feeds the sort children-first AND
         // parent-in-the-middle: the comparator then sees a parent on both
         // sides of a pair, plus the same-createdAt (localeCompare) tiebreak.
-        { file: 'b1/aa-child-c.jsonl', lines: [{ type: 'session', id: 'cccccccc-0000-4000-8000-000000000000', createdAt: 500, parentSession: '{{SID}}' }] },
-        { file: 'b1/bb-parent.jsonl', lines: [{ type: 'session', id: '{{SID}}', createdAt: 900 }] },
-        { file: 'b1/cc-child-a.jsonl', lines: [{ type: 'session', id: 'aaaaaaaa-0000-4000-8000-000000000000', createdAt: 500, parentSession: '{{SID}}' }] },
+        { file: 'b1/aa-child-c/session.jsonl', lines: [{ type: 'session', id: 'cccccccc-0000-4000-8000-000000000000', createdAt: 500, parentSession: '{{SID}}' }] },
+        { file: 'b1/bb-parent/session.jsonl', lines: [{ type: 'session', id: '{{SID}}', createdAt: 900 }] },
+        { file: 'b1/cc-child-a/session.jsonl', lines: [{ type: 'session', id: 'aaaaaaaa-0000-4000-8000-000000000000', createdAt: 500, parentSession: '{{SID}}' }] },
         // Missing id/createdAt fall back to ''/0; earliest child by createdAt.
-        { file: 'b2/orphan-fields.jsonl', lines: [{ type: 'session', parentSession: '{{SID}}' }] },
+        { file: 'b2/orphan/session.jsonl', lines: [{ type: 'session', parentSession: '{{SID}}' }] },
       ],
     })
     const result = await runScenario(
@@ -844,7 +844,7 @@ describe('runScenario', () => {
   })
 
   it('treats an empty log file as a header-less primary with default fields', { timeout: 20_000 }, async () => {
-    const { fixtureFile } = await scenario({ logs: [{ file: 'b/empty.jsonl', lines: [] }] })
+    const { fixtureFile } = await scenario({ logs: [{ file: 'b/empty/session.jsonl', lines: [] }] })
     const result = await runScenario(
       { steps: boot },
       { agent: AGENT, mode: 'replay', fixtureFile },