feat: add IPv4 address space graph with interactive tooltip

app/constants.py

@@ -56,6 +56,12 @@ WHOIS_CACHE_PATH = os.path.join(DATA_DIR, "whois_cache.json")
 # Incremental-scan state: absolute log path -> last processed byte offset
 SCAN_STATE_PATH = os.path.join(DATA_DIR, "scan_state.json")
 
+# Web UI scan-job history (web_scan.py): persisted so the dashboard keeps
+# showing recent scans across server restarts. Oldest entries are pruned
+# once SCAN_JOBS_MAX is exceeded.
+SCAN_JOBS_PATH = os.path.join(DATA_DIR, "scan_jobs.json")
+SCAN_JOBS_MAX = 50
+
 # ---------------------------------------------------------------------------
 # Log sources (configurable via environment)
 # ---------------------------------------------------------------------------

app/static/style.css

@@ -371,3 +371,129 @@ pre.content {
   border: none;
   padding: 0.25rem 0.5rem;
 }
+
+/* IPv4 graph page */
+
+.graph-controls {
+  display: flex;
+  align-items: center;
+  gap: 1.25rem;
+  flex-wrap: wrap;
+}
+
+.legend {
+  display: inline-flex;
+  align-items: center;
+  gap: 0.5rem;
+  font-size: 0.82rem;
+  color: var(--muted);
+}
+
+.sw {
+  display: inline-block;
+  width: 0.8rem;
+  height: 0.8rem;
+  border-radius: 3px;
+  border: 1px solid var(--border);
+}
+.sw.red {
+  background: #e53935;
+}
+.sw.green {
+  background: #43a047;
+}
+.sw.plain {
+  background: #fff;
+}
+
+.graph-wrap {
+  position: relative;
+  width: 100%;
+  max-width: 920px;
+  margin: 0 auto;
+}
+
+#graph-canvas {
+  display: block;
+  width: 100%;
+  height: auto;
+  aspect-ratio: 1 / 1;
+  background: #fff;
+  border: 1px solid var(--border);
+  border-radius: 8px;
+  image-rendering: pixelated;
+  cursor: crosshair;
+}
+
+.graph-overlay {
+  position: absolute;
+  inset: 0;
+  pointer-events: none;
+}
+
+.g-line {
+  position: absolute;
+  background: rgba(148, 163, 184, 0.55);
+}
+.g-v {
+  top: 0;
+  bottom: 0;
+  width: 1px;
+}
+.g-h {
+  left: 0;
+  right: 0;
+  height: 1px;
+}
+
+.g-label {
+  position: absolute;
+  transform: translate(6px, 4px);
+  font-size: 0.55rem;
+  line-height: 1;
+  color: #475569;
+  white-space: nowrap;
+  user-select: none;
+  text-shadow: 0 0 2px #fff, 0 0 2px #fff;
+}
+
+.graph-tooltip {
+  position: absolute;
+  z-index: 5;
+  background: var(--surface-2);
+  border: 1px solid var(--border);
+  border-radius: 8px;
+  padding: 0.5rem 0.65rem;
+  pointer-events: none;
+  box-shadow: 0 6px 20px rgba(0, 0, 0, 0.45);
+  max-width: 260px;
+}
+.graph-tooltip.hidden {
+  display: none;
+}
+
+.tt-head {
+  font-weight: 600;
+  margin-bottom: 0.2rem;
+}
+.tt-status {
+  font-size: 0.75rem;
+  text-transform: uppercase;
+  letter-spacing: 0.04em;
+  font-weight: 600;
+}
+.tt-status.blocked {
+  color: #e53935;
+}
+.tt-status.whitelisted {
+  color: #43a047;
+}
+.tt-status.unlisted {
+  color: var(--muted);
+}
+.tt-list {
+  margin-top: 0.35rem;
+  font-size: 0.72rem;
+  white-space: pre-line;
+  color: var(--text);
+}

app/templates/base.html

@@ -35,6 +35,11 @@
           class="{{ 'active' if request.path == '/whitelist' }}"
           >Whitelist</a
         >
+        <a
+          href="/graph"
+          class="{{ 'active' if request.path == '/graph' }}"
+          >Graph</a
+        >
       </nav>
     </header>
     <main class="container">{% block content %}{% endblock %}</main>

app/templates/graph.html

@@ -0,0 +1,227 @@
+{% extends "base.html" %} {% block title %}Graph — Intrusion Detection{%
+endblock %} {% block content %}
+<h1>IPv4 address space map</h1>
+<p class="muted">
+  One pixel per /24 network, arranged as 16×16 tiles of /8 (first octet).
+  Blocked targets are red, whitelisted ranges are green.
+</p>
+
+<div class="panel">
+  <div class="panel-head">
+    <div class="graph-controls">
+      <label class="check" style="margin-top: 0">
+        <input type="checkbox" id="show-whitelisted" checked />
+        Show whitelisted
+      </label>
+      <span class="legend">
+        <span class="sw red"></span> Blocked
+        <span class="sw green"></span> Whitelisted
+        <span class="sw plain"></span> Unlisted
+      </span>
+    </div>
+    <button class="btn ghost" id="graph-refresh">↻ Refresh</button>
+  </div>
+  <div class="graph-wrap" id="graph-wrap">
+    <canvas id="graph-canvas" width="4096" height="4096"></canvas>
+    <div class="graph-overlay" id="graph-overlay"></div>
+    <div class="graph-tooltip hidden" id="graph-tooltip"></div>
+  </div>
+  <p class="muted mono" id="graph-stats">Loading…</p>
+</div>
+{% endblock %} {% block scripts %}
+<script>
+  // Layout mirrors plot.py: 16x16 grid of /8 tiles, each 256x256 px (one pixel
+  // per /24). Blocked ranges override whitelisted ranges.
+  const TILE = 16;
+  const SIZE = 256;
+  const TOTAL = TILE * SIZE; // 4096
+
+  const V_UNLISTED = 0;
+  const V_WHITELISTED = 1;
+  const V_BLOCKED = 2;
+  const COLOR_WHITELISTED = "#43a047";
+  const COLOR_BLOCKED = "#e53935";
+
+  const canvas = document.getElementById("graph-canvas");
+  const ctx = canvas.getContext("2d");
+  const overlay = document.getElementById("graph-overlay");
+  const tooltip = document.getElementById("graph-tooltip");
+  const stats = document.getElementById("graph-stats");
+
+  // /24 index (0..2^24-1) -> value (whitelisted/blocked)
+  const marked = new Map();
+
+  function parseCIDR(s) {
+    const m = String(s)
+      .trim()
+      .match(/^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})(?:\/(\d{1,2}))?$/);
+    if (!m) return null;
+    const o = [m[1], m[2], m[3], m[4]].map(Number);
+    if (o.some((n) => n > 255)) return null;
+    const prefix = m[5] === undefined ? 32 : Number(m[5]);
+    if (prefix < 0 || prefix > 32) return null;
+    const ip = ((o[0] << 24) | (o[1] << 16) | (o[2] << 8) | o[3]) >>> 0;
+    const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0;
+    const network = (ip & mask) >>> 0;
+    const broadcast = (ip | (~mask >>> 0)) >>> 0;
+    return {
+      first24: network >>> 8,
+      last24: broadcast >>> 8,
+      label: `${o[0]}.${o[1]}.${o[2]}.${o[3]}/${prefix}`,
+    };
+  }
+
+  function idxToPos(idx) {
+    const a = (idx >>> 16) & 0xff;
+    const b = (idx >>> 8) & 0xff;
+    const c = idx & 0xff;
+    return {
+      x: (a % TILE) * SIZE + b,
+      y: Math.floor(a / TILE) * SIZE + c,
+    };
+  }
+
+  function markRange(range, value) {
+    for (let idx = range.first24; idx <= range.last24; idx++) {
+      const prev = marked.get(idx);
+      if (prev === undefined || value > prev) marked.set(idx, value);
+    }
+  }
+
+  function drawOverlay() {
+    overlay.innerHTML = "";
+    // /8 tile borders
+    for (let i = 0; i <= TILE; i++) {
+      const pct = (i / TILE) * 100;
+      const v = el("div", { class: "g-line g-v", style: `left:${pct}%` });
+      const h = el("div", { class: "g-line g-h", style: `top:${pct}%` });
+      overlay.append(v, h);
+    }
+    // first-octet labels, one per /8 tile
+    for (let a = 0; a < 256; a++) {
+      const col = a % TILE;
+      const row = Math.floor(a / TILE);
+      overlay.append(
+        el("div", {
+          class: "g-label",
+          style: `left:${(col / TILE) * 100}%;top:${(row / TILE) * 100}%`,
+          text: `${a}.0.0.0/8`,
+        }),
+      );
+    }
+  }
+
+  function render() {
+    const showWl = document.getElementById("show-whitelisted").checked;
+    ctx.fillStyle = "#ffffff";
+    ctx.fillRect(0, 0, TOTAL, TOTAL);
+    for (const [idx, value] of marked) {
+      if (!showWl && value === V_WHITELISTED) continue;
+      const p = idxToPos(idx);
+      ctx.fillStyle = value === V_BLOCKED ? COLOR_BLOCKED : COLOR_WHITELISTED;
+      ctx.fillRect(p.x, p.y, 1, 1);
+    }
+  }
+
+  // --- hover tooltip ------------------------------------------------------
+
+  function matchesIn(ranges, idx) {
+    const out = [];
+    for (const r of ranges) {
+      if (idx >= r.first24 && idx <= r.last24) out.push(r.label);
+      if (out.length >= 12) break;
+    }
+    return out;
+  }
+
+  function showTooltip(e, x, y) {
+    const tileCol = Math.floor(x / SIZE);
+    const tileRow = Math.floor(y / SIZE);
+    const a = tileRow * TILE + tileCol;
+    const b = x % SIZE;
+    const c = y % SIZE;
+    const idx = (a << 16) | (b << 8) | c;
+    const value = marked.get(idx) ?? V_UNLISTED;
+    const status =
+      value === V_BLOCKED
+        ? "blocked"
+        : value === V_WHITELISTED
+          ? "whitelisted"
+          : "unlisted";
+    const entries = matchesIn(blockedRanges, idx).concat(
+      matchesIn(whitelistedRanges, idx),
+    );
+    tooltip.innerHTML = "";
+    tooltip.append(
+      el("div", { class: "tt-head mono", text: `${a}.${b}.${c}.0/24` }),
+      el("div", { class: `tt-status ${status}`, text: status }),
+    );
+    if (entries.length) {
+      tooltip.append(
+        el("div", { class: "tt-list mono", text: entries.join("\n") }),
+      );
+    } else {
+      tooltip.append(
+        el("div", { class: "tt-list muted", text: "no matching entry" }),
+      );
+    }
+    tooltip.classList.remove("hidden");
+    const rect = canvas.getBoundingClientRect();
+    let left = e.clientX - rect.left + 14;
+    let top = e.clientY - rect.top + 14;
+    if (left + tooltip.offsetWidth > rect.width)
+      left = e.clientX - rect.left - tooltip.offsetWidth - 14;
+    if (top + tooltip.offsetHeight > rect.height)
+      top = e.clientY - rect.top - tooltip.offsetHeight - 14;
+    tooltip.style.left = `${Math.max(0, left)}px`;
+    tooltip.style.top = `${Math.max(0, top)}px`;
+  }
+
+  canvas.addEventListener("mousemove", (e) => {
+    const rect = canvas.getBoundingClientRect();
+    const lx = ((e.clientX - rect.left) * TOTAL) / rect.width;
+    const ly = ((e.clientY - rect.top) * TOTAL) / rect.height;
+    const x = Math.floor(lx);
+    const y = Math.floor(ly);
+    if (x < 0 || x >= TOTAL || y < 0 || y >= TOTAL) {
+      tooltip.classList.add("hidden");
+      return;
+    }
+    showTooltip(e, x, y);
+  });
+
+  canvas.addEventListener("mouseleave", () => {
+    tooltip.classList.add("hidden");
+  });
+
+  // --- load ---------------------------------------------------------------
+
+  let blockedRanges = [];
+  let whitelistedRanges = [];
+
+  async function loadGraph() {
+    try {
+      const d = await api("/api/graph/ipv4");
+      blockedRanges = (d.blocked || []).map(parseCIDR).filter(Boolean);
+      whitelistedRanges = (d.whitelisted || []).map(parseCIDR).filter(Boolean);
+      marked.clear();
+      for (const r of whitelistedRanges) markRange(r, V_WHITELISTED);
+      for (const r of blockedRanges) markRange(r, V_BLOCKED);
+      render();
+      drawOverlay();
+      const c = d.counts || {};
+      stats.textContent = `blocked targets: ${c.blocked ?? blockedRanges.length} · whitelisted ranges: ${c.whitelisted ?? whitelistedRanges.length} · marked /24 pixels: ${marked.size}`;
+    } catch (err) {
+      stats.textContent = `failed to load graph data: ${err.message}`;
+      toast(err.message, "error");
+    }
+  }
+
+  document
+    .getElementById("show-whitelisted")
+    .addEventListener("change", render);
+  document.getElementById("graph-refresh").addEventListener("click", loadGraph);
+
+  loadGraph();
+</script>
+{% endblock %}

app/web.py

@@ -143,6 +143,10 @@ def create_app() -> Flask:
     """Build and configure the Flask application."""
     app = Flask(__name__)
 
+    # Seed the in-memory job store from disk so the dashboard keeps showing
+    # recent scans across server restarts.
+    web_scan.load_persisted()
+
     @app.get("/")
     def index():
         return render_template("dashboard.html")
@@ -167,6 +171,10 @@ def create_app() -> Flask:
     def whitelist_page():
         return render_template("whitelist.html", whitelist_names=WHITELIST_NAMES)
 
+    @app.get("/graph")
+    def graph_page():
+        return render_template("graph.html")
+
     # --- status ------------------------------------------------------------
 
     @app.get("/api/status")
@@ -284,6 +292,35 @@ def create_app() -> Flask:
 
         return _with_router(_remove)
 
+    # --- graph --------------------------------------------------------------
+
+    @app.get("/api/graph/ipv4")
+    def graph_ipv4():
+        """Data for the IPv4 /8 tile map.
+
+        Blocked targets come from the consolidated blocklist
+        (``reports/app/data/blocklist_ipv4.txt``); whitelisted ranges from the
+        router snapshot (``data/mikrotik/address_list_ipv4.json``) filtered to
+        the configured whitelist address-list names.
+        """
+        blocked = _read_lines(
+            os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt"))
+        addr_v4 = _read_json(
+            os.path.join(MIKROTIK_DATA_DIR, "address_list_ipv4.json"))
+        whitelisted = sorted({e.get("address") for e in addr_v4
+                              if e.get("list") in WHITELIST_NAMES
+                              and e.get("address")})
+        return jsonify({
+            "blocked": blocked,
+            "whitelisted": whitelisted,
+            "counts": {
+                "blocked": len(blocked),
+                "whitelisted": len(whitelisted),
+            },
+            "blocklist_name": BLOCKLIST_NAME,
+            "whitelist_names": WHITELIST_NAMES,
+        })
+
     # --- firewall ----------------------------------------------------------
 
     @app.get("/api/firewall/state")

app/web_scan.py

@@ -8,12 +8,22 @@ whitelist refresh must not run concurrently).
 
 from __future__ import annotations
 
+import json
+import os
 import threading
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from datetime import datetime
 from typing import Any, Literal
 
+from .constants import (
+    APP_REPORT_DATA_DIR,
+    COPYPARTY_REPORT_DIR,
+    MIKROTIK_REPORT_DIR,
+    NPM_REPORT_DIR,
+    SCAN_JOBS_MAX,
+    SCAN_JOBS_PATH,
+)
 from .copyparty import CopypartyAnalyzer
 from .npm import NpmAnalyzer
 from .scan import ScanParams, ScanResult, run_scan
@@ -27,6 +37,9 @@ _scan_lock = threading.Lock()
 _jobs: dict[str, "ScanJob"] = {}
 _jobs_lock = threading.Lock()
 
+#: True once persisted jobs have been loaded from disk
+_loaded = False
+
 
 def _now() -> str:
     return datetime.now().astimezone().isoformat(timespec="seconds")
@@ -52,6 +65,7 @@ class ScanJob:
         return {
             "job_id": self.job_id,
             "kind": self.kind,
+            "params": asdict(self.params),
             "state": self.state,
             "started": self.started,
             "finished": self.finished,
@@ -66,6 +80,8 @@ class ScanJob:
                 "npm_candidates": result.npm_candidates,
                 "blocked_ips": result.blocked_ips,
                 "blocked_targets": result.blocked_targets,
+                "blocked_ipv4": result.blocked_ipv4,
+                "blocked_ipv6": result.blocked_ipv6,
                 "blocklist_v4_path": result.blocklist_v4_path,
                 "blocklist_v6_path": result.blocklist_v6_path,
                 "firewall_connected": result.firewall_connected,
@@ -87,6 +103,7 @@ def start_scan(kind: ScanKind, params: ScanParams) -> dict[str, Any]:
                   params=params, busy=True)
     with _jobs_lock:
         _jobs[job.job_id] = job
+    _save_jobs()
 
     def _run() -> None:
         job.started = _now()
@@ -129,6 +146,7 @@ def start_scan(kind: ScanKind, params: ScanParams) -> dict[str, Any]:
             job.finished = _now()
             job.busy = False
             _scan_lock.release()
+            _save_jobs()
 
     threading.Thread(target=_run, daemon=True,
                      name=f"scan-{job.job_id}").start()
@@ -137,17 +155,204 @@ def start_scan(kind: ScanKind, params: ScanParams) -> dict[str, Any]:
 
 def get_scan(job_id: str) -> dict[str, Any] | None:
     """Return a single job as a dict, or None if unknown."""
+    load_persisted()
     with _jobs_lock:
         job = _jobs.get(job_id)
     return job.to_dict() if job else None
 
 
+def _iso_key(value: str | None) -> float:
+    """ISO-8601 string -> epoch seconds (0.0 when unparseable/empty)."""
+    if not value:
+        return 0.0
+    try:
+        return datetime.fromisoformat(value).timestamp()
+    except ValueError:
+        return 0.0
+
+
+def latest_file_scan() -> dict[str, Any] | None:
+    """Build a ``done`` scan entry from the newest report files on disk.
+
+    Lets the dashboard show the latest completed scan even when it was not
+    run through the web UI (server restart, CLI-run scans). Returns None when
+    no scan outputs exist yet.
+    """
+
+    def newest(dirpath: str, suffix: str = "") -> str | None:
+        """Newest file in ``dirpath`` (optionally ending in ``suffix``)."""
+        try:
+            names = [n for n in os.listdir(dirpath)
+                     if os.path.isfile(os.path.join(dirpath, n))
+                     and (not suffix or n.endswith(suffix))]
+        except OSError:
+            return None
+        return os.path.join(dirpath, max(names)) if names else None
+
+    mikrotik = newest(MIKROTIK_REPORT_DIR, ".json")
+    cp_report = newest(COPYPARTY_REPORT_DIR, ".txt")
+    npm_report = newest(NPM_REPORT_DIR, ".txt")
+    v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")
+    v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt")
+    have_outputs = (mikrotik is not None or cp_report is not None
+                    or npm_report is not None
+                    or os.path.exists(v4_path) or os.path.exists(v6_path))
+    if not have_outputs:
+        return None
+
+    # Timestamp: prefer the newest snapshot's collected_at, else the newest
+    # report file's mtime, else now.
+    timestamp: str | None = None
+    if mikrotik:
+        try:
+            with open(mikrotik, "r", encoding="utf-8") as f:
+                timestamp = json.load(f).get("collected_at")
+        except (OSError, ValueError):
+            timestamp = None
+    if not timestamp:
+        candidates = [p for p in (mikrotik, cp_report, npm_report,
+                                  v4_path, v6_path)
+                      if p and os.path.exists(p)]
+        if candidates:
+            tz = datetime.now().astimezone().tzinfo
+            timestamp = datetime.fromtimestamp(
+                max(os.path.getmtime(p) for p in candidates),
+                tz=tz).isoformat(timespec="seconds")
+    if not timestamp:
+        timestamp = _now()
+
+    blocked_targets = 0
+    for path in (v4_path, v6_path):
+        if os.path.exists(path):
+            try:
+                with open(path, "r", encoding="utf-8") as f:
+                    blocked_targets += sum(1 for _ in f)
+            except OSError:
+                pass
+
+    return {
+        "job_id": "files-latest",
+        "kind": "consolidated",
+        "state": "done",
+        "started": timestamp,
+        "finished": timestamp,
+        "error": None,
+        "busy": False,
+        "details": {},
+        "result": {
+            "ok": True,
+            "cp_exists": cp_report is not None,
+            "npm_exists": npm_report is not None,
+            "cp_candidates": 0,
+            "npm_candidates": 0,
+            "blocked_ips": blocked_targets,
+            "blocked_targets": blocked_targets,
+            "blocked_ipv4": [],
+            "blocked_ipv6": [],
+            "blocklist_v4_path": v4_path,
+            "blocklist_v6_path": v6_path,
+            "firewall_connected": mikrotik is not None,
+            "firewall_changes": None,
+            "errors": [],
+        },
+    }
+
+
 def latest_scans(limit: int = 10) -> list[dict[str, Any]]:
-    """Return the most recently started jobs, newest first."""
+    """Return the most recently started jobs, newest first.
+
+    Appends a file-derived "latest scan" entry when the newest report files on
+    disk reflect a scan newer than any known job (e.g. after a restart or a
+    CLI run), so the dashboard always shows the latest completed scan.
+    """
+    load_persisted()
     with _jobs_lock:
         jobs = sorted(_jobs.values(),
                       key=lambda j: j.started or "", reverse=True)
-    return [j.to_dict() for j in jobs[:limit]]
+    entries = [j.to_dict() for j in jobs]
+    file_entry = latest_file_scan()
+    if file_entry is not None:
+        newest = jobs[0] if jobs else None
+        newest_ts = _iso_key(
+            (newest.finished or newest.started) if newest else None)
+        # Only surface the file-derived entry when it is genuinely newer than
+        # every known job (tolerance avoids duplicating the run that produced
+        # it, where collected_at lands a moment after the job finished).
+        if newest_ts == 0 or _iso_key(file_entry["started"]) - newest_ts > 2.0:
+            entries.append(file_entry)
+    entries.sort(key=lambda e: _iso_key(e.get("started")), reverse=True)
+    return entries[:limit]
+
+
+# ---------------------------------------------------------------------------
+# persistence
+# ---------------------------------------------------------------------------
+
+def _save_jobs() -> None:
+    """Persist the newest jobs to ``SCAN_JOBS_PATH`` (atomic tmp+rename)."""
+    with _jobs_lock:
+        jobs = sorted(_jobs.values(),
+                      key=lambda j: j.started or "", reverse=True)
+        payload = [j.to_dict() for j in jobs[:SCAN_JOBS_MAX]]
+    os.makedirs(os.path.dirname(SCAN_JOBS_PATH), exist_ok=True)
+    tmp = SCAN_JOBS_PATH + ".tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(payload, f, indent=2, sort_keys=True)
+    os.replace(tmp, SCAN_JOBS_PATH)
+
+
+def _load_jobs() -> None:
+    """Rebuild the in-memory job store from ``SCAN_JOBS_PATH``."""
+    try:
+        with open(SCAN_JOBS_PATH, "r", encoding="utf-8") as f:
+            payload = json.load(f)
+    except (OSError, ValueError, TypeError):
+        payload = []
+    if not isinstance(payload, list):
+        return
+
+    jobs: dict[str, ScanJob] = {}
+    for item in payload:
+        if not isinstance(item, dict):
+            continue
+        try:
+            params = ScanParams(**item.get("params", {}))
+        except (TypeError, ValueError):
+            continue
+        try:
+            job = ScanJob(
+                job_id=str(item["job_id"]),
+                kind=item.get("kind", "consolidated"),
+                params=params,
+                state=item.get("state", "done"),
+                started=item.get("started"),
+                finished=item.get("finished"),
+                error=item.get("error"),
+                busy=False,
+                details=item.get("details", {}) or {},
+            )
+        except (KeyError, TypeError, ValueError):
+            continue
+        result = item.get("result")
+        if isinstance(result, dict):
+            try:
+                job.result = ScanResult(**result)
+            except TypeError:
+                job.result = None
+        jobs[job.job_id] = job
+
+    with _jobs_lock:
+        _jobs.clear()
+        _jobs.update(jobs)
+
+
+def load_persisted() -> None:
+    """Load persisted jobs from disk once (idempotent)."""
+    global _loaded
+    if _loaded:
+        return
+    _load_jobs()
+    _loaded = True
 
 
 def is_scan_running() -> bool: