Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@internal/runtime/agent/agent.go`:
- Around line 557-560: After creating subWorkflowRunner via
a.createSubWorkflowRunner(ctx), immediately defer its cleanup to avoid resource
leaks (e.g., defer subWorkflowRunner.Close() or the runner's proper cleanup
method) so that any early returns (including the a.dryRun(ctx) path and other
init failures) will still release resources; place the defer directly after the
successful creation of subWorkflowRunner (near the createSubWorkflowRunner call)
rather than waiting until after a.runner.Run(...).

In `@internal/runtime/executor/dag_runner.go`:
- Around line 384-386: The derived contexts created by Execute/Retry (runCtx,
cancel := context.WithCancel(ctx)) are tracked via trackDistributedRun but
clearDistributedCancel only deletes the map entry and never calls cancel,
leaking the derived context; modify clearDistributedCancel (or the defer after
trackDistributedRun) to invoke the stored cancel function before removing it so
that successful/completed runs also cancel their derived context (ensure Stop()
behavior remains unchanged and handle nil-checks when retrieving the cancel from
the internal map).

In `@internal/subflow/runner.go`:
- Around line 143-150: The Retry method currently forwards requests without
ensuring req.StepName is present, allowing buildRetryTask to emit empty-step
retry tasks; update Runner.Retry to reject empty retry step names before
dispatching (e.g., add a check after validate(req.SubWorkflowRequest) that
returns a clear bad-request error if req.StepName == ""), so dispatchRetry and
buildRetryTask are never called with an empty StepName; reference the Retry,
dispatchRetry and buildRetryTask symbols when adding the guard and error
message.
- Around line 108-120: ShouldRun currently returns true for distributed routing
even when Run will later fail in Run.validate; update Runner.ShouldRun to mirror
the same prerequisites checked in Run.validate by also returning false if
req.RunID is empty or req.RootDAGRun is zero. In other words, inside ShouldRun
(method on Runner) keep the existing nil/dispatcher/DAG checks and the
ForceLocal/WorkerSelector/defaultMode logic but add checks for req.RunID == ""
and req.RootDAGRun == 0 so requests that would fail validation in Run (see
Run.validate) are not routed as distributed.

---

Nitpick comments:
In `@internal/runtime/executor/dag_runner.go`:
- Around line 582-661: SubDAGExecutor.Stop currently holds e.mu while calling
external cancellers/stoppers (subWorkflowRunner.Cancel,
dagCtx.DB.RequestChildCancel, process.Stop), which increases lock contention and
can block other goroutines like clearDistributedCancel; fix by under e.mu
snapshotting the maps/slices you need (distributedRuns, distributedCancels,
processes and any runIDs), then release e.mu and iterate over those snapshots to
call subWorkflowRunner.Cancel, dagCtx.DB.RequestChildCancel and process.Stop,
collecting errors as before; keep the e.cancelOnce/close(e.killed) protected as
appropriate and preserve the existing logging and error aggregation.

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@internal/runtime/executor/dag_runner.go`:
- Around line 478-486: clearDistributedCancel currently only removes the cancel
func from distributedCancels leaving the run marked active in distributedRuns so
subsequent Stop() calls still try to cancel finished child workflows; modify
clearDistributedCancel to also remove the runID from distributedRuns under the
same mutex (e.mu.mu.Lock()/Unlock())—i.e., inside the critical section fetch
cancel := e.distributedCancels[runID], delete(e.distributedCancels, runID),
delete(e.distributedRuns, runID), then unlock and call cancel() if non-nil so
the run is no longer considered active.

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@internal/runtime/executor/dag_runner_test.go`:
- Around line 779-807: Add a compile-time interface assertion to ensure
mockSubWorkflowRunner implements SubWorkflowRunner by adding a declaration like
var _ SubWorkflowRunner = (*mockSubWorkflowRunner)(nil) (placed after the
mockSubWorkflowRunner type definition); this will cause a compile-time error if
method signatures drift and matches the pattern used for mockDatabase's
assertion.