#!/usr/bin/env python3 # /// script # requires-python = ">=3.11" # dependencies = ["websockets>=13"] # /// """REM:Code agent — single-file Python implementation (tunnel-only, v2 history). uv run rem_agent.py login # one-time device login uv run rem_agent.py # run in the foreground; Ctrl-C to stop Env: REM_GATEWAY_URL, REM_GATEWAY_ORIGIN, REM_CLAUDE_BIN, REM_PROJECTS_DIR, REM_AGENT_NAME, REM_AGENT_ID. Changelog (bump VERSION below on every change): 0.3.5 compaction visibility: compaction_status SSE (phase start via 85%%-of- window prediction, phase done from system/compact_boundary), legacy compaction event from result.modelUsage drop (parity with Go), compaction markers + isCompactSummary tag in /v2/messages 0.3.4 unified namespace: every route lives under /v2 (canonical) with /api as a compatibility alias for deployed clients; create returns the full session object on both paths; GET sessions list is the merged discovery shape on both paths 0.3.3 session routes for the deployed web client: POST /api/sessions (create, returns the full session object — its absence broke create with a 404 on my.remcode.app), GET /api/sessions, DELETE/PATCH /api/sessions/{id}, GET .../log (git commit list), POST .../plan/dismiss (cross-client plan overlay dismissal) 0.3.2 validate client-supplied sessionId is a UUID (block path traversal out of the sessions dir via resume/blob/messages) 0.3.1 current model ids (sonnet-5, opus-4-8, haiku-4-5); xhigh effort for sonnet; family-based effort gate for unknown ids 0.3.0 /api/files + /api/files/content + /api/sessions/{id}/diff; utf-8 subprocess decode (Windows); warn on malformed config.json; ASCII dashes in console output 0.2.0 first-run setup (projects dir); auto-adopt CLI session on send; /api/projects + /api/claude-sessions; distinct default agent name " (py)"; peers warning on connect; REM_AGENT_ID 0.1.0 initial: tunnel client, v2 history from CLI JSONL, turn runner, permissions, SSE, push, device login """ from __future__ import annotations import argparse, asyncio, base64, contextlib, hashlib, json, logging, os, platform, random, re import shutil, subprocess, sys, time, urllib.error, urllib.parse, urllib.request, uuid, webbrowser from collections import deque from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path VERSION = "0.3.5" # keep in sync with the Changelog above APP_DIR_NAME = "rem-agent-py" DEFAULT_GATEWAY_URL = "wss://remcode.app/tunnel" DEFAULT_GATEWAY_ORIGIN = "https://my.remcode.app" RFC3339 = "%Y-%m-%dT%H:%M:%SZ" RECONNECT_MIN, RECONNECT_MAX = 1.0, 30.0 WS_CLOSE_REVOKED = 4401 # gateway: owner revoked this agent — do not reconnect SSE_KEEPALIVE, SSE_QUEUE_SIZE = 30, 256 TXN_TTL = 300 # seconds PENDING_STALE_AFTER_MS = 90_000 SUBPROCESS_LINE_LIMIT = 32 * 1024 * 1024 # Tools that require explicit approval; everything else is auto-approved # (safe default for future read-only tools). ALWAYS_ASK bypasses even # bypassPermissions — they are interactive by nature. PROMPT_TOOLS = {"Edit", "Write", "Bash", "NotebookEdit"} ALWAYS_ASK_TOOLS = {"AskUserQuestion", "ExitPlanMode", "ExitPlanModeV2"} # Verbatim copy of the Go agent's defaultModels — no discovery mechanism exists. # modelId is the alias-prefix form (no date suffix) so the substring match in # model_supports_effort also covers dated full IDs the CLI reports in sessions. DEFAULT_MODELS = [ {"id": "sonnet", "modelId": "claude-sonnet-5", "name": "Sonnet", "context": 1_000_000, "efforts": ["low", "medium", "high", "xhigh", "max", "ultracode"]}, {"id": "opus", "modelId": "claude-opus-4-8", "name": "Opus", "context": 1_000_000, "efforts": ["low", "medium", "high", "xhigh", "max", "ultracode"]}, {"id": "fable", "modelId": "claude-fable-5", "name": "Fable", "context": 1_000_000, "efforts": ["low", "medium", "high", "xhigh", "max", "ultracode"]}, {"id": "haiku", "modelId": "claude-haiku-4-5", "name": "Haiku", "context": 200_000, "efforts": []}, ] log = logging.getLogger(APP_DIR_NAME) def jdump(obj) -> str: # compact, UTF-8 raw — parity with Go's marshalNoEscape return json.dumps(obj, separators=(",", ":"), ensure_ascii=False) def opt(**kw) -> dict: # omitempty helper: keep only truthy values return {k: v for k, v in kw.items() if v} def now_ms() -> int: return int(time.time() * 1000) def now_rfc3339() -> str: return datetime.now(timezone.utc).strftime(RFC3339) def parse_rfc3339_ms(s: str) -> int: """RFC3339(Nano) → unix ms; 0 on failure. fromisoformat chokes on >6 fractional digits, so nanos are truncated first.""" if not s: return 0 v = s.strip() m = re.match(r"^(\d{4}-\d{2}-\d{2}[Tt ]\d{2}:\d{2}:\d{2})(\.\d+)?(.*)$", v) if m and m.group(2) and len(m.group(2)) > 7: v = m.group(1) + m.group(2)[:7] + (m.group(3) or "") try: dt = datetime.fromisoformat(v) except ValueError: return 0 if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return int(dt.timestamp() * 1000) def to_int(v) -> int: return int(v) if isinstance(v, (int, float)) and not isinstance(v, bool) else 0 # ─── Paths, config, identity ───────────────────────────────────────────────── # Own data dir, deliberately distinct from the Go agent's "rem-code": two # agents on one token/agent_id evict each other on the gateway. def data_dir() -> Path: if sys.platform == "darwin": base = Path.home() / "Library/Application Support" elif sys.platform == "win32": base = Path(os.environ.get("LOCALAPPDATA") or os.environ.get("APPDATA") or Path.home() / "AppData/Local") else: base = Path(os.environ.get("XDG_DATA_HOME") or Path.home() / ".local/share") d = base / APP_DIR_NAME d.mkdir(parents=True, exist_ok=True) return d def sessions_dir() -> Path: d = data_dir() / ".sessions" d.mkdir(parents=True, exist_ok=True) return d def user_agent() -> str: os_name = {"darwin": "macOS", "win32": "Windows"}.get(sys.platform, "Linux") return f"{APP_DIR_NAME}/{os_name}-{platform.machine()}/{VERSION}" @dataclass class Config: gateway_url: str gateway_origin: str claude_bin: str agent_name: str projects_dir: str def config_path() -> Path: return data_dir() / "config.json" def load_config() -> Config: """Precedence: env vars > config.json > defaults.""" file: dict = {} try: file = json.loads(config_path().read_text(encoding="utf-8")) except OSError: pass # no config yet — first_run_setup will create it except ValueError as e: # A malformed config (e.g. Windows path with unescaped backslashes) # must not silently fall back to defaults with no diagnostic. log.warning("config.json exists but failed to parse: %s - using defaults", e) env = os.environ.get return Config(env("REM_GATEWAY_URL") or file.get("gatewayUrl") or DEFAULT_GATEWAY_URL, env("REM_GATEWAY_ORIGIN") or file.get("gatewayOrigin") or DEFAULT_GATEWAY_ORIGIN, env("REM_CLAUDE_BIN") or file.get("claudeBin") or "claude", # " (py)" suffix is load-bearing: the gateway revokes tokens of # OTHER agent ids sharing one userId+agentName on every connect # (CleanupDuplicateAgents) — the bare hostname would kill the # Go agent's token. env("REM_AGENT_NAME") or file.get("agentName") or f"{platform.node() or APP_DIR_NAME} (py)", env("REM_PROJECTS_DIR") or file.get("projectsDir") or str(Path.home() / "projects")) def first_run_setup(cfg: Config) -> Config: """Interactive first run (like the Go agent): ask for the projects dir, auto-detect the Claude CLI, save config.json.""" print(f"\nWelcome to REM:Code! Settings will be saved to: {config_path()}\n") projects = "" while not projects: projects = os.path.expanduser( input(f"Projects directory [{cfg.projects_dir}]: ").strip() or cfg.projects_dir) if projects and not os.path.isdir(projects): if input(f" {projects} does not exist. Use anyway? [y/N]: ").strip().lower() not in ("y", "yes"): projects = "" claude_bin = cfg.claude_bin if found := shutil.which(claude_bin): claude_bin = found print(f"Claude CLI found at: {found}") else: claude_bin = input("Claude CLI not found in PATH. Claude CLI path: ").strip() or claude_bin config_path().write_text(json.dumps( {"projectsDir": projects, "claudeBin": claude_bin}, indent=2) + "\n", encoding="utf-8") print(f"Configuration saved to {config_path()}\n") return load_config() def credentials_path() -> Path: return data_dir() / "credentials.json" def load_credentials() -> dict: try: return json.loads(credentials_path().read_text(encoding="utf-8")) except (OSError, ValueError): return {} def save_credentials(creds: dict) -> None: credentials_path().write_text(json.dumps(creds, indent=2) + "\n", encoding="utf-8") with contextlib.suppress(OSError): os.chmod(credentials_path(), 0o600) def load_or_create_agent_id() -> str: # Precedence: REM_AGENT_ID env > config.json agentId > agent_id file > new uuid. # Set it to the Go agent's id (rem-code/agent_id) to adopt that identity, so a # phone already bound to that agent keeps reaching us — see CLAUDE.md "Identity". # Only safe when the two agents never run at once (same id → gateway eviction). if env_id := (os.environ.get("REM_AGENT_ID") or "").strip(): return env_id with contextlib.suppress(OSError, ValueError): cfg_id = json.loads(config_path().read_text(encoding="utf-8")).get("agentId") if isinstance(cfg_id, str) and cfg_id.strip(): return cfg_id.strip() p = data_dir() / "agent_id" with contextlib.suppress(OSError): if v := p.read_text(encoding="utf-8").strip(): return v v = str(uuid.uuid4()) try: p.write_text(v + "\n", encoding="utf-8") os.chmod(p, 0o600) except OSError as e: log.warning("could not save agent_id: %s", e) return v # ─── Login (device-code flow, blocking, stdlib only) ───────────────────────── def _http_post_json(url: str, body: dict | None, token: str = "", timeout: int = 15): """POST JSON → (status, parsed-or-None). Never raises.""" # Explicit UA: the gateway's bot filter 403s the default "Python-urllib/*". req = urllib.request.Request(url, data=jdump(body).encode() if body is not None else b"", method="POST", headers={"Content-Type": "application/json", "User-Agent": user_agent()}) if token: req.add_header("Authorization", "Bearer " + token) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, json.loads(resp.read().decode("utf-8", errors="replace")) except urllib.error.HTTPError as e: return e.code, None except (OSError, ValueError) as e: log.debug("POST %s failed: %s", url, e) return 0, None def cmd_login(cfg: Config) -> None: status, dc = _http_post_json(cfg.gateway_origin + "/auth/device/code", None) if status != 200 or not dc: sys.exit(f"device code request failed: HTTP {status}") full_url = f"{dc['verificationUrl']}?code={dc['userCode']}" opened = webbrowser.open(full_url) print(f"\n {'Browser opened. If it did not work, visit:' if opened else 'Open this link on any device:'}") print(f" {full_url}\n\n Code: {dc['userCode']}\n") expires_in = int(dc.get("expiresIn") or 900) print(f" Waiting for authentication... (expires in {expires_in // 60} minutes)") deadline = time.time() + expires_in while time.time() < deadline: time.sleep(max(int(dc.get("pollInterval") or 5), 1)) status, poll = _http_post_json(cfg.gateway_origin + "/auth/device/poll", {"deviceSecret": dc["deviceSecret"]}) if status == 200 and poll and poll.get("status") == "completed": save_credentials({"token": poll["token"], "slug": poll.get("slug", "")}) print(f"\n Logged in as {poll.get('slug', '')}\n Credentials: {credentials_path()}") return print(".", end="", flush=True) sys.exit("\nlogin timed out or cancelled") # ─── CLI JSONL parser (port of agent/cli_history.go) ───────────────────────── # Assistant entries sharing one Anthropic message.id merge into one Message; # tool_result user-entries fold into the originating tool_use block (no own # msgSeq); msgSeq is positional (index+1); images are content-addressed. USAGE_FIELDS = (("in", "input_tokens"), ("out", "output_tokens"), ("cacheRead", "cache_read_input_tokens"), ("cacheCreate", "cache_creation_input_tokens")) def _image_block(b: dict) -> dict | None: src = b.get("source") if not isinstance(src, dict) or src.get("type") != "base64" or not src.get("data"): return None # blobId = sha256 of the base64 STRING (not the decoded bytes) return {"type": "image", "blobId": hashlib.sha256(src["data"].encode()).hexdigest()} \ | opt(mediaType=src.get("media_type")) def _project_block(b: dict, role: str) -> dict | None: t = b.get("type") if t == "text": return {"type": "text"} | opt(text=b.get("text")) if role == "assistant": if t == "thinking": return {"type": "thinking"} | opt(text=b.get("thinking")) if t == "tool_use": out = {"type": "tool_use"} | opt(id=b.get("id"), name=b.get("name")) if b.get("input") is not None: out["input"] = b["input"] return out elif t == "image": return _image_block(b) return None def _tool_result_content(raw) -> list[dict]: if isinstance(raw, str): return [{"type": "text"} | opt(text=raw)] if not isinstance(raw, list): return [] out = [] for b in raw: if not isinstance(b, dict): continue if b.get("type") == "text": out.append({"type": "text"} | opt(text=b.get("text"))) elif b.get("type") == "image" and (cb := _image_block(b)): out.append(cb) return out def _iter_entries(path: str): """Yield (entry, message) for well-formed user/assistant JSONL lines, plus system/compact_boundary markers (with an empty message dict).""" with open(path, encoding="utf-8", errors="replace") as f: for line in f: if not (line := line.strip()): continue try: e = json.loads(line) except ValueError: continue # one bad line must not poison the parse if isinstance(e, dict) and e.get("type") in ("user", "assistant") \ and isinstance(e.get("message"), dict): yield e, e["message"] elif isinstance(e, dict) and e.get("type") == "system" \ and e.get("subtype") == "compact_boundary": yield e, {} def parse_cli_session(path: str) -> list[dict]: """Claude CLI JSONL log → list of v2 Message dicts (wire shape).""" return parse_cli_session_full(path)[0] def parse_cli_session_full(path: str) -> tuple[list[dict], list[dict]]: """Like parse_cli_session, but also returns compaction markers parsed from system/compact_boundary lines. Markers are NOT Messages and carry no msgSeq of their own (numbering must stay stable) — they anchor after the message with afterMsgSeq (0 = before the first message).""" messages: list[dict] = [] markers: list[dict] = [] ids: dict[str, int] = {} # assistant message.id → index tools: dict[str, tuple[int, int]] = {} # tool_use.id → (msgIdx, blockIdx) for e, m in _iter_entries(path): if e.get("isSidechain"): continue # sub-agent traffic ts = parse_rfc3339_ms(e.get("timestamp") or "") content = m.get("content") if e["type"] == "system": # compact_boundary (see _iter_entries) cm = e.get("compactMetadata") if isinstance(e.get("compactMetadata"), dict) else {} markers.append({"afterMsgSeq": len(messages), "ts": ts} | opt(trigger=cm.get("trigger"), preTokens=to_int(cm.get("preTokens")), postTokens=to_int(cm.get("postTokens")), durationMs=to_int(cm.get("durationMs")))) continue if e["type"] == "assistant": blocks = content if isinstance(content, list) else [] projected = [cb for b in blocks if isinstance(b, dict) and (cb := _project_block(b, "assistant"))] mid = m.get("id") or "" if (idx := ids.get(mid)) is None: messages.append({"role": "assistant", "status": "complete", "ts": ts, "content": []} | opt(tsDone=ts, cliUuid=mid, model=m.get("model"))) ids[mid] = idx = len(messages) - 1 msg = messages[idx] start = len(msg["content"]) msg["content"] += projected tools.update({cb["id"]: (idx, start + o) for o, cb in enumerate(projected) if cb["type"] == "tool_use" and cb.get("id")}) # usage / stop_reason / tsDone — last-write-wins across the group if isinstance(u := m.get("usage"), dict): msg["usage"] = {k: n for k, f in USAGE_FIELDS if (n := to_int(u.get(f)))} if m.get("stop_reason"): msg["stopReason"] = m["stop_reason"] if ts > msg.get("tsDone", 0): msg["tsDone"] = ts else: # user if e.get("isMeta"): continue # internal SDK prompts if isinstance(content, str): if content: messages.append({"role": "user", "status": "complete", "ts": ts, "content": [{"type": "text", "text": content}]} | opt(cliUuid=e.get("uuid"), isCompactSummary=bool(e.get("isCompactSummary")))) continue blocks = [b for b in content if isinstance(b, dict)] if isinstance(content, list) else [] for rb in (b for b in blocks if b.get("type") == "tool_result"): if (ref := tools.get(rb.get("tool_use_id") or "")) is None: continue # orphan tool_result — drop silently messages[ref[0]]["content"][ref[1]]["result"] = \ opt(content=_tool_result_content(rb.get("content")), isError=bool(rb.get("is_error"))) projected = [cb for b in blocks if b.get("type") != "tool_result" and (cb := _project_block(b, "user"))] if projected: messages.append({"role": "user", "status": "complete", "ts": ts, "content": projected} | opt(cliUuid=e.get("uuid"), isCompactSummary=bool(e.get("isCompactSummary")))) for i, msg in enumerate(messages): msg["msgSeq"] = i + 1 return messages, markers def extract_cli_blob(path: str, blob_id: str) -> tuple[str, bytes] | None: """Re-scan the JSONL for the image whose sha256(base64) == blob_id.""" for _, m in _iter_entries(path): content = m.get("content") if not isinstance(content, list): continue for b in content: if not isinstance(b, dict): continue candidates = [b] if b.get("type") == "tool_result" and isinstance(b.get("content"), list): candidates += [rb for rb in b["content"] if isinstance(rb, dict)] for c in candidates: src = c.get("source") if c.get("type") != "image" or not isinstance(src, dict): continue if src.get("type") != "base64" or not src.get("data"): continue if hashlib.sha256(src["data"].encode()).hexdigest() != blob_id: continue with contextlib.suppress(ValueError): return src.get("media_type") or "", base64.b64decode(src["data"]) return None # ─── Claude CLI session discovery (~/.claude/projects) ─────────────────────── def claude_projects_root() -> Path: return Path.home() / ".claude" / "projects" def _read_cwd_from_jsonl(path: Path) -> str: """First non-empty `cwd` in the first ~20 lines.""" with contextlib.suppress(OSError): with open(path, encoding="utf-8", errors="replace") as f: for _, line in zip(range(20), f): with contextlib.suppress(ValueError, AttributeError): if isinstance(cwd := json.loads(line).get("cwd"), str) and cwd: return cwd return "" _SESSION_ID_RE = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") def is_valid_session_id(sid: str) -> bool: """Canonical UUID only. A client-supplied sessionId is used verbatim as a filesystem path component (session/pending files, CLI JSONL lookup), so anything else is rejected to block path traversal out of the sessions dir.""" return bool(sid) and bool(_SESSION_ID_RE.match(sid)) def find_claude_session_file(session_id: str) -> tuple[str, str]: """Locate .jsonl across every project dir → (path, cwd).""" if not is_valid_session_id(session_id): return "", "" with contextlib.suppress(OSError): for d in claude_projects_root().iterdir(): candidate = d / f"{session_id}.jsonl" if d.is_dir() and candidate.is_file() and (cwd := _read_cwd_from_jsonl(candidate)): return str(candidate), cwd return "", "" def read_session_title(path: str) -> str: """ai-title entry, else first non-meta user text. First 64 KiB only.""" try: with open(path, encoding="utf-8", errors="replace") as f: data = f.read(64 * 1024) except OSError: return "" first_user = "" for line in data.splitlines(): try: parsed = json.loads(line) except ValueError: continue if not isinstance(parsed, dict): continue if parsed.get("type") == "ai-title": if isinstance(t := parsed.get("aiTitle"), str) and t: return t elif parsed.get("type") == "user" and not first_user and not parsed.get("isMeta"): msg = parsed.get("message") content = msg.get("content") if isinstance(msg, dict) else None if isinstance(content, str) and content: first_user = content elif isinstance(content, list): first_user = next((b["text"] for b in content if isinstance(b, dict) and b.get("type") == "text" and b.get("text")), "") return first_user[:80] + ("…" if len(first_user) > 80 else "") if first_user else "" def scan_cli_sessions(limit: int = 200) -> list[dict]: """Cheap discovery preview: top `limit` sessions by mtime, no JSONL parsing.""" raw = [] with contextlib.suppress(OSError): for d in claude_projects_root().iterdir(): if not d.is_dir(): continue with contextlib.suppress(OSError): files = [p for p in d.iterdir() if p.name.endswith(".jsonl")] project = next((c for p in sorted(files) if (c := _read_cwd_from_jsonl(p))), "") \ or d.name.replace("-", os.sep) for p in files: sid = p.name[:-len(".jsonl")] if len(sid) < 32: continue with contextlib.suppress(OSError): raw.append({"id": sid, "project": project, "path": str(p), "mtime": int(p.stat().st_mtime * 1000)}) raw.sort(key=lambda e: e["mtime"], reverse=True) raw = raw[:limit] if limit > 0 else raw for e in raw: e["title"] = read_session_title(e["path"]) return raw # ─── Overlay stores: pending / interrupted / thinking / txn ────────────────── # pending: user sends the CLI hasn't flushed to its JSONL yet (survives restart). # interrupted: partial assistant text of a turn killed mid-stream. # thinking: plaintext the CLI strips from its JSONL — live capture, memory only. def pending_path(sid: str) -> Path: return sessions_dir() / f"{sid}.pending.jsonl" def interrupted_path(sid: str) -> Path: return sessions_dir() / f"{sid}.interrupted.json" def append_pending(sid: str, entry: dict) -> None: try: with open(pending_path(sid), "a", encoding="utf-8") as f: f.write(jdump(entry) + "\n") except OSError as e: log.warning("[%s] pending write failed: %s", sid, e) def read_pending(sid: str) -> list[dict]: try: lines = pending_path(sid).read_text(encoding="utf-8").splitlines() except OSError: return [] out = [] for line in lines: with contextlib.suppress(ValueError): if line and isinstance(e := json.loads(line), dict): out.append(e) return out def write_pending(sid: str, entries: list[dict]) -> None: p = pending_path(sid) try: if not entries: p.unlink(missing_ok=True) return tmp = p.with_suffix(p.suffix + ".tmp") tmp.write_text("".join(jdump(e) + "\n" for e in entries), encoding="utf-8") tmp.replace(p) except OSError as e: log.warning("[%s] pending rewrite failed: %s", sid, e) def pending_fallback(parsed: list[dict], pending: list[dict]) -> tuple[list[dict], list[dict]]: """Prune flushed pendings by exact content match (NEVER by timestamp — a send queued behind a busy turn has an older ts than already-flushed ones); append the survivors as user messages at the tail.""" counts: dict[str, int] = {} for m in parsed: if m.get("role") == "user": for b in m.get("content", []): if b.get("type") == "text" and b.get("text"): counts[b["text"]] = counts.get(b["text"], 0) + 1 out, survivors = list(parsed), [] for e in pending: text = e.get("text") or "" if counts.get(text, 0) > 0: counts[text] -= 1 # flushed to the CLI JSONL — prune, don't duplicate continue survivors.append(e) out.append({"msgSeq": len(out) + 1, "role": "user", "status": "complete", "ts": to_int(e.get("ts")), "content": [{"type": "text", "text": text}]} | opt(clientId=e.get("clientId"), txnId=e.get("txnId"))) return out, survivors def write_interrupted(sid: str, text: str) -> None: if not text: return try: interrupted_path(sid).write_text(jdump({"text": text, "ts": now_ms()}), encoding="utf-8") except OSError as e: log.warning("[%s] interrupted write failed: %s", sid, e) def clear_interrupted(sid: str) -> None: with contextlib.suppress(OSError): interrupted_path(sid).unlink(missing_ok=True) def read_interrupted(sid: str) -> dict | None: try: e = json.loads(interrupted_path(sid).read_text(encoding="utf-8")) except (OSError, ValueError): return None return e if isinstance(e, dict) and e.get("text") else None def interrupted_merge(parsed: list[dict], e: dict | None) -> tuple[list[dict], bool]: """Append a killed turn's partial as a status:"interrupted" tail message; superseded=True when the CLI parse already contains that exact text.""" if not e or not e.get("text"): return parsed, False if any(b.get("type") == "text" and b.get("text") == e["text"] for m in parsed if m.get("role") == "assistant" for b in m.get("content", [])): return parsed, True return parsed + [{"msgSeq": len(parsed) + 1, "role": "assistant", "status": "interrupted", "ts": to_int(e.get("ts")), "content": [{"type": "text", "text": e["text"]}]}], False class ThinkStore: """sessionID → Anthropic message.id → thinking text (in-memory only: the CLI strips plaintext thinking from its JSONL, live stream is the only source).""" def __init__(self): self._m: dict[str, dict[str, str]] = {} def append_delta(self, sid: str, msg_id: str, text: str) -> None: if msg_id and text: per = self._m.setdefault(sid, {}) per[msg_id] = per.get(msg_id, "") + text def fill(self, sid: str, msgs: list[dict]) -> list[dict]: per = self._m.get(sid) or {} for m in msgs: if th := per.get(m.get("cliUuid") or ""): # full capture goes into the FIRST thinking block only if b := next((b for b in m.get("content", []) if b.get("type") == "thinking"), None): b["text"] = th return msgs class TxnCache: """Dedupe /send retries: (sessionId, txnId) → (msgSeq, clientId), 5-min TTL.""" def __init__(self): self._m: dict[str, tuple[int, str, float]] = {} def lookup(self, sid: str, txn: str): if not txn: return None now = time.time() self._m = {k: v for k, v in self._m.items() if now <= v[2]} # opportunistic GC v = self._m.get(f"{sid}:{txn}") return (v[0], v[1]) if v else None def record(self, sid: str, txn: str, msg_seq: int, client_id: str) -> None: if txn: self._m[f"{sid}:{txn}"] = (msg_seq, client_id, time.time() + TXN_TTL) # ─── MessageStore ──────────────────────────────────────────────────────────── class _MsgCache: def __init__(self): self.loaded, self.msgs, self.err = False, [], None self.markers = [] # compaction boundaries, anchored by afterMsgSeq self.next_msg_seq = 0 # monotonic; survives invalidate self.lock = asyncio.Lock() class MessageStore: def __init__(self, agent: Agent): self.agent = agent self._caches: dict[str, _MsgCache] = {} def _cache(self, sid: str) -> _MsgCache: return self._caches.setdefault(sid, _MsgCache()) async def get(self, sid: str) -> tuple[list[dict], Exception | None]: c = self._cache(sid) async with c.lock: if c.loaded: return c.msgs, c.err path, _ = await asyncio.to_thread(find_claude_session_file, sid) msgs, markers, err = [], [], None if path: try: msgs, markers = await asyncio.to_thread(parse_cli_session_full, path) except Exception as e: # surface as 500, don't crash err = e if err is None: msgs = self._merge_pending(sid, msgs) # merge order matters: msgs, superseded = interrupted_merge(msgs, read_interrupted(sid)) if superseded: clear_interrupted(sid) msgs = self.agent.think.fill(sid, msgs) c.msgs, c.markers, c.err, c.loaded = msgs, markers, err, True return c.msgs, c.err def _merge_pending(self, sid: str, parsed: list[dict]) -> list[dict]: pending = read_pending(sid) if not pending: return parsed on_disk = len(pending) s = self.agent.sessions.get(sid) # GC abandoned sends only when idle: a queued/processing send is never dropped if s is None or (not s.processing and not s.queue): pending = [e for e in pending if now_ms() - to_int(e.get("ts")) <= PENDING_STALE_AFTER_MS] out, survivors = pending_fallback(parsed, pending) if len(survivors) != on_disk: write_pending(sid, survivors) return out async def allocate_msg_seq(self, sid: str) -> int: msgs, _ = await self.get(sid) c = self._cache(sid) c.next_msg_seq = max(c.next_msg_seq, len(msgs) + 1) c.next_msg_seq += 1 return c.next_msg_seq - 1 def invalidate(self, sid: str) -> None: if c := self._caches.get(sid): c.loaded, c.msgs, c.markers, c.err = False, [], [], None async def markers(self, sid: str) -> list[dict]: await self.get(sid) # ensure cache loaded c = self._caches.get(sid) return c.markers if c else [] def is_active(self, sid: str) -> bool: c = self._caches.get(sid) return bool(c and c.loaded and c.err is None) def markers_in_page(all_markers: list[dict], page: list[dict], max_seq: int) -> list[dict]: """Markers that belong to the returned page: a marker renders before the message that FOLLOWS it (afterMsgSeq+1); a trailing marker rides the tail page (mirror of Go markersInPage).""" if not all_markers or not page: return [] first, last = page[0]["msgSeq"], page[-1]["msgSeq"] return [m for m in all_markers if first <= min(m["afterMsgSeq"] + 1, max_seq) <= last] def page_messages(msgs: list[dict], from_: int, before: int, limit: int) -> tuple[list[dict], bool]: limit = min(limit, 500) if limit > 0 else 50 if before > 0: # backward: msgSeq < before, most recent `limit`, ascending end = next((i for i, m in enumerate(msgs) if m["msgSeq"] >= before), len(msgs)) return msgs[max(end - limit, 0):end], end - limit > 0 if from_ > 0: # forward incremental: msgSeq > from start = next((i for i, m in enumerate(msgs) if m["msgSeq"] > from_), len(msgs)) return msgs[start:start + limit], start + limit < len(msgs) return msgs[-limit:], len(msgs) > limit # tail # ─── Session & SessionManager ──────────────────────────────────────────────── @dataclass class InFlightTurn: user_msg_seq: int started_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) # Both hold ONLY the CURRENT assistant message (reset on message_start) — # earlier messages of the turn are already flushed to the CLI JSONL. text: str = "" # committed text (from the `assistant` snapshot) live_text: str = "" # streamed text_delta, pre-snapshot @dataclass class QueuedTurn: text: str images: list user_msg_seq: int @dataclass class Session: id: str project: str = "" permission_mode: str = "default" name: str = "" model: str = "" effort: str = "" message_count: int = 0 created_at: str = "" updated_at: str = "" # runtime queue: deque = field(default_factory=deque) processing: bool = False pending_permission: dict | None = None in_flight: InFlightTurn | None = None allowed_tools: set = field(default_factory=set) proc: asyncio.subprocess.Process | None = None stdin_lock: asyncio.Lock = field(default_factory=asyncio.Lock) seq: int = 0 # live SSE watermark; per-run, intentionally not persisted last_result: dict | None = None ctx_tracking: dict = field(default_factory=dict) # model -> last context size compaction_seen: bool = False # compact_boundary observed this turn worker_task: asyncio.Task | None = None def __post_init__(self): self.permission_mode = self.permission_mode or "default" self.created_at = self.created_at or now_rfc3339() self.updated_at = self.updated_at or self.created_at def next_seq(self) -> int: self.seq += 1 return self.seq def project_path(self, cfg: Config) -> str: return self.project if os.path.isabs(self.project) \ else os.path.join(cfg.projects_dir, self.project) def compute_status(self) -> str: return "awaiting" if self.pending_permission is not None else \ "working" if self.processing else \ "done" if self.message_count > 0 else "idle" def info(self, agent: Agent) -> dict: d = {"id": self.id, "agentId": agent.agent_id, "name": self.name, "project": self.project, "permissionMode": self.permission_mode, "createdAt": self.created_at, "updatedAt": self.updated_at, "messageCount": self.message_count, "processing": self.processing, "pendingPermission": self.pending_permission is not None, "status": self.compute_status(), # always non-nil in the Go agent; zeroed here (no cost tracking) "cachedStats": {"models": {}, "totalCost": 0, "totalOutput": 0, "turns": 0}} \ | opt(model=self.model, effort=self.effort) if self.processing: t = self.in_flight d["turn"] = {"startedAt": t.started_at.strftime(RFC3339), "userMsgSeq": t.user_msg_seq} if t else {} if (p := self.pending_permission) is not None: d["awaiting"] = {"kind": awaiting_kind(p.get("toolName") or ""), "toolName": p.get("toolName") or ""} \ | opt(requestId=p.get("requestId")) | {"request": p} return d def meta_dict(self) -> dict: return {"id": self.id, "name": self.name, "project": self.project, "permissionMode": self.permission_mode, "createdAt": self.created_at, "updatedAt": self.updated_at, "messageCount": self.message_count} \ | opt(model=self.model, effort=self.effort) def awaiting_kind(tool: str) -> str: return {"AskUserQuestion": "question", "ExitPlanMode": "plan", "ExitPlanModeV2": "plan"} \ .get(tool, "permission") class SessionManager: def __init__(self): self.sessions: dict[str, Session] = {} self.shutting_down = False def load_all(self) -> None: for p in sessions_dir().glob("*.meta.json"): with contextlib.suppress(OSError, ValueError): m = json.loads(p.read_text(encoding="utf-8")) if isinstance(m, dict) and m.get("id"): self.sessions[m["id"]] = Session( m["id"], m.get("project", ""), m.get("permissionMode", "default"), m.get("name", ""), m.get("model", ""), m.get("effort", ""), to_int(m.get("messageCount")), m.get("createdAt", ""), m.get("updatedAt", "")) def get(self, sid: str) -> Session | None: return self.sessions.get(sid) def create(self, project: str, permission_mode: str, name: str, model: str, effort: str, sid: str = "", message_count: int = 0) -> Session: s = Session(sid or str(uuid.uuid4()), project, permission_mode, name, model, effort, message_count) self.sessions[s.id] = s self.save(s) return s def save(self, s: Session) -> None: try: (sessions_dir() / f"{s.id}.meta.json").write_text( json.dumps(s.meta_dict(), indent=2) + "\n", encoding="utf-8") except OSError as e: log.warning("[%s] meta save failed: %s", s.id, e) def delete(self, sid: str) -> None: self.sessions.pop(sid, None) with contextlib.suppress(OSError): (sessions_dir() / f"{sid}.meta.json").unlink(missing_ok=True) with contextlib.suppress(OSError): pending_path(sid).unlink(missing_ok=True) clear_interrupted(sid) def all(self) -> list[Session]: return list(self.sessions.values()) def list_infos(self, agent: Agent) -> list[dict]: return sorted((s.info(agent) for s in self.sessions.values()), key=lambda d: d["updatedAt"], reverse=True) def any_processing(self) -> bool: return any(s.processing for s in self.sessions.values()) async def wait_for_idle(self, timeout: float) -> None: deadline = time.monotonic() + timeout while self.any_processing() and time.monotonic() < deadline: await asyncio.sleep(0.3) def kill_all(self) -> None: for s in self.sessions.values(): if s.proc and s.proc.returncode is None: with contextlib.suppress(ProcessLookupError): s.proc.kill() # ─── SSE hub ───────────────────────────────────────────────────────────────── def sse_frame(event: str, data) -> str: return f"event: {event}\ndata: {data if isinstance(data, str) else jdump(data)}\n\n" class SSEClient: def __init__(self): self.id = str(uuid.uuid4()) self.q: asyncio.Queue[str] = asyncio.Queue(maxsize=SSE_QUEUE_SIZE) self.done = asyncio.Event() def enqueue(self, frame: str) -> None: # Never blocks: a slow/half-open consumer is dropped, not waited on. if not self.done.is_set(): try: self.q.put_nowait(frame) except asyncio.QueueFull: self.done.set() class Hub: def __init__(self, agent: Agent): self.agent = agent self.clients: dict[str, SSEClient] = {} def register(self, c: SSEClient) -> None: self.clients[c.id] = c def unregister(self, c: SSEClient) -> None: self.clients.pop(c.id, None) c.done.set() def broadcast(self, event: str, data) -> None: frame = sse_frame(event, data) for c in list(self.clients.values()): c.enqueue(frame) def broadcast_session_list(self) -> None: self.broadcast("sessions_update", {"type": "sessions_update", "sessions": self.agent.sessions.list_infos(self.agent)}) class Agent: def __init__(self, cfg: Config, creds: dict, agent_id: str): self.cfg, self.creds, self.agent_id = cfg, creds, agent_id self.start_time = time.time() self.sessions = SessionManager() self.hub = Hub(self) self.msgs = MessageStore(self) self.think = ThinkStore() self.txns = TxnCache() self.shutdown_started = False # ─── Claude subprocess turn runner (port of agent/claude.go) ───────────────── def resolve_claude_cmd(claude_bin: str) -> list[str]: # npm wrappers on Windows are .cmd/.bat → must be spawned via cmd.exe. if sys.platform == "win32": resolved = shutil.which(claude_bin) or claude_bin if resolved.lower().endswith((".cmd", ".bat")): return ["cmd.exe", "/c", resolved] return [resolved] return [claude_bin] def model_supports_effort(model: str) -> bool: # Unknown/renamed IDs fall back to a family gate rather than blind True — # a future haiku ID must never get --effort or the CLI rejects the turn. return next((bool(m["efforts"]) for m in DEFAULT_MODELS if model == m["id"] or m["modelId"] in model), "haiku" not in model) def ctx_window_tokens(model: str) -> int: # Context window for the compaction-start prediction; unknown ids get the # standard 200k (mirror of Go ctxWindowTokens). return 1_000_000 if "[1m]" in model else 200_000 def build_turn_args(s: Session, is_first: bool) -> list[str]: # The permission channel stays open even in bypassPermissions so always-ask # tools reach the user; bypass is enforced agent-side (handle_control_request). args = ["--print", "--verbose", "--output-format", "stream-json", "--include-partial-messages", "--input-format", "stream-json", "--permission-prompt-tool", "stdio", "--session-id" if is_first else "--resume", s.id] if s.permission_mode not in ("default", "bypassPermissions"): args += ["--permission-mode", s.permission_mode] if s.model and not s.model.startswith("<"): args += ["--model", s.model] if s.effort and model_supports_effort(s.model): args += ["--effort", "max" if s.effort == "ultracode" else s.effort] return args async def write_stdin_line(s: Session, obj: dict) -> None: if s.proc is None or s.proc.stdin is None: return async with s.stdin_lock: try: s.proc.stdin.write((jdump(obj) + "\n").encode()) await s.proc.stdin.drain() except (ConnectionError, RuntimeError, OSError) as e: log.warning("[%s] stdin write failed: %s", s.id, e) def ensure_worker(agent: Agent, s: Session) -> None: if s.worker_task is None or s.worker_task.done(): s.worker_task = asyncio.create_task(session_worker(agent, s)) async def session_worker(agent: Agent, s: Session) -> None: # Equivalent of Go's processQueue re-invocation chain: one task drains the # queue until empty; a later send re-spawns it via ensure_worker. while s.queue and not agent.sessions.shutting_down and not s.processing: await run_turn(agent, s, s.queue.popleft()) async def run_turn(agent: Agent, s: Session, turn: QueuedTurn) -> None: s.processing = True s.message_count += 1 is_first = s.message_count == 1 s.in_flight = InFlightTurn(turn.user_msg_seq) s.compaction_seen = False # Predict an imminent auto-compaction: the CLI compacts at turn START when # the context is near the model's window and streams nothing until the # compact_boundary at completion (1-3.5 min of silence) — warn clients now. ctx, model = max(((c, m) for m, c in s.ctx_tracking.items()), default=(0, "")) if ctx >= 0.85 * ctx_window_tokens(model): agent.hub.broadcast("compaction_status", { "type": "compaction_status", "sessionId": s.id, "seq": s.next_seq(), "ts": now_ms(), "phase": "start", "model": model, "preTokens": ctx}) clear_interrupted(s.id) # a new turn supersedes a previously-captured partial agent.hub.broadcast("processing", {"type": "processing", "sessionId": s.id, "msgSeq": turn.user_msg_seq}) agent.hub.broadcast_session_list() exit_code = 1 try: argv = resolve_claude_cmd(agent.cfg.claude_bin) + build_turn_args(s, is_first) log.info("[%s] spawning claude (turn %d, perm=%s)", s.id, s.message_count, s.permission_mode) try: s.proc = await asyncio.create_subprocess_exec( *argv, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=s.project_path(agent.cfg), env={**os.environ, "DISABLE_AUTOUPDATER": "1"}, limit=SUBPROCESS_LINE_LIMIT) except OSError as e: err = (f'Claude CLI not found at "{agent.cfg.claude_bin}". Install it or set ' "REM_CLAUDE_BIN.") if isinstance(e, FileNotFoundError) else f"Failed to start Claude CLI: {e}" log.error("[%s] %s", s.id, err) agent.hub.broadcast("stream", { "type": "stream", "sessionId": s.id, "ts": now_ms(), "seq": s.next_seq(), "data": {"type": "result", "subtype": "error_response", "error": err}}) return if turn.images: log.warning("[%s] images in send are not supported - sending text only", s.id) await write_stdin_line(s, {"type": "user", "message": {"role": "user", "content": turn.text}}) stderr_task = asyncio.create_task(_drain_stderr(s.proc)) await stream_stdout(agent, s, s.proc) await s.proc.wait() exit_code = s.proc.returncode if s.proc.returncode is not None else 1 if stderr := (await stderr_task).strip(): log.info("[%s] stderr: %s", s.id, stderr[:2000]) log.info("[%s] claude exited with code %d", s.id, exit_code) finally: # Mirror of the Go defer in processQueue — order matters. seq = s.next_seq() r = s.last_result or {} usage = r.get("usage") if isinstance(r.get("usage"), dict) else {} turn_done = {"type": "turn_done", "sessionId": s.id, "seq": seq} | opt( exitCode=exit_code, duration_ms=to_int(r.get("duration_ms")), num_turns=to_int(r.get("num_turns")), usage=opt(input_tokens=to_int(usage.get("input_tokens")), output_tokens=to_int(usage.get("output_tokens"))) if r else None) s.last_result = None agent.hub.broadcast("turn_done", turn_done) agent.msgs.invalidate(s.id) asyncio.get_running_loop().create_task(notify_gateway(agent, s, "turn_done", seq)) if s.in_flight and exit_code != 0 and s.in_flight.live_text: # The CLI never flushed the currently-streaming message — capture it. write_interrupted(s.id, s.in_flight.live_text) s.in_flight = s.proc = s.pending_permission = None s.processing = False s.updated_at = now_rfc3339() agent.sessions.save(s) agent.hub.broadcast_session_list() async def _drain_stderr(proc: asyncio.subprocess.Process) -> str: with contextlib.suppress(ConnectionError, OSError): if proc.stderr is not None: return (await proc.stderr.read()).decode("utf-8", errors="replace") return "" async def stream_stdout(agent: Agent, s: Session, proc: asyncio.subprocess.Process) -> None: cur_msg_id = "" while True: try: raw = await proc.stdout.readline() except ValueError: log.warning("[%s] stdout line exceeded %d bytes - skipped", s.id, SUBPROCESS_LINE_LIMIT) continue except (ConnectionError, OSError): break if not raw: break if not (line := raw.decode("utf-8", errors="replace").strip()): continue try: parsed = json.loads(line) except ValueError: log.info("[%s] stdout non-json: %s", s.id, line[:200]) continue if not isinstance(parsed, dict): continue mtype = parsed.get("type") if mtype == "control_request": await handle_control_request(agent, s, parsed) continue if mtype in ("control_cancel_request", "keep_alive"): continue assistant_text = "" if mtype == "assistant" and isinstance(msg := parsed.get("message"), dict): if isinstance(model := msg.get("model"), str) and model and not model.startswith("<"): s.model = model cur_msg_id = msg.get("id") or cur_msg_id if isinstance(content := msg.get("content"), list): assistant_text = "".join(b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text") elif mtype == "stream_event" and isinstance(ev := parsed.get("event"), dict): if ev.get("type") == "message_start": if isinstance(m := ev.get("message"), dict): cur_msg_id = m.get("id") or cur_msg_id if s.in_flight: s.in_flight.live_text = "" elif ev.get("type") == "content_block_delta" and isinstance(d := ev.get("delta"), dict): if d.get("type") == "thinking_delta" and d.get("thinking"): agent.think.append_delta(s.id, cur_msg_id, d["thinking"]) elif d.get("type") == "text_delta" and d.get("text") and s.in_flight: s.in_flight.live_text += d["text"] elif mtype == "system" and parsed.get("subtype") == "compact_boundary": # Emitted only at compaction COMPLETION; label the silent window # that preceded it. Falls through to the generic stream broadcast. pre, post, dur, trig = _parse_compact_boundary(parsed) s.compaction_seen = True if not dur and s.in_flight: dur = now_ms() - int(s.in_flight.started_at.timestamp() * 1000) agent.hub.broadcast("compaction_status", { "type": "compaction_status", "sessionId": s.id, "seq": s.next_seq(), "ts": now_ms(), "phase": "done"} | opt(preTokens=pre, postTokens=post, durationMs=dur, trigger=trig)) elif mtype == "result": s.last_result = parsed _track_context(agent, s, parsed) if proc.stdin is not None: # close stdin after result so the process exits with contextlib.suppress(ConnectionError, RuntimeError, OSError): proc.stdin.close() if s.in_flight and assistant_text: # The `assistant` snapshot commits this message's text; the delta # buffer is subsumed — reset it or the next message double-counts. s.in_flight.text += assistant_text s.in_flight.live_text = "" agent.hub.broadcast("stream", {"type": "stream", "sessionId": s.id, "data": parsed, "ts": now_ms(), "seq": s.next_seq()}) def _parse_compact_boundary(parsed: dict) -> tuple[int, int, int, str]: """compact_boundary metadata from either wire shape: live/SDK snake_case (compact_metadata.pre_tokens) or JSONL camelCase (compactMetadata.*).""" snake = parsed.get("compact_metadata") camel = parsed.get("compactMetadata") snake = snake if isinstance(snake, dict) else {} camel = camel if isinstance(camel, dict) else {} pre = to_int(snake.get("pre_tokens")) or to_int(camel.get("preTokens")) post = to_int(snake.get("post_tokens")) or to_int(camel.get("postTokens")) dur = to_int(snake.get("duration_ms")) or to_int(camel.get("durationMs")) trig = snake.get("trigger") or camel.get("trigger") or "" return pre, post, dur, trig if isinstance(trig, str) else "" def _track_context(agent: Agent, s: Session, parsed: dict) -> None: # Mirror of Go handleResultEvent's compaction bookkeeping: context per model # from result.modelUsage; a >30% drop = compaction. The legacy `compaction` # event is suppressed when the boundary already told clients this turn. mu = parsed.get("modelUsage") if not isinstance(mu, dict): return for model, u in mu.items(): if not isinstance(u, dict): continue ctx = to_int(u.get("inputTokens")) + to_int(u.get("cacheReadInputTokens")) \ + to_int(u.get("cacheCreationInputTokens")) prev = s.ctx_tracking.get(model, 0) if not s.compaction_seen and prev > 0 and ctx < prev * 0.7: log.info("[%s] compaction detected: %d -> %d", s.id, prev, ctx) agent.hub.broadcast("compaction", {"type": "compaction", "sessionId": s.id, "model": model, "prevContext": prev, "newContext": ctx, "seq": s.next_seq()}) s.ctx_tracking[model] = ctx async def handle_control_request(agent: Agent, s: Session, parsed: dict) -> None: request = parsed.get("request") if not isinstance(request, dict) or request.get("subtype") != "can_use_tool": return rid, tool = parsed.get("request_id") or "", request.get("tool_name") or "" tool_input, tuid = request.get("input"), request.get("tool_use_id") or "" log.info("[%s] control_request: can_use_tool %s", s.id, tool) if tool not in ALWAYS_ASK_TOOLS and (s.permission_mode == "bypassPermissions" or tool not in PROMPT_TOOLS or tool in s.allowed_tools): await write_stdin_line(s, {"type": "control_response", "response": { "subtype": "success", "request_id": rid, "response": {"behavior": "allow", "updatedInput": tool_input, "toolUseID": tuid}}}) return s.pending_permission = perm = { "type": "permission_request", "sessionId": s.id, "requestId": rid, "toolName": tool, "toolInput": tool_input, "toolUseId": tuid, "title": request.get("title") or "", "description": request.get("description") or ""} agent.hub.broadcast("permission_request", perm) asyncio.get_running_loop().create_task( notify_gateway(agent, s, "permission_request", 0, {"toolName": tool, "requestId": rid})) agent.hub.broadcast_session_list() async def notify_gateway(agent: Agent, s: Session, event_type: str, seq: int, extra: dict | None = None) -> None: """Optional push relay: POST /api/push/send with the gateway token.""" if not (token := agent.creds.get("token") or ""): return origin = agent.cfg.gateway_url.removesuffix("/tunnel") \ .replace("wss://", "https://", 1).replace("ws://", "http://", 1) title = " ".join(filter(None, ["/" + s.project if s.project else ""])) title = f"{title} · {s.name}" if title and s.name else title or s.name or "REM:Code" tool = (extra or {}).get("toolName") or "" body = {"type": event_type, "sessionId": s.id, "title": title, "notifId": f"{s.id}:{event_type}:{seq}", "body": {"turn_done": "Task complete", "permission_request": f"Approve {tool}?" if tool else "Needs your approval"} .get(event_type, event_type)} \ | opt(tool=tool, requestId=(extra or {}).get("requestId")) status, _ = await asyncio.to_thread(_http_post_json, origin + "/api/push/send", body, token, 10) if status == 0: log.debug("[push] gateway notify failed") # ─── HTTP-over-tunnel dispatcher ───────────────────────────────────────────── @dataclass class TunnelReq: id: str method: str path: str headers: dict query: dict # parsed: name → [values] body: bytes @dataclass class HTTPResponse: status: int = 200 headers: dict = field(default_factory=dict) body: bytes | str = b"" sse: SSEClient | None = None def parse_tunnel_request(frame: dict) -> TunnelReq: body = b"" if frame.get("bodyBase64"): with contextlib.suppress(ValueError): body = base64.b64decode(frame["bodyBase64"]) elif frame.get("body"): body = frame["body"].encode() return TunnelReq(frame.get("id") or "", (frame.get("method") or "GET").upper(), frame.get("path") or "/", frame.get("headers") or {}, urllib.parse.parse_qs(frame.get("query") or ""), body) def json_resp(obj, status: int = 200, headers: dict | None = None) -> HTTPResponse: return HTTPResponse(status, {"Content-Type": "application/json"} | (headers or {}), jdump(obj)) def json_err(msg: str, status: int) -> HTTPResponse: return json_resp({"error": msg}, status) def qint(req: TunnelReq, name: str) -> int: try: return int(req.query.get(name, [""])[0]) except ValueError: return 0 def parse_json_body(req: TunnelReq) -> dict | None: try: v = json.loads(req.body.decode("utf-8")) if req.body else {} except (ValueError, UnicodeDecodeError): return None return v if isinstance(v, dict) else None async def dispatch(agent: Agent, req: TunnelReq) -> HTTPResponse: p, meth = req.path, req.method if p == "/health": return json_resp({"status": "ok", "version": VERSION, "sessions": len(agent.sessions.sessions), "uptime": time.time() - agent.start_time}) # /v2 is the canonical namespace; /api is a compatibility alias kept for # deployed clients (web/iOS/Android) until every build calls /v2. if p.startswith("/api/"): p = "/v2/" + p[len("/api/"):] if p == "/v2/info" and meth == "GET": return handle_info(agent) if p == "/v2/events" and meth == "GET": return handle_sse(agent) if p == "/v2/projects" and meth == "GET": return json_resp(await asyncio.to_thread(list_projects, agent.cfg)) if p == "/v2/claude-sessions" and meth == "GET": return json_resp(await asyncio.to_thread( list_claude_sessions, agent.cfg, set(agent.sessions.sessions), req.query.get("project", [""])[0])) if p == "/v2/files" and meth == "GET": return await asyncio.to_thread(handle_files, agent.cfg, req.query.get("project", [""])[0], req.query.get("path", [""])[0]) if p == "/v2/files/content" and meth == "GET": return await asyncio.to_thread(handle_file_content, agent.cfg, req.query.get("project", [""])[0], req.query.get("path", [""])[0]) if p == "/v2/sessions": if meth == "POST": return await handle_create(agent, req) if meth == "GET": return await handle_sessions_list(agent) return json_err("Method not allowed", 405) if p.startswith("/v2/sessions/"): rest = p[len("/v2/sessions/"):] if meth == "POST" and rest.endswith("/plan/dismiss"): return handle_plan_dismiss(agent, rest[:-len("/plan/dismiss")]) sid, _, sub = rest.partition("/") if not sid: return json_err("Not found", 404) if sub == "": if meth == "DELETE": return handle_delete(agent, sid) if meth == "PATCH": return handle_patch(agent, req, sid) return json_err("Not found", 404) if sub == "messages" and meth == "GET": return await handle_v2_messages(agent, req, sid) if sub == "send" and meth == "POST": return await handle_v2_send(agent, req, sid) if sub.startswith("blobs/") and meth == "GET": return await handle_v2_blob(agent, sid, sub[len("blobs/"):]) if sub == "permission" and meth == "POST": return await handle_permission(agent, req, sid) if sub == "stop" and meth == "POST": return handle_stop(agent, sid) if sub == "diff" and meth == "GET": return await handle_session_diff(agent, sid) if sub == "log" and meth == "GET": return await handle_session_log(agent, sid) return json_err("Not found", 404) return json_err("Not found", 404) def handle_info(agent: Agent) -> HTTPResponse: # Fetched by the gateway right after connect (5s deadline) — must not block. ok = shutil.which(agent.cfg.claude_bin) or os.path.isfile(agent.cfg.claude_bin) info = {"agentId": agent.agent_id, "agentName": agent.cfg.agent_name, "online": True, "userAgent": user_agent(), "uptime": int(time.time() - agent.start_time), "version": VERSION, "claudeStatus": "ok" if ok else "not_found"} \ | ({} if ok else {"claudeError": f'Claude CLI not found at "{agent.cfg.claude_bin}"'}) return json_resp({"agents": [info], "models": DEFAULT_MODELS, "mcpServers": []}) def handle_sse(agent: Agent) -> HTTPResponse: c = SSEClient() agent.hub.register(c) c.enqueue(sse_frame("hello", {"clientId": c.id})) c.enqueue(sse_frame("sessions_update", {"type": "sessions_update", "sessions": agent.sessions.list_infos(agent)})) # Replay pending permissions so a (re)connecting client recovers the live # modal — ephemeral control-flow state, never part of /v2/messages. for s in agent.sessions.all(): if s.pending_permission is not None: c.enqueue(sse_frame("permission_request", s.pending_permission)) return HTTPResponse(sse=c) async def handle_sessions_list(agent: Agent) -> HTTPResponse: by_id: dict[str, dict] = {} for s in agent.sessions.all(): by_id[s.id] = {"id": s.id, "source": "rem", "lastActiveAt": parse_rfc3339_ms(s.updated_at), "state": "active" if agent.msgs.is_active(s.id) else "discovered"} \ | opt(title=s.name, project=s.project, model=s.model) for cli in await asyncio.to_thread(scan_cli_sessions, 200): if (info := by_id.get(cli["id"])) is None: info = by_id[cli["id"]] = {"id": cli["id"], "source": "cli", "state": "discovered", "lastActiveAt": cli["mtime"]} | opt(project=cli["project"]) else: info["source"] = "merged" info["lastActiveAt"] = max(info["lastActiveAt"], cli["mtime"]) if cli["project"] and not info.get("project"): info["project"] = cli["project"] if cli.get("title") and not info.get("title"): info["title"] = cli["title"] if agent.msgs.is_active(cli["id"]): info["state"] = "active" msgs, err = await agent.msgs.get(cli["id"]) if err is None: info["msgCount"] = len(msgs) return json_resp({"sessions": sorted(by_id.values(), key=lambda d: d["lastActiveAt"], reverse=True)}) def list_projects(cfg: Config) -> list[dict]: """GET /api/projects — the new-session directory picker (PWA + iOS).""" try: return [{"name": e.name, "mcpServers": []} for e in sorted(Path(cfg.projects_dir).iterdir()) if e.is_dir() and not e.name.startswith(".")] except OSError: return [] def list_claude_sessions(cfg: Config, existing: set[str], project: str) -> list[dict]: """GET /api/claude-sessions?project=X — CLI sessions available for resume.""" if not project or not is_valid_project(cfg, project): return [] abs_project = os.path.abspath(os.path.join(cfg.projects_dir, project)) d = claude_projects_root() / abs_project.replace(os.sep, "-") try: files = [p for p in d.iterdir() if p.name.endswith(".jsonl")] except OSError: return [] out = [] for p in files: sid = p.name[:-len(".jsonl")] if len(sid) < 32: continue try: mtime = datetime.fromtimestamp(p.stat().st_mtime, timezone.utc).strftime(RFC3339) except OSError: continue cs = {"id": sid, "project": project, "firstMsg": "", "model": "", "modTime": mtime, "exists": sid in existing} with contextlib.suppress(OSError): # preview from the first ~50 lines with open(p, encoding="utf-8", errors="replace") as f: for _, line in zip(range(50), f): with contextlib.suppress(ValueError, AttributeError): e = json.loads(line) msg = e.get("message") or {} if e.get("type") == "user" and not cs["firstMsg"] \ and isinstance(msg.get("content"), str): cs["firstMsg"] = msg["content"][:120] elif e.get("type") == "assistant" and not cs["model"] and msg.get("model"): cs["model"] = msg["model"] if cs["firstMsg"] and cs["model"]: break out.append(cs) out.sort(key=lambda c: c["modTime"], reverse=True) return out[:50] def is_valid_project(cfg: Config, project: str) -> bool: if not project: return False cleaned = os.path.normpath(project) if cleaned.startswith("..") or os.path.isabs(cleaned): return False full, root = os.path.abspath(os.path.join(cfg.projects_dir, cleaned)), os.path.abspath(cfg.projects_dir) return full == root or full.startswith(root + os.sep) # ─── File browser + git diff (protocol-neutral panel endpoints) ────────────── # Contracts mirror the Go agent: /api/files is a single-level os.ReadDir (NOT # git), /diff is raw `git diff HEAD` text with an X-Git-Branch header. HIDDEN_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", ".next", ".nuxt"} IMAGE_EXTS = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "gif": "image/gif", "webp": "image/webp", "svg": "image/svg+xml", "ico": "image/x-icon", "bmp": "image/bmp"} MAX_FILE_PREVIEW = 512 * 1024 def resolve_project_path(cfg: Config, project: str, sub: str) -> str: """Resolve project+sub to an absolute path, or "" when invalid. The resolved target must stay inside the project after symlink resolution (blocks traversal and symlink escape, like Go's resolveProjectPath).""" if not is_valid_project(cfg, project): return "" try: base = Path(cfg.projects_dir, project).resolve() target = (base / sub).resolve() if sub else base except OSError: return "" if target != base and base not in target.parents: return "" return str(target) def handle_files(cfg: Config, project: str, sub: str) -> HTTPResponse: d = resolve_project_path(cfg, project, sub) if not d: return json_err("invalid path", 400) try: entries = sorted(os.scandir(d), key=lambda e: e.name) except OSError: return json_err("cannot read directory", 400) dirs, files = [], [] for e in entries: try: is_dir, size = e.is_dir(), e.stat().st_size except OSError: is_dir, size = False, 0 if is_dir and (e.name.startswith(".") or e.name in HIDDEN_DIRS): continue # dot-prefixed FILES are kept, only dirs are skipped if is_dir: dirs.append({"name": e.name, "type": "dir", "size": size}) else: files.append({"name": e.name, "type": "file", "size": size} | opt(ext=e.name.rpartition(".")[2] if "." in e.name else "")) return json_resp(dirs + files) def handle_file_content(cfg: Config, project: str, sub: str) -> HTTPResponse: p = resolve_project_path(cfg, project, sub) if not p: return json_err("invalid path", 400) if not os.path.isfile(p): return json_err("not a file", 400) size = os.path.getsize(p) ext = Path(p).suffix.lstrip(".").lower() if mime := IMAGE_EXTS.get(ext): try: return HTTPResponse(200, {"Content-Type": mime}, Path(p).read_bytes()) except OSError: return json_err("cannot read file", 500) if size > MAX_FILE_PREVIEW: return json_resp({"error": "too_large", "size": size}) # 200 by contract try: content = Path(p).read_bytes().decode("utf-8", errors="replace") except OSError: return json_err("cannot read file", 500) return json_resp({"content": content, "size": size, "ext": ext}) def _git(cwd: str, *args: str) -> str: # encoding= (not text=True): Windows would otherwise decode with the locale # code page and crash on UTF-8 output. try: r = subprocess.run(["git", *args], cwd=cwd, capture_output=True, encoding="utf-8", errors="replace", timeout=30) return r.stdout if r.returncode == 0 else "" except (OSError, subprocess.SubprocessError): return "" # non-git dir / git missing → empty, like Go's execCmd def git_diff(cwd: str) -> tuple[str, str]: branch = _git(cwd, "rev-parse", "--abbrev-ref", "HEAD").strip() diff = _git(cwd, "--no-pager", "diff", "--no-ext-diff", "--no-color", "HEAD") return branch, diff async def handle_session_diff(agent: Agent, sid: str) -> HTTPResponse: if (s := agent.sessions.get(sid)) is None: return json_err("Session not found", 404) branch, diff = await asyncio.to_thread(git_diff, s.project_path(agent.cfg)) return HTTPResponse(200, {"Content-Type": "text/plain", "X-Git-Branch": branch}, diff) def git_log(cwd: str, limit: int = 100) -> list[dict]: # Field/record separators and field order are the Go agent's gitLog verbatim. fsep, rsep = "\x1f", "\x1e" fmt = fsep.join(["%H", "%h", "%an", "%ae", "%at", "%cI", "%s", "%D"]) + rsep out = _git(cwd, "--no-pager", "log", f"-n{limit}", "--pretty=format:" + fmt) commits = [] for rec in out.split(rsep): f = rec.strip("\n").split(fsep) if len(f) < 8: continue commits.append({"hash": f[0], "short": f[1], "author": f[2], "email": f[3], "ts": int(f[4]) if f[4].isdigit() else 0, # author time, unix seconds "date": f[5], # committer date, ISO 8601 "subject": f[6], "refs": f[7]}) # ref names at this commit, may be empty return commits async def handle_session_log(agent: Agent, sid: str) -> HTTPResponse: # GET /api/sessions/{id}/log — the Git panel's Log tab (commit list). if (s := agent.sessions.get(sid)) is None: return json_err("Session not found", 404) cwd = s.project_path(agent.cfg) branch = (await asyncio.to_thread(_git, cwd, "rev-parse", "--abbrev-ref", "HEAD")).strip() return json_resp({"branch": branch, "commits": await asyncio.to_thread(git_log, cwd)}) async def create_session(agent: Agent, req: TunnelReq) -> Session | HTTPResponse: if (body := parse_json_body(req)) is None: return json_err("Invalid request", 400) project, session_id = body.get("project") or "", body.get("sessionId") or "" if session_id and not is_valid_session_id(session_id): return json_err("invalid session id", 400) if not session_id: if project and not is_valid_project(agent.cfg, project): return json_err("invalid project path", 400) elif not project: _, project = await asyncio.to_thread(find_claude_session_file, session_id) # Adopting an existing CLI session starts at message_count=1 → the first # send already uses --resume instead of --session-id. s = agent.sessions.create(project, body.get("permissionMode") or "default", body.get("name") or "", body.get("model") or "", body.get("effort") or "", sid=session_id, message_count=1 if session_id else 0) agent.hub.broadcast_session_list() return s async def handle_create(agent: Agent, req: TunnelReq) -> HTTPResponse: # Clients feed the response straight into their session list, so create # returns the full session object (same shape as sessions_update entries). r = await create_session(agent, req) return r if isinstance(r, HTTPResponse) else json_resp(r.info(agent)) def handle_patch(agent: Agent, req: TunnelReq, sid: str) -> HTTPResponse: # Session settings: rename, model/effort pickers, permission-mode switch. # Unlike Go there is no live CLI to notify — a CLI process exists only # within a turn, and both spawn args and handle_control_request read the # Session fields directly. if (s := agent.sessions.get(sid)) is None: return json_err("Session not found", 404) if (body := parse_json_body(req)) is None: return json_err("Invalid request", 400) s.model = body.get("model") or s.model s.name = body.get("name") or s.name s.effort = body.get("effort") or s.effort if body.get("permissionMode") in ("plan", "default", "acceptEdits", "bypassPermissions"): s.permission_mode = body["permissionMode"] agent.sessions.save(s) agent.hub.broadcast_session_list() return json_resp({"ok": True}) def handle_delete(agent: Agent, sid: str) -> HTTPResponse: # Idempotent; the Claude CLI JSONL (source of truth) is never touched. agent.sessions.delete(sid) agent.msgs.invalidate(sid) agent.hub.broadcast_session_list() return json_resp({"ok": True}) def in_flight_building(agent: Agent, sid: str, msg_seq: int) -> tuple[dict | None, bool]: s = agent.sessions.get(sid) if s is None or not s.processing or s.in_flight is None: return None, bool(s and s.processing) # ONLY the currently-streaming message (live_text): earlier messages of the # turn are already in the CLI JSONL and returned by history. if not s.in_flight.live_text: return None, True return {"msgSeq": msg_seq, "role": "assistant", "status": "streaming", "ts": int(s.in_flight.started_at.timestamp() * 1000), "content": [{"type": "text", "text": s.in_flight.live_text}]}, True async def handle_v2_messages(agent: Agent, req: TunnelReq, sid: str) -> HTTPResponse: msgs, err = await agent.msgs.get(sid) if err is not None: return json_err(f"parse error: {err}", 500) page, has_more = page_messages(msgs, qint(req, "from"), qint(req, "before"), qint(req, "limit")) max_seq = msgs[-1]["msgSeq"] if msgs else 0 building, processing = in_flight_building(agent, sid, max_seq + 1) return json_resp({ "messages": page, "from": qint(req, "from"), "to": page[-1]["msgSeq"] if page else qint(req, "from"), "msgSeq": max_seq, "complete": True, "processing": processing, # literal null when idle — an omitted key would leave a stale building on clients "building": building, "hasMore": has_more, } | opt(markers=markers_in_page(await agent.msgs.markers(sid), page, max_seq)), headers={"Cache-Control": "no-store"}) async def handle_v2_send(agent: Agent, req: TunnelReq, sid: str) -> HTTPResponse: if (s := agent.sessions.get(sid)) is None: # Auto-adopt a CLI-discovered session on first send: clients send to # sessions straight from the discovery list without an explicit # create-with-sessionId. Unknown ids (no CLI JSONL) still 404. _, cwd = await asyncio.to_thread(find_claude_session_file, sid) if not cwd: return json_err("Session not found", 404) s = agent.sessions.create(cwd, "default", "", "", "", sid=sid, message_count=1) agent.hub.broadcast_session_list() if (body := parse_json_body(req)) is None: return json_err("Invalid request", 400) txn_id = body.get("txnId") or "" if cached := agent.txns.lookup(sid, txn_id): return json_resp({"msgSeq": cached[0], "clientId": cached[1], "txnId": txn_id}) client_id = body.get("clientId") or "rem-" + os.urandom(8).hex() text, images = body.get("text") or "", body.get("images") or [] msg_seq = await agent.msgs.allocate_msg_seq(sid) # Pending file first — it is what restart-mid-turn recovery reads back. append_pending(sid, {"text": text, "ts": now_ms()} | opt(txnId=txn_id, clientId=client_id, images=images)) s.updated_at = now_rfc3339() agent.hub.broadcast("user_message", {"type": "user_message", "sessionId": sid, "text": text, "clientId": client_id, "seq": s.next_seq(), "msgSeq": msg_seq} | opt(images=images)) s.queue.append(QueuedTurn(text, images, msg_seq)) # Drop the cached parse so a mid-turn refresh surfaces this send via the # pending fallback instead of a stale pre-send cache. agent.msgs.invalidate(sid) agent.sessions.save(s) ensure_worker(agent, s) agent.txns.record(sid, txn_id, msg_seq, client_id) return json_resp({"msgSeq": msg_seq, "clientId": client_id, "txnId": txn_id}) async def handle_v2_blob(agent: Agent, sid: str, blob_id: str) -> HTTPResponse: if not blob_id: return json_err("missing blob id", 400) path, _ = await asyncio.to_thread(find_claude_session_file, sid) if not path: return json_err("Not found", 404) try: hit = await asyncio.to_thread(extract_cli_blob, path, blob_id) except OSError as e: return json_err(f"blob read: {e}", 500) if hit is None: return json_err("Not found", 404) return HTTPResponse(200, {"Content-Type": hit[0] or "application/octet-stream", "Cache-Control": "public, max-age=31536000, immutable"}, hit[1]) async def handle_permission(agent: Agent, req: TunnelReq, sid: str) -> HTTPResponse: if (s := agent.sessions.get(sid)) is None: return json_err("Session not found", 404) if (body := parse_json_body(req)) is None: return json_err("Invalid request", 400) if s.proc is None or s.proc.stdin is None or (perm := s.pending_permission) is None: return json_err("No pending permission", 400) tool, tuid = perm.get("toolName") or "", perm.get("toolUseId") or "" behavior, updated = body.get("behavior") or "", body.get("updatedInput") if behavior == "allowAlways": s.allowed_tools.add(tool) if behavior == "deny": msg = body.get("message") or "User denied this action" # Plan deny with edits → save PLAN.md so Claude reads the file instead # of re-receiving the full plan text each round. if tool == "ExitPlanModeV2" and isinstance(updated, dict) \ and isinstance(plan := updated.get("plan"), str) and plan: try: Path(s.project_path(agent.cfg), "PLAN.md").write_text(plan, encoding="utf-8") msg += "\n\nI've edited the plan and saved it to PLAN.md — read it for the updated version." except OSError as e: log.warning("[%s] failed to write PLAN.md: %s", s.id, e) response = {"behavior": "deny", "message": msg, "toolUseID": tuid} else: response = {"behavior": "allow", "toolUseID": tuid, "updatedInput": updated if updated is not None else perm.get("toolInput")} await write_stdin_line(s, {"type": "control_response", "response": { "subtype": "success", "request_id": perm.get("requestId") or "", "response": response}}) s.pending_permission = None agent.hub.broadcast("permission_resolved", {"type": "permission_resolved", "sessionId": s.id}) agent.hub.broadcast_session_list() return json_resp({"ok": True}) def handle_stop(agent: Agent, sid: str) -> HTTPResponse: if (s := agent.sessions.get(sid)) is None: return json_err("Session not found", 404) if s.proc and s.proc.returncode is None: with contextlib.suppress(ProcessLookupError): s.proc.kill() s.queue.clear() return json_resp({"ok": True}) def handle_plan_dismiss(agent: Agent, sid: str) -> HTTPResponse: # POST /api/sessions/{id}/plan/dismiss — relays the dismissal to every # connected client so the plan overlay closes on all devices at once. if agent.sessions.get(sid) is None: return json_err("Session not found", 404) agent.hub.broadcast("plan_dismissed", {"type": "plan_dismissed", "sessionId": sid}) return json_resp({"ok": True}) # ─── Tunnel client ─────────────────────────────────────────────────────────── class TunnelClient: def __init__(self, agent: Agent): self.agent = agent self.backoff = RECONNECT_MIN async def run(self) -> None: import websockets # lazy: `login` works without the dependency import websockets.exceptions while True: try: await self._serve_connection(websockets) err: object = "connection closed" except asyncio.CancelledError: raise except websockets.exceptions.ConnectionClosed as e: if e.rcvd and e.rcvd.code == WS_CLOSE_REVOKED: log.error("[tunnel] this agent has been revoked - exiting " "(run `rem_agent.py login` to register a new one)") return err = e except Exception as e: # reconnect on any failure err = e delay = self.backoff * (1 + random.uniform(-0.25, 0.25)) log.info("[tunnel] disconnected: %s - reconnecting in %.1fs", err, delay) await asyncio.sleep(delay) self.backoff = min(self.backoff * 2, RECONNECT_MAX) async def _serve_connection(self, websockets) -> None: cfg = self.agent.cfg async with websockets.connect( cfg.gateway_url, additional_headers={"Authorization": "Bearer " + (self.agent.creds.get("token") or ""), "X-Agent-Id": self.agent.agent_id, "X-Agent-Name": cfg.agent_name}, user_agent_header=user_agent(), max_size=None, # no frame cap — parity with Go's SetReadLimit(-1) open_timeout=10) as ws: welcome = json.loads(await ws.recv()) self.backoff = RECONNECT_MIN log.info("[tunnel] connected (slug: %s, ip: %s)", welcome.get("slug", ""), welcome.get("ip", "")) log.info("[tunnel] open on your phone: %s/login", cfg.gateway_url.removesuffix("/tunnel") .replace("wss://", "https://", 1).replace("ws://", "http://", 1)) # Parallel agents on this account: same id evicts this tunnel, same # NAME gets the peer's token revoked by the gateway's dup cleanup. for p in welcome.get("peers") or []: log.warning("[tunnel] WARNING: another agent is online on this account: %s (%s)", p.get("agentName") or p.get("agentId", "?"), p.get("userAgent", "")) if p.get("agentName") and p.get("agentName") == cfg.agent_name: log.warning("[tunnel] WARNING: it has the SAME agent name - the gateway " "revokes duplicate-name tokens; set REM_AGENT_NAME") lock = asyncio.Lock() tasks: set[asyncio.Task] = set() try: while True: raw = await ws.recv() try: req = parse_tunnel_request(json.loads(raw)) except (ValueError, TypeError) as e: log.warning("[tunnel] invalid request frame: %s", e) continue t = asyncio.create_task(self._handle(ws, lock, req)) tasks.add(t) t.add_done_callback(tasks.discard) finally: for t in list(tasks): # kill in-flight handlers incl. SSE streams t.cancel() async def _send(self, ws, lock: asyncio.Lock, frame: dict) -> None: async with lock: # ws.send is not safe to call concurrently await ws.send(jdump(frame)) async def _handle(self, ws, lock: asyncio.Lock, req: TunnelReq) -> None: try: resp = await dispatch(self.agent, req) except asyncio.CancelledError: raise except Exception: log.exception("[tunnel] handler failed: %s %s", req.method, req.path) resp = json_err("internal error", 500) try: if resp.sse is not None: await self._stream_sse(ws, lock, req.id, resp.sse) return frame = {"id": req.id, "status": resp.status} | opt(headers=resp.headers) body = resp.body.encode() if isinstance(resp.body, str) else resp.body if body: try: frame["body"] = body.decode("utf-8") except UnicodeDecodeError: frame["bodyBase64"] = base64.b64encode(body).decode() await self._send(ws, lock, frame) except asyncio.CancelledError: raise except Exception as e: # socket died; the reconnect loop handles it log.debug("[tunnel] response send failed: %s", e) async def _stream_sse(self, ws, lock: asyncio.Lock, req_id: str, client: SSEClient) -> None: try: await self._send(ws, lock, { "id": req_id, "status": 200, "streaming": True, "headers": {"Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"}}) while not client.done.is_set(): try: frame = await asyncio.wait_for(client.q.get(), timeout=SSE_KEEPALIVE) except TimeoutError: frame = ": keepalive\n\n" await self._send(ws, lock, {"id": req_id, "chunk": frame}) finally: self.agent.hub.unregister(client) with contextlib.suppress(Exception): await self._send(ws, lock, {"id": req_id, "done": True}) # ─── Main / shutdown ───────────────────────────────────────────────────────── async def graceful_shutdown(agent: Agent) -> None: if agent.shutdown_started: return agent.shutdown_started = True mgr = agent.sessions mgr.shutting_down = True if mgr.any_processing(): log.info("shutting down - waiting up to 12s for turns to finish") await mgr.wait_for_idle(12) mgr.kill_all() await mgr.wait_for_idle(3) # let killed turns' finally-blocks capture interrupted text for s in mgr.all(): mgr.save(s) log.info("stopped") async def main_async(agent: Agent) -> None: agent.sessions.load_all() log.info("%s %s - data dir: %s", APP_DIR_NAME, VERSION, data_dir()) try: await TunnelClient(agent).run() except asyncio.CancelledError: # Ctrl-C: asyncio.run cancels the main task await asyncio.shield(graceful_shutdown(agent)) raise await graceful_shutdown(agent) def main() -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%H:%M:%S") ap = argparse.ArgumentParser(description="REM:Code alternative agent (Python, tunnel-only, v2 history)") ap.add_argument("command", nargs="?", default="run", choices=["run", "login"]) cfg = load_config() if ap.parse_args().command == "login": cmd_login(cfg) return if not config_path().exists(): cfg = first_run_setup(cfg) creds = load_credentials() if not creds.get("token"): print("Not logged in - starting device login.") cmd_login(cfg) creds = load_credentials() try: import websockets # noqa: F401 — fail early with a helpful message except ImportError: sys.exit("Missing dependency 'websockets'. Run via `uv run rem_agent.py` " "or `pip install websockets`.") try: asyncio.run(main_async(Agent(cfg, creds, load_or_create_agent_id()))) except KeyboardInterrupt: pass if __name__ == "__main__": main()