be59f0411e
Review finding (MAJOR-2): with dashboard.turn_isolation the plugin's agent-side code (tools, hooks) runs in the compute-host child, where _live_transports is empty and _broadcast_global_event dropped the frame at logger.debug. The child now registers its _HostTransport as a live transport for the life of run_host, so a session-less broadcast rides the existing host pipe; the parent bridge (_relay_compute_host_rpc) recognises an event frame with no session_id and fans it out via _broadcast_global_event instead of write_json, which would have dropped it on hermes serve's stdio. Minor: the late `from tui_gateway.server import …` could raise ImportError in a plugin-only process despite the "safe from any handler" docstring — caught and logged as a warning. Docs: the SDK page now tables exactly which process each call site runs in and what it reaches (hermes serve / compute-host child / stdio TUI / gateway run + chat + cron = nobody).
346 lines
18 KiB
Python
346 lines
18 KiB
Python
"""Compute-host (turn isolation) bridge: relay prompts/controls to the child process and
|
|
mirror its metadata/clarify/compress acks back into the session. Bodies are rebound onto
|
|
server.py's globals at install time (method_ctx.bind_module), so they use them bare."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import threading
|
|
|
|
from .method_ctx import HandlerRegistry, bind_module
|
|
|
|
_registry = HandlerRegistry()
|
|
|
|
_compute_host_supervisor = None
|
|
_compute_host_supervisor_lock = threading.Lock()
|
|
# Cap on how long session.compress blocks its RPC on the compute host. Must stay
|
|
# below the desktop's SESSION_COMPRESS_TIMEOUT_MS (660s) so the client gets the
|
|
# `pending` answer, not its own timeout; the late-ack path covers anything slower.
|
|
# See #97948.
|
|
_COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS = 630.0
|
|
|
|
|
|
def _turn_isolation_enabled(cfg: dict | None = None) -> bool:
|
|
if os.environ.get("HERMES_COMPUTE_HOST_CHILD") == "1":
|
|
return False
|
|
return bool((cfg or _load_dashboard_process_isolation_config()).get("turn_isolation"))
|
|
|
|
|
|
def _session_uses_compute_host(session: dict, cfg: dict | None = None) -> bool:
|
|
# Routes lazy sessions whose AIAgent was never built in-process; already-built
|
|
# sessions keep the in-process path unless a prior isolated turn marked host ownership.
|
|
return _turn_isolation_enabled(cfg) and (
|
|
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=_relay_compute_host_rpc,
|
|
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, image_paths: list[str] | None = None,
|
|
queued_prompt_generation: int | None = None, display_kind: str | None = None,
|
|
display_metadata: dict | None = None) -> dict:
|
|
with session["history_lock"]:
|
|
history = list(session.get("history", []))
|
|
history_version = int(session.get("history_version", 0))
|
|
attached_images = list(image_paths if image_paths is not None else session.get("attached_images", []))
|
|
return {
|
|
"type": "turn.start", "sid": sid, "request_id": rid,
|
|
"session_key": session.get("session_key") or sid, "text": text,
|
|
**({"display_kind": display_kind} if display_kind else {}), "history": history,
|
|
**({"display_metadata": display_metadata} if display_metadata else {}),
|
|
"history_version": history_version, "cols": int(session.get("cols", 80) or 80),
|
|
"cwd": _session_cwd(session),
|
|
"context_cwd_is_launch_artifact": _context_cwd_is_launch_artifact(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,
|
|
"auth_user_id": _session_auth_user_id(session),
|
|
"queued_prompt_generation": queued_prompt_generation,
|
|
# #101416: vouch that this process already holds the registry lease for this session, so
|
|
# the child adopts it as an inert token instead of re-claiming and being fenced out by
|
|
# our own entry ("Session ... already has a live owner"). No lease held = no vouch, and
|
|
# the child keeps its legacy self-claim path (fail-closed refusal on conflict).
|
|
"active_session_lease": _active_session_lease_vouch(session)}
|
|
|
|
|
|
def _active_session_lease_vouch(session: dict) -> dict | None:
|
|
"""``{lease_id, session_id}`` of the REAL registry lease this process holds for the session's
|
|
current stored id, else None. Qualified, not a bare bool: after a compression rotation a lease
|
|
still keyed on the old id must not let a (replacement) child borrow the continuation."""
|
|
lease = session.get("active_session_lease")
|
|
if lease is None or getattr(lease, "released", False) or not getattr(lease, "enabled", False):
|
|
return None
|
|
if str(lease.session_id) != str(session.get("session_key") or ""):
|
|
return None
|
|
return {"lease_id": str(lease.lease_id), "session_id": str(lease.session_id)}
|
|
|
|
|
|
def _metadata_mirror(session: dict | None) -> dict:
|
|
mirror = (session or {}).get("_metadata_mirror")
|
|
return mirror if isinstance(mirror, dict) else {}
|
|
|
|
|
|
def _compute_host_session_info(session: dict) -> dict:
|
|
return _session_info(session.get("agent"), session)
|
|
|
|
|
|
def _compute_host_adopt_frame_meta(session: dict, frame: dict) -> None:
|
|
"""Adopt a host frame's session_key / history_version. Caller holds history_lock.
|
|
|
|
A rotated ``session_key`` means the child compressed A->B. The child only holds an inert
|
|
borrowed token (``_install_borrowed_lease``), so the REAL registry lease is re-anchored here,
|
|
by its owner — never claimed from the child pid (authority stays singular, #103737 review)."""
|
|
new_key = str(frame.get("session_key") or "")
|
|
if new_key and new_key != str(session.get("session_key") or ""):
|
|
if not _transfer_active_session_slot(str(frame.get("sid") or ""), session, new_session_id=new_key):
|
|
logger.warning("Compression session lease did not re-anchor: sid=%s old_session_id=%s new_session_id=%s",
|
|
frame.get("sid"), session.get("session_key"), new_key)
|
|
session["session_key"] = new_key
|
|
if frame.get("history_version") is not None:
|
|
with contextlib.suppress(Exception):
|
|
session["history_version"] = max(int(session.get("history_version", 0)),
|
|
int(frame.get("history_version") or 0))
|
|
|
|
|
|
def _relay_compute_host_rpc(message: dict) -> bool:
|
|
"""Relay host frames to the client while mirroring the server→client request the host has open, so a
|
|
reconnecting client gets it back through ``open_requests``."""
|
|
params = message.get("params") if isinstance(message, dict) else None
|
|
if isinstance(message, dict) and message.get("method") == "compute_host.activity":
|
|
if isinstance(params, dict):
|
|
session = _sessions.get(str(params.get("session_id") or ""))
|
|
if session is not None:
|
|
with _history_lock(session):
|
|
if (session.get("running") and params.get("turn_id")
|
|
and session.get("_compute_host_turn_id") == params["turn_id"]):
|
|
session["_compute_host_activity_ns"] = params.get("activity_ns")
|
|
return True # Internal observation, not a client event or replay entry.
|
|
if (isinstance(message, dict) and message.get("method") == "event" and isinstance(params, dict)
|
|
and not params.get("session_id")):
|
|
# A session-less (global) event the child could not deliver itself: ``write_json`` would drop it
|
|
# on this process's stdio; fan it out to every connected client like a local broadcast.
|
|
_broadcast_global_event(str(params.get("type") or ""), params.get("payload"))
|
|
return True
|
|
if isinstance(message, dict) and isinstance(message.get("id"), str) and message.get("method") not in (None, "event"):
|
|
# A server request minted by the child: remember it against its session until it is answered/withdrawn.
|
|
session = _sessions.get(str((params or {}).get("session_id") or "")) if isinstance(params, dict) else None
|
|
if session is not None:
|
|
with _history_lock(session):
|
|
session["_compute_host_open_request"] = {
|
|
"id": message["id"], "method": message["method"], "params": dict(params)}
|
|
elif isinstance(params, dict) and params.get("type") == "request.cancel":
|
|
session = _sessions.get(str(params.get("session_id") or ""))
|
|
payload = params.get("payload")
|
|
if session is not None and isinstance(payload, dict):
|
|
with _history_lock(session):
|
|
if _open_request_matches(session, payload.get("id")):
|
|
session.pop("_compute_host_open_request", None)
|
|
return write_json(message)
|
|
|
|
|
|
def _history_lock(session: dict):
|
|
return session.get("history_lock", threading.Lock())
|
|
|
|
|
|
def _open_request_matches(session: dict, request_id) -> bool:
|
|
"""Whether ``session``'s mirrored open request is ``request_id``. Caller holds history_lock."""
|
|
mirrored = session.get("_compute_host_open_request")
|
|
return isinstance(mirrored, dict) and mirrored.get("id") == request_id
|
|
|
|
|
|
def _compute_host_request_session(request_id: str) -> tuple[str, dict] | None:
|
|
"""Find the parent mirror for one host-owned server request."""
|
|
for sid, session in list(_sessions.items()) if request_id else ():
|
|
with _history_lock(session):
|
|
if _open_request_matches(session, request_id):
|
|
return sid, session
|
|
return None
|
|
|
|
|
|
def _relay_compute_host_response(frame: dict) -> bool:
|
|
"""Forward a client's response frame to the compute-host child that owns the request. False when no
|
|
child owns that id."""
|
|
located = _compute_host_request_session(str(frame.get("id") or ""))
|
|
if located is None or not _session_uses_compute_host(located[1]):
|
|
return False
|
|
sid, session = located
|
|
with _history_lock(session):
|
|
session.pop("_compute_host_open_request", None)
|
|
try:
|
|
_get_compute_host_supervisor().respond(sid, {"frame": dict(frame)})
|
|
except Exception:
|
|
logger.debug("compute-host response relay failed sid=%s", sid, exc_info=True)
|
|
return True
|
|
|
|
|
|
def _lock_compute_host_clarify(rid: str, request_id: str, question_id: str, answer: str) -> dict | None:
|
|
"""Proxy a batch-clarify lock into the child that owns the request; keeps the parent mirror's locked
|
|
answers current for reconnect snapshots. None when the request is not host-owned."""
|
|
located = _compute_host_request_session(request_id)
|
|
if located is None or not _session_uses_compute_host(located[1]):
|
|
return None
|
|
sid, session = located
|
|
try:
|
|
ack = _get_compute_host_supervisor().respond(
|
|
sid, {"lock": {"request_id": request_id, "question_id": question_id, "answer": answer}})
|
|
except Exception as exc:
|
|
return _err(rid, 5019, f"compute-host clarify lock failed: {exc}")
|
|
if ack.get("type") == "respond.error":
|
|
return _err(rid, 5019, str(ack.get("message") or "compute-host clarify lock failed"))
|
|
response = ack.get("response")
|
|
if not isinstance(response, dict):
|
|
return _err(rid, 5019, "compute-host clarify lock returned an invalid response")
|
|
if "error" in response:
|
|
error = response["error"] if isinstance(response["error"], dict) else {}
|
|
return _err(rid, int(error.get("code") or 5000), str(error.get("message") or "clarify lock failed"))
|
|
result = response.get("result") if isinstance(response.get("result"), dict) else {}
|
|
with _history_lock(session):
|
|
if _open_request_matches(session, request_id):
|
|
mirrored = session["_compute_host_open_request"]
|
|
if result.get("status") == "expired" or result.get("remaining") == []:
|
|
session.pop("_compute_host_open_request", None)
|
|
else:
|
|
mirrored["params"]["answers"] = {**(mirrored["params"].get("answers") or {}), question_id: answer}
|
|
return _ok(rid, result)
|
|
|
|
|
|
def _apply_compute_host_metadata_mirror(session: dict, frame: dict | None) -> None:
|
|
"""Mirror host-owned session metadata: under turn isolation the host is the only
|
|
writer of live agent/history state, and UI reads must not build a second agent."""
|
|
if not isinstance(frame, dict):
|
|
return
|
|
with _history_lock(session):
|
|
_compute_host_adopt_frame_meta(session, frame)
|
|
if frame.get("message_count") is not None:
|
|
with contextlib.suppress(Exception):
|
|
session["_metadata_message_count"] = int(frame.get("message_count") or 0)
|
|
info = frame.get("session_info")
|
|
if isinstance(info, dict):
|
|
session["_metadata_mirror"] = {**_metadata_mirror(session), **info}
|
|
session["_metadata_mirror_updated_at"] = time.time()
|
|
|
|
|
|
def _on_compute_host_turn_done(rid: str, sid: str, session: dict, frame: dict) -> None:
|
|
with session["history_lock"]:
|
|
_compute_host_adopt_frame_meta(session, frame)
|
|
session["running"] = False
|
|
session["last_active"] = time.time()
|
|
_clear_inflight_turn(session)
|
|
session.pop("_compute_host_open_request", None)
|
|
if frame.get("type") == "turn.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)
|
|
# Settlement of a turn whose session was closed mid-flight: the real lease was held for it.
|
|
_release_deferred_active_session_lease(session)
|
|
info = _compute_host_session_info(session)
|
|
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, image_paths: list[str] | None = None,
|
|
queued_prompt_generation: int | None = None, display_kind: str | None = None,
|
|
display_metadata: dict | None = None) -> dict:
|
|
cfg = _load_dashboard_process_isolation_config()
|
|
frame = _compute_host_turn_frame(rid, sid, session, text, image_paths=image_paths,
|
|
queued_prompt_generation=queued_prompt_generation,
|
|
display_kind=display_kind, display_metadata=display_metadata)
|
|
# Caller JSON-RPC ids may repeat across sockets and turns. Use an opaque
|
|
# dispatch lifetime token, installed before a fast child can send activity.
|
|
turn_id = frame["turn_id"] = frame["request_id"] = uuid.uuid4().hex
|
|
with session["history_lock"]:
|
|
session["_compute_host_turn_id"] = turn_id
|
|
session.pop("_compute_host_activity_ns", None)
|
|
|
|
def _complete(done: dict) -> None:
|
|
# submit_turn reports a synchronous pipe failure via the callback before re-raising;
|
|
# leave the session untouched so prompt.submit can fail open to the in-process path.
|
|
if done.get("reason") != "send_failed":
|
|
with session["history_lock"]:
|
|
if session.get("_compute_host_turn_id") != turn_id:
|
|
return
|
|
session.pop("_compute_host_turn_id", None)
|
|
session.pop("_compute_host_activity_ns", None)
|
|
_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:
|
|
with session["history_lock"]:
|
|
if session.get("_compute_host_turn_id") == turn_id:
|
|
session.pop("_compute_host_turn_id", None)
|
|
session.pop("_compute_host_activity_ns", None)
|
|
return _err(rid, 5019, f"compute-host dispatch failed: {exc}")
|
|
with session["history_lock"]:
|
|
session["_compute_host_active"] = True
|
|
if image_paths is None:
|
|
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, timeout: float = 30.0, on_late_ack=None) -> 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, timeout=timeout,
|
|
on_late_ack=on_late_ack)
|
|
|
|
|
|
def _compute_host_compress_wait_seconds(cfg: dict | None = None) -> float:
|
|
"""RPC wait budget for a compute-host compress control: the configured compression
|
|
ceiling plus slack, capped below the desktop's RPC timeout (a fixed waiter reported
|
|
false timeouts while the host kept working); slower acks land via the late-ack path.
|
|
|
|
See #97948.
|
|
"""
|
|
from agent.conversation_compression import resolve_context_compression_timeouts
|
|
try:
|
|
compression_cfg = (cfg if cfg is not None else _load_cfg()).get("compression", {})
|
|
except Exception:
|
|
compression_cfg = {}
|
|
if not isinstance(compression_cfg, dict):
|
|
compression_cfg = {}
|
|
_idle, ceiling = resolve_context_compression_timeouts(compression_cfg)
|
|
return float(min(max(ceiling + 30.0, 120.0), _COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS))
|
|
|
|
|
|
def _adopt_late_compute_host_compress_ack(sid: str, session: dict, ack: dict, *, route_name: str) -> None:
|
|
"""Adopt a compress ack that arrived after its RPC waiter answered ``pending``: the only place
|
|
the rotated session_key / history_version / mirror can land and the client's only signal
|
|
(the same ``session.info`` + ``compacted`` edges the in-process /compress path emits). A late
|
|
``control.error`` goes out via ``error``."""
|
|
with _sessions_lock:
|
|
if _sessions.get(sid) is not session:
|
|
return
|
|
if not isinstance(ack, dict) or ack.get("type") in {"control.error", "error"}:
|
|
message = str((ack or {}).get("message") or f"compute-host {route_name} failed")
|
|
_emit("error", sid, {"message": f"compression failed: {message}"})
|
|
_status_update(sid, "ready")
|
|
return
|
|
_apply_compute_host_metadata_mirror(session, ack)
|
|
_emit("session.info", sid, _compute_host_session_info(session))
|
|
_status_update(sid, "compacted", "✓ Context compression complete")
|
|
|
|
|
|
def register(server) -> None:
|
|
"""Publish this module's helpers + handlers onto ``server``, rebound to its globals."""
|
|
bind_module(globals(), server, skip=("_",))
|