# Heartbeat · Jobs

> The three-tier job substrate — how heavy work runs without blocking a page, survives restarts, and grades its own outcomes.

---

.. toc::

## Rows, not threads

The founding rule of the substrate: *long work runs in another process,
observed by reading rows — never thread-locals.* A job is a row with a
status machine:

```
pending → running → done | error | cancelled
             ↘ cancelling ↗
```

Three tiers:

1. **Tier 1** — a thin callback (a "Run now" button) writes one
   `pending` row and returns the job id instantly. No heavy work in the
   web request.
2. **Tier 2** — the dispatcher (in the pulse worker) atomically claims
   the oldest pending row (`SELECT … FOR UPDATE SKIP LOCKED` on
   Postgres) and runs its registered body, at most three concurrently.
3. **Tier 3** — the body itself, wrapped in `hb.task(...)` so its
   lifecycle lands on the heartbeat, beating progress rows as it goes.

Because state lives in rows, a restart is **recoverable**: boot-time
orphan recovery fails anything a dead process left `running`, and a
periodic reaper closes rows whose worker went silent.

## Outcome grading

A finished job gets a severity grade from its status, its `reason`, and
its result shape — rendered identically in the cockpit, on
`/activity/<job id>`, and in the recent feed:

.. exec::docs.heartbeat.jobs_examples

## Cancellation is cooperative and cross-process

A cancel request flips the row to `cancelling`; the running body
observes it at its own boundaries and finishes as `cancelled`. The
request travels through the store — never an in-memory flag — so a
click in the web container stops work in the worker container.


```python
# File: lib/pulse/jobs.py

"""The heavy-work job substrate — JSONL-first, PG-mirrored.

Ported from the SailsBaord heartbeat substrate (services/jobs.py). The
pattern: *long work running in another process, observed from the web
container by reading rows, not thread-locals.*

A user/owner action that takes more than a moment does **not** block the
page. The three tiers:

  * **Tier-1** — a thin Dash callback validates input, calls
    :func:`create_job` (one ``pending`` row), and returns the ``job_id``
    immediately. No heavy work in the callback.
  * **Tier-2** — the dispatcher in the pulse-worker
    (``lib/pulse/worker``) :func:`claim_next` s the oldest ``pending``
    row and runs Tier-3. The web container never runs job bodies.
  * **Tier-3** — the work, registered per-kind via
    :func:`register_runner` (filled by ``lib.pulse.network_jobs``),
    wrapped in ``hb.task(...)`` so activity flows to the heartbeat.

Because state lives in rows (not an in-process set), a web **or** worker
restart is recoverable — see :func:`recover_orphans`.

Depends on ``$PULSE_DATA_DIR/jobs/`` JSONL files (source of truth,
survives a DB outage) and ``lib.db`` (optional Postgres mirror). When
``DATABASE_URL`` is unset everything runs off JSONL and the PG paths
no-op — dev and prod share one code path. **No Dash imports.**
"""
from __future__ import annotations

import json
import logging
import os
import threading
import time
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable

logger = logging.getLogger(__name__)

PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent


def _data_root() -> Path:
    return Path(os.getenv("PULSE_DATA_DIR") or (PROJECT_ROOT / "data"))


JOBS_DIR = _data_root() / "jobs"
JOBS_PATH = JOBS_DIR / "jobs.jsonl"
PROGRESS_PATH = JOBS_DIR / "progress.jsonl"

# Terminal states never transition. ``cancelling`` is the transient
# "a cancel was requested; the body will observe it and stop" — it
# lives here, deliberately NOT on the heartbeat phase enum (which is
# idle|working|error). Same vocabulary, separate scope.
ACTIVE_STATES = ("pending", "running", "cancelling")
TERMINAL_STATES = ("done", "error", "cancelled")

# Kinds whose row is created + lifecycled by an in-web-process worker,
# NOT by the dispatcher. Empty on the hub today — every network job is
# dispatcher-run. The mechanism stays so a future UI-driven batch can
# opt out of dispatch without re-porting the substrate.
UI_OWNED_KINDS: frozenset[str] = frozenset()

_write_lock = threading.Lock()

# kind → Tier-3 runner. Filled by lib.pulse.network_jobs.register_runners;
# empty here so the data layer ships inert.
_RUNNERS: dict[str, Callable[[str, dict], None]] = {}


def register_runner(kind: str, fn: Callable[[str, dict], None]) -> None:
    """Register the Tier-3 body for ``kind``.

    ``fn(job_id, params)`` runs inside the dispatcher (pulse-worker),
    is expected to wrap its work in ``hb.task(...)`` and call
    :func:`progress` / :func:`finish_job`. Re-registering replaces.
    """
    _RUNNERS[kind] = fn


def runner_for(kind: str) -> Callable[[str, dict], None] | None:
    return _RUNNERS.get(kind)


# ---------------------------------------------------------------------
# Storage helpers
# ---------------------------------------------------------------------


def _ensure_dir() -> None:
    JOBS_DIR.mkdir(parents=True, exist_ok=True)


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _now_iso() -> str:
    return _now().isoformat()


def _pg_on() -> bool:
    try:
        from lib.db import is_configured

        return bool(is_configured())
    except Exception:
        return False


def _append_jsonl(path: Path, record: dict) -> None:
    """Append one line. JSONL is the source of truth — it runs
    unconditionally and first, exactly like ``heartbeat._append_event``.
    """
    _ensure_dir()
    line = json.dumps(record, separators=(",", ":"), default=str) + "\n"
    with _write_lock:
        with path.open("a", encoding="utf-8") as fh:
            fh.write(line)


def _read_jsonl(path: Path) -> list[dict]:
    if not path.exists():
        return []
    out: list[dict] = []
    with path.open("r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                out.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return out


def _jsonl_state() -> dict[str, dict]:
    """Reconstruct latest-per-id from the append-only log.

    Low-volume by nature, so a full scan is fine — and it's only the
    fallback path; PG-configured deploys never hit it for reads.
    """
    state: dict[str, dict] = {}
    for rec in _read_jsonl(JOBS_PATH):
        jid = rec.get("id")
        if jid:
            state[jid] = rec
    return state


# ---------------------------------------------------------------------
# Single-job-per-target dedupe
#
# A double-submit (duplicate callback registration, impatient
# double-click, retry) must not spawn two rows for one action. The
# identity of a job is its target: two active jobs for the same
# (kind, params["slug"]) are the same job. Kinds without a slug param
# dedupe per-kind via an explicit dedupe key of None → no dedupe.
# ---------------------------------------------------------------------


def _job_dedupe_key(j: dict) -> str | None:
    slug = (j.get("params") or {}).get("slug")
    kind = j.get("kind")
    if not slug or not kind:
        return None
    return f"{kind}:{slug}"


def _active_jobs() -> list[dict]:
    """Every job currently in an ACTIVE state, store-of-record first."""
    rows: list[dict]
    if _pg_on():
        try:
            from lib.db import repos

            rows = repos.jobs_for_actor(clerk_id=None, limit=200)
        except Exception:
            logger.debug("_active_jobs PG failed; JSONL", exc_info=True)
            rows = list(_jsonl_state().values())
    else:
        rows = list(_jsonl_state().values())
    return [r for r in rows if r.get("status") in ACTIVE_STATES]


def find_active_peer(
    kind: str | None, slug: str | None, *, exclude_id: str | None = None,
) -> dict | None:
    """The earliest still-active job targeting the same (kind, slug).

    ``exclude_id`` drops the asking job from the set. Earliest by
    ``(created_at, id)`` so every caller independently agrees on which
    single job is the winner without a lock.
    """
    if not kind or not slug:
        return None
    key = f"{kind}:{slug}"
    peers = [
        j for j in _active_jobs()
        if _job_dedupe_key(j) == key and j.get("id") != exclude_id
    ]
    if not peers:
        return None
    peers.sort(key=lambda j: (j.get("created_at") or "", j.get("id") or ""))
    return peers[0]


def superseded_by(job_id: str) -> str | None:
    """If another active job targets this job's (kind, slug) and was
    created first, return that winner's id — this job should yield.
    """
    cur = get_job(job_id)
    if cur is None:
        return None
    kind = cur.get("kind")
    slug = (cur.get("params") or {}).get("slug")
    winner = find_active_peer(kind, slug)  # includes self
    if winner is not None and winner.get("id") != job_id:
        return winner.get("id")
    return None


# ---------------------------------------------------------------------
# Write path  (JSONL always; PG best-effort)
# ---------------------------------------------------------------------


def _persist(job: dict) -> None:
    """JSONL append (source of truth) + best-effort PG upsert."""
    _append_jsonl(JOBS_PATH, job)
    if not _pg_on():
        return
    try:
        from lib.db import repos

        if repos.get_job_row(job["id"]) is None:
            repos.insert_job(job)
        else:
            repos.update_job(job["id"], **{
                k: v for k, v in job.items() if k != "id"
            })
    except Exception:
        # A PG glitch must never break the job write path — the JSONL
        # line already landed and the dispatcher/readers fall back to
        # it. Mirrors the heartbeat's dual-write posture.
        logger.debug("job PG mirror failed (id=%s)", job.get("id"),
                     exc_info=True)


def create_job(
    *,
    kind: str,
    actor: str = "system",
    actor_clerk_id: str | None = None,
    actor_handle: str | None = None,
    title: str | None = None,
    params: dict[str, Any] | None = None,
    dedupe: bool = False,
) -> str:
    """Create a ``pending`` job and return its id immediately.

    The id is an app-generated uuid so the Tier-1 callback can return
    it and start an indicator even before the best-effort PG insert
    lands.

    ``dedupe=True`` collapses a re-submit: if an active job already
    targets the same ``(kind, params["slug"])`` its id is returned and
    **no second row is written**.
    """
    if dedupe:
        slug = (params or {}).get("slug")
        existing = find_active_peer(kind, slug)
        if existing is not None:
            logger.info(
                "create_job deduped: active %s already targets %s:%s "
                "(returning %s, no new row)",
                existing.get("id"), kind, slug, existing.get("id"),
            )
            return existing["id"]

    job_id = uuid.uuid4().hex
    job = {
        "id": job_id,
        "kind": kind,
        "status": "pending",
        "actor": actor or "system",
        "actor_clerk_id": actor_clerk_id,
        "actor_handle": actor_handle,
        "title": title or kind.replace("_", " ").title(),
        "created_at": _now_iso(),
        "started_at": None,
        "ended_at": None,
        "task_id": None,
        "cost_usd": None,
        "progress_pct": 0,
        "message": "Queued",
        "reason": None,
        "params": params or {},
        "result": None,
        "extras": None,
    }
    _persist(job)
    logger.info("job created id=%s kind=%s actor=%s", job_id, kind, actor)
    return job_id


def _patch(job_id: str, **fields: Any) -> dict | None:
    cur = get_job(job_id)
    if cur is None:
        return None
    cur.update(fields)
    _persist(cur)
    return cur


def start_job(job_id: str, *, task_id: str | None = None) -> dict | None:
    """pending → running. Called by the dispatcher right before Tier-3.
    (``claim_next`` already does this atomically in PG; this is the
    JSONL-dev equivalent / a no-op resync.)"""
    return _patch(
        job_id, status="running", started_at=_now_iso(),
        task_id=task_id, message="Running", progress_pct=5,
    )


def progress(
    job_id: str, *, pct: int | None = None, message: str | None = None,
) -> None:
    """Append a progress beat + denormalise latest onto the job row."""
    _append_jsonl(PROGRESS_PATH, {
        "job_id": job_id, "ts": _now_iso(),
        "pct": pct, "message": message,
    })
    if _pg_on():
        try:
            from lib.db import repos

            repos.append_job_progress(
                job_id, pct=pct, message=message, ts=_now_iso(),
            )
        except Exception:
            logger.debug("job progress PG mirror failed", exc_info=True)
    # Stamp a beat timestamp on EVERY progress call — the staleness reaper
    # measures from max(progress_at, started_at, created_at), so a job that
    # beats keeps itself alive.
    fields: dict[str, Any] = {"progress_at": _now_iso()}
    if pct is not None:
        fields["progress_pct"] = pct
    if message:
        fields["message"] = message[:256]
    _patch(job_id, **fields)


def finish_job(
    job_id: str,
    *,
    status: str = "done",
    cost_usd: float | None = None,
    result: dict | None = None,
    reason: str | None = None,
) -> dict | None:
    """running → done | error. Denormalises final cost onto the row."""
    if status not in ("done", "error", "cancelled"):
        status = "done"
    return _patch(
        job_id,
        status=status,
        ended_at=_now_iso(),
        cost_usd=cost_usd,
        result=result,
        reason=reason,
        progress_pct=100 if status == "done" else None,
        message={"done": "Complete", "error": "Failed",
                 "cancelled": "Cancelled"}.get(status, status),
    )


def request_cancel(job_id: str) -> bool:
    """Cancel a job. Returns True if the request was accepted.

    Two cases, because a job cancelled *before* it starts has no body
    to cooperatively stop:

      * ``pending``  → straight to ``cancelled`` (terminal). It never
        ran; just drop it. ``claim_next`` will skip it (not pending).
      * ``running``  → ``cancelling`` (transient). The Tier-3 body
        observes :func:`should_cancel` at its boundaries and finishes
        via :func:`mark_cancelled`.

    Cross-process by design: the cancel is requested in the web
    container and observed in the worker, so it MUST travel through
    the store (PG row / JSONL line), never an in-memory Event.
    """
    cur = get_job(job_id)
    if cur is None:
        return False
    state = cur.get("status")
    if state == "pending":
        mark_cancelled(job_id, reason="cancelled before start")
        return True
    if state == "running":
        _patch(job_id, status="cancelling", message="Cancelling…")
        return True
    return False


def should_cancel(job_id: str) -> bool:
    cur = get_job(job_id)
    return bool(cur and cur.get("status") == "cancelling")


def mark_cancelled(job_id: str, *, reason: str | None = None) -> dict | None:
    return finish_job(
        job_id, status="cancelled", reason=reason or "cancelled by user",
    )


def attach_task(job_id: str, task_id: str | None) -> dict | None:
    """Stamp the heartbeat ``task_id`` onto the job row — the join key
    between a job and its heartbeat_events."""
    if not task_id:
        return None
    return _patch(job_id, task_id=task_id)


# ---------------------------------------------------------------------
# Dispatcher claim + crash recovery  (Tier-2, called from the worker)
# ---------------------------------------------------------------------


def claim_next(*, kinds: list[str] | None = None) -> dict | None:
    """Atomically claim the oldest ``pending`` job → ``running``.

    PG path uses ``SELECT … FOR UPDATE SKIP LOCKED`` so a second
    worker (or two dispatcher ticks) never double-runs a job. JSONL
    dev path is single-process and serialised by ``_write_lock``.
    """
    if _pg_on():
        try:
            from lib.db import repos

            # Allowlist derived from the registered runners minus the
            # UI-owned set (empty on the hub today).
            allow = kinds
            if allow is None:
                allow = [k for k in _RUNNERS.keys() if k not in UI_OWNED_KINDS]
            claimed = repos.claim_pending_job(kinds=allow)
            if claimed is not None:
                # Mirror the claim into JSONL so the audit log + the
                # JSONL fallback agree with PG.
                _append_jsonl(JOBS_PATH, claimed)
            return claimed
        except Exception:
            logger.exception("claim_next PG failed; trying JSONL")
    with _write_lock:
        state = _jsonl_state()
        pend = sorted(
            (j for j in state.values()
             if j.get("status") == "pending"
             and (j.get("kind") not in UI_OWNED_KINDS)
             and (not kinds or j.get("kind") in kinds)),
            key=lambda j: j.get("created_at") or "",
        )
        if not pend:
            return None
        job = pend[0]
        job["status"] = "running"
        job["started_at"] = _now_iso()
        with JOBS_PATH.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(job, separators=(",", ":"),
                                default=str) + "\n")
        return job


def recover_orphans(reason: str = "orphaned by restart") -> list[dict]:
    """On worker boot: any job stuck ``running``/``cancelling`` from a
    dead process is failed. Jobs are rows; restart-safe.
    """
    if _pg_on():
        try:
            from lib.db import repos

            recovered = repos.recover_orphan_jobs(reason=reason)
            for j in recovered:
                _append_jsonl(JOBS_PATH, j)
            return recovered
        except Exception:
            logger.exception("recover_orphans PG failed; trying JSONL")
    recovered = []
    for j in _jsonl_state().values():
        if j.get("status") in ("running", "cancelling"):
            j = dict(j)
            j["status"] = "error"
            j["reason"] = reason
            j["ended_at"] = _now_iso()
            _append_jsonl(JOBS_PATH, j)
            recovered.append(j)
    return recovered


# Kinds whose Tier-3 body runs IN the web process and beats progress
# frequently — a long silence means a dead worker, so the reaper may
# close a stale ``running`` row of these kinds. Empty on the hub today;
# dispatcher-run network jobs finish in seconds and are covered by the
# stuck-``cancelling`` rule.
REAPABLE_RUNNING_KINDS: tuple[str, ...] = ()

# How long a non-terminal row may sit without a progress beat before the
# periodic reaper force-finalizes it. Override via env.
REAP_AFTER_S = float(os.environ.get("JOBS_REAP_AFTER_S", "300"))


def reap_stale_jobs(older_than_s: float = REAP_AFTER_S) -> list[dict]:
    """Periodic safety net: close rows whose worker is gone but whose
    row never reached a terminal state (``recover_orphans`` only runs on
    dispatcher BOOT). Targeting is narrow: stuck ``cancelling`` of any
    kind, plus ``running`` of :data:`REAPABLE_RUNNING_KINDS`.
    """
    reason = f"reaped: no worker progress for over {int(older_than_s)}s"
    if _pg_on():
        try:
            from lib.db import repos

            reaped = repos.reap_stale_jobs(
                older_than_s=older_than_s,
                running_kinds=REAPABLE_RUNNING_KINDS,
                reason=reason,
            )
            for j in reaped:
                _append_jsonl(JOBS_PATH, j)
            return reaped
        except Exception:
            logger.exception("reap_stale_jobs PG failed; trying JSONL")
    # JSONL fallback (dev / no DATABASE_URL): staleness from the LATEST
    # signal of life — a progress beat first, then job start, then creation.
    cutoff = datetime.now(timezone.utc) - timedelta(seconds=float(older_than_s))
    reaped = []
    for j in _jsonl_state().values():
        st = j.get("status")
        reapable = st == "cancelling" or (
            st == "running" and j.get("kind") in REAPABLE_RUNNING_KINDS
        )
        if not reapable:
            continue
        ts_raw = (j.get("progress_at") or j.get("started_at")
                  or j.get("created_at"))
        try:
            ts = datetime.fromisoformat(ts_raw) if ts_raw else None
        except (TypeError, ValueError):
            ts = None
        if ts is None:
            continue
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=timezone.utc)
        if ts < cutoff:
            j = dict(j)
            j["status"] = "cancelled"
            j["reason"] = reason
            j["ended_at"] = _now_iso()
            _append_jsonl(JOBS_PATH, j)
            reaped.append(j)
    return reaped


# ---------------------------------------------------------------------
# Tier-2 dispatcher  (runs in the pulse-worker — never the web box)
#
# Ownership mirrors the heartbeat pulse exactly:
# ``HEARTBEAT_PULSE_WORKER=1`` → defer to the sidecar (no-op here)
# unless ``force=True``. So the dispatcher runs in the single process
# in dev and in the sidecar in prod — and the prod *web* container,
# which defers the pulse, also defers the dispatcher. "The web
# container never runs job bodies" falls out of the same env flag
# that already governs the pulse, not a second mechanism.
# ---------------------------------------------------------------------

_dispatch_stop = threading.Event()
_dispatch_thread: threading.Thread | None = None
_active: set[str] = set()
_active_lock = threading.Lock()
MAX_CONCURRENT = max(1, int(os.environ.get("JOBS_MAX_CONCURRENT", "3")))


def _run_one(job: dict) -> None:
    jid, kind = job["id"], job.get("kind")
    try:
        fn = runner_for(kind)
        if fn is None:
            finish_job(
                jid, status="error",
                reason=f"no runner registered for kind={kind!r}",
            )
            return
        fn(jid, job.get("params") or {})
        # Safety net: a runner that returned without setting a terminal
        # state would leave the indicator spinning forever.
        cur = get_job(jid)
        if cur and cur.get("status") in ACTIVE_STATES:
            finish_job(jid, status="done")
    except Exception as exc:  # noqa: BLE001 — never kill the dispatcher
        logger.exception("job runner crashed id=%s kind=%s", jid, kind)
        try:
            finish_job(
                jid, status="error",
                reason=f"{type(exc).__name__}: {exc}",
            )
        except Exception:
            pass
    finally:
        with _active_lock:
            _active.discard(jid)


def _dispatch_loop(interval_s: int) -> None:
    # Boot recovery first: fail anything a dead process left in flight
    # so indicators resolve instead of hanging forever.
    try:
        orphans = recover_orphans()
        if orphans:
            logger.warning(
                "recovered %d orphaned job(s) on dispatcher boot",
                len(orphans),
            )
    except Exception:
        logger.exception("orphan recovery failed at dispatcher boot")

    _last_reap = time.monotonic()
    while not _dispatch_stop.is_set():
        try:
            # Periodic stale-job reaper (every ~60s). The boot-time
            # recover_orphans above only catches what a *restart* left
            # behind; this catches what dies while the process stays up.
            if time.monotonic() - _last_reap >= 60:
                _last_reap = time.monotonic()
                try:
                    reaped = reap_stale_jobs()
                    if reaped:
                        logger.warning(
                            "reaped %d stale job(s): %s",
                            len(reaped),
                            ", ".join(
                                f"{r.get('id', '')[:8]}({r.get('kind')})"
                                for r in reaped
                            ),
                        )
                except Exception:
                    logger.exception("periodic stale-job reap failed")

            with _active_lock:
                at_cap = len(_active) >= MAX_CONCURRENT
            if not at_cap:
                job = claim_next()
                if job is not None:
                    with _active_lock:
                        _active.add(job["id"])
                    threading.Thread(
                        target=_run_one,
                        args=(job,),
                        name=f"job-{job.get('kind')}-{job['id'][:8]}",
                        daemon=True,
                    ).start()
                    # Immediately try to claim more so a backlog drains
                    # rather than trickling one per interval.
                    continue
        except Exception:
            logger.exception("dispatch tick failed")
        if _dispatch_stop.wait(timeout=interval_s):
            return


def start_dispatcher(
    *, force: bool = False, interval_s: int = 3,
) -> threading.Thread | None:
    """Spawn the dispatcher thread (idempotent).

    Defers (returns None) when ``HEARTBEAT_PULSE_WORKER=1`` and not
    ``force`` — same rule as ``heartbeat.start_daemon``. Lazily
    registers the Tier-3 runners so whatever process owns the
    dispatcher has them, with no caller having to remember the import.
    """
    global _dispatch_thread
    if not force and os.environ.get("HEARTBEAT_PULSE_WORKER") == "1":
        return None
    if _dispatch_thread is not None and _dispatch_thread.is_alive():
        return _dispatch_thread
    try:
        from lib.pulse import network_jobs

        network_jobs.register_runners()
    except Exception:
        logger.exception("job runner registration failed")
    _dispatch_stop.clear()
    _dispatch_thread = threading.Thread(
        target=_dispatch_loop, args=(interval_s,),
        name="job-dispatcher", daemon=True,
    )
    _dispatch_thread.start()
    logger.info("job dispatcher online (max_concurrent=%d)", MAX_CONCURRENT)
    return _dispatch_thread


def stop_dispatcher() -> None:
    _dispatch_stop.set()


def dispatcher_running() -> bool:
    t = _dispatch_thread
    return bool(t and t.is_alive() and not _dispatch_stop.is_set())


# ---------------------------------------------------------------------
# Read path  (PG preferred; JSONL fallback — dev == prod code path)
# ---------------------------------------------------------------------


def get_job(job_id: str) -> dict | None:
    if _pg_on():
        try:
            from lib.db import repos

            row = repos.get_job_row(job_id)
            if row is not None:
                return row
        except Exception:
            logger.debug("get_job PG failed; trying JSONL", exc_info=True)
    return _jsonl_state().get(job_id)


def get_job_progress(job_id: str, *, limit: int = 200) -> list[dict]:
    """The per-job progress trail, oldest → newest.

    JSONL-first by design: ``progress()`` appends to ``PROGRESS_PATH``
    *unconditionally and first* (the source of truth, exactly like the
    heartbeat event log). Returns ``[]`` when there is no trail yet —
    never raises. Each row: ``{"job_id", "ts", "pct", "message"}``.
    """
    out = [
        r for r in _read_jsonl(PROGRESS_PATH)
        if r.get("job_id") == job_id
    ]
    out.sort(key=lambda r: r.get("ts") or "")
    return out[-limit:] if limit and len(out) > limit else out


def snapshot(
    *,
    clerk_id: str | None = None,
    active_limit: int = 20,
    recent_limit: int = 12,
) -> dict[str, Any]:
    """What is running, plus the recently-finished tail.
    ``clerk_id=None`` → all actors (the dashboard overview).

    Multi-process safe: the web container reads job state from the
    store, mirroring ``heartbeat.snapshot()``'s PG overlay — never a
    worker thread-local, because the worker's memory is in a different
    container.
    """
    rows: list[dict]
    if _pg_on():
        try:
            from lib.db import repos

            rows = repos.jobs_for_actor(clerk_id=clerk_id, limit=80)
        except Exception:
            logger.debug("snapshot PG failed; JSONL", exc_info=True)
            rows = list(_jsonl_state().values())
    else:
        rows = list(_jsonl_state().values())

    if clerk_id and not _pg_on():
        rows = [r for r in rows if r.get("actor_clerk_id") == clerk_id]

    rows.sort(key=lambda r: r.get("created_at") or "", reverse=True)
    active = [r for r in rows if r.get("status") in ACTIVE_STATES][:active_limit]
    recent = [r for r in rows if r.get("status") in TERMINAL_STATES][:recent_limit]
    return {
        "active": active,
        "recent": recent,
        "active_count": len(active),
        "now": _now_iso(),
    }
```

    :defaultExpanded: false
    :withExpandedButton: true

.. llms_copy::Heartbeat · Jobs


---

*Source: /docs/heartbeat-jobs*