fix(tui,gateway): /status, /context and /cron read as main's TUI on the shared owner

A TUI session's slash.exec reached the messaging gateway's handlers:
/status printed "📊 **Hermes Gateway Status**" markdown, /context the
gateway's gauge instead of the category table, and /cron was refused
(FINDINGS rows 17 / 19 / 21).

For a session created with source 'tui' the owner now renders main's
TUI text itself (gateway/session_tui_reads.py), inside the owning
profile's scope and the session's workspace:
- /status: main's session.status lines (status_report.build_status_fields)
  from the stored row and the live agent;
- /context: main's _format_live_context_output, the conversation counts,
  then render_context_breakdown_lines(grid=False) off the live agent;
- bare /cron and /cron list [--all]: the classic CLI's text, now built by
  cron_overview_lines / cron_list_lines in cli_commands_mixin (the CLI
  prints the same lines). Every /cron write stays refused.
Other surfaces keep the gateway handlers. The parity table records the
TUI reads.

Live (tmux, fake provider): all three pages match MAIN line for line,
apart from the session id format.
This commit is contained in:
Teknium committed 2026-10-13 00:35:44 -07:00
1 parent 9e933b75fb
commit 2efe028116
6 files changed
+340 -50

No files matched your search

+18
View File
@@ -103,6 +103,22 @@ def _skill_directive(name, arg, route):
'display': describe_skill_invocation(message, separator=' ')}
async def _tui_read(authority, actor, ref, name, arg):
"""Main's TUI sidecar text for a TUI session's /status, /context or /cron read, else None."""
from gateway.session_local import authorize_local_source
from gateway.session_policy import policy_for_source
from gateway.session_tui_reads import run_tui_read, tui_read
live = authority.sessions[ref.session_id]
if authorize_local_source(authority.runner, live.source) is not True:
return None
policy = policy_for_source(authority.runner, live.source)
render = tui_read(policy, name, arg)
if render is None:
return None
authority.authorize(actor, ref, 'session:read')
return await run_tui_read(authority, ref, live, policy, render, arg)
@contextmanager
def _session_workspace(runner, source):
"""The session's frozen terminal policy (``TERMINAL_CWD`` = its launch cwd), so ``/diff``
@@ -142,6 +158,8 @@ async def execute_command(connection, ref, params, *, dispatch=False):
definition = resolve_command(name)
canonical = definition.name if definition else name
verdict = command_verdict(canonical, arg) if definition else None
if definition is not None and (tui := await _tui_read(authority, actor, ref, canonical, arg)) is not None:
return tui
capability = ('session:read' if verdict == 'read' else
'session:control' if definition else 'session:submit')
authority.authorize(actor, ref, capability)
+130
View File
@@ -0,0 +1,130 @@
"""Main's TUI text for the session reads its sidecar answered itself: /status, /context, /cron.
On main the TUI's ``slash.exec`` never reached the messaging gateway's handlers: ``/status`` was
``session.status`` (``Hermes TUI Status``), ``/context`` was ``_format_live_context_output`` (the
conversation counts, then the desktop gauge's category table) and ``/cron`` ran the classic CLI's
handler in the slash worker. A TUI session on the shared owner gets the same lines here, read
from the owner's live agent, stored row and the profile's cron store; every other surface keeps
the gateway's own handlers (``gateway/session_commands.py``).
"""
from __future__ import annotations
import asyncio
import contextlib
from collections import Counter
# The TUI surfaces whose /status, /context and /cron read as main's TUI sidecar did.
_TUI_SOURCES = frozenset({'tui'})
def _cron_read(arg: str):
"""``/cron`` (overview) or ``/cron list [--all]``; None for a write (still refused)."""
tokens = arg.split()
if not tokens:
return 'overview'
if tokens[0].lower() == 'list' and set(tokens[1:]) <= {'--all', '-a'}:
return 'list-all' if tokens[1:] else 'list'
return None
def tui_read(policy, name: str, arg: str):
"""The renderer for a TUI session's read of *name*, or None to keep the gateway handler."""
if policy is None or policy.source not in _TUI_SOURCES:
return None
if name == 'status':
return _status
if name == 'context':
return _context
if name == 'cron' and _cron_read(arg) is not None:
return _cron
return None
def _stored_id(authority, ref) -> str:
with contextlib.suppress(Exception):
from hermes_state_local import local_receipt
return local_receipt(authority.db, ref.session_id)['entry']['session_id'] or ref.session_id
return ref.session_id
def _status(authority, ref, live, policy, agent, arg) -> str:
"""``tui_gateway/methods_session.py`` ``session.status`` on main, for the owner's session."""
from hermes_cli.status_report import build_status_fields, status_lines
from tui_gateway import server
stored = _stored_id(authority, ref)
row = {}
with contextlib.suppress(Exception):
row = authority.db.get_session(stored) or {}
running = authority._handle(ref).execution_state != 'idle'
fields = build_status_fields(
stored, agent, row, model=policy.current_model(authority) if agent is None else None,
provider=policy.provider if agent is None else None,
tokens=getattr(agent, 'session_total_tokens', 0) if agent is not None else None,
agent_running=running, home=authority.profile_id)
project = server._project_info_for_cwd(policy.cwd)
return '\n'.join([
'Hermes TUI Status', '', *status_lines(fields, 'session_id', 'path'),
*([f"Project: {project['name']}"] if project else []),
*status_lines(fields, 'title', 'model', 'created', 'last_activity', 'tokens', 'agent_running')])
def _context(authority, ref, live, policy, agent, arg) -> str:
"""Main's ``_format_live_context_output``: counts by role, model/provider, then the category
table off the live agent (``render_context_breakdown_lines(grid=False)``), else the stored usage."""
stored = _stored_id(authority, ref)
rows = []
with contextlib.suppress(Exception):
rows = authority.db.get_messages_as_conversation(stored, include_ancestors=True, include_row_ids=True)
from gateway.session_controls import display_projection
messages = display_projection(rows, authority.profile_id) if rows else []
breakdown: list[str] = []
if agent is not None:
with contextlib.suppress(Exception):
from agent.context_breakdown import (
compute_context_details, compute_session_context_breakdown, render_context_breakdown_lines)
conversation = authority.db.get_messages_as_conversation(stored, include_ancestors=True)
payload = compute_session_context_breakdown(agent, conversation)
details = None
if arg.strip().lower() in {'all', 'full', 'details'}:
with contextlib.suppress(Exception):
details = compute_context_details(agent)
breakdown = render_context_breakdown_lines(payload, details=details, grid=False)
lines = [f'Conversation: {len(messages)} messages' if messages else 'Conversation is empty (no messages yet).']
roles = Counter(str(msg.get('role') or 'unknown') for msg in messages)
lines.append(' ' + ', '.join(f'{r}: {roles.get(r, 0)}' for r in ('user', 'assistant', 'tool', 'system')))
if model := (getattr(agent, 'model', '') if agent is not None else '') or policy.current_model(authority):
lines.append(f'Model: {model}')
# Main read the provider off the compute-host metadata mirror, which a local session never
# filled, so its TUI always printed ``auto`` here; kept as main's text.
lines.append('Provider: auto')
if breakdown:
return '\n'.join([*lines, '', *breakdown])
compressor = getattr(agent, 'context_compressor', None) if agent is not None else None
used, maximum = (int(getattr(compressor, name, 0) or 0) for name in ('last_prompt_tokens', 'context_length'))
if used and maximum:
lines.append(f'Context usage: {used:,} / {maximum:,} tokens ({used / maximum * 100:.1f}%)')
elif used:
lines.append(f'Context usage: {used:,} tokens')
if count := int(getattr(compressor, 'compression_count', 0) or 0):
lines.append(f'Compressions: {count:,}')
return '\n'.join(lines)
def _cron(authority, ref, live, policy, agent, arg) -> str:
"""The classic CLI's ``/cron`` / ``/cron list`` text (main's TUI ran it in the slash worker)."""
from hermes_cli.cli_commands_mixin import cron_list_lines, cron_overview_lines
mode = _cron_read(arg)
lines = cron_overview_lines() if mode == 'overview' else cron_list_lines(include_disabled=mode == 'list-all')
return '\n'.join(lines).strip('\n')
async def run_tui_read(authority, ref, live, policy, render, arg) -> dict:
"""Render off the owner loop, inside the owning profile and the session's workspace."""
from gateway.session_authorities import owner_scope
from gateway.session_commands import _session_workspace
agent = authority.agent(ref)
def build():
with owner_scope(authority), _session_workspace(authority.runner, live.source):
return render(authority, ref, live, policy, agent, arg)
return {'type': 'exec', 'output': await asyncio.to_thread(build)}
+56 -46
View File
@@ -184,6 +184,60 @@ def _cron_api(**kwargs) -> dict:
return json.loads(cronjob_tool(**kwargs))
def cron_overview_lines() -> list[str]:
"""Bare ``/cron``: the command card, then the current jobs. Plain text (the shared gateway serves
the same lines to a TUI session, gateway/session_tui_reads.py)."""
out = ["", "+" + "-" * 68 + "+", "|" + _t("cron.overview_title").center(68) + "|",
"+" + "-" * 68 + "+", "", f" {_t('cron.overview_commands')}",
*_lines(_t("cron.overview_usage"), pad=" "), ""]
result = _cron_api(action="list")
jobs = result.get("jobs", []) if result.get("success") else []
if not jobs:
return [*out, f" {_t('cron.no_jobs_hint')}", ""]
from hermes_cli.cron import _next_run_row
out += [f" {_t('cron.current_jobs')}", " " + "-" * 63]
for job in jobs:
out.append(f" {job['job_id'][:12]:<12} | {job['schedule']:<15} | {job.get('repeat', '?'):<8}")
if job.get("skills"):
out.append(f" {_t('cron.skills', skills=', '.join(job['skills']))}")
out.append(f" {job.get('prompt_preview', '')}")
if job.get("next_run_at"):
# A stamp parked past the scheduler grace must not read as upcoming (#114309).
label, value = _next_run_row(job)
out.append(f" {_t('cron.next_short') if label == 'Next run' else label}: {value}")
out.append("")
return [*out, ""]
def cron_list_lines(*, include_disabled: bool = False) -> list[str]:
"""``/cron list [--all]``."""
result = _cron_api(action="list", include_disabled=include_disabled)
jobs = result.get("jobs", []) if result.get("success") else []
if not jobs:
return [_t("cron.no_jobs")]
from hermes_cli.cron import _next_run_row
out = ["", _t("cron.list_header"), "-" * 80]
for job in jobs:
out += [f" {_t('cron.id', job_id=job['job_id'])}", f" {_t('cron.name', name=job['name'])}",
f" {_t('cron.state', state=job.get('state', '?'))}",
f" {_t('cron.schedule_repeat', schedule=job['schedule'], repeat=job.get('repeat', '?'))}",
" {}: {}".format(*_next_run_row(job)) if job.get("next_run_at") else f" {_t('cron.next_run_na')}"]
if job.get("skills"):
out.append(f" {_t('cron.skills', skills=', '.join(job['skills']))}")
out.append(f" {_t('cron.prompt', prompt=job.get('prompt_preview', ''))}")
if job.get("last_run_at"):
status = job.get("last_status") or "?"
# delivery_failed: the run succeeded but delivery didn't — the reason lives
# in last_delivery_error (last_error is None).
if status == "delivery_failed" and job.get("last_delivery_error"):
status = _t("cron.status_delivery_failed", error=job["last_delivery_error"])
elif status == "error" and job.get("last_error"):
status = _t("cron.status_error", error=job["last_error"])
out.append(f" {_t('cron.last_run', when=job['last_run_at'], status=status)}")
out.append("")
return out
def _normalize_skills(values) -> list:
"""Strip, drop empties, and dedupe (order-preserving)."""
normalized = []
@@ -1378,54 +1432,10 @@ class CLICommandsMixin(CLICommandsSessionToolsMixin):
getattr(self, handler)(subcommand, opts)
def _cron_overview(self) -> None:
_pr("", "+" + "-" * 68 + "+", "|" + _t("cron.overview_title").center(68) + "|",
"+" + "-" * 68 + "+", "", f" {_t('cron.overview_commands')}",
*_lines(_t("cron.overview_usage"), pad=" "), "")
result = _cron_api(action="list")
jobs = result.get("jobs", []) if result.get("success") else []
if jobs:
from hermes_cli.cron import _next_run_row
_pr(f" {_t('cron.current_jobs')}", " " + "-" * 63)
for job in jobs:
print(f" {job['job_id'][:12]:<12} | {job['schedule']:<15} | {job.get('repeat', '?'):<8}")
if job.get("skills"):
print(f" {_t('cron.skills', skills=', '.join(job['skills']))}")
print(f" {job.get('prompt_preview', '')}")
if job.get("next_run_at"):
# A stamp parked past the scheduler grace must not read as upcoming (#114309).
label, value = _next_run_row(job)
print(f" {_t('cron.next_short') if label == 'Next run' else label}: {value}")
print()
else:
print(f" {_t('cron.no_jobs_hint')}")
print()
_pr(*cron_overview_lines())
def _cron_list(self, subcommand: str, opts: dict) -> None:
result = _cron_api(action="list", include_disabled=opts["all"])
jobs = result.get("jobs", []) if result.get("success") else []
if not jobs:
return print(_t("cron.no_jobs"))
from hermes_cli.cron import _next_run_row
print()
_pr(_t("cron.list_header"), "-" * 80)
for job in jobs:
_pr(f" {_t('cron.id', job_id=job['job_id'])}", f" {_t('cron.name', name=job['name'])}",
f" {_t('cron.state', state=job.get('state', '?'))}",
f" {_t('cron.schedule_repeat', schedule=job['schedule'], repeat=job.get('repeat', '?'))}",
" {}: {}".format(*_next_run_row(job)) if job.get("next_run_at") else f" {_t('cron.next_run_na')}")
if job.get("skills"):
print(f" {_t('cron.skills', skills=', '.join(job['skills']))}")
print(f" {_t('cron.prompt', prompt=job.get('prompt_preview', ''))}")
if job.get("last_run_at"):
status = job.get("last_status") or "?"
# delivery_failed: the run succeeded but delivery didn't — the reason lives
# in last_delivery_error (last_error is None).
if status == "delivery_failed" and job.get("last_delivery_error"):
status = _t("cron.status_delivery_failed", error=job["last_delivery_error"])
elif status == "error" and job.get("last_error"):
status = _t("cron.status_error", error=job["last_error"])
print(f" {_t('cron.last_run', when=job['last_run_at'], status=status)}")
print()
_pr(*cron_list_lines(include_disabled=opts["all"]))
def _cron_add(self, subcommand: str, opts: dict) -> None:
positionals = opts["positionals"]
+13 -1
View File
@@ -61,7 +61,19 @@ def gateway_cell(name):
return "refused"
# Reads a TUI session gets as main's TUI sidecar rendered them (gateway/session_tui_reads.py).
TUI_READS = {"status": "TUI session: main's `Hermes TUI Status`",
"context": "TUI session: main's live /context with the category table",
"cron": "TUI session: bare `/cron` and `/cron list [--all]` read the profile's jobs"}
def port_note(cmd):
if cmd.name in TUI_READS:
return "; ".join(filter(None, [TUI_READS[cmd.name], _base_note(cmd)]))
return _base_note(cmd)
def _base_note(cmd):
if cmd.name in TOGGLES:
return "port: per-session setting"
if cmd.name in NOT_A_LOCAL_READ:
@@ -110,7 +122,7 @@ def _ink_commands():
def ink_cell(cmd, gateway, ink, served):
entry = next((ink[n] for n in (cmd.name, *cmd.aliases) if n in ink), None)
if entry is None:
return "slash.exec" if gateway != "refused" else "slash.exec: refused"
return "slash.exec" if gateway != "refused" or cmd.name in TUI_READS else "slash.exec: refused"
_, block = entry
methods = set(re.findall(r"(?:rpc|request)(?:<[^>]*>)?\(\s*'([a-z_]+\.[a-z_.]+)'", block))
refused = re.findall(r"notAvailable', '([\w-]+) ([\w-]+)'\)", block)
+120
View File
@@ -0,0 +1,120 @@
"""A TUI session's /status, /context and /cron on the shared owner read as main's TUI sidecar did.
Main's TUI answered these itself (``session.status``: ``Hermes TUI Status``; ``/context``:
``_format_live_context_output`` with the category table; ``/cron``: the classic CLI's overview in
the slash worker). Through the owner a TUI session got the messaging gateway's markdown instead
(FINDINGS rows 17 / 19 / 21). Other surfaces keep the gateway's handlers.
"""
from types import SimpleNamespace
import pytest
import pytest_asyncio
@pytest_asyncio.fixture
async def owner(tmp_path, monkeypatch):
from gateway import run
from gateway.config import GatewayConfig
from gateway.session import SessionStore
from gateway.session_authority import initialize_session_authority
from gateway.session_controls import AuthorityConnection
home = tmp_path / 'home'
home.mkdir()
(home / 'config.yaml').write_text(
'model: {default: fake-model, provider: custom, base_url: "http://127.0.0.1:9/v1"}\n')
monkeypatch.setenv('HERMES_HOME', str(home))
monkeypatch.setattr(run, '_hermes_home', home)
store = SessionStore(tmp_path / 'sessions', GatewayConfig())
runner = SimpleNamespace(session_store=store, _session_db=store._db, adapters={}, _draining=False,
_cached_agent_for=lambda route: None)
runner._adapter_for_source = lambda source: runner.adapters.get(source.platform)
runner._adapters_for_profile = lambda profile: runner.adapters
authority = await initialize_session_authority(runner, profile_id=str(home), instance_id='first')
connection = AuthorityConnection(authority, object(), {'user_id': 'human'})
async def call(method, **params):
reply = await connection.dispatch({'id': 1, 'method': method, 'params': params})
assert 'result' in reply, reply
return reply['result']
yield SimpleNamespace(call=call, home=home, authority=authority, workdir=tmp_path)
async def _exec(owner, sid, command):
return (await owner.call('slash.exec', session_id=sid, command=command))['output']
@pytest.mark.asyncio
async def test_tui_status_is_main_tui_status(owner):
sid = (await owner.call('session.create', request_id='t', source='tui', cwd=str(owner.workdir)))['session_id']
out = await _exec(owner, sid, 'status')
lines = out.splitlines()
assert lines[0] == 'Hermes TUI Status' and lines[1] == '', out
assert lines[2].startswith('Session ID: ') and lines[3].startswith('Path: ')
labels = [line.split(':')[0] for line in lines[4:]]
assert labels == ['Model', 'Created', 'Last Activity', 'Tokens', 'Agent Running'], out
assert 'Model: fake-model (custom)' in out and 'Agent Running: No' in out
assert '**' not in out and 'Gateway Status' not in out
@pytest.mark.asyncio
async def test_tui_context_is_main_live_context(owner):
sid = (await owner.call('session.create', request_id='t', source='tui', cwd=str(owner.workdir)))['session_id']
out = await _exec(owner, sid, 'context')
assert out.splitlines()[:4] == ['Conversation is empty (no messages yet).',
' user: 0, assistant: 0, tool: 0, system: 0',
'Model: fake-model', 'Provider: auto'], out
assert 'Context Window' not in out
@pytest.mark.asyncio
async def test_tui_context_with_a_live_agent_renders_the_category_table(owner, monkeypatch):
sid = (await owner.call('session.create', request_id='t', source='tui', cwd=str(owner.workdir)))['session_id']
from agent import context_breakdown
agent = SimpleNamespace(model='fake-model', provider='custom')
owner.authority.runner._cached_agent_for = lambda route: agent
monkeypatch.setattr(context_breakdown, 'compute_session_context_breakdown', lambda a, m: {
'categories': [{'id': 'system', 'label': 'System prompt', 'tokens': 3000, 'color': '#fff'}],
'context_max': 128000, 'context_used': 120, 'context_percent': 0, 'estimated_total': 3000,
'context_estimated': False, 'context_source': 'provider_usage', 'model': 'fake-model'})
out = await _exec(owner, sid, 'context')
assert 'Estimated usage by category' in out and 'System prompt' in out, out
assert out.splitlines()[0] == 'Conversation is empty (no messages yet).'
@pytest.mark.asyncio
async def test_tui_cron_is_the_classic_overview_with_the_profile_jobs(owner):
from cron.jobs import create_job
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
token = set_hermes_home_override(str(owner.home))
try:
create_job('Summarize the news', '0 9 * * *', name='daily-digest')
finally:
reset_hermes_home_override(token)
sid = (await owner.call('session.create', request_id='t', source='tui', cwd=str(owner.workdir)))['session_id']
out = await _exec(owner, sid, 'cron')
assert '(^_^) Scheduled Tasks' in out and '/cron list' in out, out
assert '0 9 * * *' in out and 'Summarize the news' in out, out
listed = await _exec(owner, sid, 'cron list')
assert 'Scheduled Jobs:' in listed and 'daily-digest' in listed, listed
@pytest.mark.asyncio
async def test_tui_cron_writes_stay_refused(owner):
sid = (await owner.call('session.create', request_id='t', source='tui', cwd=str(owner.workdir)))['session_id']
from gateway.session_tui_reads import _cron_read
assert _cron_read('pause abc') is None and _cron_read('add "every 2h" x') is None
from gateway.session_controls import AuthorityConnection
connection = AuthorityConnection(owner.authority, object(), {'user_id': 'human'})
reply = await connection.dispatch({'id': 9, 'method': 'slash.exec',
'params': {'session_id': sid, 'command': 'cron remove abc'}})
assert reply.get('error', {}).get('message') == 'unsupported_command', reply
@pytest.mark.asyncio
async def test_other_surfaces_keep_the_gateway_handlers(owner):
from gateway.session_tui_reads import tui_read
for source in ('cli', 'gui', 'acp'):
policy = SimpleNamespace(source=source)
assert tui_read(policy, 'status', '') is None and tui_read(policy, 'context', '') is None
assert tui_read(None, 'status', '') is None
@@ -41,9 +41,9 @@ policy; `client-local (cli_only)` commands are terminal/UI features of a client.
| `/commands` | read | slash.exec | slash.exec | action: help | refused | messaging-only |
| `/compress` (/compact) | refused | session.mutate compress | canonical route | action: compress | session.mutate compress | |
| `/config` | refused | refused: not available yet | slash.exec: refused | unavailable: terminal | refused | client-local (cli_only) |
| `/context` (/ctx) | read | slash.exec | slash.exec | slash.exec | refused | |
| `/context` (/ctx) | read | slash.exec | slash.exec | slash.exec | refused | TUI session: main's live /context with the category table |
| `/copy` | refused | refused: not available yet | local | unavailable: terminal | refused | client-local (cli_only) |
| `/cron` | refused | refused: not available yet | slash.exec: refused | unavailable: terminal | refused | client-local (cli_only) |
| `/cron` | refused | refused: not available yet | slash.exec | unavailable: terminal | refused | TUI session: bare `/cron` and `/cron list [--all]` read the profile's jobs; client-local (cli_only) |
| `/curator` | refused | refused: not available yet | slash.exec: refused | unavailable: advanced | refused | no gateway handler (`_handle_curator_command` is missing) |
| `/debug` | refused | refused: not available yet | slash.exec: refused | slash.exec: refused | refused | uploads logs to a public paste |
| `/deny` | refused | refused: not available yet | slash.exec: refused | unavailable: messaging | refused | messaging-only |
@@ -105,7 +105,7 @@ policy; `client-local (cli_only)` commands are terminal/UI features of a client.
| `/skin` | refused | refused: not available yet | config.get, config.set | action: skin | refused | client-local (cli_only) |
| `/snapshot` (/snap) | refused | refused: not available yet | slash.exec: refused | unavailable: terminal | refused | client-local (cli_only) |
| `/start` | refused | refused: not available yet | slash.exec: refused | slash.exec: refused | refused | messaging-only |
| `/status` | read | slash.exec | canonical route | slash.exec (rpc fallback) | refused | |
| `/status` | read | slash.exec | canonical route | slash.exec (rpc fallback) | refused | TUI session: main's `Hermes TUI Status` |
| `/statusbar` (/sb) | refused | refused: not available yet | local (preference not saved) | unavailable: terminal | refused | client-local (cli_only) |
| `/steer` (/s) | refused | refused: not available yet | canonical route | slash.exec: refused | refused | |
| `/stop` | refused | session.interrupt | canonical route | action: stop | refused | |