Conversation
… SQLite store Phase 1-2 of epic #447: core types (StepMode/ErrorMode/JournalEntry with CompletableEntry invariant), DAG validation, StateStore trait + in-memory, and SqliteStore on agentflare-db-kit with append-only journal. Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
Phase 3 of epic #447: port SMG wfaas engine (DAG parallel scheduler, backoff retries, RetryIndefinitely, cancellation, graceful shutdown, event bus) fused with the durable journal — every terminal step result appended as JournalEntry::StepRun. Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
…ut+collect, variables, tokens, eviction Phase 4 of epic #447: engine executes StepMode variants (Conditional skip, Loop until/max-iterations, FanOut+Collect join via shared buffer), string-pipeline input chaining ({{input}} + output_var capture), token accounting on StepState, run-eviction cap (200), and ErrorMode::Skip turns terminal failures into skips. OpenFang test suite ported. Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
…es + complete_event Phase 5 of epic #447: Sleep/WaitEvent journaled (pending + completed entries, Restate design), in-process oneshot waiters, exactly-once complete_event with journaled pre-delivery closing the notify-before-wait race, and TTL timeout path. Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
…via journal memoization Phase 6 of epic #447: engine.recover() replays active runs over the SQLite journal, skips steps with completed entries (StepRun/Sleep/WaitEvent), and re-drives pending steps; pending Sleep re-arms idempotently; WorkflowDefinition is now Clone for re-registration. Durable-wait methods split into waits.rs to stay under the LOC gate. Crash-resume test proves a completed step never re-executes (exactly-once); racing completions resolve to one winner. Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
Phase 7 (core) of epic #447: JsonWorkflow schema compiles to engine-ready definitions with agent-prompt executors dispatching through a caller-supplied SendMessage hook (input/var templating, token accounting, retryable by default). OpenFang's four example workflows (code-review, research-and-write with conditional, brainstorm fan-out+collect, iterative-refinement loop) run as fixtures. Workspace builds with the new member. Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
… pipeline test Phase 7b: src/workflow.rs service (durable SqliteStore, shared runtime, headless-agent SendMessage hook) + mcp__flare__workflow (run/status/ complete_event/list) + agentflare workflow CLI. compile_workflow now wires OpenFang positional ordering into DAG edges (sequential chain, fan-out group -> collect). complete_event is journal-first so a different engine instance (MCP/CLI/recovery) resolves a wait. Tests: service round-trips, cross-engine event resolution, and a REAL coder->reviewer->PR pipeline that performs git/ file work in a temp repo (branch, commit, real-diff review loop, PR ref). Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
…ation Phase 7b completion: workflow service exposes async cores (no nested block_on from the daemon's async runtime) with sync wrappers for the CLI; mcp__flare__workflow handler tests cover run/status/list + error paths; real coder->reviewer->PR git flow test proves the engine drives agentflare's item-pipeline mechanics. Agentflare-Agent: 1 Agentflare-Branch: task/447 Agentflare-Item: 447
Fixes for the 8 review findings on item #447 (all except #8, documented below): - agent_send_hook now runs run_headless inside spawn_blocking within the async block, instead of synchronously before it (finding #1). - SqliteStore's StateStore impl wraps every method's rusqlite I/O in spawn_blocking so it can't stall an async executor thread (#2). - WorkflowEngine gains an optional runtime handle (with_runtime_handle); start_workflow/recover/start_cleanup_task spawn through it. src/workflow.rs now attaches WORKFLOW_RT so the MCP path (which awaits the async core directly on the daemon's own runtime) also keeps execution off that runtime, not just the CLI's block_on path (#3). - Steps whose dependency failed now get an explicit Skipped status (with a cascade to their own dependents) instead of staying stuck at Pending forever; completion signals now forward on any terminal result, not just Success/Skip, so blocked dependents actually reach the scheduler (#4). - WaitEvent waiters are now keyed per step instance (run_id:step_id:name) instead of run_id:name, so two concurrent wait_event steps sharing an event name no longer drop each other's waiter; complete_event wakes every matching waiter (#5). - cleanup_old_workflows no longer panics via .expect() on a lock/query failure; it logs and returns 0 like its siblings (#6). - write_state now does targeted UPSERTs of step_state/run_vars rows instead of DELETE-all + INSERT-all on every write (#7). Finding #8 (WorkflowRunId::new() vs db_kit::ids::new_id()) intentionally left as-is: new_id() returns a nanoid String, but WorkflowRunId wraps a Uuid and is parsed/displayed as a v7 UUID throughout the MCP/CLI surface (explicitly time-ordered). Switching would require re-typing WorkflowRunId from Uuid to String across the crate and would lose the v7 time-ordering property, which is a materially different and larger change than the 'low - consistency' severity implies. Added regression tests: concurrent_wait_events_same_name_both_resolve and dependent_of_failed_step_gets_terminal_status. cargo test -p flare-workflow (55 tests) + agentflare workflow::/ mcp_server::workflow:: tests (7 tests) pass; fmt + clippy (-D warnings -A unsafe_code -A clippy::pedantic) clean workspace-wide. Agentflare-Agent: claude-code Agentflare-Branch: task/447 Agentflare-Item: 447
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 42 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughChangesWorkflow engine
Estimated code review effort: 5 (Critical) | ~120 minutes Mergeability Score: 🔴 Critical · up to Concurrent workflow updates can overwrite one another, losing step transitions, outputs, or variables; unresolved fan-out and recovery paths can also start work too early or replay timers and events incorrectly. Merge should be blocked until these correctness issues are fixed or explicitly accepted by the owner. Sequence Diagram(s)sequenceDiagram
participant Client
participant WorkflowService
participant WorkflowEngine
participant SqliteStore
participant AgentSender
Client->>WorkflowService: run workflow definition
WorkflowService->>WorkflowEngine: compile, register, and start workflow
WorkflowEngine->>SqliteStore: persist state and journal entries
WorkflowEngine->>AgentSender: dispatch step prompt
AgentSender-->>WorkflowEngine: return step output
WorkflowEngine->>SqliteStore: persist result and status
WorkflowService-->>Client: return run identifier and status
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
…rkspace-hack Replace the RUSTSEC-flagged backoff crate with a hand-rolled exponential strategy in retry.rs (base * 1.5^n, capped at max) — the crate's own randomization was redundant with apply_jitter anyway. Regenerate agentflare-workspace-hack via cargo hakari generate to pick up flare-workflow's dependency set. Agentflare-Agent: claude-code Agentflare-Branch: task/447 Agentflare-Item: 447
Agentflare-Agent: claude-code Agentflare-Branch: task/447 Agentflare-Item: 447
getappz
changed the title
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 17
🧹 Nitpick comments (8)
src/workflow.rs (2)
142-142: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReplace
eprintln!with structured logging.This service path runs inside the MCP daemon and the CLI.
eprintln!writes to the daemon's stderr with no level or target. The crate already usestracinginflare-workflow; usetracing::info!here for consistent observability.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/workflow.rs` at line 142, Replace the eprintln! call in the workflow run-start path with tracing::info!, preserving the run_id and name fields in the structured event. Use the existing tracing integration in flare-workflow rather than writing directly to stderr.
363-363: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe exact journal-entry count makes the test brittle.
assert_eq!(status["journal_tail"].as_array().unwrap().len(), 4)couples the test to the current journaling granularity. Any added journal entry type breaks it without a behavior regression. Assert a lower bound, or assert on the entry types you require.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/workflow.rs` at line 363, Update the journal_tail assertion in the relevant workflow test to avoid requiring an exact entry count of 4. Assert only the minimum number of entries needed by the test, or validate the required journal entry types while allowing additional entries.crates/flare-workflow/src/variables.rs (1)
9-15: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winSequential replacement makes expansion order-dependent.
The loop iterates a
HashMap, so replacement order varies between runs. If a variable value itself contains{{other}}, that text may or may not be expanded, depending on iteration order. A single scan of the template would make the result deterministic and would prevent substituted content from being re-expanded.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/flare-workflow/src/variables.rs` around lines 9 - 15, Update expand_variables to perform one deterministic scan of the original template, resolving each placeholder from input or vars as it is encountered. Do not repeatedly replace the evolving result, so placeholder text inside substituted values remains literal and HashMap iteration order cannot affect expansion.src/mcp_server/workflow.rs (1)
18-63: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winError classification is inverted in two arms.
In the
runarm, invalid JSON and compile failures are caller-fixable, but Line 30 maps every error tointernal_error. In thestatusandcomplete_eventarms, Lines 43 and 56 map every error toinvalid_params, including store I/O failures that the caller cannot fix. Split the mapping so parse and validation failures returninvalid_paramsand store failures returninternal_error.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/mcp_server/workflow.rs` around lines 18 - 63, Correct error classification in the workflow dispatch arms: update run_workflow_json_async handling in "run" so JSON parsing and validation/compile failures map to invalid_params while store or execution failures map to internal_error; update workflow_status_async and complete_workflow_event_async in "status" and "complete_event" so caller-invalid inputs remain invalid_params but persistence or I/O failures map to internal_error. Use the underlying error variants or classification exposed by the workflow APIs rather than mapping every error identically.crates/flare-workflow/src/sqlite_store.rs (2)
55-101: 🚀 Performance & Scalability | 🔵 TrivialConsider an index for the status and cleanup queries.
list_active(line 349) filters onstatus, andcleanup_old_workflows(line 416) filters onstatusplusupdated_at. Both scanworkflow_runs. A composite index keeps those queries cheap as completed runs accumulate.CREATE INDEX idx_workflow_runs_status_updated ON workflow_runs (status, updated_at);Add it as a new
M::upmigration so existing databases pick it up.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/flare-workflow/src/sqlite_store.rs` around lines 55 - 101, Add a new M::up migration in migrations() that creates the composite index idx_workflow_runs_status_updated on workflow_runs(status, updated_at), ensuring existing databases receive it through the migration system.
405-433: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winReport
delete_statefailures in both cleanup paths.Line 424 and line 460 discard the
delete_stateresult withlet _ =. A failed delete leaves the run row and its journal,step_state, andrun_varsrows in place.cleanup_old_workflowsstill counts that run as deleted, so the returned count overstates the work done.cleanup_if_terminalstill returnstrue.Log the failure and exclude the run from the count.
♻️ Proposed change for `cleanup_old_workflows`
- for id in &ids { - let _ = Self::delete_state(&conn, id); - } - Ok(ids.len()) + let mut deleted = 0usize; + for id in &ids { + match Self::delete_state(&conn, id) { + Ok(()) => deleted += 1, + Err(e) => tracing::warn!(run_id = %id, error = ?e, "delete_state failed during cleanup"), + } + } + Ok(deleted)Also applies to: 439-469
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/flare-workflow/src/sqlite_store.rs` around lines 405 - 433, Handle errors from Self::delete_state in both cleanup_old_workflows and cleanup_if_terminal instead of discarding them. Log each failure with the affected workflow ID, exclude failed deletions from cleanup_old_workflows’s returned count, and ensure cleanup_if_terminal returns false when deletion fails while preserving successful behavior.crates/flare-workflow/src/executor.rs (1)
18-22: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAlign the default
is_retryablewithretry::Retryable.The default returns
truefor every error, includingWorkflowError::ShuttingDownandWorkflowError::NotFound.crates/flare-workflow/src/retry.rsclassifies both as non-retryable, andengine.rscallsstep.executor.is_retryable(&e)rather than theRetryableimpl. The engine therefore retries terminal errors during shutdown. Delegate the default to the shared classifier.♻️ Proposed change
fn is_retryable(&self, _error: &WorkflowError) -> bool { - true + crate::retry::Retryable::is_retryable(_error) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/flare-workflow/src/executor.rs` around lines 18 - 22, Update the default Executor::is_retryable implementation to delegate to the shared retry::Retryable classifier for WorkflowError instead of unconditionally returning true, while preserving custom overrides and the existing non-retryable behavior for ShuttingDown and NotFound.crates/flare-workflow/tests/engine_test.rs (1)
314-341: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThis test does not prove that cancellation stops execution.
#[tokio::test]uses the current-thread runtime, and the step callsstd::thread::sleep(3000ms), which blocks that runtime. The test's owntokio::time::sleep(50ms)cannot resume until the step returnsOk(StepResult::Success).cancel_workflowthen setsCancelledover an already finished run, so the assertion passes for the wrong reason.Use a multi-thread runtime and an async sleep, and assert that the step did not reach
Succeeded. The same blocking pattern indag_parallel_join_orders_dependents(Lines 90-93) removes any real parallelism from that test.♻️ Proposed change
-#[tokio::test] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cancellation_stops_execution() { let engine = build_engine(); - let wf = WorkflowDefinition::new("wf", "wf").add_step(step("slow", |_| { - std::thread::sleep(Duration::from_millis(3000)) - })); + let wf = WorkflowDefinition::new("wf", "wf").add_step(StepDefinition::new( + "slow", + "slow", + Arc::new(FunctionStep::new(|_: &mut WorkflowContext<Ctx>| { + Box::pin(async { + tokio::time::sleep(Duration::from_millis(3000)).await; + Ok(StepResult::Success) + }) + })), + ));Then also assert the step state:
assert_ne!( state.step_states[&StepId::new("slow")].status, StepStatus::Succeeded );🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/flare-workflow/tests/engine_test.rs` around lines 314 - 341, Update cancellation_stops_execution to use a multi-thread Tokio runtime and replace the blocking std::thread::sleep in the slow step with an async sleep, allowing cancellation to occur while execution is in progress. After cancellation, retain the Cancelled run-status assertion and also verify the slow step’s StepStatus is not Succeeded. Apply the same non-blocking sleep adjustment to dag_parallel_join_orders_dependents so its parallel execution remains real.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/flare-workflow/README.md`:
- Around line 22-37: The README usage block is not a compilable Rust doctest.
Update the example around WorkflowDefinition and WorkflowEngine with hidden
imports and setup for Arc, Duration, StepResult, WorkflowId, and Ctx, wrap the
statements in an async Result-returning function, and replace the placeholder
executor argument with a valid expression; alternatively mark the block as
non-Rust text.
In `@crates/flare-workflow/src/definition.rs`:
- Around line 282-296: Update WorkflowDefinition::validate to detect duplicate
StepId values before dependency validation and return an appropriate
ValidationError for the repeated identifier. Ensure every step id is unique so
steps_map and the engine’s step-count/progress tracking remain consistent.
In `@crates/flare-workflow/src/engine.rs`:
- Around line 1032-1038: Update the Skip handling in execute_step_with_retry so
a StepResult::Skip persists StepStatus::Skipped and completed_at before
returning, including the corresponding alternate path around the second noted
location. Keep the existing context update behavior for non-Skip results and
ensure both skip paths leave the API-visible step state terminal.
- Around line 711-721: Replace the per-step tokio::spawn call in the workflow
execution loop with self.spawn, preserving the existing async closure and
captured step execution state so agent calls and SQLite I/O run on the
executor’s configured runtime.
- Around line 487-496: Update the WaitEvent branch of the memoized match to mark
a journal entry as completed only when its resolved result is successful,
matching the existing StepRun handling below. Preserve name-based matching and
ensure EntryResult::Failure, including timeout failures from execute_wait_event,
does not memoize the step as completed.
- Around line 597-621: Separate dependency-blocked steps from intentionally
skipped steps in the workflow tracker; do not insert them into t.skipped. Update
the dependency readiness and propagation logic around are_dependencies_satisfied
and the blocked-step handling in the execution flow so a blocked ancestor keeps
all transitive dependents blocked, marking each as Skipped without launching
them or evaluating stale input.
In `@crates/flare-workflow/src/journal.rs`:
- Around line 16-26: The append function must return the sequence assigned by
its own INSERT rather than querying it afterward. Change the INSERT in append to
use SQLite RETURNING seq, read and return that value directly, and remove
next_seq or rename it to last_seq only if another caller still requires the
highest written sequence.
In `@crates/flare-workflow/src/json.rs`:
- Around line 74-76: Update default_timeout to match the 600-second run_headless
budget used by agent_send_hook, so JSON steps without an explicit timeout do not
cancel legitimate agent calls prematurely.
- Around line 130-199: Update the dependency derivation around the fan_group
handling so the first non-fan_out step after a fan-out group depends on every
member in fan_group, including when that step is not Collect. Preserve Collect’s
existing join behavior, then clear the group only after its members have been
used, and ensure later steps chain from the group-closing step.
In `@crates/flare-workflow/src/sqlite_store.rs`:
- Around line 291-328: Make SqliteStore::update atomic per run_id by acquiring a
shared per-run async lock before loading state and holding it through mutation
and write; ensure the lock is shared across all SqliteStore clones. Reuse this
same per-run lock in deletion and cleanup paths so they cannot race with
updates, while preserving the existing SQLite blocking boundaries and state
serialization behavior.
In `@crates/flare-workflow/src/store.rs`:
- Around line 222-228: Update InMemoryStore::journal to return Ok(Vec::new())
when self.journals has no entry for the given run_id, instead of propagating
WorkflowError::NotFound. Preserve cloning and returning existing journal entries
unchanged so it matches journal::read’s empty-log behavior.
In `@crates/flare-workflow/src/types.rs`:
- Around line 279-283: Update the WaitEvent variant and its journal
creation/replay paths to carry and match a step_id, preventing completion from
being shared by waits with the same name. Follow the existing StepRun and Sleep
identity handling, and add compatibility decoding for persisted rows that lack
step_id; update engine.rs, waits.rs, and all affected tests accordingly.
- Around line 236-244: Update EntryResult::success and EntryResult::from_json to
return WorkflowResult<Self>, propagating serde_json::to_vec serialization errors
instead of converting them to empty successful payloads. Adjust their callers to
handle or propagate the result so failed serialization is journaled as a real
workflow error and successful entries retain the existing payload behavior.
In `@crates/flare-workflow/src/variables.rs`:
- Around line 17-19: Update the documentation comment for capture_output to
state that it records the step output by mutating the provided vars map in place
and returns (). Remove the inaccurate claim that it returns a variables map.
In `@crates/flare-workflow/src/waits.rs`:
- Around line 30-57: Update the sleep wait flow to reuse the pending Sleep
journal entry’s persisted wake_at when one already exists, instead of always
calculating a new wake_at from Utc::now(). In the logic around the journal
lookup and JournalEntry::Sleep creation, select the existing wake_at for the
matching step_id and only create a new timestamp when no pending entry exists;
use that selected timestamp for StepWaiting and tokio::time::sleep so recovery
re-arms the original deadline.
In `@crates/flare-workflow/tests/types_test.rs`:
- Around line 197-210: Update the status list in workflow_status_serde_roundtrip
to include WorkflowStatus::Paused alongside the existing variants, ensuring its
serde serialization and deserialization roundtrip is covered.
In `@src/mcp_server/types.rs`:
- Around line 1061-1065: Restrict the db_path override in the workflow request
type and workflow_impl before passing it to SqliteStore::open_file: accept only
paths within the daemon user’s ~/.agentflare directory, or ignore/reject
client-supplied paths in production while preserving test overrides. Ensure the
default ~/.agentflare/workflows.db path remains the production fallback.
---
Nitpick comments:
In `@crates/flare-workflow/src/executor.rs`:
- Around line 18-22: Update the default Executor::is_retryable implementation to
delegate to the shared retry::Retryable classifier for WorkflowError instead of
unconditionally returning true, while preserving custom overrides and the
existing non-retryable behavior for ShuttingDown and NotFound.
In `@crates/flare-workflow/src/sqlite_store.rs`:
- Around line 55-101: Add a new M::up migration in migrations() that creates the
composite index idx_workflow_runs_status_updated on workflow_runs(status,
updated_at), ensuring existing databases receive it through the migration
system.
- Around line 405-433: Handle errors from Self::delete_state in both
cleanup_old_workflows and cleanup_if_terminal instead of discarding them. Log
each failure with the affected workflow ID, exclude failed deletions from
cleanup_old_workflows’s returned count, and ensure cleanup_if_terminal returns
false when deletion fails while preserving successful behavior.
In `@crates/flare-workflow/src/variables.rs`:
- Around line 9-15: Update expand_variables to perform one deterministic scan of
the original template, resolving each placeholder from input or vars as it is
encountered. Do not repeatedly replace the evolving result, so placeholder text
inside substituted values remains literal and HashMap iteration order cannot
affect expansion.
In `@crates/flare-workflow/tests/engine_test.rs`:
- Around line 314-341: Update cancellation_stops_execution to use a multi-thread
Tokio runtime and replace the blocking std::thread::sleep in the slow step with
an async sleep, allowing cancellation to occur while execution is in progress.
After cancellation, retain the Cancelled run-status assertion and also verify
the slow step’s StepStatus is not Succeeded. Apply the same non-blocking sleep
adjustment to dag_parallel_join_orders_dependents so its parallel execution
remains real.
In `@src/mcp_server/workflow.rs`:
- Around line 18-63: Correct error classification in the workflow dispatch arms:
update run_workflow_json_async handling in "run" so JSON parsing and
validation/compile failures map to invalid_params while store or execution
failures map to internal_error; update workflow_status_async and
complete_workflow_event_async in "status" and "complete_event" so caller-invalid
inputs remain invalid_params but persistence or I/O failures map to
internal_error. Use the underlying error variants or classification exposed by
the workflow APIs rather than mapping every error identically.
In `@src/workflow.rs`:
- Line 142: Replace the eprintln! call in the workflow run-start path with
tracing::info!, preserving the run_id and name fields in the structured event.
Use the existing tracing integration in flare-workflow rather than writing
directly to stderr.
- Line 363: Update the journal_tail assertion in the relevant workflow test to
avoid requiring an exact entry count of 4. Assert only the minimum number of
entries needed by the test, or validate the required journal entry types while
allowing additional entries.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 3d602ed2-3f54-4f9e-baa8-58f9c2b7e3f3
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (31)
Cargo.tomlagentflare-workspace-hack/Cargo.tomlcrates/flare-workflow/Cargo.tomlcrates/flare-workflow/README.mdcrates/flare-workflow/src/definition.rscrates/flare-workflow/src/engine.rscrates/flare-workflow/src/events.rscrates/flare-workflow/src/executor.rscrates/flare-workflow/src/journal.rscrates/flare-workflow/src/json.rscrates/flare-workflow/src/lib.rscrates/flare-workflow/src/retry.rscrates/flare-workflow/src/sqlite_store.rscrates/flare-workflow/src/store.rscrates/flare-workflow/src/types.rscrates/flare-workflow/src/variables.rscrates/flare-workflow/src/waits.rscrates/flare-workflow/tests/engine_test.rscrates/flare-workflow/tests/examples_test.rscrates/flare-workflow/tests/recovery_test.rscrates/flare-workflow/tests/semantics_test.rscrates/flare-workflow/tests/types_test.rscrates/flare-workflow/tests/waits_test.rssrc/cli/mod.rssrc/cli/workflow.rssrc/main.rssrc/mcp_server.rssrc/mcp_server/tests/item_commit_failure_tests.rssrc/mcp_server/types.rssrc/mcp_server/workflow.rssrc/workflow.rs
Comment on lines
+22
to
+37
| ```rust | ||
| use flare_workflow::{WorkflowDefinition, StepDefinition, WorkflowEngine, InMemoryStore}; | ||
| use flare_workflow::executor::FunctionStep; | ||
|
|
||
| let wf = WorkflowDefinition::new("wf", "wf") | ||
| .add_step(StepDefinition::new("a", "a", Arc::new(FunctionStep::new(|ctx| { | ||
| ctx.output = "hello".into(); | ||
| Box::pin(async { Ok(StepResult::Success) }) | ||
| })))) | ||
| .add_step(StepDefinition::new("b", "b", /* ... */).depends_on(&["a"])); | ||
|
|
||
| let engine = WorkflowEngine::<Ctx, InMemoryStore<Ctx>>::new(); | ||
| engine.register_workflow(wf)?; | ||
| let run = engine.start_workflow(WorkflowId::new("wf"), ctx, "input".into()).await?; | ||
| engine.wait_for_completion(run, "wf", Duration::from_secs(300)).await?; | ||
| ``` |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
# Check available code-intelligence and memory tools, then inspect the README example
# and the crate metadata without executing repository code.
set -u
printf '%s\n' '--- tool availability ---'
for tool in agentflare ctx_search ctx_read ctx_callgraph ctx_compose ctx_shell; do
if command -v "$tool" >/dev/null 2>&1; then
printf '%s: available\n' "$tool"
else
printf '%s: unavailable\n' "$tool"
fi
done
printf '%s\n' '--- memory context ---'
if command -v agentflare >/dev/null 2>&1; then
agentflare memory context 2>&1 | head -200
else
printf '%s\n' 'agentflare unavailable'
fi
printf '%s\n' '--- candidate files ---'
git ls-files | grep -E '(^|/)(README\.md|Cargo\.toml)$' | grep -E 'crates/flare-workflow|^Cargo\.toml$' || true
printf '%s\n' '--- README excerpt ---'
cat -n crates/flare-workflow/README.md | sed -n '1,90p'
printf '%s\n' '--- crate manifest ---'
cat -n crates/flare-workflow/Cargo.toml | sed -n '1,180p'
printf '%s\n' '--- workspace manifest doctest settings ---'
rg -n -C 3 'doctest|documentation|flare-workflow|edition' Cargo.toml crates/flare-workflow/Cargo.tomlRepository: getappz/agentflare
Length of output: 7346
🏁 Script executed:
set -u
printf '%s\n' '--- workflow source files ---'
git ls-files crates/flare-workflow | sed -n '1,120p'
printf '%s\n' '--- public symbols and example identifiers ---'
rg -n -C 2 'pub (struct|enum|type|trait|fn)|WorkflowDefinition|StepDefinition|WorkflowEngine|InMemoryStore|FunctionStep|StepResult|WorkflowId|Ctx|start_workflow|wait_for_completion' crates/flare-workflow/src
printf '%s\n' '--- library outline ---'
if command -v ast-grep >/dev/null 2>&1; then
ast-grep outline crates/flare-workflow/src
else
printf '%s\n' 'ast-grep unavailable'
fiRepository: getappz/agentflare
Length of output: 50374
🏁 Script executed:
set -u
printf '%s\n' '--- WorkflowContext and StepResult ---'
cat -n crates/flare-workflow/src/types.rs | sed -n '315,425p'
printf '%s\n' '--- StepDefinition constructor and dependency API ---'
cat -n crates/flare-workflow/src/definition.rs | sed -n '35,125p'
cat -n crates/flare-workflow/src/definition.rs | sed -n '145,160p'
printf '%s\n' '--- WorkflowEngine constructor and workflow methods ---'
rg -n -A 24 -B 8 'pub fn new|pub fn register_workflow|pub fn start_workflow|pub async fn wait_for_completion' crates/flare-workflow/src/engine.rs
printf '%s\n' '--- existing test setup for a compiling FunctionStep ---'
rg -n -A 35 -B 10 'FunctionStep::new|WorkflowContext|struct TestData|impl WorkflowData' crates/flare-workflow/tests crates/flare-workflow/src | head -260Repository: getappz/agentflare
Length of output: 32063
🏁 Script executed:
python3 - <<'PY'
from pathlib import Path
import re
readme = Path("crates/flare-workflow/README.md").read_text()
match = re.search(r"```rust\n(.*?)\n```", readme, re.S)
assert match, "usage Rust block not found"
block = match.group(1)
ctx_source = Path("crates/flare-workflow/src/types.rs").read_text()
engine_source = Path("crates/flare-workflow/src/engine.rs").read_text()
checks = {
"uses_top_level_await": ".await" in block,
"uses_question_mark": "?" in block,
"has_placeholder_argument": "/* ... */" in block,
"declares_arc": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bArc\b", block)),
"declares_duration": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bDuration\b", block)),
"declares_step_result": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bStepResult\b", block)),
"declares_workflow_id": bool(re.search(r"\b(?:use|let|type|struct|enum)\b.*\bWorkflowId\b", block)),
"declares_ctx_type": bool(re.search(r"\b(?:struct|type|enum)\s+Ctx\b", block)),
"declares_ctx_value": bool(re.search(r"\blet\s+ctx\b", block)),
"workflow_context_has_output": bool(re.search(r"\bpub\s+output\s*:\s*String", ctx_source)),
"start_workflow_is_async": bool(re.search(r"pub\s+async\s+fn\s+start_workflow", engine_source)),
"wait_for_completion_is_async": bool(re.search(r"pub\s+async\s+fn\s+wait_for_completion", engine_source)),
}
for name, value in checks.items():
print(f"{name}={value}")
PYRepository: getappz/agentflare
Length of output: 477
Make the usage example compile as a doctest.
The block uses undeclared Arc, Duration, StepResult, WorkflowId, and Ctx. It also contains ?, top-level .await, and /* ... */ where an executor expression is required. Add valid hidden setup and an async Result-returning example function, or mark the block as non-Rust text.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/README.md` around lines 22 - 37, The README usage block
is not a compilable Rust doctest. Update the example around WorkflowDefinition
and WorkflowEngine with hidden imports and setup for Arc, Duration, StepResult,
WorkflowId, and Ctx, wrap the statements in an async Result-returning function,
and replace the placeholder executor argument with a valid expression;
alternatively mark the block as non-Rust text.
Comment on lines
+282
to
+296
| pub fn validate(&mut self) -> Result<(), ValidationError> { | ||
| let steps_map: HashMap<&StepId, &StepDefinition<D>> = | ||
| self.steps.iter().map(|s| (&s.id, s)).collect(); | ||
|
|
||
| for step in &self.steps { | ||
| for dep in step.all_dependencies() { | ||
| if !steps_map.contains_key(dep) { | ||
| return Err(ValidationError::MissingDependency { | ||
| step: step.id.clone(), | ||
| dependency: dep.clone(), | ||
| }); | ||
| } | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Add a duplicate step-id check to validate.
steps_map collapses steps that share an id, so duplicates pass validation. The engine then uses definition.steps.len() as step_count (crates/flare-workflow/src/engine.rs Line 466) but tracks progress in HashSet<StepId> collections. With a duplicate id, total_processed() can never reach step_count, so the run ends in the deadlock branch (crates/flare-workflow/src/engine.rs Lines 665-702) with the message "Workflow deadlocked: no steps ready and none running". state.step_states also loses one entry.
Reject duplicates during validation instead.
🛠️ Proposed fix
pub enum ValidationError {
/// A step depends on another step that doesn't exist.
#[error("step '{step}' depends on non-existent step '{dependency}'")]
MissingDependency { step: StepId, dependency: StepId },
+ /// Two steps share the same id.
+ #[error("duplicate step id '{0}'")]
+ DuplicateStep(StepId),
+
/// A cycle was detected in the workflow DAG.
#[error("cycle detected involving step '{0}'")]
CycleDetected(StepId),
} pub fn validate(&mut self) -> Result<(), ValidationError> {
- let steps_map: HashMap<&StepId, &StepDefinition<D>> =
- self.steps.iter().map(|s| (&s.id, s)).collect();
+ let mut steps_map: HashMap<&StepId, &StepDefinition<D>> = HashMap::new();
+ for step in &self.steps {
+ if steps_map.insert(&step.id, step).is_some() {
+ return Err(ValidationError::DuplicateStep(step.id.clone()));
+ }
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| pub fn validate(&mut self) -> Result<(), ValidationError> { | |
| let steps_map: HashMap<&StepId, &StepDefinition<D>> = | |
| self.steps.iter().map(|s| (&s.id, s)).collect(); | |
| for step in &self.steps { | |
| for dep in step.all_dependencies() { | |
| if !steps_map.contains_key(dep) { | |
| return Err(ValidationError::MissingDependency { | |
| step: step.id.clone(), | |
| dependency: dep.clone(), | |
| }); | |
| } | |
| } | |
| } | |
| pub fn validate(&mut self) -> Result<(), ValidationError> { | |
| let mut steps_map: HashMap<&StepId, &StepDefinition<D>> = HashMap::new(); | |
| for step in &self.steps { | |
| if steps_map.insert(&step.id, step).is_some() { | |
| return Err(ValidationError::DuplicateStep(step.id.clone())); | |
| } | |
| } | |
| for step in &self.steps { | |
| for dep in step.all_dependencies() { | |
| if !steps_map.contains_key(dep) { | |
| return Err(ValidationError::MissingDependency { | |
| step: step.id.clone(), | |
| dependency: dep.clone(), | |
| }); | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/src/definition.rs` around lines 282 - 296, Update
WorkflowDefinition::validate to detect duplicate StepId values before dependency
validation and return an appropriate ValidationError for the repeated
identifier. Ensure every step id is unique so steps_map and the engine’s
step-count/progress tracking remain consistent.
Comment on lines
+487
to
+496
| let memoized = match &step.mode { | ||
| StepMode::WaitEvent { name, .. } => { | ||
| // Completed by name (journal entries carry no step id). | ||
| journal.iter().any(|e| { | ||
| matches!( | ||
| e, | ||
| JournalEntry::WaitEvent { name: n, result: Some(_) } if n == name | ||
| ) | ||
| }) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
WaitEvent memoization treats a journaled failure as success.
The match only requires result: Some(_), so any resolved WaitEvent entry marks the step completed. execute_wait_event journals a timeout as EntryResult::Failure { code: 2, .. } (crates/flare-workflow/src/waits.rs Lines 204-217). After recovery, a run whose wait timed out replays as if the wait succeeded, and the run can reach Completed. The StepRun branch below already distinguishes Success from Failure; apply the same rule here.
🛠️ Proposed fix
StepMode::WaitEvent { name, .. } => {
- // Completed by name (journal entries carry no step id).
- journal.iter().any(|e| {
- matches!(
- e,
- JournalEntry::WaitEvent { name: n, result: Some(_) } if n == name
- )
- })
+ // Completed by name (journal entries carry no step id).
+ match journal.iter().rev().find(|e| {
+ matches!(e, JournalEntry::WaitEvent { name: n, result: Some(_) } if n == name)
+ }) {
+ Some(JournalEntry::WaitEvent {
+ result: Some(EntryResult::Success(_)),
+ ..
+ }) => {
+ t.completed.insert(step.id.clone());
+ true
+ }
+ Some(JournalEntry::WaitEvent { result: Some(_), .. }) => {
+ t.failed.insert(step.id.clone());
+ true
+ }
+ _ => false,
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| let memoized = match &step.mode { | |
| StepMode::WaitEvent { name, .. } => { | |
| // Completed by name (journal entries carry no step id). | |
| journal.iter().any(|e| { | |
| matches!( | |
| e, | |
| JournalEntry::WaitEvent { name: n, result: Some(_) } if n == name | |
| ) | |
| }) | |
| } | |
| let memoized = match &step.mode { | |
| StepMode::WaitEvent { name, .. } => { | |
| // Completed by name (journal entries carry no step id). | |
| match journal.iter().rev().find(|e| { | |
| matches!(e, JournalEntry::WaitEvent { name: n, result: Some(_) } if n == name) | |
| }) { | |
| Some(JournalEntry::WaitEvent { | |
| result: Some(EntryResult::Success(_)), | |
| .. | |
| }) => { | |
| t.completed.insert(step.id.clone()); | |
| true | |
| } | |
| Some(JournalEntry::WaitEvent { result: Some(_), .. }) => { | |
| t.failed.insert(step.id.clone()); | |
| true | |
| } | |
| _ => false, | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/src/engine.rs` around lines 487 - 496, Update the
WaitEvent branch of the memoized match to mark a journal entry as completed only
when its resolved result is successful, matching the existing StepRun handling
below. Preserve name-based matching and ensure EntryResult::Failure, including
timeout failures from execute_wait_event, does not memoize the step as
completed.
Comment on lines
+597
to
+621
| if !deps_blocked_indices.is_empty() { | ||
| { | ||
| let mut t = tracker.write(); | ||
| for &idx in &deps_blocked_indices { | ||
| t.skipped.insert(definition.steps[idx].id.clone()); | ||
| } | ||
| } | ||
| for &idx in &deps_blocked_indices { | ||
| let step_id = definition.steps[idx].id.clone(); | ||
| let _ = self | ||
| .state_store | ||
| .update(run_id, |s| { | ||
| if let Some(ss) = s.step_states.get_mut(&step_id) { | ||
| ss.status = StepStatus::Skipped; | ||
| ss.last_error = | ||
| Some("skipped: upstream dependency failed".to_string()); | ||
| ss.completed_at = Some(Utc::now()); | ||
| } | ||
| }) | ||
| .await; | ||
| for &dep_idx in definition.get_dependent_indices(&step_id) { | ||
| pending_check.push_back(dep_idx); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Blocked steps are recorded as skipped, so their own dependents still execute.
Lines 600-602 insert every dependency-blocked step into t.skipped. are_dependencies_satisfied (Lines 110-114) counts a skipped dependency as satisfied. The dependents pushed at Lines 617-619 therefore pass the readiness check in Phase 1 and launch, even though a transitive ancestor failed. Only the direct dependent receives the Skipped status; grandchildren run with stale input.
Track blocked steps separately so the block cascades.
🛠️ Proposed fix
struct StepTracker {
completed: HashSet<StepId>,
failed: HashSet<StepId>,
skipped: HashSet<StepId>,
+ /// Steps that never ran because an upstream dependency failed.
+ blocked: HashSet<StepId>,
running: HashSet<StepId>, fn total_processed(&self) -> usize {
- self.completed.len() + self.failed.len() + self.skipped.len()
+ self.completed.len() + self.failed.len() + self.skipped.len() + self.blocked.len()
} fn is_step_processable(&self, step_id: &StepId, step_idx: usize) -> bool {
!self.completed.contains(step_id)
&& !self.failed.contains(step_id)
&& !self.skipped.contains(step_id)
+ && !self.blocked.contains(step_id)
&& !self.running.contains(step_id)
&& !self.waiting_until.contains_key(&step_idx)
} fn has_failed_dependency(&self, depends_on: &[StepId]) -> bool {
- depends_on.iter().any(|dep| self.failed.contains(dep))
+ depends_on
+ .iter()
+ .any(|dep| self.failed.contains(dep) || self.blocked.contains(dep))
}
fn have_all_any_deps_failed(&self, depends_on_any: &[StepId]) -> bool {
- !depends_on_any.is_empty() && depends_on_any.iter().all(|dep| self.failed.contains(dep))
+ !depends_on_any.is_empty()
+ && depends_on_any
+ .iter()
+ .all(|dep| self.failed.contains(dep) || self.blocked.contains(dep))
} let mut t = tracker.write();
for &idx in &deps_blocked_indices {
- t.skipped.insert(definition.steps[idx].id.clone());
+ t.blocked.insert(definition.steps[idx].id.clone());
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if !deps_blocked_indices.is_empty() { | |
| { | |
| let mut t = tracker.write(); | |
| for &idx in &deps_blocked_indices { | |
| t.skipped.insert(definition.steps[idx].id.clone()); | |
| } | |
| } | |
| for &idx in &deps_blocked_indices { | |
| let step_id = definition.steps[idx].id.clone(); | |
| let _ = self | |
| .state_store | |
| .update(run_id, |s| { | |
| if let Some(ss) = s.step_states.get_mut(&step_id) { | |
| ss.status = StepStatus::Skipped; | |
| ss.last_error = | |
| Some("skipped: upstream dependency failed".to_string()); | |
| ss.completed_at = Some(Utc::now()); | |
| } | |
| }) | |
| .await; | |
| for &dep_idx in definition.get_dependent_indices(&step_id) { | |
| pending_check.push_back(dep_idx); | |
| } | |
| } | |
| } | |
| if !deps_blocked_indices.is_empty() { | |
| { | |
| let mut t = tracker.write(); | |
| for &idx in &deps_blocked_indices { | |
| t.blocked.insert(definition.steps[idx].id.clone()); | |
| } | |
| } | |
| for &idx in &deps_blocked_indices { | |
| let step_id = definition.steps[idx].id.clone(); | |
| let _ = self | |
| .state_store | |
| .update(run_id, |s| { | |
| if let Some(ss) = s.step_states.get_mut(&step_id) { | |
| ss.status = StepStatus::Skipped; | |
| ss.last_error = | |
| Some("skipped: upstream dependency failed".to_string()); | |
| ss.completed_at = Some(Utc::now()); | |
| } | |
| }) | |
| .await; | |
| for &dep_idx in definition.get_dependent_indices(&step_id) { | |
| pending_check.push_back(dep_idx); | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/src/engine.rs` around lines 597 - 621, Separate
dependency-blocked steps from intentionally skipped steps in the workflow
tracker; do not insert them into t.skipped. Update the dependency readiness and
propagation logic around are_dependencies_satisfied and the blocked-step
handling in the execution flow so a blocked ancestor keeps all transitive
dependents blocked, marking each as Skipped without launching them or evaluating
stale input.
Comment on lines
+711
to
+721
| for step_idx in ready_to_launch { | ||
| let step = &definition.steps[step_idx]; | ||
| let engine = self.clone_for_execution(); | ||
| let def = Arc::clone(&definition); | ||
| let step_id = step.id.clone(); | ||
| let tx = tx.clone(); | ||
| let tracker = Arc::clone(&tracker); | ||
| let collect_buffer = Arc::clone(&collect_buffer); | ||
|
|
||
| tokio::spawn(async move { | ||
| let step = &def.steps[step_idx]; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Spawn step tasks with self.spawn, not tokio::spawn.
with_runtime_handle (Lines 190-193) exists so agent calls and SQLite I/O stay off the caller's runtime. start_workflow, recover and start_cleanup_task use self.spawn, but the per-step task uses tokio::spawn, so every executor call runs on the ambient runtime. The isolation is lost exactly where the blocking work happens.
🛠️ Proposed fix
- tokio::spawn(async move {
+ let spawner = self.clone_for_execution();
+ spawner.spawn(async move {
let step = &def.steps[step_idx];Alternatively call self.spawn(...) directly here, since self is borrowed for the whole loop.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| for step_idx in ready_to_launch { | |
| let step = &definition.steps[step_idx]; | |
| let engine = self.clone_for_execution(); | |
| let def = Arc::clone(&definition); | |
| let step_id = step.id.clone(); | |
| let tx = tx.clone(); | |
| let tracker = Arc::clone(&tracker); | |
| let collect_buffer = Arc::clone(&collect_buffer); | |
| tokio::spawn(async move { | |
| let step = &def.steps[step_idx]; | |
| for step_idx in ready_to_launch { | |
| let step = &definition.steps[step_idx]; | |
| let engine = self.clone_for_execution(); | |
| let def = Arc::clone(&definition); | |
| let step_id = step.id.clone(); | |
| let tx = tx.clone(); | |
| let tracker = Arc::clone(&tracker); | |
| let collect_buffer = Arc::clone(&collect_buffer); | |
| let spawner = self.clone_for_execution(); | |
| spawner.spawn(async move { | |
| let step = &def.steps[step_idx]; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/src/engine.rs` around lines 711 - 721, Replace the
per-step tokio::spawn call in the workflow execution loop with self.spawn,
preserving the existing async closure and captured step execution state so agent
calls and SQLite I/O run on the executor’s configured runtime.
Comment on lines
+279
to
+283
| /// Durable promise: resolved by `complete_event` or by timeout. | ||
| WaitEvent { | ||
| name: String, | ||
| result: Option<EntryResult>, | ||
| }, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
WaitEvent has no step identity, so recovery can complete the wrong step.
JournalEntry::WaitEvent records only name. The engine memoizes wait completion by name (crates/flare-workflow/src/engine.rs lines 460-619, comment "Completed by name (journal entries carry no step id)"). If two StepMode::WaitEvent steps use the same event name, one completed journal entry marks both steps complete after a crash. The second wait is skipped and its dependents run early.
Add step_id to the variant and match on it during replay, as StepRun and Sleep already do.
♻️ Proposed change
/// Durable promise: resolved by `complete_event` or by timeout.
WaitEvent {
+ step_id: StepId,
name: String,
result: Option<EntryResult>,
},Note: this changes the persisted journal payload shape, so it needs a compatibility plan for existing rows plus updates in engine.rs, waits.rs, and the tests that construct WaitEvent.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/src/types.rs` around lines 279 - 283, Update the
WaitEvent variant and its journal creation/replay paths to carry and match a
step_id, preventing completion from being shared by waits with the same name.
Follow the existing StepRun and Sleep identity handling, and add compatibility
decoding for persisted rows that lack step_id; update engine.rs, waits.rs, and
all affected tests accordingly.
Comment on lines
+17
to
+19
| /// Record a step's output under `output_var` if one is set; returns the | ||
| /// (possibly new) variables map so the caller can persist it. | ||
| pub fn capture_output(vars: &mut HashMap<String, String>, output_var: Option<&str>, output: &str) { |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the stale doc comment.
The doc states that the function "returns the (possibly new) variables map", but capture_output mutates vars in place and returns ().
📝 Proposed fix
-/// Record a step's output under `output_var` if one is set; returns the
-/// (possibly new) variables map so the caller can persist it.
+/// Record a step's output under `output_var` if one is set, mutating `vars`
+/// in place so the caller can persist it.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| /// Record a step's output under `output_var` if one is set; returns the | |
| /// (possibly new) variables map so the caller can persist it. | |
| pub fn capture_output(vars: &mut HashMap<String, String>, output_var: Option<&str>, output: &str) { | |
| /// Record a step's output under `output_var` if one is set, mutating `vars` | |
| /// in place so the caller can persist it. | |
| pub fn capture_output(vars: &mut HashMap<String, String>, output_var: Option<&str>, output: &str) { |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/src/variables.rs` around lines 17 - 19, Update the
documentation comment for capture_output to state that it records the step
output by mutating the provided vars map in place and returns (). Remove the
inaccurate claim that it returns a variables map.
Comment on lines
+197
to
+210
| #[test] | ||
| fn workflow_status_serde_roundtrip() { | ||
| for s in [ | ||
| WorkflowStatus::Pending, | ||
| WorkflowStatus::Running, | ||
| WorkflowStatus::Completed, | ||
| WorkflowStatus::Failed, | ||
| WorkflowStatus::Cancelled, | ||
| ] { | ||
| let json = serde_json::to_string(&s).unwrap(); | ||
| let back: WorkflowStatus = serde_json::from_str(&json).unwrap(); | ||
| assert_eq!(s, back); | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add WorkflowStatus::Paused to the roundtrip list.
Paused is the only status omitted. It is also the status with distinct handling in crates/flare-workflow/src/sqlite_store.rs (status_from_str, the list_active filter, and the cleanup filter), so it deserves coverage here.
💚 Proposed change
for s in [
WorkflowStatus::Pending,
WorkflowStatus::Running,
+ WorkflowStatus::Paused,
WorkflowStatus::Completed,
WorkflowStatus::Failed,
WorkflowStatus::Cancelled,
] {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| #[test] | |
| fn workflow_status_serde_roundtrip() { | |
| for s in [ | |
| WorkflowStatus::Pending, | |
| WorkflowStatus::Running, | |
| WorkflowStatus::Completed, | |
| WorkflowStatus::Failed, | |
| WorkflowStatus::Cancelled, | |
| ] { | |
| let json = serde_json::to_string(&s).unwrap(); | |
| let back: WorkflowStatus = serde_json::from_str(&json).unwrap(); | |
| assert_eq!(s, back); | |
| } | |
| } | |
| #[test] | |
| fn workflow_status_serde_roundtrip() { | |
| for s in [ | |
| WorkflowStatus::Pending, | |
| WorkflowStatus::Running, | |
| WorkflowStatus::Paused, | |
| WorkflowStatus::Completed, | |
| WorkflowStatus::Failed, | |
| WorkflowStatus::Cancelled, | |
| ] { | |
| let json = serde_json::to_string(&s).unwrap(); | |
| let back: WorkflowStatus = serde_json::from_str(&json).unwrap(); | |
| assert_eq!(s, back); | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/flare-workflow/tests/types_test.rs` around lines 197 - 210, Update the
status list in workflow_status_serde_roundtrip to include WorkflowStatus::Paused
alongside the existing variants, ensuring its serde serialization and
deserialization roundtrip is covered.
Comment on lines
+1061
to
+1065
| #[schemars( | ||
| description = "Optional SQLite store path override (defaults to ~/.agentflare/workflows.db)" | ||
| )] | ||
| #[serde(default)] | ||
| pub(crate) db_path: Option<String>, |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
db_path accepts any filesystem path from the MCP client.
workflow_impl passes this value straight to SqliteStore::open_file, which creates the file if it is missing. A client can therefore create or write a SQLite database at any path the daemon user can write, including outside the agentflare data directory. Constrain the override to a directory under ~/.agentflare, or restrict it to tests and use the default path in production.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/mcp_server/types.rs` around lines 1061 - 1065, Restrict the db_path
override in the workflow request type and workflow_impl before passing it to
SqliteStore::open_file: accept only paths within the daemon user’s ~/.agentflare
directory, or ignore/reject client-supplied paths in production while preserving
test overrides. Ensure the default ~/.agentflare/workflows.db path remains the
production fallback.
…ate race, db_path Three real bugs surfaced by the fresh full-diff CodeRabbit review on PR #472: - waits.rs execute_sleep recomputed wake_at from Utc::now() on every call, including on crash-recovery re-arm, so a durable Sleep never resumed its original deadline — every restart pushed the wake time out further. Now reuses the wake_at from an existing pending Sleep journal entry. - sqlite_store.rs SqliteStore::update released the connection lock between its load and write, so concurrent updates on the same run (e.g. two fan-out branches completing close together) could race and silently drop one's mutation. Added update_lock held across the whole cycle. - mcp_server/workflow.rs honored a client-supplied db_path in production, letting any MCP caller point the workflow store at an arbitrary file. The override is now test-only (cfg!(test)); production always uses the default ~/.agentflare/workflows.db path. Added regression tests for the first two (recover_reams_pending_sleep now asserts wake_at is preserved across the crash boundary; concurrent_updates_on_same_run_do_not_lose_writes spawns 20 concurrent updates and asserts none are lost). The db_path fix isn't independently testable — cfg!(test) is true for any test binary — so it's covered by the existing MCP tests continuing to pass with their tempdir overrides. Remaining CodeRabbit nitpicks (eprintln vs tracing, brittle journal_tail count assertion, README doctest formatting, missing composite index, discarded delete_state errors in cleanup, executor.rs default is_retryable, non-blocking cancellation test, duplicate StepId validation, WaitEvent journal entries not scoped by step_id, EntryResult::success swallowing serialization errors, InMemoryStore::journal NotFound vs empty-vec inconsistency, fan_group dependency derivation edge case) are lower-severity style/robustness items left for a follow-up pass rather than folded into this bugfix. The journal.rs 'use RETURNING instead of a second SELECT' suggestion was checked and is a non-issue here: all journal writes go through the same Arc<Mutex<Connection>>, so there's no race for it to fix. Agentflare-Agent: claude-code Agentflare-Branch: task/447 Agentflare-Item: 447
This was referenced Aug 15, 2026
Merged
getappz
added a commit
that referenced
this pull request
Aug 16, 2026
…483 bug #2) (#513) * fix(git-shim): scope-check child crash no longer blocks git as a policy denial (item #472) Agentflare-Agent: opencode Agentflare-Branch: task/472-git-shim-scope-check-subprocess-crash-de Agentflare-Item: 472 * fix(git-shim): E2E test for scope-check crash pass-through + test-only bin override Agentflare-Agent: opencode Agentflare-Branch: task/472-git-shim-scope-check-subprocess-crash-de Agentflare-Item: 472 * fix(git-shim): compile-gate scope-check bin override, fix out-of-tree bypass - AGENTFLARE_GIT_SCOPE_CHECK_BIN override is now compiled out of release builds entirely instead of only being runtime-guarded, so a stray combination of env vars can never redirect a shipped binary's scope-check. - classify_scopes() short-circuits to Clear on an empty changed-paths list before it ever checks own_target/in_own_worktree, so skipping the diff computation for an out-of-tree invoker silently turned a real OutOfTree denial into a pass. Only skip the diff when there are no enforced other-claim scopes AND the invoker isn't out-of-tree. Agentflare-Agent: claude-code_2-1-233_agent Agentflare-Branch: task/472-git-shim-scope-check-subprocess-crash-de Agentflare-Item: 472 * fix(git-shim): drain stderr concurrently, enforce path cap on merged commit set - run_in_lines_bounded now drains the child's stderr on a separate thread instead of reading stdout to EOF first, since a caller whose git invocation writes more than one pipe buffer to stderr could otherwise deadlock. Every stdout read-error path now kills+waits the child too, instead of only the TooManyLines path. - changed_paths' commit branch caps the staged diff and the unstaged diff separately, so two disjoint sets could each land at MAX_CHANGED_PATHS and union past it; enforce the cap on the merged, deduplicated set as well. - Assert the crashed-scope-check test's audit record actually carries ScopeCheckError, not Deny, closing the coverage gap that would let an incorrect Deny classification pass silently. Agentflare-Agent: claude-code_2-1-233_agent Agentflare-Branch: task/472-git-shim-scope-check-subprocess-crash-de Agentflare-Item: 472
This was referenced Aug 18, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixed 7 of the 8 code-review findings on
task/447and committed as08b8dc8:High (blocking-call findings, fixed together):
agent_send_hooknow runsrun_headlessinsidespawn_blockingwithin the returned async block, not synchronously before it.SqliteStore's entireStateStoreimpl wraps its rusqlite I/O inspawn_blocking.WorkflowEnginegained an optional runtime handle (with_runtime_handle);start_workflow/recover/start_cleanup_taskspawn through it, andsrc/workflow.rsnow attachesWORKFLOW_RTso the MCP path (which.awaits the async core directly on the daemon's own runtime) is covered too, not just the CLI'sblock_onpath.Medium:
4. Steps blocked by a failed dependency now get an explicit
Skippedstatus (cascading to their own dependents) instead of stayingPendingforever — this required also fixing completion-signal forwarding to propagate on any terminal result, not just Success/Skip, since failed steps' dependents were never even reaching the scheduler.5.
WaitEventwaiters are now keyed per step instance instead ofrun_id:name, so two concurrent waits on the same event name no longer clobber each other;complete_eventwakes every matching waiter.6.
cleanup_old_workflowsno longer panics via.expect(); it logs and returns 0 like its siblings.7.
write_statenow does targeted UPSERTs instead of DELETE-all + INSERT-all on every write (safe becausestep_states/variablesonly grow within a run's lifetime — verified no removal path exists).Skipped (documented in the commit): #8 —
WorkflowRunId::new()vsdb_kit::ids::new_id().new_id()returns a nanoidString, butWorkflowRunIdwraps aUuidand is parsed/displayed as a v7 UUID throughout the MCP/CLI surface. Switching would require re-typing it fromUuidtoStringcrate-wide and would lose intentional v7 time-ordering — a materially bigger change than the finding's "low - consistency" severity implies.Added two regression tests (
concurrent_wait_events_same_name_both_resolve,dependent_of_failed_step_gets_terminal_status) that fail without the corresponding fixes.cargo test -p flare-workflow(55 tests) and the top-levelworkflow::/mcp_server::workflow::tests (7 tests, including the real coder-reviewer-PR pipeline) all pass;fmtandclippy -D warnings -A unsafe_code -A clippy::pedanticare clean workspace-wide.Summary by CodeRabbit
workflowCLI and MCP operations for running, monitoring, listing, and completing workflow events.