feat: DAG-based workflow support

[temporaer]

Summary

Adds DAG-based workflow orchestration to jobq. Users define tasks with dependencies, submit them via the CLI or API, and a coordinator process advances the DAG — dispatching ready tasks through the standard JobQ backend (Storage Queues or Service Bus).

What this enables

• Multi-step pipelines: define tasks with depends_on relationships; the coordinator handles scheduling, fan-out, and join semantics
• Crash recovery: stuck tasks are automatically retried, orphaned children (from coordinator crashes mid-transition) are detected and repaired by periodic sweeper loops
• Output passing: tasks can set_output() and downstream tasks read it via get_upstream_output() — small values inline in Table Storage, large values spilled to blob
• Local development: LocalRunner executes workflows in-process for testing without needing queues or Table Storage
• Diagnostics: workflow doctor checks connectivity, finds stuck workflows, and surfaces stale tasks

Architecture

• Split-blob persistence — immutable definition (topology, kwargs) in {wf_id}.def.bin, mutable state (task states, counters) in {wf_id}.state.bin. Flushes are ~6x smaller than a full-state rewrite.
• Deferred-flush (in-memory snapshot) — the coordinator batches completions, applies them to an in-memory copy, pushes ready tasks immediately, then writes only the state blob once per batch
• Index Table — lightweight {prefix}WorkflowsIndex rows for listings, cancel signalling, and sweeps (not source of truth)
• Coordinator lease — sentinel blob lease enforces single-writer; second coordinator is rejected at startup
• Coordinator-owned retries — retry logic lives in the coordinator (not the queue backend), making it backend-portable (works on both Storage Queues and Service Bus)

Scalability

The coordinator is designed for deployments with tens of thousands of concurrent workflows:

• Bounded-concurrency fan-out — sweeper loops use asyncio.gather with a configurable semaphore instead of sequential iteration
• Shard-based sweep rotation — workflows are distributed across 16 partition-key shards; each sweeper tick processes one shard (full coverage in ~160s)
• Orphan-sweep skip cache — workflows whose updated_at has not moved since the last sweep are skipped entirely
• Admission control — optional --max-running-workflows limit queues excess pending workflows and activates them in FIFO order as slots open

Safety mechanisms

• ETag CAS on every state blob write — concurrent writers never silently overwrite each other
• Queue redelivery — unacked messages reappear after visibility timeout; no work is forgotten
• Idempotent completion application — duplicate completions are no-ops (keyed on task_name + attempt_no)
• Ready-repair sweep — re-pushes tasks stuck in READY state (covers crash between push and flush)

Configuration

# Required: storage account + project prefix (Table Storage for DAG state)
export JOBQ_WORKFLOW_PREFIX=mystorageaccount/MyProject

# Optional: override queue backend (defaults to same account, Storage Queues)
export JOBQ_WORKFLOW_QUEUES=sb://myservicebusns   # or another storage account

# Optional: override blob account/container for large task outputs
export JOBQ_WORKFLOW_BLOBS=myblobaccount/mycontainer

# Optional: cap concurrent running workflows
export JOBQ_MAX_RUNNING_WORKFLOWS=500

Not included

• Stress test examples (separate PR)
• Conditional/loop DAG patterns (future)

[temporaer]

Please make the following changes in this PR:

1. Remove activate_tasks() from the normal submit path

   • WorkflowClient.submit() should no longer call activate_tasks() after submit_workflow().
   • Instead, submit_workflow() should derive initial task state directly from the DAG:
     • tasks with no dependencies should be created as ready
     • tasks with dependencies should be created as pending
   • This removes an unnecessary full workflow-partition scan after submission.

2. Keep root detection fully internal

   • Do not add any user-facing “root” or “phase” concepts.
   • Root/non-root should remain derived entirely from depends_on.

3. Add efficient per-workflow task-prefix filtering

   • Support querying/filtering tasks within a single workflow by task-name prefix.
   • Assume users may categorize tasks with names like:
     • prep/featurize
     • train/rf
     • train/gnn
     • eval/compare
   • Please implement this without adding any secondary index rows or extra index tables.
   • Expose it in a user-visible way, e.g. store helper(s), CLI option(s), or both.

4. Update docs

   • Recommend prefix-based task naming conventions for grouping similar task types within a workflow.
   • Document which per-workflow queries become efficient under that convention.

5. Add/update tests

   • cover initial ready/pending task states at submission time
   • cover removal/refactor of activation behavior
   • cover per-workflow task filtering/querying by prefix

Why I’m asking for this:

• it simplifies submission by eliminating the extra activation scan
• it keeps the current single-source-of-truth row model
• it avoids the complexity and failure modes of mirrored index rows
• it gives users a documented way to structure workflows for efficient per-workflow queries

@copilot

[temporaer]

Fixed in 731b253 — all 5 items addressed:

1. Removed activate_tasks() — submit_workflow() now sets initial task state directly from the DAG (ready for roots, pending for dependents). No more initializing status or separate activation scan. Coordinator startup queries for ready tasks instead.

2. Root detection stays internal — derived entirely from depends_on, no user-facing root/phase concepts.

3. Prefix filtering — list_tasks() accepts name_prefix= using RowKey ge/lt range queries. Exposed via --prefix flag on workflow tasks CLI.

4. Docs updated — new "Task naming conventions" section with prefix examples and efficient query guidance.

5. Tests updated — test_submit_initial_task_states (replaces old two-phase test), test_batch_insert_initial_states, test_list_tasks_by_name_prefix (prefix + status combo filters). All activate_tasks() calls removed from existing tests.

[temporaer]

Here’s a concise comment you can paste into PR #75:

---

This looks great. One capability we’ll need next is conditional downstream execution based on upstream outputs.

For example, we need to express logic like:

• if result(A) > 0.28 → run B
• else → run C

So beyond pure dependency ordering, it would be helpful to support an optional condition on a task (or edge) that the coordinator evaluates once upstream dependencies are complete.

Concretely, something along these lines:

• B depends on A and runs only if A’s output satisfies > 0.28
• C depends on A and runs only if the complementary condition holds
• tasks whose condition evaluates false should move to a terminal skipped state
• skipped should be handled cleanly in downstream dependency resolution and workflow completion

I’d strongly prefer a structured predicate over arbitrary expression evaluation, e.g. a validated condition model rather than raw code/strings, since that will be safer and easier to test/document.

Not asking to block this PR on it, but I’d like to make sure the workflow model can evolve in that direction, since conditional branching is a key use case for us.

---

If you want, I can also give you:

• a shorter/blunter version, or
• a version phrased as requested follow-up work rather than general feedback.

[temporaer]

Good call — conditional branching is a natural next step. The current model should extend cleanly:

• condition field on WorkflowTask — a structured predicate (not raw code), for example {"op": "gt", "upstream": "A", "field": "mae", "value": 0.28}. The coordinator evaluates it after upstream deps complete, using the upstream's output_ref.

• skipped status — tasks whose condition evaluates false move to skipped. For downstream dep resolution, skipped would count as "not failed" (so it doesn't doom children) but "not completed" (so it doesn't satisfy dep_policy). We'd need a skip_policy or treat it like the existing dep_policy logic.

• ETag-safe — condition evaluation happens in the coordinator after increment_child_deps returns ready, so it's a single-writer check before _enqueue_task. No new concurrency concerns.

Will track as a follow-up.

[temporaer]

This is great. One specific suggestion for the condition model: please make conditions evaluate against the outputs of all direct dependencies, exposed through a templated path syntax.

For example, for a task D with depends_on=["A", "B", "prep/featurize"], the condition could reference:

• {inputs.A.score} > 0.28
• {inputs.B.loss} < 0.1
• {inputs["prep/featurize"].passed} == true

and then compose them normally:

• {inputs.A.score} > 0.28 and {inputs.B.loss} < 0.1
• {inputs["train/rf"].status} == "ok" or {inputs["train/gnn"].status} == "ok"

I’d suggest these semantics:

• inputs contains outputs of direct dependencies only, keyed by task name
• dot access for identifier-safe names: {inputs.foo.bar}
• bracket access for task names or keys containing non-word chars: {inputs["prep/featurize"].metrics["f1-score"]}
• conditions are evaluated only once all referenced dependencies are terminal
• if the condition evaluates false, the task transitions to skipped
• skipped should remain a terminal state for downstream dependency resolution / workflow completion

I’d also keep the first version intentionally small and validated:

• comparisons: == != > >= < <=
• boolean ops: and or not
• literals: numbers, strings, booleans, null
• no arbitrary code execution

That would cover the cases we need — including conditions depending on two inputs or a set of inputs — while keeping the model predictable, safe, and easy to document/test.

[temporaer]

Implemented in ad02eb8 — replaced structured TaskCondition with an expression-based condition language.

What changed:

• New condition.py module with safe AST-based expression parsing (whitelist of allowed node types — no function calls, imports, or arbitrary code)
• Conditions are now string expressions: "inputs.train.mae < 0.1"
• Multi-input references: "inputs.A.score > 0.28 and inputs.B.loss < 0.1"
• Bracket access for special chars: "inputs[\"prep/featurize\"].passed"
• Supported: comparisons (== != > >= < <= in not in), boolean ops (and or not), literals, true/false/null aliases
• validate_condition() checks syntax + ensures all inputs.X refs are in depends_on
• evaluate_condition() catches all exceptions → returns False (failing conditions skip the task)
• Coordinator fetches all upstream outputs into an inputs dict for evaluation
• Store column renamed from condition_json to condition (plain string)
• 27 new/updated tests for the expression evaluator and validation
• Docs updated with expression syntax examples

[Copilot]

Pull request overview

Adds a new ai4s.jobq.workflow module that layers DAG-based orchestration on top of the existing JobQ backends, persisting workflow/task state in Azure Table Storage and advancing workflows via a coordinator consuming completion messages.

Changes:

• Introduces workflow core components (entities/definition validation, Table Storage store, coordinator loop, workflow-aware worker processor, local runner, condition evaluator).
• Adds CLI support for ai4s-jobq workflow ... plus push --workflow, along with extensive documentation/tutorials.
• Adds tests for workflow entities/condition evaluation and Table Storage store behavior (Azurite).

Reviewed changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated 12 comments.

Show a summary per file

File	Description
vale-styles/config/vocabularies/JobQ/accept.txt	Adds workflow terminology to Vale accepted vocabulary.
tests/test_workflow_store.py	Adds Azurite-backed integration tests for WorkflowStore.
tests/test_workflow_entities.py	Adds unit tests for DAG validation, serialization, and condition evaluation.
pyproject.toml	Adds [workflow] optional dependency (azure-data-tables).
docs/workflows.md	Adds workflow architecture and user guide documentation.
docs/workflow-tutorial.md	Adds an end-to-end local tutorial using Azurite.
docs/workflow-races.md	Documents race conditions, failure modes, and mitigations.
docs/index.rst	Links new workflow documentation into Sphinx TOC.
ai4s/jobq/workflow/worker.py	Implements WorkflowShellCommandProcessor and workflow context helper for workers/scripts.
ai4s/jobq/workflow/store.py	Implements Table Storage persistence with ETag-guarded updates and query helpers.
ai4s/jobq/workflow/runner.py	Adds LocalRunner for in-process workflow execution/testing.
ai4s/jobq/workflow/entities.py	Defines workflow/task/completion dataclasses and (de)serialization helpers.
ai4s/jobq/workflow/data/workflow-definition.schema.json	Adds JSON Schema for workflow definition files.
ai4s/jobq/workflow/coordinator.py	Adds stateless coordinator loop to process completions and enqueue ready tasks.
ai4s/jobq/workflow/context.py	Adds WorkflowContext plus sync helpers for user scripts (get_upstream_output, etc.).
ai4s/jobq/workflow/condition.py	Adds restricted AST-based condition validator/evaluator with aggregates + globbing.
ai4s/jobq/workflow/client.py	Adds WorkflowClient API for submit/status/list/cancel/purge and summary aggregation.
ai4s/jobq/workflow/cli.py	Adds ai4s-jobq workflow subcommands (submit/status/list/tasks/summary/cancel/purge/run-local/coordinator).
ai4s/jobq/workflow/init.py	Exposes workflow public API surface.
ai4s/jobq/cli.py	Wires workflow commands into the main CLI and adds push --workflow plus auto processor upgrade.

[temporaer]

I think we should follow up on stage 1 + stage 2 coordinator throughput improvements before adding more coordinator parallelism.

Stage 1

• Cache workflow definitions in Coordinator

  • workflow definitions are immutable after submission, so _get_definition(workflow_id) should hit an in-memory cache after first load
  • a simple dict is probably enough to start; LRU is optional
  • this should remove at least one Table read per completion in steady state

• Add ETag retry handling to recount_summary()

  • increment_child_deps() already has a bounded retry loop, but recount_summary() does not
  • if run parallel workflow writes become

• add +off on ResourceModifiedError

• Add instrumentation

  • completions/sec
  • definition cache hit/miss
  • recount_summary() latency
  • ETag conflict counts for child updates and summary updates
  • tasks enqueued per completion

Stage 2

The bigger issue is that the hot path currently does a full recount_summary() after each completion. That makes completion cost scale with workflow size.

Please switch to:

• increment workflow summary updates in the normal completion path
• keep recount_summary() as a repair/reconciliation path
  • startup
  • sweeper
  • periodic repair
  • explicit recovery from counter drift

Important detail: please document and keep consistent whether ready counts as pending or running at workflow-summary level. The current recount logic maps ready into running; if that remains the contract, the incremental path should do the same.

Defensive testing I’d want with this

• definition cache tests:
  • repeated completions for the same workflow only load the definition once
• summary retry tests:
  • inject ResourceModifiedError into recount_summary() and verify retry/backoff works
• incremental summary transition matrix:
  • running -> completed
  • running -> failed
  • retry path back to ready
  • pending -> upstream_failed
  • skipped-task transitions
• reconciliation tests:
  • intentionally corrupt summary counters, then verify recount_summary() repairs them
• race tests:
  • two parents finishing simultaneously for one child
  • wide fan-in
• duplicate completion tests:
  • ensure counters are not double-applied

My reason for pushing on this now: the current implementation is correct, but the hot path is optimized for simplicity, not throughput. These two stages seem like the best next step before scaling out coordinator concurrency.

[temporaer]

Implemented in 489aa4f — Stage 1 + Stage 2 coordinator throughput improvements with defensive tests.

Stage 1: Definition cache, ETag retry on recount_summary(), CoordinatorMetrics instrumentation.

Stage 2: Incremental increment_summary() replaces full recount in hot path. recount_summary() kept as repair path only. ready counts as running at summary level (documented, consistent). Initial counters fixed. Clamping prevents negative drift.

20 new tests: Definition cache hit/miss, incremental summary transition matrix, diamond fan-in, failure propagation, reconciliation (corrupt counters repaired), duplicate completion idempotency, wide fan-in (10→1), initial counter accuracy.