feat(workflow): implement Epic 06 Phase 3 sequential execution engine with task nodes and run checkpointing - #152
Merged
Conversation
… with task nodes and run checkpointing
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Warning Review limit reached
Next review available in: 37 minutes You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. 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: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (16)
📝 WalkthroughWalkthroughChangesThe workflow engine now supports sequential and branching execution, task-node tool dispatch, asynchronous idempotent run starts, owner-scoped run listing, checkpoint persistence, timeouts, and failure handling. Workflow execution engine
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant WorkflowManager
participant WorkflowExecutor
participant PostgresWorkflowStore
participant TaskNodeExecutor
WorkflowManager->>PostgresWorkflowStore: Create running workflow run
WorkflowManager->>WorkflowExecutor: Start background execution
WorkflowExecutor->>PostgresWorkflowStore: Load run and checkpoint node
WorkflowExecutor->>TaskNodeExecutor: Execute ready task node
TaskNodeExecutor->>PostgresWorkflowStore: Persist node result
WorkflowExecutor->>PostgresWorkflowStore: Complete or fail workflow run
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)
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 |
Contributor
There was a problem hiding this comment.
Pull request overview
Implements Epic 06 Phase 3 of the workflow engine by adding a sequential (non-parallel) workflow execution loop with per-transition checkpointing, task-node tool execution, and run lifecycle APIs (start/get/list) with background scheduling and Postgres persistence.
Changes:
- Added
WorkflowExecutorwith sequential/branching ready-node resolution, node timeout/failure handling, and checkpoint-per-transition persistence. - Introduced
NodeExecutorprotocol +TaskNodeExecutor(tool invocation with{{dot.path}}argument templating andexecution_receipt_idpropagation). - Implemented run lifecycle in
WorkflowManager(idempotentstart_run,get_run,list_runs) with backgroundasyncio.Taskscheduling and a dedicated DB session store factory; added extensive tests and updated Epic doc.
Reviewed changes
Copilot reviewed 15 out of 15 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| docs/plans/post-mvp-v2-epic-06-workflow-engine.md | Marks Phase 3 checklist complete; adds completion record and changelog entry. |
| backend-python/tests/ai/workflow/test_task_node.py | New unit tests for TaskNodeExecutor argument templating and receipt propagation. |
| backend-python/tests/ai/workflow/test_manager.py | Removes outdated “start_run not implemented” test and points to new run tests. |
| backend-python/tests/ai/workflow/test_manager_runs.py | New tests for run lifecycle (start_run idempotency, owner scoping). |
| backend-python/tests/ai/workflow/test_executor_sequential.py | New tests for sequential/branching execution, checkpointing, failures, timeouts. |
| backend-python/tests/ai/workflow/test_deps.py | Updates DI wiring tests for node executors + background store factory. |
| backend-python/app/ai/workflow/providers/postgres.py | Implements run + node execution persistence and optimistic checkpointing. |
| backend-python/app/ai/workflow/nodes/task_node.py | Adds TaskNodeExecutor to execute tools with template resolution and receipt ID. |
| backend-python/app/ai/workflow/nodes/base.py | Adds NodeExecutionRequest and WorkflowNodeExecutionError; extends NodeExecutor. |
| backend-python/app/ai/workflow/manager.py | Implements start_run/list_runs and background execution scheduling with dedicated session. |
| backend-python/app/ai/workflow/graph/traversal.py | Implements sequential/branching ready-node resolver based on run context/current nodes. |
| backend-python/app/ai/workflow/engine/executor.py | Adds the sequential execution loop with per-node checkpointing + failure/timeout handling. |
| backend-python/app/ai/workflow/engine/background.py | Adds a process-lifetime task registry to prevent GC of scheduled run tasks. |
| backend-python/app/ai/tools/schemas.py | Extends ToolExecutionContext with optional execution_receipt_id. |
| backend-python/app/ai/deps.py | Wires WorkflowManager DI with ToolExecutor, TaskNodeExecutor, and background store factory. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+94
to
+99
| while run.status is RunStatus.RUNNING: | ||
| ready_node_ids = resolve_ready_nodes(definition, run) | ||
| if not ready_node_ids: | ||
| run = await self._complete_run(run) | ||
| break | ||
| run = await self._execute_node(run, nodes_by_id[ready_node_ids[0]]) |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (1)
backend-python/tests/ai/workflow/test_executor_sequential.py (1)
181-208: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert every checkpoint transition.
This test proves only that at least one run checkpoint occurs. It passes if the executor persists only a final checkpoint.
Record
checkpoint_run()calls and assert the running, succeeded, and completed transitions for both nodes.🤖 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 `@backend-python/tests/ai/workflow/test_executor_sequential.py` around lines 181 - 208, Update test_checkpoint_persisted_after_every_node_transition to record checkpoint_run() calls during execution and assert the running, succeeded, and completed transitions for both start and end nodes, including their expected order. Keep the existing persisted execution and checkpoint-version assertions, while ensuring the test fails if only a final checkpoint is written.
🤖 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 `@backend-python/app/ai/workflow/manager.py`:
- Around line 203-238: Update the run-start flow around the idempotency lookup
and WorkflowRun creation to normalize and validate idempotency_key before
querying or constructing the run, then replace the separate lookup/create
sequence with a store-level atomic get-or-create using the unique (owner_id,
workflow_definition_id, idempotency_key) constraint. Return the existing row for
concurrent or whitespace-equivalent requests, and call _schedule_run only when
the atomic operation reports that this caller created the row; add regression
coverage for concurrent requests and whitespace-normalized keys.
In `@backend-python/app/ai/workflow/nodes/task_node.py`:
- Line 18: Constrain workflow node IDs and trigger-input keys to the same
identifier grammar used by _PLACEHOLDER and _resolve_path, allowing only
alphanumeric and underscore segments. Update WorkflowNode validation and
trigger-input validation to reject keys such as fetch-user before they enter
context, preserving existing placeholder resolution for valid dotted paths.
In `@backend-python/app/ai/workflow/providers/postgres.py`:
- Around line 274-286: The append_node_execution method must resolve existing
records by the logical key (run_id, node_id, attempt), not only execution.id, so
crash replays update the committed execution instead of inserting a duplicate.
Update its lookup/upsert flow to reuse the existing row and preserve its ID,
while retaining normal insertion for new attempts; add a test covering replay
after a committed running checkpoint.
- Around line 174-180: The create_run method must handle concurrent identical
workflow starts atomically instead of allowing a unique-constraint
IntegrityError to escape. Replace the add/flush path in create_run with a
PostgreSQL conflict-aware insert using
uq_workflow_runs_owner_definition_idempotency, returning the inserted row or
reloading the existing matching row when a conflict occurs; preserve the commit,
refresh, and domain conversion behavior. Add a test that starts two identical
runs concurrently and verifies both return the same persisted workflow run.
In `@docs/plans/post-mvp-v2-epic-06-workflow-engine.md`:
- Line 2204: Update the WorkflowExecutor milestone changelog row to use the
correct completion date: set it to 2026-08-04 if the work completed today,
otherwise defer publication until 2026-08-05.
- Line 1218: Correct the completed API claims in the workflow-engine plan:
replace the nonexistent WorkflowExecutor.step() reference with the implemented
execute_run() API, and update the WorkflowManager.get_run() claim to reflect
that it returns WorkflowRun without node execution history. Either document an
implemented history-returning API or remove the history claim.
---
Nitpick comments:
In `@backend-python/tests/ai/workflow/test_executor_sequential.py`:
- Around line 181-208: Update
test_checkpoint_persisted_after_every_node_transition to record checkpoint_run()
calls during execution and assert the running, succeeded, and completed
transitions for both start and end nodes, including their expected order. Keep
the existing persisted execution and checkpoint-version assertions, while
ensuring the test fails if only a final checkpoint is written.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 5b5e38fa-f01f-4857-b13b-b78226a388e2
📒 Files selected for processing (15)
backend-python/app/ai/deps.pybackend-python/app/ai/tools/schemas.pybackend-python/app/ai/workflow/engine/background.pybackend-python/app/ai/workflow/engine/executor.pybackend-python/app/ai/workflow/graph/traversal.pybackend-python/app/ai/workflow/manager.pybackend-python/app/ai/workflow/nodes/base.pybackend-python/app/ai/workflow/nodes/task_node.pybackend-python/app/ai/workflow/providers/postgres.pybackend-python/tests/ai/workflow/test_deps.pybackend-python/tests/ai/workflow/test_executor_sequential.pybackend-python/tests/ai/workflow/test_manager.pybackend-python/tests/ai/workflow/test_manager_runs.pybackend-python/tests/ai/workflow/test_task_node.pydocs/plans/post-mvp-v2-epic-06-workflow-engine.md
| from app.core.caller import CallerContext | ||
|
|
||
| #: Matches a whole-string placeholder such as ``"{{trigger_input.topic}}"``. | ||
| _PLACEHOLDER = re.compile(r"^\{\{\s*([a-zA-Z0-9_]+(?:\.[a-zA-Z0-9_]+)*)\s*\}\}$") |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
fd -a -t f -e py . backend-python/app/ai/workflow/graph \
-x ast-grep outline {} --items all
rg -n -P -C 5 \
'\bclass\s+GraphValidator\b|WorkflowNode\b|node\.id\b|pattern\s*=' \
backend-python/app/ai/workflowRepository: pateatlau/fullstack-ai-platform
Length of output: 33590
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant placeholder resolution implementation and context/key usages.
sed -n '1,140p' backend-python/app/ai/workflow/nodes/task_node.py
printf '\n--- workflow context/storage references ---\n'
rg -n -C 4 'class WorkflowContext|variables\s*=|ctx\.variables|WorkflowContext|variables\[' backend-python/app/ai/workflow
printf '\n--- task node config usages ---\n'
rg -n -C 3 'arguments_template|tool_name|TaskNodeExecutor' backend-pythonRepository: pateatlau/fullstack-ai-platform
Length of output: 50390
Constrain node and trigger-input keys to the placeholder grammar.
WorkflowNode.id accepts any non-empty string, but _resolve_path only traverses exact dictionary keys after matching [a-zA-Z0-9_]+(.*) path segments. A node id such as fetch-user or an arguments_template path like variables.fetch-user.data.echo is accepted as context data but stays literal, throwing Unresolved arguments_template placeholder instead of selecting the intended value. Either reject workflow keys that include dashes, or add an unambiguous syntax for non-identifier keys.
🤖 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 `@backend-python/app/ai/workflow/nodes/task_node.py` at line 18, Constrain
workflow node IDs and trigger-input keys to the same identifier grammar used by
_PLACEHOLDER and _resolve_path, allowing only alphanumeric and underscore
segments. Update WorkflowNode validation and trigger-input validation to reject
keys such as fetch-user before they enter context, preserving existing
placeholder resolution for valid dotted paths.
| | 1.7 | 2026-08-04 | Parallel branch checkpoints: optimistic `checkpoint_version` merge + retry; prevent last-writer-wins on `context`/`current_node_ids`. Part I + Phase 5 sync. | | ||
| | 1.8 | 2026-08-04 | `apply_decision()` atomic CAS on `waiting_approval` + same-transaction run transition; no-op/conflict on duplicate decisions. Phase 7 sync. | | ||
| | 1.9 | 2026-08-05 | Phase 1 complete: canonical models/enums, `WorkflowStore` protocol, `PostgresWorkflowStore` scaffold, `WorkflowManager` skeleton, `0007_workflow_tables` migration, `WORKFLOW_ENGINE_ENABLED` + workflow config, DI wiring. 39 workflow tests; 1344 total backend passed; 89.80% coverage. Public API frozen. Phase 2 complete: `GraphValidator`, condition DSL shape validation, definition CRUD via `WorkflowManager`/`PostgresWorkflowStore`, versioning on run reference. 61 workflow tests; 1370 total backend passed; 90.00% coverage. Migration rollback CI smoke test pending. | | ||
| | 1.10 | 2026-08-05 | Phase 3 complete: `WorkflowExecutor` sequential/branching step loop, `NodeExecutor` protocol, `TaskNodeExecutor` (dot-path `arguments_template` resolution), `WorkflowManager.start_run()`/`get_run()`/`list_runs()`, per-transition checkpointing via `PostgresWorkflowStore`, background run scheduling. 88 workflow tests; 1397 total backend passed; 89.20% coverage. | |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the future completion date.
This row is dated 2026-08-05, which is after August 4, 2026. Use 2026-08-04 if this PR completed today, or publish the entry on August 5, 2026.
🤖 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 `@docs/plans/post-mvp-v2-epic-06-workflow-engine.md` at line 2204, Update the
WorkflowExecutor milestone changelog row to use the correct completion date: set
it to 2026-08-04 if the work completed today, otherwise defer publication until
2026-08-05.
… checkpoints, and identifier validation
This was referenced Aug 5, 2026
Merged
Merged
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
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
Implements Epic 06 Phase 3 — the sequential execution engine for linear/branching (non-parallel) workflow graphs.
WorkflowExecutorstep loop with per-transition checkpointing, node timeouts, and failure handlingTaskNodeExecutor(NodeExecutorprotocol) — resolves{{dot.path}}placeholders inarguments_template, delegates toToolExecutor, passesexecution_receipt_idWorkflowManager.start_run()(idempotent, requiresACTIVEdefinition),get_run(), andlist_runs()with backgroundasyncio.Taskscheduling and a dedicated DB sessionPostgresWorkflowStorerun/node-execution persistence with optimistic concurrency (checkpoint_version)app/ai/deps.py; extendsToolExecutionContextwith optionalexecution_receipt_idTest plan
uv run python -m pytest tests/ai/workflow/ -q— 88 passeduv run python -m pytest -q --cov=app --cov-fail-under=80— 1397 passed, 89.20% coverageuv run python -m ruff check app tests— cleanuv run python -m ruff format --check app tests— cleanuv run pyright app tests— cleanACTIVEdefinition and pollget_run()until terminal statusSummary by CodeRabbit
New Features
Bug Fixes
Documentation