diff --git a/README.md b/README.md index 687d39b2e..9bb72f36f 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,16 @@ yarn add @openrouter/sdk > npx skills add OpenRouterTeam/skills --skill openrouter-agent-migration > ``` +## Multi-turn conversation state + +`callModelWithState` carries conversation context across turns: each turn +loads prior history keyed by a caller-supplied `conversationId`, appends the +new input and response output, and persists it back — no manual history +management. State lives in a pluggable `ConversationStateStore` (in-memory +and file backends included), with optional TTL expiry, `expire()` sweeps, and +`clear()` for explicit deletion. See [docs/multi-turn-state.md](docs/multi-turn-state.md) +for keying, configuration, cleanup, and recovery semantics. + ## Requirements diff --git a/docs/multi-turn-state.md b/docs/multi-turn-state.md new file mode 100644 index 000000000..86dfb0192 --- /dev/null +++ b/docs/multi-turn-state.md @@ -0,0 +1,168 @@ +# Multi-Turn Conversation State + +The SDK can carry conversation context across turns so the model sees prior +user and assistant messages without the caller re-sending full history on +every request. This document describes how that state is keyed, configured, +expired, and cleared. + +> Implementation: `src/lib/conversation-state-store.ts` (store + backends), +> `src/funcs/call-model-with-state.ts` (turn-pipeline entry point). +> Integration tests: `tests/unit/multi-turn-integration.test.ts`. + +## Quick start + +```typescript +import { OpenRouter } from '@openrouter/sdk'; +import { + ConversationStateStore, + InMemoryConversationStateBackend, +} from '@openrouter/sdk/lib/conversation-state-store.js'; + +const client = new OpenRouter(); + +// One store per application (or per tenant); reuse it for every turn. +const store = new ConversationStateStore(new InMemoryConversationStateBackend(), { + ttlMs: 30 * 60 * 1000, // optional: expire idle conversations after 30 min +}); + +// Turn 1 +const first = client.callModelWithState({ + model: 'openai/gpt-5', + input: 'My name is Ada.', + conversationId: 'user-123', + stateStore: store, +}); +await first.getText(); + +// Turn 2 — the model sees turn 1's context and answers "Ada". +const second = client.callModelWithState({ + model: 'openai/gpt-5', + input: 'What is my name?', + conversationId: 'user-123', + stateStore: store, +}); +console.log(await second.getText()); +``` + +The functional entry point is also available as +`callModelWithState(client, { ... })` from `src/funcs/call-model-with-state.ts`. + +## How state is keyed + +Every conversation is identified by a caller-supplied `conversationId` string: + +- The id is **the only lookup key**. Each turn loads the state document under + that id, appends the new input and the response output to the stored + message history, and persists the result back under the same id before the + response is returned. +- Choose ids that are stable for the lifetime of the conversation — e.g. a + session id, `user-`, or a UUID minted when the chat starts. +- **Parallel conversations are fully isolated**: two ids never share state, + even when their turns interleave or run concurrently on the same store. +- The id guard: if the turn pipeline ever produces a state document whose id + differs from the accessor's bound id (e.g. the first turn creates a + placeholder state internally), the store rebinds it to your + `conversationId` on save. State can never strand under an id you don't own. + +Single-turn callers don't need any of this: `callModel` without a +`conversationId` behaves exactly as before and touches no store. + +## The state document + +```typescript +interface ConversationState { + id: string; // your conversationId + messages: BaseInputsUnion[]; // user inputs + assistant outputs, in turn order + status: 'in_progress' | 'awaiting_approval' | 'interrupted' | 'complete'; + createdAt: number; // ms epoch + updatedAt: number; // ms epoch; drives TTL expiry + previousResponseId?: string; // last Responses API response id + // ... pending tool results and metadata +} +``` + +Documents are JSON-serializable and are validated on every read. A document +that fails validation raises `CorruptedStateError` (see "Missing or corrupted +state" below). + +## Configuration + +```typescript +new ConversationStateStore(backend, { + ttlMs?: number, // idle expiry window; omit or 0 to disable + now?: () => number, // clock override (testing) +}); +``` + +| Option | Default | Effect | +|----------|--------------|--------| +| `ttlMs` | `0` (off) | `get` returns `null` for — and `expire()` deletes — conversations whose `updatedAt` is older than this. | +| `now` | `Date.now` | Clock used for expiry checks. Inject a fake clock in tests. | + +There are **no environment variables or global config flags** for multi-turn +state. Configuration is explicit per store instance — nothing changes unless +the caller constructs a store and passes it to `callModelWithState`. + +## Backends + +Two zero-dependency backends ship with the SDK: + +- **`InMemoryConversationStateBackend`** — process-local `Map`. Right for + tests and ephemeral single-process use. Documents are deep-copied on + read/write so callers can't mutate stored state by aliasing. Implements + `list()`, so store-wide `expire()` works. +- **`FileConversationStateBackend`** — JSON-file-per-conversation under a + directory (Node.js only). Ids are sanitized to prevent path traversal. + Useful for local durable execution without Redis. + +Any other backend (Redis, Postgres, …) can be implemented against the +`ConversationStateBackend` interface: + +```typescript +interface ConversationStateBackend { + load(id: string): Promise; + save(id: string, state: ConversationState): Promise; + delete(id: string): Promise; + list?(): Promise; // enables store-wide expire() +} +``` + +## Expiry and cleanup + +- **TTL expiry (read-through):** with `ttlMs` set, `store.get(id)` returns + `null` for idle-expired documents. When a turn then arrives for that id the + pipeline starts a **fresh conversation under the same id** — the model does + not see the expired history. +- **Store-wide sweep:** `store.expire()` removes every stale conversation + (requires the backend to implement `list()`) and returns the removed ids. + `store.expire(['id1', 'id2'])` sweeps only the given ids. Run this on a + timer if you want storage reclaimed rather than just hidden. +- **Explicit deletion:** `store.clear(id)` deletes one conversation + immediately, regardless of TTL. The next turn for that id starts fresh. + Clearing one conversation never affects siblings. + +## Missing or corrupted state + +| Situation | Behavior | +|-----------|----------| +| No state for `conversationId` (first turn, after `clear`, after TTL expiry) | Fresh state document created under the same id; turn proceeds normally. | +| Stored document is invalid JSON or fails schema validation | `CorruptedStateError` is thrown on the next read — the SDK fails loudly instead of silently continuing with a mangled history. | +| Backend write fails | The error propagates from the store; no partial in-memory fallback. | + +To recover a corrupted conversation, call `store.clear(id)` and let the next +turn recreate it (or restore the document from a backup and re-`put` it). + +## Testing recipes + +The integration tests in `tests/unit/multi-turn-integration.test.ts` show the +patterns, all without a live API key: + +- **3+ turn context:** queue mocked `betaResponsesSend` responses and assert + that turn N's API request input contains every earlier turn's user input and + assistant output, in order. +- **Parallel isolation:** interleave turns across two `conversationId`s on one + store and assert neither request input mentions the other conversation. +- **Expiry:** construct the store with a short `ttlMs` (or a fake `now` + clock), sleep past the TTL, and assert the next turn starts fresh. +- **Corruption:** poison the backend (or the JSON file for + `FileConversationStateBackend`) and assert `CorruptedStateError`. diff --git a/src/funcs/call-model-with-state.ts b/src/funcs/call-model-with-state.ts new file mode 100644 index 000000000..0ae8058dd --- /dev/null +++ b/src/funcs/call-model-with-state.ts @@ -0,0 +1,115 @@ +import type { $ZodObject, $ZodShape, infer as zodInfer } from 'zod/v4/core'; +import type { OpenRouterCore } from '../core.js'; +import type { CallModelInput } from '../lib/async-params.js'; +import type { + ConversationStateStore, +} from '../lib/conversation-state-store.js'; +import type { RequestOptions } from '../lib/sdks.js'; +import type { Tool } from '../lib/tool-types.js'; + +import { callModel } from './call-model.js'; +import { + createStateAccessor, +} from '../lib/conversation-state-store.js'; +import type { ModelResult } from '../lib/model-result.js'; + +/** + * Input for {@link callModelWithState}: everything `callModel` accepts, plus + * the conversation identity and the store that owns its state document. + * + * The `state` field of the base input is intentionally omitted — the accessor + * is derived from `conversationId` + `stateStore`, and supplying both would be + * contradictory. + */ +export type CallModelWithStateInput< + TTools extends readonly Tool[] = readonly Tool[], + TShared extends Record = Record, +> = Omit, 'state'> & { + /** + * Conversation/session id the turn belongs to. On the first turn (or after + * TTL expiry) the store creates a fresh state document under this id; on + * continuation turns the prior message history is loaded from it and the + * updated state — new input, response output, and tool-call outputs — is + * persisted back under the same id before the response is returned. + */ + conversationId: string; + /** + * The store that owns this conversation's state. Callers choose the backend + * (in-memory, file, Redis, …) once and pass it for every turn. + */ + stateStore: ConversationStateStore; +}; + +/** + * Multi-turn variant of `callModel` wired into the conversation state store + * (DEV-127 / docs/multi-turn-state.md). + * + * Each incoming turn: + * 1. loads the prior state by `conversationId` (missing or TTL-expired state + * falls back to a freshly created state under the same id), + * 2. appends the new turn — caller input is appended to the stored message + * history, response output and tool-call outputs are appended as they + * complete, and + * 3. persists the updated state before the response is consumed. + * + * Callers that do not supply a conversation id should use `callModel` + * directly — single-turn behavior there is unchanged. + * + * @example + * ```typescript + * const store = new ConversationStateStore(new InMemoryConversationStateBackend()); + * const first = client.callModelWithState({ + * model: 'gpt-4', + * input: 'My name is Ada.', + * conversationId: 'user-123', + * stateStore: store, + * }); + * await first.getText(); + * + * const second = client.callModelWithState({ + * model: 'gpt-4', + * input: 'What is my name?', + * conversationId: 'user-123', + * stateStore: store, + * }); + * // The model sees the first turn's context and answers "Ada". + * ``` + */ +export function callModelWithState< + TTools extends readonly Tool[], + TSharedSchema extends $ZodObject<$ZodShape> | undefined = undefined, + TShared extends Record = TSharedSchema extends $ZodObject<$ZodShape> ? zodInfer : Record, +>( + client: OpenRouterCore, + request: CallModelWithStateInput & { sharedContextSchema?: TSharedSchema }, + options?: RequestOptions, +): ModelResult { + const { conversationId, stateStore, ...rest } = request; + + if (typeof conversationId !== 'string' || conversationId.length === 0) { + throw new TypeError( + 'callModelWithState requires a non-empty "conversationId" string. ' + + 'Use callModel for single-turn requests without conversation state.', + ); + } + if (!stateStore) { + throw new TypeError( + 'callModelWithState requires a "stateStore" (ConversationStateStore) ' + + 'that owns the conversation\'s state document.', + ); + } + + const state = createStateAccessor(stateStore, conversationId); + + // The rest of the pipeline (ModelResult) already implements load/create/ + // resume, message-history merging, and per-turn persistence against the + // StateAccessor contract — the integration is supplying that accessor. + return callModel( + client, + { + ...rest, + state, + } as CallModelInput & { sharedContextSchema?: TSharedSchema }, + options, + ); +} diff --git a/src/lib/conversation-state-store.ts b/src/lib/conversation-state-store.ts new file mode 100644 index 000000000..2f3be9f7e --- /dev/null +++ b/src/lib/conversation-state-store.ts @@ -0,0 +1,316 @@ +/** + * Conversation state store. + * + * A session/conversation state manager for the `ConversationState` document + * defined by the multi-turn state design (docs/multi-turn-state.md, DEV-127 / + * PR #124). It exposes create / get / append-turn / expire / clear operations + * over a pluggable persistence backend, serializing full turn data (messages, + * tool results, and metadata) as JSON. + * + * Two backends ship with the SDK, both zero-dependency: + * + * - `InMemoryConversationStateStore` — process-local `Map` backend. Per the + * design RFC this is the documented recipe for testing and ephemeral + * single-process use; expiry is implemented by the store itself. + * - `FileConversationStateStore` — JSON-file-per-conversation backend + * (Node.js only), useful for local durable execution without Redis. + * + * Any other backend (Redis, Postgres, …) can be implemented against the + * `ConversationStateBackend` interface, and — per the design — TTL/expiry is + * delegated to that backend where available. + */ + +import { mkdir, readFile, rm, writeFile } from 'node:fs/promises'; +import { dirname, join, normalize, resolve, sep } from 'node:path'; +import { appendToMessages, createInitialState, updateState } from './conversation-state.js'; +import type * as models from '../models/index.js'; +import type { ConversationState, Tool } from './tool-types.js'; + +/** + * Minimal, dependency-free persistence backend for the state store. + * This is deliberately shaped like a subset of the design's `StateAccessor` + * contract (load/save), keyed by conversation id. + */ +export interface ConversationStateBackend { + /** Load a state document, or null if none exists for `id`. */ + load(id: string): Promise | null>; + /** Persist a state document (full-document replace). */ + save(id: string, state: ConversationState): Promise; + /** Remove a state document. No-op if absent. */ + delete(id: string): Promise; + /** Optionally enumerate known conversation ids (enables store-wide `expire`). */ + list?(): Promise; +} + +export interface ConversationStateStoreOptions { + /** + * Time-to-live in milliseconds. When set, `get` returns null for — and + * `expire` removes — conversations whose `updatedAt` is older than this. + * Omit/0 to disable expiry. + */ + ttlMs?: number; + /** Clock override for testing. Defaults to `Date.now`. */ + now?: () => number; +} + +export class ConversationStateNotFoundError extends Error { + constructor(id: string) { + super(`Conversation state not found: ${id}`); + this.name = 'ConversationStateNotFoundError'; + } +} + +export class CorruptedStateError extends Error { + constructor(id: string, detail: string) { + super(`Corrupted conversation state for "${id}": ${detail}`); + this.name = 'CorruptedStateError'; + } +} + +/** + * Serialize a state document to a JSON string. All turn data (messages, + * pending/unsent tool results, partial responses, metadata) lives on the + * document, so JSON round-tripping is lossless for the defined schema. + */ +export function serializeState( + state: ConversationState +): string { + return JSON.stringify(state); +} + +/** + * Deserialize a JSON string into a `ConversationState`, validating the + * structural invariants of the schema. Throws `CorruptedStateError` on + * missing/invalid required fields. + */ +export function deserializeState( + raw: string, + idForError = '' +): ConversationState { + let obj: unknown; + try { + obj = JSON.parse(raw); + } catch (err) { + throw new CorruptedStateError( + idForError, + `invalid JSON (${err instanceof Error ? err.message : String(err)})` + ); + } + assertValidState(obj, idForError); + return obj; +} + +function assertValidState( + obj: unknown, + idForError: string +): asserts obj is ConversationState { + if (typeof obj !== 'object' || obj === null) { + throw new CorruptedStateError(idForError, 'document is not an object'); + } + const rec = obj as Record; + if (typeof rec.id !== 'string' || rec.id.length === 0) { + throw new CorruptedStateError(idForError, 'missing or invalid "id"'); + } + if (!Array.isArray(rec.messages)) { + throw new CorruptedStateError(idForError, 'missing or invalid "messages"'); + } + const validStatuses = new Set(['in_progress', 'awaiting_approval', 'interrupted', 'complete']); + if (typeof rec.status !== 'string' || !validStatuses.has(rec.status)) { + throw new CorruptedStateError(idForError, `missing or invalid "status" (${String(rec.status)})`); + } + if (typeof rec.createdAt !== 'number' || !Number.isFinite(rec.createdAt)) { + throw new CorruptedStateError(idForError, 'missing or invalid "createdAt"'); + } + if (typeof rec.updatedAt !== 'number' || !Number.isFinite(rec.updatedAt)) { + throw new CorruptedStateError(idForError, 'missing or invalid "updatedAt"'); + } +} + +/** + * Create a `StateAccessor` bound to this store for one conversation id — + * the documented consumer-supplied accessor pattern for `callModel`. + */ +export function createStateAccessor( + store: ConversationStateStore, + id: string +): import('./tool-types.js').StateAccessor { + return { + load: () => store.get(id), + save: async (state) => { + // Guard: never let a state document leak under a different id than the + // one this accessor is bound to. ModelResult creates a random-id state + // when load() returns null; saving it would strand the conversation. + if (state.id !== id) { + await store.put({ ...state, id }); + return; + } + await store.put(state); + }, + }; +} + +/** + * Session/conversation state manager. + */ +export class ConversationStateStore { + protected readonly backend: ConversationStateBackend; + protected readonly ttlMs: number; + protected readonly now: () => number; + + constructor(backend: ConversationStateBackend, options: ConversationStateStoreOptions = {}) { + this.backend = backend; + this.ttlMs = options.ttlMs ?? 0; + this.now = options.now ?? (() => Date.now()); + } + + /** Create a new conversation and persist its initial state. Returns the id. */ + async create(id?: string): Promise> { + const state = createInitialState(id); + await this.backend.save(state.id, state); + return state; + } + + /** + * Get a conversation's current state, or null if missing or expired. + * Throws `CorruptedStateError` if the stored document fails validation. + */ + async get(id: string): Promise | null> { + const state = await this.backend.load(id); + if (state === null) return null; + assertValidState(state, id); + if (this.isExpired(state)) return null; + return state; + } + + /** + * Append one turn's items (input messages and/or tool results already + * converted to API items) to the conversation's message history and + * persist. Creates the conversation first if it does not exist yet. + */ + async appendTurn( + id: string, + newItems: models.BaseInputsUnion[], + metadata?: Partial, 'id' | 'createdAt' | 'updatedAt' | 'messages'>> + ): Promise> { + const existing = await this.get(id); + const base = existing ?? createInitialState(id); + const next = updateState(base, { + ...metadata, + messages: appendToMessages(base.messages, newItems), + }); + await this.backend.save(id, next); + return next; + } + + /** Persist a full state document (validated before write). */ + async put(state: ConversationState): Promise { + assertValidState(state, state.id ?? ''); + await this.backend.save(state.id, state); + } + + /** Remove any conversations older than the TTL. Returns ids removed. */ + async expire(ids?: string[]): Promise { + if (!this.ttlMs) return []; + const candidates = ids ?? (await this.backend.list?.()) ?? []; + if (candidates.length === 0) return []; + const removed: string[] = []; + for (const id of candidates) { + const state = await this.backend.load(id); + if (state && this.isExpired(state)) { + await this.backend.delete(id); + removed.push(id); + } + } + return removed; + } + + /** Delete a conversation's state outright. */ + async clear(id: string): Promise { + await this.backend.delete(id); + } + + protected isExpired(state: ConversationState): boolean { + return this.ttlMs > 0 && this.now() - state.updatedAt > this.ttlMs; + } +} + +/** + * In-memory backend (`Map` keyed by conversation id). Suitable for tests and + * ephemeral single-process use. Documents are deep-copied on write and read + * so callers can't mutate stored state by aliasing. + */ +export class InMemoryConversationStateBackend + implements ConversationStateBackend +{ + private readonly map = new Map(); + + async load(id: string): Promise | null> { + const raw = this.map.get(id); + if (raw === undefined) return null; + return deserializeState(raw, id); + } + + async save(id: string, state: ConversationState): Promise { + this.map.set(id, serializeState(state)); + } + + async delete(id: string): Promise { + this.map.delete(id); + } + + async list(): Promise { + return [...this.map.keys()]; + } + + keys(): string[] { + return [...this.map.keys()]; + } +} + +/** + * JSON-file-per-conversation backend under a directory (Node.js only). + * Writes are atomic-ish (write-to-temp then rename is intentionally avoided + * in favor of a single awaited writeFile, matching the design's + * fail-loudly-on-persistence-error stance). + */ +export class FileConversationStateBackend + implements ConversationStateBackend +{ + private readonly dir: string; + + constructor(dir: string) { + this.dir = resolve(dir); + } + + private pathFor(id: string): string { + // Sanitize the id to prevent path traversal; ids are `conv_` by + // default but callers may supply arbitrary strings. + const safe = normalize(id).replace(/[^a-zA-Z0-9_.-]/g, '_'); + const p = join(this.dir, `${safe}.json`); + if (!p.startsWith(this.dir + sep)) { + throw new CorruptedStateError(id, 'id resolves outside state directory'); + } + return p; + } + + async load(id: string): Promise | null> { + let raw: string; + try { + raw = await readFile(this.pathFor(id), 'utf8'); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') return null; + throw err; + } + return deserializeState(raw, id); + } + + async save(id: string, state: ConversationState): Promise { + const p = this.pathFor(id); + await mkdir(dirname(p), { recursive: true }); + await writeFile(p, serializeState(state), 'utf8'); + } + + async delete(id: string): Promise { + await rm(this.pathFor(id), { force: true }); + } +} diff --git a/src/lib/model-result.ts b/src/lib/model-result.ts index e1382aaba..e5edf97b7 100644 --- a/src/lib/model-result.ts +++ b/src/lib/model-result.ts @@ -1039,15 +1039,17 @@ export class ModelResult< let baseRequest = await this.resolveRequestForContext(initialContext); // If we have state with existing messages, use those as input + let newTurnItems: models.BaseInputsUnion[] | null = null; if (this.currentState && this.currentState.messages && Array.isArray(this.currentState.messages) && this.currentState.messages.length > 0) { // Append new input to existing messages const newInput = baseRequest.input; if (newInput) { const inputArray = Array.isArray(newInput) ? newInput : [newInput]; + newTurnItems = inputArray as models.BaseInputsUnion[]; baseRequest = { ...baseRequest, - input: appendToMessages(this.currentState.messages, inputArray as models.BaseInputsUnion[]), + input: appendToMessages(this.currentState.messages, newTurnItems), }; } else { baseRequest = { @@ -1055,6 +1057,21 @@ export class ModelResult< input: this.currentState.messages, }; } + } else if (baseRequest.input) { + // First turn of a fresh state: the whole input is the new turn. + const inputArray = Array.isArray(baseRequest.input) + ? baseRequest.input + : [baseRequest.input]; + newTurnItems = inputArray as models.BaseInputsUnion[]; + } + + // Persist the new-turn input into the stored message history so later + // turns can see it (the response output is appended separately once the + // API responds, and tool results as they execute). + if (this.stateAccessor && this.currentState && newTurnItems) { + await this.saveStateSafely({ + messages: appendToMessages(this.currentState.messages, newTurnItems), + }); } // Store resolved request with stream mode diff --git a/src/sdk/sdk.ts b/src/sdk/sdk.ts index 25257e81b..d79dcca07 100644 --- a/src/sdk/sdk.ts +++ b/src/sdk/sdk.ts @@ -36,6 +36,10 @@ import { callModel as callModelFunc, type CallModelInput, } from "../funcs/call-model.js"; +import { + callModelWithState as callModelWithStateFunc, + type CallModelWithStateInput, +} from "../funcs/call-model-with-state.js"; import type { ModelResult } from "../lib/model-result.js"; import type { RequestOptions } from "../lib/sdks.js"; import { type Tool, ToolType } from "../lib/tool-types.js"; @@ -188,5 +192,23 @@ export class OpenRouter extends ClientSDK { ): ModelResult { return callModelFunc(this, request, options); } + + /** + * Multi-turn variant of {@link callModel}: persists conversation state + * across turns via a `ConversationStateStore` keyed on `conversationId`. + */ + callModelWithState< + TTools extends readonly Tool[], + TSharedSchema extends $ZodObject<$ZodShape> | undefined = undefined, + TShared extends Record = TSharedSchema extends + $ZodObject<$ZodShape> ? zodInfer : Record, + >( + request: CallModelWithStateInput & { + sharedContextSchema?: TSharedSchema; + }, + options?: RequestOptions, + ): ModelResult { + return callModelWithStateFunc(this, request, options); + } // #endregion sdk-class-body } diff --git a/tests/unit/call-model-with-state.test.ts b/tests/unit/call-model-with-state.test.ts new file mode 100644 index 000000000..65c920b74 --- /dev/null +++ b/tests/unit/call-model-with-state.test.ts @@ -0,0 +1,292 @@ +/** + * End-to-end tests for the conversation-state integration in the turn + * pipeline (DEV-127 / task t_d2c7972c). + * + * These tests drive the normal entry point (callModelWithState -> callModel -> + * ModelResult -> betaResponsesSend) with the API call mocked at the + * betaResponsesSend boundary, and assert: + * + * - first turn: state is created under the caller's conversationId and the + * new input + response output are persisted before the turn completes + * - continuation turn: prior state is loaded and the prior turn's context is + * visible in the next API request's input + * - missing/expired state: falls back to a fresh state under the same id + * - single-turn callers (no conversation id) are unaffected: callModel keeps + * working with no store interaction + */ + +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { OpenRouterCore } from '../../src/core.js'; + +// Mock the API boundary. The path must match the import specifier used by +// model-result.ts. +vi.mock('../../src/funcs/betaResponsesSend.js', () => ({ + betaResponsesSend: vi.fn(), +})); + +import { betaResponsesSend } from '../../src/funcs/betaResponsesSend.js'; +import { callModel } from '../../src/funcs/call-model.js'; +import { callModelWithState } from '../../src/funcs/call-model-with-state.js'; +import { + ConversationStateStore, + InMemoryConversationStateBackend, + createStateAccessor, +} from '../../src/lib/conversation-state-store.js'; +import { EventStream } from '../../src/lib/event-streams.js'; + +const betaResponsesSendMock = vi.mocked(betaResponsesSend); + +// Minimal OpenRouterCore stand-in: ModelResult only passes this through to +// betaResponsesSend, which is mocked. +const client = {} as OpenRouterCore; + +function makeResponse(id: string, text: string) { + return { + id, + object: 'response', + createdAt: 1700000000, + completedAt: 1700000001, + status: 'completed', + error: null, + incompleteDetails: null, + instructions: null, + metadata: null, + model: 'test-model', + output: [ + { + type: 'message', + id: `msg_${id}`, + status: 'completed', + role: 'assistant', + content: [{ type: 'output_text', text, annotations: [] }], + }, + ], + parallelToolCalls: true, + toolChoice: 'auto', + tools: [], + temperature: 1, + topP: 1, + frequencyPenalty: 0, + presencePenalty: 0, + }; +} + +/** Build an SSE EventStream carrying a single response.completed event. */ +function sseStreamFor(response: ReturnType): EventStream { + const sseText = + `event: response.completed\n` + + `data: ${JSON.stringify({ type: 'response.completed', response })}\n\n`; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(sseText)); + controller.close(); + }, + }); + return new EventStream( + body, + (msg) => ({ done: false, value: JSON.parse(msg.data ?? '{}') }), + ) as EventStream; +} + +function queueApiResponse(response: ReturnType) { + betaResponsesSendMock.mockResolvedValueOnce({ + ok: true, + value: sseStreamFor(response), + } as Awaited>); +} + +/** The resolved request captured by the most recent mocked API call. */ +function lastApiRequestInput(): unknown { + const calls = betaResponsesSendMock.mock.calls; + const last = calls[calls.length - 1]; + const args = last?.[1] as { responsesRequest: { input?: unknown } }; + return args?.responsesRequest?.input; +} + +function makeStore(ttlMs?: number) { + return new ConversationStateStore( + new InMemoryConversationStateBackend(), + ttlMs !== undefined ? { ttlMs } : {}, + ); +} + +beforeEach(() => { + betaResponsesSendMock.mockReset(); +}); + +describe('callModelWithState — turn pipeline integration', () => { + it('first turn creates state under the conversationId and persists input + output', async () => { + const store = makeStore(); + queueApiResponse(makeResponse('resp_1', 'Hello, Ada!')); + + const result = callModelWithState(client, { + model: 'test-model', + input: 'My name is Ada.', + conversationId: 'conv-user-123', + stateStore: store, + }); + + const text = await result.getText(); + expect(text).toContain('Hello, Ada!'); + + const state = await store.get('conv-user-123'); + expect(state).not.toBeNull(); + expect(state!.id).toBe('conv-user-123'); + expect(state!.status).toBe('complete'); + expect(state!.previousResponseId).toBe('resp_1'); + + // State holds the caller's input and the response output. + const messages = state!.messages as Array | string>; + const serialized = JSON.stringify(messages); + expect(serialized).toContain('My name is Ada.'); + expect( + messages.some( + (m) => typeof m === 'object' && m.type === 'message' && m.role === 'assistant', + ), + ).toBe(true); + + // The API request carried just the first-turn input (no history yet). + expect(lastApiRequestInput()).toBe('My name is Ada.'); + }); + + it('continuation turn loads prior state and exposes earlier context to the model', async () => { + const store = makeStore(); + queueApiResponse(makeResponse('resp_1', 'Nice to meet you, Ada.')); + queueApiResponse(makeResponse('resp_2', 'Your name is Ada.')); + + const first = callModelWithState(client, { + model: 'test-model', + input: 'My name is Ada.', + conversationId: 'conv-multi', + stateStore: store, + }); + await first.getText(); + + const second = callModelWithState(client, { + model: 'test-model', + input: 'What is my name?', + conversationId: 'conv-multi', + stateStore: store, + }); + const text = await second.getText(); + expect(text).toContain('Your name is Ada.'); + + // The second API request must include the first turn's input AND output + // (context from earlier turns visible to later turns). + const secondInput = lastApiRequestInput() as Array>; + expect(Array.isArray(secondInput)).toBe(true); + + const serialized = JSON.stringify(secondInput); + expect(serialized).toContain('My name is Ada.'); // first user turn + expect(serialized).toContain('Nice to meet you, Ada.'); // first assistant reply + expect(serialized).toContain('What is my name?'); // new input + + // The accumulated state covers both turns, in order: user input (stored + // as the raw string the API received) and assistant output per turn. + const state = await store.get('conv-multi'); + const messages = state!.messages as Array | string>; + expect(messages).toHaveLength(4); + expect(messages[0]).toBe('My name is Ada.'); + expect((messages[1] as Record).role).toBe('assistant'); + expect(messages[2]).toBe('What is my name?'); + expect((messages[3] as Record).role).toBe('assistant'); + expect(state!.previousResponseId).toBe('resp_2'); + }); + + it('expired state falls back to a fresh conversation under the same id', async () => { + const store = makeStore(1); // 1ms TTL + + // Pre-seed a stale conversation. + await store.create('conv-stale'); + const seeded = await store.get('conv-stale'); + expect(seeded).not.toBeNull(); + await new Promise((r) => setTimeout(r, 5)); + expect(await store.get('conv-stale')).toBeNull(); // expired + + queueApiResponse(makeResponse('resp_fresh', 'fresh start')); + const result = callModelWithState(client, { + model: 'test-model', + input: 'Hello again.', + conversationId: 'conv-stale', + stateStore: store, + }); + await result.getText(); + + const state = await store.get('conv-stale'); + expect(state).not.toBeNull(); + // Fresh state: only the new turn's input + output, no stale history. + const messages = state!.messages as Array; + expect(messages).toHaveLength(2); + // Same conversation id retained (not a random conv_). + expect(state!.id).toBe('conv-stale'); + }); + + it('persists state under the conversationId even though ModelResult creates a random-id state on first load', async () => { + // Regression: ModelResult calls createInitialState() (random conv_) + // when load() returns null; without the accessor's id guard the document + // would be saved under that random id and the conversation stranded. + const store = makeStore(); + queueApiResponse(makeResponse('resp_guard', 'ok')); + + const result = callModelWithState(client, { + model: 'test-model', + input: 'hi', + conversationId: 'conv-guard', + stateStore: store, + }); + await result.getText(); + + const state = await store.get('conv-guard'); + expect(state).not.toBeNull(); + expect(state!.id).toBe('conv-guard'); + // Nothing saved under a random id: the backend only knows our key. + expect((store as unknown as { backend: InMemoryConversationStateBackend }).backend.keys()) + .toEqual(['conv-guard']); + }); + + it('rejects a missing conversationId', () => { + const store = makeStore(); + expect(() => + callModelWithState(client, { + model: 'test-model', + input: 'hi', + conversationId: '', + stateStore: store, + }), + ).toThrow(TypeError); + }); + + it('single-turn callModel (no conversation id) does not touch any store', async () => { + queueApiResponse(makeResponse('resp_single', 'single turn')); + + const result = callModel(client, { + model: 'test-model', + input: 'no state here', + }); + const text = await result.getText(); + expect(text).toContain('single turn'); + // Nothing to assert against a store — the point is the path works with no + // state accessor at all (existing behavior preserved). + }); + + it('createStateAccessor persists under the bound id and loads through the store', async () => { + const store = makeStore(); + const accessor = createStateAccessor(store, 'conv-direct'); + + expect(await accessor.load()).toBeNull(); + + // Simulate ModelResult's first-turn save of a random-id state. + await accessor.save({ + id: 'conv_random_zzz', + messages: [], + status: 'in_progress', + createdAt: Date.now(), + updatedAt: Date.now(), + }); + + const state = await store.get('conv-direct'); + expect(state).not.toBeNull(); + expect(state!.id).toBe('conv-direct'); + expect(await accessor.load()).not.toBeNull(); + }); +}); diff --git a/tests/unit/conversation-state-store.test.ts b/tests/unit/conversation-state-store.test.ts new file mode 100644 index 000000000..65fc7d029 --- /dev/null +++ b/tests/unit/conversation-state-store.test.ts @@ -0,0 +1,330 @@ +import { mkdtemp, rm, writeFile, readFile, chmod } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join, resolve, sep } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + ConversationStateStore, + CorruptedStateError, + FileConversationStateBackend, + InMemoryConversationStateBackend, + createStateAccessor, + deserializeState, + serializeState, +} from '../../src/lib/conversation-state-store.js'; +import { createInitialState } from '../../src/lib/conversation-state.js'; +import type { ConversationState, Tool } from '../../src/lib/tool-types.js'; + +const userMsg = (text: string) => ({ role: 'user' as const, content: text }); + +describe('ConversationStateStore (in-memory backend)', () => { + let backend: InMemoryConversationStateBackend; + let store: ConversationStateStore; + + beforeEach(() => { + backend = new InMemoryConversationStateBackend(); + store = new ConversationStateStore(backend); + }); + + describe('create', () => { + it('creates and persists an initial state', async () => { + const state = await store.create(); + expect(state.id).toMatch(/^conv_/); + expect(state.status).toBe('in_progress'); + expect(state.messages).toEqual([]); + expect(await store.get(state.id)).toEqual(state); + }); + + it('uses a caller-supplied id', async () => { + const state = await store.create('my-conv'); + expect(state.id).toBe('my-conv'); + expect(await store.get('my-conv')).toEqual(state); + }); + }); + + describe('get', () => { + it('returns null for a missing conversation', async () => { + expect(await store.get('nope')).toBeNull(); + }); + + it('returns null for expired conversations when ttlMs is set', async () => { + vi.useFakeTimers(); + try { + vi.setSystemTime(1_000_000); + const timedStore = new ConversationStateStore(backend, { ttlMs: 1000 }); + await timedStore.create('exp'); + expect(await timedStore.get('exp')).not.toBeNull(); + vi.setSystemTime(1_000_000 + 1001); // past ttl relative to updatedAt + expect(await timedStore.get('exp')).toBeNull(); + } finally { + vi.useRealTimers(); + } + }); + }); + + describe('appendTurn', () => { + it('creates the conversation if it does not exist', async () => { + const state = await store.appendTurn('new-conv', [userMsg('hello')]); + expect(state.id).toBe('new-conv'); + expect(state.messages).toHaveLength(1); + expect(state.messages[0]).toEqual(userMsg('hello')); + }); + + it('appends to existing message history and bumps updatedAt', async () => { + const created = await store.create('c1'); + await store.appendTurn('c1', [userMsg('first')]); + const after = await store.appendTurn('c1', [{ role: 'assistant' as const, content: 'second' }]); + expect(after.messages).toHaveLength(2); + expect(after.updatedAt).toBeGreaterThanOrEqual(created.updatedAt); + expect(after.id).toBe('c1'); + expect(after.createdAt).toBe(created.createdAt); + }); + + it('applies metadata updates (turn data) alongside messages', async () => { + await store.create('c2'); + const after = await store.appendTurn('c2', [userMsg('x')], { + previousResponseId: 'resp_123', + status: 'complete', + }); + expect(after.previousResponseId).toBe('resp_123'); + expect(after.status).toBe('complete'); + }); + + it('serializes tool results as part of the turn data', async () => { + await store.create('c3'); + const toolOutput = { + type: 'function_call_output' as const, + id: 'output_call_1', + callId: 'call_1', + output: JSON.stringify({ ok: true }), + }; + await store.appendTurn('c3', [toolOutput], { + unsentToolResults: [{ callId: 'call_1', name: 'do_thing', output: { ok: true } }], + }); + const loaded = await store.get('c3'); + expect(loaded?.messages[0]).toEqual(toolOutput); + expect(loaded?.unsentToolResults).toEqual([ + { callId: 'call_1', name: 'do_thing', output: { ok: true } }, + ]); + }); + }); + + describe('clear / expire', () => { + it('clear deletes the conversation', async () => { + const s = await store.create('gone'); + await store.clear(s.id); + expect(await store.get(s.id)).toBeNull(); + expect(backend.keys()).toHaveLength(0); + }); + + it('clear is a no-op for missing ids', async () => { + await expect(store.clear('nothing')).resolves.toBeUndefined(); + }); + + it('expire removes conversations past the TTL and returns their ids', async () => { + vi.useFakeTimers(); + try { + const timedStore = new ConversationStateStore(backend, { ttlMs: 100 }); + vi.setSystemTime(5_000); + await timedStore.create('old'); + vi.setSystemTime(5_200); + await timedStore.create('fresh'); // fresh gets updatedAt = 5200 + const removed = await timedStore.expire(backend.keys()); + expect(removed).toEqual(['old']); + expect(await timedStore.get('old')).toBeNull(); + expect(await timedStore.get('fresh')).not.toBeNull(); + } finally { + vi.useRealTimers(); + } + }); + + it('expire returns empty when no TTL configured', async () => { + await store.create('a'); + expect(await store.expire(backend.keys())).toEqual([]); + }); + }); + + describe('serialization', () => { + it('round-trips a full state document', () => { + const state = createInitialState('rt'); + state.previousResponseId = 'resp_x'; + state.pendingToolCalls = [{ id: 'c1', name: 'tool', arguments: { a: 1 } }]; + state.unsentToolResults = [{ callId: 'c1', name: 'tool', output: null, error: 'rejected' }]; + const raw = serializeState(state); + expect(deserializeState(raw, 'rt')).toEqual(state); + }); + + it('deserialize rejects invalid JSON', () => { + expect(() => deserializeState('{not json', 'bad')).toThrow(CorruptedStateError); + }); + + it('deserialize rejects structurally invalid documents', () => { + expect(() => deserializeState('{"id":"x"}', 'x')).toThrow(CorruptedStateError); + expect(() => + deserializeState(JSON.stringify({ id: 'x', messages: [], status: 'bogus', createdAt: 1, updatedAt: 1 }), 'x') + ).toThrow(/status/); + expect(() => + deserializeState(JSON.stringify({ id: 'x', messages: 'nope', status: 'complete', createdAt: 1, updatedAt: 1 }), 'x') + ).toThrow(/messages/); + }); + }); + + describe('error handling', () => { + it('get throws CorruptedStateError on corrupted stored state', async () => { + await store.create('corrupt'); + // Poison the backend directly with a malformed document + await backend.save('corrupt', { id: 'corrupt' } as unknown as ConversationState); + await expect(store.get('corrupt')).rejects.toThrow(CorruptedStateError); + }); + + it('put rejects invalid documents before writing', async () => { + await expect( + store.put({ id: 'x', messages: [], status: 'weird' } as unknown as ConversationState) + ).rejects.toThrow(CorruptedStateError); + expect(backend.keys()).toHaveLength(0); + }); + + it('backend persistence failures propagate loudly', async () => { + const failing = { + load: async () => null, + save: async () => { + throw new Error('disk full'); + }, + delete: async () => {}, + }; + const loudStore = new ConversationStateStore(failing); + await expect(loudStore.create()).rejects.toThrow('disk full'); + }); + + it('stored state is isolated from caller mutation (deep copy via serialization)', async () => { + const s = await store.create('iso'); + (s.messages as unknown[]).push(userMsg('mutated')); + const loaded = await store.get('iso'); + expect(loaded?.messages).toHaveLength(0); + }); + }); + + describe('concurrent access', () => { + it('sequential appendTurn calls never lose turns', async () => { + const n = 25; + for (let i = 0; i < n; i++) { + await store.appendTurn('seq', [userMsg(`turn ${i}`)]); + } + const final = await store.get('seq'); + const msgs = final?.messages as Array<{ content: string }>; + expect(msgs).toHaveLength(n); + expect(msgs.map((m) => m.content)).toEqual( + Array.from({ length: n }, (_, i) => `turn ${i}`) + ); + }); + + it('concurrent appends to distinct conversations do not interfere', async () => { + const ids = ['a', 'b', 'c', 'd']; + await Promise.all( + ids.map(async (id) => { + for (let i = 0; i < 5; i++) { + await store.appendTurn(id, [userMsg(`${id}-${i}`)]); + } + }) + ); + for (const id of ids) { + const s = await store.get(id); + const msgs = (s?.messages ?? []) as Array<{ content: string }>; + expect(msgs).toHaveLength(5); + expect(msgs.every((m) => m.content.startsWith(`${id}-`))).toBe(true); + } + }); + + it('concurrent appends to the same conversation are last-writer-wins per design (no data corruption)', async () => { + // The design (RFC §6) documents single-writer-per-conversation; concurrent + // whole-document writes are last-writer-wins. The store must never produce + // a corrupted/invalid document, regardless of which write lands last. + await store.create('contended'); + await Promise.all( + Array.from({ length: 10 }, (_, i) => + store.appendTurn('contended', [userMsg(`writer ${i}`)]) + ) + ); + const final = await store.get('contended'); + // Document is always structurally valid and internally consistent + expect(final).not.toBeNull(); + expect(final?.messages.length).toBeGreaterThanOrEqual(1); + expect(deserializeState(serializeState(final!), 'contended')).toEqual(final); + }); + }); + + describe('createStateAccessor', () => { + it('bridges the store to the StateAccessor contract used by callModel', async () => { + const accessor = createStateAccessor(store, 'acc-1'); + expect(await accessor.load()).toBeNull(); + const state = createInitialState('acc-1'); + await accessor.save(state); + expect(await accessor.load()).toEqual(state); + // And the store sees the same document + expect(await store.get('acc-1')).toEqual(state); + }); + }); +}); + +describe('FileConversationStateBackend', () => { + let dir: string; + let backend: FileConversationStateBackend; + let store: ConversationStateStore; + + beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), 'conv-state-')); + backend = new FileConversationStateBackend(dir); + store = new ConversationStateStore(backend); + }); + + afterEach(async () => { + await rm(dir, { recursive: true, force: true }); + }); + + it('persists and loads state across store instances (durability)', async () => { + await store.appendTurn('durable', [userMsg('persisted')], { previousResponseId: 'r1' }); + const freshStore = new ConversationStateStore(new FileConversationStateBackend(dir)); + const loaded = await freshStore.get('durable'); + expect(loaded?.messages).toHaveLength(1); + expect(loaded?.previousResponseId).toBe('r1'); + }); + + it('returns null for missing files and clear removes them', async () => { + expect(await store.get('absent')).toBeNull(); + await store.create('to-delete'); + await store.clear('to-delete'); + expect(await store.get('to-delete')).toBeNull(); + }); + + it('surfaces corrupted files as CorruptedStateError', async () => { + await store.create('broken'); + await writeFile(join(dir, 'broken.json'), '### not json ###', 'utf8'); + await expect(store.get('broken')).rejects.toThrow(CorruptedStateError); + }); + + it('sanitizes ids to prevent path traversal', async () => { + const evil = '../../etc/passwd'; + await store.appendTurn(evil, [userMsg('x')]); + const loaded = await store.get(evil); + expect(loaded).not.toBeNull(); + // The file lands inside the state dir with a sanitized name, never outside it + const { readdir } = await import('node:fs/promises'); + const files = await readdir(dir); + expect(files).toHaveLength(1); + expect(files[0]).toBe('.._.._etc_passwd.json'); + expect(resolve(dir, files[0]!).startsWith(resolve(dir) + sep)).toBe(true); + // The stored document keeps the original id for round-trip fidelity + const doc = JSON.parse(await readFile(join(dir, files[0]!), 'utf8')); + expect(doc.id).toBe(evil); + }); + + it('non-ENOENT read errors propagate', async () => { + await store.create('protected'); + const filePath = join(dir, 'protected.json'); + await chmod(filePath, 0o000); + try { + await expect(store.get('protected')).rejects.toThrow(); + } finally { + await chmod(filePath, 0o644); + } + }); +}); diff --git a/tests/unit/multi-turn-integration.test.ts b/tests/unit/multi-turn-integration.test.ts new file mode 100644 index 000000000..0120ef68c --- /dev/null +++ b/tests/unit/multi-turn-integration.test.ts @@ -0,0 +1,382 @@ +/** + * Multi-turn conversation integration tests (DEV-127 / task t_abd9bd76). + * + * These exercise the full turn pipeline (callModelWithState -> callModel -> + * ModelResult -> betaResponsesSend) with the API call mocked at the + * betaResponsesSend boundary, and cover the scenarios the unit tests in + * call-model-with-state.test.ts don't: + * + * - a 3+ turn conversation retaining context from EVERY earlier turn + * - parallel conversations (interleaved turns) keeping isolated state + * - state expiry/cleanup: TTL expiry fallback mid-conversation and + * store.expire()/store.clear() semantics + * - behavior when state is missing or corrupted + * + * Everything runs in the unit project (no live API key needed), so these + * tests run in CI via `npm test`. + */ + +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { OpenRouterCore } from '../../src/core.js'; + +// Mock the API boundary. The path must match the import specifier used by +// model-result.ts. +vi.mock('../../src/funcs/betaResponsesSend.js', () => ({ + betaResponsesSend: vi.fn(), +})); + +import { betaResponsesSend } from '../../src/funcs/betaResponsesSend.js'; +import { callModelWithState } from '../../src/funcs/call-model-with-state.js'; +import { + ConversationStateStore, + CorruptedStateError, + FileConversationStateBackend, + InMemoryConversationStateBackend, + createStateAccessor, +} from '../../src/lib/conversation-state-store.js'; +import { EventStream } from '../../src/lib/event-streams.js'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +const betaResponsesSendMock = vi.mocked(betaResponsesSend); + +// Minimal OpenRouterCore stand-in: ModelResult only passes this through to +// betaResponsesSend, which is mocked. +const client = {} as OpenRouterCore; + +function makeResponse(id: string, text: string) { + return { + id, + object: 'response', + createdAt: 1700000000, + completedAt: 1700000001, + status: 'completed', + error: null, + incompleteDetails: null, + instructions: null, + metadata: null, + model: 'test-model', + output: [ + { + type: 'message', + id: `msg_${id}`, + status: 'completed', + role: 'assistant', + content: [{ type: 'output_text', text, annotations: [] }], + }, + ], + parallelToolCalls: true, + toolChoice: 'auto', + tools: [], + temperature: 1, + topP: 1, + frequencyPenalty: 0, + presencePenalty: 0, + }; +} + +/** Build an SSE EventStream carrying a single response.completed event. */ +function sseStreamFor(response: ReturnType): EventStream { + const sseText = + `event: response.completed\n` + + `data: ${JSON.stringify({ type: 'response.completed', response })}\n\n`; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(sseText)); + controller.close(); + }, + }); + return new EventStream( + body, + (msg) => ({ done: false, value: JSON.parse(msg.data ?? '{}') }), + ) as EventStream; +} + +function queueApiResponse(response: ReturnType) { + betaResponsesSendMock.mockResolvedValueOnce({ + ok: true, + value: sseStreamFor(response), + } as Awaited>); +} + +/** The resolved request input captured by mocked API call `n` (0-based). */ +function apiRequestInput(n: number): unknown { + const call = betaResponsesSendMock.mock.calls[n]; + const args = call?.[1] as { responsesRequest: { input?: unknown } }; + return args?.responsesRequest?.input; +} + +function makeStore(ttlMs?: number) { + return new ConversationStateStore( + new InMemoryConversationStateBackend(), + ttlMs !== undefined ? { ttlMs } : {}, + ); +} + +/** Run one turn through the full pipeline and return its text. */ +async function runTurn( + store: ConversationStateStore, + conversationId: string, + input: string, +): Promise { + const result = callModelWithState(client, { + model: 'test-model', + input, + conversationId, + stateStore: store, + }); + return result.getText(); +} + +beforeEach(() => { + betaResponsesSendMock.mockReset(); +}); + +describe('multi-turn integration — 3+ turn conversation', () => { + it('retains context from every earlier turn across three turns', async () => { + const store = makeStore(); + queueApiResponse(makeResponse('resp_1', 'Got it: Ada, blue.')); + queueApiResponse(makeResponse('resp_2', 'Your color is blue.')); + queueApiResponse(makeResponse('resp_3', 'You are Ada and you like blue.')); + + await runTurn(store, 'conv-three', 'My name is Ada and my favorite color is blue.'); + await runTurn(store, 'conv-three', 'What is my favorite color?'); + const text = await runTurn(store, 'conv-three', 'Summarize what you know about me.'); + expect(text).toContain('Ada'); + + // Turn 3's API request must include both earlier turns' inputs AND + // outputs, in conversation order, plus the new input. + const thirdInput = apiRequestInput(2) as Array>; + expect(Array.isArray(thirdInput)).toBe(true); + const serialized = JSON.stringify(thirdInput); + expect(serialized).toContain('My name is Ada and my favorite color is blue.'); + expect(serialized).toContain('Got it: Ada, blue.'); + expect(serialized).toContain('What is my favorite color?'); + expect(serialized).toContain('Your color is blue.'); + expect(serialized).toContain('Summarize what you know about me.'); + + // Ordering: user/assistant alternate in turn order. + const texts = thirdInput.map((item) => { + if (typeof item === 'string') return item; + const content = (item as { content?: Array<{ text?: string }> }).content; + return content?.[0]?.text ?? JSON.stringify(item); + }); + const turn1User = texts.findIndex((t) => t.includes('favorite color is blue')); + const turn1Assistant = texts.findIndex((t) => t.includes('Got it: Ada')); + const turn2User = texts.findIndex((t) => t.includes('What is my favorite color?')); + const turn2Assistant = texts.findIndex((t) => t.includes('Your color is blue.')); + const turn3User = texts.findIndex((t) => t.includes('Summarize')); + expect(turn1User).toBeGreaterThanOrEqual(0); + expect(turn1Assistant).toBeGreaterThan(turn1User); + expect(turn2User).toBeGreaterThan(turn1Assistant); + expect(turn2Assistant).toBeGreaterThan(turn2User); + expect(turn3User).toBeGreaterThan(turn2Assistant); + + // Accumulated state covers all three turns and tracks the latest + // response id. + const state = await store.get('conv-three'); + expect(state).not.toBeNull(); + expect(state!.messages).toHaveLength(6); // 3 user + 3 assistant + expect(state!.previousResponseId).toBe('resp_3'); + expect(state!.status).toBe('complete'); + }); +}); + +describe('multi-turn integration — parallel conversations', () => { + it('keeps state isolated across interleaved conversations', async () => { + const store = makeStore(); + queueApiResponse(makeResponse('resp_a1', 'Hi Alice!')); + queueApiResponse(makeResponse('resp_b1', 'Hi Bob!')); + queueApiResponse(makeResponse('resp_a2', 'Alice, as I said.')); + queueApiResponse(makeResponse('resp_b2', 'Bob, as I said.')); + + // Interleave turns between two conversations. + await runTurn(store, 'conv-alice', 'I am Alice.'); + await runTurn(store, 'conv-bob', 'I am Bob.'); + await runTurn(store, 'conv-alice', 'Who am I?'); + await runTurn(store, 'conv-bob', 'Who am I?'); + + // Each conversation's second turn sees only its own first turn. + const aliceSecond = JSON.stringify(apiRequestInput(2)); + expect(aliceSecond).toContain('I am Alice.'); + expect(aliceSecond).toContain('Hi Alice!'); + expect(aliceSecond).toContain('Who am I?'); + expect(aliceSecond).not.toContain('Bob'); + + const bobSecond = JSON.stringify(apiRequestInput(3)); + expect(bobSecond).toContain('I am Bob.'); + expect(bobSecond).toContain('Hi Bob!'); + expect(bobSecond).toContain('Who am I?'); + expect(bobSecond).not.toContain('Alice'); + + // The store holds exactly two independent documents. + const alice = await store.get('conv-alice'); + const bob = await store.get('conv-bob'); + expect(alice!.messages).toHaveLength(4); + expect(bob!.messages).toHaveLength(4); + expect(alice!.previousResponseId).toBe('resp_a2'); + expect(bob!.previousResponseId).toBe('resp_b2'); + expect(JSON.stringify(alice!.messages)).not.toContain('Bob'); + expect(JSON.stringify(bob!.messages)).not.toContain('Alice'); + }); + + it('handles concurrent first turns on the same store without cross-talk', async () => { + const store = makeStore(); + queueApiResponse(makeResponse('resp_c1', 'one')); + queueApiResponse(makeResponse('resp_c2', 'two')); + + // Two conversations start truly concurrently (no awaiting between the + // pipeline entry points). + const [t1, t2] = await Promise.all([ + runTurn(store, 'conv-par-1', 'alpha'), + runTurn(store, 'conv-par-2', 'beta'), + ]); + expect(t1 + t2).toMatch(/one|two/); + + const s1 = await store.get('conv-par-1'); + const s2 = await store.get('conv-par-2'); + expect(s1!.id).toBe('conv-par-1'); + expect(s2!.id).toBe('conv-par-2'); + expect(JSON.stringify(s1!.messages)).toContain('alpha'); + expect(JSON.stringify(s1!.messages)).not.toContain('beta'); + expect(JSON.stringify(s2!.messages)).toContain('beta'); + expect(JSON.stringify(s2!.messages)).not.toContain('alpha'); + }); +}); + +describe('multi-turn integration — state expiry and cleanup', () => { + it('mid-conversation TTL expiry starts a fresh thread under the same id', async () => { + const store = makeStore(20); // 20ms TTL + queueApiResponse(makeResponse('resp_e1', 'remembered')); + queueApiResponse(makeResponse('resp_e2', 'fresh again')); + + await runTurn(store, 'conv-exp', 'First message.'); + const before = await store.get('conv-exp'); + expect(before!.messages).toHaveLength(2); + + // Let the conversation expire, then continue "the same" conversation. + await new Promise((r) => setTimeout(r, 30)); + expect(await store.get('conv-exp')).toBeNull(); // expired + + await runTurn(store, 'conv-exp', 'Second message.'); + const after = await store.get('conv-exp'); + expect(after).not.toBeNull(); + expect(after!.id).toBe('conv-exp'); + // Fresh thread: only the post-expiry turn survives; the model's second + // request must NOT contain the expired first turn. + expect(after!.messages).toHaveLength(2); + const secondInput = JSON.stringify(apiRequestInput(1)); + expect(secondInput).not.toContain('First message.'); + expect(secondInput).toContain('Second message.'); + }); + + it('store.expire() removes only stale conversations and reports removed ids', async () => { + const store = makeStore(20); + queueApiResponse(makeResponse('resp_x1', 'old')); + queueApiResponse(makeResponse('resp_x2', 'new')); + + await runTurn(store, 'conv-old', 'old turn'); + await new Promise((r) => setTimeout(r, 30)); + await runTurn(store, 'conv-new', 'new turn'); + + const removed = await store.expire(); + expect(removed).toEqual(['conv-old']); + expect(await store.get('conv-old')).toBeNull(); + expect(await store.get('conv-new')).not.toBeNull(); + }); + + it('store.clear() deletes a conversation; the next turn starts fresh', async () => { + const store = makeStore(); + queueApiResponse(makeResponse('resp_cl1', 'first')); + queueApiResponse(makeResponse('resp_cl2', 'second')); + + await runTurn(store, 'conv-clear', 'Forget me.'); + expect(await store.get('conv-clear')).not.toBeNull(); + + await store.clear('conv-clear'); + expect(await store.get('conv-clear')).toBeNull(); + + await runTurn(store, 'conv-clear', 'I am back.'); + const state = await store.get('conv-clear'); + expect(state!.messages).toHaveLength(2); + expect(JSON.stringify(state!.messages)).not.toContain('Forget me.'); + }); +}); + +describe('multi-turn integration — missing or corrupted state', () => { + it('missing state on first turn creates a fresh document under the caller id', async () => { + const store = makeStore(); + expect(await store.get('conv-never')).toBeNull(); + + queueApiResponse(makeResponse('resp_m1', 'hello')); + await runTurn(store, 'conv-never', 'First contact.'); + + const state = await store.get('conv-never'); + expect(state).not.toBeNull(); + expect(state!.id).toBe('conv-never'); + expect(state!.previousResponseId).toBe('resp_m1'); + }); + + it('store.get() throws CorruptedStateError when the backend holds invalid JSON', async () => { + // Reach into the in-memory backend and poison a document directly. + const backend = new InMemoryConversationStateBackend(); + const store = new ConversationStateStore(backend); + (backend as unknown as { map: Map }).map.set( + 'conv-bad', + '{not json', + ); + + await expect(store.get('conv-bad')).rejects.toThrow(CorruptedStateError); + }); + + it('store.get() throws CorruptedStateError when the document fails schema validation', async () => { + const backend = new InMemoryConversationStateBackend(); + const store = new ConversationStateStore(backend); + (backend as unknown as { map: Map }).map.set( + 'conv-invalid', + JSON.stringify({ id: 'conv-invalid', messages: 'not-an-array' }), + ); + + await expect(store.get('conv-invalid')).rejects.toThrow(/Corrupted conversation state/); + }); + + it('a corrupted persisted file surfaces as CorruptedStateError on the next turn', async () => { + const dir = await mkdtemp(join(tmpdir(), 'conv-state-')); + try { + const store = new ConversationStateStore(new FileConversationStateBackend(dir)); + // Seed a valid conversation through the pipeline, then corrupt the file. + queueApiResponse(makeResponse('resp_f1', 'before corruption')); + await runTurn(store, 'conv-file', 'Hi.'); + const stateFile = join(dir, 'conv-file.json'); + const good = JSON.parse(await readFile(stateFile, 'utf8')); + expect(good.id).toBe('conv-file'); + + await writeFile(stateFile, '{"id": 123, "messages": []}', 'utf8'); + + queueApiResponse(makeResponse('resp_f2', 'after corruption')); + await expect(runTurn(store, 'conv-file', 'Anyone there?')).rejects.toThrow( + CorruptedStateError, + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('clearing one conversation does not affect siblings', async () => { + const store = makeStore(); + queueApiResponse(makeResponse('resp_s1', 'keep')); + queueApiResponse(makeResponse('resp_s2', 'drop')); + + await runTurn(store, 'conv-keep', 'keep me'); + await runTurn(store, 'conv-drop', 'drop me'); + + await store.clear('conv-drop'); + expect(await store.get('conv-drop')).toBeNull(); + expect(await store.get('conv-keep')).not.toBeNull(); + + // Accessor for the cleared conversation observes the deletion. + const accessor = createStateAccessor(store, 'conv-drop'); + expect(await accessor.load()).toBeNull(); + }); +});