[V0_FILE]typescriptreact:file="app/components/demo.tsx" "use client"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { PipelineCodeWorkbench } from "@/components/pipeline-code-workbench"; import type { PipelineEvent } from "@/workflows/pipeline"; const STEP_NAMES = ["Extract", "Transform", "Validate", "Load"] as const; type DemoStatus = "idle" | "running" | "completed"; type PipelineDemoState = { status: DemoStatus; runId: string | null; currentStep: number; stepProgress: number; events: PipelineEvent[]; error: string | null; }; type PipelineLineMap = { workflowLoopLine: number; workflowStepLines: number[]; workflowDoneLine: number; stepStartLine: number; stepProgressLine: number; stepDoneLine: number; }; type PipelineDemoProps = { workflowCode: string; workflowLinesHtml: string[]; stepCode: string; stepLinesHtml: string[]; lineMap: PipelineLineMap; workflowDirective: string; stepDirective: string; }; function createInitialState(): PipelineDemoState { return { status: "idle", runId: null, currentStep: -1, stepProgress: 0, events: [], error: null, }; } function parseSseChunk(chunk: string): PipelineEvent | 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 PipelineEvent; } catch { return null; } } function stepIndexFromName(step: string): number { return STEP_NAMES.indexOf(step as (typeof STEP_NAMES)[number]); } function statusCopy(state: PipelineDemoState): string { if (state.status === "idle") { return "Ready to run the ETL pipeline and stream progress updates."; } if (state.status === "running") { return "Streaming step updates from the pipeline in real time."; } return "Pipeline finished. Every step emitted updates via a writable stream."; } function StepStatusIcon({ completed, active, }: { completed: boolean; active: boolean; }) { if (completed) { return ( ); } if (active) { return ; } return ; } export function PipelineDemo({ workflowCode, workflowLinesHtml, stepCode, stepLinesHtml, lineMap, workflowDirective, stepDirective, }: PipelineDemoProps) { const [state, setState] = useState(() => createInitialState()); const [totalMs, setTotalMs] = useState(null); const [resetKey, setResetKey] = useState(0); const abortRef = useRef(null); const completedSteps = useMemo(() => { const done = new Set(); for (const event of state.events) { if (event.type === "step_done") { done.add(event.index); } } return done; }, [state.events]); const completedCount = completedSteps.size; const activeLines = useMemo(() => { if (state.status === "idle") { return { workflow: [] as number[], step: [] as number[] }; } if (state.status === "completed") { return { workflow: lineMap.workflowDoneLine > 0 ? [lineMap.workflowDoneLine] : [], step: lineMap.stepDoneLine > 0 ? [lineMap.stepDoneLine] : [], }; } const activeStepLine = state.stepProgress >= 100 ? lineMap.stepDoneLine : state.stepProgress > 0 ? lineMap.stepProgressLine : lineMap.stepStartLine; return { workflow: lineMap.workflowLoopLine > 0 ? [lineMap.workflowLoopLine] : [], step: activeStepLine > 0 ? [activeStepLine] : [], }; }, [lineMap, state.status, state.stepProgress]); const activeTones = useMemo(() => { if (state.error) { return { workflow: "failure", step: "failure" } as const; } if (state.status === "completed") { return { workflow: "success", step: "success" } as const; } if (state.status === "running" && state.currentStep < 0) { return { workflow: "waiting", step: "waiting" } as const; } return { workflow: "active", step: "active" } as const; }, [state.currentStep, state.error, state.status]); const gutterMarks = useMemo(() => { const marks: Record = {}; for (const event of state.events) { if (event.type !== "step_done") { continue; } const line = lineMap.workflowStepLines[event.index]; if (line && line > 0) { marks[line] = "success"; } } if (state.error && lineMap.workflowLoopLine > 0) { marks[lineMap.workflowLoopLine] = "fail"; } return marks; }, [lineMap.workflowLoopLine, lineMap.workflowStepLines, state.error, state.events]); const latestMessage = useMemo(() => { for (let i = state.events.length - 1; i >= 0; i -= 1) { const event = state.events[i]; if (event.type === "step_progress") { return event.message; } } return ""; }, [state.events]); const applyEvent = useCallback((event: PipelineEvent) => { setState((prev) => { const next: PipelineDemoState = { ...prev, events: [...prev.events, event], }; switch (event.type) { case "step_start": { next.currentStep = event.index; next.stepProgress = 0; next.status = "running"; break; } case "step_progress": { const idx = stepIndexFromName(event.step); if (idx >= 0) { next.currentStep = idx; } next.stepProgress = event.percent; next.status = "running"; break; } case "step_done": { next.currentStep = event.index; next.stepProgress = 100; next.status = "running"; break; } case "pipeline_done": { next.currentStep = STEP_NAMES.length - 1; next.stepProgress = 100; next.status = "completed"; break; } default: { break; } } return next; }); if (event.type === "pipeline_done") { setTotalMs(event.totalMs); } }, []); const connectToStream = useCallback( async (runId: string, signal: AbortSignal) => { try { const response = await fetch( `/api/mock/stream?runId=${encodeURIComponent(runId)}`, { method: "GET", headers: { Accept: "text/event-stream" }, signal, } ); if (signal.aborted) { return; } if (!response.ok || !response.body) { let message = "Failed to connect to stream"; try { const data = (await response.json()) as { error?: string }; if (data.error) { message = data.error; } } catch { // Ignore invalid JSON errors. } throw new Error(message); } const reader = response.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) { applyEvent(event); } } } if (buffer.trim().length > 0) { const event = parseSseChunk(buffer.replaceAll("\r\n", "\n")); if (event) { applyEvent(event); } } } catch (error) { if (signal.aborted) { return; } const message = error instanceof Error ? error.message : "Stream connection failed"; setState((prev) => ({ ...prev, status: prev.status === "completed" ? "completed" : "idle", error: message, })); } }, [applyEvent] ); const handleStart = useCallback(async () => { abortRef.current?.abort(); const controller = new AbortController(); abortRef.current = controller; setTotalMs(null); setState({ ...createInitialState(), status: "running", }); try { const response = await fetch("/api/mock/pipeline", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ documentId: `doc-${Date.now()}` }), signal: controller.signal, }); if (controller.signal.aborted) { return; } if (!response.ok) { let message = "Failed to start pipeline"; try { const data = (await response.json()) as { error?: string }; if (data.error) { message = data.error; } } catch { // Ignore invalid JSON errors. } throw new Error(message); } const data = (await response.json()) as { runId: string }; if (controller.signal.aborted) { return; } setState((prev) => ({ ...prev, runId: data.runId, })); void connectToStream(data.runId, controller.signal); } catch (error) { if (controller.signal.aborted) { return; } const message = error instanceof Error ? error.message : "Failed to start pipeline"; setState((prev) => ({ ...prev, status: "idle", error: message, })); } }, [connectToStream]); const handleReset = useCallback(() => { abortRef.current?.abort(); abortRef.current = null; setTotalMs(null); setState(createInitialState()); setResetKey((value) => value + 1); }, []); useEffect(() => { return () => { abortRef.current?.abort(); abortRef.current = null; }; }, []); return (

Run ID: {state.runId ?? "—"}

{statusCopy(state)}

{latestMessage ? (

{latestMessage}

) : null} {state.error ? (

{state.error}

) : null} {state.status === "completed" && totalMs !== null ? (

Pipeline Complete in {totalMs}ms

) : null}

{STEP_NAMES.map((stepName, index) => { const completed = completedSteps.has(index); const active = state.status === "running" && state.currentStep === index && !completed; const progress = completed ? 100 : active ? state.stepProgress : 0; return (

{stepName}

{progress}%

); })}

{completedCount}/{STEP_NAMES.length} steps completed

); } [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: "Pipeline — Workflow DevKit Example", description: "Streaming ETL pipeline progress with getWritable() and run.getReadable()", }; export default function RootLayout({ children, }: Readonly<{ children: React.ReactNode; }>) { return ( Skip to main content {children} ); } [V0_FILE]typescript:file="app/api/pipeline/route.ts" import { NextResponse } from "next/server"; import { start } from "workflow/api"; import { pipeline } from "@/workflows/pipeline"; export async function POST(request: Request) { let body: Record; try { body = await request.json(); } catch { return NextResponse.json({ error: "Invalid JSON" }, { status: 400 }); } const documentId = typeof body.documentId === "string" ? body.documentId.trim() : ""; if (!documentId) { return NextResponse.json({ error: "documentId required" }, { status: 400 }); } try { const run = await start(pipeline, [documentId]); return NextResponse.json({ runId: run.runId }); } catch (error) { return NextResponse.json( { error: error instanceof Error ? error.message : "Unknown" }, { status: 500 } ); } } [V0_FILE]typescript:file="app/api/mock/pipeline/route.ts" import { NextResponse } from "next/server"; import { createMockPipelineRun } 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 documentId = typeof body.documentId === "string" ? body.documentId.trim() : ""; if (!documentId) { return NextResponse.json({ error: "documentId required" }, { status: 400 }); } const run = createMockPipelineRun(documentId); return NextResponse.json({ runId: run.runId }); } [V0_FILE]typescript:file="app/api/mock/stream/route.ts" import { NextResponse } from "next/server"; import { getMockPipelineRun, isMockPipelineTerminal, syncMockPipelineRun, type MockPipelineRun, } from "@/app/api/mock/store"; import type { PipelineEvent } from "@/workflows/pipeline"; 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; } function emitPipelineEvent( controller: ReadableStreamDefaultController, encoder: TextEncoder, event: PipelineEvent ): void { controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)); } export async function GET(request: Request) { const runId = parseRunId(request.url); if (!runId) { return NextResponse.json({ error: "runId required" }, { status: 400 }); } const run = getMockPipelineRun(runId); if (!run) { return NextResponse.json({ error: "Run not found" }, { status: 404 }); } syncMockPipelineRun(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 (error) { console.warn("[mock-pipeline] stream_close_failed", { attempted: "controller.close()", reason, runId, failed: error instanceof Error ? error.message : String(error), }); } }; const getSyncedRun = (): MockPipelineRun | null => { const current = getMockPipelineRun(runId); if (!current) { return null; } return syncMockPipelineRun(current); }; const emitPendingEvents = (current: MockPipelineRun): boolean => { const pending = current.events.slice(cursor); for (const event of pending) { try { emitPipelineEvent(controller, encoder, event); } catch (error) { console.warn("[mock-pipeline] stream_enqueue_failed", { attempted: "controller.enqueue()", runId, cursor, eventType: event.type, failed: error instanceof Error ? error.message : String(error), }); 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) { console.warn("[mock-pipeline] stream_run_missing", { attempted: "getMockPipelineRun(runId)", runId, cursor, }); close("run_missing"); return; } if (!emitPendingEvents(current)) { return; } if (isMockPipelineTerminal(current) && cursor >= current.events.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/store.ts" import type { PipelineEvent } from "@/workflows/pipeline"; const PIPELINE_STEPS = ["Extract", "Transform", "Validate", "Load"] as const; const STEP_DURATION_MS = 900; const STEP_PROGRESS_INTERVAL_MS = 150; const STEP_PROGRESS_POINTS = [0, 20, 40, 60, 80, 100] as const; type PipelineStep = (typeof PIPELINE_STEPS)[number]; type ScheduledPipelineEvent = { atMs: number; event: PipelineEvent; }; export type MockPipelineStatus = "running" | "completed"; export type MockPipelineRun = { runId: string; documentId: string; status: MockPipelineStatus; startedAtMs: number; completedAtMs: number | null; currentStepIndex: number; currentStepProgress: number; events: PipelineEvent[]; emittedEventCount: number; schedule: ScheduledPipelineEvent[]; }; declare global { // eslint-disable-next-line no-var var __mockPipelineRuns: Map | undefined; } const mockStoreHost = globalThis as typeof globalThis & { __mockPipelineRuns?: Map; }; function progressMessage(step: PipelineStep, pct: number): string { const messages: Record = { Extract: [ "Connecting to source...", "Reading metadata...", "Parsing fields...", "Extracting content...", "Buffering records...", "Extract complete", ], Transform: [ "Initializing rules...", "Mapping schemas...", "Converting types...", "Normalizing values...", "Applying transforms...", "Transform complete", ], Validate: [ "Loading constraints...", "Checking required fields...", "Validating types...", "Running business rules...", "Final validation...", "Validation complete", ], Load: [ "Connecting to target...", "Preparing batch...", "Writing records...", "Updating indexes...", "Flushing buffers...", "Load complete", ], }; return messages[step][pct / 20] ?? "Processing..."; } function buildSchedule(): ScheduledPipelineEvent[] { const schedule: ScheduledPipelineEvent[] = []; for (let index = 0; index < PIPELINE_STEPS.length; index += 1) { const step = PIPELINE_STEPS[index]; const stepOffsetMs = index * STEP_DURATION_MS; schedule.push({ atMs: stepOffsetMs, event: { type: "step_start", step, index, total: PIPELINE_STEPS.length, }, }); for ( let progressIndex = 0; progressIndex < STEP_PROGRESS_POINTS.length; progressIndex += 1 ) { const pct = STEP_PROGRESS_POINTS[progressIndex]; schedule.push({ atMs: stepOffsetMs + (progressIndex + 1) * STEP_PROGRESS_INTERVAL_MS, event: { type: "step_progress", step, percent: pct, message: progressMessage(step, pct), }, }); } schedule.push({ atMs: stepOffsetMs + STEP_DURATION_MS, event: { type: "step_done", step, index, total: PIPELINE_STEPS.length, durationMs: STEP_DURATION_MS, }, }); } schedule.push({ atMs: PIPELINE_STEPS.length * STEP_DURATION_MS, event: { type: "pipeline_done", totalMs: PIPELINE_STEPS.length * STEP_DURATION_MS, }, }); return schedule; } function applyEventState(run: MockPipelineRun, event: PipelineEvent, nowMs: number): void { switch (event.type) { case "step_start": { run.currentStepIndex = event.index; run.currentStepProgress = 0; return; } case "step_progress": { run.currentStepProgress = event.percent; return; } case "step_done": { run.currentStepIndex = event.index; run.currentStepProgress = 100; return; } case "pipeline_done": { run.status = "completed"; run.currentStepIndex = PIPELINE_STEPS.length - 1; run.currentStepProgress = 100; run.completedAtMs = nowMs; return; } default: { return; } } } function setStatus(run: MockPipelineRun, nextStatus: MockPipelineStatus): void { if (run.status === nextStatus) { return; } console.info("[mock-pipeline] status_transition", { runId: run.runId, from: run.status, to: nextStatus, emittedEventCount: run.emittedEventCount, currentStepIndex: run.currentStepIndex, currentStepProgress: run.currentStepProgress, }); run.status = nextStatus; } export function getMockPipelineStore(): Map { if (!mockStoreHost.__mockPipelineRuns) { mockStoreHost.__mockPipelineRuns = new Map(); } return mockStoreHost.__mockPipelineRuns; } export function createMockPipelineRun(documentId: string, nowMs: number = Date.now()): MockPipelineRun { const runId = `mock_pipeline_${nowMs}_${Math.random().toString(36).slice(2, 8)}`; const run: MockPipelineRun = { runId, documentId, status: "running", startedAtMs: nowMs, completedAtMs: null, currentStepIndex: -1, currentStepProgress: 0, events: [], emittedEventCount: 0, schedule: buildSchedule(), }; getMockPipelineStore().set(runId, run); console.info("[mock-pipeline] run_created", { runId, documentId, startedAtMs: nowMs, totalEvents: run.schedule.length, }); return run; } export function getMockPipelineRun(runId: string): MockPipelineRun | undefined { return getMockPipelineStore().get(runId); } export function syncMockPipelineRun(run: MockPipelineRun, nowMs: number = Date.now()): MockPipelineRun { const elapsedMs = Math.max(0, nowMs - run.startedAtMs); let emittedThisSync = 0; while (run.emittedEventCount < run.schedule.length) { const scheduled = run.schedule[run.emittedEventCount]; if (scheduled.atMs > elapsedMs) { break; } run.events.push(scheduled.event); run.emittedEventCount += 1; emittedThisSync += 1; applyEventState(run, scheduled.event, nowMs); } if (run.emittedEventCount >= run.schedule.length) { setStatus(run, "completed"); if (run.completedAtMs === null) { run.completedAtMs = nowMs; } } else { setStatus(run, "running"); } if (emittedThisSync > 0) { console.info("[mock-pipeline] sync_emitted_events", { runId: run.runId, emittedThisSync, emittedEventCount: run.emittedEventCount, elapsedMs, status: run.status, }); } return run; } export function isMockPipelineTerminal(run: MockPipelineRun): boolean { return run.status === "completed" && run.emittedEventCount >= run.schedule.length; } [V0_FILE]typescript:file="app/api/readable/[runId]/route.ts" import { NextRequest } from "next/server"; import { getRun } from "workflow/api"; export async function GET( _req: NextRequest, { params }: { params: Promise<{ runId: string }> } ) { const { runId } = await params; const run = await getRun(runId); const readable = run.getReadable(); return new Response(readable as unknown as ReadableStream, { headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache, no-transform", Connection: "keep-alive", "X-Accel-Buffering": "no", }, }); } [V0_FILE]typescriptreact:file="app/page.tsx" import { highlightCodeToHtmlLines } from "@/components/code-highlight-server"; import { PipelineDemo } from "./components/demo"; const stepNames = ["Extract", "Transform", "Validate", "Load"] as const; const workflowDirective = `use ${"workflow"}`; const stepDirective = `use ${"step"}`; const workflowCode = `import { getWritable } from "workflow"; export async function pipeline(documentId: string) { "${workflowDirective}"; const startMs = Date.now(); const steps = [ "Extract", "Transform", "Validate", "Load", ]; for (let i = 0; i < steps.length; i += 1) { await runPipelineStep(steps[i], i, steps.length); } const writer = getWritable().getWriter(); await writer.write({ type: "pipeline_done", totalMs: Date.now() - startMs }); writer.releaseLock(); return { status: "completed", steps: steps.length }; }`; const stepCode = `async function runPipelineStep(name: string, index: number, total: number) { "${stepDirective}"; const writer = getWritable().getWriter(); const startMs = Date.now(); try { await writer.write({ type: "step_start", step: name, index, total }); for (let pct = 0; pct <= 100; pct += 20) { await new Promise((resolve) => setTimeout(resolve, 150)); await writer.write({ type: "step_progress", step: name, percent: pct, message: getProgressMessage(name, pct), }); } await writer.write({ type: "step_done", step: name, index, total, durationMs: Date.now() - startMs, }); } finally { writer.releaseLock(); } }`; const workflowLinesHtml = highlightCodeToHtmlLines(workflowCode); const stepLinesHtml = highlightCodeToHtmlLines(stepCode); function findLine(code: string, match: string): number { const lines = code.split("\n"); const index = lines.findIndex((line) => line.includes(match)); return index === -1 ? -1 : index + 1; } const lineMap = { workflowLoopLine: findLine(workflowCode, "await runPipelineStep("), workflowStepLines: stepNames.map((name) => findLine(workflowCode, `\"${name}\"`)), workflowDoneLine: findLine(workflowCode, 'type: "pipeline_done"'), stepStartLine: findLine(stepCode, 'type: "step_start"'), stepProgressLine: findLine(stepCode, 'type: "step_progress"'), stepDoneLine: findLine(stepCode, 'type: "step_done"'), }; export default function Home() { return (

Workflow DevKit Example

Pipeline

This ETL demo streams step-level progress from inside durable steps with{" "} getWritable(){" "} and consumes it client-side through{" "} run.getReadable() . Watch Extract → Transform → Validate → Load update in real time while the code pane highlights the executing lines.

Try It

Start a mock pipeline run, follow each step as it progresses, and inspect how stream events map directly to workflow + step code.

The interactive panel uses a timestamp-based in-memory mock store. The real API routes in this app use{" "} start(pipeline){" "} and{" "} getRun(runId).getReadable() .

); } [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-react-renderer tokens (keep if any demos still use Prism) */ .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/pipeline.ts" import { getWritable } from "workflow"; export type PipelineEvent = | { type: "step_start"; step: string; index: number; total: number } | { type: "step_progress"; step: string; percent: number; message: string } | { type: "step_done"; step: string; index: number; total: number; durationMs: number; } | { type: "pipeline_done"; totalMs: number }; export async function pipeline( documentId: string ): Promise<{ status: "completed"; steps: number }> { "use workflow"; void documentId; const steps = ["Extract", "Transform", "Validate", "Load"]; const startMs = Date.now(); for (let i = 0; i < steps.length; i++) { await runPipelineStep(steps[i], i, steps.length); } const writer = getWritable().getWriter(); await writer.write({ type: "pipeline_done", totalMs: Date.now() - startMs }); writer.releaseLock(); return { status: "completed", steps: steps.length }; } async function runPipelineStep( name: string, index: number, total: number ): Promise { "use step"; const writer = getWritable().getWriter(); const startMs = Date.now(); try { await writer.write({ type: "step_start", step: name, index, total }); // Simulate work with progress updates for (let pct = 0; pct <= 100; pct += 20) { await new Promise((r) => setTimeout(r, 150)); await writer.write({ type: "step_progress", step: name, percent: pct, message: getProgressMessage(name, pct), }); } await writer.write({ type: "step_done", step: name, index, total, durationMs: Date.now() - startMs, }); } finally { writer.releaseLock(); } } function getProgressMessage(step: string, pct: number): string { const messages: Record = { Extract: [ "Connecting to source...", "Reading metadata...", "Parsing fields...", "Extracting content...", "Buffering records...", "Extract complete", ], Transform: [ "Initializing rules...", "Mapping schemas...", "Converting types...", "Normalizing values...", "Applying transforms...", "Transform complete", ], Validate: [ "Loading constraints...", "Checking required fields...", "Validating types...", "Running business rules...", "Final validation...", "Validation complete", ], Load: [ "Connecting to target...", "Preparing batch...", "Writing records...", "Updating indexes...", "Flushing buffers...", "Load complete", ], }; return messages[step]?.[pct / 20] ?? "Processing..."; } [V0_FILE]typescriptreact:file="components/pipeline-code-workbench.tsx" "use client"; import { useEffect, useMemo, useRef, useState } from "react"; type PaneKey = "workflow" | "step"; type MarkStatus = "success" | "fail"; type Tone = "active" | "waiting" | "success" | "failure"; type CopyState = "idle" | "copied" | "failed"; type ToneStyle = { border: string; bg: string; text: string; }; const EMPTY_LINE_LIST: number[] = []; const TONE_STYLES: Record = { active: { border: "border-amber-700", bg: "bg-amber-700/15", text: "text-amber-700", }, waiting: { border: "border-cyan-700", bg: "bg-cyan-700/15", text: "text-cyan-700", }, success: { border: "border-green-700", bg: "bg-green-700/15", text: "text-green-700", }, failure: { border: "border-red-700", bg: "bg-red-700/15", text: "text-red-700", }, }; function GutterGlyph({ mark, visible, }: { mark: MarkStatus; visible: boolean; }) { const color = mark === "success" ? "text-green-700" : "text-red-700"; return ( ); } function CodePane({ linesHtml, filename, directive, code, activeLines, tone, gutterMarks, resetKey, }: { linesHtml: string[]; filename: string; directive: string; code: string; activeLines: number[]; tone: Tone; gutterMarks?: Record; resetKey: number; }) { const [copyState, setCopyState] = useState("idle"); const activeLineSet = useMemo(() => new Set(activeLines), [activeLines]); const prevMarkRef = useRef>({}); useEffect(() => { prevMarkRef.current = {}; }, [resetKey]); const toneStyle = TONE_STYLES[tone]; const handleCopy = async () => { try { await navigator.clipboard.writeText(code); setCopyState("copied"); window.setTimeout(() => setCopyState("idle"), 1400); } catch { setCopyState("failed"); window.setTimeout(() => setCopyState("idle"), 1400); } }; return (

{filename}

{directive}

          
            {linesHtml.map((lineHtml, index) => {
              const lineNumber = index + 1;
              const isActive = activeLineSet.has(lineNumber);
              const liveMark = gutterMarks?.[lineNumber];
              if (liveMark) {
                prevMarkRef.current[lineNumber] = liveMark;
              }
              const stableMark = liveMark ?? prevMarkRef.current[lineNumber];
              const markVisible = liveMark !== undefined;

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

); } export type PipelineCodeWorkbenchProps = { workflowLinesHtml: string[]; stepLinesHtml: string[]; workflowCode: string; stepCode: string; workflowDirective: string; stepDirective: string; workflowFilename?: string; stepFilename?: string; activeLines: Record; activeTones?: Partial>; gutterMarks: Record; resetKey?: number; }; export function PipelineCodeWorkbench({ workflowLinesHtml, stepLinesHtml, workflowCode, stepCode, workflowDirective, stepDirective, workflowFilename = "workflows/pipeline.ts", stepFilename = "workflows/pipeline.ts", activeLines, activeTones, gutterMarks, resetKey = 0, }: PipelineCodeWorkbenchProps) { return (

); } [V0_FILE]typescript:file="components/code-highlight-server.ts" import { highlight } from "sugar-high"; export function highlightCodeToHtmlLines(code: string): string[] { return highlight(code).split("\n"); } [V0_FILE]json:file="package.json" { "name": "pipeline", "private": true, "version": "0.1.0", "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", "workflow": "4.0.1-beta.44", "sugar-high": "^0.9.5" }, "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);