Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/nooa/runtime/producers.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@ async def monitor(cmd: str):
"""Stream stdout lines from a shell command as they appear.

Yields each line (stripped) as it's written to stdout.
Output is decoded as UTF-8 with undecodable bytes replaced, so a
command that emits non-UTF-8 output does not terminate the stream.
stderr is merged into stdout. Uses ``start_new_session=True``
for process-group isolation so multiple concurrent monitors
(and the agent itself) don't contend for ptys or interfere
Expand All @@ -84,7 +86,7 @@ async def monitor(cmd: str):
assert proc.stdout is not None
try:
async for line in proc.stdout:
yield line.decode().rstrip("\n")
yield line.decode("utf-8", errors="replace").rstrip("\n")
await proc.wait()
finally:
if proc.returncode is None:
Expand Down
24 changes: 24 additions & 0 deletions tests/runtime/test_producers.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import asyncio
import os
import platform
import sys

import pytest

Expand Down Expand Up @@ -130,3 +131,26 @@ async def survivor():

await asyncio.wait_for(survivor_task, timeout=5)
assert "survived" in survivor_lines


class TestMonitorOutputDecoding:
"""Verify monitor() tolerates command output that is not valid UTF-8."""

async def test_undecodable_byte_does_not_end_the_stream(self, tmp_path):
"""A non-UTF-8 byte must not kill the producer or drop later lines."""
emitter = tmp_path / "emit.py"
emitter.write_text(
"import sys\n"
"out = sys.stdout.buffer\n"
"out.write(b'first\\n')\n"
"out.write(b'caf\\xe9\\n')\n" # latin-1 'e-acute': invalid UTF-8
"out.write(b'last\\n')\n"
"out.flush()\n",
encoding="utf-8",
)

lines = [line async for line in monitor(f'"{sys.executable}" "{emitter}"')]

# The undecodable byte becomes U+FFFD rather than being dropped, and
# "last" proves the stream continued past it.
assert lines == ["first", "caf\ufffd", "last"]