Central Scheduler History Reporter

Create a CLI utility that exercises a workflow against a central scheduler with history persistence and reports the recorded task history as JSON.

Capabilities

Run workflow against central scheduler

• Starting the CLI spins up a central scheduler with history storage enabled, launches a tiny two-step workflow through a remote worker, waits for completion, and ensures the run is recorded in scheduler history. @test
• A --port option (or env override) directs the scheduler to listen on the requested port so concurrent runs do not clash. @test

Emit history summary JSON

• After the workflow finishes, the CLI queries the scheduler's history endpoint for the latest tasks and writes a JSON array to stdout with each task's name/id, status, start time, end time, and runtime in seconds, ordered by most recent first. Exit code is 0 on success. @test
• If the scheduler cannot be reached, the CLI exits non-zero and prints a single-line error message to stderr without emitting JSON. @test

Implementation

@generates

API

from typing import Sequence

def main(argv: Sequence[str] | None = None) -> int:
    """
    CLI entrypoint that starts a central scheduler with history storage, runs a small dependent workflow via the scheduler, fetches task history, and prints a JSON summary. Returns 0 on success and non-zero on failure.
    """

Dependencies { .dependencies }

luigi { .dependency }

Provides workflow scheduling, central scheduler UI, and task history persistence/APIs.