#!/usr/bin/env python3 """ Condensate Observer — passive memory pattern watcher. Polls /proc for memory snapshots of known VPS processes, watches systemd journal for service lifecycle events, and deposits observations into NeuroGraph via ng_tract. Zero interference. Pure observation. Bunyan food later. E-T Systems / Condensate """ import json import logging import os import subprocess import sys import time from pathlib import Path from typing import Dict, List, Optional, Tuple import numpy as np import ng_tract logging.basicConfig( level=logging.INFO, format="%(asctime)s [condensate-observer] %(levelname)s %(message)s", datefmt="%Y-%m-%dT%H:%M:%S", stream=sys.stdout, ) log = logging.getLogger("condensate-observer") # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- MODULE_ID = "condensate" TRACTS_DIR = Path(os.environ.get("ET_TRACTS_DIR", os.path.expanduser("~/.et_modules/tracts"))) TRACT_PEERS = ["neurograph", "bunyan"] POLL_INTERVAL = 10 # seconds between memory polls JOURNAL_INTERVAL = 60 # seconds between journal polls EMBEDDING_DIM = 768 # name → cmdline fragment to match in /proc/{pid}/cmdline WATCHED_PROCS: Dict[str, str] = { "openclaw-gateway": "openclaw-gateway", "neurograph_rpc": "neurograph_rpc.py", "oc-usage-shim": "oc-usage-shim.py", "python3-tid": "runserver", # TID Django process } # systemd units to watch for start/stop events WATCHED_SERVICES = [ "openclaw-gateway.service", "condensate-watchdog.service", "condensate-observer.service", ] # --------------------------------------------------------------------------- # /proc helpers # --------------------------------------------------------------------------- def _read_kv_file(path: str) -> Dict[str, str]: try: with open(path) as f: d = {} for line in f: if ":" in line: k, v = line.split(":", 1) d[k.strip()] = v.strip() return d except (FileNotFoundError, ProcessLookupError, PermissionError): return {} def _kb(s: Optional[str]) -> float: if not s: return 0.0 try: return float(s.split()[0]) except (ValueError, IndexError): return 0.0 def find_pid(fragment: str) -> Optional[int]: try: for entry in os.scandir("/proc"): if not entry.name.isdigit(): continue try: cmdline = Path(f"/proc/{entry.name}/cmdline").read_bytes() if fragment.encode() in cmdline: return int(entry.name) except (FileNotFoundError, PermissionError): continue except Exception: pass return None def take_snapshot(pid: int, name: str) -> Optional[Dict]: status = _read_kv_file(f"/proc/{pid}/status") if not status: return None smaps = _read_kv_file(f"/proc/{pid}/smaps_rollup") mem = _read_kv_file("/proc/meminfo") threads_raw = status.get("Threads", "1").split() threads = int(threads_raw[0]) if threads_raw else 1 return { "pid": pid, "name": name, "rss_mb": _kb(status.get("VmRSS")) / 1024, "vsize_mb": _kb(status.get("VmSize")) / 1024, "heap_mb": _kb(smaps.get("Heap")) / 1024, "stack_kb": _kb(smaps.get("Stack")), "anon_mb": _kb(smaps.get("Anonymous")) / 1024, "threads": threads, "sys_mem_total_mb": _kb(mem.get("MemTotal")) / 1024, "sys_mem_avail_mb": _kb(mem.get("MemAvailable")) / 1024, "sys_swap_total_mb":_kb(mem.get("SwapTotal")) / 1024, "sys_swap_free_mb": _kb(mem.get("SwapFree")) / 1024, "timestamp": time.time(), } # --------------------------------------------------------------------------- # Embedding # --------------------------------------------------------------------------- # # Layout (768 dims): # [0] rss_mb / 16384 normalized RSS # [1] vsize_mb / 65536 virtual size # [2] threads / 256 thread count # [3] heap_mb / 16384 heap # [4] stack_kb / 65536 stack # [5] anon_mb / 16384 anonymous mappings # [6] rss_growth_mb_per_min rate of change, clamped −1..1 # [7] is_new_pid 1.0 on first observation of this pid # [8] is_restart 1.0 if pid changed since last poll # [9] sys_mem_used_ratio 1 − avail/total # [10] sys_swap_used_ratio swap fraction used # [11−767] zeros (reserved for future Lenia / membrane features) def encode_snapshot(snap: Dict, prev: Optional[Dict] = None, event: str = "poll") -> np.ndarray: v = np.zeros(EMBEDDING_DIM, dtype=np.float32) v[0] = snap["rss_mb"] / 16384.0 v[1] = snap["vsize_mb"] / 65536.0 v[2] = snap["threads"] / 256.0 v[3] = snap["heap_mb"] / 16384.0 v[4] = snap["stack_kb"] / 65536.0 v[5] = snap["anon_mb"] / 16384.0 if prev is not None: dt = max((snap["timestamp"] - prev["timestamp"]) / 60.0, 0.001) growth = (snap["rss_mb"] - prev["rss_mb"]) / dt v[6] = float(np.clip(growth / 100.0, -1.0, 1.0)) v[7] = 1.0 if event == "new_pid" else 0.0 v[8] = 1.0 if event == "restart" else 0.0 total = snap["sys_mem_total_mb"] if total > 0: v[9] = 1.0 - snap["sys_mem_avail_mb"] / total swap = snap["sys_swap_total_mb"] if swap > 0: v[10] = 1.0 - snap["sys_swap_free_mb"] / swap return v # --------------------------------------------------------------------------- # Tract deposit # --------------------------------------------------------------------------- def _tract_paths() -> List[str]: my_dir = TRACTS_DIR / MODULE_ID my_dir.mkdir(parents=True, exist_ok=True) return [str(my_dir / f"{peer}.tract") for peer in TRACT_PEERS] def deposit(snap: Dict, emb: np.ndarray, event: str, anomaly: bool = False): # [2026-04-12] deposit_outcome — Rust owns the bytes end-to-end. Python never touches them. meta = {k: v for k, v in snap.items() if k != "timestamp"} meta["event"] = event try: ng_tract.deposit_outcome( snap["timestamp"], MODULE_ID, snap["name"], not anomaly, emb, # numpy f32 array — zero-copy to Rust _tract_paths(), ) # Sidecar JSONL — human-readable audit trail alongside the binary tracts sidecar = TRACTS_DIR / MODULE_ID / "observations.jsonl" with open(sidecar, "a") as f: f.write(json.dumps({"ts": snap["timestamp"], **meta}) + "\n") except Exception as exc: log.warning("deposit failed (%s/%s): %s", snap["name"], event, exc) # --------------------------------------------------------------------------- # Journal watcher # --------------------------------------------------------------------------- class JournalWatcher: def __init__(self): self._last = time.time() - 5.0 def poll(self) -> List[Tuple[str, str, float]]: """Return [(unit, state, ts), ...] for lifecycle events since last call.""" since = f"@{int(self._last)}" self._last = time.time() units: List[str] = [] for svc in WATCHED_SERVICES: units += ["-u", svc] events: List[Tuple[str, str, float]] = [] try: result = subprocess.run( ["journalctl", "--since", since, *units, "--output=json", "--no-pager", "-q"], capture_output=True, text=True, timeout=10, ) for line in result.stdout.splitlines(): try: e = json.loads(line) msg = e.get("MESSAGE", "") unit = e.get("_SYSTEMD_UNIT", e.get("UNIT", "unknown")) ts = int(e.get("__REALTIME_TIMESTAMP", 0)) / 1e6 if any(w in msg for w in ("Started", "start")): events.append((unit, "started", ts)) elif any(w in msg for w in ("Stopped", "Failed", "stop")): events.append((unit, "stopped", ts)) except (json.JSONDecodeError, KeyError): continue except (subprocess.TimeoutExpired, FileNotFoundError): pass return events # --------------------------------------------------------------------------- # Process tracker # --------------------------------------------------------------------------- class ProcessTracker: def __init__(self): self._pids: Dict[str, int] = {} self._snaps: Dict[str, Dict] = {} def scan(self) -> List[Tuple[str, Optional[int], str, Optional[Dict], Optional[Dict]]]: """Scan watched processes. Returns (name, pid, event, snap, prev_snap).""" results = [] for name, fragment in WATCHED_PROCS.items(): pid = find_pid(fragment) if pid is None: if name in self._pids: log.info("gone: %s (was pid %d)", name, self._pids[name]) results.append((name, None, "gone", None, self._snaps.get(name))) del self._pids[name] self._snaps.pop(name, None) continue snap = take_snapshot(pid, name) if snap is None: continue # exited mid-scan prev = self._snaps.get(name) if name not in self._pids: event = "new_pid" log.info("new: %s pid=%d rss=%.1f MB", name, pid, snap["rss_mb"]) elif self._pids[name] != pid: event = "restart" log.info("restart: %s pid %d→%d rss=%.1f MB", name, self._pids[name], pid, snap["rss_mb"]) else: event = "poll" self._pids[name] = pid self._snaps[name] = snap results.append((name, pid, event, snap, prev)) return results # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def _lifecycle_snap(unit: str, ts: float) -> Dict: """Minimal snapshot for a journal lifecycle event.""" mem = _read_kv_file("/proc/meminfo") total = _kb(mem.get("MemTotal")) / 1024 avail = _kb(mem.get("MemAvailable")) / 1024 return { "pid": 0, "name": unit.replace(".service", ""), "rss_mb": 0, "vsize_mb": 0, "heap_mb": 0, "stack_kb": 0, "anon_mb": 0, "threads": 0, "sys_mem_total_mb": total, "sys_mem_avail_mb": avail, "sys_swap_total_mb": 0, "sys_swap_free_mb": 0, "timestamp": ts, } def main(): log.info("starting — module=%s peers=%s poll=%ds", MODULE_ID, TRACT_PEERS, POLL_INTERVAL) log.info("tract dir: %s", TRACTS_DIR / MODULE_ID) tracker = ProcessTracker() journal = JournalWatcher() last_journal = 0.0 # Initial scan — birth weights for everything already running log.info("initial scan...") for name, pid, event, snap, prev in tracker.scan(): if snap: emb = encode_snapshot(snap, prev, event) deposit(snap, emb, event) log.info(" %s pid=%s rss=%.1f MB threads=%d", name, pid, snap["rss_mb"], snap["threads"]) log.info("observing.") while True: time.sleep(POLL_INTERVAL) now = time.time() for name, pid, event, snap, prev in tracker.scan(): if snap is None: continue anomaly = snap["rss_mb"] > 12000 emb = encode_snapshot(snap, prev, event) deposit(snap, emb, event, anomaly=anomaly) if event != "poll": log.info("lifecycle: %s %s pid=%s rss=%.1f MB", event, name, pid, snap["rss_mb"]) if anomaly: log.warning("anomaly: %s rss=%.1f MB exceeds threshold", name, snap["rss_mb"]) if now - last_journal >= JOURNAL_INTERVAL: last_journal = now for unit, state, ts in journal.poll(): log.info("journal: %s → %s", unit, state) snap = _lifecycle_snap(unit, ts or now) emb = encode_snapshot(snap, event=state) deposit(snap, emb, event=state) if __name__ == "__main__": main()