Skip to content
8 changes: 8 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,12 @@

import pytest

# Tests must never inherit interactive voice/TTS state from the Hermes TUI that
# launched them. Set these before test modules import gateway code; per-test
# monkeypatches are too late for background threads that can outlive teardown.
os.environ["HERMES_VOICE"] = "0"
os.environ["HERMES_VOICE_TTS"] = "0"

# Ensure project root is importable
PROJECT_ROOT = Path(__file__).parent.parent
if str(PROJECT_ROOT) not in sys.path:
Expand Down Expand Up @@ -345,6 +351,8 @@ def _hermetic_environment(tmp_path, monkeypatch):
# 2. Blank behavioral HERMES_* vars that could change test semantics.
for name in _HERMES_BEHAVIORAL_VARS:
monkeypatch.delenv(name, raising=False)
monkeypatch.setenv("HERMES_VOICE", "0")
monkeypatch.setenv("HERMES_VOICE_TTS", "0")

# Honcho's fallback host/config resolution legitimately reads the user's
# global ~/.honcho/config.json. Keep HOME stable (subprocess tests depend
Expand Down
14 changes: 14 additions & 0 deletions tests/test_live_system_guard_self_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from __future__ import annotations

import os
import shutil
import signal
import subprocess

Expand All @@ -26,6 +27,15 @@
# A guaranteed-foreign PID: PID 1 (init). Owned by root, not us, and
# always exists. A sane guard refuses to signal it.
FOREIGN_PID = 1
SYSTEMCTL_UNAVAILABLE = pytest.mark.skipif(
shutil.which("systemctl") is None,
reason="systemctl is unavailable on this platform",
)


def test_voice_tts_is_disabled_for_every_test_process():
assert os.environ.get("HERMES_VOICE") == "0"
assert os.environ.get("HERMES_VOICE_TTS") == "0"


# ──────────────────── kill primitives ─────────────────────────
Expand Down Expand Up @@ -204,6 +214,7 @@ def test_subprocess_killall_hermes_blocked():
# ──────────────────── pass-through cases (must NOT raise) ──────


@SYSTEMCTL_UNAVAILABLE
def test_systemctl_status_passes_through():
"""Read-only systemctl probes (status/show/list-units) are fine."""
# Run with check=False so we don't fail on the gateway's exit code.
Expand All @@ -216,6 +227,7 @@ def test_systemctl_status_passes_through():
assert r is not None # Did not raise — the guard let it through.


@SYSTEMCTL_UNAVAILABLE
def test_systemctl_show_passes_through():
r = subprocess.run(
["systemctl", "--user", "show", "hermes-gateway", "--no-pager"],
Expand All @@ -226,6 +238,7 @@ def test_systemctl_show_passes_through():
assert r is not None


@SYSTEMCTL_UNAVAILABLE
def test_systemctl_list_units_passes_through():
r = subprocess.run(
["systemctl", "--user", "list-units", "fake-not-real-unit*", "--no-pager"],
Expand All @@ -236,6 +249,7 @@ def test_systemctl_list_units_passes_through():
assert r is not None


@SYSTEMCTL_UNAVAILABLE
def test_systemctl_unrelated_unit_passes_through():
"""systemctl restart of a non-hermes unit is allowed (we only protect hermes)."""
# Use --dry-run so we don't actually try to restart anything; just
Expand Down
77 changes: 75 additions & 2 deletions tests/test_tui_gateway_server.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import json
import os
import platform
import subprocess
import sys
import threading
Expand Down Expand Up @@ -8129,6 +8130,7 @@ def test_session_activate_returns_inflight_stream_before_completion(monkeypatch)
started = threading.Event()
release = threading.Event()
done = threading.Event()
monkeypatch.setattr(server, "_voice_tts_enabled", lambda: False)

class _Agent:
model = "model-live"
Expand Down Expand Up @@ -8201,9 +8203,15 @@ def _emit(event, sid, payload=None):
{"role": "user", "text": "write a long answer"},
{"role": "assistant", "text": "partial answer complete"},
]
run_thread = server._sessions["sid-live"]["_run_thread"]
run_thread.join(2)
assert not run_thread.is_alive(), "prompt worker outlived test monkeypatches"
finally:
release.set()
done.wait(2)
run_thread = server._sessions.get("sid-live", {}).get("_run_thread")
if run_thread is not None:
run_thread.join(2)
assert not run_thread.is_alive(), "prompt worker survived test cleanup"
server._sessions.pop("sid-live", None)


Expand Down Expand Up @@ -8737,15 +8745,22 @@ def _opener(url, timeout=2.0): # noqa: ARG001 — match urllib signature
return _Resp()

import urllib.request
import webbrowser

def _forbid_browser_side_effect(*_args, **_kwargs):
raise AssertionError("browser launch side effect escaped test isolation")

monkeypatch.setattr(urllib.request, "urlopen", _opener)
monkeypatch.setattr(subprocess, "Popen", _forbid_browser_side_effect)
monkeypatch.setattr("builtins.open", _forbid_browser_side_effect)
monkeypatch.setattr(webbrowser, "open", _forbid_browser_side_effect)
launched = ChromeDebugLaunch(launched=True)
with patch.dict(sys.modules, {"tools.browser_tool": fake}):
with (
patch(
"hermes_cli.browser_connect.launch_chrome_debug",
return_value=launched,
),
) as launch_chrome_debug,
patch("hermes_cli.browser_connect.local_port_in_use", return_value=False),
):
resp = server.handle_request(
Expand All @@ -8759,6 +8774,7 @@ def _opener(url, timeout=2.0): # noqa: ARG001 — match urllib signature
"Chromium-family browser launched and listening on port 9222",
]
assert os.environ["BROWSER_CDP_URL"] == "http://127.0.0.1:9222"
launch_chrome_debug.assert_called_once_with(9222, platform.system())


def test_browser_manage_connect_finds_ipv6_only_browser(monkeypatch):
Expand Down Expand Up @@ -10696,6 +10712,63 @@ def _boom():
assert usage["model"] == "x"


def test_get_usage_reports_the_agents_active_credential_label():
"""Report the entry actually installed on this agent, not pool current()."""
entries = [
types.SimpleNamespace(label="personal", runtime_api_key="personal-token", access_token=""),
types.SimpleNamespace(label="work", runtime_api_key="work-token", access_token=""),
]
pool = types.SimpleNamespace(
entries=lambda: entries,
# A shared pool may point at a subagent's lease instead.
current=lambda: entries[1],
)
agent = types.SimpleNamespace(
model="x",
api_key="personal-token",
_credential_pool=pool,
)

usage = server._get_usage(agent)

assert usage["credential_label"] == "personal"


def test_get_usage_clears_credential_label_for_single_entry_pool():
entry = types.SimpleNamespace(label="personal", runtime_api_key="token", access_token="")
agent = types.SimpleNamespace(
model="x",
api_key="token",
_credential_pool=types.SimpleNamespace(entries=lambda: [entry]),
)

usage = server._get_usage(agent)

# Usage snapshots are merged by the TUI, so an explicit empty value clears
# identity left by a previous provider/model.
assert usage["credential_label"] == ""


def test_get_usage_sanitizes_credential_label_for_single_line_status_chrome():
entries = [
types.SimpleNamespace(
label="personal\naccount\x1b",
runtime_api_key="personal-token",
access_token="",
),
types.SimpleNamespace(label="work", runtime_api_key="work-token", access_token=""),
]
agent = types.SimpleNamespace(
model="x",
api_key="personal-token",
_credential_pool=types.SimpleNamespace(entries=lambda: entries),
)

usage = server._get_usage(agent)

assert usage["credential_label"] == "personal account"


def test_persist_model_switch_preserves_sibling_model_keys(tmp_path, monkeypatch):
"""#48305: switching models from the TUI must NOT destroy sibling keys under
`model:` (model_slots, model_fallback, etc.). _persist_model_switch now uses
Expand Down
36 changes: 36 additions & 0 deletions tui_gateway/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -3581,6 +3581,39 @@ def _sync_session_key_after_compress(
pass


def _active_credential_label(agent) -> str:
"""Return the label for the pooled credential installed on ``agent``.

Pool ``current()`` is process-global and may reflect a subagent lease, so
identify the entry by the live key on this agent instead. Single-entry
pools are intentionally omitted to keep ordinary status chrome uncluttered.
"""
try:
pool = getattr(agent, "_credential_pool", None)
entries = list(pool.entries()) if pool is not None else []
if len(entries) < 2:
return ""

live_key = getattr(agent, "api_key", None)
if not live_key or callable(live_key):
return ""

for entry in entries:
entry_key = (
getattr(entry, "runtime_api_key", None)
or getattr(entry, "access_token", None)
)
if entry_key == live_key:
raw_label = str(getattr(entry, "label", "") or "")
printable = "".join(
char if char.isprintable() else " " for char in raw_label
)
return " ".join(printable.split())
except Exception:
pass
return ""


def _get_usage(agent) -> dict:
g = lambda k, fb=None: getattr(agent, k, 0) or (getattr(agent, fb, 0) if fb else 0)
usage = {
Expand All @@ -3593,6 +3626,9 @@ def _get_usage(agent) -> dict:
"total": g("session_total_tokens"),
"calls": g("session_api_calls"),
}
# The TUI merges usage snapshots, so always include the field. An empty
# value clears stale identity chrome after a model/provider switch.
usage["credential_label"] = _active_credential_label(agent)
comp = getattr(agent, "context_compressor", None)
if comp:
# context_used is the *current-window* occupancy. Do NOT fall back to
Expand Down
43 changes: 42 additions & 1 deletion ui-tui/src/__tests__/appChromeStatusRule.test.tsx
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import { stringWidth } from '@hermes/ink'
import React from 'react'
import { describe, expect, it, vi } from 'vitest'

import { StatusRule } from '../components/appChrome.js'
import { credentialStatusLabel, StatusRule } from '../components/appChrome.js'
import { DEFAULT_THEME } from '../theme.js'

type ReactNodeLike = React.ReactNode
Expand Down Expand Up @@ -104,6 +105,46 @@ const baseProps = {
voiceLabel: ''
}

describe('StatusRule active credential label', () => {
it('renders the active pooled-account label beside the model', () => {
const element = StatusRule({
...baseProps,
credentialLabel: 'personal'
})

expect(textContent(element)).toContain('opus 4.8 · personal')
})

it('does not add an account separator when no pooled label is available', () => {
const element = StatusRule({ ...baseProps })

expect(textContent(element)).toContain('opus 4.8')
expect(textContent(element)).not.toContain('opus 4.8 ·')
})

it('hides account identity before the compact-context breakpoint', () => {
const element = StatusRule({
...baseProps,
cols: 44,
credentialLabel: 'personal'
})

expect(textContent(element)).not.toContain('personal')
})

it.each([
'personal-account-with-an-excessively-long-name',
'averylongaddress@example.com',
'個人用アカウント非常に長い名前',
'🧑🏽‍💻🚀✨ personal account'
])('bounds long and wide labels without splitting graphemes: %s', label => {
const rendered = credentialStatusLabel(label)

expect(rendered).toMatch(/…$/u)
expect(stringWidth(rendered)).toBeLessThanOrEqual(20)
})
})

describe('StatusRule background-subagent indicator', () => {
it('renders ⛓ N on a wide terminal when subagents are running', () => {
const element = StatusRule({
Expand Down
13 changes: 13 additions & 0 deletions ui-tui/src/__tests__/appLayoutModel.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
import { describe, expect, it } from 'vitest'

import { statusModel } from '../components/appLayout.js'

describe('statusModel', () => {
it('uses the atomic runtime model when present', () => {
expect(statusModel('claude-fallback', 'gpt-primary')).toBe('claude-fallback')
})

it.each(['', ' ', undefined])('falls back to session info for an absent runtime model: %j', model => {
expect(statusModel(model, 'gpt-primary')).toBe('gpt-primary')
})
})
21 changes: 21 additions & 0 deletions ui-tui/src/__tests__/createGatewayEventHandler.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,27 @@ describe('createGatewayEventHandler', () => {
patchUiState({ showReasoning: true })
})

it('merges fallback model and credential identity from one completion snapshot', () => {
const onEvent = createGatewayEventHandler(buildCtx([]))

onEvent({
payload: {
text: 'fallback answer',
usage: {
calls: 1,
credential_label: 'work',
input: 10,
model: 'claude-sonnet',
output: 5,
total: 15
}
},
type: 'message.complete'
} as any)

expect(getUiState().usage).toMatchObject({ credential_label: 'work', model: 'claude-sonnet' })
})

it('archives incomplete todos into transcript flow at end of turn so they scroll up', () => {
const appended: Msg[] = []

Expand Down
Loading
Loading