240 lines
10 KiB
Python
240 lines
10 KiB
Python
|
"""Choosing agent models per launch intent (task 12): the BOARD_AGENT_MODEL
|
||
|
|
settings resolve intent → model in config, travel to the adapter as
|
||
|
|
AGENT_MODEL (absent when empty — never an empty flag value), and each
|
||
|
|
adapter renders the opaque name natively. With nothing set, launches are
|
||
|
|
byte-identical to the inherit-everything behaviour.
|
||
|
|
|
||
|
|
The `run` scripts are exercised end-to-end against stub binaries
|
||
|
|
(BOARD_CLAUDE_BIN / BOARD_OPENCODE_BIN), the same seam a live board uses.
|
||
|
|
Run with: python3 -m unittest discover -s tests
|
||
|
|
"""
|
||
|
|
|
||
|
|
import importlib.util
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import stat
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
REPO = Path(__file__).resolve().parents[1]
|
||
|
|
CORE = REPO / "manager" / "core"
|
||
|
|
CLAUDE = CORE / "adapters" / "claude"
|
||
|
|
OPENCODE = CORE / "adapters" / "opencode"
|
||
|
|
|
||
|
|
sys.path.insert(0, str(CORE))
|
||
|
|
|
||
|
|
import agents # noqa: E402
|
||
|
|
import config # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
def _load(name, path):
|
||
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
||
|
|
module = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(module)
|
||
|
|
return module
|
||
|
|
|
||
|
|
|
||
|
|
permission_config = _load("permission_config", OPENCODE / "permission_config.py")
|
||
|
|
|
||
|
|
# Neutralize any local/.env or shell leakage: process env beats .env, and
|
||
|
|
# an empty value is exactly "nothing configured".
|
||
|
|
UNSET = {"BOARD_AGENT_MODEL": "", "BOARD_AGENT_MODEL_WORK": "",
|
||
|
|
"BOARD_AGENT_MODEL_ACT_PR": "", "BOARD_AGENT_MODEL_REVIEW": ""}
|
||
|
|
|
||
|
|
|
||
|
|
def _resolve(settings: dict) -> dict:
|
||
|
|
"""config.agent_model per intent, in a fresh interpreter so the given
|
||
|
|
settings are what config reads at import."""
|
||
|
|
env = dict(os.environ)
|
||
|
|
env.update(UNSET)
|
||
|
|
env.update(settings)
|
||
|
|
out = subprocess.check_output(
|
||
|
|
[sys.executable, "-c",
|
||
|
|
"import sys; sys.path.insert(0, sys.argv[1]); import config, json; "
|
||
|
|
"print(json.dumps({m: config.agent_model(m) "
|
||
|
|
"for m in ('work', 'act-pr', 'review')}))",
|
||
|
|
str(CORE)],
|
||
|
|
env=env, text=True)
|
||
|
|
return json.loads(out)
|
||
|
|
|
||
|
|
|
||
|
|
class ModelResolution(unittest.TestCase):
|
||
|
|
def test_nothing_set_means_inherit_for_every_intent(self):
|
||
|
|
self.assertEqual(_resolve({}),
|
||
|
|
{"work": "", "act-pr": "", "review": ""})
|
||
|
|
|
||
|
|
def test_the_general_setting_covers_all_intents(self):
|
||
|
|
self.assertEqual(_resolve({"BOARD_AGENT_MODEL": "vendor-x"}),
|
||
|
|
{"work": "vendor-x", "act-pr": "vendor-x",
|
||
|
|
"review": "vendor-x"})
|
||
|
|
|
||
|
|
def test_a_per_intent_setting_beats_the_general_one_for_that_intent_only(self):
|
||
|
|
resolved = _resolve({"BOARD_AGENT_MODEL": "big",
|
||
|
|
"BOARD_AGENT_MODEL_REVIEW": "cheap"})
|
||
|
|
self.assertEqual(resolved,
|
||
|
|
{"work": "big", "act-pr": "big", "review": "cheap"})
|
||
|
|
|
||
|
|
def test_per_intent_alone_leaves_the_others_inheriting(self):
|
||
|
|
resolved = _resolve({"BOARD_AGENT_MODEL_ACT_PR": "pusher"})
|
||
|
|
self.assertEqual(resolved,
|
||
|
|
{"work": "", "act-pr": "pusher", "review": ""})
|
||
|
|
|
||
|
|
|
||
|
|
class LaunchEnv(unittest.TestCase):
|
||
|
|
"""_launch's half of the contract: AGENT_MODEL set iff a model is
|
||
|
|
configured — absent means absent, even against a leaky environment."""
|
||
|
|
|
||
|
|
def setUp(self):
|
||
|
|
self._tmp = tempfile.TemporaryDirectory()
|
||
|
|
tmp = Path(self._tmp.name)
|
||
|
|
self.capture = tmp / "env.json"
|
||
|
|
adapter = tmp / "stub-adapter"
|
||
|
|
adapter.mkdir()
|
||
|
|
run = adapter / "run"
|
||
|
|
run.write_text("#!/usr/bin/env python3\n"
|
||
|
|
"import json, os\n"
|
||
|
|
f"open({str(self.capture)!r}, 'w').write(json.dumps(dict(os.environ)))\n")
|
||
|
|
run.chmod(run.stat().st_mode | stat.S_IXUSR)
|
||
|
|
self._saved = (config.AGENT_MODEL, config.AGENT_MODELS,
|
||
|
|
config.adapter_dir, config.child_env)
|
||
|
|
config.adapter_dir = lambda: adapter
|
||
|
|
# A stray AGENT_MODEL inherited by the board process must not leak.
|
||
|
|
config.child_env = lambda: {"PATH": os.environ.get("PATH", ""),
|
||
|
|
"AGENT_MODEL": "stray-from-the-shell"}
|
||
|
|
|
||
|
|
def tearDown(self):
|
||
|
|
(config.AGENT_MODEL, config.AGENT_MODELS,
|
||
|
|
config.adapter_dir, config.child_env) = self._saved
|
||
|
|
self._tmp.cleanup()
|
||
|
|
|
||
|
|
def _launch_env(self, mode: str) -> tuple[dict, str]:
|
||
|
|
log = Path(self._tmp.name) / "job.log"
|
||
|
|
proc, log_file, model = agents._launch(
|
||
|
|
mode, "do the task", Path(self._tmp.name), "id-1", "t.md", log)
|
||
|
|
proc.wait()
|
||
|
|
log_file.close()
|
||
|
|
return json.loads(self.capture.read_text()), model
|
||
|
|
|
||
|
|
def test_no_model_configured_means_no_variable_at_all(self):
|
||
|
|
config.AGENT_MODEL = ""
|
||
|
|
config.AGENT_MODELS = {"work": "", "act-pr": "", "review": ""}
|
||
|
|
env, model = self._launch_env("work")
|
||
|
|
self.assertNotIn("AGENT_MODEL", env)
|
||
|
|
self.assertEqual(model, "")
|
||
|
|
|
||
|
|
def test_the_resolved_model_arrives_as_agent_model(self):
|
||
|
|
config.AGENT_MODEL = "big"
|
||
|
|
config.AGENT_MODELS = {"work": "", "act-pr": "", "review": "cheap"}
|
||
|
|
env, model = self._launch_env("review")
|
||
|
|
self.assertEqual(env["AGENT_MODEL"], "cheap")
|
||
|
|
self.assertEqual(model, "cheap")
|
||
|
|
env, model = self._launch_env("work")
|
||
|
|
self.assertEqual(env["AGENT_MODEL"], "big")
|
||
|
|
self.assertEqual(model, "big")
|
||
|
|
|
||
|
|
|
||
|
|
class AgentRecord(unittest.TestCase):
|
||
|
|
def test_public_record_carries_the_model_none_means_inherited(self):
|
||
|
|
base = {"id": "a", "task": "t.md", "branch": None, "worktree": None,
|
||
|
|
"status": "running", "rc": None, "started": 0.0,
|
||
|
|
"session": None, "mode": "review", "name": "Wren"}
|
||
|
|
self.assertIsNone(agents._agent_public(base)["model"])
|
||
|
|
self.assertEqual(
|
||
|
|
agents._agent_public({**base, "model": "cheap"})["model"], "cheap")
|
||
|
|
|
||
|
|
|
||
|
|
def _write_stub(directory: Path, name: str, script: str) -> Path:
|
||
|
|
stub = directory / name
|
||
|
|
stub.write_text(script, encoding="utf-8")
|
||
|
|
stub.chmod(stub.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
||
|
|
return stub
|
||
|
|
|
||
|
|
|
||
|
|
class ClaudeRunModel(unittest.TestCase):
|
||
|
|
"""AGENT_MODEL reaches the claude launch as --model; absent (or empty,
|
||
|
|
which must never happen but costs nothing to survive) = no flag."""
|
||
|
|
|
||
|
|
def _run(self, model: str | None) -> list[str]:
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
capture = Path(tmp) / "args.json"
|
||
|
|
stub = _write_stub(Path(tmp), "claude-stub",
|
||
|
|
"#!/usr/bin/env python3\n"
|
||
|
|
"import json, sys\n"
|
||
|
|
f"open({str(capture)!r}, 'w').write(json.dumps(sys.argv[1:]))\n")
|
||
|
|
wrapper = _write_stub(Path(tmp), "bin",
|
||
|
|
f"#!/usr/bin/env bash\nexec python3 {stub} \"$@\"\n")
|
||
|
|
env = dict(os.environ)
|
||
|
|
env.pop("AGENT_MODEL", None)
|
||
|
|
env.update({"BOARD_CLAUDE_BIN": str(wrapper),
|
||
|
|
"AGENT_PROMPT": "do the task", "AGENT_MODE": "work",
|
||
|
|
"AGENT_COMMANDS": "python3 -m unittest"})
|
||
|
|
if model is not None:
|
||
|
|
env["AGENT_MODEL"] = model
|
||
|
|
result = subprocess.run(["bash", str(CLAUDE / "run")], env=env,
|
||
|
|
capture_output=True, text=True)
|
||
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
||
|
|
return json.loads(capture.read_text())
|
||
|
|
|
||
|
|
def test_unset_launches_byte_identical_to_today(self):
|
||
|
|
args = self._run(None)
|
||
|
|
self.assertNotIn("--model", args)
|
||
|
|
self.assertEqual(args, self._run("")) # empty behaves like absent
|
||
|
|
|
||
|
|
def test_set_appends_model_and_changes_nothing_else(self):
|
||
|
|
args = self._run("claude-model-x")
|
||
|
|
i = args.index("--model")
|
||
|
|
self.assertEqual(args[i + 1], "claude-model-x")
|
||
|
|
self.assertEqual(args[:i] + args[i + 2:], self._run(None))
|
||
|
|
|
||
|
|
|
||
|
|
class OpencodeRunModel(unittest.TestCase):
|
||
|
|
"""AGENT_MODEL reaches the opencode launch as the generated config's
|
||
|
|
model key; absent = no key, config byte-identical to today."""
|
||
|
|
|
||
|
|
def _run(self, model: str | None) -> dict:
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
capture = Path(tmp) / "capture.json"
|
||
|
|
stub = _write_stub(Path(tmp), "opencode-stub",
|
||
|
|
"#!/usr/bin/env python3\n"
|
||
|
|
"import json, os\n"
|
||
|
|
f"open({str(capture)!r}, 'w').write("
|
||
|
|
"json.dumps(json.load(open(os.environ['OPENCODE_CONFIG']))))\n")
|
||
|
|
wrapper = _write_stub(Path(tmp), "bin",
|
||
|
|
f"#!/usr/bin/env bash\nexec python3 {stub} \"$@\"\n")
|
||
|
|
env = dict(os.environ)
|
||
|
|
env.pop("AGENT_MODEL", None)
|
||
|
|
env.update({"BOARD_OPENCODE_BIN": str(wrapper),
|
||
|
|
"AGENT_PROMPT": "do the task", "AGENT_MODE": "review",
|
||
|
|
"AGENT_COMMANDS": "python3 -m unittest"})
|
||
|
|
if model is not None:
|
||
|
|
env["AGENT_MODEL"] = model
|
||
|
|
result = subprocess.run(["bash", str(OPENCODE / "run")], env=env,
|
||
|
|
capture_output=True, text=True)
|
||
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
||
|
|
return json.loads(capture.read_text())
|
||
|
|
|
||
|
|
def test_unset_launches_byte_identical_to_today(self):
|
||
|
|
cfg = self._run(None)
|
||
|
|
self.assertNotIn("model", cfg)
|
||
|
|
self.assertEqual(cfg, permission_config.build_config(
|
||
|
|
"review", ["python3 -m unittest"]))
|
||
|
|
|
||
|
|
def test_set_lands_as_the_config_model_key_untranslated(self):
|
||
|
|
cfg = self._run("anthropic/model-x")
|
||
|
|
self.assertEqual(cfg["model"], "anthropic/model-x")
|
||
|
|
del cfg["model"]
|
||
|
|
self.assertEqual(cfg, self._run(None))
|
||
|
|
|
||
|
|
def test_build_config_only_grows_the_key_when_given_a_model(self):
|
||
|
|
commands = ["python3 -m unittest"]
|
||
|
|
self.assertNotIn("model", permission_config.build_config("work", commands))
|
||
|
|
self.assertEqual(
|
||
|
|
permission_config.build_config("work", commands, "p/m")["model"], "p/m")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|