ML stages

kernel_ai/ml/attribution/attack_map.py

@@ -124,47 +124,23 @@ def map_anomaly(anomaly: dict) -> dict | None:
             why=why,
         )
 
-    # --- Stage 1 / 2 host features ---
+    # Host-level spikes (pgfault, ctxt, TCP rates, IRQ) are traffic on this
+    # box, not ATT&CK. Only name a family when there is a process/sequence
+    # hint or an explicit miner/scanner token.
     feat = feature.lower()
-    if any(k in feat for k in ("tcp_retrans", "net_softirq", "tcp_inseg", "tcp_outseg")):
+    if "miner" in message or "xmrig" in message or any(h in feat for h in _MINER_HINTS):
         return _technique_payload(
-            TECHNIQUES["T1498"],
-            confidence=0.4,
+            TECHNIQUES["T1496"],
+            confidence=0.6,
             source="heuristic",
-            why=f"network-path pressure on {feature}",
+            why="host pressure with miner hint",
         )
-    if any(k in feat for k in ("ctxt_per_sec", "procs_running", "proc_count", "run_queue", "load1", "cpu_busy")):
-        # Resource pressure — could be DoS or miner; stay conservative.
-        if "miner" in message or "xmrig" in message:
-            tech = TECHNIQUES["T1496"]
-            conf = 0.6
-            why = "host CPU/sched pressure with miner hint"
-        else:
-            tech = TECHNIQUES["T1499"]
-            conf = 0.38
-            why = f"host sched/CPU pressure on {feature}"
-        return _technique_payload(tech, confidence=conf, source="heuristic", why=why)
-    if any(k in feat for k in ("pgfault", "pgmajfault", "pgscan", "swap_io", "psi_mem")):
+    if any(h in message or h in feat for h in _SCAN_HINTS):
         return _technique_payload(
-            TECHNIQUES["T1499"],
-            confidence=0.36,
+            TECHNIQUES["T1046"],
+            confidence=0.55,
             source="heuristic",
-            why=f"memory pressure on {feature}",
-        )
-    if "hardirq" in feat or "block_softirq" in feat:
-        return _technique_payload(
-            TECHNIQUES["T1499"],
-            confidence=0.34,
-            source="heuristic",
-            why=f"IRQ/block pressure on {feature}",
-        )
-
-    if source == "stage2_isoforest":
-        return _technique_payload(
-            TECHNIQUES["T1499"],
-            confidence=0.3,
-            source="heuristic",
-            why="IsolationForest unusual host state (family-level guess)",
+            why="host pressure with scanner hint",
         )
 
     return None

kernel_ai/ml/features.py

@@ -27,21 +27,26 @@ class FeatureSpec:
     # near-constant feature from firing on microscopic deviations.
     min_std: float
     label: str
+    # Optional per-feature z gates. 0 means "use MLConfig.z_warn / z_crit".
+    # Host chatter (minor faults, context switches) needs a higher bar than
+    # PSI / major faults — dashboard polls sat in the 6–11σ band.
+    z_warn: float = 0.0
+    z_crit: float = 0.0
 
 
 FEATURE_SPECS: dict[str, FeatureSpec] = {
     "proc_count": FeatureSpec("proc_count", "sched", 0.06, 3.0, "processes"),
     "procs_running": FeatureSpec("procs_running", "sched", 0.10, 1.0, "runnable now"),
     "procs_blocked": FeatureSpec("procs_blocked", "sched", 0.14, 1.0, "blocked (D)"),
-    "ctxt_per_sec": FeatureSpec("ctxt_per_sec", "sched", 0.18, 50.0, "context switches/s"),
+    "ctxt_per_sec": FeatureSpec("ctxt_per_sec", "sched", 0.18, 400.0, "context switches/s", 10.0, 14.0),
     "run_queue": FeatureSpec("run_queue", "sched", 0.16, 1.0, "run-queue depth"),
     "load1": FeatureSpec("load1", "sched", 0.08, 0.2, "loadavg 1m"),
     "tcp_retrans_per_sec": FeatureSpec("tcp_retrans_per_sec", "net", 0.30, 0.5, "TCP retrans/s"),
     "tcp_inseg_per_sec": FeatureSpec("tcp_inseg_per_sec", "net", 0.26, 20.0, "TCP in segs/s"),
     "tcp_outseg_per_sec": FeatureSpec("tcp_outseg_per_sec", "net", 0.34, 20.0, "TCP out segs/s"),
     "net_softirq_per_sec": FeatureSpec("net_softirq_per_sec", "net", 0.38, 20.0, "NET softirq/s"),
     "block_softirq_per_sec": FeatureSpec("block_softirq_per_sec", "fs", 0.46, 5.0, "BLOCK softirq/s"),
-    "pgfault_per_sec": FeatureSpec("pgfault_per_sec", "mm", 0.62, 50.0, "minor faults/s"),
+    "pgfault_per_sec": FeatureSpec("pgfault_per_sec", "mm", 0.62, 250.0, "minor faults/s", 12.0, 16.0),
     "pgmajfault_per_sec": FeatureSpec("pgmajfault_per_sec", "mm", 0.68, 1.0, "major faults/s"),
     "pgscan_direct_per_sec": FeatureSpec("pgscan_direct_per_sec", "mm", 0.72, 5.0, "direct reclaim/s"),
     "swap_io_per_sec": FeatureSpec("swap_io_per_sec", "mm", 0.76, 1.0, "swap io/s"),

kernel_ai/ml/retrain.py

@@ -79,10 +79,34 @@ def _refresh_markov(cfg: MLConfig, metrics: dict) -> None:
         logger.warning("Stage 8 Markov build failed: %s", exc)
 
 
+def _label_http(cfg: MLConfig, metrics: dict) -> None:
+    """Walk local nginx/app logs into ml_http_labels before champion/challenger."""
+    paths = [p for p in (cfg.http_nginx_log, cfg.http_app_log) if p]
+    readable = [p for p in paths if os.path.isfile(p) and os.access(p, os.R_OK)]
+    if not readable:
+        logger.info("HTTP label skipped: no readable log in %s", paths)
+        return
+    try:
+        from kernel_ai.ml.http_label import label_paths
+        from kernel_ai.ml.store import PostgresStore
+
+        store = PostgresStore(cfg.dsn)
+        try:
+            stats = label_paths(readable, hours=24.0, store=store)
+        finally:
+            store.close()
+        metrics["http_label_windows"] = float(stats.get("windows") or 0)
+        metrics["http_label_attempts"] = float(stats.get("attempts") or 0)
+        logger.info("HTTP labels refreshed: %s", stats)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("HTTP label failed: %s", exc)
+
+
 def _refresh_http(cfg: MLConfig, metrics: dict) -> None:
     """Best-effort Stage 9 retrain. Guardrails live in http_train (keep previous)."""
     if os.getenv("KERNEL_AI_ML_HTTP_RETRAIN", "true").lower() != "true":
         return
+    _label_http(cfg, metrics)
     try:
         from kernel_ai.ml.http_train import train as train_http
 

kernel_ai/ml/worker.py

@@ -26,17 +26,19 @@
 _HOUSEKEEPING_EVERY = 30
 
 
 def _build_anomalies(scores: dict[str, Score], cfg: MLConfig) -> list[dict]:
     """Turn high positive z-scores into Kernel DNA mutation records."""
     out: list[dict] = []
     for name, sc in scores.items():
         if sc.warm:
             continue
         # Attacks present as bursts: we flag upward deviations only.
-        if sc.z < cfg.z_warn or sc.value <= sc.mean:
-            continue
         spec = FEATURE_SPECS.get(name)
-        severity = "high" if sc.z >= cfg.z_crit else "medium"
+        warn = spec.z_warn if spec and spec.z_warn > 0 else cfg.z_warn
+        crit = spec.z_crit if spec and spec.z_crit > 0 else cfg.z_crit
+        if sc.z < warn or sc.value <= sc.mean:
+            continue
+        severity = "high" if sc.z >= crit else "medium"
         out.append(
             {
                 "source": "stage1_baseline",

kernel_ai/services/telemetry_orchestration.py

@@ -87,6 +87,43 @@ def _kernel_dna_softirq_nucleotides(limit=8):
 KERNEL_DNA_ML_SINCE_SEC = int(os.environ.get("KERNEL_DNA_ML_SINCE_SEC", "120"))
 KERNEL_DNA_ML_MAX = int(os.environ.get("KERNEL_DNA_ML_MAX", "8"))
 
+# Host-rate chatter: keep in the store, keep off the helix unless a process
+# or a named detector (sequence / HTTP / lineage) owns the row.
+_HELIX_HOLD_FEATURES = frozenset({
+    "pgfault_per_sec",
+    "ctxt_per_sec",
+    "tcp_inseg_per_sec",
+    "tcp_outseg_per_sec",
+    "net_softirq_per_sec",
+    "hardirq_per_sec",
+    "cpu_busy_pct",
+    "block_softirq_per_sec",
+    "tcp_retrans_per_sec",
+    "procs_running",
+    "proc_count",
+    "run_queue",
+    "load1",
+})
+_HELIX_ALWAYS_SOURCES = frozenset({
+    "stage4_sequence",
+    "stage5_process",
+    "stage8_sequence",
+    "stage9_http",
+    "stage9_success",
+})
+
+
+def _helix_surface(row):
+    """Whether a stored ML anomaly should become a helix mutation."""
+    source = str(row.get("source") or "")
+    if source in _HELIX_ALWAYS_SOURCES or source.startswith("stage9"):
+        return True
+    meta = row.get("meta") if isinstance(row.get("meta"), dict) else {}
+    if meta.get("pid"):
+        return True
+    feature = str(row.get("feature") or "")
+    return feature not in _HELIX_HOLD_FEATURES
+
 
 def _ml_anomalies_to_mutations(rows):
     """Map stored ML anomalies onto the Kernel DNA mutation contract.
@@ -99,6 +136,8 @@ def _ml_anomalies_to_mutations(rows):
     mutations = []
     seen = set()
     for row in rows:
+        if not _helix_surface(row):
+            continue
         feature = row.get("feature")
         if feature in seen:
             continue