From 6ab46948cb48c91c346c128ed431192d79f68a11 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Fri, 17 Jul 2026 01:11:56 -0600 Subject: [PATCH] feat(chat-routes): product turn-lifecycle seams (heartbeat, beforeTurn, lifecycle, context-gate, turn-lock) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add five optional, domain-free seams to createChatTurnRoutes so a complex product turn-orchestrator can delete its hand-rolled generator and compose the vertical. Every seam is optional — omitting all of them reproduces today's exact behavior, and the existing #194 tests are untouched. handleChatTurn stays the engine; the seams only wrap its input, its producer stream, and its settle. - turnLock: async acquire before any side effect / release once when the turn settles (drain finish, short-circuit, or throw). - contextGate: pre-producer domain-readiness short-circuit with a product Response (distinct from authorize). - beforeTurn: observe the assembled producer input and optionally rewrite the prompt / prior messages. - lifecycle: deterministic onTurnStart then exactly one of onTurnComplete(finalText, usage, durationMs) / onTurnError(error, durationMs). - heartbeat: keepalive injected while the producer is quiet; window resets on every real event. - onRawEvent: the raw producer events, before the engine frames them (telemetry). --- examples/chat-app.md | 66 ++++ src/chat-routes/turn-routes.ts | 505 ++++++++++++++++++++------ tests/chat-routes/turn-routes.test.ts | 156 ++++++++ 3 files changed, 615 insertions(+), 112 deletions(-) diff --git a/examples/chat-app.md b/examples/chat-app.md index 7fa8479..f3c0730 100644 --- a/examples/chat-app.md +++ b/examples/chat-app.md @@ -157,3 +157,69 @@ into the sandbox workspace and referenced by `path` (the ~1 MiB gateway body cap makes that two-step mandatory). Question cards render with `InteractionQuestionCard` + `useChatInteractions` and answer through `/api/chat/interactions` — see `/web-react`. + +## Advanced hooks (optional) + +A complex product turn-orchestrator does more than stream: it holds a +single-flight lock, keeps the client alive through long tool calls, gates on +domain readiness, and books telemetry. `createChatTurnRoutes` exposes five +optional seams for exactly that — **omit any one and the route behaves exactly +as above.** They compose with `authorize` / `produce` / `store` / `interactions`. + +```ts +const routes = createChatTurnRoutes({ + projectId: 'acme-agent', + authorize, store, turnStore, produce, // as above + + // 1. Single-flight lock — acquired before any side effect, released once + // when the turn settles (drain finish), on short-circuit, or on throw. + turnLock: { + acquire: async ({ identity, executionId }) => { + const got = await acquireLock(identity.tenantId, identity.sessionId, executionId) + return got.ok + ? { acquired: true, handle: got.lockId } + : { acquired: false, response: Response.json({ code: 'turn_in_flight' }, { status: 409 }) } + }, + release: (lockId) => releaseLock(lockId as string), + }, + + // 2. Domain-readiness gate — short-circuit BEFORE the producer runs (the user + // row is already persisted; return the assistant side of the turn). + contextGate: async ({ identity, prompt }) => { + const ready = await computeContextSufficiency(identity.tenantId) + return ready.ok ? { proceed: true } : { proceed: false, response: cannedAskForContext(ready.missing) } + }, + + // 3. Observe + augment the assembled input before the producer runs. + beforeTurn: async ({ prompt, priorMessages, identity }) => { + const composed = await composeSystemPromptWithCertified(identity.tenantId) + return { priorMessages: [systemMessage(composed), ...priorMessages] } + }, + + // 4. Deterministic run telemetry — start, then exactly one of complete/error. + lifecycle: { + onTurnStart: ({ identity, executionId }) => startRun(identity, executionId), + onTurnComplete: ({ finalText, usage, durationMs }) => endRun({ pass: true, finalText, usage, durationMs }), + onTurnError: ({ error, durationMs }) => endRun({ pass: false, error, durationMs }), + }, + + // 5. Keepalive while the producer is quiet (provisioning, first-token wait). + // Window resets on every real event; a chatty producer never triggers one. + heartbeat: { + intervalMs: 5_000, + event: ({ elapsedMs }) => ({ type: 'run-phase', data: { phase: 'working', heartbeat: true, elapsedMs } }), + }, + + // Raw producer events for telemetry, before the engine frames them (distinct + // from `onEvent`, which sees the engine-framed stream incl. lifecycle). + onRawEvent: (event) => emitToTrace(event), +}) +``` + +Each seam replaces a slice a hand-rolled generator otherwise owns: the lock is +the dual-scope session/workspace guard; `contextGate` is the sufficiency +short-circuit; `beforeTurn` is the prompt-composition step; `lifecycle` is the +`startRun`/`endRun`/`flush` triple (fires on failure too); `heartbeat` is the +`withHeartbeat` wrapper around silent waits. `handleChatTurn` stays the turn +engine underneath — these only wrap its input, its producer stream, and its +settle. diff --git a/src/chat-routes/turn-routes.ts b/src/chat-routes/turn-routes.ts index fa9bdc4..8206851 100644 --- a/src/chat-routes/turn-routes.ts +++ b/src/chat-routes/turn-routes.ts @@ -19,6 +19,16 @@ * Handlers are web-standard `Request → Response` (Workers, Node 18+, Deno) — * no router import. Auth/access is one injected `authorize` seam, composable * with `/app-auth` guards but not coupled to them. + * + * Five optional product seams let a complex turn-orchestrator compose the + * vertical instead of hand-rolling a generator — each omittable to the exact + * behavior above: `turnLock` (single-flight acquire/release around the turn), + * `contextGate` (pre-producer domain-readiness short-circuit), `beforeTurn` + * (observe + augment the producer input), `lifecycle` (deterministic + * start/complete/error telemetry), `heartbeat` (keepalive during silent + * producer waits), plus `onRawEvent` (the raw producer events, for telemetry). + * `handleChatTurn` stays the engine — the seams only wrap its input, its + * producer stream, and its settle. */ import { deriveExecutionId, handleChatTurn } from '@tangle-network/agent-runtime' @@ -123,6 +133,81 @@ export interface ChatTurnProduceArgs { priorMessages: PersistedChatMessageForTurn[] } +/** One event as it crosses the route: the producer's own vocabulary, or an + * injected keepalive. Same shape the engine forwards verbatim. */ +type ChatRouteEvent = { type: string; data?: Record } + +/** Keepalive emitted while the producer is quiet (long tool calls, first-token + * wait) so client watchdogs stay re-armed. One is emitted each time + * `intervalMs` elapses with no producer event; the window resets on every real + * event, so a chatty producer never triggers one. The product owns the event + * shape (`type` + `data`). Omit → no keepalives (today's behavior). */ +export interface ChatTurnHeartbeat { + intervalMs: number + event(info: { elapsedMs: number; tick: number }): ChatRouteEvent +} + +/** Patch a `beforeTurn` hook returns to augment the producer's input. Omitted + * fields keep the route-assembled value; the product's `produce` still owns + * the system prompt. */ +export interface ChatTurnInputPatch { + prompt?: string | ChatTurnPartInput[] + priorMessages?: PersistedChatMessageForTurn[] +} + +/** Pre-turn readiness verdict — proceed, or short-circuit with the product's + * own `Response` (e.g. a canned assistant reply asking for missing context). + * Distinct from `authorize`: this gates domain readiness, not access. */ +export type ChatTurnGateResult = + | { proceed: true } + | { proceed: false; response: Response } + +/** Single-flight lock verdict — acquired (with an opaque handle passed back to + * `release`), or already held (short-circuit with the product's 409-style + * `Response`). */ +export type ChatTurnLockResult = + | { acquired: true; handle?: unknown } + | { acquired: false; response: Response } + +/** Async acquire/release wrapped around the turn. `acquire` runs before any + * side effect; `release` runs once when the turn settles — including on a + * short-circuit or a throw. */ +export interface ChatTurnLock { + acquire(args: ChatTurnProduceArgs): ChatTurnLockResult | Promise + release(handle: unknown): void | Promise +} + +interface ChatTurnLifecycleBase { + identity: ChatTurnIdentity + executionId: string + turnStreamId: string + context: TContext +} +export interface ChatTurnLifecycleStart extends ChatTurnLifecycleBase { + startedAt: number +} +export interface ChatTurnLifecycleComplete extends ChatTurnLifecycleBase { + finalText: string + usage: ChatTurnUsage + durationMs: number +} +export interface ChatTurnLifecycleError extends ChatTurnLifecycleBase { + error: unknown + durationMs: number +} + +/** Deterministic run telemetry: `onTurnStart` fires before the producer runs; + * exactly one of `onTurnComplete` / `onTurnError` fires after the turn + * settles, always after `onTurnStart`. Failure is derived from the turn's own + * `error` / `session.run.failed` events (or a drain throw), not the engine's + * lifecycle envelope. Hook errors are swallowed — telemetry never fails a + * turn. */ +export interface ChatTurnLifecycle { + onTurnStart?(info: ChatTurnLifecycleStart): void | Promise + onTurnComplete?(info: ChatTurnLifecycleComplete): void | Promise + onTurnError?(info: ChatTurnLifecycleError): void | Promise +} + export interface CreateChatTurnRoutesOptions { /** Names the product in `deriveExecutionId` so retries land on the same * substrate execution. */ @@ -141,6 +226,26 @@ export interface CreateChatTurnRoutesOptions { * wrapped in `createSandboxChatProducer`. Router/openai-compat lane: the * product's own producer. May be async (box resolution). */ produce(args: ChatTurnProduceArgs): ChatTurnRouteProducer | Promise + /** Single-flight lock acquired before any side effect and released once when + * the turn settles (including short-circuit/throw). Omit → no lock. */ + turnLock?: ChatTurnLock + /** Pre-turn readiness gate that can short-circuit with a product `Response` + * before the producer runs (the user row is already persisted). Runs after + * `turnLock.acquire`, before `beforeTurn`. Omit → always proceed. */ + contextGate?(args: ChatTurnProduceArgs): ChatTurnGateResult | Promise + /** Observe the assembled producer input and optionally augment it (rewrite + * the prompt / prior messages) before the producer runs. Omit → no change. */ + beforeTurn?(args: ChatTurnProduceArgs): ChatTurnInputPatch | void | Promise + /** Deterministic run telemetry (start / complete / error) with identity and + * timing. Omit → no telemetry. */ + lifecycle?: ChatTurnLifecycle + /** Keepalive injected while the producer is quiet. Omit → no keepalives. */ + heartbeat?: ChatTurnHeartbeat + /** Observe each event the producer emits, before the engine frames it and + * before any heartbeat injection (the raw sidecar-producer events, for + * telemetry). Never alters the stream; errors are swallowed. Distinct from + * `onEvent`, which sees the engine-framed stream incl. lifecycle envelopes. */ + onRawEvent?(event: ChatRouteEvent, context: TContext): void | Promise /** Pre-persist transform of the final text (e.g. `/redact`'s `redactPII`). * Live stream is never altered. */ transformFinalText?(text: string): string | Promise @@ -227,6 +332,63 @@ function userPartsWithFiles( return toChatMessageParts([...userParts, ...fileParts.map((part) => ({ ...part }))]) } +// ── producer-stream wrappers (heartbeat + raw tap) ─────────────────────────── + +/** Fire `onRawEvent` for each producer event, before the engine frames it. + * Best-effort — a telemetry throw is logged, never propagated. */ +async function* tapRawEvents( + source: AsyncIterable, + onRawEvent: (event: ChatRouteEvent) => void | Promise, + log: (message: string, meta?: Record) => void, +): AsyncGenerator { + for await (const event of source) { + try { + await onRawEvent(event) + } catch (err) { + log('[chat-routes] onRawEvent failed', { error: err instanceof Error ? err.message : String(err) }) + } + yield event + } +} + +/** Inject a keepalive whenever `intervalMs` elapses with no source event. The + * silent window (elapsed + tick) resets on every real event, so a producer + * that keeps emitting never triggers a heartbeat. Closes the source on early + * return, matching a `for await` over it. */ +async function* withStreamHeartbeat( + source: AsyncIterable, + intervalMs: number, + makeEvent: (info: { elapsedMs: number; tick: number }) => ChatRouteEvent, +): AsyncGenerator { + const iterator = source[Symbol.asyncIterator]() + try { + let pending = iterator.next() + let windowStart = Date.now() + let tick = 0 + for (;;) { + let timer: ReturnType | undefined + const heartbeat = new Promise<'heartbeat'>((resolve) => { + timer = setTimeout(() => resolve('heartbeat'), intervalMs) + }) + const winner = await Promise.race([pending.then(() => 'event' as const), heartbeat]) + if (timer !== undefined) clearTimeout(timer) + if (winner === 'heartbeat') { + tick += 1 + yield makeEvent({ elapsedMs: Date.now() - windowStart, tick }) + continue + } + const result = await pending + if (result.done) return + yield result.value + pending = iterator.next() + windowStart = Date.now() + tick = 0 + } + } finally { + await iterator.return?.() + } +} + // ── the factory ──────────────────────────────────────────────────────────── export function createChatTurnRoutes( @@ -261,15 +423,6 @@ export function createChatTurnRoutes( })) const chatTurn = resolveChatTurn({ existingMessages, userContent: content, turnId }) - if (chatTurn.shouldInsertUserMessage) { - await options.store.appendMessage({ - threadId: payload.threadId, - role: 'user', - content, - parts: userPartsWithFiles(chatTurn.userParts, fileParts), - }) - } - const identity: ChatTurnIdentity = { tenantId, sessionId: payload.threadId, @@ -290,119 +443,247 @@ export function createChatTurnRoutes( ? [{ type: 'text', text: content }, ...fileParts] : [...fileParts] - // Durability tap: every engine event buffers (coalesced) so a dropped - // client replays the tail. Live delivery rides the Response body, not the - // tap, so `write` is intentionally absent. - const tap = createBufferedTurnTap({ - store: options.turnStore, - turnId: turnStreamId, - scopeId: payload.threadId, - coalesce: options.coalesceTurnEvents ?? coalesceDeltas, - }) - const turnMarker = { type: 'turn', turnId: turnStreamId } - await tap.onEvent(turnMarker) - - let producer: ChatTurnRouteProducer | undefined - let runFailed = false - - const result = handleChatTurn({ + // The producer input every pre-turn seam reads (and `beforeTurn` may + // rewrite). Mutated in place before the producer's deferred first pull. + let produceArgs: ChatTurnProduceArgs = { + request, + body: payload, identity, - waitUntil: ctx?.waitUntil, - log, - hooks: { - // The engine wants a synchronous producer; box resolution is async — - // defer it into the generator's first pull. - produce: () => ({ - stream: (async function* () { - producer = await options.produce({ - request, - body: payload, - identity, - context, - prompt, - executionId, - turnStreamId, - priorMessages: chatTurn.priorMessages, - }) - for await (const event of producer.stream) yield event - })(), - finalText: () => producer?.finalText() ?? '', - }), - onEvent: async (event) => { - if (event.type === 'session.run.failed' || event.type === 'error') runFailed = true - await tap.onEvent(event) - if (options.onEvent) await options.onEvent(event, context) - }, - ...(options.transformFinalText ? { transformFinalText: options.transformFinalText } : {}), - persistAssistantMessage: async ({ finalText }) => { - // The typed boundary: stream-normalizer records → stored vocabulary - // (validating projection owned by /chat-store — no cast here). - const parts = producer?.assistantParts ? toChatMessageParts(producer.assistantParts()) : undefined - if (!finalText.trim() && (!parts || parts.length === 0)) return - const usage = producer?.usage?.() ?? {} - await options.store.appendMessage({ - threadId: payload.threadId, - role: 'assistant', - content: finalText, - ...(parts && parts.length > 0 ? { parts } : {}), - ...(producer?.model ? { model: producer.model } : {}), - ...(usage.inputTokens !== undefined ? { inputTokens: usage.inputTokens } : {}), - ...(usage.outputTokens !== undefined ? { outputTokens: usage.outputTokens } : {}), - ...(usage.reasoningTokens !== undefined ? { reasoningTokens: usage.reasoningTokens } : {}), - ...(usage.cacheReadTokens !== undefined ? { cacheReadTokens: usage.cacheReadTokens } : {}), - ...(usage.cacheWriteTokens !== undefined ? { cacheWriteTokens: usage.cacheWriteTokens } : {}), - ...(usage.costUsd !== undefined ? { costUsd: usage.costUsd } : {}), - }) - }, - ...(options.onTurnComplete - ? { - onTurnComplete: ({ identity: turnIdentity, finalText }: { identity: ChatTurnIdentity; finalText: string }) => - options.onTurnComplete!({ identity: turnIdentity, finalText, context }), - } - : {}), - ...(options.traceFlush ? { traceFlush: () => options.traceFlush!(context) } : {}), - }, - }) + context, + prompt, + executionId, + turnStreamId, + priorMessages: chatTurn.priorMessages, + } - // Tee: one branch to the live client, one drained under waitUntil so the - // turn (and its buffering via onEvent) runs to completion after a client - // drop — the engine body executes as it is pulled. - const [clientBody, drainBody] = result.body.tee() - const drained = (async () => { - const reader = drainBody.getReader() + // Single-flight lock: acquire before any side effect. `release` runs + // exactly once — in the drain's `finally` on a normal turn, or right here + // on a short-circuit / throw. + let lockAcquired = false + let lockHandle: unknown + let lockReleased = false + const releaseLock = async (): Promise => { + if (!lockAcquired || lockReleased) return + lockReleased = true try { - for (;;) { - const { done } = await reader.read() - if (done) break - } - await tap.done(runFailed ? 'error' : 'complete') + await options.turnLock!.release(lockHandle) } catch (err) { - await tap.done('error') - log('[chat-routes] turn drain failed', { + log('[chat-routes] turnLock.release failed', { turnId: turnStreamId, error: err instanceof Error ? err.message : String(err), }) } - })() - if (ctx?.waitUntil) ctx.waitUntil(drained) - else void drained.catch(() => {}) + } + if (options.turnLock) { + const acquired = await options.turnLock.acquire(produceArgs) + if (!acquired.acquired) return acquired.response + lockAcquired = true + lockHandle = acquired.handle + } + + try { + if (chatTurn.shouldInsertUserMessage) { + await options.store.appendMessage({ + threadId: payload.threadId, + role: 'user', + content, + parts: userPartsWithFiles(chatTurn.userParts, fileParts), + }) + } - // Announce the replay handle before the engine's first event. - const encoder = new TextEncoder() - const marker = new ReadableStream({ - start(controller) { - controller.enqueue(encoder.encode(`${JSON.stringify(turnMarker)}\n`)) - controller.close() - }, - }) - const body = concatStreams([marker, clientBody]) + // Domain-readiness gate: may short-circuit with the product's own + // response before the producer runs. The user row above is kept (a real + // user turn); the gate's response is the assistant side of it. + if (options.contextGate) { + const gate = await options.contextGate(produceArgs) + if (!gate.proceed) { + await releaseLock() + return gate.response + } + } - return new Response(body, { - headers: { - 'Content-Type': result.contentType, - 'Cache-Control': 'no-cache', - }, - }) + // Observe + optionally augment the assembled producer input. + if (options.beforeTurn) { + const patch = await options.beforeTurn(produceArgs) + if (patch) produceArgs = { ...produceArgs, ...patch } + } + + // Durability tap: every engine event buffers (coalesced) so a dropped + // client replays the tail. Live delivery rides the Response body, not the + // tap, so `write` is intentionally absent. + const tap = createBufferedTurnTap({ + store: options.turnStore, + turnId: turnStreamId, + scopeId: payload.threadId, + coalesce: options.coalesceTurnEvents ?? coalesceDeltas, + }) + const turnMarker = { type: 'turn', turnId: turnStreamId } + await tap.onEvent(turnMarker) + + let producer: ChatTurnRouteProducer | undefined + let runFailed = false + // Data of the event that marked the run failed — handed to `onTurnError` + // when no drain throw supplies a richer cause. + let lastFailureData: Record | undefined + + const turnStartedAtMs = Date.now() + if (options.lifecycle?.onTurnStart) { + try { + await options.lifecycle.onTurnStart({ + identity, executionId, turnStreamId, context, startedAt: turnStartedAtMs, + }) + } catch (err) { + log('[chat-routes] lifecycle.onTurnStart failed', { + turnId: turnStreamId, + error: err instanceof Error ? err.message : String(err), + }) + } + } + + const result = handleChatTurn({ + identity, + waitUntil: ctx?.waitUntil, + log, + hooks: { + // The engine wants a synchronous producer; box resolution is async — + // defer it into the generator's first pull. + produce: () => ({ + stream: (async function* () { + producer = await options.produce(produceArgs) + let source: AsyncIterable = producer.stream + if (options.onRawEvent) { + source = tapRawEvents(source, (event) => options.onRawEvent!(event, context), log) + } + if (options.heartbeat) { + source = withStreamHeartbeat(source, options.heartbeat.intervalMs, options.heartbeat.event) + } + for await (const event of source) yield event + })(), + finalText: () => producer?.finalText() ?? '', + }), + onEvent: async (event) => { + if (event.type === 'session.run.failed' || event.type === 'error') { + runFailed = true + lastFailureData = event.data + } + await tap.onEvent(event) + if (options.onEvent) await options.onEvent(event, context) + }, + ...(options.transformFinalText ? { transformFinalText: options.transformFinalText } : {}), + persistAssistantMessage: async ({ finalText }) => { + // The typed boundary: stream-normalizer records → stored vocabulary + // (validating projection owned by /chat-store — no cast here). + const parts = producer?.assistantParts ? toChatMessageParts(producer.assistantParts()) : undefined + if (!finalText.trim() && (!parts || parts.length === 0)) return + const usage = producer?.usage?.() ?? {} + await options.store.appendMessage({ + threadId: payload.threadId, + role: 'assistant', + content: finalText, + ...(parts && parts.length > 0 ? { parts } : {}), + ...(producer?.model ? { model: producer.model } : {}), + ...(usage.inputTokens !== undefined ? { inputTokens: usage.inputTokens } : {}), + ...(usage.outputTokens !== undefined ? { outputTokens: usage.outputTokens } : {}), + ...(usage.reasoningTokens !== undefined ? { reasoningTokens: usage.reasoningTokens } : {}), + ...(usage.cacheReadTokens !== undefined ? { cacheReadTokens: usage.cacheReadTokens } : {}), + ...(usage.cacheWriteTokens !== undefined ? { cacheWriteTokens: usage.cacheWriteTokens } : {}), + ...(usage.costUsd !== undefined ? { costUsd: usage.costUsd } : {}), + }) + }, + ...(options.onTurnComplete + ? { + onTurnComplete: ({ identity: turnIdentity, finalText }: { identity: ChatTurnIdentity; finalText: string }) => + options.onTurnComplete!({ identity: turnIdentity, finalText, context }), + } + : {}), + ...(options.traceFlush ? { traceFlush: () => options.traceFlush!(context) } : {}), + }, + }) + + // Exactly one terminal lifecycle hook, after the turn settles. Failure is + // this route's own verdict (`runFailed` from error/failed events, or a + // drain throw), not the engine's envelope. + const fireTerminalLifecycle = async (failed: boolean, drainError: unknown): Promise => { + const lifecycle = options.lifecycle + if (!lifecycle) return + const durationMs = Date.now() - turnStartedAtMs + try { + if (failed) { + await lifecycle.onTurnError?.({ + identity, executionId, turnStreamId, context, durationMs, + error: drainError ?? lastFailureData ?? new Error('chat turn failed'), + }) + } else { + await lifecycle.onTurnComplete?.({ + identity, executionId, turnStreamId, context, durationMs, + finalText: producer?.finalText() ?? '', + usage: producer?.usage?.() ?? {}, + }) + } + } catch (err) { + log('[chat-routes] lifecycle terminal hook failed', { + turnId: turnStreamId, + error: err instanceof Error ? err.message : String(err), + }) + } + } + + // Tee: one branch to the live client, one drained under waitUntil so the + // turn (and its buffering via onEvent) runs to completion after a client + // drop — the engine body executes as it is pulled. + const [clientBody, drainBody] = result.body.tee() + const drained = (async () => { + const reader = drainBody.getReader() + let drainError: unknown + try { + for (;;) { + const { done } = await reader.read() + if (done) break + } + } catch (err) { + drainError = err + log('[chat-routes] turn drain failed', { + turnId: turnStreamId, + error: err instanceof Error ? err.message : String(err), + }) + } + const failed = runFailed || drainError !== undefined + try { + await tap.done(failed ? 'error' : 'complete') + } catch (err) { + log('[chat-routes] turn buffer finalize failed', { + turnId: turnStreamId, + error: err instanceof Error ? err.message : String(err), + }) + } + await fireTerminalLifecycle(failed, drainError) + await releaseLock() + })() + if (ctx?.waitUntil) ctx.waitUntil(drained) + else void drained.catch(() => {}) + + // Announce the replay handle before the engine's first event. + const encoder = new TextEncoder() + const marker = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(`${JSON.stringify(turnMarker)}\n`)) + controller.close() + }, + }) + const body = concatStreams([marker, clientBody]) + + return new Response(body, { + headers: { + 'Content-Type': result.contentType, + 'Cache-Control': 'no-cache', + }, + }) + } catch (err) { + // A throw before the turn began streaming (user-insert, gate, beforeTurn, + // lifecycle-start, tap setup): release the lock, then propagate. + await releaseLock() + throw err + } } async function replay(request: Request, params: { turnId: string }): Promise { diff --git a/tests/chat-routes/turn-routes.test.ts b/tests/chat-routes/turn-routes.test.ts index 07afeab..c098c09 100644 --- a/tests/chat-routes/turn-routes.test.ts +++ b/tests/chat-routes/turn-routes.test.ts @@ -246,6 +246,162 @@ describe('createChatTurnRoutes — buffered replay', () => { }) }) +describe('createChatTurnRoutes — product seams', () => { + /** Read NDJSON lines off a live body until `predicate` is satisfied or EOF. */ + async function drainUntil( + body: ReadableStream, + seen: Array>, + predicate: (seen: Array>) => boolean, + ): Promise { + const reader = body.getReader() + const decoder = new TextDecoder() + while (!predicate(seen)) { + const { value, done } = await reader.read() + if (done) break + for (const line of decoder.decode(value).split('\n').filter((l) => l.trim())) { + seen.push(JSON.parse(line) as Record) + } + } + reader.releaseLock() + } + + it('heartbeat: emits keepalives while the producer is quiet, then stops on the first real event', async () => { + let releaseGate!: () => void + const gate = new Promise((r) => { releaseGate = r }) + const { routes, ctx, pending } = makeRoutes({ + produce: () => ({ + stream: (async function* () { + await gate // silent window — keepalives should fire here + yield { type: 'text', text: 'answer' } + })(), + finalText: () => 'answer', + }), + heartbeat: { intervalMs: 5, event: ({ tick }) => ({ type: 'keepalive', data: { tick } }) }, + }) + + const res = await routes.turn(turnRequest({ threadId: 't-1', content: 'hi' }), ctx) + const seen: Array> = [] + await drainUntil(res.body!, seen, (s) => s.filter((l) => l.type === 'keepalive').length >= 2) + expect(seen.filter((l) => l.type === 'keepalive').length).toBeGreaterThanOrEqual(2) + + releaseGate() + await drainUntil(res.body!, seen, (s) => s.some((l) => l.type === 'text' && l.text === 'answer')) + await Promise.all(pending) + + const answerIndex = seen.findIndex((l) => l.type === 'text' && l.text === 'answer') + expect(answerIndex).toBeGreaterThanOrEqual(0) + // Window resets on the real event and the producer then completes with no + // further silence — no keepalive may follow the answer. + expect(seen.slice(answerIndex).filter((l) => l.type === 'keepalive')).toHaveLength(0) + }) + + it('beforeTurn: observes the assembled input and rewrites the prompt + prior messages', async () => { + const produce = vi.fn((_args: ChatTurnProduceArgs) => fakeProducer([{ type: 'text', text: 'ok' }], 'ok')) + const observed: Array = [] + const rewrittenPrior = [{ id: 'ctx', role: 'user' as const, content: 'injected', parts: null }] + const { routes, ctx, pending } = makeRoutes({ + produce, + beforeTurn: (args) => { + observed.push(args.prompt) + return { prompt: 'rewritten', priorMessages: rewrittenPrior } + }, + }) + + await readLines((await routes.turn(turnRequest({ threadId: 't-1', content: 'original' }), ctx)).body!) + await Promise.all(pending) + + expect(observed).toEqual(['original']) // observed the route-assembled prompt + const args = produce.mock.calls[0]![0] + expect(args.prompt).toBe('rewritten') + expect(args.priorMessages).toEqual(rewrittenPrior) + }) + + it('lifecycle: onTurnStart→onTurnComplete on success, onTurnStart→onTurnError on failure, always ordered', async () => { + const events: string[] = [] + const lifecycle = { + onTurnStart: () => { events.push('start') }, + onTurnComplete: (info: { finalText: string; usage: { inputTokens?: number } }) => + { events.push(`complete:${info.finalText}:${info.usage.inputTokens ?? 0}`) }, + onTurnError: () => { events.push('error') }, + } + + const ok = makeRoutes({ + lifecycle, + produce: () => fakeProducer([{ type: 'text', text: 'yo' }], 'yo', { usage: () => ({ inputTokens: 3 }) }), + }) + await readLines((await ok.routes.turn(turnRequest({ threadId: 't-1', content: 'q' }), ok.ctx)).body!) + await Promise.all(ok.pending) + expect(events).toEqual(['start', 'complete:yo:3']) + + events.length = 0 + const bad = makeRoutes({ + lifecycle, + produce: () => fakeProducer([{ type: 'error', data: { message: 'boom' } }], ''), + }) + await readLines((await bad.routes.turn(turnRequest({ threadId: 't-2', content: 'q' }), bad.ctx)).body!) + await Promise.all(bad.pending) + expect(events).toEqual(['start', 'error']) + }) + + it('contextGate: short-circuits with the product response before the producer runs', async () => { + const produce = vi.fn(() => fakeProducer([{ type: 'text', text: 'should not run' }], 'x')) + const { routes, rows, ctx } = makeRoutes({ + produce, + contextGate: async () => ({ proceed: false, response: Response.json({ needContext: true }, { status: 409 }) }), + }) + + const res = await routes.turn(turnRequest({ threadId: 't-1', content: 'hi' }), ctx) + expect(res.status).toBe(409) + expect(await res.json()).toEqual({ needContext: true }) + expect(produce).not.toHaveBeenCalled() + // The user row is still recorded (a real user turn); no assistant row. + expect(rows.filter((r) => r.role === 'user')).toHaveLength(1) + expect(rows.filter((r) => r.role === 'assistant')).toHaveLength(0) + }) + + it('turnLock: acquires before the turn and releases in finally even when the turn throws', async () => { + const order: string[] = [] + const turnLock = { + acquire: () => { order.push('acquire'); return { acquired: true as const, handle: 'h1' } }, + release: (handle: unknown) => { order.push(`release:${String(handle)}`) }, + } + const { routes, ctx } = makeRoutes({ + turnLock, + beforeTurn: () => { order.push('beforeTurn'); throw new Error('kaboom') }, + }) + + await expect(routes.turn(turnRequest({ threadId: 't-1', content: 'hi' }), ctx)).rejects.toThrow('kaboom') + expect(order).toEqual(['acquire', 'beforeTurn', 'release:h1']) + }) + + it('turnLock: rejects with the product response when already held (no producer run)', async () => { + const produce = vi.fn(() => fakeProducer([{ type: 'text', text: 'x' }], 'x')) + const { routes, rows, ctx } = makeRoutes({ + produce, + turnLock: { + acquire: () => ({ acquired: false as const, response: Response.json({ code: 'in_flight' }, { status: 409 }) }), + release: () => {}, + }, + }) + const res = await routes.turn(turnRequest({ threadId: 't-1', content: 'hi' }), ctx) + expect(res.status).toBe(409) + expect(produce).not.toHaveBeenCalled() + expect(rows).toHaveLength(0) // no user row when the lock is held + }) + + it('onRawEvent: observes producer events before the engine frames them', async () => { + const raw: string[] = [] + const { routes, ctx, pending } = makeRoutes({ + produce: () => fakeProducer([{ type: 'text', text: 'a' }, { type: 'text', text: 'b' }], 'ab'), + onRawEvent: (event) => { raw.push(event.type) }, + }) + await readLines((await routes.turn(turnRequest({ threadId: 't-1', content: 'hi' }), ctx)).body!) + await Promise.all(pending) + // Exactly the producer's own events — no engine lifecycle envelopes. + expect(raw).toEqual(['text', 'text']) + }) +}) + describe('createChatTurnRoutes — interactions composition', () => { function wireQuestion(id: string): InteractionRequestWire { return {