diff --git a/BUGS.md b/BUGS.md index 6ced6e3..a829b89 100644 --- a/BUGS.md +++ b/BUGS.md @@ -2,28 +2,22 @@ These are bugs (or missing features) I've observed while working with `multi checks`. -- [ ] No use of Cersei workflows to chain multiple prompts together. +- [ ] I have not fiddled with `max turns` yet. -- [ ] No GitHub Action available. +- [ ] No loading of skills. + +- [ ] No loading of `RULES.md` files from the `.claude` directory. - [ ] Logs no longer report the id of the check that failed (or the number of attempted retries) -- [ ] Assemble_instructions is hard-coded: src/checks/executor/mod.rs:98 (definition), called from src/checks/executor/cersei.rs:110 +- [ ] `assemble_instructions` is hard-coded: `src/checks/executor/mod.rs:98` (definition), called from `src/checks/executor/cersei.rs:110` -- [ ] No system prompt provided. +- [ ] I haven't fiddled with the system prompt yet. -- [ ] Not sure if prompt caching is enabled at all. +- [ ] Not sure if prompt caching is enabled at all, but I suspect it is. - [ ] Running 16 agents seems to nearly freeze the computer. Use an OTel profile to determine if this is true. -- [ ] No limit on max turns. - -- [ ] No loading of skills. - -- [ ] No loading of RULES.md files from the .claude directory. - -- CERSEI: `append_system_prompt()` function is dead unless routed through the separate build_system_prompt() composer. - - [ ] `Ctrl-C` (shutdown signals) needs to be handled gracefully and cross-platform. The `multi check` presenter installs a raw `libc::signal(SIGINT, …)` handler (`install_terminal_guards`, src/checks/presenter/inline.rs:224) that is Unix-only @@ -37,8 +31,14 @@ the same graceful-shutdown path (or an equivalent `tokio::signal::ctrl_c` + coor cancellation covering SIGINT/SIGTERM and Windows Ctrl-C/Ctrl-Break) while still guaranteeing the terminal is restored on the way out. +- CERSEI: `append_system_prompt()` function is dead unless routed through the separate build_system_prompt() composer. + ## Fixes +- [x] No GitHub Action available. + +- [x] No use of Cersei workflows to chain multiple prompts together. + - [x] No support for Fireworks AI. - [x] Remove the `Claude -p` executor. diff --git a/Cargo.lock b/Cargo.lock index ccd882a..4bd5f94 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,17 +17,6 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" -[[package]] -name = "ahash" -version = "0.7.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "891477e0c6a8957309ee5c45a6368af3ae14bb510732d2684ffa19af310920f9" -dependencies = [ - "getrandom 0.2.17", - "once_cell", - "version_check", -] - [[package]] name = "ahash" version = "0.8.12" @@ -132,24 +121,24 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "arc-swap" -version = "1.9.1" +version = "1.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207" +checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b" dependencies = [ "rustversion", ] [[package]] name = "arrayvec" -version = "0.7.7" +version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f02882884d3e1bc524fb12c79f107f6ad0e1cfd498c536ffb494301740995dfe" +checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" [[package]] name = "async-stream" @@ -261,9 +250,9 @@ dependencies = [ [[package]] name = "aws-lc-rs" -version = "1.17.0" +version = "1.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00" +checksum = "4342d8937fc7e5dd9b1c60292261c0670c882a2cd1719cfc11b1af41731e32ad" dependencies = [ "aws-lc-sys", "zeroize", @@ -271,14 +260,15 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.41.0" +version = "0.42.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4" +checksum = "6d9ceb1da931507a12f4fccea479dccd00da1943e1b4ae72d8e502d707361444" dependencies = [ "cc", "cmake", "dunce", "fs_extra", + "pkg-config", ] [[package]] @@ -334,9 +324,9 @@ dependencies = [ [[package]] name = "aws-sdk-cloudwatch" -version = "1.116.0" +version = "1.118.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59438bb2a3863a686277a8461fc6cc271f49c99379ed994fe6babb0584aa998d" +checksum = "6160d48fd0152d6835eda30926c86e5c4842da8be2dd80b41b7afd6076904e34" dependencies = [ "arc-swap", "aws-credential-types", @@ -363,9 +353,9 @@ dependencies = [ [[package]] name = "aws-sdk-lambda" -version = "1.129.0" +version = "1.130.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a5ff6989897073d319e106b22e0ee9abbb2c95bdc5227f5b8fcec64e2a95cfa" +checksum = "33489208bd2568aa0189cdb640343d56d6bc7fd5ba01b9e1e34d4c2a4a2bda6d" dependencies = [ "arc-swap", "aws-credential-types", @@ -1005,7 +995,7 @@ checksum = "4f4c707c6a209cbe82d10abd08e1ea8995e9ea937d2550646e02798948992be0" [[package]] name = "cersei-agent" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "anyhow", "async-trait", @@ -1032,7 +1022,7 @@ dependencies = [ [[package]] name = "cersei-compression" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "anyhow", "once_cell", @@ -1046,7 +1036,7 @@ dependencies = [ [[package]] name = "cersei-embeddings" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "async-trait", "futures", @@ -1062,7 +1052,7 @@ dependencies = [ [[package]] name = "cersei-hooks" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "async-trait", "cersei-types", @@ -1074,7 +1064,7 @@ dependencies = [ [[package]] name = "cersei-lsp" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "dashmap", "serde", @@ -1088,7 +1078,7 @@ dependencies = [ [[package]] name = "cersei-mcp" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "async-trait", "cersei-types", @@ -1102,7 +1092,7 @@ dependencies = [ [[package]] name = "cersei-memory" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "async-trait", "cersei-types", @@ -1120,7 +1110,7 @@ dependencies = [ [[package]] name = "cersei-provider" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "async-trait", "base64 0.22.1", @@ -1140,7 +1130,7 @@ dependencies = [ [[package]] name = "cersei-tools" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "async-trait", "base64 0.22.1", @@ -1184,7 +1174,7 @@ dependencies = [ [[package]] name = "cersei-types" version = "0.2.6" -source = "git+https://github.com/wack/cersei.git?branch=trunk#1045e58a0a127ef798453adcc87c2a6f667170b9" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" dependencies = [ "anyhow", "base64 0.22.1", @@ -1196,6 +1186,27 @@ dependencies = [ "uuid", ] +[[package]] +name = "cersei-workflows" +version = "0.2.6" +source = "git+https://github.com/wack/cersei.git?branch=trunk#e75278d2d22d67674c154512de6d43c4161139bf" +dependencies = [ + "async-trait", + "cersei-agent", + "cersei-tools", + "cersei-types", + "chrono", + "dashmap", + "futures", + "parking_lot", + "schemars 0.8.22", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "uuid", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -1370,7 +1381,7 @@ dependencies = [ "jetscii", "phf 0.13.1", "phf_codegen 0.13.1", - "rustc-hash 2.1.2", + "rustc-hash 2.1.3", "smallvec", "typed-arena", ] @@ -2402,9 +2413,6 @@ name = "hashbrown" version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" -dependencies = [ - "ahash 0.7.8", -] [[package]] name = "hashbrown" @@ -2635,9 +2643,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hybrid-array" -version = "0.4.12" +version = "0.4.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9155a582abd142abc056962c29e3ce5ff2ad5469f4246b537ed42c5deba857da" +checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" dependencies = [ "typenum", ] @@ -2752,7 +2760,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.5.10", + "socket2 0.6.4", "system-configuration 0.7.0", "tokio", "tower-service", @@ -2895,9 +2903,9 @@ dependencies = [ [[package]] name = "ignore" -version = "0.4.26" +version = "0.4.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b915661dd01db3f05050265b2477bcc6527b3792388e2749b41623cc592be67d" +checksum = "fe112b004901c62c2faa11f4f75e9864e0cc5af8da71c9115d184a3aa888749f" dependencies = [ "crossbeam-deque", "globset", @@ -2960,9 +2968,9 @@ dependencies = [ [[package]] name = "inotify-sys" -version = "0.1.5" +version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb" +checksum = "9ea94e891b3606826e9c998be69ddca42247dad8ad50b1649a5cb7e1c9ae06fd" dependencies = [ "libc", ] @@ -3147,9 +3155,9 @@ checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" [[package]] name = "libredox" -version = "0.1.17" +version = "0.1.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f02ab6bace2054fb888a3c16f990117b579d14a3088e472d63c6011fa185c9d3" +checksum = "c943259e342f1e06ff2da7a83eabdfe7f92ce10262688dbf1895ff0b3e6e4652" dependencies = [ "libc", ] @@ -3456,7 +3464,7 @@ dependencies = [ [[package]] name = "multitool" -version = "0.5.0" +version = "0.6.0" dependencies = [ "async-stream", "async-trait", @@ -3474,6 +3482,7 @@ dependencies = [ "cersei-provider", "cersei-tools", "cersei-types", + "cersei-workflows", "chrono", "clap 4.6.1", "comrak", @@ -3637,9 +3646,9 @@ dependencies = [ [[package]] name = "num-bigint" -version = "0.4.6" +version = "0.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9" +checksum = "c863e9ab5e7bf9c99ba75e1050f1e4d624ae87ed3532d6238ffbdc7b585dbbe6" dependencies = [ "num-integer", "num-traits", @@ -4042,7 +4051,7 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dbe55bddb694583a9db101e5ae5b31f570f2ccce312ac7d64c2e4a430510c4b3" dependencies = [ - "ahash 0.8.12", + "ahash", "async-trait", "blake2", "bytes", @@ -4076,7 +4085,7 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51dbd9509e3bb25a699bee76ba1befbffb4e733694d7e682d4bfe35a1a48cbb4" dependencies = [ - "ahash 0.8.12", + "ahash", "async-trait", "brotli", "bytes", @@ -4110,7 +4119,7 @@ dependencies = [ "serde", "serde_yaml", "sfv", - "socket2 0.5.10", + "socket2 0.6.4", "strum", "strum_macros", "thread_local", @@ -4178,7 +4187,7 @@ dependencies = [ "pingora-http", "pingora-ketama", "pingora-runtime", - "rand 0.10.1", + "rand 0.10.2", "tokio", ] @@ -4189,9 +4198,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9bb8f0df84b4b9afd63742c78e6c4b39413554f857e7d41502825e4ff9798e3e" dependencies = [ "arrayvec", - "hashbrown 0.12.3", + "hashbrown 0.17.1", "parking_lot", - "rand 0.10.1", + "rand 0.10.2", ] [[package]] @@ -4438,9 +4447,9 @@ dependencies = [ "pin-project-lite", "quinn-proto", "quinn-udp", - "rustc-hash 2.1.2", + "rustc-hash 2.1.3", "rustls 0.23.41", - "socket2 0.5.10", + "socket2 0.6.4", "thiserror 2.0.18", "tokio", "tracing", @@ -4458,7 +4467,7 @@ dependencies = [ "lru-slab", "rand 0.9.4", "ring", - "rustc-hash 2.1.2", + "rustc-hash 2.1.3", "rustls 0.23.41", "rustls-pki-types", "slab", @@ -4477,7 +4486,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.4", "tracing", "windows-sys 0.60.2", ] @@ -4526,9 +4535,9 @@ dependencies = [ [[package]] name = "rand" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ "chacha20", "getrandom 0.4.3", @@ -4886,9 +4895,9 @@ checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" [[package]] name = "rustc-hash" -version = "2.1.2" +version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" [[package]] name = "rustc_version" @@ -4975,9 +4984,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.14.1" +version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" +checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" dependencies = [ "web-time", "zeroize", @@ -6017,9 +6026,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.51" +version = "0.3.53" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85c17d80feb7334b40c484e45ed1a5273dfd8bfda537c3be2e74a06a6686f327" +checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50" dependencies = [ "deranged", "libc", @@ -6039,9 +6048,9 @@ checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" [[package]] name = "time-macros" -version = "0.2.30" +version = "0.2.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dcef1a61bdb119096e153208ec5cbec23944ce8bca13be5c7f60c634f7403935" +checksum = "c431b87111666e491a90baa837f914fb45cd5dc3c268591b0220ff5057f2085f" dependencies = [ "num-conv", "time-core", diff --git a/Cargo.toml b/Cargo.toml index 9cc1ea0..66d89c0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "multitool" -version = "0.5.0" +version = "0.6.0" edition = "2024" default-run = "multi" repository = "https://github.com/wack/multitool" @@ -92,6 +92,7 @@ cersei-agent = { git = "https://github.com/wack/cersei.git", branch = "trunk" } cersei-memory = { git = "https://github.com/wack/cersei.git", branch = "trunk" } cersei-tools = { git = "https://github.com/wack/cersei.git", branch = "trunk" } cersei-types = { git = "https://github.com/wack/cersei.git", branch = "trunk" } +cersei-workflows = { git = "https://github.com/wack/cersei.git", branch = "trunk" } textwrap = "0.16" [workspace.dependencies] diff --git a/src/checks/execution.rs b/src/checks/execution.rs index 253168e..ec799b4 100644 --- a/src/checks/execution.rs +++ b/src/checks/execution.rs @@ -1,31 +1,37 @@ //! The execution phase (M5; in-process rework in MULTI-1367; actor rework in -//! MULTI-1368). +//! MULTI-1368; cersei-workflows fan-out in MULTI-1372). //! -//! [`ExecutionActor`] receives one [`CheckDiscovered`] per validated check and -//! **immediately** offloads the (long-running) agent run onto a spawned task — -//! never `await`-ing it inline, since a Kameo actor processes one message at a -//! time to completion and awaiting here would serialize the pipeline. Each task +//! [`ExecutionActor`] **buffers** each [`CheckDiscovered`] job as discovery +//! streams it in, then — on the [`DiscoveryComplete`] sentinel — dispatches every +//! buffered check through a cersei **`foreach` workflow** whose bounded +//! concurrency is `cfg.concurrency`. The workflow replaces the hand-rolled +//! semaphore + per-check `tokio::spawn` + retry-self-message machinery: the engine +//! runs one `run-check` step per check, caps how many run at once, and each step //! creates a CoW sandbox, runs the check's agent via the injected -//! [`CheckExecutor`] (which returns the verdict inline), drops the sandbox, and -//! either forwards a [`CheckCompleted`] to reporting or — if the agent did not -//! report — asks the actor to [`RetryCheck`] it. +//! [`CheckExecutor`] (which returns the verdict inline), drops the sandbox, +//! harvests the attempt's trace, and forwards a [`CheckCompleted`] to reporting. //! -//! Bounded concurrency is enforced by a shared [`Semaphore`] acquired *inside* -//! each task (the semaphore is the cap, not the mailbox depth — decision #5). -//! Retries (`cfg.max_attempts`) are reframed as self-messages. +//! A check whose agent never reports a verdict is retried **in place** (up to +//! `cfg.max_attempts`) inside the step, escalating the attempt number so the +//! executor can vary temperature/instructions — the same retry policy the old +//! `RetryCheck` self-message implemented, now a plain loop. Buffering until the +//! sentinel is behaviorally equivalent to the old burst dispatch: discovery +//! parse-gates the whole suite before streaming a single check, so all jobs are +//! always known by the time the workflow starts. use std::path::{Path, PathBuf}; use std::sync::Arc; +use cersei_workflows::{FnStep, RunStatus, StepRegistry, Workflow, WorkflowBuilder}; use kameo::Actor; use kameo::actor::ActorRef; use kameo::message::{Context, Message}; use miette::Result; -use tokio::sync::Semaphore; +use serde_json::{Value, json}; use crate::checks::executor::{AgentOutcome, CheckExecutor, CheckReport}; use crate::checks::messages::{ - CheckCompleted, CheckDiscovered, CheckJob, DiscoveryComplete, ExecutionComplete, RetryCheck, + CheckCompleted, CheckDiscovered, CheckJob, DiscoveryComplete, ExecutionComplete, }; use crate::checks::model::{Check, CheckId, CheckOutcome, Verdict}; use crate::checks::presenter::{PresenterActor, UiEvent}; @@ -33,14 +39,14 @@ use crate::checks::reporting::ReportingActor; use crate::checks::sandbox::Sandbox; use crate::checks::trace_archive::{TraceCollector, TraceEntry}; -/// The execution actor: turns a stream of discovered checks into a stream of -/// completed checks, fanning agent runs out onto bounded background tasks. +/// The execution actor: buffers the discovered checks, then fans them out +/// through a bounded-concurrency cersei `foreach` workflow. pub(crate) struct ExecutionActor { executor: Arc, sandbox: Arc, working_dir: PathBuf, - /// The concurrency cap, shared into every spawned task. - semaphore: Arc, + /// The bounded-concurrency cap handed to the `foreach` workflow (≥1). + concurrency: usize, /// How many times to (re)run a check whose agent fails to report (≥1). max_attempts: usize, /// The downstream reporting actor. @@ -48,10 +54,13 @@ pub(crate) struct ExecutionActor { /// The display-only presenter, told of each check's lifecycle milestones. presenter: ActorRef, /// Opt-in sink for per-execution session traces (`multi check - /// --trace-archive`). `None` disables capture. Shared across the spawned - /// per-attempt tasks; every attempt (including retries) pushes its trace - /// here before signalling completion downstream. + /// --trace-archive`). `None` disables capture. Shared across the concurrent + /// `run-check` steps; every attempt (including retries) pushes its trace here + /// before signalling completion downstream. trace_collector: Option>, + /// Checks buffered as discovery streams them, drained into the workflow when + /// the [`DiscoveryComplete`] sentinel arrives. + jobs: Vec, } impl Actor for ExecutionActor { @@ -83,104 +92,84 @@ impl ExecutionActor { executor, sandbox, working_dir, - semaphore: Arc::new(Semaphore::new(concurrency.max(1))), + concurrency: concurrency.max(1), max_attempts: max_attempts.max(1), reporting, presenter, trace_collector, + jobs: Vec::new(), } } - /// Offload one attempt of `job` onto a background task. Returns immediately, - /// keeping the actor mailbox responsive; the permit is acquired *inside* the - /// task so the semaphore — not the mailbox — is the concurrency cap. - fn dispatch(&self, ctx: &mut Context, job: CheckJob, attempt: usize) { + /// Run every buffered check through a single-node `foreach` workflow whose + /// bounded concurrency is `self.concurrency`. Each array element is a `{ id }` + /// pointer into `jobs`; the `run-check` step looks the job up, drives its + /// (possibly retried) agent run, and `tell`s reporting directly — so no domain + /// type ever has to cross the workflow's JSON boundary. Awaited inline: this + /// is the actor's last message, so blocking its handler until every check + /// settles is exactly the intended barrier and needs no detached tasks. + async fn run_checks(&self, jobs: Vec) { + let count = jobs.len(); + let jobs = Arc::new(jobs); + let executor = self.executor.clone(); let sandbox = self.sandbox.clone(); - let semaphore = self.semaphore.clone(); let working_dir = self.working_dir.clone(); let reporting = self.reporting.clone(); let presenter = self.presenter.clone(); let trace_collector = self.trace_collector.clone(); - let me = ctx.actor_ref().clone(); - let id = job.id; - - tokio::spawn(async move { - // Acquire the permit inside the task. If the semaphore was closed the - // pipeline is tearing down, so just drop the work. - let Ok(_permit) = semaphore.acquire_owned().await else { - return; - }; - - // Permit acquired ⇒ the agent is about to run: mark the check Running. - let _ = presenter.tell(UiEvent::CheckStarted { id }).await; - - let mut result = run_one( - executor, - sandbox, - job.id, - job.check.clone(), - &working_dir, - attempt, - ) - .await; - - // Harvest this attempt's trace *before* signalling completion, so it - // is collected even for retried attempts (whose outcome never reaches - // reporting) and is race-free with run finalization (the collector is - // populated before the `tell` that could trigger it). - if let Some(collector) = &trace_collector - && let Some(bytes) = result.as_mut().ok().and_then(|o| o.trace_jsonl.take()) - { - collector.push(TraceEntry { - req_index: job.req_index, - req_title: job.req_title.clone(), - check_id: job.id, - check_title: job.check.title.clone(), - attempt, - bytes, - }); - } - - if has_verdict(Some(&result)) { - // The agent reported: reconcile, surface the verdict to the - // presenter, and forward straight to reporting. - let outcome = reconcile(Some(&result), &job.check.title); - let _ = presenter - .tell(UiEvent::CheckSettled { - id, - outcome: outcome.clone(), - }) + let max_attempts = self.max_attempts; + + let registry = StepRegistry::new(); + registry.register(Arc::new(FnStep::new( + "run-check", + move |input: Value, _ctx| { + let executor = executor.clone(); + let sandbox = sandbox.clone(); + let working_dir = working_dir.clone(); + let reporting = reporting.clone(); + let presenter = presenter.clone(); + let trace_collector = trace_collector.clone(); + let jobs = jobs.clone(); + async move { + let id = input.get("id").and_then(Value::as_u64).unwrap_or(0) as usize; + let job = jobs[id].clone(); + execute_check_job( + executor, + sandbox, + &working_dir, + &presenter, + &reporting, + trace_collector.as_ref(), + max_attempts, + job, + ) .await; - if let Err(err) = reporting.tell(CheckCompleted { job, outcome }).await { - tracing::debug!(?err, "reporting actor unavailable for completed check"); + Ok(json!({ "id": id })) } - } else if let Err(err) = me - .tell(RetryCheck { - job, - attempt, - last: result, - }) - .await - { - tracing::debug!(?err, "execution actor unavailable for retry"); + }, + ))); + + let def = WorkflowBuilder::new("checks") + .foreach("run-check", self.concurrency) + .commit(); + let wf = match Workflow::compile(def, ®istry) { + Ok(wf) => wf, + Err(err) => { + tracing::error!(?err, "failed to compile the checks workflow"); + return; } - }); - } + }; - /// Forward a terminal (attempts-exhausted, no-verdict) check to reporting as - /// an errored outcome. - async fn finish_errored(&self, job: CheckJob, last: Result) { - let outcome = reconcile(Some(&last), &job.check.title); - let _ = self - .presenter - .tell(UiEvent::CheckSettled { - id: job.id, - outcome: outcome.clone(), - }) - .await; - if let Err(err) = self.reporting.tell(CheckCompleted { job, outcome }).await { - tracing::debug!(?err, "reporting actor unavailable for errored check"); + let items: Vec = (0..count).map(|id| json!({ "id": id })).collect(); + match wf.start(Value::Array(items)).await { + Ok(result) if result.status == RunStatus::Success => {} + Ok(result) => tracing::warn!( + status = ?result.status, + error = ?result.error, + "checks workflow did not run to success" + ), + Err(err) => tracing::error!(?err, "checks workflow failed to run"), } } } @@ -188,33 +177,9 @@ impl ExecutionActor { impl Message for ExecutionActor { type Reply = (); - async fn handle(&mut self, msg: CheckDiscovered, ctx: &mut Context) -> Self::Reply { - self.dispatch(ctx, msg.job, 1); - } -} - -impl Message for ExecutionActor { - type Reply = (); - - async fn handle(&mut self, msg: RetryCheck, ctx: &mut Context) -> Self::Reply { - let RetryCheck { job, attempt, last } = msg; - if attempt < self.max_attempts { - tracing::info!( - check = job.id, - attempt = attempt + 1, - "retrying check whose agent did not report" - ); - let _ = self - .presenter - .tell(UiEvent::CheckRetrying { - id: job.id, - attempt: attempt as u32, - }) - .await; - self.dispatch(ctx, job, attempt + 1); - } else { - self.finish_errored(job, last).await; - } + async fn handle(&mut self, msg: CheckDiscovered, _ctx: &mut Context) -> Self::Reply { + // Buffer; the whole set is dispatched at once on `DiscoveryComplete`. + self.jobs.push(msg.job); } } @@ -223,16 +188,25 @@ impl Message for ExecutionActor { async fn handle( &mut self, - msg: DiscoveryComplete, + _msg: DiscoveryComplete, _ctx: &mut Context, ) -> Self::Reply { + // Drain the buffer and fan every check out through the workflow (awaited + // inline). Its `run-check` steps `tell` reporting one `CheckCompleted` + // each as they settle. + let jobs = std::mem::take(&mut self.jobs); + let total = jobs.len(); + if !jobs.is_empty() { + self.run_checks(jobs).await; + } + // Tell reporting how many completed checks to expect. Reporting gates - // finalization on its own count, so this is race-free regardless of - // whether agents have settled yet. + // finalization on its own fold count, so sending this after the workflow + // (every `CheckCompleted` already delivered) still finalizes correctly. if let Err(err) = self .reporting .tell(ExecutionComplete { - total_checks: msg.total_checks, + total_checks: total, }) .await { @@ -241,6 +215,87 @@ impl Message for ExecutionActor { } } +/// Drive one check to a terminal outcome: run its agent, retrying in place (up to +/// `max_attempts`) while the agent fails to report, then `tell` reporting the +/// reconciled [`CheckCompleted`]. Fires the presenter's lifecycle events in the +/// same order the old spawn+self-message path did: +/// `CheckStarted, [CheckRetrying, CheckStarted]*, CheckSettled`. +#[allow(clippy::too_many_arguments)] +async fn execute_check_job( + executor: Arc, + sandbox: Arc, + working_dir: &Path, + presenter: &ActorRef, + reporting: &ActorRef, + trace_collector: Option<&Arc>, + max_attempts: usize, + job: CheckJob, +) { + let id = job.id; + let mut attempt = 1; + loop { + // The agent is about to run: mark the check Running. + let _ = presenter.tell(UiEvent::CheckStarted { id }).await; + + let mut result = run_one( + executor.clone(), + sandbox.clone(), + job.id, + job.check.clone(), + working_dir, + attempt, + ) + .await; + + // Harvest this attempt's trace *before* signalling completion, so it is + // collected even for retried attempts (whose outcome never reaches + // reporting) and is race-free with run finalization. + if let Some(collector) = trace_collector + && let Some(bytes) = result.as_mut().ok().and_then(|o| o.trace_jsonl.take()) + { + collector.push(TraceEntry { + req_index: job.req_index, + req_title: job.req_title.clone(), + check_id: job.id, + check_title: job.check.title.clone(), + attempt, + bytes, + }); + } + + // Reported a verdict, or attempts exhausted: reconcile, surface to the + // presenter, and forward the terminal outcome to reporting. + if has_verdict(Some(&result)) || attempt >= max_attempts { + let outcome = reconcile(Some(&result), &job.check.title); + let _ = presenter + .tell(UiEvent::CheckSettled { + id, + outcome: outcome.clone(), + }) + .await; + if let Err(err) = reporting.tell(CheckCompleted { job, outcome }).await { + tracing::debug!(?err, "reporting actor unavailable for settled check"); + } + return; + } + + // No verdict and attempts remain: retry with an escalated attempt number + // (so the executor can vary temperature/instructions). + tracing::info!( + check = id, + attempt = attempt + 1, + "retrying check whose agent did not report" + ); + let _ = presenter + .tell(UiEvent::CheckRetrying { + id, + attempt: attempt as u32, + }) + .await; + attempt += 1; + } +} + /// Whether an agent outcome carries a reported verdict. fn has_verdict(outcome: Option<&Result>) -> bool { matches!(outcome, Some(Ok(o)) if o.has_verdict()) diff --git a/src/checks/messages.rs b/src/checks/messages.rs index 14e1aca..d1ce25d 100644 --- a/src/checks/messages.rs +++ b/src/checks/messages.rs @@ -13,9 +13,6 @@ use std::path::PathBuf; -use miette::Result; - -use crate::checks::executor::AgentOutcome; use crate::checks::model::{Check, CheckId, CheckOutcome}; /// One validated check, ready to run, carried end-to-end through the pipeline so @@ -48,17 +45,6 @@ pub struct DiscoveryComplete { pub total_checks: usize, } -/// Execution self-message: a spawned agent run finished without reporting a -/// verdict; re-enqueue the same check (reframing the old retry loop as messages). -pub struct RetryCheck { - pub job: CheckJob, - /// The 1-based attempt number that just finished without a verdict. - pub attempt: usize, - /// The outcome of that attempt, kept so the terminal (attempts-exhausted) - /// case can synthesize an accurate "errored" reason. - pub last: Result, -} - /// Execution -> Reporting: a single check reached a terminal verdict. pub struct CheckCompleted { pub job: CheckJob, diff --git a/src/checks/mod.rs b/src/checks/mod.rs index e3d9c5b..93c428c 100644 --- a/src/checks/mod.rs +++ b/src/checks/mod.rs @@ -17,8 +17,8 @@ //! 2. [`discovery`] — find/parse/validate `CHECKS.md` files, then stream each //! validated check downstream (strict whole-run abort on any invalid file). //! 3. [`execution`] — run each check in a CoW [`sandbox`] via a boxed -//! [`executor`], offloaded onto bounded background tasks; capture verdicts -//! through each agent's in-process judge tool. +//! [`executor`], fanned out through a bounded-concurrency cersei `foreach` +//! workflow; capture verdicts through each agent's in-process judge tool. //! 4. [`reporting`] — fold verdicts incrementally, render, and produce the exit //! code. //! @@ -177,14 +177,14 @@ fn spawn_core( /// /// Every pipeline actor is stopped this way rather than merely having its /// `ActorRef` dropped: an actor's mailbox only closes once *every* clone of its -/// `ActorRef` is gone, and [`ExecutionActor`]'s per-check background tasks each -/// hold clones of `execution`/`reporting`/`presenter` for their own lifetime -/// (see `execution::dispatch`). Relying on that implicit ref-counting to close -/// the mailbox — instead of sending an explicit `Signal::Stop`, which the actor -/// loop honors regardless of how many `ActorRef` clones are still outstanding — -/// makes teardown a race against those tasks actually finishing, rather than a -/// deterministic signal. An explicit stop is what let the presenter's shutdown -/// stay reliable; the other three actors need the same treatment. +/// `ActorRef` is gone, and the actors hold clones of one another's refs +/// (`execution` → `reporting`/`presenter`, `discovery` → all three). Relying on +/// that implicit ref-counting to close a mailbox — instead of sending an explicit +/// `Signal::Stop`, which the actor loop honors regardless of how many `ActorRef` +/// clones are still outstanding — makes teardown a race against every last ref +/// being dropped, rather than a deterministic signal. An explicit stop is what +/// let the presenter's shutdown stay reliable; the other three actors need the +/// same treatment. async fn shutdown_actor(actor: &ActorRef) { let _ = actor.stop_gracefully().await; actor.wait_for_shutdown().await;