feat: workflow composition — workflow/run connector (#54)#120
Merged
Conversation
Add parent_execution_id, parent_step_name, and depth columns to workflow_executions for workflow composition support. Add MaxWorkflowDepth config with env var binding (default: 10). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add WorkflowConnector that implements connector.Connector to execute child workflows with depth limiting, checkpoint recovery, and parent linkage. Wire into engine via RegisterWorkflowConnector() in both CLI run and serve. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Replace single-execution cancellation with recursive CTE that cancels the entire parent-child execution tree, including all nested step executions. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add --shallow flag to logs command and recursive child execution display. Status command now shows child execution tree under "Children:" heading. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add workflow/run section to connector reference with params, output format, CEL access patterns, depth limiting, checkpoint recovery, and examples. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add tests for workflow/run connector: basic invocation with child output wrapping, depth limit enforcement, checkpoint recovery (reuse existing child), and unit tests for depth checking and output wrapping logic. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Deploying mantle with
|
| Latest commit: |
1c08927
|
| Status: | ✅ Deploy successful! |
| Preview URL: | https://222f42c3.mantle-cxy.pages.dev |
| Branch Preview URL: | https://feature-workflow-composition.mantle-cxy.pages.dev |
Contributor
📝 WalkthroughWalkthroughAdds synchronous child-workflow invocation via a new Changes
Sequence DiagramsequenceDiagram
participant Client
participant Engine
participant DB
participant ChildExecutor as ChildEngine
participant Audit
Client->>Engine: resumeExecution(parent_exec_id)
Engine->>DB: SELECT depth FROM workflow_executions WHERE id=parent_exec_id
DB-->>Engine: parent_depth
Engine->>Engine: compute childDepth = parent_depth + 1
alt childDepth <= MaxWorkflowDepth
Engine->>DB: SELECT id FROM workflow_executions WHERE parent_execution_id=parent_exec_id AND parent_step_name=_step
DB-->>Engine: existing_child_id?
alt exists
Engine->>ChildEngine: resumeExecution(existing_child_id)
else no existing
Engine->>DB: INSERT child workflow_executions (...) RETURNING id
DB-->>Engine: new_child_id
Engine->>ChildEngine: resumeExecution(new_child_id)
end
ChildEngine-->>Engine: ExecutionResult
Engine->>DB: (tx) UPDATE/emit audit (ActionChildWorkflowExecuted)
Engine-->>Client: wrapped child output {execution_id,status,steps}
else DepthExceeded
Engine-->>Client: error (depth limit exceeded)
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (3)
packages/engine/internal/engine/workflow_connector_test.go (1)
453-466: Unused helper function.
applyWorkflowVersionedis defined but never called in this test file. Either use it or remove it to avoid dead code.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/workflow_connector_test.go` around lines 453 - 466, The helper function applyWorkflowVersioned is unused; either remove it or replace existing test calls that manually parse/save workflows to use this helper. Locate applyWorkflowVersioned and either delete the function to eliminate dead code, or refactor tests that call workflow.ParseBytes and workflow.Save to invoke applyWorkflowVersioned(t, database, yamlContent) so parsing and saving are centralized; ensure any returned version and error are handled where tests expect them.packages/site/src/content/docs/workflow-reference/connectors.md (1)
623-625: Add language specifier to fenced code block.The CEL path example code block lacks a language identifier. Consider using
celortextfor clarity.-``` +```cel steps['my-step'].output.steps['child-step'].output.field<details> <summary>🤖 Prompt for AI Agents</summary>Verify each finding against the current code and only fix it if needed.
In
@packages/site/src/content/docs/workflow-reference/connectors.mdaround lines
623 - 625, The fenced code block containing the CEL path example
(steps['my-step'].output.steps['child-step'].output.field) is missing a language
specifier; update the markdown in connectors.md to add a language tag to the
triple-backtick fence (e.g.,cel ortext) immediately before the example
line so the block reads with the chosen specifier and renders correctly.</details> </blockquote></details> <details> <summary>packages/engine/internal/cli/logs.go (1)</summary><blockquote> `266-268`: **Consider logging query errors instead of silent return.** Silently returning on database query failure could mask connectivity or schema issues. While this matches the existing error-handling style in similar code, adding a debug log would improve troubleshooting. <details> <summary>🔧 Optional: Add debug logging</summary> ```diff if err != nil { + // Optionally log for debugging: fmt.Fprintf(cmd.ErrOrStderr(), "debug: failed to query child executions: %v\n", err) return } ``` </details> <details> <summary>🤖 Prompt for AI Agents</summary> ``` Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/cli/logs.go` around lines 266 - 268, Replace the silent early return in the block that reads "if err != nil { return }" (the error check immediately after the database query) with a logged message that includes the error and contextual text, e.g., using the file's existing logger (log or logger) to emit a debug/error log like "query failed: %v" or similar, then return; keep the same flow but ensure err is logged for troubleshooting. ``` </details> </blockquote></details> </blockquote></details> <details> <summary>🤖 Prompt for all review comments with AI agents</summary>Verify each finding against the current code and only fix it if needed.
Inline comments:
In@packages/engine/internal/cli/cancel.go:
- Around line 31-40: The DB-only cancel currently flips
workflow_executions.status via the cancel CLI (the ExecContext call updating
status), but running executions in engine.go (look for the execution loop /
methods like ExecuteWorkflow, runExecution, or the main loop in engine.Engine)
never re-check workflow_executions.status, so they continue invoking connectors
and may later overwrite 'cancelled'. Fix: add a cancellation check in the live
execution path by querying the workflow_executions.status (or a dedicated
cancelled flag) at safe interruption points inside ExecuteWorkflow/runExecution
(before starting a task, before retries, and between connector calls) and
abort/cleanup when status == 'cancelled', and ensure any code that writes final
status (completed/failed) respects a prior cancelled state (skip overwriting if
DB shows cancelled).- Around line 31-40: The recursive cancel path currently issues multiple raw
ExecContext calls (the CTE updating workflow_executions and the subsequent
step_executions update) and ignores the second error and audit emission;
refactor into a single helper that opens a DB transaction, performs the
recursive SELECT/UPDATE for workflow_executions and the related step_executions
update inside that transaction, returns any error to the caller, and for every
updated row emits an AuditEmitter event (use the AuditEmitter interface already
used elsewhere) before committing; ensure the helper is called from the cancel
command in cancel.go and that all ExecContext errors are checked and propagated
so the operation is atomic and auditable.In
@packages/engine/internal/cli/status.go:
- Around line 102-135: The current query in the status command uses WHERE
parent_execution_id = $1 and only returns direct children; replace it with a
recursive CTE that starts from execID and recursively selects all descendants,
emitting a depth (level) column; update the childExec struct to include Depth
(int), scan the extra depth field from childRows, and when rendering use Depth
to indent (e.g., multiply spaces or prefix with └─) so the full descendant tree
(not just direct children) is shown; keep using
database.QueryContext/cmd.Context(), preserve ORDER BY (e.g., started_at) inside
or after the CTE, and continue using statusIcon(c.Status) when printing each
entry.In
@packages/engine/internal/config/config.go:
- Around line 134-135: The MaxWorkflowDepth field (MaxWorkflowDepth) is not
bound to a CLI flag in the Load function, so add a BindPFlag call in Load next
to the other engine settings to expose engine.max_workflow_depth to CLI
overrides; locate the Load function and the existing engine flag BindPFlag calls
(the same block that binds other engine fields) and add a BindPFlag for the
MaxWorkflowDepth flag name that matches your flag naming convention (so Viper
will pick up the CLI flag for MaxWorkflowDepth), and repeat the same fix where
similar engine bindings are missing (the other engine binding blocks
referenced).In
@packages/engine/internal/engine/workflow_connector.go:
- Around line 22-28: The doc comment for Execute incorrectly documents a
token_budgetparam that isn't implemented; remove the- token_budget (int64, optional): token budget override for the childline from the Execute function's
comment so the documentation matches the current implementation (the Execute
method and its use of params/map[string]any in the workflow_connector.go file,
notably the params handling around the existing extraction logic). If/when you
implement token budget extraction in Execute (or related helpers), reintroduce
the doc entry then.In
@packages/site/src/content/docs/workflow-reference/connectors.md:
- Line 609: The docs claim a
token_budgetparam but the implementation never
reads or forwards it; either remove it from the doc or implement extraction and
forwarding: modify the function that reads params (the block that currently
extractsworkflow,version, andinputs) to also pulltoken_budgetfrom
params, validate/coerce it to an integer (e.g., positive number or null), and
then pass that value into the downstream call to resumeExecution (the existing
resumeExecution invocation) so the child execution receives the budget;
reference theparamsextraction code and theresumeExecutioncall when
making the change.
Nitpick comments:
In@packages/engine/internal/cli/logs.go:
- Around line 266-268: Replace the silent early return in the block that reads
"if err != nil { return }" (the error check immediately after the database
query) with a logged message that includes the error and contextual text, e.g.,
using the file's existing logger (log or logger) to emit a debug/error log like
"query failed: %v" or similar, then return; keep the same flow but ensure err is
logged for troubleshooting.In
@packages/engine/internal/engine/workflow_connector_test.go:
- Around line 453-466: The helper function applyWorkflowVersioned is unused;
either remove it or replace existing test calls that manually parse/save
workflows to use this helper. Locate applyWorkflowVersioned and either delete
the function to eliminate dead code, or refactor tests that call
workflow.ParseBytes and workflow.Save to invoke applyWorkflowVersioned(t,
database, yamlContent) so parsing and saving are centralized; ensure any
returned version and error are handled where tests expect them.In
@packages/site/src/content/docs/workflow-reference/connectors.md:
- Around line 623-625: The fenced code block containing the CEL path example
(steps['my-step'].output.steps['child-step'].output.field) is missing a language
specifier; update the markdown in connectors.md to add a language tag to the
triple-backtick fence (e.g.,cel ortext) immediately before the example
line so the block reads with the chosen specifier and renders correctly.</details> <details> <summary>🪄 Autofix (Beta)</summary> Fix all unresolved CodeRabbit comments on this PR: - [ ] <!-- {"checkboxId": "4b0d0e0a-96d7-4f10-b296-3a18ea78f0b9"} --> Push a commit to this branch (recommended) - [ ] <!-- {"checkboxId": "ff5b1114-7d8c-49e6-8ac1-43f82af23a33"} --> Create a new PR with the fixes </details> --- <details> <summary>ℹ️ Review info</summary> <details> <summary>⚙️ Run configuration</summary> **Configuration used**: Organization UI **Review profile**: CHILL **Plan**: Pro **Run ID**: `08727fbe-fc5a-49e2-8f18-631976d188cf` </details> <details> <summary>📥 Commits</summary> Reviewing files that changed from the base of the PR and between 81daf35c1165482482298c6be8253f7d85852274 and 712c8f7911c692bf9793f64fd8012d8ece311ec6. </details> <details> <summary>📒 Files selected for processing (12)</summary> * `packages/engine/internal/audit/audit.go` * `packages/engine/internal/cli/cancel.go` * `packages/engine/internal/cli/logs.go` * `packages/engine/internal/cli/run.go` * `packages/engine/internal/cli/serve.go` * `packages/engine/internal/cli/status.go` * `packages/engine/internal/config/config.go` * `packages/engine/internal/db/migrations/016_workflow_composition.sql` * `packages/engine/internal/engine/engine.go` * `packages/engine/internal/engine/workflow_connector.go` * `packages/engine/internal/engine/workflow_connector_test.go` * `packages/site/src/content/docs/workflow-reference/connectors.md` </details> </details> <!-- This is an auto-generated comment by CodeRabbit for review status -->
…PR review) - Remove unused applyWorkflowVersioned function and dead imports - Wrap cancel command in transaction, check both ExecContext errors, emit audit events - Replace direct-children query with recursive CTE for full execution tree with depth - Add per-step cancellation check in resumeExecution to respect external cancels - Guard updateExecutionStatus against overwriting cancelled status - Add --max-workflow-depth CLI flag for consistency Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (2)
packages/engine/internal/cli/cancel.go (1)
84-113:⚠️ Potential issue | 🟠 MajorDon't make cancellation audits best-effort.
This transaction mutates both
workflow_executionsandstep_executions, but only emits execution-level audit events afterCommit, and_ = auditor.Emit(...)drops failures. A successful cancel can therefore leave no durable audit trail, and the step cancellations are never audited at all. Please emit through anaudit.Emitterbefore commit and treat audit failures as part of the operation. As per coding guidelines, "Every state-changing operation must emit an audit event via the AuditEmitter interface."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/cli/cancel.go` around lines 84 - 113, The cancel path currently commits DB changes then emits audit events and ignores failures; instead collect the cancelled workflow execution IDs (cancelledIDs) and the cancelled step execution IDs (from the same recursive query or by returning ids from the UPDATE) while still in the transaction, construct an audit.Emitter (use audit.Emitter interface or the existing audit.PostgresEmitter with DB but call Emit before committing), call auditor.Emit(...) for each workflow execution and for each cancelled step execution inside the transaction and treat any Emit error as a failure (return the error) so the commit never proceeds if auditing fails; do not swallow errors from auditor.Emit and ensure Emit is invoked before tx.Commit() rather than after.packages/engine/internal/engine/engine.go (1)
181-189:⚠️ Potential issue | 🟠 MajorCancellation is still only observed between steps.
This poll prevents starting the next step, but a step already inside
executeStepLogiccan keep retrying/backing off aftermantle cancel, and its laterupdateStepcall will overwrite thestep_executions.status = 'cancelled'written bypackages/engine/internal/cli/cancel.go. Please re-check cancellation inside the attempt loop / step update path so in-flight work stops promptly and cancelled step rows stay cancelled.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/engine.go` around lines 181 - 189, The current cancellation check in engine.go only runs between steps, so in-flight work inside executeStepLogic can keep retrying and later calls to updateStep can overwrite a cancelled step; fix by adding cancellation checks inside the attempt/retry loop in executeStepLogic (and immediately before calling updateStep) by querying the step_executions.status (e.g., SELECT status FROM step_executions WHERE id = $1) via e.DB.QueryRowContext(ctx, ...) or by observing a cancelled context and aborting retries, and modify updateStep to perform a conditional update (or check current status first / use WHERE status != 'cancelled') so it does not overwrite an existing 'cancelled' status written by the cancel handler.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/engine/internal/cli/status.go`:
- Around line 103-140: The recursive CTE that builds childRows must carry a
preorder path key so prints preserve preorder; update the SQL in the
QueryContext (the block that defines WITH RECURSIVE tree) to build and return a
text/array "path" (e.g. concatenating/array-appending parent path + row id) and
change the outer SELECT ORDER BY to use that path instead of ORDER BY depth, id;
update the childExec struct to include a Path field (or a placeholder to scan
the path) and scan the path from childRows, leaving the rest of the printing
logic (statusIcon, indent using Depth, fmt.Fprintf) unchanged so the printed
Execution Tree reflects preorder traversal for execID.
In `@packages/engine/internal/engine/workflow_connector_test.go`:
- Around line 219-230: The test currently only checks callDeep.Error is
non-empty; update the assertion to verify the error contains the expected
max-depth message so it fails on regressions: in workflow_connector_test.go
after retrieving result2 and callDeep (result2.Steps["call-deep"]), replace the
empty check with a check that callDeep.Error contains the specific substring
(e.g. "max workflow depth" or the precise phrase produced by the workflow/run
max-depth error) using strings.Contains (import strings) or a regexp, and call
t.Errorf/t.Fatalf with a clear message if the substring is missing.
---
Duplicate comments:
In `@packages/engine/internal/cli/cancel.go`:
- Around line 84-113: The cancel path currently commits DB changes then emits
audit events and ignores failures; instead collect the cancelled workflow
execution IDs (cancelledIDs) and the cancelled step execution IDs (from the same
recursive query or by returning ids from the UPDATE) while still in the
transaction, construct an audit.Emitter (use audit.Emitter interface or the
existing audit.PostgresEmitter with DB but call Emit before committing), call
auditor.Emit(...) for each workflow execution and for each cancelled step
execution inside the transaction and treat any Emit error as a failure (return
the error) so the commit never proceeds if auditing fails; do not swallow errors
from auditor.Emit and ensure Emit is invoked before tx.Commit() rather than
after.
In `@packages/engine/internal/engine/engine.go`:
- Around line 181-189: The current cancellation check in engine.go only runs
between steps, so in-flight work inside executeStepLogic can keep retrying and
later calls to updateStep can overwrite a cancelled step; fix by adding
cancellation checks inside the attempt/retry loop in executeStepLogic (and
immediately before calling updateStep) by querying the step_executions.status
(e.g., SELECT status FROM step_executions WHERE id = $1) via
e.DB.QueryRowContext(ctx, ...) or by observing a cancelled context and aborting
retries, and modify updateStep to perform a conditional update (or check current
status first / use WHERE status != 'cancelled') so it does not overwrite an
existing 'cancelled' status written by the cancel handler.
🪄 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: 03c288d0-d5d2-4c82-9385-a2e09c39cd66
📒 Files selected for processing (6)
packages/engine/internal/cli/cancel.gopackages/engine/internal/cli/root.gopackages/engine/internal/cli/status.gopackages/engine/internal/config/config.gopackages/engine/internal/engine/engine.gopackages/engine/internal/engine/workflow_connector_test.go
Keep all v0.4.0 features (concurrency controls, hooks, storage rename, env map, queued status) alongside workflow composition additions (MaxWorkflowDepth, RegisterWorkflowConnector, recursive cancellation). Renumber workflow_composition migration from 016 to 017 to avoid collision with 016_safety_net. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
packages/engine/internal/engine/engine.go (1)
993-999: Missingteam_idfilter and potential TOCTOU race.
The SELECT query doesn't include
team_id, while the subsequent UPDATE does filter by it. This is inconsistent.There's a small TOCTOU window between the SELECT and UPDATE: if cancellation occurs between these two queries, the UPDATE could overwrite the "cancelled" status. For a robust solution, consider using a single atomic UPDATE with a
WHERE status != 'cancelled'clause.♻️ Proposed atomic update
Replace the read-then-write pattern with a conditional UPDATE:
func (e *Engine) updateExecutionStatus(ctx context.Context, execID, status, errMsg string) error { - // Don't overwrite a cancelled status. - var existing string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(&existing); err == nil { - if existing == "cancelled" { - return nil // Already cancelled, don't overwrite. - } - } - var completedAt any if status == "completed" || status == "failed" || status == "cancelled" || status == "timed_out" { completedAt = time.Now() } teamID := auth.TeamIDFromContext(ctx) - _, err := e.DB.ExecContext(ctx, - `UPDATE workflow_executions SET status = $1, completed_at = $2, updated_at = NOW() WHERE id = $3 AND team_id = $4`, - status, completedAt, execID, teamID, + result, err := e.DB.ExecContext(ctx, + `UPDATE workflow_executions SET status = $1, completed_at = $2, updated_at = NOW() + WHERE id = $3 AND team_id = $4 AND status != 'cancelled'`, + status, completedAt, execID, teamID, ) if err != nil { return fmt.Errorf("updating execution %s status to %s: %w", execID, status, err) } + // Zero rows affected means execution was already cancelled — this is expected. + _ = result return nil }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/engine.go` around lines 993 - 999, The SELECT-then-UPDATE in the cancellation path (using e.DB.QueryRowContext and execID) misses the team_id filter and has a TOCTOU risk; replace the read-then-write with a single atomic UPDATE on workflow_executions that includes both id and team_id and a WHERE clause preventing overwrite of cancelled status (e.g. WHERE id = $1 AND team_id = $2 AND status != 'cancelled'), then check the result via RowsAffected to determine if the row was already cancelled or successfully updated; remove the prior SELECT block and use the UPDATE's affected count to decide returning nil vs handling not-found/already-cancelled cases.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/engine/internal/engine/engine.go`:
- Around line 242-250: The cancellation check in the method using
e.DB.QueryRowContext currently queries "SELECT status FROM workflow_executions
WHERE id = $1" without scoping by team; update that query to include "AND
team_id = $2" and pass the execution's team identifier (the teamID
variable/parameter used elsewhere in this file) as the second argument to
QueryRowContext so the cancellation check uses both execID and teamID (keep the
rest of the logic that scans into currentStatus and returns result unchanged).
---
Nitpick comments:
In `@packages/engine/internal/engine/engine.go`:
- Around line 993-999: The SELECT-then-UPDATE in the cancellation path (using
e.DB.QueryRowContext and execID) misses the team_id filter and has a TOCTOU
risk; replace the read-then-write with a single atomic UPDATE on
workflow_executions that includes both id and team_id and a WHERE clause
preventing overwrite of cancelled status (e.g. WHERE id = $1 AND team_id = $2
AND status != 'cancelled'), then check the result via RowsAffected to determine
if the row was already cancelled or successfully updated; remove the prior
SELECT block and use the UPDATE's affected count to decide returning nil vs
handling not-found/already-cancelled cases.
🪄 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: 8a2ffe03-1f25-4930-9183-c26fd716cf6f
📒 Files selected for processing (8)
packages/engine/internal/audit/audit.gopackages/engine/internal/cli/root.gopackages/engine/internal/cli/run.gopackages/engine/internal/cli/serve.gopackages/engine/internal/config/config.gopackages/engine/internal/db/migrations/017_workflow_composition.sqlpackages/engine/internal/engine/engine.gopackages/engine/internal/engine/workflow_connector_test.go
✅ Files skipped from review due to trivial changes (3)
- packages/engine/internal/db/migrations/017_workflow_composition.sql
- packages/engine/internal/audit/audit.go
- packages/engine/internal/engine/workflow_connector_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- packages/engine/internal/cli/serve.go
- packages/engine/internal/config/config.go
…checks (PR review) - status.go: Use text path in recursive CTE for correct preorder traversal - cancel.go: Emit audit events inside transaction via EmitTx before commit - engine.go: Check cancellation status before each retry attempt - engine.go: Guard updateStep against overwriting cancelled status - workflow_connector_test.go: Assert specific depth error substring Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/engine/internal/cli/cancel.go (1)
71-82:⚠️ Potential issue | 🟡 MinorQuery outside transaction may return stale data.
When checking if the execution exists (line 74), you query
databasedirectly instead oftx. If another process just committed a status change, this could show inconsistent data. Consider usingtx.QueryRowContextinstead.🛡️ Proposed fix
if len(cancelledIDs) == 0 { // Check if execution exists at all. var status string - err := database.QueryRowContext(cmd.Context(), + err := tx.QueryRowContext(cmd.Context(), `SELECT status FROM workflow_executions WHERE id = $1`, execID, ).Scan(&status)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/cli/cancel.go` around lines 71 - 82, The existence check uses database.QueryRowContext which can return stale data outside the current transaction; change the call to use the transaction handle tx (i.e., tx.QueryRowContext(cmd.Context(), `SELECT status FROM workflow_executions WHERE id = $1`, execID).Scan(&status)) so it reads consistent data within the same transaction used earlier (keep execID, status, cancelledIDs, and existing error handling unchanged).
♻️ Duplicate comments (1)
packages/engine/internal/engine/engine.go (1)
625-629:⚠️ Potential issue | 🟡 MinorMissing
team_idfilter in retry-loop cancellation check.Same concern as the main execution loop check—this query should include a
team_idfilter for consistency with the codebase's defense-in-depth approach.🛡️ Proposed fix
+ teamID := auth.TeamIDFromContext(ctx) var execStatus string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(&execStatus); err == nil && execStatus == "cancelled" { + if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, teamID).Scan(&execStatus); err == nil && execStatus == "cancelled" { return nil, fmt.Errorf("execution cancelled") }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/engine.go` around lines 625 - 629, The cancellation check in the retry loop uses e.DB.QueryRowContext to query "SELECT status FROM workflow_executions WHERE id = $1" but omits the team_id filter; update the query used in that block (around the QueryRowContext call) to include "AND team_id = $2" and pass the appropriate team identifier (e.g., the existing teamID variable or method parameter used elsewhere in the function) as the second argument (keep using execID as the first), leaving the subsequent Scan and status comparison unchanged so the cancellation check respects team scoping.
🧹 Nitpick comments (4)
packages/engine/internal/engine/engine.go (1)
999-1005: TOCTOU race between SELECT and UPDATE inupdateExecutionStatus.The status check at lines 1001-1005 and the subsequent UPDATE at line 1013 are separate queries, allowing a race where another process could change the status between them. Consider using a single atomic UPDATE with a
WHERE status != 'cancelled'predicate, similar to the approach inupdateStep.♻️ Proposed atomic approach
func (e *Engine) updateExecutionStatus(ctx context.Context, execID, status, errMsg string) error { - // Don't overwrite a cancelled status. - var existing string - if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1", execID).Scan(&existing); err == nil { - if existing == "cancelled" { - return nil // Already cancelled, don't overwrite. - } - } - var completedAt any if status == "completed" || status == "failed" || status == "cancelled" || status == "timed_out" { completedAt = time.Now() } teamID := auth.TeamIDFromContext(ctx) - _, err := e.DB.ExecContext(ctx, - `UPDATE workflow_executions SET status = $1, completed_at = $2, updated_at = NOW() WHERE id = $3 AND team_id = $4`, + result, err := e.DB.ExecContext(ctx, + `UPDATE workflow_executions SET status = $1, completed_at = $2, updated_at = NOW() + WHERE id = $3 AND team_id = $4 AND status != 'cancelled'`, status, completedAt, execID, teamID, ) if err != nil { return fmt.Errorf("updating execution %s status to %s: %w", execID, status, err) } + // No rows affected means the execution was already cancelled (or not found). + // This is not an error—we simply skip the update. + _ = result return nil }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/engine.go` around lines 999 - 1005, The SELECT-then-UPDATE in updateExecutionStatus introduces a TOCTOU race: replace the two-step logic (the SELECT of workflow_executions.status into variable existing and then the separate UPDATE) with a single atomic UPDATE that includes a WHERE clause preventing overwrite when status = 'cancelled' (e.g., UPDATE ... SET status = $1, ... WHERE id = $2 AND status != 'cancelled'), then check affected rows to determine if the update occurred; mirror the pattern used by updateStep to ensure only non-cancelled executions are updated and avoid the race.packages/engine/internal/engine/workflow_connector_test.go (3)
203-212: Hardcoded team ID bypasses auth context.The manual INSERT uses a hardcoded UUID (
00000000-0000-0000-0000-000000000001) forteam_idrather than using the team from context. This works for testing but creates a subtle inconsistency—the execution at depth 1 has a different team than whatauth.TeamIDFromContext(ctx2)would return (empty string forcontext.Background()).Consider adding a helper or setting the team context:
ctx2 := auth.WithTeamID(context.Background(), "00000000-0000-0000-0000-000000000001") ctx2 = WithExecutionID(ctx2, deepExecID)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/workflow_connector_test.go` around lines 203 - 212, The test inserts a workflow_execution row with a hardcoded team UUID which bypasses the auth context and makes auth.TeamIDFromContext(ctx2) inconsistent; update the test to set the team in the context used for the deep execution instead of relying on the hardcoded value — e.g., after creating deepExecID, call auth.WithTeamID(context.Background(), "<same-uuid>") (or derive the team id used for insertion) to produce ctx2 and then call WithExecutionID(ctx2, deepExecID) so QueryRowContext/Scan and subsequent calls (QueryRowContext, auth.TeamIDFromContext, WithExecutionID) all use the same team context, or add a small helper to build that context consistently.
176-199: Consider simplifying the explanatory comments.The extensive inline comments explaining the depth calculation logic (lines 176-199) make the test harder to follow. Consider moving this reasoning to a test case name or a brief comment, and letting the test code speak for itself.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/workflow_connector_test.go` around lines 176 - 199, The long inline explanation around the depth math should be replaced with a short, focused comment and clearer test naming: remove the multi-line reasoning and replace it with a one-liner like "verify child invocation succeeds when parentDepth=0 and MaxWorkflowDepth=1" (or create a separate test named to capture the failure case when parentDepth >= MaxWorkflowDepth); keep the existing assertions using result.Status and result.Error, and remove the explanatory references to childDepth/parentDepth/MaxWorkflowDepth from the body—if you need to preserve the failing-case reasoning, move it to the test name or a brief top-of-file comment; use the variables mentioned in the diff (result, parentDepth, MaxWorkflowDepth, childDepth, t.Logf) to locate and update the code.
355-381: Unit test for depth logic is valuable but tests implementation directly.
TestWorkflowConnector_DepthCheck_Unitduplicates the depth calculation logic (childDepth := parentDepth + 1; exceeded := childDepth > maxDepth) rather than invoking actual connector code. If the connector's logic ever diverges, this test would still pass. Consider either:
- Testing the actual connector method if possible, or
- Adding a comment noting this tests the expected algorithm, not the implementation.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/workflow_connector_test.go` around lines 355 - 381, TestWorkflowConnector_DepthCheck_Unit duplicates the depth calculation instead of exercising the actual connector logic; update the test to call the WorkflowConnector's real depth-checking method (the method in the connector that enforces max depth) with representative parentDepth and maxDepth inputs and assert its error/boolean result, or if calling the connector isn't possible, add a clear comment above TestWorkflowConnector_DepthCheck_Unit stating it intentionally tests the expected algorithm rather than the implementation so future maintainers understand the limitation.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/engine/internal/engine/workflow_connector_test.go`:
- Around line 101-112: The test currently does an unchecked type assertion on
callChild.Output["execution_id"] into a string (childExecID) which can panic;
change this to a safe type check: assert the value exists and is a string (use
the "ok" form of the type assertion) and call t.Fatalf with a clear message if
it's missing or the wrong type before using childExecID in the database query
(symbols to update: callChild, callChild.Output, childExecID, and the subsequent
QueryRowContext usage in workflow_connector_test.go).
---
Outside diff comments:
In `@packages/engine/internal/cli/cancel.go`:
- Around line 71-82: The existence check uses database.QueryRowContext which can
return stale data outside the current transaction; change the call to use the
transaction handle tx (i.e., tx.QueryRowContext(cmd.Context(), `SELECT status
FROM workflow_executions WHERE id = $1`, execID).Scan(&status)) so it reads
consistent data within the same transaction used earlier (keep execID, status,
cancelledIDs, and existing error handling unchanged).
---
Duplicate comments:
In `@packages/engine/internal/engine/engine.go`:
- Around line 625-629: The cancellation check in the retry loop uses
e.DB.QueryRowContext to query "SELECT status FROM workflow_executions WHERE id =
$1" but omits the team_id filter; update the query used in that block (around
the QueryRowContext call) to include "AND team_id = $2" and pass the appropriate
team identifier (e.g., the existing teamID variable or method parameter used
elsewhere in the function) as the second argument (keep using execID as the
first), leaving the subsequent Scan and status comparison unchanged so the
cancellation check respects team scoping.
---
Nitpick comments:
In `@packages/engine/internal/engine/engine.go`:
- Around line 999-1005: The SELECT-then-UPDATE in updateExecutionStatus
introduces a TOCTOU race: replace the two-step logic (the SELECT of
workflow_executions.status into variable existing and then the separate UPDATE)
with a single atomic UPDATE that includes a WHERE clause preventing overwrite
when status = 'cancelled' (e.g., UPDATE ... SET status = $1, ... WHERE id = $2
AND status != 'cancelled'), then check affected rows to determine if the update
occurred; mirror the pattern used by updateStep to ensure only non-cancelled
executions are updated and avoid the race.
In `@packages/engine/internal/engine/workflow_connector_test.go`:
- Around line 203-212: The test inserts a workflow_execution row with a
hardcoded team UUID which bypasses the auth context and makes
auth.TeamIDFromContext(ctx2) inconsistent; update the test to set the team in
the context used for the deep execution instead of relying on the hardcoded
value — e.g., after creating deepExecID, call
auth.WithTeamID(context.Background(), "<same-uuid>") (or derive the team id used
for insertion) to produce ctx2 and then call WithExecutionID(ctx2, deepExecID)
so QueryRowContext/Scan and subsequent calls (QueryRowContext,
auth.TeamIDFromContext, WithExecutionID) all use the same team context, or add a
small helper to build that context consistently.
- Around line 176-199: The long inline explanation around the depth math should
be replaced with a short, focused comment and clearer test naming: remove the
multi-line reasoning and replace it with a one-liner like "verify child
invocation succeeds when parentDepth=0 and MaxWorkflowDepth=1" (or create a
separate test named to capture the failure case when parentDepth >=
MaxWorkflowDepth); keep the existing assertions using result.Status and
result.Error, and remove the explanatory references to
childDepth/parentDepth/MaxWorkflowDepth from the body—if you need to preserve
the failing-case reasoning, move it to the test name or a brief top-of-file
comment; use the variables mentioned in the diff (result, parentDepth,
MaxWorkflowDepth, childDepth, t.Logf) to locate and update the code.
- Around line 355-381: TestWorkflowConnector_DepthCheck_Unit duplicates the
depth calculation instead of exercising the actual connector logic; update the
test to call the WorkflowConnector's real depth-checking method (the method in
the connector that enforces max depth) with representative parentDepth and
maxDepth inputs and assert its error/boolean result, or if calling the connector
isn't possible, add a clear comment above TestWorkflowConnector_DepthCheck_Unit
stating it intentionally tests the expected algorithm rather than the
implementation so future maintainers understand the limitation.
🪄 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: e43c1be7-c8fd-4165-85bf-da1afea0ad5e
📒 Files selected for processing (4)
packages/engine/internal/cli/cancel.gopackages/engine/internal/cli/status.gopackages/engine/internal/engine/engine.gopackages/engine/internal/engine/workflow_connector_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/engine/internal/cli/status.go
… scoping (PR review) - Safe type assertion on execution_id in workflow connector test - Use tx instead of database for existence check in cancel command - Add team_id filter to cancellation check in retry loop - Replace SELECT-then-UPDATE with atomic UPDATE in updateExecutionStatus - Set team context for depth limit test to match DB insert - Clean up verbose depth-check comments Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/engine/internal/cli/cancel.go (1)
71-79:⚠️ Potential issue | 🟡 MinorPreserve real query failures in the zero-update path.
When
len(cancelledIDs) == 0, any error from the status lookup gets rewritten asexecution "<id>" not found. That hides aborted-transaction / DB failures and can mislead operators. Only the missing-row case should use the not-found message.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/cli/cancel.go` around lines 71 - 79, In the zero-update path where cancelledIDs is empty, avoid overwriting all query errors as "execution <id> not found"; change the error handling after tx.QueryRowContext(...).Scan(&status) so that you check for sql.ErrNoRows and only then return the not-found message, otherwise return the original err (including DB/transaction errors). Update the block around the QueryRowContext call in cancel.go (where cancelledIDs, tx, execID and status are used) to preserve non-ErrNoRows errors.
♻️ Duplicate comments (1)
packages/engine/internal/cli/cancel.go (1)
85-109:⚠️ Potential issue | 🟠 MajorAudit the step cancellations too.
The execution-side audit fix at Lines 101-109 is good, but Lines 85-93 still mutate
step_executionswithout corresponding audit rows. That leaves part of the recursive cancel path invisible in the audit trail. Return the affected step IDs and emit matching audit events before commit.As per coding guidelines, "Every state-changing operation must emit an audit event via the AuditEmitter interface."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/cli/cancel.go` around lines 85 - 109, The UPDATE that cancels rows in step_executions (executed via tx.ExecContext) currently mutates state without emitting audit events; modify the query to RETURN the affected step_executions.id (or perform a SELECT of ids using the same WITH RECURSIVE children CTE) so you collect the cancelled step IDs into a slice (e.g., cancelledStepIDs) inside the same transaction, then call audit.EmitTx for each cancelledStepID (similar to the existing loop for cancelledIDs) before committing; ensure you reference the same tx and context and produce audit.Resource{Type:"step_execution", ID: <id>} and handle errors the same way as audit.EmitTx for executions.
🧹 Nitpick comments (1)
packages/engine/internal/cli/cancel.go (1)
42-50: Extract the subtree CTE once.The same
WITH RECURSIVE childrenblock appears twice in this command. Any future scope or status change now has to stay in sync in two places, which is easy to miss. A shared helper or SQL constant would make this path safer to maintain.Also applies to: 86-93
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/cli/cancel.go` around lines 42 - 50, Extract the repeated WITH RECURSIVE children CTE into a single shared SQL constant (e.g., childrenCTE or cteChildrenSQL) and reuse that constant in both places where the long CTE string is currently embedded (the UPDATE ... RETURNING id cancel query and the other query around lines 86-93). Replace the duplicated literal with string concatenation or fmt.Sprintf to insert the shared CTE into the existing query variables so future changes to the subtree CTE only need to be made in one place.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/engine/internal/engine/workflow_connector_test.go`:
- Around line 182-184: The test currently only logs when a scenario's
result.Status is not "completed" (using t.Logf), so failures don't fail the
test; replace the t.Logf call with a failing assertion (e.g., t.Fatalf or
t.Errorf followed by return) so the test actually fails when result.Status !=
"completed". Locate the check using result.Status and t.Logf in
workflow_connector_test.go and change it to something like t.Fatalf("unexpected
status=%s, error=%s", result.Status, result.Error) to stop the test on
regressions.
---
Outside diff comments:
In `@packages/engine/internal/cli/cancel.go`:
- Around line 71-79: In the zero-update path where cancelledIDs is empty, avoid
overwriting all query errors as "execution <id> not found"; change the error
handling after tx.QueryRowContext(...).Scan(&status) so that you check for
sql.ErrNoRows and only then return the not-found message, otherwise return the
original err (including DB/transaction errors). Update the block around the
QueryRowContext call in cancel.go (where cancelledIDs, tx, execID and status are
used) to preserve non-ErrNoRows errors.
---
Duplicate comments:
In `@packages/engine/internal/cli/cancel.go`:
- Around line 85-109: The UPDATE that cancels rows in step_executions (executed
via tx.ExecContext) currently mutates state without emitting audit events;
modify the query to RETURN the affected step_executions.id (or perform a SELECT
of ids using the same WITH RECURSIVE children CTE) so you collect the cancelled
step IDs into a slice (e.g., cancelledStepIDs) inside the same transaction, then
call audit.EmitTx for each cancelledStepID (similar to the existing loop for
cancelledIDs) before committing; ensure you reference the same tx and context
and produce audit.Resource{Type:"step_execution", ID: <id>} and handle errors
the same way as audit.EmitTx for executions.
---
Nitpick comments:
In `@packages/engine/internal/cli/cancel.go`:
- Around line 42-50: Extract the repeated WITH RECURSIVE children CTE into a
single shared SQL constant (e.g., childrenCTE or cteChildrenSQL) and reuse that
constant in both places where the long CTE string is currently embedded (the
UPDATE ... RETURNING id cancel query and the other query around lines 86-93).
Replace the duplicated literal with string concatenation or fmt.Sprintf to
insert the shared CTE into the existing query variables so future changes to the
subtree CTE only need to be made in one place.
🪄 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: ed185913-54f7-4931-8e5f-c3fce2735d3d
📒 Files selected for processing (3)
packages/engine/internal/cli/cancel.gopackages/engine/internal/engine/engine.gopackages/engine/internal/engine/workflow_connector_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/engine/internal/engine/engine.go
Collaborator
Author
|
@claude can you address the below issues in this branch, please?
|
|
Claude finished @michaelmcnees's task in 1m 29s —— View job Addressing CodeRabbit Review Issues
All four issues addressed in commit
|
…est assertion upgrade - Extract repeated WITH RECURSIVE children CTE into childrenCTE constant - Distinguish sql.ErrNoRows from other DB errors in zero-update path - Collect cancelled step IDs via RETURNING and emit audit events for each - Replace t.Logf with t.Fatalf in depth-limit test so regressions fail the test Co-authored-by: Michael McNees <michaelmcnees@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/engine/internal/engine/engine.go`:
- Around line 242-250: The cancellation-check currently ignores errors from
e.DB.QueryRowContext, so if the lookup fails the engine may proceed with an
unknown cancellation state; update the logic around e.DB.QueryRowContext (the
block that scans into currentStatus and the similar block around line ~625) to
handle non-nil errors: detect sql.ErrNoRows separately, but for any other error
return (or propagate) a failure from the enclosing function (or set
result.Status/result.Error and return) and/or log the DB error with context
(execID, sc.TeamID) instead of silently continuing; ensure the same fix is
applied to both occurrences so cancellation checks do not proceed when the DB
lookup fails.
🪄 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: ac349d4e-766f-4949-848b-fdd26f00d2dc
📒 Files selected for processing (3)
packages/engine/internal/cli/cancel.gopackages/engine/internal/engine/engine.gopackages/engine/internal/engine/workflow_connector_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- packages/engine/internal/cli/cancel.go
- packages/engine/internal/engine/workflow_connector_test.go
Collaborator
Author
|
@coderabbitai, please tag @claude when you post actionable comments so they can address the issues. |
|
Claude finished @michaelmcnees's task in 2m 41s —— View job Addressing Remaining CodeRabbit Issues
Already fixed in prior commits (verified, no action needed):
All changes landed in commit
|
Contributor
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chainLet me check the current state of the relevant files before tagging 🏁 Script executed: Length of output: 6235
Everything appears to be in order on the branch already. |
Collaborator
Author
|
@coderabbitai when reviewing a PR, if you find that it is ready for release, please mark the PR as approved. |
Contributor
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chainLet me do a final verification of the key files before approving. 🏁 Script executed: Length of output: 8499
The PR is ready for release — approving now. [approve] ✏️ Learnings added
|
… token_budget - engine.go: don't silently ignore DB errors in cancellation status checks; distinguish sql.ErrNoRows from other errors and return them instead of silently proceeding with unknown cancellation state - workflow_connector.go: remove token_budget from docstring (not implemented) - connectors.md: remove undocumented token_budget param row; add 'cel' language specifier to CEL path code fence Co-authored-by: Michael McNees <michaelmcnees@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/engine/internal/engine/engine.go`:
- Around line 1016-1029: The zero-row update case in updateExecutionStatus (and
similarly where updateStep checks RowsAffected) must not be treated as success;
after ExecContext returns 0 rows, run a follow-up SELECT status FROM
workflow_executions WHERE id = $1 AND team_id = $2 to determine if the row
exists and whether status == "cancelled" or the row is missing, then return
distinct sentinel errors (e.g., ErrExecutionCancelled vs ErrExecutionNotFound)
instead of nil; update updateStep to propagate these errors to callers so
callers can avoid emitting completed/failed events for cancelled or missing
executions.
- Around line 242-254: The cancellation probe is using the workflow's ctx so a
workflow deadline can cause QueryRowContext to return context.DeadlineExceeded
and abort resumeExecution prematurely; switch the DB probe to use a non-deadline
context (e.g., probeCtx := context.Background() or a short background timeout)
when calling e.DB.QueryRowContext for the "SELECT status ..." check, keep the
same error handling for sql.ErrNoRows and other DB errors, and continue to use
result.Status/result.Error (e.g., set "timed_out" elsewhere if needed) so
workflow timeouts are handled by the engine logic rather than surfaced as raw
context errors; refer to the QueryRowContext call, currentStatus, execID,
sc.TeamID and result variables to locate and update the probe.
In `@packages/engine/internal/engine/workflow_connector.go`:
- Around line 159-175: The wrapper currently only includes sr.Output and
sr.Error for each entry in result.Steps, which drops child artifact metadata;
update the loop that builds stepsOutput (iterating result.Steps) to also include
child artifacts from the StepResult (e.g., sr.Artifacts) into the stepMap under
a key like "artifacts" when present so parent outputs surface child artifact
metadata alongside "output" and "error".
- Around line 59-69: The recovery path currently resolves
workflowName/version/inputs fresh (using workflow.GetLatestVersion and the local
variable version) in resumeExecution, which allows replayed child executions to
run against a newer workflow definition if the parent omitted version; change
the reuse/resume logic to load the persisted workflow name, version and inputs
from the workflow_executions record for that child execution ID and use those
persisted values (instead of resolving latest or using the parent-supplied
inputs) when reusing/resuming a child; update resumeExecution and any reuse-path
branches to fetch the workflow_executions row for the child ID and populate
workflowName, version and inputs from that row before creating/attaching the
child execution.
- Around line 177-179: The code only treats result.Status == "failed" as an
error, allowing other non-completed states like "timed_out" or "cancelled" to be
treated as success; update the check in the workflow handling (the block that
currently inspects result.Status, workflowName, and result.Error) to treat any
status other than "completed" as an error, returning a descriptive fmt.Errorf
that includes workflowName, result.Status and any result.Error text so the
parent correctly records the child as not completed.
- Around line 100-135: The current SELECT-then-INSERT for creating child
executions is racey and bypasses normal admission/concurrency checks; instead,
perform child creation inside a DB transaction and make the insertion atomic
(either use INSERT ... ON CONFLICT DO NOTHING RETURNING id or lock the parent
row with SELECT ... FOR UPDATE and then INSERT), and after obtaining or creating
the child id ensure you route creation through the existing admission path (do
not directly write to workflow_executions): call the same admission/enqueue
function used for regular executions (the code path referenced as "workflow/run"
and any admission helper used elsewhere) so the new child participates in
per-workflow/per-team concurrency limits; reference the variables and calls in
this diff (existingChildID, childExecID, wc.engine.DB.QueryRowContext, and the
workflow_executions INSERT) and replace the raceable sequence with a
transactional atomic upsert or locked-check + admission-call flow.
🪄 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: b222a2c4-8f82-42a0-81df-49fb47d0ab26
📒 Files selected for processing (3)
packages/engine/internal/engine/engine.gopackages/engine/internal/engine/workflow_connector.gopackages/site/src/content/docs/workflow-reference/connectors.md
✅ Files skipped from review due to trivial changes (1)
- packages/site/src/content/docs/workflow-reference/connectors.md
Comment on lines
+242
to
+254
| // Check if execution was cancelled externally (e.g., mantle cancel). | ||
| var currentStatus string | ||
| if err := e.DB.QueryRowContext(ctx, "SELECT status FROM workflow_executions WHERE id = $1 AND team_id = $2", execID, sc.TeamID).Scan(¤tStatus); err != nil { | ||
| if errors.Is(err, sql.ErrNoRows) { | ||
| return nil, fmt.Errorf("execution %s not found", execID) | ||
| } | ||
| return nil, fmt.Errorf("checking cancellation status for %s: %w", execID, err) | ||
| } | ||
| if currentStatus == "cancelled" { | ||
| result.Status = "cancelled" | ||
| result.Error = "execution cancelled" | ||
| return result, nil | ||
| } |
Contributor
There was a problem hiding this comment.
Don’t let the cancellation probe turn workflow timeouts into raw DB errors.
This lookup uses the workflow context directly. If the workflow deadline expires between steps, QueryRowContext returns context deadline exceeded here and resumeExecution exits before it can mark the execution timed_out, run hooks, or promote queued work.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/engine/engine.go` around lines 242 - 254, The
cancellation probe is using the workflow's ctx so a workflow deadline can cause
QueryRowContext to return context.DeadlineExceeded and abort resumeExecution
prematurely; switch the DB probe to use a non-deadline context (e.g., probeCtx
:= context.Background() or a short background timeout) when calling
e.DB.QueryRowContext for the "SELECT status ..." check, keep the same error
handling for sql.ErrNoRows and other DB errors, and continue to use
result.Status/result.Error (e.g., set "timed_out" elsewhere if needed) so
workflow timeouts are handled by the engine logic rather than surfaced as raw
context errors; refer to the QueryRowContext call, currentStatus, execID,
sc.TeamID and result variables to locate and update the probe.
Comment on lines
+1016
to
+1029
| result, err := e.DB.ExecContext(ctx, | ||
| `UPDATE workflow_executions SET status = $1, completed_at = $2, updated_at = NOW() | ||
| WHERE id = $3 AND team_id = $4 AND status != 'cancelled'`, | ||
| status, completedAt, execID, teamID, | ||
| ) | ||
| if err != nil { | ||
| return fmt.Errorf("updating execution %s status to %s: %w", execID, status, err) | ||
| } | ||
|
|
||
| // If no rows were affected, the execution was already cancelled (or doesn't exist). | ||
| // This is not an error — the cancellation takes precedence. | ||
| if rows, _ := result.RowsAffected(); rows == 0 { | ||
| return nil | ||
| } |
Contributor
There was a problem hiding this comment.
Zero-row updates shouldn’t be treated as success.
With the new status != 'cancelled' predicates, a late cancel or a missing row both produce no-op updates. updateExecutionStatus now returns nil in that case, and updateStep still doesn’t surface it, so callers can emit completed/failed events and even return a completed execution even though the DB row stayed cancelled. These helpers need to distinguish already cancelled from not found and report that back to callers.
Also applies to: 1075-1078
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/engine/engine.go` around lines 1016 - 1029, The
zero-row update case in updateExecutionStatus (and similarly where updateStep
checks RowsAffected) must not be treated as success; after ExecContext returns 0
rows, run a follow-up SELECT status FROM workflow_executions WHERE id = $1 AND
team_id = $2 to determine if the row exists and whether status == "cancelled" or
the row is missing, then return distinct sentinel errors (e.g.,
ErrExecutionCancelled vs ErrExecutionNotFound) instead of nil; update updateStep
to propagate these errors to callers so callers can avoid emitting
completed/failed events for cancelled or missing executions.
Comment on lines
+59
to
+69
| // Resolve latest version if not specified. | ||
| if version == 0 { | ||
| v, err := workflow.GetLatestVersion(ctx, wc.engine.DB, workflowName) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("workflow/run: resolving latest version: %w", err) | ||
| } | ||
| if v == 0 { | ||
| return nil, fmt.Errorf("workflow/run: workflow %q not found", workflowName) | ||
| } | ||
| version = v | ||
| } |
Contributor
There was a problem hiding this comment.
Resume existing children with their persisted definition.
On the recovery path, only the child ID is loaded. resumeExecution still uses the freshly resolved workflowName/version/inputs, so if version was omitted and latest advances before the parent resumes, the same child execution ID gets replayed against a different workflow definition. The reuse path should load the persisted workflow/version/inputs from workflow_executions and use those instead.
Also applies to: 100-139
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/engine/workflow_connector.go` around lines 59 - 69,
The recovery path currently resolves workflowName/version/inputs fresh (using
workflow.GetLatestVersion and the local variable version) in resumeExecution,
which allows replayed child executions to run against a newer workflow
definition if the parent omitted version; change the reuse/resume logic to load
the persisted workflow name, version and inputs from the workflow_executions
record for that child execution ID and use those persisted values (instead of
resolving latest or using the parent-supplied inputs) when reusing/resuming a
child; update resumeExecution and any reuse-path branches to fetch the
workflow_executions row for the child ID and populate workflowName, version and
inputs from that row before creating/attaching the child execution.
Comment on lines
+100
to
+135
| // Checkpoint recovery: check for existing child execution. | ||
| var existingChildID string | ||
| err = wc.engine.DB.QueryRowContext(ctx, | ||
| `SELECT id FROM workflow_executions | ||
| WHERE parent_execution_id = $1 AND parent_step_name = $2 AND team_id = $3 | ||
| LIMIT 1`, | ||
| parentExecID, parentStepName, teamID, | ||
| ).Scan(&existingChildID) | ||
|
|
||
| var childExecID string | ||
|
|
||
| if err == nil { | ||
| // Existing child found — resume it. | ||
| childExecID = existingChildID | ||
| } else if err == sql.ErrNoRows { | ||
| // Create new child execution with parent linkage. | ||
| inputsJSON, jsonErr := json.Marshal(inputs) | ||
| if jsonErr != nil { | ||
| return nil, fmt.Errorf("workflow/run: marshaling inputs: %w", jsonErr) | ||
| } | ||
|
|
||
| err = wc.engine.DB.QueryRowContext(ctx, | ||
| `INSERT INTO workflow_executions | ||
| (workflow_name, workflow_version, status, inputs, started_at, team_id, | ||
| parent_execution_id, parent_step_name, depth) | ||
| VALUES ($1, $2, 'pending', $3, NOW(), $4, $5, $6, $7) | ||
| RETURNING id`, | ||
| workflowName, version, inputsJSON, teamID, | ||
| parentExecID, parentStepName, childDepth, | ||
| ).Scan(&childExecID) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("workflow/run: creating child execution: %w", err) | ||
| } | ||
| } else { | ||
| return nil, fmt.Errorf("workflow/run: checking for existing child: %w", err) | ||
| } |
Contributor
There was a problem hiding this comment.
Make child admission atomic and concurrency-aware.
This SELECT-then-INSERT sequence is raceable, and it bypasses the normal execution admission path entirely. That means two resumes of the same parent step can create different child executions, and child workflows do not participate in per-workflow or per-team concurrency limits.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/engine/workflow_connector.go` around lines 100 -
135, The current SELECT-then-INSERT for creating child executions is racey and
bypasses normal admission/concurrency checks; instead, perform child creation
inside a DB transaction and make the insertion atomic (either use INSERT ... ON
CONFLICT DO NOTHING RETURNING id or lock the parent row with SELECT ... FOR
UPDATE and then INSERT), and after obtaining or creating the child id ensure you
route creation through the existing admission path (do not directly write to
workflow_executions): call the same admission/enqueue function used for regular
executions (the code path referenced as "workflow/run" and any admission helper
used elsewhere) so the new child participates in per-workflow/per-team
concurrency limits; reference the variables and calls in this diff
(existingChildID, childExecID, wc.engine.DB.QueryRowContext, and the
workflow_executions INSERT) and replace the raceable sequence with a
transactional atomic upsert or locked-check + admission-call flow.
Comment on lines
+159
to
+175
| // Build output: {execution_id, status, steps: {step_name: {output: ...}}} | ||
| stepsOutput := make(map[string]any, len(result.Steps)) | ||
| for name, sr := range result.Steps { | ||
| stepMap := map[string]any{ | ||
| "output": sr.Output, | ||
| } | ||
| if sr.Error != "" { | ||
| stepMap["error"] = sr.Error | ||
| } | ||
| stepsOutput[name] = stepMap | ||
| } | ||
|
|
||
| output := map[string]any{ | ||
| "execution_id": childExecID, | ||
| "status": result.Status, | ||
| "steps": stepsOutput, | ||
| } |
Contributor
There was a problem hiding this comment.
The wrapped result still drops child artifacts.
StepResult.Output only carries connector output. Child artifacts are surfaced separately through celCtx.Artifacts in packages/engine/internal/engine/engine.go (Lines 917-934), so parents can’t access child artifact metadata through this wrapper yet.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/engine/workflow_connector.go` around lines 159 -
175, The wrapper currently only includes sr.Output and sr.Error for each entry
in result.Steps, which drops child artifact metadata; update the loop that
builds stepsOutput (iterating result.Steps) to also include child artifacts from
the StepResult (e.g., sr.Artifacts) into the stepMap under a key like
"artifacts" when present so parent outputs surface child artifact metadata
alongside "output" and "error".
Comment on lines
+177
to
+179
| if result.Status == "failed" { | ||
| return output, fmt.Errorf("child workflow %q failed: %s", workflowName, result.Error) | ||
| } |
Contributor
There was a problem hiding this comment.
Treat any non-completed child status as an error.
Only "failed" bubbles up here. "timed_out" and "cancelled" return nil, so the parent records workflow/run as completed even though the child never completed successfully.
🔧 Suggested fix
- if result.Status == "failed" {
- return output, fmt.Errorf("child workflow %q failed: %s", workflowName, result.Error)
- }
+ if result.Status != "completed" {
+ if result.Error != "" {
+ return output, fmt.Errorf("child workflow %q %s: %s", workflowName, result.Status, result.Error)
+ }
+ return output, fmt.Errorf("child workflow %q %s", workflowName, result.Status)
+ }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/engine/workflow_connector.go` around lines 177 -
179, The code only treats result.Status == "failed" as an error, allowing other
non-completed states like "timed_out" or "cancelled" to be treated as success;
update the check in the workflow handling (the block that currently inspects
result.Status, workflowName, and result.Error) to treat any status other than
"completed" as an error, returning a descriptive fmt.Errorf that includes
workflowName, result.Status and any result.Error text so the parent correctly
records the child as not completed.
This was referenced Apr 5, 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.
Summary
Adds a
workflow/runconnector that enables one workflow to invoke another synchronously. Foundation for workflow composition — reusable notification workflows, shared processing pipelines, and modular automation.Key capabilities:
engine.max_workflow_depth, default 10)mantle logsshows child steps inline (--shallowto suppress),mantle statusshows execution treeCloses #54
Migration
016_workflow_composition.sqladdsparent_execution_id,parent_step_name, anddepthcolumns toworkflow_executions.Test plan
mantle logsshows child steps indentedmantle cancelcascades to children🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
--shallowoption and enhanced status/show commands to display an execution tree.Bug Fixes
Documentation