From e0f9eddf616971cc579b876cda3df26ed9ba65f1 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Sat, 23 May 2026 01:30:01 +0300 Subject: [PATCH] =?UTF-8?q?feat(examples):=20chat-handler=20+=20production?= =?UTF-8?q?-trace-sink=20=E2=80=94=20the=20production-pattern=20bundle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two production patterns the prior nine examples did not cover. - examples/chat-handler — DurableChatTurnEngine.runTurn end-to-end: the durable manifest from a chat identity, runDurableTurn under the hood, session.run.* lifecycle, NDJSON ReadableStream the HTTP route forwards verbatim. Shows fresh / replay paths — same identity hits the cached run. - examples/production-trace-sink — createProductionTraceSink wiring: a per-session TraceEmitter writes spans, the sink composes a canonical ProductionRunRecord on endRun, persists via a ProductionRunRecordStore (in-memory for the demo; drizzle/D1/Postgres in production). OTLP forwarding + recordFeedback are documented in the README. Both runnable offline via pnpm tsx. README + examples/README updated. typecheck 0, Node suite 251, workerd suite 2, biome clean, build green. --- README.md | 2 + examples/README.md | 4 + examples/chat-handler/README.md | 29 +++++ examples/chat-handler/chat-handler.ts | 103 ++++++++++++++++++ examples/production-trace-sink/README.md | 28 +++++ .../production-trace-sink.ts | 72 ++++++++++++ 6 files changed, 238 insertions(+) create mode 100644 examples/chat-handler/README.md create mode 100644 examples/chat-handler/chat-handler.ts create mode 100644 examples/production-trace-sink/README.md create mode 100644 examples/production-trace-sink/production-trace-sink.ts diff --git a/README.md b/README.md index 0767f066..45cc0786 100644 --- a/README.md +++ b/README.md @@ -265,6 +265,8 @@ Runnable in [`examples/`](./examples/). Every example imports from - [`model-resolution/`](./examples/model-resolution/) — router catalog + fail-closed admission - [`durable-supervisor/`](./examples/durable-supervisor/) — cross-worker resume keystone - [`agent-into-reviewer/`](./examples/agent-into-reviewer/) — pipe one runtime's stream into a reviewer agent +- [`chat-handler/`](./examples/chat-handler/) — `DurableChatTurnEngine.runTurn` (the centerpiece production pattern) +- [`production-trace-sink/`](./examples/production-trace-sink/) — `createProductionTraceSink` data capture ## Tests diff --git a/examples/README.md b/examples/README.md index 2a02019e..c869bc89 100644 --- a/examples/README.md +++ b/examples/README.md @@ -17,6 +17,8 @@ which needs an `OPENAI_API_KEY`. | [`model-resolution/`](./model-resolution/) | `resolveChatModel` + `validateChatModelId` (fail-closed) + `getModels` | | [`durable-supervisor/`](./durable-supervisor/) | `runSupervisedTurn` — cross-worker resume keystone (fresh / resumed / replayed) | | [`agent-into-reviewer/`](./agent-into-reviewer/) | Pipe one runtime's stream into a reviewer agent (the "2-runtime" pattern) | +| [`chat-handler/`](./chat-handler/) | `DurableChatTurnEngine.runTurn` — the centerpiece production chat handler (fresh / replay paths) | +| [`production-trace-sink/`](./production-trace-sink/) | `createProductionTraceSink` — production data capture (RunRecord + OTLP + feedback) | ## Conventions @@ -45,6 +47,8 @@ pnpm tsx examples/runtime-run/runtime-run.ts pnpm tsx examples/model-resolution/model-resolution.ts pnpm tsx examples/durable-supervisor/durable-supervisor.ts pnpm tsx examples/agent-into-reviewer/agent-into-reviewer.ts +pnpm tsx examples/chat-handler/chat-handler.ts +pnpm tsx examples/production-trace-sink/production-trace-sink.ts # requires creds OPENAI_API_KEY=... pnpm tsx examples/openai-stream-backend/openai-stream-backend.ts diff --git a/examples/chat-handler/README.md b/examples/chat-handler/README.md new file mode 100644 index 00000000..91cf28c1 --- /dev/null +++ b/examples/chat-handler/README.md @@ -0,0 +1,29 @@ +# Durable chat handler + +The centerpiece production pattern every product chat handler +implements. `DurableChatTurnEngine.runTurn` composes the substrate +stack: + +- Builds the durable manifest from the chat identity (`tenantId` / + `sessionId` / `turnIndex`). +- Drives `runDurableTurn` — checkpoint + replay. +- Emits `session.run.*` lifecycle events around the producer stream. +- Returns a `ReadableStream` of NDJSON-encoded `ChatStreamEvent`s — the + shape your HTTP/SSE route forwards verbatim. + +The example shows: + +- A fresh turn streaming events (the dots are `message.part.updated`). +- A second, different turn — a fresh `runId`, the producer runs again. +- A retry of turn 0 — same identity → same `runId` → the **replay path** + emits the cached final text without re-running the producer. + +In production, `produce()` is a thin wrapper over `runAgentTaskStream(...)` +against a real backend (`createOpenAICompatibleBackend` / +`createSandboxPromptBackend`). For the cross-worker-during-turn case — +worker dies *while* streaming — use `runSupervisedTurn` (see +`examples/durable-supervisor/`). + +```bash +pnpm tsx examples/chat-handler/chat-handler.ts +``` diff --git a/examples/chat-handler/chat-handler.ts b/examples/chat-handler/chat-handler.ts new file mode 100644 index 00000000..3f4ecfdb --- /dev/null +++ b/examples/chat-handler/chat-handler.ts @@ -0,0 +1,103 @@ +/** + * Full durable chat handler — the centerpiece production pattern every + * product chat handler implements. + * + * `DurableChatTurnEngine.runTurn` composes the substrate stack: it builds + * the durable manifest from the chat identity, drives `runDurableTurn`, + * emits `session.run.*` lifecycle events around the producer stream, + * checkpoints the final text into the run store, and returns a ready-to- + * pipe `ReadableStream`. A worker crash *after* the turn finishes replays + * the cached final text; a worker crash *during* the turn is what + * `runSupervisedTurn` is for (see `examples/durable-supervisor/`). + * + * In a real product, `produce()` calls `runAgentTaskStream({ task, + * backend, input })` with a real backend (`createOpenAICompatibleBackend` + * / `createSandboxPromptBackend`). Here we yield a small scripted stream + * so the example runs offline with no LLM. + * + * Run with: + * pnpm tsx examples/chat-handler/chat-handler.ts + */ + +import type { ChatStreamEvent, DurableTurnProducer } from '@tangle-network/agent-runtime' +import { durableChatTurnEngine, InMemoryDurableRunStore } from '@tangle-network/agent-runtime' + +const store = new InMemoryDurableRunStore() + +// ── The product's `produce` hook — yields the turn's event stream + a +// finalText() once drained. In production this is a thin wrapper over +// `runAgentTaskStream(...)` against a real backend. ────────────────── +function produce(userMessage: string): DurableTurnProducer { + let accumulated = '' + const reply = userMessage.toLowerCase().includes('missing') + ? 'The 2026 return is missing Schedule B and one W-2. Please upload them.' + : `Acknowledged: "${userMessage.slice(0, 80)}". Drafting a reply.` + + async function* stream(): AsyncGenerator { + yield { type: 'message.started', data: { messageId: 'm-1' } } + for (const chunk of reply.match(/.{1,16}/g) ?? [reply]) { + accumulated += chunk + yield { + type: 'message.part.updated', + data: { messageId: 'm-1', delta: chunk, part: { type: 'text', text: accumulated } }, + } + } + yield { type: 'result', data: { finalText: accumulated } } + } + + return { stream: stream(), finalText: () => accumulated } +} + +async function runTurn(userMessage: string, turnIndex: number): Promise { + const result = durableChatTurnEngine.runTurn({ + store, + identity: { tenantId: 'demo-tenant', sessionId: 'thread-42', turnIndex }, + userMessage, + projectId: 'demo-agent', + domain: 'demo', + hooks: { produce: () => produce(userMessage) }, + }) + + // The engine returns a ReadableStream of NDJSON-encoded ChatStreamEvent + // lines — exactly what an SSE/HTTP route forwards to the client. + const reader = result.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + let final = '' + while (true) { + const { value, done } = await reader.read() + if (done) break + buffer += decoder.decode(value) + const lines = buffer.split('\n') + buffer = lines.pop() ?? '' + for (const line of lines) { + if (!line) continue + const event = JSON.parse(line) as ChatStreamEvent + if (event.type === 'message.part.updated') process.stdout.write('.') + if (event.type === 'result') final = String((event.data ?? {}).finalText ?? '') + if (event.type === 'session.run.started') console.log(`[run started ] turn=${turnIndex}`) + if (event.type === 'session.run.completed') console.log(`\n[run done ] turn=${turnIndex}`) + } + } + return final +} + +async function main() { + // Turn 1: fresh. + const t1 = await runTurn('Where do I start with my 2026 return?', 0) + console.log(`[turn 0 text ] ${t1}\n`) + + // Turn 2: a different turn — fresh manifest, fresh runId. + const t2 = await runTurn('What about the missing Schedule B?', 1) + console.log(`[turn 1 text ] ${t2}\n`) + + // Turn 3: same identity as turn 0 — durable run hits the replay path + // because step 0 of `chat:thread-42:0` is already completed. + const t3 = await runTurn('Where do I start with my 2026 return?', 0) + console.log(`[turn 0 text ] ${t3} (replayed from durable store)`) +} + +main().catch((err) => { + console.error(err) + process.exit(1) +}) diff --git a/examples/production-trace-sink/README.md b/examples/production-trace-sink/README.md new file mode 100644 index 00000000..faa2e3f0 --- /dev/null +++ b/examples/production-trace-sink/README.md @@ -0,0 +1,28 @@ +# Production trace sink + +The data-capture primitive every vertical agent's chat handler wires in +once. Until this existed, eval runs captured everything and production +captured nothing — RL training corpora, the analyst loop, and research +all ran on synthetic personas. + +What `createProductionTraceSink` does: + +- Gives the chat handler a `TraceStore` to write spans to during the + request. +- On `endRun`, composes a canonical `ProductionRunRecord` (`projectId`, + `scenarioId`, `pass`, `score`, `spanCount`, …), persists it via your + `ProductionRunRecordStore` (Drizzle / D1 / Postgres). +- Optionally ships the run as OTLP to Langfuse (`otlp.endpoint`). +- `sink.recordFeedback({ runId, label })` writes the user's thumbs-up / + thumbs-down into a `FeedbackTrajectory` — the corpus DPO/KTO trainers + consume. + +Errors are logged, never thrown — the chat handler is unaffected by a +failing OTLP collector. + +The example uses an in-memory `runRecordStore` so it runs offline. Swap +in a real DB adapter and the wiring is unchanged. + +```bash +pnpm tsx examples/production-trace-sink/production-trace-sink.ts +``` diff --git a/examples/production-trace-sink/production-trace-sink.ts b/examples/production-trace-sink/production-trace-sink.ts new file mode 100644 index 00000000..a6091e42 --- /dev/null +++ b/examples/production-trace-sink/production-trace-sink.ts @@ -0,0 +1,72 @@ +/** + * Production trace sink — the capture primitive every vertical agent's + * chat handler wires in once. + * + * Until this primitive existed, eval runs captured everything and + * production captured *nothing*. The sink closes that gap: it gives the + * chat handler a `TraceStore` to write to during the request, then on + * `endRun` composes a canonical `ProductionRunRecord`, persists it + * (Drizzle / D1 / Postgres — your DB), and ships the run as OTLP to + * Langfuse (optional). Errors are logged, never thrown. + * + * Run with: + * pnpm tsx examples/production-trace-sink/production-trace-sink.ts + */ + +import { TraceEmitter } from '@tangle-network/agent-eval' +import { + createProductionTraceSink, + type ProductionRunRecord, + type ProductionRunRecordStore, +} from '@tangle-network/agent-runtime/agent' + +// ── 1. Your DB adapter. Real wiring: drizzleRunRecordStore(db) / +// D1 / Postgres. Here: an in-memory array. ──────────────────────── +const persisted: ProductionRunRecord[] = [] +const runRecordStore: ProductionRunRecordStore = { + async append(record) { + persisted.push(record) + }, +} + +// ── 2. The sink. Omit `otlp` to skip Langfuse forwarding; omit +// `feedbackStore` to skip user-feedback persistence. The sink still +// composes + persists the RunRecord. ─────────────────────────────── +const sink = createProductionTraceSink({ projectId: 'demo-agent', runRecordStore }) + +// ── 3. One TraceEmitter per chat session — the sink's `onRunComplete` +// hook fires once at `endRun` time. The sink's TraceStore is shared +// across emitters so spans from all sessions accumulate in one +// store for the request lifetime. ─────────────────────────────────── +async function runChatSession(sessionId: string, pass: boolean, score: number) { + const emitter = new TraceEmitter(sink.traceStore, { onRunComplete: [sink.onRunComplete] }) + await emitter.startRun({ + scenarioId: sessionId, + projectId: 'demo-agent', + layer: 'app-runtime', + }) + + // …in production the chat handler emits LLM/tool spans here as the + // turn runs. The sink doesn't care what spans were emitted; it + // counts them and composes the RunRecord on endRun. + + await emitter.endRun({ pass, score }) + // The sink's onRunComplete just fired — RunRecord is in the store. +} + +async function main() { + await runChatSession('session-a', true, 0.92) + await runChatSession('session-b', false, 0.41) + + console.log(`persisted ${persisted.length} RunRecord(s):`) + for (const r of persisted) { + console.log( + ` ${r.runId.slice(0, 8)} ${r.scenarioId} pass=${r.pass} score=${r.score} spans=${r.spanCount}`, + ) + } +} + +main().catch((err) => { + console.error(err) + process.exit(1) +})