Merge branch 'codex/code-mode-typed-results' into codex/code-mode-complete-result-card

# Conflicts:
#	examples/acp-agent/tests/snapshots/advanced-toolchain/session.jsonl
#	examples/acp-agent/tests/snapshots/both-mode-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/code-mode-turn/session.jsonl
#	examples/acp-agent/tests/snapshots/code-mode-turn/stdout.expected.jsonl
#	examples/acp-agent/tests/snapshots/code-mode-workspace-context/session.jsonl
#	examples/headless-agent/tests/snapshots/advanced-toolchain/session.jsonl
#	examples/headless-agent/tests/snapshots/advanced-toolchain/stream-json.expected.jsonl
#	examples/tui-agent/tests/snapshots/code-mode/terminal.expected.txt

packages/core/README.md

@@ -15,4 +15,4 @@ The session log, system-prompt assembly, tool registry, agent vocabulary, and co
 
 `agent-loop` is the one concrete implementation of the `agent` seam and lives here because it is the harness's default product loop. It runs each driver inside `ctx.agents.withInitiator()`. Extension plugins depend on `agent`, including when they need the initiating Agent, and never on `agent-loop` directly, so the loop stays swappable.
 
-The default composition that wires this spine into a runnable agent lives in [`examples/agent-spine-demo`](../examples/agent-spine-demo/README.md): one bundle plugin that loads the control spine plus selected default capabilities (`timer` + `llm` + sessions + system-prompt + tools + agents + invariants + the local [skill family](../skill/README.md) + `tool-bash` + workspace-context + `agent-loop`) and forwards `agent-loop`'s `agents` list as its own config. It sits in `examples/` — ready-to-run demo/reference bundles — not in `core/`: `core/` ships the swappable spine pieces, while a demo bundle picks one concrete composition of them and adds a front door.
+The default composition that wires this spine into a runnable agent lives in [`examples/agent-spine-demo`](../examples/agent-spine-demo/README.md): one bundle plugin that loads the control spine plus selected default capabilities (`timer` + `llm` + sessions + fallback session titles + system-prompt + tools + agents + invariants + the local [skill family](../skill/README.md) + `tool-bash` + workspace-context + `agent-loop`) and forwards `agent-loop`'s `agents` list as its own config. It sits in `examples/` — ready-to-run demo/reference bundles — not in `core/`: `core/` ships the swappable spine pieces, while a demo bundle picks one concrete composition of them and adds a front door.

packages/core/agent-loop/README.md

@@ -29,7 +29,7 @@ The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loo
 
 ### Invariant companion
 
-The optional `@deepseek-ai/dsh-agent-loop/invariant` companion registers request reconstruction with `ctx.invariants`. The loop marks each request with an internal non-enumerable identity before freezing it; the companion then requires a live session and independently rebuilds the message boundary and folded request header from the log. Direct one-shot calls remain outside this contract even when callers freeze them or attach a session id.
+The optional `@deepseek-ai/dsh-agent-loop/invariant` companion registers request reconstruction with `ctx.invariants`. The loop records each exact frozen request in the process-local identity set owned by `dsh-llm`; the companion then requires a live session and independently rebuilds the message boundary and folded request header from the log. Direct one-shot calls remain outside this contract even when callers freeze them or attach a session id.
 
 ### Configuration (schemastery)
 

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

@@ -4,10 +4,9 @@
  */
 
 import type { Context } from 'cordis'
-import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
+import { isAgentLoopRequest, type GenerateOptions } from '@deepseek-ai/dsh-llm'
 import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
 import { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
-import { isLoopRequest } from './request-marker.ts'
 
 const PACKAGE_NAME = '@deepseek-ai/dsh-agent-loop'
 
@@ -21,7 +20,7 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant
   // Prepend prevents a short-circuiting replay listener from silencing the
   // check; correctness itself comes from the sequence-bounded reconstruction.
   ctx.on('llm/stream', (options: GenerateOptions, next) => {
-    if (!isLoopRequest(options)) return next()
+    if (!isAgentLoopRequest(options)) return next()
     if (!Object.isFrozen(options)) fail('a loop-built request must be frozen')
     if (options.sessionId === undefined) fail('a loop-built request must carry a session id')
     const session = ctx.sessions.get(options.sessionId)

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

@@ -8,14 +8,13 @@
 import type { Context } from 'cordis'
 import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message } from '@deepseek-ai/dsh-llm'
 import { isDeepStrictEqual } from 'node:util'
-import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf } from '@deepseek-ai/dsh-llm'
+import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf, markAgentLoopRequest } from '@deepseek-ai/dsh-llm'
 import { agentEvents, agentInterruptReasonOf, assembleContextFor } from '@deepseek-ai/dsh-agent'
 import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
 import { canonicalHeader } from '@deepseek-ai/dsh-session'
 import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
 import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
 import type { TransmissionLog } from './request-log.ts'
-import { markLoopRequest } from './request-marker.ts'
 import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
 import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
 import type {} from '@deepseek-ai/dsh-tools'
@@ -598,7 +597,7 @@ async function runStep(
   recordRequestHeader(session, transmission, header)
 
   // Freeze the logged header plus boundary snapshot; the prefix precedes derived history.
-  const request: GenerateOptions = deepFreeze(markLoopRequest({
+  const request: GenerateOptions = markAgentLoopRequest(deepFreeze({
     provider: header.config.provider,
     model: header.config.model,
     messages: [...header.messagePrefix ?? [], ...boundaryMessages],

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

@@ -1,22 +0,0 @@
-/** Internal identity shared by the independently bundled loop and invariant companion. */
-
-const LOOP_REQUEST = Symbol.for('@deepseek-ai/dsh-agent-loop/request')
-
-/**
- * Mark a request as owned by the agent loop before it is frozen.
- * @param request - mutable request object being assembled by the loop.
- * @returns the same request with a non-enumerable loop identity.
- */
-export function markLoopRequest<T extends object>(request: T): T {
-  Object.defineProperty(request, LOOP_REQUEST, { value: true })
-  return request
-}
-
-/**
- * Test whether a request carries the agent loop's internal identity.
- * @param request - request observed at the LLM stream boundary.
- * @returns whether the loop marked this exact request object.
- */
-export function isLoopRequest(request: object): boolean {
-  return Reflect.get(request, LOOP_REQUEST) === true
-}

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

@@ -3,7 +3,7 @@ import { Context } from 'cordis'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import InvariantService from '@deepseek-ai/dsh-invariants'
 import * as AgentLoopInvariant from '@deepseek-ai/dsh-agent-loop/invariant'
-import { markLoopRequest } from '../src/request-marker.ts'
+import { markAgentLoopRequest, type GenerateOptions } from '@deepseek-ai/dsh-llm'
 
 async function setup(): Promise<Context> {
   const ctx = new Context()
@@ -18,7 +18,8 @@ function dispatch(ctx: Context, options: unknown): void {
 }
 
 function loopRequest<T extends object>(options: T): Readonly<T> {
-  return Object.freeze(markLoopRequest(options))
+  markAgentLoopRequest(options as GenerateOptions)
+  return Object.freeze(options)
 }
 
 async function requestSetup() {
@@ -94,8 +95,10 @@ describe('request-reconstruction invariant', () => {
 
   it('rejects malformed requests carrying the loop marker', async () => {
     const { ctx, session } = await requestSetup()
+    const messages: GenerateOptions['messages'] = []
+    Object.freeze(messages)
     expect(() => {
-      dispatch(ctx, markLoopRequest({ model: 'm', messages: Object.freeze([]), sessionId: session.id }))
+      dispatch(ctx, markAgentLoopRequest({ provider: 'p', model: 'm', messages, sessionId: session.id }))
     }).toThrow(/request must be frozen/)
     expect(() => {
       dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([]) }))

packages/core/session/README.md

@@ -12,6 +12,8 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
 
 - `ctx.sessions.create(id?, { seed?, meta? }?)` validates and detaches durable seed/header data, fills the version and id, defaults `createdAt` to now, publishes the session, and binds it to the calling fiber. Persisted reconstruction supplies its original `createdAt`, `seedLength`, and `delegationDepth`.
 - `ctx.sessions.flush(session)` dispatches the awaited parallel durability checkpoint through the session's captured scope. Every listener starts and the call waits for all to settle before reporting failure; unpublished, detached, and stale objects reject.
+- `ctx.sessions.appendOutOfBand(session, type, data, trigger)` accepts only plugin event types opted into `OutOfBandSessionEventMap`. It appends directly inside an open turn; otherwise it atomically opens a zero-step plugin turn, appends, closes, and flushes. A target failure still closes and flushes the synthetic turn, and detach is deferred until the sequence settles.
+- `findLastMessageTurnEnd(events)` pairs message-triggered starts with their ends and returns the latest matched `turn/end`. Outcome consumers use this fold instead of the raw latest turn boundary because a later injection or plugin-owned zero-step turn has its own outcome.
 - `ctx.sessions.fork(source, boundary?, childSessionId?): Session` — Resolve a live session object or id, select a seed through the inclusive `boundary` event seq (default: current last event), require that boundary to be `turn/end`, and create a live child session with lineage metadata.
 - `ctx.sessions.get(id: SessionId): Session | undefined`
 - `ctx.sessions.list(): Session[]`
@@ -66,7 +68,7 @@ Durable values need one accepted representation, not a check followed by a secon
 
 The append-only log's event types, enumerated member by member — payloads, surface badges, provenance — in the generated [persistence log event catalog](../../../docs/persistence-catalog.md). Token accounting reads per-step `assistant/chunk { type: 'usage' }` records and treats `assistant/message.usage` as the committed-step fallback when no usage chunk exists; failed model-request attempts have no assistant message. Provider/model/replay provenance rides on `assistant/message`; an operational error's step is on `turn/end.reason` for `kind: 'error'`, with structured provider facts for a final model-request failure.
 
-Merge-extensible via `SessionEventMap` — a plugin declaration-merges its own types (the compaction seam's `compact/*`, bounded recovery's non-surface `llm/retry`, the hook bridges' `hook/*`); merged members appear in the same catalog.
+Merge-extensible via `SessionEventMap` — a plugin declaration-merges its own types (the compaction seam's `compact/*`, bounded recovery's non-surface `llm/retry`, the hook bridges' `hook/*`); merged members appear in the same catalog. `OutOfBandSessionEventMap` is a separate empty-by-default marker map: an event owner must merge the same key there before `appendOutOfBand()` accepts that log-only type, while surface and lifecycle types remain excluded.
 
 Also defines `TurnTriggerMap` and `TurnEndReasonMap` (merge-extensible sum types for typed turn boundaries — `kind`-tagged instead of strings). A final model-request error retains one structured `LlmFailure`; other turn errors retain message/code, and both identify the failed step.
 

packages/core/session/src/index.ts

@@ -13,7 +13,7 @@ import { scopeOf, scopeTarget } from '@deepseek-ai/dsh-scope'
 import type { Scoped } from '@deepseek-ai/dsh-scope'
 import type { Message } from '@deepseek-ai/dsh-llm'
 import { SESSION_FORMAT_VERSION, SessionId } from './types.ts'
-import type { CreateSessionOptions, EpochHeader, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts'
+import type { CreateSessionOptions, EpochHeader, OutOfBandSessionEventType, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType, TurnTrigger } from './types.ts'
 import { snapshotJsonValue } from './json.ts'
 import { SurfaceManager } from './surface.ts'
 import type { SessionSurface } from './surface.ts'
@@ -27,6 +27,27 @@ export type { SessionSurface, SurfaceFoldReplacement, SurfaceFoldResult } from '
 export { foldSurface, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
 export { canonicalHeader, foldRequestHeader, headerEquals } from './request-header.ts'
 
+/**
+ * Find the latest closed message-triggered turn, excluding injection and
+ * plugin-owned zero-step turns.
+ * @param events - session events, or an owned suffix, to inspect.
+ * @returns the latest matching turn end, or `undefined`.
+ */
+export function findLastMessageTurnEnd(
+  events: readonly SessionEvent[],
+): SessionEvent<'turn/end'> | undefined {
+  const messageTurns = new Set<number>()
+  let latest: SessionEvent<'turn/end'> | undefined
+  for (const event of events) {
+    if (event.type === 'turn/start') {
+      if (event.data.trigger.kind === 'message') messageTurns.add(event.data.turn)
+      continue
+    }
+    if (event.type === 'turn/end' && messageTurns.delete(event.data.turn)) latest = event
+  }
+  return latest
+}
+
 declare module 'cordis' {
   interface Context {
     sessions: SessionStore
@@ -227,6 +248,7 @@ interface SessionEntry {
   announced: boolean
   announcing: boolean
   appending: boolean
+  outOfBand: boolean
   detachRequested: boolean
   detach(): void
 }
@@ -401,7 +423,7 @@ export class Session {
     } finally {
       if (entry !== undefined) {
         entry.appending = false
-        if (entry.detachRequested && !entry.announcing) entry.detach()
+        if (entry.detachRequested && !entry.announcing && !entry.outOfBand) entry.detach()
       }
     }
   }
@@ -685,6 +707,7 @@ export class SessionStore extends Service {
       announced: false,
       announcing: false,
       appending: false,
+      outOfBand: false,
       detachRequested: false,
       detach: () => { this.detachEntered(entry) },
     }
@@ -697,7 +720,7 @@ export class SessionStore extends Service {
       // A lifecycle listener may own the advanced detach capability. Keep the
       // entry and its publication hooks live until synchronous creation or append
       // publication unwinds, then publish the paired disposal edge.
-      if (entry.announcing || entry.appending) {
+      if (entry.announcing || entry.appending || entry.outOfBand) {
         entry.detachRequested = true
         return
       }
@@ -751,7 +774,7 @@ export class SessionStore extends Service {
       }
     } finally {
       entry.announcing = false
-      if (entry.detachRequested && !entry.appending) entry.detach()
+      if (entry.detachRequested && !entry.appending && !entry.outOfBand) entry.detach()
     }
   }
 
@@ -795,6 +818,87 @@ export class SessionStore extends Service {
     if (failure !== undefined) throw failure.reason
   }
 
+  /**
+   * Append one plugin-declared log-only event without borrowing the agent
+   * loop's lifecycle. An open turn receives the event directly and remains
+   * responsible for its ordinary checkpoint. A closed log receives one
+   * zero-step turn around the event, followed by an awaited flush.
+   *
+   * Once the synthetic `turn/start` commits, this method always attempts its
+   * matching `turn/end` and flush, including when the target append fails.
+   * Detachment requested by an event or flush listener is deferred until that
+   * sequence settles, so publication cannot switch from a live scoped session
+   * to an unobserved bare `Session` halfway through the update.
+   *
+   * @param session - exact live session that owns the target log.
+   * @param type - event type opted into {@link OutOfBandSessionEventMap} by its owner.
+   * @param data - typed JSON payload for the target event.
+   * @param trigger - plugin-owned turn trigger used only when the log is closed.
+   * @returns the accepted target event with its assigned sequence and timestamp.
+   * @throws when the session is detached, another out-of-band append is active,
+   *   event acceptance fails, the synthetic turn cannot close, or flushing fails.
+   */
+  async appendOutOfBand<T extends OutOfBandSessionEventType>(
+    session: Session,
+    type: T,
+    data: SessionEventMap[T],
+    trigger: TurnTrigger,
+  ): Promise<SessionEvent<T>> {
+    const entry = this.liveEntryFor(session)
+    if (entry.outOfBand) {
+      throw new Error(`session "${session.id}" already has an out-of-band append in progress`)
+    }
+    entry.outOfBand = true
+    // `T` is excluded from SurfaceEventType by OutOfBandSessionEventType, but
+    // TypeScript does not reduce Session.append's conditional rest parameter
+    // through a generic intersection. Preserve that proven two-argument call
+    // shape without widening the public Session.append overload.
+    const appendLogOnly = session.append.bind(session) as unknown as <K extends OutOfBandSessionEventType>(
+      eventType: K,
+      eventData: SessionEventMap[K],
+    ) => SessionEvent<K>
+    try {
+      const lastBoundary = session.events.findLast(event => event.type === 'turn/start' || event.type === 'turn/end')
+      if (lastBoundary?.type === 'turn/start') {
+        return appendLogOnly(type, data)
+      }
+
+      const lastStart = session.events.findLast(event => event.type === 'turn/start')
+      const turn = (lastStart?.data.turn ?? 0) + 1
+      let accepted: SessionEvent<T> | undefined
+      let failure: unknown
+      let opened = false
+      try {
+        session.append('turn/start', { turn, trigger })
+        opened = true
+        accepted = appendLogOnly(type, data)
+      } catch (error: unknown) {
+        failure = error
+      } finally {
+        if (opened) {
+          // The only target types admitted by OutOfBandSessionEventMap are
+          // log-only plugin events, so the synthetic turn remains open here.
+          session.append('turn/end', { turn, reason: { kind: 'completed' } })
+          try {
+            await this.flush(session)
+          } catch (error: unknown) {
+            if (failure === undefined) failure = error
+          }
+        }
+      }
+      if (failure !== undefined) {
+        // eslint-disable-next-line @typescript-eslint/only-throw-error -- preserve an arbitrary flush-listener rejection exactly
+        throw failure
+      }
+      /* v8 ignore next -- accepted is assigned unless an append failure was captured above. */
+      if (accepted === undefined) throw new Error('out-of-band append completed without an accepted event')
+      return accepted
+    } finally {
+      entry.outOfBand = false
+      if (entry.detachRequested && !entry.announcing && !entry.appending) entry.detach()
+    }
+  }
+
   /** Return the exact live entry; detached/prepared objects reject. */
   private liveEntryFor(session: Session): SessionEntry {
     const entry = attachments.get(session)

packages/core/session/src/types.ts

@@ -274,9 +274,23 @@ export interface SessionEventMap {
   'request/header': { header: EpochHeader; reason: RequestHeaderReason }
 }
 
+/**
+ * Marker map for plugin-owned log-only events accepted by
+ * `SessionStore.appendOutOfBand()`. A plugin extends this map with the same key
+ * it adds to {@link SessionEventMap}; surface and lifecycle events stay
+ * ineligible unless their owner explicitly opts them into this narrow seam.
+ */
+export interface OutOfBandSessionEventMap {}
+
 /** The appendable event-type keys of {@link SessionEventMap}, plugin-merged extensions included. */
 export type SessionEventType = keyof SessionEventMap
 
+/** Plugin-declared non-surface event types accepted by `SessionStore.appendOutOfBand()`. */
+export type OutOfBandSessionEventType = Exclude<
+  Extract<SessionEventType, keyof OutOfBandSessionEventMap>,
+  SurfaceEventType
+>
+
 /**
  * The subset of {@link SessionEventType} values whose events produce LLM
  * messages and are eligible to appear on the ordered surface. Only these

packages/core/session/tests/out-of-band.spec.ts

@@ -0,0 +1,226 @@
+import { Context } from 'cordis'
+import { describe, expect, it } from 'vitest'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
+
+declare module '@deepseek-ai/dsh-session' {
+  interface SessionEventMap {
+    'test/log-only': { value: string }
+  }
+
+  interface OutOfBandSessionEventMap {
+    'test/log-only': true
+  }
+
+}
+
+const updateTrigger = { kind: 'injection', source: { kind: 'plugin', plugin: 'test' } } as const
+
+describe('SessionStore.appendOutOfBand', () => {
+  it('joins an open turn without adding a boundary or flushing it', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('open'))
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+    session.append('turn/start', {
+      turn: 1,
+      trigger: { kind: 'message', source: { kind: 'user' } },
+    })
+
+    const event = await ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'inside' },
+      updateTrigger,
+    )
+
+    expect(event).toMatchObject({ type: 'test/log-only', seq: 1, data: { value: 'inside' } })
+    expect(session.events.map(item => item.type)).toEqual(['turn/start', 'test/log-only'])
+    expect(flushes).toBe(0)
+  })
+
+  it('wraps a closed log in one zero-step turn and flushes the balanced update', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('closed'))
+    const flushedTypes: string[][] = []
+    ctx.on('session/flush', (flushed) => {
+      flushedTypes.push(flushed.events.map(event => event.type))
+    })
+
+    const first = await ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'first' },
+      updateTrigger,
+    )
+    const second = await ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'second' },
+      updateTrigger,
+    )
+
+    expect(first.seq).toBe(1)
+    expect(second.seq).toBe(4)
+    expect(session.events).toMatchObject([
+      { type: 'turn/start', seq: 0, data: { turn: 1, trigger: updateTrigger } },
+      { type: 'test/log-only', seq: 1, data: { value: 'first' } },
+      { type: 'turn/end', seq: 2, data: { turn: 1, reason: { kind: 'completed' } } },
+      { type: 'turn/start', seq: 3, data: { turn: 2, trigger: updateTrigger } },
+      { type: 'test/log-only', seq: 4, data: { value: 'second' } },
+      { type: 'turn/end', seq: 5, data: { turn: 2, reason: { kind: 'completed' } } },
+    ])
+    expect(flushedTypes).toEqual([
+      ['turn/start', 'test/log-only', 'turn/end'],
+      ['turn/start', 'test/log-only', 'turn/end', 'turn/start', 'test/log-only', 'turn/end'],
+    ])
+  })
+
+  it('closes and flushes a zero-step turn when the target event is rejected', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('rejected'))
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+
+    await expect(ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 1n } as never,
+      updateTrigger,
+    )).rejects.toThrow(/non-JSON-serializable/)
+
+    expect(session.events).toMatchObject([
+      { type: 'turn/start', data: { turn: 1 } },
+      { type: 'turn/end', data: { turn: 1, reason: { kind: 'completed' } } },
+    ])
+    expect(flushes).toBe(1)
+  })
+
+  it('does not flush when the synthetic turn cannot open', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('start-failure'))
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+
+    await expect(ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'unreachable' },
+      { ...updateTrigger, invalid: 1n } as never,
+    )).rejects.toThrow(/non-JSON-serializable/)
+
+    expect(session.events).toEqual([])
+    expect(flushes).toBe(0)
+  })
+
+  it('preserves a target rejection when the balancing flush also rejects', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('target-and-flush-failure'))
+    ctx.on('session/flush', () => { throw new Error('disk failed') })
+
+    await expect(ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 1n } as never,
+      updateTrigger,
+    )).rejects.toThrow(/non-JSON-serializable/)
+
+    expect(session.events.map(event => event.type)).toEqual([
+      'turn/start',
+      'turn/end',
+    ])
+  })
+
+  it('keeps the session attached through publication and its flush', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.prepare(SessionId('dispose'))
+    const detach = ctx.sessions.enter(session)
+    ctx.sessions.announce(session)
+    let liveDuringFlush = false
+    ctx.on('session/event', (_observed, event) => {
+      if (event.type === 'turn/start') detach()
+    })
+    ctx.on('session/flush', () => {
+      liveDuringFlush = ctx.sessions.get(session.id) === session
+    })
+
+    await ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'last' },
+      updateTrigger,
+    )
+
+    expect(session.events.map(event => event.type)).toEqual([
+      'turn/start',
+      'test/log-only',
+      'turn/end',
+    ])
+    expect(liveDuringFlush).toBe(true)
+    expect(ctx.sessions.get(session.id)).toBeUndefined()
+  })
+
+  it('rejects detached sessions before opening a turn', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.prepare(SessionId('detached'))
+
+    await expect(ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'nope' },
+      updateTrigger,
+    )).rejects.toThrow('session "detached" is not live in this store')
+    expect(session.events).toEqual([])
+  })
+
+  it('leaves a balanced log when the durability checkpoint rejects', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('flush-failure'))
+    ctx.on('session/flush', () => { throw new Error('disk failed') })
+
+    await expect(ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'accepted' },
+      updateTrigger,
+    )).rejects.toThrow('disk failed')
+    expect(session.events.map(event => event.type)).toEqual([
+      'turn/start',
+      'test/log-only',
+      'turn/end',
+    ])
+  })
+
+  it('rejects overlapping updates while the first append is still settling', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('overlap'))
+    let release!: () => void
+    const checkpoint = new Promise<void>((resolve) => {
+      release = resolve
+    })
+    ctx.on('session/flush', () => checkpoint)
+
+    const first = ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'first' },
+      updateTrigger,
+    )
+    await expect(ctx.sessions.appendOutOfBand(
+      session,
+      'test/log-only',
+      { value: 'overlap' },
+      updateTrigger,
+    )).rejects.toThrow(/out-of-band append in progress/)
+    release()
+    await expect(first).resolves.toMatchObject({ data: { value: 'first' } })
+  })
+})

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

@@ -1,7 +1,13 @@
 import { describe, expect, expectTypeOf, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import { CallId } from '@deepseek-ai/dsh-llm'
-import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
+import SessionStore, {
+  findLastMessageTurnEnd,
+  SESSION_FORMAT_VERSION,
+  Session,
+  SessionEvent,
+  SessionId,
+} from '@deepseek-ai/dsh-session'
 import type { CreateSessionOptions, SessionEventType, SessionHeader, SessionSurface, TodoItem } from '@deepseek-ai/dsh-session'
 
 describe('Session', () => {
@@ -48,6 +54,42 @@ describe('Session', () => {
     expect(structuredClone(turnEnd.data.reason)).toEqual({ kind: 'max-tokens' })
   })
 
+  it('finds the latest message-turn outcome past later non-message turns', () => {
+    const session = new Session(SessionId('message-turn-outcome'))
+    expect(findLastMessageTurnEnd(session.events)).toBeUndefined()
+    session.append('turn/start', {
+      turn: 1,
+      trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'before' } },
+    })
+    session.append('context/message', {
+      content: [{ type: 'text', text: 'before' }],
+      source: { kind: 'plugin', plugin: 'before' },
+    }, { surfaceOp: 'append' })
+    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    expect(findLastMessageTurnEnd(session.events)).toBeUndefined()
+
+    session.append('turn/start', {
+      turn: 2,
+      trigger: { kind: 'message', source: { kind: 'user' } },
+    })
+    session.append('user/message', {
+      content: [{ type: 'text', text: 'bounded prompt' }],
+      source: { kind: 'user' },
+    }, { surfaceOp: 'append' })
+    const messageEnd = session.append('turn/end', { turn: 2, reason: { kind: 'max-tokens' } })
+    session.append('turn/start', {
+      turn: 3,
+      trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'after' } },
+    })
+    session.append('context/message', {
+      content: [{ type: 'text', text: 'after' }],
+      source: { kind: 'plugin', plugin: 'after' },
+    }, { surfaceOp: 'append' })
+    session.append('turn/end', { turn: 3, reason: { kind: 'completed' } })
+
+    expect(findLastMessageTurnEnd(session.events)).toBe(messageEnd)
+  })
+
   it('round-trips the coarse aborted turn outcome', () => {
     const session = new Session(SessionId('aborted'))
     session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })