From d490dbe4faa77b637721af0d14da89c0792a55b2 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 3 Apr 2026 17:24:51 -0700 Subject: [PATCH] refactor: simplify run storage and retire sqlite metadata --- AGENTS.md | 1 - Cargo.lock | 520 ------------------ .../fabro-cli/src/commands/pr/create.rs | 2 +- lib/crates/fabro-cli/src/commands/pr/list.rs | 4 +- lib/crates/fabro-cli/src/commands/pr/mod.rs | 3 +- lib/crates/fabro-cli/src/commands/run/cp.rs | 2 +- lib/crates/fabro-cli/src/commands/run/diff.rs | 5 +- lib/crates/fabro-cli/src/commands/run/logs.rs | 5 +- lib/crates/fabro-cli/src/commands/run/mod.rs | 16 +- .../fabro-cli/src/commands/run/preview.rs | 2 +- .../fabro-cli/src/commands/run/resume.rs | 3 +- lib/crates/fabro-cli/src/commands/run/ssh.rs | 3 +- lib/crates/fabro-cli/src/commands/run/wait.rs | 14 +- .../fabro-cli/src/commands/runs/inspect.rs | 5 +- .../fabro-cli/src/commands/runs/list.rs | 36 +- lib/crates/fabro-cli/src/commands/runs/rm.rs | 25 +- .../fabro-cli/src/commands/store/dump.rs | 39 +- .../fabro-cli/src/commands/system/df.rs | 51 +- .../fabro-cli/src/commands/system/prune.rs | 12 +- lib/crates/fabro-cli/tests/it/cmd/rewind.rs | 2 +- .../fabro-cli/tests/it/cmd/server_start.rs | 1 - .../fabro-cli/tests/it/cmd/server_status.rs | 1 - .../fabro-cli/tests/it/cmd/server_stop.rs | 1 - lib/crates/fabro-cli/tests/it/cmd/support.rs | 10 +- .../fabro-cli/tests/it/cmd/system_df.rs | 18 +- .../tests/it/scenario/server_lifecycle.rs | 1 - .../fabro-cli/tests/it/workflow/real_cli.rs | 1 - lib/crates/fabro-db/Cargo.toml | 22 - .../migrations/001_create_workflow_runs.sql | 8 - .../002_rename_logs_dir_to_run_dir.sql | 1 - lib/crates/fabro-db/src/lib.rs | 124 ----- lib/crates/fabro-db/src/migrate.rs | 48 -- lib/crates/fabro-db/src/workflow_run.rs | 11 - lib/crates/fabro-server/Cargo.toml | 2 - lib/crates/fabro-server/src/serve.rs | 7 +- lib/crates/fabro-server/src/server.rs | 102 ++-- lib/crates/fabro-server/src/sessions.rs | 8 - lib/crates/fabro-server/tests/it/api.rs | 25 +- lib/crates/fabro-server/tests/it/helpers.rs | 6 +- .../tests/it/openapi_conformance.rs | 6 +- .../fabro-server/tests/it/pagination.rs | 6 +- lib/crates/fabro-store/src/lib.rs | 6 +- lib/crates/fabro-store/src/run_state.rs | 62 +-- lib/crates/fabro-store/src/runtime.rs | 6 - lib/crates/fabro-store/src/slate/catalog.rs | 46 +- lib/crates/fabro-store/src/slate/mod.rs | 346 +++++++----- lib/crates/fabro-store/src/slate/run_store.rs | 69 +-- lib/crates/fabro-store/src/types.rs | 72 ++- .../src/operations/rebuild_meta.rs | 6 +- .../src/pipeline/pull_request.rs | 50 +- lib/crates/fabro-workflow/src/run_dump.rs | 61 +- lib/crates/fabro-workflow/src/run_lookup.rs | 244 ++++---- 52 files changed, 766 insertions(+), 1361 deletions(-) delete mode 100644 lib/crates/fabro-db/Cargo.toml delete mode 100644 lib/crates/fabro-db/migrations/001_create_workflow_runs.sql delete mode 100644 lib/crates/fabro-db/migrations/002_rename_logs_dir_to_run_dir.sql delete mode 100644 lib/crates/fabro-db/src/lib.rs delete mode 100644 lib/crates/fabro-db/src/migrate.rs delete mode 100644 lib/crates/fabro-db/src/workflow_run.rs diff --git a/AGENTS.md b/AGENTS.md index a714d7710..f28f0ae63 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -57,7 +57,6 @@ Fabro is an AI-powered workflow orchestration platform. Workflows are defined as - **fabro-llm** — Unified LLM client with providers: Anthropic, OpenAI, Gemini, OpenAI-compatible, plus retry/middleware/streaming - **fabro-api** — Auto-generated Rust types and reqwest HTTP client from OpenAPI spec (build.rs + progenitor) - **fabro-github** — GitHub App auth (JWT signing, installation tokens, PR creation) -- **fabro-db** — SQLite with WAL mode, schema migrations - **fabro-mcp** — Model Context Protocol client/server - **fabro-slack** — Slack integration (socket mode, blocks API) - **fabro-devcontainer** — Parses `.devcontainer/devcontainer.json` for container setup diff --git a/Cargo.lock b/Cargo.lock index 095c67621..2e2f4f415 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -303,15 +303,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "atoi" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528" -dependencies = [ - "num-traits", -] - [[package]] name = "atomic" version = "0.6.1" @@ -487,12 +478,6 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "base64ct" -version = "1.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" - [[package]] name = "bincode" version = "1.3.3" @@ -522,9 +507,6 @@ name = "bitflags" version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" -dependencies = [ - "serde_core", -] [[package]] name = "block-buffer" @@ -623,12 +605,6 @@ version = "1.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec" -[[package]] -name = "byteorder" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" - [[package]] name = "bytes" version = "1.11.1" @@ -817,12 +793,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "const-oid" -version = "0.9.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" - [[package]] name = "convert_case" version = "0.10.0" @@ -894,21 +864,6 @@ dependencies = [ "libc", ] -[[package]] -name = "crc" -version = "3.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" -dependencies = [ - "crc-catalog", -] - -[[package]] -name = "crc-catalog" -version = "2.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" - [[package]] name = "crc32fast" version = "1.5.0" @@ -1223,17 +1178,6 @@ dependencies = [ "uuid", ] -[[package]] -name = "der" -version = "0.7.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" -dependencies = [ - "const-oid", - "pem-rfc7468", - "zeroize", -] - [[package]] name = "der-parser" version = "9.0.0" @@ -1316,7 +1260,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", - "const-oid", "crypto-common", "subtle", ] @@ -1414,9 +1357,6 @@ name = "either" version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" -dependencies = [ - "serde", -] [[package]] name = "email_address" @@ -1479,17 +1419,6 @@ dependencies = [ "libc", ] -[[package]] -name = "etcetera" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "136d1b5283a1ab77bd9257427ffd09d8667ced0570b6f938942bc7568ed5b943" -dependencies = [ - "cfg-if", - "home", - "windows-sys 0.48.0", -] - [[package]] name = "event-listener" version = "5.4.1" @@ -1695,17 +1624,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "fabro-db" -version = "0.176.2" -dependencies = [ - "chrono", - "sqlx", - "thiserror 2.0.18", - "tokio", - "tracing", -] - [[package]] name = "fabro-devcontainer" version = "0.176.2" @@ -1944,7 +1862,6 @@ dependencies = [ "fabro-agent", "fabro-api", "fabro-config", - "fabro-db", "fabro-github", "fabro-graphviz", "fabro-hooks", @@ -1977,7 +1894,6 @@ dependencies = [ "serde_json", "serde_yaml", "sha2", - "sqlx", "tempfile", "tokio", "tokio-rustls", @@ -2521,17 +2437,6 @@ dependencies = [ "futures-util", ] -[[package]] -name = "futures-intrusive" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f" -dependencies = [ - "futures-core", - "lock_api", - "parking_lot", -] - [[package]] name = "futures-io" version = "0.3.32" @@ -2737,15 +2642,6 @@ dependencies = [ "foldhash 0.2.0", ] -[[package]] -name = "hashlink" -version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" -dependencies = [ - "hashbrown 0.15.5", -] - [[package]] name = "headers" version = "0.4.1" @@ -2806,15 +2702,6 @@ dependencies = [ "digest", ] -[[package]] -name = "home" -version = "0.5.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" -dependencies = [ - "windows-sys 0.61.2", -] - [[package]] name = "hostname" version = "0.4.2" @@ -3439,9 +3326,6 @@ name = "lazy_static" version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" -dependencies = [ - "spin", -] [[package]] name = "leb128fmt" @@ -3467,12 +3351,6 @@ dependencies = [ "pkg-config", ] -[[package]] -name = "libm" -version = "0.2.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" - [[package]] name = "libredox" version = "0.1.12" @@ -3484,17 +3362,6 @@ dependencies = [ "redox_syscall 0.7.3", ] -[[package]] -name = "libsqlite3-sys" -version = "0.30.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" -dependencies = [ - "cc", - "pkg-config", - "vcpkg", -] - [[package]] name = "libz-sys" version = "1.1.24" @@ -3911,22 +3778,6 @@ dependencies = [ "num-traits", ] -[[package]] -name = "num-bigint-dig" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7" -dependencies = [ - "lazy_static", - "libm", - "num-integer", - "num-iter", - "num-traits", - "rand 0.8.5", - "smallvec", - "zeroize", -] - [[package]] name = "num-cmp" version = "0.1.0" @@ -3986,7 +3837,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" dependencies = [ "autocfg", - "libm", ] [[package]] @@ -4452,15 +4302,6 @@ dependencies = [ "serde_core", ] -[[package]] -name = "pem-rfc7468" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412" -dependencies = [ - "base64ct", -] - [[package]] name = "percent-encoding" version = "2.3.2" @@ -4580,27 +4421,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" -[[package]] -name = "pkcs1" -version = "0.7.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f" -dependencies = [ - "der", - "pkcs8", - "spki", -] - -[[package]] -name = "pkcs8" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" -dependencies = [ - "der", - "spki", -] - [[package]] name = "pkg-config" version = "0.3.32" @@ -5218,26 +5038,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "rsa" -version = "0.9.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" -dependencies = [ - "const-oid", - "digest", - "num-bigint-dig", - "num-integer", - "num-traits", - "pkcs1", - "pkcs8", - "rand_core 0.6.4", - "signature", - "spki", - "subtle", - "zeroize", -] - [[package]] name = "rust-embed" version = "8.11.0" @@ -5888,7 +5688,6 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ - "digest", "rand_core 0.6.4", ] @@ -6000,9 +5799,6 @@ name = "smallvec" version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" -dependencies = [ - "serde", -] [[package]] name = "socket2" @@ -6023,208 +5819,6 @@ dependencies = [ "lock_api", ] -[[package]] -name = "spki" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" -dependencies = [ - "base64ct", - "der", -] - -[[package]] -name = "sqlx" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc" -dependencies = [ - "sqlx-core", - "sqlx-macros", - "sqlx-mysql", - "sqlx-postgres", - "sqlx-sqlite", -] - -[[package]] -name = "sqlx-core" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6" -dependencies = [ - "base64", - "bytes", - "chrono", - "crc", - "crossbeam-queue", - "either", - "event-listener", - "futures-core", - "futures-intrusive", - "futures-io", - "futures-util", - "hashbrown 0.15.5", - "hashlink", - "indexmap 2.13.0", - "log", - "memchr", - "once_cell", - "percent-encoding", - "serde", - "serde_json", - "sha2", - "smallvec", - "thiserror 2.0.18", - "tokio", - "tokio-stream", - "tracing", - "url", -] - -[[package]] -name = "sqlx-macros" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2d452988ccaacfbf5e0bdbc348fb91d7c8af5bee192173ac3636b5fb6e6715d" -dependencies = [ - "proc-macro2", - "quote", - "sqlx-core", - "sqlx-macros-core", - "syn 2.0.117", -] - -[[package]] -name = "sqlx-macros-core" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b" -dependencies = [ - "dotenvy", - "either", - "heck 0.5.0", - "hex", - "once_cell", - "proc-macro2", - "quote", - "serde", - "serde_json", - "sha2", - "sqlx-core", - "sqlx-mysql", - "sqlx-postgres", - "sqlx-sqlite", - "syn 2.0.117", - "tokio", - "url", -] - -[[package]] -name = "sqlx-mysql" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aa003f0038df784eb8fecbbac13affe3da23b45194bd57dba231c8f48199c526" -dependencies = [ - "atoi", - "base64", - "bitflags", - "byteorder", - "bytes", - "chrono", - "crc", - "digest", - "dotenvy", - "either", - "futures-channel", - "futures-core", - "futures-io", - "futures-util", - "generic-array", - "hex", - "hkdf", - "hmac", - "itoa", - "log", - "md-5", - "memchr", - "once_cell", - "percent-encoding", - "rand 0.8.5", - "rsa", - "serde", - "sha1", - "sha2", - "smallvec", - "sqlx-core", - "stringprep", - "thiserror 2.0.18", - "tracing", - "whoami", -] - -[[package]] -name = "sqlx-postgres" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46" -dependencies = [ - "atoi", - "base64", - "bitflags", - "byteorder", - "chrono", - "crc", - "dotenvy", - "etcetera", - "futures-channel", - "futures-core", - "futures-util", - "hex", - "hkdf", - "hmac", - "home", - "itoa", - "log", - "md-5", - "memchr", - "once_cell", - "rand 0.8.5", - "serde", - "serde_json", - "sha2", - "smallvec", - "sqlx-core", - "stringprep", - "thiserror 2.0.18", - "tracing", - "whoami", -] - -[[package]] -name = "sqlx-sqlite" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2d12fe70b2c1b4401038055f90f151b78208de1f9f89a7dbfd41587a10c3eea" -dependencies = [ - "atoi", - "chrono", - "flume", - "futures-channel", - "futures-core", - "futures-executor", - "futures-intrusive", - "futures-util", - "libsqlite3-sys", - "log", - "percent-encoding", - "serde", - "serde_urlencoded", - "sqlx-core", - "thiserror 2.0.18", - "tracing", - "url", -] - [[package]] name = "sse-stream" version = "0.2.1" @@ -6287,17 +5881,6 @@ version = "2.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b3c8667cd96245cbb600b8dec5680a7319edd719c5aa2b5d23c6bff94f39765" -[[package]] -name = "stringprep" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" -dependencies = [ - "unicode-bidi", - "unicode-normalization", - "unicode-properties", -] - [[package]] name = "strsim" version = "0.10.0" @@ -7022,12 +6605,6 @@ version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" -[[package]] -name = "unicode-bidi" -version = "0.3.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" - [[package]] name = "unicode-general-category" version = "1.1.0" @@ -7040,21 +6617,6 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" -[[package]] -name = "unicode-normalization" -version = "0.1.25" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8" -dependencies = [ - "tinyvec", -] - -[[package]] -name = "unicode-properties" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" - [[package]] name = "unicode-segmentation" version = "1.12.0" @@ -7257,12 +6819,6 @@ dependencies = [ "wit-bindgen", ] -[[package]] -name = "wasite" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b" - [[package]] name = "wasm-bindgen" version = "0.2.114" @@ -7441,16 +6997,6 @@ dependencies = [ "rustls-pki-types", ] -[[package]] -name = "whoami" -version = "1.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d4a4db5077702ca3015d3d02d74974948aba2ad9e12ab7df718ee64ccd7e97d" -dependencies = [ - "libredox", - "wasite", -] - [[package]] name = "winapi" version = "0.3.9" @@ -7683,15 +7229,6 @@ dependencies = [ "windows-targets 0.42.2", ] -[[package]] -name = "windows-sys" -version = "0.48.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" -dependencies = [ - "windows-targets 0.48.5", -] - [[package]] name = "windows-sys" version = "0.52.0" @@ -7743,21 +7280,6 @@ dependencies = [ "windows_x86_64_msvc 0.42.2", ] -[[package]] -name = "windows-targets" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c" -dependencies = [ - "windows_aarch64_gnullvm 0.48.5", - "windows_aarch64_msvc 0.48.5", - "windows_i686_gnu 0.48.5", - "windows_i686_msvc 0.48.5", - "windows_x86_64_gnu 0.48.5", - "windows_x86_64_gnullvm 0.48.5", - "windows_x86_64_msvc 0.48.5", -] - [[package]] name = "windows-targets" version = "0.52.6" @@ -7815,12 +7337,6 @@ version = "0.42.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8" -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8" - [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" @@ -7839,12 +7355,6 @@ version = "0.42.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43" -[[package]] -name = "windows_aarch64_msvc" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc" - [[package]] name = "windows_aarch64_msvc" version = "0.52.6" @@ -7863,12 +7373,6 @@ version = "0.42.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f" -[[package]] -name = "windows_i686_gnu" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e" - [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -7899,12 +7403,6 @@ version = "0.42.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060" -[[package]] -name = "windows_i686_msvc" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406" - [[package]] name = "windows_i686_msvc" version = "0.52.6" @@ -7923,12 +7421,6 @@ version = "0.42.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36" -[[package]] -name = "windows_x86_64_gnu" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e" - [[package]] name = "windows_x86_64_gnu" version = "0.52.6" @@ -7947,12 +7439,6 @@ version = "0.42.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3" -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc" - [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" @@ -7971,12 +7457,6 @@ version = "0.42.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0" -[[package]] -name = "windows_x86_64_msvc" -version = "0.48.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" - [[package]] name = "windows_x86_64_msvc" version = "0.52.6" diff --git a/lib/crates/fabro-cli/src/commands/pr/create.rs b/lib/crates/fabro-cli/src/commands/pr/create.rs index b83ae8709..0436979f5 100644 --- a/lib/crates/fabro-cli/src/commands/pr/create.rs +++ b/lib/crates/fabro-cli/src/commands/pr/create.rs @@ -32,7 +32,7 @@ async fn create_from( let storage_dir = base.parent().unwrap_or(base); let store = store::build_store(storage_dir)?; let run = resolve_run_combined(store.as_ref(), base, &args.run_id).await?; - let run_store = store::open_run_reader(storage_dir, &run.run_id).await?; + let run_store = store::open_run_reader(storage_dir, &run.run_id()).await?; let state = run_store.state().await?; let record = state.run.context("Failed to load run record from store")?; diff --git a/lib/crates/fabro-cli/src/commands/pr/list.rs b/lib/crates/fabro-cli/src/commands/pr/list.rs index 99bb58935..e4d3d2a34 100644 --- a/lib/crates/fabro-cli/src/commands/pr/list.rs +++ b/lib/crates/fabro-cli/src/commands/pr/list.rs @@ -49,10 +49,10 @@ async fn list_from( let mut entries: Vec<(String, PullRequestRecord)> = Vec::new(); for run in &runs { - if let Ok(run_store) = store.open_run_reader(&run.run_id).await { + if let Ok(run_store) = store.open_run_reader(&run.run_id()).await { if let Ok(state) = run_store.state().await { if let Some(record) = state.pull_request { - entries.push((run.run_id.to_string(), record)); + entries.push((run.run_id().to_string(), record)); } } } diff --git a/lib/crates/fabro-cli/src/commands/pr/mod.rs b/lib/crates/fabro-cli/src/commands/pr/mod.rs index de94f919f..b2fb21019 100644 --- a/lib/crates/fabro-cli/src/commands/pr/mod.rs +++ b/lib/crates/fabro-cli/src/commands/pr/mod.rs @@ -38,8 +38,9 @@ pub(crate) async fn load_pr_record( let storage_dir = base.parent().unwrap_or(base); let store = store::build_store(storage_dir)?; let run = resolve_run_combined(store.as_ref(), base, run_id).await?; + let run_id = run.run_id(); let run_dir = run.path; - let run_store = store::open_run_reader(storage_dir, &run.run_id).await?; + let run_store = store::open_run_reader(storage_dir, &run_id).await?; let state = run_store.state().await?; let record = state.pull_request.with_context(|| { format!("No pull request found in store. Create one first with: fabro pr create {run_id}") diff --git a/lib/crates/fabro-cli/src/commands/run/cp.rs b/lib/crates/fabro-cli/src/commands/run/cp.rs index 4f0299216..2407ad19f 100644 --- a/lib/crates/fabro-cli/src/commands/run/cp.rs +++ b/lib/crates/fabro-cli/src/commands/run/cp.rs @@ -130,7 +130,7 @@ async fn load_sandbox( ) -> Result> { let store = store::build_store(storage_dir)?; let run = resolve_run_combined(store.as_ref(), base, run_prefix).await?; - let run_store = store::open_run_reader(storage_dir, &run.run_id).await?; + let run_store = store::open_run_reader(storage_dir, &run.run_id()).await?; let record = run_store .state() .await? diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index 2c2f1e3d3..abc7cbb5a 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -18,13 +18,14 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> { let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; + let run_id = run.run_id(); + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_id).await?; let patch = resolve_diff(&run.path, &run_store, &args).await?; if globals.json { let mut value = serde_json::json!({ - "run_id": run.run_id, + "run_id": run_id, "node": args.node, }); if args.shortstat { diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 41da15366..df31e853b 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -24,14 +24,15 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - info!(run_id = %run.run_id, "Showing logs"); + let run_id = run.run_id(); + info!(run_id = %run_id, "Showing logs"); let since_cutoff = match &args.since { Some(value) => Some(parse_since(value)?), None => None, }; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_id).await?; let (all_lines, last_seq) = match run_store.list_events().await { Ok(events) => { let last_seq = events.last().map_or(0, |event| event.seq); diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 048eea894..76d5e66e6 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -59,15 +59,12 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run_info = resolve_run_combined(store.as_ref(), &base, &run).await?; - let child = start::start_run( - &run_info.path, - &run_info.run_id, - &cli_settings.storage_dir(), - false, - ) - .await?; + let run_id = run_info.run_id(); + let child = + start::start_run(&run_info.path, &run_id, &cli_settings.storage_dir(), false) + .await?; if globals.json { - print_json_pretty(&serde_json::json!({ "run_id": run_info.run_id }))?; + print_json_pretty(&serde_json::json!({ "run_id": run_id }))?; } else { eprintln!("Started engine process (PID {})", child.id()); } @@ -79,10 +76,11 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run_info = resolve_run_combined(store.as_ref(), &base, &run).await?; + let run_id = run_info.run_id(); let exit_code = attach::attach_run( &run_info.path, Some(cli_settings.storage_dir().as_path()), - Some(&run_info.run_id), + Some(&run_id), false, styles, None, diff --git a/lib/crates/fabro-cli/src/commands/run/preview.rs b/lib/crates/fabro-cli/src/commands/run/preview.rs index 126b3d605..afabcdef9 100644 --- a/lib/crates/fabro-cli/src/commands/run/preview.rs +++ b/lib/crates/fabro-cli/src/commands/run/preview.rs @@ -13,7 +13,7 @@ pub(crate) async fn run(args: PreviewArgs, globals: &GlobalArgs) -> Result<()> { let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id()).await?; let record = run_store .state() .await? diff --git a/lib/crates/fabro-cli/src/commands/run/resume.rs b/lib/crates/fabro-cli/src/commands/run/resume.rs index 3990071ae..41d6b2143 100644 --- a/lib/crates/fabro-cli/src/commands/run/resume.rs +++ b/lib/crates/fabro-cli/src/commands/run/resume.rs @@ -21,10 +21,9 @@ pub(crate) async fn resume_command( let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; + let run_id = run.run_id(); let run_dir = run.path; - let run_id = run.run_id; - if launcher_pid_alive(&run_dir) { bail!("an engine process is still running for this run — cannot resume"); } diff --git a/lib/crates/fabro-cli/src/commands/run/ssh.rs b/lib/crates/fabro-cli/src/commands/run/ssh.rs index 5b5c2a916..5204feb52 100644 --- a/lib/crates/fabro-cli/src/commands/run/ssh.rs +++ b/lib/crates/fabro-cli/src/commands/run/ssh.rs @@ -17,7 +17,8 @@ pub(crate) async fn run(args: SshArgs, globals: &GlobalArgs) -> Result<()> { let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; + let run_id = run.run_id(); + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_id).await?; let record = run_store .state() .await? diff --git a/lib/crates/fabro-cli/src/commands/run/wait.rs b/lib/crates/fabro-cli/src/commands/run/wait.rs index bb05069a8..3280cf133 100644 --- a/lib/crates/fabro-cli/src/commands/run/wait.rs +++ b/lib/crates/fabro-cli/src/commands/run/wait.rs @@ -24,7 +24,8 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) let store = store::build_store(&cli_settings.storage_dir())?; let run_info = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - info!(run_id = %run_info.run_id, "Waiting for run to complete"); + let run_id = run_info.run_id(); + info!(run_id = %run_id, "Waiting for run to complete"); let deadline = args .timeout @@ -33,8 +34,7 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) let started_waiting_at = std::time::Instant::now(); let final_status = loop { - let run_store = - store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_id).await?; let status = run_store.state().await?.status.map(|record| record.status); let status = status.unwrap_or_else(|| { if started_waiting_at.elapsed() < WAIT_STARTUP_GRACE { @@ -54,7 +54,7 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) bail!( "Timed out after {}s waiting for run '{}'", args.timeout.unwrap(), - run_info.run_id + run_id ); } std::thread::sleep(interval.min(dl - now)); @@ -63,16 +63,16 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) } }; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_id).await?; let conclusion = run_store.state().await?.conclusion; if globals.json { - let json_value = build_json_output(final_status, &run_info.run_id, conclusion.as_ref()); + let json_value = build_json_output(final_status, &run_id, conclusion.as_ref()); let mut out = std::io::stdout().lock(); serde_json::to_writer_pretty(&mut out, &json_value)?; writeln!(out)?; } else { - print_human_output(final_status, &run_info.run_id, conclusion.as_ref(), styles); + print_human_output(final_status, &run_id, conclusion.as_ref(), styles); } if final_status == RunStatus::Succeeded { diff --git a/lib/crates/fabro-cli/src/commands/runs/inspect.rs b/lib/crates/fabro-cli/src/commands/runs/inspect.rs index fd4ab398d..6a26b4124 100644 --- a/lib/crates/fabro-cli/src/commands/runs/inspect.rs +++ b/lib/crates/fabro-cli/src/commands/runs/inspect.rs @@ -28,8 +28,9 @@ pub(crate) async fn run(args: &InspectArgs, globals: &GlobalArgs) -> Result<()> let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; - let output = inspect_run_store(&run.run_id, &run.path, run.status, &run_store).await; + let run_id = run.run_id(); + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_id).await?; + let output = inspect_run_store(&run_id, &run.path, run.status(), &run_store).await; let json = serde_json::to_string_pretty(&[output])?; println!("{json}"); Ok(()) diff --git a/lib/crates/fabro-cli/src/commands/runs/list.rs b/lib/crates/fabro-cli/src/commands/runs/list.rs index 2c5343186..21a52617a 100644 --- a/lib/crates/fabro-cli/src/commands/runs/list.rs +++ b/lib/crates/fabro-cli/src/commands/runs/list.rs @@ -42,13 +42,32 @@ pub(crate) async fn list_command( ); if globals.json { - println!("{}", serde_json::to_string_pretty(&filtered)?); + let json_rows: Vec<_> = filtered + .iter() + .map(|run| { + serde_json::json!({ + "run_id": run.run_id(), + "dir_name": run.dir_name, + "workflow_name": run.workflow_name(), + "workflow_slug": run.workflow_slug(), + "status": run.status(), + "status_reason": run.status_reason(), + "start_time": run.start_time(), + "labels": run.labels(), + "duration_ms": run.duration_ms(), + "total_cost": run.total_cost(), + "host_repo_path": run.host_repo_path(), + "goal": run.goal(), + }) + }) + .collect(); + println!("{}", serde_json::to_string_pretty(&json_rows)?); return Ok(()); } if args.quiet { for run in &filtered { - println!("{}", run.run_id); + println!("{}", run.run_id()); } return Ok(()); } @@ -79,7 +98,7 @@ pub(crate) async fn list_command( let rows: Vec> = display_runs .iter() .map(|run| { - let duration_display = match run.duration_ms { + let duration_display = match run.duration_ms() { Some(ms) => format_duration_ms(ms), None => match run.start_time_dt { Some(start) => { @@ -92,20 +111,19 @@ pub(crate) async fn list_command( }, }; let dir_display = run - .host_repo_path - .as_deref() + .host_repo_path() .map_or_else(|| "-".to_string(), |p| tilde_path(Path::new(p))); - let run_id = run.run_id.to_string(); + let run_id = run.run_id().to_string(); vec![ short_run_id(&run_id) .cell() .foreground_color(color_if(use_color, Color::Ansi256(8))), - run.workflow_name.clone().cell(), - status_cell(run.status, use_color), + run.workflow_name().cell(), + status_cell(run.status(), use_color), dir_display.cell(), duration_display.cell(), - truncate_goal(&run.goal, 50) + truncate_goal(&run.goal(), 50) .cell() .foreground_color(color_if(use_color, Color::Ansi256(8))), ] diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index 4571d108f..cab426cac 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -48,12 +48,12 @@ async fn remove_from( } }; - if run.status.is_active() && !args.force { - let run_id = run.run_id.to_string(); + if run.status().is_active() && !args.force { + let run_id = run.run_id().to_string(); let error = format!( "cannot remove active run {} (status: {}, use -f to force)", short_run_id(&run_id), - run.status + run.status() ); if !globals.json { eprintln!("{error}"); @@ -66,7 +66,7 @@ async fn remove_from( continue; } - let run_id = run.run_id.to_string(); + let run_id = run.run_id().to_string(); if let Err(err) = remove_run_dir_with_cleanup(store, &run).await { if !globals.json { eprintln!("error: {identifier}: {err}"); @@ -113,11 +113,12 @@ pub(crate) async fn remove_run_with_cleanup(store: &SlateStore, run: &RunInfo) - } async fn remove_run_dir_with_cleanup(store: &SlateStore, run: &RunInfo) -> Result<()> { - let run_store = match store.open_run_reader(&run.run_id).await { + let run_id = run.run_id(); + let run_store = match store.open_run_reader(&run_id).await { Ok(run_store) => Some(run_store), Err(err) => { warn!( - run_id = %run.run_id, + run_id = %run_id, error = %err, "failed to open run store during removal" ); @@ -127,13 +128,13 @@ async fn remove_run_dir_with_cleanup(store: &SlateStore, run: &RunInfo) -> Resul if let Some(run_store) = run_store.as_ref() { if let Err(err) = append_workflow_event( run_store, - &run.run_id, + &run_id, &WorkflowRunEvent::RunRemoving { reason: None }, ) .await { warn!( - run_id = %run.run_id, + run_id = %run_id, error = %err, "failed to append removing status event" ); @@ -145,11 +146,11 @@ async fn remove_run_dir_with_cleanup(store: &SlateStore, run: &RunInfo) -> Resul match reconnect_sandbox(&record).await { Ok(sandbox) => { if let Err(err) = sandbox.cleanup().await { - warn!(run_id = %run.run_id, error = %err, "sandbox cleanup failed"); + warn!(run_id = %run_id, error = %err, "sandbox cleanup failed"); } } Err(err) => { - warn!(run_id = %run.run_id, error = %err, "sandbox reconnect failed"); + warn!(run_id = %run_id, error = %err, "sandbox reconnect failed"); } } } @@ -161,9 +162,9 @@ async fn remove_run_dir_with_cleanup(store: &SlateStore, run: &RunInfo) -> Resul async fn delete_run_store_state(store: &SlateStore, run: &RunInfo) -> Result<()> { store - .delete_run(&run.run_id) + .delete_run(&run.run_id()) .await - .with_context(|| format!("failed to delete store state for {}", run.run_id)) + .with_context(|| format!("failed to delete store state for {}", run.run_id())) } async fn load_sandbox_record( diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 03fc1dcd0..363ebca66 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -20,19 +20,20 @@ pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) -> let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; + let run_id = run.run_id(); + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_id).await?; let file_count = export_run(&run_store, &args.output).await?; if globals.json { print_json_pretty(&serde_json::json!({ - "run_id": run.run_id, + "run_id": run_id, "output_dir": absolute_or_current(&args.output), "file_count": file_count, }))?; } else { println!( "Exported {file_count} files for run {} to {}", - run.run_id, + run_id, args.output.display() ); } @@ -298,7 +299,7 @@ mod tests { visit: 2, }; append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RunCreated { run_id, @@ -318,7 +319,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::WorkflowRunStarted { name: "night-sky".to_string(), @@ -333,7 +334,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RunRunning { reason: status_record.reason, @@ -343,7 +344,7 @@ mod tests { .unwrap(); for checkpoint in [&first_checkpoint, &second_checkpoint] { append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::CheckpointCompleted { node_id: checkpoint.current_node.clone(), @@ -375,7 +376,7 @@ mod tests { .unwrap(); } append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::SandboxInitialized { working_directory: sandbox.working_directory.clone(), @@ -388,7 +389,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::Prompt { stage: "code".to_string(), @@ -402,7 +403,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::PromptCompleted { node_id: "code".to_string(), @@ -415,7 +416,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::StageCompleted { node_id: "code".to_string(), @@ -446,7 +447,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::CommandStarted { node_id: "code".to_string(), @@ -458,7 +459,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::CommandCompleted { node_id: "code".to_string(), @@ -472,7 +473,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RetroStarted { prompt: Some("How did it go?".to_string()), @@ -483,7 +484,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RetroCompleted { duration_ms: 50, @@ -494,7 +495,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::WorkflowRunCompleted { duration_ms: conclusion.duration_ms, @@ -542,7 +543,7 @@ mod tests { .unwrap(); let output = tempfile::tempdir().unwrap(); - let file_count = export_run(run.as_ref(), output.path()).await.unwrap(); + let file_count = export_run(&run, output.path()).await.unwrap(); assert_eq!(file_count, 22); let exported_run: RunRecord = read_json(&output.path().join("run.json")); @@ -640,7 +641,7 @@ mod tests { let run = store.create_run(&run_id, created_at, None).await.unwrap(); let run_record = sample_run_record(run_id, created_at); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RunCreated { run_id, @@ -672,7 +673,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let output = temp.path().join("dump"); - let err = export_run(run.as_ref(), &output).await.unwrap_err(); + let err = export_run(&run, &output).await.unwrap_err(); assert!(err.to_string().contains("asset filename")); assert!(!output.exists()); } diff --git a/lib/crates/fabro-cli/src/commands/system/df.rs b/lib/crates/fabro-cli/src/commands/system/df.rs index bc63be093..be069d829 100644 --- a/lib/crates/fabro-cli/src/commands/system/df.rs +++ b/lib/crates/fabro-cli/src/commands/system/df.rs @@ -88,17 +88,17 @@ async fn df_from( for run in &runs { let size = dir_size(&run.path); total_run_size += size; - if run.status.is_active() { + if run.status().is_active() { active_count += 1; } else { reclaimable_run_size += size; } if args.verbose { run_details.push(RunSizeInfo { - run_id: run.run_id.to_string(), - workflow_name: run.workflow_name.clone(), - status: run.status, - start_time: run.start_time.clone(), + run_id: run.run_id().to_string(), + workflow_name: run.workflow_name(), + status: run.status(), + start_time: run.start_time(), start_time_dt: run.start_time_dt, size, }); @@ -122,29 +122,6 @@ async fn df_from( } } - let mut db_count = 0u64; - let mut total_db_size = 0u64; - if let Ok(entries) = std::fs::read_dir(data_dir) { - for entry in entries.flatten() { - let path = entry.path(); - if !path.is_file() { - continue; - } - let name = entry.file_name().to_string_lossy().to_string(); - if std::path::Path::new(&name) - .extension() - .is_some_and(|ext| ext.eq_ignore_ascii_case("db")) - || name.ends_with(".db-wal") - || name.ends_with(".db-shm") - { - if let Ok(meta) = path.metadata() { - db_count += 1; - total_db_size += meta.len(); - } - } - } - } - let run_reclaim_pct = if total_run_size > 0 { #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] // f64-to-integer: percentage is 0-100 @@ -172,13 +149,6 @@ async fn df_from( size_bytes: total_log_size, reclaimable_bytes: Some(total_log_size), }, - SummaryRow { - r#type: "databases".to_string(), - count: db_count, - active: None, - size_bytes: total_db_size, - reclaimable_bytes: Some(0), - }, ]; let runs = args.verbose.then(|| { run_details @@ -195,7 +165,7 @@ async fn df_from( }); print_json_pretty(&DfOutput { summary, - total_size_bytes: total_run_size + total_log_size + total_db_size, + total_size_bytes: total_run_size + total_log_size, total_reclaimable_bytes: reclaimable_run_size + total_log_size, runs, })?; @@ -235,15 +205,6 @@ async fn df_from( .cell() .justify(Justify::Right), ], - vec![ - "Databases".cell(), - db_count.cell().justify(Justify::Right), - "-".cell().justify(Justify::Right), - format_size(total_db_size).cell().justify(Justify::Right), - format!("{} (0%)", format_size(0)) - .cell() - .justify(Justify::Right), - ], ]; let summary_table = summary_rows .table() diff --git a/lib/crates/fabro-cli/src/commands/system/prune.rs b/lib/crates/fabro-cli/src/commands/system/prune.rs index be85a47e8..fca3dcfa9 100644 --- a/lib/crates/fabro-cli/src/commands/system/prune.rs +++ b/lib/crates/fabro-cli/src/commands/system/prune.rs @@ -81,7 +81,7 @@ async fn prune_from( }); } - filtered.retain(|run| !run.status.is_active()); + filtered.retain(|run| !run.status().is_active()); if filtered.is_empty() { if globals.json { @@ -108,9 +108,9 @@ async fn prune_from( let rows: Vec = filtered .iter() .map(|run| PruneRunRow { - run_id: run.run_id.to_string(), + run_id: run.run_id().to_string(), dir_name: run.dir_name.clone(), - workflow_name: run.workflow_name.clone(), + workflow_name: run.workflow_name(), size_bytes: dir_size(&run.path), }) .collect(); @@ -119,7 +119,7 @@ async fn prune_from( if args.yes { for run in &filtered { - info!(run_id = %run.run_id, path = %run.path.display(), "deleting run"); + info!(run_id = %run.run_id(), path = %run.path.display(), "deleting run"); remove_run_with_cleanup(store, run).await?; } if globals.json { @@ -149,8 +149,8 @@ async fn prune_from( } for run in &filtered { - debug!(run_id = %run.run_id, "would delete run (dry-run)"); - println!("would delete: {} ({})", run.dir_name, run.workflow_name); + debug!(run_id = %run.run_id(), "would delete run (dry-run)"); + println!("would delete: {} ({})", run.dir_name, run.workflow_name()); } eprintln!( "\n{} run(s) would be deleted ({} freed). Pass --yes to confirm.", diff --git a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs index de58d14b7..5facf751c 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs @@ -191,7 +191,7 @@ fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() { "rewind should clear pull request" ); assert!( - state.nodes.is_empty(), + state.is_empty(), "rewind should clear node state that belonged to the prior execution" ); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs index 67d013274..f53fd4fbf 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_status.rs b/lib/crates/fabro-cli/tests/it/cmd/server_status.rs index 13039a893..1a9785060 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_status.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_status.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs b/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs index 49a6f5c65..42cabafaf 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 9d84ff0b6..433db3a6d 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -784,15 +784,13 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git let state = run_state(&run.run_dir); assert!( state - .nodes - .iter() - .any(|((node_id, _), node)| node_id == "step_one" && node.diff.is_some()) + .iter_nodes() + .any(|(node, state)| node.node_id == "step_one" && state.diff.is_some()) ); assert!( state - .nodes - .iter() - .any(|((node_id, _), node)| node_id == "step_two" && node.diff.is_some()) + .iter_nodes() + .any(|(node, state)| node.node_id == "step_two" && state.diff.is_some()) ); } GitWorkflowKind::Noop => { diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs index 43a75fa5f..2e4d4fa83 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs @@ -30,13 +30,11 @@ fn help() { } #[test] -fn system_df_summarizes_runs_logs_and_databases() { +fn system_df_summarizes_runs_and_logs() { let context = test_context!(); setup_completed_dry_run(&context); std::fs::create_dir_all(context.storage_dir.join("logs")).unwrap(); std::fs::write(context.storage_dir.join("logs/cli.log"), b"log line\n").unwrap(); - std::fs::write(context.storage_dir.join("fabro.db"), b"db").unwrap(); - std::fs::write(context.storage_dir.join("fabro.db-wal"), b"wal").unwrap(); let mut filters = context.filters(); filters.push(( @@ -50,10 +48,9 @@ fn system_df_summarizes_runs_logs_and_databases() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (0%) - Logs 1 - [SIZE] [SIZE] (100%) - Databases 2 - [SIZE] [SIZE] (0%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (0%) + Logs 1 - [SIZE] [SIZE] (100%) Data directory: [STORAGE_DIR] ----- stderr ----- @@ -82,10 +79,9 @@ fn system_df_verbose_lists_runs_with_reclaimable_marker() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (0%) - Logs 0 - [SIZE] [SIZE] (0%) - Databases 0 - [SIZE] [SIZE] (0%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (0%) + Logs 0 - [SIZE] [SIZE] (0%) Data directory: [STORAGE_DIR] diff --git a/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs index 83677825e..051c4e024 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs b/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs index 2db5868a7..f00d9a377 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs @@ -32,7 +32,6 @@ async fn run_real_cli_test(provider: Provider, model: &str) { &context, None, &emitter, - workspace.path(), &env, None, ) diff --git a/lib/crates/fabro-db/Cargo.toml b/lib/crates/fabro-db/Cargo.toml deleted file mode 100644 index 3c8c124fc..000000000 --- a/lib/crates/fabro-db/Cargo.toml +++ /dev/null @@ -1,22 +0,0 @@ -[package] -name = "fabro-db" -edition.workspace = true -version.workspace = true -publish = false -license.workspace = true -description = "SQLite persistence layer for Fabro" - -[lib] -doctest = false - -[lints] -workspace = true - -[dependencies] -sqlx = { workspace = true, features = ["sqlite"] } -chrono.workspace = true -thiserror.workspace = true -tracing.workspace = true - -[dev-dependencies] -tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/lib/crates/fabro-db/migrations/001_create_workflow_runs.sql b/lib/crates/fabro-db/migrations/001_create_workflow_runs.sql deleted file mode 100644 index 5cef6005d..000000000 --- a/lib/crates/fabro-db/migrations/001_create_workflow_runs.sql +++ /dev/null @@ -1,8 +0,0 @@ -CREATE TABLE workflow_runs ( - id TEXT PRIMARY KEY NOT NULL, - title TEXT NOT NULL DEFAULT '', - logs_dir TEXT NOT NULL, - work_dir TEXT NOT NULL, - created_at TEXT NOT NULL DEFAULT (datetime('now')), - updated_at TEXT NOT NULL DEFAULT (datetime('now')) -); diff --git a/lib/crates/fabro-db/migrations/002_rename_logs_dir_to_run_dir.sql b/lib/crates/fabro-db/migrations/002_rename_logs_dir_to_run_dir.sql deleted file mode 100644 index 29c7c7e03..000000000 --- a/lib/crates/fabro-db/migrations/002_rename_logs_dir_to_run_dir.sql +++ /dev/null @@ -1 +0,0 @@ -ALTER TABLE workflow_runs RENAME COLUMN logs_dir TO run_dir; diff --git a/lib/crates/fabro-db/src/lib.rs b/lib/crates/fabro-db/src/lib.rs deleted file mode 100644 index e796670b3..000000000 --- a/lib/crates/fabro-db/src/lib.rs +++ /dev/null @@ -1,124 +0,0 @@ -mod migrate; -pub mod workflow_run; - -use std::path::Path; -use std::time::Duration; - -use sqlx::SqlitePool; -use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions}; -use tracing::debug; - -pub use migrate::initialize_db; -pub use workflow_run::WorkflowRun; - -/// Connect to a SQLite database at the given path, creating it if it doesn't exist. -pub async fn connect(path: &Path) -> Result { - debug!(path = %path.display(), "Connecting to SQLite database"); - let options = SqliteConnectOptions::new() - .filename(path) - .create_if_missing(true) - .journal_mode(SqliteJournalMode::Wal) - .busy_timeout(Duration::from_secs(5)) - .foreign_keys(true); - - SqlitePoolOptions::new().connect_with(options).await -} - -/// Connect to an in-memory SQLite database (for tests). -pub async fn connect_memory() -> Result { - let options = SqliteConnectOptions::new() - .filename(":memory:") - .journal_mode(SqliteJournalMode::Wal) - .busy_timeout(Duration::from_secs(5)) - .foreign_keys(true); - - SqlitePoolOptions::new() - .max_connections(1) - .connect_with(options) - .await -} - -#[cfg(test)] -mod tests { - use super::*; - use chrono::Utc; - - #[tokio::test] - async fn connect_memory_returns_working_pool() { - let pool = connect_memory().await.unwrap(); - let row: (i64,) = sqlx::query_as("SELECT 1").fetch_one(&pool).await.unwrap(); - assert_eq!(row.0, 1); - } - - #[tokio::test] - async fn initialize_db_creates_workflow_runs_table() { - let pool = connect_memory().await.unwrap(); - initialize_db(&pool).await.unwrap(); - - let row: (String,) = sqlx::query_as( - "SELECT name FROM sqlite_master WHERE type='table' AND name='workflow_runs'", - ) - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(row.0, "workflow_runs"); - } - - #[tokio::test] - async fn initialize_db_sets_user_version() { - let pool = connect_memory().await.unwrap(); - initialize_db(&pool).await.unwrap(); - - let row: (i64,) = sqlx::query_as("PRAGMA user_version") - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(row.0, 2); - } - - #[tokio::test] - async fn initialize_db_is_idempotent() { - let pool = connect_memory().await.unwrap(); - initialize_db(&pool).await.unwrap(); - initialize_db(&pool).await.unwrap(); - - let row: (i64,) = sqlx::query_as("PRAGMA user_version") - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(row.0, 2); - } - - #[tokio::test] - async fn workflow_run_round_trips_through_sql() { - let pool = connect_memory().await.unwrap(); - initialize_db(&pool).await.unwrap(); - - let now = Utc::now(); - let now_str = now.format("%Y-%m-%d %H:%M:%S").to_string(); - - sqlx::query( - "INSERT INTO workflow_runs (id, title, run_dir, work_dir, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)", - ) - .bind("run-1") - .bind("My Run") - .bind("/tmp/logs") - .bind("/tmp/work") - .bind(&now_str) - .bind(&now_str) - .execute(&pool) - .await - .unwrap(); - - let run: WorkflowRun = sqlx::query_as("SELECT * FROM workflow_runs WHERE id = ?") - .bind("run-1") - .fetch_one(&pool) - .await - .unwrap(); - - assert_eq!(run.id, "run-1"); - assert_eq!(run.title, "My Run"); - assert_eq!(run.run_dir, "/tmp/logs"); - assert_eq!(run.work_dir, "/tmp/work"); - } -} diff --git a/lib/crates/fabro-db/src/migrate.rs b/lib/crates/fabro-db/src/migrate.rs deleted file mode 100644 index d7420e326..000000000 --- a/lib/crates/fabro-db/src/migrate.rs +++ /dev/null @@ -1,48 +0,0 @@ -use sqlx::SqlitePool; -use tracing::{debug, info}; - -const CURRENT_VERSION: i64 = 2; - -const MIGRATION_001: &str = include_str!("../migrations/001_create_workflow_runs.sql"); -const MIGRATION_002: &str = include_str!("../migrations/002_rename_logs_dir_to_run_dir.sql"); - -/// Apply all pending migrations to the database. -/// -/// Uses `PRAGMA user_version` to track which migrations have been applied. -pub async fn initialize_db(pool: &SqlitePool) -> Result<(), sqlx::Error> { - let row: (i64,) = sqlx::query_as("PRAGMA user_version") - .fetch_one(pool) - .await?; - let from_version = row.0; - - if from_version < CURRENT_VERSION { - info!( - from_version = from_version, - to_version = CURRENT_VERSION, - "Running database migrations" - ); - let mut tx = pool.begin().await?; - - if from_version < 1 { - sqlx::query(MIGRATION_001).execute(&mut *tx).await?; - } - - if from_version < 2 { - sqlx::query(MIGRATION_002).execute(&mut *tx).await?; - } - - sqlx::query(&format!("PRAGMA user_version = {CURRENT_VERSION}")) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - info!(version = CURRENT_VERSION, "Database migrations complete"); - } else { - debug!( - version = from_version, - "Database already at current version" - ); - } - - Ok(()) -} diff --git a/lib/crates/fabro-db/src/workflow_run.rs b/lib/crates/fabro-db/src/workflow_run.rs deleted file mode 100644 index 712ed9497..000000000 --- a/lib/crates/fabro-db/src/workflow_run.rs +++ /dev/null @@ -1,11 +0,0 @@ -use chrono::{DateTime, Utc}; - -#[derive(Debug, sqlx::FromRow)] -pub struct WorkflowRun { - pub id: String, - pub title: String, - pub run_dir: String, - pub work_dir: String, - pub created_at: DateTime, - pub updated_at: DateTime, -} diff --git a/lib/crates/fabro-server/Cargo.toml b/lib/crates/fabro-server/Cargo.toml index ab82689f1..b47305cae 100644 --- a/lib/crates/fabro-server/Cargo.toml +++ b/lib/crates/fabro-server/Cargo.toml @@ -26,7 +26,6 @@ fabro-model = { path = "../fabro-model" } fabro-retro = { path = "../fabro-retro" } fabro-types = { path = "../fabro-types" } fabro-util = { path = "../fabro-util" } -fabro-db = { path = "../fabro-db" } fabro-api = { path = "../fabro-api" } fabro-store = { path = "../fabro-store" } chrono.workspace = true @@ -35,7 +34,6 @@ axum.workspace = true axum-extra.workspace = true cookie.workspace = true dirs.workspace = true -sqlx.workspace = true tower = "0.5" tokio-stream = { workspace = true, features = ["sync"] } base64.workspace = true diff --git a/lib/crates/fabro-server/src/serve.rs b/lib/crates/fabro-server/src/serve.rs index 60e8f987f..0c53e596c 100644 --- a/lib/crates/fabro-server/src/serve.rs +++ b/lib/crates/fabro-server/src/serve.rs @@ -103,7 +103,7 @@ pub async fn serve_command( } }; - // Initialize data directory and SQLite database + // Initialize data directory and storage let config_path = args.config.clone(); let disk_settings = load_server_settings(config_path.as_deref())?; let data_dir = storage_dir_override.unwrap_or_else(|| resolve_storage_dir(&disk_settings)); @@ -115,9 +115,6 @@ pub async fn serve_command( dry_run_mode, ))); std::fs::create_dir_all(&data_dir)?; - let db = fabro_db::connect(&data_dir.join("fabro.db")).await?; - fabro_db::initialize_db(&db).await?; - let (auth_mode, client_auth, max_concurrent_runs) = { let cfg = shared_settings.read().expect("config lock poisoned"); let api = cfg.api.clone().unwrap_or_default(); @@ -144,7 +141,7 @@ pub async fn serve_command( Duration::from_millis(1), )); let state = - create_app_state_with_store(db, Arc::clone(&shared_settings), max_concurrent_runs, store); + create_app_state_with_store(Arc::clone(&shared_settings), max_concurrent_runs, store); spawn_scheduler(Arc::clone(&state)); let router = build_router(state, auth_mode); diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 50a6f659d..4fecd54b1 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -129,7 +129,6 @@ pub struct AppState { runs: Mutex>, aggregate_usage: Mutex, store: StoreHandle, - pub db: sqlx::SqlitePool, max_concurrent_runs: usize, scheduler_notify: Notify, pub sessions: SessionStore, @@ -406,18 +405,16 @@ async fn get_aggregate_usage( (StatusCode::OK, Json(response)).into_response() } -/// Create an `AppState` with the given LLM spec factory and database pool. -pub fn create_app_state(db: sqlx::SqlitePool) -> Arc { - create_app_state_with_options(db, Settings::default(), 5) +/// Create an `AppState` with default settings. +pub fn create_app_state() -> Arc { + create_app_state_with_options(Settings::default(), 5) } #[doc(hidden)] pub fn create_app_state_with_registry_factory( - db: sqlx::SqlitePool, registry_factory_override: impl Fn(Arc) -> HandlerRegistry + Send + Sync + 'static, ) -> Arc { build_app_state( - db, Arc::new(RwLock::new(Settings::default())), Some(Box::new(registry_factory_override)), 5, @@ -425,14 +422,12 @@ pub fn create_app_state_with_registry_factory( ) } -/// Create an `AppState` with the given database pool, settings, and concurrency limit. +/// Create an `AppState` with the given settings and concurrency limit. pub fn create_app_state_with_options( - db: sqlx::SqlitePool, settings: Settings, max_concurrent_runs: usize, ) -> Arc { create_app_state_with_store( - db, Arc::new(RwLock::new(settings)), max_concurrent_runs, test_store(), @@ -448,16 +443,14 @@ fn test_store() -> StoreHandle { } pub fn create_app_state_with_store( - db: sqlx::SqlitePool, settings: Arc>, max_concurrent_runs: usize, store: StoreHandle, ) -> Arc { - build_app_state(db, settings, None, max_concurrent_runs, store) + build_app_state(settings, None, max_concurrent_runs, store) } fn build_app_state( - db: sqlx::SqlitePool, settings: Arc>, registry_factory_override: Option>, max_concurrent_runs: usize, @@ -467,7 +460,6 @@ fn build_app_state( runs: Mutex::new(HashMap::new()), aggregate_usage: Mutex::new(UsageAccumulator::default()), store, - db, max_concurrent_runs, scheduler_notify: Notify::new(), sessions: new_session_store(), @@ -1679,14 +1671,8 @@ mod tests { ) } - async fn test_db() -> sqlx::SqlitePool { - let pool = fabro_db::connect_memory().await.unwrap(); - fabro_db::initialize_db(&pool).await.unwrap(); - pool - } - - fn test_app_with(db: sqlx::SqlitePool) -> Router { - let state = create_app_state(db); + fn test_app_with() -> Router { + let state = create_app_state(); build_router(state, AuthMode::Disabled) } @@ -1706,7 +1692,7 @@ mod tests { #[tokio::test] async fn test_model_unknown_returns_404() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let req = Request::builder() .method("POST") @@ -1721,7 +1707,7 @@ mod tests { #[tokio::test] async fn test_model_known_returns_200_with_status() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let req = Request::builder() .method("POST") @@ -1757,7 +1743,7 @@ mod tests { webhooks: None, }); let app = build_router( - create_app_state_with_options(test_db().await, settings, 5), + create_app_state_with_options(settings, 5), AuthMode::Disabled, ); @@ -1782,7 +1768,7 @@ mod tests { #[tokio::test] async fn logout_redirects_to_login_page() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let response = app .oneshot( @@ -1807,7 +1793,7 @@ mod tests { #[tokio::test] async fn static_favicon_is_served() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let response = app .oneshot( @@ -1831,7 +1817,7 @@ mod tests { #[tokio::test] async fn test_model_dry_run_returns_ok() { - let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5); + let state = create_app_state_with_options(dry_run_settings(), 5); let app = build_router(state, AuthMode::Disabled); let req = Request::builder() @@ -1851,7 +1837,7 @@ mod tests { #[tokio::test] async fn test_model_dry_run_unknown_returns_404() { - let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5); + let state = create_app_state_with_options(dry_run_settings(), 5); let app = build_router(state, AuthMode::Disabled); let req = Request::builder() @@ -1867,7 +1853,7 @@ mod tests { #[tokio::test] async fn post_runs_starts_run_and_returns_id() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let req = Request::builder() .method("POST") @@ -1888,7 +1874,7 @@ mod tests { #[tokio::test] async fn post_runs_invalid_dot_returns_bad_request() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let req = Request::builder() .method("POST") @@ -1905,7 +1891,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn get_run_status_returns_status() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = test_app_with_scheduler(state); // Start a run @@ -1949,7 +1935,7 @@ mod tests { #[tokio::test] async fn get_run_status_not_found() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let missing_run_id = fixtures::RUN_64; let req = Request::builder() @@ -1964,7 +1950,7 @@ mod tests { #[tokio::test] async fn get_questions_returns_empty_list() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); // Start a run @@ -1998,7 +1984,7 @@ mod tests { #[tokio::test] async fn submit_answer_not_found_run() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let missing_run_id = fixtures::RUN_64; let req = Request::builder() @@ -2016,7 +2002,7 @@ mod tests { #[tokio::test] async fn get_events_not_found() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let missing_run_id = fixtures::RUN_64; let req = Request::builder() @@ -2031,7 +2017,7 @@ mod tests { #[tokio::test] async fn get_checkpoint_returns_null_initially() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); // Start a run @@ -2061,7 +2047,7 @@ mod tests { #[tokio::test] async fn get_context_returns_map() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); // Start a run @@ -2094,7 +2080,7 @@ mod tests { #[tokio::test] async fn cancel_run_succeeds() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); // Start a run @@ -2129,7 +2115,7 @@ mod tests { #[tokio::test] async fn cancel_nonexistent_run_returns_not_found() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let missing_run_id = fixtures::RUN_64; let req = Request::builder() @@ -2144,7 +2130,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn get_events_returns_sse_stream() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = test_app_with_scheduler(state); // Start a run @@ -2195,7 +2181,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn run_completes_and_status_is_completed() { - let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5); + let state = create_app_state_with_options(dry_run_settings(), 5); let app = test_app_with_scheduler(state); // Start a run @@ -2234,7 +2220,7 @@ mod tests { #[tokio::test] async fn get_graph_returns_svg() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); // Start a run @@ -2286,7 +2272,7 @@ mod tests { #[tokio::test] async fn get_graph_not_found() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let missing_run_id = fixtures::RUN_64; let req = Request::builder() @@ -2301,7 +2287,7 @@ mod tests { #[tokio::test] async fn list_runs_returns_started_run() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); // List should be empty initially @@ -2350,7 +2336,7 @@ mod tests { #[tokio::test] async fn get_aggregate_usage_returns_zeros_initially() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(Arc::clone(&state), AuthMode::Disabled); let req = Request::builder() @@ -2373,7 +2359,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn aggregate_usage_increments_after_run_completes() { - let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5); + let state = create_app_state_with_options(dry_run_settings(), 5); let app = test_app_with_scheduler(state); // Start a run @@ -2424,7 +2410,7 @@ mod tests { #[tokio::test] async fn post_runs_returns_queued_status() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(state, AuthMode::Disabled); let req = Request::builder() @@ -2497,7 +2483,7 @@ mod tests { }), ..Default::default() }; - let state = create_app_state_with_options(test_db().await, settings.clone(), 5); + let state = create_app_state_with_options(settings.clone(), 5); let app = build_router(Arc::clone(&state), AuthMode::Disabled); let req = Request::builder() @@ -2542,7 +2528,7 @@ mod tests { let output_path = output_dir.path().join("executed.txt"); let dot = command_dot(&format!("printf snapshot > {}", output_path.display())); let initial_settings = dry_run_settings(); - let state = create_app_state_with_options(test_db().await, initial_settings.clone(), 5); + let state = create_app_state_with_options(initial_settings.clone(), 5); let app = build_router(Arc::clone(&state), AuthMode::Disabled); let req = Request::builder() @@ -2575,7 +2561,7 @@ mod tests { #[tokio::test] async fn cancel_queued_run_succeeds() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(state, AuthMode::Disabled); // Submit a run (no scheduler, stays queued) @@ -2623,7 +2609,7 @@ mod tests { }), ..Default::default() }; - let state = create_app_state_with_options(test_db().await, settings, 5); + let state = create_app_state_with_options(settings, 5); let app = build_router(Arc::clone(&state), AuthMode::Disabled); let req = Request::builder() @@ -2688,7 +2674,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cancel_before_run_transitions_to_running_closes_event_stream() { - let state = create_app_state_with_registry_factory(test_db().await, |interviewer| { + let state = create_app_state_with_registry_factory(|interviewer| { std::thread::sleep(std::time::Duration::from_millis(200)); fabro_workflow::handler::default_registry(interviewer, || None) }); @@ -2731,7 +2717,7 @@ mod tests { #[tokio::test] async fn queue_position_reported_for_queued_runs() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(state, AuthMode::Disabled); // Submit two runs (no scheduler, both stay queued) @@ -2773,7 +2759,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn concurrency_limit_respected() { - let state = create_app_state_with_options(test_db().await, Settings::default(), 1); + let state = create_app_state_with_options(Settings::default(), 1); let app = test_app_with_scheduler(state); // Submit two runs with max_concurrent_runs=1 @@ -2822,7 +2808,7 @@ mod tests { #[tokio::test] async fn submit_answer_to_queued_run_returns_conflict() { - let state = create_app_state(test_db().await); + let state = create_app_state(); let app = build_router(state, AuthMode::Disabled); let req = Request::builder() @@ -2854,7 +2840,7 @@ mod tests { #[tokio::test] async fn create_completion_non_streaming_returns_json() { - let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5); + let state = create_app_state_with_options(dry_run_settings(), 5); let app = build_router(state, AuthMode::Disabled); let req = Request::builder() @@ -2884,7 +2870,7 @@ mod tests { #[tokio::test] async fn create_completion_streaming_returns_sse() { - let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5); + let state = create_app_state_with_options(dry_run_settings(), 5); let app = build_router(state, AuthMode::Disabled); let req = Request::builder() @@ -2915,7 +2901,7 @@ mod tests { #[tokio::test] async fn create_completion_missing_messages_returns_422() { - let app = test_app_with(test_db().await); + let app = test_app_with(); let req = Request::builder() .method("POST") diff --git a/lib/crates/fabro-server/src/sessions.rs b/lib/crates/fabro-server/src/sessions.rs index beb3bba62..481220992 100644 --- a/lib/crates/fabro-server/src/sessions.rs +++ b/lib/crates/fabro-server/src/sessions.rs @@ -432,16 +432,8 @@ mod tests { use crate::jwt_auth::AuthMode; use crate::server::{build_router, create_app_state_with_options}; - async fn test_db() -> sqlx::SqlitePool { - let pool = fabro_db::connect_memory().await.unwrap(); - fabro_db::initialize_db(&pool).await.unwrap(); - pool - } - async fn dry_run_app() -> axum::Router { - let db = test_db().await; let state = create_app_state_with_options( - db, fabro_types::Settings { dry_run: Some(true), ..Default::default() diff --git a/lib/crates/fabro-server/tests/it/api.rs b/lib/crates/fabro-server/tests/it/api.rs index 88850eff8..53403ba0c 100644 --- a/lib/crates/fabro-server/tests/it/api.rs +++ b/lib/crates/fabro-server/tests/it/api.rs @@ -22,12 +22,6 @@ mod mtls_e2e { use fabro_server::tls::{ClientAuth, build_rustls_config}; use tokio::net::TcpListener; - async fn test_db() -> sqlx::SqlitePool { - let pool = fabro_db::connect_memory().await.unwrap(); - fabro_db::initialize_db(&pool).await.unwrap(); - pool - } - /// Generate a complete CA + server cert + client cert PKI in `dir`. /// Returns paths: (ca_cert, server_cert, server_key, client_cert_pem, client_key_pem) fn generate_pki(dir: &Path, ca_cn: &str, server_cn: &str, client_cn: &str) -> PkiPaths { @@ -188,7 +182,7 @@ mod mtls_e2e { let rustls_config = build_rustls_config(tls_settings, client_auth); let tls_acceptor = tokio_rustls::TlsAcceptor::from(rustls_config); - let state = create_app_state(test_db().await); + let state = create_app_state(); let router = build_router(state, auth_mode); tokio::spawn(async move { @@ -419,7 +413,6 @@ mod mtls_e2e { // =========================================================================== mod server_lifecycle { - use super::super::helpers::test_db; use super::api; use std::sync::Arc; use std::time::Duration; @@ -531,7 +524,7 @@ mod server_lifecycle { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn full_http_lifecycle_approve_and_complete() { - let state = create_app_state_with_registry_factory(test_db().await, gate_registry); + let state = create_app_state_with_registry_factory(gate_registry); fabro_server::server::spawn_scheduler(Arc::clone(&state)); let app = build_router( Arc::clone(&state), @@ -601,7 +594,7 @@ mod server_lifecycle { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn full_http_lifecycle_cancel() { - let state = create_app_state_with_registry_factory(test_db().await, gate_registry); + let state = create_app_state_with_registry_factory(gate_registry); fabro_server::server::spawn_scheduler(Arc::clone(&state)); let app = build_router( Arc::clone(&state), @@ -654,7 +647,6 @@ mod server_lifecycle { // =========================================================================== mod sse_events { - use super::super::helpers::test_db; use super::api; use std::sync::Arc; use std::time::Duration; @@ -734,7 +726,7 @@ mod sse_events { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn sse_stream_contains_expected_event_types() { - let state = create_app_state_with_options(test_db().await, dry_run_settings(), 5); + let state = create_app_state_with_options(dry_run_settings(), 5); fabro_server::server::spawn_scheduler(Arc::clone(&state)); let app = build_router( Arc::clone(&state), @@ -837,7 +829,6 @@ mod sse_events { // =========================================================================== mod serve_dry_run { - use super::super::helpers::test_db; use super::api; use std::sync::Arc; use std::time::Duration; @@ -857,7 +848,6 @@ mod serve_dry_run { /// Build the router exactly as `serve_command` does in dry-run mode. async fn dry_run_app() -> axum::Router { let state = create_app_state_with_options( - test_db().await, fabro_types::Settings { dry_run: Some(true), ..Default::default() @@ -975,7 +965,6 @@ mod serve_dry_run { } mod route_prefixes { - use super::super::helpers::test_db; use axum::body::Body; use axum::http::{Method, Request, StatusCode}; use fabro_server::server::{build_router, create_app_state}; @@ -989,7 +978,7 @@ mod route_prefixes { #[tokio::test] async fn old_unversioned_routes_return_404() { let app = build_router( - create_app_state(test_db().await), + create_app_state(), fabro_server::jwt_auth::AuthMode::Disabled, ); @@ -1009,7 +998,7 @@ mod route_prefixes { #[tokio::test] async fn root_and_health_stay_at_root() { let app = build_router( - create_app_state(test_db().await), + create_app_state(), fabro_server::jwt_auth::AuthMode::Disabled, ); @@ -1040,7 +1029,7 @@ mod route_prefixes { #[tokio::test] async fn moved_routes_not_at_root_of_api_prefix() { let app = build_router( - create_app_state(test_db().await), + create_app_state(), fabro_server::jwt_auth::AuthMode::Disabled, ); diff --git a/lib/crates/fabro-server/tests/it/helpers.rs b/lib/crates/fabro-server/tests/it/helpers.rs index 4d4f72dd0..5762a188b 100644 --- a/lib/crates/fabro-server/tests/it/helpers.rs +++ b/lib/crates/fabro-server/tests/it/helpers.rs @@ -1,5 +1,3 @@ -pub(crate) async fn test_db() -> sqlx::SqlitePool { - let pool = fabro_db::connect_memory().await.unwrap(); - fabro_db::initialize_db(&pool).await.unwrap(); - pool +pub(crate) fn test_app_state() -> std::sync::Arc { + fabro_server::server::create_app_state() } diff --git a/lib/crates/fabro-server/tests/it/openapi_conformance.rs b/lib/crates/fabro-server/tests/it/openapi_conformance.rs index cb904fe1e..0de4c733b 100644 --- a/lib/crates/fabro-server/tests/it/openapi_conformance.rs +++ b/lib/crates/fabro-server/tests/it/openapi_conformance.rs @@ -7,7 +7,7 @@ clippy::manual_let_else )] -use super::helpers::test_db; +use super::helpers::test_app_state; use std::collections::BTreeSet; use axum::body::Body; @@ -17,7 +17,7 @@ use fabro_config::sandbox::SandboxSettings; use fabro_hooks::*; use fabro_sandbox::daytona::*; use fabro_server::jwt_auth::AuthMode; -use fabro_server::server::{build_router, create_app_state}; +use fabro_server::server::build_router; use fabro_server::server_config::*; use tower::ServiceExt; @@ -65,7 +65,7 @@ fn methods_for_path_item(item: &openapiv3::PathItem) -> Vec { #[tokio::test] async fn all_spec_routes_are_routable() { let spec = load_spec(); - let state = create_app_state(test_db().await); + let state = test_app_state(); let app = build_router(state, AuthMode::Disabled); let mut checked = 0; diff --git a/lib/crates/fabro-server/tests/it/pagination.rs b/lib/crates/fabro-server/tests/it/pagination.rs index c2916db5d..cf549a675 100644 --- a/lib/crates/fabro-server/tests/it/pagination.rs +++ b/lib/crates/fabro-server/tests/it/pagination.rs @@ -2,11 +2,11 @@ #![allow(clippy::absolute_paths)] -use super::helpers::test_db; +use super::helpers::test_app_state; use axum::body::Body; use axum::http::{Request, StatusCode}; use fabro_server::jwt_auth::AuthMode; -use fabro_server::server::{build_router, create_app_state}; +use fabro_server::server::build_router; use tower::ServiceExt; async fn get_json(app: axum::Router, uri: &str) -> serde_json::Value { @@ -93,7 +93,7 @@ const ENDPOINTS: &[PaginatedEndpoint] = &[ #[tokio::test] async fn paginated_endpoints_return_correct_shape() { - let state = create_app_state(test_db().await); + let state = test_app_state(); let app = build_router(state, AuthMode::Disabled); for ep in ENDPOINTS { diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index b6f49dbce..7cc889c19 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -12,13 +12,13 @@ mod types; pub use error::{Result, StoreError}; pub use run_state::{NodeState, RunProjection}; pub use runtime::RuntimeState; -pub use slate::{SlateRunStore, SlateStore}; -pub use types::{CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, RunSummary}; +pub use slate::{NodeAsset, SlateRunStore, SlateStore}; +pub use types::{CatalogRecord, EventEnvelope, EventPayload, NodeVisit, NodeVisitRef, RunSummary}; pub type StoreHandle = Arc; #[derive(Debug, Default, Clone, PartialEq, Eq)] -pub struct ListRunsQuery { +pub(crate) struct ListRunsQuery { pub start: Option>, pub end: Option>, } diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 2130e6626..6888f2e00 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -6,7 +6,9 @@ use chrono::{DateTime, Utc}; use serde::de::DeserializeOwned; use serde_json::Value; -use crate::{CatalogRecord, EventEnvelope, NodeVisitRef, Result, RunSummary, StoreError}; +use crate::{ + CatalogRecord, EventEnvelope, NodeVisit, NodeVisitRef, Result, RunSummary, StoreError, +}; use fabro_types::{ Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro, RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StageUsage, @@ -28,8 +30,7 @@ pub struct RunProjection { pub sandbox: Option, pub final_patch: Option, pub pull_request: Option, - pub nodes: HashMap<(String, u32), NodeState>, - pub last_git_sha: Option, + nodes: HashMap, } #[derive(Debug, Clone, Default)] @@ -37,7 +38,6 @@ pub struct NodeState { pub prompt: Option, pub response: Option, pub status: Option, - pub outcome: Option>>, pub provider_used: Option, pub diff: Option, pub script_invocation: Option, @@ -123,26 +123,16 @@ impl RunProjection { self.status = Some(run_status_record(RunStatus::Succeeded, &properties, ts)?); self.conclusion = Some(conclusion_from_completed(&properties, ts)?); self.final_patch = optional_string(&properties, "final_patch"); - self.last_git_sha = optional_string(&properties, "final_git_commit_sha") - .or_else(|| self.last_git_sha.clone()); } "run.failed" => { self.status = Some(run_status_record(RunStatus::Failed, &properties, ts)?); self.conclusion = Some(conclusion_from_failed(&properties, ts)); - self.last_git_sha = optional_string(&properties, "git_commit_sha") - .or_else(|| self.last_git_sha.clone()); } "run.rewound" => { self.reset_for_rewind(); - self.last_git_sha = optional_string(&properties, "run_commit_sha") - .or_else(|| self.last_git_sha.clone()); } "checkpoint.completed" => { let checkpoint = checkpoint_from_properties(&properties, ts)?; - self.last_git_sha = checkpoint - .git_commit_sha - .clone() - .or_else(|| self.last_git_sha.clone()); if let Some(node_id) = value.get("node_id").and_then(Value::as_str) { let visit = checkpoint .node_visits @@ -210,7 +200,6 @@ impl RunProjection { let node = self.node_mut(node_id, visit); node.response = response; node.status = Some(status); - node.outcome = Some(outcome); } "stage.failed" => { let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { @@ -226,18 +215,6 @@ impl RunProjection { failure_reason: failure_reason.clone(), timestamp: ts, }); - node.outcome = Some(Outcome { - status: StageStatus::Fail, - preferred_label: None, - suggested_next_ids: Vec::new(), - context_updates: HashMap::new(), - jump_to_node: None, - notes: None, - failure, - usage: None, - files_touched: Vec::new(), - duration_ms: None, - }); } "agent.session.started" | "agent.cli.started" => { let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { @@ -279,15 +256,29 @@ impl RunProjection { } pub fn node(&self, node: &NodeVisitRef<'_>) -> Option<&NodeState> { - self.nodes.get(&(node.node_id.to_string(), node.visit)) + self.nodes.get(&node.into_owned()) + } + + pub fn iter_nodes(&self) -> impl Iterator, &NodeState)> { + self.nodes + .iter() + .map(|(node, state)| (node.as_ref(), state)) + } + + pub fn is_empty(&self) -> bool { + self.nodes.is_empty() + } + + pub fn set_node(&mut self, node: NodeVisitRef<'_>, state: NodeState) { + self.nodes.insert(node.into_owned(), state); } pub fn list_node_visits(&self, node_id: &str) -> Vec { let mut visits = self .nodes .keys() - .filter(|(current_node_id, _)| current_node_id == node_id) - .map(|(_, visit)| *visit) + .filter(|node| node.node_id == node_id) + .map(|node| node.visit) .collect::>(); visits.sort_unstable(); visits.dedup(); @@ -332,14 +323,19 @@ impl RunProjection { } fn node_mut(&mut self, node_id: &str, visit: u32) -> &mut NodeState { - self.nodes.entry((node_id.to_string(), visit)).or_default() + self.nodes + .entry(NodeVisit { + node_id: node_id.to_string(), + visit, + }) + .or_default() } fn current_visit_for(&self, node_id: &str) -> Option { self.nodes .keys() - .filter(|(current_node_id, _)| current_node_id == node_id) - .map(|(_, visit)| *visit) + .filter(|node| node.node_id == node_id) + .map(|node| node.visit) .max() } diff --git a/lib/crates/fabro-store/src/runtime.rs b/lib/crates/fabro-store/src/runtime.rs index c72c0b8f8..28efd7f32 100644 --- a/lib/crates/fabro-store/src/runtime.rs +++ b/lib/crates/fabro-store/src/runtime.rs @@ -13,11 +13,6 @@ impl RuntimeState { } } - #[must_use] - pub fn root(&self) -> &Path { - &self.root - } - #[must_use] pub fn runtime_dir(&self) -> PathBuf { self.root.join("runtime") @@ -71,7 +66,6 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let state = RuntimeState::new(dir.path()); - assert_eq!(state.root(), dir.path()); assert_eq!(state.runtime_dir(), dir.path().join("runtime")); assert_eq!( state.interview_request_path(), diff --git a/lib/crates/fabro-store/src/slate/catalog.rs b/lib/crates/fabro-store/src/slate/catalog.rs index 69d1adaed..4d873512a 100644 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ b/lib/crates/fabro-store/src/slate/catalog.rs @@ -4,10 +4,16 @@ use chrono::{DateTime, Utc}; use futures::TryStreamExt; use object_store::ObjectStore; use object_store::path::Path; +use serde::{Deserialize, Serialize}; use crate::{CatalogRecord, ListRunsQuery, Result}; use fabro_types::RunId; +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct StartIndexRecord { + run_id: RunId, +} + pub(crate) async fn write_catalog( store: Arc, base_prefix: &str, @@ -29,7 +35,7 @@ pub(crate) async fn write_catalog( store .put( &by_start_path(base_prefix, created_at, run_id), - bytes.into(), + serde_json::to_vec(&StartIndexRecord { run_id: *run_id })?.into(), ) .await?; Ok(record) @@ -52,7 +58,10 @@ pub(crate) async fn list_catalogs( let metas = store.list(Some(&prefix)).try_collect::>().await?; let mut records = Vec::new(); for meta in metas { - let Some(record) = read_catalog_path(store.clone(), meta.location).await? else { + let Some(index) = read_start_index_path(store.clone(), meta.location).await? else { + continue; + }; + let Some(record) = read_locator(store.clone(), base_prefix, &index.run_id).await? else { continue; }; if let Some(start) = query.start { @@ -99,6 +108,22 @@ pub(crate) async fn read_catalog_path( } } +#[cfg(test)] +fn by_start_record(run_id: &RunId) -> StartIndexRecord { + StartIndexRecord { run_id: *run_id } +} + +async fn read_start_index_path( + store: Arc, + path: Path, +) -> Result> { + match store.get(&path).await { + Ok(result) => Ok(Some(serde_json::from_slice(&result.bytes().await?)?)), + Err(object_store::Error::NotFound { .. }) => Ok(None), + Err(err) => Err(err.into()), + } +} + #[cfg(test)] pub(super) mod test_support { use super::*; @@ -125,7 +150,12 @@ pub(super) mod test_support { for record in canonical.values() { let path = by_start_path(base_prefix, record.created_at, &record.run_id); if !object_exists(store.clone(), &path).await? { - store.put(&path, serde_json::to_vec(record)?.into()).await?; + store + .put( + &path, + serde_json::to_vec(&by_start_record(&record.run_id))?.into(), + ) + .await?; } } @@ -136,16 +166,16 @@ pub(super) mod test_support { let mut seen = HashSet::new(); for meta in by_start_metas { let location = meta.location.clone(); - let Some(record) = read_catalog_path(store.clone(), location.clone()).await? else { + let Some(index) = read_start_index_path(store.clone(), location.clone()).await? else { delete_if_exists(store.clone(), &location).await?; continue; }; - let expected = canonical.get(&record.run_id).map(|canonical_record| { - by_start_path(base_prefix, canonical_record.created_at, &record.run_id) + let expected = canonical.get(&index.run_id).map(|canonical_record| { + by_start_path(base_prefix, canonical_record.created_at, &index.run_id) }); match expected { Some(expected) if expected == location => { - seen.insert(record.run_id); + seen.insert(index.run_id); } _ => { delete_if_exists(store.clone(), &location).await?; @@ -158,7 +188,7 @@ pub(super) mod test_support { store .put( &by_start_path(base_prefix, record.created_at, &record.run_id), - serde_json::to_vec(record)?.into(), + serde_json::to_vec(&by_start_record(&record.run_id))?.into(), ) .await?; } diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 4ac1d497c..4a0ea42d3 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -16,8 +16,14 @@ use tokio::sync::Mutex; use crate::keys; use crate::{CatalogRecord, ListRunsQuery, Result, RunSummary, StoreError}; use fabro_types::RunId; -pub use run_store::SlateRunStore; use run_store::SlateRunStoreInner; +pub use run_store::{NodeAsset, SlateRunStore}; + +#[derive(Clone, Copy)] +enum RunOpenMode { + Writer, + Reader, +} #[derive(Clone)] pub struct SlateStore { @@ -104,55 +110,61 @@ impl SlateStore { weak.upgrade().map(SlateRunStore::from_inner) } - async fn open_run_store(&self, record: &CatalogRecord) -> Result> { - if let Some(active) = self.get_active_run(&record.run_id).await { - if active.matches_record(record) { - return Ok(Some(active)); + async fn open_existing_run( + &self, + record: &CatalogRecord, + mode: RunOpenMode, + ) -> Result> { + if matches!(mode, RunOpenMode::Writer) { + if let Some(active) = self.get_active_run(&record.run_id).await { + if active.record() == *record { + return Ok(Some(active)); + } + return Err(StoreError::Other(format!( + "active run cache mismatch for run_id {:?}", + record.run_id + ))); } - return Err(StoreError::Other(format!( - "active run cache mismatch for run_id {:?}", - record.run_id - ))); } if !self.db_prefix_has_objects(&record.db_prefix).await? { return Ok(None); } - let db = self.open_db(&record.db_prefix).await?; - let has_init = match SlateRunStore::validate_init(&db, record).await { - Ok(has_init) => has_init, - Err(err) => { - let _ = db.close().await; - return Err(err); + match mode { + RunOpenMode::Writer => { + let db = self.open_db(&record.db_prefix).await?; + let has_init = match SlateRunStore::validate_init(&db, record).await { + Ok(has_init) => has_init, + Err(err) => { + let _ = db.close().await; + return Err(err); + } + }; + if !has_init { + let _ = db.close().await; + return Ok(None); + } + let run_store = SlateRunStore::open_writer(record.clone(), db).await?; + self.cache_active_run(&run_store).await; + Ok(Some(run_store)) } - }; - if !has_init { - let _ = db.close().await; - return Ok(None); - } - let run_store = SlateRunStore::open_writer(record.clone(), db).await?; - self.cache_active_run(&run_store).await; - Ok(Some(run_store)) - } - - async fn open_run_reader_store(&self, record: &CatalogRecord) -> Result> { - if !self.db_prefix_has_objects(&record.db_prefix).await? { - return Ok(None); - } - let reader = self.open_reader(&record.db_prefix).await?; - let has_init = match SlateRunStore::validate_init(&reader, record).await { - Ok(has_init) => has_init, - Err(err) => { - let _ = reader.close().await; - return Err(err); + RunOpenMode::Reader => { + let reader = self.open_reader(&record.db_prefix).await?; + let has_init = match SlateRunStore::validate_init(&reader, record).await { + Ok(has_init) => has_init, + Err(err) => { + let _ = reader.close().await; + return Err(err); + } + }; + if !has_init { + let _ = reader.close().await; + return Ok(None); + } + SlateRunStore::open_reader(record.clone(), reader) + .await + .map(Some) } - }; - if !has_init { - let _ = reader.close().await; - return Ok(None); } - SlateRunStore::open_reader(record.clone(), reader) - .await - .map(Some) } async fn delete_db_prefix(&self, db_prefix: &str) -> Result<()> { @@ -237,7 +249,7 @@ impl SlateStore { .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; let run_store = self - .open_run_store(&locator) + .open_existing_run(&locator, RunOpenMode::Writer) .await? .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; Ok(run_store) @@ -249,19 +261,28 @@ impl SlateStore { .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; let run_store = self - .open_run_reader_store(&locator) + .open_existing_run(&locator, RunOpenMode::Reader) .await? .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; Ok(run_store) } - pub async fn list_runs(&self, query: &ListRunsQuery) -> Result> { + pub async fn list_runs(&self) -> Result> { + self.list_runs_in_range(None, None).await + } + + pub async fn list_runs_in_range( + &self, + start: Option>, + end: Option>, + ) -> Result> { + let query = ListRunsQuery { start, end }; let catalogs = - catalog::list_catalogs(self.object_store.clone(), &self.base_prefix, query).await?; + catalog::list_catalogs(self.object_store.clone(), &self.base_prefix, &query).await?; let mut summaries = Vec::new(); for record in catalogs { if let Some(active) = self.get_active_run(&record.run_id).await { - if !active.matches_record(&record) { + if active.record() != record { return Err(StoreError::Other(format!( "active run cache mismatch for run_id {:?}", record.run_id @@ -295,59 +316,67 @@ impl SlateStore { active.close().await?; } - if let Some(locator) = - catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? + if let Some(record) = + catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id) + .await? + .or(active_record) { - delete_path( - self.object_store.clone(), - &catalog::by_start_path(&self.base_prefix, locator.created_at, run_id), - ) - .await?; - self.delete_db_prefix(&locator.db_prefix).await?; - delete_path( - self.object_store.clone(), - &catalog::by_id_path(&self.base_prefix, run_id), - ) - .await?; - return Ok(()); + return self.delete_run_record(&record).await; } - if let Some(record) = active_record { - delete_path( - self.object_store.clone(), - &catalog::by_start_path(&self.base_prefix, record.created_at, run_id), - ) - .await?; - self.delete_db_prefix(&record.db_prefix).await?; - delete_path( - self.object_store.clone(), - &catalog::by_id_path(&self.base_prefix, run_id), - ) - .await?; - return Ok(()); - } + self.repair_delete_run(run_id).await + } +} +impl SlateStore { + async fn delete_run_record(&self, record: &CatalogRecord) -> Result<()> { + delete_path( + self.object_store.clone(), + &catalog::by_start_path(&self.base_prefix, record.created_at, &record.run_id), + ) + .await?; + self.delete_db_prefix(&record.db_prefix).await?; + delete_path( + self.object_store.clone(), + &catalog::by_id_path(&self.base_prefix, &record.run_id), + ) + .await?; + Ok(()) + } + + async fn repair_delete_run(&self, run_id: &RunId) -> Result<()> { let by_start_prefix = Path::from(format!("{}by-start", self.base_prefix)); - let metas = self + let db_prefix = Path::from(format!("{}db", self.base_prefix)); + let expected_name = format!("{run_id}.json"); + + for meta in self .object_store .list(Some(&by_start_prefix)) .try_collect::>() - .await?; - let expected_name = format!("{run_id}.json"); - for meta in metas { - if meta.location.filename() != Some(expected_name.as_str()) { - continue; - } - let Some(record) = - catalog::read_catalog_path(self.object_store.clone(), meta.location.clone()) - .await? - else { + .await? + { + if meta.location.filename() == Some(expected_name.as_str()) { delete_path(self.object_store.clone(), &meta.location).await?; - continue; - }; - self.delete_db_prefix(&record.db_prefix).await?; - delete_path(self.object_store.clone(), &meta.location).await?; + } } + + let db_run_segment = format!("/{run_id}/"); + for meta in self + .object_store + .list(Some(&db_prefix)) + .try_collect::>() + .await? + { + if meta.location.to_string().contains(&db_run_segment) { + delete_path(self.object_store.clone(), &meta.location).await?; + } + } + + delete_path( + self.object_store.clone(), + &catalog::by_id_path(&self.base_prefix, run_id), + ) + .await?; Ok(()) } } @@ -573,6 +602,11 @@ mod tests { store.head(path).await.is_ok() } + async fn read_json_value(store: Arc, path: &Path) -> serde_json::Value { + let result = store.get(path).await.unwrap(); + serde_json::from_slice(&result.bytes().await.unwrap()).unwrap() + } + async fn seed_db( object_store: Arc, record: &CatalogRecord, @@ -653,8 +687,14 @@ mod tests { let by_start = catalog::by_start_path("runs/", created_at, &test_run_id("run-1")); assert!(object_exists(object_store.clone(), &by_id).await); assert!(object_exists(object_store.clone(), &by_start).await); + assert_eq!( + read_json_value(object_store.clone(), &by_start).await, + serde_json::json!({ + "run_id": test_run_id("run-1").to_string(), + }) + ); - let summary = store.list_runs(&ListRunsQuery::default()).await.unwrap(); + let summary = store.list_runs().await.unwrap(); assert_eq!(summary.len(), 1); assert_eq!(summary[0].catalog.run_id, test_run_id("run-1")); assert_eq!(summary[0].workflow_name, Some("night-sky".to_string())); @@ -717,16 +757,10 @@ mod tests { .unwrap(); assert!(store.open_run(&test_run_id("run-1")).await.is_ok()); - assert!( - store - .list_runs(&ListRunsQuery::default()) - .await - .unwrap() - .is_empty() - ); + assert!(store.list_runs().await.unwrap().is_empty()); repair_catalog_for_tests(&store).await.unwrap(); - let listed = store.list_runs(&ListRunsQuery::default()).await.unwrap(); + let listed = store.list_runs().await.unwrap(); assert_eq!(listed.len(), 1); assert!( object_exists( @@ -737,6 +771,65 @@ mod tests { ); } + #[tokio::test] + async fn list_runs_uses_by_start_index_to_resolve_canonical_by_id_record() { + let (object_store, store) = make_store(); + let created_at = dt("2026-03-27T12:00:00Z"); + let record = CatalogRecord { + run_id: test_run_id("run-1"), + created_at, + db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")), + run_dir: None, + }; + + let db = seed_db(object_store.clone(), &record, true).await; + db.put( + keys::event_key(1, created_at.timestamp_millis()), + serde_json::to_vec(&event_payload( + "run-1", + "2026-03-27T12:00:00Z", + "run.created", + None, + serde_json::json!({ + "settings": sample_run_record("run-1", created_at).settings, + "graph": sample_run_record("run-1", created_at).graph, + "workflow_slug": sample_run_record("run-1", created_at).workflow_slug, + "working_directory": sample_run_record("run-1", created_at).working_directory, + "host_repo_path": sample_run_record("run-1", created_at).host_repo_path, + "base_branch": sample_run_record("run-1", created_at).base_branch, + "labels": sample_run_record("run-1", created_at).labels, + }), + )) + .unwrap(), + ) + .await + .unwrap(); + db.close().await.unwrap(); + + object_store + .put( + &catalog::by_id_path("runs/", &test_run_id("run-1")), + serde_json::to_vec(&record).unwrap().into(), + ) + .await + .unwrap(); + object_store + .put( + &catalog::by_start_path("runs/", created_at, &test_run_id("run-1")), + serde_json::json!({ + "run_id": test_run_id("run-1").to_string(), + }) + .to_string() + .into(), + ) + .await + .unwrap(); + + let listed = store.list_runs().await.unwrap(); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].catalog, record); + } + #[tokio::test] async fn reopen_recovers_event_sequences() { let (_object_store, store) = make_store(); @@ -826,13 +919,7 @@ mod tests { .unwrap(); assert!(store.open_run(&test_run_id("run-1")).await.is_err()); - assert!( - store - .list_runs(&ListRunsQuery::default()) - .await - .unwrap() - .is_empty() - ); + assert!(store.list_runs().await.unwrap().is_empty()); } #[tokio::test] @@ -885,7 +972,7 @@ mod tests { .await .unwrap(); - let listed = store.list_runs(&ListRunsQuery::default()).await.unwrap(); + let listed = store.list_runs().await.unwrap(); assert_eq!(listed.len(), 1); let reopened = store.open_run(&test_run_id("run-1")).await.unwrap(); @@ -1186,12 +1273,20 @@ mod tests { assert_eq!( run.list_all_assets().await.unwrap(), vec![ - ( - "artifact-only".to_string(), - 7, - "logs/output.txt".to_string() - ), - ("code".to_string(), 2, "src/lib.rs".to_string()) + crate::slate::NodeAsset { + node: crate::NodeVisit { + node_id: "artifact-only".to_string(), + visit: 7, + }, + filename: "logs/output.txt".to_string(), + }, + crate::slate::NodeAsset { + node: crate::NodeVisit { + node_id: "code".to_string(), + visit: 2, + }, + filename: "src/lib.rs".to_string(), + } ] ); } @@ -1493,7 +1588,7 @@ mod tests { Some("diff --git a/src/lib.rs b/src/lib.rs\n") ); assert_eq!(state.pull_request, Some(pull_request.clone())); - assert!(state.nodes.keys().any(|(id, _)| id == "code")); + assert!(state.iter_nodes().any(|(node, _)| node.node_id == "code")); let node_state = state .node(&node) .expect("node state should exist for code:2"); @@ -1865,7 +1960,7 @@ mod tests { .map(|checkpoint| checkpoint.current_node.as_str()), Some("plan") ); - assert!(state.nodes.is_empty()); + assert!(state.is_empty()); } #[tokio::test] @@ -1876,12 +1971,15 @@ mod tests { .await .unwrap(); - let invalid_missing: EventPayload = serde_json::from_value(serde_json::json!({ + let invalid_missing_err = serde_json::from_value::(serde_json::json!({ "run_id": "run-1" })) - .unwrap(); - let err = run.append_event(&invalid_missing).await.unwrap_err(); - assert!(matches!(err, StoreError::InvalidEvent(_))); + .unwrap_err(); + assert!( + invalid_missing_err + .to_string() + .contains("missing or non-string required field: id") + ); let invalid_run_id: EventPayload = serde_json::from_value(serde_json::json!({ "id": "evt-invalid-run", @@ -2018,7 +2116,7 @@ mod tests { .await .unwrap(); - let all = store.list_runs(&ListRunsQuery::default()).await.unwrap(); + let all = store.list_runs().await.unwrap(); assert_eq!(all.len(), 2); assert_eq!(all[0].catalog.run_id, test_run_id("run-late")); assert_eq!(all[0].workflow_name, Some("night-sky".to_string())); @@ -2033,10 +2131,10 @@ mod tests { assert_eq!(all[1].status, None); let filtered = store - .list_runs(&ListRunsQuery { - start: Some(dt("2026-03-27T11:00:00Z")), - end: Some(dt("2026-03-27T13:00:00Z")), - }) + .list_runs_in_range( + Some(dt("2026-03-27T11:00:00Z")), + Some(dt("2026-03-27T13:00:00Z")), + ) .await .unwrap(); assert_eq!(filtered.len(), 1); diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index a1e19ced7..4b5df4e6b 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -15,32 +15,30 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; use crate::run_state::EventProjectionCache; use crate::{ - CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, Result, RunProjection, RunSummary, - StoreError, + CatalogRecord, EventEnvelope, EventPayload, NodeVisit, NodeVisitRef, Result, RunProjection, + RunSummary, StoreError, }; -use fabro_types::RunId; - #[derive(Clone)] pub struct SlateRunStore { inner: Arc, } +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct NodeAsset { + pub node: NodeVisit, + pub filename: String, +} + impl std::fmt::Debug for SlateRunStore { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("SlateRunStore") - .field("run_id", &self.inner.run_id) - .field("created_at", &self.inner.created_at) - .field("db_prefix", &self.inner.db_prefix) - .field("run_dir", &self.inner.run_dir) + .field("record", &self.inner.record) .finish_non_exhaustive() } } pub(crate) struct SlateRunStoreInner { - run_id: RunId, - created_at: DateTime, - db_prefix: String, - run_dir: Option, + record: CatalogRecord, db: SlateRunDb, event_seq: AtomicU32, close_lock: Mutex<()>, @@ -57,10 +55,7 @@ impl SlateRunStore { let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?; Ok(Self { inner: Arc::new(SlateRunStoreInner { - run_id: record.run_id, - created_at: record.created_at, - db_prefix: record.db_prefix, - run_dir: record.run_dir, + record, db: SlateRunDb::Writer(db), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -73,10 +68,7 @@ impl SlateRunStore { let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?; Ok(Self { inner: Arc::new(SlateRunStoreInner { - run_id: record.run_id, - created_at: record.created_at, - db_prefix: record.db_prefix, - run_dir: record.run_dir, + record, db: SlateRunDb::Reader(Box::new(db)), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -94,23 +86,11 @@ impl SlateRunStore { } pub(crate) fn record(&self) -> CatalogRecord { - CatalogRecord { - run_id: self.inner.run_id, - created_at: self.inner.created_at, - db_prefix: self.inner.db_prefix.clone(), - run_dir: self.inner.run_dir.clone(), - } - } - - pub(crate) fn matches_record(&self, record: &CatalogRecord) -> bool { - self.inner.run_id == record.run_id - && self.inner.created_at == record.created_at - && self.inner.db_prefix == record.db_prefix - && self.inner.run_dir == record.run_dir + self.inner.record.clone() } pub(crate) fn created_at(&self) -> DateTime { - self.inner.created_at + self.inner.record.created_at } pub(crate) async fn close(&self) -> Result<()> { @@ -168,7 +148,13 @@ impl SlateRunStore { impl SlateRunStore { pub async fn append_event(&self, payload: &EventPayload) -> Result { - payload.validate(&self.inner.run_id)?; + if payload.run_id() != self.inner.record.run_id.to_string() { + return Err(StoreError::InvalidEvent(format!( + "payload run_id {:?} does not match store run_id {:?}", + payload.run_id(), + self.inner.record.run_id + ))); + } let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst); self.inner .db @@ -267,7 +253,7 @@ impl SlateRunStore { .await } - pub async fn list_all_assets(&self) -> Result> { + pub async fn list_all_assets(&self) -> Result> { self.inner.db.list_all_assets().await } @@ -327,7 +313,7 @@ impl SlateRunDb { } } - async fn list_all_assets(&self) -> Result> { + async fn list_all_assets(&self) -> Result> { match self { Self::Writer(db) => list_all_assets(db).await, Self::Reader(db) => list_all_assets(db.as_ref()).await, @@ -418,7 +404,7 @@ where Ok(artifact_ids) } -async fn list_all_assets(db: &R) -> Result> +async fn list_all_assets(db: &R) -> Result> where R: DbRead + Sync, { @@ -428,10 +414,13 @@ where let mut assets = Vec::new(); while let Some(entry) = iter.next().await? { let key = key_to_string(&entry.key)?; - let Some(asset) = keys::parse_node_asset_key(&key) else { + let Some((node_id, visit, filename)) = keys::parse_node_asset_key(&key) else { continue; }; - assets.push(asset); + assets.push(NodeAsset { + node: NodeVisit { node_id, visit }, + filename, + }); } assets.sort(); Ok(assets) diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index 9d286d4f3..cb263d6c8 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; +use serde::{Deserialize, Deserializer, Serialize}; use crate::{Result, StoreError}; use fabro_types::{RunId, RunStatus, StatusReason}; @@ -12,6 +12,32 @@ pub struct NodeVisitRef<'a> { pub visit: u32, } +impl<'a> NodeVisitRef<'a> { + #[must_use] + pub fn into_owned(self) -> NodeVisit { + NodeVisit { + node_id: self.node_id.to_string(), + visit: self.visit, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +pub struct NodeVisit { + pub node_id: String, + pub visit: u32, +} + +impl NodeVisit { + #[must_use] + pub fn as_ref(&self) -> NodeVisitRef<'_> { + NodeVisitRef { + node_id: &self.node_id, + visit: self.visit, + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct CatalogRecord { pub run_id: RunId, @@ -36,7 +62,7 @@ pub struct RunSummary { pub total_cost: Option, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize)] #[serde(transparent)] pub struct EventPayload(serde_json::Value); @@ -48,6 +74,17 @@ impl EventPayload { } pub(crate) fn validate(&self, expected_run_id: &RunId) -> Result<()> { + self.validate_shape()?; + if self.run_id() == expected_run_id.to_string() { + return Ok(()); + } + Err(StoreError::InvalidEvent(format!( + "payload run_id {:?} does not match store run_id {expected_run_id:?}", + self.run_id() + ))) + } + + fn validate_shape(&self) -> Result<()> { let obj = self.0.as_object().ok_or_else(|| { StoreError::InvalidEvent("event payload must be a JSON object".into()) })?; @@ -62,22 +99,14 @@ impl EventPayload { } } } - - match obj.get("run_id") { - Some(serde_json::Value::String(run_id)) if run_id == &expected_run_id.to_string() => { - Ok(()) - } - Some(serde_json::Value::String(run_id)) => Err(StoreError::InvalidEvent(format!( - "payload run_id {run_id:?} does not match store run_id {expected_run_id:?}" - ))), - _ => Err(StoreError::InvalidEvent( - "missing or non-string required field: run_id".into(), - )), - } + Ok(()) } - pub fn into_inner(self) -> serde_json::Value { - self.0 + #[must_use] + pub fn run_id(&self) -> &str { + self.0["run_id"] + .as_str() + .expect("EventPayload::run_id called before validation") } pub fn as_value(&self) -> &serde_json::Value { @@ -85,6 +114,17 @@ impl EventPayload { } } +impl<'de> Deserialize<'de> for EventPayload { + fn deserialize(deserializer: D) -> std::result::Result + where + D: Deserializer<'de>, + { + let payload = Self(serde_json::Value::deserialize(deserializer)?); + payload.validate_shape().map_err(serde::de::Error::custom)?; + Ok(payload) + } +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct EventEnvelope { pub seq: u32, diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index f6ce009fc..f5d057162 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -5,9 +5,7 @@ use std::path::PathBuf; use anyhow::{Context, Result, bail}; use fabro_checkpoint::branch::BranchStore; use fabro_checkpoint::git::Store as GitStore; -use fabro_store::{ - ListRunsQuery, NodeVisitRef, SlateRunStore as DurableRunStore, SlateStore as DurableStore, -}; +use fabro_store::{NodeVisitRef, SlateRunStore as DurableRunStore, SlateStore as DurableStore}; use fabro_types::RunId; use git2::{Repository, Signature}; use ulid::Ulid; @@ -184,7 +182,7 @@ pub async fn find_run_id_by_prefix_or_store( let current_repo_root = canonical_repo_root(repo)?; let mut matches = Vec::new(); - for summary in fabro_store.list_runs(&ListRunsQuery::default()).await? { + for summary in fabro_store.list_runs().await? { if summary.catalog.run_id.to_string() == prefix { if summary.host_repo_path.is_none() { return Ok(summary.catalog.run_id); diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index c88213d9c..8be8fc69d 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -206,14 +206,13 @@ fn parse_dot_summary(dot: &str) -> (String, usize, usize) { /// directory scan behavior. fn read_plan_text(state: &RunProjection) -> Option { let mut plan_nodes = state - .nodes - .iter() - .filter_map(|((node_id, visit), node)| { - node_id.starts_with("plan").then_some(( - node_id.as_str(), - *visit, - node.response.as_deref(), - )) + .iter_nodes() + .filter_map(|(node, node_state)| { + let node_id = node.node_id; + let visit = node.visit; + node_id + .starts_with("plan") + .then_some((node_id, visit, node_state.response.as_deref())) }) .collect::>(); plan_nodes.sort_by(|left, right| left.0.cmp(right.0).then(left.1.cmp(&right.1))); @@ -908,8 +907,11 @@ mod tests { #[test] fn read_plan_text_found() { let mut state = RunProjection::default(); - state.nodes.insert( - ("plan".to_string(), 1), + state.set_node( + fabro_store::NodeVisitRef { + node_id: "plan", + visit: 1, + }, fabro_store::NodeState { response: Some("This is the plan".to_string()), ..Default::default() @@ -923,8 +925,11 @@ mod tests { #[test] fn read_plan_text_prefix_match() { let mut state = RunProjection::default(); - state.nodes.insert( - ("planning".to_string(), 1), + state.set_node( + fabro_store::NodeVisitRef { + node_id: "planning", + visit: 1, + }, fabro_store::NodeState { response: Some("Planning content".to_string()), ..Default::default() @@ -938,15 +943,21 @@ mod tests { #[test] fn read_plan_text_prefers_alphabetically_first_plan_node() { let mut state = RunProjection::default(); - state.nodes.insert( - ("planning".to_string(), 1), + state.set_node( + fabro_store::NodeVisitRef { + node_id: "planning", + visit: 1, + }, fabro_store::NodeState { response: Some("Planning content".to_string()), ..Default::default() }, ); - state.nodes.insert( - ("plan".to_string(), 1), + state.set_node( + fabro_store::NodeVisitRef { + node_id: "plan", + visit: 1, + }, fabro_store::NodeState { response: Some("Plan content".to_string()), ..Default::default() @@ -960,8 +971,11 @@ mod tests { #[test] fn read_plan_text_not_found() { let mut state = RunProjection::default(); - state.nodes.insert( - ("implement".to_string(), 1), + state.set_node( + fabro_store::NodeVisitRef { + node_id: "implement", + visit: 1, + }, fabro_store::NodeState::default(), ); diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index 007403038..1ad570518 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -2,7 +2,7 @@ use std::io::Write; use std::path::{Component, Path, PathBuf}; use anyhow::{Context, Result, bail}; -use fabro_store::{NodeVisitRef, RunProjection, SlateRunStore}; +use fabro_store::{RunProjection, SlateRunStore}; use crate::git::MetadataStore; @@ -43,60 +43,65 @@ impl RunDump { #[must_use] pub fn metadata_checkpoint(state: &RunProjection) -> Self { let mut entries = Vec::new(); - let mut keys: Vec<_> = state.nodes.keys().collect(); + let mut keys: Vec<_> = state + .iter_nodes() + .map(|(node, _)| node.into_owned()) + .collect(); keys.sort(); - for (node_id, visit) in keys { - let Some(node) = state.nodes.get(&(node_id.clone(), *visit)) else { + for node_key in keys { + let Some(node) = state.node(&node_key.as_ref()) else { continue; }; + let node_id = node_key.node_id.as_str(); + let visit = node_key.visit; if let Some(prompt) = node.prompt.as_ref() { entries.push(RunDumpEntry::text( - metadata_node_file_path(node_id, *visit, "prompt.md"), + metadata_node_file_path(node_id, visit, "prompt.md"), prompt.clone(), )); } if let Some(response) = node.response.as_ref() { entries.push(RunDumpEntry::text( - metadata_node_file_path(node_id, *visit, "response.md"), + metadata_node_file_path(node_id, visit, "response.md"), response.clone(), )); } if let Some(status) = node.status.as_ref() { push_json_entry_path( &mut entries, - &PathBuf::from(metadata_node_file_path(node_id, *visit, "status.json")), + &PathBuf::from(metadata_node_file_path(node_id, visit, "status.json")), status, ); } if let Some(provider_used) = node.provider_used.as_ref() { entries.push(RunDumpEntry::json( - metadata_node_file_path(node_id, *visit, "provider_used.json"), + metadata_node_file_path(node_id, visit, "provider_used.json"), provider_used.clone(), )); } if let Some(diff) = node.diff.as_ref() { entries.push(RunDumpEntry::text( - metadata_node_file_path(node_id, *visit, "diff.patch"), + metadata_node_file_path(node_id, visit, "diff.patch"), diff.clone(), )); } if let Some(script_invocation) = node.script_invocation.as_ref() { entries.push(RunDumpEntry::json( - metadata_node_file_path(node_id, *visit, "script_invocation.json"), + metadata_node_file_path(node_id, visit, "script_invocation.json"), script_invocation.clone(), )); } if let Some(script_timing) = node.script_timing.as_ref() { entries.push(RunDumpEntry::json( - metadata_node_file_path(node_id, *visit, "script_timing.json"), + metadata_node_file_path(node_id, visit, "script_timing.json"), script_timing.clone(), )); } if let Some(parallel_results) = node.parallel_results.as_ref() { entries.push(RunDumpEntry::json( - metadata_node_file_path(node_id, *visit, "parallel_results.json"), + metadata_node_file_path(node_id, visit, "parallel_results.json"), parallel_results.clone(), )); } @@ -142,14 +147,19 @@ impl RunDump { push_json_entry(&mut entries, "sandbox.json", record); } - let mut node_keys: Vec<_> = state.nodes.keys().cloned().collect(); + let mut node_keys: Vec<_> = state + .iter_nodes() + .map(|(node, _)| node.into_owned()) + .collect(); node_keys.sort(); - for (node_id, visit) in &node_keys { - let node = &state.nodes[&(node_id.clone(), *visit)]; - let node_id_segment = validate_single_path_segment("node id", node_id)?; + for node_key in &node_keys { + let node = state + .node(&node_key.as_ref()) + .with_context(|| format!("missing node {:?} in projection", node_key))?; + let node_id_segment = validate_single_path_segment("node id", &node_key.node_id)?; let base = PathBuf::from("nodes") .join(node_id_segment) - .join(format!("visit-{visit}")); + .join(format!("visit-{}", node_key.visit)); if let Some(prompt) = node.prompt.as_ref() { entries.push(RunDumpEntry::text_path( @@ -218,26 +228,23 @@ impl RunDump { )); } - for (node_id, visit, filename) in run_store.list_all_assets().await? { - let node_id_segment = validate_single_path_segment("node id", &node_id)?; - let filename_path = validate_relative_path("asset filename", &filename)?; - let node = NodeVisitRef { - node_id: &node_id, - visit, - }; + for asset in run_store.list_all_assets().await? { + let node_id_segment = validate_single_path_segment("node id", &asset.node.node_id)?; + let filename_path = validate_relative_path("asset filename", &asset.filename)?; let data = run_store - .get_asset(&node, &filename) + .get_asset(&asset.node.as_ref(), &asset.filename) .await? .with_context(|| { format!( - "asset {filename:?} for node {node_id:?} visit {visit} is missing from the store" + "asset {:?} for node {:?} visit {} is missing from the store", + asset.filename, asset.node.node_id, asset.node.visit ) })?; entries.push(RunDumpEntry::bytes_path( &PathBuf::from("artifacts") .join("nodes") .join(node_id_segment) - .join(format!("visit-{visit}")) + .join(format!("visit-{}", asset.node.visit)) .join(filename_path), data.to_vec(), )); diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index f656482ef..5d727df63 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -3,31 +3,26 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result, bail}; use chrono::{DateTime, Utc}; -use fabro_store::{ListRunsQuery, SlateStore}; +use fabro_store::{RunSummary, SlateStore}; use fabro_types::RunId; use serde::Serialize; use crate::run_status::{RunStatus, StatusReason}; +#[derive(Debug, Clone)] +struct RunLocalState { + dir_name: String, + start_time_dt: Option>, + end_time: Option>, + path: PathBuf, + is_orphan: bool, +} + #[derive(Debug, Clone, Serialize)] pub struct RunInfo { - pub run_id: RunId, + #[serde(skip)] + summary: Option, pub dir_name: String, - pub workflow_name: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub workflow_slug: Option, - pub status: RunStatus, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub status_reason: Option, - pub start_time: String, - pub labels: HashMap, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub duration_ms: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub total_cost: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub host_repo_path: Option, - pub goal: String, #[serde(skip)] pub start_time_dt: Option>, #[serde(skip)] @@ -38,6 +33,98 @@ pub struct RunInfo { pub is_orphan: bool, } +impl RunInfo { + fn new(summary: Option, local: RunLocalState) -> Self { + Self { + summary, + dir_name: local.dir_name, + start_time_dt: local.start_time_dt, + end_time: local.end_time, + path: local.path, + is_orphan: local.is_orphan, + } + } + + pub fn run_id(&self) -> RunId { + self.summary + .as_ref() + .map(|summary| summary.catalog.run_id) + .or_else(|| parse_run_id(&self.dir_name)) + .expect("RunInfo must have a run id") + } + + pub fn workflow_name(&self) -> String { + self.summary + .as_ref() + .and_then(|summary| summary.workflow_name.clone()) + .unwrap_or_else(|| "[no run record]".to_string()) + } + + pub fn workflow_slug(&self) -> Option<&str> { + self.summary + .as_ref() + .and_then(|summary| summary.workflow_slug.as_deref()) + } + + pub fn status(&self) -> RunStatus { + self.summary + .as_ref() + .and_then(|summary| summary.status) + .unwrap_or(RunStatus::Dead) + } + + pub fn status_reason(&self) -> Option { + self.summary + .as_ref() + .and_then(|summary| summary.status_reason) + } + + pub fn start_time(&self) -> String { + self.summary + .as_ref() + .and_then(|summary| summary.start_time.or(Some(summary.catalog.created_at))) + .or(self.start_time_dt) + .map(|time| time.to_rfc3339()) + .unwrap_or_default() + } + + pub fn labels(&self) -> &HashMap { + if let Some(summary) = self.summary.as_ref() { + &summary.labels + } else { + empty_labels() + } + } + + pub fn duration_ms(&self) -> Option { + self.summary + .as_ref() + .and_then(|summary| summary.duration_ms) + } + + pub fn total_cost(&self) -> Option { + self.summary.as_ref().and_then(|summary| summary.total_cost) + } + + pub fn host_repo_path(&self) -> Option<&str> { + self.summary + .as_ref() + .and_then(|summary| summary.host_repo_path.as_deref()) + } + + pub fn goal(&self) -> String { + self.summary + .as_ref() + .and_then(|summary| summary.goal.clone()) + .unwrap_or_default() + } +} + +fn empty_labels() -> &'static HashMap { + static EMPTY: std::sync::OnceLock> = std::sync::OnceLock::new(); + EMPTY.get_or_init(HashMap::new) +} + pub fn default_storage_dir() -> PathBuf { dirs::home_dir() .expect("could not determine home directory") @@ -48,10 +135,6 @@ pub fn logs_base(storage_dir: &Path) -> PathBuf { storage_dir.join("logs") } -pub fn default_logs_base() -> PathBuf { - logs_base(&default_storage_dir()) -} - pub fn runs_base(storage_dir: &Path) -> PathBuf { storage_dir.join("runs") } @@ -81,35 +164,25 @@ fn scan_orphan_runs(base: &Path) -> Result> { .ok() .and_then(|m| m.modified().ok()) .map(|time| -> DateTime { time.into() }); - let mtime = mtime_dt.map(|dt| dt.to_rfc3339()).unwrap_or_default(); let run_id = std::fs::read_to_string(path.join("id.txt")) .ok() .and_then(|s| parse_run_id(&s)) .or_else(|| parse_run_id(&dir_name)); - let Some(run_id) = run_id else { + if run_id.is_none() { continue; - }; + } - let status_info = StatusInfo::simple(RunStatus::Dead); - runs.push(RunInfo { - run_id, - dir_name, - workflow_name: "[no run record]".to_string(), - workflow_slug: None, - status: status_info.status, - status_reason: status_info.reason, - start_time: mtime, - labels: HashMap::new(), - duration_ms: status_info.duration_ms, - total_cost: status_info.total_cost, - host_repo_path: None, - start_time_dt: mtime_dt, - end_time: status_info.end_time, - path, - goal: String::new(), - is_orphan: true, - }); + runs.push(RunInfo::new( + None, + RunLocalState { + dir_name, + start_time_dt: mtime_dt, + end_time: None, + path, + is_orphan: true, + }, + )); } runs.sort_by(|a, b| b.start_time_dt.cmp(&a.start_time_dt)); @@ -119,12 +192,12 @@ fn scan_orphan_runs(base: &Path) -> Result> { pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result> { let mut runs_by_id: HashMap = HashMap::new(); - if let Ok(store_runs) = store.list_runs(&ListRunsQuery::default()).await { + if let Ok(store_runs) = store.list_runs().await { for summary in store_runs { - let Some(run_info) = run_info_from_summary(&summary) else { + let Some(run_info) = run_info_from_summary(summary) else { continue; }; - runs_by_id.insert(summary.catalog.run_id, run_info); + runs_by_id.insert(run_info.run_id(), run_info); } } @@ -134,9 +207,9 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result>(); for run in scan_orphan_runs(base)? .into_iter() - .filter(|run| run.is_orphan && !store_run_ids.contains(&run.run_id)) + .filter(|run| run.is_orphan && !store_run_ids.contains(&run.run_id())) { - runs_by_id.insert(run.run_id, run); + runs_by_id.insert(run.run_id(), run); } let mut runs: Vec<_> = runs_by_id.into_values().collect(); @@ -144,7 +217,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result Option { +fn run_info_from_summary(summary: RunSummary) -> Option { let run_dir = summary.catalog.run_dir.as_deref()?; let path = PathBuf::from(run_dir); if !path.exists() { @@ -154,7 +227,6 @@ fn run_info_from_summary(summary: &fabro_store::RunSummary) -> Option { .file_name() .map(|name| name.to_string_lossy().to_string())?; let start_time_dt = summary.catalog.created_at; - let start_time = summary.start_time.unwrap_or(start_time_dt); let end_time = if summary.status.is_some_and(RunStatus::is_terminal) { summary.duration_ms.and_then(|duration_ms| { Some(start_time_dt + chrono::Duration::milliseconds(i64::try_from(duration_ms).ok()?)) @@ -163,47 +235,16 @@ fn run_info_from_summary(summary: &fabro_store::RunSummary) -> Option { None }; - Some(RunInfo { - run_id: summary.catalog.run_id, - dir_name, - workflow_name: summary - .workflow_name - .clone() - .unwrap_or_else(|| "[starting]".to_string()), - workflow_slug: summary.workflow_slug.clone(), - status: summary.status.unwrap_or(RunStatus::Dead), - status_reason: summary.status_reason, - start_time: start_time.to_rfc3339(), - labels: summary.labels.clone(), - duration_ms: summary.duration_ms, - total_cost: summary.total_cost, - host_repo_path: summary.host_repo_path.clone(), - goal: summary.goal.clone().unwrap_or_default(), - start_time_dt: Some(start_time_dt), - end_time, - path, - is_orphan: false, - }) -} - -struct StatusInfo { - status: RunStatus, - reason: Option, - end_time: Option>, - duration_ms: Option, - total_cost: Option, -} - -impl StatusInfo { - fn simple(status: RunStatus) -> Self { - Self { - status, - reason: None, - end_time: None, - duration_ms: None, - total_cost: None, - } - } + Some(RunInfo::new( + Some(summary), + RunLocalState { + dir_name, + start_time_dt: Some(start_time_dt), + end_time, + path, + is_orphan: false, + }, + )) } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -222,24 +263,25 @@ pub fn filter_runs( ) -> Vec { runs.iter() .filter(|run| { - if status_filter == StatusFilter::RunningOnly && !run.status.is_active() { + if status_filter == StatusFilter::RunningOnly && !run.status().is_active() { return false; } if run.is_orphan && !include_orphans { return false; } if let Some(before) = before { - if !run.start_time.is_empty() && run.start_time.as_str() >= before { + let start_time = run.start_time(); + if !start_time.is_empty() && start_time.as_str() >= before { return false; } } if let Some(pattern) = workflow { - if !run.workflow_name.contains(pattern) { + if !run.workflow_name().contains(pattern) { return false; } } for (key, value) in labels { - match run.labels.get(key) { + match run.labels().get(key) { Some(current) if current == value => {} _ => return false, } @@ -261,7 +303,7 @@ pub async fn resolve_run_combined( let id_matches: Vec<_> = runs .iter() - .filter(|run| run_id_matches(run.run_id, identifier)) + .filter(|run| run_id_matches(run.run_id(), identifier)) .collect(); match id_matches.len() { @@ -269,7 +311,7 @@ pub async fn resolve_run_combined( count if count > 1 => { let ids: Vec = id_matches .iter() - .map(|run| run.run_id.to_string()) + .map(|run| run.run_id().to_string()) .collect(); bail!( "Ambiguous prefix '{identifier}': {count} runs match: {}", @@ -282,12 +324,12 @@ pub async fn resolve_run_combined( let id_lower = identifier.to_lowercase(); let id_collapsed = collapse_separators(&id_lower); let workflow_match = runs.iter().filter(|run| !run.is_orphan).find(|run| { - if let Some(slug) = &run.workflow_slug { + if let Some(slug) = run.workflow_slug() { if slug.to_lowercase() == id_lower { return true; } } - let name_lower = run.workflow_name.to_lowercase(); + let name_lower = run.workflow_name().to_lowercase(); name_lower.contains(&id_lower) || collapse_separators(&name_lower).contains(&id_collapsed) }); @@ -400,10 +442,10 @@ mod tests { let runs = scan_runs_combined(&store, temp.path()).await.unwrap(); let run = runs .iter() - .find(|run| run.run_id == fixtures::RUN_1) + .find(|run| run.run_id() == fixtures::RUN_1) .expect("run should be listed"); - assert_eq!(run.status, RunStatus::Submitted); + assert_eq!(run.status(), RunStatus::Submitted); assert!(!run.is_orphan); } }