"""darwin-term - standalone WS→PTY→SSH terminal bridge for Darwin agent panes.

Wire protocol (compatible with standard xterm.js client):

  WS /ws?target=&agent=&mode=&session=&cwd=[&token=]

    client→server:

      binary frames  - raw keystrokes/input bytes sent to PTY

      text JSON      - {"t":"resize","cols":N,"rows":N}  → TIOCSWINSZ

    server→client:

      binary frames  - raw PTY output

      text (string)  - banner lines + JSON control:

                        {"_cockpit":"session","name":<tmux-name>}  (persistent mode only)

Auth: DARWIN_TERM_TOKEN env var. Pass as ?token= query param or

      Authorization: Bearer <token> header. Missing/wrong → close 4403.

      (Future: replace with Authentik SSO.)

Targets: ct215 (REDACTED-IP), ct241 (REDACTED-IP), ct247 (REDACTED-IP)

SSH key: /root/.ssh/termius_homelab  (root auth, key-based, no password)

Port: 9894

"""

import asyncio

import fcntl

import json

import logging

import os

import pty

import re

import select

import shlex

import struct

import termios

import uuid

from contextlib import asynccontextmanager

import uvicorn

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query

from fastapi.responses import JSONResponse

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

logger = logging.getLogger("darwin-term")

SSH_KEY = os.environ.get("DARWIN_TERM_SSH_KEY", "/root/.ssh/termius_homelab")

KNOWN_HOSTS = "/root/.ssh/known_hosts"

DARWIN_TERM_TOKEN = os.environ.get("DARWIN_TERM_TOKEN", "")

COMMON_DIRS = [

    {"label": "Default (home ~)", "path": ""},

    {"label": "Projects - /shared/projects", "path": "/shared/projects"},

    {"label": "Oversight - /shared/projects/Oversight", "path": "/shared/projects/Oversight"},

]

TARGETS: dict[str, dict] = {

    "ct215": {

        "ip": "REDACTED-IP",

        "label": "Claude Code - CT215",

        "agents": ["claude", "shell"],

        "default_agent": "claude",

        "default_cwd": "/shared/projects",

    },

    "ct241": {

        "ip": "REDACTED-IP",

        "label": "Codex - CT241",

        "agents": ["codex", "shell"],

        "default_agent": "codex",

        "default_cwd": "/shared/projects",

    },

    "ct247": {

        "ip": "REDACTED-IP",

        "label": "OpenCode / GLM - CT247",

        "agents": ["opencode", "shell"],

        "default_agent": "opencode",

        "default_cwd": "",

    },

}

AGENT_CMD: dict[str, str] = {

    "claude": "CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=0 IS_SANDBOX=1 claude --dangerously-skip-permissions",

    "codex": "codex",

    "opencode": "opencode",

}

SESSION_PREFIXES = tuple(f"{a}-" for a in list(AGENT_CMD) + ["shell"])

_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]{1,80}$")

_CWD_RE  = re.compile(r"^[A-Za-z0-9_./~ -]{1,256}$")

def _check_token(token_param: str | None, auth_header: str | None) -> bool:

    """Return True if the request carries the correct bearer token."""

    if not DARWIN_TERM_TOKEN:

        return True

    candidate = token_param or ""

    if not candidate and auth_header and auth_header.lower().startswith("bearer "):

        candidate = auth_header[7:]

    return candidate == DARWIN_TERM_TOKEN

def _ssh_argv(ip: str) -> list[str]:

    return [

        "ssh", "-tt",

        "-i", SSH_KEY,

        "-o", "BatchMode=yes",

        "-o", "ConnectTimeout=10",

        "-o", "StrictHostKeyChecking=accept-new",

        "-o", f"UserKnownHostsFile={KNOWN_HOSTS}",

        "-o", "ServerAliveInterval=20",

        f"root@{ip}",

    ]

def _tmux_attach(session: str, payload: str) -> str:

    """Create-or-attach a tmux session with deep history and mouse OFF.

    `mouse off` keeps terminal output in xterm.js's own scrollback (the browser

    viewport) instead of being captured into tmux copy-mode - which touch can't

    drive - so the phone can finger-scroll terminal history. A large history-limit

    still keeps the underlying tmux buffer deep across reattach."""

    q = shlex.quote(session)

    create = f"tmux new-session -d -A -s {q} {shlex.quote(payload)}"

    core = (

        "tmux set-option -g mouse off >/dev/null 2>&1; "

        "tmux set-option -g history-limit 50000 >/dev/null 2>&1; "

        "tmux set-option -g window-size smallest >/dev/null 2>&1; "

        "tmux set-option -g aggressive-resize on >/dev/null 2>&1"

    )

    skin = "tmux set-option -g status-style 'bg=colour234,fg=colour37' >/dev/null 2>&1 || true"

    attach = f"tmux attach-session -t {q}"

    return f"{create}; {core}; {skin}; {attach}"

def _remote_command(agent: str, mode: str, session: str, cwd: str | None) -> str:

    if agent == "shell":

        inner = "bash -l"

        if cwd:

            inner = f"bash -lc {shlex.quote(f'cd {shlex.quote(cwd)} && exec bash -l')}"

        if mode == "persistent":

            payload = inner if cwd else "bash -l"

            return _tmux_attach(session, payload)

        return inner

    base = AGENT_CMD[agent]

    if cwd:

        base = f"cd {shlex.quote(cwd)} && {base}"

    run = f"bash -lc {shlex.quote(base)}"

    if mode == "persistent":

        return _tmux_attach(session, run)

    return run

def _valid_session(name: str) -> bool:

    return bool(name) and bool(_NAME_RE.match(name)) and name.startswith(SESSION_PREFIXES)

def _read_fd(fd: int) -> bytes | None:

    """Blocking read via select; None = no data yet, b'' = EOF."""

    r, _, _ = select.select([fd], [], [], 1.0)

    if r:

        try:

            data = os.read(fd, 65536)

            return data if data else b""

        except OSError:

            return b""

    return None

async def _run_ssh(ip: str, remote_cmd: str, timeout: float = 12.0) -> tuple[int, str, str]:

    argv = [

        "ssh", "-i", SSH_KEY,

        "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",

        "-o", "StrictHostKeyChecking=accept-new",

        "-o", f"UserKnownHostsFile={KNOWN_HOSTS}",

        f"root@{ip}", remote_cmd,

    ]

    proc = await asyncio.create_subprocess_exec(

        *argv,

        stdout=asyncio.subprocess.PIPE,

        stderr=asyncio.subprocess.PIPE,

    )

    try:

        out, err = await asyncio.wait_for(proc.communicate(), timeout=timeout)

    except asyncio.TimeoutError:

        try:

            proc.kill()

        except ProcessLookupError:

            pass

        return -1, "", "timeout"

    return proc.returncode, out.decode(errors="replace"), err.decode(errors="replace")

async def _list_sessions() -> list[dict]:

    results: list[dict] = []

    async def one(key: str, meta: dict):

        _rc, out, _err = await _run_ssh(

            meta["ip"],

            "tmux list-sessions -F '#{session_name}|#{session_created}|#{session_attached}' 2>/dev/null || true",

        )

        for line in out.splitlines():

            parts = line.strip().split("|")

            if not parts or not parts[0]:

                continue

            name = parts[0]

            if not name.startswith(SESSION_PREFIXES):

                continue

            results.append({

                "target": key,

                "label": meta["label"],

                "session": name,

                "agent": name.split("-", 1)[0],

                "created": parts[1] if len(parts) > 1 else "",

                "attached": (parts[2] == "1") if len(parts) > 2 else False,

            })

    await asyncio.gather(*(one(k, m) for k, m in TARGETS.items()))

    results.sort(key=lambda r: (r["target"], r["session"]))

    return results

@asynccontextmanager

async def lifespan(_app: FastAPI):

    if not DARWIN_TERM_TOKEN:

        logger.warning("DARWIN_TERM_TOKEN is not set - service is OPEN, set it in /opt/darwin/.env")

    else:

        logger.info("darwin-term auth: token configured (length %d)", len(DARWIN_TERM_TOKEN))

    logger.info("SSH key: %s", SSH_KEY)

    yield

app = FastAPI(title="darwin-term", version="1.0.0", lifespan=lifespan)

@app.get("/health")

async def health():

    return {"status": "ok", "service": "darwin-term"}

@app.get("/targets")

async def targets(

    token: str | None = Query(default=None),

    authorization: str | None = None,

):

    if not _check_token(token, authorization):

        return JSONResponse({"error": "unauthorized"}, status_code=403)

    return {

        "targets": [

            {

                "key": k,

                "ip": m["ip"],

                "label": m["label"],

                "agents": m["agents"],

                "default_agent": m["default_agent"],

                "default_cwd": m.get("default_cwd", ""),

                "dirs": COMMON_DIRS,

            }

            for k, m in TARGETS.items()

        ],

        "agent_cmds": AGENT_CMD,

        "common_dirs": COMMON_DIRS,

    }

@app.get("/sessions")

async def sessions(

    token: str | None = Query(default=None),

    authorization: str | None = None,

):

    if not _check_token(token, authorization):

        return JSONResponse({"error": "unauthorized"}, status_code=403)

    return {"sessions": await _list_sessions()}

@app.post("/kill")

async def kill(

    target: str | None = Query(default=None),

    session: str | None = Query(default=None),

    token: str | None = Query(default=None),

    authorization: str | None = None,

):

    if not _check_token(token, authorization):

        return JSONResponse({"error": "unauthorized"}, status_code=403)

    if target not in TARGETS:

        return JSONResponse({"error": "unknown target"}, status_code=400)

    if not _valid_session(session or ""):

        return JSONResponse({"error": "invalid session"}, status_code=400)

    meta = TARGETS[target]

    rc, _out, err = await _run_ssh(

        meta["ip"],

        f"tmux kill-session -t {session} 2>&1 || true",

    )

    return {"ok": rc == 0, "session": session, "detail": err.strip()}

@app.websocket("/ws")

async def terminal_ws(websocket: WebSocket):

    q = websocket.query_params

    token_param = q.get("token")

    auth_header = websocket.headers.get("authorization")

    if not _check_token(token_param, auth_header):

        await websocket.close(code=4403)

        return

    target  = q.get("target", "")

    agent   = q.get("agent", "shell")

    mode    = "persistent" if q.get("mode") == "persistent" else "ephemeral"

    session = q.get("session", "")

    cwd     = q.get("cwd") or None

    if target not in TARGETS:

        await websocket.accept()

        await websocket.send_text("\r\n[darwin-term] unknown target\r\n")

        await websocket.close()

        return

    meta = TARGETS[target]

    if agent not in ("shell",) and agent not in meta["agents"]:

        agent = meta["default_agent"]

    if cwd and not _CWD_RE.match(cwd):

        await websocket.accept()

        await websocket.send_text("\r\n[darwin-term] invalid cwd\r\n")

        await websocket.close()

        return

    if mode == "persistent":

        if _valid_session(session):

            pass

        else:

            session = f"{agent}-{uuid.uuid4().hex[:8]}"

    else:

        session = ""

    remote_cmd = _remote_command(agent, mode, session, cwd)

    argv = _ssh_argv(meta["ip"]) + [remote_cmd]

    await websocket.accept()

    banner = (

        f"[darwin-term] {meta['label']} · {agent} · {mode}"

        + (f" · {session}" if session else "")

        + "\r\n"

    )

    await websocket.send_text(banner)

    if mode == "persistent":

        await websocket.send_text(json.dumps({"_cockpit": "session", "name": session}) + "\r\n")

    loop = asyncio.get_running_loop()

    master_fd, slave_fd = pty.openpty()

    flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)

    fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)

    try:

        fcntl.ioctl(master_fd, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0))

    except OSError:

        pass

    proc = await asyncio.create_subprocess_exec(

        *argv,

        stdin=slave_fd,

        stdout=slave_fd,

        stderr=slave_fd,

        preexec_fn=os.setsid,

        env={**os.environ, "TERM": "xterm-256color"},

    )

    os.close(slave_fd)

    def _set_winsize(rows: int, cols: int):

        try:

            fcntl.ioctl(master_fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))

        except OSError:

            pass

    stop = asyncio.Event()

    async def pty_to_ws():

        try:

            while not stop.is_set():

                try:

                    data = await loop.run_in_executor(None, _read_fd, master_fd)

                except OSError:

                    break

                if data is None:

                    continue

                if data == b"":

                    break

                try:

                    await websocket.send_bytes(data)

                except Exception:

                    break

        finally:

            stop.set()

    async def ws_to_pty():

        try:

            while not stop.is_set():

                msg = await websocket.receive()

                if msg.get("type") == "websocket.disconnect":

                    break

                if msg.get("bytes") is not None:

                    os.write(master_fd, msg["bytes"])

                elif msg.get("text") is not None:

                    text = msg["text"]

                    if text.startswith('{"t":"resize"'):

                        try:

                            c = json.loads(text)

                            _set_winsize(int(c.get("rows", 24)), int(c.get("cols", 80)))

                            continue

                        except Exception:

                            pass

                    os.write(master_fd, text.encode())

        except WebSocketDisconnect:

            pass

        except Exception:

            pass

        finally:

            stop.set()

    reader = asyncio.create_task(pty_to_ws())

    writer = asyncio.create_task(ws_to_pty())

    try:

        await stop.wait()

    finally:

        reader.cancel()

        writer.cancel()

        try:

            proc.terminate()

        except ProcessLookupError:

            pass

        try:

            os.close(master_fd)

        except OSError:

            pass

        try:

            await websocket.close()

        except Exception:

            pass

        logger.info("session closed: target=%s agent=%s mode=%s session=%s", target, agent, mode, session or "-")

if __name__ == "__main__":

    uvicorn.run(

        "main:app",

        host="0.0.0.0",

        port=9894,

        log_level="info",

    )