v0.4.0 — The Safety Net Update#77
Merged
Merged
Conversation
Add a Version field to the Config struct, validated after Viper unmarshal: - Missing or 0 defaults to version 1 (backward compat) - Version 1 is accepted - Version 2+ returns a hard error directing users to upgrade mantle No env var binding — version is config-file-only. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…#75) The `tmp` config section and `TmpStorage` interface were misleadingly named -- they store execution artifacts, not temporary files. Rename to `storage`/ `StorageConfig`/`Storage` throughout the codebase. A deprecation fallback reads the old `tmp` YAML section and logs a warning, so existing configs continue to work during migration. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Replace team-scoped ReEncryptAll with a package-level RotateAll function that operates globally across all teams and accepts a caller-managed transaction. Update the CLI rotate-key command to manage its own tx lifecycle (begin/commit/rollback). Add ActionSecretKeyRotated audit action. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…49, #30) Add max_parallel_executions, on_limit, and hooks to Workflow struct; add max_parallel to Step struct; add HooksConfig type with on_success, on_failure, on_finish blocks. Validate all new fields including hook step uniqueness, depends_on rejection, and duration parsing. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…otion (#49) Add per-workflow and per-team concurrency limits using Postgres advisory locks. Executions that exceed limits are either queued or rejected based on the on_limit policy. Queued executions are promoted FIFO when a slot opens. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Register `hooks` and `execution` as top-level CEL variables so hook steps can reference hook outputs (hooks.<name>.output) and execution metadata (execution.status, execution.failed_step, etc.). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Implement lifecycle hooks that run after main workflow steps complete. Hook failures are best-effort and never alter workflow execution status. - Add hook audit actions (hook.step.started/completed/failed) - Add hook Prometheus metrics (hook_steps_total, hook_steps_failed_total) - Exclude hook steps from main DAG query (hook_block IS NULL filter) - Create hooks.go with executeHooks/executeHookBlock/recordHookStep - Wire hooks into resumeExecution with timeout/cancellation handling Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add `mantle retry <execution-id>` command that creates a new execution resuming from the failure point of a previous run. Completed upstream steps are copied from the original; the failed step and downstream re-execute with fresh hooks. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add `mantle rollback <workflow> [--to-version N]` command that creates a new version with the content of a previous version. Preserves full version history via rollback_of column. Default target is the second most recent version. Emits workflow.rolled_back audit event. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Wrap rollback read+insert in a transaction to prevent race conditions - Add name format, timeout, and retry validation to hook steps - Log audit emit errors instead of silently discarding them - Add partial indexes for retry lookups and hook step filtering Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Replace the monolithic UNIQUE(execution_id, step_name, attempt) constraint with two partial indexes — one for main steps (WHERE hook_block IS NULL) and one for hook steps (WHERE hook_block IS NOT NULL). This lets hook steps use their natural names instead of "on_failure:cleanup" prefixes, with the hook_block column providing disambiguation. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Also fixes the execution status check constraint to include 'queued' and 'timed_out' statuses needed by the concurrency control feature. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughAdds workflow concurrency (queued/pending), retry and rollback CLI commands, lifecycle hooks, secret key rotation, storage renames (tmp→storage), symlink-hardening for filesystem storage, audit/metrics, DB migration, and extensive engine, CLI, and test changes. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant CLI
participant Engine
participant DB
participant Auditor
Client->>CLI: mantle run / CLI run command
CLI->>Engine: ExecuteWithOptions(..., Force?)
Engine->>DB: BeginTx + CheckConcurrencyLimits (advisory locks)
alt limit exceeded & on_limit == "queue"
DB-->>Engine: indicate queued
Engine->>DB: insert execution (status="queued") + commit
Engine-->>CLI: respond Status=queued
else allowed
DB-->>Engine: allowed
Engine->>DB: insert execution (status="pending") + commit
Engine->>Engine: run execution (steps)
Engine->>DB: update execution terminal status
Engine->>DB: PromoteQueued(workflow) / PromoteQueuedByTeam(team)
alt queued execution promoted
DB-->>Engine: promoted execution id
Engine->>Auditor: EmitTx promotion event
Engine->>DB: update promoted -> pending
end
Engine-->>CLI: return result (completed/failed/timed_out)
end
sequenceDiagram
participant User
participant CLI
participant Engine
participant DB
participant Auditor
User->>CLI: mantle retry <execution-id> [--from-step] [--force]
CLI->>Engine: RetryExecution(execID, fromStep, force)
Engine->>DB: load original execution, inputs, workflow
Engine->>DB: BeginTx
Engine->>DB: CheckConcurrencyLimits -> queued or allowed
Engine->>DB: insert new execution (status pending|queued) and commit
Engine->>DB: BeginTx
Engine->>DB: update newExecution.retried_from_execution_id
Engine->>DB: copy upstream step outputs into new execution (upsert DO NOTHING)
Engine->>Auditor: EmitTx retry event
Engine->>DB: commit
alt new execution queued
Engine-->>CLI: return Status=queued
else pending
Engine->>Engine: resumeExecution(newExecID, inputs)
Engine-->>CLI: return execution result
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 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: 23
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
packages/engine/internal/artifact/storage.go (1)
31-78:⚠️ Potential issue | 🔴 CriticalSymlinked path components under
BasePathcan escape the storage directory.The path validation in
Put(lines 32-45),DeleteByPrefix(lines 84-95), andDelete(lines 101-111) uses only lexical checks withfilepath.Absandfilepath.Rel. These do not resolve symlinks. If any directory component underBasePathis a symlink,os.Createandos.RemoveAllwill follow it, allowing writes and deletes outside the intended storage root.For example, if
BasePathcontains/artifacts/link → /etc/sensitive, aPutwith key"link/file.txt"will write to/etc/sensitive/file.txtdespite passing the lexical boundary check.Add
os.EvalSymlinksto canonicalize the destination path and re-validate it is under the resolvedBasePathbefore any filesystem operations. Include a regression test for a symlinked child directory in all three methods.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/artifact/storage.go` around lines 31 - 78, The path validation is lexical and misses symlinked components; update FilesystemStorage.Put, DeleteByPrefix, and Delete to canonicalize paths with os.EvalSymlinks (or filepath.EvalSymlinks) before performing operations: first eval the configured fs.BasePath to a resolvedBase, then build destPath, eval that to resolvedDest, and re-check that resolvedDest has resolvedBase as a prefix (reject if not); perform MkdirAll/Open/Create/Remove only after this symlink-aware validation. Add regression tests exercising a symlinked child directory under BasePath for Put, DeleteByPrefix, and Delete to ensure writes/deletes cannot escape the storage root.packages/engine/internal/engine/toolsteps.go (1)
24-33:⚠️ Potential issue | 🔴 CriticalAdd the same
hook_block IS NULLpredicate to the fallback lookup.The INSERT conflict target correctly scopes to non-hook rows, but the fallback SELECT can return a hook step row if it shares the same
(execution_id, step_name, attempt). When idempotent retries occur, this causes cached LLM responses to attach to the wrong step ID.Minimal fix
err = ts.DB.QueryRowContext(ctx, ` SELECT id FROM step_executions - WHERE execution_id = $1 AND step_name = $2 AND attempt = 1 + WHERE execution_id = $1 AND step_name = $2 AND attempt = 1 + AND hook_block IS NULL `, executionID, stepName).Scan(&id)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/engine/toolsteps.go` around lines 24 - 33, The fallback SELECT in the idempotent insert path can return a hook row; update the QueryRowContext call that selects id from step_executions (the fallback after detecting sql.ErrNoRows) to include the same hook_block IS NULL predicate used in the INSERT conflict target so it only returns non-hook rows for the given executionID and stepName (where attempt = 1), i.e. modify the SELECT used in ts.DB.QueryRowContext to add "AND hook_block IS NULL" to match the INSERT semantics.packages/site/src/content/docs/configuration.md (1)
196-206:⚠️ Potential issue | 🟡 MinorAdd
version: 1to the remainingmantle.yamlexamples.The Production and Server Mode snippets still omit the top-level version field, so copying either example produces a config that contradicts the v0.4.0 requirement documented above.
Also applies to: 224-234
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/site/src/content/docs/configuration.md` around lines 196 - 206, The remaining mantle.yaml examples (the Production and Server Mode snippets shown as YAML blocks) are missing the top-level version field; update each YAML example by adding a top line "version: 1" above the existing keys so the examples comply with the v0.4.0 requirement and won't produce invalid configs when copied.packages/engine/internal/workflow/workflow.go (1)
46-60:⚠️ Potential issue | 🟠 Major
steps[].max_parallelis currently a no-op.This field now parses and validates, but
packages/engine/internal/engine/concurrency.go:21-31only enforces workflow/team limits, and the step execution path inpackages/engine/internal/engine/engine.go:226-252never readsStep.MaxParallel. Users can setmax_paralleland get no runtime enforcement. Either wire the per-step advisory-lock logic before exposing this field, or reject/document it as unsupported for now.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/workflow/workflow.go` around lines 46 - 60, Step.MaxParallel is parsed but never enforced at runtime; update the step execution path in the engine (where individual steps are launched) to use the concurrency package's advisory-lock APIs (the same functions currently enforcing workflow/team limits) so that each step requests a per-step slot limited by Step.MaxParallel (use Step.MaxParallel when >0, otherwise fall back to existing workflow/team limits), and ensure the lock is released when the step completes or fails. Reference Step.MaxParallel and the engine's step execution function and the concurrency package's acquire/release lock functions when making the change.
🤖 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/retry.go`:
- Around line 104-116: The failure message picks a failed step by iterating the
map result.Steps (non-deterministic); change it to scan the ordered slice
returned by orderedSteps(result) to find the first failed step in execution
order: call orderedSteps(result), iterate that slice to locate the first step
with sr.Status == "failed" and use its name in the fmt.Errorf("workflow failed
at step %q: %s", failedStep, result.Error) return; if none found keep the
generic "workflow failed: %s" error path. Ensure you reference result.Status and
result.Error as before.
In `@packages/engine/internal/cli/rotate_key.go`:
- Around line 84-87: Avoid printing raw key material and accepting it via argv:
stop using the --new-key flag to accept raw key material (recommend reading from
stdin or a file path) and remove any unconditional fmt.Fprintf that prints
newKey to stdout; specifically, change the code paths that reference the
flag/variable newKey and the fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n",
newKey) call so that if the user supplied the key you never echo it back, and if
you must output a generated key only write it to a secure destination
(stdin/stdout only when explicitly requested via a --output-file or --print-once
flag) or write it to a file; also update the warning message printing (the
fmt.Fprintln that warns the key is sensitive) to remain but do not include the
key itself—apply the same change for the other occurrence at the later print
(line ~93) so no raw key material appears in process args, shell history, or
logs.
- Around line 68-82: Move the audit emission to occur before committing the
transaction and make it a required step: call audit.PostgresEmitter.Emit (using
an emitter backed by the active tx/transaction rather than the outer database)
and check its error, returning the error so the commit is not performed; only
after successful Emit should you call tx.Commit(). Also avoid using the
cancelable cmd.Context() for the emit—use a stable context (e.g.
context.Background() or a derived non-cancelable context) so emitter failures or
cancellation cannot slip past and allow a commit without an audit record.
In `@packages/engine/internal/cli/run.go`:
- Line 146: Update the flag help text for the --force flag (where
cmd.Flags().Bool("force", ...) is declared in run.go) to explicitly warn that
forcing will bypass per-workflow and per-team concurrency limits and advisory
lock protection, will not queue executions, and may exceed configured limits;
replace the terse "Bypass concurrency limits" message with a more descriptive
warning covering these implications.
In `@packages/engine/internal/config/config.go`:
- Around line 290-297: The code currently calls v.Sub("tmp").Unmarshal(&legacy)
without ensuring v.Sub("tmp") is non-nil; update the deprecated-fallback block
to first assign sub := v.Sub("tmp"), check if sub != nil before calling
sub.Unmarshal(&legacy), and only set cfg.Storage and call slog.Warn when
Unmarshal succeeds; reference the StorageConfig struct, cfg.Storage,
v.Sub("tmp"), Unmarshal, and slog.Warn in your change.
In `@packages/engine/internal/db/migrations/016_safety_net.sql`:
- Around line 47-51: Before re-adding the old chk_execution_status constraint on
table workflow_executions, add a data migration step that updates any rows with
status = 'queued' to the legacy state (e.g., 'pending') and status = 'timed_out'
to the legacy state (e.g., 'failed') so the new CHECK will pass; perform these
UPDATE statements against the workflow_executions.status column prior to
dropping/adding the chk_execution_status constraint and ensure changes are
committed before recreating the constraint.
- Around line 37-44: The down migration must avoid re-creating a global
uniqueness that conflicts with hook rows; instead of re-adding the monolithic
constraint step_executions_execution_id_step_name_attempt_key on (execution_id,
step_name, attempt), create a partial unique index that only enforces uniqueness
for main steps (i.e., WHERE hook_block IS NULL) or include hook_block in the
key; update the step_executions rollback block to DROP the old indexes and then
CREATE UNIQUE INDEX (or ALTER by CREATE UNIQUE INDEX CONCURRENTLY) e.g. create a
unique index on (execution_id, step_name, attempt) WHERE hook_block IS NULL so
hook rows can reuse the same name without causing the down migration to fail.
- Around line 3-8: The migration added a new persisted status "timed_out" but
the engine still records timeouts as "failed"; update the engine code so
timeouts are actually persisted and treated as terminal: modify the timeout
handling in function(s) that set execution status (e.g., updateExecutionStatus
and the timeout/hook routing logic that currently sets mainStatus to
"timed_out") to persist "timed_out" instead of "failed", and add "timed_out" to
any terminal-status checks (the function that currently checks terminal statuses
around lines where isTerminal/updateExecutionStatus is implemented) so the
runtime treats timed_out as terminal; ensure any place mapping statuses or
computing terminalness includes "timed_out" consistently.
In `@packages/engine/internal/engine/concurrency.go`:
- Around line 94-118: PromoteQueued currently updates executions from queued to
pending without emitting an audit event; change PromoteQueued to accept an audit
emitter (e.g., audit.AuditEmitter or audit.Emitter) and modify the UPDATE query
in PromoteQueued to use RETURNING id so you can capture the promoted execution
ID; after confirming RowsAffected/returned ID > 0 emit an audit.Event (Timestamp
time.Now(), Actor "engine", Action audit.ActionExecutionPromoted, Resource type
"workflow_execution" with the returned ID, and Metadata with the workflow name),
then update metrics.ExecutionsQueued.WithLabelValues(workflowName).Dec(); ensure
errors from the DB call and from the emitter are handled/returned and update the
function signature and all call sites accordingly.
- Around line 120-138: PromoteQueuedByTeam currently updates a queued execution
to pending but does not decrement metrics.ExecutionsQueued (causing metric
drift); update PromoteQueuedByTeam to fetch the promoted row's id and workflow
name (or at minimum the workflow name) in the same transaction/SELECT used to
choose the row, then after the UPDATE call
metrics.ExecutionsQueued.WithLabelValues(workflowName).Dec() (or use/emit a
separate team-scoped metric if you decide not to obtain workflowName); ensure
this mirrors the behavior in PromoteQueued by referencing the same metrics logic
and performs the metric decrement only when an execution was actually promoted.
- Around line 21-92: CheckConcurrencyLimits currently uses pg_advisory_xact_lock
but the code in engine.go commits the transaction before calling
createExecution, creating a TOCTOU window; fix by moving the insertion into the
same transaction that called CheckConcurrencyLimits so the advisory lock covers
both the count check and the insert. Concretely: change createExecution (or add
createExecutionTx) to accept and use the active *sql.Tx, call that from the same
transaction that invoked CheckConcurrencyLimits (do the insert before
committing), and ensure engine.go uses the transactional variant and only
commits after the insert so the advisory lock serialises check+insert
atomically.
In `@packages/engine/internal/engine/engine.go`:
- Around line 329-340: PromoteQueued and PromoteQueuedByTeam errors are
currently only logged; increment a failure metric when these return an error so
promotion failures can be alerted on. In the PromoteQueued/PromoteQueuedByTeam
error branches (after log.Printf) call your telemetry/metrics API to increment a
counter (e.g., "promotion_failures") and attach labels for workflow
(workflowName) and team (sc.TeamID) and optionally the error class, using the
same promoteCtx/e.DB context if required; ensure the metric call does not change
the best-effort behavior (ignore metric errors).
- Around line 97-119: There is a TOCTOU race between CheckConcurrencyLimits and
the later createExecution call: the advisory lock/transaction (tx from
e.DB.BeginTx) is committed before createExecution executes, allowing concurrent
requests to pass the check; fix by performing the execution creation while still
holding the same transaction/lock instead of committing first — i.e., move or
add the createExecution call (or whatever inserts the execution row) into the
same tx scope that runs CheckConcurrencyLimits (use tx for the insert or use
SELECT ... FOR UPDATE/pg_advisory_lock semantics within the same tx) and only
commit after the execution record has been created; refer to
CheckConcurrencyLimits, tx, createExecution, opts.Force,
wf.MaxParallelExecutions, and e.MaxConcurrentExecutionsPerTeam when applying
this change.
In `@packages/engine/internal/engine/hooks_test.go`:
- Around line 42-47: The test's on_success assertion is vacuous because both
main-step and notify-success hit the same httptest server (server) so called
becomes true on the first request; update the test to either run a dedicated
hook server or assert the server received two requests. Specifically, modify the
test around the httptest.NewServer and called variable used by
executeHooks/on_success (and the main-step/notify-success requests) so that you
track requestCount (or create a separate hookServer for notify-success) and
assert requestCount == 2 after executeHooks returns (or assert the hookServer
was called once), ensuring executeHooks actually dispatched the on_success hook.
- Around line 42-47: Tests use unsynchronized shared variables (called,
hookCalled, finishCallCount) that are written by httptest.Server handler
goroutines; replace these with concurrency-safe primitives: use sync/atomic's
atomic.Bool for boolean flags (atomic.Bool.Store(true) in the handler and
atomic.Bool.Load() in assertions) and atomic.Int32/Int64 for counters
(atomic.AddInt32 / atomic.LoadInt32) or alternatively use channels to signal the
test from the handler. Update imports to include sync/atomic, change all
reads/writes of called, hookCalled, finishCallCount to the corresponding atomic
methods (or channel send/receive) and assert on the atomic-loaded values in the
test body.
In `@packages/engine/internal/engine/hooks.go`:
- Around line 44-54: The current logic logs but ignores an invalid
wf.Hooks.Timeout, leaving hooks unbounded; update the hook timeout handling so
that when time.ParseDuration(wf.Hooks.Timeout) returns an error you set a
sensible default duration (e.g., defaultHookTimeout), log that you're falling
back to it, then create the timeout context via context.WithTimeout(ctx, dur)
and defer cancel only when the timeout is applied; reference wf.Hooks.Timeout,
time.ParseDuration, context.WithTimeout and cancel to locate and update the
code, and add/define a named default (defaultHookTimeout) if one does not exist.
In `@packages/engine/internal/engine/retry.go`:
- Around line 131-174: The current e.DB.QueryContext call in retry.go can return
multiple attempts per step and may copy an earlier attempt; modify the SQL so
you only select the latest attempt per step (e.g., replace the query with a
DISTINCT ON query: SELECT DISTINCT ON (step_name) step_name, status, output,
error FROM step_executions WHERE execution_id = $1 AND hook_block IS NULL ORDER
BY step_name, attempt DESC (or ORDER BY step_name, started_at DESC) ), keeping
the same parameters and the rest of the loop logic (references:
e.DB.QueryContext call, step_executions table, and the rows loop that scans
stepName/status/output/error). Ensure the WHERE clause (hook_block IS NULL)
remains and that ordering chooses the newest attempt so the inserted row
reflects the latest completed output.
- Around line 86-108: The concurrency check acquires advisory locks inside a
transaction via CheckConcurrencyLimits but currently commits before inserting
the execution, creating a TOCTOU race; modify createExecution to accept an
optional *sql.Tx (e.g., createExecution(ctx, tx *sql.Tx, ...) or similar) and,
when a tx is provided, perform the execution INSERT using that tx so the insert
occurs before the transaction commit (and thus before locks are released).
Update the caller code in the retry flow and ExecuteWithOptions to pass the
active tx returned by e.DB.BeginTx into createExecution and only commit after
createExecution returns, ensuring the advisory-locked check and the insert are
atomic.
In `@packages/engine/internal/metrics/metrics.go`:
- Around line 48-59: Remove the redundant HookStepsFailedTotal metric: delete
the HookStepsFailedTotal promauto.NewCounterVec declaration and remove any
references/usages of HookStepsFailedTotal elsewhere in the codebase so consumers
query HookStepsTotal with label status="failed" instead; ensure imports/builds
still compile and remove unused prometheus/promauto references if they become
unused after deletion.
In `@packages/engine/internal/secret/store_test.go`:
- Around line 273-290: The test begins a DB transaction with store.DB.BeginTx
and then calls RotateAll(ctx, tx, oldEncryptor, newEnc); to avoid causing a
secondary error (Commit on a closed tx) add defer tx.Rollback() immediately
after successful BeginTx to ensure cleanup, replace the non-fatal t.Errorf count
assertion with t.Fatalf so the test stops on failure, and only call tx.Commit()
after the fatal-guarded checks so Commit() runs only when the assertions passed;
update both locations that follow this pattern (the RotateAll() test blocks
referenced) accordingly.
In `@packages/engine/internal/secret/store.go`:
- Around line 196-245: RotateAll currently uses SELECT ... FOR UPDATE which
won't block concurrent INSERTs, so acquire a Postgres advisory lock at the start
of RotateAll to serialize rotation with credential writers: call
tx.ExecContext(ctx, "SELECT pg_advisory_xact_lock($1)", <fixed_int64_key>)
before querying rows in RotateAll and release automatically at txn end; also
ensure all credential-writing code paths (any functions that INSERT/UPDATE the
credentials table) take the same advisory lock (pg_advisory_xact_lock with the
same key) during writes so inserts are serialized with RotateAll and no new rows
are missed.
In `@packages/engine/internal/workflow/validate.go`:
- Around line 570-675: validateHookSteps currently checks names, actions,
timeouts, retry and depends_on but omits CEL expression validation for
Step.Params and Step.If, so hook steps with invalid CEL only fail at runtime;
update validateHooks (and/or validateHookSteps) to call the same CEL validation
logic used for main workflow steps (the CEL validator invoked for params/if in
main step validation) for each hook Step (type Step) after basic validation,
validating Step.Params and Step.If and appending ValidationError entries with
the appropriate field prefix (e.g., prefix + ".params" and prefix + ".if") on
parse/compile failures so hook CEL expressions are caught at config validation
time.
In `@packages/site/src/content/docs/cli-reference/workflow-commands.md`:
- Around line 336-385: Add language specifiers "text" to the fenced code blocks
in the "mantle rollback" and "mantle retry" docs: update the usage block and the
three error-example blocks under "mantle rollback" (`Usage:` block, `Error:
workflow "fetch-and-summarize"...`, and `Error: version 5 not found...`) to use
```text fences, and apply the same change to the three code blocks in the
`mantle retry` section to ensure consistent Markdown language tagging.
---
Outside diff comments:
In `@packages/engine/internal/artifact/storage.go`:
- Around line 31-78: The path validation is lexical and misses symlinked
components; update FilesystemStorage.Put, DeleteByPrefix, and Delete to
canonicalize paths with os.EvalSymlinks (or filepath.EvalSymlinks) before
performing operations: first eval the configured fs.BasePath to a resolvedBase,
then build destPath, eval that to resolvedDest, and re-check that resolvedDest
has resolvedBase as a prefix (reject if not); perform
MkdirAll/Open/Create/Remove only after this symlink-aware validation. Add
regression tests exercising a symlinked child directory under BasePath for Put,
DeleteByPrefix, and Delete to ensure writes/deletes cannot escape the storage
root.
In `@packages/engine/internal/engine/toolsteps.go`:
- Around line 24-33: The fallback SELECT in the idempotent insert path can
return a hook row; update the QueryRowContext call that selects id from
step_executions (the fallback after detecting sql.ErrNoRows) to include the same
hook_block IS NULL predicate used in the INSERT conflict target so it only
returns non-hook rows for the given executionID and stepName (where attempt =
1), i.e. modify the SELECT used in ts.DB.QueryRowContext to add "AND hook_block
IS NULL" to match the INSERT semantics.
In `@packages/engine/internal/workflow/workflow.go`:
- Around line 46-60: Step.MaxParallel is parsed but never enforced at runtime;
update the step execution path in the engine (where individual steps are
launched) to use the concurrency package's advisory-lock APIs (the same
functions currently enforcing workflow/team limits) so that each step requests a
per-step slot limited by Step.MaxParallel (use Step.MaxParallel when >0,
otherwise fall back to existing workflow/team limits), and ensure the lock is
released when the step completes or fails. Reference Step.MaxParallel and the
engine's step execution function and the concurrency package's acquire/release
lock functions when making the change.
In `@packages/site/src/content/docs/configuration.md`:
- Around line 196-206: The remaining mantle.yaml examples (the Production and
Server Mode snippets shown as YAML blocks) are missing the top-level version
field; update each YAML example by adding a top line "version: 1" above the
existing keys so the examples comply with the v0.4.0 requirement and won't
produce invalid configs when copied.
🪄 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: ASSERTIVE
Plan: Pro
Run ID: 4da75f4d-434f-4787-aef7-2cfa1ee31e79
📒 Files selected for processing (38)
CLAUDE.mdpackages/engine/docker-compose.ymlpackages/engine/internal/artifact/reaper.gopackages/engine/internal/artifact/reaper_test.gopackages/engine/internal/artifact/storage.gopackages/engine/internal/artifact/storage_test.gopackages/engine/internal/audit/audit.gopackages/engine/internal/cel/cel.gopackages/engine/internal/cel/cel_test.gopackages/engine/internal/cli/cancel.gopackages/engine/internal/cli/retry.gopackages/engine/internal/cli/rollback.gopackages/engine/internal/cli/root.gopackages/engine/internal/cli/rotate_key.gopackages/engine/internal/cli/run.gopackages/engine/internal/cli/serve.gopackages/engine/internal/config/config.gopackages/engine/internal/config/config_test.gopackages/engine/internal/db/migrations/016_safety_net.sqlpackages/engine/internal/engine/concurrency.gopackages/engine/internal/engine/concurrency_test.gopackages/engine/internal/engine/engine.gopackages/engine/internal/engine/hooks.gopackages/engine/internal/engine/hooks_test.gopackages/engine/internal/engine/orchestrator.gopackages/engine/internal/engine/retry.gopackages/engine/internal/engine/retry_test.gopackages/engine/internal/engine/toolsteps.gopackages/engine/internal/metrics/metrics.gopackages/engine/internal/secret/store.gopackages/engine/internal/secret/store_test.gopackages/engine/internal/server/server.gopackages/engine/internal/workflow/validate.gopackages/engine/internal/workflow/validate_hooks_test.gopackages/engine/internal/workflow/workflow.gopackages/site/src/content/docs/cli-reference/workflow-commands.mdpackages/site/src/content/docs/configuration.mdpackages/site/src/content/docs/workflow-reference/index.md
Comment on lines
84
to
87
| fmt.Fprintf(cmd.OutOrStdout(), "Re-encrypted %d credential(s).\n", count) | ||
| fmt.Fprintln(cmd.ErrOrStderr(), "WARNING: The following key is sensitive. Store it securely and clear your terminal.") | ||
| fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n", newKey) | ||
| fmt.Fprintln(cmd.OutOrStdout(), "Update MANTLE_ENCRYPTION_KEY to the new key and restart.") |
Contributor
There was a problem hiding this comment.
Avoid exposing the new master key via argv and stdout.
--new-key puts raw key material in shell history, process listings, and CI logs, and Line 86 then echoes it back to stdout even when the caller supplied the key. Prefer stdin or a file-based input, and do not re-print user-provided keys.
Also applies to: 93-93
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/cli/rotate_key.go` around lines 84 - 87, Avoid
printing raw key material and accepting it via argv: stop using the --new-key
flag to accept raw key material (recommend reading from stdin or a file path)
and remove any unconditional fmt.Fprintf that prints newKey to stdout;
specifically, change the code paths that reference the flag/variable newKey and
the fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n", newKey) call so that if the
user supplied the key you never echo it back, and if you must output a generated
key only write it to a secure destination (stdin/stdout only when explicitly
requested via a --output-file or --print-once flag) or write it to a file; also
update the warning message printing (the fmt.Fprintln that warns the key is
sensitive) to remain but do not include the key itself—apply the same change for
the other occurrence at the later print (line ~93) so no raw key material
appears in process args, shell history, or logs.
Comment on lines
+570
to
+675
| // validateHooks validates the hooks configuration block. | ||
| func validateHooks(hooks *HooksConfig) []ValidationError { | ||
| var errs []ValidationError | ||
|
|
||
| // Validate hooks timeout. | ||
| if hooks.Timeout != "" { | ||
| d, err := time.ParseDuration(hooks.Timeout) | ||
| if err != nil { | ||
| errs = append(errs, ValidationError{ | ||
| Field: "hooks.timeout", | ||
| Message: fmt.Sprintf("invalid duration: %v", err), | ||
| }) | ||
| } else if d <= 0 { | ||
| errs = append(errs, ValidationError{ | ||
| Field: "hooks.timeout", | ||
| Message: "timeout must be a positive duration", | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| // Validate each hook block. Each block has its own name namespace. | ||
| errs = append(errs, validateHookSteps(hooks.OnSuccess, "hooks.on_success")...) | ||
| errs = append(errs, validateHookSteps(hooks.OnFailure, "hooks.on_failure")...) | ||
| errs = append(errs, validateHookSteps(hooks.OnFinish, "hooks.on_finish")...) | ||
|
|
||
| return errs | ||
| } | ||
|
|
||
| // validateHookSteps validates a slice of hook steps within a single block. | ||
| func validateHookSteps(steps []Step, blockPrefix string) []ValidationError { | ||
| var errs []ValidationError | ||
| seen := make(map[string]bool) | ||
|
|
||
| for i, step := range steps { | ||
| prefix := fmt.Sprintf("%s[%d]", blockPrefix, i) | ||
|
|
||
| if step.Name == "" { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".name", | ||
| Message: "hook step name is required", | ||
| }) | ||
| } else { | ||
| if !namePattern.MatchString(step.Name) { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".name", | ||
| Message: "hook step name must match ^[a-z][a-z0-9-]*$", | ||
| }) | ||
| } | ||
| if seen[step.Name] { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".name", | ||
| Message: fmt.Sprintf("duplicate hook step name %q", step.Name), | ||
| }) | ||
| } | ||
| seen[step.Name] = true | ||
| } | ||
|
|
||
| if step.Action == "" { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".action", | ||
| Message: "hook step action is required", | ||
| }) | ||
| } | ||
|
|
||
| // Validate timeout. | ||
| if step.Timeout != "" { | ||
| d, err := time.ParseDuration(step.Timeout) | ||
| if err != nil { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".timeout", | ||
| Message: fmt.Sprintf("invalid duration: %v", err), | ||
| }) | ||
| } else if d <= 0 { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".timeout", | ||
| Message: "timeout must be a positive duration", | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| // Validate retry policy. | ||
| if step.Retry != nil { | ||
| if step.Retry.MaxAttempts <= 0 { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".retry.max_attempts", | ||
| Message: "max_attempts must be greater than 0", | ||
| }) | ||
| } | ||
| if step.Retry.Backoff != "" && !validBackoffTypes[step.Retry.Backoff] { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".retry.backoff", | ||
| Message: fmt.Sprintf("backoff must be one of: fixed, exponential (got %q)", step.Retry.Backoff), | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| if len(step.DependsOn) > 0 { | ||
| errs = append(errs, ValidationError{ | ||
| Field: prefix + ".depends_on", | ||
| Message: "hook steps do not support depends_on — use a child workflow for complex error handling", | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| return errs | ||
| } |
Contributor
There was a problem hiding this comment.
Missing CEL expression validation for hook steps.
Hook steps support params and if fields (they use the same Step struct), but validateHookSteps doesn't validate CEL expressions within these fields. Main workflow steps get CEL validation at lines 413-430, but hook steps bypass this check entirely. Invalid CEL expressions in hooks would only fail at runtime.
🐛 Proposed fix to add CEL validation for hooks
Add CEL validation to the validateHooks function after validating hook steps:
func validateHooks(hooks *HooksConfig) []ValidationError {
var errs []ValidationError
// ... existing timeout and step validation ...
errs = append(errs, validateHookSteps(hooks.OnSuccess, "hooks.on_success")...)
errs = append(errs, validateHookSteps(hooks.OnFailure, "hooks.on_failure")...)
errs = append(errs, validateHookSteps(hooks.OnFinish, "hooks.on_finish")...)
+ // Validate CEL expressions in hook steps.
+ celEval, celErr := mantleCEL.NewEvaluator()
+ if celErr == nil {
+ errs = append(errs, validateHookStepsCEL(celEval, hooks.OnSuccess, "hooks.on_success")...)
+ errs = append(errs, validateHookStepsCEL(celEval, hooks.OnFailure, "hooks.on_failure")...)
+ errs = append(errs, validateHookStepsCEL(celEval, hooks.OnFinish, "hooks.on_finish")...)
+ }
return errs
}
+func validateHookStepsCEL(eval *mantleCEL.Evaluator, steps []Step, blockPrefix string) []ValidationError {
+ var errs []ValidationError
+ for i, step := range steps {
+ prefix := fmt.Sprintf("%s[%d]", blockPrefix, i)
+ if step.If != "" {
+ if err := eval.CompileCheck(step.If); err != nil {
+ errs = append(errs, ValidationError{
+ Field: prefix + ".if",
+ Message: fmt.Sprintf("invalid CEL expression: %v", err),
+ })
+ }
+ }
+ errs = append(errs, validateCELInParams(eval, step.Params, prefix+".params")...)
+ }
+ return errs
+}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // validateHooks validates the hooks configuration block. | |
| func validateHooks(hooks *HooksConfig) []ValidationError { | |
| var errs []ValidationError | |
| // Validate hooks timeout. | |
| if hooks.Timeout != "" { | |
| d, err := time.ParseDuration(hooks.Timeout) | |
| if err != nil { | |
| errs = append(errs, ValidationError{ | |
| Field: "hooks.timeout", | |
| Message: fmt.Sprintf("invalid duration: %v", err), | |
| }) | |
| } else if d <= 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: "hooks.timeout", | |
| Message: "timeout must be a positive duration", | |
| }) | |
| } | |
| } | |
| // Validate each hook block. Each block has its own name namespace. | |
| errs = append(errs, validateHookSteps(hooks.OnSuccess, "hooks.on_success")...) | |
| errs = append(errs, validateHookSteps(hooks.OnFailure, "hooks.on_failure")...) | |
| errs = append(errs, validateHookSteps(hooks.OnFinish, "hooks.on_finish")...) | |
| return errs | |
| } | |
| // validateHookSteps validates a slice of hook steps within a single block. | |
| func validateHookSteps(steps []Step, blockPrefix string) []ValidationError { | |
| var errs []ValidationError | |
| seen := make(map[string]bool) | |
| for i, step := range steps { | |
| prefix := fmt.Sprintf("%s[%d]", blockPrefix, i) | |
| if step.Name == "" { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".name", | |
| Message: "hook step name is required", | |
| }) | |
| } else { | |
| if !namePattern.MatchString(step.Name) { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".name", | |
| Message: "hook step name must match ^[a-z][a-z0-9-]*$", | |
| }) | |
| } | |
| if seen[step.Name] { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".name", | |
| Message: fmt.Sprintf("duplicate hook step name %q", step.Name), | |
| }) | |
| } | |
| seen[step.Name] = true | |
| } | |
| if step.Action == "" { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".action", | |
| Message: "hook step action is required", | |
| }) | |
| } | |
| // Validate timeout. | |
| if step.Timeout != "" { | |
| d, err := time.ParseDuration(step.Timeout) | |
| if err != nil { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".timeout", | |
| Message: fmt.Sprintf("invalid duration: %v", err), | |
| }) | |
| } else if d <= 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".timeout", | |
| Message: "timeout must be a positive duration", | |
| }) | |
| } | |
| } | |
| // Validate retry policy. | |
| if step.Retry != nil { | |
| if step.Retry.MaxAttempts <= 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".retry.max_attempts", | |
| Message: "max_attempts must be greater than 0", | |
| }) | |
| } | |
| if step.Retry.Backoff != "" && !validBackoffTypes[step.Retry.Backoff] { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".retry.backoff", | |
| Message: fmt.Sprintf("backoff must be one of: fixed, exponential (got %q)", step.Retry.Backoff), | |
| }) | |
| } | |
| } | |
| if len(step.DependsOn) > 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".depends_on", | |
| Message: "hook steps do not support depends_on — use a child workflow for complex error handling", | |
| }) | |
| } | |
| } | |
| return errs | |
| } | |
| // validateHooks validates the hooks configuration block. | |
| func validateHooks(hooks *HooksConfig) []ValidationError { | |
| var errs []ValidationError | |
| // Validate hooks timeout. | |
| if hooks.Timeout != "" { | |
| d, err := time.ParseDuration(hooks.Timeout) | |
| if err != nil { | |
| errs = append(errs, ValidationError{ | |
| Field: "hooks.timeout", | |
| Message: fmt.Sprintf("invalid duration: %v", err), | |
| }) | |
| } else if d <= 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: "hooks.timeout", | |
| Message: "timeout must be a positive duration", | |
| }) | |
| } | |
| } | |
| // Validate each hook block. Each block has its own name namespace. | |
| errs = append(errs, validateHookSteps(hooks.OnSuccess, "hooks.on_success")...) | |
| errs = append(errs, validateHookSteps(hooks.OnFailure, "hooks.on_failure")...) | |
| errs = append(errs, validateHookSteps(hooks.OnFinish, "hooks.on_finish")...) | |
| // Validate CEL expressions in hook steps. | |
| celEval, celErr := mantleCEL.NewEvaluator() | |
| if celErr == nil { | |
| errs = append(errs, validateHookStepsCEL(celEval, hooks.OnSuccess, "hooks.on_success")...) | |
| errs = append(errs, validateHookStepsCEL(celEval, hooks.OnFailure, "hooks.on_failure")...) | |
| errs = append(errs, validateHookStepsCEL(celEval, hooks.OnFinish, "hooks.on_finish")...) | |
| } | |
| return errs | |
| } | |
| // validateHookSteps validates a slice of hook steps within a single block. | |
| func validateHookSteps(steps []Step, blockPrefix string) []ValidationError { | |
| var errs []ValidationError | |
| seen := make(map[string]bool) | |
| for i, step := range steps { | |
| prefix := fmt.Sprintf("%s[%d]", blockPrefix, i) | |
| if step.Name == "" { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".name", | |
| Message: "hook step name is required", | |
| }) | |
| } else { | |
| if !namePattern.MatchString(step.Name) { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".name", | |
| Message: "hook step name must match ^[a-z][a-z0-9-]*$", | |
| }) | |
| } | |
| if seen[step.Name] { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".name", | |
| Message: fmt.Sprintf("duplicate hook step name %q", step.Name), | |
| }) | |
| } | |
| seen[step.Name] = true | |
| } | |
| if step.Action == "" { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".action", | |
| Message: "hook step action is required", | |
| }) | |
| } | |
| // Validate timeout. | |
| if step.Timeout != "" { | |
| d, err := time.ParseDuration(step.Timeout) | |
| if err != nil { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".timeout", | |
| Message: fmt.Sprintf("invalid duration: %v", err), | |
| }) | |
| } else if d <= 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".timeout", | |
| Message: "timeout must be a positive duration", | |
| }) | |
| } | |
| } | |
| // Validate retry policy. | |
| if step.Retry != nil { | |
| if step.Retry.MaxAttempts <= 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".retry.max_attempts", | |
| Message: "max_attempts must be greater than 0", | |
| }) | |
| } | |
| if step.Retry.Backoff != "" && !validBackoffTypes[step.Retry.Backoff] { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".retry.backoff", | |
| Message: fmt.Sprintf("backoff must be one of: fixed, exponential (got %q)", step.Retry.Backoff), | |
| }) | |
| } | |
| } | |
| if len(step.DependsOn) > 0 { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".depends_on", | |
| Message: "hook steps do not support depends_on — use a child workflow for complex error handling", | |
| }) | |
| } | |
| } | |
| return errs | |
| } | |
| func validateHookStepsCEL(eval *mantleCEL.Evaluator, steps []Step, blockPrefix string) []ValidationError { | |
| var errs []ValidationError | |
| for i, step := range steps { | |
| prefix := fmt.Sprintf("%s[%d]", blockPrefix, i) | |
| if step.If != "" { | |
| if err := eval.CompileCheck(step.If); err != nil { | |
| errs = append(errs, ValidationError{ | |
| Field: prefix + ".if", | |
| Message: fmt.Sprintf("invalid CEL expression: %v", err), | |
| }) | |
| } | |
| } | |
| errs = append(errs, validateCELInParams(eval, step.Params, prefix+".params")...) | |
| } | |
| return errs | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/workflow/validate.go` around lines 570 - 675,
validateHookSteps currently checks names, actions, timeouts, retry and
depends_on but omits CEL expression validation for Step.Params and Step.If, so
hook steps with invalid CEL only fail at runtime; update validateHooks (and/or
validateHookSteps) to call the same CEL validation logic used for main workflow
steps (the CEL validator invoked for params/if in main step validation) for each
hook Step (type Step) after basic validation, validating Step.Params and Step.If
and appending ValidationError entries with the appropriate field prefix (e.g.,
prefix + ".params" and prefix + ".if") on parse/compile failures so hook CEL
expressions are caught at config validation time.
…y lock (PR review) - Replace --new-key flag with --new-key-file to avoid exposing secrets in process arguments; add --output-key for secure file-based key output - Never echo back user-provided keys; only print auto-generated keys - Move audit emission inside the transaction via new audit.EmitTx() so the audit record commits atomically with the rotation - Use context.Background() for audit emit to prevent cancellation loss - Add pg_advisory_xact_lock in RotateAll to serialize with credential writers - Add defer tx.Rollback() in tests; upgrade count assertions to t.Fatalf Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Close TOCTOU race where advisory lock was released between concurrency check and execution insert by performing both in the same transaction. Add createExecutionTx for transactional inserts. Update PromoteQueued and PromoteQueuedByTeam to emit audit events and return promoted IDs. Add PromotionFailuresTotal metric and ActionExecutionPromoted constant. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…nts (PR review)
- Default 5m timeout on hook parse failure instead of proceeding without one
- Atomic variables in hooks_test.go; fix vacuous on_success assertion with request counter
- Remove redundant HookStepsFailedTotal metric (use HookStepsTotal status=failed)
- Down migration: partial unique index, data migration for queued/timed_out statuses
- Persist timed_out as terminal status in updateExecutionStatus
- Deterministic failed-step lookup via orderedSteps; DISTINCT ON in retry step copy
- Nil check on v.Sub("tmp") in config fallback
- CEL expression validation for hook step params and if conditions
- Add hook_block IS NULL to toolsteps fallback SELECT
- Improved --force flag help text
- Code block language specifiers in workflow-commands.md
- Add version: 1 to Production and Server Mode YAML examples
- Symlink-aware path validation in FilesystemStorage (EvalSymlinks)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Deploying mantle with
|
| Latest commit: |
63c8853
|
| Status: | ✅ Deploy successful! |
| Preview URL: | https://5cbf6baa.mantle-cxy.pages.dev |
| Branch Preview URL: | https://feature-v040-safety-net.mantle-cxy.pages.dev |
Contributor
There was a problem hiding this comment.
Actionable comments posted: 15
♻️ Duplicate comments (3)
packages/engine/internal/db/migrations/016_safety_net.sql (1)
42-46:⚠️ Potential issue | 🔴 CriticalThe down migration cannot create this index after dropping
hook_block.Line 42 removes the column, so Line 45 fails immediately because the partial-index predicate still references
hook_block. The legacy schema also cannot represent hook rows at all, so rollback needs to reconcile/deletehook_block IS NOT NULLrows before dropping the column, then recreate the old global uniqueness on the remaining data.Suggested fix
-ALTER TABLE step_executions DROP COLUMN IF EXISTS hook_block; --- Restore the original unique constraint as a partial index (WHERE hook_block IS NULL) --- to avoid failures if any hook rows remain. -CREATE UNIQUE INDEX step_executions_execution_id_step_name_attempt_key - ON step_executions (execution_id, step_name, attempt) WHERE hook_block IS NULL; +DELETE FROM step_executions WHERE hook_block IS NOT NULL; +ALTER TABLE step_executions DROP COLUMN IF EXISTS hook_block; +ALTER TABLE step_executions + ADD CONSTRAINT step_executions_execution_id_step_name_attempt_key + UNIQUE (execution_id, step_name, attempt);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/db/migrations/016_safety_net.sql` around lines 42 - 46, The down migration fails because it drops the hook_block column and then tries to create a partial index that references hook_block; fix by first reconciling/deleting hook rows, then drop the column, then recreate the original global unique index. Concretely: in the rollback for step_executions, remove any rows where hook_block IS NOT NULL (or otherwise migrate them away) so the legacy schema can represent remaining data, then DROP COLUMN hook_block, and finally CREATE UNIQUE INDEX step_executions_execution_id_step_name_attempt_key ON step_executions (execution_id, step_name, attempt) without a WHERE predicate.packages/engine/internal/secret/store.go (1)
197-200:⚠️ Potential issue | 🔴 CriticalThis lock does not yet protect all credential writes.
RotateAllnow takes the advisory key, but(*Store).Createin this file still inserts directly throughs.DBwithout taking the same lock. A credential created during rotation can still be skipped and left encrypted with the old master key. Please route credential writers through the same advisory-lock/transaction path before relying on this as serialized rotation.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/secret/store.go` around lines 197 - 200, Create currently writes credentials directly via s.DB while RotateAll uses a transaction that acquires the advisory lock (SELECT pg_advisory_xact_lock($1) with key 0x4D414E544C45), so credential creation can race with rotation; update (*Store).Create to route writes through the same serialized path by either accepting/using the same tx that acquires the advisory lock or by performing the same pg_advisory_xact_lock and using a transaction around the insert, ensuring Create uses the same lock key and transaction semantics as RotateAll (or factor a helper used by both) so writers and rotation are mutually serialized.packages/engine/internal/config/config.go (1)
290-297:⚠️ Potential issue | 🟠 MajorThe legacy
tmpfallback currently breaks precedence.Assigning
cfg.Storage = legacyoverwrites anystorage.*values that already came from env vars or flags wheneverstorage.typeis still empty. The fallback should only fill missing fields, or only run when nostorage.*key was set at all.As per coding guidelines, "Configuration should use mantle.yaml with CLI flag and env var overrides."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/config/config.go` around lines 290 - 297, The fallback from the deprecated "tmp" section currently replaces the entire cfg.Storage (StorageConfig) and thus clobbers any values set via env/flags; change the logic in config.go so that when v.Sub("tmp") unmarshals into a legacy StorageConfig you merge missing fields into cfg.Storage rather than assigning it wholesale, or only apply the legacy values when no "storage.*" keys were set; implement a field-by-field copy (or a utility merge) that preserves non-zero values already present in cfg.Storage (e.g., keep existing cfg.Storage.Type/other fields if set) and only set fields from legacy when cfg.Storage has their zero value. Ensure this happens in the same block where v.Sub("tmp") and sub.Unmarshal(&legacy) are used and add a unit test or validation to cover env/flag overrides vs. tmp fallback.
🤖 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/artifact/storage.go`:
- Around line 90-104: The fallback that handles non-existent target paths
compares resolvedBase to filepath.Abs(target) which can retain symlinks in the
base and cause false "prefix escapes base path" errors; in the code around
resolvedTarget/resolvedBase (where filepath.EvalSymlinks, filepath.Abs and
filepath.Rel are used) change the fallback to compute the candidate path
relative to resolvedBase by taking the cleaned relative suffix of target (e.g.,
relSuffix := filepath.Clean(strings.TrimPrefix(target, baseUnresolved)) or by
finding the deepest existing parent of target and EvalSymlinks on that parent)
and then join that suffix to resolvedBase (e.g., filepath.Join(resolvedBase,
relSuffix)) before calling filepath.Rel, and apply the same fix in the other
identical section (lines ~120-133) so comparisons are always against a
fully-evaluated canonical base.
- Around line 31-48: In FilesystemStorage.Put validate the provided key before
calling os.MkdirAll: ensure key is not absolute (use filepath.IsAbs) and perform
a lexical clean (filepath.Clean) then reject any cleaned key that is "." or
starts with ".." or has a leading ".." path segment (e.g.,
strings.HasPrefix(clean, ".."+string(os.PathSeparator)) or clean == ".."); only
after this relative-only check proceed to build destPath and run MkdirAll, then
keep the existing EvalSymlinks/resolvedBase/resolvedDest and filepath.Rel checks
as the second runtime/symlink guard (refer to FilesystemStorage.Put, destPath,
MkdirAll, resolvedBase, resolvedDest, and rel).
In `@packages/engine/internal/audit/postgres.go`:
- Around line 33-37: PostgresEmitter.EmitTx currently writes directly with
emitEvent and therefore skips the same auth enrichment performed in
PostgresEmitter.Emit; update EmitTx to call p.enrichFromContext(ctx, event) (or
otherwise apply the same tx-aware enrichment helper) before calling emitEvent so
that auth_method and other extractor-derived metadata are included for
transactional writes; ensure you reference PostgresEmitter.EmitTx,
PostgresEmitter.enrichFromContext, and emitEvent so the transactional audit path
mirrors PostgresEmitter.Emit.
In `@packages/engine/internal/cli/retry.go`:
- Around line 39-53: The retry path constructs a new engine.Engine but never
applies the configured per-team concurrency cap, so ensure you copy
cfg.Engine.MaxConcurrentExecutionsPerTeam onto the created engine instance (eng)
after engine.New(database) and before using it; specifically set the engine's
concurrency field/property (the Engine instance created in this block) from
cfg.Engine.MaxConcurrentExecutionsPerTeam so the retry command respects the same
per-team limit enforced by run/serve wiring and honors CLI/env overrides.
- Around line 73-76: The JSON path currently returns immediately from
json.NewEncoder(cmd.OutOrStdout()).Encode(result) before the code checks
result.Status == "failed", so `mantle retry --output json` can exit 0 on
failures; change the flow in the retry handling function so you first compute
the execution error (the variable/error produced when result.Status == "failed"
or similar), then always write the result (use
json.NewEncoder(...).Encode(result) when outputFormat == "json" or the text
branch otherwise), and finally return the computed error; update both the JSON
branch around json.NewEncoder(cmd.OutOrStdout()).Encode(result) and the second
JSON block later (lines ~104-116) to follow this pattern so emission and error
return are decoupled.
In `@packages/engine/internal/cli/run.go`:
- Line 88: The code only treats result.Status == "failed" as an error after
calling eng.ExecuteWithOptions (the variable named result), so executions that
end with "timed_out" or "cancelled" return exit 0; update the post-execution
status check(s) that inspect result.Status (the branch handling "failed") to
also treat "timed_out" and "cancelled" as error conditions and return a non-zero
exit (do the same for the duplicate check around lines 125-137), preserving
existing error messaging but mapping these additional statuses into the error
path.
In `@packages/engine/internal/config/config.go`:
- Around line 136-137: The engine config misses setting the default for
engine.max_concurrent_executions_per_team (documented default 10), so update the
config loader (the Load/Unmarshal path in
packages/engine/internal/config/config.go) to call
v.SetDefault("engine.max_concurrent_executions_per_team", 10) before
unmarshalling into the struct so that MaxConcurrentExecutionsPerTeam on the
EngineConfig is 10 instead of 0; apply the same v.SetDefault addition in the
other loader location referenced around the duplicate spot (the other
Load/initialization call near line 257) so CheckConcurrencyLimits won’t treat 0
as “unlimited.”
- Around line 281-288: The current logic treats cfg.Version == 0 as a fallback
to 1 even when the user explicitly set "version: 0"; change it so the default to
1 only occurs when the source doesn't set the key (use v.IsSet("version") to
detect this), and if the version key is set and cfg.Version != 1 return the
existing unsupported-version error (thus rejecting explicit 0). Update the block
that references cfg.Version and use v.IsSet("version") to gate setting
cfg.Version = 1 and to enforce that any provided value other than 1 is rejected.
In `@packages/engine/internal/engine/concurrency_test.go`:
- Line 31: The tests only exercise CheckConcurrencyLimits with
teamMaxConcurrent=0 so the team-level path never runs; update the test cases in
concurrency_test.go to include at least one call to CheckConcurrencyLimits where
teamMaxConcurrent is non-zero (e.g., pass a positive value for
teamMaxConcurrent) and assert the expected queue/reject behavior for that path.
Locate the existing CheckConcurrencyLimits calls (the calls using "wf-a", 5,
"queue", 0 and the similar calls later in the file) and add or modify one case
to use a non-zero teamMaxConcurrent to exercise the per-team advisory-lock/count
branch and verify its result.
In `@packages/engine/internal/engine/engine.go`:
- Around line 286-314: The current timeout check uses the parent ctx (ctx.Err())
which can be nil even when a step timed out; update the logic to determine
terminalStatus from the step execution result instead—have executeStepLogic
return an explicit timeout indicator (e.g., a field on stepResult like TimedOut
or a wrapped timeout error) and change the branch that sets
terminalStatus/result.Error to use stepResult.TimedOut or timeout-specific error
text in stepResult.Error (and fallback to ctx.Err() only if stepResult gives no
info); update references in this block (failedStepName, stepResult,
executeStepLogic, result.Status, result.Error, updateExecutionStatus) to use
that propagated timeout flag so step-level timeouts are correctly classified as
"timed_out".
In `@packages/engine/internal/engine/hooks.go`:
- Around line 37-42: The CEL execution context currently hardcodes the
"failed_in" field to "steps" in the celCtx.Execution map, which can be
misleading; change the initial value of "failed_in" in celCtx.Execution (the map
built in hooks.go) to an empty string (or a more neutral default) and ensure the
existing hook-failure path that updates "failed_in" to the hook block name still
writes into the same key so that when no failure occurs the field is blank and
when a hook step fails it contains the actual block name (update any code that
reads/sets celCtx.Execution["failed_in"] accordingly).
In `@packages/engine/internal/engine/retry.go`:
- Around line 185-197: The audit emission for retries is currently done via
e.Auditor.Emit after the DB transaction commits (see e.Auditor.Emit in
retry.go), which can leave the new execution without an audit if the emitter
fails; move this emission into the same transaction using the transactional
helper (e.g., audit.EmitTx or the pattern used in rotate_key.go) so the audit
write is part of the DB transaction — update retry.go to call the transactional
audit emitter inside the transaction scope (using the same ctx/tx and the same
metadata: originalExecID, retryStep, workflowName, version) and remove the
post-commit e.Auditor.Emit call.
In `@packages/engine/internal/workflow/validate.go`:
- Around line 618-620: validateHookSteps currently resets its local seen map per
block which allows duplicate hook step names across hooks.OnSuccess,
hooks.OnFailure, and hooks.OnFinish; add a cross-block uniqueness check in
validateHooks by maintaining a single shared seen map (or set) across calls to
validateHookSteps so any name reused in different hook blocks is rejected.
Specifically, update validateHooks to create and pass a shared seen container
into validateHookSteps (or perform an additional check after the three append
calls) so names in hooks.OnSuccess, hooks.OnFailure, and hooks.OnFinish are
unique across all blocks and cause a validation error if duplicated.
In `@packages/site/src/content/docs/cli-reference/workflow-commands.md`:
- Around line 295-296: The docs incorrectly describe `--force` as overriding
execution status; instead update the CLI docs and the error example to state
that `--force` bypasses concurrency limits (a concurrency override) to match the
behavior implemented in packages/engine/internal/cli/retry.go (the flag wiring
for --force). Locate references to `--force` in workflow-commands.md (including
the table row and the example around lines ~324–326) and change the description
and any example text/errors to say it allows retry despite concurrency limits
(e.g., “Bypass concurrency limits to start a retry even when the concurrency
limiter would prevent it”) rather than saying it retries completed or cancelled
executions. Ensure wording references the concurrency bypass behavior rather
than status override so docs match the implementation in retry.go.
---
Duplicate comments:
In `@packages/engine/internal/config/config.go`:
- Around line 290-297: The fallback from the deprecated "tmp" section currently
replaces the entire cfg.Storage (StorageConfig) and thus clobbers any values set
via env/flags; change the logic in config.go so that when v.Sub("tmp")
unmarshals into a legacy StorageConfig you merge missing fields into cfg.Storage
rather than assigning it wholesale, or only apply the legacy values when no
"storage.*" keys were set; implement a field-by-field copy (or a utility merge)
that preserves non-zero values already present in cfg.Storage (e.g., keep
existing cfg.Storage.Type/other fields if set) and only set fields from legacy
when cfg.Storage has their zero value. Ensure this happens in the same block
where v.Sub("tmp") and sub.Unmarshal(&legacy) are used and add a unit test or
validation to cover env/flag overrides vs. tmp fallback.
In `@packages/engine/internal/db/migrations/016_safety_net.sql`:
- Around line 42-46: The down migration fails because it drops the hook_block
column and then tries to create a partial index that references hook_block; fix
by first reconciling/deleting hook rows, then drop the column, then recreate the
original global unique index. Concretely: in the rollback for step_executions,
remove any rows where hook_block IS NOT NULL (or otherwise migrate them away) so
the legacy schema can represent remaining data, then DROP COLUMN hook_block, and
finally CREATE UNIQUE INDEX step_executions_execution_id_step_name_attempt_key
ON step_executions (execution_id, step_name, attempt) without a WHERE predicate.
In `@packages/engine/internal/secret/store.go`:
- Around line 197-200: Create currently writes credentials directly via s.DB
while RotateAll uses a transaction that acquires the advisory lock (SELECT
pg_advisory_xact_lock($1) with key 0x4D414E544C45), so credential creation can
race with rotation; update (*Store).Create to route writes through the same
serialized path by either accepting/using the same tx that acquires the advisory
lock or by performing the same pg_advisory_xact_lock and using a transaction
around the insert, ensuring Create uses the same lock key and transaction
semantics as RotateAll (or factor a helper used by both) so writers and rotation
are mutually serialized.
🪄 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: ASSERTIVE
Plan: Pro
Run ID: 22a7bac7-7621-47dc-99dd-d779259bd3fd
📒 Files selected for processing (21)
packages/engine/internal/artifact/storage.gopackages/engine/internal/audit/audit.gopackages/engine/internal/audit/postgres.gopackages/engine/internal/cli/retry.gopackages/engine/internal/cli/rotate_key.gopackages/engine/internal/cli/run.gopackages/engine/internal/config/config.gopackages/engine/internal/db/migrations/016_safety_net.sqlpackages/engine/internal/engine/concurrency.gopackages/engine/internal/engine/concurrency_test.gopackages/engine/internal/engine/engine.gopackages/engine/internal/engine/hooks.gopackages/engine/internal/engine/hooks_test.gopackages/engine/internal/engine/retry.gopackages/engine/internal/engine/toolsteps.gopackages/engine/internal/metrics/metrics.gopackages/engine/internal/secret/store.gopackages/engine/internal/secret/store_test.gopackages/engine/internal/workflow/validate.gopackages/site/src/content/docs/cli-reference/workflow-commands.mdpackages/site/src/content/docs/configuration.md
Comment on lines
+92
to
+108
| if err := tx.Commit(); err != nil { | ||
| return fmt.Errorf("committing transaction: %w", err) | ||
| } | ||
|
|
||
| fmt.Fprintf(cmd.OutOrStdout(), "Re-encrypted %d credential(s).\n", count) | ||
| fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n", newKey) | ||
|
|
||
| // Output the new key securely. | ||
| if !userProvided { | ||
| if outputKey != "" { | ||
| if err := os.WriteFile(outputKey, []byte(newKey+"\n"), 0600); err != nil { | ||
| return fmt.Errorf("writing key to file: %w", err) | ||
| } | ||
| fmt.Fprintf(cmd.OutOrStdout(), "New key written to: %s\n", outputKey) | ||
| } else { | ||
| fmt.Fprintln(cmd.ErrOrStderr(), "WARNING: The following key is sensitive. Store it securely and clear your terminal.") | ||
| fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n", newKey) | ||
| } |
Contributor
There was a problem hiding this comment.
Post-commit key file write failure could orphan the new key.
If tx.Commit() succeeds (line 92) but os.WriteFile(outputKey, ...) fails (line 101), the rotation is persisted but the user doesn't have the new key. Consider writing the key file before committing the transaction, or at minimum, print the key to stderr as a fallback when file write fails.
🛡️ Proposed fix to ensure key availability
+ // Write key file before commit so we can rollback if it fails.
+ if !userProvided && outputKey != "" {
+ if err := os.WriteFile(outputKey, []byte(newKey+"\n"), 0600); err != nil {
+ return fmt.Errorf("writing key to file: %w", err)
+ }
+ }
+
if err := tx.Commit(); err != nil {
return fmt.Errorf("committing transaction: %w", err)
}
fmt.Fprintf(cmd.OutOrStdout(), "Re-encrypted %d credential(s).\n", count)
- // Output the new key securely.
- if !userProvided {
- if outputKey != "" {
- if err := os.WriteFile(outputKey, []byte(newKey+"\n"), 0600); err != nil {
- return fmt.Errorf("writing key to file: %w", err)
- }
- fmt.Fprintf(cmd.OutOrStdout(), "New key written to: %s\n", outputKey)
- } else {
+ if !userProvided {
+ if outputKey != "" {
+ fmt.Fprintf(cmd.OutOrStdout(), "New key written to: %s\n", outputKey)
+ } else {
fmt.Fprintln(cmd.ErrOrStderr(), "WARNING: The following key is sensitive. Store it securely and clear your terminal.")
fmt.Fprintf(cmd.OutOrStdout(), "New key: %s\n", newKey)
}
}…y, validation - Config: distinguish missing vs explicit version 0 via v.IsSet; field-by-field tmp->storage merge to preserve env/flag overrides - CLI: run/retry exit non-zero on failed/timed_out/cancelled; JSON output no longer masks failure exit codes - Retry: set MaxConcurrentExecutionsPerTeam; audit emitted inside transaction; link+copy+audit in single tx - Hooks: failed_in defaults to "" (empty), set to "steps" only when main steps fail; hook failures update to block name as before - Audit: EmitTx accepts optional PostgresEmitter for context enrichment - Storage: reject absolute/.. keys before MkdirAll; fix symlink fallback in Delete/DeleteByPrefix to use resolvedBase - Validation: cross-block hook step name uniqueness (shared CEL ns) - Secret store: Create acquires same advisory lock as RotateAll - Engine: step-level timeout detection via errors.Is(DeadlineExceeded) returns timed_out status; migration adds timed_out to step status - Migration down: delete hook rows before dropping column, restore proper unique constraint instead of partial index - Tests: team-level concurrency, updated timeout/validation expectations - Docs: --force description corrected for retry command Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
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/secret/store.go (1)
34-50:⚠️ Potential issue | 🔴 Critical
Createencrypts outside the advisory lock, allowing old-key ciphertext to be inserted afterRotateAllcommits.The encryption at lines 34–37 happens before the transaction and advisory lock are acquired. A concurrent
Createcan encrypt with the current key (K1), then block waiting for the advisory lock at line 49. Meanwhile,RotateAllacquires the lock, rotates all credentials to a new key (K2), and commits. WhenCreatefinally acquires the lock and inserts, the credential is encrypted with K1 but was not included inRotateAll's snapshot. Once callers switch to K2 decryption, that row becomes unreadable.Move encryption inside the transaction after acquiring the advisory lock (lines 49–51), or add key version tracking to the schema and re-encrypt if the version changed while waiting for the lock.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/secret/store.go` around lines 34 - 50, The Create flow encrypts plaintext before acquiring the advisory lock, so concurrent RotateAll can change keys and leave the new row encrypted with the old key; fix by moving the call to s.Encryptor.Encrypt(plaintext) into the transaction after acquiring the advisory lock (i.e., after tx is created and after tx.ExecContext(... "SELECT pg_advisory_xact_lock(...)") succeeds) inside the Create implementation, or alternatively implement key-version tracking on the credential rows and re-encrypt with the current key before insert if the version changed while waiting; ensure the code references the same tx for the insert so the newly encrypted ciphertext is committed under the advisory lock with RotateAll.
♻️ Duplicate comments (2)
packages/site/src/content/docs/cli-reference/workflow-commands.md (1)
324-326:⚠️ Potential issue | 🟡 MinorThe retry error example still treats
--forceas a status override.The flag table now says
--forceonly bypasses concurrency limits, but this example still tells users it can retry a completed execution. Drop theUse --force to retry anywaysuffix so the docs match the command.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/site/src/content/docs/cli-reference/workflow-commands.md` around lines 324 - 326, The example error message currently ends with "Use --force to retry anyway" which contradicts the flag table that says --force only bypasses concurrency limits; update the example error string (the code block containing "Error: execution abc123-def456 is not in a failed state (status: completed). Use --force to retry anyway.") to remove the "Use --force to retry anyway" suffix so it reads "Error: execution abc123-def456 is not in a failed state (status: completed)." and ensure the surrounding text/examples referencing the --force flag remain consistent.packages/engine/internal/config/config.go (1)
136-137:⚠️ Potential issue | 🟡 MinorMissing default for
engine.max_concurrent_executions_per_team.Documentation specifies the default should be
10, but nov.SetDefault()call exists in the code. Without it, the field will default to0(unlimited) if not explicitly configured, contradicting the documented behavior.Add:
+v.SetDefault("engine.max_concurrent_executions_per_team", 10)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/config/config.go` around lines 136 - 137, The config struct field MaxConcurrentExecutionsPerTeam lacks a viper default, so add a v.SetDefault call for the key "engine.max_concurrent_executions_per_team" with value 10 in the configuration initialization (the same place where other defaults are set, e.g., alongside v.SetDefault calls for Budget and other engine keys) so the MaxConcurrentExecutionsPerTeam field gets the documented default of 10 when not provided.
🤖 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/retry.go`:
- Around line 129-132: The retry command reads an "--output" flag but never
registers it, causing runtime errors; add a String flag registration for
"output" on the same command object (cmd) alongside the other flags (e.g., next
to cmd.Flags().String("from-step", ...), cmd.Flags().Bool("force", ...),
cmd.Flags().BoolP("verbose", ...)), using an empty string default and a brief
description like "Output format (e.g. json|text)"; ensure the flag name matches
exactly ("output") so the code that reads the flag (the --output usage at line
58) returns the expected value.
- Around line 93-97: The JSON encoding error is currently ignored; change the
json.NewEncoder(cmd.OutOrStdout()).Encode(result) call to capture its error
(e.g., err := json.NewEncoder(cmd.OutOrStdout()).Encode(result)) and handle it
instead of discarding it: if err != nil then write the error to the command
stderr (cmd.ErrOrStderr()) or return the error (or wrap it) so the caller sees
the failure, otherwise continue to return exitErr as before; update the block
around outputFormat == "json" and the Encode(result) call to implement this
error check and proper propagation.
In `@packages/engine/internal/cli/run.go`:
- Around line 113-115: The current branch that handles outputFormat == "json"
ignores errors from json.NewEncoder(...).Encode(result) and always returns
exitErr; change this so the Encode error is captured and returned: if Encode
returns an error, return that error (or if exitErr is non-nil, wrap/combine them
into a single error that preserves both messages). Update the json output block
in run.go (the code using json.NewEncoder(cmd.OutOrStdout()).Encode(result) and
return exitErr) to capture encodeErr, and return encodeErr when non-nil, or a
combined error when both encodeErr and exitErr exist.
---
Outside diff comments:
In `@packages/engine/internal/secret/store.go`:
- Around line 34-50: The Create flow encrypts plaintext before acquiring the
advisory lock, so concurrent RotateAll can change keys and leave the new row
encrypted with the old key; fix by moving the call to
s.Encryptor.Encrypt(plaintext) into the transaction after acquiring the advisory
lock (i.e., after tx is created and after tx.ExecContext(... "SELECT
pg_advisory_xact_lock(...)") succeeds) inside the Create implementation, or
alternatively implement key-version tracking on the credential rows and
re-encrypt with the current key before insert if the version changed while
waiting; ensure the code references the same tx for the insert so the newly
encrypted ciphertext is committed under the advisory lock with RotateAll.
---
Duplicate comments:
In `@packages/engine/internal/config/config.go`:
- Around line 136-137: The config struct field MaxConcurrentExecutionsPerTeam
lacks a viper default, so add a v.SetDefault call for the key
"engine.max_concurrent_executions_per_team" with value 10 in the configuration
initialization (the same place where other defaults are set, e.g., alongside
v.SetDefault calls for Budget and other engine keys) so the
MaxConcurrentExecutionsPerTeam field gets the documented default of 10 when not
provided.
In `@packages/site/src/content/docs/cli-reference/workflow-commands.md`:
- Around line 324-326: The example error message currently ends with "Use
--force to retry anyway" which contradicts the flag table that says --force only
bypasses concurrency limits; update the example error string (the code block
containing "Error: execution abc123-def456 is not in a failed state (status:
completed). Use --force to retry anyway.") to remove the "Use --force to retry
anyway" suffix so it reads "Error: execution abc123-def456 is not in a failed
state (status: completed)." and ensure the surrounding text/examples referencing
the --force flag remain consistent.
🪄 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: ASSERTIVE
Plan: Pro
Run ID: 2b7daf3f-1cdc-40e6-908b-00037de704e7
📒 Files selected for processing (16)
packages/engine/internal/artifact/storage.gopackages/engine/internal/audit/postgres.gopackages/engine/internal/cli/retry.gopackages/engine/internal/cli/run.gopackages/engine/internal/config/config.gopackages/engine/internal/config/config_test.gopackages/engine/internal/db/migrations/016_safety_net.sqlpackages/engine/internal/engine/concurrency_test.gopackages/engine/internal/engine/engine.gopackages/engine/internal/engine/engine_test.gopackages/engine/internal/engine/hooks.gopackages/engine/internal/engine/retry.gopackages/engine/internal/secret/store.gopackages/engine/internal/workflow/validate.gopackages/engine/internal/workflow/validate_hooks_test.gopackages/site/src/content/docs/cli-reference/workflow-commands.md
Comment on lines
+129
to
+132
| cmd.Flags().String("from-step", "", "Step to retry from (default: first failed step)") | ||
| cmd.Flags().Bool("force", false, "Bypass concurrency limits") | ||
| cmd.Flags().BoolP("verbose", "v", false, "Show step outputs and durations") | ||
| return cmd |
Contributor
There was a problem hiding this comment.
Missing --output flag registration.
The command reads --output at line 58 but doesn't register it. This will cause a runtime error or return empty string.
🔧 Proposed fix
cmd.Flags().String("from-step", "", "Step to retry from (default: first failed step)")
cmd.Flags().Bool("force", false, "Bypass concurrency limits")
cmd.Flags().BoolP("verbose", "v", false, "Show step outputs and durations")
+ cmd.Flags().StringP("output", "o", "text", "Output format (text or json)")
return cmd📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| cmd.Flags().String("from-step", "", "Step to retry from (default: first failed step)") | |
| cmd.Flags().Bool("force", false, "Bypass concurrency limits") | |
| cmd.Flags().BoolP("verbose", "v", false, "Show step outputs and durations") | |
| return cmd | |
| cmd.Flags().String("from-step", "", "Step to retry from (default: first failed step)") | |
| cmd.Flags().Bool("force", false, "Bypass concurrency limits") | |
| cmd.Flags().BoolP("verbose", "v", false, "Show step outputs and durations") | |
| cmd.Flags().StringP("output", "o", "text", "Output format (text or json)") | |
| return cmd |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/cli/retry.go` around lines 129 - 132, The retry
command reads an "--output" flag but never registers it, causing runtime errors;
add a String flag registration for "output" on the same command object (cmd)
alongside the other flags (e.g., next to cmd.Flags().String("from-step", ...),
cmd.Flags().Bool("force", ...), cmd.Flags().BoolP("verbose", ...)), using an
empty string default and a brief description like "Output format (e.g.
json|text)"; ensure the flag name matches exactly ("output") so the code that
reads the flag (the --output usage at line 58) returns the expected value.
Comment on lines
113
to
+115
| if outputFormat == "json" { | ||
| return json.NewEncoder(cmd.OutOrStdout()).Encode(result) | ||
| _ = json.NewEncoder(cmd.OutOrStdout()).Encode(result) | ||
| return exitErr |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cd packages/engine && find . -name "run.go" -path "*/cli/*" -type fRepository: dvflw/mantle
Length of output: 77
🏁 Script executed:
cd packages/engine && head -150 internal/cli/run.go | tail -80Repository: dvflw/mantle
Length of output: 2600
🏁 Script executed:
cd packages/engine && wc -l internal/cli/run.goRepository: dvflw/mantle
Length of output: 79
Don't swallow JSON output failures.
If Encode fails in --output json mode, successful runs currently return nil even though no usable JSON was written. Return the encoding error, and combine it with exitErr when both need to be surfaced.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/engine/internal/cli/run.go` around lines 113 - 115, The current
branch that handles outputFormat == "json" ignores errors from
json.NewEncoder(...).Encode(result) and always returns exitErr; change this so
the Encode error is captured and returned: if Encode returns an error, return
that error (or if exitErr is non-nil, wrap/combine them into a single error that
preserves both messages). Update the json output block in run.go (the code using
json.NewEncoder(cmd.OutOrStdout()).Encode(result) and return exitErr) to capture
encodeErr, and return encodeErr when non-nil, or a combined error when both
encodeErr and exitErr exist.
… round 3) - Handle JSON encoding errors in retry.go and run.go instead of discarding - Move credential encryption inside advisory lock in store.Create to prevent race with concurrent RotateAll - Fix misleading --force error example in CLI docs Not applied: --output flag registration (already exists as persistent flag on root command), max_concurrent_executions_per_team default of 10 (spec says 0 = unlimited, Go zero value is correct)
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
packages/engine/internal/cli/retry.go (1)
131-134:⚠️ Potential issue | 🟠 MajorMissing
--outputflag registration.Line 58 reads
cmd.Flags().GetString("output")but the flag is never registered. This causes--output jsonto silently fail (returns empty string, defaulting to text mode).🔧 Proposed fix
cmd.Flags().String("from-step", "", "Step to retry from (default: first failed step)") cmd.Flags().Bool("force", false, "Bypass concurrency limits") cmd.Flags().BoolP("verbose", "v", false, "Show step outputs and durations") + cmd.Flags().StringP("output", "o", "text", "Output format: text or json") return cmd🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/engine/internal/cli/retry.go` around lines 131 - 134, The code calls cmd.Flags().GetString("output") but never registers the flag; add a registration for the "output" flag alongside the existing flags (where cmd is configured in retry.go) — e.g., call cmd.Flags().String("output", "text", "Output format (text|json)") so GetString("output") returns the intended value; ensure the flag name matches exactly "output" and place the registration near the other flag registrations (the block that sets "from-step", "force", "verbose") so the --output option is recognized.
🤖 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/site/src/content/docs/cli-reference/workflow-commands.md`:
- Around line 379-383: Update the example error text in the docs to match the
actual error format produced by the rollback code: replace the shown message
"Error: version 5 not found for workflow \"fetch-and-summarize\"" with the
implementation's format "Error: workflow \"fetch-and-summarize\" version 5 not
found" so it matches the error string used in
packages/engine/cmd/cli/rollback.go (the "workflow %q version %d not found"
message).
---
Duplicate comments:
In `@packages/engine/internal/cli/retry.go`:
- Around line 131-134: The code calls cmd.Flags().GetString("output") but never
registers the flag; add a registration for the "output" flag alongside the
existing flags (where cmd is configured in retry.go) — e.g., call
cmd.Flags().String("output", "text", "Output format (text|json)") so
GetString("output") returns the intended value; ensure the flag name matches
exactly "output" and place the registration near the other flag registrations
(the block that sets "from-step", "force", "verbose") so the --output option is
recognized.
🪄 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: ASSERTIVE
Plan: Pro
Run ID: 8f0421d6-b6e3-4365-8d62-1acf1d588cc1
📒 Files selected for processing (4)
packages/engine/internal/cli/retry.gopackages/engine/internal/cli/run.gopackages/engine/internal/secret/store.gopackages/site/src/content/docs/cli-reference/workflow-commands.md
7 tasks
4 tasks
This was referenced Apr 17, 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
version: 1field for future-proof config validation,tmpsection renamed tostoragewith deprecation fallback ([Infra] Config file versioning (version: 1 in mantle.yaml) #52, [Breaking] Rename tmp config section to storage #75)RotateAllwith caller-managed transactions, audit events, stderr key warning ([Security] Secret key rotation (mantle secrets rotate-key) #51)queuedstatus, completion-triggered + backup queue promotion ([Feature] Workflow concurrency controls (max parallel executions) #49)on_success,on_failure,on_finishblocks with separate timeout budget, best-effort semantics, full CEL context ([Feature] on_failure workflow-level error handler #30)mantle retrycreates new execution with copied upstream checkpoints,--forceflag for concurrency bypass ([Feature] Execution retry from failed step (mantle retry) #48)mantle rollbackinserts old content as new version withrollback_oftraceability ([Feature] Workflow rollback command (mantle rollback) #50)Issues Closed
Closes #52, closes #75, closes #51, closes #49, closes #30, closes #48, closes #50
Migration
Single migration
016_safety_net.sqladds columns toteams,workflow_executions,step_executions,workflow_definitionswith partial unique indexes for hook step disambiguation.Breaking:
tmpconfig section renamed tostorage. Oldtmpsection still works with deprecation warning in v0.4.0; will be removed in a future release.Review History
Test plan
go test ./...— 20 packages, 0 failures (including 15 new tests for concurrency/hooks/retry)go build ./cmd/mantle— binary builds cleango vet ./...— 0 findingsmantle init && mantle validate <workflow> && mantle apply <workflow>mantle retryandmantle rollbackcommands work end-to-endmax_parallel_executions: 1🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Behavior Changes
Configuration
Documentation