Merge PR6 (snapshot golden removal + PR5 trace-events) into PR7

packages/core/agent-loop/src/loop.ts

@@ -38,8 +38,8 @@ function toError(error: unknown): CodedError {
  * caller's try/catch), OR end the stream with a finish-error/aborted chunk
  * (the only option for adapters that can't throw mid-stream, e.g.
  * library-backed ones). This translates the latter into a thrown step error
- * so the turn ends error/aborted with a logged `error` event, never as a
- * normal `completed` assistant message.
+ * so the turn ends error/aborted (the failure recorded on `turn/end.reason`),
+ * never as a normal `completed` assistant message.
  *
  * `FinishReason` is merge-extensible (plugins/adapters can add `kind`s), so
  * the switch handles the known terminal-failure kinds and treats every other
@@ -322,15 +322,22 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
   const failTurn = (err: CodedError): void => {
     if (errorReported) return
     errorReported = true
-    // Set `reason` here so the durable failure is captured before closeTurn
-    // appends turn/end. The step number rides along so the operational error's
-    // location survives in the durable log.
-    reason = { kind: 'error', step, ...errorData(err) }
+    // Set the error reason ONLY while the turn is still open — closeTurn appends
+    // turn/end with it. If the turn has already ended (the only way here: a
+    // throwing agent/turn-end listener after closeTurn(true) already appended
+    // turn/end), the reason can no longer affect the durable log, so log the late
+    // throw directly instead — otherwise the listener exception would vanish.
+    if (!turnEnded) {
+      reason = { kind: 'error', step, ...errorData(err) }
+    } else {
+      ctx.logger.warn(`agent "${agent.id}": agent/turn-end listener threw after turn ${turn} closed: ${err.message}`)
+    }
     try {
       ctx.emit('agent/error', agent, turn, step, err)
     } catch {
-      // contained: the error is already captured on `reason`; a throwing
-      // agent/error listener must not prevent the turn from closing.
+      // contained: the error is already captured (on `reason`, or via the logger
+      // above); a throwing agent/error listener must not prevent the turn from
+      // closing.
     }
   }
 
@@ -609,7 +616,14 @@ async function runStep(
   let message: Message = assembler.message()
   message = await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message))
 
-  session.append('assistant/message', { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) })
+  // Same content-or-usage guard as the max-tokens branch: a step that finishes
+  // with neither assembled content nor usage (e.g. a bare `stop` finish that
+  // streamed nothing) records no assistant/message — an empty-content message
+  // exists only to host usage, and deriveMessages() skips it either way, so
+  // appending one with no usage would be a pure trace-only row.
+  if (message.content.length > 0 || assembler.usage) {
+    session.append('assistant/message', { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) })
+  }
 
   // --- Tool execution (sequential; parallel execution is a TODO) ---
   // ToolRegistry.execute converts tool failures (including aborts) into

packages/core/agent-loop/tests/loop.spec.ts

@@ -487,6 +487,25 @@ describe('agent loop', () => {
     expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
   })
 
+  it('appends no assistant/message for a normal stop finish with empty content and no usage', async () => {
+    // A clean `stop` finish that streamed nothing assembled (no blocks) and
+    // carried no usage chunk has nothing to record: the content-or-usage guard
+    // on the normal step path suppresses a pure trace-only empty assistant/message.
+    const adapter = new MockAdapter([[{ type: 'finish', reason: { kind: 'stop' } }]])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+
+    const reasons: TurnEndReason[] = []
+    ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    expect(reasons).toEqual([{ kind: 'completed' }])
+    expect(agent.session.events.some(e => e.type === 'assistant/message')).toBe(false)
+    expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
+  })
+
   it('keeps safe max-tokens assistant content while dropping truncated tool calls', async () => {
     const callId = CallId('c1')
     const adapter = new MockAdapter([[

packages/core/agent-loop/tests/review-fixes.spec.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import LlmService, { CallId, MessageSource, StreamChunk } from '@deepseek-ai/dsh-llm'
 import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
@@ -806,6 +806,7 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
     ctx.on('agent/turn-end', () => { if (!threw) { threw = true; throw new Error('boom turn-end') } })
     const errors: Error[] = []
     ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
+    const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
 
     send(agent, 'go')
     await waitForIdle(ctx, agent)
@@ -815,6 +816,9 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
     expect(c.errors).toBe(0) // NO session error event (it would be post-turn/end)
     expect(agent.session.events.at(-1)?.type).toBe('turn/end') // last event is the boundary
     expect(errors.map(e => e.message)).toEqual(['boom turn-end']) // surfaced via agent/error
+    // The late throw is also logged directly: failTurn's turn-already-ended
+    // branch warns so a throwing turn-end listener after turn/end never vanishes.
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining('agent/turn-end listener threw after turn 1 closed'))
     // The whole log is loadable (nothing dropped): a fresh replay sees the turn.
     const replay = new Session(SessionId('replay'), [...agent.session.events])
     expect(replay.deriveMessages().map(m => m.role)).toEqual(['user', 'assistant'])

packages/core/session/README.md

@@ -37,7 +37,7 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
 - `session.append(type, data): SessionEvent` — synchronous, never blocks on I/O. **Throws** if `data` is not losslessly JSON-serializable (BigInt, function, symbol, undefined, non-finite number, circular ref, or an exotic object like Map/Set/Date) — the event log is the durable source of truth, so this invariant is enforced at the source (exported as `isJsonValue` for backends to reuse on their replay/fork entry points).
 - `session.deriveMessages(): Message[]` — derive the LLM message history from the event log. Raw `assistant/chunk` events are skipped; `context/message` and `steering/message` render as tagged synthetic user messages.
 - `session.events`, `session.seq`, `session.id`
-- `session.header: SessionHeader` — immutable creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`). Kept out of the event log (a storage concern, not replayable state); a minimal v1 header is synthesized for bare `Session` construction.
+- `session.header: SessionHeader` — immutable creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`). Kept out of the event log (a storage concern, not replayable state); a minimal header (stamped with the current `SESSION_FORMAT_VERSION`) is synthesized for bare `Session` construction.
 
 ### Metadata types (`types.ts`)
 

packages/core/session/src/index.ts

@@ -9,7 +9,7 @@
 import { Context, Service } from 'cordis'
 import { isAbsolute } from 'node:path'
 import type { ContentBlock, Message, MessageSource } from '@deepseek-ai/dsh-llm'
-import { SessionId } from './types.ts'
+import { SESSION_FORMAT_VERSION, SessionId } from './types.ts'
 import type { CreateSessionOptions, SessionEvent, SessionEventMap, SessionEventType, SessionHeader } from './types.ts'
 import { isJsonValue } from './json.ts'
 
@@ -79,9 +79,10 @@ export class Session {
   /**
    * Immutable creation metadata (format version, cwd, lineage). Supplied by
    * the store via `ctx.sessions.create()`. When a `Session` is constructed
-   * bare (tests, ad-hoc replay), a minimal v1 header is synthesized so
-   * `session.header` is always present. Kept out of the event log — it is a
-   * storage concern, not replayable conversation state.
+   * bare (tests, ad-hoc replay), a minimal header is synthesized (stamped with
+   * the current {@link SESSION_FORMAT_VERSION}) so `session.header` is always
+   * present. Kept out of the event log — it is a storage concern, not
+   * replayable conversation state.
    */
   readonly header: SessionHeader
 
@@ -112,7 +113,7 @@ export class Session {
       // structuredClone can never hit a non-cloneable value here.
       this.log = seed.map(event => structuredClone(event))
     }
-    this.header = header ?? { version: 1, id, createdAt: Date.now() }
+    this.header = header ?? { version: SESSION_FORMAT_VERSION, id, createdAt: Date.now() }
   }
 
   get events(): readonly SessionEvent[] {
@@ -180,8 +181,7 @@ export class Session {
     const messages: Message[] = []
     for (const event of this.log) {
       // Intentionally non-exhaustive: only message-producing events derive
-      // history; turn/step boundaries, chunks, usage, and errors are
-      // trace/replay data.
+      // history; turn/step boundaries and chunks are trace/replay data.
       // eslint-disable-next-line @typescript-eslint/switch-exhaustiveness-check
       switch (event.type) {
         case 'user/message': {
@@ -284,7 +284,7 @@ export class SessionStore extends Service {
       throw new Error(`session cwd must be an absolute path, got "${cwd}"`)
     }
     const header: SessionHeader = {
-      version: 1,
+      version: SESSION_FORMAT_VERSION,
       id: sessionId,
       createdAt: options?.meta?.createdAt ?? Date.now(),
       ...cwd !== undefined ? { cwd } : {},

packages/core/session/src/types.ts

@@ -9,6 +9,23 @@ export function SessionId(id: string): SessionId {
   return id as SessionId
 }
 
+/**
+ * The on-disk session format version, stamped into every newly-written
+ * {@link SessionHeader} and enforced by every persistence backend on load. The
+ * single source of truth for the version — write sites and the load-time check
+ * all read it.
+ *
+ * It is **`0`** deliberately: while the harness is unreleased the on-disk format
+ * is **unstable / pre-release, with no compatibility implied**. Breaking changes
+ * to the persisted {@link SessionEventMap} shape (folding fields onto an event,
+ * removing a variant, …) happen freely and do NOT bump this — v0 absorbs all
+ * pre-release churn, and a backend simply REJECTS any log not at v0 (there is no
+ * migration; no persisted user data exists to preserve). A real, monotonically
+ * bumped version policy begins at the first tagged release, when a specific
+ * format boundary becomes worth distinguishing.
+ */
+export const SESSION_FORMAT_VERSION = 0
+
 /**
  * Immutable session metadata — written once at creation and never rewritten.
  *
@@ -19,7 +36,11 @@ export function SessionId(id: string): SessionId {
  * metadata) writes such a header.
  */
 export interface SessionHeader {
-  /** On-disk format version; a persistence backend rejects unknown versions. */
+  /**
+   * On-disk format version, stamped from {@link SESSION_FORMAT_VERSION} when the
+   * session is created. A persistence backend rejects any other version on load
+   * (no migration — see the constant).
+   */
   version: number
   /** The session's id (mirrors the {@link Session}'s id). */
   id: SessionId

packages/core/session/tests/session.spec.ts

@@ -1,7 +1,7 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import { CallId } from '@deepseek-ai/dsh-llm'
-import SessionStore, { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
+import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
 
 describe('Session', () => {
   it('derives message history from the event log', () => {
@@ -255,11 +255,11 @@ describe('SessionStore', () => {
     expect(ctx.sessions.get(SessionId('lifecycle'))).toBeUndefined()
   })
 
-  it('synthesizes a minimal v1 header for a bare-created session', async () => {
+  it('synthesizes a minimal current-version header for a bare-created session', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
     const session = ctx.sessions.create(SessionId('plain'))
-    expect(session.header).toMatchObject({ version: 1, id: 'plain' })
+    expect(session.header).toMatchObject({ version: SESSION_FORMAT_VERSION, id: 'plain' })
     expect(typeof session.header.createdAt).toBe('number')
     expect(session.header.cwd).toBeUndefined()
     expect(session.header.parentSession).toBeUndefined()
@@ -272,7 +272,7 @@ describe('SessionStore', () => {
       meta: { cwd: '/work/project', parentSession: SessionId('parent') },
     })
     expect(session.header).toMatchObject({
-      version: 1,
+      version: SESSION_FORMAT_VERSION,
       id: 'child',
       cwd: '/work/project',
       parentSession: 'parent',
@@ -288,9 +288,9 @@ describe('SessionStore', () => {
     expect(ctx.sessions.get(SessionId('rel'))).toBeUndefined()
   })
 
-  it('a bare Session() constructed without the store still exposes a v1 header', () => {
+  it('a bare Session() constructed without the store still exposes a current-version header', () => {
     const session = new Session(SessionId('bare'))
-    expect(session.header).toMatchObject({ version: 1, id: 'bare' })
+    expect(session.header).toMatchObject({ version: SESSION_FORMAT_VERSION, id: 'bare' })
     expect(typeof session.header.createdAt).toBe('number')
   })