fix(flows): inherit flow_env in sub-flow predicates

[hugocasa]

Summary

Predicates inside a BranchOne / for-loop sub-flow couldn't read flow_env (e.g., skip_if: flow_env.X === true returned false even when X was true). Sub-flows constructed by payload_from_modules for branches/loops don't carry the parent's flow_env in their own FlowValue, so handle_flow saw flow.flow_env = None and propagated None to every eval. Input transforms with simple flow_env.X happened to work because they hit a regex API fast-path that walks to the root, but anything wrapped in Boolean(...) (skip_if, stop_after_if, retry_if, branchone predicates) or any complex expression (Math.min(flow_env.X, 10)) fell through to QuickJS and used the local None.

Same root cause also broke four eval calls in update_flow_status_after_job_completion_internal that hardcoded None for flow_env: stop_after_if, stop_after_all_iters_if, and both evaluate_retry paths.

Changes

• New helper fetch_root_flow_env in worker_flow.rs — a recursive CTE walks COALESCE(flow_innermost_root_job, parent_job) up the job chain (depth ≤ 50), prefers flow_version.value, falls back to raw_flow, and returns the closest ancestor's flow_env. Recursion is needed because parallel for-loop iterations and imported sub-flows reset flow_innermost_root_job to None.
• handle_flow — when flow.flow_env.is_none() and the job has a parent_job, inherit via the lookup before resolving $var:/$res:.
• update_flow_status_after_job_completion_internal — same lookup (lazy: only when current/failure module has stop_after_if / stop_after_all_iters_if / retry), then thread the env into:
  • compute_bool_from_expr for stop_after_if
  • evaluate_stop_after_all_iters_if (new flow_env param, both call sites)
  • both evaluate_retry calls (continue_on_error + should_retry closure)
• Imported flows (FlowModuleValue::Flow { path }) keep their own flow_env if defined (status quo: replace, no merge); if undefined they now inherit from the parent.
• 9 regression tests in flow_engine_parity.rs, all verified to fail on main and pass with the fix:
  • test_flow_env_skip_if_inside_branchone — the original bug
  • test_flow_env_skip_if_nested_branchone — recursion past depth 1 (branch-in-branch)
  • test_flow_env_skip_if_in_parallel_forloop — flow_innermost_root_job=None case
  • test_flow_env_complex_input_transform_in_branch — QuickJS fallback path inside sub-flow
  • test_flow_env_imported_flow_uses_own_env — pins option (i): own env wins
  • test_flow_env_imported_flow_inherits_when_unset — imported flow with no env inherits from parent
  • test_flow_env_in_stop_after_if, test_flow_env_in_retry_if, test_flow_env_in_stop_after_all_iters_if

Test plan

• Create a flow with flow_env: { ENV: "production" }, add a BranchOne with an inner step whose Skip condition is flow_env.ENV === "production". Run the flow → inner step is skipped.
• Same flow but use Stop early if instead → flow stops at that step.
• Same flow but use Retry condition on a failing step → retries fire.
• In a non-trivial step input transform inside a branch, use Math.min(flow_env.LIMIT, 10) → resolves correctly.
• Repeat the skip_if test with a parallel for-loop instead of BranchOne → still skips.
• Import a flow that has its own flow_env from a parent flow with a different flow_env → imported flow uses its own values.

---

Generated with Claude Code

---

Summary by cubic

Fixes missing flow_env in sub-flow predicates and respects imported-flow scoping. Predicates and stop/retry logic now inherit the nearest flow_env, with DB lookups only when a predicate needs it.

• Bug Fixes

  • Added fetch_root_flow_env that recursively walks flow_innermost_root_job -> parent_job (depth-capped) and returns the closest ancestor’s flow_env, preferring flow_version.value over raw_flow; avoids jumping via root_job so imported flows keep their scope.
  • In handle_flow, inherit flow_env for sub-flows before resolving $var:/$res:.
  • In update_flow_status_after_job_completion_internal, lazily resolve flow_env only if stop_after_if/stop_after_all_iters_if/retry_if are set, then pass it to compute_bool_from_expr, evaluate_stop_after_all_iters_if (new param), and evaluate_retry.
  • Imported flows keep their own flow_env; if unset, they inherit from the parent. Predicates inside imported flows and nested branches now see the correct scope.
  • Added regression tests for nested/parallel branches and loops, complex input transforms, imported flows (own/inherit/nested branch), and stop_after_if/stop_after_all_iters_if/retry_if.

• Refactors

  • Removed the GET /jobs/flow_env_by_flow_job_id API and client helper; no HTTP calls for flow_env.
  • Simplified JS eval fast-path to only optimize results.*; flow_env is read directly in QuickJS (no wrapping/await).

^{Written for commit 732e123. Summary will update on new commits.}

[hugocasa]

/ai review this PR

[cloudflare-workers-and-pages]

Deploying windmill with    Cloudflare Pages

Latest commit:	732e123
Status:	✅  Deploy successful!
Preview URL:	https://6253cd94.windmill.pages.dev
Branch Preview URL:	https://flow-env-skip-eval.windmill.pages.dev

View logs

[claude]

Claude finished @hugocasa's task in 4m 32s —— View job

---

Review

Solid, well-scoped fix with strong test coverage. The architectural split between fetch_root_flow_env (DB lookup with recursive CTE) and resolve_flow_env_for_status_update (lookup + $var:/$res: interpolation) mirrors the existing pattern in handle_flow, and the lazy needs_flow_env gating keeps the common path free of extra queries. Below are findings worth considering — all minor, none blocking.

Suggestions

1. needs_flow_env check is broader than necessary — worker_flow.rs:472-477

evaluate_retry only consults flow_env when retry.retry_if is set (see worker_flow.rs:2274-2286). The current check fires the lookup whenever m.retry.is_some() / failure_module.retry.is_some(), even for retries without a retry_if expression — a common case. Tightening to m.retry.as_ref().is_some_and(|r| r.retry_if.is_some()) (and the same for the failure module) avoids the extra DB round-trip + transform_json for these. Fix this →

2. Silent error in get_mini_pulled_job failure — worker_flow.rs:2398-2401

let mini = match get_mini_pulled_job(db, &flow_job_id).await {
    Ok(Some(j)) => j,
    _ => return Some(env),
};

The catch-all _ swallows both Ok(None) (job missing — fine) and Err(e) (DB issue — should be logged). Consider matching Err(e) explicitly with a tracing::warn! so DB issues during status updates don't go silent.

3. Recursive CTE omits root_job — worker_flow.rs:2444

The walk uses COALESCE(c.flow_innermost_root_job, c.parent_job), but the canonical helper get_root_job_id (common.rs:1325-1330) and the existing API query (jobs.rs:475) both prefer root_job first. Including it as the primary preference would shorten the walk in the non-reset cases (and the recursion still handles parallel-loop / imported sub-flows where it's reset to NULL). Likely minor in practice — your bench depth is small — but it would align with the rest of the codebase.

4. Empty-env optimization inconsistency — worker_flow.rs:2395-2397 vs worker_flow.rs:2632-2654

resolve_flow_env_for_status_update short-circuits on env.is_empty() before calling transform_json, but handle_flow doesn't. Either harmonize (skip in both) or drop the optimization (transform_json on an empty map is cheap anyway).

5. Unnecessary clone in resolve_flow_env_for_status_update — worker_flow.rs:2390-2394

let env = if let Some(ref e) = flow_value.flow_env {
    e.clone()
} else { ... };

When local env is set, the entire HashMap<String, Box<RawValue>> is cloned. Since transform_json takes &HashMap<...>, this could be Cow<'_, HashMap<...>> or two branches that pass by reference. Minor — not a hot path — but worth a thought.

6. Magic number 50 for recursion depth — worker_flow.rs:2445

The cycle/runaway guard c.depth < 50 would benefit from a named constant + comment explaining "max nested sub-flow depth before bailing." Tiny readability win.

Things I checked and look good

• All four None sites in update_flow_status_after_job_completion_internal are now threaded with resolved_flow_env.as_ref() (lines 627, 911, 1131, 1217, 1599); the new param on evaluate_stop_after_all_iters_if is wired at both call sites.
• Imported-flow precedence (own env wins, falls back to parent only when unset) is pinned by test_flow_env_imported_flow_uses_own_env and ..._inherits_when_unset — matches the comment in handle_flow:2615-2625 ("status quo: replace, no merge").
• The CTE correctly returns the closest ancestor with a non-null flow_env (ORDER BY chain.depth ASC LIMIT 1), and prefers flow_version.value over the row-pinned raw_flow — consistent with the API endpoint at jobs.rs:466-471.
• if env.is_empty() correctly handles the Some({}) case from the SQL IS NOT NULL check (a JSON {} passes that filter).
• Test coverage is thorough: branches/nested branches/parallel loops/QuickJS-fallback transforms/imported flows/stop_after_if/retry_if/stop_after_all_iters_if are all covered with both before/after assertions and prose explaining the failure mode.
  · branch: flow-env-skip-eval

[claude]

Claude finished @hugocasa's task in 1m 35s —— View job

---

Re-review (commits since prior review: 0ac94de)

Good to merge — 0ac94de cleanly addresses four of my earlier suggestions; the two remaining nits are minor and not blocking.

What changed since prior review

refactor(flows): align flow_env lookup with get_root_job_id and tighten gate — all backend/worker_flow.rs:

Prior finding	Status
#1 needs_flow_env over-eager for retry without retry_if	✅ Fixed via retry_uses_flow_env closure (worker_flow.rs:473-482)
#2 Silent error on get_mini_pulled_job failure	✅ Fixed — Err(e) now matched explicitly with tracing::warn! (worker_flow.rs:2406-2409)
#3 Recursive CTE omits root_job	✅ Fixed — COALESCE(c.root_job, c.flow_innermost_root_job, c.parent_job) aligns with jobs.rs:475 and jobs.rs:581 (worker_flow.rs:2459)
#6 Magic number 50	✅ Fixed — extracted as MAX_FLOW_ENV_LOOKUP_DEPTH constant with comment (worker_flow.rs:2429-2431), passed as $3 parameter

Spot-checks on the new commit

• CTE select-list / column-list match: the chain CTE column list (id, root_job, parent_job, flow_innermost_root_job, runnable_id, runnable_path, raw_flow, depth) matches both the anchor and recursive SELECT lists. ✓
• Parameter binding: MAX_FLOW_ENV_LOOKUP_DEPTH: i32 bound as $3; c.depth < $3 is well-typed (depth is implicitly int4 from the 0 literal + + 1). ✓
• Sqlx cache regenerated: three new query JSONs added, the prior one deleted — cache is consistent with the SQL change.
• Tightened gate is sound: evaluate_retry body at worker_flow.rs:2274-2286 confirms flow_env is only consulted inside the retry_if branch — gating on retry_if.is_some() is exact, no false negatives.

Remaining minor nits (not blocking — same as prior review)

P2 — Empty-env optimization inconsistency — worker_flow.rs:2400-2402 vs handle_flow:2632-2654

resolve_flow_env_for_status_update short-circuits on env.is_empty() before the transform_json call; handle_flow doesn't. Either harmonize or just drop the optimization — transform_json on an empty map is cheap.

P2 — Unnecessary clone of local flow_env — worker_flow.rs:2395-2399

let env = if let Some(ref e) = flow_value.flow_env {
    e.clone()
} else {
    fetch_root_flow_env(db, flow_job_id, workspace_id).await?
};

When the local env is set we clone the entire HashMap<String, Box<RawValue>> even though transform_json only needs &HashMap<…>. A Cow<'_, HashMap<…>> or two passes-by-reference would avoid the copy. Not a hot path, so this is purely a polish item.

AGENTS.md compliance

• backend/CLAUDE.md rules: validation, sqlx cache discipline — followed (3 query JSONs added, 1 removed in latest commit, all consistent with the schema change).
• No banned patterns triggered.

Test coverage

Backend Rust tests in flow_engine_parity.rs already cover the new behavior comprehensively (9 regression tests pinning branches/nested branches/parallel for-loops/QuickJS-fallback transforms/imported flows/stop_after_if/retry_if/stop_after_all_iters_if). The latest commit is a pure refactor of internal lookup logic — no new test surface needed; the existing tests exercise both the pre-root_job and post-root_job walks via the parallel-loop scenario where flow_innermost_root_job resets to NULL.

Manual verification: the test plan in the PR body is appropriate. Worth specifically re-running the parallel for-loop case (flow_innermost_root_job=NULL) end-to-end in the UI, since the CTE's preference change (now starting from root_job) affects that path; the unit test confirms correctness but a manual sanity check of "create a flow with flow_env, wrap a step in a parallel for-loop with skip_if: flow_env.X === true, verify skip fires" covers the integration path.

· branch: flow-env-skip-eval
·

[cubic-dev-ai]

No issues found across 5 files

[github-actions]

Codex Review

Should address issues before merging: imported-flow own env skipped in nested subflows

Findings

[P1] backend/windmill-worker/src/worker_flow.rs:2459 - The ancestor walk now follows root_job before flow_innermost_root_job / parent_job, so a branch or loop sub-flow inside an imported flow skips over the imported-flow job and reads the top-level parent’s flow_env instead. The PR explicitly preserves “imported flow own env wins”, but for parent flow_env.KEY = "parent" importing a flow with its own flow_env.KEY = "imported" and an inner BranchOne/loop predicate, the generated sub-flow job has root_job = top parent and flow_innermost_root_job = imported flow job; this CTE jumps straight to the top parent and returns the wrong env. The walk needs to visit the nearest flow ancestor before the global root, or otherwise carry the imported flow’s env into generated sub-flows.

Test coverage

Backend coverage is strong for root inheritance, branch/loop predicates, retry/stop predicates, and top-level imported flow env precedence. It is missing the nested imported-flow case above: imported flow defines its own flow_env, contains a BranchOne or for-loop sub-flow, and the parent has a conflicting env value.

Manual verification should cover that scenario through the flow runner: run a parent flow that imports a flow with its own flow_env, where the imported flow’s inner BranchOne step has skip_if: flow_env.KEY === "imported". The inner step should use the imported flow’s env and skip even when the parent sets KEY differently.

[github-actions]

Pi Review

Good to merge

The second commit (0ac94de) addressed four of the six suggestions from the prior Claude review:

• ✅ Tightened the needs_flow_env gate — retry_uses_flow_env now checks retry_if.is_some(), not just retry.is_some(), avoiding unnecessary DB queries for retry configs without a condition expression.
• ✅ Added explicit Err(e) match for get_mini_pulled_job failures in resolve_flow_env_for_status_update, with tracing::warn!.
• ✅ The recursive CTE now uses COALESCE(c.root_job, c.flow_innermost_root_job, c.parent_job), matching the priority order in get_root_job_id.
• ✅ Magic number 50 replaced with named constant MAX_FLOW_ENV_LOOKUP_DEPTH and a doc comment.

The remaining two suggestions from Claude (empty-env short-circuit consistency, unnecessary clone) are truly minor and don't rise to even P2 under the review policy — transform_json on an empty map is cheap, and this is not a hot path.

What I verified

• All four None→resolved_flow_env.as_ref() replacements in update_flow_status_after_job_completion_internal are wired correctly to compute_bool_from_expr (line 632), evaluate_stop_after_all_iters_if (lines 917, 1138), and both evaluate_retry call sites (lines 1223, 1607).
• The evaluate_stop_after_all_iters_if signature change adds flow_env: Option<&HashMap<String, Box<RawValue>>> and both call sites pass the parameter.
• handle_flow inherits flow_env via fetch_root_flow_env when flow.flow_env.is_none() and flow_job.parent_job.is_some(), with FlowModuleValue::Flow (imported flows) benefiting automatically — own env wins when set, inherited when absent.
• The recursive CTE uses parameterized queries ($1, $2, $3) — no SQL injection risk.
• resolved_flow_env.as_ref() correctly produces Option<&HashMap<String, Box<RawValue>>> for all downstream callers.
• All 9 regression tests verify the exact bug scenarios from the PR description.

Test coverage

Backend — 9 regression tests added to flow_engine_parity.rs covering: BranchOne skip_if, nested BranchOne, complex input transforms in sub-flows, parallel for-loop iterations, imported flows with own env, imported flows inheriting env, stop_after_if, retry_if, and stop_after_all_iters_if. The tests insert flow/flow_version rows via raw SQL for the imported-flow cases and run complete flow executions. This is thorough coverage for the changed evaluation paths.

Manual verification needed

• Run cargo test --features deno_core from backend/ to confirm the 9 new tests pass (they're gated behind the deno_core feature).
• Run ./update_sqlx.sh from backend/ to ensure the offline query cache includes the new recursive CTE query (the 3 new .sqlx JSON files are for the test-only INSERT queries; the runtime CTE query in fetch_root_flow_env must be cached too).

[github-actions]

Now I have a complete understanding of the code. Let me write the review.

Pi Review

Should address issues before merging: imported flow's flow_env lost for predicates inside nested BranchOne/loop sub-flows

---

P1 — BranchOne/loop predicates inside imported flows see parent's flow_env, not the imported flow's own env

File: backend/windmill-worker/src/worker_flow.rs, lines 2437–2475 (fetch_root_flow_env)

The fetch_root_flow_env function uses COALESCE(current_job.root_job, current_job.flow_innermost_root_job, current_job.parent_job, current_job.id) to resolve the ancestor to read flow_env from. The root_job column is always propagated by push as get_root_job_id(&flow_job), which follows the root_job chain to the ultimate root of the entire execution chain. For a BranchOne sub-flow spawned inside an imported flow, this means root_job jumps to the top-level parent flow, skipping the imported flow's job entirely.

Scenario:

1. Parent flow (id=100, flow_env.KEY="parent") imports a flow at f/system/imported (id=101, flow_env.KEY="imported")
2. The imported flow contains a BranchOne with a step that has skip_if: flow_env.KEY === "imported"
3. When the BranchOne sub-flow (id=102) runs handle_flow, flow.flow_env is None (from payload_from_modules), so it calls fetch_root_flow_env
4. fetch_root_flow_env resolves COALESCE(root_job=100, flow_innermost_root_job=101, ...) → picks 100 → returns {KEY: "parent"}
5. Predicate evaluates "parent" === "imported" → false — the step is not skipped, contradicting the user's intent

The imported flow's own flow_env is preserved when its direct modules are evaluated in handle_flow (line 2630: flow.flow_env.is_none() is false, so no lookup is triggered). But any sub-flow spawned by a BranchOne/for-loop inside the imported flow inherits the parent's env instead.

Root cause: root_job tracks the ultimate root of the execution chain, not the nearest flow-context boundary. The recursive CTE in the prior commit (0ac94de) had the same issue — COALESCE(c.root_job, c.flow_innermost_root_job, c.parent_job) also preferred root_job and would skip the imported flow. The correct walk for flow_env is COALESCE(flow_innermost_root_job, parent_job) — the chain of flow-context jobs, not the root_job shortcut.

This is covered by the test test_flow_env_imported_flow_uses_own_env but only for direct modules in the imported flow (which work correctly). The scenario of a BranchOne/for-loop predicate inside the imported flow is untested. The test test_flow_env_imported_flow_inherits_when_unset covers inheritance only when the imported flow has no flow_env — a simpler case where any ancestor's env is acceptable.

Fix direction: Either (a) revert to a CTE that walks COALESCE(flow_innermost_root_job, parent_job) without involving root_job, returning the first non-null flow_env, or (b) restructure the single-hop query to prefer flow_innermost_root_job and if the resolved job has no flow_env in its raw_flow, fall back to root_job. Option (a) is more robust but requires going back to a CTE. A test for the imported flow + nested BranchOne scenario should be added.

---

P2 — Test gap: imported flow with own flow_env + nested BranchOne/loop predicate

File: backend/tests/flow_engine_parity.rs, around line 2586

The existing imported-flow tests cover:

• test_flow_env_imported_flow_uses_own_env — direct step in imported flow reads flow_env.KEY (works, no sub-flow involved)
• test_flow_env_imported_flow_inherits_when_unset — imported flow without flow_env has a skip_if on a direct module (works, handle_flow for the imported flow job inherits)

Neither test has a BranchOne or for-loop inside the imported flow. The scenario that breaks — imported flow with its own flow_env, containing a BranchOne/for-loop whose sub-flow predicate references flow_env — is missing. This should be added alongside whatever fix is chosen for the P1 above.

---

What I verified

• All four None→resolved_flow_env.as_ref() replacements in update_flow_status_after_job_completion_internal are correctly wired to compute_bool_from_expr (line 635), evaluate_stop_after_all_iters_if (lines 919, 1139), and both evaluate_retry call sites (lines 1222, 1607). ✓
• The evaluate_stop_after_all_iters_if signature adds flow_env: Option<&HashMap<String, Box<RawValue>>> and both call sites pass it. ✓
• handle_flow resolves flow_env via inherited_env when flow.flow_env.is_none() and parent_job.is_some(), wiring it through env_source → transform_json correctly. ✓
• resolve_flow_env_for_status_update guards access with needs_flow_env — the gating on retry_if.is_some() (not just retry.is_some()) is exact per the evaluate_retry body at line 2274. ✓
• retry without retry_if is correctly excluded from triggering the DB query. ✓
• Error logging for get_mini_pulled_job failures is present via tracing::warn!. ✓
• The sqlx offline cache matches the current query (single-hop JOIN with COALESCE(root_job, flow_innermost_root_job, parent_job, id)). ✓
• SQL uses parameterized queries — no injection risk. ✓
• The get_flow_env_by_flow_job_id API endpoint (jobs.rs:455–576) uses the same COALESCE(root_job, flow_innermost_root_job, parent_job, id) pattern, so the same P1 applies there for frontend flow_env.X lookups — however, input transforms inside sub-flows go through handle_flow's resolved flow_env (the main code path here), which is where this PR intervenes. The API endpoint affects only frontend fetches and is out of scope for this PR.

Test coverage

Backend — 9 regression tests added to flow_engine_parity.rs, gated on #[cfg(feature = "deno_core")], covering: simple BranchOne skip_if, nested BranchOne, complex input transforms (QuickJS path), parallel for-loop iterations, imported flow with own env (direct step), imported flow inheriting env (direct skip_if), stop_after_if, retry_if, and stop_after_all_iters_if. Tests use sqlx::test(fixtures("base")) and run complete flow executions via RunJob::run_until_complete.

Not covered: imported flow with its own flow_env + BranchOne/for-loop predicate inside (the P1 described above). Also missing is the equivalent scenario for stop_after_if / stop_after_all_iters_if / retry_if predicates inside a sub-flow of an imported flow.

Manual verification needed

• Run cargo test --features deno_core from backend/ to confirm the 9 new tests pass.
• The PR's test plan is appropriate for the top-level and direct-module cases. For the imported flow + nested sub-flow scenario, manually test: parent flow with flow_env.KEY = "parent" imports a flow with flow_env.KEY = "imported" containing a BranchOne that has an inner step with skip_if: flow_env.KEY === "imported". The step should be skipped. This currently returns the wrong result per the P1 finding.

[github-actions]

Pi Review

Good to merge — the two latest commits (54915e5 and 732e123) addressed all P1 and P2 findings from prior reviews. The code is correct, well-tested, and clean.

What changed across the PR (5 commits)

The PR progressed through three design iterations driven by prior review findings:

1. 25c9fd0 — initial fix with single-hop COALESCE(root_job, flow_innermost_root_job, parent_job) query
2. 0ac94de — addressed Claude's review (tightened needs_flow_env gate, named constant for depth, root_job priority in COALESCE, explicit error logging)
3. ef9d870 — dropped recursive CTE, kept single-hop via root_job propagation
4. 54915e5 — reverted to recursive CTE, now walking COALESCE(flow_innermost_root_job, parent_job) without root_job, to respect imported-flow scope boundaries. Added test_flow_env_imported_flow_with_nested_branch covering the exact scenario the earlier Pi and Codex reviews flagged as P1
5. 732e123 — cleanup: removed the now-unused get_flow_env_by_flow_job_id API endpoint and AuthedClient::get_flow_env_by_flow_job_id method (dead code from the deno_core era), updated jseval regexes and handle_full_regex to drop flow_env from the API fast-path dispatch, and cleaned up a stale test comment

What I verified

• fetch_root_flow_env walk is correct: COALESCE(c.flow_innermost_root_job, c.parent_job) walks one flow-context scope at a time. ORDER BY depth ASC LIMIT 1 returns the closest ancestor with flow_env. For an imported flow with its own flow_env, the imported flow's job is found at a shorter depth than the parent — imported-flow scope is preserved.
• test_flow_env_imported_flow_with_nested_branch (line 2771 of flow_engine_parity.rs): imported flow with {KEY: "imported"}, parent with {KEY: "parent"}, BranchOne inside the imported flow with skip_if: flow_env.KEY === 'imported'. Assertions confirm the imported flow's env is used (marker passed through, leaf skipped).
• All four eval sites in update_flow_status_after_job_completion_internal correctly pass resolved_flow_env.as_ref(): compute_bool_from_expr at line 635, evaluate_stop_after_all_iters_if at lines 919 and 1139, evaluate_retry at lines 1223 and 1607.
• handle_flow inherits via fetch_root_flow_env when flow.flow_env.is_none() && parent_job.is_some(), then threads env_source → resolved_env → flow_env down to all six eval_timeout call sites.
• retry_uses_flow_env gate is exact: evaluate_retry (line 2267) only consults flow_env inside retry.retry_if.is_some() (line 2279). Gating on retry_if.is_some() has no false negatives.
• jseval regex changes: RE and RE_FULL no longer capture flow_env — correct, because QuickJS now receives it as an in-memory object via eval_quickjs_inner (lines 379–391 of lib.rs). try_exact_property_access still handles flow_env for the in-process fast-path. No remaining callers of the removed endpoint.
• Death of the API endpoint: get_flow_env_by_flow_job_id removed from jobs.rs (router registration + handler body, 128 lines) and client.rs (method, 20 lines). Zero remaining references in source.
• sqlx cache: 3 new .json cache files added, 1 removed — consistent with the query changes (the CTE query_scalar! + 2 test INSERT macros, the old single-hop query deleted).

Test coverage

Backend — 10 regression tests in flow_engine_parity.rs covering: BranchOne skip_if, nested BranchOne, parallel for-loop, complex QuickJS-fallback input transforms, imported flows with own env, imported flows with no env (inherit), imported flow with nested BranchOne, stop_after_if, retry_if, and stop_after_all_iters_if. The imported-flow tests insert flow/flow_version rows directly to simulate persisted imported flows and run full flow executions end-to-end.

Manual verification: the test plan in the PR body is appropriate. The critical scenario to verify end-to-end in the UI is the imported flow with nested BranchOne (the scenario uncovered in the prior review round): create a parent flow that imports a flow with its own flow_env, where the imported flow contains a BranchOne with skip_if: flow_env.KEY === 'imported' while the parent sets KEY differently. The imported flow's predicate must use its own flow_env, not the parent's.