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>
This commit is contained in:
brooklyn! 2026-07-16 15:24:03 -04:00 committed by GitHub
parent 91ed8e4a99
commit 7d27a31ce7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 3757 additions and 39 deletions

View file

@ -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

625
scripts/iso-certify.py Executable file
View file

@ -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())

View file

@ -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

View file

@ -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()

View file

@ -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()

View file

@ -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"}

681
tui_gateway/compute_host.py Normal file
View file

@ -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())

View file

@ -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",
]

View file

@ -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 <name> to rename this session."
if name == "effort":
return "Use /reasoning <effort> 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

View file

@ -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",
]