[V0_FILE]typescriptreact:file="app/components/namespaced-streams-code-workbench.tsx" "use client"; import { useEffect, useMemo, useRef, useState } from "react"; type HighlightTone = "amber" | "violet" | "green" | "red" | "cyan"; type CopyState = "idle" | "copied" | "failed"; type HighlightStyle = { border: string; bg: string; text: string; }; const HIGHLIGHT_STYLES: Record = { amber: { border: "border-amber-700", bg: "bg-amber-700/15", text: "text-amber-700", }, violet: { border: "border-violet-700", bg: "bg-violet-700/15", text: "text-violet-700", }, green: { border: "border-green-700", bg: "bg-green-700/15", text: "text-green-700", }, red: { border: "border-red-700", bg: "bg-red-700/15", text: "text-red-700", }, cyan: { border: "border-cyan-700", bg: "bg-cyan-700/15", text: "text-cyan-700", }, }; const CHECKMARK_POINTS = "3,8.5 7,12.5 14,4.5"; type GutterMarkKind = "success" | "fail"; type CodePaneProps = { code: string; htmlLines: string[]; activeLines: number[]; gutterMarks: Record; filename: string; subtitle: string; tone: HighlightTone; }; function CodePane({ code, htmlLines, activeLines, gutterMarks, filename, subtitle, tone, }: CodePaneProps) { const [copyState, setCopyState] = useState("idle"); const activeLineSet = useMemo(() => new Set(activeLines), [activeLines]); const gutterMarkMap = useMemo( () => new Map(Object.entries(gutterMarks).map(([k, v]) => [Number(k), v])), [gutterMarks] ); const prevMarkRef = useRef>(new Map()); const isFullReset = gutterMarkMap.size === 0 && activeLines.length === 0; useEffect(() => { if (isFullReset) { prevMarkRef.current.clear(); return; } gutterMarkMap.forEach((kind, line) => { prevMarkRef.current.set(line, kind); }); }, [gutterMarkMap, isFullReset]); const highlightStyle = HIGHLIGHT_STYLES[tone]; const handleCopy = async () => { try { await navigator.clipboard.writeText(code); setCopyState("copied"); window.setTimeout(() => setCopyState("idle"), 1200); } catch { setCopyState("failed"); window.setTimeout(() => setCopyState("idle"), 1200); } }; return (

{filename}

{subtitle}

          
            {htmlLines.map((lineHtml, index) => {
              const lineNumber = index + 1;
              const isActive = activeLineSet.has(lineNumber);
              const markKind = gutterMarkMap.get(lineNumber);
              const displayKind = isFullReset
                ? undefined
                : markKind ?? prevMarkRef.current.get(lineNumber);

              return (
                
                  
                    
                  
                  
                    {lineNumber}
                  
                   0 ? lineHtml : " ",
                    }}
                  />
                
              );
            })}

); } export type NamespacedStreamsCodeWorkbenchProps = { workflowCode: string; workflowHtmlLines: string[]; workflowActiveLines: number[]; workflowGutterMarks: Record; workflowTone: HighlightTone; stepCode: string; stepHtmlLines: string[]; stepActiveLines: number[]; stepGutterMarks: Record; stepTone: HighlightTone; }; export function NamespacedStreamsCodeWorkbench({ workflowCode, workflowHtmlLines, workflowActiveLines, workflowGutterMarks, workflowTone, stepCode, stepHtmlLines, stepActiveLines, stepGutterMarks, stepTone, }: NamespacedStreamsCodeWorkbenchProps) { return (

); } [V0_FILE]typescriptreact:file="app/components/demo.tsx" "use client"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { NamespacedStreamsCodeWorkbench } from "./namespaced-streams-code-workbench"; type DraftEvent = { type: "chunk"; text: string; idx: number }; type TelemetryEvent = | { type: "start"; runId: string; name: string } | { type: "tokens"; input: number; output: number } | { type: "done"; totalMs: number; totalTokens: number }; type DemoStatus = "idle" | "running" | "completed"; type HighlightTone = "amber" | "violet" | "green" | "red" | "cyan"; type WorkflowLineMap = { draftWriteLines: number[]; telemetryWriteLines: number[]; telemetryStartLine: number[]; telemetryDoneLine: number[]; buildOutlineCall: number[]; writeSectionsCall: number[]; returnLine: number[]; }; type StepLineMap = { buildOutlineBody: number[]; writeSectionsBody: number[]; }; type DemoProps = { workflowCode: string; workflowHtmlLines: string[]; workflowLineMap: WorkflowLineMap; stepCode: string; stepHtmlLines: string[]; stepLineMap: StepLineMap; }; type DemoState = { status: DemoStatus; runId: string | null; draftEvents: DraftEvent[]; telemetryEvents: TelemetryEvent[]; error: string | null; }; function createInitialState(): DemoState { return { status: "idle", runId: null, draftEvents: [], telemetryEvents: [], error: null, }; } function parseSseChunk(chunk: string): T | null { const payload = chunk .split("\n") .map((line) => line.trim()) .filter((line) => line.startsWith("data:")) .map((line) => line.slice(5).trim()) .join("\n"); if (!payload) return null; try { return JSON.parse(payload) as T; } catch { return null; } } function computeTotalTokens(events: TelemetryEvent[]): { totalInput: number; totalOutput: number; } { let totalInput = 0; let totalOutput = 0; for (const e of events) { if (e.type === "tokens") { totalInput += e.input; totalOutput += e.output; } } return { totalInput, totalOutput }; } export function NamespacedStreamsDemo({ workflowCode, workflowHtmlLines, workflowLineMap, stepCode, stepHtmlLines, stepLineMap, }: DemoProps) { const [state, setState] = useState(() => createInitialState()); const abortRef = useRef(null); const draftScrollRef = useRef(null); const telemetryScrollRef = useRef(null); useEffect(() => { return () => { abortRef.current?.abort(); abortRef.current = null; }; }, []); // Auto-scroll draft pane useEffect(() => { const el = draftScrollRef.current; if (el) el.scrollTop = el.scrollHeight; }, [state.draftEvents.length]); // Auto-scroll telemetry pane useEffect(() => { const el = telemetryScrollRef.current; if (el) el.scrollTop = el.scrollHeight; }, [state.telemetryEvents.length]); const applyDraftEvent = useCallback((event: DraftEvent) => { setState((prev) => { const next = { ...prev, draftEvents: [...prev.draftEvents, event] }; return next; }); }, []); const applyTelemetryEvent = useCallback((event: TelemetryEvent) => { setState((prev) => { const next = { ...prev, telemetryEvents: [...prev.telemetryEvents, event], }; if (event.type === "done") { next.status = "completed"; } return next; }); }, []); const connectSse = useCallback( async ( url: string, onEvent: (e: T) => void, signal: AbortSignal ) => { try { const res = await fetch(url, { headers: { Accept: "text/event-stream" }, signal, }); if (signal.aborted) return; if (!res.ok || !res.body) return; const reader = res.body.getReader(); const decoder = new TextDecoder(); let buffer = ""; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const chunks = buffer.replaceAll("\r\n", "\n").split("\n\n"); buffer = chunks.pop() ?? ""; for (const chunk of chunks) { const event = parseSseChunk(chunk); if (event) onEvent(event); } } if (buffer.trim()) { const event = parseSseChunk(buffer.replaceAll("\r\n", "\n")); if (event) onEvent(event); } } catch { // Abort or network error - ignore. } }, [] ); const handleStart = useCallback(async () => { abortRef.current?.abort(); const controller = new AbortController(); abortRef.current = controller; setState({ ...createInitialState(), status: "running" }); try { const res = await fetch("/api/mock/start", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ topic: "Durable Workflows" }), signal: controller.signal, }); if (controller.signal.aborted) return; if (!res.ok) { const data = (await res.json().catch(() => ({}))) as { error?: string; }; throw new Error(data.error ?? "Failed to start"); } const data = (await res.json()) as { runId: string }; if (controller.signal.aborted) return; setState((prev) => ({ ...prev, runId: data.runId })); // Connect to both SSE streams simultaneously void connectSse( `/api/mock/stream/draft?runId=${encodeURIComponent(data.runId)}`, applyDraftEvent, controller.signal ); void connectSse( `/api/mock/stream/telemetry?runId=${encodeURIComponent(data.runId)}`, applyTelemetryEvent, controller.signal ); } catch (error) { if (controller.signal.aborted) return; const message = error instanceof Error ? error.message : "Failed to start"; setState((prev) => ({ ...prev, status: "idle", error: message })); } }, [applyDraftEvent, applyTelemetryEvent, connectSse]); const handleReset = useCallback(() => { abortRef.current?.abort(); abortRef.current = null; setState(createInitialState()); }, []); // Compute the last telemetry event for determining phase const lastTelemetryType = useMemo(() => { if (state.telemetryEvents.length === 0) return null; return state.telemetryEvents[state.telemetryEvents.length - 1].type; }, [state.telemetryEvents]); const lastDraftIdx = useMemo(() => { if (state.draftEvents.length === 0) return -1; return state.draftEvents[state.draftEvents.length - 1].idx; }, [state.draftEvents]); // Determine which phase we are in for code highlighting const phase = useMemo< "idle" | "telemetry_start" | "outline" | "sections" | "telemetry_done" | "completed" >(() => { if (state.status === "idle") return "idle"; if (state.status === "completed") return "completed"; // Running phase determination based on event arrival if (state.draftEvents.length === 0 && state.telemetryEvents.length === 0) { return "telemetry_start"; } if ( state.telemetryEvents.length > 0 && state.draftEvents.length === 0 ) { return "telemetry_start"; } if (lastDraftIdx === 0 && state.draftEvents.length <= 1) { return "outline"; } if (lastTelemetryType === "done") { return "telemetry_done"; } return "sections"; }, [state.status, state.draftEvents.length, state.telemetryEvents.length, lastDraftIdx, lastTelemetryType]); // Code workbench state const codeState = useMemo(() => { const wfMarks: Record = {}; const stepMarks: Record = {}; let workflowActiveLines: number[] = []; let stepActiveLines: number[] = []; let workflowTone: HighlightTone = "amber"; let stepTone: HighlightTone = "amber"; if (phase === "idle") { return { workflowActiveLines: [], stepActiveLines: [], workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, workflowTone: "amber" as HighlightTone, stepTone: "amber" as HighlightTone, }; } if (phase === "telemetry_start") { workflowActiveLines = workflowLineMap.telemetryStartLine; workflowTone = "violet"; stepTone = "violet"; } else if (phase === "outline") { // buildOutline step is running, draft.write happening workflowActiveLines = workflowLineMap.buildOutlineCall; stepActiveLines = stepLineMap.buildOutlineBody; workflowTone = "amber"; stepTone = "green"; // Mark telemetry start line as done for (const ln of workflowLineMap.telemetryStartLine) wfMarks[ln] = "success"; } else if (phase === "sections") { // writeSections step, alternating draft.write and telemetry.write const lastDraftArrived = state.draftEvents.length > 1; const lastTelemetryArrived = state.telemetryEvents.length > 1 && state.telemetryEvents[state.telemetryEvents.length - 1].type === "tokens"; if (lastTelemetryArrived && !lastDraftArrived) { workflowActiveLines = workflowLineMap.telemetryWriteLines; workflowTone = "violet"; } else { workflowActiveLines = workflowLineMap.draftWriteLines; workflowTone = "amber"; } stepActiveLines = stepLineMap.writeSectionsBody; stepTone = "green"; // Mark prior phases as done for (const ln of workflowLineMap.telemetryStartLine) wfMarks[ln] = "success"; for (const ln of workflowLineMap.buildOutlineCall) wfMarks[ln] = "success"; for (const ln of stepLineMap.buildOutlineBody) stepMarks[ln] = "success"; } else if (phase === "telemetry_done" || phase === "completed") { workflowActiveLines = workflowLineMap.returnLine; workflowTone = "green"; stepTone = "green"; // Mark everything as done for (const ln of workflowLineMap.telemetryStartLine) wfMarks[ln] = "success"; for (const ln of workflowLineMap.buildOutlineCall) wfMarks[ln] = "success"; for (const ln of workflowLineMap.draftWriteLines) wfMarks[ln] = "success"; for (const ln of workflowLineMap.telemetryWriteLines) wfMarks[ln] = "success"; for (const ln of workflowLineMap.writeSectionsCall) wfMarks[ln] = "success"; for (const ln of workflowLineMap.telemetryDoneLine) wfMarks[ln] = "success"; for (const ln of workflowLineMap.returnLine) wfMarks[ln] = "success"; for (const ln of stepLineMap.buildOutlineBody) stepMarks[ln] = "success"; for (const ln of stepLineMap.writeSectionsBody) stepMarks[ln] = "success"; } return { workflowActiveLines, stepActiveLines, workflowGutterMarks: wfMarks, stepGutterMarks: stepMarks, workflowTone, stepTone, }; }, [phase, state.draftEvents.length, state.telemetryEvents, workflowLineMap, stepLineMap]); const tokenSummary = useMemo( () => computeTotalTokens(state.telemetryEvents), [state.telemetryEvents] ); const isRunning = state.status === "running"; const isCompleted = state.status === "completed"; return (

{state.error && (

{state.error}

)} {/* Controls */}

{state.status !== "idle" && ( )} Topic: "Durable Workflows"

{state.status === "idle" ? "Ready to generate. Two namespaced streams will update simultaneously." : isRunning ? "Streaming draft content and telemetry from two namespaces..." : "Generation complete. Both streams delivered independently."}

{/* Two-pane stream visualization */}

{/* Draft stream pane */}

0 ? "bg-green-700 animate-pulse" : isCompleted ? "bg-green-700" : "bg-gray-500/70" }`} aria-hidden="true" /> draft namespace

{state.draftEvents.length === 0 ? (

Waiting for draft chunks...

) : (

{state.draftEvents.map((event) => (

chunk {event.idx}

{event.text}

))}

)}

{/* Telemetry stream pane */}

0 ? "bg-violet-700 animate-pulse" : isCompleted ? "bg-violet-700" : "bg-gray-500/70" }`} aria-hidden="true" /> telemetry namespace

{state.telemetryEvents.length === 0 ? (

Waiting for telemetry events...

) : (

{state.telemetryEvents.map((event, i) => ( ))}

)}

{/* Run info bar */} {state.runId && (

Run ID: {state.runId} {isCompleted && ( <> | Tokens: {tokenSummary.totalInput + tokenSummary.totalOutput} | Completed )}

)} {/* Code workbench */}

); } function TelemetryRow({ event }: { event: TelemetryEvent }) { if (event.type === "start") { return (

); } if (event.type === "tokens") { const total = event.input + event.output; const inputPct = total > 0 ? Math.round((event.input / total) * 100) : 50; return (

TOKENS {event.input} in / {event.output} out

); } if (event.type === "done") { return (

DONE {event.totalTokens} tokens in {(event.totalMs / 1000).toFixed(1)}s

); } return null; } [V0_FILE]typescript:file="app/components/code-highlight-server.ts" import { highlight } from "sugar-high"; export function highlightCodeToHtmlLines(code: string): string[] { return highlight(code).split("\n"); } [V0_FILE]typescriptreact:file="app/layout.tsx" import type { Metadata } from "next"; import { Geist, Geist_Mono } from "next/font/google"; import "./globals.css"; const geistSans = Geist({ variable: "--font-geist-sans", subsets: ["latin"], }); const geistMono = Geist_Mono({ variable: "--font-geist-mono", subsets: ["latin"], }); export const metadata: Metadata = { title: "Namespaced Streams - Workflow DevKit Example", description: "Demonstrates multiple named writable streams per workflow run with getWritable({ namespace })", }; export default function RootLayout({ children, }: Readonly<{ children: React.ReactNode; }>) { return ( Skip to main content {children} ); } [V0_FILE]typescript:file="app/api/mock/stream/draft/route.ts" import { NextResponse } from "next/server"; import { getMockStreamRun, syncMockStreamRun, type MockStreamRun, } from "@/app/api/mock/store"; const STREAM_TICK_MS = 75; function parseRunId(url: string): string | null { const { searchParams } = new URL(url); const runId = searchParams.get("runId"); if (!runId) return null; const trimmed = runId.trim(); return trimmed.length > 0 ? trimmed : null; } export async function GET(request: Request) { const runId = parseRunId(request.url); if (!runId) { return NextResponse.json({ error: "runId required" }, { status: 400 }); } const run = getMockStreamRun(runId); if (!run) { return NextResponse.json({ error: "Run not found" }, { status: 404 }); } syncMockStreamRun(run); const encoder = new TextEncoder(); let closeFromCancel: ((reason: string) => void) | null = null; const stream = new ReadableStream({ start(controller) { let closed = false; let cursor = 0; let timer: ReturnType | null = null; let onAbort: (() => void) | null = null; const clearTimer = () => { if (timer) { clearTimeout(timer); timer = null; } }; const close = (reason: string) => { if (closed) return; closed = true; clearTimer(); if (onAbort) { request.signal.removeEventListener("abort", onAbort); onAbort = null; } try { controller.close(); } catch { // Stream already closed. } void reason; }; const getSyncedRun = (): MockStreamRun | null => { const current = getMockStreamRun(runId); if (!current) return null; return syncMockStreamRun(current); }; const emitPending = (current: MockStreamRun): boolean => { const pending = current.draftEvents.slice(cursor); for (const event of pending) { try { controller.enqueue( encoder.encode(`data: ${JSON.stringify(event)}\n\n`) ); } catch { close("enqueue_failed"); return false; } cursor += 1; } return true; }; const pump = () => { if (closed) return; if (request.signal.aborted) { close("request_aborted"); return; } const current = getSyncedRun(); if (!current) { close("run_missing"); return; } if (!emitPending(current)) return; if ( current.status === "completed" && cursor >= current.draftEvents.length ) { close("terminal_state"); return; } timer = setTimeout(pump, STREAM_TICK_MS); }; onAbort = () => close("request_abort_signal"); request.signal.addEventListener("abort", onAbort, { once: true }); closeFromCancel = close; pump(); }, cancel() { closeFromCancel?.("consumer_cancelled"); }, }); return new Response(stream, { headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache, no-transform", Connection: "keep-alive", "X-Accel-Buffering": "no", }, }); } [V0_FILE]typescript:file="app/api/mock/stream/telemetry/route.ts" import { NextResponse } from "next/server"; import { getMockStreamRun, syncMockStreamRun, type MockStreamRun, } from "@/app/api/mock/store"; const STREAM_TICK_MS = 75; function parseRunId(url: string): string | null { const { searchParams } = new URL(url); const runId = searchParams.get("runId"); if (!runId) return null; const trimmed = runId.trim(); return trimmed.length > 0 ? trimmed : null; } export async function GET(request: Request) { const runId = parseRunId(request.url); if (!runId) { return NextResponse.json({ error: "runId required" }, { status: 400 }); } const run = getMockStreamRun(runId); if (!run) { return NextResponse.json({ error: "Run not found" }, { status: 404 }); } syncMockStreamRun(run); const encoder = new TextEncoder(); let closeFromCancel: ((reason: string) => void) | null = null; const stream = new ReadableStream({ start(controller) { let closed = false; let cursor = 0; let timer: ReturnType | null = null; let onAbort: (() => void) | null = null; const clearTimer = () => { if (timer) { clearTimeout(timer); timer = null; } }; const close = (reason: string) => { if (closed) return; closed = true; clearTimer(); if (onAbort) { request.signal.removeEventListener("abort", onAbort); onAbort = null; } try { controller.close(); } catch { // Stream already closed. } void reason; }; const getSyncedRun = (): MockStreamRun | null => { const current = getMockStreamRun(runId); if (!current) return null; return syncMockStreamRun(current); }; const emitPending = (current: MockStreamRun): boolean => { const pending = current.telemetryEvents.slice(cursor); for (const event of pending) { try { controller.enqueue( encoder.encode(`data: ${JSON.stringify(event)}\n\n`) ); } catch { close("enqueue_failed"); return false; } cursor += 1; } return true; }; const pump = () => { if (closed) return; if (request.signal.aborted) { close("request_aborted"); return; } const current = getSyncedRun(); if (!current) { close("run_missing"); return; } if (!emitPending(current)) return; if ( current.status === "completed" && cursor >= current.telemetryEvents.length ) { close("terminal_state"); return; } timer = setTimeout(pump, STREAM_TICK_MS); }; onAbort = () => close("request_abort_signal"); request.signal.addEventListener("abort", onAbort, { once: true }); closeFromCancel = close; pump(); }, cancel() { closeFromCancel?.("consumer_cancelled"); }, }); return new Response(stream, { headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache, no-transform", Connection: "keep-alive", "X-Accel-Buffering": "no", }, }); } [V0_FILE]typescript:file="app/api/mock/start/route.ts" import { NextResponse } from "next/server"; import { createMockStreamRun } from "@/app/api/mock/store"; export async function POST(request: Request) { let body: Record; try { body = await request.json(); } catch { return NextResponse.json({ error: "Invalid JSON" }, { status: 400 }); } const topic = typeof body.topic === "string" ? body.topic.trim() : ""; if (!topic) { return NextResponse.json({ error: "topic required" }, { status: 400 }); } const run = createMockStreamRun(topic); return NextResponse.json({ runId: run.runId }); } [V0_FILE]typescript:file="app/api/mock/status/route.ts" import { NextResponse } from "next/server"; import { getMockStreamRun, syncMockStreamRun } from "@/app/api/mock/store"; export async function GET(request: Request) { const { searchParams } = new URL(request.url); const runId = searchParams.get("runId"); if (!runId) { return NextResponse.json({ error: "runId required" }, { status: 400 }); } const run = getMockStreamRun(runId); if (!run) { return NextResponse.json({ error: "Run not found" }, { status: 404 }); } syncMockStreamRun(run); return NextResponse.json({ runId: run.runId, status: run.status, topic: run.topic, draftCount: run.draftEvents.length, telemetryCount: run.telemetryEvents.length, workflowRunId: run.runId, sectionCount: run.draftEvents.length, }); } [V0_FILE]typescript:file="app/api/mock/store.ts" export type DraftEvent = { type: "chunk"; text: string; idx: number }; export type TelemetryEvent = | { type: "start"; runId: string; name: string } | { type: "tokens"; input: number; output: number } | { type: "done"; totalMs: number; totalTokens: number }; export type MockRunStatus = "running" | "completed"; export type ScheduledEvent = { atMs: number; namespace: "draft" | "telemetry"; draftEvent?: DraftEvent; telemetryEvent?: TelemetryEvent; }; export type MockStreamRun = { runId: string; topic: string; status: MockRunStatus; startedAtMs: number; completedAtMs: number | null; schedule: ScheduledEvent[]; draftEvents: DraftEvent[]; telemetryEvents: TelemetryEvent[]; /** How many schedule items have been processed so far */ emittedCount: number; }; declare global { // eslint-disable-next-line no-var var __mockStreamRuns: Map | undefined; } const mockStoreHost = globalThis as typeof globalThis & { __mockStreamRuns?: Map; }; function getStore(): Map { if (!mockStoreHost.__mockStreamRuns) { mockStoreHost.__mockStreamRuns = new Map(); } return mockStoreHost.__mockStreamRuns; } function buildSchedule(runId: string, topic: string): ScheduledEvent[] { const schedule: ScheduledEvent[] = []; // Telemetry: "start" event at 0ms schedule.push({ atMs: 0, namespace: "telemetry", telemetryEvent: { type: "start", runId, name: "generatePost" }, }); // Draft chunk 0 (outline) at 600ms schedule.push({ atMs: 600, namespace: "draft", draftEvent: { type: "chunk", idx: 0, text: `# ${topic}\n\n## Outline\n1. Introduction\n2. Key Concepts\n3. Implementation\n4. Best Practices`, }, }); // Telemetry: tokens for outline at 700ms schedule.push({ atMs: 700, namespace: "telemetry", telemetryEvent: { type: "tokens", input: 45, output: 120 }, }); // Sections with alternating draft/telemetry const sections = [ `## Introduction\nAn overview of ${topic} and why it matters for modern applications...`, `## Key Concepts\nThe fundamental building blocks: durable execution, deterministic replay, and step boundaries...`, `## Implementation\nHere's how to build it step by step with proper error handling and idempotency...`, `## Best Practices\nTesting strategies, monitoring, and production deployment patterns...`, ]; let cursor = 1300; // Start sections after outline step completes for (let i = 0; i < sections.length; i++) { // Draft chunk schedule.push({ atMs: cursor, namespace: "draft", draftEvent: { type: "chunk", idx: i + 1, text: sections[i] }, }); // Telemetry: tokens follows draft by ~150ms schedule.push({ atMs: cursor + 150, namespace: "telemetry", telemetryEvent: { type: "tokens", input: 80 + i * 10, output: 150 + i * 30 }, }); cursor += 600; } // Telemetry: done at the end const totalMs = cursor + 200; schedule.push({ atMs: totalMs, namespace: "telemetry", telemetryEvent: { type: "done", totalMs, totalTokens: 945 }, }); return schedule; } export function createMockStreamRun(topic: string): MockStreamRun { const runId = `mock_ns_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; const nowMs = Date.now(); const run: MockStreamRun = { runId, topic, status: "running", startedAtMs: nowMs, completedAtMs: null, schedule: buildSchedule(runId, topic), draftEvents: [], telemetryEvents: [], emittedCount: 0, }; getStore().set(runId, run); return run; } export function getMockStreamRun(runId: string): MockStreamRun | undefined { return getStore().get(runId); } export function syncMockStreamRun( run: MockStreamRun, nowMs: number = Date.now() ): MockStreamRun { const elapsedMs = Math.max(0, nowMs - run.startedAtMs); while (run.emittedCount < run.schedule.length) { const scheduled = run.schedule[run.emittedCount]; if (scheduled.atMs > elapsedMs) { break; } if (scheduled.namespace === "draft" && scheduled.draftEvent) { run.draftEvents.push(scheduled.draftEvent); } if (scheduled.namespace === "telemetry" && scheduled.telemetryEvent) { run.telemetryEvents.push(scheduled.telemetryEvent); } run.emittedCount += 1; } // Check if all events have been emitted if (run.emittedCount >= run.schedule.length && run.status !== "completed") { run.status = "completed"; run.completedAtMs = nowMs; } return run; } export function isMockStreamTerminal(run: MockStreamRun): boolean { return ( run.status === "completed" && run.emittedCount >= run.schedule.length ); } [V0_FILE]typescript:file="app/api/start/route.ts" import { NextResponse } from "next/server"; import { start } from "workflow/api"; import { generatePost } from "@/workflows/namespaced-streams"; export async function POST(request: Request) { const { topic } = await request.json(); if (!topic || typeof topic !== "string") { return NextResponse.json({ error: "topic required" }, { status: 400 }); } const run = await start(generatePost, [topic]); const result = await run.returnValue; return NextResponse.json({ runId: run.runId, result }); } [V0_FILE]typescriptreact:file="app/page.tsx" import { highlightCodeToHtmlLines } from "./components/code-highlight-server"; import { NamespacedStreamsDemo } from "./components/demo"; const directiveUseWorkflow = `"use ${"workflow"}"`; const directiveUseStep = `"use ${"step"}"`; const workflowCode = `import { getWritable, getWorkflowMetadata } from "workflow"; export async function generatePost(topic: string) { ${directiveUseWorkflow}; const { workflowRunId } = getWorkflowMetadata(); const draft = getWritable({ namespace: "draft" }).getWriter(); const telemetry = getWritable({ namespace: "telemetry" }).getWriter(); const startedAt = Date.now(); try { await telemetry.write({ type: "start", runId: workflowRunId, name: "generatePost" }); const outline = await buildOutline(topic); await draft.write({ type: "chunk", idx: 0, text: outline }); await telemetry.write({ type: "tokens", input: 45, output: 120 }); const sections = await writeSections(topic, outline); for (let i = 0; i < sections.length; i++) { await draft.write({ type: "chunk", idx: i + 1, text: sections[i] }); await telemetry.write({ type: "tokens", input: 80 + i * 10, output: 150 + i * 30 }); } const totalMs = Date.now() - startedAt; await telemetry.write({ type: "done", totalMs, totalTokens: 945 }); return { status: "completed", workflowRunId, sectionCount: sections.length + 1 }; } finally { draft.releaseLock(); telemetry.releaseLock(); } }`; const stepCode = `async function buildOutline(topic: string) { ${directiveUseStep}; return \`# \${topic}\\n\\n## Outline\\n1. Introduction\\n2. Key Concepts\\n3. Implementation\\n4. Best Practices\`; } async function writeSections(topic: string, outline: string) { ${directiveUseStep}; return [ \`## Introduction\\nAn overview of \${topic}...\`, \`## Key Concepts\\nThe fundamental building blocks...\`, \`## Implementation\\nHere's how to build it...\`, \`## Best Practices\\nTesting strategies...\`, ]; }`; function findLines(code: string, match: string): number[] { const lines = code.split("\n"); const result: number[] = []; for (let i = 0; i < lines.length; i++) { if (lines[i].includes(match)) { result.push(i + 1); } } return result; } function findLine(code: string, match: string): number[] { const lines = code.split("\n"); for (let i = 0; i < lines.length; i++) { if (lines[i].includes(match)) { return [i + 1]; } } return []; } const workflowLineMap = { draftWriteLines: findLines(workflowCode, "await draft.write("), telemetryWriteLines: findLines(workflowCode, "await telemetry.write("), telemetryStartLine: findLine(workflowCode, 'type: "start"'), telemetryDoneLine: findLine(workflowCode, 'type: "done"'), buildOutlineCall: findLine(workflowCode, "await buildOutline("), writeSectionsCall: findLine(workflowCode, "await writeSections("), returnLine: findLine(workflowCode, 'status: "completed"'), }; const stepLineMap = { buildOutlineBody: findLine(stepCode, "return `# ${topic}"), writeSectionsBody: findLine(stepCode, "return ["), }; const workflowHtmlLines = highlightCodeToHtmlLines(workflowCode); const stepHtmlLines = highlightCodeToHtmlLines(stepCode); export default function Home() { return (

Workflow DevKit Example

Namespaced Streams

AI content generation needs two simultaneous outputs: the user-facing draft and internal telemetry. Traditional approach: a websocket multiplex layer with message envelopes and ordering bugs. With Workflow DevKit, call{" "} {"getWritable({ namespace })"}{" "} twice and get two independent, typed streams from one workflow run. Pair with{" "} getWorkflowMetadata(){" "} for run-level context and{" "} run.returnValue{" "} to await the final result from the API route.

Try It

Start a mock content generation run and watch two namespaced streams update simultaneously: a “draft” stream carrying generated text and a “telemetry” stream carrying token counts and timing.

The interactive panel uses a timestamp-based in-memory mock store. The real API routes in this app use{" "} {"getWritable({ namespace: \"draft\" })"}{" "} and{" "} {"getWritable({ namespace: \"telemetry\" })"} .

); } [V0_FILE]css:file="app/globals.css" @import "tailwindcss"; @theme { --color-background-100: #0a0a0a; --color-background-200: #171717; --color-gray-100: #1a1a1a; --color-gray-200: #1f1f1f; --color-gray-300: #292929; --color-gray-400: #737373; --color-gray-500: #525252; --color-gray-600: #404040; --color-gray-700: #262626; --color-gray-800: #171717; --color-gray-900: #a1a1a1; --color-gray-1000: #ededed; --color-blue-700: #0070F3; --color-red-700: #E5484D; --color-amber-700: #FFB224; --color-green-700: #46A758; --color-violet-700: #8B5CF6; --color-cyan-700: #06B6D4; --font-sans: var(--font-geist-sans); --font-mono: var(--font-geist-mono); } body { background: #0a0a0a; color: #ededed; font-family: var(--font-sans), -apple-system, BlinkMacSystemFont, sans-serif; } /* sugar-high syntax tokens */ :root, .dark { --sh-class: #52a8ff; --sh-identifier: #ededed; --sh-sign: #a1a1a1; --sh-property: #ff6166; --sh-entity: #62c073; --sh-jsxliterals: #ededed; --sh-string: #62c073; --sh-keyword: #f75f8f; --sh-comment: #a1a1a1; } /* Prism token colors */ .token.keyword, .token.keyword.control-flow { color: #f75f8f; } .token.plain, .token.parameter, .token.punctuation, .token.plain-text, .token.property-access, .token.operator, .token.tag.punctuation { color: #ededed; } .token.function, .token.function.maybe-class-name { color: #bf7af0; } .token.tag { color: #62c073; } .token.literal-property.property { color: #ff6166; } .token.string, .token.boolean { color: #62c073; } .token.comment { color: #a1a1a1; } .token.class-name { color: #52a8ff; } .token.number { color: #62c073; } .token.attr-name { color: #ff6166; } .token.attr-value { color: #62c073; } .token.builtin { color: #52a8ff; } .token.template-string { color: #62c073; } .token.interpolation { color: #ededed; } [V0_FILE]javascript:file="postcss.config.mjs" export default { plugins: { "@tailwindcss/postcss": {}, }, }; [V0_FILE]typescript:file="workflows/namespaced-streams.ts" import { getWritable, getWorkflowMetadata } from "workflow"; export type DraftEvent = { type: "chunk"; text: string; idx: number }; export type TelemetryEvent = | { type: "start"; runId: string; name: string } | { type: "tokens"; input: number; output: number } | { type: "done"; totalMs: number; totalTokens: number }; export type GenerateResult = { status: "completed"; workflowRunId: string; sectionCount: number; }; export async function generatePost(topic: string): Promise { "use workflow"; const { workflowRunId } = getWorkflowMetadata(); const draft = getWritable({ namespace: "draft" }).getWriter(); const telemetry = getWritable({ namespace: "telemetry" }).getWriter(); const startedAt = Date.now(); try { await telemetry.write({ type: "start", runId: workflowRunId, name: "generatePost" }); const outline = await buildOutline(topic); await draft.write({ type: "chunk", idx: 0, text: outline }); await telemetry.write({ type: "tokens", input: 45, output: 120 }); const sections = await writeSections(topic, outline); for (let i = 0; i < sections.length; i++) { await draft.write({ type: "chunk", idx: i + 1, text: sections[i] }); await telemetry.write({ type: "tokens", input: 80 + i * 10, output: 150 + i * 30 }); } const totalMs = Date.now() - startedAt; await telemetry.write({ type: "done", totalMs, totalTokens: 945 }); return { status: "completed", workflowRunId, sectionCount: sections.length + 1 }; } finally { draft.releaseLock(); telemetry.releaseLock(); } } async function buildOutline(topic: string): Promise { "use step"; return `# ${topic}\n\n## Outline\n1. Introduction\n2. Key Concepts\n3. Implementation\n4. Best Practices`; } async function writeSections(topic: string, outline: string): Promise { "use step"; void outline; return [ `## Introduction\nAn overview of ${topic} and why it matters for modern applications...`, `## Key Concepts\nThe fundamental building blocks: durable execution, deterministic replay, and step boundaries...`, `## Implementation\nHere's how to build it step by step with proper error handling and idempotency...`, `## Best Practices\nTesting strategies, monitoring, and production deployment patterns...`, ]; } [V0_FILE]json:file="package.json" { "name": "namespaced-streams", "version": "0.1.0", "private": true, "scripts": { "dev": "next dev", "build": "next build", "start": "next start", "lint": "next lint" }, "dependencies": { "next": "16.0.10", "react": "19.2.3", "react-dom": "19.2.3", "sugar-high": "^0.9.5", "workflow": "4.1.0-beta.53" }, "devDependencies": { "@tailwindcss/postcss": "^4", "@types/node": "^20", "@types/react": "^19", "@types/react-dom": "^19", "tailwindcss": "^4", "typescript": "^5" } } [V0_FILE]json:file="tsconfig.json" { "compilerOptions": { "target": "ES2017", "lib": ["dom", "dom.iterable", "esnext"], "allowJs": true, "skipLibCheck": true, "strict": true, "noEmit": true, "esModuleInterop": true, "module": "esnext", "moduleResolution": "bundler", "resolveJsonModule": true, "isolatedModules": true, "jsx": "react-jsx", "incremental": true, "plugins": [ { "name": "next" }, { "name": "workflow" } ], "paths": { "@/*": ["./*"] } }, "include": [ "next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts", ".next/dev/types/**/*.ts", "**/*.mts" ], "exclude": ["node_modules"] } [V0_FILE]typescript:file="next.config.ts" import { withWorkflow } from "workflow/next"; import type { NextConfig } from "next"; const nextConfig: NextConfig = {}; export default withWorkflow(nextConfig);