Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 129 additions & 0 deletions tests/test_tui_gateway_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
from pathlib import Path
from unittest.mock import patch

import pytest

from hermes_constants import reset_hermes_home_override, set_hermes_home_override
from hermes_cli.active_sessions import active_session_registry_snapshot
from tui_gateway import server
Expand Down Expand Up @@ -2002,6 +2004,60 @@ def test_notification_event_routing_by_session_key(monkeypatch):
assert server._notification_event_belongs_elsewhere(mine, {"session_key": "ghost"}) is False


def test_prompt_submit_rejects_negative_truncate_ordinal(monkeypatch):
"""A negative truncate_before_user_ordinal must be rejected, not honoured.

The handler validates the upper bound (`ordinal >= len(user_indices)`) but a
negative ordinal would otherwise slip through and hit Python negative
indexing: `user_indices[-1]` selects the LAST user turn, truncating history
to everything before it and persisting that loss via replace_messages — an
unrecoverable overwrite of the session DB. Reject it on the safe 4018 path
and leave the in-memory history and the DB untouched.
"""
replaced = []

class _FakeDB:
def replace_messages(self, key, messages):
replaced.append((key, list(messages)))

history = [
{"role": "user", "content": "first"},
{"role": "assistant", "content": "ok"},
{"role": "user", "content": "second"},
{"role": "assistant", "content": "done"},
]
server._sessions["trunc-sid"] = _session(history=list(history))
monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
# If the guard ever lets a negative ordinal through, these would run and the
# session would be marked busy; failing here makes that regression loud.
monkeypatch.setattr(
server, "_start_agent_build", lambda *a, **k: pytest.fail("must not start a turn")
)
monkeypatch.setattr(
server, "_start_inflight_turn", lambda *a, **k: pytest.fail("must not start a turn")
)

try:
resp = server.handle_request(
{
"id": "1",
"method": "prompt.submit",
"params": {
"session_id": "trunc-sid",
"text": "next",
"truncate_before_user_ordinal": -1,
},
}
)
assert resp["error"]["code"] == 4018
# History and the DB are left exactly as they were — no silent loss.
assert server._sessions["trunc-sid"]["history"] == history
assert server._sessions["trunc-sid"]["running"] is False
assert replaced == []
finally:
server._sessions.pop("trunc-sid", None)


def test_session_create_does_not_persist_empty_row(monkeypatch):
"""session.create must NOT eagerly write a DB row.

Expand Down Expand Up @@ -4186,6 +4242,21 @@ def test_commands_catalog_includes_tui_mouse_command():
assert "/mouse" in tui_pairs


def test_commands_catalog_does_not_list_alias_collisions_as_pairs():
resp = server.handle_request(
{"id": "1", "method": "commands.catalog", "params": {}}
)

pairs = dict(resp["result"]["pairs"])
tui_cat = next(c for c in resp["result"]["categories"] if c["name"] == "TUI")
tui_pairs = dict(tui_cat["pairs"])
canon = resp["result"]["canon"]

assert canon["/compact"] == "/compress"
assert "/compact" not in pairs
assert "/compact" not in tui_pairs


def test_commands_catalog_filters_gateway_only_commands_and_keeps_status_visible():
resp = server.handle_request(
{"id": "1", "method": "commands.catalog", "params": {}}
Expand Down Expand Up @@ -8474,3 +8545,61 @@ def fake_agent(**kwargs):

assert agent.model == "gpt-5.5"
assert captured["provider"] == "deepseek"


def test_get_usage_does_not_substitute_cumulative_total_for_context_used():
"""An external context engine that does not report last_prompt_tokens must
not have the cumulative lifetime session_total_tokens shown as its current
context occupancy — that substitution produced impossible 1.9m/120k (100%)
status-bar readings (#50421). With no real current occupancy known,
context_used/percent stay unset rather than wrong."""
agent = types.SimpleNamespace(
model="test-model",
session_total_tokens=1_900_000,
context_compressor=types.SimpleNamespace(
last_prompt_tokens=0,
context_length=120_000,
compression_count=0,
),
)
usage = server._get_usage(agent)
assert usage.get("context_used") != 1_900_000
assert "context_used" not in usage
assert "context_percent" not in usage


def test_get_usage_reports_real_current_occupancy():
"""When the compressor reports a real current prompt size, context_used is
that value (not the cumulative total) and the percent is sane."""
agent = types.SimpleNamespace(
model="test-model",
session_total_tokens=1_900_000,
context_compressor=types.SimpleNamespace(
last_prompt_tokens=60_000,
context_length=120_000,
compression_count=2,
),
)
usage = server._get_usage(agent)
assert usage["context_used"] == 60_000
assert usage["context_max"] == 120_000
assert usage["context_percent"] == 50


def test_get_usage_clamps_post_compression_sentinel():
"""Right after a compression, last_prompt_tokens is the -1 sentinel
(conversation_compression sets it until the next real usage report). It is
truthy, so `or 0` doesn't neutralize it — the guard must clamp <0 to 0 so
the transitional turn emits no gauge instead of leaking context_used=-1."""
agent = types.SimpleNamespace(
model="test-model",
session_total_tokens=4_000_000,
context_compressor=types.SimpleNamespace(
last_prompt_tokens=-1,
context_length=1_048_576,
compression_count=6,
),
)
usage = server._get_usage(agent)
assert "context_used" not in usage
assert "context_percent" not in usage
58 changes: 53 additions & 5 deletions tui_gateway/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,14 @@ def __init__(self, session_key: str, model: str):
self._closed = False
from hermes_cli._subprocess_compat import windows_hide_flags

# start_new_session=True detaches the slash worker into its own
# process group / session. Without this, the worker inherits the
# gateway's pgid (= TUI parent PID). When mcp_tool's
# _kill_orphaned_mcp_children races with slash_worker spawn and sweeps
# the gateway's child set, it captures the worker PID, records the
# inherited pgid, and killpg() then kills the TUI parent itself.
# See agent/lsp/client.py for the symmetric LSP server fix and
# tools/mcp_tool.py _filter_mcp_children for defense-in-depth.
self.proc = subprocess.Popen(
argv,
stdin=subprocess.PIPE,
Expand All @@ -296,6 +304,7 @@ def __init__(self, session_key: str, model: str):
# Tier-1 secrets (gateway/GitHub/infra) are still stripped (#29157).
env=hermes_subprocess_env(inherit_credentials=True),
creationflags=windows_hide_flags(),
start_new_session=True,
)
threading.Thread(target=self._drain_stdout, daemon=True).start()
threading.Thread(target=self._drain_stderr, daemon=True).start()
Expand Down Expand Up @@ -2964,12 +2973,31 @@ def _get_usage(agent) -> dict:
}
comp = getattr(agent, "context_compressor", None)
if comp:
ctx_used = getattr(comp, "last_prompt_tokens", 0) or usage["total"] or 0
# context_used is the *current-window* occupancy. Do NOT fall back to
# usage["total"] (cumulative lifetime session_total_tokens): for an
# external context engine that doesn't report last_prompt_tokens that
# substitution showed lifetime totals as the live context fill, yielding
# impossible readings such as 1.9m/120k clamped to 100% (#50421).
#
# Per the issue, populate context_used/percent only from a *real*
# current-occupancy value and "leave it unknown otherwise" — so a falsy
# last_prompt_tokens (0 or missing, i.e. an engine that doesn't track
# per-window occupancy) intentionally emits no gauge rather than a
# fabricated 0% or the old cumulative reading. The built-in compressor
# always reports a real last_prompt_tokens once a turn runs, so it is
# unaffected.
# Clamp the -1 "compression just ran, awaiting real usage" sentinel
# (conversation_compression.py) to 0 so the transitional turn reads as
# unknown (no gauge) instead of leaking context_used=-1. Matches the
# CLI status-bar path (cli.py _get_status_bar_snapshot).
last_prompt = getattr(comp, "last_prompt_tokens", 0) or 0
if last_prompt < 0:
last_prompt = 0
ctx_max = getattr(comp, "context_length", 0) or 0
if ctx_max:
usage["context_used"] = ctx_used
if ctx_max and last_prompt:
usage["context_used"] = last_prompt
usage["context_max"] = ctx_max
usage["context_percent"] = max(0, min(100, round(ctx_used / ctx_max * 100)))
usage["context_percent"] = max(0, min(100, round(last_prompt / ctx_max * 100)))
usage["compressions"] = getattr(comp, "compression_count", 0) or 0
# Live count of background/async subagents still running (delegate_task
# batches + background single delegations). Mirrors the classic CLI status
Expand Down Expand Up @@ -8121,7 +8149,12 @@ def _(rid, params: dict) -> dict:
return _err(rid, 4004, "truncate_before_user_ordinal must be an integer")
history = session.get("history", [])
user_indices = [i for i, m in enumerate(history) if m.get("role") == "user"]
if ordinal >= len(user_indices):
# Reject out-of-range ordinals on BOTH ends. A negative value would
# otherwise sail past the upper-bound check and hit Python's negative
# indexing below (user_indices[-1] -> the LAST user turn), silently
# truncating history to everything before it and persisting that loss
# via replace_messages — an unrecoverable overwrite of the session DB.
if ordinal < 0 or ordinal >= len(user_indices):
return _err(rid, 4018, "target user message is no longer in session history")
truncated = history[: user_indices[ordinal]]
session["history"] = truncated
Expand Down Expand Up @@ -11156,6 +11189,12 @@ def _(rid, params: dict) -> dict:
cat_map[cat].append([c, desc])

for name, desc, cat in _TUI_EXTRA:
# ``pairs`` are concrete commands; aliases live only in ``canon``.
# If a registry alias now owns the same slash text (for example
# /compact -> /compress), emitting a TUI extra with that name makes
# the catalog internally inconsistent for clients that merge both.
if canon.get(name.lower(), name) != name:
continue
all_pairs.append([name, desc])
if cat not in cat_map:
cat_map[cat] = []
Expand Down Expand Up @@ -11310,19 +11349,28 @@ def _(rid, params: dict) -> dict:
if name in qcmds:
qc = qcmds[name]
if qc.get("type") == "exec":
# Sanitize env to prevent credential leakage —
# quick commands run in the TUI server process which
# has all API keys in os.environ.
from tools.environments.local import _sanitize_subprocess_env
sanitized_env = _sanitize_subprocess_env(os.environ.copy())
r = subprocess.run(
qc.get("command", ""),
shell=True,
capture_output=True,
text=True,
timeout=30,
stdin=subprocess.DEVNULL,
env=sanitized_env,
)
output = (
(r.stdout or "")
+ ("\n" if r.stdout and r.stderr else "")
+ (r.stderr or "")
).strip()[:4000]
if output:
from agent.redact import redact_sensitive_text
output = redact_sensitive_text(output)
if r.returncode != 0:
return _err(
rid,
Expand Down
Loading