diff --git a/src/agent_env/agents/agents/opencode/opencode_bridge.py b/src/agent_env/agents/agents/opencode/opencode_bridge.py index 797a0f1d5..ac3fd86c3 100644 --- a/src/agent_env/agents/agents/opencode/opencode_bridge.py +++ b/src/agent_env/agents/agents/opencode/opencode_bridge.py @@ -458,13 +458,16 @@ async def run_opencode( # has no session_id, so OpenCode mints one, which the caller captures from the # result and passes back on the next turn. # - # The message is a variadic positional and --file is an array flag; if - # --file comes before the message, yargs greedily swallows the message into - # the file array ("File not found: "). So: message first, files last. + # The prompt goes on stdin, not argv: Linux caps one argument at 128 KiB (MAX_ARG_STRLEN), and a longer prompt + # fails exec with E2BIG before the CLI starts. `opencode run` reads stdin when it isn't a TTY and sends it + # unchanged. A positional message it would put before stdin ("\n"), quoted if it has a space, so + # there is none. + # + # Were a positional message added, it must come before --file: the message is a variadic positional and --file + # is an array flag, so yargs would swallow a message after --file into the file array ("File not found: "). cmd = ["opencode", "run", "--format", "json", "--model", f"litellm/{model}", "--dir", workspace] if session_id: cmd += ["-s", session_id] - cmd.append(prompt) for path in (attachment_paths or []): cmd += ["--file", path] @@ -480,7 +483,7 @@ async def run_opencode( proc = await _started(asyncio.create_subprocess_exec( *cmd, - stdin=asyncio.subprocess.DEVNULL, + stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, @@ -489,7 +492,7 @@ async def run_opencode( _RUNNING.add(proc) parser = _EventParser(event_callback) try: - stderr = await asyncio.wait_for(_stream_cli(proc, parser.on_line), timeout=timeout_s) + stderr = await asyncio.wait_for(_stream_cli(proc, prompt.encode(), parser.on_line), timeout=timeout_s) except asyncio.CancelledError: _kill_cli(proc) await _reap(proc) @@ -527,12 +530,25 @@ async def run_opencode( _STDOUT_READ_BYTES = 64 * 1024 -async def _stream_cli(proc, on_line: Callable[[str], None]) -> bytes: - """``proc.communicate()``, handing ``on_line`` each stdout line as it arrives. +async def _stream_cli(proc, stdin_data: bytes, on_line: Callable[[str], None]) -> bytes: + """``proc.communicate(stdin_data)``, handing ``on_line`` each stdout line as it arrives. Lines are split as ``stdout.decode(errors="replace").splitlines()`` splits the whole output, so they are the same lines; a line has no length limit, as with ``communicate``. Returns stderr once the CLI has exited.""" + + async def feed_stdin() -> None: + try: + proc.stdin.write(stdin_data) + await proc.stdin.drain() + except (BrokenPipeError, ConnectionResetError): + # A CLI that exits before reading its input, as on a config it rejects, closes stdin; its stderr says why. + pass + finally: + proc.stdin.close() + + # Fed while stdout is read, so a CLI that prints before it has read its whole input never deadlocks the turn. + stdin_task = asyncio.create_task(feed_stdin()) stderr_task = asyncio.create_task(proc.stderr.read()) try: # A line's pieces, joined once when its newline arrives: re-joining them on every read @@ -548,10 +564,12 @@ async def _stream_cli(proc, on_line: Callable[[str], None]) -> bytes: pieces.append(rest) for line in b"".join(pieces).decode(errors="replace").splitlines(): on_line(line) + await stdin_task stderr = await stderr_task await proc.wait() return stderr finally: + stdin_task.cancel() stderr_task.cancel() diff --git a/tst/unit/agents/opencode/fake_cli.py b/tst/unit/agents/opencode/fake_cli.py index 7506a6de5..b7eea97db 100644 --- a/tst/unit/agents/opencode/fake_cli.py +++ b/tst/unit/agents/opencode/fake_cli.py @@ -1,5 +1,5 @@ -"""A scripted `opencode` process for the bridge tests: fixed stdout, stderr and exit code. ``hang`` -keeps its output open, as a CLI that never finishes.""" +"""A scripted `opencode` process for the bridge tests: fixed stdout, stderr and exit code, and a record of its stdin. +``hang`` keeps its output open, as a CLI that never finishes.""" from __future__ import annotations import asyncio @@ -13,8 +13,24 @@ def _reader(data: bytes, *, eof: bool = True) -> asyncio.StreamReader: return reader +class FakeStdin: + def __init__(self) -> None: + self.data = b"" + self.closed = False + + def write(self, data: bytes) -> None: + self.data += data + + async def drain(self) -> None: + return None + + def close(self) -> None: + self.closed = True + + class FakeCLI: def __init__(self, stdout: bytes = b"", stderr: bytes = b"", returncode: int = 0, *, hang: bool = False): + self.stdin = FakeStdin() self.stdout = _reader(stdout, eof=not hang) self.stderr = _reader(stderr, eof=not hang) self.returncode = returncode diff --git a/tst/unit/agents/opencode/test_opencode_bridge_builtin_tools.py b/tst/unit/agents/opencode/test_opencode_bridge_builtin_tools.py index 2c3a4f0f4..75bee09fe 100644 --- a/tst/unit/agents/opencode/test_opencode_bridge_builtin_tools.py +++ b/tst/unit/agents/opencode/test_opencode_bridge_builtin_tools.py @@ -35,11 +35,12 @@ def _permission(selection: dict | None, servers: dict | None = None) -> dict: @pytest.fixture def spawned(monkeypatch): """Every spawn's argv and environment, each answered by a CLI that finishes its turn.""" - calls: list[tuple[list[str], dict]] = [] + calls: list[tuple[list[str], dict, FakeCLI]] = [] async def _exec(*cmd, **kwargs): - calls.append((list(cmd), kwargs["env"])) - return FakeCLI(json.dumps(_STOP).encode() + b"\n") + cli = FakeCLI(json.dumps(_STOP).encode() + b"\n") + calls.append((list(cmd), kwargs["env"], cli)) + return cli monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) monkeypatch.setattr(opencode_bridge, "_context_xdg", lambda _context: {"XDG_DATA_HOME": "/ctx/share"}) @@ -47,7 +48,8 @@ async def _exec(*cmd, **kwargs): def test_no_selection_leaves_the_config_and_argv_as_they_were(monkeypatch, spawned): - """The whole config and argv of a turn without a selection, as origin/main builds them.""" + """The whole config and argv of a turn without a selection, as origin/main builds them, except that the prompt + goes on stdin rather than as the positional message.""" monkeypatch.setenv("LITELLM_BASE_URL", "https://gateway.example") monkeypatch.delenv("LITELLM_EXTRA_HEADERS", raising=False) @@ -55,9 +57,10 @@ def test_no_selection_leaves_the_config_and_argv_as_they_were(monkeypatch, spawn prompt="go", context_id="ctx", session_id="ses_0", model="m", mcp_servers={"env": {"url": "https://gw/mcp"}}, effort="high", attachment_paths=["/w/a.png"], workspace="/w")) - ((argv, env),) = spawned - assert argv == ["opencode", "run", "--format", "json", "--model", "litellm/m", "--dir", "/w", "-s", "ses_0", "go", + ((argv, env, cli),) = spawned + assert argv == ["opencode", "run", "--format", "json", "--model", "litellm/m", "--dir", "/w", "-s", "ses_0", "--file", "/w/a.png"] + assert cli.stdin.data == b"go" assert env["OPENCODE_CONFIG_CONTENT"] == json.dumps({ "$schema": "https://opencode.ai/config.json", "provider": {"litellm": { @@ -98,9 +101,10 @@ def test_the_selection_reaches_the_cli(spawned): asyncio.run(opencode_bridge.run_opencode(prompt="go", context_id="ctx", builtin_tools=tools)) - ((argv, env),) = spawned + ((argv, env, cli),) = spawned assert json.loads(env["OPENCODE_CONFIG_CONTENT"])["permission"] == {"*": "deny", **_CHECKS, "question": "deny"} - assert argv[-1] == "go" + assert "go" not in argv + assert cli.stdin.data == b"go" def test_turns_get_the_selection_the_agent_config_set(monkeypatch, oc): diff --git a/tst/unit/agents/opencode/test_opencode_bridge_config.py b/tst/unit/agents/opencode/test_opencode_bridge_config.py index a299da749..d66c7fc58 100644 --- a/tst/unit/agents/opencode/test_opencode_bridge_config.py +++ b/tst/unit/agents/opencode/test_opencode_bridge_config.py @@ -97,14 +97,23 @@ def test_the_runs_role_replaces_the_role_a_server_is_registered_with_in_any_case assert "headers" not in without_role["peer_agents"] +class _Spawns(list): + """Each spawn's argv and environment; ``clis`` has the CLI that answered each, with what it read on stdin.""" + + def __init__(self) -> None: + super().__init__() + self.clis: list[FakeCLI] = [] + + @pytest.fixture def spawned(monkeypatch): """Every spawn's argv and environment, each answered by a CLI that finishes its turn.""" - calls: list[tuple[list[str], dict]] = [] + calls = _Spawns() async def _exec(*cmd, **kwargs): calls.append((list(cmd), kwargs["env"])) - return FakeCLI(json.dumps(_STOP).encode() + b"\n") + calls.clis.append(FakeCLI(json.dumps(_STOP).encode() + b"\n")) + return calls.clis[-1] monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) monkeypatch.setattr(opencode_bridge, "_context_xdg", lambda _context: {"XDG_DATA_HOME": "/ctx/share"}) @@ -121,7 +130,7 @@ def test_a_turn_runs_in_its_workspace_with_tls_checks_on(monkeypatch, spawned): attachment_paths=["/w/a.png"], workspace="/w")) ((argv, env),) = spawned - assert argv == ["opencode", "run", "--format", "json", "--model", "litellm/m", "--dir", "/w", "-s", "ses_0", "go", + assert argv == ["opencode", "run", "--format", "json", "--model", "litellm/m", "--dir", "/w", "-s", "ses_0", "--file", "/w/a.png"] assert "NODE_TLS_REJECT_UNAUTHORIZED" not in env and env["XDG_DATA_HOME"] == "/ctx/share" options = json.loads(env["OPENCODE_CONFIG_CONTENT"])["provider"]["litellm"]["options"] @@ -150,8 +159,8 @@ def test_the_system_prompt_reaches_the_model_as_instructions_and_the_output_form prompt="go", context_id="ctx", system_prompt="be brief", output_format={"type": "json_schema", "schema": schema})) ((argv, env),) = spawned - assert argv[-1] == ('go\n\nReturn only a JSON object matching this schema, with no Markdown or ' - 'surrounding text: {"type":"object"}') + assert spawned.clis[0].stdin.data == (b'go\n\nReturn only a JSON object matching this schema, with no Markdown or ' + b'surrounding text: {"type":"object"}') (instructions,) = json.loads(env["OPENCODE_CONFIG_CONTENT"])["instructions"] assert Path(instructions) == opencode_bridge.context_data_dir("ctx") / "system-prompt.md" assert Path(instructions).read_text() == "be brief" diff --git a/tst/unit/agents/opencode/test_opencode_bridge_stdin.py b/tst/unit/agents/opencode/test_opencode_bridge_stdin.py new file mode 100644 index 000000000..bbfd98ca5 --- /dev/null +++ b/tst/unit/agents/opencode/test_opencode_bridge_stdin.py @@ -0,0 +1,215 @@ +"""The prompt goes to `opencode run` on stdin, never argv: Linux caps one argument at 128 KiB (MAX_ARG_STRLEN), and a +longer prompt fails exec with E2BIG before the CLI starts. opencode-ai 1.17.18 reads stdin when it isn't a TTY and, +with no positional message, sends it to the model unchanged.""" + +import asyncio +import json +import shlex + +import pytest + +from tst.unit.agents.opencode.agent import opencode_bridge +from tst.unit.agents.opencode.fake_cli import FakeCLI, FakeStdin + +MAX_ARG_STRLEN = 131072 # Linux single-argument limit (PAGE_SIZE * 32) + +_STOP = { + "type": "step_finish", + "sessionID": "ses_1", + "part": {"type": "step-finish", "reason": "stop"}, +} +_TEXT = { + "type": "text", + "sessionID": "ses_1", + "part": {"type": "text", "text": '{"ok": true}'}, +} + + +def _lines(*events: dict) -> bytes: + return b"".join(json.dumps(event).encode() + b"\n" for event in events) + + +@pytest.fixture +def spawned(monkeypatch): + """Every spawn's argv, keyword arguments and scripted CLI, each CLI finishing its turn.""" + calls: list[dict] = [] + + async def _exec(*cmd, **kwargs): + proc = FakeCLI(_lines(_TEXT, _STOP)) + calls.append({"cmd": list(cmd), "kwargs": kwargs, "proc": proc}) + return proc + + monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) + monkeypatch.setattr(opencode_bridge, "_context_xdg", lambda _context: {}) + return calls + + +def test_a_first_turn_sends_its_prompt_on_stdin_and_not_in_argv(spawned): + result = asyncio.run( + opencode_bridge.run_opencode( + prompt="summarise the filing", context_id="ctx", model="m", workspace="/w" + ) + ) + + (call,) = spawned + assert call["cmd"] == [ + "opencode", + "run", + "--format", + "json", + "--model", + "litellm/m", + "--dir", + "/w", + ] + assert call["kwargs"]["stdin"] is asyncio.subprocess.PIPE + assert ( + call["proc"].stdin.data == b"summarise the filing" and call["proc"].stdin.closed + ) + assert result.error is None and result.session_id == "ses_1" + + +def test_a_resumed_turn_names_its_session_and_sends_its_prompt_on_stdin(spawned): + asyncio.run( + opencode_bridge.run_opencode( + prompt="and the next one", + context_id="ctx", + session_id="ses_0", + model="m", + attachment_paths=["/w/a.png"], + workspace="/w", + ) + ) + + (call,) = spawned + assert call["cmd"] == [ + "opencode", + "run", + "--format", + "json", + "--model", + "litellm/m", + "--dir", + "/w", + "-s", + "ses_0", + "--file", + "/w/a.png", + ] + assert call["proc"].stdin.data == b"and the next one" + + +def test_a_one_mib_prompt_and_its_output_format_reach_stdin_byte_for_byte(spawned): + prompt = " médical record line\n" * (1024 * 1024 // 22) + "\n\n" + schema = { + "type": "object", + "properties": {"ok": {"type": "boolean"}}, + "required": ["ok"], + } + + result = asyncio.run( + opencode_bridge.run_opencode( + prompt=prompt, + context_id="ctx", + session_id="ses_0", + output_format={"type": "json_schema", "schema": schema}, + ) + ) + + (call,) = spawned + assert max(len(arg.encode()) for arg in call["cmd"]) < 4096 + suffix = ( + "\n\nReturn only a JSON object matching this schema, with no Markdown or surrounding text: " + + json.dumps(schema, separators=(",", ":")) + ) + assert call["proc"].stdin.data == (prompt + suffix).encode() + assert result.structured_output == {"ok": True} + + +def test_a_real_cli_reads_a_one_mib_prompt_whole_from_stdin(monkeypatch, tmp_path): + # Exec'd for real: on main the prompt is one argument, over Linux's per-argument limit and macOS's 1 MiB total. + received = tmp_path / "stdin" + stop = shlex.quote(_lines(_STOP).decode()) + fake_opencode = tmp_path / "opencode" + fake_opencode.write_text( + f"#!/bin/sh\ncat > {shlex.quote(str(received))}\nprintf '%s' {stop}\n" + ) + fake_opencode.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:/usr/bin:/bin") + monkeypatch.setattr(opencode_bridge, "_context_xdg", lambda _context: {}) + prompt = "x" * (1024 * 1024 + 7) + + result = asyncio.run( + opencode_bridge.run_opencode( + prompt=prompt, context_id="ctx", workspace=str(tmp_path) + ) + ) + + assert result.error is None, result.error + assert received.read_bytes() == prompt.encode() + + +def test_a_cli_that_exits_before_reading_its_prompt_still_ends_the_turn( + monkeypatch, tmp_path +): + # A CLI that rejects its config exits without reading stdin; the bridge's write then hits a closed pipe. + spawn = asyncio.create_subprocess_exec + + async def cli_that_never_reads(*_cmd, **kwargs): + return await spawn("sh", "-c", "echo 'Error: bad config' >&2; exit 1", **kwargs) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", cli_that_never_reads) + monkeypatch.setattr(opencode_bridge, "_context_xdg", lambda _context: {}) + + result = asyncio.run( + opencode_bridge.run_opencode( + prompt="x" * (4 * 1024 * 1024), context_id="ctx", timeout_s=30 + ) + ) + + assert result.error == "Error: bad config" + + +def test_a_real_cli_that_prints_before_reading_its_prompt_does_not_deadlock( + monkeypatch, tmp_path +): + # The CLI fills its stdout and stderr pipes before it reads stdin. Were the prompt written to completion before + # stdout and stderr were read, both sides would block on a full pipe until the turn timed out. + stop = shlex.quote(_lines(_STOP).decode()) + fake_opencode = tmp_path / "opencode" + fake_opencode.write_text( + "#!/bin/sh\n" + "head -c 300000 /dev/zero | tr '\\0' a; echo\n" + "head -c 300000 /dev/zero | tr '\\0' b >&2\n" + "cat > /dev/null\n" + f"printf '%s' {stop}\n" + ) + fake_opencode.chmod(0o755) + monkeypatch.setenv("PATH", f"{tmp_path}:/usr/bin:/bin") + monkeypatch.setattr(opencode_bridge, "_context_xdg", lambda _context: {}) + + result = asyncio.run( + opencode_bridge.run_opencode( + prompt="x" * (1024 * 1024), + context_id="ctx", + workspace=str(tmp_path), + timeout_s=20, + ) + ) + + assert result.error is None, result.error + + +def test_a_failed_prompt_write_fails_the_turn(spawned, monkeypatch): + # Only a CLI that has closed stdin (broken pipe, connection reset) is tolerated; any other write error means the + # CLI may have got part of the prompt, so the turn must not report success. + async def drain_fails(_self): + raise OSError(5, "Input/output error") + + monkeypatch.setattr(FakeStdin, "drain", drain_fails) + + with pytest.raises(OSError, match="Input/output error"): + asyncio.run(opencode_bridge.run_opencode(prompt="summarise", context_id="ctx")) + + (call,) = spawned + assert call["proc"].stdin.closed