{workflow.id}

\n {workflow.description &&

{workflow.description}

}\n

\n {run && (\n

\n \n {run.status}\n

\n Run ID: {run.runId.slice(0, 12)}...\n

\n )}\n

\n \n \n \n {\n if (node.type === 'input') return '#C8E6C9'\n if (node.type === 'output') return '#E0E0E0'\n const status = node.data.status\n return status === 'success' ? '#C8E6C9' :\n status === 'failed' ? '#FFCDD2' :\n status === 'running' ? '#BBDEFB' : '#F5F5F5'\n }}\n />\n \n

\n )\n}\n\nexport default WorkflowVisualizer\n",[194,195,101,76],"c554fe78-d1c8-11f0-8d7b-42dde27851f2",["D",1764932068219],".workflow-visualizer {\n width: 100%;\n height: 100%;\n display: flex;\n flex-direction: column;\n background: white;\n}\n\n.visualizer-header {\n padding: 20px 24px;\n border-bottom: 1px solid #e0e0e0;\n background: white;\n display: flex;\n justify-content: space-between;\n align-items: center;\n}\n\n.visualizer-header h1 {\n font-size: 24px;\n font-weight: 600;\n margin-bottom: 4px;\n}\n\n.visualizer-header p {\n font-size: 14px;\n color: #666;\n}\n\n.run-info {\n text-align: right;\n}\n\n.run-status {\n display: flex;\n align-items: center;\n gap: 8px;\n margin-bottom: 6px;\n font-weight: 600;\n text-transform: capitalize;\n}\n\n.status-indicator {\n width: 12px;\n height: 12px;\n border-radius: 50%;\n}\n\n.status-indicator.running {\n background: #2196F3;\n animation: pulse 1.5s ease-in-out infinite;\n}\n\n.status-indicator.success {\n background: #4CAF50;\n}\n\n.status-indicator.failed {\n background: #F44336;\n}\n\n@keyframes pulse {\n 0%, 100% { opacity: 1; }\n 50% { opacity: 0.5; }\n}\n\n.run-meta {\n font-size: 12px;\n color: #666;\n}\n\n.run-meta code {\n background: #f5f5f5;\n padding: 2px 6px;\n border-radius: 3px;\n font-family: monospace;\n}\n\n.react-flow__node {\n font-size: 12px;\n}\n",[194,195,101,88],"c554fc02-d1c8-11f0-8d7b-42dde27851f2",["D",1764932068219],"import { memo } from 'react'\nimport { Handle, Position, NodeProps } from 'reactflow'\nimport './StepNode.css'\n\ninterface StepNodeData {\n label: string\n description?: string\n status: 'pending' | 'running' | 'success' | 'failed'\n output?: unknown\n error?: string\n index: number\n}\n\nfunction StepNode({ data }: NodeProps) {\n const { label, description, status, output, error, index } = data\n\n const getStatusIcon = () => {\n switch (status) {\n case 'success': return '✓'\n case 'failed': return '✗'\n case 'running': return '●'\n default: return '○'\n }\n }\n\n const getStatusColor = () => {\n switch (status) {\n case 'success': return '#4CAF50'\n case 'failed': return '#F44336'\n case 'running': return '#2196F3'\n default: return '#999'\n }\n }\n\n return (\n

\n \n\n

#{index}

\n \n {getStatusIcon()}\n

\n\n

{label}

\n\n {description && (\n

{description}

\n )}\n\n {status !== 'pending' && (\n

\n {status === 'running' && (\n

\n Executing...\n

\n )}\n\n {status === 'success' && output && (\n

Output:

{JSON.stringify(output, null, 2)}

\n )}\n\n {status === 'failed' && error && (\n

Error:

{error}

\n )}\n

\n )}\n\n \n \n )\n}\n\nexport default memo(StepNode)\n",[194,195,101,94],"c554f8d8-d1c8-11f0-8d7b-42dde27851f2",["D",1764932068219],".step-node {\n background: white;\n border: 2px solid #e0e0e0;\n border-radius: 8px;\n padding: 12px;\n min-width: 220px;\n max-width: 320px;\n box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1);\n transition: all 0.2s;\n}\n\n.step-node:hover {\n box-shadow: 0 4px 12px rgba(0, 0, 0, 0.15);\n transform: translateY(-2px);\n}\n\n.step-node.running {\n border-color: #2196F3;\n background: #E3F2FD;\n}\n\n.step-node.success {\n border-color: #4CAF50;\n background: #E8F5E9;\n}\n\n.step-node.failed {\n border-color: #F44336;\n background: #FFEBEE;\n}\n\n.step-header {\n display: flex;\n justify-content: space-between;\n align-items: center;\n margin-bottom: 8px;\n}\n\n.step-index {\n font-size: 11px;\n font-weight: 600;\n color: #999;\n background: #f5f5f5;\n padding: 2px 6px;\n border-radius: 3px;\n}\n\n.step-status-icon {\n font-size: 16px;\n font-weight: bold;\n}\n\n.step-label {\n font-weight: 600;\n font-size: 14px;\n margin-bottom: 4px;\n word-break: break-word;\n}\n\n.step-description {\n font-size: 12px;\n color: #666;\n margin-bottom: 8px;\n line-height: 1.4;\n}\n\n.step-details {\n margin-top: 8px;\n padding-top: 8px;\n border-top: 1px solid #e0e0e0;\n}\n\n.step-running {\n display: flex;\n align-items: center;\n gap: 8px;\n color: #2196F3;\n font-size: 12px;\n font-weight: 500;\n}\n\n.spinner {\n width: 14px;\n height: 14px;\n border: 2px solid #E3F2FD;\n border-top-color: #2196F3;\n border-radius: 50%;\n animation: spin 0.8s linear infinite;\n}\n\n@keyframes spin {\n to { transform: rotate(360deg); }\n}\n\n.step-output,\n.step-error {\n font-size: 11px;\n}\n\n.output-label,\n.error-label {\n font-weight: 600;\n margin-bottom: 4px;\n font-size: 10px;\n text-transform: uppercase;\n letter-spacing: 0.5px;\n}\n\n.output-label {\n color: #4CAF50;\n}\n\n.error-label {\n color: #F44336;\n}\n\n.step-output pre {\n background: #f9f9f9;\n padding: 8px;\n border-radius: 4px;\n overflow-x: auto;\n max-height: 120px;\n overflow-y: auto;\n font-size: 10px;\n line-height: 1.4;\n margin: 0;\n white-space: pre-wrap;\n word-wrap: break-word;\n}\n\n.error-message {\n background: #FFF3E0;\n padding: 8px;\n border-radius: 4px;\n color: #D84315;\n font-size: 11px;\n line-height: 1.4;\n}\n\n.react-flow__handle {\n width: 8px;\n height: 8px;\n background: #2196F3;\n border: 2px solid white;\n}\n",[194,195,101,99],"pulls","branches","forks","envVars",{"_95":26,"_224":26},"connections",{"_95":26,"_223":26},"unfilledCount","emptyCount",{"_33":609,"_13":539,"_74":534,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":610,"_39":26,"_77":102,"_191":-5,"_79":30,"_192":-5,"_282":611,"_17":612},{"_33":602,"_13":603,"_74":604,"_43":534,"_35":36,"_188":40,"_189":24,"_37":605,"_39":26,"_77":78,"_191":606,"_79":30,"_192":-5,"_282":607,"_17":608},{"_33":595,"_13":596,"_74":597,"_43":534,"_35":36,"_188":40,"_189":24,"_37":598,"_39":26,"_77":78,"_191":599,"_79":30,"_192":-5,"_282":600,"_17":601},{"_33":588,"_13":589,"_74":590,"_43":534,"_35":36,"_188":40,"_189":24,"_37":591,"_39":26,"_77":78,"_191":592,"_79":30,"_192":-5,"_282":593,"_17":594},{"_33":581,"_13":582,"_74":583,"_43":534,"_35":36,"_188":40,"_189":24,"_37":584,"_39":26,"_77":78,"_191":585,"_79":30,"_192":-5,"_282":586,"_17":587},{"_33":574,"_13":575,"_74":576,"_43":534,"_35":36,"_188":40,"_189":24,"_37":577,"_39":26,"_77":78,"_191":578,"_79":30,"_192":-5,"_282":579,"_17":580},{"_33":567,"_13":568,"_74":569,"_43":534,"_35":36,"_188":40,"_189":24,"_37":570,"_39":26,"_77":78,"_191":571,"_79":30,"_192":-5,"_282":572,"_17":573},{"_33":560,"_13":561,"_74":562,"_43":534,"_35":36,"_188":40,"_189":24,"_37":563,"_39":26,"_77":78,"_191":564,"_79":30,"_192":-5,"_282":565,"_17":566},{"_33":553,"_13":554,"_74":555,"_43":534,"_35":36,"_188":40,"_189":24,"_37":556,"_39":26,"_77":78,"_191":557,"_79":30,"_192":-5,"_282":558,"_17":559},{"_33":547,"_13":418,"_74":548,"_43":534,"_35":36,"_188":40,"_189":24,"_37":549,"_39":26,"_77":78,"_191":550,"_79":30,"_192":-5,"_282":551,"_17":552},{"_33":540,"_13":541,"_74":542,"_43":534,"_35":36,"_188":40,"_189":24,"_37":543,"_39":26,"_77":78,"_191":544,"_79":30,"_192":-5,"_282":545,"_17":546},{"_33":531,"_13":532,"_74":533,"_43":534,"_35":36,"_188":40,"_189":24,"_37":535,"_39":26,"_77":78,"_191":536,"_79":30,"_192":-5,"_282":537,"_17":538},{"_33":527,"_13":483,"_74":478,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":528,"_39":26,"_77":102,"_191":-5,"_79":30,"_192":-5,"_282":529,"_17":530},{"_33":520,"_13":521,"_74":522,"_43":478,"_35":36,"_188":40,"_189":24,"_37":523,"_39":26,"_77":78,"_191":524,"_79":30,"_192":-5,"_282":525,"_17":526},{"_33":514,"_13":515,"_74":114,"_43":478,"_35":36,"_188":40,"_189":24,"_37":516,"_39":26,"_77":502,"_191":517,"_79":30,"_192":-5,"_282":518,"_17":519},{"_33":506,"_13":507,"_74":508,"_43":478,"_35":36,"_188":40,"_189":24,"_37":509,"_39":26,"_77":510,"_191":511,"_79":30,"_192":-5,"_282":512,"_17":513},{"_33":498,"_13":499,"_74":500,"_43":478,"_35":36,"_188":40,"_189":24,"_37":501,"_39":26,"_77":502,"_191":503,"_79":30,"_192":-5,"_282":504,"_17":505},{"_33":491,"_13":492,"_74":493,"_43":478,"_35":36,"_188":40,"_189":24,"_37":494,"_39":26,"_77":78,"_191":495,"_79":30,"_192":-5,"_282":496,"_17":497},{"_33":484,"_13":485,"_74":486,"_43":478,"_35":36,"_188":40,"_189":24,"_37":487,"_39":26,"_77":78,"_191":488,"_79":30,"_192":-5,"_282":489,"_17":490},{"_33":475,"_13":476,"_74":477,"_43":478,"_35":36,"_188":40,"_189":24,"_37":479,"_39":26,"_77":78,"_191":480,"_79":30,"_192":-5,"_282":481,"_17":482},{"_33":471,"_13":194,"_74":372,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":472,"_39":26,"_77":102,"_191":-5,"_79":30,"_192":-5,"_282":473,"_17":474},{"_33":467,"_13":195,"_74":187,"_43":372,"_35":36,"_188":40,"_189":24,"_37":468,"_39":26,"_77":102,"_191":-5,"_79":30,"_192":-5,"_282":469,"_17":470},{"_33":186,"_13":101,"_74":100,"_43":187,"_35":36,"_188":40,"_189":24,"_37":464,"_39":26,"_77":102,"_191":-5,"_79":30,"_192":-5,"_282":465,"_17":466},{"_33":212,"_13":99,"_74":98,"_43":100,"_35":36,"_188":40,"_189":24,"_37":461,"_39":26,"_77":21,"_191":214,"_79":30,"_192":-5,"_282":462,"_17":463},{"_33":208,"_13":94,"_74":93,"_43":100,"_35":36,"_188":40,"_189":24,"_37":458,"_39":26,"_77":78,"_191":210,"_79":30,"_192":-5,"_282":459,"_17":460},{"_33":204,"_13":88,"_74":87,"_43":100,"_35":36,"_188":40,"_189":24,"_37":455,"_39":26,"_77":21,"_191":206,"_79":30,"_192":-5,"_282":456,"_17":457},{"_33":200,"_13":76,"_74":75,"_43":100,"_35":36,"_188":40,"_189":24,"_37":452,"_39":26,"_77":78,"_191":202,"_79":30,"_192":-5,"_282":453,"_17":454},{"_33":445,"_13":446,"_74":447,"_43":187,"_35":36,"_188":40,"_189":24,"_37":448,"_39":26,"_77":21,"_191":449,"_79":30,"_192":-5,"_282":450,"_17":451},{"_33":438,"_13":439,"_74":440,"_43":187,"_35":36,"_188":40,"_189":24,"_37":441,"_39":26,"_77":78,"_191":442,"_79":30,"_192":-5,"_282":443,"_17":444},{"_33":431,"_13":432,"_74":433,"_43":187,"_35":36,"_188":40,"_189":24,"_37":434,"_39":26,"_77":21,"_191":435,"_79":30,"_192":-5,"_282":436,"_17":437},{"_33":424,"_13":425,"_74":426,"_43":187,"_35":36,"_188":40,"_189":24,"_37":427,"_39":26,"_77":78,"_191":428,"_79":30,"_192":-5,"_282":429,"_17":430},{"_33":417,"_13":418,"_74":419,"_43":187,"_35":36,"_188":40,"_189":24,"_37":420,"_39":26,"_77":78,"_191":421,"_79":30,"_192":-5,"_282":422,"_17":423},{"_33":411,"_13":363,"_74":412,"_43":372,"_35":36,"_188":40,"_189":24,"_37":413,"_39":26,"_77":21,"_191":414,"_79":30,"_192":-5,"_282":415,"_17":416},{"_33":405,"_13":314,"_74":406,"_43":372,"_35":36,"_188":40,"_189":24,"_37":407,"_39":26,"_77":21,"_191":408,"_79":30,"_192":-5,"_282":409,"_17":410},{"_33":398,"_13":399,"_74":400,"_43":372,"_35":36,"_188":40,"_189":24,"_37":401,"_39":26,"_77":21,"_191":402,"_79":30,"_192":-5,"_282":403,"_17":404},{"_33":391,"_13":392,"_74":393,"_43":372,"_35":36,"_188":40,"_189":24,"_37":394,"_39":26,"_77":21,"_191":395,"_79":30,"_192":-5,"_282":396,"_17":397},{"_33":384,"_13":385,"_74":386,"_43":372,"_35":36,"_188":40,"_189":24,"_37":387,"_39":26,"_77":21,"_191":388,"_79":30,"_192":-5,"_282":389,"_17":390},{"_33":377,"_13":378,"_74":379,"_43":372,"_35":36,"_188":40,"_189":24,"_37":380,"_39":26,"_77":21,"_191":381,"_79":30,"_192":-5,"_282":382,"_17":383},{"_33":369,"_13":370,"_74":371,"_43":372,"_35":36,"_188":40,"_189":24,"_37":373,"_39":26,"_77":78,"_191":374,"_79":30,"_192":-5,"_282":375,"_17":376},{"_33":362,"_13":363,"_74":364,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":365,"_39":26,"_77":21,"_191":366,"_79":30,"_192":-5,"_282":367,"_17":368},{"_33":355,"_13":356,"_74":357,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":358,"_39":26,"_77":21,"_191":359,"_79":30,"_192":-5,"_282":360,"_17":361},{"_33":348,"_13":349,"_74":350,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":351,"_39":26,"_77":21,"_191":352,"_79":30,"_192":-5,"_282":353,"_17":354},{"_33":341,"_13":342,"_74":343,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":344,"_39":26,"_77":21,"_191":345,"_79":30,"_192":-5,"_282":346,"_17":347},{"_33":334,"_13":335,"_74":336,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":337,"_39":26,"_77":21,"_191":338,"_79":30,"_192":-5,"_282":339,"_17":340},{"_33":327,"_13":328,"_74":329,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":330,"_39":26,"_77":21,"_191":331,"_79":30,"_192":-5,"_282":332,"_17":333},{"_33":320,"_13":321,"_74":322,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":323,"_39":26,"_77":21,"_191":324,"_79":30,"_192":-5,"_282":325,"_17":326},{"_33":313,"_13":314,"_74":315,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":316,"_39":26,"_77":21,"_191":317,"_79":30,"_192":-5,"_282":318,"_17":319},{"_33":306,"_13":307,"_74":308,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":309,"_39":26,"_77":21,"_191":310,"_79":30,"_192":-5,"_282":311,"_17":312},{"_33":299,"_13":300,"_74":301,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":302,"_39":26,"_77":21,"_191":303,"_79":30,"_192":-5,"_282":304,"_17":305},{"_33":292,"_13":293,"_74":294,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":295,"_39":26,"_77":21,"_191":296,"_79":30,"_192":-5,"_282":297,"_17":298},{"_33":285,"_13":286,"_74":287,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":288,"_39":26,"_77":78,"_191":289,"_79":30,"_192":-5,"_282":290,"_17":291},{"_33":277,"_13":278,"_74":279,"_43":-5,"_35":36,"_188":40,"_189":24,"_37":280,"_39":26,"_77":21,"_191":281,"_79":30,"_192":-5,"_282":283,"_17":284},"c555774a-d1c8-11f0-8d7b-42dde27851f2","workflow-engines-research-report.md","019aee26-23ac-778a-bcd7-30e8897d176c",["D",1764932068219],"# Code-First Workflow Engines: Architectural Patterns and Design Decisions\n\n**Research Report** \n*November 2025*\n\n---\n\n## Executive Summary\n\nThis report analyzes the architectural patterns and design decisions across modern code-first workflow engines. Based on research of major platforms including Temporal, Airflow, Prefect, Dagster, and others, we identify seven core architectural patterns and key trade-offs that define this space.\n\n**Key Finding:** Despite superficial differences, all code-first workflow engines converge on similar foundational patterns—queue-based task distribution, state machine cores, and separation of orchestration from execution—while diverging significantly in their approach to developer experience, durability guarantees, and domain specialization.\n\n---\n\n## Table of Contents\n\n1. [Introduction](#1-introduction)\n2. [Core Architectural Patterns](#2-core-architectural-patterns)\n3. [Shared Foundations](#3-shared-foundations)\n4. [Key Design Trade-offs](#4-key-design-trade-offs)\n5. [Evolution and Trends](#5-evolution-and-trends)\n6. [Recommendations](#6-recommendations)\n7. [References](#7-references)\n\n---\n\n## 1. Introduction\n\n### 1.1 What are Code-First Workflow Engines?\n\nCode-first workflow engines allow developers to define complex, multi-step processes using general-purpose programming languages rather than visual designers or XML configurations. These engines handle orchestration, state management, failure recovery, and coordination of distributed tasks.\n\n### 1.2 Research Scope\n\nThis analysis covers:\n- **Durable Execution Engines**: Temporal, Cadence, Restate\n- **Data Orchestrators**: Airflow, Prefect, Dagster\n- **General-Purpose Engines**: Windmill, n8n, Tork\n- **Platform-Specific**: Azure Durable Functions, AWS Step Functions\n\n### 1.3 Why This Matters\n\nWorkflow engines are critical infrastructure for:\n- Distributed transaction coordination (e.g., payment processing, order fulfillment)\n- Data pipeline orchestration (ETL, ML training)\n- Long-running processes (multi-day workflows, human-in-the-loop)\n- Microservice coordination and saga pattern implementation\n\n---\n\n## 2. Core Architectural Patterns\n\n### 2.1 State Persistence & Durability Models\n\n#### Event Sourcing + State Machines\n\nWorkflow engines fundamentally operate as state machines that store workflow states and track execution history. The underlying implementation varies significantly:\n\n**Durable Execution Approach (Temporal/Cadence)**\n\nDurable execution engines like Temporal automatically preserve the full state of workflows across failures by persisting every step. This enables workflows to be recovered, replayed, or paused at any arbitrary point. The key innovation is that developers can write code that appears synchronous (including sleeping for 30 days) while the engine handles all persistence and recovery automatically.\n\n**Event-Driven Approach (Infinitic, custom engines)**\n\nEvent-driven engines maintain workflow history and trigger tasks based on events, with the workflow engine maintaining state while workers remain stateless. This pattern separates concerns: the engine handles coordination while workers handle execution.\n\n#### Stateless vs. Stateful Engines\n\n**Stateless Engines** (WorkflowEngine .NET, many custom implementations):\n- Store process states in external persistent storage (SQL, NoSQL)\n- Automatically save and load processes from storage\n- Enable horizontal scaling without state synchronization concerns\n- Simpler worker implementation but require reliable storage\n\n**Stateful Engines** (Temporal with history service):\n- Maintain complete execution history in-memory with durable backing\n- Can replay workflows from history to reconstruct state\n- More complex coordination but richer debugging capabilities\n\n---\n\n### 2.2 Task Distribution Architecture\n\n#### The Universal Queue-Based Worker Pattern\n\nEvery practical workflow engine uses queues to dispatch tasks to workers, avoiding direct task calls to handle flow control, availability, and slowness issues. This is perhaps the most universal pattern across all workflow engines.\n\n**Core Components:**\n\n1. **Coordinator/Orchestrator**\n - Manages workflow lifecycle\n - Tracks job state and failures\n - Routes tasks to appropriate workers\n - Handles retries and timeout logic\n\n2. **Workers**\n - Stateless processes that poll task queues\n - Execute tasks and return results\n - Can be scaled independently of coordinator\n - No state synchronization required between workers\n\n3. **Task Queues**\n - Message brokers (RabbitMQ, Kafka, cloud-native queues)\n - Decouple coordination from execution\n - Provide natural backpressure when workers are busy\n - Enable automatic retry when workers crash\n\n4. **Matching Service** (in sophisticated systems like Temporal)\n - Routes workflow and activity tasks to available workers\n - Acts as task queue manager for the system\n - Handles load balancing and partition management\n\n#### Task Queue Partitioning\n\nTask queues are partitioned for horizontal scaling, allowing multiple nodes to independently handle task delivery and avoid bottlenecks. A single task queue might receive tasks from thousands of workflows and be polled by hundreds of workers. Partitioning prevents hotspots and allows linear scaling.\n\n**Temporal's Approach:**\n- 10,000+ shards where each shard stores both workflow state and a queue\n- Enables transactional updates across state and queue\n- Partitions distribute load across multiple matching service nodes\n\n#### Worker Architecture Details\n\nWorkers typically implement:\n- **Pollers**: Internal threads that long-poll the cluster for new tasks\n- **Thread Pools**: Execute activity tasks with configurable concurrency\n- **Slot-Based Concurrency**: Limit simultaneous task execution\n- **Heartbeating**: Keep coordinator informed of task progress\n\n---\n\n### 2.3 Transactional Consistency\n\n#### The Critical Challenge\n\nWorkflow engines must transactionally update state, task queues, and timers together to avoid race conditions where state and queues become inconsistent. Consider these failure scenarios:\n\n1. **State updates succeed, queue write fails**: System believes task is outstanding but task was never queued\n2. **Queue write succeeds, state update is slow**: Task gets delivered and processed before state reflects it was queued\n3. **Timer fires, but state is inconsistent**: Workflow may execute duplicate operations\n\nWithout transactional guarantees, these edge cases proliferate.\n\n#### Solutions\n\n**Transactional Outbox Pattern (Temporal)**\n\nTemporal uses a partitioned architecture where each shard storing workflow state also stores a queue, enabling transactional commits followed by transfer to the queueing subsystem. The outbox pattern ensures that:\n- State changes and queue writes happen atomically within a shard\n- Transfer to external queues can retry safely\n- Deduplication handles any resulting duplicates\n\n**Single Database Approach (simpler systems)**\n\nMany workflow engines use a single database or even a single process to simplify transactional requirements. This works but doesn't scale:\n- Limited throughput (single writer)\n- Limited partition tolerance\n- Database becomes bottleneck\n- Suitable for small to medium deployments\n\n**Saga Pattern for Cross-Service Transactions**\n\nFor workflows spanning multiple services, engines implement saga patterns to maintain eventual consistency without distributed transactions.\n\n---\n\n### 2.4 Workflow Definition Paradigms\n\n#### Code-First vs. Config-First\n\nThe fundamental choice in workflow definition approach shapes the entire developer experience:\n\n**Code-First Approach**\n\nWorkflows defined in general-purpose programming languages (Python, Go, Java, TypeScript) that are executed by the engine.\n\n*Advantages:*\n- Full programming language power (loops, conditionals, functions)\n- Type safety and IDE support\n- Easier unit testing and debugging\n- Version control friendly\n- Can leverage existing libraries and frameworks\n\n*Disadvantages:*\n- Airflow workflows require redeployment when they change since they're written in Python\n- Requires programming knowledge\n- Can't be easily edited by non-technical users\n- Potential for overcomplication\n\n**Examples**: Temporal, Prefect, Dagster, Airflow\n\n**Config-First Approach**\n\nWorkflows written in YAML/JSON don't require engine redeployment when changed.\n\n*Advantages:*\n- Keeping rules as data instead of code means no recompilation and allows runtime changes\n- Non-technical users can understand and edit workflows\n- Clear separation between workflow logic and implementation\n- Easier to generate workflows programmatically\n\n*Disadvantages:*\n- Limited expressiveness (no or limited control flow)\n- Version control and diff tracking more challenging\n- YAML complexity can become unwieldy\n- Limited type safety\n\n**Examples**: AWS Step Functions (JSON), many traditional BPM engines, Kubernetes workflows\n\n#### The Hybrid Revolution\n\nModern systems increasingly adopt hybrid approaches:\n\n**Dagster Components**\n\nDagster Components allow mixing declarative YAML DSLs with imperative Python code for common patterns. Teams can:\n- Use YAML for standardized, repeated patterns (e.g., data extraction)\n- Fall back to Python for complex custom logic\n- Create organization-specific DSLs\n- Abstract away implementation details while maintaining flexibility\n\n**Windmill's Approach**\n\nWindmill lets engineers write scripts in multiple languages (Python, TypeScript, Go) and compose them into workflows via visual DAG editor. This provides:\n- Script development in familiar languages\n- Visual composition of workflows\n- Auto-generated UIs and APIs\n- Best of both worlds: code flexibility with visual orchestration\n\n---\n\n### 2.5 Orchestration vs. Choreography\n\nDistributed workflows can be coordinated through two fundamental patterns:\n\n#### Orchestration Pattern\n\nA central orchestrator tells participants what local transactions to execute. Temporal follows the orchestrator variant of the Saga pattern where a centralized coordinator tracks workflow progress and dispatches commands.\n\n**Characteristics:**\n- Single service/component coordinates entire workflow\n- Explicit workflow definition visible in one place\n- Clear causality and execution order\n- Easier debugging and monitoring\n\n**Advantages:**\n- Centralized visibility and control\n- Single source of truth for workflow state\n- Easier to understand end-to-end flow\n- Better for complex workflows with many decision points\n\n**Disadvantages:**\n- Single point of failure (requires HA setup)\n- Orchestrator can become bottleneck\n- Tight coupling to orchestrator\n- Orchestrator must know about all participants\n\n**Best For:**\n- Complex workflows with many steps\n- Workflows requiring centralized decision logic\n- When observability is critical\n- Financial transactions, order processing\n\n#### Choreography Pattern\n\nEach local transaction publishes domain events that trigger local transactions in other services. Services react to each other's operations without a central conductor, avoiding single point of failure.\n\n**Characteristics:**\n- Decentralized coordination\n- Services respond to events autonomously\n- No central coordinator required\n- Implicit workflow through event chains\n\n**Advantages:**\n- No single point of failure\n- Loose coupling between services\n- Natural event-driven architecture fit\n- Services can evolve independently\n\n**Disadvantages:**\n- Harder to track overall workflow state\n- Implicit dependencies difficult to visualize\n- Debugging spans multiple services\n- Risk of cyclic dependencies\n\n**Best For:**\n- Event-driven architectures\n- Loosely coupled microservices\n- When services should be autonomous\n- Simpler workflows with clear event flows\n\n#### Hybrid Patterns\n\nMany real-world systems combine both:\n- Orchestration for critical paths requiring visibility\n- Choreography for loosely coupled background processes\n- Orchestration within bounded contexts, choreography across them\n\n---\n\n### 2.6 Data Orchestration-Specific Patterns\n\nData orchestration engines have evolved domain-specific patterns:\n\n#### Asset-Centric vs. Task-Centric\n\n**Asset-Centric Model (Dagster)**\n\nDagster treats data assets (tables, models, reports) as first-class citizens with dependencies, lineage, and state management central to orchestration. The focus shifts from \"what tasks to run\" to \"what data should exist.\"\n\n*Key Concepts:*\n- **Software-Defined Assets**: Tables, ML models defined as code\n- **Asset Materialization**: Creating or updating an asset\n- **Asset Lineage**: Automatic dependency tracking\n- **Asset Metadata**: Quality metrics, ownership, documentation\n- **Freshness Policies**: Declarative data SLAs\n\n*Benefits:*\n- Natural fit for data platform thinking\n- Built-in data cataloging\n- Asset-level observability\n- Incremental materialization\n- Clear ownership boundaries\n\n**Task-Centric Model (Airflow, Prefect)**\n\nTraditional DAG-based approach where tasks and their dependencies are primary abstractions. Prefect's task-based model is imperative - developers define exact sequence of steps rather than describing desired end states.\n\n*Key Concepts:*\n- **Tasks**: Individual units of work\n- **DAGs**: Directed acyclic graphs of task dependencies\n- **Operators**: Reusable task templates\n- **XComs**: Cross-task communication\n- **Task Groups**: Logical grouping of related tasks\n\n*Benefits:*\n- Intuitive for ETL thinking\n- Rich operator ecosystem\n- Fine-grained control over execution\n- Familiar to most data engineers\n\n#### Scheduler Architecture Differences\n\n**Airflow:**\n- Centralized scheduler with metadata database\n- Distributed workers (Celery, Kubernetes, LocalExecutor)\n- DAGs parsed continuously from Python files\n- Shared metadata store across components\n\n**Prefect:**\n- Separates orchestration (Cloud/Server) from execution\n- Agents poll for work and execute in any environment\n- Scheduler doesn't need network ingress to workers\n- Hybrid and edge-friendly architecture\n\n**Dagster:**\n- Separates daemon (schedulers/sensors) from user code\n- User code packaged as gRPC servers\n- Loaded dynamically into web UI (Dagit)\n- Isolation of dependency trees reduces scheduler load\n\n---\n\n### 2.7 Failure Handling & Compensation\n\n#### The Saga Pattern for Distributed Transactions\n\nSagas enable orchestrating multi-step workflows across services, with each step providing compensating actions for rollback when failures occur. This is critical in microservices where traditional ACID transactions don't span services.\n\n**Saga Characteristics:**\n\nSagas are coordinating objects implemented as state machines that listen for events and instruct other parts of the system. The key distinction between saga variants:\n\n- **Pure Sagas**: No internal state; decisions based entirely on incoming messages\n- **Process Managers**: Maintain state; act as state machines making decisions based on state and events\n\n**Implementation Pattern:**\n\n1. **Forward Operations**: Each step in workflow performs business operation\n2. **Compensating Actions**: Each step defines how to undo its operation\n3. **Coordination**: Orchestrator or choreography coordinates execution\n4. **Rollback**: On failure, execute compensating actions in reverse\n\n**Example: Order Processing**\n\n```\nNormal Flow:\n1. Reserve Inventory → Compensate: Release Inventory\n2. Charge Payment → Compensate: Refund Payment \n3. Ship Order → Compensate: Cancel Shipment\n\nFailure Scenario:\n1. Reserve Inventory ✓\n2. Charge Payment ✓\n3. Ship Order ✗ (fails)\n→ Execute: Refund Payment, Release Inventory\n```\n\n#### Built-In Retry & Durability\n\nModern workflow engines provide sophisticated retry mechanisms:\n\n**Temporal Activities:**\n- Built-in support for timeouts and retries\n- Configurable retry policies (exponential backoff, max attempts)\n- Automatic recovery with resume from last checkpoint\n- Activities can run for days with automatic heartbeating\n\n**Benefits of Durable Execution:**\n- Simpler codebase with fewer bugs\n- Comprehensive test coverage through replay capabilities\n- Can simulate failures and test compensation logic\n- Time travel debugging\n\n---\n\n### 2.8 Scaling Dimensions\n\n#### Horizontal Scalability Strategies\n\nModern workflow engines must scale along multiple dimensions:\n\n**1. Sharding for Parallelism**\n\nSharding enables executing millions of tasks by distributing workflow state across many partitions. Temporal uses 10,000+ shards where each shard stores both workflow state and a queue for transactional updates.\n\n**2. Worker Pool Scaling**\n\nWorkers scale independently:\n- Add workers to increase task throughput\n- Remove workers to reduce costs\n- No state coordination needed\n- Auto-scaling based on queue depth\n\n**3. Task Queue Separation**\n\nMultiple task queues for different workloads allow isolation of CPU-intensive vs. I/O-bound tasks with independent autoscaling. For example:\n- GPU-intensive ML training on dedicated workers\n- I/O-bound API calls on different workers\n- Prevents resource contention\n- Independent scaling policies per workload type\n\n**4. Database Scaling**\n\nDifferent approaches to scaling the persistence layer:\n- **Sharding**: Partition state across multiple databases\n- **Replication**: Read replicas for query workloads\n- **Caching**: In-memory caches for hot workflows\n- **Time-series optimization**: Archival of old execution data\n\n#### Worker Architecture for Scale\n\nWorkers implement sophisticated patterns:\n\n**Pollers**\n- Internal threads that long-poll the cluster for new tasks\n- Configurable poller count controls task pickup speed\n- Long-polling reduces network overhead vs. short polling\n- Pollers don't execute tasks, just retrieve them\n\n**Execution Threads**\n- Separate thread pool for actual task execution\n- Configurable concurrency limits per worker\n- Slot-based concurrency prevents overload\n- Graceful degradation under pressure\n\n**Heartbeating**\n- Workers periodically report task progress\n- Coordinator can detect worker failures\n- Enables task timeout and reassignment\n- Critical for long-running tasks\n\n---\n\n## 3. Shared Foundations\n\nDespite significant architectural differences, code-first workflow engines converge on several fundamental patterns:\n\n### 3.1 Queue-Based Task Distribution\n\n**Universal adoption** of message queues for task dispatch:\n- Decouples coordination from execution\n- Provides natural backpressure\n- Enables retry and failure recovery\n- Facilitates horizontal scaling\n\nNo successful workflow engine attempts direct synchronous task invocation at scale.\n\n### 3.2 State Machine Core\n\nAll workflow engines model execution as state transitions:\n- Workflows define valid states and transitions\n- Events trigger state changes\n- State changes may trigger actions (task dispatch)\n- Current state determines valid next actions\n\nEven engines with imperative code interfaces (Temporal) internally maintain state machines.\n\n### 3.3 Retry and Timeout Mechanisms\n\nBuilt-in fault tolerance is non-negotiable:\n- Configurable retry policies\n- Exponential backoff\n- Max retry limits\n- Timeout detection\n- Automatic task reassignment on failure\n\nThese shouldn't be the application's responsibility.\n\n### 3.4 Event-Driven Triggers\n\nMoving beyond pure scheduling to reactive execution:\n- Scheduled triggers (cron-like)\n- Event triggers (message arrives, file uploaded)\n- API triggers (webhook, HTTP request)\n- Manual triggers (human intervention)\n- Sensor triggers (external system state change)\n\nModern workflows react to the world, not just the clock.\n\n### 3.5 Observability Focus\n\nRich UIs for monitoring, debugging, and understanding execution state:\n- Real-time workflow status\n- Execution history and audit trails\n- Performance metrics and bottleneck detection\n- Error tracking and alerting\n- Dependency visualization\n\nDurable execution ensures every step of code execution can be tracked and viewed, enabling quick diagnosis of issues.\n\n### 3.6 Separation of Concerns\n\nCoordination logic separate from business logic:\n- Orchestration engine handles: scheduling, retries, state, queuing\n- Application code handles: business logic, data transformation\n- Clean interfaces between layers\n- Business logic doesn't deal with infrastructure concerns\n\nThis separation is critical for testability and maintainability.\n\n---\n\n## 4. Key Design Trade-offs\n\n### 4.1 Developer Experience vs. Operational Complexity\n\n**The Trade-off:**\n\nAirflow has significant operational overhead (scheduler, workers, metadata DB, message queue) but provides a vast ecosystem of operators and integrations. Prefect emphasizes simplicity with a \"negative engineering\" philosophy—assuming developers know how to code and removing unnecessary abstractions. Dagster takes a first-principles approach with the full development lifecycle in mind, from development to deployment to monitoring.\n\n**Implications:**\n\n- **Airflow**: Best for teams with DevOps resources, need for battle-tested stability\n- **Prefect**: Best for agile teams wanting rapid iteration\n- **Dagster**: Best for data platform teams wanting end-to-end lifecycle management\n\nEarly workflow systems like AWS Simple Workflow Service (SWF) struggled with adoption despite being conceptually advanced due to challenging developer experience. The lesson: power without usability leads to low adoption.\n\n### 4.2 Flexibility vs. Guardrails\n\n**The Spectrum:**\n\n- **Full Code Flexibility** (Temporal): Any code in workflow, full language power\n - Pro: Can express anything\n - Con: Can shoot yourself in foot, hard to enforce standards\n\n- **Constrained Frameworks** (Airflow with TaskFlow API): Structured patterns with some flexibility\n - Pro: Best practices built-in\n - Con: Sometimes need to work around framework\n\n- **Declarative DSLs** (YAML workflows): Limited to defined constructs\n - Pro: Prevents entire classes of errors\n - Con: May not express all needs\n\nModern platforms aim to provide flexibility with visibility and guardrails enterprises need to move fast without breaking things.\n\n### 4.3 Monolithic vs. Distributed\n\n**Historical Context:**\n\nTraditional workflow engines were often designed to run as single process and not distributed, limiting scalability. This was acceptable when workflows were departmental tools, not production infrastructure.\n\n**Modern Requirements:**\n\n- Multi-region deployment\n- Fault tolerance across availability zones\n- Elastic scaling\n- Multi-tenancy\n\nModern engines embrace distributed architecture from the start, but this adds complexity.\n\n### 4.4 Real-Time vs. Batch\n\n**Historical Focus:**\n\nAirflow historically focused on scheduled batch jobs (nightly ETL, daily reports). This made sense when data was batch-oriented.\n\n**Modern Reality:**\n\nPrefect and Dagster offer superior event-native design for real-time pipelines. Airflow 3.0 (April 2025) improved event-driven capabilities but still carries batch-oriented legacy.\n\n**Trade-off:**\n\n- Batch-oriented systems optimize for throughput and resource efficiency\n- Real-time systems optimize for latency and responsiveness\n- Hard to excel at both simultaneously\n\n### 4.5 Generalization vs. Specialization\n\n**General-Purpose Engines** (Temporal, Prefect):\n- Can orchestrate anything\n- No domain assumptions\n- Broadly applicable\n\n**Domain-Specific Engines** (Dagster for data):\n- Optimized for specific use cases\n- Domain-specific features (asset lineage, data quality)\n- Steeper learning curve for other domains\n\nThe trend is toward general engines with domain-specific extensions.\n\n---\n\n## 5. Evolution and Trends\n\n### 5.1 Historical Evolution\n\n**Phase 1: Traditional BPM (2000s)**\n- BPMN/BPEL based\n- XML configuration\n- Heavy enterprise focus\n- Visual designers\n- Examples: jBPM, Activiti, Camunda\n\n**Phase 2: Code-First Data (2010s)**\n- Python-based DAGs\n- Data engineering focus\n- Scheduled batch jobs\n- Example: Airflow (2014)\n\n**Phase 3: Modern Durable Execution (2015-2020)**\n- Workflow-as-code\n- Built-in resilience\n- Language-native SDKs\n- Examples: Cadence (2016), Temporal (2019)\n\n**Phase 4: Asset-Centric & Declarative (2020+)**\n- Data as first-class citizen\n- Hybrid code/config approaches\n- Developer experience focus\n- Examples: Dagster Components (2025), modern Prefect\n\n### 5.2 Current Trends\n\n**1. Convergence on Event-Driven**\n\nAll engines moving toward event-native architectures:\n- Airflow 3.0 added event-driven scheduling\n- Prefect built event-driven from start\n- Temporal supports signals and queries for external events\n\n**2. AI/ML Workload Support**\n\nSpecialized features for AI/ML:\n- Long-running training jobs\n- GPU resource management\n- Model versioning and lineage\n- Experiment tracking integration\n\n**3. Hybrid Cloud & Edge**\n\nWorkflows must run anywhere:\n- Multi-cloud support\n- Edge deployment (Prefect's agent model)\n- Air-gapped environments\n- Kubernetes-native\n\n**4. Developer Experience Revolution**\n\nFocus on DX improvements:\n- Local development modes\n- Fast feedback loops\n- Testing frameworks\n- Time-travel debugging\n- Branch deployments (Dagster)\n\n**5. Platform Consolidation**\n\nEngines absorbing adjacent concerns:\n- Data cataloging (Dagster)\n- Data quality checks\n- Cost management\n- Observability\n- Reducing tool sprawl\n\n### 5.3 Future Directions\n\n**Predicted Evolutions:**\n\n1. **AI-Assisted Workflow Development**\n - LLM-generated workflows from natural language\n - Automatic optimization suggestions\n - Intelligent error recovery\n\n2. **Multi-Model Orchestration**\n - Seamless mix of code, config, visual\n - DSLs for common patterns\n - Full code for edge cases\n\n3. **Deeper Cloud Integration**\n - Native serverless execution\n - Zero-operations deployment\n - Automatic scaling without config\n\n4. **Cross-Organization Workflows**\n - Workflow federation\n - Inter-company orchestration\n - Privacy-preserving coordination\n\n---\n\n## 6. Recommendations\n\n### 6.1 Choosing a Workflow Engine\n\n**For Data/Analytics Teams:**\n\n- **Start with Dagster** if:\n - Building a modern data platform\n - Need strong data lineage and cataloging\n - Have strong Python skills\n - Want asset-centric thinking\n\n- **Choose Airflow** if:\n - Already invested in Airflow ecosystem\n - Need 2000+ community operators\n - Have DevOps resources for operation\n - Batch-oriented workflows\n\n- **Consider Prefect** if:\n - Want rapid development\n - Need hybrid/edge deployment\n - Prefer minimal operational overhead\n - Dynamic, event-driven workflows\n\n**For Distributed Systems/Microservices:**\n\n- **Choose Temporal** if:\n - Need rock-solid reliability guarantees\n - Implementing saga patterns\n - Long-running processes (days/weeks)\n - Multi-step transactions across services\n - Critical business processes (payments, orders)\n\n**For General Automation:**\n\n- **Choose Windmill** if:\n - Need multi-language support\n - Want low-code components with code flexibility\n - Building internal tools\n - Small to medium team\n\n- **Choose n8n** if:\n - Need extensive API integrations\n - Low-code with code fallback\n - Business process automation\n - Non-technical users involved\n\n### 6.2 Architectural Guidelines\n\n**When Building Custom Workflows:**\n\n1. **Always use queues for task distribution**\n - Don't do synchronous RPC between coordinator and workers\n - Queues provide natural backpressure and retry\n\n2. **Ensure transactional consistency**\n - Use outbox pattern or single-database approach\n - Never update state and queues without transactions\n - Edge cases will bite you\n\n3. **Make workers stateless**\n - All state in coordinator/database\n - Enables trivial horizontal scaling\n - Simplifies failure recovery\n\n4. **Implement comprehensive observability**\n - Log all state transitions\n - Provide execution history\n - Enable time-travel debugging\n - Visualize dependencies\n\n5. **Support local development**\n - Workflows should run on developer laptop\n - Fast feedback loops\n - Testability is critical\n\n6. **Design for failure**\n - Assume anything can fail anytime\n - Implement compensating actions\n - Make operations idempotent\n - Test failure scenarios\n\n### 6.3 Migration Strategies\n\n**From Homegrown to Managed Engine:**\n\n1. **Start with new workflows** - Don't migrate everything\n2. **Run in parallel** - Prove stability before switching\n3. **Choose based on team skills** - Don't fight your team's strengths\n4. **Consider operational burden** - Factor in ongoing costs\n5. **Evaluate vendor lock-in** - Prefer open-source or standard APIs\n\n**Incremental Migration Path:**\n\nMany engines (especially Dagster) support observing existing systems:\n- Wrap existing Airflow DAGs\n- Gradually migrate workflows\n- Maintain visibility across both systems\n- Reduce risk of big-bang migration\n\n---\n\n## 7. References\n\n### Primary Sources\n\n**Temporal/Durable Execution:**\n- Temporal Engineering Blog: \"Designing a Workflow Engine from First Principles\"\n- Temporal Documentation: \"How Temporal Works\"\n- \"Durable Execution in Distributed Systems\" by Temporal\n\n**Data Orchestration:**\n- \"Orchestration Showdown: Dagster vs Prefect vs Airflow\" (ZenML, November 2025)\n- \"Dagster vs Airflow: Comparing Top Data Orchestration Tools\" (DataCamp, 2024)\n- \"Airflow vs Prefect vs Dagster\" (Galaxy, 2025)\n\n**Patterns & Architecture:**\n- \"Saga and Process Manager\" (Event-Driven.io)\n- Microservices.io: \"Pattern: Saga\"\n- \"Events, Sagas and Workflows: Managing Long-Running Processes\"\n\n**Community Resources:**\n- awesome-workflow-engines (GitHub)\n- SE Radio: \"Maxim Fateev on Durable Execution with Temporal\"\n- Various Stack Overflow discussions on workflow patterns\n\n### Key Papers & Talks\n\n- Hector Garcia-Molina & Kenneth Salem (1987): Original Saga pattern paper\n- Systems @ Scale 2021: Workflow engine design principles\n- Various conference talks on Temporal, Airflow, Dagster evolution\n\n---\n\n## Appendix A: Comparison Matrix\n\n| Feature | Temporal | Airflow | Prefect | Dagster |\n|---------|----------|---------|---------|---------|\n| **Primary Use Case** | Distributed transactions | Data pipelines | General workflows | Data platforms |\n| **Definition Style** | Code-first (Python, Go, Java, TS) | Code-first (Python DAGs) | Code-first (Python) | Asset-centric (Python) |\n| **State Model** | Durable execution, full replay | Metadata DB, task state | Event-driven, state streams | Asset materialization state |\n| **Scheduling** | Event + cron | Primarily cron | Event-native | Asset-aware, event-driven |\n| **Deployment** | Self-hosted or cloud | Self-hosted (complex) | Cloud or self-hosted | Self-hosted or cloud |\n| **Scalability** | 10,000+ shards | Horizontal via executors | Agent-based elastic | gRPC user code isolation |\n| **Best For** | Mission-critical transactions | Established ETL pipelines | Rapid development | Modern data platforms |\n| **Learning Curve** | Moderate | Steep | Gentle | Moderate |\n| **Operational Overhead** | Moderate | High | Low | Moderate |\n| **Community Size** | Growing | Very large | Growing | Growing rapidly |\n\n---\n\n## Appendix B: Glossary\n\n**Activity**: A single unit of work in a workflow, typically a function or service call.\n\n**Asset**: In data orchestration, a data object like a table or model that has dependencies and lineage.\n\n**Choreography**: Coordination pattern where services react to events without central orchestrator.\n\n**Compensating Action**: Undo operation for a completed workflow step, used in saga rollback.\n\n**DAG (Directed Acyclic Graph)**: Structure representing workflow steps and their dependencies.\n\n**Durable Execution**: Programming model where code execution state persists automatically across failures.\n\n**Orchestration**: Coordination pattern where central service directs other services.\n\n**Saga**: Long-running transaction pattern with compensating actions for rollback.\n\n**Shard**: Partition of workflow state for horizontal scaling.\n\n**Task Queue**: Message queue holding work items for workers to process.\n\n**Worker**: Stateless process that executes workflow tasks.\n\n**Workflow**: Multi-step process with defined coordination and error handling.\n\n---\n\n## Appendix C: Code Examples\n\n### Example 1: Temporal Workflow (Python)\n\n```python\nfrom datetime import timedelta\nfrom temporalio import workflow, activity\n\n@activity.defn\nasync def send_email(recipient: str) -> None:\n # Send email logic\n print(f\"Email sent to {recipient}\")\n\n@workflow.defn\nclass SubscriptionWorkflow:\n @workflow.run\n async def run(self, user_id: str) -> str:\n # Day 1: Welcome email\n await workflow.execute_activity(\n send_email,\n user_id,\n start_to_close_timeout=timedelta(minutes=5)\n )\n \n # Wait 7 days\n await workflow.sleep(timedelta(days=7))\n \n # Day 8: Follow-up email\n await workflow.execute_activity(\n send_email,\n user_id,\n start_to_close_timeout=timedelta(minutes=5)\n )\n \n return f\"Campaign complete for {user_id}\"\n```\n\n### Example 2: Airflow DAG (Python)\n\n```python\nfrom airflow import DAG\nfrom airflow.operators.python import PythonOperator\nfrom datetime import datetime, timedelta\n\ndef extract_data():\n # Extract logic\n return \"data\"\n\ndef transform_data(ti):\n data = ti.xcom_pull(task_ids='extract')\n # Transform logic\n return f\"transformed_{data}\"\n\ndef load_data(ti):\n data = ti.xcom_pull(task_ids='transform')\n # Load logic\n print(f\"Loading {data}\")\n\nwith DAG(\n 'etl_pipeline',\n schedule_interval=timedelta(days=1),\n start_date=datetime(2025, 1, 1),\n catchup=False\n) as dag:\n \n extract = PythonOperator(\n task_id='extract',\n python_callable=extract_data\n )\n \n transform = PythonOperator(\n task_id='transform',\n python_callable=transform_data\n )\n \n load = PythonOperator(\n task_id='load',\n python_callable=load_data\n )\n \n extract >> transform >> load\n```\n\n### Example 3: Dagster Assets (Python)\n\n```python\nfrom dagster import asset, AssetExecutionContext\n\n@asset\ndef raw_customers(context: AssetExecutionContext):\n \"\"\"Extract raw customer data\"\"\"\n # Fetch from API\n return fetch_customers()\n\n@asset(deps=[raw_customers])\ndef cleaned_customers(context: AssetExecutionContext):\n \"\"\"Clean and validate customer data\"\"\"\n # Load raw data\n raw = load_asset(\"raw_customers\")\n # Clean data\n return clean_data(raw)\n\n@asset(deps=[cleaned_customers])\ndef customer_metrics(context: AssetExecutionContext):\n \"\"\"Calculate customer metrics\"\"\"\n cleaned = load_asset(\"cleaned_customers\")\n return calculate_metrics(cleaned)\n```\n\n### Example 4: Prefect Flow (Python)\n\n```python\nfrom prefect import flow, task\nfrom datetime import timedelta\n\n@task(retries=3, retry_delay_seconds=60)\ndef fetch_data(url: str):\n # Fetch logic with automatic retries\n return fetch_from_api(url)\n\n@task\ndef process_data(data):\n # Processing logic\n return processed_data\n\n@flow(log_prints=True)\ndef etl_flow(source_url: str):\n raw_data = fetch_data(source_url)\n processed = process_data(raw_data)\n return processed\n\n# Run flow\nif __name__ == \"__main__\":\n etl_flow(\"https://api.example.com/data\")\n```\n\n---\n\n## Document Information\n\n**Report Version**: 1.0 \n**Publication Date**: November 2025 \n**Author**: Research Analysis \n**Status**: Final\n\n**Change Log**:\n- v1.0 (Nov 2025): Initial comprehensive report\n\n**Contact**: For questions or feedback about this report, please refer to the research methodology and sources cited.\n\n---\n\n*End of Report*\n","children",[],[278],"c55573e4-d1c8-11f0-8d7b-42dde27851f2","test_migration.ts","019aee26-23ac-778a-bcd7-2e1fa43af29c",["D",1764932068219],"/**\n * Migration test script\n *\n * This script demonstrates how the migration works.\n * Note: This is for documentation purposes. In Val Town, you would run this\n * directly in your environment.\n */\n\nimport { generateMigrationSQL } from \"./backend/migrations.ts\";\n\nconsole.log(\"=== wrkflw Table Migration Test ===\\n\");\n\n// Test 1: Generate migration SQL\nconsole.log(\"1. Generated Migration SQL (without dropping old table):\");\nconsole.log(\"---\");\nconst sql = generateMigrationSQL(false);\nconsole.log(sql);\nconsole.log(\"\\n\");\n\n// Test 2: Show what would happen during automatic migration\nconsole.log(\"2. Automatic Migration Process:\");\nconsole.log(\"---\");\nconsole.log(\"When you call WorkflowStorage.init(), the following happens:\");\nconsole.log(\" a) Check if 'workflow_runs' table exists\");\nconsole.log(\" b) Create 'wrkflw_workflow_runs' table if needed\");\nconsole.log(\" c) Copy all data from old table to new table\");\nconsole.log(\" d) Create indexes on new table\");\nconsole.log(\" e) Old table is preserved for safety\");\nconsole.log(\"\\n\");\n\n// Test 3: Manual migration example\nconsole.log(\"3. Manual Migration Example:\");\nconsole.log(\"---\");\nconsole.log(\"import { runMigrations } from './backend/migrations.ts';\");\nconsole.log(\"\");\nconsole.log(\"// Run migration (keeps old table)\");\nconsole.log(\"const results = await runMigrations();\");\nconsole.log(\"console.log(results[0].message);\");\nconsole.log(\"console.log(`Migrated ${results[0].rowCount} rows`);\");\nconsole.log(\"\\n\");\n\n// Test 4: Table naming verification\nconsole.log(\"4. Table Naming Scheme:\");\nconsole.log(\"---\");\nconsole.log(\"Old table name: workflow_runs\");\nconsole.log(\"New table name: wrkflw_workflow_runs\");\nconsole.log(\"Old index names:\");\nconsole.log(\" - idx_workflow_runs_workflow_id\");\nconsole.log(\" - idx_workflow_runs_status\");\nconsole.log(\"New index names:\");\nconsole.log(\" - idx_wrkflw_workflow_runs_workflow_id\");\nconsole.log(\" - idx_wrkflw_workflow_runs_status\");\nconsole.log(\"\\n\");\n\nconsole.log(\"=== All table names now use wrkflw_ prefix ===\");\n",[],[286],"c5556caa-d1c8-11f0-8d7b-42dde27851f2","deno.json","019aee26-23ac-778a-bcd7-28a2a9679a70",["D",1764932068219],"{\n\t\"$schema\": \"https://raw.githubusercontent.com/denoland/deno/348900b8b79f4a434cab4c74b3bc8d4d2fa8ee74/cli/schemas/config-file.v1.json\",\n\t\"compilerOptions\": {\n\t\t\"lib\": [\n\t\t\t\"dom\",\n\t\t\t\"dom.iterable\",\n\t\t\t\"dom.asynciterable\",\n\t\t\t\"deno.ns\",\n\t\t\t\"deno.unstable\"\n\t\t],\n\t\t\"noImplicitAny\": false,\n\t\t\"strict\": false,\n\t\t\"types\": [\"https://www.val.town/types/valtown.d.ts\"]\n\t},\n\t\"experimental\": {\n\t\t\"unstable-node-globals\": true,\n\t\t\"unstable-sloppy-imports\": true,\n\t\t\"unstable-temporal\": true,\n\t\t\"unstable-worker-options\": true\n\t},\n\t\"imports\": {\n\t\t\"@biomejs/biome\": \"npm:@biomejs/biome@^2.3.5\"\n\t},\n\t\"lint\": {\n\t\t\"files\": {\n\t\t\t\"include\": [\"deno:/https/esm.town/**/*\"]\n\t\t},\n\t\t\"rules\": {\n\t\t\t\"exclude\": [\"no-explicit-any\", \"no-import-prefix\"]\n\t\t}\n\t},\n\t\"lock\": false,\n\t\"nodeModulesDir\": \"auto\"\n}\n",[],[293],"c5556a52-d1c8-11f0-8d7b-42dde27851f2","biome.json","019aee26-23ac-778a-bcd7-263ffe65eddc",["D",1764932068219],"{\n\t\"$schema\": \"https://biomejs.dev/schemas/2.3.5/schema.json\",\n\t\"vcs\": { \"enabled\": true, \"clientKind\": \"git\", \"useIgnoreFile\": true },\n\t\"files\": { \"ignoreUnknown\": false },\n\t\"formatter\": { \"enabled\": true, \"indentStyle\": \"tab\" },\n\t\"linter\": { \"enabled\": true, \"rules\": { \"recommended\": true } },\n\t\"javascript\": { \"formatter\": { \"quoteStyle\": \"double\" } },\n\t\"assist\": {\n\t\t\"enabled\": true,\n\t\t\"actions\": { \"source\": { \"organizeImports\": \"on\" } }\n\t}\n}\n",[],[300],"c5556160-d1c8-11f0-8d7b-42dde27851f2","SCOUT.md","019aee26-23ac-778a-bcd7-23ee619b2506",["D",1764932068219],"# wrkflw - Codebase Overview\n\n**Last Updated:** November 20, 2025\n\n## Project Summary\n\nwrkflw is a TypeScript-first workflow engine for Val Town, inspired by Mastra. It enables building strongly-typed, composable workflows that run on Val Town's serverless platform with SQLite-backed persistence.\n\n## Architecture Overview\n\n```\n┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐\n│ Workflow │ │ Execution │ │ SQLite │\n│ Builder │───▶│ Engine │───▶│ Storage │\n│ (workflow.ts) │ │ (engine.ts) │ │ (storage.ts) │\n└─────────────────┘ └─────────────────┘ └─────────────────┘\n │ │ │\n ▼ ▼ ▼\n┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐\n│ Step │ │ Workflow │ │ API Server │\n│ Definition │ │ Run │ │ (Optional) │\n│ (step.ts) │ │ (run.ts) │ │ (api-server) │\n└─────────────────┘ └─────────────────┘ └─────────────────┘\n```\n\n## Core Components\n\n### 1. **Step System** (`backend/step.ts`)\n- Define reusable workflow steps with Zod schemas\n- Strong typing for inputs/outputs\n- Access to workflow context (previous results, state)\n\n### 2. **Workflow Builder** (`backend/workflow.ts`)\n- Fluent API: `.then(step).then(step).commit()`\n- Automatic type inference through step chains\n- Compile-time type checking\n\n### 3. **Execution Engine** (`backend/engine.ts`)\n- Sequential step execution\n- State persistence after each step\n- Error handling and result capture\n\n### 4. **Storage Layer** (`backend/storage.ts`)\n- SQLite-backed workflow snapshots\n- Track execution path and step results\n- Query runs by workflow ID or status\n\n### 5. **Prebuilt Steps** (`backend/prebuilt-steps.ts`)\n- HTTP requests (GET, POST)\n- Data transformation (map, filter, pick)\n- Utilities (delay, logger, template)\n\n## Key Features\n\n- **Type Safety**: Full TypeScript inference across workflows\n- **Val Town Native**: Works with HTTP, Cron, Email triggers\n- **Visualization**: Mermaid diagrams + React canvas visualizer\n- **Persistence**: ACID-compliant SQLite storage\n- **Composability**: Steps and workflows as reusable modules\n\n## Quick Start Tasks\n\nWhen resuming development, start here:\n\n1. **Run Examples**: `deno run --allow-all examples/simple-workflow.ts`\n2. **Test Types**: `deno check backend/index.ts`\n3. **Visualize**: `deno run --allow-all examples/api-server-example-local.ts`\n4. **Frontend Dev**: `cd frontend && npm run dev`\n\n## Development Commands\n\n```bash\n# Backend (Deno)\ndeno lint # Lint backend code\ndeno fmt # Format backend code\ndeno check backend/index.ts # Type check\n\n# Frontend (Node)\ncd frontend && npm run dev # Start visualizer\ncd frontend && npm run build # Build for production\n\n# Code Quality\nnpx @biomejs/biome check --write # Fix all issues\n```\n\n## Project Structure\n\n```\nbackend/ # Core workflow engine (Deno)\n ├── index.ts # Main exports\n ├── types.ts # TypeScript definitions\n ├── step.ts # Step creation\n ├── workflow.ts # Workflow builder\n ├── engine.ts # Execution engine\n ├── storage.ts # SQLite persistence\n └── prebuilt-steps.ts # Ready-to-use steps\n\nexamples/ # Usage examples\n ├── simple-workflow.ts\n ├── http-trigger.ts\n └── api-server-example-local.ts\n\nfrontend/ # React visualizer (Vite)\n └── src/components/ # Canvas workflow viewer\n```\n\n## Common Patterns\n\n### Creating a Workflow\n```typescript\nconst workflow = createWorkflow({\n id: 'my-workflow',\n inputSchema: z.object({ data: z.string() }),\n outputSchema: z.object({ result: z.string() }),\n})\n .then(step1)\n .then(step2)\n .commit();\n```\n\n### Running a Workflow\n```typescript\nconst run = await workflow.createRun();\nconst result = await run.start({ inputData: { data: \"hello\" } });\n```\n\n### Accessing Previous Results\n```typescript\nexecute: async ({ inputData, getStepResult }) => {\n const previousResult = getStepResult(step1);\n return { result: previousResult.value + 1 };\n}\n```\n\n## Next Steps for Development\n\n1. **Add New Prebuilt Steps**: Extend `backend/prebuilt-steps.ts`\n2. **Enhance Visualization**: Improve React canvas in `frontend/`\n3. **Advanced Control Flow**: Parallel execution, branching\n4. **Error Recovery**: Retry logic, fallback steps\n5. **Performance**: Optimize SQLite queries, add indexes\n\n## Key Files to Understand\n\n- `backend/types.ts` - Core type definitions\n- `backend/workflow.ts` - Workflow builder implementation\n- `backend/engine.ts` - Execution logic\n- `examples/simple-workflow.ts` - Basic usage pattern\n- `AGENTS.md` - Development guidelines and commands",[],[307],"c5554cde-d1c8-11f0-8d7b-42dde27851f2","README.md","019aee26-23ac-778a-bcd7-1f5c37cb1e94",["D",1764932068219],"# wrkflw\n\nA **TypeScript-first workflow engine** for Val Town, inspired by [Mastra](https://github.com/mastra-ai/mastra).\n\nBuild strongly-typed, composable workflows that run on Val Town's serverless platform.\n\n## Features\n\n- **Strongly Typed**: Full TypeScript inference across workflow definitions\n- **Composable**: Build workflows from reusable steps\n- **Trigger Agnostic**: Works with HTTP, Cron, or Email Val Town triggers\n- **Durable State**: SQLite-backed persistence with ACID guarantees\n- **Runtime Validation**: Zod schema validation for inputs and outputs\n- **Error Handling**: Structured error tracking and recovery\n- **Visual Workflows**: Generate Mermaid diagrams of workflow structure and execution state\n\n## Quick Start\n\n### 1. Define Steps\n\n```typescript\nimport { createStep } from \"./backend/index.ts\";\nimport { z } from \"npm:zod@^3.23\";\n\nconst fetchUser = createStep({\n id: 'fetch-user',\n inputSchema: z.object({ userId: z.string() }),\n outputSchema: z.object({ name: z.string(), email: z.string() }),\n execute: async ({ inputData }) => {\n const user = await db.getUser(inputData.userId);\n return { name: user.name, email: user.email };\n },\n});\n```\n\n### 2. Build Workflows\n\n```typescript\nimport { createWorkflow } from \"./backend/index.ts\";\n\nconst emailWorkflow = createWorkflow({\n id: 'email-campaign',\n inputSchema: z.object({ userId: z.string() }),\n outputSchema: z.object({ sent: z.boolean() }),\n})\n .then(fetchUser) // Type: { userId: string } -> { name, email }\n .then(generateEmail) // Type: { name, email } -> { subject, body }\n .then(sendEmail) // Type: { subject, body } -> { sent }\n .commit();\n```\n\n### 3. Execute Workflows\n\n```typescript\n// Create a run\nconst run = await emailWorkflow.createRun();\n\n// Execute\nconst result = await run.start({\n inputData: { userId: 'user-123' },\n});\n\nconsole.log(result); // { sent: true }\n```\n\n## Prebuilt Steps\n\nwrkflw includes a collection of ready-to-use steps for common tasks:\n\n### HTTP/API Steps\n\n**`httpGet`** - Make GET requests to APIs\n```typescript\nimport { httpGet, createWorkflow } from \"./backend/index.ts\";\n\nconst workflow = createWorkflow({\n id: 'fetch-api-data',\n inputSchema: z.object({\n url: z.string().url(),\n headers: z.record(z.string()).optional(),\n queryParams: z.record(z.string()).optional(),\n }),\n outputSchema: z.object({\n status: z.number(),\n data: z.unknown(),\n headers: z.record(z.string()),\n }),\n})\n .then(httpGet)\n .commit();\n```\n\n**`httpPost`** - POST JSON data to APIs\n```typescript\nimport { httpPost } from \"./backend/index.ts\";\n\n// Use in workflow with inputData: { url, body, headers? }\n```\n\n### String/Template Steps\n\n**`template`** - String interpolation with variables\n```typescript\nimport { template } from \"./backend/index.ts\";\n\nconst workflow = createWorkflow({\n inputSchema: z.object({\n template: z.string(),\n variables: z.record(z.union([z.string(), z.number(), z.boolean()])),\n }),\n outputSchema: z.object({ result: z.string() }),\n})\n .then(template)\n .commit();\n\n// Input: { template: \"Hello {{name}}!\", variables: { name: \"Alice\" } }\n// Output: { result: \"Hello Alice!\" }\n```\n\n### Data Transformation Steps\n\n**`filterArray`** - Filter array items by condition\n```typescript\nimport { filterArray } from \"./backend/index.ts\";\n\n// Input: {\n// array: [1, 2, 3, 4, 5],\n// filterFn: \"(n) => n % 2 === 0\"\n// }\n// Output: {\n// filtered: [2, 4],\n// originalCount: 5,\n// filteredCount: 2\n// }\n```\n\n**`mapArray`** - Transform array items\n```typescript\nimport { mapArray } from \"./backend/index.ts\";\n\n// Input: {\n// array: [1, 2, 3],\n// mapFn: \"(n) => n * 2\"\n// }\n// Output: { mapped: [2, 4, 6], count: 3 }\n```\n\n**`pickFields`** - Extract specific fields from objects\n```typescript\nimport { pickFields } from \"./backend/index.ts\";\n\n// Input: {\n// object: { id: 1, name: \"Alice\", secret: \"xyz\" },\n// fields: [\"id\", \"name\"]\n// }\n// Output: { result: { id: 1, name: \"Alice\" } }\n```\n\n### Utility Steps\n\n**`delay`** - Wait for a specified duration\n```typescript\nimport { delay } from \"./backend/index.ts\";\n\n// Input: { durationMs: 2000 }\n// Waits 2 seconds, then outputs: { durationMs: 2000, completedAt: \"...\" }\n```\n\n**`logger`** - Structured logging at different levels\n```typescript\nimport { logger } from \"./backend/index.ts\";\n\n// Input: {\n// level: \"info\",\n// message: \"Processing data\",\n// data: { count: 10 }\n// }\n// Logs to console and passes data through\n```\n\n### Complete Example with Prebuilt Steps\n\n```typescript\nimport {\n createWorkflow,\n createStep,\n httpGet,\n filterArray,\n mapArray,\n template\n} from \"./backend/index.ts\";\n\n// Helper step to prepare API request\nconst prepareRequest = createStep({\n id: \"prepare-request\",\n inputSchema: z.object({ userId: z.number() }),\n outputSchema: z.object({\n url: z.string(),\n queryParams: z.record(z.string()).optional(),\n }),\n execute: async ({ inputData }) => ({\n url: \"https://api.example.com/posts\",\n queryParams: { userId: String(inputData.userId) },\n }),\n});\n\n// Build workflow using prebuilt steps\nconst workflow = createWorkflow({\n id: 'process-user-posts',\n inputSchema: z.object({ userId: z.number() }),\n outputSchema: z.object({ titles: z.array(z.string()) }),\n})\n .then(prepareRequest) // Convert input to API request\n .then(httpGet) // Fetch from API (prebuilt)\n // Add more steps as needed\n .commit();\n```\n\nSee `examples/prebuilt-steps-simple.ts` for more examples.\n\n## Installation\n\n### For Val Town\n\n1. Create a new val or project in Val Town\n2. Copy the `backend/` directory into your val\n3. Import and use:\n\n```typescript\nimport { createStep, createWorkflow } from \"./backend/index.ts\";\n```\n\n### For Local Development\n\n```bash\ngit clone \ncd wrkflw\ndeno run --allow-all examples/simple-workflow.ts\n```\n\n## Examples\n\n### HTTP Trigger\n\nCreate a new HTTP val in Val Town:\n\n```typescript\nimport { emailWorkflow } from \"./workflows/email-campaign.ts\";\n\nexport default async function(req: Request) {\n const url = new URL(req.url);\n const userId = url.searchParams.get(\"userId\");\n \n if (!userId) {\n return Response.json({ error: \"Missing userId\" }, { status: 400 });\n }\n \n const run = await emailWorkflow.createRun();\n const result = await run.start({ inputData: { userId } });\n \n return Response.json({ success: true, result });\n}\n```\n\n### Cron Trigger\n\nCreate a new Cron val in Val Town:\n\n```typescript\nimport { welcomeWorkflow } from \"./workflows/welcome.ts\";\n\nexport default async function(interval: Interval) {\n // Get users who signed up in the last hour\n const newUsers = await getRecentUsers();\n \n for (const user of newUsers) {\n const run = await welcomeWorkflow.createRun();\n await run.start({ inputData: { userId: user.id } });\n console.log(`Sent welcome email to ${user.id}`);\n }\n}\n```\n\n### Email Trigger\n\nCreate a new Email val in Val Town:\n\n```typescript\nimport { supportWorkflow } from \"./workflows/support.ts\";\n\nexport default async function(email: Email) {\n const run = await supportWorkflow.createRun();\n \n const result = await run.start({\n inputData: {\n from: email.from,\n subject: email.subject,\n body: email.text,\n },\n });\n \n console.log(`Processed support email: ${result.ticketId}`);\n}\n```\n\n## API Reference\n\n### `createStep(config)`\n\nCreate a typed workflow step.\n\n```typescript\ncreateStep({\n id: 'step-id',\n description: 'Optional description',\n inputSchema: z.object({ /* input type */ }),\n outputSchema: z.object({ /* output type */ }),\n execute: async ({ inputData, getStepResult, getInitData }) => {\n // Your logic here\n return { /* matches outputSchema */ };\n },\n});\n```\n\n**Execute Context:**\n- `inputData` - Validated input for this step\n- `getStepResult(step)` - Get output from a previous step\n- `getInitData()` - Get the workflow's initial input\n- `state` - Workflow-level state (if defined)\n- `setState(newState)` - Update workflow state\n- `runId` - Current run ID\n- `workflowId` - Current workflow ID\n\n### `createWorkflow(config)`\n\nCreate a typed workflow builder.\n\n```typescript\ncreateWorkflow({\n id: 'workflow-id',\n description: 'Optional description',\n inputSchema: z.object({ /* input type */ }),\n outputSchema: z.object({ /* output type */ }),\n stateSchema: z.object({ /* optional state */ }),\n})\n .then(step1)\n .then(step2)\n .commit();\n```\n\n**Builder Methods:**\n- `.then(step)` - Add a sequential step\n- `.commit()` - Finalize the workflow\n\n### `workflow.createRun(runId?)`\n\nCreate a new workflow run instance.\n\n```typescript\nconst run = await workflow.createRun(); // Auto-generated ID\n// or\nconst run = await workflow.createRun('custom-run-id');\n```\n\n### `run.start(params)`\n\nExecute the workflow.\n\n```typescript\nconst result = await run.start({\n inputData: { /* matches workflow.inputSchema */ },\n initialState: { /* optional, matches workflow.stateSchema */ },\n});\n```\n\n### `run.getSnapshot()`\n\nGet the current state snapshot.\n\n```typescript\nconst snapshot = await run.getSnapshot();\nconsole.log({\n status: snapshot.status, // 'running' | 'success' | 'failed'\n stepResults: snapshot.stepResults, // Results from each step\n executionPath: snapshot.executionPath, // Steps executed\n});\n```\n\n## Type Inference\n\nThe workflow builder automatically infers types through the chain:\n\n```typescript\nconst step1 = createStep({\n inputSchema: z.object({ a: z.string() }),\n outputSchema: z.object({ b: z.number() }),\n execute: async ({ inputData }) => ({ b: 42 }),\n});\n\nconst step2 = createStep({\n inputSchema: z.object({ b: z.number() }), // Must match step1's output\n outputSchema: z.object({ c: z.boolean() }),\n execute: async ({ inputData }) => ({ c: true }),\n});\n\nconst workflow = createWorkflow({\n inputSchema: z.object({ a: z.string() }),\n outputSchema: z.object({ c: z.boolean() }),\n})\n .then(step1) // Accepts { a: string }\n .then(step2) // Accepts { b: number } from step1\n .commit();\n\n// TypeScript knows the output is { c: boolean }\n```\n\nIf step types don't match, TypeScript will show a compile error:\n\n```typescript\nconst incompatibleStep = createStep({\n inputSchema: z.object({ x: z.string() }), // Error: doesn't match step1 output\n outputSchema: z.object({ y: z.string() }),\n execute: async ({ inputData }) => ({ y: \"value\" }),\n});\n\nworkflow.then(step1).then(incompatibleStep); // TypeScript error\n```\n\n## Error Handling\n\nErrors are automatically captured and persisted:\n\n```typescript\ntry {\n const result = await run.start({ inputData });\n} catch (error) {\n // Workflow failed - snapshot saved with error\n const snapshot = await run.getSnapshot();\n console.log(snapshot.status); // 'failed'\n console.log(snapshot.error); // Error message\n}\n```\n\n## State Management\n\nDefine workflow-level state that persists across steps:\n\n```typescript\nconst workflow = createWorkflow({\n id: 'stateful-workflow',\n inputSchema: z.object({ start: z.number() }),\n outputSchema: z.object({ total: z.number() }),\n stateSchema: z.object({ counter: z.number() }),\n});\n\nconst incrementStep = createStep({\n id: 'increment',\n inputSchema: z.object({ start: z.number() }),\n outputSchema: z.object({ value: z.number() }),\n execute: async ({ inputData, state, setState }) => {\n const newCounter = (state?.counter || 0) + inputData.start;\n setState({ counter: newCounter });\n return { value: newCounter };\n },\n});\n```\n\n## Storage\n\nAll workflow runs are persisted to Val Town's SQLite database:\n\n```typescript\nimport { WorkflowStorage } from \"./backend/index.ts\";\n\nconst storage = new WorkflowStorage();\nawait storage.init();\n\n// List all runs\nconst runs = await storage.listRuns();\n\n// List runs for specific workflow\nconst workflowRuns = await storage.listRuns('my-workflow-id');\n\n// Get runs by status\nconst failedRuns = await storage.getRunsByStatus('failed');\n```\n\n## Workflow Visualization\n\nwrkflw includes built-in visualization capabilities to help you understand and debug your workflows. You can generate visual diagrams in multiple formats (currently Mermaid, with more formats planned).\n\n### Visualize Workflow Structure\n\nGenerate a diagram showing the static structure of your workflow:\n\n```typescript\nconst workflow = createWorkflow({\n id: 'data-processing',\n inputSchema: z.object({ url: z.string() }),\n outputSchema: z.object({ saved: z.boolean() }),\n})\n .then(fetchData)\n .then(processData)\n .then(saveResults)\n .commit();\n\n// Generate Mermaid diagram\nconst diagram = workflow.visualize(\"mermaid\");\nconsole.log(diagram);\n```\n\nOutput:\n```mermaid\ngraph TD\n Start([data-processing])\n fetch_data[fetch-data
Fetch data from API]\n Start --> fetch_data\n process_data[process-data
Process and transform data]\n fetch_data --> process_data\n save_results[save-results
Save to database]\n process_data --> save_results\n End([End])\n save_results --> End\n```\n\n### Visualize Execution State\n\nGenerate a diagram showing the current state of a workflow run:\n\n```typescript\nconst run = await workflow.createRun();\nawait run.start({ inputData: { url: \"https://api.example.com/data\" } });\n\n// Visualize execution state\nconst executionDiagram = await run.visualize(\"mermaid\", {\n highlightCurrentStep: true,\n showStatus: true,\n});\n\nconsole.log(executionDiagram);\n```\n\nThe execution visualization includes:\n- ✓ Checkmarks for completed steps\n- ✗ X marks for failed steps\n- ● Dots for currently running steps\n- Color-coded step states (success, failed, running, pending)\n- Error messages for failed steps\n\n### Visualization Options\n\nBoth workflow and execution visualizations support various options:\n\n```typescript\n// Workflow structure options\nworkflow.visualize(\"mermaid\", {\n includeDescriptions: true, // Show step descriptions (default: true)\n includeSchemas: false, // Show schema info (not yet implemented)\n});\n\n// Execution state options\nawait run.visualize(\"mermaid\", {\n includeDescriptions: true, // Show step descriptions (default: true)\n highlightCurrentStep: true, // Highlight the current/last step (default: true)\n showStatus: true, // Show status icons (default: true)\n showResults: false, // Show step results (default: false)\n});\n```\n\n### Rendering Diagrams\n\nMermaid diagrams can be rendered in multiple ways:\n\n1. **GitHub/GitLab**: Paste into markdown files (rendered automatically)\n2. **Mermaid Live Editor**: Copy to [mermaid.live](https://mermaid.live)\n3. **VS Code**: Use the Mermaid Preview extension\n4. **Documentation sites**: Most support Mermaid natively\n\n### Using Visualization Functions Directly\n\nYou can also use the visualization functions directly without calling methods on workflow/run instances:\n\n```typescript\nimport { visualizeWorkflow, visualizeExecution } from \"./backend/index.ts\";\n\n// Visualize workflow structure\nconst diagram = visualizeWorkflow(workflow, \"mermaid\", {\n includeDescriptions: true,\n});\n\n// Visualize execution with snapshot\nconst snapshot = await run.getSnapshot();\nif (snapshot) {\n const executionDiagram = visualizeExecution(\n workflow,\n snapshot,\n \"mermaid\",\n { showStatus: true }\n );\n}\n```\n\n### Future Visualization Formats\n\nThe visualization API is designed to be extensible. Planned formats include:\n\n- **JSON**: Graph data for custom visualization libraries (D3, Cytoscape)\n- **ASCII**: Text-based diagrams for CLI/terminal output\n- **DOT**: Graphviz format for advanced layout algorithms\n\nSee `examples/workflow-visualization.ts` for a complete example.\n\n## Canvas Visualizer (React App)\n\nwrkflw includes a powerful React-based canvas visualizer that provides real-time, interactive workflow visualization. Unlike the text-based Mermaid diagrams, the canvas visualizer offers a dynamic, web-based UI for monitoring and debugging workflows.\n\n### Features\n\n- **Interactive Canvas**: Pan, zoom, and navigate workflow diagrams with smooth animations\n- **Real-time Updates**: Watch workflow execution live with auto-refresh\n- **Step Details**: Click on nodes to view outputs, errors, and execution status\n- **Execution History**: Browse past workflow runs and their results\n- **Multiple Workflows**: Switch between different workflows seamlessly\n- **Status Indicators**: Visual feedback for running, success, and failed steps\n\n### Quick Start\n\n**Option 1: Embedded Visualizer (Easiest)**\n```bash\n# Single server with embedded React visualizer\ndeno run --allow-all examples/api-server-example-local.ts\n# Open http://localhost:8000\n```\n\n**Option 2: Full ReactFlow Canvas (Most Features)**\n\n1. **Start the API Server**:\n```bash\n# From the root directory\ndeno run --allow-all examples/api-server-example-local.ts\n```\n\n2. **Start the Frontend** (in a separate terminal):\n```bash\ncd frontend\nnpm install\nnpm run dev\n```\n\n3. **Open the Visualizer**:\nVisit `http://localhost:3000` in your browser\n\n### Usage\n\n```typescript\nimport { createApiServer, WorkflowStorage } from \"./backend/index.ts\";\n\n// Initialize storage and API server\nconst storage = new WorkflowStorage();\nawait storage.init();\nconst apiServer = createApiServer(storage);\n\n// Register your workflows\napiServer.registerWorkflow(dataProcessingWorkflow, {\n url: \"https://api.example.com/data\" // Example input\n});\n\n// Start the server\nawait apiServer.serve(8000);\n```\n\nThe canvas visualizer will automatically:\n- Display all registered workflows in the sidebar\n- Show workflow structure with connected nodes\n- Update in real-time as workflows execute\n- Highlight running steps with animations\n- Show success/failure states with color coding\n- Display step outputs and error messages\n\n### API Endpoints\n\nThe API server exposes the following endpoints for the visualizer:\n\n```\nGET /api/workflows - List all registered workflows\nGET /api/workflows/:id - Get workflow details\nGET /api/workflows/:id/runs - List runs for a workflow\nPOST /api/workflows/:id/runs - Create and start a new run\nGET /api/runs/:runId - Get detailed run status\n```\n\n### Architecture\n\n```\n┌──────────────────┐ HTTP ┌──────────────────┐\n│ React Frontend │ ◄─────────► │ Deno API Server │\n│ (Port 3000) │ /api/* │ (Port 8000) │\n└──────────────────┘ └──────────────────┘\n │ │\n │ React Flow │ Workflow Engine\n │ Canvas Rendering │ SQLite Storage\n ▼ ▼\n Visualizer UI Execution State\n```\n\nSee `frontend/README.md` for detailed documentation and customization options.\n\n## Project Structure\n\n```\nwrkflw/\n├── backend/\n│ ├── index.ts # Main exports\n│ ├── types.ts # TypeScript types\n│ ├── step.ts # Step creation\n│ ├── workflow.ts # Workflow builder\n│ ├── engine.ts # Execution engine\n│ ├── storage.ts # SQLite persistence\n│ ├── run.ts # Run management\n│ ├── visualize.ts # Workflow visualization\n│ ├── api-server.ts # API server for canvas visualizer\n│ └── prebuilt-steps.ts # Ready-to-use steps\n├── frontend/\n│ ├── src/\n│ │ ├── components/\n│ │ │ ├── WorkflowVisualizer.tsx # Main canvas component\n│ │ │ ├── StepNode.tsx # Custom step node\n│ │ │ ├── WorkflowVisualizer.css\n│ │ │ └── StepNode.css\n│ │ ├── App.tsx # Main app component\n│ │ ├── types.ts # Frontend types\n│ │ └── main.tsx # Entry point\n│ ├── package.json\n│ ├── vite.config.ts\n│ └── README.md\n├── examples/\n│ ├── simple-workflow.ts\n│ ├── http-trigger.ts\n│ ├── cron-trigger.ts\n│ ├── prebuilt-steps-simple.ts\n│ ├── workflow-visualization.ts\n│ └── api-server-example.ts # Canvas visualizer demo\n├── PLAN.md\n└── README.md\n```\n\n## Complete Example\n\nHere's a full example of a workflow that fetches data, processes it, and sends a notification:\n\n```typescript\nimport { createWorkflow, createStep } from \"./backend/index.ts\";\nimport { z } from \"npm:zod@^3.23\";\n\n// Step 1: Fetch data from an API\nconst fetchData = createStep({\n id: 'fetch-data',\n inputSchema: z.object({ url: z.string() }),\n outputSchema: z.object({ data: z.array(z.any()) }),\n execute: async ({ inputData }) => {\n const response = await fetch(inputData.url);\n const data = await response.json();\n return { data };\n },\n});\n\n// Step 2: Process the data\nconst processData = createStep({\n id: 'process-data',\n inputSchema: z.object({ data: z.array(z.any()) }),\n outputSchema: z.object({ \n count: z.number(), \n summary: z.string() \n }),\n execute: async ({ inputData }) => {\n return {\n count: inputData.data.length,\n summary: `Processed ${inputData.data.length} items`,\n };\n },\n});\n\n// Step 3: Send notification\nconst sendNotification = createStep({\n id: 'send-notification',\n inputSchema: z.object({ count: z.number(), summary: z.string() }),\n outputSchema: z.object({ sent: z.boolean() }),\n execute: async ({ inputData, getStepResult }) => {\n // Access original data from first step\n const originalData = getStepResult(fetchData);\n \n console.log(`Sending notification: ${inputData.summary}`);\n // Send to Slack, Discord, etc.\n \n return { sent: true };\n },\n});\n\n// Build the workflow\nexport const dataProcessingWorkflow = createWorkflow({\n id: 'data-processing',\n inputSchema: z.object({ url: z.string() }),\n outputSchema: z.object({ sent: z.boolean() }),\n})\n .then(fetchData)\n .then(processData)\n .then(sendNotification)\n .commit();\n\n// Use in an HTTP val\nexport default async function(req: Request) {\n const { url } = await req.json();\n \n const run = await dataProcessingWorkflow.createRun();\n const result = await run.start({ inputData: { url } });\n \n return Response.json(result);\n}\n```\n\n## Architecture\n\n```\n┌─────────────────────────────────────────────────────────┐\n│ Val Town Trigger │\n│ (HTTP / Cron / Email val) │\n└─────────────────┬───────────────────────────────────────┘\n │\n ▼\n┌─────────────────────────────────────────────────────────┐\n│ Workflow Instance │\n│ - createRun() creates a new execution │\n│ - start() begins execution │\n└─────────────────┬───────────────────────────────────────┘\n │\n ▼\n┌─────────────────────────────────────────────────────────┐\n│ Execution Engine │\n│ - Validates inputs with Zod schemas │\n│ - Executes steps sequentially │\n│ - Captures errors and results │\n└─────────────────┬───────────────────────────────────────┘\n │\n ▼\n┌─────────────────────────────────────────────────────────┐\n│ Step Execution Context │\n│ - inputData: validated input │\n│ - getStepResult(): access previous steps │\n│ - getInitData(): access workflow input │\n│ - state/setState: workflow-level state │\n└─────────────────┬───────────────────────────────────────┘\n │\n ▼\n┌─────────────────────────────────────────────────────────┐\n│ SQLite Storage │\n│ - Saves snapshot after each step │\n│ - Tracks execution path and results │\n│ - Enables debugging and recovery │\n└─────────────────────────────────────────────────────────┘\n```\n\n## Roadmap\n\n**Phase 1 (Complete):** Core engine with sequential execution \n**Phase 2:** Advanced control flow (parallel, branching, loops) \n**Phase 3:** Durability (suspend/resume, sleep) \n**Phase 4:** Enhanced features (foreach, nested workflows, retries) \n**Phase 5:** Observability (streaming, debugging UI)\n\n## Contributing\n\nThis is an experimental project for learning and personal automation. Contributions, ideas, and feedback are welcome.\n\n## Inspiration\n\nThis project is heavily inspired by [Mastra](https://github.com/mastra-ai/mastra), adapted for Val Town's serverless environment and optimized for personal automation use cases.\n\n## License\n\nMIT\n",[],[314],"c5553f82-d1c8-11f0-8d7b-42dde27851f2","PLAN.md","019aee26-23ac-778a-bcd7-19eed8a7a7b4",["D",1764932068219],"# Val Town Workflow Engine - Implementation Plan\n\n## Project Overview\n\nBuild a **TypeScript-first workflow engine** for Val Town inspired by Mastra's architecture, optimized for learning and personal automation use cases.\n\n---\n\n## Design Principles\n\n1. **Strong Typing**: Full TypeScript inference across workflow definitions\n2. **Composable**: Steps and workflows are composable units\n3. **Flexible Triggering**: Workflows as modules callable from any Val Town trigger (HTTP, Cron, Email)\n4. **Simple State Management**: SQLite-backed with ACID guarantees\n5. **Val Town Native**: Embrace platform constraints (no external dependencies)\n\n---\n\n## Architecture Overview\n\n```mermaid\ngraph TD\n A[Workflow Definition] -->|createWorkflow| B[Workflow Builder]\n B -->|.then| C[Step Chain]\n B -->|.commit| D[Executable Workflow]\n \n E[Step Definition] -->|createStep| F[Step Instance]\n F --> C\n \n D -->|.start| G[Execution Engine]\n G --> H[Step Execution]\n H --> I[SQLite State Store]\n \n J[HTTP Trigger] --> D\n K[Cron Trigger] --> D\n L[Email Trigger] --> D\n \n style A fill:#e1f5ff\n style D fill:#fff4e1\n style I fill:#ffe1f5\n```\n\n---\n\n## Phase 1: Core Engine (MVP)\n\n### Components to Build\n\n#### 1. **Step Definition** (`step.ts`)\n\n```typescript\ninterface StepConfig {\n id: TId;\n description?: string;\n inputSchema: z.ZodSchema;\n outputSchema: z.ZodSchema;\n stateSchema?: z.ZodSchema;\n execute: ExecuteFunction;\n}\n\ninterface ExecuteFunction {\n (context: {\n inputData: TInput;\n state?: TState;\n setState?: (state: TState) => void;\n getStepResult: (step: T) => z.infer;\n getInitData: () => any;\n }): Promise | TOutput;\n}\n\nfunction createStep(\n config: StepConfig\n): Step;\n```\n\n**Example Usage:**\n```typescript\nconst fetchUser = createStep({\n id: 'fetch-user',\n inputSchema: z.object({ userId: z.string() }),\n outputSchema: z.object({ \n name: z.string(), \n email: z.string() \n }),\n execute: async ({ inputData }) => {\n const user = await db.getUser(inputData.userId);\n return { name: user.name, email: user.email };\n }\n});\n```\n\n---\n\n#### 2. **Workflow Builder** (`workflow.ts`)\n\n```typescript\ninterface WorkflowConfig {\n id: TId;\n description?: string;\n inputSchema: z.ZodSchema;\n outputSchema: z.ZodSchema;\n stateSchema?: z.ZodSchema;\n}\n\nclass WorkflowBuilder {\n then(\n step: TStep\n ): WorkflowBuilder, TInput, TStep['outputSchema']>;\n \n commit(): Workflow;\n}\n\nfunction createWorkflow(\n config: WorkflowConfig\n): WorkflowBuilder<{}, TInput, TInput>;\n```\n\n**Example Usage:**\n```typescript\nconst emailWorkflow = createWorkflow({\n id: 'email-campaign',\n inputSchema: z.object({ userId: z.string() }),\n outputSchema: z.object({ sent: z.boolean() })\n})\n .then(fetchUser)\n .then(generateEmail)\n .then(sendEmail)\n .commit();\n\n// Fully typed!\ntype Output = z.infer;\n```\n\n---\n\n#### 3. **Execution Engine** (`engine.ts`)\n\n```typescript\ninterface ExecutionContext {\n workflowId: string;\n runId: string;\n executionPath: number[]; // Current position in step array\n stepResults: Record;\n state: Record;\n}\n\ninterface StepResult {\n status: 'success' | 'failed' | 'running';\n output?: T;\n error?: Error;\n timestamp: number;\n}\n\nclass ExecutionEngine {\n async execute(params: {\n workflow: Workflow;\n runId: string;\n inputData: TInput;\n initialState?: any;\n }): Promise {\n // 1. Initialize execution context\n // 2. Iterate through steps sequentially\n // 3. Execute each step with context\n // 4. Persist state after each step\n // 5. Return final output\n }\n \n private async executeStep(\n step: Step,\n context: ExecutionContext,\n inputData: any\n ): Promise;\n}\n```\n\n---\n\n#### 4. **State Persistence** (`storage.ts`)\n\n```typescript\ninterface WorkflowSnapshot {\n runId: string;\n workflowId: string;\n status: 'running' | 'success' | 'failed';\n executionPath: number[];\n stepResults: Record;\n state: Record;\n result?: any;\n error?: string;\n timestamp: number;\n}\n\nclass WorkflowStorage {\n constructor(private sqlite: typeof import('@std/sqlite'));\n\n async init(): Promise {\n await this.sqlite.execute(`\n CREATE TABLE IF NOT EXISTS wrkflw_workflow_runs (\n run_id TEXT PRIMARY KEY,\n workflow_id TEXT NOT NULL,\n status TEXT NOT NULL,\n execution_path TEXT,\n step_results TEXT,\n state TEXT,\n result TEXT,\n error TEXT,\n created_at INTEGER DEFAULT (unixepoch()),\n updated_at INTEGER DEFAULT (unixepoch())\n )\n `);\n }\n \n async saveSnapshot(snapshot: WorkflowSnapshot): Promise;\n async loadSnapshot(runId: string): Promise;\n async listRuns(workflowId?: string): Promise;\n}\n```\n\n---\n\n#### 5. **Run Management** (`run.ts`)\n\n```typescript\nclass WorkflowRun {\n constructor(\n private workflow: Workflow,\n private runId: string,\n private engine: ExecutionEngine,\n private storage: WorkflowStorage\n ) {}\n \n async start(params: {\n inputData: TInput;\n initialState?: any;\n }): Promise {\n return this.engine.execute({\n workflow: this.workflow,\n runId: this.runId,\n inputData: params.inputData,\n initialState: params.initialState\n });\n }\n}\n\n// On Workflow class\nclass Workflow {\n async createRun(runId?: string): Promise> {\n const id = runId || crypto.randomUUID();\n return new WorkflowRun(this, id, this.engine, this.storage);\n }\n}\n```\n\n---\n\n### Directory Structure\n\n```\nwrkflw/\n├── backend/\n│ ├── index.ts # Main entry point / exports\n│ ├── step.ts # Step definition & creation\n│ ├── workflow.ts # Workflow builder & class\n│ ├── engine.ts # Execution engine\n│ ├── storage.ts # SQLite persistence\n│ ├── run.ts # Run management\n│ └── types.ts # Shared TypeScript types\n├── examples/\n│ ├── email-workflow.ts # Example workflow\n│ ├── http-trigger.ts # HTTP trigger example\n│ └── cron-trigger.ts # Cron trigger example\n├── PLAN.md\n├── README.md\n└── deno.json\n```\n\n---\n\n## Implementation Tasks\n\n### Milestone 1: Type System & Builders\n\n- [ ] Define core types (`Step`, `Workflow`, `ExecutionContext`, `StepResult`)\n- [ ] Implement `createStep()` with Zod schema validation\n- [ ] Implement `createWorkflow()` with fluent builder\n- [ ] Add `.then()` method with type inference\n- [ ] Add `.commit()` to finalize workflow\n- [ ] Test type inference chain (compile-time checks)\n\n### Milestone 2: Execution Engine\n\n- [ ] Implement `ExecutionEngine` class\n- [ ] Sequential step execution loop\n- [ ] Build execution context with step results\n- [ ] Implement `getStepResult()` helper\n- [ ] Implement `getInitData()` helper\n- [ ] Error handling and result wrapping\n\n### Milestone 3: State Persistence\n\n- [ ] Create SQLite schema for workflow runs\n- [ ] Implement `WorkflowStorage` class\n- [ ] Add `saveSnapshot()` after each step\n- [ ] Add `loadSnapshot()` for debugging\n- [ ] Add `listRuns()` for observability\n\n### Milestone 4: Run Management\n\n- [ ] Implement `WorkflowRun` class\n- [ ] Add `createRun()` to Workflow\n- [ ] Add `start()` method\n- [ ] Generate/validate run IDs\n- [ ] Connect engine + storage\n\n### Milestone 5: Examples & Documentation\n\n- [ ] Create simple email workflow example\n- [ ] Create HTTP trigger val example\n- [ ] Create Cron trigger val example\n- [ ] Write README with usage examples\n- [ ] Document type inference patterns\n\n---\n\n## Example End-to-End Usage\n\n### 1. Define Workflow\n\n```typescript\n// backend/workflows/email-campaign.ts\nimport { createWorkflow, createStep } from \"../index.ts\";\nimport { z } from \"npm:zod\";\nimport { sqlite } from \"https://esm.town/v/std/sqlite\";\n\nconst fetchUser = createStep({\n id: \"fetch-user\",\n inputSchema: z.object({ userId: z.string() }),\n outputSchema: z.object({ name: z.string(), email: z.string() }),\n execute: async ({ inputData }) => {\n const result = await sqlite.execute({\n sql: \"SELECT name, email FROM users WHERE id = ?\",\n args: [inputData.userId],\n });\n return {\n name: result.rows[0][0] as string,\n email: result.rows[0][1] as string,\n };\n },\n});\n\nconst generateEmail = createStep({\n id: \"generate-email\",\n inputSchema: z.object({ name: z.string(), email: z.string() }),\n outputSchema: z.object({ subject: z.string(), body: z.string() }),\n execute: async ({ inputData }) => {\n return {\n subject: `Hello ${inputData.name}!`,\n body: `Welcome to our service, ${inputData.name}!`,\n };\n },\n});\n\nconst sendEmail = createStep({\n id: \"send-email\",\n inputSchema: z.object({\n email: z.string(),\n subject: z.string(),\n body: z.string(),\n }),\n outputSchema: z.object({ sent: z.boolean(), messageId: z.string() }),\n execute: async ({ inputData, getStepResult }) => {\n // Access previous step results\n const user = getStepResult(fetchUser);\n \n // Send email logic\n const result = await sendEmailAPI({\n to: user.email,\n subject: inputData.subject,\n body: inputData.body,\n });\n \n return { sent: true, messageId: result.id };\n },\n});\n\nexport const emailCampaign = createWorkflow({\n id: \"email-campaign\",\n inputSchema: z.object({ userId: z.string() }),\n outputSchema: z.object({ sent: z.boolean(), messageId: z.string() }),\n})\n .then(fetchUser)\n .then(generateEmail)\n .then(sendEmail)\n .commit();\n```\n\n### 2. HTTP Trigger\n\n```typescript\n// backend/triggers/http.ts\nimport { emailCampaign } from \"../workflows/email-campaign.ts\";\n\nexport default async function(req: Request) {\n const url = new URL(req.url);\n const userId = url.searchParams.get(\"userId\");\n \n if (!userId) {\n return Response.json({ error: \"Missing userId\" }, { status: 400 });\n }\n \n // Create and start a run\n const run = await emailCampaign.createRun();\n const result = await run.start({ inputData: { userId } });\n \n return Response.json({\n success: true,\n runId: run.runId,\n result\n });\n}\n```\n\n### 3. Cron Trigger\n\n```typescript\n// backend/triggers/cron.ts\nimport { emailCampaign } from \"../workflows/email-campaign.ts\";\nimport { sqlite } from \"https://esm.town/v/std/sqlite\";\n\nexport default async function(interval: Interval) {\n // Get users who signed up today\n const users = await sqlite.execute({\n sql: `SELECT user_id FROM users \n WHERE date(created_at) = date('now')`,\n });\n \n // Run workflow for each user\n for (const row of users.rows) {\n const userId = row[0] as string;\n const run = await emailCampaign.createRun();\n \n try {\n await run.start({ inputData: { userId } });\n console.log(`Sent welcome email to ${userId}`);\n } catch (error) {\n console.error(`Failed to send to ${userId}:`, error);\n }\n }\n}\n```\n\n---\n\n## Technical Decisions\n\n### Type Inference Strategy\n\nFollowing Mastra's approach:\n- Use generic type parameters on builders\n- Thread `TPrevOutput` through `.then()` chain\n- Each step's output becomes next step's input type\n- Zod schemas provide runtime validation + type extraction\n\n### State Management (Phase 1)\n\n**Simplified approach for MVP:**\n- SQLite for all workflow state (no blob storage)\n- JSON serialization for step results\n- Snapshot after each step execution\n- Single database transaction per step\n\n### Error Handling (Phase 1)\n\n**Basic error wrapping:**\n- Try/catch around step execution\n- Capture error in `StepResult`\n- Persist failed state to SQLite\n- Throw error to caller (no retry logic yet)\n\n### Execution Model (Phase 1)\n\n**Synchronous execution:**\n- Run executes all steps in single Val Town invocation\n- No suspend/resume (Phase 2)\n- No parallel execution (Phase 2)\n- Sequential only with `.then()`\n\n---\n\n## Future Extensions (Post-MVP)\n\n### Phase 2: Advanced Control Flow\n- `.parallel([step1, step2])` - Concurrent execution\n- `.branch([[condition, stepA], ...])` - Conditional branching\n- `.map({ ... })` - Data transformation between steps\n\n### Phase 3: Durability\n- `.sleep(duration)` - Time-based delays via Cron\n- Suspend/resume for long-running workflows\n- Restart from last checkpoint\n\n### Phase 4: Advanced Features\n- `.foreach(step, { concurrency })` - Array iteration\n- Nested workflows (workflows as steps)\n- Retry logic and exponential backoff\n- Workflow-level state (beyond step results)\n\n### Phase 5: Observability\n- Event streaming for step progress\n- Run history and debugging UI\n- Time-travel debugging\n- Execution visualizations\n\n---\n\n## Success Criteria\n\n**Phase 1 MVP is complete when:**\n1. ✅ Can define strongly-typed steps with Zod schemas\n2. ✅ Can chain steps with `.then()` and type inference works\n3. ✅ Can execute workflows and get typed output\n4. ✅ Step results are accessible via `getStepResult()`\n5. ✅ Workflow state persists to SQLite after each step\n6. ✅ Can trigger workflows from HTTP and Cron vals\n7. ✅ Examples demonstrate personal automation use cases\n\n---\n\n## Open Questions\n\n1. **Validation**: Should we validate schemas at runtime (dev mode) or trust TypeScript?\n2. **Errors**: Should failed steps stop workflow or allow continuation?\n3. **Logging**: Use `console.log` or build custom logger?\n4. **Testing**: Add test examples or just examples?\n\n---\n\n## Next Steps\n\n1. Set up project structure\n2. Implement type system (`types.ts`)\n3. Build `createStep()` function\n4. Build `createWorkflow()` builder\n5. Test type inference manually\n6. Implement execution engine\n7. Add SQLite storage\n8. Create examples\n\n---\n\n**Ready to start building!** 🚀\n",[],[321],"c5553a96-d1c8-11f0-8d7b-42dde27851f2","MIGRATION.md","019aee26-23ac-778a-bcd7-15512bf7dee0",["D",1764932068219],"# Database Migration Guide\n\n## Table Naming Normalization\n\nAll tables in the wrkflw library now use the `wrkflw_` prefix for consistency and namespace clarity. This guide explains how to migrate existing data.\n\n## Affected Tables\n\n- `workflow_runs` → `wrkflw_workflow_runs`\n\n## Automatic Migration\n\nThe library includes automatic migration functionality that runs during initialization. When you call `WorkflowStorage.init()`, the migration will:\n\n1. Check if the old `workflow_runs` table exists\n2. Create the new `wrkflw_workflow_runs` table if it doesn't exist\n3. Copy all data from the old table to the new table\n4. Preserve the old table for safety (no data loss)\n\n### How It Works\n\nThe migration is **automatic and safe**:\n\n```typescript\nimport { WorkflowStorage } from \"./backend/storage.ts\";\n\nconst storage = new WorkflowStorage();\nawait storage.init(); // Migration happens automatically here\n```\n\nBy default, the old table is **not deleted** to ensure data safety. Both tables will exist after migration.\n\n## Manual Migration Options\n\n### Option 1: Using the Migration API\n\nIf you want more control over the migration process:\n\n```typescript\nimport { runMigrations, migrateWorkflowRunsTable } from \"./backend/migrations.ts\";\n\n// Migrate with old table preserved (recommended)\nconst result1 = await migrateWorkflowRunsTable(false);\nconsole.log(result1.message);\nconsole.log(`Migrated ${result1.rowCount} rows`);\n\n// Migrate and drop old table (use with caution)\nconst result2 = await migrateWorkflowRunsTable(true);\n```\n\n### Option 2: Using SQL Directly\n\nGenerate a SQL migration script for manual execution:\n\n```typescript\nimport { generateMigrationSQL } from \"./backend/migrations.ts\";\n\n// Generate SQL without dropping old table\nconst sql = generateMigrationSQL(false);\nconsole.log(sql);\n\n// Or generate SQL with old table cleanup\nconst sql = generateMigrationSQL(true);\nconsole.log(sql);\n```\n\nYou can then execute this SQL manually in your SQLite database.\n\n**Note**: The generated SQL script assumes you have an existing `workflow_runs` table to migrate from. If the table doesn't exist (fresh installation), the data migration step (Step 3) will fail, which is expected - you can create just the new table structure without the migration step.\n\n### Example SQL Migration Script\n\n```sql\n-- Create new table with wrkflw_ prefix\nCREATE TABLE IF NOT EXISTS wrkflw_workflow_runs (\n run_id TEXT PRIMARY KEY,\n workflow_id TEXT NOT NULL,\n status TEXT NOT NULL,\n execution_path TEXT,\n step_results TEXT,\n state TEXT,\n input_data TEXT,\n result TEXT,\n error TEXT,\n created_at INTEGER DEFAULT (unixepoch()),\n updated_at INTEGER DEFAULT (unixepoch())\n);\n\n-- Create indexes\nCREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_workflow_id\nON wrkflw_workflow_runs(workflow_id);\n\nCREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_status\nON wrkflw_workflow_runs(status);\n\n-- Copy data from old table\n-- NOTE: This step will fail if 'workflow_runs' table doesn't exist.\n-- For fresh installations, this is expected - skip this step.\nINSERT OR IGNORE INTO wrkflw_workflow_runs (\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error,\n created_at, updated_at\n)\nSELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error,\n created_at, updated_at\nFROM workflow_runs;\n\n-- Optional: Drop old table (only after verifying migration succeeded)\n-- DROP TABLE workflow_runs;\n-- DROP INDEX IF EXISTS idx_workflow_runs_workflow_id;\n-- DROP INDEX IF EXISTS idx_workflow_runs_status;\n```\n\n## Migration Safety\n\nThe migration process is designed to be **safe and non-destructive**:\n\n1. **No data loss**: Old table is preserved by default\n2. **Idempotent**: Can be run multiple times safely\n3. **Uses INSERT OR IGNORE**: Won't overwrite existing data in new table\n4. **Atomic operations**: Each step is a complete SQL transaction\n\n## Cleaning Up Old Tables\n\nAfter verifying that your application works correctly with the new table:\n\n1. Check that all data was migrated successfully\n2. Verify your application is using the new table\n3. Optionally, drop the old table manually:\n\n```typescript\nimport { sqlite } from \"https://esm.town/v/std/sqlite\";\n\n// Only do this after thorough verification!\nawait sqlite.execute(\"DROP TABLE IF EXISTS workflow_runs\");\nawait sqlite.execute(\"DROP INDEX IF EXISTS idx_workflow_runs_workflow_id\");\nawait sqlite.execute(\"DROP INDEX IF EXISTS idx_workflow_runs_status\");\n```\n\nOr using the migration API:\n\n```typescript\nimport { migrateWorkflowRunsTable } from \"./backend/migrations.ts\";\n\n// This will drop the old table\nawait migrateWorkflowRunsTable(true);\n```\n\n## Troubleshooting\n\n### Migration doesn't run\n\n- Ensure you're calling `WorkflowStorage.init()` before using storage\n- Check that you have the latest version with migration support\n\n### Data appears missing\n\n- Check if data is in the old table: `SELECT COUNT(*) FROM workflow_runs;`\n- Check if data is in the new table: `SELECT COUNT(*) FROM wrkflw_workflow_runs;`\n- The migration uses `INSERT OR IGNORE`, so existing records won't be duplicated\n\n### Performance concerns\n\n- The migration runs only once during initialization\n- For very large datasets (10,000+ rows), consider running migration during off-peak hours\n- The migration is fast for typical workflow run counts\n\n## Questions?\n\nIf you encounter any issues during migration, please open an issue on the GitHub repository.\n",[],[328],"c55532c6-d1c8-11f0-8d7b-42dde27851f2","FRONTEND_COMPARISON.md","019aee26-23ac-778a-bcd7-108d628cafce",["D",1764932068219],"# Frontend Visualization Comparison\n\nThis document explains the two different frontend implementations for the workflow visualizer and when to use each.\n\n## Overview\n\nThere are two versions of the workflow visualizer:\n\n1. **Local ReactFlow Frontend** - Full-featured interactive flow diagram for local development\n2. **Val Town Frontend** - Simplified embedded version for Val Town HTTP vals\n\n---\n\n## 1. Local ReactFlow Frontend\n\n**Location:** `/frontend/` directory\n\n### Technology Stack\n- **React** 18+ with TypeScript\n- **ReactFlow** - Interactive node-based UI library\n- **Vite** - Build tool and dev server\n- **CSS** - Modular component stylesheets\n\n### Features\n- ✨ **Interactive Flow Diagram** - Drag, zoom, and pan canvas\n- 🎨 **Custom Step Nodes** with rich visualizations:\n - Step number badges (#1, #2, etc.)\n - Status icons (✓ success, ✗ failed, ● running, ○ pending)\n - Expandable output/error displays\n - Animated spinners for running steps\n- 🔗 **Animated Edges** - Color-coded connections between steps\n - Green for successful steps\n - Red for failed steps\n - Blue animated edges for running steps\n - Gray for pending steps\n- 🗺️ **MiniMap & Controls** - Easy navigation for large workflows\n- 📊 **Visual Graph Layout** - Automatic vertical step arrangement\n- 🎯 **Start/End Nodes** - Clear workflow boundaries\n\n### Visual Style\n- **Light theme** - White background (#FFFFFF)\n- **Material Design colors** - Green/red/blue status indicators\n- **Professional layout** - Clean, spacious design\n\n### Use Cases\n- **Local development** - Full-featured workflow development\n- **Desktop applications** - Rich visualization needs\n- **Complex workflows** - Many steps or branches\n- **Interactive debugging** - Detailed step inspection\n\n### Setup & Running\n```bash\ncd frontend\nnpm install\nnpm run dev\n# Visit http://localhost:3000\n```\n\nThe frontend expects the API server to be running separately on port 8000.\n\n### Architecture\n```\nfrontend/\n├── src/\n│ ├── App.tsx # Main app with sidebar\n│ ├── components/\n│ │ ├── WorkflowVisualizer.tsx # ReactFlow canvas\n│ │ └── StepNode.tsx # Custom step node component\n│ ├── types.ts # TypeScript interfaces\n│ └── *.css # Component styles\n├── index.html\n└── package.json\n```\n\n---\n\n## 2. Val Town Frontend\n\n**Location:** `examples/api-server-example.http.ts` (embedded)\n\n### Technology Stack\n- **React** 18.2.0 from CDN (esm.sh)\n- **Plain HTML/CSS** - No build step required\n- **Import Maps** - Modern ESM imports\n- **Inline JavaScript** - Single-file deployment\n\n### Features\n- 📝 **List-based Layout** - Simple vertical step list\n- 🎨 **Color-coded Borders** - Visual step status indicators\n - Blue border for pending steps\n - Green border for completed steps\n - Red border for failed steps\n- 🏷️ **Status Badges** - Clear running/success/failed indicators\n- 📊 **Step Details** - Expandable output and error information\n- 🔄 **Auto-refresh** - Optional automatic run updates\n- 📱 **Responsive** - Works on all screen sizes\n\n### Visual Style\n- **Dark theme** - Near-black background (#0a0a0a)\n- **Modern aesthetic** - Minimalist design\n- **Compact layout** - Efficient use of space\n\n### Use Cases\n- **Val Town deployment** - Self-contained HTTP vals\n- **Quick prototyping** - No build step needed\n- **Simple workflows** - Sequential step execution\n- **Embedded views** - Lightweight visualization\n\n### Deployment\n\n**For Val Town:**\nSimply copy the contents of `examples/api-server-example.http.ts` to a Val Town HTTP val. Everything (API + frontend) is included in a single file.\n\n**For Local Development:**\nRun the local server version for testing without deploying:\n```bash\ndeno run --allow-all examples/api-server-example-local.ts\n# Open http://localhost:8000\n```\n\nThis gives you the same embedded visualizer but running locally on port 8000.\n\n### Architecture\n```\napi-server-example.http.ts # Val Town deployment version\napi-server-example-local.ts # Local development version\n├── Workflow definitions\n├── Hono API routes\n└── Embedded HTML with:\n ├── \n\n\n

\n\n \n\n \n \n\n\n `;\n return c.html(html);\n});\n\n// ============================================================================\n// Export for Val Town HTTP\n// ============================================================================\n\nexport default app.fetch; // This is the entry point for HTTP vals",[],[483,515],"c554c278-d1c8-11f0-8d7b-42dde27851f2","api-server-example-local.ts","019aee26-23ab-7319-83f5-bb96aec4cd9b",["D",1764932068219],"/**\n * Example: API Server for Workflow Visualization (Local Deno Server)\n *\n * This example demonstrates how to run a local API server that:\n * 1. Sets up workflows with an API server using Hono\n * 2. Serves workflow data via REST API\n * 3. Creates and monitors workflow runs\n * 4. Serves a React-based frontend visualizer\n *\n * Run this example locally:\n * deno run --allow-all examples/api-server-example-local.ts\n *\n * Then open http://localhost:8000 to see the visualizer\n *\n * Note: For Val Town deployment, use api-server-example.http.ts instead\n */\n\nimport { Hono } from \"https://esm.sh/hono@4\";\nimport { z } from \"npm:zod@^3.23\";\nimport { createStep, createWorkflow } from \"../backend/index.ts\";\nimport { WorkflowStorage } from \"../backend/storage.ts\";\n\n// Initialize storage\nconst storage = new WorkflowStorage();\nawait storage.init();\n\n// Create Hono app\nconst app = new Hono();\n\n// Unwrap Hono errors to see original error details\napp.onError((err, c) => {\n\tthrow err;\n});\n\n// In-memory workflow registry\nconst workflows = new Map();\n\n// ============================================================================\n// Example Workflow 1: Data Processing Pipeline\n// ============================================================================\n\nconst fetchData = createStep({\n\tid: \"fetch-data\",\n\tdescription: \"Fetch data from external API\",\n\tinputSchema: z.object({ url: z.string() }),\n\toutputSchema: z.object({ data: z.array(z.record(z.unknown())) }),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(`Fetching data from ${inputData.url}...`);\n\n\t\t// Simulate API call\n\t\tawait new Promise((resolve) => setTimeout(resolve, 1000));\n\n\t\treturn {\n\t\t\tdata: [\n\t\t\t\t{ id: 1, name: \"Alice\", score: 95 },\n\t\t\t\t{ id: 2, name: \"Bob\", score: 87 },\n\t\t\t\t{ id: 3, name: \"Charlie\", score: 92 },\n\t\t\t],\n\t\t};\n\t},\n});\n\nconst processData = createStep({\n\tid: \"process-data\",\n\tdescription: \"Transform and filter data\",\n\tinputSchema: z.object({ data: z.array(z.record(z.unknown())) }),\n\toutputSchema: z.object({ processed: z.array(z.record(z.unknown())) }),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(\"Processing data...\");\n\n\t\t// Simulate processing\n\t\tawait new Promise((resolve) => setTimeout(resolve, 800));\n\n\t\tconst processed = inputData.data\n\t\t\t.filter((item: any) => item.score >= 90)\n\t\t\t.map((item: any) => ({\n\t\t\t\t...item,\n\t\t\t\tgrade: \"A\",\n\t\t\t\ttimestamp: Date.now(),\n\t\t\t}));\n\n\t\treturn { processed };\n\t},\n});\n\nconst saveResults = createStep({\n\tid: \"save-results\",\n\tdescription: \"Save processed data to database\",\n\tinputSchema: z.object({ processed: z.array(z.record(z.unknown())) }),\n\toutputSchema: z.object({ saved: z.number(), success: z.boolean() }),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(\"Saving results...\");\n\n\t\t// Simulate database save\n\t\tawait new Promise((resolve) => setTimeout(resolve, 600));\n\n\t\treturn {\n\t\t\tsaved: inputData.processed.length,\n\t\t\tsuccess: true,\n\t\t};\n\t},\n});\n\nconst dataProcessingWorkflow = createWorkflow({\n\tid: \"data-processing\",\n\tdescription: \"Fetch, process, and save data pipeline\",\n\tinputSchema: z.object({ url: z.string() }),\n\toutputSchema: z.object({ saved: z.number(), success: z.boolean() }),\n})\n\t.then(fetchData)\n\t.then(processData)\n\t.then(saveResults)\n\t.commit();\n\nworkflows.set(dataProcessingWorkflow.id, {\n\tworkflow: dataProcessingWorkflow,\n\texampleInput: { url: \"https://api.example.com/data\" },\n});\n\n// ============================================================================\n// Example Workflow 2: Email Campaign\n// ============================================================================\n\nconst fetchUsers = createStep({\n\tid: \"fetch-users\",\n\tdescription: \"Fetch users from database\",\n\tinputSchema: z.object({ segment: z.string() }),\n\toutputSchema: z.object({\n\t\tusers: z.array(\n\t\t\tz.object({ id: z.string(), email: z.string(), name: z.string() }),\n\t\t),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(`Fetching users in segment: ${inputData.segment}`);\n\t\tawait new Promise((resolve) => setTimeout(resolve, 700));\n\n\t\treturn {\n\t\t\tusers: [\n\t\t\t\t{ id: \"1\", email: \"alice@example.com\", name: \"Alice\" },\n\t\t\t\t{ id: \"2\", email: \"bob@example.com\", name: \"Bob\" },\n\t\t\t\t{ id: \"3\", email: \"charlie@example.com\", name: \"Charlie\" },\n\t\t\t],\n\t\t};\n\t},\n});\n\nconst generateEmails = createStep({\n\tid: \"generate-emails\",\n\tdescription: \"Generate personalized email content\",\n\tinputSchema: z.object({\n\t\tusers: z.array(\n\t\t\tz.object({ id: z.string(), email: z.string(), name: z.string() }),\n\t\t),\n\t}),\n\toutputSchema: z.object({\n\t\temails: z.array(\n\t\t\tz.object({ to: z.string(), subject: z.string(), body: z.string() }),\n\t\t),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(\"Generating personalized emails...\");\n\t\tawait new Promise((resolve) => setTimeout(resolve, 900));\n\n\t\tconst emails = inputData.users.map((user) => ({\n\t\t\tto: user.email,\n\t\t\tsubject: `Hello ${user.name}!`,\n\t\t\tbody: `Hi ${user.name}, we have exciting news for you...`,\n\t\t}));\n\n\t\treturn { emails };\n\t},\n});\n\nconst sendEmails = createStep({\n\tid: \"send-emails\",\n\tdescription: \"Send emails via email service\",\n\tinputSchema: z.object({\n\t\temails: z.array(\n\t\t\tz.object({ to: z.string(), subject: z.string(), body: z.string() }),\n\t\t),\n\t}),\n\toutputSchema: z.object({ sent: z.number(), failed: z.number() }),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(\"Sending emails...\");\n\t\tawait new Promise((resolve) => setTimeout(resolve, 1200));\n\n\t\treturn {\n\t\t\tsent: inputData.emails.length,\n\t\t\tfailed: 0,\n\t\t};\n\t},\n});\n\nconst emailCampaignWorkflow = createWorkflow({\n\tid: \"email-campaign\",\n\tdescription: \"Automated email campaign workflow\",\n\tinputSchema: z.object({ segment: z.string() }),\n\toutputSchema: z.object({ sent: z.number(), failed: z.number() }),\n})\n\t.then(fetchUsers)\n\t.then(generateEmails)\n\t.then(sendEmails)\n\t.commit();\n\nworkflows.set(emailCampaignWorkflow.id, {\n\tworkflow: emailCampaignWorkflow,\n\texampleInput: { segment: \"premium-users\" },\n});\n\n// ============================================================================\n// Example Workflow 3: Report Generation (with potential failure)\n// ============================================================================\n\nconst collectMetrics = createStep({\n\tid: \"collect-metrics\",\n\tdescription: \"Collect performance metrics\",\n\tinputSchema: z.object({ period: z.string() }),\n\toutputSchema: z.object({ metrics: z.record(z.number()) }),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(`Collecting metrics for period: ${inputData.period}`);\n\t\tawait new Promise((resolve) => setTimeout(resolve, 500));\n\n\t\treturn {\n\t\t\tmetrics: {\n\t\t\t\tusers: 1250,\n\t\t\t\trevenue: 45600,\n\t\t\t\tconversions: 234,\n\t\t\t},\n\t\t};\n\t},\n});\n\nconst analyzeMetrics = createStep({\n\tid: \"analyze-metrics\",\n\tdescription: \"Analyze metrics and generate insights\",\n\tinputSchema: z.object({ metrics: z.record(z.number()) }),\n\toutputSchema: z.object({\n\t\tinsights: z.array(z.string()),\n\t\ttrends: z.record(z.string()),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(\"Analyzing metrics...\");\n\t\tawait new Promise((resolve) => setTimeout(resolve, 800));\n\n\t\t// Randomly fail sometimes to demonstrate error handling\n\t\tif (Math.random() < 0.3) {\n\t\t\tthrow new Error(\"Analysis service temporarily unavailable\");\n\t\t}\n\n\t\treturn {\n\t\t\tinsights: [\n\t\t\t\t\"User growth increased by 15%\",\n\t\t\t\t\"Revenue trending upward\",\n\t\t\t\t\"Conversion rate improved\",\n\t\t\t],\n\t\t\ttrends: {\n\t\t\t\tusers: \"increasing\",\n\t\t\t\trevenue: \"increasing\",\n\t\t\t\tconversions: \"stable\",\n\t\t\t},\n\t\t};\n\t},\n});\n\nconst generateReport = createStep({\n\tid: \"generate-report\",\n\tdescription: \"Generate PDF report\",\n\tinputSchema: z.object({\n\t\tinsights: z.array(z.string()),\n\t\ttrends: z.record(z.string()),\n\t}),\n\toutputSchema: z.object({ reportUrl: z.string(), pages: z.number() }),\n\texecute: async ({ inputData }) => {\n\t\tconsole.log(\"Generating report...\");\n\t\tawait new Promise((resolve) => setTimeout(resolve, 1000));\n\n\t\treturn {\n\t\t\treportUrl: `https://reports.example.com/report-${Date.now()}.pdf`,\n\t\t\tpages: 12,\n\t\t};\n\t},\n});\n\nconst reportWorkflow = createWorkflow({\n\tid: \"report-generation\",\n\tdescription: \"Monthly performance report generation\",\n\tinputSchema: z.object({ period: z.string() }),\n\toutputSchema: z.object({ reportUrl: z.string(), pages: z.number() }),\n})\n\t.then(collectMetrics)\n\t.then(analyzeMetrics)\n\t.then(generateReport)\n\t.commit();\n\nworkflows.set(reportWorkflow.id, {\n\tworkflow: reportWorkflow,\n\texampleInput: { period: \"2024-11\" },\n});\n\n// ============================================================================\n// API Routes\n// ============================================================================\n\n// GET /api/workflows - List all workflows\napp.get(\"/api/workflows\", (c) => {\n\tconst workflowList = Array.from(workflows.entries()).map(\n\t\t([id, { workflow, exampleInput }]) => ({\n\t\t\tid: workflow.id,\n\t\t\tdescription: workflow.config.description,\n\t\t\tsteps: workflow.stepFlow.map((entry: any) => ({\n\t\t\t\tid: entry.step.id,\n\t\t\t\tdescription: entry.step.description,\n\t\t\t})),\n\t\t\texampleInput,\n\t\t}),\n\t);\n\treturn c.json(workflowList);\n});\n\n// GET /api/workflows/:id - Get workflow details\napp.get(\"/api/workflows/:id\", (c) => {\n\tconst workflowId = c.req.param(\"id\");\n\tconst entry = workflows.get(workflowId);\n\n\tif (!entry) {\n\t\treturn c.json({ error: \"Workflow not found\" }, 404);\n\t}\n\n\tconst { workflow, exampleInput } = entry;\n\treturn c.json({\n\t\tid: workflow.id,\n\t\tdescription: workflow.config.description,\n\t\tsteps: workflow.stepFlow.map((entry: any) => ({\n\t\t\tid: entry.step.id,\n\t\t\tdescription: entry.step.description,\n\t\t})),\n\t\texampleInput,\n\t});\n});\n\n// GET /api/workflows/:id/runs - List runs for a workflow\napp.get(\"/api/workflows/:id/runs\", async (c) => {\n\tconst workflowId = c.req.param(\"id\");\n\tconst snapshots = await storage.listRuns(workflowId);\n\tconst runs = snapshots.map((snapshot) => ({\n\t\trunId: snapshot.runId,\n\t\tworkflowId: snapshot.workflowId,\n\t\tstatus: snapshot.status,\n\t\texecutionPath: snapshot.executionPath,\n\t\tstepResults: snapshot.stepResults,\n\t\tinputData: snapshot.inputData,\n\t\tresult: snapshot.result,\n\t\terror: snapshot.error,\n\t\tcreatedAt: snapshot.timestamp,\n\t\tupdatedAt: snapshot.timestamp,\n\t}));\n\treturn c.json(runs);\n});\n\n// POST /api/workflows/:id/runs - Create a new run\napp.post(\"/api/workflows/:id/runs\", async (c) => {\n\tconst workflowId = c.req.param(\"id\");\n\tconst entry = workflows.get(workflowId);\n\n\tif (!entry) {\n\t\treturn c.json({ error: \"Workflow not found\" }, 404);\n\t}\n\n\tconst body = await c.req.json();\n\tconst { inputData } = body;\n\n\t// Create and start the run\n\tconst run = await entry.workflow.createRun();\n\n\t// Start execution in the background\n\trun.start({ inputData }).catch((error: Error) => {\n\t\tconsole.error(`Run ${run.runId} failed:`, error);\n\t});\n\n\t// Return the initial run state\n\tconst snapshot = await storage.loadSnapshot(run.runId);\n\tif (!snapshot) {\n\t\treturn c.json({ error: \"Failed to create run\" }, 500);\n\t}\n\n\treturn c.json(\n\t\t{\n\t\t\trunId: snapshot.runId,\n\t\t\tworkflowId: snapshot.workflowId,\n\t\t\tstatus: snapshot.status,\n\t\t\texecutionPath: snapshot.executionPath,\n\t\t\tstepResults: snapshot.stepResults,\n\t\t\tinputData: snapshot.inputData,\n\t\t\tresult: snapshot.result,\n\t\t\terror: snapshot.error,\n\t\t\tcreatedAt: snapshot.timestamp,\n\t\t\tupdatedAt: snapshot.timestamp,\n\t\t},\n\t\t201,\n\t);\n});\n\n// GET /api/runs/:runId - Get run details\napp.get(\"/api/runs/:runId\", async (c) => {\n\tconst runId = c.req.param(\"runId\");\n\tconst snapshot = await storage.loadSnapshot(runId);\n\n\tif (!snapshot) {\n\t\treturn c.json({ error: \"Run not found\" }, 404);\n\t}\n\n\treturn c.json({\n\t\trunId: snapshot.runId,\n\t\tworkflowId: snapshot.workflowId,\n\t\tstatus: snapshot.status,\n\t\texecutionPath: snapshot.executionPath,\n\t\tstepResults: snapshot.stepResults,\n\t\tinputData: snapshot.inputData,\n\t\tresult: snapshot.result,\n\t\terror: snapshot.error,\n\t\tcreatedAt: snapshot.timestamp,\n\t\tupdatedAt: snapshot.timestamp,\n\t});\n});\n\n// ============================================================================\n// Frontend\n// ============================================================================\n\napp.get(\"/\", (c) => {\n\tconst html = `\n\n\n \n \n Workflow Visualizer\n \n \n\n\n

\n\n \n\n \n \n\n\n `;\n\treturn c.html(html);\n});\n\n// ============================================================================\n// Start Local Server\n// ============================================================================\n\nconst PORT = 8000;\n\nconsole.log(\"\\n=======================================================\");\nconsole.log(\"Workflow API Server with Embedded Visualizer\");\nconsole.log(\"=======================================================\\n\");\nconsole.log(\"Registered workflows:\");\nconsole.log(\" - data-processing: Fetch, process, and save data\");\nconsole.log(\" - email-campaign: Automated email campaign\");\nconsole.log(\" - report-generation: Monthly report generation\\n\");\nconsole.log(\"API Endpoints:\");\nconsole.log(\" GET /api/workflows - List all workflows\");\nconsole.log(\" GET /api/workflows/:id - Get workflow details\");\nconsole.log(\" GET /api/workflows/:id/runs - List runs for workflow\");\nconsole.log(\" POST /api/workflows/:id/runs - Create new run\");\nconsole.log(\" GET /api/runs/:runId - Get run details\\n\");\nconsole.log(`Server starting on http://localhost:${PORT}`);\nconsole.log(`Open http://localhost:${PORT} to view the visualizer\\n`);\nconsole.log(\"=======================================================\\n\");\n\nDeno.serve({ port: PORT }, app.fetch);\n",[],[483,521],"c554f78e-d1c8-11f0-8d7b-42dde27851f2",["D",1764932068219],[238,239,240,241,242,243,244],[483],"c554be04-d1c8-11f0-8d7b-42dde27851f2","workflow.ts","019aee26-23ab-7319-83f5-b161e39b501a","019aee26-23ab-7319-83f5-b45aa4f27aaf",["D",1764932068219],"import type { z } from \"npm:zod@^3.23\";\nimport { ExecutionEngine } from \"./engine.ts\";\nimport { WorkflowRun } from \"./run.ts\";\nimport { WorkflowStorage } from \"./storage.ts\";\nimport type { Step, StepFlowEntry, WorkflowConfig } from \"./types.ts\";\nimport {\n\ttype VisualizationFormat,\n\ttype VisualizationOptions,\n\tvisualizeWorkflow,\n} from \"./visualize.ts\";\n\n/**\n * Workflow builder with fluent API and type inference\n */\nexport class WorkflowBuilder<\n\tTSteps extends Record>,\n\tTWorkflowId extends string,\n\tTInput,\n\tTPrevOutput,\n\tTState = unknown,\n> {\n\tconstructor(\n\t\tprivate config: WorkflowConfig,\n\t\tprivate steps: TSteps,\n\t\tprivate stepFlow: StepFlowEntry[],\n\t) {}\n\n\t/**\n\t * Add a sequential step to the workflow\n\t * The step's input must match the previous step's output\n\t */\n\tthen>(\n\t\tstep: TStep,\n\t): WorkflowBuilder<\n\t\tTSteps & Record,\n\t\tTWorkflowId,\n\t\tTInput,\n\t\tz.infer,\n\t\tTState\n\t> {\n\t\tconst newSteps = { ...this.steps, [step.id]: step } as TSteps &\n\t\t\tRecord;\n\t\tconst newStepFlow = [...this.stepFlow, { type: \"step\" as const, step }];\n\n\t\treturn new WorkflowBuilder(this.config, newSteps, newStepFlow);\n\t}\n\n\t/**\n\t * Finalize the workflow and make it executable\n\t */\n\tcommit(): Workflow {\n\t\treturn new Workflow(this.config, this.steps, this.stepFlow);\n\t}\n}\n\n/**\n * Executable workflow instance\n */\nexport class Workflow<\n\tTSteps extends Record>,\n\tTWorkflowId extends string,\n\tTInput,\n\tTOutput,\n\tTState = unknown,\n> {\n\tprivate engine: ExecutionEngine;\n\tprivate storage: WorkflowStorage;\n\n\tconstructor(\n\t\tpublic readonly config: WorkflowConfig<\n\t\t\tTWorkflowId,\n\t\t\tTInput,\n\t\t\tTOutput,\n\t\t\tTState\n\t\t>,\n\t\tpublic readonly steps: TSteps,\n\t\tpublic readonly stepFlow: StepFlowEntry[],\n\t) {\n\t\tthis.storage = new WorkflowStorage();\n\t\tthis.engine = new ExecutionEngine(this.storage);\n\t}\n\n\t/**\n\t * Initialize the workflow (creates database tables)\n\t */\n\tasync init(): Promise {\n\t\tawait this.storage.init();\n\t}\n\n\t/**\n\t * Create a new workflow run instance\n\t */\n\tasync createRun(\n\t\trunId?: string,\n\t): Promise> {\n\t\t// Initialize storage if not already done\n\t\tawait this.init();\n\n\t\tconst id = runId || crypto.randomUUID();\n\t\treturn new WorkflowRun(this, id, this.engine, this.storage);\n\t}\n\n\t/**\n\t * Get the workflow ID\n\t */\n\tget id(): TWorkflowId {\n\t\treturn this.config.id as TWorkflowId;\n\t}\n\n\t/**\n\t * Get the input schema\n\t */\n\tget inputSchema(): z.ZodSchema {\n\t\treturn this.config.inputSchema;\n\t}\n\n\t/**\n\t * Get the output schema\n\t */\n\tget outputSchema(): z.ZodSchema {\n\t\treturn this.config.outputSchema as z.ZodSchema;\n\t}\n\n\t/**\n\t * Get the state schema\n\t */\n\tget stateSchema(): z.ZodSchema | undefined {\n\t\treturn this.config.stateSchema;\n\t}\n\n\t/**\n\t * Generate a visual representation of this workflow\n\t *\n\t * @param format - The visualization format (default: \"mermaid\")\n\t * @param options - Visualization options\n\t * @returns Visualization string in the specified format\n\t *\n\t * @example\n\t * ```typescript\n\t * const diagram = workflow.visualize(\"mermaid\");\n\t * console.log(diagram);\n\t * ```\n\t */\n\tvisualize(\n\t\tformat: VisualizationFormat = \"mermaid\",\n\t\toptions: VisualizationOptions = {},\n\t): string {\n\t\treturn visualizeWorkflow(this, format, options);\n\t}\n}\n\n/**\n * Create a new workflow with strong typing\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({\n * id: 'email-campaign',\n * inputSchema: z.object({ userId: z.string() }),\n * outputSchema: z.object({ sent: z.boolean() })\n * })\n * .then(fetchUser)\n * .then(generateEmail)\n * .then(sendEmail)\n * .commit();\n * ```\n */\nexport function createWorkflow<\n\tTId extends string,\n\tTInput,\n\tTOutput,\n\tTState = unknown,\n>(\n\tconfig: WorkflowConfig,\n): WorkflowBuilder<{}, TId, TInput, TInput, TState> {\n\treturn new WorkflowBuilder(config, {}, []);\n}\n",[],[539,532],"backend","c554b558-d1c8-11f0-8d7b-42dde27851f2","visualize.ts","019aee26-23ab-7319-83f5-adc979ec66df",["D",1764932068219],"import type { Workflow } from \"./workflow.ts\";\nimport type { WorkflowSnapshot } from \"./types.ts\";\n\n/**\n * Visualization format types\n */\nexport type VisualizationFormat = \"mermaid\" | \"json\" | \"ascii\" | \"dot\";\n\n/**\n * Options for visualization\n */\nexport interface VisualizationOptions {\n\t/** Include schema information in visualization */\n\tincludeSchemas?: boolean;\n\t/** Include step descriptions */\n\tincludeDescriptions?: boolean;\n}\n\n/**\n * Options for execution visualization\n */\nexport interface ExecutionVisualizationOptions extends VisualizationOptions {\n\t/** Highlight the current/last executed step */\n\thighlightCurrentStep?: boolean;\n\t/** Show step execution status */\n\tshowStatus?: boolean;\n\t/** Show step results */\n\tshowResults?: boolean;\n}\n\n/**\n * Visualize a workflow's structure\n *\n * @param workflow - The workflow to visualize\n * @param format - The output format (default: \"mermaid\")\n * @param options - Visualization options\n * @returns Visualization string in the specified format\n *\n * @example\n * ```typescript\n * const diagram = visualizeWorkflow(workflow, \"mermaid\");\n * console.log(diagram);\n * ```\n */\nexport function visualizeWorkflow(\n\tworkflow: Workflow,\n\tformat: VisualizationFormat = \"mermaid\",\n\toptions: VisualizationOptions = {},\n): string {\n\tswitch (format) {\n\t\tcase \"mermaid\":\n\t\t\treturn generateMermaidWorkflow(workflow, options);\n\t\tcase \"json\":\n\t\t\tthrow new Error(\"JSON format not yet implemented\");\n\t\tcase \"ascii\":\n\t\t\tthrow new Error(\"ASCII format not yet implemented\");\n\t\tcase \"dot\":\n\t\t\tthrow new Error(\"DOT format not yet implemented\");\n\t\tdefault:\n\t\t\tthrow new Error(`Unknown visualization format: ${format}`);\n\t}\n}\n\n/**\n * Visualize a workflow execution with state information\n *\n * @param workflow - The workflow being executed\n * @param snapshot - The execution snapshot\n * @param format - The output format (default: \"mermaid\")\n * @param options - Visualization options\n * @returns Visualization string in the specified format\n *\n * @example\n * ```typescript\n * const snapshot = await run.getSnapshot();\n * if (snapshot) {\n * const diagram = visualizeExecution(workflow, snapshot, \"mermaid\", {\n * highlightCurrentStep: true,\n * showStatus: true\n * });\n * console.log(diagram);\n * }\n * ```\n */\nexport function visualizeExecution(\n\tworkflow: Workflow,\n\tsnapshot: WorkflowSnapshot,\n\tformat: VisualizationFormat = \"mermaid\",\n\toptions: ExecutionVisualizationOptions = {},\n): string {\n\tswitch (format) {\n\t\tcase \"mermaid\":\n\t\t\treturn generateMermaidExecution(workflow, snapshot, options);\n\t\tcase \"json\":\n\t\t\tthrow new Error(\"JSON format not yet implemented\");\n\t\tcase \"ascii\":\n\t\t\tthrow new Error(\"ASCII format not yet implemented\");\n\t\tcase \"dot\":\n\t\t\tthrow new Error(\"DOT format not yet implemented\");\n\t\tdefault:\n\t\t\tthrow new Error(`Unknown visualization format: ${format}`);\n\t}\n}\n\n/**\n * Generate Mermaid diagram for workflow structure\n */\nfunction generateMermaidWorkflow(\n\tworkflow: Workflow,\n\toptions: VisualizationOptions,\n): string {\n\tconst lines: string[] = [];\n\tconst { includeDescriptions = true } = options;\n\n\t// Start with graph definition\n\tlines.push(\"graph TD\");\n\n\t// Add workflow title/description as comment\n\tif (workflow.config.description) {\n\t\tlines.push(` %% ${workflow.config.description}`);\n\t}\n\n\t// Start node\n\tconst startLabel = workflow.config.id;\n\tlines.push(` Start([${startLabel}])`);\n\n\t// Generate step nodes\n\tconst stepIds = workflow.stepFlow.map((entry) => entry.step.id);\n\n\tstepIds.forEach((stepId, index) => {\n\t\tconst step = workflow.steps[stepId];\n\t\tif (!step) return;\n\n\t\tconst nodeId = sanitizeNodeId(stepId);\n\t\tlet label = stepId;\n\n\t\tif (includeDescriptions && step.description) {\n\t\t\tlabel = `${stepId}
${step.description}`;\n\t\t}\n\n\t\t// Create step node\n\t\tlines.push(` ${nodeId}[${label}]`);\n\n\t\t// Connect from Start to first step, or from previous step\n\t\tif (index === 0) {\n\t\t\tlines.push(` Start --> ${nodeId}`);\n\t\t} else {\n\t\t\tconst prevNodeId = sanitizeNodeId(stepIds[index - 1]);\n\t\t\tlines.push(` ${prevNodeId} --> ${nodeId}`);\n\t\t}\n\t});\n\n\t// End node\n\tlines.push(\" End([End])\");\n\tif (stepIds.length > 0) {\n\t\tconst lastNodeId = sanitizeNodeId(stepIds[stepIds.length - 1]);\n\t\tlines.push(` ${lastNodeId} --> End`);\n\t} else {\n\t\tlines.push(\" Start --> End\");\n\t}\n\n\t// Add styling\n\tlines.push(\"\");\n\tlines.push(\" classDef default fill:#e1f5ff,stroke:#01579b,stroke-width:2px\");\n\tlines.push(\" classDef startEnd fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px\");\n\tlines.push(\" class Start,End startEnd\");\n\n\treturn lines.join(\"\\n\");\n}\n\n/**\n * Generate Mermaid diagram for workflow execution with state\n */\nfunction generateMermaidExecution(\n\tworkflow: Workflow,\n\tsnapshot: WorkflowSnapshot,\n\toptions: ExecutionVisualizationOptions,\n): string {\n\tconst lines: string[] = [];\n\tconst {\n\t\tincludeDescriptions = true,\n\t\thighlightCurrentStep = true,\n\t\tshowStatus = true,\n\t\tshowResults = false,\n\t} = options;\n\n\t// Start with graph definition\n\tlines.push(\"graph TD\");\n\n\t// Add execution info as comment\n\tlines.push(` %% Workflow: ${workflow.config.id}`);\n\tlines.push(` %% Run ID: ${snapshot.runId}`);\n\tlines.push(` %% Status: ${snapshot.status}`);\n\tif (snapshot.error) {\n\t\tlines.push(` %% Error: ${snapshot.error}`);\n\t}\n\n\t// Start node\n\tconst startLabel = workflow.config.id;\n\tconst startStatus = snapshot.status !== \"running\" || snapshot.executionPath.length > 0\n\t\t? \"✓\"\n\t\t: \"●\";\n\tlines.push(` Start([${startStatus} ${startLabel}])`);\n\n\t// Generate step nodes with execution state\n\tconst stepIds = workflow.stepFlow.map((entry) => entry.step.id);\n\tconst executedStepIndices = new Set(snapshot.executionPath);\n\tconst lastExecutedIndex = snapshot.executionPath.length > 0\n\t\t? snapshot.executionPath[snapshot.executionPath.length - 1]\n\t\t: -1;\n\n\tstepIds.forEach((stepId, index) => {\n\t\tconst step = workflow.steps[stepId];\n\t\tif (!step) return;\n\n\t\tconst nodeId = sanitizeNodeId(stepId);\n\t\tconst stepResult = snapshot.stepResults[stepId];\n\t\tconst isExecuted = executedStepIndices.has(index);\n\t\tconst isCurrent = highlightCurrentStep && index === lastExecutedIndex &&\n\t\t\tsnapshot.status === \"running\";\n\n\t\t// Build label\n\t\tlet label = stepId;\n\t\tlet statusIcon = \"\";\n\n\t\tif (showStatus && stepResult) {\n\t\t\tswitch (stepResult.status) {\n\t\t\t\tcase \"success\":\n\t\t\t\t\tstatusIcon = \"✓ \";\n\t\t\t\t\tbreak;\n\t\t\t\tcase \"failed\":\n\t\t\t\t\tstatusIcon = \"✗ \";\n\t\t\t\t\tbreak;\n\t\t\t\tcase \"running\":\n\t\t\t\t\tstatusIcon = \"● \";\n\t\t\t\t\tbreak;\n\t\t\t}\n\t\t} else if (isExecuted) {\n\t\t\tstatusIcon = \"✓ \";\n\t\t} else if (isCurrent) {\n\t\t\tstatusIcon = \"● \";\n\t\t}\n\n\t\tlabel = `${statusIcon}${stepId}`;\n\n\t\tif (includeDescriptions && step.description) {\n\t\t\tlabel += `
${step.description}`;\n\t\t}\n\n\t\tif (showResults && stepResult && stepResult.status === \"failed\") {\n\t\t\tconst errorMsg = stepResult.error.message.substring(0, 50);\n\t\t\tlabel += `
${errorMsg}`;\n\t\t}\n\n\t\t// Create step node\n\t\tlines.push(` ${nodeId}[${label}]`);\n\n\t\t// Connect from Start to first step, or from previous step\n\t\tif (index === 0) {\n\t\t\tlines.push(` Start --> ${nodeId}`);\n\t\t} else {\n\t\t\tconst prevNodeId = sanitizeNodeId(stepIds[index - 1]);\n\t\t\tlines.push(` ${prevNodeId} --> ${nodeId}`);\n\t\t}\n\t});\n\n\t// End node\n\tconst endStatus = snapshot.status === \"success\"\n\t\t? \"✓\"\n\t\t: snapshot.status === \"failed\"\n\t\t? \"✗\"\n\t\t: \"○\";\n\tlines.push(` End([${endStatus} End])`);\n\tif (stepIds.length > 0) {\n\t\tconst lastNodeId = sanitizeNodeId(stepIds[stepIds.length - 1]);\n\t\tlines.push(` ${lastNodeId} --> End`);\n\t} else {\n\t\tlines.push(\" Start --> End\");\n\t}\n\n\t// Add styling based on execution state\n\tlines.push(\"\");\n\tlines.push(\" classDef default fill:#e1f5ff,stroke:#01579b,stroke-width:2px\");\n\tlines.push(\" classDef startEnd fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px\");\n\tlines.push(\" classDef success fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px\");\n\tlines.push(\" classDef failed fill:#ffcdd2,stroke:#c62828,stroke-width:2px\");\n\tlines.push(\" classDef running fill:#fff9c4,stroke:#f57f17,stroke-width:3px\");\n\tlines.push(\" classDef pending fill:#f5f5f5,stroke:#9e9e9e,stroke-width:1px\");\n\n\t// Apply classes\n\tlines.push(\" class Start,End startEnd\");\n\n\tstepIds.forEach((stepId, index) => {\n\t\tconst nodeId = sanitizeNodeId(stepId);\n\t\tconst stepResult = snapshot.stepResults[stepId];\n\t\tconst isExecuted = executedStepIndices.has(index);\n\t\tconst isCurrent = highlightCurrentStep && index === lastExecutedIndex &&\n\t\t\tsnapshot.status === \"running\";\n\n\t\tif (isCurrent) {\n\t\t\tlines.push(` class ${nodeId} running`);\n\t\t} else if (stepResult) {\n\t\t\tif (stepResult.status === \"success\") {\n\t\t\t\tlines.push(` class ${nodeId} success`);\n\t\t\t} else if (stepResult.status === \"failed\") {\n\t\t\t\tlines.push(` class ${nodeId} failed`);\n\t\t\t}\n\t\t} else if (!isExecuted) {\n\t\t\tlines.push(` class ${nodeId} pending`);\n\t\t}\n\t});\n\n\treturn lines.join(\"\\n\");\n}\n\n/**\n * Sanitize step ID to be a valid Mermaid node ID\n */\nfunction sanitizeNodeId(id: string): string {\n\t// Replace hyphens and special characters with underscores\n\treturn id.replace(/[^a-zA-Z0-9_]/g, \"_\");\n}\n",[],[539,541],"c554b2a6-d1c8-11f0-8d7b-42dde27851f2","019aee26-23ab-7319-83f5-abf61837f105",["D",1764932068219],"import type { z } from \"npm:zod@^3.23\";\n\n/**\n * Step execution result with status tracking\n */\nexport type StepResult =\n\t| { status: \"success\"; output: T; timestamp: number }\n\t| { status: \"failed\"; error: Error; timestamp: number }\n\t| { status: \"running\"; timestamp: number };\n\n/**\n * Step execution context provided to execute function\n */\nexport interface StepExecutionContext {\n\t/** Input data for this step */\n\tinputData: TInput;\n\n\t/** Workflow-level state (if defined) */\n\tstate?: TState;\n\n\t/** Update workflow state */\n\tsetState?: (state: TState) => void;\n\n\t/** Get the result of a previous step */\n\tgetStepResult: >(\n\t\tstep: T,\n\t) => z.infer;\n\n\t/** Get the initial workflow input data */\n\tgetInitData: () => T;\n\n\t/** Current run ID */\n\trunId: string;\n\n\t/** Current workflow ID */\n\tworkflowId: string;\n}\n\n/**\n * Step execute function signature\n */\nexport type ExecuteFunction = (\n\tcontext: StepExecutionContext,\n) => Promise | TOutput;\n\n/**\n * Step configuration\n */\nexport interface StepConfig {\n\tid: TId;\n\tdescription?: string;\n\tinputSchema: z.ZodSchema;\n\toutputSchema: z.ZodSchema;\n\tstateSchema?: z.ZodSchema;\n\texecute: ExecuteFunction;\n}\n\n/**\n * Step instance\n */\nexport interface Step {\n\tid: TId;\n\tdescription?: string;\n\tinputSchema: z.ZodSchema;\n\toutputSchema: z.ZodSchema;\n\tstateSchema?: z.ZodSchema;\n\texecute: ExecuteFunction;\n}\n\n/**\n * Workflow step entry in execution graph\n */\nexport interface StepFlowEntry {\n\ttype: \"step\";\n\tstep: Step;\n}\n\n/**\n * Workflow configuration\n */\nexport interface WorkflowConfig<\n\tTId extends string,\n\tTInput,\n\tTOutput,\n\tTState = unknown,\n> {\n\tid: TId;\n\tdescription?: string;\n\tinputSchema: z.ZodSchema;\n\toutputSchema: z.ZodSchema;\n\tstateSchema?: z.ZodSchema;\n}\n\n/**\n * Execution context tracking workflow execution state\n */\nexport interface ExecutionContext {\n\tworkflowId: string;\n\trunId: string;\n\texecutionPath: number[];\n\tstepResults: Record>;\n\tstate: TState;\n\tinputData: unknown;\n}\n\n/**\n * Workflow snapshot for persistence\n */\nexport interface WorkflowSnapshot {\n\trunId: string;\n\tworkflowId: string;\n\tstatus: \"running\" | \"success\" | \"failed\";\n\texecutionPath: number[];\n\tstepResults: Record>;\n\tstate: TState;\n\tinputData: unknown;\n\tresult?: unknown;\n\terror?: string;\n\ttimestamp: number;\n}\n",[],[539,418],"c554abb2-d1c8-11f0-8d7b-42dde27851f2","storage.ts","019aee26-23ab-7319-83f5-a77abfe2c5ef",["D",1764932068219],"import { runMigrations } from \"./migrations.ts\";\nimport type { WorkflowSnapshot } from \"./types.ts\";\n\n/**\n * SQLite-backed storage for workflow state\n */\nexport class WorkflowStorage {\n\tprivate initialized = false;\n\tprivate initPromise: Promise | null = null;\n\n\t/**\n\t * Initialize the storage (create tables)\n\t */\n\tasync init(): Promise {\n\t\tif (this.initialized) {\n\t\t\treturn;\n\t\t}\n\n\t\t// If initialization is already in progress, wait for it to complete\n\t\tif (this.initPromise) {\n\t\t\treturn this.initPromise;\n\t\t}\n\n\t\t// Start initialization and store the promise\n\t\tthis.initPromise = (async () => {\n\t\t\t// Note: In Val Town, import sqlite dynamically to avoid errors outside Val Town\n\t\t\tconst { sqlite } = await import(\"https://esm.town/v/std/sqlite\");\n\n\t\t\t// Run migrations to ensure tables are up to date\n\t\t\tawait runMigrations({ dropOldTables: false });\n\n\t\t\tawait sqlite.execute(`\n CREATE TABLE IF NOT EXISTS wrkflw_workflow_runs (\n run_id TEXT PRIMARY KEY,\n workflow_id TEXT NOT NULL,\n status TEXT NOT NULL,\n execution_path TEXT,\n step_results TEXT,\n state TEXT,\n input_data TEXT,\n result TEXT,\n error TEXT,\n created_at INTEGER DEFAULT (unixepoch()),\n updated_at INTEGER DEFAULT (unixepoch())\n )\n `);\n\n\t\t\t// Create index for querying by workflow_id\n\t\t\tawait sqlite.execute(`\n CREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_workflow_id\n ON wrkflw_workflow_runs(workflow_id)\n `);\n\n\t\t\t// Create index for querying by status\n\t\t\tawait sqlite.execute(`\n CREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_status\n ON wrkflw_workflow_runs(status)\n `);\n\n\t\t\tthis.initialized = true;\n\t\t})();\n\n\t\treturn this.initPromise;\n\t}\n\n\t/**\n\t * Save a workflow snapshot\n\t */\n\tasync saveSnapshot(snapshot: WorkflowSnapshot): Promise {\n\t\tconst { sqlite } = await import(\"https://esm.town/v/std/sqlite\");\n\n\t\tawait sqlite.execute({\n\t\t\tsql: `\n INSERT INTO wrkflw_workflow_runs (\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error, updated_at\n ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, unixepoch())\n ON CONFLICT(run_id) DO UPDATE SET\n status = excluded.status,\n execution_path = excluded.execution_path,\n step_results = excluded.step_results,\n state = excluded.state,\n result = excluded.result,\n error = excluded.error,\n updated_at = unixepoch()\n `,\n\t\t\targs: [\n\t\t\t\tsnapshot.runId,\n\t\t\t\tsnapshot.workflowId,\n\t\t\t\tsnapshot.status,\n\t\t\t\tJSON.stringify(snapshot.executionPath),\n\t\t\t\tJSON.stringify(snapshot.stepResults),\n\t\t\t\tJSON.stringify(snapshot.state),\n\t\t\t\tJSON.stringify(snapshot.inputData),\n\t\t\t\tsnapshot.result ? JSON.stringify(snapshot.result) : null,\n\t\t\t\tsnapshot.error ?? null,\n\t\t\t],\n\t\t});\n\t}\n\n\t/**\n\t * Load a workflow snapshot by run ID\n\t */\n\tasync loadSnapshot(runId: string): Promise {\n\t\tconst { sqlite } = await import(\"https://esm.town/v/std/sqlite\");\n\n\t\tconst result = await sqlite.execute({\n\t\t\tsql: `\n SELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error, updated_at\n FROM wrkflw_workflow_runs\n WHERE run_id = ?\n `,\n\t\t\targs: [runId],\n\t\t});\n\n\t\tif (result.rows.length === 0) {\n\t\t\treturn null;\n\t\t}\n\n\t\tconst row = result.rows[0];\n\n\t\treturn {\n\t\t\trunId: row[0] as string,\n\t\t\tworkflowId: row[1] as string,\n\t\t\tstatus: row[2] as \"running\" | \"success\" | \"failed\",\n\t\t\texecutionPath: JSON.parse(row[3] as string),\n\t\t\tstepResults: JSON.parse(row[4] as string),\n\t\t\tstate: JSON.parse(row[5] as string),\n\t\t\tinputData: JSON.parse(row[6] as string),\n\t\t\tresult: row[7] ? JSON.parse(row[7] as string) : undefined,\n\t\t\terror: row[8] as string | undefined,\n\t\t\ttimestamp: row[9] as number,\n\t\t};\n\t}\n\n\t/**\n\t * List workflow runs, optionally filtered by workflow ID\n\t */\n\tasync listRuns(workflowId?: string): Promise {\n\t\tconst { sqlite } = await import(\"https://esm.town/v/std/sqlite\");\n\n\t\tconst result = workflowId\n\t\t\t? await sqlite.execute({\n\t\t\t\t\tsql: `\n SELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error, updated_at\n FROM wrkflw_workflow_runs\n WHERE workflow_id = ?\n ORDER BY updated_at DESC\n `,\n\t\t\t\t\targs: [workflowId],\n\t\t\t\t})\n\t\t\t: await sqlite.execute(`\n SELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error, updated_at\n FROM wrkflw_workflow_runs\n ORDER BY updated_at DESC\n `);\n\n\t\treturn result.rows.map((row) => ({\n\t\t\trunId: row[0] as string,\n\t\t\tworkflowId: row[1] as string,\n\t\t\tstatus: row[2] as \"running\" | \"success\" | \"failed\",\n\t\t\texecutionPath: JSON.parse(row[3] as string),\n\t\t\tstepResults: JSON.parse(row[4] as string),\n\t\t\tstate: JSON.parse(row[5] as string),\n\t\t\tinputData: JSON.parse(row[6] as string),\n\t\t\tresult: row[7] ? JSON.parse(row[7] as string) : undefined,\n\t\t\terror: row[8] as string | undefined,\n\t\t\ttimestamp: row[9] as number,\n\t\t}));\n\t}\n\n\t/**\n\t * Delete a workflow run\n\t */\n\tasync deleteRun(runId: string): Promise {\n\t\tconst { sqlite } = await import(\"https://esm.town/v/std/sqlite\");\n\n\t\tawait sqlite.execute({\n\t\t\tsql: `DELETE FROM wrkflw_workflow_runs WHERE run_id = ?`,\n\t\t\targs: [runId],\n\t\t});\n\t}\n\n\t/**\n\t * Get workflow runs by status\n\t */\n\tasync getRunsByStatus(\n\t\tstatus: \"running\" | \"success\" | \"failed\",\n\t\tworkflowId?: string,\n\t): Promise {\n\t\tconst { sqlite } = await import(\"https://esm.town/v/std/sqlite\");\n\n\t\tconst result = workflowId\n\t\t\t? await sqlite.execute({\n\t\t\t\t\tsql: `\n SELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error, updated_at\n FROM wrkflw_workflow_runs\n WHERE status = ? AND workflow_id = ?\n ORDER BY updated_at DESC\n `,\n\t\t\t\t\targs: [status, workflowId],\n\t\t\t\t})\n\t\t\t: await sqlite.execute({\n\t\t\t\t\tsql: `\n SELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error, updated_at\n FROM wrkflw_workflow_runs\n WHERE status = ?\n ORDER BY updated_at DESC\n `,\n\t\t\t\t\targs: [status],\n\t\t\t\t});\n\n\t\treturn result.rows.map((row) => ({\n\t\t\trunId: row[0] as string,\n\t\t\tworkflowId: row[1] as string,\n\t\t\tstatus: row[2] as \"running\" | \"success\" | \"failed\",\n\t\t\texecutionPath: JSON.parse(row[3] as string),\n\t\t\tstepResults: JSON.parse(row[4] as string),\n\t\t\tstate: JSON.parse(row[5] as string),\n\t\t\tinputData: JSON.parse(row[6] as string),\n\t\t\tresult: row[7] ? JSON.parse(row[7] as string) : undefined,\n\t\t\terror: row[8] as string | undefined,\n\t\t\ttimestamp: row[9] as number,\n\t\t}));\n\t}\n}\n",[],[539,554],"c554aa9a-d1c8-11f0-8d7b-42dde27851f2","step.ts","019aee26-23ab-7319-83f5-a0036a9c4b41",["D",1764932068219],"import type { Step, StepConfig } from \"./types.ts\";\n\n/**\n * Create a workflow step with strong typing\n *\n * @example\n * ```typescript\n * const fetchUser = createStep({\n * id: 'fetch-user',\n * inputSchema: z.object({ userId: z.string() }),\n * outputSchema: z.object({ name: z.string(), email: z.string() }),\n * execute: async ({ inputData }) => {\n * const user = await db.getUser(inputData.userId);\n * return { name: user.name, email: user.email };\n * }\n * });\n * ```\n */\nexport function createStep(\n\tconfig: StepConfig,\n): Step {\n\treturn {\n\t\tid: config.id,\n\t\tdescription: config.description,\n\t\tinputSchema: config.inputSchema,\n\t\toutputSchema: config.outputSchema,\n\t\tstateSchema: config.stateSchema,\n\t\texecute: config.execute,\n\t};\n}\n",[],[539,561],"c554a7e8-d1c8-11f0-8d7b-42dde27851f2","run.ts","019aee26-23ab-7319-83f5-9fa41357f00b",["D",1764932068219],"import type { ExecutionEngine } from \"./engine.ts\";\nimport type { WorkflowStorage } from \"./storage.ts\";\nimport type { WorkflowSnapshot } from \"./types.ts\";\nimport type { Workflow } from \"./workflow.ts\";\nimport {\n\ttype ExecutionVisualizationOptions,\n\ttype VisualizationFormat,\n\tvisualizeExecution,\n} from \"./visualize.ts\";\n\n/**\n * A workflow run instance\n */\nexport class WorkflowRun {\n\tconstructor(\n\t\tprivate workflow: Workflow,\n\t\tpublic readonly runId: string,\n\t\tprivate engine: ExecutionEngine,\n\t\tprivate storage: WorkflowStorage,\n\t) {}\n\n\t/**\n\t * Start the workflow execution\n\t */\n\tasync start(params: {\n\t\tinputData: TInput;\n\t\tinitialState?: TState;\n\t}): Promise {\n\t\tconst { inputData, initialState } = params;\n\n\t\treturn await this.engine.execute({\n\t\t\tworkflow: this.workflow,\n\t\t\trunId: this.runId,\n\t\t\tinputData,\n\t\t\tinitialState,\n\t\t});\n\t}\n\n\t/**\n\t * Get the current snapshot of this run\n\t */\n\tasync getSnapshot(): Promise | null> {\n\t\treturn await this.storage.loadSnapshot(this.runId) as WorkflowSnapshot | null;\n\t}\n\n\t/**\n\t * Get the workflow ID\n\t */\n\tget workflowId(): string {\n\t\treturn this.workflow.id;\n\t}\n\n\t/**\n\t * Generate a visual representation of this workflow run's execution state\n\t *\n\t * @param format - The visualization format (default: \"mermaid\")\n\t * @param options - Visualization options\n\t * @returns Visualization string in the specified format, or null if no snapshot exists\n\t *\n\t * @example\n\t * ```typescript\n\t * const diagram = await run.visualize(\"mermaid\", {\n\t * highlightCurrentStep: true,\n\t * showStatus: true\n\t * });\n\t * if (diagram) {\n\t * console.log(diagram);\n\t * }\n\t * ```\n\t */\n\tasync visualize(\n\t\tformat: VisualizationFormat = \"mermaid\",\n\t\toptions: ExecutionVisualizationOptions = {},\n\t): Promise {\n\t\tconst snapshot = await this.getSnapshot();\n\t\tif (!snapshot) {\n\t\t\treturn null;\n\t\t}\n\t\treturn visualizeExecution(this.workflow, snapshot, format, options);\n\t}\n}\n",[],[539,568],"c5549f64-d1c8-11f0-8d7b-42dde27851f2","prebuilt-steps.ts","019aee26-23ab-7319-83f5-9bf20021cf69",["D",1764932068219],"/**\n * Prebuilt workflow steps for common tasks\n *\n * These steps provide ready-to-use functionality for:\n * - HTTP/API requests\n * - String templating\n * - Timing/delays\n * - Logging\n * - Data transformation\n */\n\nimport { z } from \"npm:zod@^3.23\";\nimport { createStep } from \"./step.ts\";\n\n// ============================================================================\n// HTTP/API Steps\n// ============================================================================\n\n/**\n * Make a GET request to a URL\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(httpGet)\n * .commit();\n *\n * await run.start({\n * inputData: {\n * url: 'https://api.example.com/users',\n * headers: { 'Authorization': 'Bearer token' }\n * }\n * });\n * ```\n */\nexport const httpGet = createStep({\n\tid: \"http-get\",\n\tdescription: \"Make a GET request and return the JSON response\",\n\tinputSchema: z.object({\n\t\turl: z.string().url(),\n\t\theaders: z.record(z.string()).optional(),\n\t\tqueryParams: z.record(z.string()).optional(),\n\t}),\n\toutputSchema: z.object({\n\t\tstatus: z.number(),\n\t\tdata: z.unknown(),\n\t\theaders: z.record(z.string()),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconst { url, headers, queryParams } = inputData;\n\n\t\t// Build URL with query parameters\n\t\tconst urlObj = new URL(url);\n\t\tif (queryParams) {\n\t\t\tObject.entries(queryParams).forEach(([key, value]) => {\n\t\t\t\turlObj.searchParams.append(key, value);\n\t\t\t});\n\t\t}\n\n\t\tconst response = await fetch(urlObj.toString(), {\n\t\t\tmethod: \"GET\",\n\t\t\theaders: headers || {},\n\t\t});\n\n\t\tconst data = await response.json();\n\n\t\t// Convert Headers object to plain object\n\t\tconst responseHeaders: Record = {};\n\t\tresponse.headers.forEach((value, key) => {\n\t\t\tresponseHeaders[key] = value;\n\t\t});\n\n\t\treturn {\n\t\t\tstatus: response.status,\n\t\t\tdata,\n\t\t\theaders: responseHeaders,\n\t\t};\n\t},\n});\n\n/**\n * Make a POST request with JSON body\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(httpPost)\n * .commit();\n *\n * await run.start({\n * inputData: {\n * url: 'https://api.example.com/users',\n * body: { name: 'John', email: 'john@example.com' }\n * }\n * });\n * ```\n */\nexport const httpPost = createStep({\n\tid: \"http-post\",\n\tdescription: \"Make a POST request with JSON body and return the response\",\n\tinputSchema: z.object({\n\t\turl: z.string().url(),\n\t\tbody: z.unknown(),\n\t\theaders: z.record(z.string()).optional(),\n\t}),\n\toutputSchema: z.object({\n\t\tstatus: z.number(),\n\t\tdata: z.unknown(),\n\t\theaders: z.record(z.string()),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconst { url, body, headers } = inputData;\n\n\t\tconst response = await fetch(url, {\n\t\t\tmethod: \"POST\",\n\t\t\theaders: {\n\t\t\t\t\"Content-Type\": \"application/json\",\n\t\t\t\t...headers,\n\t\t\t},\n\t\t\tbody: JSON.stringify(body),\n\t\t});\n\n\t\tconst data = await response.json();\n\n\t\t// Convert Headers object to plain object\n\t\tconst responseHeaders: Record = {};\n\t\tresponse.headers.forEach((value, key) => {\n\t\t\tresponseHeaders[key] = value;\n\t\t});\n\n\t\treturn {\n\t\t\tstatus: response.status,\n\t\t\tdata,\n\t\t\theaders: responseHeaders,\n\t\t};\n\t},\n});\n\n// ============================================================================\n// String/Template Steps\n// ============================================================================\n\n/**\n * Template string interpolation with variable substitution\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(template)\n * .commit();\n *\n * await run.start({\n * inputData: {\n * template: 'Hello {{name}}, your order #{{orderId}} is ready!',\n * variables: { name: 'John', orderId: '12345' }\n * }\n * });\n * // Output: { result: 'Hello John, your order #12345 is ready!' }\n * ```\n */\nexport const template = createStep({\n\tid: \"template\",\n\tdescription: \"Interpolate variables into a template string using {{variable}} syntax\",\n\tinputSchema: z.object({\n\t\ttemplate: z.string(),\n\t\tvariables: z.record(z.union([z.string(), z.number(), z.boolean()])),\n\t}),\n\toutputSchema: z.object({\n\t\tresult: z.string(),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconst { template: templateStr, variables } = inputData;\n\n\t\tconst result = templateStr.replace(/\\{\\{(\\w+)\\}\\}/g, (match, key) => {\n\t\t\tif (key in variables) {\n\t\t\t\treturn String(variables[key]);\n\t\t\t}\n\t\t\treturn match; // Keep original if variable not found\n\t\t});\n\n\t\treturn { result };\n\t},\n});\n\n// ============================================================================\n// Utility Steps\n// ============================================================================\n\n/**\n * Delay/sleep for a specified duration\n *\n * Useful for rate limiting, waiting between API calls, or timing-sensitive workflows\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(delay)\n * .then(apiCall)\n * .commit();\n *\n * await run.start({\n * inputData: { durationMs: 2000 } // Wait 2 seconds\n * });\n * ```\n */\nexport const delay = createStep({\n\tid: \"delay\",\n\tdescription: \"Wait for a specified duration in milliseconds\",\n\tinputSchema: z.object({\n\t\tdurationMs: z.number().min(0).max(300000), // Max 5 minutes\n\t}),\n\toutputSchema: z.object({\n\t\tdurationMs: z.number(),\n\t\tcompletedAt: z.string(),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconst { durationMs } = inputData;\n\n\t\tawait new Promise((resolve) => setTimeout(resolve, durationMs));\n\n\t\treturn {\n\t\t\tdurationMs,\n\t\t\tcompletedAt: new Date().toISOString(),\n\t\t};\n\t},\n});\n\n/**\n * Structured logging step\n *\n * Logs data at different levels and passes input through unchanged.\n * Useful for debugging workflow execution.\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(fetchData)\n * .then(logger) // Log the data for debugging\n * .then(processData)\n * .commit();\n *\n * await run.start({\n * inputData: {\n * level: 'info',\n * message: 'Processing user data',\n * data: { userId: '123' }\n * }\n * });\n * ```\n */\nexport const logger = createStep({\n\tid: \"logger\",\n\tdescription: \"Log data at different levels (debug, info, warn, error) and pass through\",\n\tinputSchema: z.object({\n\t\tlevel: z.enum([\"debug\", \"info\", \"warn\", \"error\"]).default(\"info\"),\n\t\tmessage: z.string(),\n\t\tdata: z.unknown().optional(),\n\t}),\n\toutputSchema: z.object({\n\t\tlevel: z.enum([\"debug\", \"info\", \"warn\", \"error\"]),\n\t\tmessage: z.string(),\n\t\tdata: z.unknown().optional(),\n\t\ttimestamp: z.string(),\n\t}),\n\texecute: async ({ inputData, runId, workflowId }) => {\n\t\tconst { level, message, data } = inputData;\n\t\tconst timestamp = new Date().toISOString();\n\n\t\tconst logEntry = {\n\t\t\ttimestamp,\n\t\t\tlevel,\n\t\t\tworkflowId,\n\t\t\trunId,\n\t\t\tmessage,\n\t\t\t...(data && { data }),\n\t\t};\n\n\t\t// Use appropriate console method\n\t\tswitch (level) {\n\t\t\tcase \"debug\":\n\t\t\t\tconsole.debug(logEntry);\n\t\t\t\tbreak;\n\t\t\tcase \"info\":\n\t\t\t\tconsole.info(logEntry);\n\t\t\t\tbreak;\n\t\t\tcase \"warn\":\n\t\t\t\tconsole.warn(logEntry);\n\t\t\t\tbreak;\n\t\t\tcase \"error\":\n\t\t\t\tconsole.error(logEntry);\n\t\t\t\tbreak;\n\t\t}\n\n\t\treturn {\n\t\t\tlevel,\n\t\t\tmessage,\n\t\t\tdata,\n\t\t\ttimestamp,\n\t\t};\n\t},\n});\n\n// ============================================================================\n// Data Transformation Steps\n// ============================================================================\n\n/**\n * Filter array items based on a condition\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(filterArray)\n * .commit();\n *\n * await run.start({\n * inputData: {\n * array: [\n * { name: 'Alice', age: 30 },\n * { name: 'Bob', age: 17 },\n * { name: 'Charlie', age: 25 }\n * ],\n * filterFn: '(item) => item.age >= 18'\n * }\n * });\n * // Output: { filtered: [{ name: 'Alice', age: 30 }, { name: 'Charlie', age: 25 }] }\n * ```\n */\nexport const filterArray = createStep({\n\tid: \"filter-array\",\n\tdescription: \"Filter array items using a JavaScript filter function (as string)\",\n\tinputSchema: z.object({\n\t\tarray: z.array(z.unknown()),\n\t\tfilterFn: z.string(), // JavaScript function as string: \"(item) => condition\"\n\t}),\n\toutputSchema: z.object({\n\t\tfiltered: z.array(z.unknown()),\n\t\toriginalCount: z.number(),\n\t\tfilteredCount: z.number(),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconst { array, filterFn } = inputData;\n\n\t\t// Create function from string\n\t\t// Note: In production, consider security implications of eval\n\t\t// For Val Town personal automation, this provides flexibility\n\t\tconst filterFunction = eval(filterFn);\n\n\t\tconst filtered = array.filter(filterFunction);\n\n\t\treturn {\n\t\t\tfiltered,\n\t\t\toriginalCount: array.length,\n\t\t\tfilteredCount: filtered.length,\n\t\t};\n\t},\n});\n\n/**\n * Transform array items using a map function\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(mapArray)\n * .commit();\n *\n * await run.start({\n * inputData: {\n * array: [1, 2, 3, 4],\n * mapFn: '(item) => item * 2'\n * }\n * });\n * // Output: { mapped: [2, 4, 6, 8] }\n * ```\n */\nexport const mapArray = createStep({\n\tid: \"map-array\",\n\tdescription: \"Transform array items using a JavaScript map function (as string)\",\n\tinputSchema: z.object({\n\t\tarray: z.array(z.unknown()),\n\t\tmapFn: z.string(), // JavaScript function as string: \"(item) => transformed\"\n\t}),\n\toutputSchema: z.object({\n\t\tmapped: z.array(z.unknown()),\n\t\tcount: z.number(),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconst { array, mapFn } = inputData;\n\n\t\t// Create function from string\n\t\tconst mapFunction = eval(mapFn);\n\n\t\tconst mapped = array.map(mapFunction);\n\n\t\treturn {\n\t\t\tmapped,\n\t\t\tcount: mapped.length,\n\t\t};\n\t},\n});\n\n/**\n * Extract specific fields from an object\n *\n * Useful for simplifying data flow between steps by removing unnecessary fields\n *\n * @example\n * ```typescript\n * const workflow = createWorkflow({ ... })\n * .then(pickFields)\n * .commit();\n *\n * await run.start({\n * inputData: {\n * object: { id: 1, name: 'Alice', email: 'alice@example.com', password: 'secret' },\n * fields: ['id', 'name', 'email']\n * }\n * });\n * // Output: { result: { id: 1, name: 'Alice', email: 'alice@example.com' } }\n * ```\n */\nexport const pickFields = createStep({\n\tid: \"pick-fields\",\n\tdescription: \"Extract specific fields from an object\",\n\tinputSchema: z.object({\n\t\tobject: z.record(z.unknown()),\n\t\tfields: z.array(z.string()),\n\t}),\n\toutputSchema: z.object({\n\t\tresult: z.record(z.unknown()),\n\t}),\n\texecute: async ({ inputData }) => {\n\t\tconst { object, fields } = inputData;\n\n\t\tconst result: Record = {};\n\n\t\tfor (const field of fields) {\n\t\t\tif (field in object) {\n\t\t\t\tresult[field] = object[field];\n\t\t\t}\n\t\t}\n\n\t\treturn { result };\n\t},\n});\n",[],[539,575],"c55492e4-d1c8-11f0-8d7b-42dde27851f2","migrations.ts","019aee26-23ab-7319-83f5-9521e3d5c69a",["D",1764932068219],"/**\n * Database migration utilities for wrkflw\n *\n * This module provides SQL migration functionality to help users\n * migrate their existing tables to the new naming scheme with wrkflw_ prefix.\n */\n\n/**\n * Migrates the old workflow_runs table to wrkflw_workflow_runs\n *\n * This function:\n * 1. Checks if the old table exists\n * 2. Creates the new table with wrkflw_ prefix\n * 3. Copies all data from the old table to the new table\n * 4. Optionally drops the old table (if dropOldTable is true)\n *\n * @param dropOldTable - Whether to drop the old table after migration (default: false for safety)\n * @returns Migration summary with row count and status\n */\nexport async function migrateWorkflowRunsTable(\n\tdropOldTable = false,\n): Promise<{ migrated: boolean; rowCount: number; message: string }> {\n\tconst { sqlite } = await import(\"https://esm.town/v/std/sqlite\");\n\n\t// Check both tables in a single query for efficiency\n\tconst tableCheck = await sqlite.execute({\n\t\tsql: `\n SELECT name FROM sqlite_master\n WHERE type='table' AND name IN ('workflow_runs', 'wrkflw_workflow_runs')\n `,\n\t\targs: [],\n\t});\n\n\tconst existingTables = tableCheck.rows.map((row) => row[0] as string);\n\tconst oldTableExists = existingTables.includes(\"workflow_runs\");\n\tconst newTableExists = existingTables.includes(\"wrkflw_workflow_runs\");\n\n\t// If new table exists and old table doesn't, migration already completed\n\tif (newTableExists && !oldTableExists) {\n\t\treturn {\n\t\t\tmigrated: false,\n\t\t\trowCount: 0,\n\t\t\tmessage:\n\t\t\t\t\"Migration already completed - using 'wrkflw_workflow_runs' table\",\n\t\t};\n\t}\n\n\t// If old table doesn't exist, nothing to migrate\n\tif (!oldTableExists) {\n\t\treturn {\n\t\t\tmigrated: false,\n\t\t\trowCount: 0,\n\t\t\tmessage: \"No migration needed - old table 'workflow_runs' does not exist\",\n\t\t};\n\t}\n\n\t// Create new table if it doesn't exist\n\tif (!newTableExists) {\n\t\tawait sqlite.execute(`\n CREATE TABLE wrkflw_workflow_runs (\n run_id TEXT PRIMARY KEY,\n workflow_id TEXT NOT NULL,\n status TEXT NOT NULL,\n execution_path TEXT,\n step_results TEXT,\n state TEXT,\n input_data TEXT,\n result TEXT,\n error TEXT,\n created_at INTEGER DEFAULT (unixepoch()),\n updated_at INTEGER DEFAULT (unixepoch())\n )\n `);\n\n\t\t// Create indexes\n\t\tawait sqlite.execute(`\n CREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_workflow_id\n ON wrkflw_workflow_runs(workflow_id)\n `);\n\n\t\tawait sqlite.execute(`\n CREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_status\n ON wrkflw_workflow_runs(status)\n `);\n\t}\n\n\t// Get row count from old table\n\tconst countResult = await sqlite.execute({\n\t\tsql: \"SELECT COUNT(*) as count FROM workflow_runs\",\n\t\targs: [],\n\t});\n\tconst rowCount = countResult.rows[0][0] as number;\n\n\t// If old table is empty, no migration needed\n\tif (rowCount === 0) {\n\t\treturn {\n\t\t\tmigrated: false,\n\t\t\trowCount: 0,\n\t\t\tmessage:\n\t\t\t\t\"No migration needed - old table 'workflow_runs' exists but is empty\",\n\t\t};\n\t}\n\n\t// Check if new table already has data\n\tconst newTableCountResult = await sqlite.execute({\n\t\tsql: \"SELECT COUNT(*) as count FROM wrkflw_workflow_runs\",\n\t\targs: [],\n\t});\n\tconst newTableRowCount = newTableCountResult.rows[0][0] as number;\n\n\t// Copy data from old table to new table\n\tconst _insertResult = await sqlite.execute(`\n INSERT OR IGNORE INTO wrkflw_workflow_runs (\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error,\n created_at, updated_at\n )\n SELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error,\n created_at, updated_at\n FROM workflow_runs\n `);\n\n\t// Calculate how many rows were actually inserted\n\tconst finalCountResult = await sqlite.execute({\n\t\tsql: \"SELECT COUNT(*) as count FROM wrkflw_workflow_runs\",\n\t\targs: [],\n\t});\n\tconst finalRowCount = finalCountResult.rows[0][0] as number;\n\tconst rowsInserted = finalRowCount - newTableRowCount;\n\tconst rowsSkipped = rowCount - rowsInserted;\n\n\t// Build informative message\n\tlet message = \"\";\n\tif (rowsSkipped > 0) {\n\t\tmessage = `Migration completed with conflicts: inserted ${rowsInserted} new rows, skipped ${rowsSkipped} duplicate rows (conflicts detected - new table already had ${newTableRowCount} rows)`;\n\t} else {\n\t\tmessage = `Successfully migrated ${rowsInserted} rows from 'workflow_runs' to 'wrkflw_workflow_runs'`;\n\t}\n\n\t// Optionally drop old table\n\tif (dropOldTable) {\n\t\tawait sqlite.execute(\"DROP TABLE workflow_runs\");\n\t\t// Also drop old indexes if they exist\n\t\tawait sqlite.execute(\"DROP INDEX IF EXISTS idx_workflow_runs_workflow_id\");\n\t\tawait sqlite.execute(\"DROP INDEX IF EXISTS idx_workflow_runs_status\");\n\n\t\treturn {\n\t\t\tmigrated: true,\n\t\t\trowCount,\n\t\t\tmessage: `Successfully migrated ${rowCount} rows from 'workflow_runs' to 'wrkflw_workflow_runs' and dropped old table`,\n\t\t};\n\t}\n\n\treturn {\n\t\tmigrated: true,\n\t\trowCount,\n\t\tmessage: `Successfully migrated ${rowCount} rows from 'workflow_runs' to 'wrkflw_workflow_runs'. Old table preserved for safety.`,\n\t};\n}\n\n/**\n * Runs all pending migrations\n *\n * This function should be called during application initialization\n * to ensure all tables are up to date with the latest schema.\n *\n * @param options - Migration options\n * @param options.dropOldTables - Whether to drop old tables after migration\n * @returns Array of migration results\n */\nexport async function runMigrations(\n\toptions: { dropOldTables?: boolean } = {},\n): Promise> {\n\tconst results = [];\n\n\t// Run workflow_runs migration\n\tconst workflowRunsResult = await migrateWorkflowRunsTable(\n\t\toptions.dropOldTables ?? false,\n\t);\n\tresults.push(workflowRunsResult);\n\n\treturn results;\n}\n\n/**\n * Export SQL script for manual migration\n *\n * Generates SQL statements that can be run manually to migrate tables.\n *\n * **IMPORTANT**: This SQL script assumes the old `workflow_runs` table exists.\n * If you're setting up a fresh installation (no existing data), you only need\n * to create the new table - the data migration step can be skipped.\n *\n * Before running this script, verify the old table exists:\n * ```sql\n * SELECT name FROM sqlite_master WHERE type='table' AND name='workflow_runs';\n * ```\n *\n * @param dropOldTable - Whether to include DROP TABLE statements\n * @returns SQL migration script as a string\n */\nexport function generateMigrationSQL(dropOldTable = false): string {\n\tconst sql = `\n-- Migration Script: Normalize table naming to wrkflw_ prefix\n-- Generated: ${new Date().toISOString()}\n--\n-- IMPORTANT: This script is designed for migrating FROM an existing workflow_runs table.\n-- If you're setting up a fresh installation without existing data, you can skip\n-- the data migration step (Step 3) as it will have no data to copy.\n--\n-- Before running this script, you can verify if the old table exists:\n-- SELECT name FROM sqlite_master WHERE type='table' AND name='workflow_runs';\n\n-- Step 1: Create new table with wrkflw_ prefix\nCREATE TABLE IF NOT EXISTS wrkflw_workflow_runs (\n run_id TEXT PRIMARY KEY,\n workflow_id TEXT NOT NULL,\n status TEXT NOT NULL,\n execution_path TEXT,\n step_results TEXT,\n state TEXT,\n input_data TEXT,\n result TEXT,\n error TEXT,\n created_at INTEGER DEFAULT (unixepoch()),\n updated_at INTEGER DEFAULT (unixepoch())\n);\n\n-- Step 2: Create indexes for new table\nCREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_workflow_id\nON wrkflw_workflow_runs(workflow_id);\n\nCREATE INDEX IF NOT EXISTS idx_wrkflw_workflow_runs_status\nON wrkflw_workflow_runs(status);\n\n-- Step 3: Copy data from old table to new table\n-- NOTE: This step will fail if the 'workflow_runs' table doesn't exist.\n-- If you don't have an existing workflow_runs table, this is expected and you can\n-- ignore this step - there's no data to migrate.\nINSERT OR IGNORE INTO wrkflw_workflow_runs (\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error,\n created_at, updated_at\n)\nSELECT\n run_id, workflow_id, status, execution_path,\n step_results, state, input_data, result, error,\n created_at, updated_at\nFROM workflow_runs;\n${\n\tdropOldTable\n\t\t? `\n-- Step 4: Drop old table and indexes (CAUTION: This is irreversible!)\nDROP TABLE IF EXISTS workflow_runs;\nDROP INDEX IF EXISTS idx_workflow_runs_workflow_id;\nDROP INDEX IF EXISTS idx_workflow_runs_status;\n`\n\t\t: `\n-- Step 4: Old table preserved for safety\n-- To drop the old table manually, run:\n-- DROP TABLE workflow_runs;\n-- DROP INDEX IF EXISTS idx_workflow_runs_workflow_id;\n-- DROP INDEX IF EXISTS idx_workflow_runs_status;\n`\n}\n-- Migration complete!\n`.trim();\n\n\treturn sql;\n}\n",[],[539,582],"c5549104-d1c8-11f0-8d7b-42dde27851f2","index.ts","019aee26-23ab-7319-83f5-90d41faeb6f1",["D",1764932068219],"/**\n * wrkflw - A TypeScript-first workflow engine for Val Town\n *\n * Inspired by Mastra, optimized for Val Town platform\n */\n\nexport { ExecutionEngine } from \"./engine.ts\";\nexport { WorkflowRun } from \"./run.ts\";\n// Core exports\nexport { createStep } from \"./step.ts\";\nexport { WorkflowStorage } from \"./storage.ts\";\n// Migration exports\nexport {\n\tgenerateMigrationSQL,\n\tmigrateWorkflowRunsTable,\n\trunMigrations,\n} from \"./migrations.ts\";\n// Type exports\nexport type {\n\tExecuteFunction,\n\tExecutionContext,\n\tStep,\n\tStepConfig,\n\tStepExecutionContext,\n\tStepFlowEntry,\n\tStepResult,\n\tWorkflowConfig,\n\tWorkflowSnapshot,\n} from \"./types.ts\";\nexport { createWorkflow, Workflow, WorkflowBuilder } from \"./workflow.ts\";\n\n// Prebuilt steps\nexport {\n\tdelay,\n\tfilterArray,\n\thttpGet,\n\thttpPost,\n\tlogger,\n\tmapArray,\n\tpickFields,\n\ttemplate,\n} from \"./prebuilt-steps.ts\";\n\n// Visualization exports\nexport {\n\tvisualizeExecution,\n\tvisualizeWorkflow,\n\ttype ExecutionVisualizationOptions,\n\ttype VisualizationFormat,\n\ttype VisualizationOptions,\n} from \"./visualize.ts\";\n\n// API Server exports\nexport { ApiServer, createApiServer } from \"./api-server.ts\";\n",[],[539,589],"c5548a7e-d1c8-11f0-8d7b-42dde27851f2","engine.ts","019aee26-23ab-7319-83f5-8e765f6e3a59",["D",1764932068219],"import type { WorkflowStorage } from \"./storage.ts\";\nimport type {\n\tExecutionContext,\n\tStep,\n\tStepExecutionContext,\n\tStepResult,\n} from \"./types.ts\";\nimport type { Workflow } from \"./workflow.ts\";\n\n/**\n * Execution engine that runs workflows step by step\n */\nexport class ExecutionEngine {\n\tconstructor(private storage: WorkflowStorage) {}\n\n\t/**\n\t * Execute a workflow from start to finish\n\t */\n\tasync execute(params: {\n\t\tworkflow: Workflow;\n\t\trunId: string;\n\t\tinputData: TInput;\n\t\tinitialState?: TState;\n\t}): Promise {\n\t\tconst { workflow, runId, inputData, initialState } = params;\n\n\t\t// Validate input data at runtime\n\t\tconst validatedInput = workflow.inputSchema.parse(inputData);\n\n\t\t// Initialize execution context\n\t\tconst context: ExecutionContext = {\n\t\t\tworkflowId: workflow.id,\n\t\t\trunId,\n\t\t\texecutionPath: [],\n\t\t\tstepResults: {},\n\t\t\tstate:\n\t\t\t\tinitialState ??\n\t\t\t\t(workflow.stateSchema ? ({} as TState) : undefined as unknown as TState),\n\t\t\tinputData: validatedInput,\n\t\t};\n\n\t\t// Save initial snapshot\n\t\tawait this.storage.saveSnapshot({\n\t\t\trunId,\n\t\t\tworkflowId: workflow.id,\n\t\t\tstatus: \"running\",\n\t\t\texecutionPath: context.executionPath,\n\t\t\tstepResults: context.stepResults,\n\t\t\tstate: context.state,\n\t\t\tinputData: validatedInput,\n\t\t\ttimestamp: Date.now(),\n\t\t});\n\n\t\ttry {\n\t\t\t// Execute all steps sequentially\n\t\t\tlet currentOutput: unknown = validatedInput;\n\n\t\t\tfor (let i = 0; i < workflow.stepFlow.length; i++) {\n\t\t\t\tcontext.executionPath.push(i);\n\n\t\t\t\tconst entry = workflow.stepFlow[i];\n\n\t\t\t\tif (entry.type === \"step\") {\n\t\t\t\t\tconst result = await this.executeStep(\n\t\t\t\t\t\tentry.step,\n\t\t\t\t\t\tcontext,\n\t\t\t\t\t\tcurrentOutput,\n\t\t\t\t\t);\n\n\t\t\t\t\tcontext.stepResults[entry.step.id] = result;\n\n\t\t\t\t\tif (result.status === \"failed\") {\n\t\t\t\t\t\t// Save failed state\n\t\t\t\t\t\tawait this.storage.saveSnapshot({\n\t\t\t\t\t\t\trunId,\n\t\t\t\t\t\t\tworkflowId: workflow.id,\n\t\t\t\t\t\t\tstatus: \"failed\",\n\t\t\t\t\t\t\texecutionPath: context.executionPath,\n\t\t\t\t\t\t\tstepResults: context.stepResults,\n\t\t\t\t\t\t\tstate: context.state,\n\t\t\t\t\t\t\tinputData: validatedInput,\n\t\t\t\t\t\t\terror: result.error.message,\n\t\t\t\t\t\t\ttimestamp: Date.now(),\n\t\t\t\t\t\t});\n\n\t\t\t\t\t\tthrow result.error;\n\t\t\t\t\t}\n\n\t\t\t\t\tif (result.status === \"success\") {\n\t\t\t\t\t\tcurrentOutput = result.output;\n\t\t\t\t\t}\n\n\t\t\t\t\t// Save progress after each step\n\t\t\t\t\tawait this.storage.saveSnapshot({\n\t\t\t\t\t\trunId,\n\t\t\t\t\t\tworkflowId: workflow.id,\n\t\t\t\t\t\tstatus: \"running\",\n\t\t\t\t\t\texecutionPath: context.executionPath,\n\t\t\t\t\t\tstepResults: context.stepResults,\n\t\t\t\t\t\tstate: context.state,\n\t\t\t\t\t\tinputData: validatedInput,\n\t\t\t\t\t\ttimestamp: Date.now(),\n\t\t\t\t\t});\n\t\t\t\t}\n\t\t\t}\n\n\t\t\t// Validate output\n\t\t\tconst validatedOutput = workflow.outputSchema.parse(currentOutput);\n\n\t\t\t// Save success state\n\t\t\tawait this.storage.saveSnapshot({\n\t\t\t\trunId,\n\t\t\t\tworkflowId: workflow.id,\n\t\t\t\tstatus: \"success\",\n\t\t\t\texecutionPath: context.executionPath,\n\t\t\t\tstepResults: context.stepResults,\n\t\t\t\tstate: context.state,\n\t\t\t\tinputData: validatedInput,\n\t\t\t\tresult: validatedOutput,\n\t\t\t\ttimestamp: Date.now(),\n\t\t\t});\n\n\t\t\treturn validatedOutput;\n\t\t} catch (error) {\n\t\t\t// If error wasn't already saved (e.g., validation error)\n\t\t\tawait this.storage.saveSnapshot({\n\t\t\t\trunId,\n\t\t\t\tworkflowId: workflow.id,\n\t\t\t\tstatus: \"failed\",\n\t\t\t\texecutionPath: context.executionPath,\n\t\t\t\tstepResults: context.stepResults,\n\t\t\t\tstate: context.state,\n\t\t\t\tinputData: validatedInput,\n\t\t\t\terror: error instanceof Error ? error.message : String(error),\n\t\t\t\ttimestamp: Date.now(),\n\t\t\t});\n\n\t\t\tthrow error;\n\t\t}\n\t}\n\n\t/**\n\t * Execute a single step\n\t */\n\tprivate async executeStep(\n\t\tstep: Step,\n\t\tcontext: ExecutionContext,\n\t\tinputData: unknown,\n\t): Promise> {\n\t\tconst timestamp = Date.now();\n\n\t\ttry {\n\t\t\t// Validate input data at runtime\n\t\t\tconst validatedInput = step.inputSchema.parse(inputData);\n\n\t\t\t// Build step execution context\n\t\t\tconst stepContext: StepExecutionContext = {\n\t\t\t\tinputData: validatedInput,\n\t\t\t\tstate: context.state,\n\t\t\t\tsetState: (newState: TState) => {\n\t\t\t\t\tcontext.state = newState;\n\t\t\t\t},\n\t\t\t\tgetStepResult: >(targetStep: T) => {\n\t\t\t\t\tconst result = context.stepResults[targetStep.id];\n\t\t\t\t\tif (!result) {\n\t\t\t\t\t\tthrow new Error(`Step ${targetStep.id} has not been executed yet`);\n\t\t\t\t\t}\n\t\t\t\t\tif (result.status !== \"success\") {\n\t\t\t\t\t\tthrow new Error(\n\t\t\t\t\t\t\t`Step ${targetStep.id} did not complete successfully`,\n\t\t\t\t\t\t);\n\t\t\t\t\t}\n\t\t\t\t\treturn result.output;\n\t\t\t\t},\n\t\t\t\tgetInitData: () => context.inputData as any,\n\t\t\t\trunId: context.runId,\n\t\t\t\tworkflowId: context.workflowId,\n\t\t\t};\n\n\t\t\t// Execute the step\n\t\t\tconst output = await step.execute(stepContext);\n\n\t\t\t// Validate output at runtime\n\t\t\tconst validatedOutput = step.outputSchema.parse(output);\n\n\t\t\treturn {\n\t\t\t\tstatus: \"success\",\n\t\t\t\toutput: validatedOutput,\n\t\t\t\ttimestamp,\n\t\t\t};\n\t\t} catch (error) {\n\t\t\treturn {\n\t\t\t\tstatus: \"failed\",\n\t\t\t\terror: error instanceof Error ? error : new Error(String(error)),\n\t\t\t\ttimestamp,\n\t\t\t};\n\t\t}\n\t}\n}\n",[],[539,596],"c5547106-d1c8-11f0-8d7b-42dde27851f2","api-server.ts","019aee26-23ab-7319-83f5-89b04d783661",["D",1764932068219],"import { Workflow } from \"./workflow.ts\";\nimport { WorkflowStorage } from \"./storage.ts\";\nimport { WorkflowSnapshot } from \"./types.ts\";\n\ninterface ApiWorkflowData {\n id: string;\n description?: string;\n steps: Array<{ id: string; description?: string }>;\n exampleInput?: Record;\n}\n\ninterface ApiRunData {\n runId: string;\n workflowId: string;\n status: \"running\" | \"success\" | \"failed\";\n executionPath: number[];\n stepResults: Record;\n inputData: unknown;\n result?: unknown;\n error?: string;\n createdAt: number;\n updatedAt: number;\n}\n\nexport class ApiServer {\n private workflows: Map; exampleInput?: Record }> = new Map();\n private storage: WorkflowStorage;\n\n constructor(storage: WorkflowStorage) {\n this.storage = storage;\n }\n\n registerWorkflow(\n workflow: Workflow,\n exampleInput?: Record,\n ) {\n this.workflows.set(workflow.id, { workflow: workflow as any, exampleInput });\n }\n\n private serializeWorkflow(workflowId: string): ApiWorkflowData | null {\n const entry = this.workflows.get(workflowId);\n if (!entry) return null;\n\n const { workflow, exampleInput } = entry;\n return {\n id: workflow.id,\n description: workflow.description,\n steps: workflow.steps.map((step) => ({\n id: step.id,\n description: step.description,\n })),\n exampleInput,\n };\n }\n\n private serializeRun(snapshot: WorkflowSnapshot): ApiRunData {\n return {\n runId: snapshot.runId,\n workflowId: snapshot.workflowId,\n status: snapshot.status,\n executionPath: snapshot.executionPath,\n stepResults: snapshot.stepResults,\n inputData: snapshot.inputData,\n result: snapshot.result,\n error: snapshot.error,\n createdAt: snapshot.timestamp,\n updatedAt: snapshot.timestamp,\n };\n }\n\n async handleRequest(request: Request): Promise {\n const url = new URL(request.url);\n const path = url.pathname;\n const method = request.method;\n\n // CORS headers\n const corsHeaders = {\n \"Access-Control-Allow-Origin\": \"*\",\n \"Access-Control-Allow-Methods\": \"GET, POST, OPTIONS\",\n \"Access-Control-Allow-Headers\": \"Content-Type\",\n \"Content-Type\": \"application/json\",\n };\n\n // Handle preflight requests\n if (method === \"OPTIONS\") {\n return new Response(null, { headers: corsHeaders });\n }\n\n try {\n // GET /api/workflows - List all workflows\n if (path === \"/api/workflows\" && method === \"GET\") {\n const workflows = Array.from(this.workflows.keys()).map((id) =>\n this.serializeWorkflow(id)\n ).filter((w): w is ApiWorkflowData => w !== null);\n\n return new Response(JSON.stringify(workflows), { headers: corsHeaders });\n }\n\n // GET /api/workflows/:id - Get workflow details\n const workflowMatch = path.match(/^\\/api\\/workflows\\/([^\\/]+)$/);\n if (workflowMatch && method === \"GET\") {\n const workflowId = workflowMatch[1];\n const workflow = this.serializeWorkflow(workflowId);\n\n if (!workflow) {\n return new Response(JSON.stringify({ error: \"Workflow not found\" }), {\n status: 404,\n headers: corsHeaders,\n });\n }\n\n return new Response(JSON.stringify(workflow), { headers: corsHeaders });\n }\n\n // GET /api/workflows/:id/runs - List runs for a workflow\n const runsMatch = path.match(/^\\/api\\/workflows\\/([^\\/]+)\\/runs$/);\n if (runsMatch && method === \"GET\") {\n const workflowId = runsMatch[1];\n const snapshots = await this.storage.listRuns(workflowId);\n const runs = snapshots.map((s) => this.serializeRun(s));\n\n return new Response(JSON.stringify(runs), { headers: corsHeaders });\n }\n\n // POST /api/workflows/:id/runs - Create a new run\n if (runsMatch && method === \"POST\") {\n const workflowId = runsMatch[1];\n const entry = this.workflows.get(workflowId);\n\n if (!entry) {\n return new Response(JSON.stringify({ error: \"Workflow not found\" }), {\n status: 404,\n headers: corsHeaders,\n });\n }\n\n const body = await request.json();\n const { inputData } = body;\n\n // Create and start the run in the background\n const run = await entry.workflow.createRun();\n\n // Start execution (don't await - let it run in background)\n run.start({ inputData }).catch((error) => {\n console.error(`Run ${run.runId} failed:`, error);\n });\n\n // Return the initial run state\n const snapshot = await this.storage.loadSnapshot(run.runId);\n if (!snapshot) {\n return new Response(JSON.stringify({ error: \"Failed to create run\" }), {\n status: 500,\n headers: corsHeaders,\n });\n }\n\n return new Response(JSON.stringify(this.serializeRun(snapshot)), {\n status: 201,\n headers: corsHeaders,\n });\n }\n\n // GET /api/runs/:runId - Get run details\n const runMatch = path.match(/^\\/api\\/runs\\/([^\\/]+)$/);\n if (runMatch && method === \"GET\") {\n const runId = runMatch[1];\n const snapshot = await this.storage.loadSnapshot(runId);\n\n if (!snapshot) {\n return new Response(JSON.stringify({ error: \"Run not found\" }), {\n status: 404,\n headers: corsHeaders,\n });\n }\n\n return new Response(JSON.stringify(this.serializeRun(snapshot)), {\n headers: corsHeaders,\n });\n }\n\n // 404 for unmatched routes\n return new Response(JSON.stringify({ error: \"Not found\" }), {\n status: 404,\n headers: corsHeaders,\n });\n } catch (error) {\n console.error(\"API Error:\", error);\n return new Response(\n JSON.stringify({ error: error instanceof Error ? error.message : \"Internal server error\" }),\n {\n status: 500,\n headers: corsHeaders,\n },\n );\n }\n }\n\n async serve(port = 8000) {\n console.log(`API server starting on http://localhost:${port}`);\n\n const server = Deno.serve({ port }, (request) => this.handleRequest(request));\n\n return server;\n }\n}\n\n// Helper function to create and configure an API server\nexport function createApiServer(storage: WorkflowStorage): ApiServer {\n return new ApiServer(storage);\n}\n",[],[539,603],"c554c156-d1c8-11f0-8d7b-42dde27851f2",["D",1764932068219],[226,227,228,229,230,231,232,233,234,235,236],[539],["D",1764932068219],{"_11":12,"_154":82},{"_11":12,"_149":150,"_81":82,"_31":80,"_151":152,"_153":30},{"_39":40,"_13":16,"_33":36,"_138":104,"_139":71,"_140":141,"_142":108,"_143":144,"_145":146,"_147":14,"_148":12},"ctx",{"_52":-5,"_175":650,"_651":652,"_653":654,"_655":30,"_656":30,"_657":30,"_658":24,"_659":-5,"_660":-7,"_661":-7,"_662":26,"_663":30,"_664":30,"_665":-5,"_666":30,"_667":30,"_668":30,"_669":-5},"ENV",{"_621":622,"_623":624,"_625":16,"_626":627,"_628":629,"_630":631,"_632":633,"_634":635,"_636":637,"_638":639,"_640":641,"_642":643,"_644":645,"_646":647,"_648":649},"REACT_APP_SENTRY_DSN","https://586020d86f05489db63a554690acf1e9@o1398359.ingest.sentry.io/6724499","RENDER_GIT_COMMIT","fdb67791413f4386b4dbf18f354029a617f6160f","RENDER_GIT_BRANCH","POSTHOG_API_TOKEN","phc_nmoyuAkStEZuUp69rYlg1DZgrIt19Gx0Y4Hwf8tGgJP","POSTHOG_API_HOST","https://wart.val.town","WEB_API_TEMPLATE","https://{val}.web.val.run","APP_URL","https://www.val.town","MODULE_URL","https://esm.town","VANITY_WEB_API_TEMPLATE","https://{label}.val.run/","DENOLS_URL","https://lsp.val.town","NODE_ENV","production","CLOUDFLARE_TURNSTILE_SITE_KEY","0x4AAAAAACtaIZJH7s9rBMZ6","GOOGLE_CLIENT_ID","875989696899-jtsqkhlbdv5jrlsk8631p7ipjg3c6o62.apps.googleusercontent.com","STRIPE_PUBLISHABLE_KEY","pk_live_51N2zvWEpZTTLipIkVEwXrUG9AT0yeOOEYxszarZT1jpsZV58Ly6wlg3SltZpHN26CUalOBmUSY7adQRQiSkCZz2500qSXNkf8Y","API_URL","https://api.val.town",[],"preferencesKeybindings","default","preferencesTheme","light","preferencesFixers","preferencesCopilot","preferencesSelectionAi","preferencesCloseBrackets","preferencesChatPanelState","preferencesDprint","preferencesDenofmt","notificationCount","flagDenols","isBusinessOrgMember","businessOrgHandle","isBusinessOrgAdmin","isHigherTownieTier","navCollapsed","currentOrgId"]

History · components