hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(stream-json): verbatim text deltas, closed protocol on init failure, per-call tool keys

Browse Source 
- on_text_delta dropped whitespace-only deltas, so concatenating the `text`
  events no longer reproduced the answer (a newline between paragraphs was
  lost). Only None/"" (the turn-end sentinel) is skipped now.
- The emitter was attached only after credentials + agent init succeeded, so a
  missing key or unknown provider exited 1 with an EMPTY stdout and the
  provider error rendered through ChatConsole (stdout). The emitter is now
  built before _ensure_runtime_credentials/_init_agent; that path closes the
  protocol with init + a failed `result` (exit_code 1, error) and the
  credential error goes to stderr whenever stdout is machine-readable
  (tool_progress_mode == "off", i.e. -Q and stream-json).
- _tool_started was keyed by tool name, so concurrent same-name calls
  clobbered each other's start time; key on tool_call_id when the caller
  passes one and surface it on tool_use/tool_result.

Live: `hermes chat -q … --format stream-json` with no provider and with a dead
custom base_url both yield pure JSONL (`system` + `result`, exit 1).
This commit is contained in:
teknium1
2026-09-14 21:20:16 -07:00
committed by Teknium
parent ee0666bf0e
commit aa75d3724f
4 changed files with 57 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
cli.py
View File

@@ -4491,6 +4491,12 @@ def _run_single_query_mode(cli, query, image, quiet, oneshot, stream_json: bool
if quiet:
# Quiet mode: suppress banner, spinner, tool previews.
cli.tool_progress_mode = "off"
emitter = None
if stream_json:
# Built BEFORE credentials/agent init so a failed start still closes the protocol
# (init + result) instead of exiting 1 with an empty stdout.
from hermes_cli.stream_json import StreamJsonEmitter
emitter = StreamJsonEmitter(model=getattr(cli, "model", "") or "", session_id=cli.session_id or "")
if cli._ensure_runtime_credentials():
effective_query: Any = _route_single_query_images(
cli, query, query, single_query_images, single_query_image_urls
@@ -4504,12 +4510,13 @@ def _run_single_query_mode(cli, query, image, quiet, oneshot, stream_json: bool
request_overrides=turn_route.get("request_overrides"),
):
_configure_quiet_agent(cli.agent)
emitter = None
if stream_json:
from hermes_cli.stream_json import StreamJsonEmitter
emitter = StreamJsonEmitter.attach(cli.agent, session_id=cli.session_id or "")
if emitter is not None:
emitter.attach(cli.agent)
_run_quiet_single_query(cli, effective_query, emitter=emitter)
if emitter is not None:
emitter.emit_result({"failed": True, "error": "credentials or agent init failed"},
session_id=cli.session_id or "", exit_code=1)
sys.exit(1) # credentials or agent init failed
# No welcome banner (~420 ms cold); session id / resume hint come from _print_exit_summary().
_query_label = query or ("[image attached]" if single_query_images else "")

5
hermes_cli/cli_agent_setup_mixin.py
View File

@@ -194,7 +194,10 @@ class CLIAgentSetupMixin:
_primary_exc = None
if runtime is None:
message = format_runtime_provider_error(_primary_exc) if _primary_exc else "Provider resolution failed."
ChatConsole().print(f"[bold red]{message}[/]")
if getattr(self, "tool_progress_mode", "full") == "off":
print(message, file=sys.stderr) # quiet/stream-json: stdout is machine-readable
else:
ChatConsole().print(f"[bold red]{message}[/]")
return False
api_key = runtime.get("api_key")
base_url = runtime.get("base_url")

28
hermes_cli/stream_json.py
View File

@@ -43,33 +43,39 @@ class StreamJsonEmitter:
self._tool_started: dict[str, float] = {}
self._emit({"type": "system", "subtype": "init", "model": model, "session_id": session_id})
@classmethod
def attach(cls, agent, *, session_id: str = "") -> "StreamJsonEmitter":
"""Emit ``init`` and route the agent's streaming/tool callbacks into this emitter."""
emitter = cls(model=getattr(agent, "model", "") or "", session_id=session_id)
agent.stream_delta_callback = emitter.on_text_delta
agent.tool_progress_callback = emitter.on_tool_progress
return emitter
def attach(self, agent) -> "StreamJsonEmitter":
"""Route the agent's streaming/tool callbacks into this emitter (``init`` was already written at
construction, before credentials/agent init, so a failed start still yields init + result)."""
agent.stream_delta_callback = self.on_text_delta
agent.tool_progress_callback = self.on_tool_progress
return self
def on_text_delta(self, text: str | None) -> None:
if text and str(text).strip():
self._emit({"type": "text", "text": text})
# Only None/"" (the turn-end sentinel) is dropped: whitespace deltas are part of the text, and
# a consumer concatenating ``text`` events must reproduce the answer byte for byte.
if text:
self._emit({"type": "text", "text": str(text)})
def on_tool_progress(self, event_type: str, tool_name: str | None = None, preview: Any = None, args: Any = None,
**kwargs: Any) -> None:
"""``tool.started`` → ``tool_use`` (with ``input`` when the args are a dict); ``tool.completed`` →
``tool_result``. Other progress events (reasoning, output risk) are not part of the protocol."""
name = tool_name or "unknown"
# Parallel same-name calls would clobber each other's start time under a name-only key.
key = kwargs.get("tool_call_id") or name
if event_type == "tool.started":
self._tool_started[name] = time.time()
self._tool_started[key] = time.time()
payload: dict[str, Any] = {"type": "tool_use", "name": name}
if kwargs.get("tool_call_id"):
payload["tool_call_id"] = kwargs["tool_call_id"]
if isinstance(args, dict):
payload["input"] = args
self._emit(payload)
elif event_type == "tool.completed":
duration = kwargs.get("duration") or (time.time() - self._tool_started.pop(name, time.time()))
duration = kwargs.get("duration") or (time.time() - self._tool_started.pop(key, time.time()))
output = str(kwargs.get("result") or "")
self._emit({"type": "tool_result", "name": name,
**({"tool_call_id": kwargs["tool_call_id"]} if kwargs.get("tool_call_id") else {}),
"output": output if len(output) <= _TOOL_OUTPUT_CAP else output[:_TOOL_OUTPUT_CAP] + "...",
"duration_ms": int(float(duration) * 1000), "is_error": bool(kwargs.get("is_error", False))})

38
tests/hermes_cli/test_stream_json.py
View File

@@ -16,24 +16,34 @@ def _events(capsys):
def test_emitter_event_stream_is_valid_jsonl(capsys):
emitter = StreamJsonEmitter(model="test-model", session_id="s-1")
emitter.on_text_delta("hel")
emitter.on_text_delta(" ") # whitespace-only deltas carry no information
emitter.on_text_delta("\n ") # whitespace deltas are part of the answer and must be forwarded verbatim
emitter.on_text_delta("lo")
emitter.on_text_delta(None) # the turn-end sentinel the agent sends
emitter.on_tool_progress("tool.started", "read_file", "preview", {"path": "x"})
emitter.on_text_delta("")
emitter.on_tool_progress("tool.started", "read_file", "preview", {"path": "x"}, tool_call_id="call-a")
emitter.on_tool_progress("tool.started", "read_file", "preview", {"path": "y"}, tool_call_id="call-b")
emitter.on_tool_progress("reasoning.available", "_thinking", "hmm", None) # not part of the protocol
emitter.on_tool_progress("tool.completed", "read_file", None, None, duration=0.5, is_error=False, result="x" * 6000)
emitter.on_tool_progress("tool.completed", "read_file", None, None, tool_call_id="call-b", result="y")
emitter.on_tool_progress("tool.completed", "read_file", None, None, tool_call_id="call-a", duration=0.5,
is_error=False, result="x" * 6000)
code = emitter.emit_result({"final_response": "", "failed": True, "error": "boom", "input_tokens": 3}, exit_code=0)
events = _events(capsys)
assert [e["type"] for e in events] == ["system", "text", "tool_use", "tool_result", "result"]
assert [e["type"] for e in events] == ["system", "text", "text", "text", "tool_use", "tool_use", "tool_result",
"tool_result", "result"]
assert "".join(e["text"] for e in events if e["type"] == "text") == "hel\n lo"
assert events[0]["subtype"] == "init" and events[0]["model"] == "test-model"
assert events[2]["input"] == {"path": "x"}
assert events[3]["duration_ms"] == 500 and events[3]["output"].endswith("...") and len(events[3]["output"]) == 5003
assert events[4]["input"] == {"path": "x"} and events[4]["tool_call_id"] == "call-a"
# concurrent same-name calls: each result pairs with its own start, not the last-started one
assert [e["tool_call_id"] for e in events if e["type"] == "tool_result"] == ["call-b", "call-a"]
assert events[6]["duration_ms"] < 500
assert events[7]["duration_ms"] == 500 and events[7]["output"].endswith("...") and len(events[7]["output"]) == 5003
assert code == 1 and events[-1] == {**events[-1], "exit_code": 1, "error": "boom", "session_id": "s-1"}
assert events[-1]["tokens"]["input"] == 3
assert all("timestamp" in e for e in events)
def _run_stream_json_chat(monkeypatch, capsys, run_conversation):
def _run_stream_json_chat(monkeypatch, capsys, run_conversation, credentials_ok=True):
"""parser → cmd_chat → cli.main → quiet single-query path with a deterministic fake agent."""
import cli
import hermes_cli.main as cli_entry
@@ -58,7 +68,7 @@ def _run_stream_json_chat(monkeypatch, capsys, run_conversation):
return True
def _ensure_runtime_credentials(self):
return True
return credentials_ok
def _resolve_turn_agent_config(self, _query):
return {"signature": "r", "model": None, "runtime": None, "request_overrides": None}
@@ -103,13 +113,15 @@ def _interrupted_turn(_agent):
raise KeyboardInterrupt
@pytest.mark.parametrize("turn, exit_code, types", [
(_ok_turn, 0, ["system", "text", "tool_use", "tool_result", "result"]),
(_interrupted_turn, 130, ["system", "result"]),
@pytest.mark.parametrize("turn, credentials_ok, exit_code, types", [
(_ok_turn, True, 0, ["system", "text", "tool_use", "tool_result", "result"]),
(_interrupted_turn, True, 130, ["system", "result"]),
(_ok_turn, False, 1, ["system", "result"]), # credentials fail before the agent exists
])
def test_chat_stream_json_implies_quiet_and_closes_with_result(monkeypatch, capsys, turn, exit_code, types):
def test_chat_stream_json_implies_quiet_and_closes_with_result(monkeypatch, capsys, turn, credentials_ok, exit_code,
types):
"""No ``-Q`` needed; stdout is only JSONL; the stream always ends in a ``result`` carrying the exit code."""
code, events = _run_stream_json_chat(monkeypatch, capsys, turn)
code, events = _run_stream_json_chat(monkeypatch, capsys, turn, credentials_ok=credentials_ok)
assert code == exit_code
assert [e["type"] for e in events] == types
assert events[-1]["exit_code"] == exit_code and events[-1]["session_id"] == "session-123"