from __future__ import annotations import json import unittest from datetime import datetime, timezone from ai_triage.models import ( FailureEvent, MonitorTarget, WorkflowJob, WorkflowRun, WorkflowStep, ) from ai_triage.orchestrator import Orchestrator SCENARIO_LOG = """ DEBUG: do step, case:case-1, operation:ClusterService_GetCluster, order:2, request_type:http, saved_placeholder:map[token:secret] DEBUG: http request, url:https://api.example.com/v1/clusters/1, method:GET, request_body:{}, resp_header:map[X-Debug-Trace-Id:[trace-1]], resp_body:{"state":"CREATING"}, status_code:200 response is unexpected, expression:{"status_code":200,"cel":"body.state == \\"ACTIVE\\""} do action failed, scenario:sce_cluster.yaml, case:case-1, order:2 """ INTERRUPTED_LOG = """ DEBUG: do step, case:case-export, operation:ExportService_ListExports, order:2, request_type:http, saved_placeholder:map[] DEBUG: http request, url:https://api.example.com/v1beta2/tidbs/1/exports, method:GET, request_body:{}, resp_header:map[], resp_body:{"state":"RUNNING"}, status_code:200 signal: interrupt ##[error]The operation was canceled. Complete job ##[warning]Node.js 20 is deprecated. """ DIRECT_FAILURE_THEN_INTERRUPTED_LOG = """ DEBUG: do step, case:case-export, operation:ExportService_ListExports, order:2, request_type:http, saved_placeholder:map[] DEBUG: http request, url:https://api.example.com/v1beta2/tidbs/1/exports, method:GET, request_body:{}, resp_header:map[], resp_body:{"error":"upstream unavailable"}, status_code:503 response is unexpected, expression:{"status_code":200} signal: interrupt ##[error]The operation was canceled. """ TARGET = MonitorTarget( id="dedicated", enabled=True, workflow="scenario.yaml", branch="main", ) def failure_event( failed_step: str = "Run scenario cases", *, job_conclusion: str = "failure", step_conclusion: str | None = None, steps: tuple[WorkflowStep, ...] | None = None, with_job: bool = True, ) -> FailureEvent: run = WorkflowRun( id=101, status="completed", conclusion="failure", run_attempt=2, html_url="https://github.example/runs/101", created_at=datetime(2026, 7, 18, 1, tzinfo=timezone.utc), updated_at=datetime(2026, 7, 18, 2, tzinfo=timezone.utc), ) job = None if with_job: job = WorkflowJob( id=201, name="nonprod (sce_cluster.yaml)", status="completed", conclusion=job_conclusion, steps=steps if steps is not None else ( WorkflowStep( 13, failed_step, "completed", step_conclusion or job_conclusion, ), ), html_url="https://github.example/jobs/201", ) return FailureEvent.create(TARGET, run, job) class FakeGitHub: def __init__( self, *, fail_log: bool = False, raw_log: str = SCENARIO_LOG, tail_log: str | None = None, ) -> None: self.fail_log = fail_log self.raw_log = raw_log self.tail_log = raw_log if tail_log is None else tail_log self.tail_downloads = 0 self.full_downloads = 0 self.requested_tail_bytes: list[int] = [] @property def log_downloads(self) -> int: return self.tail_downloads + self.full_downloads def download_job_log_tail(self, job_id: int, *, max_bytes: int) -> str: self.tail_downloads += 1 self.requested_tail_bytes.append(max_bytes) if self.fail_log: raise RuntimeError("tail unavailable token=must-not-leak") return self.tail_log def download_job_log(self, job_id: int, *, max_bytes: int) -> str: self.full_downloads += 1 if self.fail_log: raise RuntimeError("log unavailable token=must-not-leak") return self.raw_log class FakeTriage: def __init__(self) -> None: self.allowed: list[bool] = [] self.owners = {"infrastructure": "infra-owner"} def classify(self, evidence: object, scenario_step_allowed: bool) -> dict[str, object]: self.allowed.append(scenario_step_allowed) return { "category": "product_error" if scenario_step_allowed else "infrastructure_error", "confidence": 0.9, "summary": "classified", "root_cause": "cause", "recommendations": ["fix"], "evidence": ["evidence"], "owner": "unassigned", "owner_type": "component" if scenario_step_allowed else "infrastructure", "ai_used": scenario_step_allowed, } class FakeOwnerRouter: def route(self, triage: dict[str, object]) -> dict[str, object]: return { **dict(triage), "owner": "oncall@example.com", "owner_type": "product_oncall", "owner_emails": ["oncall@example.com"], "owner_open_ids": ["ou_oncall"], "oncall_team": "global_control", "oncall_team_display": "SOE - Global & Dedicated", } class OrchestratorTests(unittest.TestCase): def test_triage_event_returns_redacted_job_report_without_notifying(self) -> None: event = failure_event() triage = FakeTriage() result = Orchestrator(FakeGitHub(), triage).triage_event(event, TARGET) self.assertEqual([True], triage.allowed) self.assertEqual(event.event_id, result.report["event_id"]) self.assertEqual(201, result.report["job_id"]) self.assertEqual("product_error", result.report["incidents"][0]["triage"]["category"]) self.assertEqual("https://api.example.com/v1/clusters/1", result.report["incidents"][0]["evidence"]["request_url"]) self.assertEqual(event.event_id, result.notification["event_id"]) self.assertEqual( "dedicated/run-101-attempt-2/job-201", result.relative_directory ) self.assertNotIn("secret", json.dumps(result.report)) def test_product_owner_routing_keeps_mention_id_but_redacts_email(self) -> None: result = Orchestrator( FakeGitHub(), FakeTriage(), owner_router=FakeOwnerRouter(), ).triage_event(failure_event(), TARGET) triage = result.notification["incidents"][0]["triage"] serialized = json.dumps(result.notification) self.assertEqual("product_oncall", triage["owner_type"]) self.assertEqual(["ou_oncall"], triage["owner_open_ids"]) self.assertEqual("SOE - Global & Dedicated", triage["oncall_team_display"]) self.assertNotIn("oncall@example.com", serialized) def test_reads_only_last_max_log_bytes_once(self) -> None: github = FakeGitHub(tail_log=SCENARIO_LOG) result = Orchestrator( github, FakeTriage(), max_log_bytes=2_000_000, ).triage_event(failure_event(), TARGET) self.assertEqual(1, github.tail_downloads) self.assertEqual([2_000_000], github.requested_tail_bytes) self.assertEqual(0, github.full_downloads) self.assertEqual( "sce_cluster.yaml", result.report["incidents"][0]["evidence"]["scenario_yaml"], ) def test_reads_last_max_log_bytes_to_preserve_step_context(self) -> None: class RangeAwareGitHub(FakeGitHub): def download_job_log_tail(self, job_id: int, *, max_bytes: int) -> str: self.tail_downloads += 1 self.requested_tail_bytes.append(max_bytes) return self.raw_log[-max_bytes:] long_retry_log = ( "DEBUG: do step, case:, operation:ClusterService_GetCluster, " "order:1, request_type:http, saved_placeholder:map[]\n" + ("retrying\n" * 20_000) + "\n".join(SCENARIO_LOG.strip().splitlines()[1:]) ) github = RangeAwareGitHub(raw_log=long_retry_log) result = Orchestrator( github, FakeTriage(), max_log_bytes=2_000_000, ).triage_event(failure_event(), TARGET) evidence = result.report["incidents"][0]["evidence"] self.assertEqual([2_000_000], github.requested_tail_bytes) self.assertEqual("ClusterService_GetCluster", evidence["operation"]) self.assertEqual(1.0, evidence["order"]) def test_extracts_related_metadata_for_notification_report(self) -> None: github = FakeGitHub( tail_log=( "DEBUG: do step, case:case-1, operation:ClusterService_GetCluster, order:2, request_type:http, saved_placeholder:map[token:secret]\n" "DEBUG: http request, url:https://api.example.com/v1/clusters/1, method:GET, request_body:{}, resp_header:map[], resp_body:{\"state\":\"CREATING\"}, status_code:200\n" "response is unexpected, expression:{\"status_code\":200,\"cel\":\"body.state == \\\"ACTIVE\\\"\"}\n" "clusterId:c1 resourcePoolId:p1 changefeedId:cf1\n" ) ) result = Orchestrator(github, FakeTriage()).triage_event(failure_event(), TARGET) metadata = result.report["incidents"][0]["evidence"]["related_metadata"] self.assertIn("c1", tuple(metadata["clusterId"])) self.assertEqual(("p1",), tuple(metadata["resourcePoolId"])) self.assertEqual(("cf1",), tuple(metadata["changefeedId"])) def test_uninformative_tail_does_not_trigger_a_second_download(self) -> None: github = FakeGitHub( tail_log=( "Complete job\n" "##[error]Process completed with exit code 1." ), raw_log=SCENARIO_LOG, ) triage = FakeTriage() result = Orchestrator( github, triage, max_log_bytes=2_000_000, ).triage_event(failure_event(), TARGET) self.assertEqual(1, github.tail_downloads) self.assertEqual([2_000_000], github.requested_tail_bytes) self.assertEqual(0, github.full_downloads) self.assertEqual([True], triage.allowed) self.assertIsNone( result.report["incidents"][0]["evidence"]["scenario_yaml"] ) def test_non_scenario_failure_is_forced_to_infrastructure(self) -> None: triage = FakeTriage() result = Orchestrator(FakeGitHub(), triage).triage_event( failure_event("Install mysqlsh"), TARGET ) self.assertEqual([False], triage.allowed) self.assertEqual( "infrastructure_error", result.report["incidents"][0]["triage"]["category"], ) def test_timed_out_step_is_forced_to_infrastructure(self) -> None: triage = FakeTriage() result = Orchestrator(FakeGitHub(), triage).triage_event( failure_event(job_conclusion="timed_out"), TARGET ) self.assertEqual([False], triage.allowed) self.assertEqual( "infrastructure_error", result.report["incidents"][0]["triage"]["category"], ) def test_cancelled_signal_interrupt_uses_ai_but_remains_infrastructure( self, ) -> None: triage = FakeTriage() result = Orchestrator( FakeGitHub(raw_log=INTERRUPTED_LOG), triage, ).triage_event( failure_event( job_conclusion="failure", step_conclusion="cancelled", ), TARGET, ) incident = result.report["incidents"][0] self.assertEqual([True], triage.allowed) self.assertIn("signal: interrupt", incident["evidence"]["error"]) self.assertEqual( "infrastructure_error", incident["triage"]["category"], ) self.assertEqual("infrastructure", incident["triage"]["owner_type"]) self.assertEqual("infra-owner", incident["triage"]["owner"]) self.assertEqual("gha-workflow", incident["triage"]["component"]) self.assertEqual( "sibling-failure-interrupted-job", incident["triage"]["problem_key"], ) self.assertEqual("unknown", incident["triage"]["oncall_team"]) self.assertTrue(incident["triage"]["ai_used"]) def test_direct_failure_evidence_wins_over_later_signal_interrupt(self) -> None: triage = FakeTriage() result = Orchestrator( FakeGitHub(raw_log=DIRECT_FAILURE_THEN_INTERRUPTED_LOG), triage, ).triage_event( failure_event( job_conclusion="failure", step_conclusion="cancelled", ), TARGET, ) incident = result.report["incidents"][0] self.assertEqual([True], triage.allowed) self.assertEqual(503, incident["evidence"]["actual_status"]) self.assertEqual("product_error", incident["triage"]["category"]) self.assertEqual("component", incident["triage"]["owner_type"]) def test_multiple_failed_scenario_steps_invoke_ai_only_once(self) -> None: triage = FakeTriage() steps = ( WorkflowStep( 13, "Run scenario cases", "completed", "failure", ), WorkflowStep( 14, "Run scenario cases", "completed", "failure", ), ) result = Orchestrator(FakeGitHub(), triage).triage_event( failure_event(steps=steps), TARGET, ) self.assertEqual(1, triage.allowed.count(True)) self.assertEqual(2, len(result.report["incidents"])) def test_missing_log_is_redacted_and_still_triaged(self) -> None: result = Orchestrator(FakeGitHub(fail_log=True), FakeTriage()).triage_event( failure_event(), TARGET ) serialized = json.dumps(result.report) self.assertIn("job log unavailable", serialized) self.assertNotIn("must-not-leak", serialized) def test_run_without_failed_job_creates_synthetic_infrastructure_incident(self) -> None: github = FakeGitHub() triage = FakeTriage() result = Orchestrator(github, triage).triage_event( failure_event(with_job=False), TARGET ) self.assertEqual(0, github.log_downloads) self.assertEqual([False], triage.allowed) self.assertIsNone(result.report["job_id"]) self.assertEqual( "infrastructure_error", result.report["incidents"][0]["triage"]["category"], ) self.assertEqual( "dedicated/run-101-attempt-2/run", result.relative_directory ) if __name__ == "__main__": unittest.main()