Merge remote-tracking branch 'origin/master' into worktree/provider-routed-llm-adapters

# Conflicts: # docs/cordis-catalog/services.md # docs/event-producer-consumer.md # examples/acp-agent/tests/snapshots/advanced-toolchain/session.1.jsonl # examples/acp-agent/tests/snapshots/advanced-toolchain/session.2.jsonl # 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/permission-switching/session.jsonl # examples/acp-agent/tests/snapshots/skill-load/session.jsonl # examples/acp-agent/tests/snapshots/text-turn/session.jsonl
2026-07-15 10:15:06 +08:00
parent 310aa9992f 34bb75b2d2
commit 35cae1d389
112 changed files with 3986 additions and 572 deletions
@@ -8,7 +8,6 @@
 import { Context } from 'cordis'
 import { interruptedTurnClosers, SESSION_FORMAT_VERSION, snapshotJsonValue } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
-import { seedCoversPrefix } from './index.ts'

 /**
 * A stored session's header, valid contiguous event prefix, and optional opaque
@@ -110,6 +109,15 @@ async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unkn
  return errors
 }

+/** Whether a live session seed reproduces a persisted prefix exactly. */
+function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
+  return prefix.length <= seed.length
+    && prefix.every((event, index) => {
+      const seedEvent = seed[index]
+      return seedEvent !== undefined && JSON.stringify(seedEvent) === JSON.stringify(event)
+    })
+}
+
 /**
 * Owns the backend-agnostic session write-path orchestration. A backend
 * constructs one (`new PersistenceCoordinator(ctx, this)`), implements
@@ -134,10 +142,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
  private chains = new Map<SessionId, Promise<unknown>>()
  /**
   * Init promises keyed by live session object, preventing an id-reusing
-   * replacement from inheriting stale initialization. Readonly access supports
-   * backend white-box tests.
+   * replacement from inheriting stale initialization. Flush is the public
+   * observation boundary; callers do not inspect this bookkeeping directly.
   */
-  readonly inits = new Map<Session, Promise<void>>()
+  private inits = new Map<Session, Promise<void>>()

  constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
    this.installWritePath()
@@ -6,7 +6,6 @@
 */

 import { Context, Service } from 'cordis'
-import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'

 // Re-export the metadata vocabulary so consumers import it from the seam.
@@ -22,35 +21,6 @@ declare module 'cordis' {
  }
 }

-/**
- * Check whether a live seed exactly reproduces a durable prefix, including full
- * payloads. This distinguishes resume/HMR rebinding from an id collision.
- * @param seed - the live session's creation-time event snapshot.
- * @param prefix - the persisted prefix the seed must reproduce.
- * @returns `true` when the prefix fits within the seed and every event matches by JSON text.
- */
-export function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
-  return prefix.length <= seed.length
-    && prefix.every((event, index) => {
-      const seedEvent = seed[index]
-      return seedEvent !== undefined && JSON.stringify(seedEvent) === JSON.stringify(event)
-    })
-}
-
-/**
- * Reject a batch that is not wholly losslessly JSON-serializable. Live session
- * appends already enforce this; persistence append paths also accept replay or
- * direct batches that may bypass a live session instance. Validation uses the
- * same one-pass materializer as the coordinator, so getters are read once.
- * @param events - the complete event batch to validate.
- */
-export function assertSerializable(events: readonly SessionEvent[]): void {
-  const snapshot = snapshotJsonValue(events)
-  if (snapshot === undefined) {
-    throw new Error('session event batch is not losslessly JSON-serializable because it contains non-JSON-serializable data')
-  }
-}
-
 /**
 * Durable append-only session storage. Implementations preserve contiguous,
 * losslessly JSON-serializable events; {@link append} resolves only after
@@ -13,7 +13,6 @@ import { describe, expect, it } from 'vitest'
 import { Context, type Fiber } from 'cordis'
 import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
-import type { SessionPersistence } from '../src/index.ts'
 import { meta, oneTurnLog, appendLog } from './contract.ts'

 /**
@@ -39,11 +38,6 @@ export interface CoordinatorFixture {
 const WORK = '/w'
 const OTHER = '/other'

-/** The per-session init map a backend exposes for white-box init awaits. */
-function inits(persistence: SessionPersistence): Map<Session, Promise<void>> {
-  return (persistence as unknown as { inits: Map<Session, Promise<void>> }).inits
-}
-
 /** Append a whole event log to a live session, event by event (drives session/event). */
 function send(session: Session, events: readonly SessionEvent[]): void {
  appendLog(session, events)
@@ -168,7 +162,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        const seed = oneTurnLog()
        // A fork: a brand-new id whose seed came from elsewhere.
        const forked = ctx.sessions.create(SessionId('forked'), { seed, meta: { cwd: WORK } })
-        await inits(ctx.sessionPersistence).get(forked) // onCreated persisted the seed
+        await ctx.sessions.flush(forked) // onCreated persisted the seed
        const loaded = await ctx.sessionPersistence.load(SessionId('forked'))
        expect(loaded.events).toEqual(seed)
        // A flush with no NEW events must not double-write.
@@ -197,7 +191,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
      try {
        const loaded = await second.ctx.sessionPersistence.load(SessionId('resumed'))
        const s2 = second.ctx.sessions.create(SessionId('resumed'), { seed: loaded.events, meta: { cwd: WORK } })
-        await inits(second.ctx.sessionPersistence).get(s2) // let onCreated adopt
+        await second.ctx.sessions.flush(s2) // let onCreated adopt
        s2.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
        s2.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
        await second.ctx.parallel('session/flush', s2)
@@ -370,7 +364,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
      try {
        const s2 = second.ctx.sessions.create(SessionId('collide'), { meta: { cwd: WORK } })
        s2.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
-        await expect(inits(second.ctx.sessionPersistence).get(s2))
+        await expect(second.ctx.sessions.flush(s2))
          .rejects.toThrow(/already has a persisted log|id collision/)
      } finally {
        await second.fiber.dispose()
@@ -388,14 +382,14 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
          firstSession = inner.sessions.create(SessionId('abandoned'), { meta: { cwd: WORK } })
        }, { inject: ['sessions'] }))
-        await inits(ctx.sessionPersistence).get(firstSession) // register the lazy state
+        await ctx.sessions.flush(firstSession) // register the lazy state
        await firstFiber.dispose() // disposed before any append → never materialized

        let reuse!: Session
        await ctx.plugin(Object.assign((inner: Context) => {
          reuse = inner.sessions.create(SessionId('abandoned'), { meta: { cwd: WORK } })
        }, { inject: ['sessions'] }))
-        await expect(inits(ctx.sessionPersistence).get(reuse)).resolves.toBeUndefined()
+        await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
        reuse.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
        reuse.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
        await ctx.parallel('session/flush', reuse)
@@ -415,7 +409,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
          first = inner.sessions.create(SessionId('buffered'), { meta: { cwd: WORK } })
        }, { inject: ['sessions'] }))
-        await inits(ctx.sessionPersistence).get(first)
+        await ctx.sessions.flush(first)
        // Append a turn but do NOT flush — events sit in the write-behind buffer.
        first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
        first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
@@ -425,7 +419,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        await ctx.plugin(Object.assign((inner: Context) => {
          reuse = inner.sessions.create(SessionId('buffered'), { meta: { cwd: WORK } })
        }, { inject: ['sessions'] }))
-        await expect(inits(ctx.sessionPersistence).get(reuse)).rejects.toThrow(/already bound to a different live session/)
+        await expect(ctx.sessions.flush(reuse)).rejects.toThrow(/already bound to a different live session/)
      } finally {
        await fiber.dispose()
        await fix.cleanup()
@@ -462,7 +456,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        // A live session with that id arrives and claims it (cursor 0 matches
        // trivially), persisting its seed.
        const live = ctx.sessions.create(SessionId('lazy-claim'), { seed: oneTurnLog(), meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).resolves.toBeUndefined()
+        await expect(ctx.sessions.flush(live)).resolves.toBeUndefined()
        const loaded = await ctx.sessionPersistence.load(SessionId('lazy-claim'))
        expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
      } finally {
@@ -487,7 +481,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        await ctx.plugin(Object.assign((inner: Context) => {
          fresh = inner.sessions.create(SessionId('preview'), { meta: { cwd: WORK } })
        }, { inject: ['sessions'] }))
-        await expect(inits(ctx.sessionPersistence).get(fresh))
+        await expect(ctx.sessions.flush(fresh))
          .rejects.toThrow(/do not match this live session|already has a persisted log|id collision/)
      } finally {
        await fiber.dispose()
@@ -511,7 +505,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
          { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
          { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
        ], meta: { cwd: WORK } })
-        await inits(ctx.sessionPersistence).get(cont)
+        await ctx.sessions.flush(cont)
        const loaded = await ctx.sessionPersistence.load(SessionId('claim'))
        expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
      } finally {
@@ -531,7 +525,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        // cwd scope is the fence (without it, WORK events would append under the
        // OTHER header). Rejected as a collision.
        const live = ctx.sessions.create(SessionId('wrong-cwd-claim'), { seed: oneTurnLog(), meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+        await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
      } finally {
        await fiber.dispose()
        await fix.cleanup()
@@ -549,7 +543,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        // A live session whose SEED matches the loaded prefix but whose cwd is
        // WORK must still be rejected — the cwd guard runs before the seed check.
        const live = ctx.sessions.create(SessionId('wrong-cwd-load'), { seed: events, meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+        await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
      } finally {
        await fiber.dispose()
        await fix.cleanup()
@@ -565,7 +559,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
        // A live session reusing the id but WITH cwd WORK is a cwd mismatch
        // (undefined vs WORK) and must be rejected.
        const live = ctx.sessions.create(SessionId('no-cwd-state'), { seed: oneTurnLog(), meta: { cwd: WORK } })
-        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+        await expect(ctx.sessions.flush(live)).rejects.toThrow(/different cwd|id collision/)
      } finally {
        await fiber.dispose()
        await fix.cleanup()
@@ -1,9 +1,9 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import SessionStore, { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
-import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
-  SessionPersistence, PersistenceCoordinator, assertSerializable, seedCoversPrefix,
+  SessionPersistence, PersistenceCoordinator,
  type PersistenceBackend, type StoredPrefix,
 } from '../src/index.ts'
 import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
@@ -53,11 +53,6 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
    return this.coordinator.load(id)
  }

-  /** White-box accessor: await a specific session's onCreated init. */
-  get inits(): Map<Session, Promise<void>> {
-    return this.coordinator.inits
-  }
-
  // --- PersistenceBackend hooks (the Map storage primitives) ---

  // A Map-backed store has no torn tails, so `tornMarker` is never set. Ids are
@@ -157,38 +152,3 @@ describe('SessionPersistence service registration', () => {
    await fiber.dispose()
  })
 })
-
-describe('shared persistence helpers', () => {
-  it('accepts a seed that reproduces the persisted prefix exactly', () => {
-    const log = oneTurnLog()
-    expect(seedCoversPrefix(log, log.slice(0, 3))).toBe(true)
-    expect(seedCoversPrefix(log, [])).toBe(true)
-  })
-
-  it('rejects a prefix longer than the seed', () => {
-    const log = oneTurnLog()
-    expect(seedCoversPrefix(log.slice(0, 2), log)).toBe(false)
-  })
-
-  it('rejects a same-envelope event with mutated data', () => {
-    const log = oneTurnLog()
-    const tampered = structuredClone(log)
-    const event = tampered[1]!
-    tampered[1] = {
-      ...event,
-      data: { ...event.data, content: [{ type: 'text', text: 'tampered' }] },
-    } as SessionEvent
-    expect(seedCoversPrefix(tampered, log.slice(0, 2))).toBe(false)
-  })
-
-  it('accepts JSON-serializable event data', () => {
-    expect(() => { assertSerializable(oneTurnLog()) }).not.toThrow()
-  })
-
-  it('rejects a batch containing non-JSON-serializable event data', () => {
-    const bad = [
-      { type: 'user/message', seq: 0, time: 1, data: { content: 1n } },
-    ] as unknown as SessionEvent[]
-    expect(() => { assertSerializable(bad) }).toThrow(/batch is not losslessly JSON-serializable/)
-  })
-})