From 4aa3494d7c7a0ea50926b56f21c34bd33d00b556 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Thu, 16 Jul 2026 16:25:48 -0600 Subject: [PATCH] fix(conversation): resume participant sessions --- docs/api/index.md | 91 +++++--- docs/canonical-api.md | 2 +- docs/durability-adapters.md | 33 ++- src/conversation/run-conversation.test.ts | 261 +++++++++++++++++++++- src/conversation/run-conversation.ts | 143 ++++++++++-- src/conversation/types.ts | 11 +- src/run.ts | 57 ++--- src/sessions.ts | 72 +++++- 8 files changed, 557 insertions(+), 113 deletions(-) diff --git a/docs/api/index.md b/docs/api/index.md index dd62d123..e30dfab8 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -1135,7 +1135,7 @@ Defined in: [errors.ts:117](https://github.com/tangle-network/agent-runtime/blob ### InMemoryRuntimeSessionStore -Defined in: [sessions.ts:41](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L41) +Defined in: [sessions.ts:111](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L111) In-memory `RuntimeSessionStore` for single-process use and tests. @@ -1161,7 +1161,7 @@ In-memory `RuntimeSessionStore` for single-process use and tests. > **get**(`sessionId`): `RuntimeSession` \| `undefined` -Defined in: [sessions.ts:45](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L45) +Defined in: [sessions.ts:115](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L115) ###### Parameters @@ -1181,7 +1181,7 @@ Defined in: [sessions.ts:45](https://github.com/tangle-network/agent-runtime/blo > **put**(`session`): `void` -Defined in: [sessions.ts:49](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L49) +Defined in: [sessions.ts:119](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L119) ###### Parameters @@ -1201,7 +1201,7 @@ Defined in: [sessions.ts:49](https://github.com/tangle-network/agent-runtime/blo > **appendEvent**(`sessionId`, `event`): `void` -Defined in: [sessions.ts:53](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L53) +Defined in: [sessions.ts:123](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L123) ###### Parameters @@ -1225,7 +1225,7 @@ Defined in: [sessions.ts:53](https://github.com/tangle-network/agent-runtime/blo > **listEvents**(`sessionId`): [`RuntimeStreamEvent`](#runtimestreamevent)[] -Defined in: [sessions.ts:59](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L59) +Defined in: [sessions.ts:129](https://github.com/tangle-network/agent-runtime/blob/main/src/sessions.ts#L129) ###### Parameters @@ -5098,17 +5098,25 @@ Deterministic turn identifier — stable across retries of the same logical turn so caching gateways and trace backends can dedupe. Shape: `${runId}.t${index}.${speakerSlug}`. +##### sessionId? + +> `optional` **sessionId?**: `string` + +Defined in: [conversation/types.ts:147](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L147) + +Backend session used for this turn. Present on turns recorded by session-aware runners. + ##### text > **text**: `string` -Defined in: [conversation/types.ts:146](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L146) +Defined in: [conversation/types.ts:148](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L148) ##### usage? > `optional` **usage?**: `object` -Defined in: [conversation/types.ts:152](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L152) +Defined in: [conversation/types.ts:154](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L154) Aggregated backend usage for this turn alone. Populated from any `llm_call` stream events the backend emitted; `undefined` when the @@ -5138,7 +5146,7 @@ backend reports no usage. > **attempts**: `number` -Defined in: [conversation/types.ts:164](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L164) +Defined in: [conversation/types.ts:166](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L166) Number of attempts that ran before this turn committed. `1` is the common case; higher means the call policy retried after transient @@ -5148,19 +5156,19 @@ failures. > **startedAt**: `string` -Defined in: [conversation/types.ts:165](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L165) +Defined in: [conversation/types.ts:167](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L167) ##### endedAt > **endedAt**: `string` -Defined in: [conversation/types.ts:166](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L166) +Defined in: [conversation/types.ts:168](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L168) *** ### Conversation -Defined in: [conversation/types.ts:170](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L170) +Defined in: [conversation/types.ts:172](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L172) #### Stable @@ -5170,19 +5178,19 @@ Defined in: [conversation/types.ts:170](https://github.com/tangle-network/agent- > **participants**: readonly [`ConversationParticipant`](#conversationparticipant)[] -Defined in: [conversation/types.ts:171](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L171) +Defined in: [conversation/types.ts:173](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L173) ##### policy > **policy**: [`ConversationPolicy`](#conversationpolicy) -Defined in: [conversation/types.ts:172](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L172) +Defined in: [conversation/types.ts:174](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L174) *** ### RunConversationOptions -Defined in: [conversation/types.ts:176](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L176) +Defined in: [conversation/types.ts:178](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L178) #### Stable @@ -5192,7 +5200,7 @@ Defined in: [conversation/types.ts:176](https://github.com/tangle-network/agent- > **seed**: `string` -Defined in: [conversation/types.ts:178](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L178) +Defined in: [conversation/types.ts:180](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L180) First message kicking off the conversation. Routes to the first speaker. @@ -5200,7 +5208,7 @@ First message kicking off the conversation. Routes to the first speaker. > `optional` **runId?**: `string` -Defined in: [conversation/types.ts:185](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L185) +Defined in: [conversation/types.ts:187](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L187) Optional run identifier for cross-participant trace correlation. Auto- generated when omitted. Reusing a runId against the same `journal` @@ -5211,7 +5219,7 @@ continues from the first un-recorded turn. > `optional` **signal?**: `AbortSignal` -Defined in: [conversation/types.ts:187](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L187) +Defined in: [conversation/types.ts:189](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L189) Cancellation signal — aborts mid-stream and halts with `{ kind: 'abort' }`. @@ -5219,7 +5227,7 @@ Cancellation signal — aborts mid-stream and halts with `{ kind: 'abort' }`. > `optional` **onEvent?**: (`event`) => `void` \| `Promise`\<`void`\> -Defined in: [conversation/types.ts:194](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L194) +Defined in: [conversation/types.ts:196](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L196) Event sink for per-turn micro-events. Distinct from the result transcript: the sink fires for every text-delta, every turn-start/end, and the @@ -5240,18 +5248,29 @@ without waiting for the conversation to finish. > `optional` **journal?**: [`ConversationJournal`](#conversationjournal) -Defined in: [conversation/types.ts:201](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L201) +Defined in: [conversation/types.ts:203](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L203) Optional durable transcript. When set, the runner persists every committed turn before yielding `turn_end`. Reusing the same `runId` against the same journal resumes from the last committed turn — so a driver process crash mid-run loses zero acknowledged turns. +##### sessionStore? + +> `optional` **sessionStore?**: [`RuntimeSessionStore`](#runtimesessionstore) + +Defined in: [conversation/types.ts:210](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L210) + +Stores each participant's backend session. The runner keeps an in-memory +store for one invocation when omitted. Reuse a durable store with the same +`runId` and journal after a process restart. Backends implementing `resume` +continue their provider session; other backends receive the full transcript. + ##### propagatedHeaders? > `optional` **propagatedHeaders?**: `Readonly`\<`Record`\<`string`, `string`\>\> -Defined in: [conversation/types.ts:208](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L208) +Defined in: [conversation/types.ts:217](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L217) Headers to forward verbatim to every participant backend call (gateway propagation: `X-Tangle-Forwarded-Authorization`, run/turn correlation, @@ -5262,7 +5281,7 @@ their `AgentBackendContext`; backends that ignore the field still work. > `optional` **inboundDepth?**: `number` -Defined in: [conversation/types.ts:214](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L214) +Defined in: [conversation/types.ts:223](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L223) Inbound depth at the point this driver was invoked. The runner increments it on every outbound participant call; gateways refuse at @@ -5272,7 +5291,7 @@ increments it on every outbound participant call; gateways refuse at > `optional` **parentTurnId?**: `string` -Defined in: [conversation/types.ts:221](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L221) +Defined in: [conversation/types.ts:230](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L230) Parent turn id when this conversation is *inside* another turn (i.e. the driver is itself a participant via `createConversationBackend`). The @@ -5283,7 +5302,7 @@ so trace stitching survives nested orchestration. ### ConversationResult -Defined in: [conversation/types.ts:225](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L225) +Defined in: [conversation/types.ts:234](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L234) #### Stable @@ -5293,49 +5312,49 @@ Defined in: [conversation/types.ts:225](https://github.com/tangle-network/agent- > **runId**: `string` -Defined in: [conversation/types.ts:226](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L226) +Defined in: [conversation/types.ts:235](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L235) ##### transcript > **transcript**: [`ConversationTurn`](#conversationturn)[] -Defined in: [conversation/types.ts:227](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L227) +Defined in: [conversation/types.ts:236](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L236) ##### turns > **turns**: `number` -Defined in: [conversation/types.ts:228](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L228) +Defined in: [conversation/types.ts:237](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L237) ##### spentCreditsCents > **spentCreditsCents**: `number` -Defined in: [conversation/types.ts:229](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L229) +Defined in: [conversation/types.ts:238](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L238) ##### halted > **halted**: [`HaltReason`](#haltreason) -Defined in: [conversation/types.ts:230](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L230) +Defined in: [conversation/types.ts:239](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L239) ##### durationMs > **durationMs**: `number` -Defined in: [conversation/types.ts:231](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L231) +Defined in: [conversation/types.ts:240](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L240) ##### startedAt > **startedAt**: `string` -Defined in: [conversation/types.ts:232](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L232) +Defined in: [conversation/types.ts:241](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L241) ##### endedAt > **endedAt**: `string` -Defined in: [conversation/types.ts:233](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L233) +Defined in: [conversation/types.ts:242](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L242) *** @@ -11170,7 +11189,7 @@ Defined in: [conversation/types.ts:100](https://github.com/tangle-network/agent- > **ConversationStreamEvent** = \{ `type`: `"conversation_start"`; `runId`: `string`; `participants`: readonly `string`[]; `seed`: `string`; `timestamp`: `string`; \} \| \{ `type`: `"conversation_resumed"`; `runId`: `string`; `participants`: readonly `string`[]; `transcript`: readonly [`ConversationTurn`](#conversationturn)[]; `timestamp`: `string`; \} \| \{ `type`: `"turn_start"`; `runId`: `string`; `index`: `number`; `speaker`: `string`; `turnId`: `string`; `attempt`: `number`; `timestamp`: `string`; \} \| \{ `type`: `"turn_text_delta"`; `runId`: `string`; `index`: `number`; `speaker`: `string`; `turnId`: `string`; `text`: `string`; `timestamp?`: `string`; \} \| \{ `type`: `"turn_retry"`; `runId`: `string`; `index`: `number`; `speaker`: `string`; `turnId`: `string`; `attempt`: `number`; `reason`: `string`; `timestamp`: `string`; \} \| \{ `type`: `"turn_end"`; `runId`: `string`; `turn`: [`ConversationTurn`](#conversationturn); `timestamp`: `string`; \} \| \{ `type`: `"conversation_end"`; `runId`: `string`; `result`: [`ConversationResult`](#conversationresult); `timestamp`: `string`; \} -Defined in: [conversation/types.ts:237](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L237) +Defined in: [conversation/types.ts:246](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/types.ts#L246) #### Stable @@ -12976,7 +12995,7 @@ imports `@cloudflare/workers-types` directly (peer-style typing). > **runConversation**(`conversation`, `options`): `Promise`\<[`ConversationResult`](#conversationresult)\> -Defined in: [conversation/run-conversation.ts:65](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/run-conversation.ts#L65) +Defined in: [conversation/run-conversation.ts:67](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/run-conversation.ts#L67) #### Parameters @@ -12998,7 +13017,7 @@ Defined in: [conversation/run-conversation.ts:65](https://github.com/tangle-netw > **runConversationStream**(`conversation`, `options`): `AsyncIterable`\<[`ConversationStreamEvent`](#conversationstreamevent)\> -Defined in: [conversation/run-conversation.ts:84](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/run-conversation.ts#L84) +Defined in: [conversation/run-conversation.ts:86](https://github.com/tangle-network/agent-runtime/blob/main/src/conversation/run-conversation.ts#L86) Streaming conversation orchestrator: drives N participants in turn through their own backends, enforcing `maxTurns` / `maxCreditsCents` / `haltOn`, yielding per-event stream markers. @@ -14257,7 +14276,7 @@ Stamp cross-cutting defaults onto adapter-projected RunRecords without > **runAgentTask**\<`TState`, `TAction`, `TActionResult`, `TEval`\>(`options`): `Promise`\<[`AgentTaskRunResult`](#agenttaskrunresult)\<`TState`, `TAction`, `TActionResult`, `TEval`\>\> -Defined in: [run.ts:88](https://github.com/tangle-network/agent-runtime/blob/main/src/run.ts#L88) +Defined in: [run.ts:86](https://github.com/tangle-network/agent-runtime/blob/main/src/run.ts#L86) Single-shot task lifecycle for adapter-driven tasks: readiness-gated, emits the runtime lifecycle event vocabulary, session-store pluggable. @@ -14297,7 +14316,7 @@ Single-shot task lifecycle for adapter-driven tasks: readiness-gated, emits the > **runAgentTaskStream**\<`TInput`\>(`options`): `AsyncIterable`\<[`RuntimeStreamEvent`](#runtimestreamevent)\> -Defined in: [run.ts:206](https://github.com/tangle-network/agent-runtime/blob/main/src/run.ts#L206) +Defined in: [run.ts:204](https://github.com/tangle-network/agent-runtime/blob/main/src/run.ts#L204) Streaming task lifecycle: delegates execution to an `AgentExecutionBackend` (model API, sandbox, or custom iterable) and yields lifecycle events as they happen. diff --git a/docs/canonical-api.md b/docs/canonical-api.md index 00489ddf..3fb6acc5 100644 --- a/docs/canonical-api.md +++ b/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) | diff --git a/docs/durability-adapters.md b/docs/durability-adapters.md index 2a9c9b21..b3036cca 100644 --- a/docs/durability-adapters.md +++ b/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 diff --git a/src/conversation/run-conversation.test.ts b/src/conversation/run-conversation.test.ts index e0d88e68..77fd9b60 100644 --- a/src/conversation/run-conversation.test.ts +++ b/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', () => { diff --git a/src/conversation/run-conversation.ts b/src/conversation/run-conversation.ts index 54b1e001..a0976095 100644 --- a/src/conversation/run-conversation.ts +++ b/src/conversation/run-conversation.ts @@ -10,6 +10,8 @@ * forward events as they arrive. Both share one driving loop. * * Distributed-systems primitives layered on top of the loop: + * - **Participant sessions**: one backend session per participant is resumed + * between turns; stateless backends receive one reconstructed transcript. * - **Idempotent turn ids** — `turnId(runId, index, speaker)` stays stable * across retries so caching gateways can dedupe. * - **Durable journal** — optional `ConversationJournal` persists every @@ -33,12 +35,12 @@ import type { KnowledgeReadinessReport } from '@tangle-network/agent-eval' import { BackendTransportError } from '../errors' -import { newRuntimeSession, nowIso, touchSession } from '../sessions' +import { InMemoryRuntimeSessionStore, nowIso, startOrResumeRuntimeSession } from '../sessions' import type { AgentBackendContext, AgentBackendInput, AgentTaskSpec, - RuntimeSession, + RuntimeSessionStore, } from '../types' import { type BackendCallPolicy, @@ -49,7 +51,7 @@ import { sleep, } from './call-policy' import { buildForwardHeaders, FORWARD_HEADERS } from './headers' -import { turnId as deriveTurnId } from './turn-id' +import { turnId as deriveTurnId, slugifySpeaker } from './turn-id' import type { Conversation, ConversationParticipant, @@ -102,6 +104,7 @@ export async function* runConversationStream( let spentCreditsCents = 0 let startedAt = nowIso() let resumed = false + const sessionStore = options.sessionStore ?? new InMemoryRuntimeSessionStore() if (options.journal) { const prior = await options.journal.loadRun(runId) @@ -143,6 +146,7 @@ export async function* runConversationStream( } } const startedAtMs = Date.now() + const participantSessions = sessionIdsFrom(transcript) if (resumed) { yield { @@ -267,11 +271,14 @@ export async function* runConversationStream( for await (const delta of driveSingleAttempt({ speaker, participants: conversation.participants, + seed: options.seed, input: currentInput, turnIndex, runId, turnId: tid, transcript, + participantSessions, + sessionStore, signal: perAttempt.signal, aggregator: localAgg, propagatedHeaders: buildForwardHeaders({ @@ -331,6 +338,9 @@ export async function* runConversationStream( } const turn = aggregator.toTurn({ turnId: tid, attempts: attemptCount }) + if (turn.sessionId) { + bindParticipantSession(participantSessions, turn.speaker, turn.sessionId) + } transcript.push(turn) spentCreditsCents += centsFromUsd(turn.usage?.costUsd ?? 0) if (options.journal) { @@ -385,11 +395,14 @@ export async function* runConversationStream( interface SingleAttemptArgs { speaker: ConversationParticipant participants: readonly ConversationParticipant[] + seed: string input: string turnIndex: number runId: string turnId: string transcript: readonly ConversationTurn[] + participantSessions: Map + sessionStore: RuntimeSessionStore signal: AbortSignal aggregator: TurnAggregator propagatedHeaders: Record @@ -410,10 +423,8 @@ async function* driveSingleAttempt( }, } const knowledge = passingReadiness(task.id) - const messages = buildMessagesFor(args.speaker.name, args.transcript, args.input) - const backendInput: AgentBackendInput = { task, message: args.input, messages } - const startCtx: Omit & { requestedSessionId?: string } = { + const startCtx: Omit = { task, knowledge, signal: args.signal, @@ -421,14 +432,21 @@ async function* driveSingleAttempt( turnId: args.turnId, propagatedHeaders: args.propagatedHeaders, } - const session: RuntimeSession = args.speaker.backend.start - ? touchSession(await args.speaker.backend.start(backendInput, startCtx)) - : newRuntimeSession(args.speaker.backend.kind, undefined, { - runId: args.runId, - turnIndex: args.turnIndex, - turnId: args.turnId, - speaker: args.speaker.name, - }) + const previousSessionId = args.participantSessions.get(args.speaker.name) + const opened = await startOrResumeRuntimeSession({ + backend: args.speaker.backend, + input: () => fullConversationInput(task, args.speaker.name, args.seed, args.transcript), + continuationInput: () => continuedConversationInput(task, args.speaker.name, args.transcript), + context: startCtx, + store: args.sessionStore, + sessionId: + previousSessionId ?? participantSessionId(args.runId, args.speaker.name, args.participants), + resume: previousSessionId !== undefined && args.speaker.backend.resume !== undefined, + validateSession: (session) => + assertParticipantSessionAvailable(args.participantSessions, args.speaker.name, session.id), + }) + const session = opened.session + args.aggregator.recordSession(session.id) const streamCtx: AgentBackendContext = { task, @@ -440,7 +458,7 @@ async function* driveSingleAttempt( propagatedHeaders: args.propagatedHeaders, } - for await (const event of args.speaker.backend.stream(backendInput, streamCtx)) { + for await (const event of args.speaker.backend.stream(opened.input, streamCtx)) { if (args.signal.aborted) { // Surface the abort so the outer retry/halt logic can react. The signal // either fires because of caller-cancel (propagate as-is) or because of @@ -481,6 +499,7 @@ async function* driveSingleAttempt( class TurnAggregator { private text = '' private adoptedFinal = false + private sessionId: string | undefined private usage: | { tokensIn?: number @@ -514,6 +533,10 @@ class TurnAggregator { return this.text.trim().length > 0 } + recordSession(sessionId: string): void { + this.sessionId = sessionId + } + recordUsage(event: { model?: string tokensIn?: number @@ -535,6 +558,7 @@ class TurnAggregator { index: this.base.index, speaker: this.base.speaker, turnId: meta.turnId, + ...(this.sessionId ? { sessionId: this.sessionId } : {}), text: this.text.trim(), usage: this.usage, attempts: meta.attempts, @@ -548,14 +572,15 @@ class TurnAggregator { * Build the participant's POV of the transcript so an OpenAI-compatible * backend sees its own turns as `assistant` and everyone else's as `user`, * with explicit speaker tags so 3+ party conversations stay disambiguated. - * The seed / current input is appended as the trailing user message. */ function buildMessagesFor( speakerName: string, + seed: string, transcript: readonly ConversationTurn[], - currentInput: string, ): Array<{ role: string; content: string }> { - const messages: Array<{ role: string; content: string }> = [] + const messages: Array<{ role: string; content: string }> = seed + ? [{ role: 'user', content: seed }] + : [] for (const turn of transcript) { if (turn.speaker === speakerName) { messages.push({ role: 'assistant', content: turn.text }) @@ -563,10 +588,90 @@ function buildMessagesFor( messages.push({ role: 'user', content: `[${turn.speaker}] ${turn.text}` }) } } - if (currentInput) messages.push({ role: 'user', content: currentInput }) return messages } +function fullConversationInput( + task: AgentTaskSpec, + speakerName: string, + seed: string, + transcript: readonly ConversationTurn[], +): AgentBackendInput { + const transcriptText = transcript.map((turn) => `[${turn.speaker}] ${turn.text}`) + return { + task, + message: [seed, ...transcriptText].filter((part) => part.length > 0).join('\n\n'), + messages: buildMessagesFor(speakerName, seed, transcript), + } +} + +function continuedConversationInput( + task: AgentTaskSpec, + speakerName: string, + transcript: readonly ConversationTurn[], +): AgentBackendInput { + let lastOwnTurn = -1 + for (let index = transcript.length - 1; index >= 0; index -= 1) { + if (transcript[index]?.speaker === speakerName) { + lastOwnTurn = index + break + } + } + const unseen = transcript.slice(lastOwnTurn + 1) + const message = + unseen.length > 0 + ? unseen.map((turn) => `[${turn.speaker}] ${turn.text}`).join('\n\n') + : 'Continue.' + return { task, message, messages: [{ role: 'user', content: message }] } +} + +function sessionIdsFrom(transcript: readonly ConversationTurn[]): Map { + const sessions = new Map() + for (const turn of transcript) { + if (turn.sessionId) bindParticipantSession(sessions, turn.speaker, turn.sessionId) + } + return sessions +} + +function bindParticipantSession( + sessions: Map, + speakerName: string, + sessionId: string, +): void { + assertParticipantSessionAvailable(sessions, speakerName, sessionId) + sessions.set(speakerName, sessionId) +} + +function assertParticipantSessionAvailable( + sessions: ReadonlyMap, + speakerName: string, + sessionId: string, +): void { + for (const [boundSpeaker, boundSessionId] of sessions) { + if (boundSpeaker !== speakerName && boundSessionId === sessionId) { + throw new BackendTransportError( + 'conversation', + `session '${sessionId}' is already bound to participant '${boundSpeaker}'`, + ) + } + } +} + +function participantSessionId( + runId: string, + speakerName: string, + participants: readonly ConversationParticipant[], +): string { + const participantIndex = participants.findIndex((participant) => participant.name === speakerName) + if (participantIndex < 0) { + throw new BackendTransportError( + 'conversation', + `participant '${speakerName}' is not registered in this conversation`, + ) + } + return `${runId}.participant.${participantIndex}.${slugifySpeaker(speakerName)}` +} + /** * True when this participant should forward the caller's * `X-Tangle-Forwarded-Authorization` on outbound calls (the "pass-through" diff --git a/src/conversation/types.ts b/src/conversation/types.ts index 5cd3d526..222cdf38 100644 --- a/src/conversation/types.ts +++ b/src/conversation/types.ts @@ -12,7 +12,7 @@ * @stable */ -import type { AgentExecutionBackend } from '../types' +import type { AgentExecutionBackend, RuntimeSessionStore } from '../types' import type { BackendCallPolicy } from './call-policy' import type { PropagatedHeaders } from './headers' import type { ConversationJournal } from './journal' @@ -143,6 +143,8 @@ export interface ConversationTurn { * `${runId}.t${index}.${speakerSlug}`. */ turnId: string + /** Backend session used for this turn. Present on turns recorded by session-aware runners. */ + sessionId?: string text: string /** * Aggregated backend usage for this turn alone. Populated from any @@ -199,6 +201,13 @@ export interface RunConversationOptions { * driver process crash mid-run loses zero acknowledged turns. */ journal?: ConversationJournal + /** + * Stores each participant's backend session. The runner keeps an in-memory + * store for one invocation when omitted. Reuse a durable store with the same + * `runId` and journal after a process restart. Backends implementing `resume` + * continue their provider session; other backends receive the full transcript. + */ + sessionStore?: RuntimeSessionStore /** * Headers to forward verbatim to every participant backend call (gateway * propagation: `X-Tangle-Forwarded-Authorization`, run/turn correlation, diff --git a/src/run.ts b/src/run.ts index 6d2413b2..148d4260 100644 --- a/src/run.ts +++ b/src/run.ts @@ -61,12 +61,11 @@ export function applyRunRecordDefaults( } import { normalizeBackendStreamEvent } from './backends' -import { BackendTransportError, SessionMismatchError } from './errors' +import { BackendTransportError } from './errors' import { decideKnowledgeReadiness } from './readiness' -import { newRuntimeSession, nowIso, touchSession } from './sessions' +import { nowIso, startOrResumeRuntimeSession, touchSession } from './sessions' import type { AgentBackendInput, - AgentExecutionBackend, AgentKnowledgeProvider, AgentRuntimeEventSink, AgentTaskContext, @@ -76,7 +75,6 @@ import type { BackendErrorDetail, RunAgentTaskOptions, RunAgentTaskStreamOptions, - RuntimeSession, RuntimeStreamEvent, } from './types' @@ -248,22 +246,16 @@ export async function* runAgentTaskStream>( return { ...event, timestamp: nowIso() } } -async function startBackendSession( - backend: AgentExecutionBackend, - input: TInput, - context: { task: AgentTaskSpec; knowledge: KnowledgeReadinessReport; signal?: AbortSignal }, - requestedSessionId?: string, -): Promise { - if (backend.start) return backend.start(input, { ...context, requestedSessionId }) - return newRuntimeSession(backend.kind, requestedSessionId) -} - -async function resumeBackendSession( - backend: AgentExecutionBackend, - session: RuntimeSession, - input: TInput, - context: { task: AgentTaskSpec; knowledge: KnowledgeReadinessReport; signal?: AbortSignal }, -): Promise { - if (session.backend !== backend.kind) { - throw new SessionMismatchError(session.backend, backend.kind) - } - if (backend.resume) return backend.resume(session, input, context) - return touchSession({ ...session, status: 'active' }) -} - function buildReadiness( task: AgentTaskSpec, provider: AgentKnowledgeProvider | undefined, diff --git a/src/sessions.ts b/src/sessions.ts index fedcdb85..357df2b7 100644 --- a/src/sessions.ts +++ b/src/sessions.ts @@ -8,7 +8,15 @@ * @stable */ -import type { RuntimeSession, RuntimeSessionStore, RuntimeStreamEvent } from './types' +import { SessionMismatchError } from './errors' +import type { + AgentBackendContext, + AgentBackendInput, + AgentExecutionBackend, + RuntimeSession, + RuntimeSessionStore, + RuntimeStreamEvent, +} from './types' /** @internal */ export function newRuntimeSession( @@ -37,6 +45,68 @@ export function nowIso(): string { return new Date().toISOString() } +interface StartOrResumeRuntimeSessionOptions { + backend: AgentExecutionBackend + input: TInput | (() => TInput) + /** Smaller input used only when the backend can continue its native session. */ + continuationInput?: TInput | (() => TInput) + context: Omit + store?: RuntimeSessionStore + sessionId?: string + resume?: boolean + validateSession?: (session: RuntimeSession) => void +} + +interface RuntimeSessionCall { + session: RuntimeSession + input: TInput + resumed: boolean +} + +/** Start or resume one backend session and persist its current identity. @internal */ +export async function startOrResumeRuntimeSession( + options: StartOrResumeRuntimeSessionOptions, +): Promise> { + const existing = + options.resume && options.sessionId ? await options.store?.get(options.sessionId) : undefined + if (!existing) { + const input = resolveSessionInput(options.input) + const session = options.backend.start + ? await options.backend.start(input, { + ...options.context, + requestedSessionId: options.sessionId, + }) + : newRuntimeSession(options.backend.kind, options.sessionId) + assertSessionBackend(session, options.backend.kind) + options.validateSession?.(session) + await options.store?.put(session) + return { session, input, resumed: false } + } + + assertSessionBackend(existing, options.backend.kind) + const continued = options.backend.resume !== undefined + const input = resolveSessionInput( + continued ? (options.continuationInput ?? options.input) : options.input, + ) + const session = options.backend.resume + ? await options.backend.resume(existing, input, options.context) + : touchSession({ ...existing, status: 'active' }) + assertSessionBackend(session, options.backend.kind) + options.validateSession?.(session) + await options.store?.put(session) + return { session, input, resumed: true } +} + +function resolveSessionInput( + input: TInput | (() => TInput), +): TInput { + return typeof input === 'function' ? input() : input +} + +function assertSessionBackend(session: RuntimeSession, backend: string): void { + if (session.backend !== backend) throw new SessionMismatchError(session.backend, backend) +} + /** In-memory `RuntimeSessionStore` for single-process use and tests. @stable */ export class InMemoryRuntimeSessionStore implements RuntimeSessionStore { private readonly sessions = new Map()