Add agent collector loop: drain monitors into the SQLite store

ober

7d1d0036ffa54438685305496d1cd7f9bebe1488

diff --git a/Makefile b/Makefile
index 471d338..245665a 100644
--- a/Makefile
+++ b/Makefile
@@ -8,7 +8,7 @@ SCHEME ?= $(JERBOA)/.chez/bin/scheme
 BUILD  ?= build/rust
 TYPED  := $(wildcard typed/*.ss)
 
-.PHONY: rust test ffi-demo kernels-check triage-check triage-store-check analytics-check detect-check storage-check entity-check threats-check geoip-check sigma-check yaml-rules-check buffer-check dns-sniffer-check suspicious-check netconn-check kernmod-check selinux-check container-check dns-servers-check sensitive-path-check dtrace-parse-check proc-linux-check freebsd-parse-check event-meta-check config-check event-danger-check persistence-check file-change-check webshell-check platform-mounts-check analyze-cli-check collector-cli-check event-summary-check ioc-check frame-check correlate-check revshell-check cron-check logtamper-check detection-rules-check ipaddr-check auth-check lolbin-check dga-check calendar-check monitor-process-check monitor-network-check monitor-files-check monitor-dns-check monitor-manager-check event-json-check checks clean
+.PHONY: rust test ffi-demo kernels-check triage-check triage-store-check analytics-check detect-check storage-check entity-check threats-check geoip-check sigma-check yaml-rules-check buffer-check dns-sniffer-check suspicious-check netconn-check kernmod-check selinux-check container-check dns-servers-check sensitive-path-check dtrace-parse-check proc-linux-check freebsd-parse-check event-meta-check config-check event-danger-check persistence-check file-change-check webshell-check platform-mounts-check analyze-cli-check collector-cli-check event-summary-check ioc-check frame-check correlate-check revshell-check cron-check logtamper-check detection-rules-check ipaddr-check auth-check lolbin-check dga-check calendar-check monitor-process-check monitor-network-check monitor-files-check monitor-dns-check monitor-manager-check event-json-check collector-check checks clean
 # Combined libdir path so sibling libraries `(jsecmon ...)` resolve to ./jsecmon
 # (a second --libdirs would replace, not append, the jerboa one).
 LIBDIRS := "$(JERBOA)/lib:$(CURDIR)"
@@ -379,6 +379,13 @@ monitor-manager-check:
 event-json-check:
 	$(SCHEME) --libdirs $(LIBDIRS) --script examples/event_json_check.ss
 
+# Agent collector loop (secmon agent storage path): boot + a tick driven through
+# the monitors into a real temp SQLite store, then queried back to assert events
+# are serialized, summarized, pid/name-extracted, and persisted. No typed kernels
+# (no rust dep) -- only the native sqlite loader, like entity-check.
+collector-check:
+	$(LOADER_ENV) $(SCHEME) --libdirs $(LIBDIRS) --script examples/collector_check.ss
+
 # Everything that runs through the Jerboa side of the bridge, one shot.
 checks: kernels-check
 	$(SCHEME) --libdirs $(LIBDIRS) --script examples/triage_check.ss
@@ -433,6 +440,7 @@ checks: kernels-check
 	$(SCHEME) --libdirs $(LIBDIRS) --script examples/monitor_dns_check.ss
 	$(SCHEME) --libdirs $(LIBDIRS) --script examples/monitor_manager_check.ss
 	$(SCHEME) --libdirs $(LIBDIRS) --script examples/event_json_check.ss
+	$(LOADER_ENV) $(SCHEME) --libdirs $(LIBDIRS) --script examples/collector_check.ss
 
 clean:
 	rm -rf $(BUILD)
diff --git a/examples/collector_check.ss b/examples/collector_check.ss
new file mode 100644
index 0000000..e9c5ea1
--- /dev/null
+++ b/examples/collector_check.ss
@@ -0,0 +1,104 @@
+;;; Behaviour check for the agent collector loop (secmon agent storage path).
+;;;
+;;; Wires the manager's four monitors to fixture providers and drives boot + a
+;;; tick through `run-boot!`/`run-tick!` into a REAL temp SQLite store, then
+;;; queries the rows back to assert the whole agent body works end to end: events
+;;; are serialized, summarized, pid/name-extracted, and persisted — no OS access.
+;;;
+;;; Run from the repo root with the repo on the libdir path:
+;;;   scheme --libdirs $JERBOA/lib --libdirs . --script examples/collector_check.ss
+
+(import (jerboa prelude)
+        (jsecmon monitor-process)
+        (jsecmon monitor-network)
+        (jsecmon monitor-files)
+        (jsecmon monitor-dns)
+        (jsecmon monitor-manager)
+        (jsecmon storage)
+        (jsecmon collector))
+
+(def fails 0)
+(def (check label got want)
+  (let ((ok (equal? got want)))
+    (unless ok (set! fails (+ fails 1)))
+    (displayln (if ok "  ok   " "  FAIL ") label " => " got
+               (if ok "" (str "  (want " want ")")))))
+(def (one db type) (car (query-events db (make-filter "event_type" type))))
+(def (count db type) (length (query-events db (make-filter "event_type" type))))
+
+;; --- a temp store -------------------------------------------------------------
+(def db-path (str "/tmp/jsec-collector-" (random 1000000000) ".db"))
+(when (file-exists? db-path) (delete-file db-path))
+(def db (store-open db-path))
+
+;; --- the same fixture monitor-set as monitor_manager_check --------------------
+(def *pids* '(1 100))
+(def procs (make-hash-table))
+(hash-put! procs 1   (make-proc-info 1   0    0 "/sbin/init"     "systemd" '("/sbin/init") #f '()))
+(hash-put! procs 100 (make-proc-info 100 1 1000 "/usr/bin/gitea" "gitea"   '("/usr/bin/gitea") #f '()))
+(def pp (make-mon-provider (lambda () *pids*) (lambda (p) (hash-get procs p)) "host"))
+
+(def l-ssh (make-conn-info "tcp" "0.0.0.0" 22 "0.0.0.0" 0 "LISTEN" #f "sshd"))
+(def c-web (make-conn-info "tcp" "10.0.0.2" 50000 "93.184.216.34" 443 "ESTABLISHED" 1234 "curl"))
+(def c-dns (make-conn-info "udp" "10.0.0.2" 40000 "8.8.8.8"       53  "ESTABLISHED" 1234 "curl"))
+(def np (make-net-provider (lambda () (list c-web c-dns)) (lambda () (list l-ssh)) "host"))
+
+(def fs (make-hash-table))
+(def (fs-get p) (or (hash-get fs p) (make-file-state #f 0 0 0 0 #f)))
+(hash-put! fs "/etc/passwd" (make-file-state "p1" #o644 0 0 100 #t))
+(def fp (make-file-provider fs-get (lambda (p) #f) "host"))
+
+(def mset (make-monitor-set
+           "host"
+           (make-monitor pp) pp
+           (make-network-monitor np) np
+           (make-file-monitor fp 'linux) fp '("/etc/passwd")
+           (make-dns-monitor '("8.8.8.8") "host")))
+
+(def col (open-collector db "agent-1" "host"))
+
+;; --- boot: one agent_start row ------------------------------------------------
+(displayln "boot stores exactly the agent_start event:")
+(check "boot inserts 1"   (run-boot! col mset "0.1.0" 1000) 1)
+(check "count = 1"        (store-count db) 1)
+(check "agent_start summary" (hash-get (one db "agent_start") "summary")
+       "agent started on host")
+
+;; a critical file changes after baseline -> the tick will flag it
+(hash-put! fs "/etc/passwd" (make-file-state "p2" #o644 0 0 100 #t))
+
+;; --- tick: the seven manager events, all persisted ----------------------------
+(displayln "one tick persists every monitor's events:")
+(check "tick inserts 7"   (run-tick! col mset 2000) 7)
+(check "count = 8"        (store-count db) 8)
+(check "2 process_start"  (count db "process_start") 2)
+(check "1 listening_port" (count db "listening_port") 1)
+(check "2 network_connection" (count db "network_connection") 2)
+(check "1 suspicious_file_change" (count db "suspicious_file_change") 1)
+(check "1 dns_query"      (count db "dns_query") 1)
+
+;; --- the rows carry derived pid/name/summary + parsed data --------------------
+(displayln "rows carry extracted pid/process_name, summary, and parsed data:")
+(def dq (one db "dns_query"))
+(check "dns pid extracted"   (hash-get dq "pid") 1234)
+(check "dns name extracted"  (hash-get dq "process_name") "curl")
+(check "dns summary"         (hash-get dq "summary") "<unknown> UDP query")
+(check "dns data parsed"     (hash-get (hash-get dq "data") "server_addr") "8.8.8.8")
+
+(def sf (one db "suspicious_file_change"))
+(check "file path stored"    (hash-get (hash-get sf "data") "path") "/etc/passwd")
+(check "file has no pid"     (hash-get sf "pid") #f)
+(check "file summary=reason" (string-contains (hash-get sf "summary") "Critical") 0)
+
+;; --- a replay of the same tick is deduped by UNIQUE(host,source,seq) ----------
+(displayln "no-change tick (within dns dedup window) stores nothing:")
+(check "tick inserts 0"   (run-tick! col mset 2500) 0)
+(check "count still 8"    (store-count db) 8)
+
+(store-close db)
+(delete-file db-path)
+
+(newline)
+(if (= fails 0)
+    (displayln "OK: the collector boots, ticks, and persists every event with derived fields.")
+    (begin (displayln fails " FAILURES") (exit 1)))
diff --git a/jsecmon/collector.ss b/jsecmon/collector.ss
new file mode 100644
index 0000000..64cbc9d
--- /dev/null
+++ b/jsecmon/collector.ss
@@ -0,0 +1,106 @@
+#!chezscheme
+;;; jsecmon agent collector loop (secmon's agent storage path), untyped.
+;;;
+;;; The seam where the monitor runtime meets storage. secmon's agent drains the
+;;; mpsc channel of SecurityEvents and, for each, derives (category, severity,
+;;; pid, process_name, summary, json) and INSERTs a row. jsecmon already has the
+;;; pieces — monitor-tick produces the structured event hashes, event-json turns
+;;; one into its flat JSON `data`, event-summary derives pid/name/summary from
+;;; that data, storage writes the row — this module ties them into the loop.
+;;;
+;;;   `collect-event!` is the per-event store step; `run-boot!`/`run-tick!` drive
+;;;   a monitor-set's startup and one poll cycle into the store. All of this is
+;;;   pure over an injected db handle + monitor-set, so the whole agent body is
+;;;   testable against a temp SQLite db with fixture providers
+;;;   (examples/collector_check.ss). `run-agent` is the thin live shell: real
+;;;   clock, the Linux monitor-set, a poll/sleep loop — Linux-only, untested here.
+;;;
+;;; Faithfulness points the Rust pins:
+;;;   * seq is a per-(host,source) monotonic counter; the events table's
+;;;     UNIQUE(host,source,seq) + store-event's INSERT OR IGNORE make a replay a
+;;;     no-op, so collect-events! returns only the count actually inserted.
+;;;   * pid / process_name are derived FROM the flat data via extract-pid /
+;;;     extract-process-name (secmon's store path does the same), not read off the
+;;;     event's routing scalars — so they cover source_pid / spawned_process / …
+;;;     for every variant, and become SQL NULL when the data carries none.
+;;;   * summary is build-summary over the SAME parsed data object the row stores.
+
+(library (jsecmon collector)
+  (export open-collector collector?
+          collector-db collector-source collector-host collector-seq
+          collect-event! collect-events!
+          run-boot! run-tick!
+          now-ms run-agent)
+  (import (except (chezscheme)
+                  make-hash-table hash-table?
+                  sort sort!
+                  printf fprintf
+                  path-extension path-absolute?
+                  with-input-from-string with-output-to-string
+                  iota 1+ 1-
+                  partition
+                  make-date make-time)
+          (except (jerboa prelude) meta atom?)
+          (jsecmon monitor-manager)
+          (jsecmon event-json)
+          (jsecmon event-summary)
+          (jsecmon storage))
+
+  ;; db: an open store handle; source: this agent's reporter label; host: the
+  ;; default hostname for events that don't carry one; seq: the monotonic counter.
+  (defstruct collector (db source host seq))
+
+  (def (open-collector db source host)
+    (make-collector db source host 0))
+
+  (def (next-seq! col)
+    (let ((n (+ 1 (collector-seq col))))
+      (collector-seq-set! col n)
+      n))
+
+  ;; store one monitor event row; returns #t iff a new row was inserted.
+  (def (collect-event! col ev)
+    (let* ((seq       (next-seq! col))
+           (type      (hash-get ev "type"))
+           (severity  (or (hash-get ev "severity") "info"))
+           (host      (or (hash-get ev "host") (collector-host col)))
+           (ts        (or (hash-get ev "ts") 0))
+           (data-hash (event-data-hash ev))
+           (data-json (json-object->string data-hash))
+           (summary   (build-summary type data-hash))
+           (pid       (extract-pid data-hash))
+           (pname     (extract-process-name data-hash)))
+      (store-event (collector-db col) seq host (collector-source col)
+                   ts type severity pid pname summary data-json)))
+
+  ;; store a batch (e.g. one tick's events); returns the number inserted.
+  (def (collect-events! col events)
+    (fold-left (lambda (n ev) (if (collect-event! col ev) (+ n 1) n)) 0 events))
+
+  ;; one-time startup: baseline + agent_start, stored.
+  (def (run-boot! col mset version now)
+    (collect-events! col (monitor-boot mset version now)))
+
+  ;; one poll cycle: every monitor scans, results stored.
+  (def (run-tick! col mset now)
+    (collect-events! col (monitor-tick mset now)))
+
+  ;; --- the live shell (Linux): real clock + the platform monitor-set ---------
+
+  ;; epoch milliseconds (secmon's event timestamps are ms).
+  (def (now-ms)
+    (let ((t (current-time)))
+      (+ (* 1000 (time-second t)) (quotient (time-nanosecond t) 1000000))))
+
+  ;; the agent: open the store, wire the Linux providers, boot, then poll every
+  ;; `interval-secs`. Long-lived; the caller runs it as the agent process.
+  (def (run-agent db-path source version interval-secs)
+    (let* ((db   (store-open db-path))
+           (mset (make-linux-monitor-set))
+           (host (monitor-set-hostname mset))
+           (col  (open-collector db source host)))
+      (run-boot! col mset version (now-ms))
+      (let loop ()
+        (run-tick! col mset (now-ms))
+        (sleep (make-time 'time-duration 0 interval-secs))
+        (loop)))))