tenant.openagents/omega

1from __future__ import annotations
2
3import unittest
4
5from zed_eval import benchmarks, config, harness_command
6
7
8def make_run_request(benchmark_id: str, **overrides: object) -> dict[str, object]:
9    benchmark = benchmarks.get_benchmark(benchmark_id)
10    request: dict[str, object] = {
11        "run_id": "run-1",
12        "namespace": "tester",
13        "experiment_name": benchmark_id,
14        "benchmark": benchmarks.benchmark_metadata(benchmark),
15        "volume_name": "agent-evals",
16        "api_secret_name": "agent-evals-llm-providers",
17        "build_id": "bld-test",
18        "agent_model": "anthropic/claude-sonnet-4-6",
19        "judge_preset": benchmark.default_judge or "leaderboard",
20    }
21    request.update(overrides)
22    return request
23
24
25def option_values(command: list[str], option: str) -> list[str]:
26    return [
27        command[index + 1]
28        for index, argument in enumerate(command[:-1])
29        if argument == option
30    ]
31
32
33class BenchmarkResolutionTests(unittest.TestCase):
34    def test_group_expands_to_parts(self) -> None:
35        self.assertEqual(
36            benchmarks.resolve_benchmarks(["swe-atlas"]),
37            ["swe-atlas-qna", "swe-atlas-rf", "swe-atlas-tw"],
38        )
39
40    def test_aliases_and_dedup(self) -> None:
41        self.assertEqual(
42            benchmarks.resolve_benchmarks(["qna", "tb21", "rf,tw"]),
43            [
44                "swe-atlas-qna",
45                "terminal-bench-2.1",
46                "swe-atlas-rf",
47                "swe-atlas-tw",
48            ],
49        )
50
51
52class HarnessCommandTests(unittest.TestCase):
53    def test_terminal_bench_is_harbor_registry_no_judge(self) -> None:
54        command = harness_command.build_harness_command(
55            make_run_request("terminal-bench-2.1"), "/tmp/jobs"
56        )
57        self.assertEqual(command[0], "harbor")
58        self.assertIn("terminal-bench/terminal-bench-2-1", command)
59        self.assertIn("-d", command)
60        # Test-scored benchmarks must not wire up an LLM judge.
61        self.assertFalse(any(arg.startswith("EVAL_MODEL=") for arg in command))
62        self.assertNotIn("--verifier-import-path", command)
63
64    def test_swe_atlas_rf_wires_judge(self) -> None:
65        command = harness_command.build_harness_command(
66            make_run_request("swe-atlas-rf"), "/tmp/jobs"
67        )
68        verifier_env = option_values(command, "--ve")
69        agent_env = option_values(command, "--ae")
70
71        self.assertEqual(command[0], "harbor")
72        self.assertIn("scale-ai/swe-atlas-rf", command)
73        self.assertIn(
74            config.JUDGE_PROXY_VERIFIER_IMPORT_PATH,
75            option_values(command, "--verifier-import-path"),
76        )
77        self.assertTrue(any(arg.startswith("EVAL_MODEL=") for arg in verifier_env))
78        self.assertTrue(
79            any(arg.startswith("ZED_JUDGE_UPSTREAM=") for arg in verifier_env)
80        )
81        self.assertTrue(
82            any(arg.startswith("ZED_JUDGE_AUTH_ENV=") for arg in verifier_env)
83        )
84        self.assertIn("ZED_JUDGE_MAX_TOKENS=8192", verifier_env)
85        self.assertFalse(any(arg.startswith("ZED_JUDGE_") for arg in agent_env))
86
87    def test_deepswe_uses_pier_path_without_cli_allowlist(self) -> None:
88        command = harness_command.build_harness_command(
89            make_run_request("deepswe"), "/tmp/jobs"
90        )
91        self.assertEqual(command[0], "pier")
92        self.assertIn("-p", command)
93        self.assertIn("/tmp/datasets/deepswe/tasks", command)
94        # Pier uses the Pier-native agent class.
95        self.assertIn("zed_eval.pier_agent:ZedPierAgent", command)
96        # The Pier network allowlist is declared by the agent, not the CLI;
97        # `pier run` has no --agent-allow-host option.
98        self.assertNotIn("--agent-allow-host", command)
99
100    def test_harbor_benchmarks_use_harbor_agent(self) -> None:
101        command = harness_command.build_harness_command(
102            make_run_request("swe-atlas-rf"), "/tmp/jobs"
103        )
104        self.assertIn("zed_eval.agent:ZedAgent", command)
105
106    def test_eval_cli_timeout_override(self) -> None:
107        command = harness_command.build_harness_command(
108            make_run_request("swe-atlas-rf", eval_cli_timeout=123), "/tmp/jobs"
109        )
110        self.assertIn("EVAL_CLI_TIMEOUT=123", command)
111
112    def test_extra_env_is_forwarded_to_harness_agent(self) -> None:
113        command = harness_command.build_harness_command(
114            make_run_request(
115                "swe-atlas-rf",
116                extra_env={
117                    "ZED_EVAL_INSTRUCTION_SUFFIX_FILE": "/data/prompts/checklist.md"
118                },
119            ),
120            "/tmp/jobs",
121        )
122
123        self.assertIn(
124            "ZED_EVAL_INSTRUCTION_SUFFIX_FILE=/data/prompts/checklist.md",
125            option_values(command, "--ae"),
126        )
127
128    def test_swe_atlas_tw_uses_path_dataset(self) -> None:
129        command = harness_command.build_harness_command(
130            make_run_request("swe-atlas-tw"), "/tmp/jobs"
131        )
132        self.assertIn("-p", command)
133        self.assertIn("/tmp/datasets/swe-atlas-tw/data/tw", command)
134
135
136if __name__ == "__main__":
137    unittest.main()
138