Develop

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),

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",
 ]

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:

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 @@
         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 @@
             metrics = {}
 
     _refresh_stide(cfg, metrics)
+    _refresh_markov(cfg, metrics)
     logger.info("retrain complete: %s", metrics)
     return 0
 

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 @@
         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 _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 @@
         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)
@@ -117,7 +138,7 @@
         for pid, name in samples.items():
             self._append(int(pid), str(name))
 
     def update_stream(self, events) -> int:
         """Ingest an ordered L2 event stream (Stage 6). Returns events consumed."""
         n = 0
         for ev in events or []:
@@ -128,7 +149,8 @@
                 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 @@
         """
         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))

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,13 +92,24 @@ 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
     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
 
 

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,