Problem

When DAGU_DEFAULT_EXECUTION_MODE=distributed, DAG parameters (e.g., content_hash) passed via the enqueue API were silently lost. The DAG would run on workers with no runtime params, causing failures for DAGs that require them.

Root Cause

In DAGExecutor.ExecuteDAG(), the distributed execution path (when shouldUseDistributedExecution(dag) is true) built task options with WithWorkerSelector, WithPreviousStatus, and WithBaseConfig — but never called WithTaskParams(). This meant task.Params was empty when dispatched to the coordinator.

Meanwhile, the local execution path correctly extracted params from previousStatus.ParamsList via spec.QuoteRuntimeParams() and injected them into the subprocess env.

Fix

Added a WithTaskParams(previousStatus.Params) call in the distributed branch of ExecuteDAG(), before CreateTask(). This mirrors how the API's dispatchStartToCoordinator correctly passes params, ensuring the worker's loadDAG() receives them via spec.WithParams(task.Params).

Tests

All related tests pass: internal/service/scheduler, internal/service/worker, internal/dispatch.

Summary by cubic

Fixes dropped DAG runtime params in distributed queue dispatch. We now append executor.WithTaskParams(previousStatus.Params) to task options before CreateTask(), so workers receive params (e.g., content_hash) when DAGU_DEFAULT_EXECUTION_MODE=distributed.

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

Summary by CodeRabbit

• Bug Fixes
  • Fixed task parameter propagation in distributed execution scenarios, ensuring parameters are correctly preserved and passed through execution workflows when applicable.