diff --git a/kernel_ai/api/rest.py b/kernel_ai/api/rest.py
index 30a870f..9c5646c 100644
--- a/kernel_ai/api/rest.py
+++ b/kernel_ai/api/rest.py
@@ -6,6 +6,8 @@
 
 _API_ROUTES = [
     ("/syscalls-realtime", "syscalls_realtime", h.syscalls_realtime, None),
+    ("/syscall/<name>", "syscall_detail", h.syscall_detail, None),
+    ("/irq/<irq>", "irq_detail", h.irq_detail, None),
     ("/kernel-data", "kernel_data", h.kernel_data, None),
     ("/io-pulse", "io_pulse", h.io_pulse, None),
     ("/process-kernel-map", "process_kernel_map", h.process_kernel_map, None),
diff --git a/kernel_ai/http/api.py b/kernel_ai/http/api.py
index f40ed76..dd808c8 100644
--- a/kernel_ai/http/api.py
+++ b/kernel_ai/http/api.py
@@ -4,6 +4,7 @@
     get_execution_context,
     io_open_files,
     io_pulse,
+    irq_detail,
     kernel_data,
     kernel_dna,
     ml_anomalies,
@@ -12,6 +13,7 @@
     process_kernel_map,
     sentry_test,
     siem_alerts,
+    syscall_detail,
     syscalls_realtime,
 )
 from kernel_ai.http.api_handlers.network_system import (
@@ -78,6 +80,7 @@
     "ingest_frontend_logs",
     "io_open_files",
     "io_pulse",
+    "irq_detail",
     "isolation_context",
     "kernel_data",
     "kernel_dna",
@@ -91,6 +94,7 @@
     "security_realtime",
     "sentry_test",
     "siem_alerts",
+    "syscall_detail",
     "syscalls_realtime",
     "traceroute_info",
 ]
diff --git a/kernel_ai/ml/config.py b/kernel_ai/ml/config.py
index ca24702..0929e0c 100644
--- a/kernel_ai/ml/config.py
+++ b/kernel_ai/ml/config.py
@@ -149,8 +149,16 @@ class MLConfig:
     seq_mismatch_warn: float = _env_float("KERNEL_AI_ML_SEQ_MISMATCH_WARN", 0.30)
     seq_mismatch_crit: float = _env_float("KERNEL_AI_ML_SEQ_MISMATCH_CRIT", 0.55)
     seq_cooldown_sec: float = _env_float("KERNEL_AI_ML_SEQ_COOLDOWN_SEC", 30.0)
+    # How long the event source may stay silent before it is reported. A dead feed
+    # used to be indistinguishable from a quiet host: on 2026-08-11 the kernel audit
+    # switch went off for 30 hours and Stage 4 kept re-scoring its frozen window,
+    # emitting one anomaly per cooldown (120/hour) the whole time.
+    seq_stale_warn_sec: float = _env_float("KERNEL_AI_ML_SEQ_STALE_WARN_SEC", 300.0)
     # Flush newly observed n-grams to the store every N seconds (profile growth).
     seq_flush_sec: float = _env_float("KERNEL_AI_ML_SEQ_FLUSH_SEC", 30.0)
+    # Forget a pid whose last event is this many events old — bounds memory and
+    # keeps windows of long-gone processes out of the candidate set.
+    seq_pid_idle_events: int = _env_int("KERNEL_AI_ML_SEQ_PID_IDLE_EVENTS", 5000)
     # Training keeps n-grams seen at least this many times (frequency-based poison
     # guard: a one-off attack sequence never enters the "normal" profile).
     seq_min_ngram_count: int = _env_int("KERNEL_AI_ML_SEQ_MIN_COUNT", 3)
@@ -191,6 +199,9 @@ class MLConfig:
     stage8_score_warn: float = _env_float("KERNEL_AI_ML_STAGE8_SCORE_WARN", 3.0)
     stage8_score_crit: float = _env_float("KERNEL_AI_ML_STAGE8_SCORE_CRIT", 5.0)
     stage8_cooldown_sec: float = _env_float("KERNEL_AI_ML_STAGE8_COOLDOWN_SEC", 30.0)
+    # Heartbeat of the raw window score (no anomaly attached): lets the thresholds
+    # above be derived from the live distribution of a given host.
+    stage8_log_every_sec: float = _env_float("KERNEL_AI_ML_STAGE8_LOG_EVERY_SEC", 60.0)
 
     @property
     def alpha(self) -> float:
diff --git a/kernel_ai/ml/retrain.py b/kernel_ai/ml/retrain.py
index 299d2ec..58efd27 100644
--- a/kernel_ai/ml/retrain.py
+++ b/kernel_ai/ml/retrain.py
@@ -1,6 +1,6 @@
-"""Stage 3 + Stage 4 auto-retrain orchestrator (run by a systemd timer).
+"""Stage 3 + Stage 4 + Stage 8 auto-retrain orchestrator (run by a systemd timer).
 
-    measure drift  ->  decide  ->  IsolationForest on clean data  ->  STIDE profile
+    measure drift -> decide -> IsolationForest on clean data -> STIDE -> Markov
 
 Decision:
   * default: always retrain IsolationForest (the timer cadence IS the schedule).
@@ -47,12 +47,44 @@ def _refresh_stide(cfg: MLConfig, metrics: dict) -> None:
         logger.warning("STIDE profile build failed: %s", exc)
 
 
+def _refresh_markov(cfg: MLConfig, metrics: dict) -> None:
+    """Best-effort Stage 8 rebuild from the same n-gram corpus as STIDE.
+
+    Without this the transition model freezes on the day it was trained while the
+    vocabulary underneath it keeps growing. Note that a rebuild can move the score
+    scale, and ``KERNEL_AI_ML_STAGE8_SCORE_WARN`` is tuned against a scale — the
+    numbers below are logged so a drift away from the tuned thresholds is visible
+    (the worker's periodic window-score line is the other half of that check).
+    """
+    if not getattr(cfg, "enable_stage8", False):
+        return
+    if (getattr(cfg, "stage8_backend", "markov") or "markov") != "markov":
+        return
+    try:
+        from kernel_ai.ml.sequence_deep.train_markov import train_markov
+
+        seq_meta = train_markov(cfg, use_ngrams=True)
+        metrics["stage8_transitions"] = float(seq_meta.get("n_transitions") or 0)
+        logger.info(
+            "Stage 8 Markov refreshed: transitions=%s vocab=%s common=%s rare=%s",
+            seq_meta.get("n_transitions"),
+            seq_meta.get("vocab"),
+            seq_meta.get("corpus_common_neg_avg_logprob"),
+            seq_meta.get("corpus_rare_neg_avg_logprob"),
+        )
+    except SystemExit as exc:
+        logger.info("Stage 8 Markov not rebuilt: %s", exc)
+    except Exception as exc:  # noqa: BLE001 - deep sequence model is optional
+        logger.warning("Stage 8 Markov build failed: %s", exc)
+
+
 def run(*, only_if_drift: bool, min_samples: int, stide_only: bool = False) -> int:
     cfg = MLConfig()
     metrics: dict = {}
 
     if stide_only:
         _refresh_stide(cfg, metrics)
+        _refresh_markov(cfg, metrics)
         logger.info("stide-only retrain complete: %s", metrics)
         return 0
 
@@ -81,6 +113,7 @@ def run(*, only_if_drift: bool, min_samples: int, stide_only: bool = False) -> i
             metrics = {}
 
     _refresh_stide(cfg, metrics)
+    _refresh_markov(cfg, metrics)
     logger.info("retrain complete: %s", metrics)
     return 0
 
diff --git a/kernel_ai/ml/sequence.py b/kernel_ai/ml/sequence.py
index 75a691d..94e1f3d 100644
--- a/kernel_ai/ml/sequence.py
+++ b/kernel_ai/ml/sequence.py
@@ -33,6 +33,14 @@
 # Separator for serialising an n-gram tuple into a stable string key.
 _SEP = "|"
 
+# Synthetic events from the Stage 6 collector's demo mode carry these comms. They
+# may be scored (that is how the demo proves the wiring is alive) but must never
+# enter the persisted profile: on 2026-08-03 a single demo minute wrote ~34k
+# observations of connect→vfork→memfd_create→userfaultfd into ml_syscall_ngrams,
+# making the fileless-payload syscalls we watch for the most "normal" chain on the
+# host, far above any frequency-based poison guard.
+DEMO_COMMS = frozenset({"demo", "mimic", "novel"})
+
 
 class SyscallSampler:
     """Sample the current syscall of running processes from procfs."""
@@ -79,6 +87,15 @@ def __init__(self, n: int = 3, window: int = 400, max_pids: int = 4096) -> None:
         self.n = max(2, n)
         self.window = window
         self.max_pids = max_pids
+        # Monotonic count of samples ever appended. The rolling windows below keep
+        # their contents when the event source dies, so scoring them again would
+        # re-report the same window forever; callers compare this counter against
+        # the value they last scored to tell fresh evidence from a frozen feed.
+        self.ingested = 0
+        # Value of ``ingested`` when each pid last produced something. A pid window
+        # survives the process that filled it, so scorers need to tell "this pid is
+        # still doing that" from "this pid died holding an odd-looking window".
+        self._pid_stamp: dict[int, int] = {}
         self._hist: dict[int, deque[str]] = {}
         # Global rolling window (profile growth / Stage 8 / fallback scoring).
         self._recent: deque[str] = deque(maxlen=window)
@@ -91,8 +108,11 @@ def __init__(self, n: int = 3, window: int = 400, max_pids: int = 4096) -> None:
     def _drop_pid(self, pid: int) -> None:
         self._hist.pop(pid, None)
         self._recent_by_pid.pop(pid, None)
+        self._pid_stamp.pop(pid, None)
 
-    def _append(self, pid: int, name: str) -> None:
+    def _append(self, pid: int, name: str, *, learn: bool = True) -> None:
+        self.ingested += 1
+        self._pid_stamp[pid] = self.ingested
         hist = self._hist.get(pid)
         if hist is None:
             hist = deque(maxlen=self.n)
@@ -101,7 +121,8 @@ def _append(self, pid: int, name: str) -> None:
         if len(hist) == self.n:
             key = _SEP.join(hist)
             self._recent.append(key)
-            self._pending[key] = self._pending.get(key, 0) + 1
+            if learn:
+                self._pending[key] = self._pending.get(key, 0) + 1
             pid_win = self._recent_by_pid.get(pid)
             if pid_win is None:
                 pid_win = deque(maxlen=self.window)
@@ -128,7 +149,8 @@ def update_stream(self, events) -> int:
                 continue
             if not name:
                 continue
-            self._append(pid, name)
+            comm = getattr(ev, "comm", None) if not isinstance(ev, dict) else ev.get("comm")
+            self._append(pid, name, learn=str(comm or "") not in DEMO_COMMS)
             n += 1
             if len(self._hist) > self.max_pids:
                 # Bound memory: drop oldest pid histories opportunistically.
@@ -148,6 +170,34 @@ def recent_tokens(self) -> list[str]:
         """
         return ngrams_to_tokens(self.recent(), n=self.n)
 
+    def pid_stamps(self) -> dict[int, int]:
+        """``{pid: ingest counter when it last emitted}`` — freshness per pid."""
+        return dict(self._pid_stamp)
+
+    def evict_idle(self, *, older_than: int) -> int:
+        """Forget pids silent for the last ``older_than`` ingested events."""
+        cutoff = self.ingested - max(1, int(older_than))
+        dead = [pid for pid, stamp in self._pid_stamp.items() if stamp < cutoff]
+        for pid in dead:
+            self._drop_pid(pid)
+        return len(dead)
+
+    def recent_tokens_by_pid(self, *, min_len: int) -> list[tuple[int, list[str]]]:
+        """Per-pid syscall token streams, for scorers that work on tokens (Stage 8).
+
+        Averaging over the global mix hides short chains: a busy daemon emitting
+        thousands of ``accept4`` transitions drags any window average back to normal,
+        so a five-syscall hostile burst on another pid moves it by ~0.1. Stage 4 hit
+        exactly this and moved to per-pid windows; the deep scorer needs the same.
+        """
+        need = max(2, int(min_len))
+        out: list[tuple[int, list[str]]] = []
+        for pid, dq in self._recent_by_pid.items():
+            tokens = ngrams_to_tokens(list(dq), n=self.n)
+            if len(tokens) >= need:
+                out.append((pid, tokens))
+        return out
+
     def recent_by_pid(self, *, min_len: int) -> list[tuple[int, list[str]]]:
         """Return ``(pid, ngram_window)`` for pids with enough recent n-grams."""
         need = max(1, int(min_len))
diff --git a/kernel_ai/ml/sequence_deep/markov.py b/kernel_ai/ml/sequence_deep/markov.py
index 07d2225..251a2a2 100644
--- a/kernel_ai/ml/sequence_deep/markov.py
+++ b/kernel_ai/ml/sequence_deep/markov.py
@@ -19,18 +19,32 @@ class MarkovScorer:
     # counts[prev][nxt] = count
     _counts: dict[str, dict[str, int]] = field(default_factory=dict)
     _row_totals: dict[str, int] = field(default_factory=dict)
+    # How often each token appears as a destination, and the grand total. Used to
+    # back off when the *source* token was never seen at all.
+    _token_totals: dict[str, int] = field(default_factory=dict)
+    _total: int = 0
 
     @property
     def ready(self) -> bool:
         return bool(self._counts)
 
-    def observe(self, tokens: list[str]) -> None:
-        if len(tokens) < 2:
+    def observe(self, tokens: list[str], *, weight: int = 1) -> None:
+        """Fold a sequence into the table; ``weight`` is how often it was seen.
+
+        Weights matter: the corpus is a table of n-grams with counts spanning five
+        orders of magnitude (``accept4|accept4|accept4`` 256k vs a chain seen three
+        times). Materialising repeats to represent that is wasteful, and capping them
+        flattens the model into "everything is roughly equally likely" — which is the
+        one thing a transition model must not believe.
+        """
+        if len(tokens) < 2 or weight <= 0:
             return
         for a, b in zip(tokens, tokens[1:]):
             row = self._counts.setdefault(a, {})
-            row[b] = row.get(b, 0) + 1
-            self._row_totals[a] = self._row_totals.get(a, 0) + 1
+            row[b] = row.get(b, 0) + weight
+            self._row_totals[a] = self._row_totals.get(a, 0) + weight
+            self._token_totals[b] = self._token_totals.get(b, 0) + weight
+            self._total += weight
 
     def score_window(self, tokens: list[str]) -> dict | None:
         """Return neg-avg-logprob style score, or None if untrained / too short."""
@@ -45,8 +59,15 @@ def score_window(self, tokens: list[str]) -> dict | None:
             row_total = self._row_totals.get(prev, 0)
             # Laplace-ish smoothing so unseen transitions are finite but rare.
             vocab = max(1, len(self._counts))
-            cnt = (self._counts.get(prev) or {}).get(nxt, 0)
-            prob = (cnt + 1.0) / (row_total + vocab) if row_total else 1.0 / vocab
+            if row_total:
+                cnt = (self._counts.get(prev) or {}).get(nxt, 0)
+                prob = (cnt + 1.0) / (row_total + vocab)
+            else:
+                # This syscall was never seen as a source at all. Backing off to
+                # ``1/vocab`` made such a step look *certain* in a small model (and
+                # cheap in a large one); score it against the whole corpus instead,
+                # so a chain of never-seen syscalls reads as the surprise it is.
+                prob = (self._token_totals.get(nxt, 0) + 1.0) / (self._total + vocab)
             lp = math.log(max(prob, 1e-12))
             total += lp
             n += 1
@@ -71,6 +92,8 @@ def state_dict(self) -> dict:
             "meta": dict(self.meta),
             "counts": self._counts,
             "row_totals": self._row_totals,
+            "token_totals": self._token_totals,
+            "total": self._total,
         }
 
     @classmethod
@@ -78,6 +101,15 @@ def from_state(cls, state: dict) -> "MarkovScorer":
         m = cls(order=int(state.get("order") or 1), meta=dict(state.get("meta") or {}))
         m._counts = {k: dict(v) for k, v in (state.get("counts") or {}).items()}
         m._row_totals = {k: int(v) for k, v in (state.get("row_totals") or {}).items()}
+        token_totals = state.get("token_totals")
+        if token_totals is None:
+            # Artifact from before the back-off was added: derive it from the table.
+            token_totals = {}
+            for row in m._counts.values():
+                for nxt, cnt in row.items():
+                    token_totals[nxt] = token_totals.get(nxt, 0) + int(cnt)
+        m._token_totals = {k: int(v) for k, v in token_totals.items()}
+        m._total = int(state.get("total") or sum(m._row_totals.values()))
         return m
 
 
diff --git a/kernel_ai/ml/sequence_deep/scorer.py b/kernel_ai/ml/sequence_deep/scorer.py
index efff877..6873e20 100644
--- a/kernel_ai/ml/sequence_deep/scorer.py
+++ b/kernel_ai/ml/sequence_deep/scorer.py
@@ -92,6 +92,8 @@ def build_anomaly(self, score: dict, cfg: Any) -> dict:
         severity = "high" if neg >= crit else "medium"
         worst = score.get("worst_tokens") or []
         why = " → ".join(str(t) for t in worst) if worst else score.get("model", "deep-seq")
+        pid = score.get("pid")
+        pid_bit = f" pid={pid}" if pid else ""
         return {
             "source": "stage8_sequence",
             "feature": "syscall_seq_deep",
@@ -104,11 +106,12 @@ def build_anomaly(self, score: dict, cfg: Any) -> dict:
             "baseline_std": None,
             "position": 0.18,
             "message": (
-                f"Deep sequence model ({score.get('model', '?')}): "
+                f"Deep sequence model ({score.get('model', '?')}){pid_bit}: "
                 f"neg_avg_logprob={neg:.2f} (warn={warn}); unlikely transition near {why}"
             ),
             "meta": {
                 "stage": 8,
+                "pid": pid,
                 "model": score.get("model"),
                 "perplexity": score.get("perplexity"),
                 "neg_avg_logprob": neg,
diff --git a/kernel_ai/ml/sequence_deep/train_markov.py b/kernel_ai/ml/sequence_deep/train_markov.py
index 75cdc32..6579f5b 100644
--- a/kernel_ai/ml/sequence_deep/train_markov.py
+++ b/kernel_ai/ml/sequence_deep/train_markov.py
@@ -58,21 +58,19 @@ def load_corpus_file(path: str | Path) -> list[list[str]]:
     return sequences
 
 
-def load_corpus_ngrams(dsn: str, *, n: int, min_count: int = 1) -> list[list[str]]:
+def load_corpus_ngrams(dsn: str, *, n: int, min_count: int = 1) -> list[tuple[list[str], int]]:
+    """Return ``(tokens, weight)`` where the weight is the observed n-gram count."""
     from kernel_ai.ml.store import fetch_ngram_counts
 
     counts = fetch_ngram_counts(dsn, n=n)
-    sequences: list[list[str]] = []
+    sequences: list[tuple[list[str], int]] = []
     for key, cnt in counts.items():
         if cnt < min_count:
             continue
         toks = [t for t in str(key).split("|") if t]
         if len(toks) < 2:
             continue
-        # Cap repeats so a hot n-gram cannot dominate the table entirely.
-        reps = min(int(cnt), 50)
-        for _ in range(reps):
-            sequences.append(toks)
+        sequences.append((toks, int(cnt)))
     return sequences
 
 
@@ -83,8 +81,8 @@ def load_corpus_synthetic(*, repeats: int = 40) -> list[list[str]]:
     return sequences
 
 
-def _count_transitions(sequences: list[list[str]]) -> int:
-    return sum(max(0, len(s) - 1) for s in sequences)
+def _count_transitions(sequences: list[tuple[list[str], int]]) -> int:
+    return sum(max(0, len(s) - 1) * w for s, w in sequences)
 
 
 def train_markov(
@@ -97,14 +95,14 @@ def train_markov(
 ) -> dict:
     """Fit Markov + encoder and persist artifact. Returns metrics dict."""
     cfg = cfg or MLConfig()
-    sequences: list[list[str]] = []
+    sequences: list[tuple[list[str], int]] = []
     source = "empty"
 
     if corpus_path:
-        sequences = load_corpus_file(corpus_path)
+        sequences = [(seq, 1) for seq in load_corpus_file(corpus_path)]
         source = f"file:{corpus_path}"
     elif use_synthetic:
-        sequences = load_corpus_synthetic()
+        sequences = [(seq, 1) for seq in load_corpus_synthetic()]
         source = "synthetic"
     elif use_ngrams:
         try:
@@ -112,7 +110,7 @@ def train_markov(
             source = "ml_syscall_ngrams"
         except Exception as exc:  # noqa: BLE001
             logger.warning("ngram corpus unavailable (%s) — falling back to synthetic", exc)
-            sequences = load_corpus_synthetic()
+            sequences = [(seq, 1) for seq in load_corpus_synthetic()]
             source = "synthetic_fallback"
 
     n_trans = _count_transitions(sequences)
@@ -124,9 +122,9 @@ def train_markov(
 
     encoder = SequenceEncoder()
     markov = MarkovScorer(order=1, meta={"stage": 8, "source": source})
-    for seq in sequences:
+    for seq, weight in sequences:
         encoder.fit(seq)
-        markov.observe(seq)
+        markov.observe(seq, weight=weight)
 
     artifact = {
         "encoder": encoder.state_dict(),
@@ -136,7 +134,14 @@ def train_markov(
     os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
     joblib.dump(artifact, out_path)
 
-    # Quick self-check: normal-ish window vs weird jump.
+    # Self-check against the corpus itself. The synthetic probes below stay for
+    # continuity, but they are meaningless for an audit-sourced model: they are made
+    # of tokens (read/write/close) the audit allowlist never emits, so they only
+    # measure the unknown-token penalty. The corpus probes compare the most and the
+    # least frequent real sequences, which is what separation actually means here.
+    ranked = sorted(sequences, key=lambda sw: sw[1], reverse=True)
+    common_score = markov.score_window(ranked[0][0]) if ranked else None
+    rare_score = markov.score_window(ranked[-1][0]) if ranked else None
     normal_score = markov.score_window(_SYNTHETIC_NORMAL[0])
     weird_score = markov.score_window(["openat", "execve", "connect", "dup2"])
     metrics = {
@@ -146,6 +151,9 @@ def train_markov(
         "vocab": len(encoder.token_to_id),
         "states": len(markov._counts),
         "path": out_path,
+        "corpus_common": "|".join(ranked[0][0]) if ranked else None,
+        "corpus_common_neg_avg_logprob": (common_score or {}).get("neg_avg_logprob"),
+        "corpus_rare_neg_avg_logprob": (rare_score or {}).get("neg_avg_logprob"),
         "normal_neg_avg_logprob": (normal_score or {}).get("neg_avg_logprob"),
         "weird_neg_avg_logprob": (weird_score or {}).get("neg_avg_logprob"),
     }
diff --git a/kernel_ai/ml/worker.py b/kernel_ai/ml/worker.py
index 2b1092f..d1c7190 100644
--- a/kernel_ai/ml/worker.py
+++ b/kernel_ai/ml/worker.py
@@ -157,6 +157,11 @@ def __init__(self, cfg: MLConfig | None = None) -> None:
         self._seq_model_mtime: float | None = None
         self._last_seq_emit = 0.0
         self._last_seq_flush = 0.0
+        # Ingest counter at the last scoring, and when evidence last arrived.
+        self._seq_scored_at = -1
+        self._seq_last_event_at = 0.0
+        self._seq_stale_logged = False
+        self._seq_pid_marks: dict[int, int] = {}
         self._seq_source = (self.cfg.seq_source or "procfs").strip().lower()
         if self.cfg.enable_stage4 and self._seq_source != "off":
             from kernel_ai.ml.sequence import NgramTracker, SyscallSampler
@@ -206,6 +211,9 @@ def __init__(self, cfg: MLConfig | None = None) -> None:
         # Stage 8 — deep sequence stub (Markov/LSTM). No-op without artifact.
         self.deep_scorer = None
         self._last_stage8_emit = 0.0
+        self._last_stage8_log = 0.0
+        self._stage8_scored_at = -1
+        self._stage8_pid_marks: dict[int, int] = {}
         if self.cfg.enable_stage8:
             from kernel_ai.ml.sequence_deep import DeepSequenceScorer
 
@@ -260,6 +268,62 @@ def _maybe_load_seq_model(self) -> None:
         except Exception as exc:  # noqa: BLE001 - keep running without Stage 4
             logger.warning("failed to load STIDE profile: %s", exc)
 
+    def _sequence_evidence_is_fresh(self, now: float) -> bool:
+        """True when syscalls arrived since the last time a window was scored.
+
+        The rolling windows outlive their source: when the stream stops they keep
+        their last contents, so re-scoring them produces the same verdict forever —
+        one anomaly per cooldown, indefinitely. That is what happened while the
+        kernel audit switch was off on 2026-08-11: for 30 hours with zero events
+        Stage 4 reported a steady 120 anomalies an hour about a frozen window.
+        """
+        ingested = self.seq_tracker.ingested
+        if self._seq_last_event_at == 0.0:
+            self._seq_last_event_at = now
+        if ingested != self._seq_scored_at:
+            self._seq_scored_at = ingested
+            self._seq_last_event_at = now
+            if self._seq_stale_logged:
+                logger.info("syscall stream resumed (source=%s)", self._seq_source)
+                self._seq_stale_logged = False
+            return True
+        silence = now - self._seq_last_event_at
+        if silence >= self.cfg.seq_stale_warn_sec and not self._seq_stale_logged:
+            # Said once per outage, and filebeat ships it: a silent feed should
+            # look like a problem rather than like a calm system.
+            logger.warning(
+                "syscall stream silent for %.0fs (source=%s): sequence scoring "
+                "paused until events resume",
+                silence,
+                self._seq_source,
+            )
+            self._seq_stale_logged = True
+        return False
+
+    def _fresh_pid_windows(
+        self,
+        entries: list[tuple[int, list[str]]],
+        marks: dict[int, int],
+        stamps: dict[int, int],
+    ) -> list[tuple[int, list[str]]]:
+        """Keep only pid windows that changed since this scorer last looked.
+
+        The per-pid twin of the stream guard. A window outlives its process, so a
+        short-lived pid that ended on an odd chain stays the loudest thing on the
+        host and gets re-reported every cooldown — the same anomaly, forever, about
+        a process that no longer exists.
+        """
+        fresh: list[tuple[int, list[str]]] = []
+        for pid, window in entries:
+            stamp = stamps.get(pid)
+            if stamp is None or marks.get(pid) == stamp:
+                continue
+            marks[pid] = stamp
+            fresh.append((pid, window))
+        for gone in [pid for pid in marks if pid not in stamps]:
+            marks.pop(gone, None)
+        return fresh
+
     def _tick_sequence(self) -> dict | None:
         """Ingest syscalls, grow the n-gram vocabulary, and score the window."""
         if self.seq_tracker is None:
@@ -289,16 +353,23 @@ def _tick_sequence(self) -> dict | None:
             pending = self.seq_tracker.drain_pending()
             if pending:
                 self.store.upsert_ngram_counts(self.cfg.seq_n, pending)
+            self.seq_tracker.evict_idle(older_than=self.cfg.seq_pid_idle_events)
             self._last_seq_flush = now
 
         if self.seq_model is None:
             return None
 
+        if not self._sequence_evidence_is_fresh(now):
+            return None
+
         # Prefer per-pid windows (classic STIDE): high-volume connect/setuid
         # spam from one process must not dilute a hostile chain on another.
         best: tuple[float, int, list[str], int | None] | None = None
-        for pid, window in self.seq_tracker.recent_by_pid(
-            min_len=self.cfg.seq_pid_min_window
+        stamps = self.seq_tracker.pid_stamps()
+        for pid, window in self._fresh_pid_windows(
+            self.seq_tracker.recent_by_pid(min_len=self.cfg.seq_pid_min_window),
+            self._seq_pid_marks,
+            stamps,
         ):
             mismatch, misses = self.seq_model.score_window(window)
             if best is None or mismatch > best[0]:
@@ -328,17 +399,58 @@ def _tick_stage8(self) -> dict | None:
         self.deep_scorer.maybe_reload()
         if not self.deep_scorer.ready:
             return None
-        tokens = self.seq_tracker.recent_tokens()
-        if len(tokens) < max(8, self.cfg.stage8_window // 4):
-            return None
-        tokens = tokens[-self.cfg.stage8_window :]
-        score = self.deep_scorer.score_tokens(tokens)
-        if not score:
+        # Same frozen-window trap as Stage 4, and Stage 8 keeps its own mark because
+        # both stages read the tracker in the same tick.
+        ingested = self.seq_tracker.ingested
+        if ingested == self._stage8_scored_at:
             return None
-        neg = float(score.get("neg_avg_logprob") or score.get("perplexity") or 0.0)
+        self._stage8_scored_at = ingested
+
+        # Score per pid and keep the least likely one; the global mix is only a
+        # fallback for hosts where no single pid has enough history yet.
+        min_tokens = max(8, self.cfg.stage8_window // 4)
+        best: tuple[float, dict, int | None] | None = None
+        for pid, tokens in self._fresh_pid_windows(
+            self.seq_tracker.recent_tokens_by_pid(min_len=min_tokens),
+            self._stage8_pid_marks,
+            self.seq_tracker.pid_stamps(),
+        ):
+            score = self.deep_scorer.score_tokens(tokens[-self.cfg.stage8_window :])
+            if not score:
+                continue
+            neg = float(score.get("neg_avg_logprob") or score.get("perplexity") or 0.0)
+            if best is None or neg > best[0]:
+                best = (neg, score, pid)
+        if best is None:
+            tokens = self.seq_tracker.recent_tokens()
+            if len(tokens) < min_tokens:
+                return None
+            score = self.deep_scorer.score_tokens(tokens[-self.cfg.stage8_window :])
+            if not score:
+                return None
+            best = (
+                float(score.get("neg_avg_logprob") or score.get("perplexity") or 0.0),
+                score,
+                None,
+            )
+        neg, score, pid = best
+        score["pid"] = pid
+        now = time.time()
+        # A heartbeat of the score itself: thresholds for a fresh host should be read
+        # off the live distribution, not guessed, and later it shows the model is
+        # still scoring rather than quietly dormant.
+        if (now - self._last_stage8_log) >= self.cfg.stage8_log_every_sec:
+            self._last_stage8_log = now
+            logger.info(
+                "Stage 8 window score: pid=%s neg_avg_logprob=%.3f (warn=%.2f crit=%.2f) worst=%s",
+                pid,
+                neg,
+                self.cfg.stage8_score_warn,
+                self.cfg.stage8_score_crit,
+                ",".join(str(t) for t in (score.get("worst_tokens") or [])[:3]),
+            )
         if neg < self.cfg.stage8_score_warn:
             return None
-        now = time.time()
         if (now - self._last_stage8_emit) < self.cfg.stage8_cooldown_sec:
             return None
         self._last_stage8_emit = now
diff --git a/tests/test_api_contracts.py b/tests/test_api_contracts.py
index d3b3268..02e0ea8 100644
--- a/tests/test_api_contracts.py
+++ b/tests/test_api_contracts.py
@@ -2,6 +2,8 @@
 
 import os
 
+import pytest
+
 from kernel_ai.contracts.api_contracts import (
     validate_crypto_realtime_response,
     validate_execution_context_response,
@@ -86,3 +88,52 @@ def test_isolation_context_contract():
     resp = _client().get("/api/isolation-context")
     assert resp.status_code == 200
     validate_isolation_context_response(resp.get_json())
+
+
+def _filesystem_blocks_payload():
+    return {
+        "timestamp": "2026-08-12T10:00:00",
+        "mounts": [{"mountpoint": "/", "used_percent": 41.0}],
+        "devices": [{"name": "vda", "write_bps": 0.0, "read_bps": 0.0}],
+        "writepath": {"stages": [], "hot": "block"},
+        "writeback": {"dirty_mb": 0.4, "writeback_mb": 0.0},
+        "io_scheduler": {"device": "vda", "scheduler": "none"},
+        "meta": {"used_percent": 41.0, "mount_count": 1},
+    }
+
+
+def test_filesystem_blocks_contract_allows_missing_io_scheduler():
+    # A box with no real block device reports no scheduler, and that is valid.
+    payload = _filesystem_blocks_payload()
+    payload["io_scheduler"] = None
+    validate_filesystem_blocks_response(payload)
+
+
+def _drop(payload, *path):
+    target = payload
+    for step in path[:-1]:
+        target = target[step]
+    del target[path[-1]]
+    return payload
+
+
+def _stale_grid_payload():
+    # The grid shape this endpoint served before the write-path rewrite.
+    return {"timestamp": "2026-08-12T10:00:00", "rows": 4, "cols": 4,
+            "zones": [], "blocks": [], "meta": {}}
+
+
+@pytest.mark.parametrize(
+    "payload, expected",
+    [
+        (_stale_grid_payload(), "missing key 'mounts'"),
+        (_drop(_filesystem_blocks_payload(), "mounts"), "missing key 'mounts'"),
+        (_drop(_filesystem_blocks_payload(), "writeback"), "missing key 'writeback'"),
+        (_drop(_filesystem_blocks_payload(), "writepath", "hot"), r"writepath: missing key 'hot'"),
+        (_drop(_filesystem_blocks_payload(), "meta", "mount_count"),
+         r"meta: missing key 'mount_count'"),
+    ],
+)
+def test_filesystem_blocks_contract_rejects_drift(payload, expected):
+    with pytest.raises(ValueError, match=expected):
+        validate_filesystem_blocks_response(payload)
diff --git a/tests/test_core_observability_subsystems.py b/tests/test_core_observability_subsystems.py
new file mode 100644
index 0000000..1b2ea35
--- /dev/null
+++ b/tests/test_core_observability_subsystems.py
@@ -0,0 +1,107 @@
+"""Subsystem load must report measured values, never padded ones.
+
+The panel these numbers feed used to floor CPU at 50% and pin I/O wait to 100%,
+so the assertions below are deliberately about the arithmetic and about the
+warming state, not merely about the shape of the payload.
+"""
+
+from kernel_ai.services import core_observability as svc
+
+
+def _reset_baseline():
+    svc._SUBSYSTEM_PREV.update({"ts": None, "cpu": None, "disk": None, "net": None, "net_peak": 0.0})
+
+
+def _stub_proc(monkeypatch, cpu_times, memory=(4 * 1024**3, 8 * 1024**3)):
+    monkeypatch.setattr(svc.platform, "system", lambda: "Linux")
+    monkeypatch.setattr(svc, "_read_cpu_times", lambda: cpu_times)
+    monkeypatch.setattr(svc, "_read_memory_used", lambda: memory)
+    monkeypatch.setattr(svc, "_read_loadavg", lambda: (3, [1.5, 1.25, 1.0]))
+    monkeypatch.setattr(svc, "_read_tcp_inuse", lambda: 31)
+    monkeypatch.setattr(svc, "_read_mount_count", lambda: 52)
+    monkeypatch.setattr(svc.psutil, "disk_io_counters", lambda: None)
+    monkeypatch.setattr(svc.psutil, "net_io_counters", lambda: None)
+
+
+def test_first_call_marks_rates_warming_instead_of_inventing_them(monkeypatch):
+    _reset_baseline()
+    # user, nice, system, idle, iowait
+    _stub_proc(monkeypatch, [1000, 0, 500, 8000, 500])
+
+    out = svc.get_kernel_subsystem_status()
+
+    for key in ("process_scheduler", "file_system", "network_stack"):
+        assert out[key]["warming"] is True, key
+        assert out[key]["value"] is None, key
+    # Memory is a level, not a rate, so it is answerable straight away.
+    assert out["memory_management"]["warming"] is False
+    assert out["memory_management"]["value"] == 50.0
+
+
+def test_cpu_and_iowait_come_from_the_delta_between_polls(monkeypatch):
+    _reset_baseline()
+    _stub_proc(monkeypatch, [1000, 0, 500, 8000, 500])
+    svc.get_kernel_subsystem_status()
+
+    # Over the next interval: 70 busy, 20 idle, 10 waiting on I/O out of 100.
+    _stub_proc(monkeypatch, [1060, 0, 510, 8020, 510])
+    out = svc.get_kernel_subsystem_status()
+
+    assert out["process_scheduler"]["value"] == 70.0
+    assert out["process_scheduler"]["usage"] == 70
+    assert out["file_system"]["value"] == 10.0
+    assert out["process_scheduler"]["warming"] is False
+
+
+def test_idle_machine_reports_idle_rather_than_a_floor(monkeypatch):
+    _reset_baseline()
+    _stub_proc(monkeypatch, [1000, 0, 500, 8000, 500])
+    svc.get_kernel_subsystem_status()
+
+    # Nothing but idle jiffies accumulated between the two polls.
+    _stub_proc(monkeypatch, [1000, 0, 500, 8100, 500])
+    out = svc.get_kernel_subsystem_status()
+
+    assert out["process_scheduler"]["value"] == 0.0
+    assert out["process_scheduler"]["usage"] == 0
+    assert out["file_system"]["value"] == 0.0
+    assert out["file_system"]["usage"] == 0
+
+
+def test_scheduler_row_carries_the_load_average(monkeypatch):
+    _reset_baseline()
+    _stub_proc(monkeypatch, [1000, 0, 500, 8000, 500])
+    out = svc.get_kernel_subsystem_status()
+
+    assert out["process_scheduler"]["load"] == [1.5, 1.25, 1.0]
+    assert out["process_scheduler"]["detail"] == 3
+    assert out["process_scheduler"]["detail_unit"] == "runnable"
+
+
+def test_network_throughput_is_scaled_against_the_busiest_second_seen(monkeypatch):
+    class Net:
+        def __init__(self, sent, recv):
+            self.bytes_sent = sent
+            self.bytes_recv = recv
+
+    _reset_baseline()
+    _stub_proc(monkeypatch, [1000, 0, 500, 8000, 500])
+    monkeypatch.setattr(svc.psutil, "net_io_counters", lambda: Net(0, 0))
+    monkeypatch.setattr(svc.time, "time", lambda: 1000.0)
+    svc.get_kernel_subsystem_status()
+
+    # 4 MiB moved over one second becomes both the reading and the full-scale mark.
+    monkeypatch.setattr(svc.psutil, "net_io_counters", lambda: Net(2 * 1024**2, 2 * 1024**2))
+    monkeypatch.setattr(svc.time, "time", lambda: 1001.0)
+    peak = svc.get_kernel_subsystem_status()["network_stack"]
+    assert peak["value"] == 4 * 1024**2
+    assert peak["usage"] == 100
+    assert peak["detail"] == 31
+    assert peak["detail_unit"] == "sockets"
+
+    # A quiet second afterwards must read as quiet, not as full scale.
+    monkeypatch.setattr(svc.psutil, "net_io_counters", lambda: Net(2 * 1024**2, 2 * 1024**2))
+    monkeypatch.setattr(svc.time, "time", lambda: 1002.0)
+    quiet = svc.get_kernel_subsystem_status()["network_stack"]
+    assert quiet["value"] == 0
+    assert quiet["usage"] == 0
diff --git a/tests/test_gunicorn_conf.py b/tests/test_gunicorn_conf.py
new file mode 100644
index 0000000..421b565
--- /dev/null
+++ b/tests/test_gunicorn_conf.py
@@ -0,0 +1,44 @@
+"""The child_exit hook runs on every worker exit, including the ones systemd
+triggers on restart, so anything raised here lands in the journal of a healthy
+box and looks like a crash."""
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+_CONF = Path(__file__).resolve().parents[1] / "gunicorn.conf.py"
+
+
+def _load():
+    spec = importlib.util.spec_from_file_location("gunicorn_conf", _CONF)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class _Worker:
+    pid = 4242
+
+
+def test_no_multiproc_dir_means_nothing_to_clean_up(monkeypatch):
+    # prometheus_client joins the directory with the pid, so an unset variable
+    # used to raise TypeError on the None path.
+    monkeypatch.delenv("PROMETHEUS_MULTIPROC_DIR", raising=False)
+    _load().child_exit(None, _Worker())
+
+
+def test_a_blank_multiproc_dir_counts_as_unset(monkeypatch):
+    monkeypatch.setenv("PROMETHEUS_MULTIPROC_DIR", "   ")
+    _load().child_exit(None, _Worker())
+
+
+def test_configured_multiproc_dir_still_reaps_the_worker(monkeypatch, tmp_path):
+    multiprocess = pytest.importorskip("prometheus_client.multiprocess")
+    monkeypatch.setenv("PROMETHEUS_MULTIPROC_DIR", str(tmp_path))
+    reaped = []
+    monkeypatch.setattr(multiprocess, "mark_process_dead", reaped.append)
+
+    _load().child_exit(None, _Worker())
+
+    assert reaped == [_Worker.pid]
diff --git a/tests/test_kernel_maps_service.py b/tests/test_kernel_maps_service.py
index 7eecb7a..98e0069 100644
--- a/tests/test_kernel_maps_service.py
+++ b/tests/test_kernel_maps_service.py
@@ -3,9 +3,50 @@
 from kernel_ai.services import kernel_maps as svc
 
 
-def test_syscall_names_contains_common_entries():
-    assert svc.SYSCALL_NAMES[0] == "read"
-    assert svc.SYSCALL_NAMES[41] == "socket"
+def test_bundled_table_is_the_x86_64_numbering():
+    assert svc.SYSCALL_NAMES_X86_64[0] == "read"
+    assert svc.SYSCALL_NAMES_X86_64[41] == "socket"
+
+
+def test_running_table_comes_from_auditd(monkeypatch):
+    svc.get_syscall_names.cache_clear()
+    monkeypatch.setattr(svc.shutil, "which", lambda _name: "/usr/bin/ausyscall")
+    monkeypatch.setattr(
+        svc.subprocess, "run",
+        lambda *_a, **_k: type("R", (), {"stdout": "Using aarch64 syscall table:\n0\tio_setup\n73\tppoll\n"})()
+    )
+    table = svc.get_syscall_names()
+    # The same number means flock on x86_64: borrowing that name here would
+    # mislabel every row on arm64.
+    assert table == {0: "io_setup", 73: "ppoll"}
+    svc.get_syscall_names.cache_clear()
+
+
+def test_bundled_table_is_used_only_on_x86_64(monkeypatch):
+    svc.get_syscall_names.cache_clear()
+    monkeypatch.setattr(svc.shutil, "which", lambda _name: None)
+    monkeypatch.setattr(svc.platform, "machine", lambda: "x86_64")
+    assert svc.get_syscall_names()[73] == "flock"
+
+    svc.get_syscall_names.cache_clear()
+    monkeypatch.setattr(svc.platform, "machine", lambda: "aarch64")
+    # Nothing trustworthy left: callers show syscall_<number> instead of a name
+    # taken from another architecture.
+    assert svc.get_syscall_names() == {}
+    svc.get_syscall_names.cache_clear()
+
+
+def test_unreadable_auditd_output_does_not_break_resolution(monkeypatch):
+    svc.get_syscall_names.cache_clear()
+    monkeypatch.setattr(svc.shutil, "which", lambda _name: "/usr/bin/ausyscall")
+
+    def _boom(*_a, **_k):
+        raise OSError("no exec")
+
+    monkeypatch.setattr(svc.subprocess, "run", _boom)
+    monkeypatch.setattr(svc.platform, "machine", lambda: "x86_64")
+    assert svc.get_syscall_names()[0] == "read"
+    svc.get_syscall_names.cache_clear()
 
 
 def test_map_syscall_to_subsystem():
diff --git a/tests/test_ml_collectors.py b/tests/test_ml_collectors.py
index 46f5038..b4c655b 100644
--- a/tests/test_ml_collectors.py
+++ b/tests/test_ml_collectors.py
@@ -1,5 +1,7 @@
 """Tests for Stage 6 syscall event contract + n-gram stream ingest."""
 
+from types import SimpleNamespace
+
 from kernel_ai.ml.collectors.base import (
     AUDIT_ARCH_AARCH64,
     AUDIT_ARCH_X86_64,
@@ -83,3 +85,126 @@ def test_ngram_tracker_recent_by_pid_not_diluted():
     global_m, _ = model.score_window(tracker.recent())
     pid_m, _ = model.score_window(by_pid[2])
     assert pid_m > global_m
+
+
+def _events(pid, names, start=0.0):
+    return [
+        SyscallEvent(ts=start + i * 0.01, pid=pid, uid=0, comm="x", syscall=name)
+        for i, name in enumerate(names)
+    ]
+
+
+def test_tracker_counts_every_ingested_sample():
+    tracker = NgramTracker(n=3, window=50)
+    assert tracker.ingested == 0
+    tracker.update_stream(_events(7, ["clone", "openat", "execve"]))
+    assert tracker.ingested == 3
+    # An empty drain is what a dead feed looks like: nothing new to score.
+    tracker.update_stream([])
+    assert tracker.ingested == 3
+    tracker.update({9: "connect"})
+    assert tracker.ingested == 4
+
+
+def test_demo_events_are_scored_but_never_learned():
+    """A wiring demo must not be able to teach the profile that ptrace is normal."""
+    tracker = NgramTracker(n=3, window=50)
+    tracker.update_stream(
+        [
+            SyscallEvent(ts=i * 0.01, pid=99, uid=0, comm="novel", syscall=name)
+            for i, name in enumerate(["ptrace", "memfd_create", "userfaultfd", "connect"])
+        ]
+    )
+    assert "ptrace|memfd_create|userfaultfd" in tracker.recent()
+    assert tracker.drain_pending() == {}
+
+    tracker.update_stream(_events(7, ["clone", "openat", "execve"]))
+    assert tracker.drain_pending() == {"clone|openat|execve": 1}
+
+
+def _guard_worker(stale_after=300.0):
+    """A worker stripped to the fields the freshness guard touches (no DB)."""
+    from kernel_ai.ml.worker import MLWorker
+
+    worker = MLWorker.__new__(MLWorker)
+    worker.cfg = SimpleNamespace(seq_stale_warn_sec=stale_after)
+    worker.seq_tracker = NgramTracker(n=3, window=50)
+    worker._seq_scored_at = -1
+    worker._seq_last_event_at = 0.0
+    worker._seq_stale_logged = False
+    worker._seq_source = "socket"
+    return worker
+
+
+def test_a_frozen_window_is_scored_only_once():
+    worker = _guard_worker()
+    worker.seq_tracker.update_stream(_events(7, ["clone", "openat", "execve"]))
+
+    assert worker._sequence_evidence_is_fresh(1000.0) is True
+    # The stream stopped. The window still holds those three syscalls, and before
+    # the guard it was re-reported once per cooldown for as long as the feed was down.
+    assert worker._sequence_evidence_is_fresh(1002.0) is False
+    assert worker._sequence_evidence_is_fresh(1004.0) is False
+
+    worker.seq_tracker.update_stream(_events(7, ["connect"], start=5.0))
+    assert worker._sequence_evidence_is_fresh(1006.0) is True
+
+
+def test_a_dead_pid_window_is_not_reported_again():
+    """A short-lived pid that ended on an odd chain must not be re-reported forever."""
+    worker = _guard_worker()
+    tracker = worker.seq_tracker
+    tracker.update_stream(_events(42, ["ptrace", "memfd_create", "execve", "connect"]))
+    tracker.update_stream(_events(7, ["accept4", "accept4", "accept4"], start=1.0))
+
+    marks: dict[int, int] = {}
+    first = worker._fresh_pid_windows(
+        tracker.recent_by_pid(min_len=1), marks, tracker.pid_stamps()
+    )
+    assert {pid for pid, _ in first} == {42, 7}
+
+    # The live pid keeps working, the dead one does not: only the live one returns.
+    tracker.update_stream(_events(7, ["accept4"], start=2.0))
+    second = worker._fresh_pid_windows(
+        tracker.recent_by_pid(min_len=1), marks, tracker.pid_stamps()
+    )
+    assert [pid for pid, _ in second] == [7]
+
+    # Nothing new at all: nobody is scored.
+    assert worker._fresh_pid_windows(
+        tracker.recent_by_pid(min_len=1), marks, tracker.pid_stamps()
+    ) == []
+
+
+def test_idle_pids_are_evicted():
+    tracker = NgramTracker(n=3, window=50)
+    tracker.update_stream(_events(42, ["ptrace", "memfd_create", "execve"]))
+    tracker.update_stream(_events(7, ["accept4"] * 20, start=1.0))
+
+    assert tracker.evict_idle(older_than=10) == 1
+    assert set(tracker.pid_stamps()) == {7}
+    assert [pid for pid, _ in tracker.recent_by_pid(min_len=1)] == [7]
+
+
+def test_a_silent_stream_is_reported_once_and_on_recovery(caplog):
+    worker = _guard_worker(stale_after=300.0)
+    worker.seq_tracker.update_stream(_events(7, ["clone", "openat", "execve"]))
+    worker._sequence_evidence_is_fresh(1000.0)
+
+    with caplog.at_level("INFO", logger="kernel_ai.ml.worker"):
+        # Inside the grace period the silence is not worth a line yet.
+        worker._sequence_evidence_is_fresh(1200.0)
+        assert not caplog.records
+
+        worker._sequence_evidence_is_fresh(1400.0)
+        assert [r.levelname for r in caplog.records] == ["WARNING"]
+        assert "silent for 400s" in caplog.records[0].getMessage()
+
+        # Still silent: one line per outage, not one per tick.
+        worker._sequence_evidence_is_fresh(1500.0)
+        assert len(caplog.records) == 1
+
+        worker.seq_tracker.update_stream(_events(7, ["connect"], start=9.0))
+        worker._sequence_evidence_is_fresh(1502.0)
+        assert [r.levelname for r in caplog.records] == ["WARNING", "INFO"]
+        assert "resumed" in caplog.records[1].getMessage()
diff --git a/tests/test_ml_stage8.py b/tests/test_ml_stage8.py
index b7decfd..f46e51b 100644
--- a/tests/test_ml_stage8.py
+++ b/tests/test_ml_stage8.py
@@ -105,6 +105,82 @@ def test_build_anomaly_contract():
     assert anom["severity"] == "medium"
 
 
+def test_observe_weight_shapes_the_distribution():
+    """A chain seen 1000x must not look as likely as one seen 3x."""
+    hot = MarkovScorer()
+    hot.observe(["accept4", "accept4", "accept4"], weight=1000)
+    hot.observe(["accept4", "execve", "connect"], weight=3)
+
+    common = hot.score_window(["accept4", "accept4", "accept4"])
+    rare = hot.score_window(["accept4", "execve", "connect"])
+    assert rare["neg_avg_logprob"] > common["neg_avg_logprob"] * 2
+
+    # Same two chains without weights: the model believes both are ordinary.
+    flat = MarkovScorer()
+    flat.observe(["accept4", "accept4", "accept4"])
+    flat.observe(["accept4", "execve", "connect"])
+    flat_common = flat.score_window(["accept4", "accept4", "accept4"])
+    flat_rare = flat.score_window(["accept4", "execve", "connect"])
+    assert abs(flat_rare["neg_avg_logprob"] - flat_common["neg_avg_logprob"]) < 0.2
+
+
+def test_ngram_corpus_carries_counts_as_weights(monkeypatch):
+    from kernel_ai.ml.sequence_deep import train_markov as tm
+
+    monkeypatch.setattr(
+        "kernel_ai.ml.store.fetch_ngram_counts",
+        lambda dsn, n: {"accept4|accept4|accept4": 2500, "clone|execve|connect": 4},
+    )
+    corpus = tm.load_corpus_ngrams("dsn", n=3, min_count=3)
+    assert sorted(w for _, w in corpus) == [4, 2500]
+    assert tm._count_transitions(corpus) == 2 * 2500 + 2 * 4
+
+
+def test_per_pid_token_windows_isolate_a_short_chain():
+    """A hostile burst on one pid must be scored on its own, not averaged into noise."""
+    from kernel_ai.ml.collectors.base import SyscallEvent
+    from kernel_ai.ml.sequence import NgramTracker
+
+    tracker = NgramTracker(n=3, window=200)
+    tracker.update_stream(
+        [SyscallEvent(ts=i * 0.01, pid=1, uid=0, comm="nginx", syscall="accept4") for i in range(300)]
+        + [
+            SyscallEvent(ts=100 + i * 0.01, pid=2, uid=0, comm="evil", syscall=name)
+            for i, name in enumerate(["memfd_create", "execve", "connect", "ptrace"] * 5)
+        ]
+    )
+    by_pid = dict(tracker.recent_tokens_by_pid(min_len=8))
+    assert set(by_pid) == {1, 2}
+    assert set(by_pid[1]) == {"accept4"}
+    assert "ptrace" in by_pid[2]
+
+    m = MarkovScorer()
+    m.observe(["accept4"] * 50, weight=1000)
+    hostile = m.score_window(by_pid[2])["neg_avg_logprob"]
+    mixed = m.score_window(tracker.recent_tokens())["neg_avg_logprob"]
+    assert hostile > mixed
+
+
+def test_nightly_retrain_refreshes_markov_only_when_stage8_is_on(monkeypatch):
+    from kernel_ai.ml import retrain
+
+    calls = []
+    monkeypatch.setattr(
+        "kernel_ai.ml.sequence_deep.train_markov.train_markov",
+        lambda cfg, **kw: calls.append(kw) or {"n_transitions": 1234, "vocab": 21},
+    )
+
+    off = SimpleNamespace(enable_stage8=False, stage8_backend="markov")
+    retrain._refresh_markov(off, {})
+    assert calls == []
+
+    metrics: dict = {}
+    on = SimpleNamespace(enable_stage8=True, stage8_backend="markov")
+    retrain._refresh_markov(on, metrics)
+    assert calls == [{"use_ngrams": True}]
+    assert metrics["stage8_transitions"] == 1234
+
+
 def test_corpus_file_train(tmp_path):
     corpus = tmp_path / "norm.txt"
     corpus.write_text(
diff --git a/tests/test_process_inspect_service.py b/tests/test_process_inspect_service.py
index 7197a7d..4812f3a 100644
--- a/tests/test_process_inspect_service.py
+++ b/tests/test_process_inspect_service.py
@@ -62,3 +62,120 @@ def connections(self):
     descriptors = out["descriptors"]
     assert [item["fd"] for item in descriptors] == [0, 1, 2, 7, 19]
     assert [item["type"] for item in descriptors] == ["stdin", "stdout", "stderr", "socket", "pipe"]
+
+
+class _FakeAncestor:
+    """Minimal psutil.Process stand-in for lineage walking."""
+
+    def __init__(self, pid, name, create_time, parent=None):
+        self.pid = pid
+        self._name = name
+        self._create_time = create_time
+        self._parent = parent
+
+    def as_dict(self, _fields):
+        return {
+            "pid": self.pid,
+            "name": self._name,
+            "create_time": self._create_time,
+            "status": "sleeping",
+            "username": "alex",
+        }
+
+    def cmdline(self):
+        return ["/usr/bin/" + self._name]
+
+    def name(self):
+        return self._name
+
+    def parent(self):
+        return self._parent
+
+    def children(self):
+        return []
+
+
+def test_get_process_lineage_orders_oldest_first(monkeypatch):
+    init = _FakeAncestor(1, "systemd", 1000.0)
+    shell = _FakeAncestor(50, "bash", 2000.0, parent=init)
+    leaf = _FakeAncestor(900, "python3", 3000.0, parent=shell)
+
+    monkeypatch.setattr(svc.psutil, "Process", lambda _pid: leaf)
+    monkeypatch.setattr(svc.psutil, "boot_time", lambda: 999.0)
+
+    out = svc.get_process_lineage_info(900)
+    assert [row["pid"] for row in out["chain"]] == [1, 50, 900]
+    assert [row["name"] for row in out["chain"]] == ["systemd", "bash", "python3"]
+    assert out["depth"] == 3
+    assert out["truncated"] is False
+    assert out["chain"][0]["create_time"] == 1000.0
+
+
+def test_get_process_lineage_breaks_parent_cycle(monkeypatch):
+    a = _FakeAncestor(10, "a", 1000.0)
+    b = _FakeAncestor(11, "b", 1100.0, parent=a)
+    a._parent = b  # pathological /proc race: parent chain loops back
+
+    monkeypatch.setattr(svc.psutil, "Process", lambda _pid: b)
+    monkeypatch.setattr(svc.psutil, "boot_time", lambda: 999.0)
+
+    out = svc.get_process_lineage_info(11)
+    assert [row["pid"] for row in out["chain"]] == [10, 11]
+
+
+def test_activity_counters_expose_deltas_source(monkeypatch, tmp_path):
+    class _Times:
+        user = 1.25
+        system = 0.5
+
+    class _FakeProc:
+        def cpu_times(self):
+            return _Times()
+
+    status = tmp_path / "status"
+    status.write_text(
+        "Name:\tnginx\nThreads:\t4\n"
+        "voluntary_ctxt_switches:\t120\n"
+        "nonvoluntary_ctxt_switches:\t7\n"
+    )
+    io_file = tmp_path / "io"
+    io_file.write_text("read_bytes: 4096\nwrite_bytes: 8192\n")
+
+    real_open = open
+
+    def fake_open(path, *args, **kwargs):
+        if path == "/proc/77/status":
+            return real_open(status, *args, **kwargs)
+        if path == "/proc/77/io":
+            return real_open(io_file, *args, **kwargs)
+        return real_open(path, *args, **kwargs)
+
+    monkeypatch.setattr(svc.psutil, "Process", lambda _pid: _FakeProc())
+    monkeypatch.setattr("builtins.open", fake_open)
+
+    out = svc.get_process_activity_counters(77)
+    assert out["ctx_voluntary"] == 120
+    assert out["ctx_nonvoluntary"] == 7
+    assert out["num_threads"] == 4
+    assert out["cpu_user"] == 1.25
+    assert out["read_bytes"] == 4096
+    assert out["io_readable"] is True
+    assert out["ts"] > 0
+
+
+def test_activity_counters_survive_unreadable_io(monkeypatch):
+    class _FakeProc:
+        def cpu_times(self):
+            raise svc.psutil.AccessDenied(77)
+
+    def fake_open(path, *args, **kwargs):
+        raise PermissionError(path)
+
+    monkeypatch.setattr(svc.psutil, "Process", lambda _pid: _FakeProc())
+    monkeypatch.setattr("builtins.open", fake_open)
+
+    out = svc.get_process_activity_counters(77)
+    assert out["io_readable"] is False
+    assert out["read_bytes"] is None
+    assert out["cpu_user"] is None
+    assert out["ctx_voluntary"] is None
diff --git a/tests/test_syscall_anatomy_service.py b/tests/test_syscall_anatomy_service.py
new file mode 100644
index 0000000..6104fe5
--- /dev/null
+++ b/tests/test_syscall_anatomy_service.py
@@ -0,0 +1,119 @@
+"""Tests for ``kernel_ai.services.syscall_anatomy``."""
+
+import json
+
+from kernel_ai.services import syscall_anatomy as sa
+
+
+def _pretend_x86(monkeypatch, symbols):
+    monkeypatch.setattr(sa.platform, "machine", lambda: "x86_64")
+    monkeypatch.setattr(sa, "kernel_symbols", lambda: frozenset(symbols))
+
+
+def test_the_chain_follows_the_call_from_userspace_to_where_it_sleeps(monkeypatch):
+    _pretend_x86(monkeypatch, {
+        "entry_SYSCALL_64", "do_syscall_64", "__x64_sys_epoll_wait", "do_epoll_wait", "ep_poll",
+    })
+    out = sa.describe("epoll_wait", nr=232, subsystem="net", wchans=[("ep_poll", 11)], sampled=12)
+    assert [c["symbol"] for c in out["chain"]] == [
+        "epoll_wait()",
+        "entry_SYSCALL_64",
+        "do_syscall_64",
+        "__x64_sys_epoll_wait",
+        "do_epoll_wait",
+        "ep_poll",
+    ]
+    assert out["chain"][-1]["stage"] == "sleep"
+    # The tally covers the sample, not everyone parked in a busy call.
+    assert out["chain"][-1]["note"] == "11 of 12 sampled"
+    assert out["abi"].startswith("nr in rax")
+
+
+def test_a_symbol_this_kernel_does_not_have_is_left_out(monkeypatch):
+    # do_epoll_wait is inlined on some builds; the card must not claim it.
+    _pretend_x86(monkeypatch, {"entry_SYSCALL_64", "do_syscall_64", "__x64_sys_epoll_wait"})
+    out = sa.describe("epoll_wait", nr=232, subsystem="net", wchans=[])
+    assert [c["symbol"] for c in out["chain"]] == [
+        "epoll_wait()", "entry_SYSCALL_64", "do_syscall_64", "__x64_sys_epoll_wait",
+    ]
+
+
+def test_the_sleeping_function_marks_a_symbol_already_on_the_chain(monkeypatch):
+    _pretend_x86(monkeypatch, {"entry_SYSCALL_64", "do_syscall_64", "__x64_sys_select", "do_select"})
+    out = sa.describe("select", nr=23, subsystem="fs", wchans=[("do_select", 4)])
+    sleeping = [c for c in out["chain"] if c["stage"] == "sleep"]
+    assert [c["symbol"] for c in sleeping] == ["do_select"]
+    # …and it is not repeated further down the chain.
+    assert [c["symbol"] for c in out["chain"]].count("do_select") == 1
+
+
+def test_an_undocumented_call_still_gets_its_number_and_its_chain(monkeypatch):
+    _pretend_x86(monkeypatch, {"entry_SYSCALL_64", "do_syscall_64", "__x64_sys_landlock_add_rule"})
+    out = sa.describe("landlock_add_rule", nr=445, subsystem="kernel", wchans=[])
+    assert out["nr"] == 445
+    assert out["signature"] == ""
+    assert "__x64_sys_landlock_add_rule" in [c["symbol"] for c in out["chain"]]
+
+
+def test_userspace_is_the_one_stage_no_table_can_confirm(monkeypatch):
+    _pretend_x86(monkeypatch, {"entry_SYSCALL_64", "do_syscall_64"})
+    out = sa.describe("read", nr=0, subsystem="fs", wchans=[])
+    assert out["chain"][0]["confirmed"] is False
+    assert all(c["confirmed"] for c in out["chain"][1:])
+
+
+def test_without_kallsyms_the_chain_keeps_only_what_it_can_stand_behind(monkeypatch):
+    _pretend_x86(monkeypatch, set())
+    out = sa.describe("read", nr=0, subsystem="fs", wchans=[("pipe_read", 2)])
+    assert out["symbols_confirmed"] is False
+    # The entry and dispatch stages are architecture facts, and the sleeping
+    # function came from the kernel itself; the handler symbol did not.
+    assert [c["symbol"] for c in out["chain"]] == [
+        "read()", "entry_SYSCALL_64", "do_syscall_64", "pipe_read",
+    ]
+
+
+def _forget_symbols(monkeypatch):
+    monkeypatch.setitem(sa._SYMBOL_CACHE, "names", None)
+
+
+def test_the_published_set_is_used_when_kallsyms_is_walled_off(tmp_path, monkeypatch):
+    # ProtectKernelTunables=yes closes /proc/kallsyms to the backend; the root
+    # collector reads it and leaves the names where the backend can get them.
+    _forget_symbols(monkeypatch)
+    path = tmp_path / "ksyms.json"
+    path.write_text(json.dumps(["__x64_sys_read", "ksys_read"]), encoding="utf-8")
+    monkeypatch.setattr(sa, "_SYMBOLS_SNAPSHOT", str(path))
+    monkeypatch.setattr(sa, "_symbols_from_kallsyms", lambda: frozenset())
+    assert sa.kernel_symbols() == frozenset({"__x64_sys_read", "ksys_read"})
+
+
+def test_kallsyms_answers_when_nothing_was_published(tmp_path, monkeypatch):
+    _forget_symbols(monkeypatch)
+    monkeypatch.setattr(sa, "_SYMBOLS_SNAPSHOT", str(tmp_path / "absent.json"))
+    monkeypatch.setattr(sa, "_symbols_from_kallsyms", lambda: frozenset({"do_syscall_64"}))
+    assert sa.kernel_symbols() == frozenset({"do_syscall_64"})
+
+
+def test_an_empty_answer_is_not_cached_as_the_truth(tmp_path, monkeypatch):
+    # A backend that started before the collector must not stay blind until it
+    # is restarted.
+    _forget_symbols(monkeypatch)
+    monkeypatch.setattr(sa, "_SYMBOLS_SNAPSHOT", str(tmp_path / "absent.json"))
+    monkeypatch.setattr(sa, "_symbols_from_kallsyms", lambda: frozenset())
+    assert sa.kernel_symbols() == frozenset()
+    monkeypatch.setattr(sa, "_symbols_from_kallsyms", lambda: frozenset({"do_syscall_64"}))
+    assert sa.kernel_symbols() == frozenset({"do_syscall_64"})
+
+
+def test_publishing_writes_a_readable_list(tmp_path, monkeypatch):
+    monkeypatch.setattr(sa, "_symbols_from_kallsyms", lambda: frozenset({"ksys_read", "do_select"}))
+    path = tmp_path / "ksyms.json"
+    assert sa.publish_symbols(str(path)) == 2
+    assert json.loads(path.read_text(encoding="utf-8")) == ["do_select", "ksys_read"]
+
+
+def test_descriptor_arguments_are_named_only_where_they_exist():
+    assert sa.fd_argument("read") == 0
+    assert sa.fd_argument("poll") is None
+    assert sa.fd_argument("nothing_like_this") is None
diff --git a/tests/test_syscalls_service.py b/tests/test_syscalls_service.py
index 683b9ec..519b024 100644
--- a/tests/test_syscalls_service.py
+++ b/tests/test_syscalls_service.py
@@ -1,5 +1,8 @@
 """Tests for ``kernel_ai.services.syscalls``."""
 
+import json
+import time
+
 from kernel_ai.services import syscalls as svc
 
 
@@ -26,6 +29,176 @@ def test_get_real_system_calls_linux_empty_proc(monkeypatch):
     assert isinstance(out, list)
 
 
+def _stub_proc(monkeypatch, parked, comms=None, kernel_threads=()):
+    """Pretend /proc holds the given {pid: syscall_line} set of tasks.
+
+    Anything named in ``kernel_threads`` gets an empty command line, which is
+    how the real thing tells a kthread from a process.
+    """
+    monkeypatch.setattr(svc.platform, "system", lambda: "Linux")
+    monkeypatch.setattr(svc.os, "listdir", lambda _path: list(parked))
+    monkeypatch.setattr(svc.os.path, "exists", lambda _path: True)
+
+    real_open = open
+
+    def fake_open(path, *args, **kwargs):
+        text = None
+        for pid, line in parked.items():
+            if path == f"/proc/{pid}/syscall":
+                text = line
+            elif path == f"/proc/{pid}/comm":
+                text = (comms or {}).get(pid, f"task{pid}")
+            elif path == f"/proc/{pid}/cmdline":
+                text = b"" if pid in kernel_threads else b"/usr/bin/task\x00"
+        if text is None:
+            return real_open(path, *args, **kwargs)
+
+        class _Handle:
+            def read(self_inner, size=-1):
+                return text[:size] if size and size > 0 else text
+
+            def __enter__(self_inner):
+                return self_inner
+
+            def __exit__(self_inner, *_exc):
+                return False
+
+        return _Handle()
+
+    monkeypatch.setattr("builtins.open", fake_open)
+
+
+def test_waiters_name_the_processes_parked_in_each_syscall(monkeypatch):
+    _stub_proc(
+        monkeypatch,
+        {"11": "0 0x3 0x0", "12": "0 0x4 0x0", "13": "1 0x1 0x0"},
+        {"11": "nginx", "12": "nginx", "13": "bash"},
+    )
+    out = svc.get_real_system_calls(
+        syscall_names={0: "read", 1: "write"},
+        map_syscall_to_subsystem_fn=lambda _name: "fs",
+        kernel_dna_max_procs=10,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    rows = {row["name"]: row for row in out}
+    assert rows["read"]["count"] == 2
+    assert [(w["pid"], w["comm"]) for w in rows["read"]["waiters"]] == [(11, "nginx"), (12, "nginx")]
+    assert [(w["pid"], w["comm"]) for w in rows["write"]["waiters"]] == [(13, "bash")]
+
+
+def test_waiter_list_is_capped_but_the_count_stays_honest(monkeypatch):
+    pids = [str(100 + i) for i in range(svc._MAX_WAITERS_PER_SYSCALL + 5)]
+    _stub_proc(monkeypatch, {pid: "0 0x3 0x0" for pid in pids})
+    out = svc.get_real_system_calls(
+        syscall_names={0: "read"},
+        map_syscall_to_subsystem_fn=lambda _name: "fs",
+        kernel_dna_max_procs=100,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    assert out[0]["count"] == len(pids)
+    assert len(out[0]["waiters"]) == svc._MAX_WAITERS_PER_SYSCALL
+
+
+def test_running_and_unreadable_tasks_are_not_counted_as_waiters(monkeypatch):
+    _stub_proc(monkeypatch, {"21": "running", "22": "-1", "23": "0 0x3 0x0"})
+    out = svc.get_real_system_calls(
+        syscall_names={0: "read"},
+        map_syscall_to_subsystem_fn=lambda _name: "fs",
+        kernel_dna_max_procs=10,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    assert len(out) == 1
+    assert out[0]["count"] == 1
+    assert [w["pid"] for w in out[0]["waiters"]] == [23]
+
+
+def test_comm_is_none_when_the_process_exits_mid_sample(monkeypatch):
+    _stub_proc(monkeypatch, {"31": "0 0x3 0x0"})
+    monkeypatch.setattr(svc, "_read_comm", lambda _pid: None)
+    out = svc.get_real_system_calls(
+        syscall_names={0: "read"},
+        map_syscall_to_subsystem_fn=lambda _name: "fs",
+        kernel_dna_max_procs=10,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    assert out[0]["waiters"] == [{"pid": 31, "comm": None}]
+
+
+def test_kernel_threads_are_not_parked_in_syscall_zero(monkeypatch):
+    # kthreads read as "0 0x0 …" because no call is in flight; counting them
+    # would invent a top row with every kthread on the box parked in it.
+    _stub_proc(
+        monkeypatch,
+        {"2": "0 0x0 0x0 0x0 0x0 0x0 0x0 0x0 0x0", "3": "0 0x0 0x0 0x0 0x0 0x0 0x0 0x0 0x0", "40": "0 0x3 0x0"},
+        kernel_threads=("2", "3"),
+    )
+    out = svc.get_real_system_calls(
+        syscall_names={0: "read"},
+        map_syscall_to_subsystem_fn=lambda _name: "fs",
+        kernel_dna_max_procs=10,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    assert out[0]["count"] == 1
+    assert [w["pid"] for w in out[0]["waiters"]] == [40]
+
+
+def _write_snapshot(tmp_path, monkeypatch, payload):
+    path = tmp_path / "syscalls.json"
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    monkeypatch.setattr(svc, "_SYSCALLS_SNAPSHOT", str(path))
+    return path
+
+
+def test_a_fresh_collector_snapshot_is_preferred_over_self_sampling(tmp_path, monkeypatch):
+    _write_snapshot(tmp_path, monkeypatch, {
+        "ts": time.time(),
+        "tasks_total": 199,
+        "blocked_total": 51,
+        "syscalls": [{"name": "futex", "nr": 202, "count": 30, "waiters": []}],
+    })
+    sample = svc.get_syscall_sample(
+        syscall_names={},
+        map_syscall_to_subsystem_fn=lambda _name: "sched",
+        kernel_dna_max_procs=10,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    assert sample["scope"] == "machine"
+    assert sample["blocked_total"] == 51
+    assert sample["syscalls"][0]["name"] == "futex"
+    # The collector samples; naming the subsystem is the app's job.
+    assert sample["syscalls"][0]["subsystem"] == "sched"
+
+
+def test_a_stale_snapshot_is_refused_rather_than_shown_as_now(tmp_path, monkeypatch):
+    _write_snapshot(tmp_path, monkeypatch, {
+        "ts": time.time() - (svc._SNAPSHOT_MAX_AGE + 30),
+        "syscalls": [{"name": "futex", "count": 30, "waiters": []}],
+    })
+    _stub_proc(monkeypatch, {"7": "0 0x3 0x0"})
+    sample = svc.get_syscall_sample(
+        syscall_names={0: "read"},
+        map_syscall_to_subsystem_fn=lambda _name: "fs",
+        kernel_dna_max_procs=10,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    assert sample["scope"] == "self"
+    assert [row["name"] for row in sample["syscalls"]] == ["read"]
+
+
+def test_without_a_collector_the_sample_says_it_only_saw_itself(monkeypatch):
+    monkeypatch.setattr(svc, "_SYSCALLS_SNAPSHOT", "/nonexistent/syscalls.json")
+    _stub_proc(monkeypatch, {"7": "0 0x3 0x0", "8": "0 0x4 0x0"})
+    sample = svc.get_syscall_sample(
+        syscall_names={0: "read"},
+        map_syscall_to_subsystem_fn=lambda _name: "fs",
+        kernel_dna_max_procs=10,
+        fallback_mock_calls_fn=lambda: [],
+    )
+    assert sample["source"] == "backend"
+    assert sample["scope"] == "self"
+    assert sample["blocked_total"] == 2
+
+
 def test_get_softirq_nucleotides_handles_read_error(monkeypatch):
     def _boom(*_args, **_kwargs):
         raise OSError("nope")
diff --git a/tests/test_telemetry_orchestration_service.py b/tests/test_telemetry_orchestration_service.py
index 28d0b9f..0e1bd1c 100644
--- a/tests/test_telemetry_orchestration_service.py
+++ b/tests/test_telemetry_orchestration_service.py
@@ -14,7 +14,8 @@ def fake_get_real_system_calls(**kwargs):
     out = svc.get_real_system_calls()
 
     assert out == [{"name": "read"}]
-    assert called["syscall_names"] is svc.SYSCALL_NAMES
+    # The table of the running kernel, not a bundled one for another machine.
+    assert called["syscall_names"] == svc.get_syscall_names()
     assert callable(called["map_syscall_to_subsystem_fn"])
     assert callable(called["fallback_mock_calls_fn"])