fix(output): write --json as UTF-8 regardless of the console codepage (#546)

src/keboola_agent_cli/commands/agent.py

@@ -23,7 +23,7 @@
 import typer
 
 from ..errors import ConfigError, ErrorCode
-from ..output import OutputFormatter
+from ..output import OutputFormatter, write_machine_output
 from ..server.agents_store import AgentAction, Trigger
 from ..services.agent_service import AgentService
 from ._helpers import check_cli_permission, get_formatter, get_service
@@ -394,8 +394,11 @@ def _render_stream_event(formatter: OutputFormatter, evt: dict[str, Any]) -> Non
     multi-line summary with exit code, elapsed time, response preview.
     """
     if formatter.json_mode:
-        sys.stdout.write(json.dumps(evt, ensure_ascii=False) + "\n")
-        sys.stdout.flush()
+        # `ensure_ascii=False` keeps the event text readable, which means raw
+        # non-ASCII reaches stdout -- so it must not go through the console's
+        # encoder (issue #546). `write_machine_output` also flushes, which a
+        # stream consumer depends on.
+        write_machine_output(json.dumps(evt, ensure_ascii=False))
         return
     event = evt.get("event", "?")
     data = evt.get("data") or {}

src/keboola_agent_cli/commands/http_client.py

@@ -16,12 +16,12 @@
 from __future__ import annotations
 
 import json
-import sys
 from typing import Any
 
 import typer
 
 from ..constants import HTTP_DEFAULT_TIMEOUT
+from ..output import write_machine_output
 from ..services.http_forwarder_service import (
     ForwardedResponse,
     ForwarderError,
@@ -41,15 +41,20 @@
 def _print_json(_console: Any, data: Any) -> None:
     """Human-mode renderer: pipe-safe pretty JSON.
 
-    Uses ``sys.stdout.write`` instead of ``console.print`` because Rich
-    can soft-wrap long strings or escape markup, which breaks downstream
+    Writes straight to stdout instead of ``console.print`` because Rich can
+    soft-wrap long strings or escape markup, which breaks downstream
     ``json.loads`` consumers. Real-world case: an AI agent piped
     ``kbagent http get /openapi.json`` into ``python3 -c "json.load(sys.stdin)"``
     and hit ``JSONDecodeError`` because Rich had reflowed lines. The output
     of ``kbagent http`` is virtually always parsed by something downstream
     (an LLM, a script, jq) -- so machine-clean stdout is the contract.
+
+    Through :func:`write_machine_output` so that contract also survives a
+    non-UTF-8 console (issue #546). ``json.dumps`` escapes non-ASCII under its
+    ``ensure_ascii`` default, so this path cannot crash today; it uses the
+    helper so it stays correct if that default is ever dropped.
     """
-    sys.stdout.write(json.dumps(data, indent=2) + "\n")
+    write_machine_output(json.dumps(data, indent=2))
 
 
 def _resolve_service(ctx: typer.Context) -> HttpForwarderService:

src/keboola_agent_cli/output.py

@@ -14,6 +14,33 @@
 from .models import ErrorResponse, SuccessResponse
 
 
+def write_machine_output(text: str) -> None:
+    """Write a machine-readable line to stdout as UTF-8, whatever the console is.
+
+    ``--json`` exists to be piped into another program, so its bytes must not
+    depend on the terminal's active codepage. On a default Czech / Polish /
+    Hungarian Windows console (cp1250), ``sys.stdout.write`` raised
+    ``UnicodeEncodeError`` on any non-ASCII character in the payload -- an arrow
+    in a flow name was enough to make ``kbagent --json flow list`` unusable
+    (issue #546). Pydantic's ``model_dump_json`` emits raw non-ASCII, unlike
+    ``json.dumps``, whose ``ensure_ascii`` default escapes it away.
+
+    Writing through ``sys.stdout.buffer`` bypasses the text layer's encoder
+    entirely, so the codepage never gets a say. The text layer is flushed first
+    so anything already written through it keeps its place in the stream. A
+    stdout with no binary buffer (captured or replaced streams) has no encoder
+    to bypass, so the plain write is already correct there.
+    """
+    payload = text + "\n"
+    buffer = getattr(sys.stdout, "buffer", None)
+    if buffer is None:
+        sys.stdout.write(payload)
+        return
+    sys.stdout.flush()
+    buffer.write(payload.encode("utf-8"))
+    buffer.flush()
+
+
 class OutputFormatter:
     """Formats CLI output as either JSON (for machines/agents) or Rich (for humans).
 
@@ -54,7 +81,7 @@ def output(
         """
         if self.json_mode:
             response = SuccessResponse(status="ok", data=data)
-            sys.stdout.write(response.model_dump_json(indent=2) + "\n")
+            write_machine_output(response.model_dump_json(indent=2))
         else:
             if human_formatter is not None:
                 human_formatter(self.console, data)
@@ -99,7 +126,7 @@ def error(
                 "status": "error",
                 "error": err.model_dump(exclude_none=True),
             }
-            sys.stdout.write(json.dumps(error_envelope, indent=2) + "\n")
+            write_machine_output(json.dumps(error_envelope, indent=2))
         else:
             self.err_console.print(f"[bold red]Error:[/bold red] {message}")
 
@@ -111,7 +138,7 @@ def success(self, message: str) -> None:
         """
         if self.json_mode:
             response = SuccessResponse(status="ok", data={"message": message})
-            sys.stdout.write(response.model_dump_json(indent=2) + "\n")
+            write_machine_output(response.model_dump_json(indent=2))
         else:
             self.console.print(f"[bold green]Success:[/bold green] {message}")
 

tests/test_output.py

@@ -1,10 +1,13 @@
 """Tests for OutputFormatter with JSON and Rich dual mode."""
 
+import io
 import json
 import sys
+from collections.abc import Callable
 from io import StringIO
 from typing import cast
 
+import pytest
 from rich.console import Console
 
 from keboola_agent_cli.output import (
@@ -20,6 +23,7 @@
     format_query_results,
     format_tool_result,
     format_tools_table,
+    write_machine_output,
 )
 
 
@@ -1074,3 +1078,129 @@ def test_falls_back_to_csv_preview_for_full_export(self) -> None:
         assert "Results:" in output
         assert "id,name" in output
         assert "alice" in output
+
+
+class TestMachineOutputIsAlwaysUtf8:
+    """`--json` must not depend on the console codepage (issue #546).
+
+    On a default Czech/Polish/Hungarian Windows console (cp1250) a single
+    non-ASCII character in the data -- an arrow in a flow name was the real
+    report -- made `kbagent --json flow list` raise UnicodeEncodeError instead
+    of printing JSON. The codepage is simulated here, so these run everywhere.
+
+    Only the two pydantic paths reproduce the crash: `model_dump_json` emits
+    raw non-ASCII, whereas `json.dumps` escapes it to `\\uXXXX` under its
+    `ensure_ascii` default. The `error()` test below therefore pins the
+    invariant rather than the bug -- it is what fails if anyone later turns
+    `ensure_ascii` off.
+    """
+
+    ARROW_NAME = "extract → transform"
+
+    @staticmethod
+    def _cp1250_stdout() -> io.TextIOWrapper:
+        """A stdout whose text layer cannot encode the payload, like cp1250."""
+        return io.TextIOWrapper(io.BytesIO(), encoding="cp1250", newline="")
+
+    def _capture(self, monkeypatch: pytest.MonkeyPatch, action: Callable[[], None]) -> bytes:
+        stream = self._cp1250_stdout()
+        monkeypatch.setattr(sys, "stdout", stream)
+        action()
+        stream.flush()
+        return cast(io.BytesIO, stream.buffer).getvalue()
+
+    def test_output_survives_a_codepage_that_cannot_encode_the_data(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        formatter = OutputFormatter(json_mode=True)
+
+        written = self._capture(monkeypatch, lambda: formatter.output({"name": self.ARROW_NAME}))
+
+        # Bytes are UTF-8 and round-trip, rather than being lost or escaped.
+        assert json.loads(written.decode("utf-8"))["data"]["name"] == self.ARROW_NAME
+
+    def test_success_survives_it_too(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        formatter = OutputFormatter(json_mode=True)
+
+        written = self._capture(monkeypatch, lambda: formatter.success(self.ARROW_NAME))
+
+        assert json.loads(written.decode("utf-8"))["data"]["message"] == self.ARROW_NAME
+
+    def test_error_survives_it_too(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """Passes pre-fix too -- `json.dumps` escapes the arrow away.
+
+        Kept as the guard on the invariant: it starts failing the moment the
+        error envelope stops escaping non-ASCII.
+        """
+        formatter = OutputFormatter(json_mode=True)
+
+        written = self._capture(
+            monkeypatch, lambda: formatter.error(self.ARROW_NAME, error_code="ERROR")
+        )
+
+        assert self.ARROW_NAME in json.loads(written.decode("utf-8"))["error"]["message"]
+
+    def test_a_text_stream_without_a_binary_buffer_still_works(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Captured/replaced stdout has no encoder to bypass -- write plainly."""
+        stream = StringIO()
+        monkeypatch.setattr(sys, "stdout", stream)
+
+        write_machine_output('{"ok": true}')
+
+        assert stream.getvalue() == '{"ok": true}\n'
+
+    def test_human_mode_output_written_earlier_keeps_its_place(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Flushing the text layer first stops the two writers reordering."""
+        stream = self._cp1250_stdout()
+        monkeypatch.setattr(sys, "stdout", stream)
+
+        sys.stdout.write("first\n")
+        write_machine_output("second")
+        stream.flush()
+
+        assert cast(io.BytesIO, stream.buffer).getvalue() == b"first\nsecond\n"
+
+
+class TestStreamedAgentEventsAreUtf8:
+    """`--json ... --stream` writes NDJSON of its own, outside OutputFormatter.
+
+    `_render_stream_event` builds each line with `json.dumps(..., ensure_ascii
+    =False)` -- deliberately, so event text stays readable -- which means raw
+    non-ASCII reaches stdout. That is precisely the property that crashes on a
+    cp1250 console, so this path needs the same UTF-8 write as the formatter
+    (issue #546).
+    """
+
+    def test_event_with_non_ascii_does_not_crash_on_a_cp1250_console(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from keboola_agent_cli.commands.agent import _render_stream_event
+
+        stream = io.TextIOWrapper(io.BytesIO(), encoding="cp1250", newline="")
+        monkeypatch.setattr(sys, "stdout", stream)
+        formatter = OutputFormatter(json_mode=True)
+
+        _render_stream_event(formatter, {"event": "log", "data": {"msg": "načítám → hotovo"}})
+        stream.flush()
+
+        written = cast(io.BytesIO, stream.buffer).getvalue()
+        assert json.loads(written.decode("utf-8"))["data"]["msg"] == "načítám → hotovo"
+
+    def test_each_event_is_flushed_so_consumers_see_it_immediately(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A stream consumer reads line by line; buffering would stall it."""
+        from keboola_agent_cli.commands.agent import _render_stream_event
+
+        stream = io.TextIOWrapper(io.BytesIO(), encoding="cp1250", newline="")
+        monkeypatch.setattr(sys, "stdout", stream)
+        formatter = OutputFormatter(json_mode=True)
+
+        _render_stream_event(formatter, {"event": "init", "data": {}})
+
+        # Readable without an explicit flush by the test.
+        assert cast(io.BytesIO, stream.buffer).getvalue().endswith(b"\n")