api: add per-run workerSelector to start and enqueue

Merged
YO
6d ago
+1619-583

What

A caller can route one run to a worker set without editing the DAG.

  • POST /dags/{fileName}/start, POST /dags/{fileName}/enqueue, POST /dag-runs, and POST /dag-runs/enqueue take workerSelector (object of string → string).
  • dagu start and dagu enqueue take --worker-label key=value (repeatable), the flag dagu exec already had; the remote CLI forwards it.
  • The selector replaces the DAG's worker_selector for that run, including its "local" value, and follows the same rules as the YAML key: label-key and label-value charset, non-empty values, and no DAGs with approval steps. Invalid input is a 400 (or a CLI error) before anything is created.
  • The selector is kept in the run's DAG snapshot, so queued dispatch, a retry, and a reschedule from the snapshot route the way the run was routed. A reschedule from the current DAG file uses the file, as before.

Why

Operators scale worker pools by label and sometimes need one run to go to a different pool than its DAG names (a hot standby, a machine with the right device attached, a test of a new pool) without editing and re-syncing the DAG. Until now only the embedded engine API and dagu exec could do that.

Also fixed

The coordinator built the attempt's DAG snapshot from the task's YAML and dropped the task's own worker_selector. Runs dispatched directly (the frontend's distributed start, the embedded engine's WithWorkerSelector) were routed correctly once, but a retry or a queued re-dispatch of the same attempt read the YAML's selector instead. The snapshot now records the task's selector.

How it travels

Start and enqueue run dagu start / dagu enqueue as a subprocess that reloads the DAG from its file, so the override goes along as --worker-label flags, the way the queue override goes along as --queue. The subprocess applies it to the loaded DAG before intake persists the snapshot.

Tests

  • internal/ir: ApplyWorkerSelector replaces and clears ForceLocal, keeps its own copy, refuses approval DAGs and invalid labels; FormatWorkerSelector.
  • internal/cmd: --worker-label on enqueue and start lands in the persisted snapshot; invalid pairs refused; refused together with --worker-id.
  • internal/launcher: sorted, deterministic --worker-label arguments.
  • internal/service/frontend/api/v1: all four endpoints persist the selector; invalid and approval cases return 400.
  • internal/service/coordinator: root and sub attempts persist the task selector.
  • internal/intg/distr: a run enqueued with zone=b lands on worker-b, not the DAG's zone=a worker; with no matching worker it stays queued with WorkerReady=False / NoMatchingWorker until one registers.

Docs: skills/dagu/references/cli.md, steptypes.md, MCP reference, regenerated llms.txt, api.gen.go, and ui/src/api/v1/schema.ts.

Notes for reviewers

  • No proto change: Task.worker_selector already existed.
  • Retry and reschedule endpoints take no new parameter on purpose; they reuse the snapshot.
  • Two Windows-only test failures in dagruns_artifacts_internal_test.go are pre-existing (symlink creation needs a privilege the account lacks) and unrelated; a separate small PR makes them skip cleanly.

Summary by cubic

Adds a per-run workerSelector to start and enqueue so a caller can route one run to a different worker set than its DAG names, without editing the DAG file.

  • POST /dags/{fileName}/start, POST /dags/{fileName}/enqueue, POST /dag-runs, and POST /dag-runs/enqueue accept workerSelector; dagu start and dagu enqueue accept --worker-label (repeatable), which remote invocations forward.
  • The selector replaces the DAG's worker_selector for that run — including its "local" value — validated under the same rules as the YAML field; invalid input is a 400 (or a CLI error) before anything is created. An explicitly empty --worker-label is refused, and it can't combine with --worker-id.
  • workerSelector is stored in the run's snapshot, so queued dispatch, retries, and reschedules keep the run routed as it was; a reschedule from the current DAG file still uses the file.
  • Runtime-param resolution rebuilds the DAG from its YAML, so dispatch, selected-step seeding, and reschedule-from-snapshot now reapply the selector to the resolved DAG. dagu exec with labels now requires queues enabled.
  • Fixes the coordinator dropping a task's own worker_selector when building the attempt snapshot from the task YAML, so direct dispatch and embedded WithWorkerSelector runs were routed correctly only once and fell back to the YAML on retry or re-dispatch.

Summary by CodeRabbit

  • New Features
    • Added per-run worker selection using repeatable --worker-label options in dagu start and dagu enqueue, or the workerSelector field in API requests.
    • Overrides replace the DAG’s worker selector and are retained for retries, reschedules, and queued dispatch.
  • Validation
    • Invalid labels are rejected. Per-run overrides aren’t supported for DAGs with approval steps, and --worker-label can’t be combined with --worker-id.