synapse_swarm.py

#!/usr/bin/env python3
"""
synapse_swarm.py — Portable minimal session daemon for the Synapse_COR bundle.
Version: 1.0.0
License: MIT

Provides OpenSwarm v2.0.8-compatible session HTTP API backed by the bundle's
portable call_llm_with_fallback() instead of FUTRON cascade-v3.

No FUTRON installation required. Works standalone.

Environment variables:
  SYNAPSE_SWARM_PORT      HTTP port (default: 9055)
  SYNAPSE_SWARM_HOST      Bind hostname (default: 127.0.0.1)
  SYNAPSE_SWARM_HOME      State directory (default: ~/.synapse-swarm)
  SYNAPSE_FORGE_URL       External LLM endpoint (optional — auto-detected)
  SYNAPSE_FORGE_API_KEY   API key (optional)
  SYNAPSE_FORGE_MODEL     Default model

Subcommands:
  serve  --port 9055 [--hostname 127.0.0.1] [--log-level INFO]
  start | stop | status | logs [-f]
  sessions list | show <id> | delete <id> | gc
"""

import argparse
import datetime
import json
import os
import re
import signal
import socket
import subprocess
import sys
import time
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from threading import Lock
from urllib.parse import urlparse

# ── Paths ──────────────────────────────────────────────────────────────────────
HOME = Path.home()
SWARM_HOME = Path(os.environ.get("SYNAPSE_SWARM_HOME", str(HOME / ".synapse-swarm")))
SESSIONS_DIR = SWARM_HOME / "sessions"
PID_FILE = SWARM_HOME / "swarm.pid"
LOG_FILE = SWARM_HOME / "serve.log"

SWARM_HOME.mkdir(parents=True, exist_ok=True)
SESSIONS_DIR.mkdir(parents=True, exist_ok=True)

_sessions_lock = Lock()
_start_time = time.time()
_log_level = 20  # INFO


def _log(level: str, msg: str):
    levels = {"DEBUG": 10, "INFO": 20, "WARN": 30, "ERROR": 40}
    if levels.get(level, 20) < _log_level:
        return
    ts = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
    line = f"[{ts}] [{level}] {msg}\n"
    sys.stderr.write(line)
    try:
        with open(LOG_FILE, "a") as f:
            f.write(line)
    except OSError:
        pass


def info(m): _log("INFO", m)
def warn(m): _log("WARN", m)
def error(m): _log("ERROR", m)


# ── LLM dispatch (portable — no FUTRON dependency) ────────────────────────────
def _call_llm(prompt: str, model_id: str = "") -> tuple[str, int]:
    """
    Portable LLM dispatch for the bundle.
    Tries: bundle's call_llm_with_fallback → Gemini free → Pollinations → error.
    """
    try:
        bundle_dir = Path(__file__).parent
        sys.path.insert(0, str(bundle_dir))
        from synapse_forge import call_llm_with_fallback  # type: ignore
        reply = call_llm_with_fallback(prompt)
        if reply and len(reply.strip()) >= 1:
            return reply.strip(), 0
    except Exception as e:
        warn(f"bundle LLM failed: {e}")

    # Gemini 2.5 Flash free tier — no API key required with key from env
    gemini_key = os.environ.get("GEMINI_API_KEY", "")
    if gemini_key:
        try:
            import urllib.request
            payload = json.dumps({
                "contents": [{"role": "user", "parts": [{"text": prompt}]}]
            }).encode()
            req = urllib.request.Request(
                f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={gemini_key}",
                data=payload,
                headers={"Content-Type": "application/json"},
                method="POST",
            )
            with urllib.request.urlopen(req, timeout=30) as r:
                data = json.loads(r.read())
                text = data["candidates"][0]["content"]["parts"][0]["text"]
                return text.strip(), 0
        except Exception as e:
            warn(f"Gemini fallback failed: {e}")

    # Pollinations — truly zero-config free
    try:
        import urllib.request
        payload = json.dumps({
            "model": "openai-fast",
            "messages": [{"role": "user", "content": prompt}],
        }).encode()
        req = urllib.request.Request(
            "https://api.pollinations.ai/v1/chat/completions",
            data=payload,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=30) as r:
            data = json.loads(r.read())
            return data["choices"][0]["message"]["content"].strip(), 0
    except Exception as e:
        warn(f"Pollinations fallback failed: {e}")

    return "[synapse-swarm: all LLM providers exhausted]", 1


# ── Session helpers ────────────────────────────────────────────────────────────
def _session_path(sid: str) -> Path:
    return SESSIONS_DIR / f"{sid}.json"


def _load_session(sid: str):
    p = _session_path(sid)
    if not p.exists():
        return None
    try:
        with open(p) as f:
            return json.load(f)
    except Exception:
        return None


def _save_session(session: dict):
    p = _session_path(session["id"])
    with _sessions_lock:
        tmp = str(p) + ".tmp"
        with open(tmp, "w") as f:
            json.dump(session, f, indent=2)
        os.replace(tmp, str(p))


def _list_sessions() -> list:
    result = []
    for p in sorted(SESSIONS_DIR.glob("*.json")):
        try:
            with open(p) as f:
                s = json.load(f)
            result.append({
                "id": s["id"], "slug": s.get("slug", ""),
                "projectID": s.get("projectID", ""),
                "created": s.get("created", ""), "updated": s.get("updated", ""),
                "message_count": len(s.get("messages", [])),
            })
        except Exception:
            pass
    return result


def _slug(n=3) -> str:
    import random
    words = ["alpha","bravo","charlie","delta","echo","forge","ghost",
             "hotel","india","juliet","kilo","lima","mike","nova",
             "oscar","papa","romeo","sierra","tango","uniform"]
    return "-".join(random.choice(words) for _ in range(n))


def _new_session(project_id="", title="") -> dict:
    sid = str(uuid.uuid4())
    now = datetime.datetime.utcnow().isoformat() + "Z"
    session = {
        "id": sid, "slug": _slug(),
        "projectID": project_id,
        "title": title or f"session-{sid[:8]}",
        "created": now, "updated": now,
        "messages": [],
    }
    _save_session(session)
    return session


def _extract_text(parts: list) -> str:
    for p in parts:
        if p.get("type") == "text":
            return p.get("text", "")
    return ""


def _build_prompt(messages: list, new_text: str) -> str:
    parts = []
    for msg in messages:
        label = "HUMAN" if msg.get("role") == "user" else "ASSISTANT"
        text = _extract_text(msg.get("parts", []))
        if text:
            parts.append(f"{label}: {text}")
    parts.append(f"HUMAN: {new_text}")
    full = "\n\n".join(parts)
    if len(full) > 32_000:
        full = "...[history truncated]\n\n" + full[-32_000:]
    return full


def _make_envelope(reply_text: str, model_id: str = "") -> dict:
    return {
        "id": str(uuid.uuid4()),
        "role": "assistant",
        "modelID": model_id or "synapse-swarm/portable",
        "parts": [
            {"type": "reasoning", "text": ""},
            {"type": "text", "text": reply_text},
            {"type": "step-finish", "finishReason": "stop"},
        ],
    }


# ── HTTP handler ──────────────────────────────────────────────────────────────
class SwarmHandler(BaseHTTPRequestHandler):

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

    def _json(self, code, obj):
        body = json.dumps(obj, indent=2).encode()
        self.send_response(code)
        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 _empty(self, code):
        self.send_response(code)
        self.send_header("Content-Length", "0")
        self.end_headers()

    def _body(self) -> dict:
        n = int(self.headers.get("Content-Length", 0))
        if n == 0:
            return {}
        try:
            return json.loads(self.rfile.read(n))
        except Exception:
            return {}

    def do_OPTIONS(self):
        self.send_response(200)
        for k, v in [("Access-Control-Allow-Origin", "*"),
                     ("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS"),
                     ("Access-Control-Allow-Headers", "Content-Type, Authorization"),
                     ("Content-Length", "0")]:
            self.send_header(k, v)
        self.end_headers()

    def do_GET(self):
        path = urlparse(self.path).path.rstrip("/")
        if path == "/health":
            self._json(200, {
                "ok": True, "daemon": "synapse-swarm", "version": "1.0.0",
                "portable": True,
                "sessions": len(list(SESSIONS_DIR.glob("*.json"))),
                "uptime_s": int(time.time() - _start_time),
            })
        elif path == "/session":
            self._json(200, _list_sessions())
        else:
            m = re.match(r"^/session/([a-z0-9\-]+)$", path)
            if m:
                s = _load_session(m.group(1))
                self._json(200, s) if s else self._json(404, {"error": "not found"})
            else:
                self._json(404, {"error": f"not found: {path}"})

    def do_POST(self):
        path = urlparse(self.path).path.rstrip("/")
        if path == "/session":
            body = self._body()
            s = _new_session(body.get("projectID", ""), body.get("title", ""))
            self._json(200, {"id": s["id"], "slug": s["slug"],
                             "projectID": s["projectID"], "created": s["created"]})
            return
        m = re.match(r"^/session/([a-z0-9\-]+)/message$", path)
        if m:
            sid = m.group(1)
            session = _load_session(sid)
            if not session:
                self._json(404, {"error": "session not found"})
                return
            body = self._body()
            parts = body.get("parts", [])
            model_id = body.get("modelID", "")
            new_text = _extract_text(parts)
            if not new_text:
                self._json(400, {"error": "no text in parts"})
                return
            now = datetime.datetime.utcnow().isoformat() + "Z"
            session["messages"].append({
                "id": str(uuid.uuid4()), "role": body.get("role", "user"),
                "modelID": model_id, "parts": parts, "created": now,
            })
            prompt = _build_prompt(session["messages"][:-1], new_text)
            reply_text, rc = _call_llm(prompt, model_id)
            env = _make_envelope(reply_text, model_id)
            env["created"] = datetime.datetime.utcnow().isoformat() + "Z"
            session["messages"].append(env)
            session["updated"] = env["created"]
            _save_session(session)
            self._json(200, env)
            return
        self._json(404, {"error": f"not found: {path}"})

    def do_DELETE(self):
        path = urlparse(self.path).path.rstrip("/")
        m = re.match(r"^/session/([a-z0-9\-]+)$", path)
        if m:
            p = _session_path(m.group(1))
            if not p.exists():
                self._json(404, {"error": "not found"})
                return
            p.unlink()
            self._empty(204)
            return
        self._json(404, {"error": f"not found: {path}"})


# ── Server + daemon control ────────────────────────────────────────────────────
def cmd_serve(port: int, hostname: str, log_level: str):
    global _log_level
    _log_level = {"DEBUG": 10, "INFO": 20, "WARN": 30, "ERROR": 40}.get(log_level.upper(), 20)
    with open(PID_FILE, "w") as f:
        f.write(str(os.getpid()))

    def _cleanup(sig, frame):
        info("synapse-swarm shutting down")
        try: PID_FILE.unlink()
        except Exception: pass
        sys.exit(0)

    signal.signal(signal.SIGTERM, _cleanup)
    signal.signal(signal.SIGINT, _cleanup)
    server = HTTPServer((hostname, port), SwarmHandler)
    info(f"synapse-swarm (portable) listening on http://{hostname}:{port}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        _cleanup(None, None)


def _is_running():
    if not PID_FILE.exists():
        return None
    try:
        pid = int(PID_FILE.read_text().strip())
        os.kill(pid, 0)
        return pid
    except Exception:
        return None


def cmd_start(port=9055, hostname="127.0.0.1", log_level="INFO"):
    if _is_running():
        print(f"[swarm] already running (PID {_is_running()})")
        return
    proc = subprocess.Popen(
        [sys.executable, __file__, "serve", "--port", str(port),
         "--hostname", hostname, "--log-level", log_level],
        stdout=open(LOG_FILE, "a"), stderr=subprocess.STDOUT, start_new_session=True,
    )
    deadline = time.time() + 10
    while time.time() < deadline:
        try:
            s = socket.create_connection((hostname, port), timeout=1); s.close()
            print(f"[swarm] LIVE at http://{hostname}:{port} (PID {proc.pid})")
            return
        except OSError:
            time.sleep(0.5)
    print(f"[swarm] started (PID {proc.pid}) — check {LOG_FILE}")


def cmd_stop():
    pid = _is_running()
    if not pid:
        print("[swarm] not running"); return
    os.kill(pid, signal.SIGTERM)
    time.sleep(1)
    if _is_running():
        os.kill(pid, signal.SIGKILL)
    try: PID_FILE.unlink()
    except Exception: pass
    print(f"[swarm] stopped (was PID {pid})")


def cmd_status(port=9055, hostname="127.0.0.1"):
    pid = _is_running()
    if not pid:
        print("[swarm] STOPPED"); sys.exit(1)
    print(f"[swarm] RUNNING (PID {pid})")
    try:
        import urllib.request
        with urllib.request.urlopen(f"http://{hostname}:{port}/health", timeout=2) as r:
            d = json.loads(r.read())
        print(f"  HTTP: OK  sessions={d['sessions']}  uptime={d['uptime_s']}s")
    except Exception as e:
        print(f"  HTTP: not responding ({e})")


def main():
    p = argparse.ArgumentParser(prog="synapse-swarm",
        description="Portable session daemon (bundle variant — no FUTRON required)")
    sub = p.add_subparsers(dest="cmd")

    ps = sub.add_parser("serve")
    ps.add_argument("--port", type=int, default=int(os.environ.get("SYNAPSE_SWARM_PORT", 9055)))
    ps.add_argument("--hostname", default=os.environ.get("SYNAPSE_SWARM_HOST", "127.0.0.1"))
    ps.add_argument("--log-level", default="INFO")

    pst = sub.add_parser("start")
    pst.add_argument("--port", type=int, default=9055)
    pst.add_argument("--hostname", default="127.0.0.1")
    pst.add_argument("--log-level", default="INFO")

    sub.add_parser("stop")
    sub.add_parser("status")
    pl = sub.add_parser("logs")
    pl.add_argument("-f", "--follow", action="store_true")

    psess = sub.add_parser("sessions")
    ss = psess.add_subparsers(dest="sessions_cmd")
    ss.add_parser("list")
    psh = ss.add_parser("show"); psh.add_argument("id")
    pdel = ss.add_parser("delete"); pdel.add_argument("id")
    pgc = ss.add_parser("gc"); pgc.add_argument("--max-age-days", type=int, default=30)

    args = p.parse_args()
    if args.cmd == "serve":
        cmd_serve(args.port, args.hostname, args.log_level)
    elif args.cmd == "start":
        cmd_start(args.port, args.hostname, args.log_level)
    elif args.cmd == "stop":
        cmd_stop()
    elif args.cmd == "status":
        cmd_status()
    elif args.cmd == "logs":
        if LOG_FILE.exists():
            subprocess.run(["tail", "-f" if args.follow else "-100", str(LOG_FILE)])
    elif args.cmd == "sessions":
        if args.sessions_cmd == "list":
            sessions = _list_sessions()
            if not sessions: print("[swarm] no sessions"); return
            for s in sessions:
                print(f"{s['id'][:8]}  {s.get('slug',''):20s}  msgs={s.get('message_count',0)}")
        elif args.sessions_cmd == "show":
            s = _load_session(args.id)
            print(json.dumps(s, indent=2) if s else f"not found: {args.id}")
        elif args.sessions_cmd == "delete":
            p2 = _session_path(args.id)
            if p2.exists(): p2.unlink(); print(f"deleted {args.id}")
            else: print(f"not found: {args.id}")
        elif args.sessions_cmd == "gc":
            cutoff = time.time() - (args.max_age_days * 86400)
            removed = sum(1 for p2 in SESSIONS_DIR.glob("*.json")
                          if p2.stat().st_mtime < cutoff and p2.unlink() is None)
            print(f"[swarm] gc: removed {removed} sessions")
    else:
        p.print_help(); sys.exit(1)


if __name__ == "__main__":
    main()