tenant.openagents/omega

1from __future__ import annotations
2
3import unittest
4
5from zed_eval import report
6from zed_eval.report import TrialRecord
7
8
9def trial(
10    task: str,
11    passed: bool | None,
12    *,
13    steps: int,
14    tool_calls: dict[str, int],
15    input_tokens: int = 0,
16    output_tokens: int = 0,
17) -> TrialRecord:
18    return TrialRecord(
19        task_name=task,
20        started_at=None,
21        passed=passed,
22        error_reason=None,
23        input_tokens=input_tokens,
24        output_tokens=output_tokens,
25        step_count=steps,
26        total_tool_calls=sum(tool_calls.values()),
27        tool_calls=dict(tool_calls),
28        has_result=True,
29    )
30
31
32class SliceMetricsTests(unittest.TestCase):
33    def test_means_over_subset(self) -> None:
34        records = [
35            trial("a", True, steps=4, tool_calls={"read_file": 2, "edit_file": 1}),
36            trial("b", True, steps=6, tool_calls={"read_file": 4, "edit_file": 3}),
37        ]
38        metrics = report.slice_metrics(records)
39        self.assertEqual(metrics["n"], 2)
40        self.assertEqual(metrics["mean_steps"], 5.0)
41        self.assertEqual(metrics["mean_tool_calls"], 5.0)
42        self.assertEqual(metrics["mean_tool_calls_by_tool"]["read_file"], 3.0)
43        self.assertEqual(metrics["mean_tool_calls_by_tool"]["edit_file"], 2.0)
44
45    def test_empty_slice(self) -> None:
46        metrics = report.slice_metrics([])
47        self.assertEqual(metrics["n"], 0)
48        self.assertIsNone(metrics["mean_steps"])
49
50
51class ReportConditioningTests(unittest.TestCase):
52    def test_on_success_excludes_failures_and_errors(self) -> None:
53        records = [
54            trial(
55                "pass1", True, steps=3, tool_calls={"read_file": 1}, input_tokens=100
56            ),
57            trial(
58                "fail1", False, steps=9, tool_calls={"read_file": 9}, input_tokens=900
59            ),
60            trial("err1", None, steps=1, tool_calls={"read_file": 1}),
61        ]
62        scored = [r for r in records if r.passed is not None]
63        passing = [r for r in scored if r.passed]
64
65        on_success = report.slice_metrics(passing)
66        overall = report.slice_metrics(scored)
67
68        # Conditioned on success: only the passing trial counts.
69        self.assertEqual(on_success["n"], 1)
70        self.assertEqual(on_success["mean_steps"], 3.0)
71        # Overall (scored) includes the failure, inflating the mean.
72        self.assertEqual(overall["n"], 2)
73        self.assertEqual(overall["mean_steps"], 6.0)
74
75    def test_pass_rate_single_attempt(self) -> None:
76        scored = [
77            trial("a", True, steps=1, tool_calls={}),
78            trial("b", False, steps=1, tool_calls={}),
79            trial("c", True, steps=1, tool_calls={}),
80            trial("d", True, steps=1, tool_calls={}),
81        ]
82        rate, sem, attempts = report.pass_rate_with_sem(scored)
83        self.assertAlmostEqual(rate, 0.75)
84        self.assertIsNone(sem)
85        self.assertEqual(attempts, 1)
86
87
88class TimeoutVerdictTests(unittest.TestCase):
89    def _result(self, exc_type: str) -> dict:
90        return {"exception_info": {"exception_type": exc_type}}
91
92    def test_timeout_excluded_by_default(self) -> None:
93        passed, reason = report.trial_verdict(self._result("AgentTimeoutError"))
94        self.assertIsNone(passed)
95        self.assertEqual(reason, "AgentTimeoutError")
96
97    def test_timeout_counts_as_failure_when_enabled(self) -> None:
98        passed, reason = report.trial_verdict(
99            self._result("AgentTimeoutError"), timeout_is_failure=True
100        )
101        self.assertIs(passed, False)
102        self.assertEqual(reason, "AgentTimeoutError")
103
104    def test_other_exceptions_stay_errored_even_when_enabled(self) -> None:
105        # A genuine harness/infra exception is never a scored failure.
106        passed, _ = report.trial_verdict(
107            self._result("SandboxCreateError"), timeout_is_failure=True
108        )
109        self.assertIsNone(passed)
110
111
112if __name__ == "__main__":
113    unittest.main()
114