fix(agents): give opencode its prompt on stdin, not as an argument

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: <prompt>"). 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 ("<message>\n<stdin>"), 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: <prompt>").
     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()
 
 

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

tst/unit/agents/opencode/test_opencode_bridge_builtin_tools.py

@@ -35,29 +35,32 @@ 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"})
     return calls
 
 
 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)
 
     asyncio.run(opencode_bridge.run_opencode(
         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):

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"