LeetLLM
MediumSchedulingPython 3

Workflow State Machine

Track dependent jobs through legal workflow states with retry and cancellation behavior.

40m3 sample tests7 hidden tests

Implement WorkflowRunner, an in-memory job tracker for dependent work. Jobs move through a small set of states:

text
1pending -> running -> succeeded 2pending -> running -> failed 3running -> retrying -> pending 4pending/running/retrying/failed -> canceled

Requirements

  • add_job(job_id, deps=None, max_attempts=1) registers a job.
  • ready() returns pending jobs whose dependencies have succeeded, in insertion order.
  • transition(job_id, state, reason=None) validates state transitions and returns the final state.
  • Starting a job increments its attempt count.
  • If a running job fails and still has attempts left, move it to retrying; otherwise move it to failed.
  • status(job_id) returns the current state, attempts, and history.

Example

python
1runner = WorkflowRunner() 2runner.add_job("extract") 3runner.add_job("train", deps=["extract"], max_attempts=2) 4 5assert runner.ready() == ["extract"] 6runner.transition("extract", "running") 7runner.transition("extract", "succeeded") 8assert runner.ready() == ["train"]

Constraints

  • Keep it single-process and in memory.
  • Use deterministic ordering.
  • Raise ValueError for invalid transitions or unknown jobs.

Editor