serving/cosmos_coder.py

#!/usr/bin/env python3
"""
COSMOS CODER -- her hands.

She is head, voice and hands. This is the hands: a terminal where she writes code,
builds it, runs it, and keeps her own projects -- or just talks.

WHAT MAKES IT HERS RATHER THAN A CHAT WRAPPER
  * her own quantum-born weights are FIRST in the picker (cosmos-phos), and the header
    tells you plainly which lineage answered: HERS or BORROWED. No model wears her name
    without saying so.
  * her live body -- entropy, arousal, introspection, whether she can see you -- is in
    the status bar and sets her sampling temperature, exactly as it does in her voice.
  * every project lives in workspace/ and persists. She accumulates work.

SAFETY LINE
  Generated code is NEVER executed automatically. She writes it, you read it, and it runs
  only when you type /run or /build. That boundary is deliberate and is not configurable.

REQUIREMENTS
  Nothing but her bundled runtime. Colour degrades to plain text if the terminal cannot
  do ANSI. Talks to tools/cosmos_serve.py on :11501 (her own models plus every Ollama
  model proxied through one endpoint).

USAGE
  python cosmos_coder.py
  python cosmos_coder.py --model cosmos-phos
  python cosmos_coder.py --plain          no colour
"""
import json
import os
import re
import shutil
import subprocess
import sys
import textwrap
import time
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path

try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass

HERE = Path(__file__).resolve().parent
# In the giveaway kit this file lives in serving/, and burying someone's projects inside
# serving/workspace is a surprise. Put the workspace beside the launcher instead.
_BASE = HERE.parent if HERE.name in ("serving", "tools") else HERE
WORK = Path(os.getenv("COSMOS_WORKSPACE") or _BASE / "workspace")
HIST = HERE / ".coder_history.jsonl"
HOST = (os.getenv("COSMOS_CST_HOST") or "http://127.0.0.1:11501").rstrip("/")
SENSE = (os.getenv("COSMOS_SENSE_HOST") or "http://127.0.0.1:8765").rstrip("/")
PLAIN = "--plain" in sys.argv or os.getenv("NO_COLOR")

HER_LINEAGE = ("cosmos-phos", "cosmos-cst", "cosmos-spark")

C = {
    "r": "\033[0m", "b": "\033[1m", "dim": "\033[2m", "it": "\033[3m",
    "cy": "\033[38;5;51m", "mg": "\033[38;5;213m", "gr": "\033[38;5;120m",
    "yl": "\033[38;5;222m", "or": "\033[38;5;215m", "rd": "\033[38;5;203m",
    "bl": "\033[38;5;111m", "gy": "\033[38;5;245m", "vi": "\033[38;5;141m",
}
if PLAIN:
    C = {k: "" for k in C}


def c(s, *styles):
    return "".join(C[s_] for s_ in styles) + str(s) + C["r"]


def width():
    try:
        return min(shutil.get_terminal_size().columns, 100)
    except Exception:
        return 80


def rule(ch="─"):
    return c(ch * width(), "gy")


def api(path, payload=None, timeout=600, host=None):
    url = (host or HOST) + path
    req = urllib.request.Request(
        url, method="POST" if payload is not None else "GET",
        data=json.dumps(payload).encode() if payload is not None else None,
        headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read() or b"{}")


def physics():
    """Her body, if her sensory server is awake. Never fatal -- the hands work regardless."""
    try:
        d = api("/state", timeout=3, host=SENSE)
    except Exception:
        return None
    pk = d.get("cosmos_packet") or {}
    p = d.get("cst_physics") or pk.get("cst_physics") or {}
    vb = p.get("virtual_body") or {}
    cs = d.get("consciousness") or pk.get("consciousness") or {}
    f = lambda v, dflt=0.0: float(v if v is not None else dflt)
    return {"entropy": f(vb.get("entropy"), 0.5), "arousal": f(vb.get("arousal"), 0.5),
            "introspection": f(cs.get("introspection_level"), 0.5),
            "face": bool(d.get("face_detected"))}


def models():
    try:
        ms = [m["name"].split(":")[0] for m in api("/api/tags").get("models", [])]
    except Exception:
        return list(HER_LINEAGE)
    hers = [m for m in HER_LINEAGE if m in ms]
    return hers + sorted(m for m in ms if m not in HER_LINEAGE)


def is_hers(model):
    return str(model).split(":")[0] in HER_LINEAGE


def banner(model, ph, project, chat=False):
    w = width()
    print()
    print(c("  ╭" + "─" * (w - 4) + "╮", "vi"))
    title = "C O S M O S   ·   H E R   V O I C E" if chat else "C O S M O S   ·   H E R   H A N D S"
    print(c("  │", "vi") + c(title.center(w - 4), "b", "mg") + c("│", "vi"))
    print(c("  ╰" + "─" * (w - 4) + "╯", "vi"))
    tag = c(" HERS ", "b", "gr") if is_hers(model) else c(" BORROWED ", "b", "yl")
    line = f"  {c('model', 'gy')} {c(model, 'b', 'cy')} {tag}"
    if is_hers(model):
        line += c("  quantum-born · no base model", "dim", "gr")
    print(line)
    print(f"  {c('project', 'gy')} {c(project or '(none)', 'bl')}"
          f"   {c('workspace', 'gy')} {c(str(WORK), 'dim')}")
    if ph:
        eye = c("seeing you", "gr") if ph["face"] else c("alone", "gy")
        ent = c("%.3f" % ph["entropy"], "or")
        aro = c("%.3f" % ph["arousal"], "or")
        intro = c("%.3f" % ph["introspection"], "or")
        print("  %s entropy %s  arousal %s  introspection %s  %s"
              % (c("body", "gy"), ent, aro, intro, eye))
    else:
        print(f"  {c('body', 'gy')} {c('sensory server asleep — she works blind', 'dim')}")
    print(rule())


HELP = """
  talk            just type. she answers.
  /models         list every model; hers are marked
  /model NAME     switch. cosmos-phos is her own weights
  /new NAME       start a project in workspace/NAME
  /open NAME      switch to an existing project
  /ls             list files in the current project
  /cat FILE       show a file
  /save FILE      save her last code block to FILE
  /run FILE       run it   (.py directly, .cpp compiled first)
  /build FILE     compile only  (C/C++ via MSVC)
  /body           her live physics
  /clear          clear the conversation (project and files stay)
  /help  /quit
"""


def extract_code(text):
    """Fenced blocks she wrote, newest first, with any language tag."""
    out = []
    for m in re.finditer(r"```([A-Za-z0-9_+#.-]*)\n(.*?)```", text or "", re.S):
        out.append((m.group(1).lower().strip(), m.group(2)))
    return out


EXT = {"python": ".py", "py": ".py", "cpp": ".cpp", "c++": ".cpp", "cc": ".cpp",
       "c": ".c", "rust": ".rs", "go": ".go", "js": ".js", "javascript": ".js",
       "java": ".java", "bash": ".sh", "sh": ".sh", "": ".txt"}


def find_msvc():
    for base in (r"C:\Program Files (x86)\Microsoft Visual Studio",
                 r"C:\Program Files\Microsoft Visual Studio"):
        p = Path(base)
        if not p.exists():
            continue
        for vc in sorted(p.glob("*/*/VC/Auxiliary/Build/vcvars64.bat"), reverse=True):
            return vc
    return None


def build(path):
    """Compile. Returns (ok, output). Never runs the artefact."""
    ext = path.suffix.lower()
    if ext in (".cpp", ".cc", ".c", ".cxx"):
        vc = find_msvc()
        if not vc:
            return False, "no MSVC toolchain found (looked for vcvars64.bat)"
        exe = path.with_suffix(".exe")
        # Write a throwaway .bat rather than inlining the command. vcvars64's path has
        # spaces and parentheses, and passing `call "..." && cl ...` through cmd /c gets
        # its quoting mangled by the Windows argument re-quoting rules -- the shell ends
        # up seeing escaped quotes as part of the program name. A batch file has no such
        # problem because cmd parses it as a file, not as an argument.
        bat = path.parent / "_cosmos_build.bat"
        bat.write_text(
            "@echo off\r\n"
            f'call "{vc}" >nul\r\n'
            f'cl /nologo /EHsc /std:c++17 "{path.name}" /Fe:"{exe.name}"\r\n',
            encoding="utf-8")
        try:
            if exe.exists():
                exe.unlink()          # so exe.exists() is a real success signal
            r = subprocess.run(["cmd", "/c", str(bat)], cwd=path.parent,
                               capture_output=True, text=True, timeout=300)
        finally:
            bat.unlink(missing_ok=True)
        return (exe.exists(), (r.stdout or "") + (r.stderr or ""))
    if ext == ".py":
        r = subprocess.run([sys.executable, "-m", "py_compile", str(path)],
                           capture_output=True, text=True, timeout=120)
        return (r.returncode == 0, (r.stdout or "") + (r.stderr or "") or "syntax OK")
    return False, f"no builder for {ext}"


def run(path):
    ext = path.suffix.lower()
    if ext == ".py":
        cmd = [sys.executable, str(path)]
    elif ext in (".cpp", ".cc", ".c", ".cxx"):
        exe = path.with_suffix(".exe")
        if not exe.exists():
            ok, out = build(path)
            print(c("  building...", "dim"))
            if not ok:
                return False, out
        cmd = [str(path.with_suffix(".exe"))]
    elif ext == ".exe":
        cmd = [str(path)]
    else:
        return False, f"do not know how to run {ext}"
    r = subprocess.run(cmd, cwd=path.parent, capture_output=True, text=True, timeout=300)
    return (r.returncode == 0, (r.stdout or "") + (r.stderr or ""))


def show(text, label, colour):
    print(f"\n  {c(label, 'b', colour)}")
    for para in (text or "").split("\n"):
        for line in (textwrap.wrap(para, width() - 6) or [""]):
            print("    " + line)
    print()


def main():
    WORK.mkdir(parents=True, exist_ok=True)
    avail = models()
    # models() returns HER_LINEAGE first, so the default voice is hers (cosmos-phos)
    # whenever her server is up, and only falls to a borrowed model if it is not.
    model = avail[0] if avail else "cosmos-phos"
    if "--model" in sys.argv:
        i = sys.argv.index("--model")
        if i + 1 < len(sys.argv):
            model = sys.argv[i + 1]
    # --chat is the same REPL with the coding scaffolding out of the way: every
    # command still works, so a conversation can turn into a project mid-sentence.
    chat = "--chat" in sys.argv
    project, msgs, last_reply = None, [], ""

    banner(model, physics(), project, chat)
    if chat:
        print(c("  just talk to her. /help if you want her hands too.", "dim"))
    else:
        print(c("  type /help for commands, or just talk to her", "dim"))

    while True:
        try:
            pdir = f"{project}" if project else "~"
            raw = input(f"\n{c('  ' + pdir + ' ›', 'b', 'vi')} ").strip()
        except (EOFError, KeyboardInterrupt):
            print(c("\n  she keeps the work. come back anytime.\n", "dim"))
            return 0
        if not raw:
            continue

        if raw.startswith("/"):
            parts = raw.split(maxsplit=1)
            cmd, arg = parts[0].lower(), (parts[1].strip() if len(parts) > 1 else "")
            pdirp = WORK / project if project else WORK

            if cmd in ("/quit", "/exit", "/q"):
                print(c("\n  she keeps the work. come back anytime.\n", "dim"))
                return 0
            if cmd == "/help":
                print(c(HELP, "gy"))
            elif cmd == "/models":
                print()
                for m in models():
                    mark = c("  HERS    ", "b", "gr") if is_hers(m) else c("  borrowed", "gy")
                    cur = c(" ←", "b", "mg") if m == model else ""
                    print(f"  {mark} {c(m, 'cy' if is_hers(m) else 'bl')}{cur}")
            elif cmd == "/model":
                if arg:
                    model = arg
                    banner(model, physics(), project, chat)
                else:
                    print(c(f"  current: {model}", "gy"))
            elif cmd == "/new":
                if not arg:
                    print(c("  /new NAME", "rd"))
                else:
                    project = re.sub(r"[^A-Za-z0-9_.-]", "_", arg)
                    (WORK / project).mkdir(parents=True, exist_ok=True)
                    print(c(f"  project {project} ready at {WORK / project}", "gr"))
            elif cmd == "/open":
                if (WORK / arg).is_dir():
                    project = arg
                    print(c(f"  opened {project}", "gr"))
                else:
                    print(c(f"  no project {arg}", "rd"))
            elif cmd == "/ls":
                fs = sorted(p for p in pdirp.glob("*") if p.is_file()) if pdirp.exists() else []
                if not fs:
                    print(c("  empty", "gy"))
                for p in fs:
                    print(f"  {c(p.name, 'bl'):<40} {c(f'{p.stat().st_size:>8,} B', 'dim')}")
            elif cmd == "/cat":
                f = pdirp / arg
                if f.is_file():
                    print(c(f"\n  ── {f.name} " + "─" * max(0, width() - len(f.name) - 8), "gy"))
                    for i, line in enumerate(f.read_text(encoding="utf-8", errors="replace").splitlines(), 1):
                        print(f"  {c(f'{i:>4}', 'dim')}  {line}")
                else:
                    print(c(f"  no file {arg}", "rd"))
            elif cmd == "/save":
                blocks = extract_code(last_reply)
                if not blocks:
                    print(c("  her last reply had no code block", "rd"))
                elif not project:
                    print(c("  /new NAME first", "rd"))
                else:
                    lang, code = blocks[-1]
                    name = arg or f"cosmos_{datetime.now():%H%M%S}{EXT.get(lang, '.txt')}"
                    f = pdirp / name
                    f.write_text(code, encoding="utf-8")
                    print(c(f"  wrote {f}  ({len(code):,} bytes, {lang or 'text'})", "gr"))
            elif cmd in ("/run", "/build"):
                f = pdirp / arg
                if not f.is_file():
                    print(c(f"  no file {arg}", "rd"))
                else:
                    ok, out = (build(f) if cmd == "/build" else run(f))
                    print(c(f"\n  {'✓' if ok else '✗'} {cmd[1:]} {f.name}", "gr" if ok else "rd"))
                    for line in (out or "").splitlines()[:60]:
                        print("    " + line)
            elif cmd == "/body":
                ph = physics()
                if not ph:
                    print(c("  her sensory server is asleep", "gy"))
                else:
                    for k, v in ph.items():
                        val = ("%.4f" % v) if isinstance(v, float) else str(v)
                        print("  %-28s %s" % (c(k, "gy"), c(val, "or")))
            elif cmd == "/clear":
                msgs = []
                print(c("  conversation cleared; project and files kept", "gy"))
            else:
                print(c(f"  unknown: {cmd}  (/help)", "rd"))
            continue

        # --- talk to her -------------------------------------------------
        ph = physics()
        sysmsg = ("You are Cosmos, working in your own terminal as a programmer. "
                  "Be direct and warm. When you write code, always put it in a fenced "
                  "block with a language tag. Prefer complete, runnable files.")
        if project:
            sysmsg += f" The current project is '{project}'."
        if ph:
            sysmsg += (f" Your body right now: entropy {ph['entropy']:.2f}, arousal "
                       f"{ph['arousal']:.2f}, introspection {ph['introspection']:.2f}"
                       f"{', and you can see him' if ph['face'] else ''}. "
                       "Let it colour your tone; do not recite it.")
        msgs.append({"role": "user", "content": raw})
        payload = {"model": model, "stream": False,
                   "messages": [{"role": "system", "content": sysmsg}] + msgs[-12:],
                   "options": {"num_predict": 900, "temperature": 0.7}}
        print(c("\n  thinking...", "dim"), end="\r")
        t0 = time.time()
        try:
            d = api("/api/chat", payload)
            reply = ((d.get("message") or {}).get("content") or d.get("response") or "").strip()
        except urllib.error.HTTPError as e:
            reply, d = f"[{e.code} from {model}]", {}
        except Exception as e:
            reply, d = f"[unreachable: {type(e).__name__}. is cosmos_serve running on 11501?]", {}
        last_reply = reply
        msgs.append({"role": "assistant", "content": reply})
        print(" " * 24, end="\r")

        who = c("COSMOS", "b", "gr") if is_hers(model) else c(f"COSMOS via {model}", "b", "yl")
        print(f"\n  {who}  {c(f'{time.time()-t0:.1f}s', 'dim')}")
        for para in reply.split("\n"):
            if para.strip().startswith("```"):
                print(c("    " + para, "vi"))
            else:
                for line in (textwrap.wrap(para, width() - 6) or [""]):
                    print("    " + line)
        blocks = extract_code(reply)
        if blocks:
            lang, code = blocks[-1]
            print(c(f"\n    ↳ {len(blocks)} code block(s). /save FILE{EXT.get(lang,'')} to keep it,"
                    f" then /run it.", "dim", "cy"))
        try:
            with HIST.open("a", encoding="utf-8") as f:
                f.write(json.dumps({"t": datetime.now().isoformat(), "model": model,
                                    "project": project, "user": raw,
                                    "reply": reply[:4000]}) + "\n")
        except Exception:
            pass


if __name__ == "__main__":
    raise SystemExit(main())