feat(examples): chat-handler + production-trace-sink

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
 

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

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
+```

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<ChatStreamEvent> {
+  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<ChatStreamEvent, void, unknown> {
+    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<string> {
+  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)
+})

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
+```

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)
+})