From 3a3d89ef8004b4b3192863dd5fab63701f5ce8c9 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Wed, 28 Oct 2026 13:55:36 -0700 Subject: [PATCH] fix(agents): opencode runs as many turns as memory fits, and takes its system prompt as instructions Each `opencode run` peaks near 900 MB, so a modal_vm (8 GB, no swap) running every concurrent turn at once ran out of memory, and its kernel killed one: "opencode exited with code -9", which failed the A2A validator's mcp probe on whichever turn it hit. Turns past the cap (memory, or the cgroup's limit, past 1 GB, at 900 MB a turn; OPENCODE_MAX_CONCURRENT_TURNS sets it) now wait for one to finish. A system prompt prepended to the message read as the user's own words: the validator's "respond with this token" system prompt, still set when its snapshot probe planted a token, made the model take the planted turn for a prompt injection and refuse to recall it. It's now written to the context's data directory and named in the config's `instructions`, which opencode puts in the system message. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../agents/agents/opencode/README.md | 13 ++-- .../agents/agents/opencode/a2a_server.py | 61 +++++++++++++------ .../agents/agents/opencode/opencode_bridge.py | 23 +++++-- .../agents/opencode/test_message_contract.py | 37 ++++++++++- .../opencode/test_opencode_bridge_config.py | 19 +++++- 5 files changed, 122 insertions(+), 31 deletions(-) diff --git a/src/agent_env/agents/agents/opencode/README.md b/src/agent_env/agents/agents/opencode/README.md index b087b08f1..a630fcfea 100644 --- a/src/agent_env/agents/agents/opencode/README.md +++ b/src/agent_env/agents/agents/opencode/README.md @@ -7,11 +7,16 @@ context: the protocol package comes from PyPI (`requirements.txt`), Node is pinn Each task runs `opencode run --format json` once in `/app`. A context's first task starts an OpenCode session, and each later task resumes it with `-s `; each context keeps its sessions in a data directory of its own, under -`$HOME/sessions/`. Text parts are the prompt, after the task's `system_prompt`, and file parts are written under -`/app/inputs//` and attached with `--file`; a PDF goes as its pages' PNGs, rendered with `pdftoppm` where it is -installed. Turns of different contexts run at once, and a context's turns one at a time. +`$HOME/sessions/`. Text parts are the prompt, and file parts are written under `/app/inputs//` and attached with +`--file`; a PDF goes as its pages' PNGs, rendered with `pdftoppm` where it is installed. Turns of different contexts run +at once, as many as memory fits at about 900 MB each past the first 1 GB (`OPENCODE_MAX_CONCURRENT_TURNS` sets the +number), and a context's turns one at a time. -Each run's settings reach the CLI as `OPENCODE_CONFIG_CONTENT`, so nothing is written for it on disk: +Each run's settings reach the CLI as `OPENCODE_CONFIG_CONTENT`, so nothing but the system prompt is written for it on +disk: + +- the task's `system_prompt`, written to `system-prompt.md` in the context's data directory and named in + `instructions`, so it reaches the model as system instructions rather than as the user's words; - the model endpoint, `$LITELLM_BASE_URL/v1`, as an OpenAI-compatible provider with its key from `LITELLM_API_KEY`, `[model.headers]` (`LITELLM_EXTRA_HEADERS`) and the run's `project_id` and `task_id` as `x-litellm-customer-id` and diff --git a/src/agent_env/agents/agents/opencode/a2a_server.py b/src/agent_env/agents/agents/opencode/a2a_server.py index 898b3e355..6bd5b93b3 100644 --- a/src/agent_env/agents/agents/opencode/a2a_server.py +++ b/src/agent_env/agents/agents/opencode/a2a_server.py @@ -20,6 +20,7 @@ import subprocess import tempfile import uuid +from collections.abc import Mapping from contextlib import asynccontextmanager from pathlib import Path from typing import Any @@ -265,6 +266,30 @@ async def lifespan(app: Starlette): _INSTALL_COMMANDS_HOST = [_UNPACK, _ENDPOINT + f"bash {{work_dir}}/agent/install.sh host {{agent_name}} {{a2a_port}} {_HOST_WORKDIR}"] +# Each `opencode run` peaks near 900 MB, and the server, its bridge and the OS need about 1 GB more. A VM that ran +# every concurrent turn at once ran out of memory and its kernel killed one (exit -9), so turns past the cap wait. +TURN_MEMORY_MB, RESERVED_MEMORY_MB = 900, 1024 + + +def max_concurrent_turns(environ: Mapping[str, str] = os.environ) -> int: + """How many `opencode run` processes may run at once: OPENCODE_MAX_CONCURRENT_TURNS, else as many as memory fits.""" + if configured := environ.get("OPENCODE_MAX_CONCURRENT_TURNS"): + return max(1, int(configured)) + return max(1, (memory_mb() - RESERVED_MEMORY_MB) // TURN_MEMORY_MB) + + +def memory_mb(cgroup_limit: Path = Path("/sys/fs/cgroup/memory.max")) -> int: + """The memory this server's processes may use: the machine's, or its cgroup's limit when that's lower.""" + total = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") + try: + limit = cgroup_limit.read_text().strip() + except OSError: + limit = "max" + if limit.isdigit(): + total = min(total, int(limit)) + return total // 2**20 + + @a2a_agent( identity=AgentIdentity( name="OpenCode", @@ -305,6 +330,7 @@ class OpenCodeAgent(DurableSessionRefs, AgentEnvAgent): """Bridges A2A tasks onto `opencode run`.""" def __init__(self) -> None: + self._turn_slots = asyncio.Semaphore(max_concurrent_turns()) self._turn_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary() self._changelog: dict[str, Any] | None = None self._changelog_lock = asyncio.Lock() @@ -337,23 +363,24 @@ def record(event: dict) -> None: except FilePartMaterializeError as exc: return TaskResult.failure("filepart_error", str(exc), error_type="infra_error") try: - result = await run_opencode( - prompt=prompt, - context_id=context_id, - session_id=session_id, - model=request.config.model, - effort=request.config.effort, - mcp_servers=dict(request.mcp_servers), - system_prompt=request.config.system_prompt, - output_format=request.config.output_format, - timeout_s=request.config.timeout_seconds, - project_id=request.config.project_id, - task_id=request.config.task_id, - role=request.config.role, - attachment_paths=attachments, - event_callback=record, - workspace=str(WORKSPACE_DIR), - ) + async with self._turn_slots: + result = await run_opencode( + prompt=prompt, + context_id=context_id, + session_id=session_id, + model=request.config.model, + effort=request.config.effort, + mcp_servers=dict(request.mcp_servers), + system_prompt=request.config.system_prompt, + output_format=request.config.output_format, + timeout_s=request.config.timeout_seconds, + project_id=request.config.project_id, + task_id=request.config.task_id, + role=request.config.role, + attachment_paths=attachments, + event_callback=record, + workspace=str(WORKSPACE_DIR), + ) except Exception: logger.exception("OpenCode invocation failed") return TaskResult.failure( diff --git a/src/agent_env/agents/agents/opencode/opencode_bridge.py b/src/agent_env/agents/agents/opencode/opencode_bridge.py index 665b51f8b..6d150bfa9 100644 --- a/src/agent_env/agents/agents/opencode/opencode_bridge.py +++ b/src/agent_env/agents/agents/opencode/opencode_bridge.py @@ -203,6 +203,7 @@ def _build_config( role: str | None, effort: str | None = None, environ: Mapping[str, str] = os.environ, + instructions: list[str] | None = None, ) -> dict: """Assemble the inline OpenCode config passed via OPENCODE_CONFIG_CONTENT. @@ -254,6 +255,8 @@ def _build_config( "permission": {"*": "allow", "question": "deny"}, "autoupdate": False, } + if instructions: + config["instructions"] = list(instructions) if mcp_servers: mcp: dict[str, dict] = {} @@ -267,6 +270,14 @@ def _build_config( return config +def _system_prompt_file(context_id: str, system_prompt: str) -> Path: + """The conversation's system prompt as the file its config's ``instructions`` name, written for this turn.""" + path = context_data_dir(context_id) / "system-prompt.md" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(system_prompt) + return path + + _PROCESS_CLEANUP_TIMEOUT_S = 5 _RUNNING: set = set() @@ -362,11 +373,10 @@ async def run_opencode( if attribution: _record_attribution(project_id, task_id) - # OpenCode's `run` has no system-prompt flag and its config `instructions` - # field takes file paths (not inline text), so we prepend the caller's - # system prompt to the message. The model follows it either way. - if system_prompt: - prompt = f"{system_prompt}\n\n{prompt}" + # OpenCode's `run` has no system-prompt flag, and its config `instructions` takes files, which reach the model as + # system instructions. Prepended to the message instead, a system prompt reads as the user's own words, and a + # model can take an instruction there for a prompt injection and refuse it. + instructions = [str(_system_prompt_file(context_id, system_prompt))] if system_prompt else None if output_format is not None: schema = validate_output_format(output_format) prompt += ( @@ -374,7 +384,8 @@ async def run_opencode( + json.dumps(schema, separators=(",", ":")) ) - config = _build_config(model, mcp_servers, provider_headers(os.environ, attribution), role, effort) + config = _build_config(model, mcp_servers, provider_headers(os.environ, attribution), role, effort, + instructions=instructions) # Resume a specific session by id (multi-turn continuity). `--continue` # does NOT replay history to the model; `-s ` does. First turn diff --git a/tst/unit/agents/opencode/test_message_contract.py b/tst/unit/agents/opencode/test_message_contract.py index 48aa4ee10..6a0acb574 100644 --- a/tst/unit/agents/opencode/test_message_contract.py +++ b/tst/unit/agents/opencode/test_message_contract.py @@ -10,7 +10,7 @@ from agentenv_protocol.a2a_agent import Uploaded from tst.unit.agents.opencode.agent import a2a_server, opencode_kit -from tst.unit.agents.opencode.conftest import result +from tst.unit.agents.opencode.conftest import Harness, result class Turns: @@ -129,6 +129,41 @@ async def imported(_context, _path): assert [call["session_id"] for call in turns.calls] == [None, "ses_imported"] +@pytest.mark.asyncio +async def test_turns_past_the_cap_wait_for_one_to_finish(monkeypatch, turns): + """Each `opencode run` peaks near 900 MB, and a VM that ran every concurrent turn at once had its kernel kill one.""" + monkeypatch.setenv("OPENCODE_MAX_CONCURRENT_TURNS", "2") + oc = Harness(a2a_server.OpenCodeChangelogAgent()) + turns.release = asyncio.Event() + async with oc.client() as client: + sent = [asyncio.create_task(oc.send(client, "go", f"ctx-{n}")) for n in range(3)] + await _until(lambda: len(turns.calls) == 2) + await asyncio.sleep(0.1) + assert len(turns.calls) == 2 + turns.release.set() + tasks = await asyncio.gather(*sent) + + assert [task["status"]["state"] for task in tasks] == ["completed"] * 3 + + +def test_the_cap_is_what_memory_fits_past_what_the_server_needs(monkeypatch): + monkeypatch.setattr(a2a_server, "memory_mb", lambda: 8192) + assert a2a_server.max_concurrent_turns({}) == 7 + monkeypatch.setattr(a2a_server, "memory_mb", lambda: 1500) + assert a2a_server.max_concurrent_turns({}) == 1 + assert a2a_server.max_concurrent_turns({"OPENCODE_MAX_CONCURRENT_TURNS": "3"}) == 3 + assert a2a_server.max_concurrent_turns({"OPENCODE_MAX_CONCURRENT_TURNS": "0"}) == 1 + + +def test_a_cgroup_limit_below_the_machines_memory_is_the_memory(tmp_path): + machine = a2a_server.memory_mb(tmp_path / "no-cgroup") + limit = tmp_path / "memory.max" + limit.write_text("max\n") + assert a2a_server.memory_mb(limit) == machine + limit.write_text(f"{512 * 2**20}\n") + assert a2a_server.memory_mb(limit) == min(512, machine) + + @pytest.mark.asyncio async def test_a_turn_gets_the_sdks_peer_tools_with_its_registered_servers(monkeypatch, oc, turns): monkeypatch.setenv("A2A_PORT", "8000") diff --git a/tst/unit/agents/opencode/test_opencode_bridge_config.py b/tst/unit/agents/opencode/test_opencode_bridge_config.py index 29a9f5cb1..a299da749 100644 --- a/tst/unit/agents/opencode/test_opencode_bridge_config.py +++ b/tst/unit/agents/opencode/test_opencode_bridge_config.py @@ -9,6 +9,7 @@ """ import asyncio import json +from pathlib import Path import pytest @@ -140,12 +141,24 @@ def test_a_turn_with_no_model_of_its_own_runs_the_default(monkeypatch, spawned): assert json.loads(env["OPENCODE_CONFIG_CONTENT"])["model"] == "litellm/provider/fallback" -def test_the_system_prompt_and_the_output_format_lead_and_follow_the_message(spawned): +def test_the_system_prompt_reaches_the_model_as_instructions_and_the_output_format_follows_the_message(spawned): + """Prepended to the message, a system prompt read as the user's own words, which a model can refuse as an + injection.""" schema = {"type": "object"} asyncio.run(opencode_bridge.run_opencode( prompt="go", context_id="ctx", system_prompt="be brief", output_format={"type": "json_schema", "schema": schema})) - ((argv, _env),) = spawned - assert argv[-1] == ('be brief\n\ngo\n\nReturn only a JSON object matching this schema, with no Markdown or ' + ((argv, env),) = spawned + assert argv[-1] == ('go\n\nReturn only a JSON object matching this schema, with no Markdown or ' '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" + + +def test_a_turn_with_no_system_prompt_names_no_instructions(spawned): + asyncio.run(opencode_bridge.run_opencode(prompt="go", context_id="ctx")) + + ((_argv, env),) = spawned + assert "instructions" not in json.loads(env["OPENCODE_CONFIG_CONTENT"])