fix(agents): opencode runs as many turns as memory fits, and takes its system prompt as instructions

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 <session_id>`; 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/<task>/` 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/<task>/` 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

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(

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,19 +373,19 @@ 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 += (
             "\n\nReturn only a JSON object matching this schema, with no Markdown or surrounding text: "
             + 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 <session_id>` does. First turn

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")

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"])