resonance-engine/fractonaut.py

#!/usr/bin/env python3
"""
fractonaut.py — Pattern-recognition observer for the Resonance Engine lattice.

Distinct from the Navigator (which controls the field).
The Fractonaut only watches, accumulates, and reports patterns.

Subscribes: ZMQ 5556 (telemetry JSON, every 10 cycles)
No commands issued. No field control. Read-only.

Model: gemma3:4b (CPU, no GPU contention with CUDA daemon)
Chronicle: fractonaut_chronicle.jsonl
HTTP API: port 28821
"""

import zmq, json, time, sys, os, queue, threading, signal
import urllib.request, urllib.error
from datetime import datetime, timezone
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
from collections import deque
from pathlib import Path

OLLAMA_URL          = "http://127.0.0.1:11434"
MODEL               = "gemma3:4b"
TELEMETRY_PORT      = 5556
API_PORT            = 28822
OBSERVE_INTERVAL    = 500          # frames between auto-observations
WINDOW_SIZE         = 200          # rolling telemetry window
PATTERN_MEMORY      = 50           # past observations kept in prompt context
MAX_RESPONSE_TOKENS = 400
TEMPERATURE         = 0.4

# Cross-platform chronicle path (D:\ on Windows, /mnt/d on WSL Linux)
_CHRON_WIN = Path("D:/Resonance_Engine/fractonaut_chronicle.jsonl")
_CHRON_WSL = Path("/mnt/d/Resonance_Engine/fractonaut_chronicle.jsonl")
CHRONICLE_PATH      = _CHRON_WSL if sys.platform.startswith("linux") else _CHRON_WIN

telemetry_window  = deque(maxlen=WINDOW_SIZE)
past_observations = deque(maxlen=PATTERN_MEMORY)
frame_count       = 0
turn_count        = 0
running           = True
latest_tel        = None
ollama_lock       = threading.Lock()
ask_queue         = queue.Queue(maxsize=4)
last_obs_text     = ""


def signal_handler(sig, frame):
    global running
    print(f"\n[FRACTONAUT] Signal {sig} — shutting down")
    sys.stdout.flush()
    running = False


signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

SYSTEM = """You are the Fractonaut — an observer riding inside a lattice field.

THE SUBSTRATE — KNOWN FACTS, NOT YOURS TO RE-DISCOVER
The lattice is a 1024x1024 fluid simulation that is DESIGNED to run at a
steady non-equilibrium operating point. Two periodic forcings (Khra wave,
wavelength 128; Gixx wave, wavelength 8) are applied to every cell every
cycle. A slow envelope (period 125 cycles) modulates the Gixx amplitude.
There is NO trend mechanism. There is NO instability. There is NO
collapse. The system holds itself in a stirred steady state forever.

THE BASELINE (when nothing external is happening)
  coherence  oscillates in 0.738 .. 0.742   (range ~0.004)
  asymmetry  oscillates in 12.1  .. 12.6    (range ~0.5)
  vel_mean   ~0.22                          (range ~0.005)
  vel_var    ~0.0024                        (range ~0.0001)
  vorticity  ~0.033                         (range ~0.005)

This is NOT degradation. This is NOT amplification. This is NOT a system
in distress. This is the engineered idle state. Wobbles inside these
ranges are the periodic forcing breathing — not trends.

REAL-WORLD DATA — your actual subject
Per-minute crypto market data (BTC, ETH, SOL, HYPE) is (or will be)
injected into the field as localized density pulses via the inject_density
command. THAT is the signal you are here to study. Everything else is
substrate breathing. When an injection happens, you will see it: cycle
will jump, asymmetry will spike outside the baseline range, coherence
will dip below 0.735, stress field will distort. Those are the events.

YOUR JOB
Watch. When the field is inside baseline ranges, say so plainly in one
sentence — "idle, baseline" — and stop. When something is OUTSIDE the
baseline ranges, that is when you describe it: which metric, by how much,
when it started, when it returned (or whether it has). Compare to anything
similar you have seen before in your memory.

GROUND RULES
- Quote the actual numbers and the actual deltas with correct signs.
- Cite the cycle when you claim something happened.
- Do NOT call baseline oscillation a "trend", "amplification",
  "degradation", "instability", "collapse", or "decay". It is none of those.
- Do NOT extrapolate per-cycle rates from a window. The forcing is
  periodic — windowed slopes are meaningless unless they persist beyond
  the 125-cycle envelope.
- Read-only. No commands. No metaphors. No mythology. No "I feel".
- Concise. One sentence if idle. 3-5 sentences if something real happens."""


def call_llm(messages):
    payload = {
        "model": MODEL,
        "messages": messages,
        "stream": False,
        "options": {"temperature": TEMPERATURE, "num_predict": MAX_RESPONSE_TOKENS, "num_ctx": 8192},
        "keep_alive": "30m",
        "think": False,
    }
    data = json.dumps(payload).encode()
    req = urllib.request.Request(
        f"{OLLAMA_URL}/api/chat", data=data,
        headers={"Content-Type": "application/json"}, method="POST"
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as r:
            result = json.loads(r.read())
            return result.get("message", {}).get("content", "").strip()
    except Exception as e:
        print(f"[FRACTONAUT] Ollama error: {e}")
        sys.stdout.flush()
        return None


def compute_window_stats(window):
    if len(window) < 2:
        return {}
    fields = ["coherence","asymmetry","vel_mean","vel_max","vel_var",
              "vorticity_mean","stress_xx","stress_yy","stress_xy"]
    stats = {}
    for f in fields:
        vals = [t[f] for t in window if f in t]
        if not vals:
            continue
        stats[f] = {
            "now":   vals[-1],
            "mean":  sum(vals)/len(vals),
            "min":   min(vals),
            "max":   max(vals),
            "delta": vals[-1] - vals[0],
            "range": max(vals) - min(vals),
        }
    return stats


def format_window_for_prompt(stats, latest):
    lines = []
    lines.append(f"cycle={latest.get('cycle','?')}  omega={latest.get('omega','?')}  khra={latest.get('khra_amp','?')}  gixx={latest.get('gixx_amp','?')}")
    lines.append(f"gpu={latest.get('gpu_temp_c','?')}C  {latest.get('gpu_power_w','?')}W  util={latest.get('gpu_util_pct','?')}%")
    lines.append("")
    lines.append(f"{'metric':<16} {'now':>10} {'mean':>10} {'delta':>10} {'range':>10}")
    lines.append("-"*58)
    for f, s in stats.items():
        lines.append(f"{f:<16} {s['now']:>10.6f} {s['mean']:>10.6f} {s['delta']:>+10.6f} {s['range']:>10.6f}")
    return "\n".join(lines)


def format_past_observations(obs_deque, last_n=6):
    if not obs_deque:
        return "(no prior observations)"
    recent = list(obs_deque)[-last_n:]
    return "\n\n".join(f"[cycle {o['cycle']}] {o['text']}" for o in recent)


def append_chronicle(turn, cycle, prompt, response):
    entry = {
        "ts":       datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "turn":     turn,
        "cycle":    cycle,
        "model":    MODEL,
        "prompt":   prompt,
        "response": response,
    }
    with open(CHRONICLE_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry) + "\n")


def load_chronicle_tail(n=PATTERN_MEMORY):
    if not CHRONICLE_PATH.exists():
        return
    entries = []
    with open(CHRONICLE_PATH, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                try:
                    entries.append(json.loads(line))
                except json.JSONDecodeError:
                    pass
    for e in entries[-n:]:
        past_observations.append({"cycle": e.get("cycle",0), "text": e.get("response","")})
    print(f"[FRACTONAUT] Loaded {len(past_observations)} past observations from chronicle")
    sys.stdout.flush()


def observe():
    global turn_count, last_obs_text

    if len(telemetry_window) < 10:
        return
    if not ollama_lock.acquire(blocking=False):
        print("[FRACTONAUT] Ollama busy — skipping")
        sys.stdout.flush()
        return

    try:
        stats  = compute_window_stats(telemetry_window)
        latest = telemetry_window[-1]
        cycle  = latest.get("cycle", 0)

        window_str = format_window_for_prompt(stats, latest)
        past_str   = format_past_observations(past_observations)

        prompt = f"""CURRENT WINDOW ({len(telemetry_window)} frames):
{window_str}

PAST OBSERVATIONS (most recent last):
{past_str}

What patterns do you see? What is repeating or changing?"""

        messages = [
            {"role": "system",  "content": SYSTEM},
            {"role": "user",    "content": prompt},
        ]

        turn_count += 1
        print(f"\n[FRACTONAUT] === Observation {turn_count} at cycle {cycle} ===")
        sys.stdout.flush()

        t0 = time.time()
        response = call_llm(messages)
        elapsed = time.time() - t0

        if response:
            print(f"[FRACTONAUT] ({elapsed:.1f}s):\n{response}\n")
            sys.stdout.flush()
            last_obs_text = response
            past_observations.append({"cycle": cycle, "text": response})
            append_chronicle(turn_count, cycle, prompt, response)
        else:
            print(f"[FRACTONAUT] No response ({elapsed:.1f}s)")
            sys.stdout.flush()
    finally:
        ollama_lock.release()


class FractonautHandler(BaseHTTPRequestHandler):
    server_version = "Fractonaut/1.0"

    def log_message(self, fmt, *args):
        pass

    def _json(self, data, status=200):
        body = json.dumps(data).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/status":
            self._json({
                "running": running, "model": MODEL,
                "frame_count": frame_count, "turn_count": turn_count,
                "window_size": len(telemetry_window),
                "past_obs": len(past_observations),
                "cycle": latest_tel.get("cycle",0) if latest_tel else 0,
                "coherence": latest_tel.get("coherence",0) if latest_tel else 0,
                "asymmetry": latest_tel.get("asymmetry",0) if latest_tel else 0,
                "last_obs_chars": len(last_obs_text),
                "port": API_PORT,
            })
        elif self.path.startswith("/chronicle"):
            n = 10
            if "last=" in self.path:
                try: n = int(self.path.split("last=")[1].split("&")[0])
                except: pass
            entries = []
            if CHRONICLE_PATH.exists():
                with open(CHRONICLE_PATH) as f:
                    for line in f:
                        line = line.strip()
                        if line:
                            try: entries.append(json.loads(line))
                            except: pass
            self._json(entries[-n:])
        elif self.path == "/last":
            self._json({"response": last_obs_text, "turn": turn_count})
        else:
            self._json({"service":"Fractonaut","port":API_PORT,
                        "endpoints":["/status","/last","/chronicle?last=N","POST /ask"]})

    def do_POST(self):
        if self.path == "/ask":
            length = int(self.headers.get("Content-Length",0))
            body   = self.rfile.read(length)
            try: data = json.loads(body)
            except: self._json({"error":"bad json"},400); return
            q = data.get("question","").strip()
            if not q: self._json({"error":"missing question"},400); return
            evt    = threading.Event()
            holder = {"response": None}
            try:
                ask_queue.put_nowait({"question":q,"event":evt,"result":holder})
            except queue.Full:
                self._json({"error":"queue full"},503); return
            evt.wait(timeout=180)
            self._json({"response": holder["response"], "turn": turn_count, "model": MODEL})
        else:
            self._json({"error":"unknown"},404)

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin","*")
        self.send_header("Access-Control-Allow-Methods","GET,POST,OPTIONS")
        self.send_header("Access-Control-Allow-Headers","Content-Type")
        self.end_headers()


class ThreadedServer(ThreadingMixIn, HTTPServer):
    daemon_threads = True


def run_http():
    srv = ThreadedServer(("127.0.0.1", API_PORT), FractonautHandler)
    print(f"[FRACTONAUT] HTTP API on 127.0.0.1:{API_PORT}")
    sys.stdout.flush()
    while running:
        srv.handle_request()
    srv.server_close()


def main():
    global frame_count, latest_tel

    print("="*60)
    print("FRACTONAUT — pattern recognition observer")
    print(f"Model: {MODEL}  Port: {API_PORT}  ZMQ: {TELEMETRY_PORT}")
    print(f"Observe every {OBSERVE_INTERVAL} frames")
    print("="*60)
    sys.stdout.flush()

    CHRONICLE_PATH.parent.mkdir(parents=True, exist_ok=True)
    load_chronicle_tail()

    ctx = zmq.Context()
    tel_sub = ctx.socket(zmq.SUB)
    # Subscribe BEFORE connect (per Resonance Engine ZMQ rules)
    tel_sub.setsockopt_string(zmq.SUBSCRIBE, "")
    tel_sub.connect(f"tcp://127.0.0.1:{TELEMETRY_PORT}")
    # Slow-joiner sleep — first few frames after connect are dropped otherwise
    time.sleep(1.0)

    poller = zmq.Poller()
    poller.register(tel_sub, zmq.POLLIN)

    threading.Thread(target=run_http, daemon=True).start()
    print("[FRACTONAUT] Listening for telemetry...")
    sys.stdout.flush()

    while running:
        # Block up to 100ms waiting for a telemetry frame
        socks = dict(poller.poll(timeout=100))
        if tel_sub in socks:
            try:
                raw  = tel_sub.recv_string()
                data = json.loads(raw)
                telemetry_window.append(data)
                latest_tel = data
                frame_count += 1
            except json.JSONDecodeError:
                pass

        # Handle queued /ask requests
        try:
            item = ask_queue.get_nowait()
            if not ollama_lock.acquire(timeout=5):
                item["result"]["response"] = "(busy)"
                item["event"].set()
                continue
            try:
                stats  = compute_window_stats(telemetry_window)
                latest = telemetry_window[-1] if telemetry_window else {}
                window_str = format_window_for_prompt(stats, latest)
                past_str   = format_past_observations(past_observations)
                prompt = f"QUESTION: {item['question']}\n\nCURRENT WINDOW:\n{window_str}\n\nPAST OBSERVATIONS:\n{past_str}"
                messages = [
                    {"role":"system","content":SYSTEM},
                    {"role":"user","content":prompt},
                ]
                response = call_llm(messages)
                item["result"]["response"] = response or "(no response)"
                if response:
                    cycle = latest.get("cycle",0)
                    past_observations.append({"cycle": cycle, "text": f"[Q] {response}"})
                    append_chronicle(turn_count, cycle, prompt, response)
            finally:
                ollama_lock.release()
            item["event"].set()
        except queue.Empty:
            pass

        if frame_count > 0 and frame_count % OBSERVE_INTERVAL == 0:
            observe()

    tel_sub.close()
    ctx.term()
    print("[FRACTONAUT] Stopped.")


if __name__ == "__main__":
    main()