From 7d27a31ce779139d8182afb44756e937d39768a6 Mon Sep 17 00:00:00 2001 From: brooklyn! Date: Thu, 16 Jul 2026 15:24:03 -0400 Subject: [PATCH] feat(dashboard): isolate turns in compute host (#65895) Add the flag-gated compute-host supervisor, delta/control protocol, PPID orphan guard, inline fail-open path, synthetic GIL-heavy turn seam, and AC-4 certify harness. Verified on current origin/main: 343 focused tests pass; Ruff and diff checks pass; 360s AC-4 run with six heavy lanes passes at 6.11ms serving p99 with zero stalls and valid load. Co-authored-by: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> --- hermes_cli/config.py | 5 + scripts/iso-certify.py | 625 ++++++++++++++++ tests/test_tui_gateway_server.py | 466 ++++++++++++ tests/tui_gateway/test_compute_host.py | 85 +++ tests/tui_gateway/test_compute_host_phase1.py | 336 +++++++++ tests/tui_gateway/test_iso_certify_seam.py | 110 +++ tui_gateway/compute_host.py | 681 +++++++++++++++++ tui_gateway/host_supervisor.py | 567 ++++++++++++++ tui_gateway/server.py | 690 +++++++++++++++++- tui_gateway/synthetic_turn.py | 231 ++++++ 10 files changed, 3757 insertions(+), 39 deletions(-) create mode 100755 scripts/iso-certify.py create mode 100644 tests/tui_gateway/test_compute_host.py create mode 100644 tests/tui_gateway/test_compute_host_phase1.py create mode 100644 tests/tui_gateway/test_iso_certify_seam.py create mode 100644 tui_gateway/compute_host.py create mode 100644 tui_gateway/host_supervisor.py create mode 100644 tui_gateway/synthetic_turn.py diff --git a/hermes_cli/config.py b/hermes_cli/config.py index bf825664e..f8e48b6e0 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1994,6 +1994,11 @@ DEFAULT_CONFIG = { # Web dashboard settings "dashboard": { "theme": "default", # Dashboard visual theme: "default", "midnight", "ember", "mono", "cyberpunk", "rose" + # Process-isolation rollout controls. Runtime reads these through the + # raw config loader, so tui_gateway.server also owns explicit defaults. + "turn_isolation": False, + "compute_host_heartbeat_secs": 15, + "compute_host_respawn_max": 3, # Hide the token/cost analytics surfaces (Analytics page, token bars and # cost figures on the Models page) by default. The numbers shown there # are a local debug estimate: they only count successful main-agent diff --git a/scripts/iso-certify.py b/scripts/iso-certify.py new file mode 100755 index 000000000..c9b217d18 --- /dev/null +++ b/scripts/iso-certify.py @@ -0,0 +1,625 @@ +#!/usr/bin/env python3 +"""iso-certify — AC-4 dashboard turn-isolation certify harness. + +Certifies Mechanism-B process isolation +(``docs/desktop/2026-07-04-dashboard-process-isolation-PRD.md``, AC-4): under +6 concurrent heavy agent turns, the dashboard's HTTP/ws SERVING plane must stay +responsive (p99 < 1s) with zero event-loop stalls. + +What it does +------------ +1. Spawns a SCRATCH dashboard (``hermes dashboard``) bound to loopback on a + free port, with an ISOLATED ``HERMES_HOME`` (temp dir, minimal seeded state). + It NEVER touches the live :9119 dashboard / ai.hermes.dashboard / live + state.db. Loopback bind ⇒ no auth gate (web_server.should_require_auth). +2. Arms the synthetic GIL-heavy turn seam (``HERMES_ISO_CERTIFY_SYNTH_TURN=1``, + see ``tui_gateway/synthetic_turn.py``) so 6 concurrent turns reproduce the + ``take_gil`` interpreter-contention regime WITHOUT real model calls. A + network/sleep stub would release the GIL and NOT reproduce the incident, so + it would be a fake green — the synthetic turn is pure-Python CPU on purpose. +3. Drives 6 concurrent heavy turns over ws (session.create → prompt.submit), + and CONCURRENTLY probes the serving path — a ws ``session.list`` round-trip + AND a REST ``GET /api/status`` — every 500ms, timing each. +4. Reports p50/p95/p99 latency for both probes + the count of probes over the + 1s threshold ("serving stalls") + the count of ``event loop stalled`` / + ``ws write slow`` lines the dashboard logged during the run. + +Verdict +------- +The AC-4 verdict comes ONLY from the heavy run: PASS iff serving p99 < 1s AND +zero serving stalls AND zero ``event loop stalled`` log lines over the sustained +window. ``--dry-run`` runs ONE short light turn as a plumbing smoke test and is +explicitly NOT a verdict (a dry-run green is a fake green — the spec says so). + +Turn isolation is controlled by the ``dashboard.turn_isolation`` config knob in +the scratch HERMES_HOME; ``--isolation on|off`` sets it. Run BOTH: + iso-certify --isolation off # baseline: expect stalls + iso-certify --isolation on # the measurement that decides AC-4 +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import socket +import subprocess +import sys +import tempfile +import threading +import time +import shutil +import urllib.request +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Any + +try: + from websockets.sync.client import connect as ws_connect +except Exception as exc: # pragma: no cover - dependency guard + print(f"iso-certify requires the 'websockets' package: {exc}", file=sys.stderr) + raise + +REPO_ROOT = Path(__file__).resolve().parents[1] +_READY_RE = re.compile(r"HERMES_(?:DASHBOARD|BACKEND)_READY port=(\d+)") +_STALL_LOG_RE = re.compile(r"event loop stalled|ws write slow \(loop stalled") + + +# ── stats ────────────────────────────────────────────────────────────── +def percentile(values: list[float], pct: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + if len(ordered) == 1: + return ordered[0] + rank = (len(ordered) - 1) * (pct / 100.0) + lo = int(rank) + hi = min(lo + 1, len(ordered) - 1) + frac = rank - lo + return ordered[lo] * (1.0 - frac) + ordered[hi] * frac + + +def summarize(values: list[float]) -> dict[str, float]: + return { + "count": len(values), + "p50_ms": percentile(values, 50), + "p95_ms": percentile(values, 95), + "p99_ms": percentile(values, 99), + "max_ms": max(values) if values else 0.0, + } + + +def free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return int(s.getsockname()[1]) + + +# ── scratch HERMES_HOME ───────────────────────────────────────────────── +def seed_scratch_home(home: Path, *, isolation: str, heartbeat_secs: int, respawn_max: int) -> None: + """Write a minimal config.yaml with the isolation knob set.""" + home.mkdir(parents=True, exist_ok=True) + (home / "state").mkdir(parents=True, exist_ok=True) + (home / "logs").mkdir(parents=True, exist_ok=True) + cfg = { + # Pin the model/provider to what SyntheticHeavyAgent reports so the + # per-turn _sync_agent_model_with_config sees a match and no-ops — a + # mismatch would try (and fail) a real model switch, erroring the turn + # before its heavy loop runs (which would be a FALSE green: serving + # stays responsive because no heavy compute happened). The synthetic + # seam means no real API call is ever made regardless. + "provider": "synthetic", + "model": "synthetic-heavy", + "dashboard": { + "turn_isolation": (isolation == "on"), + "compute_host_heartbeat_secs": heartbeat_secs, + "compute_host_respawn_max": respawn_max, + }, + # Keep memory/mem0/skills side-machinery from reaching out. + "memory": {"enabled": False}, + } + # config.yaml is the canonical config; write it directly. + import yaml # provided by the runtime venv + + (home / "config.yaml").write_text(yaml.safe_dump(cfg, sort_keys=True), encoding="utf-8") + # A stub .env so credential resolution doesn't spelunk the real home. + (home / ".env").write_text("OPENAI_API_KEY=sk-synthetic-not-used\n", encoding="utf-8") + + +# ── dashboard process ─────────────────────────────────────────────────── +class ScratchDashboard: + def __init__( + self, + *, + home: Path, + port: int, + isolation: str, + env_extra: dict[str, str], + ) -> None: + self.home = home + self.port = port + self.isolation = isolation + self.env_extra = env_extra + self.proc: subprocess.Popen[str] | None = None + self.actual_port = port + self.log_lines: list[str] = [] + self._log_lock = threading.Lock() + self._ready = threading.Event() + + @property + def stall_log_count(self) -> int: + with self._log_lock: + return sum(1 for ln in self.log_lines if _STALL_LOG_RE.search(ln)) + + def _drain(self, stream: Any) -> None: + for raw in stream: + line = raw.rstrip("\n") + with self._log_lock: + self.log_lines.append(line) + m = _READY_RE.search(line) + if m: + self.actual_port = int(m.group(1)) + self._ready.set() + + def __enter__(self) -> "ScratchDashboard": + venv_py = REPO_ROOT / "venv" / "bin" / "python" + python = str(venv_py) if venv_py.exists() else sys.executable + env = dict(os.environ) + env.update(self.env_extra) + env["HERMES_HOME"] = str(self.home) + env["HOME"] = str(self.home.parent) if str(self.home.parent) else env.get("HOME", "") + env["HERMES_HOME"] = str(self.home) + env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "") + env["HERMES_ISO_CERTIFY_SYNTH_TURN"] = "1" + cmd = [ + python, "-m", "hermes_cli.main", "dashboard", + "--no-open", "--host", "127.0.0.1", "--port", str(self.port), + ] + self.proc = subprocess.Popen( + cmd, cwd=str(REPO_ROOT), env=env, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, + ) + threading.Thread(target=self._drain, args=(self.proc.stdout,), name="dash-log", daemon=True).start() + if not self._ready.wait(timeout=90.0): + self._dump_tail() + raise RuntimeError("scratch dashboard did not become ready within 90s") + # Give uvicorn a beat to actually bind the ws route. + time.sleep(1.0) + return self + + def __exit__(self, *_exc: object) -> None: + if self.proc is None: + return + try: + self.proc.terminate() + self.proc.wait(timeout=10) + except Exception: + try: + self.proc.kill() + except Exception: + pass + + def _dump_tail(self, n: int = 40) -> None: + with self._log_lock: + tail = self.log_lines[-n:] + sys.stderr.write("---- scratch dashboard log tail ----\n") + for ln in tail: + sys.stderr.write(ln + "\n") + sys.stderr.write("------------------------------------\n") + + +# ── ws client (one connection = one lane) ─────────────────────────────── +class WSClient: + def __init__(self, port: int, token: str) -> None: + self.url = f"ws://127.0.0.1:{port}/api/ws?token={token}" + self.ws = ws_connect( + self.url, + open_timeout=15, + max_size=None, + additional_headers={"Origin": f"http://127.0.0.1:{port}"}, + ) + self._id = 0 + self._lock = threading.Lock() + # Drain the gateway.ready event. + self._recv_until(lambda o: o.get("method") == "event", timeout=10) + + def _next_id(self) -> str: + with self._lock: + self._id += 1 + return f"r{self._id}" + + def _recv_until(self, pred, timeout: float) -> dict: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + raw = self.ws.recv(timeout=max(0.05, deadline - time.monotonic())) + except TimeoutError: + continue + try: + obj = json.loads(raw) + except Exception: + continue + if pred(obj): + return obj + raise TimeoutError("ws recv predicate timed out") + + def rpc(self, method: str, params: dict, timeout: float = 30.0) -> dict: + rid = self._next_id() + self.ws.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params})) + return self._recv_until(lambda o: o.get("id") == rid, timeout=timeout) + + def send_only(self, method: str, params: dict) -> str: + rid = self._next_id() + self.ws.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params})) + return rid + + def close(self) -> None: + try: + self.ws.close() + except Exception: + pass + + +# ── heavy-turn lane ───────────────────────────────────────────────────── +def drive_heavy_turn(port: int, token: str, turn_spec: dict, stop_at: float, results: list[dict]) -> None: + """One lane: create a session, submit heavy turns until the deadline.""" + lane: dict[str, Any] = {"turns": 0, "errors": [], "turn_durations_s": [], "min_deltas": None} + try: + cli = WSClient(port, token) + except Exception as exc: + lane["errors"].append(f"connect: {exc}") + results.append(lane) + return + try: + resp = cli.rpc("session.create", {"cols": 80, "source": "iso-certify"}, timeout=30) + sid = ((resp.get("result") or {}).get("session_id")) or ((resp.get("result") or {}).get("id")) + if not sid: + lane["errors"].append(f"session.create bad resp: {resp}") + results.append(lane) + return + lane["sid"] = sid + spec_text = json.dumps(turn_spec) + while time.monotonic() < stop_at: + # prompt.submit returns immediately ("streaming"); the turn runs async. + r = cli.rpc("prompt.submit", {"session_id": sid, "text": spec_text}, timeout=30) + if r.get("error"): + lane["errors"].append(f"prompt.submit: {r['error']}") + break + # Wait for the REAL turn boundary. The isolated path emits + # session.info MID-turn (metadata mirror), so waiting on it would + # false-complete a turn in <1s and inflate the count while NO heavy + # compute ran — the acceptance-gate "proxy not effect" trap. The + # turn is done only on message.complete. Count deltas + duration so + # a fast-erroring turn (e.g. a failed model switch) is caught, not + # masked as sustained load. + per_turn_budget = turn_spec.get("duration_s", 8.0) + 30.0 + deadline = time.monotonic() + per_turn_budget + turn_start = time.monotonic() + deltas = 0 + started = False + done = False + errored = False + while time.monotonic() < deadline: + try: + o = cli._recv_until( + lambda o: o.get("method") == "event" + and (o.get("params") or {}).get("type") + in {"message.start", "message.delta", "message.complete", "error"}, + timeout=max(0.5, deadline - time.monotonic()), + ) + except TimeoutError: + break + ptype = (o.get("params") or {}).get("type") + if ptype == "message.start": + started = True + turn_start = time.monotonic() + deltas = 0 + continue + # prompt.submit's RPC response may race with a duplicate terminal + # event from the previous turn. Only events after this turn's + # message.start can satisfy or score its load boundary. + if not started: + continue + if ptype == "message.delta": + deltas += 1 + continue + if ptype == "error": + msg = str(((o.get("params") or {}).get("payload") or {}).get("message") or "") + lane["errors"].append(f"turn error: {msg[:160]}") + errored = True + break + if ptype == "message.complete": + done = True + break + turn_dur = time.monotonic() - turn_start + if done: + lane["turns"] += 1 + lane["turn_durations_s"].append(round(turn_dur, 2)) + lane["min_deltas"] = deltas if lane["min_deltas"] is None else min(lane["min_deltas"], deltas) + if errored: + break + except Exception as exc: + lane["errors"].append(f"lane: {exc}") + finally: + cli.close() + results.append(lane) + + +# ── serving-path probes ───────────────────────────────────────────────── +def warmup_serving(port: int, token: str, rounds: int = 6) -> None: + """Prime serving-path caches before the measured window. + + The first ``/api/status`` on a freshly-booted process pays one-time + cold-start cost (config-version check, gateway-health probe, DB connect) that + a real dashboard — warm for hours before the incident regime — never pays + during a stall. AC-4 measures sustained-load responsiveness, not cold boot, + so we hit both serving endpoints a few times UNMEASURED first. This is not a + green-washing shortcut: the measured window still runs the full 6-lane heavy + load; warmup only removes a one-time boot artifact from the p99. + """ + rest_url = f"http://127.0.0.1:{port}/api/status" + warm_ws: WSClient | None = None + try: + warm_ws = WSClient(port, token) + except Exception: + warm_ws = None + for _ in range(rounds): + try: + with urllib.request.urlopen(rest_url, timeout=30) as fh: + fh.read() + except Exception: + pass + if warm_ws is not None: + try: + warm_ws.rpc("session.list", {"limit": 20}, timeout=30) + except Exception: + pass + time.sleep(0.2) + if warm_ws is not None: + warm_ws.close() + + +def probe_loop(port: int, token: str, stop_at: float, cadence_s: float, ws_samples: list[float], rest_samples: list[float]) -> None: + """Probe ws session.list + REST /api/status every ``cadence_s`` seconds.""" + probe_ws: WSClient | None = None + try: + probe_ws = WSClient(port, token) + except Exception: + probe_ws = None + rest_url = f"http://127.0.0.1:{port}/api/status" + next_tick = time.monotonic() + while time.monotonic() < stop_at: + # ws round-trip (session.list — the serving-plane DB read AC-4 protects). + if probe_ws is not None: + t0 = time.perf_counter() + try: + probe_ws.rpc("session.list", {"limit": 20}, timeout=30) + ws_samples.append((time.perf_counter() - t0) * 1000.0) + except Exception: + # A failed/timed-out probe is itself a serving stall; record the + # elapsed as the sample so it counts against p99. + ws_samples.append((time.perf_counter() - t0) * 1000.0) + try: + probe_ws.close() + probe_ws = WSClient(port, token) + except Exception: + probe_ws = None + # REST round-trip. + t1 = time.perf_counter() + try: + with urllib.request.urlopen(rest_url, timeout=30) as fh: + fh.read() + rest_samples.append((time.perf_counter() - t1) * 1000.0) + except Exception: + rest_samples.append((time.perf_counter() - t1) * 1000.0) + next_tick += cadence_s + sleep_for = next_tick - time.monotonic() + if sleep_for > 0: + time.sleep(sleep_for) + else: + next_tick = time.monotonic() + if probe_ws is not None: + probe_ws.close() + + +# ── run ───────────────────────────────────────────────────────────────── +def run_certify(args: argparse.Namespace) -> dict[str, Any]: + port = free_port() + import secrets + token = secrets.token_urlsafe(24) + parent_tmp = Path(tempfile.mkdtemp(prefix="iso-certify-")) + home = parent_tmp / "hermes-home" + seed_scratch_home( + home, + isolation=args.isolation, + heartbeat_secs=args.heartbeat_secs, + respawn_max=args.respawn_max, + ) + + concurrency = 1 if args.dry_run else args.concurrency + duration_s = 3.0 if args.dry_run else args.duration_s + turn_duration = 0.5 if args.dry_run else args.turn_duration_s + threshold_ms = args.threshold_ms + + turn_spec = { + "duration_s": turn_duration, + "delta_interval_s": args.delta_interval_s, + "tokens_per_delta": args.tokens_per_delta, + "chunk": args.chunk, + } + + result: dict[str, Any] = { + "mode": "dry-run" if args.dry_run else "heavy", + "isolation": args.isolation, + "concurrency": concurrency, + "run_duration_s": duration_s, + "turn_spec": turn_spec, + "threshold_ms": threshold_ms, + "scratch_home": str(home), + "port": port, + } + + try: + with ScratchDashboard( + home=home, port=port, isolation=args.isolation, + env_extra={"HERMES_DASHBOARD_SESSION_TOKEN": token}, + ) as dash: + actual_port = dash.actual_port + result["port"] = actual_port + ws_samples: list[float] = [] + rest_samples: list[float] = [] + lane_results: list[dict] = [] + stall_before = dash.stall_log_count + + # Prime serving-path caches so a one-time cold-start artifact does + # not count as a serving stall (only for the real heavy run; a + # dry-run stays a raw plumbing smoke). + if not args.dry_run: + warmup_serving(actual_port, token) + stall_before = dash.stall_log_count + + stop_at = time.monotonic() + duration_s + probe_thread = threading.Thread( + target=probe_loop, + args=(actual_port, token, stop_at, args.cadence_s, ws_samples, rest_samples), + name="probe", daemon=True, + ) + probe_thread.start() + + with ThreadPoolExecutor(max_workers=concurrency) as pool: + for _ in range(concurrency): + pool.submit(drive_heavy_turn, actual_port, token, turn_spec, stop_at, lane_results) + # Pool context waits for all lanes. + probe_thread.join(timeout=30) + + stall_after = dash.stall_log_count + total_turns = sum(l.get("turns", 0) for l in lane_results) + lane_errors = [e for l in lane_results for e in l.get("errors", [])] + all_durations = [d for l in lane_results for d in l.get("turn_durations_s", [])] + min_deltas = [l["min_deltas"] for l in lane_results if l.get("min_deltas") is not None] + lanes_with_turn = sum(1 for l in lane_results if l.get("turns", 0) > 0) + + ws_stat = summarize(ws_samples) + rest_stat = summarize(rest_samples) + ws_over = sum(1 for v in ws_samples if v > threshold_ms) + rest_over = sum(1 for v in rest_samples if v > threshold_ms) + serving_stalls = ws_over + rest_over + log_stalls = stall_after - stall_before + + # Load validity — the run only certifies anything if the offered load + # was REAL: every lane completed ≥1 turn, and the completed turns + # actually held ~the requested heavy duration and streamed deltas. + # A fast-erroring/short turn is NOT sustained GIL load, so a green off + # it would be a proxy (serving stays fast because nothing burned). + median_turn_dur = percentile(all_durations, 50) if all_durations else 0.0 + min_turn_dur = min(all_durations) if all_durations else 0.0 + worst_min_deltas = min(min_deltas) if min_deltas else 0 + expected_turn_s = float(turn_spec.get("duration_s", 8.0)) + load_valid = ( + lanes_with_turn >= concurrency + and total_turns >= concurrency + and min_turn_dur >= 0.7 * expected_turn_s + and worst_min_deltas >= 5 + ) + + result.update({ + "ws_probe": ws_stat, + "rest_probe": rest_stat, + "ws_probes_over_threshold": ws_over, + "rest_probes_over_threshold": rest_over, + "serving_stalls": serving_stalls, + "event_loop_stall_log_lines": log_stalls, + "heavy_turns_completed": total_turns, + "lanes_with_turn": lanes_with_turn, + "median_turn_duration_s": round(median_turn_dur, 2), + "min_turn_duration_s": round(min_turn_dur, 2), + "worst_lane_min_deltas": worst_min_deltas, + "load_valid": load_valid, + "lane_errors": lane_errors[:20], + }) + + # AC-4 verdict — heavy run only. A dry-run reports but never PASSes. + serving_p99 = max(ws_stat["p99_ms"], rest_stat["p99_ms"]) + result["serving_p99_ms"] = serving_p99 + if args.dry_run: + result["verdict"] = "SMOKE-OK" if (total_turns > 0 and not lane_errors) else "SMOKE-FAIL" + result["is_verdict"] = False + else: + serving_ok = ( + serving_p99 < threshold_ms + and serving_stalls == 0 + and log_stalls == 0 + and probe_thread_samples_ok(ws_samples, rest_samples) + ) + if not load_valid: + # Cannot certify: the offered load was not the incident + # regime. Never a PASS; report INCONCLUSIVE, not FAIL, so it + # is not read as "isolation broke serving". + result["verdict"] = "INCONCLUSIVE" + result.setdefault("notes", []).append( + "load invalid: lanes/turn-duration/deltas below the sustained-heavy-load floor — " + "not the AC-4 incident regime, verdict cannot certify" + ) + else: + result["verdict"] = "PASS" if serving_ok else "FAIL" + result["is_verdict"] = True + finally: + if not args.keep_home: + shutil.rmtree(parent_tmp, ignore_errors=True) + else: + result["scratch_home_kept"] = str(parent_tmp) + + return result + + +def probe_thread_samples_ok(ws_samples: list[float], rest_samples: list[float]) -> bool: + """Guard against a blind gate: require the probes actually ran. + + A run that produced no probe samples saw NOTHING — it must not PASS. This is + the tri-state INCONCLUSIVE floor (an empty timeline is never a green). + """ + return len(ws_samples) >= 3 and len(rest_samples) >= 3 + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description="AC-4 dashboard turn-isolation certify harness") + p.add_argument("--isolation", choices=["on", "off"], default="on", + help="set dashboard.turn_isolation in the scratch HERMES_HOME") + p.add_argument("--dry-run", action="store_true", + help="1 short light turn plumbing smoke — NOT an AC-4 verdict") + p.add_argument("--concurrency", type=int, default=6, help="concurrent heavy-turn lanes (AC-4: 6)") + p.add_argument("--duration-s", type=float, default=600.0, dest="duration_s", + help="sustained run window seconds (AC-4: ~600 = 10 min)") + p.add_argument("--turn-duration-s", type=float, default=12.0, dest="turn_duration_s", + help="wall seconds of GIL-holding compute per heavy turn") + p.add_argument("--delta-interval-s", type=float, default=0.05, dest="delta_interval_s", + help="streamed-delta cadence per heavy turn") + p.add_argument("--tokens-per-delta", type=int, default=512, dest="tokens_per_delta") + p.add_argument("--chunk", type=int, default=20000, help="pure-Python ops per interrupt-check chunk") + p.add_argument("--cadence-s", type=float, default=0.5, dest="cadence_s", + help="serving-path probe cadence (AC-4: 500ms)") + p.add_argument("--threshold-ms", type=float, default=1000.0, dest="threshold_ms", + help="serving p99 threshold (AC-4: <1s)") + p.add_argument("--heartbeat-secs", type=int, default=15, dest="heartbeat_secs") + p.add_argument("--respawn-max", type=int, default=3, dest="respawn_max") + p.add_argument("--keep-home", action="store_true", help="do not delete the scratch HERMES_HOME on exit") + p.add_argument("--json-out", type=Path, help="write JSON metrics to this path") + args = p.parse_args(argv) + + result = run_certify(args) + text = json.dumps(result, indent=2, sort_keys=True) + print(text) + if args.json_out: + args.json_out.parent.mkdir(parents=True, exist_ok=True) + args.json_out.write_text(text + "\n", encoding="utf-8") + + # Exit code: 0 only on a real PASS (or a clean dry-run smoke); non-zero + # otherwise. A dry-run is never treated as a verdict for automation. + if result.get("is_verdict"): + return 0 if result.get("verdict") == "PASS" else 1 + return 0 if result.get("verdict") == "SMOKE-OK" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index be46a5e01..adaeeefb9 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -130,6 +130,390 @@ def test_handoff_fail_marks_only_inflight_rows(monkeypatch): server._sessions.pop(sid, None) +def test_dashboard_process_isolation_config_defaults_without_default_merge(monkeypatch): + """tui_gateway.server::_load_cfg is raw YAML, so defaults live at read site.""" + monkeypatch.setattr(server, "_load_cfg", lambda: {}) + + assert server._load_dashboard_process_isolation_config() == { + "turn_isolation": False, + "compute_host_heartbeat_secs": 15, + "compute_host_respawn_max": 3, + } + + +def test_dashboard_process_isolation_config_coerces_raw_values(): + cfg = { + "dashboard": { + "turn_isolation": "yes", + "compute_host_heartbeat_secs": "30", + "compute_host_respawn_max": "0", + } + } + + assert server._load_dashboard_process_isolation_config(cfg) == { + "turn_isolation": True, + "compute_host_heartbeat_secs": 30, + "compute_host_respawn_max": 0, + } + + malformed = {"dashboard": "enabled"} + assert server._load_dashboard_process_isolation_config(malformed) == { + "turn_isolation": False, + "compute_host_heartbeat_secs": 15, + "compute_host_respawn_max": 3, + } + + +def test_default_config_seeds_dashboard_process_isolation_keys(): + from hermes_cli.config import DEFAULT_CONFIG + + dashboard = DEFAULT_CONFIG["dashboard"] + assert dashboard["turn_isolation"] is False + assert dashboard["compute_host_heartbeat_secs"] == 15 + assert dashboard["compute_host_respawn_max"] == 3 + + +def test_prompt_submit_dispatches_to_compute_host_when_turn_isolation_enabled(monkeypatch): + class FakeSupervisor: + def __init__(self): + self.frames = [] + self.callback = None + + def submit_turn(self, frame, *, on_complete=None): + self.frames.append(frame) + self.callback = on_complete + return frame["request_id"] + + fake_supervisor = FakeSupervisor() + seed_history = [{"role": "user", "content": "previous"}] + server._sessions["iso-sid"] = _session(history=list(seed_history)) + server._sessions["iso-sid"]["agent"] = None + server._sessions["iso-sid"]["agent_ready"] = threading.Event() + parent_writes = {"ensure_session": 0, "persist_seed": 0} + monkeypatch.setattr( + server, + "_load_cfg", + lambda: {"dashboard": {"turn_isolation": True}}, + ) + monkeypatch.setattr( + server, + "_ensure_session_db_row", + lambda _session: parent_writes.__setitem__( + "ensure_session", parent_writes["ensure_session"] + 1 + ), + ) + monkeypatch.setattr( + server, + "_persist_branch_seed", + lambda _session: parent_writes.__setitem__( + "persist_seed", parent_writes["persist_seed"] + 1 + ), + ) + monkeypatch.setattr(server, "_get_compute_host_supervisor", lambda _cfg=None: fake_supervisor) + + try: + resp = server.handle_request( + { + "id": "submit", + "method": "prompt.submit", + "params": {"session_id": "iso-sid", "text": "hello"}, + } + ) + assert resp["result"] == {"status": "streaming", "turn_isolation": True} + assert fake_supervisor.frames[0]["type"] == "turn.start" + assert fake_supervisor.frames[0]["sid"] == "iso-sid" + assert fake_supervisor.frames[0]["text"] == "hello" + assert fake_supervisor.frames[0]["history"] == seed_history + assert server._sessions["iso-sid"]["history"] == seed_history + assert parent_writes == {"ensure_session": 0, "persist_seed": 0} + assert server._sessions["iso-sid"]["running"] is True + + fake_supervisor.callback( + { + "type": "turn.end", + "sid": "iso-sid", + "request_id": "submit", + "history_version": 1, + } + ) + assert server._sessions["iso-sid"]["running"] is False + assert server._sessions["iso-sid"]["history_version"] == 1 + finally: + server._sessions.pop("iso-sid", None) + + +def test_prompt_submit_fails_open_inline_when_compute_host_dispatch_breaks(monkeypatch): + class _BrokenSupervisor: + def submit_turn(self, frame, *, on_complete=None): + if on_complete is not None: + on_complete( + { + "type": "turn.error", + "request_id": frame["request_id"], + "reason": "send_failed", + "message": "broken pipe", + } + ) + raise BrokenPipeError("broken pipe") + + class _ImmediateThread: + def __init__(self, target=None, **_kwargs): + self._target = target + + def start(self): + assert self._target is not None + self._target() + + session = _session(agent=None, agent_ready=threading.Event()) + server._sessions["iso-fallback"] = session + inline_calls = [] + monkeypatch.setattr(server, "_load_cfg", lambda: {"dashboard": {"turn_isolation": True}}) + monkeypatch.setattr(server, "_get_compute_host_supervisor", lambda _cfg=None: _BrokenSupervisor()) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda _session: None) + monkeypatch.setattr(server, "_persist_branch_seed", lambda _session: None) + monkeypatch.setattr(server, "_start_agent_build", lambda _sid, _session: None) + monkeypatch.setattr(server, "_wait_agent", lambda _session, _rid: None) + monkeypatch.setattr( + server, + "_run_prompt_submit", + lambda rid, sid, _session, text: inline_calls.append((rid, sid, text)), + ) + monkeypatch.setattr(server.threading, "Thread", _ImmediateThread) + + try: + resp = server.handle_request( + { + "id": "fallback-turn", + "method": "prompt.submit", + "params": {"session_id": "iso-fallback", "text": "hello"}, + } + ) + finally: + server._sessions.pop("iso-fallback", None) + + assert resp == { + "jsonrpc": "2.0", + "id": "fallback-turn", + "result": {"status": "streaming"}, + } + assert inline_calls == [("fallback-turn", "iso-fallback", "hello")] + assert session.get("_compute_host_active") is not True + + +def test_compute_host_turn_end_updates_metadata_mirror(monkeypatch): + session = _session( + agent=None, + agent_ready=threading.Event(), + history=[{"role": "user", "content": "serving process must not read this"}], + _compute_host_active=True, + ) + server._sessions["iso-sid"] = session + emitted = [] + monkeypatch.setattr(server, "_emit", lambda event, sid, payload=None: emitted.append((event, sid, payload))) + + try: + server._on_compute_host_turn_done( + "turn-1", + "iso-sid", + session, + { + "type": "turn.end", + "sid": "iso-sid", + "request_id": "turn-1", + "session_key": "rotated-session-key", + "history_version": 4, + "message_count": 3, + "session_info": { + "model": "host-model", + "provider": "host-provider", + "system_prompt": "host system prompt", + "tools": {"core": ["terminal"]}, + "usage": {"total": 140, "context_used": 80, "context_max": 1000}, + }, + }, + ) + + assert session["session_key"] == "rotated-session-key" + assert session["history_version"] == 4 + assert session["_metadata_mirror"]["model"] == "host-model" + info = server._session_info(None, session) + assert info["model"] == "host-model" + assert info["provider"] == "host-provider" + assert info["system_prompt"] == "host system prompt" + assert info["tools"] == {"core": ["terminal"]} + assert info["usage"]["total"] == 140 + assert "credential_warning" not in info + assert emitted[-1] == ("session.info", "iso-sid", info) + finally: + server._sessions.pop("iso-sid", None) + + +def test_slash_exec_compress_flag_on_applies_host_control_mirror(monkeypatch): + class _ExplodingWorker: + def __init__(self, *args, **kwargs): + raise AssertionError("slash worker should not run for isolated /compress") + + class _FakeSupervisor: + def __init__(self): + self.controls = [] + + def control(self, sid, *, route_name, payload=None, wait=True, timeout=30.0): + self.controls.append((sid, route_name, dict(payload or {}), wait)) + return { + "type": "control.ack", + "sid": sid, + "request_id": (payload or {}).get("request_id", "control-1"), + "route_name": route_name, + "output": "Compressed 4 → 2 messages", + "session_key": "host-rotated-key", + "history_version": 9, + "message_count": 2, + "session_info": { + "model": "host-model", + "provider": "host-provider", + "usage": {"total": 42}, + }, + } + + fake = _FakeSupervisor() + session = _session(agent=None, agent_ready=threading.Event(), _compute_host_active=True) + server._sessions["sid"] = session + monkeypatch.setattr(server, "_load_cfg", lambda: {"dashboard": {"turn_isolation": True}}) + monkeypatch.setattr(server, "_get_compute_host_supervisor", lambda _cfg=None: fake) + monkeypatch.setattr(server, "_SlashWorker", _ExplodingWorker) + monkeypatch.setattr(server, "_compress_session_history", lambda *a, **k: (_ for _ in ()).throw(AssertionError("parent compressed"))) + monkeypatch.setattr(server, "_sync_session_key_after_compress", lambda *a, **k: (_ for _ in ()).throw(AssertionError("parent identity guard ran"))) + + try: + resp = server.handle_request( + { + "id": "1", + "method": "slash.exec", + "params": {"command": "compress focus", "session_id": "sid"}, + } + ) + finally: + server._sessions.pop("sid", None) + + assert resp["result"]["output"] == "Compressed 4 → 2 messages" + assert fake.controls[0][1] == "slash.compress" + assert fake.controls[0][2]["command"] == "/compress focus" + assert session["session_key"] == "host-rotated-key" + assert session["history_version"] == 9 + assert server._session_info(None, session)["model"] == "host-model" + + +def test_prompt_submit_golden_transcript_matches_flag_off_and_on(monkeypatch): + class _ImmediateThread: + def __init__(self, target=None, daemon=None, **_kwargs): + self._target = target + + def start(self): + assert self._target is not None + self._target() + + class _Agent: + model = "gold-model" + provider = "gold-provider" + session_id = "session-key" + session_input_tokens = 10 + session_output_tokens = 5 + session_prompt_tokens = 10 + session_completion_tokens = 5 + session_total_tokens = 15 + session_api_calls = 1 + context_compressor = None + + def clear_interrupt(self): + return None + + def run_conversation(self, prompt, conversation_history=None, stream_callback=None, **_kwargs): + if stream_callback is not None: + stream_callback("hi") + return { + "final_response": "hi", + "messages": [ + {"role": "user", "content": prompt}, + {"role": "assistant", "content": "hi"}, + ], + } + + fixed_info = {"model": "gold-model", "provider": "gold-provider", "usage": {"total": 15}} + usage = server._get_usage(_Agent()) + monkeypatch.setattr(server.threading, "Thread", _ImmediateThread) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda _session: None) + monkeypatch.setattr(server, "_persist_branch_seed", lambda _session: None) + monkeypatch.setattr(server, "_session_info", lambda _agent, _session=None: dict(fixed_info)) + monkeypatch.setattr(server, "make_stream_renderer", lambda _cols: None) + monkeypatch.setattr(server, "render_message", lambda _raw, _cols: None) + fake_title = types.ModuleType("agent.title_generator") + setattr(fake_title, "maybe_auto_title", lambda *args, **kwargs: None) + monkeypatch.setitem(sys.modules, "agent.title_generator", fake_title) + + def run_flag_off(): + events = [] + monkeypatch.setattr(server, "_emit", lambda event, sid, payload=None: events.append((event, sid, payload))) + monkeypatch.setattr(server, "_load_cfg", lambda: {"dashboard": {"turn_isolation": False}}) + server._sessions["sid"] = _session( + agent=_Agent(), model_override={"model": "gold-model", "provider": "gold-provider"} + ) + try: + response = server.handle_request( + {"id": "turn-1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "hello"}} + ) + assert response["result"]["status"] == "streaming" + return events + finally: + server._sessions.pop("sid", None) + + def run_flag_on(): + events = [] + monkeypatch.setattr(server, "_emit", lambda event, sid, payload=None: events.append((event, sid, payload))) + monkeypatch.setattr(server, "_load_cfg", lambda: {"dashboard": {"turn_isolation": True}}) + + class _FakeSupervisor: + def submit_turn(self, frame, *, on_complete=None): + sid = frame["sid"] + server._emit("message.start", sid) + server._emit("message.delta", sid, {"text": "hi"}) + server._emit("message.complete", sid, {"text": "hi", "usage": usage, "status": "complete"}) + server._emit("session.info", sid, dict(fixed_info)) + if on_complete is not None: + on_complete( + { + "type": "turn.end", + "sid": sid, + "request_id": frame["request_id"], + "session_key": "session-key", + "history_version": 1, + "message_count": 2, + "session_info": dict(fixed_info), + "session_info_emitted": True, + } + ) + return frame["request_id"] + + monkeypatch.setattr(server, "_get_compute_host_supervisor", lambda _cfg=None: _FakeSupervisor()) + session = _session( + agent=None, + agent_ready=threading.Event(), + _compute_host_active=True, + model_override={"model": "gold-model", "provider": "gold-provider"}, + ) + session["agent"] = None + server._sessions["sid"] = session + try: + response = server.handle_request( + {"id": "turn-1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "hello"}} + ) + assert response["result"]["status"] == "streaming" + return events + finally: + server._sessions.pop("sid", None) + + assert run_flag_on() == run_flag_off() + + def test_session_context_explicit_cwd_for_ephemeral_task(monkeypatch, tmp_path): """Background/preview tasks use ephemeral ids absent from `_sessions`, so the parent workspace is passed explicitly; it must pin instead of clearing back @@ -4858,6 +5242,88 @@ def test_session_compress_syncs_session_key_after_rotation(monkeypatch): server._sessions.pop("sid", None) +def test_slash_exec_r7_read_commands_use_metadata_mirror_flag_on(monkeypatch): + class _ExplodingWorker: + def __init__(self, *args, **kwargs): + raise AssertionError("slash worker should not run for isolated read commands") + + history_from_db = [ + {"role": "user", "content": "live question from state db"}, + {"role": "assistant", "content": "live answer from state db"}, + ] + + class _DB: + def get_session(self, key): + assert key == "session-key" + return { + "title": "Live title", + "started_at": 1_700_000_000, + "updated_at": 1_700_000_060, + "pinned": True, + } + + def get_messages_as_conversation(self, key, include_ancestors=True): + assert key == "session-key" + assert include_ancestors is True + return list(history_from_db) + + server._sessions["sid"] = _session( + agent=None, + history=[{"role": "user", "content": "stale parent mirror"}], + _compute_host_active=True, + _metadata_mirror={ + "model": "host-model", + "provider": "host-provider", + "system_prompt": "host system prompt", + "tools": {"core": ["terminal", "read_file"]}, + "usage": { + "model": "host-model", + "input": 100, + "output": 20, + "reasoning": 5, + "prompt": 120, + "completion": 20, + "total": 140, + "calls": 2, + "context_used": 80, + "context_max": 1000, + "context_percent": 8, + "compressions": 1, + }, + }, + _metadata_message_count=2, + ) + monkeypatch.setattr(server, "_SlashWorker", _ExplodingWorker) + monkeypatch.setattr(server, "_get_db", lambda: _DB()) + monkeypatch.setattr(server, "_load_cfg", lambda: {"dashboard": {"turn_isolation": True}}) + + cases = { + "usage": "Total tokens: 140", + "history": "live question from state db", + "prompt": "host system prompt", + "status": "Tokens: 140", + "context": "Context usage: ~80 / 1,000 tokens", + "tools": "terminal", + "help": "/status", + } + + try: + for command, expected in cases.items(): + resp = server.handle_request( + { + "id": command, + "method": "slash.exec", + "params": {"command": command, "session_id": "sid"}, + } + ) + assert "result" in resp, (command, resp) + assert expected in resp["result"]["output"] + assert "stale parent mirror" not in resp["result"]["output"] + assert "(._.)" not in resp["result"]["output"] + finally: + server._sessions.pop("sid", None) + + def test_prompt_submit_sets_approval_session_key(monkeypatch): from tools.approval import get_current_session_key diff --git a/tests/tui_gateway/test_compute_host.py b/tests/tui_gateway/test_compute_host.py new file mode 100644 index 000000000..eb748a593 --- /dev/null +++ b/tests/tui_gateway/test_compute_host.py @@ -0,0 +1,85 @@ +import json +import os +import queue +import subprocess +import sys +import threading +from pathlib import Path + + +def _stdout_queue(proc: subprocess.Popen) -> queue.Queue[dict]: + out: queue.Queue[dict] = queue.Queue() + assert proc.stdout is not None + + def drain() -> None: + for line in proc.stdout or []: + out.put(json.loads(line)) + + threading.Thread(target=drain, daemon=True).start() + return out + + +def _read_json_line(out: queue.Queue[dict], timeout: float = 2.0) -> dict: + try: + return out.get(timeout=timeout) + except queue.Empty as exc: + raise AssertionError("timed out waiting for compute host JSON") from exc + + +def test_compute_host_line_json_seed_turn_interrupt(): + repo = Path(__file__).resolve().parents[2] + env = dict(os.environ) + env["PYTHONPATH"] = str(repo) + os.pathsep + env.get("PYTHONPATH", "") + proc = subprocess.Popen( + [sys.executable, "-m", "tui_gateway.compute_host"], + cwd=str(repo), + env=env, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + assert proc.stdin is not None + out = _stdout_queue(proc) + try: + hello = _read_json_line(out) + assert hello["type"] == "hello" + assert hello["host_pid"] == proc.pid + + proc.stdin.write(json.dumps({"type": "session.seed", "sid": "s1", "request_id": "seed"}) + "\n") + proc.stdin.flush() + assert _read_json_line(out)["type"] == "session.seeded" + + proc.stdin.write( + json.dumps( + { + "type": "turn.start", + "sid": "s1", + "request_id": "turn", + "prompt": "hello", + "delta_count": 3, + "delay_s": 0, + } + ) + + "\n" + ) + proc.stdin.flush() + + seen = [] + while True: + frame = _read_json_line(out) + seen.append(frame["type"]) + if frame["type"] == "turn.end": + assert frame["history_version"] == 1 + assert frame["message_count"] == 2 + break + assert seen.count("delta") == 3 + + proc.stdin.write(json.dumps({"type": "shutdown", "request_id": "stop"}) + "\n") + proc.stdin.flush() + assert _read_json_line(out)["type"] == "shutdown.ack" + proc.wait(timeout=2) + finally: + if proc.poll() is None: + proc.kill() diff --git a/tests/tui_gateway/test_compute_host_phase1.py b/tests/tui_gateway/test_compute_host_phase1.py new file mode 100644 index 000000000..473c1fb10 --- /dev/null +++ b/tests/tui_gateway/test_compute_host_phase1.py @@ -0,0 +1,336 @@ +import io +import json +import os +import sys +import threading +import time +from pathlib import Path + +import pytest + +from tui_gateway.compute_host import ComputeHost, _default_workers +from tui_gateway.host_supervisor import ( + MUTATOR_ROUTE_TABLE, + HostSupervisor, + append_log_record, +) + + +def _json_lines(out: io.StringIO) -> list[dict]: + frames = [] + for line in out.getvalue().splitlines(): + if line.strip(): + frames.append(json.loads(line)) + return frames + + +def _wait_for_frame(out: io.StringIO, predicate, timeout: float = 2.0) -> dict: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + for frame in _json_lines(out): + if predicate(frame): + return frame + time.sleep(0.01) + raise AssertionError(f"timed out waiting for frame; saw={_json_lines(out)}") + + +def test_compute_host_workers_inherit_tui_pool_env_or_8(monkeypatch): + monkeypatch.delenv("HERMES_TUI_RPC_POOL_WORKERS", raising=False) + monkeypatch.delenv("HERMES_COMPUTE_HOST_WORKERS", raising=False) + assert _default_workers() == 8 + + monkeypatch.setenv("HERMES_TUI_RPC_POOL_WORKERS", "11") + assert _default_workers() == 11 + + # Dead-RC tombstone: malformed env falls back to 8, not the old except-branch 4. + monkeypatch.setenv("HERMES_TUI_RPC_POOL_WORKERS", "not-an-int") + assert _default_workers() == 8 + + +def test_compute_host_frame_protocol_round_trip(): + out = io.StringIO() + host = ComputeHost(stdout=out, max_workers=2, heartbeat_secs=0) + try: + host.handle_frame({"type": "session.seed", "sid": "alpha", "request_id": "seed", "history": []}) + host.handle_frame( + { + "type": "turn.start", + "sid": "alpha", + "request_id": "turn-1", + "prompt": "hello", + "delta_count": 3, + "delay_s": 0, + } + ) + + end = _wait_for_frame(out, lambda f: f.get("type") == "turn.end" and f.get("request_id") == "turn-1") + assert end["history_version"] == 1 + frames = _json_lines(out) + assert [f["type"] for f in frames if f.get("request_id") == "turn-1"] == [ + "turn.started", + "delta", + "delta", + "delta", + "turn.end", + ] + finally: + host.close() + + +def test_compute_host_interrupt_control_is_not_queued_behind_turn(): + out = io.StringIO() + host = ComputeHost(stdout=out, max_workers=1, heartbeat_secs=0) + try: + host.handle_frame({"type": "session.seed", "sid": "alpha", "request_id": "seed", "history": []}) + host.handle_frame( + { + "type": "turn.start", + "sid": "alpha", + "request_id": "turn-slow", + "prompt": "hello", + "delta_count": 200, + "delay_s": 0.01, + } + ) + _wait_for_frame(out, lambda f: f.get("type") == "delta" and f.get("request_id") == "turn-slow") + + host.handle_frame({"type": "interrupt", "sid": "alpha", "request_id": "stop-1"}) + ack = _wait_for_frame(out, lambda f: f.get("type") == "interrupt.ack" and f.get("request_id") == "stop-1") + assert ack["applied"] is True + + end = _wait_for_frame(out, lambda f: f.get("type") == "turn.end" and f.get("request_id") == "turn-slow") + assert end["interrupted"] is True + typed = [f["type"] for f in _json_lines(out)] + assert typed.index("interrupt.ack") < typed.index("turn.end") + finally: + host.close() + + +def test_compute_host_flushes_sessions_on_orphan_shutdown(monkeypatch): + from tui_gateway import server + + out = io.StringIO() + host = ComputeHost(stdout=out, max_workers=1, heartbeat_secs=0) + session = {"session_key": "key"} + calls: list[tuple[dict, str]] = [] + server._sessions["flush-sid"] = session + monkeypatch.setattr( + server, + "_finalize_session", + lambda sess, end_reason="tui_close": calls.append((sess, end_reason)), + ) + try: + host.flush_all_sessions(reason="orphan") + assert calls == [(session, "compute_host_orphan")] + finally: + server._sessions.pop("flush-sid", None) + host.close() + + +def test_compute_host_parent_guard_exits_when_parent_pid_changes(monkeypatch): + out = io.StringIO() + host = ComputeHost(stdout=out, max_workers=1, heartbeat_secs=0) + host._parent_pid = 111 + monkeypatch.setattr(os, "getppid", lambda: 222) + + def _exit(code): + raise SystemExit(code) + + monkeypatch.setattr(os, "_exit", _exit) + + with pytest.raises(SystemExit) as exc_info: + host._parent_guard_loop() + + assert exc_info.value.code == 0 + orphan = next(frame for frame in _json_lines(out) if frame.get("type") == "orphan") + assert orphan["old_ppid"] == 111 + assert orphan["ppid"] == 222 + assert isinstance(orphan["host_ns"], int) + + +def test_mutator_route_table_matches_prd_inventory(): + assert MUTATOR_ROUTE_TABLE == { + "prompt.submit": "turn-path", + "session.interrupt": "turn-path", + "reload.mcp": "run-concurrent", + "session.compress": "idle-gated", + "prompt.submit.truncate": "idle-gated", + "slash.model": "idle-gated", + "slash.personality": "idle-gated", + "slash.prompt": "idle-gated", + "slash.compress": "idle-gated", + "session.reset": "idle-gated", + "session.history.reload": "idle-gated", + "slash.retry": "idle-gated", + } + + +def test_compute_host_compress_control_runs_identity_guard_in_host(monkeypatch): + from tui_gateway import server + + out = io.StringIO() + host = ComputeHost(stdout=out, max_workers=1, heartbeat_secs=0) + + class _Agent: + model = "host-model" + provider = "host-provider" + tools = [] + _cached_system_prompt = "" + session_input_tokens = 1 + session_output_tokens = 1 + session_prompt_tokens = 1 + session_completion_tokens = 1 + session_total_tokens = 2 + session_api_calls = 1 + context_compressor = None + + session = { + "agent": _Agent(), + "session_key": "before-key", + "history": [ + {"role": "user", "content": "before"}, + {"role": "assistant", "content": "before"}, + ], + "history_lock": threading.Lock(), + "history_version": 2, + "running": False, + "manual_compression_lock": threading.Lock(), + } + calls: dict[str, object] = {} + + def _compress(sess, focus_topic=None, **_kwargs): + assert sess is session + calls["compress_focus"] = focus_topic + with sess["history_lock"]: + sess["history"] = [{"role": "summary", "content": "compressed in host"}] + sess["history_version"] = 3 + + def _sync(sid, sess): + assert sess is session + calls["sync"] = sid + sess["session_key"] = "after-key" + + server._sessions["sid"] = session + monkeypatch.setenv("HERMES_COMPUTE_HOST_CHILD", "1") + monkeypatch.setattr(server, "_compress_session_history", _compress) + monkeypatch.setattr(server, "_sync_session_key_after_compress", _sync) + monkeypatch.setattr(server, "_emit", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + server, + "_session_info", + lambda _agent, _session=None: { + "model": "host-model", + "provider": "host-provider", + "usage": {"total": 2}, + }, + ) + + try: + host.handle_frame( + { + "type": "control", + "sid": "sid", + "request_id": "compress-1", + "route_name": "slash.compress", + "command": "/compress focus", + } + ) + ack = _wait_for_frame( + out, + lambda f: f.get("type") == "control.ack" and f.get("request_id") == "compress-1", + ) + finally: + server._sessions.pop("sid", None) + host.close() + + assert calls == {"compress_focus": "focus", "sync": "sid"} + assert ack["route_name"] == "slash.compress" + assert ack["session_key"] == "after-key" + assert ack["history_version"] == 3 + assert ack["message_count"] == 1 + assert ack["session_info"]["model"] == "host-model" + + +def test_append_log_record_single_write_lines(tmp_path): + path = tmp_path / "agent.log" + + def writer(i: int) -> None: + append_log_record(path, f"line-{i:03d}-" + ("x" * 2000)) + + threads = [threading.Thread(target=writer, args=(i,)) for i in range(32)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + lines = path.read_text(encoding="utf-8").splitlines() + assert len(lines) == 32 + assert sorted(line.split("-", 2)[1] for line in lines) == [f"{i:03d}" for i in range(32)] + assert all(line.endswith("x" * 2000) for line in lines) + + +def test_supervisor_startup_reconcile_pid_reuse_guard(tmp_path, monkeypatch): + registry = tmp_path / "dashboard-compute-host.json" + registry.write_text(json.dumps({"host_pid": os.getpid(), "boot_id": "stale"}), encoding="utf-8") + + killed: list[int] = [] + supervisor = HostSupervisor(registry_path=registry, argv=[sys.executable, "-c", ""], autostart=False) + monkeypatch.setattr(supervisor, "_pid_matches_compute_host", lambda _pid: False) + monkeypatch.setattr(supervisor, "_terminate_pid", lambda pid, **_kw: killed.append(pid)) + + result = supervisor.reconcile_startup_orphan() + + assert result == "pid-reuse-ignored" + assert killed == [] + assert not registry.exists() + + +def test_supervisor_crash_emits_turn_error_and_respawns(tmp_path): + script = tmp_path / "fake_host.py" + script.write_text( + """ +import json, os, sys +print(json.dumps({'type':'hello','host_pid':os.getpid(),'boot_id':'boot-1','build_sha':'test','hermes_home':os.environ.get('HERMES_HOME','')}), flush=True) +for raw in sys.stdin: + frame=json.loads(raw) + if frame.get('type') == 'shutdown': + print(json.dumps({'type':'shutdown.ack','request_id':frame.get('request_id')}), flush=True) + break + if frame.get('type') == 'turn.start': + print(json.dumps({'type':'turn.started','sid':frame.get('sid'),'request_id':frame.get('request_id')}), flush=True) + sys.stdout.flush() + os._exit(7) +""".strip(), + encoding="utf-8", + ) + registry = tmp_path / "dashboard-compute-host.json" + completions: list[dict] = [] + rpc_events: list[dict] = [] + supervisor = HostSupervisor( + registry_path=registry, + argv=[sys.executable, str(script)], + rpc_sink=rpc_events.append, + respawn_max=2, + heartbeat_secs=1, + expected_build_sha="test", + autostart=False, + ) + try: + supervisor.start() + supervisor.submit_turn( + {"type": "turn.start", "sid": "sid-1", "request_id": "turn-1", "text": "hello"}, + on_complete=completions.append, + ) + deadline = time.monotonic() + 5 + while time.monotonic() < deadline and not completions: + time.sleep(0.02) + assert completions, "host crash did not complete pending turn" + assert completions[0]["type"] == "turn.error" + assert completions[0]["reason"] == "crash" + + deadline = time.monotonic() + 5 + while time.monotonic() < deadline and not supervisor.is_running(): + time.sleep(0.02) + assert supervisor.is_running() + finally: + supervisor.shutdown() diff --git a/tests/tui_gateway/test_iso_certify_seam.py b/tests/tui_gateway/test_iso_certify_seam.py new file mode 100644 index 000000000..3a9a41342 --- /dev/null +++ b/tests/tui_gateway/test_iso_certify_seam.py @@ -0,0 +1,110 @@ +"""Tests for the AC-4 isolation certify seam + harness helpers. + +The synthetic heavy-turn agent (``tui_gateway/synthetic_turn.py``) is a test +seam: dead unless ``HERMES_ISO_CERTIFY_SYNTH_TURN=1``. These tests pin (a) the +dead-when-unset contract, (b) that an armed turn holds for the requested wall +duration and streams deltas, (c) that interrupt aborts it promptly, and (d) the +harness percentile math. +""" + +from __future__ import annotations + +import importlib.util +import threading +import time +from pathlib import Path + +import pytest + +from tui_gateway.synthetic_turn import ( + SyntheticHeavyAgent, + maybe_build_synthetic_agent, + synth_turn_armed, +) + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _load_iso_certify(): + path = REPO_ROOT / "scripts" / "iso-certify.py" + spec = importlib.util.spec_from_file_location("iso_certify", path) + assert spec and spec.loader + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def test_synth_seam_dead_when_env_unset(monkeypatch): + monkeypatch.delenv("HERMES_ISO_CERTIFY_SYNTH_TURN", raising=False) + assert synth_turn_armed() is False + assert maybe_build_synthetic_agent("sid") is None + + +def test_synth_seam_armed_builds_agent(monkeypatch): + monkeypatch.setenv("HERMES_ISO_CERTIFY_SYNTH_TURN", "1") + assert synth_turn_armed() is True + agent = maybe_build_synthetic_agent("sid", {"model": "custom-x"}) + assert isinstance(agent, SyntheticHeavyAgent) + assert agent.model == "custom-x" + + +def test_synth_turn_holds_duration_and_streams(): + agent = SyntheticHeavyAgent("s1") + deltas: list[str] = [] + spec = '{"duration_s": 0.4, "delta_interval_s": 0.05, "tokens_per_delta": 100}' + t0 = time.monotonic() + result = agent.run_conversation(spec, stream_callback=deltas.append) + elapsed = time.monotonic() - t0 + # Held for ~the requested wall time (allow generous upper bound under load). + assert 0.35 <= elapsed <= 1.5, elapsed + assert result["interrupted"] is False + assert len(deltas) >= 3 + # Token accounting advanced (the 100K-token heavy-turn proxy). + assert agent.session_output_tokens == 100 * len(deltas) + assert agent.session_api_calls == 1 + # Role alternation preserved in the produced messages. + roles = [m["role"] for m in result["messages"]] + assert roles[-2:] == ["user", "assistant"] + + +def test_synth_turn_interrupt_aborts_promptly(): + agent = SyntheticHeavyAgent("s2") + + def _stop(): + time.sleep(0.2) + agent.interrupt() + + threading.Thread(target=_stop, daemon=True).start() + t0 = time.monotonic() + result = agent.run_conversation('{"duration_s": 10.0}') + elapsed = time.monotonic() - t0 + assert result["interrupted"] is True + assert elapsed < 1.5, elapsed + + +def test_synth_turn_non_json_prompt_uses_defaults(monkeypatch): + monkeypatch.setenv("HERMES_ISO_CERTIFY_DURATION_S", "0.2") + agent = SyntheticHeavyAgent("s3") + t0 = time.monotonic() + agent.run_conversation("just a plain prompt, not json") + assert time.monotonic() - t0 >= 0.15 + + +def test_harness_percentile_and_guard(): + iso = _load_iso_certify() + assert iso.percentile([], 99) == 0.0 + assert iso.percentile([5.0], 99) == 5.0 + vals = [float(i) for i in range(1, 101)] # 1..100 + assert 98.0 <= iso.percentile(vals, 99) <= 100.0 + assert iso.percentile(vals, 50) == pytest.approx(50.5, abs=0.6) + # The empty-timeline INCONCLUSIVE floor: too few probe samples never PASSes. + assert iso.probe_thread_samples_ok([1.0, 2.0], [1.0, 2.0, 3.0]) is False + assert iso.probe_thread_samples_ok([1.0, 2.0, 3.0], [1.0, 2.0, 3.0]) is True + + +def test_harness_summarize_shape(): + iso = _load_iso_certify() + s = iso.summarize([10.0, 20.0, 30.0]) + assert s["count"] == 3 + assert s["max_ms"] == 30.0 + assert set(s) == {"count", "p50_ms", "p95_ms", "p99_ms", "max_ms"} diff --git a/tui_gateway/compute_host.py b/tui_gateway/compute_host.py new file mode 100644 index 000000000..bc2a42d4f --- /dev/null +++ b/tui_gateway/compute_host.py @@ -0,0 +1,681 @@ +"""Persistent dashboard compute-host process. + +Phase 0 used this module as a deterministic line-JSON spike. Phase 1 keeps the +same transport and turns it into the long-lived child that owns live AIAgent +objects when ``dashboard.turn_isolation`` is enabled. +""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import json +import os +import signal +import subprocess +import sys +import threading +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable + + +def now_ns() -> int: + return time.perf_counter_ns() + + +@dataclass +class SpikeAgent: + """A deterministic AIAgent-shaped object for pipe/interrupt measurements.""" + + session_id: str + history: list[dict[str, str]] = field(default_factory=list) + _interrupt: threading.Event = field(default_factory=threading.Event) + + def clear_interrupt(self) -> None: + self._interrupt.clear() + + def interrupt(self) -> None: + self._interrupt.set() + + def run_conversation( + self, + prompt: str, + *, + conversation_history: list[dict[str, str]] | None = None, + stream_callback: Callable[[str], None] | None = None, + delta_count: int = 24, + delay_s: float = 0.001, + ) -> dict[str, Any]: + base_history = list(conversation_history if conversation_history is not None else self.history) + chunks: list[str] = [] + interrupted = False + for index in range(max(0, int(delta_count))): + if self._interrupt.is_set(): + interrupted = True + break + chunk = f"{self.session_id}:{prompt}:{index:04d} " + chunks.append(chunk) + if stream_callback is not None: + stream_callback(chunk) + if delay_s > 0: + time.sleep(delay_s) + if self._interrupt.is_set(): + interrupted = True + final = "".join(chunks) + if interrupted: + final += "[interrupted]" + messages = [ + *base_history, + {"role": "user", "content": prompt}, + {"role": "assistant", "content": final}, + ] + self.history = messages + return {"final_response": final, "messages": messages, "interrupted": interrupted} + + +@dataclass +class HostSession: + sid: str + agent: SpikeAgent + history_version: int = 0 + running: bool = False + lock: threading.Lock = field(default_factory=threading.Lock) + + +class _HostTransport: + def __init__(self, emit: Callable[[dict[str, Any]], None]) -> None: + self._emit = emit + + def write(self, obj: dict) -> bool: + sid = "" + try: + if obj.get("method") == "event": + sid = str(((obj.get("params") or {}).get("session_id")) or "") + except Exception: + sid = "" + self._emit({"type": "rpc", "sid": sid, "message": obj}) + return True + + def close(self) -> None: + return None + + +def _repo_root() -> Path: + return Path(__file__).resolve().parents[1] + + +def _build_sha() -> str: + try: + return subprocess.check_output( + ["git", "rev-parse", "HEAD"], + cwd=str(_repo_root()), + text=True, + stderr=subprocess.DEVNULL, + timeout=2, + ).strip() + except Exception: + return "unknown" + + +class ComputeHost: + def __init__( + self, + *, + stdout: Any = None, + max_workers: int | None = None, + heartbeat_secs: int | float | None = None, + ) -> None: + self._stdout = stdout or sys.stdout + self._write_lock = threading.Lock() + self._sessions: dict[str, HostSession] = {} + self._executor = concurrent.futures.ThreadPoolExecutor( + max_workers=max_workers or _default_workers(), + thread_name_prefix="compute-host-turn", + ) + self._closed = threading.Event() + self._parent_pid = os.getppid() + self._boot_id = uuid.uuid4().hex + self._progress_counter = 0 + self._progress_lock = threading.Lock() + self._turn_futures: set[concurrent.futures.Future] = set() + self._turn_futures_lock = threading.Lock() + self._transport = _HostTransport(self.emit) + self._heartbeat_secs = ( + float(heartbeat_secs) + if heartbeat_secs is not None + else float(os.environ.get("HERMES_COMPUTE_HOST_HEARTBEAT_SECS") or "15") + ) + if self._heartbeat_secs > 0: + threading.Thread(target=self._heartbeat_loop, name="compute-host-heartbeat", daemon=True).start() + threading.Thread(target=self._parent_guard_loop, name="compute-host-ppid-guard", daemon=True).start() + + def emit(self, frame: dict[str, Any]) -> None: + frame.setdefault("host_ns", now_ns()) + data = json.dumps(frame, separators=(",", ":"), ensure_ascii=False) + with self._write_lock: + print(data, file=self._stdout, flush=True) + + def close(self) -> None: + self._closed.set() + self._executor.shutdown(wait=False, cancel_futures=True) + + def shutdown(self, *, reason: str = "shutdown", wait: float = 10.0) -> None: + self._closed.set() + self.flush_all_sessions(reason=reason) + deadline = time.monotonic() + max(0.0, wait) + while time.monotonic() < deadline: + with self._turn_futures_lock: + pending = [f for f in self._turn_futures if not f.done()] + if not pending: + break + time.sleep(0.05) + self._executor.shutdown(wait=False, cancel_futures=True) + + def flush_all_sessions(self, *, reason: str = "shutdown") -> None: + try: + from tui_gateway import server + except Exception: + return + for session in list(getattr(server, "_sessions", {}).values()): + try: + server._finalize_session(session, end_reason=f"compute_host_{reason}") + except Exception: + pass + + def handle_frame(self, frame: dict[str, Any]) -> None: + kind = str(frame.get("type") or "") + if kind == "session.seed": + self._handle_seed(frame) + elif kind == "turn.start": + self._handle_turn_start(frame) + elif kind == "interrupt": + self._handle_interrupt(frame) + elif kind == "reload_mcp": + self._handle_reload_mcp(frame) + elif kind == "control": + self._handle_control(frame) + elif kind == "shutdown": + self.emit({"type": "shutdown.ack", "request_id": frame.get("request_id")}) + # Explicit supervisor/test shutdown is a clean child-process close; + # SIGTERM and orphan paths are the durability flush paths. + self._closed.set() + self._executor.shutdown(wait=False, cancel_futures=True) + else: + self.emit( + { + "type": "error", + "request_id": frame.get("request_id"), + "message": f"unknown frame type: {kind}", + } + ) + + # ── Phase-0 deterministic spike frames ───────────────────────────── + + def _handle_seed(self, frame: dict[str, Any]) -> None: + sid = str(frame.get("sid") or "") + if not sid: + self.emit({"type": "error", "request_id": frame.get("request_id"), "message": "sid required"}) + return + history = frame.get("history") + if not isinstance(history, list): + history = [] + self._sessions[sid] = HostSession(sid=sid, agent=SpikeAgent(sid, list(history))) + self.emit({"type": "session.seeded", "sid": sid, "request_id": frame.get("request_id")}) + + def _handle_turn_start(self, frame: dict[str, Any]) -> None: + sid = str(frame.get("sid") or "") + if sid in self._sessions: + self._handle_spike_turn_start(frame) + return + future = self._executor.submit(self._run_real_turn, dict(frame)) + with self._turn_futures_lock: + self._turn_futures.add(future) + future.add_done_callback(self._turn_futures.discard) + + def _handle_spike_turn_start(self, frame: dict[str, Any]) -> None: + sid = str(frame.get("sid") or "") + session = self._sessions.get(sid) + if session is None: + self.emit({"type": "turn.error", "sid": sid, "request_id": frame.get("request_id"), "message": "unknown session"}) + return + with session.lock: + if session.running: + self.emit({"type": "turn.error", "sid": sid, "request_id": frame.get("request_id"), "message": "session busy"}) + return + session.running = True + future = self._executor.submit(self._run_spike_turn, session, dict(frame)) + with self._turn_futures_lock: + self._turn_futures.add(future) + future.add_done_callback(self._turn_futures.discard) + + def _handle_interrupt(self, frame: dict[str, Any]) -> None: + sid = str(frame.get("sid") or "") + spike = self._sessions.get(sid) + if spike is not None: + spike.agent.interrupt() + self.emit( + { + "type": "interrupt.ack", + "sid": sid, + "request_id": frame.get("request_id"), + "applied": True, + "applied_ns": now_ns(), + } + ) + return + try: + from tui_gateway import server + + session = server._sessions.get(sid) + if session is None: + self.emit({"type": "interrupt.ack", "sid": sid, "request_id": frame.get("request_id"), "applied": False}) + return + agent = session.get("agent") + if agent is not None and hasattr(agent, "interrupt"): + agent.interrupt() + with session.get("history_lock", threading.Lock()): + session["_turn_cancel_requested"] = True + session["queued_prompt"] = None + self.emit({"type": "interrupt.ack", "sid": sid, "request_id": frame.get("request_id"), "applied": True, "applied_ns": now_ns()}) + except Exception as exc: + self.emit({"type": "interrupt.ack", "sid": sid, "request_id": frame.get("request_id"), "applied": False, "message": str(exc)}) + + def _run_spike_turn(self, session: HostSession, frame: dict[str, Any]) -> None: + request_id = frame.get("request_id") or uuid.uuid4().hex + prompt = str(frame.get("prompt") or frame.get("text") or "") + try: + delta_count = int(frame.get("delta_count", 24)) + except (TypeError, ValueError): + delta_count = 24 + try: + delay_s = float(frame.get("delay_s", 0.001)) + except (TypeError, ValueError): + delay_s = 0.001 + with session.lock: + history = list(session.agent.history) + session.agent.clear_interrupt() + self.emit({"type": "turn.started", "sid": session.sid, "request_id": request_id, "started_ns": now_ns()}) + + def stream(delta: str) -> None: + self._bump_progress() + self.emit( + { + "type": "delta", + "sid": session.sid, + "request_id": request_id, + "text": delta, + "emitted_ns": now_ns(), + } + ) + + try: + result = session.agent.run_conversation( + prompt, + conversation_history=history, + stream_callback=stream, + delta_count=delta_count, + delay_s=delay_s, + ) + with session.lock: + session.history_version += 1 + session.running = False + history_version = session.history_version + self._bump_progress() + self.emit( + { + "type": "turn.end", + "sid": session.sid, + "request_id": request_id, + "history_version": history_version, + "message_count": len(result.get("messages") or []), + "interrupted": bool(result.get("interrupted")), + "ended_ns": now_ns(), + } + ) + except Exception as exc: # pragma: no cover - defensive host boundary + with session.lock: + session.running = False + self.emit({"type": "turn.error", "sid": session.sid, "request_id": request_id, "message": str(exc)}) + + # ── Real dashboard turn path ─────────────────────────────────────── + + def _run_real_turn(self, frame: dict[str, Any]) -> None: + sid = str(frame.get("sid") or "") + request_id = str(frame.get("request_id") or uuid.uuid4().hex) + if not sid: + self.emit({"type": "turn.error", "sid": sid, "request_id": request_id, "message": "sid required"}) + return + try: + from tui_gateway import server + + session = self._ensure_server_session(server, frame) + with session["history_lock"]: + if session.get("running"): + self.emit({"type": "turn.error", "sid": sid, "request_id": request_id, "message": "session busy"}) + return + session["running"] = True + session["_turn_cancel_requested"] = False + session["last_active"] = time.time() + server._start_inflight_turn(session, frame.get("text") if "text" in frame else frame.get("prompt")) + self.emit({"type": "turn.started", "sid": sid, "request_id": request_id, "started_ns": now_ns()}) + try: + server._ensure_session_db_row(session) + except Exception: + pass + try: + import hermes_undo + + hermes_undo.on_user_message_appended(session["session_key"]) + except Exception: + pass + try: + server._persist_branch_seed(session) + except Exception: + pass + text = frame.get("text") if "text" in frame else frame.get("prompt", "") + server._run_prompt_submit(request_id, sid, session, text) + run_thread = session.get("_run_thread") + if run_thread is not None and hasattr(run_thread, "join"): + run_thread.join() + with session["history_lock"]: + history_version = int(session.get("history_version", 0)) + message_count = len(session.get("history") or []) + interrupted = bool(session.get("_turn_cancel_requested")) + session_key = str(session.get("session_key") or "") + session_info = server._session_info(session.get("agent"), session) + self._bump_progress() + self.emit( + { + "type": "turn.end", + "sid": sid, + "request_id": request_id, + "history_version": history_version, + "session_key": session_key, + "message_count": message_count, + "interrupted": interrupted, + "ended_ns": now_ns(), + "session_info": session_info, + "session_info_emitted": True, + } + ) + except Exception as exc: + try: + from tui_gateway import server + + session = server._sessions.get(sid) + if session is not None: + with session.get("history_lock", threading.Lock()): + session["running"] = False + server._clear_inflight_turn(session) + except Exception: + pass + self.emit({"type": "turn.error", "sid": sid, "request_id": request_id, "reason": "exception", "message": str(exc)}) + + def _ensure_server_session(self, server: Any, frame: dict[str, Any]) -> dict: + sid = str(frame.get("sid") or "") + key = str(frame.get("session_key") or sid) + session = server._sessions.get(sid) + if session is not None: + session["transport"] = self._transport + if frame.get("cols") is not None: + session["cols"] = int(frame.get("cols") or 80) + if frame.get("cwd"): + session["cwd"] = str(frame.get("cwd")) + if frame.get("profile_home"): + session["profile_home"] = str(frame.get("profile_home")) + if isinstance(frame.get("attached_images"), list): + session["attached_images"] = list(frame.get("attached_images") or []) + return session + + history = frame.get("history") if isinstance(frame.get("history"), list) else [] + profile_home = str(frame.get("profile_home") or "") + session_db = None + home_token = None + try: + if profile_home: + from hermes_constants import set_hermes_home_override + from hermes_state import SessionDB + + home_token = set_hermes_home_override(profile_home) + session_db = SessionDB(db_path=Path(profile_home) / "state.db") + agent = server._make_agent( + sid, + key, + session_id=key, + model_override=frame.get("model_override"), + reasoning_config_override=frame.get("reasoning_config_override"), + service_tier_override=frame.get("service_tier_override"), + platform_override=frame.get("source"), + session_db=session_db, + ) + finally: + if home_token is not None: + try: + from hermes_constants import reset_hermes_home_override + + reset_hermes_home_override(home_token) + except Exception: + pass + try: + from tui_gateway.transport import bind_transport, reset_transport + + token = bind_transport(self._transport) + try: + server._init_session( + sid, + key, + agent, + list(history), + cols=int(frame.get("cols") or 80), + cwd=str(frame.get("cwd") or "") or None, + session_db=session_db, + source=frame.get("source"), + ) + finally: + reset_transport(token) + except Exception: + # If _init_session's side machinery (slash worker, approval notify) is + # unavailable, keep a minimal host-owned session rather than failing + # the turn after the expensive agent build succeeded. + server._sessions[sid] = { + "agent": agent, + "session_key": key, + "history": list(history), + "history_lock": threading.Lock(), + "history_version": int(frame.get("history_version") or 0), + "inflight_turn": None, + "created_at": time.time(), + "last_active": time.time(), + "running": False, + "attached_images": [], + "image_counter": 0, + "cwd": str(frame.get("cwd") or os.getcwd()), + "cols": int(frame.get("cols") or 80), + "slash_worker": None, + "show_reasoning": server._load_show_reasoning(), + "tool_progress_mode": server._load_tool_progress_mode(), + "edit_snapshots": {}, + "tool_started_at": {}, + "model_override": frame.get("model_override"), + "source": server._sanitize_client_source(frame.get("source")), + "transport": self._transport, + } + session = server._sessions[sid] + session["transport"] = self._transport + session["profile_home"] = profile_home or session.get("profile_home") + if isinstance(frame.get("attached_images"), list): + session["attached_images"] = list(frame.get("attached_images") or []) + if frame.get("model_override") is not None: + session["model_override"] = frame.get("model_override") + return session + + def _handle_reload_mcp(self, frame: dict[str, Any]) -> None: + sid = str(frame.get("sid") or "") + request_id = frame.get("request_id") + try: + from tui_gateway import server + + resp = server.handle_request({"id": request_id, "method": "reload.mcp", "params": {"session_id": sid, "confirm": True}}) + self.emit({"type": "reload_mcp.ack", "sid": sid, "request_id": request_id, "response": resp}) + except Exception as exc: + self.emit({"type": "control.error", "sid": sid, "request_id": request_id, "message": str(exc)}) + + def _handle_control(self, frame: dict[str, Any]) -> None: + sid = str(frame.get("sid") or "") + request_id = frame.get("request_id") + route_name = str(frame.get("route_name") or "") + try: + from tui_gateway import server + from tui_gateway.host_supervisor import MUTATOR_ROUTE_TABLE + + route = MUTATOR_ROUTE_TABLE.get(route_name) + if route is None: + self.emit({"type": "control.error", "sid": sid, "request_id": request_id, "message": f"unclassified route: {route_name}"}) + return + session = server._sessions.get(sid) + if session is None: + self.emit({"type": "control.error", "sid": sid, "request_id": request_id, "message": "session not found"}) + return + if route == "idle-gated" and session.get("running"): + self.emit({"type": "control.error", "sid": sid, "request_id": request_id, "message": "session busy"}) + return + if route_name == "reload.mcp": + self._handle_reload_mcp({**frame, "type": "reload_mcp"}) + return + command = str(frame.get("command") or "") + output = "" + if command: + output = server._mirror_slash_side_effects(sid, session, command) + with session["history_lock"]: + history_version = int(session.get("history_version", 0)) + message_count = len(session.get("history") or []) + session_key = str(session.get("session_key") or "") + self.emit( + { + "type": "control.ack", + "sid": sid, + "request_id": request_id, + "route_name": route_name, + "output": output, + "session_key": session_key, + "history_version": history_version, + "message_count": message_count, + "session_info": server._session_info(session.get("agent"), session), + } + ) + except Exception as exc: + self.emit({"type": "control.error", "sid": sid, "request_id": request_id, "message": str(exc)}) + + def _bump_progress(self) -> None: + with self._progress_lock: + self._progress_counter += 1 + + def _heartbeat_loop(self) -> None: + while not self._closed.wait(self._heartbeat_secs): + with self._turn_futures_lock: + active_turns = sum(1 for f in self._turn_futures if not f.done()) + with self._progress_lock: + counter = self._progress_counter + self.emit( + { + "type": "hb", + "active_turns": active_turns, + "progress_counter": counter, + "rss_mb": _rss_mb(os.getpid()), + } + ) + + def _parent_guard_loop(self) -> None: + while not self._closed.wait(1.0): + ppid = os.getppid() + if ppid in {0, 1} or (self._parent_pid and ppid != self._parent_pid): + self.emit({"type": "orphan", "old_ppid": self._parent_pid, "ppid": ppid}) + self.shutdown(reason="orphan") + os._exit(0) + + +def _rss_mb(pid: int) -> float: + try: + out = subprocess.check_output(["ps", "-o", "rss=", "-p", str(pid)], text=True, stderr=subprocess.DEVNULL, timeout=2).strip() + return int(out.splitlines()[-1].strip()) / 1024.0 if out else 0.0 + except Exception: + return 0.0 + + +def _default_workers() -> int: + try: + return max(2, int(os.environ.get("HERMES_TUI_RPC_POOL_WORKERS") or "8")) + except (TypeError, ValueError): + return 8 + + +def run_host(stdin: Any = None, stdout: Any = None) -> None: + os.environ["HERMES_COMPUTE_HOST_CHILD"] = "1" + stdin = stdin or sys.stdin + host = ComputeHost(stdout=stdout or sys.stdout) + shutting_down = threading.Event() + + def _signal_handler(_signum, _frame) -> None: + if shutting_down.is_set(): + return + shutting_down.set() + host.shutdown(reason="sigterm") + raise SystemExit(0) + + try: + signal.signal(signal.SIGTERM, _signal_handler) + signal.signal(signal.SIGINT, _signal_handler) + except Exception: + pass + + host.emit( + { + "type": "hello", + "host_pid": os.getpid(), + "boot_id": host._boot_id, + "build_sha": _build_sha(), + "cwd": os.getcwd(), + "hermes_home": os.environ.get("HERMES_HOME", ""), + } + ) + + def _reader() -> None: + for raw in stdin: + if host._closed.is_set(): + break + try: + frame = json.loads(raw) + except json.JSONDecodeError as exc: + host.emit({"type": "error", "message": f"invalid json: {exc}"}) + continue + if not isinstance(frame, dict): + host.emit({"type": "error", "message": "frame must be an object"}) + continue + host.handle_frame(frame) + if frame.get("type") == "shutdown": + os._exit(0) + if host._closed.is_set(): + break + + reader = threading.Thread(target=_reader, name="compute-host-control-reader", daemon=True) + reader.start() + try: + while not host._closed.wait(0.2): + if not reader.is_alive(): + break + finally: + host.shutdown(reason="stdin_closed", wait=2.0) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Dashboard compute-host process") + parser.parse_args(argv) + run_host() + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/tui_gateway/host_supervisor.py b/tui_gateway/host_supervisor.py new file mode 100644 index 000000000..941d9f69b --- /dev/null +++ b/tui_gateway/host_supervisor.py @@ -0,0 +1,567 @@ +"""Supervisor for the dashboard compute-host child process. + +The dashboard process owns sockets and JSON-RPC dispatch. When +``dashboard.turn_isolation`` is enabled, agent turns move behind one persistent +``python -m tui_gateway.compute_host`` child so compute-heavy agent threads do +not contend with the serving process' event loop for the same GIL. +""" + +from __future__ import annotations + +import json +import logging +import os +import queue +import signal +import subprocess +import sys +import threading +import time +import uuid +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from hermes_constants import get_hermes_home +from tools.environments.local import hermes_subprocess_env + +logger = logging.getLogger(__name__) +_Thread = threading.Thread + +MUTATOR_ROUTE_TABLE: dict[str, str] = { + "prompt.submit": "turn-path", + "session.interrupt": "turn-path", + "reload.mcp": "run-concurrent", + "session.compress": "idle-gated", + "prompt.submit.truncate": "idle-gated", + "slash.model": "idle-gated", + "slash.personality": "idle-gated", + "slash.prompt": "idle-gated", + "slash.compress": "idle-gated", + "session.reset": "idle-gated", + "session.history.reload": "idle-gated", + "slash.retry": "idle-gated", +} + +_REGISTRY_NAME = "dashboard-compute-host.json" +_RESPAWN_WINDOW_SECS = 300.0 +_SHUTDOWN_TIMEOUT_SECS = 10.0 + + +def append_log_record(path: str | Path, record: str) -> None: + """Append one log record using O_APPEND and exactly one os.write call.""" + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + text = record if record.endswith("\n") else f"{record}\n" + data = text.encode("utf-8", errors="replace") + fd = os.open(str(p), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + try: + os.write(fd, data) + finally: + os.close(fd) + + +def _repo_root() -> Path: + return Path(__file__).resolve().parents[1] + + +def _build_sha() -> str: + try: + return subprocess.check_output( + ["git", "rev-parse", "HEAD"], + cwd=str(_repo_root()), + text=True, + stderr=subprocess.DEVNULL, + timeout=2, + ).strip() + except Exception: + return "unknown" + + +def _default_registry_path() -> Path: + return get_hermes_home() / "state" / _REGISTRY_NAME + + +def _pid_alive(pid: int) -> bool: + if pid <= 0: + return False + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + except Exception: + return False + + +def _pid_command(pid: int) -> str: + if pid <= 0: + return "" + # Linux fast path. + proc_cmdline = Path("/proc") / str(pid) / "cmdline" + try: + data = proc_cmdline.read_bytes() + if data: + return data.replace(b"\x00", b" ").decode("utf-8", errors="replace") + except Exception: + pass + try: + return subprocess.check_output( + ["ps", "-p", str(pid), "-o", "command="], + text=True, + stderr=subprocess.DEVNULL, + timeout=2, + ).strip() + except Exception: + return "" + + +def is_compute_host_identity(pid: int) -> bool: + cmd = _pid_command(pid) + return "tui_gateway.compute_host" in cmd + + +class HostSupervisor: + """Own one persistent compute-host child and relay its frames.""" + + def __init__( + self, + *, + registry_path: str | Path | None = None, + argv: list[str] | None = None, + cwd: str | Path | None = None, + env: dict[str, str] | None = None, + rpc_sink: Callable[[dict], None] | None = None, + respawn_max: int = 3, + heartbeat_secs: int = 15, + expected_build_sha: str | None = None, + expected_hermes_home: str | None = None, + autostart: bool = True, + ) -> None: + self.registry_path = Path(registry_path) if registry_path is not None else _default_registry_path() + self.argv = argv or [sys.executable, "-m", "tui_gateway.compute_host"] + self.cwd = Path(cwd) if cwd is not None else _repo_root() + self.env = env + self.rpc_sink = rpc_sink or (lambda _obj: None) + self.respawn_max = max(0, int(respawn_max)) + self.heartbeat_secs = max(1, int(heartbeat_secs)) + self.expected_build_sha = expected_build_sha if expected_build_sha is not None else _build_sha() + self.expected_hermes_home = expected_hermes_home if expected_hermes_home is not None else str(get_hermes_home()) + + self._lock = threading.RLock() + self._proc: subprocess.Popen[str] | None = None + self._stdout_thread: threading.Thread | None = None + self._stderr_thread: threading.Thread | None = None + self._wait_thread: threading.Thread | None = None + self._hello_event = threading.Event() + self._hello: dict[str, Any] = {} + self._closing = False + self._stopped_respawning = False + self._restart_times: list[float] = [] + self._pending_turns: dict[str, tuple[str, Callable[[dict], None] | None]] = {} + self._pending_controls: dict[str, queue.Queue[dict]] = {} + self._stderr_tail: list[str] = [] + self._last_progress_counter = 0 + + if autostart: + self.start() + + @property + def pid(self) -> int: + proc = self._proc + return int(proc.pid or 0) if proc is not None else 0 + + @property + def hello(self) -> dict[str, Any]: + return dict(self._hello) + + def is_running(self) -> bool: + proc = self._proc + return proc is not None and proc.poll() is None and not self._stopped_respawning + + def start(self) -> None: + with self._lock: + if self.is_running(): + return + self._closing = False + self.reconcile_startup_orphan() + self._spawn_locked(reason="startup") + + def shutdown(self) -> None: + with self._lock: + self._closing = True + proc = self._proc + if proc is None: + return + try: + if proc.poll() is None and proc.stdin is not None: + self._send_frame({"type": "shutdown", "request_id": f"shutdown-{uuid.uuid4().hex}"}) + proc.wait(timeout=_SHUTDOWN_TIMEOUT_SECS) + except Exception: + self._terminate_process(proc) + finally: + self._remove_registry() + + def reconcile_startup_orphan(self) -> str: + """Terminate a stale registered host, guarding against PID reuse.""" + try: + data = json.loads(self.registry_path.read_text(encoding="utf-8")) + except FileNotFoundError: + return "none" + except Exception: + self._remove_registry() + return "invalid-registry" + + try: + pid = int(data.get("host_pid") or 0) + except Exception: + pid = 0 + if pid <= 0 or not _pid_alive(pid): + self._remove_registry() + return "not-running" + if not self._pid_matches_compute_host(pid): + # PID was reused by another process. Never signal it. + self._remove_registry() + return "pid-reuse-ignored" + + self._terminate_pid(pid, timeout=_SHUTDOWN_TIMEOUT_SECS) + self._remove_registry() + return "terminated" + + def submit_turn( + self, + frame: dict[str, Any], + *, + on_complete: Callable[[dict], None] | None = None, + ) -> str: + self.start() + request_id = str(frame.get("request_id") or uuid.uuid4().hex) + sid = str(frame.get("sid") or "") + payload = dict(frame) + payload["type"] = "turn.start" + payload["request_id"] = request_id + with self._lock: + self._pending_turns[request_id] = (sid, on_complete) + try: + self._send_frame(payload) + except Exception as exc: + with self._lock: + self._pending_turns.pop(request_id, None) + err = { + "type": "turn.error", + "sid": sid, + "request_id": request_id, + "reason": "send_failed", + "message": str(exc), + } + if on_complete is not None: + on_complete(err) + raise + return request_id + + def interrupt(self, sid: str, *, request_id: str | None = None) -> None: + self.start() + self._send_frame({"type": "interrupt", "sid": sid, "request_id": request_id or uuid.uuid4().hex}) + + def reload_mcp(self, sid: str, *, request_id: str | None = None) -> dict: + return self.control( + sid, + route_name="reload.mcp", + payload={"type": "reload_mcp", "sid": sid, "request_id": request_id or uuid.uuid4().hex}, + wait=True, + ) + + def control( + self, + sid: str, + *, + route_name: str, + payload: dict[str, Any] | None = None, + wait: bool = True, + timeout: float = 30.0, + ) -> dict: + if route_name not in MUTATOR_ROUTE_TABLE: + raise ValueError(f"unclassified host mutator route: {route_name}") + self.start() + request_id = str((payload or {}).get("request_id") or uuid.uuid4().hex) + frame = dict(payload or {}) + frame.setdefault("type", "control") + frame["sid"] = sid + frame["route_name"] = route_name + frame["request_id"] = request_id + q: queue.Queue[dict] | None = None + if wait: + q = queue.Queue(maxsize=1) + with self._lock: + self._pending_controls[request_id] = q + self._send_frame(frame) + if not wait or q is None: + return {"status": "sent", "request_id": request_id} + try: + return q.get(timeout=timeout) + finally: + with self._lock: + self._pending_controls.pop(request_id, None) + + def _spawn_locked(self, *, reason: str) -> None: + if self._stopped_respawning: + raise RuntimeError("compute host respawn disabled after crash loop") + self._hello_event.clear() + self._hello = {} + env = hermes_subprocess_env(inherit_credentials=True) + env.update(os.environ) + if self.env: + env.update(self.env) + env["HERMES_COMPUTE_HOST_HEARTBEAT_SECS"] = str(self.heartbeat_secs) + env.setdefault("PYTHONPATH", str(_repo_root())) + if str(_repo_root()) not in env["PYTHONPATH"].split(os.pathsep): + env["PYTHONPATH"] = str(_repo_root()) + os.pathsep + env["PYTHONPATH"] + proc = subprocess.Popen( + self.argv, + cwd=str(self.cwd), + env=env, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + start_new_session=True, + ) + self._proc = proc + self._stdout_thread = _Thread(target=self._drain_stdout, args=(proc,), name="compute-host-stdout", daemon=True) + self._stderr_thread = _Thread(target=self._drain_stderr, args=(proc,), name="compute-host-stderr", daemon=True) + self._wait_thread = _Thread(target=self._wait_for_exit, args=(proc,), name="compute-host-wait", daemon=True) + self._stdout_thread.start() + self._stderr_thread.start() + self._wait_thread.start() + if not self._hello_event.wait(timeout=10.0): + self._terminate_process(proc) + raise RuntimeError(f"compute host did not send hello; stderr={self._stderr_tail[-5:]}") + self._validate_hello() + self._persist_registry() + logger.info("compute host started pid=%s reason=%s", proc.pid, reason) + + def _validate_hello(self) -> None: + hello = self._hello + if not hello: + raise RuntimeError("compute host missing hello") + got_home = str(hello.get("hermes_home") or "") + if got_home and got_home != self.expected_hermes_home: + raise RuntimeError(f"compute host HERMES_HOME mismatch: {got_home} != {self.expected_hermes_home}") + got_sha = str(hello.get("build_sha") or "") + if self.expected_build_sha != "unknown" and got_sha not in {"", "unknown", self.expected_build_sha}: + raise RuntimeError(f"compute host build mismatch: {got_sha} != {self.expected_build_sha}") + + def _persist_registry(self) -> None: + self.registry_path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.registry_path.with_suffix(self.registry_path.suffix + ".tmp") + payload = { + "host_pid": self.pid, + "boot_id": self._hello.get("boot_id") or "", + "build_sha": self._hello.get("build_sha") or "", + "started_at": time.time(), + "argv": self.argv, + } + tmp.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8") + tmp.replace(self.registry_path) + + def _remove_registry(self) -> None: + try: + self.registry_path.unlink() + except FileNotFoundError: + pass + except Exception: + logger.debug("failed to remove compute host registry", exc_info=True) + + def _send_frame(self, frame: dict[str, Any]) -> None: + with self._lock: + proc = self._proc + if proc is None or proc.poll() is not None or proc.stdin is None: + raise RuntimeError("compute host is not running") + proc.stdin.write(json.dumps(frame, separators=(",", ":"), ensure_ascii=False) + "\n") + proc.stdin.flush() + + def _drain_stdout(self, proc: subprocess.Popen[str]) -> None: + assert proc.stdout is not None + for raw in proc.stdout: + try: + frame = json.loads(raw) + except json.JSONDecodeError: + logger.warning("compute host emitted invalid json: %r", raw[:200]) + continue + if isinstance(frame, dict): + self._handle_host_frame(frame) + + def _drain_stderr(self, proc: subprocess.Popen[str]) -> None: + assert proc.stderr is not None + for raw in proc.stderr: + text = raw.rstrip("\n") + if text: + self._stderr_tail = (self._stderr_tail + [text])[-80:] + logger.warning("compute host stderr: %s", text) + + def _handle_host_frame(self, frame: dict[str, Any]) -> None: + ftype = str(frame.get("type") or "") + if ftype == "hello": + self._hello = dict(frame) + self._hello_event.set() + return + if ftype == "hb": + self._last_progress_counter = int(frame.get("progress_counter") or self._last_progress_counter) + logger.debug("compute host heartbeat: %s", frame) + return + if ftype == "rpc": + message = frame.get("message") + if isinstance(message, dict): + self.rpc_sink(message) + return + if ftype in {"turn.end", "turn.error"}: + self._complete_turn(frame) + return + if ftype in {"control.ack", "control.error", "interrupt.ack", "reload_mcp.ack", "shutdown.ack"}: + request_id = str(frame.get("request_id") or "") + with self._lock: + q = self._pending_controls.get(request_id) + if q is not None: + try: + q.put_nowait(frame) + except queue.Full: + pass + return + if ftype == "error" and frame.get("request_id"): + request_id = str(frame.get("request_id") or "") + with self._lock: + q = self._pending_controls.get(request_id) + if q is not None: + try: + q.put_nowait(frame) + except queue.Full: + pass + + def _complete_turn(self, frame: dict[str, Any]) -> None: + request_id = str(frame.get("request_id") or "") + with self._lock: + pending = self._pending_turns.pop(request_id, None) + if pending is None: + return + _sid, cb = pending + if cb is not None: + try: + cb(frame) + except Exception: + logger.exception("compute host turn completion callback failed") + + def _wait_for_exit(self, proc: subprocess.Popen[str]) -> None: + code = proc.wait() + if self._closing: + return + with self._lock: + if self._proc is not proc: + return + self._proc = None + self._remove_registry() + self._fail_pending_turns(reason="crash", message=f"compute host exited with code {code}") + self._maybe_respawn_after_crash() + + def _fail_pending_turns(self, *, reason: str, message: str) -> None: + with self._lock: + pending = self._pending_turns + self._pending_turns = {} + for request_id, (sid, cb) in pending.items(): + frame = { + "type": "turn.error", + "sid": sid, + "request_id": request_id, + "reason": reason, + "message": message, + } + self.rpc_sink( + { + "jsonrpc": "2.0", + "method": "event", + "params": { + "type": "error", + "session_id": sid, + "payload": {"message": message, "reason": reason}, + }, + } + ) + if cb is not None: + try: + cb(frame) + except Exception: + logger.exception("compute host error callback failed") + + def _maybe_respawn_after_crash(self) -> None: + now = time.monotonic() + self._restart_times = [t for t in self._restart_times if now - t <= _RESPAWN_WINDOW_SECS] + if len(self._restart_times) >= self.respawn_max: + self._stopped_respawning = True + logger.error("compute host crash loop: max %s restarts per 5min reached; not respawning", self.respawn_max) + return + self._restart_times.append(now) + # Small bounded backoff; tests and first recovery stay quick. + delay = min(5.0, 0.25 * (2 ** max(0, len(self._restart_times) - 1))) + + def _respawn() -> None: + time.sleep(delay) + with self._lock: + if self._closing or self._stopped_respawning or self._proc is not None: + return + try: + self._spawn_locked(reason="crash") + except Exception: + logger.exception("compute host respawn failed") + + _Thread(target=_respawn, name="compute-host-respawn", daemon=True).start() + + def _pid_matches_compute_host(self, pid: int) -> bool: + return is_compute_host_identity(pid) + + def _terminate_pid(self, pid: int, *, timeout: float = _SHUTDOWN_TIMEOUT_SECS) -> None: + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + return + except Exception: + logger.debug("failed to SIGTERM compute host pid=%s", pid, exc_info=True) + return + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not _pid_alive(pid): + return + time.sleep(0.05) + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + return + except Exception: + logger.debug("failed to SIGKILL compute host pid=%s", pid, exc_info=True) + + def _terminate_process(self, proc: subprocess.Popen[str]) -> None: + if proc.poll() is not None: + return + try: + proc.terminate() + proc.wait(timeout=_SHUTDOWN_TIMEOUT_SECS) + return + except Exception: + pass + try: + proc.kill() + except Exception: + pass + try: + proc.wait(timeout=2) + except Exception: + pass + + +__all__ = [ + "MUTATOR_ROUTE_TABLE", + "HostSupervisor", + "append_log_record", + "is_compute_host_identity", +] diff --git a/tui_gateway/server.py b/tui_gateway/server.py index fd3385979..47b10a0d4 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -237,7 +237,7 @@ try: 2, int(os.environ.get("HERMES_TUI_RPC_POOL_WORKERS") or "8") ) except (ValueError, TypeError): - _rpc_pool_workers = 4 + _rpc_pool_workers = 8 _pool = concurrent.futures.ThreadPoolExecutor( max_workers=_rpc_pool_workers, thread_name_prefix="tui-rpc", @@ -1144,6 +1144,180 @@ def _emit(event: str, sid: str, payload: dict | None = None): write_json({"jsonrpc": "2.0", "method": "event", "params": params}) +_compute_host_supervisor = None +_compute_host_supervisor_lock = threading.Lock() + + +def _inside_compute_host_child() -> bool: + return os.environ.get("HERMES_COMPUTE_HOST_CHILD") == "1" + + +def _turn_isolation_enabled(cfg: dict | None = None) -> bool: + if _inside_compute_host_child(): + return False + isolation_cfg = cfg or _load_dashboard_process_isolation_config() + return bool(isolation_cfg.get("turn_isolation")) + + +def _session_uses_compute_host(session: dict, cfg: dict | None = None) -> bool: + if not _turn_isolation_enabled(cfg): + return False + # Phase 1 routes lazy/dashboard sessions whose live AIAgent has not been + # built inside the serving process. Already-built in-process sessions keep + # the historical path unless a prior isolated turn marked host ownership. + return bool(session.get("_compute_host_active")) or ( + session.get("agent") is None and session.get("agent_ready") is not None + ) + + +def _get_compute_host_supervisor(cfg: dict | None = None): + global _compute_host_supervisor + isolation_cfg = cfg or _load_dashboard_process_isolation_config() + with _compute_host_supervisor_lock: + if _compute_host_supervisor is None: + from tui_gateway.host_supervisor import HostSupervisor + + _compute_host_supervisor = HostSupervisor( + rpc_sink=write_json, + heartbeat_secs=int(isolation_cfg.get("compute_host_heartbeat_secs") or 15), + respawn_max=int(isolation_cfg.get("compute_host_respawn_max") or 3), + ) + return _compute_host_supervisor + + +def _compute_host_turn_frame(rid: str, sid: str, session: dict, text: Any) -> dict: + with session["history_lock"]: + history = list(session.get("history", [])) + history_version = int(session.get("history_version", 0)) + attached_images = list(session.get("attached_images", [])) + return { + "type": "turn.start", + "sid": sid, + "request_id": rid, + "session_key": session.get("session_key") or sid, + "text": text, + "history": history, + "history_version": history_version, + "cols": int(session.get("cols", 80) or 80), + "cwd": _session_cwd(session), + "profile_home": session.get("profile_home") or "", + "model_override": session.get("model_override"), + "reasoning_config_override": session.get("create_reasoning_override"), + "service_tier_override": session.get("create_service_tier_override"), + "source": _session_source(session), + "attached_images": attached_images, + } + + +def _metadata_mirror(session: dict | None) -> dict: + mirror = (session or {}).get("_metadata_mirror") + return mirror if isinstance(mirror, dict) else {} + + +def _apply_compute_host_metadata_mirror(session: dict, frame: dict | None) -> None: + """Mirror host-owned session metadata in the serving process. + + The compute host is the only writer of live agent/history state while turn + isolation is active. The serving process keeps read metadata from the last + host frame so UI reads do not construct a second in-process agent. + """ + if not isinstance(frame, dict): + return + with session.get("history_lock", threading.Lock()): + if frame.get("session_key"): + session["session_key"] = str(frame.get("session_key")) + if frame.get("history_version") is not None: + try: + session["history_version"] = max( + int(session.get("history_version", 0)), + int(frame.get("history_version") or 0), + ) + except Exception: + pass + if frame.get("message_count") is not None: + try: + session["_metadata_message_count"] = int(frame.get("message_count") or 0) + except Exception: + pass + info = frame.get("session_info") + if isinstance(info, dict): + mirror = dict(_metadata_mirror(session)) + mirror.update(info) + session["_metadata_mirror"] = mirror + session["_metadata_mirror_updated_at"] = time.time() + + +def _on_compute_host_turn_done(rid: str, sid: str, session: dict, frame: dict) -> None: + is_error = frame.get("type") == "turn.error" + with session["history_lock"]: + if frame.get("session_key"): + session["session_key"] = str(frame.get("session_key")) + if frame.get("history_version") is not None: + try: + session["history_version"] = max( + int(session.get("history_version", 0)), + int(frame.get("history_version") or 0), + ) + except Exception: + pass + session["running"] = False + session["last_active"] = time.time() + _clear_inflight_turn(session) + if is_error: + message = str(frame.get("message") or "compute host turn failed") + _emit("message.complete", sid, {"text": f"Error: {message}", "status": "error"}) + _apply_compute_host_metadata_mirror(session, frame) + try: + info = _session_info(session.get("agent"), session) + except TypeError: + info = _session_info(session.get("agent")) + if not frame.get("session_info_emitted"): + _emit("session.info", sid, info) + _drain_queued_prompt(rid, sid, session) + + +def _submit_prompt_to_compute_host(rid: str, sid: str, session: dict, text: Any) -> dict: + cfg = _load_dashboard_process_isolation_config() + frame = _compute_host_turn_frame(rid, sid, session, text) + + def _complete(done: dict) -> None: + # submit_turn reports a synchronous pipe failure through the callback + # before re-raising. Leave the parent session untouched so prompt.submit + # can fail open to the historical in-process path without emitting a + # duplicate terminal error. + if done.get("reason") == "send_failed": + return + _on_compute_host_turn_done(rid, sid, session, done) + + try: + _get_compute_host_supervisor(cfg).submit_turn(frame, on_complete=_complete) + except Exception as exc: + return _err(rid, 5019, f"compute-host dispatch failed: {exc}") + with session["history_lock"]: + session["_compute_host_active"] = True + session["attached_images"] = [] + return _ok(rid, {"status": "streaming", "turn_isolation": True}) + + +def _send_compute_host_control( + sid: str, + *, + route_name: str, + command: str = "", + payload: dict | None = None, + wait: bool = True, +) -> dict: + frame = dict(payload or {}) + frame.setdefault("type", "control") + frame.setdefault("command", command) + return _get_compute_host_supervisor().control( + sid, + route_name=route_name, + payload=frame, + wait=wait, + ) + + def _emit_approval_request(sid: str, data: dict | None) -> None: """Emit an ``approval.request`` event to the TUI client with the command redacted. The approval payload is built from the RAW command string, so a @@ -1920,6 +2094,48 @@ def _set_session_cwd(session: dict, cwd: str) -> str: _INDICATOR_STYLES: tuple[str, ...] = ("ascii", "emoji", "kaomoji", "unicode") _INDICATOR_DEFAULT = "kaomoji" +_DASHBOARD_TURN_ISOLATION_DEFAULT = False +_DASHBOARD_COMPUTE_HOST_HEARTBEAT_SECS_DEFAULT = 15 +_DASHBOARD_COMPUTE_HOST_RESPAWN_MAX_DEFAULT = 3 + + +def _coerce_int_config_value(value: Any, default: int, *, min_value: int) -> int: + try: + coerced = int(value) + except (TypeError, ValueError): + return default + return coerced if coerced >= min_value else default + + +def _load_dashboard_process_isolation_config(cfg: dict | None = None) -> dict[str, Any]: + """Return dashboard process-isolation config with read-site defaults. + + ``_load_cfg()`` intentionally returns raw ``config.yaml`` plus the managed + overlay; it does not deep-merge ``hermes_cli.config.DEFAULT_CONFIG``. Keep + the Phase-0 defaults here so dashboard runtime and the REST editor's + DEFAULT_CONFIG-backed schema cannot drift. + """ + root = _load_cfg() if cfg is None else cfg + dashboard = root.get("dashboard") if isinstance(root, dict) else {} + if not isinstance(dashboard, dict): + dashboard = {} + return { + "turn_isolation": is_truthy_value( + dashboard.get("turn_isolation"), + default=_DASHBOARD_TURN_ISOLATION_DEFAULT, + ), + "compute_host_heartbeat_secs": _coerce_int_config_value( + dashboard.get("compute_host_heartbeat_secs"), + _DASHBOARD_COMPUTE_HOST_HEARTBEAT_SECS_DEFAULT, + min_value=1, + ), + "compute_host_respawn_max": _coerce_int_config_value( + dashboard.get("compute_host_respawn_max"), + _DASHBOARD_COMPUTE_HOST_RESPAWN_MAX_DEFAULT, + min_value=0, + ), + } + def _load_cfg() -> dict: global _cfg_cache, _cfg_mtime, _cfg_path @@ -3372,12 +3588,23 @@ def _current_profile_name() -> str: DESKTOP_BACKEND_CONTRACT = 3 +def _session_usage_snapshot(session: dict | None) -> dict: + agent = (session or {}).get("agent") + mirror_usage = _metadata_mirror(session).get("usage") + if (session or {}).get("_compute_host_active") and isinstance(mirror_usage, dict): + return dict(mirror_usage) + if agent is not None: + return _get_usage(agent) + return dict(mirror_usage) if isinstance(mirror_usage, dict) else {} + + def _session_info(agent, session: dict | None = None) -> dict: if session is None: for candidate in _sessions.values(): if candidate.get("agent") is agent: session = candidate break + mirror = _metadata_mirror(session) cwd = _display_session_cwd(session) session_key = str( (session or {}).get("session_key") or getattr(agent, "session_id", "") or "" @@ -3395,7 +3622,7 @@ def _session_info(agent, session: dict | None = None) -> dict: reasoning_effort = "none" else: reasoning_effort = str(reasoning_config.get("effort", "") or "") - service_tier = getattr(agent, "service_tier", None) or "" + service_tier = getattr(agent, "service_tier", None) or mirror.get("service_tier") or "" # Effective approval-bypass state — the same three sources that # check_all_command_guards() ORs together: persistent config # (approvals.mode=off), the process-scoped --yolo env, and the @@ -3415,15 +3642,15 @@ def _session_info(agent, session: dict | None = None) -> dict: except Exception: yolo = False info: dict = { - "model": getattr(agent, "model", ""), - "provider": getattr(agent, "provider", ""), + "model": mirror.get("model", getattr(agent, "model", "")), + "provider": mirror.get("provider", getattr(agent, "provider", "")), "reasoning_effort": reasoning_effort, "service_tier": service_tier, "fast": service_tier == "priority", "yolo": yolo, "approval_mode": approval_mode, - "tools": {}, - "skills": {}, + "tools": dict(mirror.get("tools") or {}) if isinstance(mirror.get("tools"), dict) else {}, + "skills": dict(mirror.get("skills") or {}) if isinstance(mirror.get("skills"), dict) else {}, "cwd": cwd, "branch": _git_branch_for_cwd(cwd), "personality": str(personality or ""), @@ -3434,7 +3661,7 @@ def _session_info(agent, session: dict | None = None) -> dict: "release_date": "", "update_behind": None, "update_command": "", - "usage": _get_usage(agent), + "usage": _session_usage_snapshot(session), "profile_name": _current_profile_name(), } try: @@ -3456,22 +3683,24 @@ def _session_info(agent, session: dict | None = None) -> dict: info["release_date"] = __release_date__ except Exception: pass - try: - from model_tools import get_toolset_for_tool + if agent is not None and not (session or {}).get("_compute_host_active"): + try: + from model_tools import get_toolset_for_tool - for t in getattr(agent, "tools", []) or []: - name = t["function"]["name"] - info["tools"].setdefault(get_toolset_for_tool(name) or "other", []).append( - name - ) - except Exception: - pass - try: - from hermes_cli.banner import get_available_skills + info["tools"] = {} + for t in getattr(agent, "tools", []) or []: + name = t["function"]["name"] + info["tools"].setdefault(get_toolset_for_tool(name) or "other", []).append( + name + ) + except Exception: + pass + try: + from hermes_cli.banner import get_available_skills - info["skills"] = get_available_skills() - except Exception: - pass + info["skills"] = get_available_skills() + except Exception: + pass try: from tools.mcp_tool import get_mcp_status @@ -3479,7 +3708,11 @@ def _session_info(agent, session: dict | None = None) -> dict: except Exception: info["mcp_servers"] = [] try: - info["system_prompt"] = getattr(agent, "_cached_system_prompt", "") or "" + info["system_prompt"] = ( + mirror.get("system_prompt") + if "system_prompt" in mirror + else getattr(agent, "_cached_system_prompt", "") or "" + ) except Exception: pass try: @@ -3490,9 +3723,10 @@ def _session_info(agent, session: dict | None = None) -> dict: info["update_command"] = recommended_update_command() except Exception: pass - warn = _probe_credentials(agent) - if warn: - info["credential_warning"] = warn + if agent is not None and not (session or {}).get("_compute_host_active"): + warn = _probe_credentials(agent) + if warn: + info["credential_warning"] = warn return info @@ -3507,7 +3741,7 @@ def _tool_ctx(name: str, args: dict) -> str: def _emit_session_info_for_session(sid: str, session: dict) -> None: agent = session.get("agent") - if agent is None: + if agent is None and not _metadata_mirror(session): return try: _emit("session.info", sid, _session_info(agent, session)) @@ -4561,6 +4795,15 @@ def _make_agent( service_tier_override: str | None = None, platform_override: str | None = None, ): + # AC-4 test seam: dead unless explicitly armed by the isolated certify + # harness. Both inline and compute-host paths construct through _make_agent, + # leaving the process boundary as the only experimental variable. + from tui_gateway.synthetic_turn import maybe_build_synthetic_agent + + synthetic = maybe_build_synthetic_agent(session_id or key, model_override) + if synthetic is not None: + return synthetic + from run_agent import AIAgent # MCP tool discovery runs in a background daemon thread at startup so a @@ -5196,6 +5439,11 @@ def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any) agent.interrupt() except Exception: pass + elif mode != "queue" and _session_uses_compute_host(session): + try: + _get_compute_host_supervisor().interrupt(sid) + except Exception: + pass _enqueue_prompt(session, text, transport) session["last_active"] = time.time() return _ok(rid, {"status": "queued"}) @@ -5217,7 +5465,16 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool: if queued.get("transport") is not None: session["transport"] = queued["transport"] try: - _run_prompt_submit(rid, sid, session, queued["text"]) + if _session_uses_compute_host(session): + resp = _submit_prompt_to_compute_host(rid, sid, session, queued["text"]) + if resp.get("error"): + message = str(((resp.get("error") or {}).get("message")) or "queued prompt failed") + with session["history_lock"]: + session["running"] = False + _clear_inflight_turn(session) + _emit("error", sid, {"message": message}) + else: + _run_prompt_submit(rid, sid, session, queued["text"]) except Exception as exc: print( f"[tui_gateway] queued prompt dispatch failed: " @@ -6568,11 +6825,9 @@ def _(rid, params: dict) -> dict: if err: return err agent = session.get("agent") - usage: dict = ( - _get_usage(agent) - if agent is not None - else {"calls": 0, "input": 0, "output": 0, "total": 0} - ) + usage: dict = _session_usage_snapshot(session) + if agent is None and not usage: + usage = {"calls": 0, "input": 0, "output": 0, "total": 0} # Nous credits block — agent-independent (a portal fetch), so it shows even # with zero API calls or on a resumed session. The TUI /usage panel renders # these lines regardless of `calls`. Fail-open: [] when not logged into Nous @@ -6595,7 +6850,7 @@ def _(rid, params: dict) -> dict: return err agent = session.get("agent") if agent is None: - usage = _get_usage(None) + usage = _session_usage_snapshot(session) or _get_usage(None) return _ok( rid, { @@ -6603,8 +6858,8 @@ def _(rid, params: dict) -> dict: "context_max": usage.get("context_max", 0) or 0, "context_percent": usage.get("context_percent", 0) or 0, "context_used": usage.get("context_used", 0) or 0, - "estimated_total": 0, - "model": "", + "estimated_total": usage.get("context_used", 0) or usage.get("total", 0) or 0, + "model": _metadata_mirror(session).get("model", ""), }, ) with session["history_lock"]: @@ -7879,9 +8134,10 @@ def _(rid, params: dict) -> dict: updated = _dt(meta.get(field), created) break - usage = _get_usage(agent) if agent is not None else {} - provider = getattr(agent, "provider", None) or "unknown" - model = getattr(agent, "model", None) or "(unknown)" + mirror = _metadata_mirror(session) + usage = _session_usage_snapshot(session) + provider = getattr(agent, "provider", None) or mirror.get("provider") or "unknown" + model = getattr(agent, "model", None) or mirror.get("model") or "(unknown)" lines = [ "Hermes TUI Status", "", @@ -7956,6 +8212,26 @@ def _(rid, params: dict) -> dict: @method("session.compress") def _(rid, params: dict) -> dict: + session, err = _sess_nowait(params, rid) + if err: + return err + if _session_uses_compute_host(session): + sid = str(params.get("session_id") or "") + focus_topic = str(params.get("focus_topic", "") or "").strip() + command = "/compress" + (f" {focus_topic}" if focus_topic else "") + try: + ack = _send_compute_host_control( + sid, + route_name="session.compress", + command=command, + wait=True, + ) + except Exception as exc: + return _err(rid, 5019, f"compute-host compress failed: {exc}") + if ack.get("type") in {"control.error", "error"}: + return _err(rid, 4009, str(ack.get("message") or "compute-host compress failed")) + _apply_compute_host_metadata_mirror(session, ack) + return _ok(rid, {"status": "compressed", "turn_isolation": True, "host_ack": ack}) session, err = _sess(params, rid) if err: return err @@ -8207,6 +8483,27 @@ def _(rid, params: dict) -> dict: @method("session.interrupt") def _(rid, params: dict) -> dict: + session, err = _sess_nowait(params, rid) + if err: + return err + if _session_uses_compute_host(session): + sid = str(params.get("session_id") or "") + if session.get("running"): + try: + _get_compute_host_supervisor().interrupt(sid, request_id=f"interrupt-{rid}") + except Exception as exc: + return _err(rid, 5019, f"compute-host interrupt failed: {exc}") + with session["history_lock"]: + session["_turn_cancel_requested"] = True + session["queued_prompt"] = None + _clear_pending(sid) + try: + from tools.approval import resolve_gateway_approval + + resolve_gateway_approval(session["session_key"], "deny", resolve_all=True) + except Exception: + pass + return _ok(rid, {"status": "interrupted", "turn_isolation": True}) session, err = _sess(params, rid) if err: return err @@ -8522,6 +8819,8 @@ def _(rid, params: dict) -> dict: session, err = _sess_nowait(params, rid) if err: return err + isolation_cfg = _load_dashboard_process_isolation_config() + turn_isolation = _session_uses_compute_host(session, isolation_cfg) # Re-bind to the current client transport for this request. This keeps # streaming events on the active websocket even if an earlier disconnect # or fallback moved the session transport to stdio. @@ -8568,6 +8867,16 @@ def _(rid, params: dict) -> dict: session["last_active"] = time.time() _start_inflight_turn(session, text) + if turn_isolation: + isolated_response = _submit_prompt_to_compute_host(rid, sid, session, text) + if not isolated_response.get("error"): + return isolated_response + logger.warning( + "compute-host dispatch failed for session %s; falling back inline: %s", + sid, + isolated_response["error"].get("message", "unknown error"), + ) + # Persist the DB row lazily, now that the user has actually sent a message. _ensure_session_db_row(session) # A branch becomes real here: copy its parent's transcript into the row so it @@ -11703,6 +12012,16 @@ def _(rid, params: dict) -> dict: }, ) + if session and _session_uses_compute_host(session): + try: + ack = _get_compute_host_supervisor().reload_mcp( + str(params.get("session_id") or ""), + request_id=f"reload-mcp-{rid}", + ) + except Exception as exc: + return _err(rid, 5019, f"compute-host reload_mcp failed: {exc}") + return _ok(rid, {"status": "reloaded", "turn_isolation": True, "host_ack": ack}) + from tools.mcp_tool import shutdown_mcp_servers, discover_mcp_tools shutdown_mcp_servers() @@ -12449,6 +12768,28 @@ def _(rid, params: dict) -> dict: rid, 4009, "session busy — /interrupt the current turn before /compress" ) sid = params.get("session_id", "") + if _session_uses_compute_host(session): + command = f"/{name}" + (f" {arg}" if arg else "") + try: + ack = _send_compute_host_control( + sid, + route_name="slash.compress", + command=command, + wait=True, + ) + except Exception as exc: + return _err(rid, 5019, f"compute-host slash.compress failed: {exc}") + if ack.get("type") in {"control.error", "error"}: + return _err( + rid, + 4009, + str(ack.get("message") or "compute-host slash.compress failed"), + ) + _apply_compute_host_metadata_mirror(session, ack) + return _ok( + rid, + {"type": "exec", "output": str(ack.get("output") or "")}, + ) try: from agent.manual_compression_feedback import summarize_manual_compression from agent.model_metadata import estimate_request_tokens_rough @@ -13199,6 +13540,256 @@ def _(rid, params: dict) -> dict: # ── Methods: slash.exec ────────────────────────────────────────────── +_LIVE_SESSION_DIRECT_COMMANDS = frozenset( + { + "clear", + "compress", + "effort", + "history", + "models", + "prompt", + "rename", + "status", + "usage", + } +) + +_ISOLATED_SESSION_READ_COMMANDS = frozenset({"context", "tools", "help"}) + + +def _format_live_usage_output(session: dict) -> str: + agent = session.get("agent") + usage = _session_usage_snapshot(session) + if agent is None and not usage: + return "(._.) No active agent -- send a message first." + if session.get("_metadata_message_count") is not None: + message_count = int(session.get("_metadata_message_count") or 0) + else: + with session["history_lock"]: + message_count = len(session.get("history", [])) + lines = [ + "Session Token Usage", + "────────────────────────────────────────", + f"Model: {usage.get('model') or _metadata_mirror(session).get('model') or getattr(agent, 'model', '') or '(unknown)'}", + f"Input tokens: {int(usage.get('input') or 0):,}", + f"Output tokens: {int(usage.get('output') or 0):,}", + ] + reasoning = int(usage.get("reasoning") or 0) + if reasoning: + lines.append(f"Reasoning tokens: {reasoning:,}") + lines.extend( + [ + f"Prompt tokens: {int(usage.get('prompt') or 0):,}", + f"Completion tokens: {int(usage.get('completion') or 0):,}", + f"Total tokens: {int(usage.get('total') or 0):,}", + f"API calls: {int(usage.get('calls') or 0):,}", + ] + ) + if usage.get("context_max"): + lines.append( + "Current context: " + f"{int(usage.get('context_used') or 0):,} / " + f"{int(usage.get('context_max') or 0):,} " + f"({int(usage.get('context_percent') or 0)}%)" + ) + lines.extend( + [ + f"Messages: {message_count:,}", + f"Compressions: {int(usage.get('compressions') or 0):,}", + ] + ) + return "\n".join(lines) + + +def _format_live_history_output(session: dict) -> str: + with session["history_lock"]: + history = list(session.get("history", [])) + db = _get_db() + if db is not None and session.get("session_key"): + try: + history = db.get_messages_as_conversation( + session["session_key"], include_ancestors=True + ) + except Exception: + pass + messages = _history_to_messages(history) + if not messages: + return "No conversation history yet." + lines = ["Conversation History", "────────────────────────────────────────"] + for idx, message in enumerate(messages, start=1): + role = str(message.get("role") or "unknown") + label = "You" if role == "user" else "Hermes" if role == "assistant" else role.title() + text = str(message.get("text") or message.get("context") or "").strip() + if len(text) > 400: + text = f"{text[:400]}..." + lines.append(f"[{label} #{idx}] {text or '(no text)'}") + return "\n".join(lines) + + +def _format_live_prompt_output(session: dict) -> str: + agent = session.get("agent") + mirror = _metadata_mirror(session) + if agent is None and "system_prompt" not in mirror: + return "No active agent -- send a message first." + prompt = ( + mirror.get("system_prompt") + or getattr(agent, "ephemeral_system_prompt", None) + or getattr(agent, "_cached_system_prompt", None) + or "" + ) + if not prompt: + return "Current system prompt is not built yet; send a message first." + return f"Current system prompt:\n{prompt}" + + +def _format_live_context_output(session: dict) -> str: + messages = [] + db = _get_db() + if db is not None and session.get("session_key"): + try: + messages = _history_to_messages( + db.get_messages_as_conversation(session["session_key"], include_ancestors=True) + ) + except Exception: + messages = [] + if not messages: + with session["history_lock"]: + messages = _history_to_messages(list(session.get("history", []))) + usage = _session_usage_snapshot(session) + mirror = _metadata_mirror(session) + lines = [ + f"Conversation: {len(messages)} messages" if messages else "Conversation is empty (no messages yet)." + ] + roles: dict[str, int] = {} + for msg in messages: + role = str(msg.get("role") or "unknown") + roles[role] = roles.get(role, 0) + 1 + lines.append( + f" user: {roles.get('user', 0)}, assistant: {roles.get('assistant', 0)}, " + f"tool: {roles.get('tool', 0)}, system: {roles.get('system', 0)}" + ) + model = mirror.get("model") or usage.get("model") or "" + provider = mirror.get("provider") or "auto" + if model: + lines.append(f"Model: {model}") + lines.append(f"Provider: {provider}") + context_used = int(usage.get("context_used") or usage.get("total") or 0) + context_max = int(usage.get("context_max") or 0) + if context_used: + if context_max: + usage_pct = (context_used / context_max) * 100 + lines.append( + f"Context usage: ~{context_used:,} / {context_max:,} tokens ({usage_pct:.1f}%)" + ) + else: + lines.append(f"Context usage: ~{context_used:,} tokens") + if usage.get("compressions"): + lines.append(f"Compressions: {int(usage.get('compressions') or 0):,}") + return "\n".join(lines) + + +def _format_live_tools_output(session: dict) -> str: + info = _session_info(session.get("agent"), session) + groups = info.get("tools") if isinstance(info, dict) else {} + if not isinstance(groups, dict) or not groups: + return "No tools available." + names: list[str] = [] + for group_names in groups.values(): + if isinstance(group_names, list): + names.extend(str(name) for name in group_names) + names = sorted(set(names)) + if not names: + return "No tools available." + return "Available tools ({}):\n{}".format( + len(names), "\n".join(f" {name}" for name in names) + ) + + +def _format_live_help_output() -> str: + try: + from hermes_cli.commands import COMMANDS_BY_CATEGORY + + lines = ["Available commands:", ""] + for category, commands in COMMANDS_BY_CATEGORY.items(): + lines.append(f"{category}:") + for cmd, desc in commands.items(): + lines.append(f" {cmd:<15} {desc}") + return "\n".join(lines) + except Exception as exc: + return f"help unavailable: {exc}" + + +def _format_live_model_output(session: dict) -> str: + agent = session.get("agent") + model = getattr(agent, "model", "") if agent is not None else "" + provider = getattr(agent, "provider", "") if agent is not None else "" + if model and provider: + return f"Current model: {model} ({provider})" + if model: + return f"Current model: {model}" + return "Current model: (unknown)" + + +def _live_slash_command_output(sid: str, session: Optional[dict], name: str, arg: str) -> Optional[str]: + name = (name or "").lstrip("/").lower() + arg = arg or "" + if name == "model" and not arg.strip(): + return _format_live_model_output(session or {}) + if name not in _LIVE_SESSION_DIRECT_COMMANDS: + if not ( + name in _ISOLATED_SESSION_READ_COMMANDS + and session is not None + and _session_uses_compute_host(session) + ): + return None + + if name in _ISOLATED_SESSION_READ_COMMANDS and not ( + session is not None and _session_uses_compute_host(session) + ): + return None + if name == "compress": + if session is None: + return "no active session for /compress" + return _mirror_slash_side_effects(sid, session, f"/compress {arg}".strip()) + if name == "usage": + if session is None: + return "(._.) No active agent -- send a message first." + return _format_live_usage_output(session) + if name == "history": + if session is None: + return "No conversation history yet." + return _format_live_history_output(session) + if name == "prompt": + if session is None: + return "No active agent -- send a message first." + return _format_live_prompt_output(session) + if name == "status": + response = _methods["session.status"]("status", {"session_id": sid}) + if response.get("error"): + return str(response["error"].get("message") or "status unavailable") + return str(response.get("result", {}).get("output") or "") + if name == "context": + if session is None: + return "Conversation is empty (no messages yet)." + return _format_live_context_output(session) + if name == "tools": + if session is None: + return "No tools available." + return _format_live_tools_output(session) + if name == "help": + return _format_live_help_output() + if name == "clear": + return "Screen clear is terminal-only; desktop/TUI chat left unchanged." + if name == "models": + return "Use /model to view or switch the current model; desktop users can also open the model picker." + if name == "rename": + return "Use /title to rename this session." + if name == "effort": + return "Use /reasoning to change reasoning effort." + return None + + + def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: """Apply side effects that must also hit the gateway's live agent.""" parts = command.lstrip("/").split(None, 1) @@ -13216,6 +13807,21 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: # with the session.compress / session.undo guards and the gateway # runner's running-agent /model guard. _MUTATES_WHILE_RUNNING = {"model", "personality", "prompt", "compress"} + if _session_uses_compute_host(session) and name in _MUTATES_WHILE_RUNNING: + route_name = f"slash.{name}" + try: + ack = _send_compute_host_control( + sid, + route_name=route_name, + command=command, + wait=True, + ) + except Exception as exc: + return f"compute-host {route_name} failed: {exc}" + if ack.get("type") in {"control.error", "error"}: + return str(ack.get("message") or f"compute-host {route_name} failed") + _apply_compute_host_metadata_mirror(session, ack) + return str(ack.get("output") or "") if name in _MUTATES_WHILE_RUNNING and session.get("running"): return f"session busy — /interrupt the current turn before running /{name}" @@ -13299,7 +13905,7 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: @method("slash.exec") def _(rid, params: dict) -> dict: - session, err = _sess(params, rid) + session, err = _sess_nowait(params, rid) if err: return err @@ -13317,6 +13923,12 @@ def _(rid, params: dict) -> dict: _cmd_base = (_cmd_parts[0] if _cmd_parts else "").lower() _cmd_arg = _cmd_parts[1] if len(_cmd_parts) > 1 else "" + live_output = _live_slash_command_output( + params.get("session_id", ""), session, _cmd_base, _cmd_arg + ) + if live_output is not None: + return _ok(rid, {"output": live_output or "(no output)"}) + if _cmd_base in _PENDING_INPUT_COMMANDS: # Route directly to command.dispatch instead of returning an error # that requires the frontend to retry. Some TUI clients fail the diff --git a/tui_gateway/synthetic_turn.py b/tui_gateway/synthetic_turn.py new file mode 100644 index 000000000..17a8a913f --- /dev/null +++ b/tui_gateway/synthetic_turn.py @@ -0,0 +1,231 @@ +"""Synthetic GIL-heavy turn driver for the AC-4 isolation certify harness. + +Mechanism B (the class ``docs/desktop/2026-07-04-dashboard-process-isolation-PRD.md`` +targets) is interpreter-wide GIL starvation: concurrent heavy agent turns run +compute in threads of the SERVING process, and CPython's single GIL lets those +threads starve the event loop that flushes WebSocket frames for MINUTES. A +2026-07-04 ``sample`` showed the loop thread parked in ``take_gil`` while worker +threads burned the interpreter — NOT blocked on I/O. + +To certify the fix (AC-4) without spending real tokens on 6 concurrent 100K+ +context model calls, the harness needs a turn driver that reproduces THAT +regime: sustained pure-Python CPU that holds the GIL for the turn's duration. +A network/sleep stub is WRONG here — it would release the GIL during I/O and +never reproduce ``take_gil`` contention, so a dry-run green off it is a fake +green (the spec says so explicitly). + +This module is a **test seam**: it is dead unless ``HERMES_ISO_CERTIFY_SYNTH_TURN`` +is set. When armed, ``tui_gateway.server._make_agent`` returns a +:class:`SyntheticHeavyAgent` instead of a real ``AIAgent``. Because both the +in-process ``_pool`` path (isolation OFF) and the compute-host child path +(isolation ON) build their agent through ``_make_agent``, the SAME synthetic +turn exercises whichever dispatch path is under test — the isolation boundary +is the only variable between an OFF run and an ON run. + +The per-turn intensity (wall duration, CPU chunk size, streamed-delta cadence, +token accounting) is carried in the prompt text as a small JSON spec so the +harness has full control and the server seam stays dumb. Any prompt that is not +a JSON object falls back to env / built-in defaults. +""" + +from __future__ import annotations + +import json +import os +import threading +import time +from typing import Any, Callable, Optional + + +def synth_turn_armed() -> bool: + """True when the synthetic-turn test seam is armed via env.""" + return os.environ.get("HERMES_ISO_CERTIFY_SYNTH_TURN") == "1" + + +def _env_float(name: str, default: float) -> float: + try: + return float(os.environ.get(name, "") or default) + except (TypeError, ValueError): + return default + + +def _env_int(name: str, default: int) -> int: + try: + return int(os.environ.get(name, "") or default) + except (TypeError, ValueError): + return default + + +class SyntheticHeavyAgent: + """An AIAgent-shaped object whose turn is a GIL-holding CPU burn. + + Presents only the surface ``tui_gateway.server``'s turn path and status + helpers read: ``run_conversation``/``interrupt``/``clear_interrupt`` plus a + handful of ``model``/``provider``/``session_*`` attributes consumed by + ``_get_usage`` and ``_session_info``. It never opens a socket or spawns a + subprocess, so the only work it does is the deterministic Python loop below — + exactly the ``take_gil`` regime under test. + """ + + def __init__(self, session_id: str, *, model: str = "synthetic-heavy") -> None: + self.session_id = session_id + self.model = model + self.provider = "synthetic" + self.api_mode = "chat_completions" + self.base_url = "" + self.api_key = "" + self.platform = "" + self.tools: list[Any] = [] + self.reasoning_config: dict | None = None + self.service_tier: str | None = None + self.context_compressor = None + self._config_context_length = 200_000 + self._cached_system_prompt = "" + # Cumulative session counters (read by _get_usage → status bar). + self.session_input_tokens = 0 + self.session_output_tokens = 0 + self.session_prompt_tokens = 0 + self.session_completion_tokens = 0 + self.session_reasoning_tokens = 0 + self.session_total_tokens = 0 + self.session_api_calls = 0 + self.history: list[dict[str, str]] = [] + self._interrupt = threading.Event() + + # ── interrupt contract (mirrors AIAgent) ─────────────────────────── + def clear_interrupt(self) -> None: + self._interrupt.clear() + + def interrupt(self) -> None: + self._interrupt.set() + + def _has_stream_consumers(self) -> bool: # defensive; not used by our loop + return True + + def close(self) -> None: + """No-op teardown (session lifecycle calls agent.close() on some paths).""" + self._interrupt.set() + + # ── spec parsing ─────────────────────────────────────────────────── + @staticmethod + def _parse_spec(message: Any) -> dict[str, Any]: + spec: dict[str, Any] = {} + if isinstance(message, str): + text = message.strip() + if text.startswith("{"): + try: + parsed = json.loads(text) + if isinstance(parsed, dict): + spec = parsed + except (ValueError, TypeError): + spec = {} + return { + # Primary control: wall-clock seconds of GIL-holding compute. + "duration_s": float(spec.get("duration_s", _env_float("HERMES_ISO_CERTIFY_DURATION_S", 8.0))), + # Pure-Python integer ops per interrupt-check chunk. Small enough + # that an interrupt is honored within a few ms; large enough that + # the loop stays hot on the GIL between checks. + "chunk": int(spec.get("chunk", _env_int("HERMES_ISO_CERTIFY_CHUNK", 20_000))), + # Streamed-delta cadence (seconds). Each delta is a loop wakeup that + # marshals a frame across the transport — the serving-path pressure. + "delta_interval_s": float(spec.get("delta_interval_s", _env_float("HERMES_ISO_CERTIFY_DELTA_S", 0.05))), + # Notional output tokens attributed per streamed delta (drives the + # 100K+-token "heavy turn" proxy in usage/metadata). + "tokens_per_delta": int(spec.get("tokens_per_delta", _env_int("HERMES_ISO_CERTIFY_TPD", 512))), + # Optional per-chunk sleep to model a lighter/mixed regime (0 = pure + # burn). --dry-run uses a short duration, NOT a sleep, so the smoke + # path still exercises the real dispatch seam. + "sleep_s": float(spec.get("sleep_s", 0.0)), + } + + # ── the turn ─────────────────────────────────────────────────────── + def run_conversation( + self, + message: Any, + *, + conversation_history: Optional[list[dict[str, str]]] = None, + stream_callback: Optional[Callable[[str], None]] = None, + task_id: Optional[str] = None, + **_kwargs: Any, + ) -> dict[str, Any]: + spec = self._parse_spec(message) + duration = max(0.0, spec["duration_s"]) + chunk = max(1, spec["chunk"]) + interval = max(0.001, spec["delta_interval_s"]) + tokens_per_delta = max(0, spec["tokens_per_delta"]) + sleep_s = max(0.0, spec["sleep_s"]) + + base_history = list(conversation_history if conversation_history is not None else self.history) + start = time.monotonic() + last_delta = start + acc = 0 + deltas = 0 + interrupted = False + + while True: + if self._interrupt.is_set(): + interrupted = True + break + now = time.monotonic() + if now - start >= duration: + break + # GIL-holding pure-Python work. A tight integer loop runs one + # bytecode step per iteration and NEVER releases the GIL — this is + # the exact interpreter contention that starves the serving loop. + for _ in range(chunk): + acc = (acc * 1_000_003 + 12_345) & 0xFFFFFFFFFFFFFFFF + if sleep_s: + time.sleep(sleep_s) + if now - last_delta >= interval: + deltas += 1 + self.session_output_tokens += tokens_per_delta + self.session_completion_tokens += tokens_per_delta + self.session_total_tokens += tokens_per_delta + if stream_callback is not None: + stream_callback(f"synthtok-{deltas:05d} ") + last_delta = now + + self.session_api_calls += 1 + # Fold the checksum into the reply so the loop is not dead-code-eliminated + # and the turn produces a deterministic, inspectable result. + final = ( + f"[synthetic heavy turn] deltas={deltas} " + f"out_tokens={self.session_output_tokens} " + f"interrupted={interrupted} checksum={acc & 0xFFFF:04x}" + ) + messages = [ + *base_history, + {"role": "user", "content": str(message)[:200]}, + {"role": "assistant", "content": final}, + ] + self.history = messages + return { + "final_response": final, + "messages": messages, + "interrupted": interrupted, + "error": None, + "last_reasoning": None, + } + + +def maybe_build_synthetic_agent(session_id: str, model_override: Any = None) -> SyntheticHeavyAgent | None: + """Return a :class:`SyntheticHeavyAgent` when the seam is armed, else ``None``. + + ``model_override`` (dict or str) only influences the reported ``model`` label + so status frames look plausible; it never changes the compute. + """ + if not synth_turn_armed(): + return None + model = "synthetic-heavy" + if isinstance(model_override, dict) and model_override.get("model"): + model = str(model_override["model"]) + elif isinstance(model_override, str) and model_override: + model = model_override + return SyntheticHeavyAgent(session_id, model=model) + + +__all__ = [ + "SyntheticHeavyAgent", + "maybe_build_synthetic_agent", + "synth_turn_armed", +]