feat(workflow): durable DAG workflow worker - #380
Merged
Conversation
A new `workflow` worker that orchestrates multi-agent pipelines as a durable directed-acyclic graph on the iii engine. - DAG core: typed node/run state, dependency validation, multi-input joins, topological scheduling, and crash-safe durable persistence with reconcile. - Execution: per-node (optionally nested) agent sessions, fast-wake on turn-completed, caller notify callbacks, and a sweep loop with timeouts. - Permissions: inherit-by-default node permissions, per-node router deny. - Fire-and-forget `workflow::start`: callers get the run_id back immediately and receive the outcome via `reply_to` / `notify` or by polling `workflow::status`; no blocking `await` path. - Workspace picker control plane for the shell (#376). - Add `llm-router` and `workflow` dependencies to iii.worker.yaml and workflow.yaml; remove outdated observability and workflow plan docs. Claude-Session: https://claude.ai/code/session_019pMG6VY8cKLW2EzF1FEkV9
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
skill-check — worker0 verified, 27 skipped (no docs/).
Four for four. Nicely done. |
|
Warning Review limit reached
Next review available in: 12 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: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (11)
📝 WalkthroughWalkthroughIntroduces a new Changesworkflow worker: new deterministic DAG orchestrator
Sequence Diagram(s)sequenceDiagram
participant Caller
participant workflow_start
participant workflow_tick
participant harness
participant iii_state
participant events
Caller->>workflow_start: StartRequest{definition, input}
workflow_start->>iii_state: get_idem (idempotency check)
workflow_start->>iii_state: put_run (Running record)
workflow_start->>workflow_tick: enqueue_tick(step=0)
loop each tick
workflow_tick->>iii_state: get_run
workflow_tick->>harness: harness::status (reconcile running nodes)
workflow_tick->>iii_state: put_node_result (on completion)
workflow_tick->>workflow_tick: expand_ready_fanouts + decide
alt Fire nodes
workflow_tick->>harness: harness::send (fire_node)
workflow_tick->>iii_state: put_run
else Finalize
workflow_tick->>harness: cascade_stop_running
workflow_tick->>events: emit_run_completed + emit_notify + emit_reply
workflow_tick->>iii_state: put_run (terminal status)
else Park
workflow_tick->>iii_state: put_run
end
end
rect rgba(200, 100, 50, 0.5)
Note over workflow_tick,iii_state: sweep: reconcile + timeout + GC
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ytallo
approved these changes
Jun 30, 2026
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
🧹 Nitpick comments (4)
workflow/src/functions/start.rs (1)
896-908: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winPersist the def after the depth check to avoid an orphaned
def_ref.
put_def(Line 899) runs beforecaller_workflow_depthand theMAX_WORKFLOW_DEPTHrejection (Lines 903-908). When the depth cap trips, the function returnsErrbut the definition is already persisted with no corresponding run record, so nothing references it and run-scoped GC won't reclaim it.caller_workflow_depthonly needscaller_session_id, so it can run beforeput_def.♻️ Reorder depth check before persisting the def
let run_id = new_run_id(); let _guard = deps.locks.guard(&run_id).await; - state::put_def(&deps.iii, &run_id, &req.definition).await?; - // Bound sub-workflow nesting: a node that opted into `workflow::start` could // otherwise recurse (sub-workflow → node → sub-workflow → …) without limit. let depth = caller_workflow_depth(deps, caller_session_id.as_deref()).await?; if depth > MAX_WORKFLOW_DEPTH { return Err(WorkflowError::InvalidDef(format!( "sub-workflow nesting depth {depth} exceeds the cap of {MAX_WORKFLOW_DEPTH}" ))); } + + state::put_def(&deps.iii, &run_id, &req.definition).await?;🤖 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 `@workflow/src/functions/start.rs` around lines 896 - 908, The depth-cap validation in start workflow should happen before persisting the definition, because `state::put_def` currently runs before `caller_workflow_depth` and can leave an orphaned `def_ref` if `MAX_WORKFLOW_DEPTH` is exceeded. Move the `caller_workflow_depth(deps, caller_session_id.as_deref())` check ahead of `state::put_def` in `start`, then only store the def after the depth passes so rejected sub-workflows do not create unreferenced persisted data.workflow/tests/orchestration.rs (3)
171-186: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAssert the classifier’s failure message, not just the variant.
This helper re-injects
errorinto the checkpoint after only checkingNodeOutcome::Failed(_), so a regression whereclassify_terminal("completed", ..., Some(err))rewrites or drops the message would still pass here. DestructureNodeOutcome::Failed(msg)and persist/assertmsginstead.Proposed change
fn complete_with_error(record: &mut WorkflowRunRecord, node_uid_str: &str, error: &str) { - // classify_terminal("completed", garbage_value, error) → must be Failed - let outcome = classify_terminal( + let outcome = classify_terminal( "completed", Some(json!({"unexpected": "garbage"})), Some(error.to_string()), ); - assert!( - matches!(outcome, NodeOutcome::Failed(_)), - "classify_terminal should return Failed when result_error is set" - ); + let msg = match outcome { + NodeOutcome::Failed(msg) => msg, + other => panic!("expected Failed, got {other:?}"), + }; + assert_eq!(msg, error, "classify_terminal should preserve result_error"); if let Some(cp) = record.nodes.get_mut(node_uid_str) { cp.state = NodeState::Failed; - cp.result_error = Some(error.to_string()); + cp.result_error = Some(msg); } }🤖 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 `@workflow/tests/orchestration.rs` around lines 171 - 186, Update the test helper complete_with_error in workflow/tests/orchestration.rs so it verifies the actual failure payload from classify_terminal, not just that the result is NodeOutcome::Failed. Destructure the NodeOutcome::Failed message returned by classify_terminal("completed", ..., Some(error.to_string())) and assert it matches the expected error string before storing it in WorkflowRunRecord; keep the checkpoint update on cp.result_error aligned with that same asserted message.
614-616: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winThe “non-empty ready frontier” case is not actually exercised here.
With
abort = true, this record never reaches a materialized ready node: firstplanis stillRunning, and later you calldecideafterplancompletes without expanding thereadfanout. So this test validates “abort beats running/quiescence”, but not “abort beats a ready frontier” as the comment says.Proposed change
- // decide must return Finalize(Cancelled) regardless of the Running node or - // any ready frontier. + // decide must return Finalize(Cancelled) regardless of the Running node. let decision = decide(&def, &record); @@ - // Simulate completing plan with a result (abort still set) — decide must - // still return Finalize(Cancelled), not Fire or Park. + // Materialize the next ready fanout, then assert abort still wins over a + // non-empty frontier. complete(&mut record, &mut results, "plan", json!({"docs": ["x"]})); + dag::expand_ready_fanouts(&def, &mut record, &results); + assert!( + !dag::ready_frontier(&def, &record).is_empty(), + "sanity: abort precedence check should have a ready frontier" + ); let decision2 = decide(&def, &record);Also applies to: 633-636
🤖 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 `@workflow/tests/orchestration.rs` around lines 614 - 616, The abort test in decide() is only covering Running/quiescent states, not a non-empty ready frontier, so the comment is misleading. Update the orchestration test around decide and the related read fanout setup so a materialized ready node is actually present before calling decide, while keeping the abort=true assertion; this should exercise the ready-frontier path in addition to the existing Running case. Use the existing plan, record, and decide helpers in orchestration.rs to build the frontier explicitly rather than relying on the current sequence.
553-567: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAssert the join payload before completing
d.This test currently proves the barrier semantics, but not the
InputFrom::Manycontract. Becausedrive_stepnever callsdag::gather_input,dstill fires here even if the join payload starts dropping one branch or uses the wrong object shape. Add an assertion thatdag::gather_input(&def, &record, "d", &results)contains bothbandckeyed results before completingd.Proposed change
// Step 3: both b and c done → d's join fires. let step3 = drive_step(&def, &mut record, &results); match &step3 { TickDecision::Fire(uids) => { assert_eq!( uids, &vec!["d".to_string()], "step 3 must fire d after join" ); } other => panic!("expected Fire([d]) at step 3, got {:?}", other), } + + let joined = dag::gather_input(&def, &record, "d", &results); + assert_eq!( + joined, + json!({ + "b": {"x": 2}, + "c": {"x": 3} + }), + "d must receive both branch results keyed by dependency id" + ); // Complete d. complete(&mut record, &mut results, "d", json!({"x": 4}));🤖 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 `@workflow/tests/orchestration.rs` around lines 553 - 567, The test around drive_step and complete for node d only checks that the join fires, but it does not verify the InputFrom::Many payload shape. Before completing d, call dag::gather_input with def, record, "d", and results, then assert the returned payload includes both b and c keyed results with the expected object structure. Keep the existing fire assertion in orchestration.rs, but add the payload assertion in the same test so the join contract is validated directly.
🤖 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 `@workflow/README.md`:
- Around line 79-95: The iii-state concurrency note is hard-coded to sdk 0.19.2
while the workspace pins iii-sdk via the workflow crate, so update the README
section to match the pinned iii-sdk version or remove the explicit version
reference entirely. Edit the iii-state/WorkflowRunRecord explanation and the
“Blocked prerequisite” note so the rationale stays aligned with the actual
dependency version in workflow/Cargo.toml and does not go stale.
In `@workflow/src/configuration.rs`:
- Around line 186-208: The harness-hook registration path in the configuration
callback marks `bound` as true before `bind_turn_completed`,
`bind_pre_trigger_hook`, and `bind_pre_generate_hook` succeed, which blocks all
future retries if any bind fails. Move the `bound` transition so it only happens
after all three binds succeed, or otherwise reset it when a bind is
missing/failed, and keep the existing logging in the `configuration`
registration logic to report partial binding state. Use the `bound` atomic and
the three `bind_*` helpers as the key points to update so transient failures can
be retried on later `engine::*-available` events.
- Around line 395-400: The sweep schedule update in apply_config is being
committed even when the cron rebind does not succeed, causing the live snapshot
to drift from the active trigger. Update the workflow configuration path around
rebind_slot/bind_sweep and apply_config so the new sweep_expression is only
written after a successful rebind, and skip or preserve the old value when
bind_sweep returns None or fails. Keep the existing old.sweep_expression
comparison and tracing::info flow, but gate the snapshot commit on the rebind
outcome so later config changes can retry correctly.
In `@workflow/src/dag.rs`:
- Around line 397-398: The fanout path in dag::gather_input is currently
returning record.input.clone() for fanout_item, so read#i children get the
top-level input instead of the materialized fanout element. Update gather_input
to resolve fanout_item from record.fanout_src using the item’s uid/index, and
adjust the tick.rs caller so it passes the materialized uid rather than only the
base id. Make sure the dispatch input for fanout children is built from the
per-item materialized value, not the workflow input.
In `@workflow/src/events.rs`:
- Around line 98-106: The request payload built in the reply path currently
places `run` at the top level, but `harness::send` only reads
`SendRequest.options`, so the wake/passive flag is being ignored. Update the
payload construction in this reply flow to put `run` inside `options` alongside
`functions` before calling `harness::send`, using the existing `reply.functions`
and `wake` values so the handler receives the intended `run:true/false`
behavior.
In `@workflow/src/functions/inject_guidance.rs`:
- Around line 17-21: Update WORKFLOW_GUIDANCE in inject_guidance so it matches
the real default node policy enforced by tick::node_functions and
normalize_functions: remove or soften the claim that omitting agent.functions
means a node can call ["*"], and instead describe the actual inherited/allowed
behavior including the implicit control-plane and router-generate denials. Keep
the guidance aligned with the runtime rules so the wording around
agent.functions, default inheritance, and least-privilege examples in
WORKFLOW_GUIDANCE does not promise capabilities that the workflow engine will
reject.
In `@workflow/src/functions/node_result.rs`:
- Around line 25-30: The NodeResultResponse serialization is omitting the result
field when it is None because of the skip_serializing_if on the result member,
which breaks the expected explicit null response. Update NodeResultResponse so
workflow::node-result always serializes result as null when absent, and remove
or adjust the serde attribute on result to preserve the public contract.
In `@workflow/src/functions/start.rs`:
- Around line 564-601: The dependency validation in start workflow setup allows
node refs in input.from without matching depends_on, so scheduling and reads can
diverge. Update the validation around the node/dependency check to require every
node referenced by InputFrom::One/InputFrom::Many (and any fanout.over) to also
appear in depends_on, or otherwise reject the definition when they are missing.
Use the existing consumed-set logic in the node loop to compare against
depends_on and extend the InvalidDef message to point at the mismatch.
In `@workflow/src/functions/sweep.rs`:
- Around line 65-76: GC sweep should also clean up the workflow session reverse
index, not just the run record. Update the run deletion path used by sweep,
especially state::delete_run() and any helper it calls, to remove each
checkpoint’s session_id -> run_id entry before deleting the run. Use the
existing workflow state structures around state::delete_run and the
workflow::wake index maintenance to locate the session index cleanup, and ensure
the sweep loop in sweep() continues to call the same deletion path so orphaned
index rows cannot remain.
In `@workflow/src/functions/tick.rs`:
- Around line 128-136: The build_opening helper is embedding input_json directly
inside the workflow_input fence, which lets upstream content break out of the
boundary if it contains a closing tag. Update build_opening in tick.rs to escape
or neutralize any workflow_input tag delimiters in the payload before formatting
the fenced string, while keeping the structure of build_opening and its template
handling unchanged.
In `@workflow/src/locks.rs`:
- Around line 19-33: The per-run mutex cache in WorkflowLocks::guard leaves
strong Arc entries in the map forever, so completed run_ids are never reclaimed.
Update the WorkflowLocks map/guard flow to avoid retaining strong ownership
after a lock is no longer needed, either by storing Weak references or by
removing the entry once the last guard is dropped. Make the change around
WorkflowLocks::guard and the map field so stale run entries can be evicted
safely without breaking concurrent lock acquisition.
In `@workflow/src/main.rs`:
- Around line 93-96: The startup path in main currently aborts on any
fetch_config() error, but worker binaries should stay available by warning and
falling back to WorkerConfig::default(). Update the config-loading block around
configuration::fetch_config(&iii) to catch errors, emit tracing::warn! with
enough context, and continue using WorkerConfig::default() instead of
propagating the failure through anyhow::Error::msg/context.
In `@workflow/src/state.rs`:
- Around line 171-177: The cleanup logic in state::delete_record is dropping
errors from state_delete for child results and the definition, then always
deleting the run record via state_delete on SCOPE_RUN. Update this flow so the
run record is only removed after all child deletes succeed: propagate or
aggregate failures from the loop over record.nodes and the SCOPE_DEF delete, and
return early on any error before calling the final run delete. Use the existing
state_delete, SCOPE_RESULT, SCOPE_DEF, SCOPE_RUN, and def_key symbols to keep
the retry anchor intact.
In `@workflow/src/types.rs`:
- Around line 55-58: The node id constraint in the types definition is missing
`/`, which breaks the `<run_id>/<node_uid>` storage contract used by result
references. Update the node id validation near the `nodes: BTreeMap<String,
NodeDef>` definition and any related checks in `ids.rs`/`result_ref` handling to
reject `/` as well, or consistently escape node ids before composing storage
keys so `node_uid` round-trips safely.
---
Nitpick comments:
In `@workflow/src/functions/start.rs`:
- Around line 896-908: The depth-cap validation in start workflow should happen
before persisting the definition, because `state::put_def` currently runs before
`caller_workflow_depth` and can leave an orphaned `def_ref` if
`MAX_WORKFLOW_DEPTH` is exceeded. Move the `caller_workflow_depth(deps,
caller_session_id.as_deref())` check ahead of `state::put_def` in `start`, then
only store the def after the depth passes so rejected sub-workflows do not
create unreferenced persisted data.
In `@workflow/tests/orchestration.rs`:
- Around line 171-186: Update the test helper complete_with_error in
workflow/tests/orchestration.rs so it verifies the actual failure payload from
classify_terminal, not just that the result is NodeOutcome::Failed. Destructure
the NodeOutcome::Failed message returned by classify_terminal("completed", ...,
Some(error.to_string())) and assert it matches the expected error string before
storing it in WorkflowRunRecord; keep the checkpoint update on cp.result_error
aligned with that same asserted message.
- Around line 614-616: The abort test in decide() is only covering
Running/quiescent states, not a non-empty ready frontier, so the comment is
misleading. Update the orchestration test around decide and the related read
fanout setup so a materialized ready node is actually present before calling
decide, while keeping the abort=true assertion; this should exercise the
ready-frontier path in addition to the existing Running case. Use the existing
plan, record, and decide helpers in orchestration.rs to build the frontier
explicitly rather than relying on the current sequence.
- Around line 553-567: The test around drive_step and complete for node d only
checks that the join fires, but it does not verify the InputFrom::Many payload
shape. Before completing d, call dag::gather_input with def, record, "d", and
results, then assert the returned payload includes both b and c keyed results
with the expected object structure. Keep the existing fire assertion in
orchestration.rs, but add the payload assertion in the same test so the join
contract is validated directly.
🪄 Autofix (Beta)
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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: bf8830af-1161-4a5b-9325-b9620f37633e
⛔ Files ignored due to path filters (1)
workflow/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (31)
harness/iii.worker.yamlworkflow/.gitignoreworkflow/Cargo.tomlworkflow/README.mdworkflow/iii.worker.yamlworkflow/src/config.rsworkflow/src/configuration.rsworkflow/src/dag.rsworkflow/src/error.rsworkflow/src/events.rsworkflow/src/functions/inject_guidance.rsworkflow/src/functions/mod.rsworkflow/src/functions/node_result.rsworkflow/src/functions/stamp_reply.rsworkflow/src/functions/start.rsworkflow/src/functions/status.rsworkflow/src/functions/stop.rsworkflow/src/functions/sweep.rsworkflow/src/functions/tick.rsworkflow/src/functions/wake.rsworkflow/src/ids.rsworkflow/src/lib.rsworkflow/src/locks.rsworkflow/src/main.rsworkflow/src/manifest.rsworkflow/src/reconcile.rsworkflow/src/state.rsworkflow/src/telemetry.rsworkflow/src/timeout.rsworkflow/src/types.rsworkflow/tests/orchestration.rs
- start: fold each node's input.from/fanout.over reads into depends_on before
validate+persist, so a node never schedules before a node it reads (was: the
read resolved to null while the run still reported success); reserve '/' in
node ids too (it's the run_id/node_uid storage-key separator).
- state(delete_run): propagate child deletes so the run record survives as the
retry anchor on a transient failure, and clear the session reverse-index so GC
stops leaking index rows.
- locks: hold Weak (not Arc) per-run mutexes so a finished run's lock frees
instead of pinning forever.
- configuration: release the harness-hook bind claim on a partial bind so a
later registry-change event retries; commit a new sweep_expression only after
the cron rebind actually succeeds.
- tick(build_opening): escape angle brackets in the untrusted fenced input so
upstream output can't forge a </workflow_input> break-out.
- events(reply): drop the dead top-level `run` field — harness::send has no such
field (verified against harness send.rs) and always drives a turn; the captured
caller policy already rides in options.functions.
- node-result: serialize an explicit `{ "result": null }` instead of dropping
the key, matching the documented contract.
- main: warn and fall back to WorkerConfig::default() on a config-fetch error
instead of aborting boot (matches the other worker binaries).
- inject_guidance: correct the default-node-policy wording (reach minus the
workflow control plane), and README: sdk 0.19.2 -> 0.20.0 (pinned iii-sdk).
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
2 participants
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.
Summary
A new
workflowworker that orchestrates multi-agent pipelines as a durable directed-acyclic graph on the iii engine.workflow::start— callers get therun_idback immediately and receive the outcome viareply_to/notifyor by pollingworkflow::status; no blockingawaitpath.llm-routerandworkflowdeps intoiii.worker.yaml/workflow.yaml; remove outdated observability and workflow plan docs.Correctness model
Single-writer-per-run: every path that reads-mutates-writes a run record (
tick,reconcile, sweep,stop) holds the per-run lock, closing the read-modify-write race. Idempotency for duplicate deliveries comes from deterministic child-session and node-result ids. Horizontal scaling is a deployment choice (shardtickbyrun_id) — no record-level CAS exists in iii-state yet, so true multi-writer-per-run is intentionally out of scope until an engineif_match/state::casop lands.Test plan
workflow/tests/orchestration.rs(~800 lines) covers DAG scheduling, joins, persistence/reconcile, and timeout sweep.cargo test -p workflowSee
workflow/README.mdfor the full result-delivery and scaling notes.Summary by CodeRabbit
New Features
Bug Fixes
Documentation