mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
65effb1e9f
The Ink TUI, Desktop and the gateway-attached CLI send /model as a session.mutate model mutation straight to the session owner, so the earlier handlers never saw it and `reset` was frozen into the policy as a model name (the provider then got `model: reset`). prepare_model now treats `reset` as a pick of the owning profile's configured model, read fresh from config.yaml (session-scoped, never written back, not held behind the cost ask). A session already on that default gets a `noop` with an "already on the configured default" note and nothing is committed. `reset` with any other selection field is refused like on every other surface. A messaging chat viewed through the owner clears its override through the gateway's own reset; a reset queued behind a running turn shows the default it will adopt. The TUI, Desktop and the attached CLI print the owner's note.
651 lines
36 KiB
Python
651 lines
36 KiB
Python
"""Classic terminal presentation of authority events and fenced controls."""
|
||
import asyncio
|
||
from contextlib import suppress
|
||
import logging
|
||
import sys
|
||
import uuid
|
||
|
||
from hermes_cli.gateway_client import GatewayClientError
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
_PREVIEW_CHARS = 80
|
||
|
||
|
||
def _pending_preview(text):
|
||
"""One terminal-safe line of a pending admission's prompt: escape sequences removed, every
|
||
other non-printable (C0/C1, bidi and zero-width format chars, newlines) folded to a space,
|
||
capped at ``_PREVIEW_CHARS``. Empty when the row carries no text."""
|
||
from tools.ansi_strip import strip_ansi
|
||
if not isinstance(text, str):
|
||
return ""
|
||
line = " ".join("".join(ch if ch.isprintable() else " " for ch in strip_ansi(text)).split())
|
||
if len(line) > _PREVIEW_CHARS:
|
||
line = line[:_PREVIEW_CHARS - 1] + "…"
|
||
return line
|
||
|
||
|
||
def _quoted(row, fallback):
|
||
"""A pending row's own line: its quoted, indented prompt (so it can never start with
|
||
``/discard``), or ``fallback`` naming its id when it carries no text."""
|
||
preview = _pending_preview(row.get("text"))
|
||
return f' "{preview}"' if preview else f" ({fallback} {row['admission_id']})"
|
||
|
||
|
||
_UNKNOWN_OUTCOME = ("Hermes couldn't confirm how this turn ended (usually because it restarted mid-turn). "
|
||
"Resume this session and use the /discard <id> command it shows to drop the turn. "
|
||
"Don't send the same message again until you've checked.")
|
||
|
||
_BLOCKING_CONTROL_EVENTS = frozenset({
|
||
"approval.request", "approval.settled", "clarify.request", "clarify.settled",
|
||
"session.info",
|
||
})
|
||
|
||
|
||
def pending_lines(pending, *, interactive=True):
|
||
"""The owner's pending admissions as shown on attach (the classic CLI shows them before a turn).
|
||
A lost turn is never replayed, so the person must be told what it was and the exact command
|
||
that drops it; a queued prompt is not in the transcript yet, so it is quoted too. A one-shot
|
||
(``interactive=False``) cannot take ``/discard``, so its caller names the remedy instead."""
|
||
lost = [row for row in pending if row["status"] == "unknown"]
|
||
queued = [row for row in pending if row["status"] == "queued"]
|
||
lines = []
|
||
if len(lost) == 1:
|
||
lines.append("⏸ This chat is paused. Hermes couldn't confirm how the last turn ended "
|
||
"(usually because it restarted mid-turn):")
|
||
if _pending_preview(lost[0].get("text")):
|
||
lines.append(_quoted(lost[0], "lost turn"))
|
||
lines.append(" Hermes won't re-run it by itself, since it may already have done part of the work.")
|
||
if interactive:
|
||
lines += [" To drop it and continue, type:", f"/discard {lost[0]['admission_id']}"]
|
||
elif lost:
|
||
lines.append(f"⏸ This chat is paused. Hermes couldn't confirm how {len(lost)} turns ended "
|
||
"(usually because it restarted mid-turn):")
|
||
lines.append(" Hermes won't re-run them by itself, since they may already have done part of the work.")
|
||
if interactive:
|
||
lines.append(" To continue, drop each one with the /discard line under it:")
|
||
for row in lost:
|
||
lines.append(_quoted(row, "lost turn"))
|
||
if interactive:
|
||
lines.append(f"/discard {row['admission_id']}")
|
||
if queued:
|
||
count = "1 message is" if len(queued) == 1 else f"{len(queued)} messages are"
|
||
after = {0: "to run", 1: "and will run once that turn is dropped"}.get(
|
||
len(lost), "and will run once those turns are dropped")
|
||
lines.append(f" {count} waiting {after}:")
|
||
lines.extend(_quoted(row, "queued message") for row in queued)
|
||
return lines
|
||
|
||
|
||
def lost_admission(snapshot, admission_id):
|
||
"""``/discard``'s row: the resume snapshot is the only source of the generation the authority
|
||
stamped on a turn lost across an owner restart."""
|
||
lost = next((row for row in snapshot.get("pending", [])
|
||
if row["admission_id"] == admission_id and row["status"] == "unknown"), None)
|
||
if lost is None:
|
||
raise GatewayClientError("No unknown (lost) admission with that id; resume to refresh")
|
||
return lost
|
||
|
||
|
||
class GatewayChatView:
|
||
def __init__(self, client, snapshot, *, quiet=False, emitter=None, usage_file=None):
|
||
self.usage_file = usage_file
|
||
self.client = client
|
||
self.session_id = snapshot["stored_session_id"]
|
||
self.generation = snapshot.get("execution_generation", 0)
|
||
self.prompts = {p["prompt_id"]: p for p in snapshot.get("prompts", [])}
|
||
self.pending = snapshot.get("pending", [])
|
||
self.model = str((snapshot.get("info") or {}).get("model") or "Hermes").split("/")[-1]
|
||
# ``--format stream-json``: stdout belongs to the JSONL protocol, so every human line is
|
||
# replaced by an emitter event and the terminal record carries the exit code.
|
||
self.emitter = emitter
|
||
self.quiet = quiet or emitter is not None
|
||
self.finite = False
|
||
self.unattended = False # `-z`: the classic one-shot auto-approves; `-q` stays single-query
|
||
self.resume_footer = False # `chat -q` without -Q: main's "Resume this session with:" block
|
||
self.launch_api_key = None # `--api-key` of this terminal's launch; never durable (/new reuses it)
|
||
self.finite_admission = None
|
||
self._finite_events = []
|
||
self.streams = {}
|
||
self.completions = {}
|
||
# ``gateway_replay_gap``: admission -> the reply segment shown when a gap cut it; own unsettled turns.
|
||
self.cut, self.mine, self.gaps, self._attaching = {}, {}, 0, asyncio.Lock()
|
||
self.changed = asyncio.Event()
|
||
self.failure = None
|
||
# Set once with ``failure``: the interactive composer races its line read against it so a
|
||
# dead owner ends the REPL immediately instead of on the next keypress.
|
||
self.failed = asyncio.Event()
|
||
from hermes_cli.gateway_mutations import PreparedMutations
|
||
self.mutations = PreparedMutations()
|
||
# The interactive composer (set by ``run``); None in one-shot / non-TTY runs, where a
|
||
# guarded model switch stays a refusal instead of a prompt.
|
||
self._composer = None
|
||
# `/undo` puts the removed message back in the composer for the next prompt.
|
||
self.prefill = ""
|
||
|
||
def unknown_admissions(self):
|
||
return [row["admission_id"] for row in self.pending if row["status"] == "unknown"]
|
||
|
||
def show_pending(self):
|
||
for line in pending_lines(self.pending, interactive=not self.finite):
|
||
print(line, file=sys.stderr)
|
||
|
||
def show_prompt(self, prompt):
|
||
print(f"\n{prompt.get('description') or prompt.get('question') or 'Approval required'}", file=sys.stderr)
|
||
if prompt.get("command"):
|
||
print(prompt["command"], file=sys.stderr)
|
||
command = "/approve" if prompt["kind"] == "approval" else "/answer"
|
||
print(f"{command} {prompt['prompt_id']} <{'|'.join(prompt.get('choices', [])) or 'answer'}>", file=sys.stderr)
|
||
|
||
async def render(self):
|
||
while True:
|
||
event = await self.client.events.get()
|
||
if isinstance(event, Exception):
|
||
self._fail(event)
|
||
return
|
||
params = event.get("params", {})
|
||
if params.get("session_id") != self.session_id:
|
||
continue
|
||
kind, payload = params.get("type"), params.get("payload", {})
|
||
self.generation = params.get("execution_generation", self.generation)
|
||
if kind == "session.replay_gap":
|
||
await self._reattach()
|
||
if self.failure:
|
||
return
|
||
continue
|
||
admission = params.get("admission_id") or payload.get("admission_id")
|
||
if self.finite and kind in _BLOCKING_CONTROL_EVENTS:
|
||
# Gates and unknown executions block the session FIFO, not merely one
|
||
# admission's output. Track them even when another admission owns the
|
||
# event so a queued one-shot can detach instead of waiting forever.
|
||
self._dispatch_event(kind, admission, payload)
|
||
self.changed.set()
|
||
continue
|
||
if self.finite and admission:
|
||
if self.finite_admission is None:
|
||
# The owner can publish before prompt.submit's receipt reaches this client.
|
||
# Hold admission-scoped output until we know which admission this invocation owns.
|
||
self._finite_events.append((kind, admission, payload))
|
||
continue
|
||
if admission != self.finite_admission:
|
||
continue
|
||
self._dispatch_event(kind, admission, payload)
|
||
self.changed.set()
|
||
|
||
def _fail(self, error):
|
||
self.failure = error
|
||
self.failed.set()
|
||
self.changed.set()
|
||
|
||
async def _reattach(self):
|
||
"""The owner retired this view's subscription; its turns run on. Re-attach (a refusal fails the
|
||
view as before), show the controls the hole hid, and recover the turns in admission order."""
|
||
sid = self.session_id
|
||
async with self._attaching: # a submit acked meanwhile recovers against this snapshot, which covers it
|
||
self.gaps += 1
|
||
try:
|
||
snapshot = await self.client.rpc("session.resume", session_id=sid)
|
||
except Exception: # health: allow BLE001 -- any failed re-attach ends the wait, never hangs it
|
||
if self.session_id == sid:
|
||
self._fail(GatewayClientError("session_replay_gap"))
|
||
return
|
||
if self.session_id != sid:
|
||
return # /new or /branch moved this view meanwhile
|
||
self.generation = snapshot.get("execution_generation", self.generation)
|
||
self.pending, prompts = snapshot.get("pending", self.pending), {p["prompt_id"]: p for p in snapshot.get("prompts", [])}
|
||
for prompt_id in prompts.keys() - self.prompts.keys() if not self.finite else ():
|
||
self.show_prompt(prompts[prompt_id])
|
||
self.prompts = prompts
|
||
for admission in dict.fromkeys([*self.streams, *self.mine]): # the running turn first: one FIFO
|
||
await self._recover(admission, sid)
|
||
self.changed.set() # after its own turn settled: a control it lists may end a one-shot's wait
|
||
|
||
async def _recover(self, admission, sid):
|
||
"""Queued: it lost nothing. Running: cut until its live completion. Unknown: its completion was the
|
||
one lost, so an own turn settles as unknown now. Unlisted: it ended in the hole."""
|
||
from hermes_cli.gateway_replay_gap import pending_status, settled
|
||
status = pending_status(self.pending, admission)
|
||
if status == "unknown" and admission in self.mine:
|
||
self._complete(admission, {"outcome": "unknown", "text": _UNKNOWN_OUTCOME})
|
||
self.changed.set()
|
||
if not admission or status in ("queued", "unknown") or admission in self.completions:
|
||
return
|
||
self.cut.setdefault(admission, self.streams.pop(admission, ""))
|
||
if status or any(kind == "message.complete" and own == admission for kind, own, _ in self._finite_events):
|
||
return # (its completion already arrived, held for the submit ack)
|
||
try:
|
||
receipt = await self.client.rpc("prompt.receipt", session_id=sid, admission_id=admission, include_result=True)
|
||
except Exception: # health: allow BLE001 -- unreadable, like a receipt without its reply
|
||
receipt = {"status": "terminal"}
|
||
verdict = settled(receipt) if self.session_id == sid and admission not in self.completions else False
|
||
if verdict:
|
||
outcome = {"interrupted": "cancelled"}.get(receipt.get("outcome"), receipt.get("outcome"))
|
||
self._complete(admission, {"outcome": outcome, "_receipt": receipt,
|
||
"text": (receipt.get("result") or {}).get("final_response") or ""})
|
||
self.changed.set()
|
||
elif verdict is None and self.finite: # unreadable: fails as before (interactive: the next gap retries)
|
||
self._fail(GatewayClientError("session_replay_gap"))
|
||
|
||
async def _read_line(self, prompt, symbol):
|
||
"""The next composer line, or ``""`` once the event stream failed first (``self.failure``
|
||
is then set). The pending read is cancelled (prompt_toolkit restores the terminal) so the
|
||
caller reports the unknown outcome without waiting for a keypress."""
|
||
default, self.prefill = self.prefill, ""
|
||
reader = asyncio.ensure_future(prompt.prompt_async(symbol, **({"default": default} if default else {})))
|
||
failed = asyncio.ensure_future(self.failed.wait())
|
||
try:
|
||
await asyncio.wait({reader, failed}, return_when=asyncio.FIRST_COMPLETED)
|
||
finally:
|
||
failed.cancel()
|
||
if reader.done():
|
||
return reader.result()
|
||
reader.cancel()
|
||
with suppress(asyncio.CancelledError):
|
||
await reader
|
||
return ""
|
||
|
||
def _dispatch_event(self, kind, admission, payload):
|
||
handler = {
|
||
"message.delta": self._delta, "message.complete": self._complete,
|
||
"tool.start": self._tool_start, "tool.complete": self._tool_complete,
|
||
"approval.request": self._request, "clarify.request": self._request,
|
||
"approval.settled": self._settled, "clarify.settled": self._settled,
|
||
"session.info": self._session_info,
|
||
"status.update": self._notice, "notification.show": self._notice,
|
||
}.get(kind)
|
||
if handler:
|
||
handler(admission, payload)
|
||
|
||
def _session_info(self, admission, payload):
|
||
self.pending = payload.get("pending", self.pending)
|
||
self.generation = payload.get("execution_generation", self.generation)
|
||
|
||
def _notice(self, admission, payload):
|
||
# The owner already applied display.warning_notifications; it publishes only what shows.
|
||
if not self.quiet and str(payload.get("text") or "").strip():
|
||
print(str(payload["text"]).strip(), file=sys.stderr, flush=True)
|
||
|
||
def _tool_start(self, admission, payload):
|
||
# Deltas before a tool call are interim commentary the final reply does not repeat;
|
||
# close that segment so `_complete` measures only the final's own stream.
|
||
if admission in self.cut:
|
||
self.cut[admission] = ""
|
||
if self.streams.pop(admission, None) and not self.quiet:
|
||
print(flush=True)
|
||
if self.emitter is not None:
|
||
self.emitter.on_tool_progress("tool.started", payload.get("tool_name"), None, payload.get("args"),
|
||
tool_call_id=payload.get("tool_call_id") or None)
|
||
elif not self.quiet:
|
||
# Same line shape the in-process CLI prints: the tool's emoji and its primary argument.
|
||
from agent.display import build_tool_preview, get_tool_emoji
|
||
name = payload.get("tool_name") or payload.get("name") or "tool"
|
||
preview = build_tool_preview(name, payload.get("args") or {}, max_len=0)
|
||
print(f"{get_tool_emoji(name)} {name}{f': {preview}' if preview else ''}", flush=True)
|
||
|
||
def _tool_complete(self, admission, payload):
|
||
if self.emitter is not None:
|
||
self.emitter.on_tool_progress("tool.completed", payload.get("tool_name"), None, payload.get("args"),
|
||
tool_call_id=payload.get("tool_call_id") or None,
|
||
result=payload.get("result"), is_error=payload.get("is_error", False))
|
||
|
||
def _delta(self, admission, payload):
|
||
text = payload.get("text") or payload.get("delta") or payload.get("content") or ""
|
||
if not isinstance(text, str) or admission in self.cut:
|
||
return # cut: live text stopped at the replay gap; the settlement shows the reply once
|
||
if self.emitter is not None:
|
||
self.emitter.on_text_delta(text)
|
||
elif not self.quiet:
|
||
self.streams[admission] = self.streams.get(admission, "") + text
|
||
# No flush: under the live composer, patch_stdout line-buffers this so a redraw
|
||
# (a resize mid-stream) never interleaves with a half-written line. Complete lines
|
||
# still appear as they stream; the last one lands with `_complete`.
|
||
print(text, end="")
|
||
|
||
def _complete(self, admission, payload):
|
||
text = payload.get("text") or payload.get("content") or ""
|
||
streamed, cut = self.streams.pop(admission, ""), self.cut.pop(admission, None)
|
||
self.mine.pop(admission, None)
|
||
if cut is not None:
|
||
from hermes_cli.gateway_replay_gap import owed_reply
|
||
payload = {**payload, "replay_gap": True}
|
||
if not self.quiet and text:
|
||
print(owed_reply(cut, text), flush=True)
|
||
elif not self.quiet:
|
||
# The final's deltas carry the agent's segment break (leading blank lines after a
|
||
# tool call) that the settled text has trimmed; a match modulo that edge whitespace
|
||
# is the same reply already on screen.
|
||
if not streamed:
|
||
print(text, flush=True)
|
||
elif text.startswith(streamed):
|
||
print(text[len(streamed):], flush=True)
|
||
elif text.strip() == streamed.strip():
|
||
print(flush=True)
|
||
else:
|
||
print("\n" + text, flush=True)
|
||
# A turn the owner's restart/stop interrupted names that cause (#134889 F1); other
|
||
# message.complete warnings stay Ink/Desktop-only as before.
|
||
warning = payload.get("warning")
|
||
if payload.get("status") == "interrupted" and isinstance(warning, str) and warning.strip():
|
||
print(warning.strip(), file=sys.stderr, flush=True)
|
||
self.completions[admission] = payload
|
||
|
||
def _request(self, admission, payload):
|
||
self.prompts[payload["prompt_id"]] = payload
|
||
# Finite invocations only need the session-blocking state so they can
|
||
# detach cleanly; do not print another admission's control details.
|
||
if not self.finite:
|
||
self.show_prompt(payload)
|
||
|
||
def _settled(self, admission, payload):
|
||
self.prompts.pop(payload["prompt_id"], None)
|
||
|
||
async def submit(self, text):
|
||
self.mutations.retire(self.session_id)
|
||
sid, gaps = self.session_id, self.gaps
|
||
receipt = await self.client.rpc("prompt.submit", session_id=self.session_id,
|
||
input_id=uuid.uuid4().hex, text=text,
|
||
**({"finite": True} if self.finite else {}),
|
||
**({"unattended": True} if self.finite and self.unattended else {}))
|
||
if (receipt or {}).get("voice_stopped") is True:
|
||
# TUI-26: a bare stop phrase typed into this LOCAL session's voice chat (another viewer
|
||
# turned it on) ended the chat on the owner; nothing was admitted and no turn follows.
|
||
if not self.quiet:
|
||
from agent.i18n import t
|
||
print(t("cli.voice.stop_phrase_typed"), flush=True)
|
||
return receipt
|
||
admission = (receipt or {}).get("admission_id")
|
||
if admission and admission not in self.completions and self.session_id == sid:
|
||
self.mine[admission] = None
|
||
if self.gaps != gaps: # a re-attach left after this submit: its snapshot covers it
|
||
async with self._attaching:
|
||
if admission not in self.cut and not self.failure:
|
||
await self._recover(admission, sid)
|
||
return receipt
|
||
|
||
async def command(self, text):
|
||
command, _, rest = text.partition(" ")
|
||
if command in {"/quit", "/exit", "/detach"}:
|
||
return False
|
||
if command == "/stop":
|
||
await self.client.rpc("session.interrupt", session_id=self.session_id,
|
||
execution_generation=self.generation)
|
||
return True
|
||
if command in {"/approve", "/answer"}:
|
||
prompt_id, _, answer = rest.partition(" ")
|
||
prompt = self.prompts.get(prompt_id)
|
||
expected = "approval" if command == "/approve" else "clarify"
|
||
if not prompt or prompt["kind"] != expected:
|
||
raise GatewayClientError("No matching pending control; resume to refresh")
|
||
await self.client.rpc(expected + ".respond", session_id=self.session_id,
|
||
execution_generation=prompt["execution_generation"], prompt_id=prompt_id,
|
||
**({"choice": answer} if expected == "approval" else {"answer": answer}))
|
||
return True
|
||
if command == "/discard":
|
||
snapshot = await self.client.rpc("session.resume", session_id=self.session_id)
|
||
lost = lost_admission(snapshot, rest.strip())
|
||
await self.client.rpc("prompt.resolve_unknown", session_id=self.session_id,
|
||
admission_id=lost["admission_id"], execution_generation=lost["execution_generation"])
|
||
return True
|
||
if command in {'/branch', '/model', '/compress'}:
|
||
from hermes_cli.gateway_mutations import slash_mutation
|
||
operation, payload = slash_mutation(command, rest.strip())
|
||
original = self.session_id
|
||
confirm = self._confirm_model_switch if operation == 'model' and self._composer is not None else None
|
||
result = await self.mutations.apply(self.client, original, operation, payload, confirm=confirm)
|
||
if result.get('status') == 'cancelled':
|
||
return True
|
||
if result.get('status') == 'preview':
|
||
self.mutations.acknowledge(original, operation, payload)
|
||
print('\n'.join(result['lines']))
|
||
return True
|
||
if result.get('status') == 'noop':
|
||
# Nothing removable (gateway/session_mutation_compress.py), or a `/model reset` already
|
||
# on the configured default (its note): main's report, nothing committed.
|
||
self.mutations.acknowledge(original, operation, payload)
|
||
summary = result.get('summary') or {'note': result.get('note')}
|
||
print('\n'.join(filter(None, (summary.get('headline'), summary.get('token_line'), summary.get('note')))))
|
||
return True
|
||
target = result.get('branched_session_id', original)
|
||
await self.adopt(await self.client.rpc('session.resume', session_id=target))
|
||
self.mutations.acknowledge(original, operation, payload)
|
||
if operation == 'model' and result.get('model'):
|
||
self.model = str(result['model']).split('/')[-1]
|
||
print(f"{operation}: {target}")
|
||
return True
|
||
if command == "/yolo":
|
||
# This session's approval bypass on the owner (same verb as the TUI's /yolo and Desktop).
|
||
word = rest.strip().lower()
|
||
if word not in {"", "on", "off"}:
|
||
raise GatewayClientError("Usage: /yolo [on|off]")
|
||
result = await self.client.rpc("config.set", session_id=self.session_id, key="yolo",
|
||
**({"value": "1" if word == "on" else "0"} if word else {}))
|
||
print(f"YOLO {'on' if result.get('value') == '1' else 'off'} for this session")
|
||
return True
|
||
from hermes_cli import gateway_chat_commands
|
||
if command == "/help":
|
||
print(gateway_chat_commands.help_text())
|
||
return True
|
||
return await gateway_chat_commands.run_command(self, command, rest)
|
||
|
||
async def adopt(self, snapshot):
|
||
"""Attach this view to another session's snapshot (a branch, /new)."""
|
||
self.session_id = snapshot['stored_session_id']
|
||
self.generation = snapshot['execution_generation']
|
||
self.prompts = {p['prompt_id']: p for p in snapshot.get('prompts', [])}
|
||
self.pending = snapshot.get('pending', [])
|
||
self.model = str((snapshot.get('info') or {}).get('model') or self.model).split('/')[-1]
|
||
if not self.finite:
|
||
# Bare `hermes -c` in this terminal continues the session it is now on, as the
|
||
# in-process CLI re-wrote its breadcrumb on every session switch.
|
||
from hermes_cli.terminal_breadcrumbs import write_breadcrumb
|
||
await asyncio.to_thread(write_breadcrumb, self.session_id)
|
||
|
||
async def _confirm_model_switch(self, refusal):
|
||
"""The owner refused a guarded model target (cost / data policy / large context): ask as
|
||
the in-process CLI does (``_confirm_expensive_model_switch``: switch anyway once, or cancel)
|
||
on this composer. Only an explicit yes applies; anything else keeps the current model."""
|
||
from agent.i18n import t
|
||
from hermes_cli.gateway_mutations import confirm_choice, confirmation_title
|
||
choices = [("once", t("cli.model.choice_switch_anyway"), t("cli.model.desc_switch_anyway")),
|
||
("cancel", t("cli.model.choice_cancel"), t("cli.model.desc_keep_current_model"))]
|
||
print(f"\n!!! {confirmation_title(refusal)} !!!\n{refusal['confirm_message']}\n")
|
||
for index, (_, label, detail) in enumerate(choices, 1):
|
||
print(f" {index}. {label} \u2014 {detail}")
|
||
try:
|
||
raw = await self._read_line(self._composer, t("cli.model.confirm_choice_prompt"))
|
||
except (KeyboardInterrupt, EOFError):
|
||
raw = ""
|
||
if not self.failure and confirm_choice(raw, choices) == "once":
|
||
return True
|
||
print(t("cli.model.switch_cancelled"))
|
||
return False
|
||
|
||
def _detach(self, message):
|
||
"""One-shot cannot go on without a human: say why on stderr, exit 3, and in stream-json
|
||
mode close the protocol with a failed ``result`` (a consumer parsing stdout must never be
|
||
left without a terminal record)."""
|
||
print(message, file=sys.stderr)
|
||
if self.emitter is not None:
|
||
return self.emitter.emit_result({"failed": True, "error": message}, session_id=self.session_id, exit_code=3)
|
||
return 3
|
||
|
||
def _finite_block(self, admission):
|
||
"""Why a finite viewer cannot keep waiting for *admission*, or None. Its own lost turn
|
||
settles as the unknown completion would, so the exit code never depends on frame order;
|
||
only a still-queued input behind someone else's unknown turn is retained and detached."""
|
||
own = next((row["status"] for row in self.pending if row["admission_id"] == admission), None)
|
||
if own == "unknown":
|
||
self.completions[admission] = {"outcome": "unknown", "text": _UNKNOWN_OUTCOME}
|
||
return None
|
||
if own == "queued" and self.unknown_admissions():
|
||
return ("This chat is paused behind a turn whose ending Hermes couldn't confirm. "
|
||
"Your message is saved and will run once you resume this session and /discard "
|
||
"that turn; don't send it again.")
|
||
if self.prompts and (own or not self.gaps): # a control blocks only a turn still pending
|
||
return "Input required; detached without cancelling. Resume this session interactively."
|
||
return None
|
||
|
||
async def _settled_result(self, admission):
|
||
"""The structured result the owner committed with this admission's settlement — the same
|
||
dict the in-process one-shot got from ``run_conversation`` (best-effort, never raises)."""
|
||
from websockets.exceptions import WebSocketException
|
||
try:
|
||
receipt = await self.client.rpc("prompt.receipt", session_id=self.session_id,
|
||
admission_id=admission, include_result=True)
|
||
return dict(receipt.get("result") or {})
|
||
# Transport loss / refusal, or a receipt whose ``result`` is not a mapping.
|
||
except (GatewayClientError, OSError, TimeoutError, WebSocketException,
|
||
AttributeError, TypeError, ValueError) as exc:
|
||
# The exit code and ledger still follow the settled outcome; a missing receipt is not fatal.
|
||
logger.debug("settled result for %s unavailable: %s", admission, exc)
|
||
return {}
|
||
|
||
def _print_exit_ids(self):
|
||
"""The finite run's durable id on stderr (automation wrappers parse it; it names the
|
||
physical row a compaction may have advanced) and, for `chat -q`, main's resume block. Main's
|
||
`-z` (unattended) printed only the reply: no id line."""
|
||
if not self.unattended:
|
||
print(f"\nsession_id: {self.session_id}", file=sys.stderr, flush=True)
|
||
if self.resume_footer:
|
||
print(resume_footer(self.session_id), flush=True)
|
||
|
||
@staticmethod
|
||
def _report_silent_interrupt(text, result):
|
||
"""A restart or stop drain settles a finite turn interrupted with no reply: say so on stderr
|
||
instead of exiting 130 silently. It is never replayed."""
|
||
if text or not result.get("interrupted"):
|
||
return
|
||
reason = result.get("interrupt_message")
|
||
print(f"Turn interrupted{f' ({reason})' if isinstance(reason, str) and reason else ''}; "
|
||
"it was not retried. Resume this session to continue.", file=sys.stderr)
|
||
|
||
def _report_no_reply(self, exit_code):
|
||
"""`-z` exit 1 (a completed turn with no reply text): the in-process one-shot's stderr line."""
|
||
if self.unattended and exit_code == 1:
|
||
print("hermes -z: no final response was produced; treating the run as failed.", file=sys.stderr)
|
||
|
||
def _write_usage_file(self, result, outcome):
|
||
"""``-z --usage-file``: the same JSON ledger the in-process one-shot wrote."""
|
||
from hermes_cli.oneshot import _write_usage_file
|
||
result = {**result, "session_id": result.get("session_id") or self.session_id}
|
||
failure = None if outcome in ("completed", "cancelled") else (result.get("error") or outcome or "failed")
|
||
_write_usage_file(self.usage_file, result, failure=failure)
|
||
|
||
async def run(self, query=None, *, oneshot=False):
|
||
self.quiet = self.quiet or oneshot
|
||
self.finite = oneshot
|
||
self.show_pending()
|
||
for prompt in self.prompts.values():
|
||
self.show_prompt(prompt)
|
||
if oneshot and self.unknown_admissions():
|
||
# A new admission would queue behind the unknown row and never run; refuse BEFORE
|
||
# submitting so nothing is left queued for the next interactive resume to find.
|
||
return self._detach(f"Your message was not sent. To continue, run `hermes sessions discard "
|
||
f"{self.session_id} --yes`, or resume this session in an interactive chat "
|
||
"and /discard the paused turn there; then send it again.")
|
||
renderer = asyncio.create_task(self.render())
|
||
try:
|
||
receipt = await self.submit(query) if query else None
|
||
if oneshot:
|
||
admission = _oneshot_admission(receipt)
|
||
if admission is None:
|
||
return 0 # the voice chat ended; no turn was admitted, so none is awaited
|
||
self.finite_admission = admission
|
||
for kind, event_admission, payload in self._finite_events:
|
||
if event_admission == admission and admission not in self.completions:
|
||
self._dispatch_event(kind, event_admission, payload)
|
||
self._finite_events.clear()
|
||
self.changed.set()
|
||
while admission not in self.completions:
|
||
self.changed.clear()
|
||
if self.failure:
|
||
raise self.failure
|
||
blocked = self._finite_block(admission)
|
||
if blocked:
|
||
return self._detach(blocked)
|
||
if admission not in self.completions:
|
||
await self.changed.wait()
|
||
terminal = self.completions[admission]
|
||
outcome = terminal.get("outcome")
|
||
text = terminal.get("text") or terminal.get("content") or ""
|
||
# The outcome only says failed/cancelled; a turn stopped by --max-turns settles
|
||
# 'completed' with ``completed: False`` in its committed result. Judge both with the
|
||
# in-process exit contracts: `-z` 0/2/130 (1 = no text), `chat -q`/`-Q` 0/1/130.
|
||
from hermes_cli.oneshot import _oneshot_exit_code
|
||
from hermes_cli.turn_exit import turn_exit_code
|
||
recovered = terminal.get("_receipt")
|
||
result = dict(recovered.get("result") or {}) if recovered else await self._settled_result(admission)
|
||
result.update(failed=bool(result.get("failed")) or outcome not in ("completed", "cancelled"),
|
||
interrupted=bool(result.get("interrupted")) or outcome == "cancelled")
|
||
exit_code = (_oneshot_exit_code(text, result) if self.unattended
|
||
else turn_exit_code(result, kanban_worker=False))
|
||
if self.usage_file:
|
||
self._write_usage_file(result, outcome)
|
||
if self.emitter is not None:
|
||
return self.emitter.emit_result({**result, "final_response": text,
|
||
"replay_gap": bool(terminal.get("replay_gap"))},
|
||
session_id=self.session_id, exit_code=exit_code)
|
||
self._report_silent_interrupt(text, result)
|
||
print(text, flush=True)
|
||
self._report_no_reply(exit_code)
|
||
# Same stderr exit contract as the legacy -Q path: automation wrappers read the
|
||
# durable id from this line, and it names the physical row (a compaction may have
|
||
# advanced it past the row printed at start).
|
||
self._print_exit_ids()
|
||
return exit_code
|
||
from prompt_toolkit import PromptSession
|
||
from prompt_toolkit.patch_stdout import patch_stdout
|
||
from hermes_cli.skin_engine import get_active_prompt_symbol, get_active_skin
|
||
welcome = "Welcome to Hermes Agent! Type your message or /help for commands."
|
||
print(get_active_skin().get_branding("welcome", welcome), flush=True)
|
||
# The classic status bar's leading segments: model, then the attached session.
|
||
prompt = PromptSession(erase_when_done=True,
|
||
bottom_toolbar=lambda: f" \u2624 {self.model} \u2502 {self.session_id} ")
|
||
prompt_symbol = get_active_prompt_symbol("❯ ")
|
||
self._composer = prompt if sys.stdin.isatty() else None
|
||
with patch_stdout():
|
||
while not self.failure:
|
||
try:
|
||
# Empty when the stream failed first: the loop guard then raises it.
|
||
text = (await self._read_line(prompt, prompt_symbol)).strip()
|
||
if not text:
|
||
continue
|
||
if text.startswith("/"):
|
||
if not await self.command(text):
|
||
return 0
|
||
else:
|
||
# Same scrollback shape as the in-process CLI: the typed prompt line is
|
||
# erased on submit and the message lands as a `●` preview row.
|
||
print(f"\n{'─' * 40}\n● {text}", flush=True)
|
||
await self.submit(text)
|
||
except KeyboardInterrupt:
|
||
print("Use /stop to interrupt execution, /quit to detach.")
|
||
except EOFError:
|
||
return 0
|
||
except GatewayClientError as exc:
|
||
print(f"Error: {exc}", file=sys.stderr)
|
||
raise self.failure
|
||
finally:
|
||
renderer.cancel()
|
||
with suppress(asyncio.CancelledError):
|
||
await renderer
|
||
|
||
|
||
def _oneshot_admission(receipt):
|
||
"""The one-shot query's admission id; None when the owner admitted nothing because the query was a
|
||
bare stop phrase that ended the session's LOCAL voice chat (TUI-26 ``{voice_stopped: true}``)."""
|
||
if receipt is None:
|
||
raise GatewayClientError("One-shot requires a query")
|
||
return None if receipt.get("voice_stopped") is True else receipt["admission_id"]
|
||
|
||
|
||
def resume_footer(session_id):
|
||
"""The classic single-query exit block (``_print_exit_summary``): how to continue this session.
|
||
A non-default profile needs ``-p`` because session ids are profile-scoped."""
|
||
from agent.i18n import t
|
||
from hermes_cli.profiles import get_active_profile_name
|
||
profile = get_active_profile_name()
|
||
flag = "" if profile in ("default", "custom") else f" -p {profile}"
|
||
return f"\n{t('cli.session.exit_resume_hint')}\n hermes --resume {session_id}{flag}"
|