diff --git a/cli.py b/cli.py index 05a383ad9d..a43659c718 100644 --- a/cli.py +++ b/cli.py @@ -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 "") diff --git a/hermes_cli/cli_agent_setup_mixin.py b/hermes_cli/cli_agent_setup_mixin.py index f5ecb52e71..c780085ff7 100644 --- a/hermes_cli/cli_agent_setup_mixin.py +++ b/hermes_cli/cli_agent_setup_mixin.py @@ -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") diff --git a/hermes_cli/stream_json.py b/hermes_cli/stream_json.py index 6fc85227c6..d134f3eff2 100644 --- a/hermes_cli/stream_json.py +++ b/hermes_cli/stream_json.py @@ -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))}) diff --git a/tests/hermes_cli/test_stream_json.py b/tests/hermes_cli/test_stream_json.py index 9327f26a15..e7c8df5d11 100644 --- a/tests/hermes_cli/test_stream_json.py +++ b/tests/hermes_cli/test_stream_json.py @@ -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"