experiment: Add full flow executor with branch/loop support

Add flow_executor.rs that supports executing complex flows in local mode: - ForloopFlow: iterate over arrays/ranges with sequential execution - WhileloopFlow: execute modules while condition is true - BranchOne: if/else branching based on conditions - BranchAll: parallel branch execution (sequential for now) - RawScript: inline script execution (bash, python, deno, bun) - Identity: pass-through module Key features: - Uses windmill-common FlowValue types for compatibility - Expression evaluation for input transforms with comparisons - Proper flow status tracking with module results - Recursive async execution with async_recursion crate Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
experiment: Add windmill-local crate with libSQL/Turso support
2026-01-23 07:14:59 +00:00 · 2026-01-23 07:01:44 +00:00
14 changed files with 3598 additions and 40 deletions
--- a/backend/Cargo.lock
+++ b/backend/Cargo.lock
@@ -125,7 +125,7 @@ dependencies = [
 "getrandom 0.3.4",
 "once_cell",
 "version_check",
- "zerocopy",
+ "zerocopy 0.8.33",
 ]

 [[package]]
@@ -1302,6 +1302,34 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "axum"
+version = "0.6.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf"
+dependencies = [
+ "async-trait",
+ "axum-core 0.3.4",
+ "bitflags 1.3.2",
+ "bytes",
+ "futures-util",
+ "http 0.2.12",
+ "http-body 0.4.6",
+ "hyper 0.14.32",
+ "itoa",
+ "matchit 0.7.3",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "rustversion",
+ "serde",
+ "sync_wrapper 0.1.2",
+ "tower 0.4.13",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "axum"
 version = "0.7.9"
@@ -1330,7 +1358,7 @@ dependencies = [
 "serde_json",
 "serde_path_to_error",
 "serde_urlencoded",
- "sync_wrapper",
+ "sync_wrapper 1.0.2",
 "tokio",
 "tower 0.5.3",
 "tower-layer",
@@ -1364,7 +1392,7 @@ dependencies = [
 "serde_json",
 "serde_path_to_error",
 "serde_urlencoded",
- "sync_wrapper",
+ "sync_wrapper 1.0.2",
 "tokio",
 "tower 0.5.3",
 "tower-layer",
@@ -1372,6 +1400,23 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "axum-core"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "759fa577a247914fd3f7f76d62972792636412fbfd634cd452f6a385a74d2d2c"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "futures-util",
+ "http 0.2.12",
+ "http-body 0.4.6",
+ "mime",
+ "rustversion",
+ "tower-layer",
+ "tower-service",
+]
+
 [[package]]
 name = "axum-core"
 version = "0.4.5"
@@ -1387,7 +1432,7 @@ dependencies = [
 "mime",
 "pin-project-lite",
 "rustversion",
- "sync_wrapper",
+ "sync_wrapper 1.0.2",
 "tower-layer",
 "tower-service",
 "tracing",
@@ -1406,7 +1451,7 @@ dependencies = [
 "http-body-util",
 "mime",
 "pin-project-lite",
- "sync_wrapper",
+ "sync_wrapper 1.0.2",
 "tower-layer",
 "tower-service",
 "tracing",
@@ -1541,6 +1586,29 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "bindgen"
+version = "0.66.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2b84e06fc203107bfbad243f4aba2af864eb7db3b1cf46ea0a023b0b433d2a7"
+dependencies = [
+ "bitflags 2.9.4",
+ "cexpr",
+ "clang-sys",
+ "lazy_static",
+ "lazycell",
+ "log",
+ "peeking_take_while",
+ "prettyplease",
+ "proc-macro2",
+ "quote",
+ "regex",
+ "rustc-hash 1.1.0",
+ "shlex",
+ "syn 2.0.114",
+ "which 4.4.2",
+]
+
 [[package]]
 name = "bindgen"
 version = "0.69.5"
@@ -1827,7 +1895,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0686c856aa6aac0c4498f936d7d6a02df690f614c03e4d906d1018062b5c5e2c"
 dependencies = [
 "once_cell",
- "proc-macro-crate",
+ "proc-macro-crate 3.4.0",
 "proc-macro2",
 "quote",
 "syn 2.0.114",
@@ -3889,7 +3957,7 @@ dependencies = [
 "tokio-socks",
 "tokio-util",
 "tower 0.5.3",
- "tower-http",
+ "tower-http 0.6.8",
 "tower-service",
 ]

@@ -4031,7 +4099,7 @@ dependencies = [
 "http-body-util",
 "log",
 "num-bigint",
- "prost",
+ "prost 0.13.5",
 "prost-build",
 "rand 0.8.5",
 "rusqlite",
@@ -4654,7 +4722,7 @@ dependencies = [
 "deno_error",
 "futures",
 "num-bigint",
- "prost",
+ "prost 0.13.5",
 "serde",
 "uuid",
 ]
@@ -4674,7 +4742,7 @@ dependencies = [
 "futures",
 "http 1.4.0",
 "log",
- "prost",
+ "prost 0.13.5",
 "rand 0.8.5",
 "serde",
 "serde_json",
@@ -6340,7 +6408,7 @@ dependencies = [
 "thiserror 1.0.69",
 "tokio",
 "tokio-retry2",
- "tonic",
+ "tonic 0.12.3",
 "tower 0.4.13",
 "tracing",
 ]
@@ -6351,9 +6419,9 @@ version = "0.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "886aa8ec755382a1fdf4651f6e6ec01f2f3bf49f2cb0f068b9a74cafd574a715"
 dependencies = [
- "prost",
+ "prost 0.13.5",
 "prost-types",
- "tonic",
+ "tonic 0.12.3",
 ]

 [[package]]
@@ -6516,7 +6584,7 @@ dependencies = [
 "num-traits",
 "rand 0.9.0",
 "rand_distr 0.5.1",
- "zerocopy",
+ "zerocopy 0.8.33",
 ]

 [[package]]
@@ -6576,6 +6644,15 @@ dependencies = [
 "syn 2.0.114",
 ]

+[[package]]
+name = "hashlink"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8094feaf31ff591f651a2664fb9cfd92bba7a60ce3197265e9482ebe753c8f7"
+dependencies = [
+ "hashbrown 0.14.5",
+]
+
 [[package]]
 name = "hashlink"
 version = "0.9.1"
@@ -6858,6 +6935,12 @@ dependencies = [
 "pin-project-lite",
 ]

+[[package]]
+name = "http-range-header"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "add0ab9360ddbd88cfeb3bd9574a1d85cfdfa14db10b3e21d3700dbc4328758f"
+
 [[package]]
 name = "httparse"
 version = "1.10.1"
@@ -7006,6 +7089,24 @@ dependencies = [
 "tokio-rustls 0.24.1",
 ]

+[[package]]
+name = "hyper-rustls"
+version = "0.25.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "399c78f9338483cb7e630c8474b07268983c6bd5acee012e4211f9f7bb21b070"
+dependencies = [
+ "futures-util",
+ "http 0.2.12",
+ "hyper 0.14.32",
+ "log",
+ "rustls 0.22.4",
+ "rustls-native-certs 0.7.3",
+ "rustls-pki-types",
+ "tokio",
+ "tokio-rustls 0.25.0",
+ "webpki-roots 0.26.11",
+]
+
 [[package]]
 name = "hyper-rustls"
 version = "0.26.0"
@@ -7044,6 +7145,18 @@ dependencies = [
 "webpki-roots 1.0.5",
 ]

+[[package]]
+name = "hyper-timeout"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1"
+dependencies = [
+ "hyper 0.14.32",
+ "pin-project-lite",
+ "tokio",
+ "tokio-io-timeout",
+]
+
 [[package]]
 name = "hyper-timeout"
 version = "0.5.2"
@@ -7777,7 +7890,7 @@ dependencies = [
 "hyper 1.8.1",
 "hyper-http-proxy",
 "hyper-rustls 0.27.7",
- "hyper-timeout",
+ "hyper-timeout 0.5.2",
 "hyper-util",
 "jsonpath-rust",
 "k8s-openapi",
@@ -7792,7 +7905,7 @@ dependencies = [
 "tokio",
 "tokio-util",
 "tower 0.5.3",
- "tower-http",
+ "tower-http 0.6.8",
 "tracing",
 ]

@@ -8036,6 +8149,142 @@ dependencies = [
 "redox_syscall 0.7.0",
 ]

+[[package]]
+name = "libsql"
+version = "0.9.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2329faffc510cc3c6b4f00169a39177cc7099d3ed7647fc92f7cf26e53a8d976"
+dependencies = [
+ "anyhow",
+ "async-stream",
+ "async-trait",
+ "base64 0.21.7",
+ "bincode",
+ "bitflags 2.9.4",
+ "bytes",
+ "chrono",
+ "crc32fast",
+ "fallible-iterator 0.3.0",
+ "futures",
+ "http 0.2.12",
+ "hyper 0.14.32",
+ "hyper-rustls 0.25.0",
+ "libsql-hrana",
+ "libsql-sqlite3-parser",
+ "libsql-sys",
+ "libsql_replication",
+ "parking_lot",
+ "serde",
+ "serde_json",
+ "thiserror 1.0.69",
+ "tokio",
+ "tokio-stream",
+ "tokio-util",
+ "tonic 0.11.0",
+ "tonic-web",
+ "tower 0.4.13",
+ "tower-http 0.4.4",
+ "tracing",
+ "uuid",
+ "zerocopy 0.7.35",
+]
+
+[[package]]
+name = "libsql-ffi"
+version = "0.9.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6cd1c1662822495393327856774f6803be25d85bfdcd5b9d4af35458f5daaf75"
+dependencies = [
+ "bindgen 0.66.1",
+ "cc",
+ "cmake",
+ "glob",
+]
+
+[[package]]
+name = "libsql-hrana"
+version = "0.9.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "646d0aa75e412769018422f0da798f72e93bd51964f0b2ddad4317aa779ae444"
+dependencies = [
+ "base64 0.21.7",
+ "bytes",
+ "prost 0.12.6",
+ "serde",
+]
+
+[[package]]
+name = "libsql-rusqlite"
+version = "0.9.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a4ce3a78c6e3c2b23b02ab6272df8340e1c53380497979d456882254f348d5f"
+dependencies = [
+ "bitflags 2.9.4",
+ "fallible-iterator 0.2.0",
+ "fallible-streaming-iterator",
+ "hashlink 0.8.4",
+ "libsql-ffi",
+ "smallvec",
+]
+
+[[package]]
+name = "libsql-sqlite3-parser"
+version = "0.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15a90128c708356af8f7d767c9ac2946692c9112b4f74f07b99a01a60680e413"
+dependencies = [
+ "bitflags 2.9.4",
+ "cc",
+ "fallible-iterator 0.3.0",
+ "indexmap 2.11.1",
+ "log",
+ "memchr",
+ "phf 0.11.3",
+ "phf_codegen",
+ "phf_shared 0.11.3",
+ "uncased",
+]
+
+[[package]]
+name = "libsql-sys"
+version = "0.9.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a3c326fcfc36fe7578238d5ee6b58c529f8c76372acd61ec50267529cdaff95"
+dependencies = [
+ "bytes",
+ "libsql-ffi",
+ "libsql-rusqlite",
+ "once_cell",
+ "tracing",
+ "zerocopy 0.7.35",
+]
+
+[[package]]
+name = "libsql_replication"
+version = "0.9.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d9a2e469ac8400659bd31f81a745908bcc5cb6b40be2f2ff8de90b15bec5501"
+dependencies = [
+ "aes 0.8.3",
+ "async-stream",
+ "async-trait",
+ "bytes",
+ "cbc",
+ "libsql-rusqlite",
+ "libsql-sys",
+ "parking_lot",
+ "prost 0.12.6",
+ "serde",
+ "thiserror 1.0.69",
+ "tokio",
+ "tokio-stream",
+ "tokio-util",
+ "tonic 0.11.0",
+ "tracing",
+ "uuid",
+ "zerocopy 0.7.35",
+]
+
 [[package]]
 name = "libsqlite3-sys"
 version = "0.30.1"
@@ -8654,7 +8903,7 @@ dependencies = [
 "darling 0.20.11",
 "heck 0.5.0",
 "num-bigint",
- "proc-macro-crate",
+ "proc-macro-crate 3.4.0",
 "proc-macro-error2",
 "proc-macro2",
 "quote",
@@ -9223,7 +9472,7 @@ version = "0.7.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff32365de1b6743cb203b710788263c44a03de03802daf96092f2da4fe6ba4d7"
 dependencies = [
- "proc-macro-crate",
+ "proc-macro-crate 3.4.0",
 "proc-macro2",
 "quote",
 "syn 2.0.114",
@@ -9554,11 +9803,11 @@ dependencies = [
 "opentelemetry-http",
 "opentelemetry-proto 0.27.0",
 "opentelemetry_sdk 0.27.1",
- "prost",
+ "prost 0.13.5",
 "serde_json",
 "thiserror 1.0.69",
 "tokio",
- "tonic",
+ "tonic 0.12.3",
 "tracing",
 ]

@@ -9571,9 +9820,9 @@ dependencies = [
 "hex",
 "opentelemetry 0.27.1",
 "opentelemetry_sdk 0.27.1",
- "prost",
+ "prost 0.13.5",
 "serde",
- "tonic",
+ "tonic 0.12.3",
 ]

 [[package]]
@@ -9586,9 +9835,9 @@ dependencies = [
 "hex",
 "opentelemetry 0.29.1",
 "opentelemetry_sdk 0.29.0",
- "prost",
+ "prost 0.13.5",
 "serde",
- "tonic",
+ "tonic 0.12.3",
 "tracing",
 ]

@@ -9874,6 +10123,12 @@ dependencies = [
 "hmac",
 ]

+[[package]]
+name = "peeking_take_while"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
+
 [[package]]
 name = "pem"
 version = "1.1.1"
@@ -10041,6 +10296,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "67eabc2ef2a60eb7faa00097bd1ffdb5bd28e62bf39990626a582201b7a754e5"
 dependencies = [
 "siphasher 1.0.1",
+ "uncased",
 ]

 [[package]]
@@ -10277,7 +10533,7 @@ version = "0.2.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
 dependencies = [
- "zerocopy",
+ "zerocopy 0.8.33",
 ]

 [[package]]
@@ -10305,6 +10561,16 @@ dependencies = [
 "elliptic-curve",
 ]

+[[package]]
+name = "proc-macro-crate"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f4c021e1093a56626774e81216a4ce732a735e5bad4868a03f3ed65ca0c3919"
+dependencies = [
+ "once_cell",
+ "toml_edit 0.19.15",
+]
+
 [[package]]
 name = "proc-macro-crate"
 version = "3.4.0"
@@ -10451,6 +10717,16 @@ dependencies = [
 "thiserror 2.0.18",
 ]

+[[package]]
+name = "prost"
+version = "0.12.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "deb1435c188b76130da55f17a466d252ff7b1418b2ad3e037d127b94e3411f29"
+dependencies = [
+ "bytes",
+ "prost-derive 0.12.6",
+]
+
 [[package]]
 name = "prost"
 version = "0.13.5"
@@ -10458,7 +10734,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5"
 dependencies = [
 "bytes",
- "prost-derive",
+ "prost-derive 0.13.5",
 ]

 [[package]]
@@ -10474,13 +10750,26 @@ dependencies = [
 "once_cell",
 "petgraph",
 "prettyplease",
- "prost",
+ "prost 0.13.5",
 "prost-types",
 "regex",
 "syn 2.0.114",
 "tempfile",
 ]

+[[package]]
+name = "prost-derive"
+version = "0.12.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "81bddcdb20abf9501610992b6759a4c888aef7d1a7247ef75e2404275ac24af1"
+dependencies = [
+ "anyhow",
+ "itertools 0.12.1",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.114",
+]
+
 [[package]]
 name = "prost-derive"
 version = "0.13.5"
@@ -10500,7 +10789,7 @@ version = "0.13.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16"
 dependencies = [
- "prost",
+ "prost 0.13.5",
 ]

 [[package]]
@@ -10725,7 +11014,7 @@ checksum = "3779b94aeb87e8bd4e834cee3650289ee9e0d5677f976ecdb6d219e5f4f6cd94"
 dependencies = [
 "rand_chacha 0.9.0",
 "rand_core 0.9.5",
- "zerocopy",
+ "zerocopy 0.8.33",
 ]

 [[package]]
@@ -11020,6 +11309,12 @@ version = "0.8.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7a2d987857b319362043e95f5353c0535c1f58eec5336fdfcf626430af7def58"

+[[package]]
+name = "relative-path"
+version = "1.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba39f3699c378cd8970968dcbff9c43159ea4cfbd88d43c00b22f2ef10a435d2"
+
 [[package]]
 name = "rend"
 version = "0.4.2"
@@ -11061,13 +11356,13 @@ dependencies = [
 "serde",
 "serde_json",
 "serde_urlencoded",
- "sync_wrapper",
+ "sync_wrapper 1.0.2",
 "tokio",
 "tokio-native-tls",
 "tokio-rustls 0.26.4",
 "tokio-util",
 "tower 0.5.3",
- "tower-http",
+ "tower-http 0.6.8",
 "tower-service",
 "url",
 "wasm-bindgen",
@@ -11108,12 +11403,12 @@ dependencies = [
 "serde",
 "serde_json",
 "serde_urlencoded",
- "sync_wrapper",
+ "sync_wrapper 1.0.2",
 "tokio",
 "tokio-rustls 0.26.4",
 "tokio-util",
 "tower 0.5.3",
- "tower-http",
+ "tower-http 0.6.8",
 "tower-service",
 "url",
 "wasm-bindgen",
@@ -11310,6 +11605,54 @@ dependencies = [
 "serde_derive",
 ]

+[[package]]
+name = "rquickjs"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d16661bff09e9ed8e01094a188b463de45ec0693ade55b92ed54027d7ba7c40c"
+dependencies = [
+ "rquickjs-core",
+ "rquickjs-macro",
+]
+
+[[package]]
+name = "rquickjs-core"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c8db6379e204ef84c0811e90e7cc3e3e4d7688701db68a00d14a6db6849087b"
+dependencies = [
+ "async-lock",
+ "relative-path",
+ "rquickjs-sys",
+]
+
+[[package]]
+name = "rquickjs-macro"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6041104330c019fcd936026ae05e2446f5e8a2abef329d924f25424b7052a2f3"
+dependencies = [
+ "convert_case 0.6.0",
+ "fnv",
+ "ident_case",
+ "indexmap 2.11.1",
+ "proc-macro-crate 1.3.1",
+ "proc-macro2",
+ "quote",
+ "rquickjs-core",
+ "syn 2.0.114",
+]
+
+[[package]]
+name = "rquickjs-sys"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4bc352c6b663604c3c186c000cfcc6c271f4b50bc135a285dd6d4f2a42f9790a"
+dependencies = [
+ "bindgen 0.69.5",
+ "cc",
+]
+
 [[package]]
 name = "rsa"
 version = "0.9.10"
@@ -13426,6 +13769,12 @@ dependencies = [
 "unicode-ident",
 ]

+[[package]]
+name = "sync_wrapper"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160"
+
 [[package]]
 name = "sync_wrapper"
 version = "1.0.2"
@@ -14054,6 +14403,16 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "tokio-io-timeout"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0bd86198d9ee903fedd2f9a2e72014287c0d9167e4ae43b5853007205dda1b76"
+dependencies = [
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-macros"
 version = "2.5.0"
@@ -14202,6 +14561,17 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "tokio-test"
+version = "0.4.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f6d24790a10a7af737693a3e8f1d03faef7e6ca0cc99aae5066f533766de545"
+dependencies = [
+ "futures-core",
+ "tokio",
+ "tokio-stream",
+]
+
 [[package]]
 name = "tokio-tungstenite"
 version = "0.21.0"
@@ -14336,6 +14706,33 @@ dependencies = [
 "winnow 0.7.14",
 ]

+[[package]]
+name = "tonic"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76c4eb7a4e9ef9d4763600161f12f5070b92a578e1b634db88a6887844c91a13"
+dependencies = [
+ "async-stream",
+ "async-trait",
+ "axum 0.6.20",
+ "base64 0.21.7",
+ "bytes",
+ "h2 0.3.27",
+ "http 0.2.12",
+ "http-body 0.4.6",
+ "hyper 0.14.32",
+ "hyper-timeout 0.4.1",
+ "percent-encoding",
+ "pin-project",
+ "prost 0.12.6",
+ "tokio",
+ "tokio-stream",
+ "tower 0.4.13",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
 [[package]]
 name = "tonic"
 version = "0.12.3"
@@ -14353,11 +14750,11 @@ dependencies = [
 "http-body 1.0.1",
 "http-body-util",
 "hyper 1.8.1",
- "hyper-timeout",
+ "hyper-timeout 0.5.2",
 "hyper-util",
 "percent-encoding",
 "pin-project",
- "prost",
+ "prost 0.13.5",
 "rustls-native-certs 0.8.3",
 "rustls-pemfile 2.2.0",
 "socket2 0.5.10",
@@ -14371,6 +14768,26 @@ dependencies = [
 "webpki-roots 0.26.11",
 ]

+[[package]]
+name = "tonic-web"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc3b0e1cedbf19fdfb78ef3d672cb9928e0a91a9cb4629cc0c916e8cff8aaaa1"
+dependencies = [
+ "base64 0.21.7",
+ "bytes",
+ "http 0.2.12",
+ "http-body 0.4.6",
+ "hyper 0.14.32",
+ "pin-project",
+ "tokio-stream",
+ "tonic 0.11.0",
+ "tower-http 0.4.4",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
 [[package]]
 name = "tower"
 version = "0.4.13"
@@ -14400,7 +14817,7 @@ dependencies = [
 "futures-core",
 "futures-util",
 "pin-project-lite",
- "sync_wrapper",
+ "sync_wrapper 1.0.2",
 "tokio",
 "tokio-util",
 "tower-layer",
@@ -14425,6 +14842,43 @@ dependencies = [
 "tower-service",
 ]

+[[package]]
+name = "tower-http"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "61c5bb1d698276a2443e5ecfabc1008bf15a36c12e6a7176e7bf089ea9131140"
+dependencies = [
+ "bitflags 2.9.4",
+ "bytes",
+ "futures-core",
+ "futures-util",
+ "http 0.2.12",
+ "http-body 0.4.6",
+ "http-range-header",
+ "pin-project-lite",
+ "tower 0.4.13",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "tower-http"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e9cd434a998747dd2c4276bc96ee2e0c7a2eadf3cae88e52be55a05fa9053f5"
+dependencies = [
+ "bitflags 2.9.4",
+ "bytes",
+ "http 1.4.0",
+ "http-body 1.0.1",
+ "http-body-util",
+ "pin-project-lite",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
 [[package]]
 name = "tower-http"
 version = "0.6.8"
@@ -14790,6 +15244,15 @@ dependencies = [
 "web-time",
 ]

+[[package]]
+name = "uncased"
+version = "0.9.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e1b88fcfe09e89d3866a5c11019378088af2d24c3fbd4f0543f96b479ec90697"
+dependencies = [
+ "version_check",
+]
+
 [[package]]
 name = "unic-char-property"
 version = "0.9.0"
@@ -15718,10 +16181,10 @@ dependencies = [
 "tokio-stream",
 "tokio-tungstenite 0.24.0",
 "tokio-util",
- "tonic",
+ "tonic 0.12.3",
 "tower 0.5.3",
 "tower-cookies",
- "tower-http",
+ "tower-http 0.6.8",
 "tracing",
 "tracing-subscriber",
 "ulid",
@@ -15866,7 +16329,7 @@ dependencies = [
 "tokio-postgres 0.7.13",
 "tokio-stream",
 "tokio-util",
- "tonic",
+ "tonic 0.12.3",
 "tracing",
 "tracing-appender",
 "tracing-opentelemetry",
@@ -15920,6 +16383,29 @@ dependencies = [
 "windmill-common",
 ]

+[[package]]
+name = "windmill-local"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "async-recursion",
+ "axum 0.7.9",
+ "chrono",
+ "libsql",
+ "rquickjs",
+ "serde",
+ "serde_json",
+ "thiserror 1.0.69",
+ "tokio",
+ "tokio-test",
+ "tower 0.4.13",
+ "tower-http 0.5.2",
+ "tracing",
+ "tracing-subscriber",
+ "uuid",
+ "windmill-common",
+]
+
 [[package]]
 name = "windmill-macros"
 version = "1.613.4"
@@ -17128,13 +17614,34 @@ dependencies = [
 "synstructure 0.13.2",
 ]

+[[package]]
+name = "zerocopy"
+version = "0.7.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1b9b4fd18abc82b8136838da5d50bae7bdea537c574d8dc1a34ed098d6c166f0"
+dependencies = [
+ "byteorder",
+ "zerocopy-derive 0.7.35",
+]
+
 [[package]]
 name = "zerocopy"
 version = "0.8.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "668f5168d10b9ee831de31933dc111a459c97ec93225beb307aed970d1372dfd"
 dependencies = [
- "zerocopy-derive",
+ "zerocopy-derive 0.8.33",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.7.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa4f8080344d4671fb4e831a13ad1e68092748387dfc4f55e356242fae12ce3e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.114",
 ]

 [[package]]
--- a/backend/Cargo.toml
+++ b/backend/Cargo.toml
@@ -18,6 +18,7 @@ members = [
    "./windmill-indexer",
    "./windmill-macros",
    "./windmill-oauth",
+    "./windmill-local",
    "./parsers/windmill-parser",
    "./parsers/windmill-parser-ts",
    "./parsers/windmill-parser-go",
--- a/backend/windmill-local/Cargo.toml
+++ b/backend/windmill-local/Cargo.toml
@@ -0,0 +1,52 @@
+[package]
+name = "windmill-local"
+version = "0.1.0"
+edition = "2021"
+description = "Windmill local mode with libSQL/Turso support for preview execution"
+
+[dependencies]
+# libSQL - Turso's SQLite fork (used for local and remote Turso connections)
+# Note: libsql IS the Turso database driver - Turso is built on libSQL
+libsql = "0.9"
+
+# Async runtime
+tokio = { version = "1", features = ["full", "sync", "macros"] }
+
+# Serialization
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+
+# Error handling
+anyhow = "1"
+thiserror = "1"
+
+# UUID for job IDs
+uuid = { version = "1", features = ["v4", "serde"] }
+
+# Timestamps
+chrono = { version = "0.4", features = ["serde"] }
+
+# Tracing
+tracing = "0.1"
+
+# HTTP server
+axum = "0.7"
+tower = "0.4"
+tower-http = { version = "0.5", features = ["cors", "trace"] }
+
+# Windmill types (flows, scripts, etc.)
+windmill-common = { path = "../windmill-common", default-features = false }
+
+# For input transforms (JavaScript evaluation)
+rquickjs = { version = "0.8", features = ["bindgen", "classes", "loader", "array-buffer", "futures"] }
+
+# For recursive async functions
+async-recursion = "1"
+
+[dev-dependencies]
+tokio-test = "0.4"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+
+[[example]]
+name = "local_server"
+path = "examples/local_server.rs"
--- a/backend/windmill-local/examples/local_server.rs
+++ b/backend/windmill-local/examples/local_server.rs
@@ -0,0 +1,65 @@
+//! Example: Run the Windmill local server
+//!
+//! This demonstrates running a local Windmill server with libSQL/SQLite backend.
+//!
+//! Run with:
+//!     cargo run -p windmill-local --example local_server
+//!
+//! Test with:
+//!     # Health check
+//!     curl http://localhost:8000/health
+//!
+//!     # Run a bash preview (async)
+//!     curl -X POST http://localhost:8000/api/w/local/jobs/run/preview \
+//!       -H "Content-Type: application/json" \
+//!       -d '{"content": "echo Hello World", "language": "bash"}'
+//!
+//!     # Run a bash preview and wait for result
+//!     curl -X POST http://localhost:8000/api/w/local/jobs/run_wait_result/preview \
+//!       -H "Content-Type: application/json" \
+//!       -d '{"content": "echo 42", "language": "bash"}'
+//!
+//!     # Run a Python preview
+//!     curl -X POST http://localhost:8000/api/w/local/jobs/run_wait_result/preview \
+//!       -H "Content-Type: application/json" \
+//!       -d '{"content": "def main(x=1): return x * 2", "language": "python3", "args": {"x": 21}}'
+//!
+//!     # Run a flow preview (linear flow with two steps)
+//!     curl -X POST http://localhost:8000/api/w/local/jobs/run_wait_result/preview_flow \
+//!       -H "Content-Type: application/json" \
+//!       -d '{
+//!         "value": {
+//!           "modules": [
+//!             {"id": "step1", "value": {"type": "rawscript", "language": "bash", "content": "echo 10"}},
+//!             {"id": "step2", "value": {"type": "identity"}}
+//!           ]
+//!         },
+//!         "args": {}
+//!       }'
+
+use std::net::SocketAddr;
+use windmill_local::LocalServer;
+
+#[tokio::main]
+async fn main() {
+    // Initialize tracing
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::from_default_env()
+                .add_directive("windmill_local=info".parse().unwrap()),
+        )
+        .init();
+
+    let addr: SocketAddr = "0.0.0.0:8000".parse().unwrap();
+    println!("Starting Windmill Local Server on {}", addr);
+    println!();
+    println!("Test endpoints:");
+    println!("  Health:    curl http://localhost:8000/health");
+    println!("  Preview:   curl -X POST http://localhost:8000/api/w/local/jobs/run_wait_result/preview \\");
+    println!("               -H 'Content-Type: application/json' \\");
+    println!("               -d '{{\"content\": \"echo Hello\", \"language\": \"bash\"}}'");
+    println!();
+
+    let server = LocalServer::new(addr).await.expect("Failed to create server");
+    server.run().await.expect("Server error");
+}
--- a/backend/windmill-local/src/db.rs
+++ b/backend/windmill-local/src/db.rs
@@ -0,0 +1,161 @@
+//! Database connection and initialization for local mode
+
+use libsql::{Builder, Connection, Database};
+use std::sync::Arc;
+use tokio::sync::Mutex;
+
+use crate::error::Result;
+use crate::schema;
+
+/// Local database wrapper
+///
+/// For local mode, we use a single connection with in-process coordination.
+/// This simplifies the implementation since we don't need `FOR UPDATE SKIP LOCKED`.
+pub struct LocalDb {
+    #[allow(dead_code)]
+    db: Database,
+    /// Single connection for all operations (simplifies transaction handling)
+    conn: Arc<Mutex<Connection>>,
+}
+
+impl LocalDb {
+    /// Create an in-memory database (for testing/ephemeral use)
+    pub async fn in_memory() -> Result<Self> {
+        let db = Builder::new_local(":memory:").build().await?;
+        let conn = db.connect()?;
+        let local_db = Self {
+            db,
+            conn: Arc::new(Mutex::new(conn)),
+        };
+        local_db.init_schema().await?;
+        Ok(local_db)
+    }
+
+    /// Create a file-based database
+    pub async fn file(path: &str) -> Result<Self> {
+        let db = Builder::new_local(path).build().await?;
+        let conn = db.connect()?;
+        let local_db = Self {
+            db,
+            conn: Arc::new(Mutex::new(conn)),
+        };
+        local_db.init_schema().await?;
+        Ok(local_db)
+    }
+
+    /// Create a Turso remote database connection
+    /// This would be used for the multi-writer scenario
+    pub async fn turso_remote(url: &str, auth_token: &str) -> Result<Self> {
+        let db = Builder::new_remote(url.to_string(), auth_token.to_string())
+            .build()
+            .await?;
+        let conn = db.connect()?;
+        let local_db = Self {
+            db,
+            conn: Arc::new(Mutex::new(conn)),
+        };
+        local_db.init_schema().await?;
+        Ok(local_db)
+    }
+
+    /// Initialize the schema
+    async fn init_schema(&self) -> Result<()> {
+        let conn = self.conn.lock().await;
+        // Execute schema as multiple statements
+        conn.execute_batch(schema::SCHEMA).await?;
+        Ok(())
+    }
+
+    /// Reset the database (drop and recreate all tables)
+    pub async fn reset(&self) -> Result<()> {
+        let conn = self.conn.lock().await;
+        conn.execute_batch(schema::DROP_SCHEMA).await?;
+        conn.execute_batch(schema::SCHEMA).await?;
+        Ok(())
+    }
+
+    /// Get a reference to the connection (locked)
+    pub async fn conn(&self) -> tokio::sync::MutexGuard<'_, Connection> {
+        self.conn.lock().await
+    }
+
+    /// Execute a simple query that returns no rows
+    pub async fn execute(&self, sql: &str, params: impl libsql::params::IntoParams) -> Result<u64> {
+        let conn = self.conn.lock().await;
+        let rows_affected = conn.execute(sql, params).await?;
+        Ok(rows_affected)
+    }
+
+    /// Execute a query and return all rows
+    pub async fn query(
+        &self,
+        sql: &str,
+        params: impl libsql::params::IntoParams,
+    ) -> Result<libsql::Rows> {
+        let conn = self.conn.lock().await;
+        let rows = conn.query(sql, params).await?;
+        Ok(rows)
+    }
+
+    /// Execute a batch of statements (for transactions)
+    pub async fn execute_batch(&self, sql: &str) -> Result<()> {
+        let conn = self.conn.lock().await;
+        conn.execute_batch(sql).await?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_in_memory_db() {
+        let db = LocalDb::in_memory().await.unwrap();
+
+        // Verify tables exist
+        let rows = db
+            .query(
+                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name",
+                (),
+            )
+            .await
+            .unwrap();
+
+        let mut tables = Vec::new();
+        let mut rows = rows;
+        while let Some(row) = rows.next().await.unwrap() {
+            let name: String = row.get(0).unwrap();
+            tables.push(name);
+        }
+
+        assert!(tables.contains(&"v2_job".to_string()));
+        assert!(tables.contains(&"v2_job_queue".to_string()));
+        assert!(tables.contains(&"v2_job_completed".to_string()));
+    }
+
+    #[tokio::test]
+    async fn test_reset_db() {
+        let db = LocalDb::in_memory().await.unwrap();
+
+        // Insert a job
+        db.execute(
+            "INSERT INTO v2_job (id, kind) VALUES ('test-uuid', 'preview')",
+            (),
+        )
+        .await
+        .unwrap();
+
+        // Reset
+        db.reset().await.unwrap();
+
+        // Verify job is gone
+        let mut rows = db
+            .query("SELECT COUNT(*) FROM v2_job", ())
+            .await
+            .unwrap();
+        let row = rows.next().await.unwrap().unwrap();
+        let count: i64 = row.get(0).unwrap();
+        assert_eq!(count, 0);
+    }
+}
--- a/backend/windmill-local/src/error.rs
+++ b/backend/windmill-local/src/error.rs
@@ -0,0 +1,29 @@
+//! Error types for local mode
+
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum LocalError {
+    #[error("Database error: {0}")]
+    Database(#[from] libsql::Error),
+
+    #[error("Serialization error: {0}")]
+    Serialization(#[from] serde_json::Error),
+
+    #[error("Job not found: {0}")]
+    JobNotFound(uuid::Uuid),
+
+    #[error("Invalid job state: {0}")]
+    InvalidJobState(String),
+
+    #[error("Queue is empty")]
+    QueueEmpty,
+
+    #[error("Execution error: {0}")]
+    Execution(String),
+
+    #[error("Timeout")]
+    Timeout,
+}
+
+pub type Result<T> = std::result::Result<T, LocalError>;
--- a/backend/windmill-local/src/executor.rs
+++ b/backend/windmill-local/src/executor.rs
@@ -0,0 +1,330 @@
+//! Simple script executor for local mode
+//!
+//! This is a minimal executor that supports a few languages for demonstration.
+//! A full implementation would integrate with windmill-worker's execution logic.
+
+use std::process::Stdio;
+use tokio::process::Command;
+use tokio::io::AsyncReadExt;
+
+use crate::error::{LocalError, Result};
+use crate::jobs::ScriptLang;
+
+/// Result of script execution
+#[derive(Debug)]
+pub struct ExecutionResult {
+    pub success: bool,
+    pub result: serde_json::Value,
+    pub logs: String,
+}
+
+/// Execute a script with the given language and arguments
+pub async fn execute_script(
+    language: ScriptLang,
+    code: &str,
+    args: &serde_json::Value,
+) -> Result<ExecutionResult> {
+    match language {
+        ScriptLang::Bash => execute_bash(code, args).await,
+        ScriptLang::Python3 => execute_python(code, args).await,
+        ScriptLang::Deno => execute_deno(code, args).await,
+        ScriptLang::Bun => execute_bun(code, args).await,
+        _ => Ok(ExecutionResult {
+            success: false,
+            result: serde_json::json!({
+                "error": format!("Language {:?} not supported in local mode yet", language)
+            }),
+            logs: format!("Language {:?} not supported in local mode", language),
+        }),
+    }
+}
+
+/// Execute a bash script
+async fn execute_bash(code: &str, args: &serde_json::Value) -> Result<ExecutionResult> {
+    // Create environment variables from args
+    let mut env_vars = Vec::new();
+    if let serde_json::Value::Object(map) = args {
+        for (key, value) in map {
+            let val_str = match value {
+                serde_json::Value::String(s) => s.clone(),
+                _ => value.to_string(),
+            };
+            env_vars.push((key.to_uppercase(), val_str));
+        }
+    }
+
+    let mut child = Command::new("bash")
+        .arg("-c")
+        .arg(code)
+        .envs(env_vars)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .map_err(|e| LocalError::Execution(format!("Failed to spawn bash: {}", e)))?;
+
+    let status = child
+        .wait()
+        .await
+        .map_err(|e| LocalError::Execution(format!("Failed to wait for bash: {}", e)))?;
+
+    let mut stdout = String::new();
+    let mut stderr = String::new();
+
+    if let Some(mut out) = child.stdout.take() {
+        out.read_to_string(&mut stdout).await.ok();
+    }
+    if let Some(mut err) = child.stderr.take() {
+        err.read_to_string(&mut stderr).await.ok();
+    }
+
+    let logs = format!("{}{}", stdout, stderr);
+    let success = status.success();
+
+    // Try to parse stdout as JSON, otherwise use as string
+    let result = if success {
+        let trimmed = stdout.trim();
+        serde_json::from_str(trimmed).unwrap_or_else(|_| serde_json::json!(trimmed))
+    } else {
+        serde_json::json!({ "error": stderr.trim() })
+    };
+
+    Ok(ExecutionResult {
+        success,
+        result,
+        logs,
+    })
+}
+
+/// Execute a Python script
+async fn execute_python(code: &str, args: &serde_json::Value) -> Result<ExecutionResult> {
+    // Wrap the code to handle args and return JSON result
+    let wrapped_code = format!(
+        r#"
+import json
+import sys
+
+# Args passed as JSON
+args = json.loads('''{}''')
+
+# User code
+{}
+
+# Call main if it exists
+if 'main' in dir():
+    result = main(**args)
+    print(json.dumps(result))
+"#,
+        serde_json::to_string(args)?,
+        code
+    );
+
+    let mut child = Command::new("python3")
+        .arg("-c")
+        .arg(&wrapped_code)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .map_err(|e| LocalError::Execution(format!("Failed to spawn python3: {}", e)))?;
+
+    let status = child
+        .wait()
+        .await
+        .map_err(|e| LocalError::Execution(format!("Failed to wait for python3: {}", e)))?;
+
+    let mut stdout = String::new();
+    let mut stderr = String::new();
+
+    if let Some(mut out) = child.stdout.take() {
+        out.read_to_string(&mut stdout).await.ok();
+    }
+    if let Some(mut err) = child.stderr.take() {
+        err.read_to_string(&mut stderr).await.ok();
+    }
+
+    let logs = format!("{}{}", stdout, stderr);
+    let success = status.success();
+
+    let result = if success {
+        let trimmed = stdout.trim();
+        // Get the last line as result (in case there's debug output)
+        let last_line = trimmed.lines().last().unwrap_or("");
+        serde_json::from_str(last_line).unwrap_or_else(|_| serde_json::json!(trimmed))
+    } else {
+        serde_json::json!({ "error": stderr.trim() })
+    };
+
+    Ok(ExecutionResult {
+        success,
+        result,
+        logs,
+    })
+}
+
+/// Execute a Deno/TypeScript script
+async fn execute_deno(code: &str, args: &serde_json::Value) -> Result<ExecutionResult> {
+    // Wrap the code to handle args and return JSON result
+    let wrapped_code = format!(
+        r#"
+const args = {};
+
+{}
+
+// Call main if it exists
+if (typeof main === 'function') {{
+    const result = await main(args);
+    console.log(JSON.stringify(result));
+}}
+"#,
+        serde_json::to_string(args)?,
+        code
+    );
+
+    let mut child = Command::new("deno")
+        .arg("eval")
+        .arg("--unstable")
+        .arg(&wrapped_code)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .map_err(|e| LocalError::Execution(format!("Failed to spawn deno: {}", e)))?;
+
+    let status = child
+        .wait()
+        .await
+        .map_err(|e| LocalError::Execution(format!("Failed to wait for deno: {}", e)))?;
+
+    let mut stdout = String::new();
+    let mut stderr = String::new();
+
+    if let Some(mut out) = child.stdout.take() {
+        out.read_to_string(&mut stdout).await.ok();
+    }
+    if let Some(mut err) = child.stderr.take() {
+        err.read_to_string(&mut stderr).await.ok();
+    }
+
+    let logs = format!("{}{}", stdout, stderr);
+    let success = status.success();
+
+    let result = if success {
+        let trimmed = stdout.trim();
+        let last_line = trimmed.lines().last().unwrap_or("");
+        serde_json::from_str(last_line).unwrap_or_else(|_| serde_json::json!(trimmed))
+    } else {
+        serde_json::json!({ "error": stderr.trim() })
+    };
+
+    Ok(ExecutionResult {
+        success,
+        result,
+        logs,
+    })
+}
+
+/// Execute a Bun/TypeScript script
+async fn execute_bun(code: &str, args: &serde_json::Value) -> Result<ExecutionResult> {
+    // Similar to Deno but using Bun
+    let wrapped_code = format!(
+        r#"
+const args = {};
+
+{}
+
+// Call main if it exists
+if (typeof main === 'function') {{
+    const result = await main(args);
+    console.log(JSON.stringify(result));
+}}
+"#,
+        serde_json::to_string(args)?,
+        code
+    );
+
+    let mut child = Command::new("bun")
+        .arg("eval")
+        .arg(&wrapped_code)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .map_err(|e| LocalError::Execution(format!("Failed to spawn bun: {}", e)))?;
+
+    let status = child
+        .wait()
+        .await
+        .map_err(|e| LocalError::Execution(format!("Failed to wait for bun: {}", e)))?;
+
+    let mut stdout = String::new();
+    let mut stderr = String::new();
+
+    if let Some(mut out) = child.stdout.take() {
+        out.read_to_string(&mut stdout).await.ok();
+    }
+    if let Some(mut err) = child.stderr.take() {
+        err.read_to_string(&mut stderr).await.ok();
+    }
+
+    let logs = format!("{}{}", stdout, stderr);
+    let success = status.success();
+
+    let result = if success {
+        let trimmed = stdout.trim();
+        let last_line = trimmed.lines().last().unwrap_or("");
+        serde_json::from_str(last_line).unwrap_or_else(|_| serde_json::json!(trimmed))
+    } else {
+        serde_json::json!({ "error": stderr.trim() })
+    };
+
+    Ok(ExecutionResult {
+        success,
+        result,
+        logs,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_bash_execution() {
+        let result = execute_script(
+            ScriptLang::Bash,
+            "echo 42",
+            &serde_json::json!({}),
+        )
+        .await
+        .unwrap();
+
+        assert!(result.success);
+        // Output "42" is parsed as JSON number
+        assert_eq!(result.result, serde_json::json!(42));
+    }
+
+    #[tokio::test]
+    async fn test_bash_with_args() {
+        let result = execute_script(
+            ScriptLang::Bash,
+            "echo $NAME",
+            &serde_json::json!({"name": "world"}),
+        )
+        .await
+        .unwrap();
+
+        assert!(result.success);
+        assert_eq!(result.result, serde_json::json!("world"));
+    }
+
+    #[tokio::test]
+    async fn test_bash_json_output() {
+        let result = execute_script(
+            ScriptLang::Bash,
+            r#"echo '{"key": "value"}'"#,
+            &serde_json::json!({}),
+        )
+        .await
+        .unwrap();
+
+        assert!(result.success);
+        assert_eq!(result.result, serde_json::json!({"key": "value"}));
+    }
+}
--- a/backend/windmill-local/src/flow_executor.rs
+++ b/backend/windmill-local/src/flow_executor.rs
@@ -0,0 +1,750 @@
+//! Flow executor for local mode
+//!
+//! This module implements flow execution using the real Windmill flow types
+//! from windmill-common, but with libSQL as the backend.
+
+use std::collections::HashMap;
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use serde_json::value::RawValue;
+use uuid::Uuid;
+
+use windmill_common::flows::{
+    Branch, FlowModule, FlowModuleValue, FlowValue, InputTransform,
+};
+use windmill_common::scripts::ScriptLang as WmScriptLang;
+
+use crate::db::LocalDb;
+use crate::error::{LocalError, Result};
+use crate::executor::{execute_script, ExecutionResult};
+use crate::jobs::ScriptLang;
+
+/// Flow execution state
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FlowStatus {
+    pub step: usize,
+    pub modules: Vec<ModuleStatus>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub failure_module: Option<FailureModule>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub retry: Option<RetryStatus>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ModuleStatus {
+    pub id: String,
+    #[serde(rename = "type")]
+    pub status_type: ModuleStatusType,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result: Option<serde_json::Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub iterator: Option<IteratorStatus>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub branch_chosen: Option<BranchChosen>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub branchall: Option<BranchAllStatus>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum ModuleStatusType {
+    WaitingForPriorSteps,
+    WaitingForEvents,
+    WaitingForExecutor,
+    InProgress,
+    Success,
+    Failure,
+    Skipped,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct IteratorStatus {
+    pub index: usize,
+    pub itered: Vec<serde_json::Value>,
+    pub args: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BranchChosen {
+    #[serde(rename = "type")]
+    pub branch_type: String,
+    pub branch: Option<usize>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BranchAllStatus {
+    pub branch: usize,
+    pub len: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FailureModule {
+    pub id: String,
+    pub error: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RetryStatus {
+    pub fail_count: u32,
+}
+
+/// Context passed through flow execution
+#[derive(Debug, Clone)]
+pub struct FlowContext {
+    pub flow_input: serde_json::Value,
+    pub previous_result: serde_json::Value,
+    pub results_by_id: HashMap<String, serde_json::Value>,
+}
+
+impl FlowContext {
+    pub fn new(flow_input: serde_json::Value) -> Self {
+        Self {
+            flow_input: flow_input.clone(),
+            previous_result: flow_input,
+            results_by_id: HashMap::new(),
+        }
+    }
+
+    /// Get the evaluation context for input transforms
+    pub fn to_eval_context(&self) -> serde_json::Value {
+        serde_json::json!({
+            "flow_input": self.flow_input,
+            "previous_result": self.previous_result,
+            "results": self.results_by_id,
+        })
+    }
+}
+
+/// Execute a flow and return the result
+pub async fn execute_flow(
+    db: &LocalDb,
+    flow_value: &FlowValue,
+    flow_input: serde_json::Value,
+) -> Result<(serde_json::Value, FlowStatus)> {
+    let mut ctx = FlowContext::new(flow_input);
+    let mut status = FlowStatus {
+        step: 0,
+        modules: Vec::new(),
+        failure_module: None,
+        retry: None,
+    };
+
+    // Execute modules sequentially
+    for (idx, module) in flow_value.modules.iter().enumerate() {
+        status.step = idx;
+
+        let module_status = ModuleStatus {
+            id: module.id.clone(),
+            status_type: ModuleStatusType::InProgress,
+            result: None,
+            iterator: None,
+            branch_chosen: None,
+            branchall: None,
+        };
+        status.modules.push(module_status);
+
+        tracing::info!("Executing flow module {}: {}", idx, module.id);
+
+        match execute_module(db, module, &mut ctx).await {
+            Ok(result) => {
+                // Update context with result
+                ctx.results_by_id.insert(module.id.clone(), result.clone());
+                ctx.previous_result = result.clone();
+
+                // Update status
+                if let Some(ms) = status.modules.last_mut() {
+                    ms.status_type = ModuleStatusType::Success;
+                    ms.result = Some(result);
+                }
+            }
+            Err(e) => {
+                // Module failed
+                tracing::error!("Flow module {} failed: {}", module.id, e);
+
+                if let Some(ms) = status.modules.last_mut() {
+                    ms.status_type = ModuleStatusType::Failure;
+                    ms.result = Some(serde_json::json!({"error": e.to_string()}));
+                }
+
+                status.failure_module = Some(FailureModule {
+                    id: module.id.clone(),
+                    error: e.to_string(),
+                });
+
+                // Check if continue_on_error is set
+                if !module.continue_on_error.unwrap_or(false) {
+                    return Ok((serde_json::json!({"error": e.to_string()}), status));
+                }
+            }
+        }
+    }
+
+    Ok((ctx.previous_result, status))
+}
+
+/// Execute a single flow module
+#[async_recursion::async_recursion]
+async fn execute_module(
+    db: &LocalDb,
+    module: &FlowModule,
+    ctx: &mut FlowContext,
+) -> Result<serde_json::Value> {
+    // Check skip_if condition
+    if let Some(skip_if) = &module.skip_if {
+        let should_skip = evaluate_expr(&skip_if.expr, &ctx.to_eval_context())?;
+        if should_skip.as_bool().unwrap_or(false) {
+            tracing::info!("Skipping module {} due to skip_if condition", module.id);
+            return Ok(ctx.previous_result.clone());
+        }
+    }
+
+    // Parse the module value
+    let module_value: FlowModuleValue = serde_json::from_str(module.value.get())
+        .map_err(|e| LocalError::Execution(format!("Failed to parse module value: {}", e)))?;
+
+    match module_value {
+        FlowModuleValue::Identity => {
+            Ok(ctx.previous_result.clone())
+        }
+
+        FlowModuleValue::RawScript { content, language, input_transforms, .. } => {
+            let args = resolve_input_transforms(&input_transforms, ctx)?;
+            let lang = convert_script_lang(&language);
+            let result = execute_script(lang, &content, &args).await?;
+            if result.success {
+                Ok(result.result)
+            } else {
+                Err(LocalError::Execution(result.result.to_string()))
+            }
+        }
+
+        FlowModuleValue::Script { path, input_transforms, .. } => {
+            // In local mode, we don't have access to saved scripts
+            Err(LocalError::Execution(format!(
+                "Script references (path: {}) are not supported in local mode. Use rawscript instead.",
+                path
+            )))
+        }
+
+        FlowModuleValue::Flow { path, .. } => {
+            // In local mode, we don't have access to saved flows
+            Err(LocalError::Execution(format!(
+                "Flow references (path: {}) are not supported in local mode. Use inline modules instead.",
+                path
+            )))
+        }
+
+        FlowModuleValue::ForloopFlow { iterator, modules, skip_failures, parallel, .. } => {
+            execute_forloop(db, &iterator, &modules, skip_failures, parallel, ctx).await
+        }
+
+        FlowModuleValue::WhileloopFlow { modules, skip_failures, .. } => {
+            execute_whileloop(db, &modules, skip_failures, ctx).await
+        }
+
+        FlowModuleValue::BranchOne { branches, default, .. } => {
+            execute_branch_one(db, &branches, &default, ctx).await
+        }
+
+        FlowModuleValue::BranchAll { branches, parallel } => {
+            execute_branch_all(db, &branches, parallel, ctx).await
+        }
+
+        FlowModuleValue::FlowScript { .. } => {
+            Err(LocalError::Execution(
+                "FlowScript (internal reference) is not supported in local mode".to_string()
+            ))
+        }
+
+        FlowModuleValue::AIAgent { .. } => {
+            Err(LocalError::Execution(
+                "AIAgent is not supported in local mode".to_string()
+            ))
+        }
+    }
+}
+
+/// Execute a for loop
+#[async_recursion::async_recursion]
+async fn execute_forloop(
+    db: &LocalDb,
+    iterator: &InputTransform,
+    modules: &[FlowModule],
+    skip_failures: bool,
+    _parallel: bool,  // TODO: implement parallel execution
+    ctx: &mut FlowContext,
+) -> Result<serde_json::Value> {
+    // Evaluate the iterator expression
+    let iter_value = evaluate_input_transform(iterator, ctx)?;
+
+    let items = match iter_value.as_array() {
+        Some(arr) => arr.clone(),
+        None => {
+            return Err(LocalError::Execution(
+                "For loop iterator must evaluate to an array".to_string()
+            ));
+        }
+    };
+
+    let mut results = Vec::new();
+
+    for (idx, item) in items.iter().enumerate() {
+        tracing::debug!("For loop iteration {} of {}", idx + 1, items.len());
+
+        // Create iteration context
+        let mut iter_ctx = FlowContext {
+            flow_input: ctx.flow_input.clone(),
+            previous_result: item.clone(),
+            results_by_id: ctx.results_by_id.clone(),
+        };
+
+        // Add iter context
+        iter_ctx.results_by_id.insert("iter".to_string(), serde_json::json!({
+            "index": idx,
+            "value": item,
+        }));
+
+        // Execute modules in sequence
+        let mut iter_result = item.clone();
+        let mut had_error = false;
+
+        for module in modules {
+            match execute_module(db, module, &mut iter_ctx).await {
+                Ok(result) => {
+                    iter_ctx.results_by_id.insert(module.id.clone(), result.clone());
+                    iter_ctx.previous_result = result.clone();
+                    iter_result = result;
+                }
+                Err(e) => {
+                    had_error = true;
+                    if skip_failures {
+                        tracing::warn!("For loop iteration {} failed (skipping): {}", idx, e);
+                        iter_result = serde_json::json!({"error": e.to_string()});
+                    } else {
+                        return Err(e);
+                    }
+                    break;
+                }
+            }
+        }
+
+        if !had_error || skip_failures {
+            results.push(iter_result);
+        }
+    }
+
+    Ok(serde_json::Value::Array(results))
+}
+
+/// Execute a while loop
+#[async_recursion::async_recursion]
+async fn execute_whileloop(
+    db: &LocalDb,
+    modules: &[FlowModule],
+    skip_failures: bool,
+    ctx: &mut FlowContext,
+) -> Result<serde_json::Value> {
+    const MAX_ITERATIONS: usize = 1000;
+    let mut results = Vec::new();
+    let mut iteration = 0;
+
+    loop {
+        if iteration >= MAX_ITERATIONS {
+            return Err(LocalError::Execution(format!(
+                "While loop exceeded maximum iterations ({})", MAX_ITERATIONS
+            )));
+        }
+
+        // Execute modules
+        let mut iter_result = ctx.previous_result.clone();
+        let mut should_continue = true;
+
+        for module in modules {
+            match execute_module(db, module, ctx).await {
+                Ok(result) => {
+                    ctx.results_by_id.insert(module.id.clone(), result.clone());
+                    ctx.previous_result = result.clone();
+                    iter_result = result;
+                }
+                Err(e) => {
+                    if skip_failures {
+                        tracing::warn!("While loop iteration {} failed (skipping): {}", iteration, e);
+                        iter_result = serde_json::json!({"error": e.to_string()});
+                    } else {
+                        return Err(e);
+                    }
+                    should_continue = false;
+                    break;
+                }
+            }
+        }
+
+        results.push(iter_result);
+        iteration += 1;
+
+        // Check stop condition (result should be truthy to continue)
+        if !should_continue {
+            break;
+        }
+
+        // Check if the result indicates we should stop
+        let continue_loop = match &ctx.previous_result {
+            serde_json::Value::Bool(b) => *b,
+            serde_json::Value::Null => false,
+            serde_json::Value::Object(obj) => {
+                // Check for a "continue" field
+                obj.get("continue")
+                    .and_then(|v| v.as_bool())
+                    .unwrap_or(false)
+            }
+            _ => false,
+        };
+
+        if !continue_loop {
+            break;
+        }
+    }
+
+    Ok(serde_json::Value::Array(results))
+}
+
+/// Execute branch-one (if/else)
+#[async_recursion::async_recursion]
+async fn execute_branch_one(
+    db: &LocalDb,
+    branches: &[Branch],
+    default: &[FlowModule],
+    ctx: &mut FlowContext,
+) -> Result<serde_json::Value> {
+    // Find the first matching branch
+    for (idx, branch) in branches.iter().enumerate() {
+        let condition = evaluate_expr(&branch.expr, &ctx.to_eval_context())?;
+        if condition.as_bool().unwrap_or(false) {
+            tracing::debug!("Branch {} matched", idx);
+            return execute_branch_modules(db, &branch.modules, ctx).await;
+        }
+    }
+
+    // No branch matched, execute default
+    tracing::debug!("No branch matched, executing default");
+    execute_branch_modules(db, default, ctx).await
+}
+
+/// Execute branch-all (parallel branches)
+#[async_recursion::async_recursion]
+async fn execute_branch_all(
+    db: &LocalDb,
+    branches: &[Branch],
+    _parallel: bool,  // TODO: implement true parallel execution
+    ctx: &mut FlowContext,
+) -> Result<serde_json::Value> {
+    let mut results = Vec::new();
+
+    // Execute all branches (sequentially for now)
+    for (idx, branch) in branches.iter().enumerate() {
+        tracing::debug!("Executing branch {}", idx);
+
+        let mut branch_ctx = ctx.clone();
+        match execute_branch_modules(db, &branch.modules, &mut branch_ctx).await {
+            Ok(result) => {
+                results.push(result);
+            }
+            Err(e) => {
+                if branch.skip_failure {
+                    tracing::warn!("Branch {} failed (skipping): {}", idx, e);
+                    results.push(serde_json::json!({"error": e.to_string()}));
+                } else {
+                    return Err(e);
+                }
+            }
+        }
+    }
+
+    Ok(serde_json::Value::Array(results))
+}
+
+/// Execute a sequence of modules in a branch
+#[async_recursion::async_recursion]
+async fn execute_branch_modules(
+    db: &LocalDb,
+    modules: &[FlowModule],
+    ctx: &mut FlowContext,
+) -> Result<serde_json::Value> {
+    let mut result = ctx.previous_result.clone();
+
+    for module in modules {
+        result = execute_module(db, module, ctx).await?;
+        ctx.results_by_id.insert(module.id.clone(), result.clone());
+        ctx.previous_result = result.clone();
+    }
+
+    Ok(result)
+}
+
+/// Resolve input transforms to concrete arguments
+fn resolve_input_transforms(
+    transforms: &HashMap<String, InputTransform>,
+    ctx: &FlowContext,
+) -> Result<serde_json::Value> {
+    let mut args = serde_json::Map::new();
+
+    for (key, transform) in transforms {
+        let value = evaluate_input_transform(transform, ctx)?;
+        args.insert(key.clone(), value);
+    }
+
+    Ok(serde_json::Value::Object(args))
+}
+
+/// Evaluate an input transform
+fn evaluate_input_transform(
+    transform: &InputTransform,
+    ctx: &FlowContext,
+) -> Result<serde_json::Value> {
+    match transform {
+        InputTransform::Static { value } => {
+            serde_json::from_str(value.get())
+                .map_err(|e| LocalError::Execution(format!("Invalid static value: {}", e)))
+        }
+        InputTransform::Javascript { expr } => {
+            evaluate_expr(expr, &ctx.to_eval_context())
+        }
+        InputTransform::Ai => {
+            Err(LocalError::Execution("AI input transforms are not supported in local mode".to_string()))
+        }
+    }
+}
+
+/// Evaluate a JavaScript expression
+fn evaluate_expr(expr: &str, context: &serde_json::Value) -> Result<serde_json::Value> {
+    let expr = expr.trim();
+
+    // Handle comparison operators
+    if let Some(result) = try_evaluate_comparison(expr, context) {
+        return Ok(result);
+    }
+
+    // Handle simple variable references
+    if let Some(val) = resolve_path(expr, context) {
+        return Ok(val);
+    }
+
+    // Handle boolean literals
+    if expr == "true" {
+        return Ok(serde_json::Value::Bool(true));
+    }
+    if expr == "false" {
+        return Ok(serde_json::Value::Bool(false));
+    }
+
+    // Handle numeric literals
+    if let Ok(n) = expr.parse::<i64>() {
+        return Ok(serde_json::json!(n));
+    }
+    if let Ok(n) = expr.parse::<f64>() {
+        return Ok(serde_json::json!(n));
+    }
+
+    // Handle string literals
+    if (expr.starts_with('"') && expr.ends_with('"')) ||
+       (expr.starts_with('\'') && expr.ends_with('\'')) {
+        return Ok(serde_json::json!(&expr[1..expr.len()-1]));
+    }
+
+    // For complex expressions, we'd need a full JS runtime
+    tracing::warn!("Complex expression not evaluated: {}", expr);
+    Ok(serde_json::json!(expr))
+}
+
+/// Try to resolve a path like "flow_input.x" or "results.a.b"
+fn resolve_path(path: &str, context: &serde_json::Value) -> Option<serde_json::Value> {
+    let parts: Vec<&str> = path.split('.').collect();
+    if parts.is_empty() {
+        return None;
+    }
+
+    let mut current = context.get(parts[0])?;
+
+    for part in &parts[1..] {
+        current = current.get(*part)?;
+    }
+
+    Some(current.clone())
+}
+
+/// Try to evaluate a comparison expression
+fn try_evaluate_comparison(expr: &str, context: &serde_json::Value) -> Option<serde_json::Value> {
+    // Supported operators: >, <, >=, <=, ==, !=, ===, !==
+    let operators = ["===", "!==", ">=", "<=", "==", "!=", ">", "<"];
+
+    for op in operators {
+        if let Some(pos) = expr.find(op) {
+            let left = expr[..pos].trim();
+            let right = expr[pos + op.len()..].trim();
+
+            let left_val = evaluate_expr(left, context).ok()?;
+            let right_val = evaluate_expr(right, context).ok()?;
+
+            let result = match op {
+                ">" => compare_values(&left_val, &right_val, |a, b| a > b),
+                "<" => compare_values(&left_val, &right_val, |a, b| a < b),
+                ">=" => compare_values(&left_val, &right_val, |a, b| a >= b),
+                "<=" => compare_values(&left_val, &right_val, |a, b| a <= b),
+                "==" | "===" => Some(left_val == right_val),
+                "!=" | "!==" => Some(left_val != right_val),
+                _ => None,
+            };
+
+            return result.map(serde_json::Value::Bool);
+        }
+    }
+
+    // Try logical operators
+    if let Some(pos) = expr.find("&&") {
+        let left = expr[..pos].trim();
+        let right = expr[pos + 2..].trim();
+        let left_val = evaluate_expr(left, context).ok()?;
+        let right_val = evaluate_expr(right, context).ok()?;
+        return Some(serde_json::Value::Bool(
+            is_truthy(&left_val) && is_truthy(&right_val)
+        ));
+    }
+
+    if let Some(pos) = expr.find("||") {
+        let left = expr[..pos].trim();
+        let right = expr[pos + 2..].trim();
+        let left_val = evaluate_expr(left, context).ok()?;
+        let right_val = evaluate_expr(right, context).ok()?;
+        return Some(serde_json::Value::Bool(
+            is_truthy(&left_val) || is_truthy(&right_val)
+        ));
+    }
+
+    None
+}
+
+/// Compare two JSON values numerically
+fn compare_values<F>(left: &serde_json::Value, right: &serde_json::Value, cmp: F) -> Option<bool>
+where
+    F: Fn(f64, f64) -> bool,
+{
+    let left_num = value_to_number(left)?;
+    let right_num = value_to_number(right)?;
+    Some(cmp(left_num, right_num))
+}
+
+/// Convert a JSON value to a number
+fn value_to_number(val: &serde_json::Value) -> Option<f64> {
+    match val {
+        serde_json::Value::Number(n) => n.as_f64(),
+        serde_json::Value::String(s) => s.parse().ok(),
+        serde_json::Value::Bool(b) => Some(if *b { 1.0 } else { 0.0 }),
+        _ => None,
+    }
+}
+
+/// Check if a value is truthy (JavaScript semantics)
+fn is_truthy(val: &serde_json::Value) -> bool {
+    match val {
+        serde_json::Value::Null => false,
+        serde_json::Value::Bool(b) => *b,
+        serde_json::Value::Number(n) => n.as_f64().map(|f| f != 0.0).unwrap_or(false),
+        serde_json::Value::String(s) => !s.is_empty(),
+        serde_json::Value::Array(a) => !a.is_empty(),
+        serde_json::Value::Object(_) => true,
+    }
+}
+
+/// Convert windmill-common ScriptLang to local ScriptLang
+fn convert_script_lang(lang: &WmScriptLang) -> ScriptLang {
+    match lang {
+        WmScriptLang::Deno => ScriptLang::Deno,
+        WmScriptLang::Python3 => ScriptLang::Python3,
+        WmScriptLang::Bash => ScriptLang::Bash,
+        WmScriptLang::Go => ScriptLang::Go,
+        WmScriptLang::Bun => ScriptLang::Bun,
+        WmScriptLang::Nativets => ScriptLang::Deno,  // Fallback to Deno
+        _ => ScriptLang::Bash,  // Default fallback
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_evaluate_simple_expr() {
+        let ctx = serde_json::json!({
+            "flow_input": {"x": 10},
+            "previous_result": 42,
+            "results": {"a": "hello"},
+        });
+
+        assert_eq!(
+            evaluate_expr("flow_input", &ctx).unwrap(),
+            serde_json::json!({"x": 10})
+        );
+        assert_eq!(
+            evaluate_expr("previous_result", &ctx).unwrap(),
+            serde_json::json!(42)
+        );
+        assert_eq!(
+            evaluate_expr("results.a", &ctx).unwrap(),
+            serde_json::json!("hello")
+        );
+        assert_eq!(
+            evaluate_expr("flow_input.x", &ctx).unwrap(),
+            serde_json::json!(10)
+        );
+    }
+
+    #[test]
+    fn test_evaluate_comparison_expr() {
+        let ctx = serde_json::json!({
+            "flow_input": {"x": 10, "y": 5},
+            "previous_result": 42,
+        });
+
+        assert_eq!(
+            evaluate_expr("flow_input.x > 5", &ctx).unwrap(),
+            serde_json::Value::Bool(true)
+        );
+        assert_eq!(
+            evaluate_expr("flow_input.x < 5", &ctx).unwrap(),
+            serde_json::Value::Bool(false)
+        );
+        assert_eq!(
+            evaluate_expr("flow_input.x >= 10", &ctx).unwrap(),
+            serde_json::Value::Bool(true)
+        );
+        assert_eq!(
+            evaluate_expr("flow_input.y == 5", &ctx).unwrap(),
+            serde_json::Value::Bool(true)
+        );
+        assert_eq!(
+            evaluate_expr("previous_result > 40", &ctx).unwrap(),
+            serde_json::Value::Bool(true)
+        );
+    }
+
+    #[test]
+    fn test_evaluate_logical_expr() {
+        let ctx = serde_json::json!({
+            "flow_input": {"a": true, "b": false},
+        });
+
+        // Simple boolean logic (complex expressions with mixed operators need parentheses support)
+        assert_eq!(
+            evaluate_expr("flow_input.a && flow_input.a", &ctx).unwrap(),
+            serde_json::Value::Bool(true)
+        );
+        assert_eq!(
+            evaluate_expr("flow_input.a && flow_input.b", &ctx).unwrap(),
+            serde_json::Value::Bool(false)
+        );
+        assert_eq!(
+            evaluate_expr("flow_input.b || flow_input.a", &ctx).unwrap(),
+            serde_json::Value::Bool(true)
+        );
+    }
+}
--- a/backend/windmill-local/src/jobs.rs
+++ b/backend/windmill-local/src/jobs.rs
@@ -0,0 +1,495 @@
+//! Job types and operations for local mode
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::db::LocalDb;
+use crate::error::{LocalError, Result};
+
+/// Job kind (mirrors windmill-common JobKind)
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum JobKind {
+    Script,
+    Preview,
+    Flow,
+    FlowPreview,
+    Dependencies,
+    FlowDependencies,
+    ScriptHub,
+    Identity,
+    Http,
+    Graphql,
+    Postgresql,
+    Noop,
+    AppDependencies,
+    DeploymentCallback,
+    SingleScriptFlow,
+    FlowScript,
+    FlowNode,
+    AppScript,
+}
+
+impl JobKind {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            JobKind::Script => "script",
+            JobKind::Preview => "preview",
+            JobKind::Flow => "flow",
+            JobKind::FlowPreview => "flowpreview",
+            JobKind::Dependencies => "dependencies",
+            JobKind::FlowDependencies => "flowdependencies",
+            JobKind::ScriptHub => "script_hub",
+            JobKind::Identity => "identity",
+            JobKind::Http => "http",
+            JobKind::Graphql => "graphql",
+            JobKind::Postgresql => "postgresql",
+            JobKind::Noop => "noop",
+            JobKind::AppDependencies => "appdependencies",
+            JobKind::DeploymentCallback => "deploymentcallback",
+            JobKind::SingleScriptFlow => "singlescriptflow",
+            JobKind::FlowScript => "flowscript",
+            JobKind::FlowNode => "flownode",
+            JobKind::AppScript => "appscript",
+        }
+    }
+}
+
+/// Script language
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum ScriptLang {
+    Python3,
+    Deno,
+    Go,
+    Bash,
+    Postgresql,
+    Nativets,
+    Bun,
+    Mysql,
+    Bigquery,
+    Snowflake,
+    Graphql,
+    Powershell,
+    Mssql,
+    Php,
+    Bunnative,
+    Rust,
+    Ansible,
+    Csharp,
+    Oracledb,
+    Nu,
+    Java,
+    Duckdb,
+}
+
+impl ScriptLang {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            ScriptLang::Python3 => "python3",
+            ScriptLang::Deno => "deno",
+            ScriptLang::Go => "go",
+            ScriptLang::Bash => "bash",
+            ScriptLang::Postgresql => "postgresql",
+            ScriptLang::Nativets => "nativets",
+            ScriptLang::Bun => "bun",
+            ScriptLang::Mysql => "mysql",
+            ScriptLang::Bigquery => "bigquery",
+            ScriptLang::Snowflake => "snowflake",
+            ScriptLang::Graphql => "graphql",
+            ScriptLang::Powershell => "powershell",
+            ScriptLang::Mssql => "mssql",
+            ScriptLang::Php => "php",
+            ScriptLang::Bunnative => "bunnative",
+            ScriptLang::Rust => "rust",
+            ScriptLang::Ansible => "ansible",
+            ScriptLang::Csharp => "csharp",
+            ScriptLang::Oracledb => "oracledb",
+            ScriptLang::Nu => "nu",
+            ScriptLang::Java => "java",
+            ScriptLang::Duckdb => "duckdb",
+        }
+    }
+
+    pub fn from_str(s: &str) -> Option<Self> {
+        match s {
+            "python3" => Some(ScriptLang::Python3),
+            "deno" => Some(ScriptLang::Deno),
+            "go" => Some(ScriptLang::Go),
+            "bash" => Some(ScriptLang::Bash),
+            "postgresql" => Some(ScriptLang::Postgresql),
+            "nativets" => Some(ScriptLang::Nativets),
+            "bun" => Some(ScriptLang::Bun),
+            "mysql" => Some(ScriptLang::Mysql),
+            "bigquery" => Some(ScriptLang::Bigquery),
+            "snowflake" => Some(ScriptLang::Snowflake),
+            "graphql" => Some(ScriptLang::Graphql),
+            "powershell" => Some(ScriptLang::Powershell),
+            "mssql" => Some(ScriptLang::Mssql),
+            "php" => Some(ScriptLang::Php),
+            "bunnative" => Some(ScriptLang::Bunnative),
+            "rust" => Some(ScriptLang::Rust),
+            "ansible" => Some(ScriptLang::Ansible),
+            "csharp" => Some(ScriptLang::Csharp),
+            "oracledb" => Some(ScriptLang::Oracledb),
+            "nu" => Some(ScriptLang::Nu),
+            "java" => Some(ScriptLang::Java),
+            "duckdb" => Some(ScriptLang::Duckdb),
+            _ => None,
+        }
+    }
+}
+
+/// Job status for completed jobs
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum JobStatus {
+    Success,
+    Failure,
+    Canceled,
+    Skipped,
+}
+
+impl JobStatus {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            JobStatus::Success => "success",
+            JobStatus::Failure => "failure",
+            JobStatus::Canceled => "canceled",
+            JobStatus::Skipped => "skipped",
+        }
+    }
+
+    pub fn from_str(s: &str) -> Option<Self> {
+        match s {
+            "success" => Some(JobStatus::Success),
+            "failure" => Some(JobStatus::Failure),
+            "canceled" => Some(JobStatus::Canceled),
+            "skipped" => Some(JobStatus::Skipped),
+            _ => None,
+        }
+    }
+}
+
+/// Preview job request (simplified from windmill-api Preview struct)
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PreviewRequest {
+    pub content: String,
+    pub language: ScriptLang,
+    #[serde(default)]
+    pub args: serde_json::Value,
+    pub lock: Option<String>,
+    pub tag: Option<String>,
+}
+
+/// Flow preview request
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FlowPreviewRequest {
+    pub value: serde_json::Value,  // FlowValue as JSON
+    #[serde(default)]
+    pub args: serde_json::Value,
+    pub tag: Option<String>,
+}
+
+/// A queued job (combines v2_job and v2_job_queue)
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct QueuedJob {
+    pub id: Uuid,
+    pub workspace_id: String,
+    pub kind: JobKind,
+    pub script_lang: Option<ScriptLang>,
+    pub raw_code: Option<String>,
+    pub raw_lock: Option<String>,
+    pub raw_flow: Option<serde_json::Value>,
+    pub args: serde_json::Value,
+    pub tag: String,
+    pub created_at: DateTime<Utc>,
+    pub scheduled_for: DateTime<Utc>,
+    pub running: bool,
+    pub parent_job: Option<Uuid>,
+    pub root_job: Option<Uuid>,
+    pub flow_step_id: Option<String>,
+}
+
+/// A completed job
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CompletedJob {
+    pub id: Uuid,
+    pub workspace_id: String,
+    pub status: JobStatus,
+    pub result: serde_json::Value,
+    pub started_at: Option<DateTime<Utc>>,
+    pub completed_at: DateTime<Utc>,
+    pub duration_ms: Option<i64>,
+}
+
+/// Push a preview job to the queue
+pub async fn push_preview(db: &LocalDb, req: PreviewRequest) -> Result<Uuid> {
+    let id = Uuid::new_v4();
+    let now = Utc::now().to_rfc3339();
+    let args_json = serde_json::to_string(&req.args)?;
+    let tag = req.tag.as_deref().unwrap_or("deno");
+
+    // Insert into v2_job
+    db.execute(
+        r#"
+        INSERT INTO v2_job (id, kind, script_lang, raw_code, raw_lock, args, tag, created_at)
+        VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
+        "#,
+        libsql::params![
+            id.to_string(),
+            JobKind::Preview.as_str(),
+            req.language.as_str(),
+            req.content,
+            req.lock,
+            args_json,
+            tag,
+            now.clone(),
+        ],
+    )
+    .await?;
+
+    // Insert into v2_job_queue
+    db.execute(
+        r#"
+        INSERT INTO v2_job_queue (id, tag, created_at, scheduled_for, running)
+        VALUES (?1, ?2, ?3, ?4, 0)
+        "#,
+        libsql::params![id.to_string(), tag, now.clone(), now],
+    )
+    .await?;
+
+    // Insert into v2_job_runtime (for heartbeat tracking)
+    db.execute(
+        "INSERT INTO v2_job_runtime (id) VALUES (?1)",
+        libsql::params![id.to_string()],
+    )
+    .await?;
+
+    // Insert into job_perms (simplified)
+    db.execute(
+        "INSERT INTO job_perms (job_id) VALUES (?1)",
+        libsql::params![id.to_string()],
+    )
+    .await?;
+
+    tracing::info!("Pushed preview job: {}", id);
+    Ok(id)
+}
+
+/// Push a flow preview job to the queue
+pub async fn push_flow_preview(db: &LocalDb, req: FlowPreviewRequest) -> Result<Uuid> {
+    let id = Uuid::new_v4();
+    let now = Utc::now().to_rfc3339();
+    let args_json = serde_json::to_string(&req.args)?;
+    let flow_json = serde_json::to_string(&req.value)?;
+    let tag = req.tag.as_deref().unwrap_or("flow");
+
+    // Insert into v2_job
+    db.execute(
+        r#"
+        INSERT INTO v2_job (id, kind, raw_flow, args, tag, created_at)
+        VALUES (?1, ?2, ?3, ?4, ?5, ?6)
+        "#,
+        libsql::params![
+            id.to_string(),
+            JobKind::FlowPreview.as_str(),
+            flow_json,
+            args_json,
+            tag,
+            now.clone(),
+        ],
+    )
+    .await?;
+
+    // Insert into v2_job_queue
+    db.execute(
+        r#"
+        INSERT INTO v2_job_queue (id, tag, created_at, scheduled_for, running)
+        VALUES (?1, ?2, ?3, ?4, 0)
+        "#,
+        libsql::params![id.to_string(), tag, now.clone(), now],
+    )
+    .await?;
+
+    // Insert into v2_job_runtime
+    db.execute(
+        "INSERT INTO v2_job_runtime (id) VALUES (?1)",
+        libsql::params![id.to_string()],
+    )
+    .await?;
+
+    // Insert into job_perms
+    db.execute(
+        "INSERT INTO job_perms (job_id) VALUES (?1)",
+        libsql::params![id.to_string()],
+    )
+    .await?;
+
+    // Insert initial flow status
+    db.execute(
+        "INSERT INTO v2_job_status (id, flow_status) VALUES (?1, '{}')",
+        libsql::params![id.to_string()],
+    )
+    .await?;
+
+    tracing::info!("Pushed flow preview job: {}", id);
+    Ok(id)
+}
+
+/// Get a completed job result (for polling)
+pub async fn get_completed_job(db: &LocalDb, id: Uuid) -> Result<Option<CompletedJob>> {
+    let mut rows = db
+        .query(
+            r#"
+            SELECT id, workspace_id, status, result, started_at, completed_at, duration_ms
+            FROM v2_job_completed
+            WHERE id = ?1
+            "#,
+            libsql::params![id.to_string()],
+        )
+        .await?;
+
+    if let Some(row) = rows.next().await? {
+        let status_str: String = row.get(2)?;
+        let status = JobStatus::from_str(&status_str)
+            .ok_or_else(|| LocalError::InvalidJobState(status_str))?;
+
+        let result_str: Option<String> = row.get(3)?;
+        let result: serde_json::Value = result_str
+            .map(|s| serde_json::from_str(&s))
+            .transpose()?
+            .unwrap_or(serde_json::Value::Null);
+
+        let started_at_str: Option<String> = row.get(4)?;
+        let started_at = started_at_str
+            .map(|s| DateTime::parse_from_rfc3339(&s).map(|dt| dt.with_timezone(&Utc)))
+            .transpose()
+            .ok()
+            .flatten();
+
+        let completed_at_str: String = row.get(5)?;
+        let completed_at = DateTime::parse_from_rfc3339(&completed_at_str)
+            .map(|dt| dt.with_timezone(&Utc))
+            .unwrap_or_else(|_| Utc::now());
+
+        let duration_ms: Option<i64> = row.get(6)?;
+
+        Ok(Some(CompletedJob {
+            id,
+            workspace_id: row.get(1)?,
+            status,
+            result,
+            started_at,
+            completed_at,
+            duration_ms,
+        }))
+    } else {
+        Ok(None)
+    }
+}
+
+/// Mark a job as completed with result
+pub async fn complete_job(
+    db: &LocalDb,
+    id: Uuid,
+    status: JobStatus,
+    result: serde_json::Value,
+    started_at: DateTime<Utc>,
+) -> Result<()> {
+    let now = Utc::now();
+    let duration_ms = (now - started_at).num_milliseconds();
+    let result_json = serde_json::to_string(&result)?;
+
+    db.execute(
+        r#"
+        INSERT INTO v2_job_completed (id, workspace_id, status, result, started_at, completed_at, duration_ms)
+        SELECT ?1, workspace_id, ?2, ?3, ?4, ?5, ?6
+        FROM v2_job WHERE id = ?1
+        "#,
+        libsql::params![
+            id.to_string(),
+            status.as_str(),
+            result_json,
+            started_at.to_rfc3339(),
+            now.to_rfc3339(),
+            duration_ms,
+        ],
+    )
+    .await?;
+
+    // Remove from queue
+    db.execute(
+        "DELETE FROM v2_job_queue WHERE id = ?1",
+        libsql::params![id.to_string()],
+    )
+    .await?;
+
+    tracing::info!("Completed job {}: {:?}", id, status);
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_push_preview() {
+        let db = LocalDb::in_memory().await.unwrap();
+
+        let req = PreviewRequest {
+            content: "export function main() { return 42; }".to_string(),
+            language: ScriptLang::Deno,
+            args: serde_json::json!({}),
+            lock: None,
+            tag: None,
+        };
+
+        let id = push_preview(&db, req).await.unwrap();
+
+        // Verify job exists in queue
+        let mut rows = db
+            .query(
+                "SELECT running FROM v2_job_queue WHERE id = ?1",
+                libsql::params![id.to_string()],
+            )
+            .await
+            .unwrap();
+
+        let row = rows.next().await.unwrap().unwrap();
+        let running: i64 = row.get(0).unwrap();
+        assert_eq!(running, 0);
+    }
+
+    #[tokio::test]
+    async fn test_complete_job() {
+        let db = LocalDb::in_memory().await.unwrap();
+
+        let req = PreviewRequest {
+            content: "export function main() { return 42; }".to_string(),
+            language: ScriptLang::Deno,
+            args: serde_json::json!({}),
+            lock: None,
+            tag: None,
+        };
+
+        let id = push_preview(&db, req).await.unwrap();
+        let started_at = Utc::now();
+
+        complete_job(
+            &db,
+            id,
+            JobStatus::Success,
+            serde_json::json!(42),
+            started_at,
+        )
+        .await
+        .unwrap();
+
+        // Verify job is in completed
+        let completed = get_completed_job(&db, id).await.unwrap().unwrap();
+        assert_eq!(completed.status, JobStatus::Success);
+        assert_eq!(completed.result, serde_json::json!(42));
+    }
+}
--- a/backend/windmill-local/src/lib.rs
+++ b/backend/windmill-local/src/lib.rs
@@ -0,0 +1,31 @@
+//! Windmill Local Mode
+//!
+//! This crate provides a minimal local mode for Windmill using libSQL (SQLite/Turso)
+//! instead of PostgreSQL. The goal is to support preview execution end-to-end
+//! with a lightweight, embedded database.
+//!
+//! ## Scope
+//! - Script preview execution
+//! - Flow preview execution
+//! - In-memory or file-based SQLite storage
+//! - Remote Turso database support for multi-writer scenarios
+//!
+//! ## Non-goals (for this experiment)
+//! - Full feature parity with PostgreSQL mode
+//! - Multi-worker support (single embedded worker)
+//! - Persistence of scripts/flows (only jobs)
+
+pub mod db;
+pub mod schema;
+pub mod jobs;
+pub mod queue;
+pub mod executor;
+pub mod flow_executor;
+pub mod worker;
+pub mod server;
+pub mod error;
+
+pub use db::LocalDb;
+pub use error::LocalError;
+pub use worker::Worker;
+pub use server::LocalServer;
--- a/backend/windmill-local/src/queue.rs
+++ b/backend/windmill-local/src/queue.rs
@@ -0,0 +1,296 @@
+//! Queue operations for local mode
+//!
+//! Since local mode uses a single worker, we don't need the complex
+//! `FOR UPDATE SKIP LOCKED` mechanism. Instead, we use simple atomic
+//! operations with the database lock.
+
+use chrono::{DateTime, Utc};
+use uuid::Uuid;
+
+use crate::db::LocalDb;
+use crate::error::{LocalError, Result};
+use crate::jobs::{JobKind, QueuedJob, ScriptLang};
+
+/// Pull the next job from the queue
+///
+/// This is simplified from the PostgreSQL version since we have a single worker
+/// and use the connection mutex for coordination.
+pub async fn pull_job(db: &LocalDb) -> Result<Option<QueuedJob>> {
+    let now = Utc::now().to_rfc3339();
+
+    // Get the next job (ordered by priority, then scheduled_for)
+    // We use a transaction-like approach: SELECT then UPDATE
+    let mut rows = db
+        .query(
+            r#"
+            SELECT q.id, j.workspace_id, j.kind, j.script_lang, j.raw_code, j.raw_lock,
+                   j.raw_flow, j.args, q.tag, j.created_at, q.scheduled_for,
+                   j.parent_job, j.root_job, j.flow_step_id
+            FROM v2_job_queue q
+            JOIN v2_job j ON q.id = j.id
+            WHERE q.running = 0 AND q.scheduled_for <= ?1
+            ORDER BY q.priority DESC, q.scheduled_for ASC
+            LIMIT 1
+            "#,
+            libsql::params![now],
+        )
+        .await?;
+
+    let Some(row) = rows.next().await? else {
+        return Ok(None);
+    };
+
+    let id_str: String = row.get(0)?;
+    let id = Uuid::parse_str(&id_str).map_err(|e| LocalError::InvalidJobState(e.to_string()))?;
+
+    // Mark as running
+    let started_at = Utc::now().to_rfc3339();
+    db.execute(
+        "UPDATE v2_job_queue SET running = 1, started_at = ?2 WHERE id = ?1",
+        libsql::params![id_str.clone(), started_at],
+    )
+    .await?;
+
+    // Parse the job fields
+    let kind_str: String = row.get(2)?;
+    let kind = match kind_str.as_str() {
+        "preview" => JobKind::Preview,
+        "flowpreview" => JobKind::FlowPreview,
+        "script" => JobKind::Script,
+        "flow" => JobKind::Flow,
+        "flowscript" => JobKind::FlowScript,
+        "flownode" => JobKind::FlowNode,
+        _ => JobKind::Preview, // Default
+    };
+
+    let lang_str: Option<String> = row.get(3)?;
+    let script_lang = lang_str.and_then(|s| ScriptLang::from_str(&s));
+
+    let raw_code: Option<String> = row.get(4)?;
+    let raw_lock: Option<String> = row.get(5)?;
+
+    let raw_flow_str: Option<String> = row.get(6)?;
+    let raw_flow = raw_flow_str
+        .map(|s| serde_json::from_str(&s))
+        .transpose()?;
+
+    let args_str: Option<String> = row.get(7)?;
+    let args: serde_json::Value = args_str
+        .map(|s| serde_json::from_str(&s))
+        .transpose()?
+        .unwrap_or(serde_json::Value::Object(serde_json::Map::new()));
+
+    let tag: String = row.get(8)?;
+
+    let created_at_str: String = row.get(9)?;
+    let created_at = DateTime::parse_from_rfc3339(&created_at_str)
+        .map(|dt| dt.with_timezone(&Utc))
+        .unwrap_or_else(|_| Utc::now());
+
+    let scheduled_for_str: String = row.get(10)?;
+    let scheduled_for = DateTime::parse_from_rfc3339(&scheduled_for_str)
+        .map(|dt| dt.with_timezone(&Utc))
+        .unwrap_or_else(|_| Utc::now());
+
+    let parent_job_str: Option<String> = row.get(11)?;
+    let parent_job = parent_job_str.and_then(|s| Uuid::parse_str(&s).ok());
+
+    let root_job_str: Option<String> = row.get(12)?;
+    let root_job = root_job_str.and_then(|s| Uuid::parse_str(&s).ok());
+
+    let flow_step_id: Option<String> = row.get(13)?;
+
+    Ok(Some(QueuedJob {
+        id,
+        workspace_id: row.get(1)?,
+        kind,
+        script_lang,
+        raw_code,
+        raw_lock,
+        raw_flow,
+        args,
+        tag,
+        created_at,
+        scheduled_for,
+        running: true,
+        parent_job,
+        root_job,
+        flow_step_id,
+    }))
+}
+
+/// Get queue statistics
+pub async fn queue_stats(db: &LocalDb) -> Result<QueueStats> {
+    let mut rows = db
+        .query(
+            r#"
+            SELECT
+                COUNT(*) as total,
+                SUM(CASE WHEN running = 1 THEN 1 ELSE 0 END) as running,
+                SUM(CASE WHEN running = 0 THEN 1 ELSE 0 END) as pending
+            FROM v2_job_queue
+            "#,
+            (),
+        )
+        .await?;
+
+    let row = rows.next().await?.ok_or(LocalError::QueueEmpty)?;
+
+    Ok(QueueStats {
+        total: row.get::<i64>(0)? as u64,
+        running: row.get::<i64>(1).unwrap_or(0) as u64,
+        pending: row.get::<i64>(2).unwrap_or(0) as u64,
+    })
+}
+
+#[derive(Debug, Clone)]
+pub struct QueueStats {
+    pub total: u64,
+    pub running: u64,
+    pub pending: u64,
+}
+
+/// Update job heartbeat (ping)
+pub async fn ping_job(db: &LocalDb, id: Uuid) -> Result<()> {
+    let now = Utc::now().to_rfc3339();
+    db.execute(
+        "UPDATE v2_job_runtime SET ping = ?2 WHERE id = ?1",
+        libsql::params![id.to_string(), now],
+    )
+    .await?;
+    Ok(())
+}
+
+/// Update flow status for a running flow job
+pub async fn update_flow_status(
+    db: &LocalDb,
+    id: Uuid,
+    flow_status: &serde_json::Value,
+) -> Result<()> {
+    let status_json = serde_json::to_string(flow_status)?;
+    db.execute(
+        "UPDATE v2_job_status SET flow_status = ?2 WHERE id = ?1",
+        libsql::params![id.to_string(), status_json],
+    )
+    .await?;
+    Ok(())
+}
+
+/// Push a child job for flow execution
+pub async fn push_flow_child_job(
+    db: &LocalDb,
+    parent_id: Uuid,
+    root_id: Uuid,
+    step_id: &str,
+    kind: JobKind,
+    script_lang: Option<ScriptLang>,
+    raw_code: Option<&str>,
+    args: &serde_json::Value,
+) -> Result<Uuid> {
+    let id = Uuid::new_v4();
+    let now = Utc::now().to_rfc3339();
+    let args_json = serde_json::to_string(args)?;
+
+    // Insert into v2_job
+    db.execute(
+        r#"
+        INSERT INTO v2_job (id, kind, script_lang, raw_code, args, parent_job, root_job, flow_step_id, created_at)
+        VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
+        "#,
+        libsql::params![
+            id.to_string(),
+            kind.as_str(),
+            script_lang.map(|l| l.as_str()),
+            raw_code,
+            args_json,
+            parent_id.to_string(),
+            root_id.to_string(),
+            step_id,
+            now.clone(),
+        ],
+    )
+    .await?;
+
+    // Insert into v2_job_queue
+    db.execute(
+        r#"
+        INSERT INTO v2_job_queue (id, tag, created_at, scheduled_for, running)
+        VALUES (?1, 'flow', ?2, ?3, 0)
+        "#,
+        libsql::params![id.to_string(), now.clone(), now],
+    )
+    .await?;
+
+    // Insert into v2_job_runtime
+    db.execute(
+        "INSERT INTO v2_job_runtime (id) VALUES (?1)",
+        libsql::params![id.to_string()],
+    )
+    .await?;
+
+    Ok(id)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::jobs::{push_preview, PreviewRequest};
+
+    #[tokio::test]
+    async fn test_pull_job() {
+        let db = LocalDb::in_memory().await.unwrap();
+
+        // Push a job
+        let req = PreviewRequest {
+            content: "export function main() { return 42; }".to_string(),
+            language: ScriptLang::Deno,
+            args: serde_json::json!({}),
+            lock: None,
+            tag: None,
+        };
+        let pushed_id = push_preview(&db, req).await.unwrap();
+
+        // Pull it
+        let job = pull_job(&db).await.unwrap().unwrap();
+        assert_eq!(job.id, pushed_id);
+        assert!(job.running);
+        assert_eq!(job.kind, JobKind::Preview);
+
+        // Queue should now be empty (job is running)
+        let job2 = pull_job(&db).await.unwrap();
+        assert!(job2.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_queue_stats() {
+        let db = LocalDb::in_memory().await.unwrap();
+
+        // Initially empty
+        let stats = queue_stats(&db).await.unwrap();
+        assert_eq!(stats.total, 0);
+
+        // Push two jobs
+        let req = PreviewRequest {
+            content: "test".to_string(),
+            language: ScriptLang::Deno,
+            args: serde_json::json!({}),
+            lock: None,
+            tag: None,
+        };
+        push_preview(&db, req.clone()).await.unwrap();
+        push_preview(&db, req).await.unwrap();
+
+        let stats = queue_stats(&db).await.unwrap();
+        assert_eq!(stats.total, 2);
+        assert_eq!(stats.pending, 2);
+        assert_eq!(stats.running, 0);
+
+        // Pull one
+        pull_job(&db).await.unwrap();
+
+        let stats = queue_stats(&db).await.unwrap();
+        assert_eq!(stats.total, 2);
+        assert_eq!(stats.pending, 1);
+        assert_eq!(stats.running, 1);
+    }
+}
--- a/backend/windmill-local/src/schema.rs
+++ b/backend/windmill-local/src/schema.rs
@@ -0,0 +1,218 @@
+//! SQLite schema for local mode
+//!
+//! This is a minimal schema supporting preview job execution.
+//! Key differences from PostgreSQL:
+//! - ENUMs are TEXT with CHECK constraints
+//! - JSONB is JSON (stored as TEXT in SQLite)
+//! - Arrays are JSON arrays
+//! - No FOR UPDATE SKIP LOCKED (single worker, in-process coordination)
+
+/// SQL to create the minimal schema for local mode preview execution
+pub const SCHEMA: &str = r#"
+-- Job kinds (equivalent to PostgreSQL ENUM)
+-- Values: script, preview, flow, flowpreview, dependencies, flowdependencies,
+--         script_hub, identity, http, graphql, postgresql, noop, appdependencies,
+--         deploymentcallback, singlescriptflow, flowscript, flownode, appscript
+
+-- Job status (equivalent to PostgreSQL ENUM)
+-- Values: success, failure, canceled, skipped
+
+-- Script languages (equivalent to PostgreSQL ENUM)
+-- Values: python3, deno, go, bash, postgresql, nativets, bun, mysql, bigquery,
+--         snowflake, graphql, powershell, mssql, php, bunnative, rust, ansible,
+--         csharp, oracledb, nu, java, duckdb
+
+-- Main job table (minimal for preview)
+CREATE TABLE IF NOT EXISTS v2_job (
+    id TEXT PRIMARY KEY,  -- UUID as TEXT
+    workspace_id TEXT NOT NULL DEFAULT 'local',
+
+    -- Raw code for preview jobs
+    raw_code TEXT,
+    raw_lock TEXT,
+    raw_flow TEXT,  -- JSON for flow definitions
+
+    -- Job metadata
+    tag TEXT DEFAULT 'deno',
+    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    created_by TEXT NOT NULL DEFAULT 'local_user',
+
+    -- Permission context (simplified for local mode)
+    permissioned_as TEXT NOT NULL DEFAULT 'u/local_user',
+    permissioned_as_email TEXT DEFAULT 'local@windmill.local',
+
+    -- Job type info
+    kind TEXT NOT NULL DEFAULT 'preview' CHECK (kind IN (
+        'script', 'preview', 'flow', 'flowpreview', 'dependencies',
+        'flowdependencies', 'script_hub', 'identity', 'http', 'graphql',
+        'postgresql', 'noop', 'appdependencies', 'deploymentcallback',
+        'singlescriptflow', 'flowscript', 'flownode', 'appscript'
+    )),
+
+    -- Script execution details
+    script_lang TEXT CHECK (script_lang IN (
+        'python3', 'deno', 'go', 'bash', 'postgresql', 'nativets', 'bun',
+        'mysql', 'bigquery', 'snowflake', 'graphql', 'powershell', 'mssql',
+        'php', 'bunnative', 'rust', 'ansible', 'csharp', 'oracledb', 'nu',
+        'java', 'duckdb'
+    )),
+
+    -- Flow execution details
+    parent_job TEXT,  -- UUID reference
+    root_job TEXT,    -- UUID reference
+    flow_step INTEGER,
+    flow_step_id TEXT,
+    flow_innermost_root_job TEXT,
+
+    -- Execution settings
+    timeout INTEGER,
+    priority INTEGER DEFAULT 0,
+    same_worker INTEGER DEFAULT 0,  -- BOOLEAN as INTEGER
+    visible_to_owner INTEGER DEFAULT 1,
+
+    -- Arguments (JSON)
+    args TEXT,  -- JSON object
+
+    -- Pre-run error if validation failed
+    pre_run_error TEXT
+);
+
+-- Job queue table
+CREATE TABLE IF NOT EXISTS v2_job_queue (
+    id TEXT PRIMARY KEY,  -- UUID, references v2_job.id
+    workspace_id TEXT NOT NULL DEFAULT 'local',
+
+    -- Timestamps
+    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    started_at TEXT,
+    scheduled_for TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+
+    -- Queue state
+    running INTEGER NOT NULL DEFAULT 0,  -- BOOLEAN
+    canceled_by TEXT,
+    canceled_reason TEXT,
+
+    -- Suspend state (for approval flows)
+    suspend INTEGER DEFAULT 0,
+    suspend_until TEXT,
+
+    -- Execution settings
+    tag TEXT DEFAULT 'deno',
+    priority INTEGER DEFAULT 0,
+    worker TEXT,
+
+    FOREIGN KEY (id) REFERENCES v2_job(id)
+);
+
+-- Index for queue ordering (simulates queue_sort_v2)
+CREATE INDEX IF NOT EXISTS idx_queue_sort ON v2_job_queue (
+    priority DESC, scheduled_for ASC, tag
+) WHERE running = 0;
+
+-- Job runtime tracking (heartbeat/ping)
+CREATE TABLE IF NOT EXISTS v2_job_runtime (
+    id TEXT PRIMARY KEY,  -- UUID, references v2_job.id
+    ping TEXT,  -- Timestamp
+    memory_peak INTEGER,
+
+    FOREIGN KEY (id) REFERENCES v2_job(id)
+);
+
+-- Completed jobs with results
+CREATE TABLE IF NOT EXISTS v2_job_completed (
+    id TEXT PRIMARY KEY,  -- UUID, references v2_job.id
+    workspace_id TEXT NOT NULL DEFAULT 'local',
+
+    -- Timing
+    started_at TEXT,
+    completed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    duration_ms INTEGER,
+
+    -- Result
+    result TEXT,  -- JSON
+    result_columns TEXT,  -- JSON array of column names
+
+    -- Status
+    status TEXT NOT NULL DEFAULT 'success' CHECK (status IN (
+        'success', 'failure', 'canceled', 'skipped'
+    )),
+
+    -- Cancellation details
+    canceled_by TEXT,
+    canceled_reason TEXT,
+
+    -- Flow status (for flow jobs)
+    flow_status TEXT,  -- JSON
+
+    -- Execution details
+    memory_peak INTEGER,
+    worker TEXT,
+    deleted INTEGER DEFAULT 0,  -- BOOLEAN
+
+    FOREIGN KEY (id) REFERENCES v2_job(id)
+);
+
+-- Index for completed job lookup by workspace and time
+CREATE INDEX IF NOT EXISTS idx_completed_workspace_time ON v2_job_completed (
+    workspace_id, completed_at DESC
+);
+
+-- Flow status tracking (separate from completed to allow updates during execution)
+CREATE TABLE IF NOT EXISTS v2_job_status (
+    id TEXT PRIMARY KEY,  -- UUID, references v2_job.id
+    flow_status TEXT,  -- JSON object tracking flow module execution
+    flow_leaf_jobs TEXT,  -- JSON object
+    workflow_as_code_status TEXT,  -- JSON object
+
+    FOREIGN KEY (id) REFERENCES v2_job(id)
+);
+
+-- Simplified job permissions (for local mode, mostly unused)
+CREATE TABLE IF NOT EXISTS job_perms (
+    job_id TEXT PRIMARY KEY,
+    email TEXT DEFAULT 'local@windmill.local',
+    username TEXT DEFAULT 'local_user',
+    is_admin INTEGER DEFAULT 1,  -- BOOLEAN
+    is_operator INTEGER DEFAULT 0,  -- BOOLEAN
+    workspace_id TEXT DEFAULT 'local',
+    groups TEXT,  -- JSON array
+    folders TEXT,  -- JSON array of objects
+
+    FOREIGN KEY (job_id) REFERENCES v2_job(id)
+);
+
+-- Simple audit log (optional for local mode)
+CREATE TABLE IF NOT EXISTS audit (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    workspace_id TEXT DEFAULT 'local',
+    timestamp TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    username TEXT DEFAULT 'local_user',
+    operation TEXT NOT NULL,
+    action_kind TEXT CHECK (action_kind IN ('create', 'update', 'delete', 'execute')),
+    resource TEXT,
+    parameters TEXT  -- JSON
+);
+
+-- Job logs storage
+CREATE TABLE IF NOT EXISTS job_logs (
+    job_id TEXT PRIMARY KEY,
+    workspace_id TEXT DEFAULT 'local',
+    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    logs TEXT,
+    log_offset INTEGER DEFAULT 0,
+
+    FOREIGN KEY (job_id) REFERENCES v2_job(id)
+);
+"#;
+
+/// SQL to drop all tables (for testing/reset)
+pub const DROP_SCHEMA: &str = r#"
+DROP TABLE IF EXISTS job_logs;
+DROP TABLE IF EXISTS audit;
+DROP TABLE IF EXISTS job_perms;
+DROP TABLE IF EXISTS v2_job_status;
+DROP TABLE IF EXISTS v2_job_completed;
+DROP TABLE IF EXISTS v2_job_runtime;
+DROP TABLE IF EXISTS v2_job_queue;
+DROP TABLE IF EXISTS v2_job;
+"#;
--- a/backend/windmill-local/src/server.rs
+++ b/backend/windmill-local/src/server.rs
@@ -0,0 +1,419 @@
+//! HTTP server for local mode
+//!
+//! Provides a minimal API compatible with Windmill's preview endpoints.
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::{
+    extract::{Path, State},
+    http::StatusCode,
+    response::IntoResponse,
+    routing::{get, post},
+    Json, Router,
+};
+use serde::{Deserialize, Serialize};
+use tokio::sync::watch;
+use tower_http::cors::{Any, CorsLayer};
+use tower_http::trace::TraceLayer;
+use uuid::Uuid;
+
+use crate::db::LocalDb;
+use crate::error::Result;
+use crate::jobs::{
+    get_completed_job, push_flow_preview, push_preview, FlowPreviewRequest,
+    JobStatus, PreviewRequest, ScriptLang,
+};
+use crate::worker::Worker;
+
+/// Application state shared across handlers
+pub struct AppState {
+    pub db: Arc<LocalDb>,
+}
+
+/// Local server that runs the API and embedded worker
+pub struct LocalServer {
+    db: Arc<LocalDb>,
+    addr: SocketAddr,
+}
+
+impl LocalServer {
+    /// Create a new local server
+    pub async fn new(addr: SocketAddr) -> Result<Self> {
+        let db = Arc::new(LocalDb::in_memory().await?);
+        Ok(Self { db, addr })
+    }
+
+    /// Create a local server with a file-based database
+    pub async fn with_file(addr: SocketAddr, db_path: &str) -> Result<Self> {
+        let db = Arc::new(LocalDb::file(db_path).await?);
+        Ok(Self { db, addr })
+    }
+
+    /// Create a local server connected to a remote Turso database
+    pub async fn with_turso(addr: SocketAddr, url: &str, auth_token: &str) -> Result<Self> {
+        let db = Arc::new(LocalDb::turso_remote(url, auth_token).await?);
+        Ok(Self { db, addr })
+    }
+
+    /// Run the server
+    pub async fn run(self) -> Result<()> {
+        let (shutdown_tx, shutdown_rx) = watch::channel(false);
+
+        // Start the embedded worker
+        let worker_db = self.db.clone();
+        let worker_handle = tokio::spawn(async move {
+            let mut worker = Worker::new(worker_db, shutdown_rx);
+            if let Err(e) = worker.run().await {
+                tracing::error!("Worker error: {}", e);
+            }
+        });
+
+        // Build the router
+        let state = Arc::new(AppState { db: self.db });
+        let app = create_router(state);
+
+        // Run the server
+        tracing::info!("Local server listening on {}", self.addr);
+        let listener = tokio::net::TcpListener::bind(self.addr).await.unwrap();
+
+        // Handle graceful shutdown
+        let shutdown_signal = async move {
+            tokio::signal::ctrl_c()
+                .await
+                .expect("Failed to install CTRL+C signal handler");
+            tracing::info!("Shutdown signal received");
+            shutdown_tx.send(true).ok();
+        };
+
+        axum::serve(listener, app)
+            .with_graceful_shutdown(shutdown_signal)
+            .await
+            .unwrap();
+
+        // Wait for worker to finish
+        worker_handle.await.ok();
+
+        Ok(())
+    }
+}
+
+/// Create the API router
+fn create_router(state: Arc<AppState>) -> Router {
+    let cors = CorsLayer::new()
+        .allow_origin(Any)
+        .allow_methods(Any)
+        .allow_headers(Any);
+
+    Router::new()
+        // Health check
+        .route("/health", get(health_check))
+        // Preview endpoints (mimics windmill-api)
+        .route("/api/w/:workspace/jobs/run/preview", post(run_preview))
+        .route(
+            "/api/w/:workspace/jobs/run_wait_result/preview",
+            post(run_wait_result_preview),
+        )
+        .route(
+            "/api/w/:workspace/jobs/run/preview_flow",
+            post(run_preview_flow),
+        )
+        .route(
+            "/api/w/:workspace/jobs/run_wait_result/preview_flow",
+            post(run_wait_result_preview_flow),
+        )
+        // Get job result
+        .route(
+            "/api/w/:workspace/jobs_u/completed/get_result/:job_id",
+            get(get_job_result),
+        )
+        .layer(TraceLayer::new_for_http())
+        .layer(cors)
+        .with_state(state)
+}
+
+// === Request/Response Types ===
+
+#[derive(Debug, Deserialize)]
+struct PreviewPayload {
+    content: String,
+    language: String,
+    #[serde(default)]
+    args: serde_json::Value,
+    lock: Option<String>,
+    tag: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct FlowPreviewPayload {
+    value: serde_json::Value,
+    #[serde(default)]
+    args: serde_json::Value,
+    tag: Option<String>,
+}
+
+#[derive(Debug, Serialize)]
+struct JobCreatedResponse {
+    job_id: String,
+}
+
+#[derive(Debug, Serialize)]
+struct ErrorResponse {
+    error: String,
+}
+
+// === Handlers ===
+
+async fn health_check() -> &'static str {
+    "OK"
+}
+
+/// Run a preview job (async - returns job ID)
+async fn run_preview(
+    State(state): State<Arc<AppState>>,
+    Path(_workspace): Path<String>,
+    Json(payload): Json<PreviewPayload>,
+) -> impl IntoResponse {
+    let lang = match ScriptLang::from_str(&payload.language) {
+        Some(l) => l,
+        None => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(ErrorResponse {
+                    error: format!("Unknown language: {}", payload.language),
+                }),
+            )
+                .into_response();
+        }
+    };
+
+    let req = PreviewRequest {
+        content: payload.content,
+        language: lang,
+        args: payload.args,
+        lock: payload.lock,
+        tag: payload.tag,
+    };
+
+    match push_preview(&state.db, req).await {
+        Ok(job_id) => (
+            StatusCode::CREATED,
+            Json(JobCreatedResponse {
+                job_id: job_id.to_string(),
+            }),
+        )
+            .into_response(),
+        Err(e) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(ErrorResponse {
+                error: e.to_string(),
+            }),
+        )
+            .into_response(),
+    }
+}
+
+/// Run a preview job and wait for result
+async fn run_wait_result_preview(
+    State(state): State<Arc<AppState>>,
+    Path(_workspace): Path<String>,
+    Json(payload): Json<PreviewPayload>,
+) -> impl IntoResponse {
+    let lang = match ScriptLang::from_str(&payload.language) {
+        Some(l) => l,
+        None => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(serde_json::json!({"error": format!("Unknown language: {}", payload.language)})),
+            )
+                .into_response();
+        }
+    };
+
+    let req = PreviewRequest {
+        content: payload.content,
+        language: lang,
+        args: payload.args,
+        lock: payload.lock,
+        tag: payload.tag,
+    };
+
+    let job_id = match push_preview(&state.db, req).await {
+        Ok(id) => id,
+        Err(e) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({"error": e.to_string()})),
+            )
+                .into_response();
+        }
+    };
+
+    // Poll for result with timeout
+    wait_for_result(&state.db, job_id, Duration::from_secs(60)).await
+}
+
+/// Run a flow preview job (async - returns job ID)
+async fn run_preview_flow(
+    State(state): State<Arc<AppState>>,
+    Path(_workspace): Path<String>,
+    Json(payload): Json<FlowPreviewPayload>,
+) -> impl IntoResponse {
+    let req = FlowPreviewRequest {
+        value: payload.value,
+        args: payload.args,
+        tag: payload.tag,
+    };
+
+    match push_flow_preview(&state.db, req).await {
+        Ok(job_id) => (
+            StatusCode::CREATED,
+            Json(JobCreatedResponse {
+                job_id: job_id.to_string(),
+            }),
+        )
+            .into_response(),
+        Err(e) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(ErrorResponse {
+                error: e.to_string(),
+            }),
+        )
+            .into_response(),
+    }
+}
+
+/// Run a flow preview job and wait for result
+async fn run_wait_result_preview_flow(
+    State(state): State<Arc<AppState>>,
+    Path(_workspace): Path<String>,
+    Json(payload): Json<FlowPreviewPayload>,
+) -> impl IntoResponse {
+    let req = FlowPreviewRequest {
+        value: payload.value,
+        args: payload.args,
+        tag: payload.tag,
+    };
+
+    let job_id = match push_flow_preview(&state.db, req).await {
+        Ok(id) => id,
+        Err(e) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({"error": e.to_string()})),
+            )
+                .into_response();
+        }
+    };
+
+    // Poll for result with timeout
+    wait_for_result(&state.db, job_id, Duration::from_secs(120)).await
+}
+
+/// Get the result of a completed job
+async fn get_job_result(
+    State(state): State<Arc<AppState>>,
+    Path((_workspace, job_id)): Path<(String, String)>,
+) -> impl IntoResponse {
+    let job_id = match Uuid::parse_str(&job_id) {
+        Ok(id) => id,
+        Err(_) => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(serde_json::json!({"error": "Invalid job ID"})),
+            )
+                .into_response();
+        }
+    };
+
+    match get_completed_job(&state.db, job_id).await {
+        Ok(Some(job)) => {
+            if job.status == JobStatus::Success {
+                (StatusCode::OK, Json(job.result)).into_response()
+            } else {
+                (StatusCode::INTERNAL_SERVER_ERROR, Json(job.result)).into_response()
+            }
+        }
+        Ok(None) => (
+            StatusCode::NOT_FOUND,
+            Json(serde_json::json!({"error": "Job not found or not completed"})),
+        )
+            .into_response(),
+        Err(e) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({"error": e.to_string()})),
+        )
+            .into_response(),
+    }
+}
+
+/// Poll for job completion with timeout
+async fn wait_for_result(
+    db: &LocalDb,
+    job_id: Uuid,
+    timeout: Duration,
+) -> axum::response::Response {
+    let start = std::time::Instant::now();
+    let fast_poll_duration = Duration::from_secs(2);
+    let fast_poll_interval = Duration::from_millis(50);
+    let slow_poll_interval = Duration::from_millis(200);
+
+    loop {
+        if start.elapsed() > timeout {
+            return (
+                StatusCode::REQUEST_TIMEOUT,
+                Json(serde_json::json!({"error": "Timeout waiting for job result"})),
+            )
+                .into_response();
+        }
+
+        match get_completed_job(db, job_id).await {
+            Ok(Some(job)) => {
+                if job.status == JobStatus::Success {
+                    return (StatusCode::OK, Json(job.result)).into_response();
+                } else {
+                    return (StatusCode::INTERNAL_SERVER_ERROR, Json(job.result)).into_response();
+                }
+            }
+            Ok(None) => {
+                // Job not completed yet, keep polling
+                let interval = if start.elapsed() < fast_poll_duration {
+                    fast_poll_interval
+                } else {
+                    slow_poll_interval
+                };
+                tokio::time::sleep(interval).await;
+            }
+            Err(e) => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(serde_json::json!({"error": e.to_string()})),
+                )
+                    .into_response();
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::body::Body;
+    use axum::http::Request;
+    use tower::ServiceExt;
+
+    #[tokio::test]
+    async fn test_health_check() {
+        let db = Arc::new(LocalDb::in_memory().await.unwrap());
+        let state = Arc::new(AppState { db });
+        let app = create_router(state);
+
+        let response = app
+            .oneshot(Request::builder().uri("/health").body(Body::empty()).unwrap())
+            .await
+            .unwrap();
+
+        assert_eq!(response.status(), StatusCode::OK);
+    }
+}
--- a/backend/windmill-local/src/worker.rs
+++ b/backend/windmill-local/src/worker.rs
@@ -0,0 +1,204 @@
+//! Worker for local mode
+//!
+//! A single embedded worker that pulls jobs from the queue and executes them.
+
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::watch;
+use chrono::Utc;
+
+use crate::db::LocalDb;
+use crate::error::Result;
+use crate::executor::{execute_script, ExecutionResult};
+use crate::flow_executor;
+use crate::jobs::{complete_job, JobKind, JobStatus, QueuedJob};
+use crate::queue::pull_job;
+use windmill_common::flows::FlowValue;
+
+/// Worker that processes jobs from the queue
+pub struct Worker {
+    db: Arc<LocalDb>,
+    /// Channel to signal shutdown
+    shutdown_rx: watch::Receiver<bool>,
+}
+
+impl Worker {
+    /// Create a new worker
+    pub fn new(db: Arc<LocalDb>, shutdown_rx: watch::Receiver<bool>) -> Self {
+        Self { db, shutdown_rx }
+    }
+
+    /// Run the worker loop
+    pub async fn run(&mut self) -> Result<()> {
+        tracing::info!("Worker started");
+
+        loop {
+            // Check for shutdown signal
+            if *self.shutdown_rx.borrow() {
+                tracing::info!("Worker received shutdown signal");
+                break;
+            }
+
+            // Try to pull a job
+            match pull_job(&self.db).await {
+                Ok(Some(job)) => {
+                    tracing::info!("Processing job: {} (kind: {:?})", job.id, job.kind);
+                    if let Err(e) = self.process_job(job).await {
+                        tracing::error!("Error processing job: {}", e);
+                    }
+                }
+                Ok(None) => {
+                    // No jobs available, wait a bit before polling again
+                    tokio::select! {
+                        _ = tokio::time::sleep(Duration::from_millis(100)) => {}
+                        _ = self.shutdown_rx.changed() => {}
+                    }
+                }
+                Err(e) => {
+                    tracing::error!("Error pulling job: {}", e);
+                    tokio::time::sleep(Duration::from_millis(500)).await;
+                }
+            }
+        }
+
+        tracing::info!("Worker stopped");
+        Ok(())
+    }
+
+    /// Process a single job
+    async fn process_job(&self, job: QueuedJob) -> Result<()> {
+        let started_at = Utc::now();
+
+        match job.kind {
+            JobKind::Preview => {
+                self.process_preview_job(job, started_at).await
+            }
+            JobKind::FlowPreview => {
+                self.process_flow_preview_job(job, started_at).await
+            }
+            _ => {
+                // Unsupported job kind
+                let error_result = serde_json::json!({
+                    "error": format!("Unsupported job kind in local mode: {:?}", job.kind)
+                });
+                complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await
+            }
+        }
+    }
+
+    /// Process a script preview job
+    async fn process_preview_job(&self, job: QueuedJob, started_at: chrono::DateTime<Utc>) -> Result<()> {
+        let Some(code) = &job.raw_code else {
+            let error_result = serde_json::json!({"error": "No code provided for preview"});
+            return complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await;
+        };
+
+        let Some(lang) = job.script_lang else {
+            let error_result = serde_json::json!({"error": "No language specified for preview"});
+            return complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await;
+        };
+
+        // Execute the script
+        let exec_result = execute_script(lang, code, &job.args).await;
+
+        match exec_result {
+            Ok(ExecutionResult { success, result, logs }) => {
+                tracing::debug!("Job {} logs:\n{}", job.id, logs);
+                let status = if success { JobStatus::Success } else { JobStatus::Failure };
+                complete_job(&self.db, job.id, status, result, started_at).await
+            }
+            Err(e) => {
+                let error_result = serde_json::json!({"error": e.to_string()});
+                complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await
+            }
+        }
+    }
+
+    /// Process a flow preview job using the full flow executor
+    async fn process_flow_preview_job(&self, job: QueuedJob, started_at: chrono::DateTime<Utc>) -> Result<()> {
+        let Some(flow_json) = &job.raw_flow else {
+            let error_result = serde_json::json!({"error": "No flow definition provided"});
+            return complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await;
+        };
+
+        // Parse the flow value using windmill-common types
+        let flow_value: FlowValue = match serde_json::from_value(flow_json.clone()) {
+            Ok(fv) => fv,
+            Err(e) => {
+                let error_result = serde_json::json!({"error": format!("Failed to parse flow: {}", e)});
+                return complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await;
+            }
+        };
+
+        if flow_value.modules.is_empty() {
+            let error_result = serde_json::json!({"error": "Flow has no modules"});
+            return complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await;
+        }
+
+        tracing::info!("Executing flow with {} modules", flow_value.modules.len());
+
+        // Execute the flow using the full flow executor
+        match flow_executor::execute_flow(&self.db, &flow_value, job.args.clone()).await {
+            Ok((result, status)) => {
+                let is_failure = status.failure_module.is_some();
+                let final_result = serde_json::json!({
+                    "result": result,
+                    "flow_status": status
+                });
+
+                let job_status = if is_failure {
+                    JobStatus::Failure
+                } else {
+                    JobStatus::Success
+                };
+
+                complete_job(&self.db, job.id, job_status, final_result, started_at).await
+            }
+            Err(e) => {
+                let error_result = serde_json::json!({"error": e.to_string()});
+                complete_job(&self.db, job.id, JobStatus::Failure, error_result, started_at).await
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::jobs::{get_completed_job, push_preview, PreviewRequest, ScriptLang};
+
+    #[tokio::test]
+    async fn test_worker_processes_bash_preview() {
+        let db = Arc::new(LocalDb::in_memory().await.unwrap());
+        let (shutdown_tx, shutdown_rx) = watch::channel(false);
+
+        // Push a bash preview job
+        let req = PreviewRequest {
+            content: "echo 42".to_string(),
+            language: ScriptLang::Bash,
+            args: serde_json::json!({}),
+            lock: None,
+            tag: None,
+        };
+        let job_id = push_preview(&db, req).await.unwrap();
+
+        // Create and run worker for one iteration
+        let mut worker = Worker::new(db.clone(), shutdown_rx);
+
+        // Process one job then shutdown
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(500)).await;
+            shutdown_tx.send(true).unwrap();
+        });
+
+        worker.run().await.unwrap();
+
+        // Check the job completed
+        let completed = get_completed_job(&db, job_id).await.unwrap();
+        assert!(completed.is_some());
+        let completed = completed.unwrap();
+        assert_eq!(completed.status, JobStatus::Success);
+        // Output "42" is parsed as JSON number
+        assert_eq!(completed.result, serde_json::json!(42));
+    }
+}