fix(conversation): resume participant sessions

docs/canonical-api.md

@@ -62,7 +62,7 @@ A general "loop" primitive is the single most common modelling error in this rep
 | Run a genome through a topology shape over the keystone Supervisor, end-to-end | `runPersonified({ persona, shape, task, budget })` — `/loops` | a hand-rolled `createSupervisor().run` + seam-wiring helper |
 | Loop a worker over one evolving artifact, K rounds, stop-when-good | `loopUntil(seed, spec)` as the `shape` — `/loops` | a `while(!done){runWorker();decide()}` hand-loop or "multi-attempt refine driver" |
 | Run a worker agent under test conversing with a **simulated-user persona**, K rounds, worker-only metered | `runPersonaConversation({ worker, persona, backendFor, systemPromptOf })` — root `.` (also `/loops`) | a hand-rolled per-agent `dispatchWithSurface` bridge / eval-dispatch loop |
-| Run **two `AgentProfile`s head-to-head** over a persistent transcript | `runConversation(...)` — root `.` | a hand-rolled two-agent turn loop |
+| Run **two `AgentProfile`s head-to-head** with a separate resumable session for each actor | `runConversation(...)` from root `.` | a hand-rolled two-agent turn loop |
 | Drop a persona⟷agent conversation into an eval matrix as its dispatch | `runPersonaDispatch` → `runProfileMatrix({ dispatch })` — root `.` / `agent-eval/campaign` | a per-agent custom dispatch bridge |
 | Best-of-N / parallel-research / map-reduce at equal compute | `fanout(items, opts)` — `/loops` | `Promise.all` over N calls + manual argmax/merge (bypasses the budget pool → breaks equal-k) |
 | Produce-then-gate with a real checker | `verify(spec)` — `/loops` | "generate, then self-check with the same model, ship if ok" (collapses selector+judge) |

docs/durability-adapters.md

@@ -182,22 +182,35 @@ await journal.migrate()
 
 ## Resuming a run
 
-The runner replays the journal automatically. Two requirements:
+The runner replays the journal automatically.
+Three values control what resumes:
 
 1. Pass the same `runId` to `runConversation` (or `runConversationStream`).
-2. Pass the same `journal` instance (or a fresh one pointed at the same backing store).
+2. Pass the same `journal` instance, or a fresh instance pointed at the same backing store.
+3. Pass the same durable `RuntimeSessionStore` to continue each participant's provider session.
 
-```ts
-// Process A — runs for a while, then crashes mid-turn
-await runConversation(conv, { runId: 'conv_abc', seed: 'hello', journal })
-// ^ assume the process dies after journal recorded turns 0–2 but before turn 3
+The package includes `InMemoryRuntimeSessionStore` for one process.
+For process restarts, implement the two required `RuntimeSessionStore` methods, `get` and `put`, against the same database as your application.
+The example below assumes that implementation is named `sessionStore`.
 
-// Process B — restarts, same runId, same journal
-const resumed = await runConversation(conv, { runId: 'conv_abc', seed: 'hello', journal })
-// ^ runner replays turns 0–2 from the journal, picks up at turn 3, finishes the run
+```ts
+// Process A records turns 0 through 2, then exits before turn 3 starts.
+await runConversation(conv, { runId: 'conv_abc', seed: 'hello', journal, sessionStore })
+
+// Process B opens the same transcript and participant sessions.
+const resumed = await runConversation(conv, {
+  runId: 'conv_abc',
+  seed: 'hello',
+  journal,
+  sessionStore,
+})
 ```
 
-`onEvent` receives a `conversation_resumed` event when the runner finds a non-empty journal entry, so any UI / SSE subscriber can re-hydrate the existing transcript before live deltas resume.
+Each committed turn records the provider session used by its speaker.
+When the session exists in `sessionStore` and its backend implements `resume`, the next turn continues that provider session with only the unseen conversation turns.
+Without a shared session store, the transcript still resumes, but the actor starts a new provider session with the full transcript as context.
+
+`onEvent` receives a `conversation_resumed` event when the runner finds a non-empty journal entry, so any UI or SSE subscriber can rehydrate the existing transcript before live deltas resume.
 
 ## Operational notes
 

src/conversation/run-conversation.test.ts

@@ -2,9 +2,16 @@ import { describe, expect, it } from 'vitest'
 
 import { createIterableBackend } from '../backends'
 import { ValidationError } from '../errors'
-import type { AgentExecutionBackend, RuntimeStreamEvent } from '../types'
+import { InMemoryRuntimeSessionStore } from '../sessions'
+import type {
+  AgentBackendInput,
+  AgentExecutionBackend,
+  RuntimeSession,
+  RuntimeStreamEvent,
+} from '../types'
 import { createConversationBackend } from './conversation-backend'
 import { defineConversation } from './define-conversation'
+import { InMemoryConversationJournal } from './journal'
 import { runConversation, runConversationStream } from './run-conversation'
 import type { ConversationStreamEvent } from './types'
 
@@ -104,6 +111,57 @@ function backendErrorEventBackend(name: string): AgentExecutionBackend {
   })
 }
 
+interface SessionCall {
+  phase: 'start' | 'resume' | 'stream'
+  input: AgentBackendInput
+  sessionId: string
+}
+
+function sessionBackend(
+  name: string,
+  replies: readonly string[],
+  calls: SessionCall[],
+): AgentExecutionBackend {
+  let replyIndex = 0
+  return createIterableBackend({
+    kind: `session-${name}`,
+    start(input, context) {
+      const session = testSession(`provider-${name}`)
+      calls.push({ phase: 'start', input, sessionId: session.id })
+      expect(context.requestedSessionId).toBeDefined()
+      return session
+    },
+    resume(session, input) {
+      calls.push({ phase: 'resume', input, sessionId: session.id })
+      return { ...session, status: 'active', updatedAt: new Date().toISOString() }
+    },
+    async *stream(input, context) {
+      calls.push({ phase: 'stream', input, sessionId: context.session.id })
+      const text = replies[replyIndex]
+      replyIndex += 1
+      if (text === undefined) throw new Error(`session backend '${name}' has no reply`)
+      yield {
+        type: 'text_delta',
+        task: context.task,
+        session: context.session,
+        text,
+        timestamp: new Date().toISOString(),
+      } satisfies RuntimeStreamEvent
+    },
+  })
+}
+
+function testSession(id: string): RuntimeSession {
+  const now = new Date().toISOString()
+  return {
+    id,
+    backend: id.replace('provider-', 'session-'),
+    status: 'active',
+    createdAt: now,
+    updatedAt: now,
+  }
+}
+
 describe('defineConversation', () => {
   const okBackend = fakeBackend('x', ['hi'])
 
@@ -215,6 +273,207 @@ describe('runConversation — happy path', () => {
     const result = await runConversation(conv, { seed: 'go' })
     expect(result.transcript.map((t) => t.speaker)).toEqual(['a', 'b', 'c', 'a'])
   })
+
+  it('resumes one provider session per participant with only unseen turns', async () => {
+    const calls: SessionCall[] = []
+    const conv = defineConversation({
+      participants: [
+        { name: 'author', backend: sessionBackend('author', ['a-1', 'a-2'], calls) },
+        { name: 'critic', backend: sessionBackend('critic', ['c-1', 'c-2'], calls) },
+      ],
+      policy: { maxTurns: 4 },
+    })
+
+    const result = await runConversation(conv, {
+      seed: 'solve the task',
+      runId: 'persistent-actors',
+    })
+
+    expect(calls.filter((call) => call.phase === 'start')).toHaveLength(2)
+    expect(calls.filter((call) => call.phase === 'resume')).toHaveLength(2)
+    expect(result.transcript.map((turn) => turn.sessionId)).toEqual([
+      'provider-author',
+      'provider-critic',
+      'provider-author',
+      'provider-critic',
+    ])
+
+    const streams = calls.filter((call) => call.phase === 'stream')
+    expect(streams[0]?.input.message).toBe('solve the task')
+    expect(streams[1]?.input.message).toContain('[author] a-1')
+    expect(streams[1]?.input.message).toContain('solve the task')
+    expect(streams[2]?.input.message).toBe('[critic] c-1')
+    expect(streams[3]?.input.message).toBe('[author] a-2')
+  })
+
+  it('reconstructs stateless messages once without repeating the previous response', async () => {
+    const inputs: AgentBackendInput[] = []
+    const backend = (name: string, reply: string): AgentExecutionBackend =>
+      createIterableBackend({
+        kind: `stateless-${name}`,
+        async *stream(input, context) {
+          inputs.push(input)
+          yield {
+            type: 'text_delta',
+            task: context.task,
+            session: context.session,
+            text: reply,
+            timestamp: new Date().toISOString(),
+          } satisfies RuntimeStreamEvent
+        },
+      })
+    const conv = defineConversation({
+      participants: [
+        { name: 'author', backend: backend('author', 'draft') },
+        { name: 'critic', backend: backend('critic', 'review') },
+      ],
+      policy: { maxTurns: 2 },
+    })
+
+    await runConversation(conv, { seed: 'solve the task', runId: 'stateless-history' })
+
+    expect(inputs[0]?.messages).toEqual([{ role: 'user', content: 'solve the task' }])
+    expect(inputs[1]?.messages).toEqual([
+      { role: 'user', content: 'solve the task' },
+      { role: 'user', content: '[author] draft' },
+    ])
+  })
+
+  it('continues actor sessions after the conversation driver restarts', async () => {
+    const calls: SessionCall[] = []
+    const journal = new InMemoryConversationJournal()
+    const sessionStore = new InMemoryRuntimeSessionStore()
+    const conv = defineConversation({
+      participants: [
+        { name: 'author', backend: sessionBackend('author', ['a-1', 'a-2'], calls) },
+        { name: 'critic', backend: sessionBackend('critic', ['c-1', 'c-2'], calls) },
+      ],
+      policy: { maxTurns: 4 },
+    })
+    const options = {
+      seed: 'solve the task',
+      runId: 'driver-restart',
+      journal,
+      sessionStore,
+    }
+
+    let committedTurns = 0
+    for await (const event of runConversationStream(conv, options)) {
+      if (event.type === 'turn_end') committedTurns += 1
+      if (committedTurns === 2) break
+    }
+    const result = await runConversation(conv, options)
+
+    expect(result.transcript.map((turn) => turn.text)).toEqual(['a-1', 'c-1', 'a-2', 'c-2'])
+    expect(calls.filter((call) => call.phase === 'start')).toHaveLength(2)
+    expect(calls.filter((call) => call.phase === 'resume')).toHaveLength(2)
+  })
+
+  it('does not bind a participant to a session from a failed attempt', async () => {
+    let starts = 0
+    let resumes = 0
+    let streams = 0
+    const retrying = createIterableBackend({
+      kind: 'retry-session',
+      start() {
+        starts += 1
+        const now = new Date().toISOString()
+        return {
+          id: `attempt-${starts}`,
+          backend: 'retry-session',
+          status: 'active',
+          createdAt: now,
+          updatedAt: now,
+        } satisfies RuntimeSession
+      },
+      resume(session) {
+        resumes += 1
+        return session
+      },
+      async *stream(_input, context) {
+        streams += 1
+        if (streams === 1) throw new Error('ECONNRESET')
+        yield {
+          type: 'text_delta',
+          task: context.task,
+          session: context.session,
+          text: 'recovered',
+          timestamp: new Date().toISOString(),
+        } satisfies RuntimeStreamEvent
+      },
+    })
+    const conv = defineConversation({
+      participants: [
+        { name: 'author', backend: retrying },
+        { name: 'critic', backend: fakeBackend('critic', ['unused']) },
+      ],
+      policy: {
+        maxTurns: 1,
+        defaultCallPolicy: { maxRetries: 1, retryBackoffMs: 0 },
+      },
+    })
+
+    const result = await runConversation(conv, { seed: 'solve the task' })
+
+    expect(result.transcript[0]).toMatchObject({
+      text: 'recovered',
+      sessionId: 'attempt-2',
+      attempts: 2,
+    })
+    expect({ starts, resumes, streams }).toEqual({ starts: 2, resumes: 0, streams: 2 })
+  })
+
+  it('rejects a provider session shared by two participants', async () => {
+    let sessionWrites = 0
+    const sessionStore = {
+      get: () => undefined,
+      put: () => {
+        sessionWrites += 1
+      },
+    }
+    const backend = createIterableBackend({
+      kind: 'shared-session',
+      start() {
+        const now = new Date().toISOString()
+        return {
+          id: 'provider-shared',
+          backend: 'shared-session',
+          status: 'active',
+          createdAt: now,
+          updatedAt: now,
+        } satisfies RuntimeSession
+      },
+      resume(session) {
+        return session
+      },
+      async *stream(_input, context) {
+        yield {
+          type: 'text_delta',
+          task: context.task,
+          session: context.session,
+          text: 'response',
+          timestamp: new Date().toISOString(),
+        } satisfies RuntimeStreamEvent
+      },
+    })
+    const conv = defineConversation({
+      participants: [
+        { name: 'author', backend },
+        { name: 'critic', backend },
+      ],
+      policy: { maxTurns: 2 },
+    })
+
+    const result = await runConversation(conv, { seed: 'solve the task', sessionStore })
+
+    expect(result.transcript).toHaveLength(1)
+    expect(result.halted).toMatchObject({
+      kind: 'participant_error',
+      participant: 'critic',
+      message: expect.stringContaining("already bound to participant 'author'"),
+    })
+    expect(sessionWrites).toBe(1)
+  })
 })
 
 describe('runConversation — halting', () => {