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*