"""Covers the full model/effort tuning chain: config catalog -> /agents -> the two task-creating entry points -> the CLI argv each adapter finally builds. """ from pathlib import Path import pytest from fastapi.testclient import TestClient from dual_agent.adapters.base import AgentRequest from dual_agent.adapters.claude import ClaudeAdapter from dual_agent.adapters.codex import CodexAdapter from dual_agent.adapters.fake import FakeAgentAdapter from dual_agent.advisor import Advisor from dual_agent.api import create_app from dual_agent.cli import _binary from dual_agent.domain import Agents, AgentTuning, ModelCatalog from dual_agent.infra.process import ProcessResult, ProcessRunner from dual_agent.persistence import TaskStore from dual_agent.policy import CommandPolicy from dual_agent.services import Orchestrator CATALOG = { "codex": ModelCatalog( models=["", "gpt-5.6-luna"], efforts=["", "low", "high"], default_model="gpt-5.6-luna", ), "claude": ModelCatalog(models=["", "opus"], efforts=["", "high"], default_effort="high"), } # --------------------------------------------------------------------- config catalog def test_agents_catalog_loads_per_provider_from_dual_agent_yml(tmp_path): config = tmp_path / ".dual-agent.yml" config.write_text( "agents:\n" " planner: codex\n" " implementer: claude\n" " catalog:\n" " codex:\n" " models: ['', 'gpt-5.6-luna']\n" " efforts: ['', 'low', 'high']\n" " default_model: gpt-5.6-luna\n" " claude:\n" " models: ['', 'opus']\n" " efforts: ['', 'high']\n", encoding="utf-8", ) policy = CommandPolicy.from_file(config) assert policy.agents.catalog["codex"].models == ["", "gpt-5.6-luna"] assert policy.agents.catalog["codex"].default_model == "gpt-5.6-luna" assert policy.agents.catalog["claude"].efforts == ["", "high"] def test_agents_catalog_rejects_malformed_entry(tmp_path): config = tmp_path / ".dual-agent.yml" config.write_text( "agents:\n catalog:\n codex:\n models: not-a-list\n", encoding="utf-8", ) with pytest.raises(Exception): CommandPolicy.from_file(config) def test_validate_tuning_accepts_catalog_values_rejects_others(): agents = Agents(catalog=CATALOG) agents.validate_tuning("planner", AgentTuning(model="gpt-5.6-luna", effort="low")) # no raise with pytest.raises(ValueError): agents.validate_tuning("planner", AgentTuning(model="not-a-real-model")) with pytest.raises(ValueError): agents.validate_tuning("implementer", AgentTuning(effort="not-a-real-effort")) def test_validate_tuning_empty_choice_always_allowed(): Agents().validate_tuning("planner", AgentTuning()) # no catalog at all, still fine # --------------------------------------------------------------------------- /agents class _FakeAdapter: def __init__(self, model="", effort=""): self.model = model self.effort = effort class _AgentsOnlyOrchestrator: """Just enough surface for the /agents route; no store, no adapter_factory calls.""" def __init__(self, agents_config, planner, implementer, tunable=True): self.agents_config = agents_config self.planner = planner self.implementer = implementer self.adapter_factory = (lambda role, tuning: None) if tunable else None def test_agents_endpoint_has_no_hardcoded_efforts_and_reflects_config(): orchestrator = _AgentsOnlyOrchestrator( Agents(catalog=CATALOG), _FakeAdapter(model="gpt-5.6-luna"), _FakeAdapter() ) client = TestClient(create_app(orchestrator)) body = client.get("/agents").json() assert "efforts" not in body assert body["planner"]["models"] == ["", "gpt-5.6-luna"] assert body["planner"]["efforts"] == ["", "low", "high"] assert body["implementer"]["models"] == ["", "opus"] assert body["implementer"]["default_effort"] == "high" assert body["tunable"] is True def test_agents_endpoint_changes_with_config_alone(): other_catalog = {"codex": ModelCatalog(models=[""], efforts=[""]), "claude": ModelCatalog(models=[""], efforts=[""])} orchestrator = _AgentsOnlyOrchestrator(Agents(catalog=other_catalog), _FakeAdapter(), _FakeAdapter()) body = TestClient(create_app(orchestrator)).get("/agents").json() assert body["planner"]["models"] == [""] assert body["planner"]["efforts"] == [""] # ------------------------------------------------------------- the two entry points class _CountingWorkspace: """Proves a rejected tuning never reaches the workspace, let alone a CLI.""" def __init__(self): self.calls = 0 def create(self, task_id, repo_path): self.calls += 1 destination = Path(repo_path) / "wt" / task_id destination.mkdir(parents=True, exist_ok=True) return destination, "deadbeef" def _build(tmp_path): store = TaskStore(tmp_path / ".dual-agent") workspace = _CountingWorkspace() factory_calls = [] def adapter_factory(role, tuning): factory_calls.append((role, tuning)) return FakeAgentAdapter({}) orchestrator = Orchestrator( store, FakeAgentAdapter({}), FakeAgentAdapter({}), workspace, test_runner=lambda task, path: True, agents=Agents(catalog=CATALOG), adapter_factory=adapter_factory, ) advisor = Advisor(store, orchestrator.planner, orchestrator.implementer, orchestrator) client = TestClient(create_app(orchestrator, advisor)) return client, orchestrator, workspace, factory_calls def test_post_tasks_rejects_unknown_model_before_touching_workspace(tmp_path): client, orchestrator, workspace, factory_calls = _build(tmp_path) response = client.post( "/tasks", json={ "repo_path": str(tmp_path / "repo"), "goal": "do it", "tuning": {"planner": {"model": "not-a-real-model", "effort": ""}}, }, ) assert response.status_code == 400 assert workspace.calls == 0 assert factory_calls == [] tasks_dir = orchestrator.store.root / "tasks" assert not tasks_dir.is_dir() or not list(tasks_dir.iterdir()) def test_post_tasks_accepts_catalog_model_and_effort(tmp_path): client, orchestrator, workspace, factory_calls = _build(tmp_path) response = client.post( "/tasks", json={ "repo_path": str(tmp_path / "repo"), "goal": "do it", "tuning": {"planner": {"model": "gpt-5.6-luna", "effort": "low"}}, }, ) assert response.status_code == 201 assert workspace.calls == 1 def test_post_discussions_rejects_unknown_effort_before_creating_a_discussion(tmp_path): client, orchestrator, workspace, factory_calls = _build(tmp_path) response = client.post( "/discussions", json={ "repo_path": str(tmp_path / "repo"), "opening": "hi", "tuning": {"implementer": {"model": "", "effort": "not-a-real-effort"}}, }, ) assert response.status_code == 400 assert factory_calls == [] discussions_dir = orchestrator.store.root / "discussions" assert not discussions_dir.is_dir() or not list(discussions_dir.iterdir()) def test_post_discussions_accepts_catalog_model(tmp_path): client, orchestrator, workspace, factory_calls = _build(tmp_path) response = client.post( "/discussions", json={ "repo_path": str(tmp_path / "repo"), "opening": "hi", "tuning": {"implementer": {"model": "opus", "effort": ""}}, }, ) assert response.status_code == 201 discussions_dir = orchestrator.store.root / "discussions" assert len(list(discussions_dir.iterdir())) == 1 # ------------------------------------------------------------------------ CLI argv def test_codex_tuning_argv_shape(): adapter = CodexAdapter(model="gpt-5.6-luna", effort="high") assert adapter._tuning() == ["--model", "gpt-5.6-luna", "-c", "model_reasoning_effort=high"] def test_claude_tuning_argv_shape(): adapter = ClaudeAdapter(model="opus", effort="high") assert adapter._tuning() == ["--model", "opus", "--effort", "high"] # ---------------------------------------------------------- catalog default -> CLI def test_binary_options_fall_back_to_catalog_default_when_nothing_chosen(): """A changed default_model/default_effort in .dual-agent.yml must reach the CLI, not just relabel /agents, or the base (non-task-tuned) adapter goes stale.""" policy = CommandPolicy( commands={}, agents=Agents( catalog={"codex": ModelCatalog(models=["", "gpt-6-nova"], efforts=["", "high"], default_model="gpt-6-nova", default_effort="high")} ), ) options = _binary(policy, "codex") assert options["model"] == "gpt-6-nova" assert options["effort"] == "high" def test_binary_options_prefer_an_explicit_choice_over_the_catalog_default(): policy = CommandPolicy( commands={}, agents=Agents(catalog={"codex": ModelCatalog(models=["", "gpt-6-nova", "gpt-7"], default_model="gpt-6-nova")}), ) options = _binary(policy, "codex", AgentTuning(model="gpt-7", effort="")) assert options["model"] == "gpt-7" def _fake_process_run(captured): def run(self, argv, cwd, timeout_s, name=None, on_start=None, stdout_suffix="log"): captured["argv"] = argv stdout = self.log_root / "out.txt" stderr = self.log_root / "err.txt" stdout.write_text("", encoding="utf-8") stderr.write_text("", encoding="utf-8") return ProcessResult("SUCCESS", 0, stdout, stderr, 1) return run def test_codex_adapter_run_sends_model_and_effort_to_the_cli(tmp_path, monkeypatch): captured = {} monkeypatch.setattr(ProcessRunner, "run", _fake_process_run(captured)) adapter = CodexAdapter(log_root=tmp_path / "logs", model="gpt-5.6-luna", effort="high") adapter.run(AgentRequest("run1", "PLAN", tmp_path, "goal")) argv = captured["argv"] assert argv[argv.index("--model") + 1] == "gpt-5.6-luna" assert "model_reasoning_effort=high" in argv def test_claude_adapter_run_sends_model_and_effort_to_the_cli(tmp_path, monkeypatch): captured = {} monkeypatch.setattr(ProcessRunner, "run", _fake_process_run(captured)) adapter = ClaudeAdapter(log_root=tmp_path / "logs", model="opus", effort="high") adapter.run(AgentRequest("run1", "PLAN", tmp_path, "goal")) argv = captured["argv"] assert argv[argv.index("--model") + 1] == "opus" assert argv[argv.index("--effort") + 1] == "high"