Job events and HTTP recovery

Application WebSocket provides live Job snapshots. HTTP is the durable state source: after event loss, keep querying the same application Job id. This is the P1 API flow; P2 delivers the formal Web Monitor.

Observe Morning song

Start the independent API from the repository root. Use a separate fake data directory. It produces identified CPU fixtures, not real inference performance or music quality.

uv sync --project services/api --locked --python 3.12.13
$env:MUSIC_API_RUNTIME_MODE = "fake"
uv run --project services/api --no-sync music-api serve --data-dir data/application-events-fake --port 8000

Create a Morning song application Job through the transcription guide or generation guide. Retain its Project and Job ids. In another terminal, replace the placeholders and use a new output file:

uv run --project services/api --no-sync python services/api/examples/watch_job.py --project-id <PROJECT_ID> --job-id <JOB_ID> --output data/job-events/first.json

The example prints received status/phase/progress and saves the original domain events plus final HTTP Job. If the Job already finished, its first snapshot is the durable terminal state. The example reads the same Job without creating, retrying or cancelling work or modifying Assets.

Complete source:

"""Watch one application Job over WS, then recover its persistent state by HTTP."""

import argparse
from datetime import datetime, timezone
import json
import math
from pathlib import Path
import time
from urllib.parse import urlsplit, urlunsplit
from uuid import UUID

import httpx
from websockets.exceptions import ConnectionClosed
from websockets.sync.client import connect


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", default="http://127.0.0.1:8000")
    parser.add_argument("--project-id", required=True, type=UUID)
    parser.add_argument("--job-id", required=True, type=UUID)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--disconnect-after-first", action="store_true", help="Close WS after its initial snapshot and prove HTTP recovery")
    parser.add_argument("--wait-seconds", type=float, default=180)
    arguments = parser.parse_args()
    if arguments.output.exists() or not math.isfinite(arguments.wait_seconds) or arguments.wait_seconds <= 0:
        parser.error("Use a new output file and a positive wait window")
    address = urlsplit(arguments.url)
    if address.scheme != "http" or address.hostname not in {"127.0.0.1", "localhost", "::1"} or address.username or address.password:
        parser.error("Use the configured local HTTP application address")
    path = f"/projects/{arguments.project_id}/jobs/{arguments.job_id}"
    websocket_url = urlunsplit(("ws", address.netloc, path + "/events", "", ""))
    began = datetime.now(timezone.utc)
    deadline = time.monotonic() + arguments.wait_seconds
    messages, websocket_outcome = [], "initializing"
    try:
        with connect(websocket_url, proxy=None, open_timeout=5, close_timeout=1, max_size=1024 * 1024) as websocket:
            while True:
                value = json.loads(websocket.recv(timeout=max(0.01, deadline - time.monotonic())))
                messages.append(value)
                job = value["job"]
                print(json.dumps({"status": job["status"], "phase": job["phase"], "progress": job["progress"]}))
                if arguments.disconnect_after_first:
                    websocket_outcome = "deliberately disconnected after initial snapshot"
                    break
                if job["status"] in {"completed", "failed", "cancelled"}:
                    websocket_outcome = "terminal snapshot received"
                    break
    except (ConnectionClosed, TimeoutError, OSError):
        websocket_outcome = "subscription unavailable; HTTP recovery required"
    with httpx.Client(base_url=arguments.url, timeout=10, trust_env=False) as client:
        while True:
            response = client.get(path)
            response.raise_for_status()
            recovered = response.json()
            if recovered["status"] in {"completed", "failed", "cancelled"}:
                break
            if time.monotonic() >= deadline:
                break
            time.sleep(min(0.2, max(0, deadline - time.monotonic())))
    arguments.output.parent.mkdir(parents=True, exist_ok=True)
    with arguments.output.open("x", encoding="utf-8") as output:
        json.dump({"started_at": began.isoformat(), "finished_at": datetime.now(timezone.utc).isoformat(),
                   "websocket_outcome": websocket_outcome, "events": messages, "http_job": recovered}, output, indent=2)
        output.write("\n")
    print("Saved domain events and HTTP recovery. Keep this same application Job id if it still needs confirmation.")


if __name__ == "__main__":
    main()

Complete example: services/api/examples/watch_job.py ↗

Use the event contract

Connect WS /projects/{project_id}/jobs/{job_id}/events. Each JSON contains type="job.updated", a live sequence, and job. The Job shares the Pydantic JobRead returned by GET /projects/{project_id}/jobs/{job_id}: application ids, five states, domain phase, progress, results and recovery errors. Runtime prompt/node ids and raw captions stay out of messages. A wrong Project or missing Job reference rejects the connection handshake with HTTP 403; querying the same HTTP Job address returns 404. Check the Project and Job ids before reconnecting.

cancel_requested and recovery_required also come from the same durable Job. A queued/running cancellation intent still needs confirmation. After a lost intent-commit acknowledgement, events can report the persisted intent and recovery flag. Read the same Job, then explicitly recover through the cancel/retry guide. Only confirmed cancellation sends a cancelled terminal snapshot.

The connection subscribes before reading a fresh persisted snapshot. Buffered older events cannot replace a newer snapshot with queued. Sequence orders the current live channel; it is not a permanent event cursor. Reconnect starts with a new durable snapshot and resets old live sequence tracking. HTTP recovers missed terminal state. Slow clients may skip intermediate snapshots; terminal recovery does not depend on replay.

Only sourced phases are shown. Pinned plugin captions can confirm loading_model, transcribing, planning_score, generating_semantic, synthesizing or decoding_audio; unconfirmed phase stays null. Phases have no fixed percentages and need not be monotonic. ComfyUI node bars contain heuristic shares and token ceilings, so current Transcribe/Generate never convert those bars into Job fractions. Only reliable finite 0–1 readings for the whole Job can become progress; otherwise it stays null.

After confirming native completion, the application durably records running/saving, then validates, imports and associates the whole output set. Completed follows those steps. Cancelled/failed/completed states remain stable. Duplicate, out-of-order or late native hints cannot directly complete a Job or repeat import. Native WS loss still permits owned HTTP queue/history queries.

Reproduce disconnect recovery

Use a new output file for the same Job and deliberately disconnect after its initial snapshot:

uv run --project services/api --no-sync python services/api/examples/watch_job.py --project-id <PROJECT_ID> --job-id <JOB_ID> --disconnect-after-first --output data/job-events/disconnected.json

Inspect websocket_outcome and http_job. If work remains active, continue GET using the same Job id. A timeout ends this observation window and does not authorize automatic resubmission. A failed Job retains error/recovery instructions; incomplete outputs cannot become a successful result. Download original files through application Asset ids in the completed result.

Each WebSocket observes one Job in its owning Project and closes normally after its terminal snapshot. Client disconnect does not cancel inference. A separate recovery issue delivers active-job reconciliation after API restart; this slice does not promise resumed active execution. Only the GPU owner validates real Runtime with fresh source receipts and a separate data directory. See the Job event verification record.

Source version: 1ce3268Page source: docs/guides/api-job-events.en.md