M2: onboarding loop: candidates, review queue, versioned rules, backlog re-parse

.github/workflows/ci.yml

@@ -1,4 +1,4 @@
-# M1 CI (Task 12). Unit tests only — no container services; the live E2E gate
+# M1+M2 CI. Unit tests only — no container services; the live E2E gate
 # is `python scripts/smoke.py`, run locally against the compose stack.
 name: ci
 
@@ -18,8 +18,9 @@ jobs:
           pip install -r services/collector/requirements.txt \
                       -r services/pipeline/requirements.txt \
                       -r services/gateway/requirements.txt \
+                      -r services/onboarding/requirements.txt \
                       -r simulator/requirements.txt
-          pip install pytest ruff==0.16.8
+          pip install pytest~=8.4 ruff==0.16.8
       - name: Lint
         run: ruff check .
       - name: Unit tests (libs + services + simulator)
@@ -36,7 +37,7 @@ jobs:
         with:
           node-version: 22
       - name: Install
-        run: npm ci || npm i
+        run: npm ci
       - name: Test
         run: npm run test --if-present
       - name: Build

.gitignore

@@ -22,7 +22,6 @@ ENV/
 
 # Env / secrets
 .env
-deploy/.env
 *.env.local
 
 # OS / editor cruft
@@ -42,14 +41,10 @@ Thumbs.db
 .ollama
 #zip files
 *.zip
-testdata/registered_rules.json
 #references
 references/
 
 # MVP build artifacts
 node_modules/
 web/dist/
-.venv/
-*.egg-info/
-__pycache__/
-.pytest_cache/
+.pytest_cache/

README.md

@@ -1,36 +1,77 @@
 # ULPF — Universal Log Pre-processing Framework
 
-SIH 2026 PS 26156 MVP. Status: M1 walking skeleton complete — the milestone
-exit is `python scripts/smoke.py` (all 5 checks PASS).
+SIH 2026 PS 26156 MVP. Status: M2 onboarding loop complete — the milestone
+exit is `python scripts/smoke.py` (all 8 checks PASS).
 
-- `services/` — collector, pipeline, gateway (+ onboarding, drift in M2/M3)
+- `services/` — collector, pipeline, gateway, onboarding (drift in M3)
 - `libs/ulpf-core` — fingerprinting, parsing, IDs, hash chain (single implementation)
 - `libs/ocsf-schema` — curated OCSF subset schema + validator
 - `web/` — React dashboard
 - `simulator/` — multi-vendor traffic player
 - `deploy/` — docker-compose, migrations
-- `scripts/smoke.py` — M1 acceptance gate (live E2E incl. forced replay)
+- `scripts/smoke.py` — M1+M2 acceptance gate (live E2E incl. forced replay + the onboarding hero loop)
 - `legacy-demo/` — the original hackathon demo, frozen. See its README.
 
-## Quickstart (M1)
+## Collector ingest
+
+Syslog UDP + TCP on 5514, HTTP `POST /v1/ingest` on 8080 (≤1000 lines/request).
+Oversized input is handled asymmetrically by design: UDP drops just the one
+oversized datagram (each datagram is an independent event), while TCP closes
+the connection — a byte stream has no line boundary to resynchronize to.
+
+## Quickstart
+
+```bash
+cd deploy && docker compose --profile slm-4b up -d --build
+```
+
+Brings up the M1 pipeline plus the M2 onboarding loop and the default SLM
+tier: `qwen3:4b` behind the `ollama-4b` sidecar (runs on 8GB boxes; the model
+is pulled by the sidecar on first start — a one-time multi-GB download).
+16GB boxes may upgrade: `--profile slm-8b` plus `ULPF_OLLAMA_URL=http://ollama-8b:11434`
+and `ULPF_OLLAMA_MODEL=qwen3:8b` in `deploy/.env` (see `.env.example`). The
+ollama sidecars publish **no host port** — they are reachable only inside the
+compose network (the onboarding service is their only client).
+
+Hero loop (unknown appliance → candidate rule): play the unknown `newapp`
+corpus at the collector, then watch the Review Queue while onboarding splits
+samples, asks the SLM, validates, and stores the candidate. On the default
+`qwen3:4b` tier the SLM may store no rule on modest hardware — in that case
+the fingerprint stays backlogged until a candidate lands (author one via the
+dashboard or `POST /api/rules/manual`); see docs/m2-slm-sanity.md for the
+measured tier numbers. The full lifecycle — grants, version minting, backlog
+re-parse — is recorded in docs/m2-rule-lifecycle.md:
+
+```bash
+docker compose --profile sim run --rm simulator --eps 25 --duration 20 --mode new-appliance
+open http://localhost:3000        # Review Queue: the newapp fingerprint lands as pending_review when a candidate stores
+# sustained live traffic for the dashboard (`run` needs the sim profile re-activated):
+docker compose --profile sim run --rm simulator --eps 20 --duration 300 --loop
+```
+
+Acceptance gate (starts the stack itself — bring the SLM tier profile so the
+onboarding sidecar is up):
 
 ```bash
-cd deploy && docker compose --profile sim up -d --build
-python scripts/smoke.py          # end-to-end acceptance incl. forced replay
-open http://localhost:3000       # Overview: live feed + raw traceability
-# sustained live traffic for the dashboard (the --profile sim one-shot above
-# exits after its single ~56-line pass):
-docker compose run --rm simulator --eps 20 --duration 300 --loop
+python scripts/smoke.py          # end-to-end acceptance incl. forced replay + hero loop
 ```
 
-The smoke runs the simulator over the golden corpora, then verifies: lossless
-ingest (raw delta == sent), the parsed/unparsed split per fingerprint, raw
-traceability (API + SQL join with recomputed sha256), forced replay
-idempotency (Kafka consumer-group delete + pipeline restart: raw events
-unchanged, raw_batches grown), and an independent recomputation of every
-raw_batches merkle root plus full per-partition chain linkage.
+The smoke runs the simulator over the golden corpora, then verifies:
+lossless ingest (raw delta == sent), the parsed/unparsed split per
+fingerprint, raw traceability (API + SQL join with recomputed sha256),
+forced replay idempotency (Kafka consumer-group delete + pipeline restart:
+raw events unchanged, raw_batches grown), an independent recomputation of
+every raw_batches merkle root plus full per-partition chain linkage (A–E),
+then the M2 onboarding loop: the newapp hero loop through approve → backlog
+re-parse (SLM candidate if one stores within `ULPF_SMOKE_SLM_WAIT_S`
+(default 300s), a locally pre-validated hand-authored candidate otherwise),
+the deterministic zenwall manual rule (G), and a replay-stability check with
+active rules, samples included (H).
 
-Smoke host deps (Python 3.13 tested): `psycopg[binary]` only. No confluent-kafka
+Smoke host deps (Python 3.13 tested): `psycopg[binary]` plus the two libs
+importable — `pip install -e libs/ulpf-core -e libs/ocsf-schema` (the smoke
+computes fingerprints and pre-validates hand-authored rules with the same
+`ulpf_core` gate the services run). No confluent-kafka
 on the host: Kafka interaction goes through `docker compose exec redpanda rpk`,
 and the simulator runs in its own container (the services'
 `confluent-kafka==2.5.0` pin is in-image only, py3.12). The smoke is

bench/fingerprint_fragmentation.py

@@ -24,12 +24,11 @@
 from __future__ import annotations
 
 import random
-import re
 import sys
-from collections.abc import Callable
 from pathlib import Path
 
-from ulpf_core.fingerprint import _classify, fingerprint_id, shape_hash
+from ulpf_core.fingerprint import fingerprint_id, shape_hash
+from ulpf_core.mutations import mutate_line
 
 ROOT = Path(__file__).resolve().parents[1]
 GOLDEN = ROOT / "libs" / "ulpf-core" / "tests" / "data" / "golden"
@@ -38,98 +37,10 @@
 SEED = 20260919
 MUTATIONS = 100
 
-# --- mutation machinery ------------------------------------------------------
-# Timestamps mutate first and park behind \x00 sentinels so later numeric
-# passes cannot chew freshly generated values into shape-breaking forms.
-# IP:port mutates before the IP pass so the ":port" tail does not lose the
-# digit its lookbehind needs.
-
-_ISO_TS_RE = re.compile(
-    r"(\d{4})-(\d{2})-(\d{2})(?:T(\d{2}):(\d{2}):(\d{2}))?([+-]\d{2}:?\d{2}|Z)?"
-)
-_SYSLOG_TS_RE = re.compile(
-    r"\b((?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec))\s+(\d{1,2})\s+(\d{2}):(\d{2}):(\d{2})\b"
-)
-_PORT_RE = re.compile(r"(?<=\d):\d{1,5}\b")              # port tail of IP:port
-_IP_RE = re.compile(r"\d{1,3}(?:\.\d{1,3}){3}")
-_KV_NUM_RE = re.compile(r"(?<==)\d{1,5}\b")               # key=12345 values
-_JSON_NUM_RE = re.compile(r'(?<=":)\d{1,5}\b')            # "key":12345 values
-_HOSTNAME_RE = re.compile(r"\b[a-z][a-z0-9-]*\d{1,4}\b")  # fw01, vpn-gw01, eth0
-_SENTINEL_RE = re.compile(r"\x00A(\d+)\x00")
-_MONTH_RE = re.compile(r"^(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)$")
-_MARKUP_CHARS = set('{}"<>')
-
-
-def _rand_iso_ts(match: re.Match[str], rng: random.Random) -> str:
-    text = f"{rng.randint(2020, 2030):04d}-{rng.randint(1, 12):02d}-{rng.randint(1, 28):02d}"
-    if match.group(4):
-        text += f"T{rng.randint(0, 23):02d}:{rng.randint(0, 59):02d}:{rng.randint(0, 59):02d}"
-    return text + (match.group(7) or "")
-
-
-def _rand_syslog_ts(match: re.Match[str], rng: random.Random) -> str:
-    # Keep the month word so the mutant still detects as syslog.
-    return (
-        f"{match.group(1)} {rng.randint(1, 28)} "
-        f"{rng.randint(0, 23):02d}:{rng.randint(0, 59):02d}:{rng.randint(0, 59):02d}"
-    )
-
-
-def _rand_ip(rng: random.Random) -> str:
-    return ".".join(str(rng.randint(1, 254)) for _ in range(4))
-
-
-def mutate_line(line: str, rng: random.Random) -> str:
-    """Return a same-format mutant of *line* with randomized values."""
-    stash: list[str] = []
-
-    def _swap(make: Callable[[re.Match[str]], str]) -> Callable[[re.Match[str]], str]:
-        def _sub(match: re.Match[str]) -> str:
-            stash.append(make(match))
-            return f"\x00A{len(stash) - 1}\x00"
-
-        return _sub
-
-    line = _ISO_TS_RE.sub(_swap(lambda m: _rand_iso_ts(m, rng)), line)
-    line = _SYSLOG_TS_RE.sub(_swap(lambda m: _rand_syslog_ts(m, rng)), line)
-    line = _PORT_RE.sub(_swap(lambda m: f":{rng.randint(1, 65535)}"), line)
-    line = _IP_RE.sub(_swap(lambda m: _rand_ip(rng)), line)
-    line = _KV_NUM_RE.sub(_swap(lambda m: str(rng.randint(1, 65535))), line)
-    line = _JSON_NUM_RE.sub(_swap(lambda m: str(rng.randint(1, 65535))), line)
-    line = _HOSTNAME_RE.sub(_swap(lambda m: f"h{rng.randint(0, 9999):04d}"), line)
-
-    # Free-text runs (message bodies) get a random word count. Structural
-    # tokens (markup, sentinels, syslog month, anything classified) are kept.
-    if " " in line:
-        kept: list[str] = []
-        run: list[str] = []
-
-        def _flush() -> None:
-            if run:
-                kept.extend(rng.choice(WORDS) for _ in range(rng.randint(1, 6)))
-                run.clear()
-
-        for token in line.split():
-            if (
-                _SENTINEL_RE.fullmatch(token)
-                or _MONTH_RE.fullmatch(token)
-                or _MARKUP_CHARS.intersection(token)
-                or _classify(token) != "W"
-            ):
-                _flush()
-                kept.append(token)
-            else:
-                run.append(token)
-        _flush()
-        line = " ".join(kept)
-
-    return _SENTINEL_RE.sub(lambda m: stash[int(m.group(1))], line)
-
-
-WORDS = [
-    "alpha", "bravo", "delta", "echo", "fox", "golf",
-    "hotel", "india", "juliet", "kilo", "lima", "mike",
-]
+# Mutation machinery (regexes, rand helpers, WORDS, mutate_line) lives in
+# ulpf_core.mutations since M2 — single definition shared with the candidate
+# probe. The bench passes vary_text=True to keep its historical behavior
+# (free-text message word counts are mutated too).
 
 
 def run_bench(mutations: int = MUTATIONS, seed: int = SEED) -> dict:
@@ -164,7 +75,7 @@ def run_bench(mutations: int = MUTATIONS, seed: int = SEED) -> dict:
         for _ in range(mutations):
             base = rng.choice(lines)
             fp = fingerprint_id(base)
-            mutant = mutate_line(base, rng)
+            mutant = mutate_line(base, rng, vary_text=True)
             mutant_hashes.add(shape_hash(mutant))
             ok = shape_hash(mutant) == canonical[fp]
             if not fp.startswith("auto_"):