# AI Triage Monitor Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Replace workflow dispatching with a restart-safe watcher that continuously discovers new failed GitHub Actions jobs, triages each job, and notifies one configured Feishu topic group. **Architecture:** A `Monitor` polls configured workflow targets and inserts unique failure events into SQLite. A staged event state machine separates GitHub discovery, AI triage, report persistence, and Feishu delivery so retries resume at the failed stage. Existing evidence parsing, redaction, ACP validation, classification gates, and report rendering remain the trusted core. **Tech Stack:** Python 3.11 standard library (`sqlite3`, `urllib`, `signal`, `threading`, `tomllib`, `unittest`), GitHub REST API `2026-03-10`, OpenCode ACP, Feishu custom-bot webhook. ## Global Constraints - Do not dispatch, rerun, cancel, or otherwise mutate GitHub Actions. - Do not modify, stop, or deploy to the existing EC2 service. - Add no third-party runtime dependencies. - Never persist raw logs, resolved secrets, GitHub tokens, Feishu webhook URLs, or ACP credentials. - Preserve the non-overridable infrastructure classification gate. - Use event identity `(target_id, run_id, run_attempt, job_id)`. - Treat Feishu delivery as at least once and include a stable event ID in every report and notification. - Run tests before implementation for every behavior change and observe the expected failure. ## File map - `src/ai_triage/models.py`: immutable monitor configuration and GitHub/event value objects. - `src/ai_triage/config.py`: strict TOML parsing for `[monitor]`, `[feishu]`, and `[[targets]]`. - `src/ai_triage/github.py`: read-only workflow-run, current-attempt job, and job-log APIs. - `src/ai_triage/state.py`: SQLite schema and leased stage transitions; no network or filesystem report writes. - `src/ai_triage/orchestrator.py`: convert one claimed failure event into a redacted report and aggregate notification payload. - `src/ai_triage/monitor.py`: bootstrap, discovery, stage execution, retry isolation, continuous loop, and shutdown. - `src/ai_triage/feishu.py`: one job-level topic-group message with stable event ID and links. - `src/ai_triage/report.py`: unchanged serializers invoked with a unique event directory. - `src/ai_triage/cli.py`: `validate` and `watch` command composition. - `README.md`, `config.example.toml`: watcher operation and the committed safe inert example; `config.toml` is an ignored local deployment file. --- ### Task 1: Monitor configuration and models **Files:** - Modify: `ai_triage/src/ai_triage/models.py` - Modify: `ai_triage/src/ai_triage/config.py` - Modify: `ai_triage/tests/test_config.py` - Modify: `ai_triage/config.example.toml` **Interfaces:** - Produces: `MonitorConfig`, `MonitorTarget`, `WorkflowRun.created_at`, `WorkflowRun.updated_at`, `FailureEvent`, and `AppConfig.monitor`/`AppConfig.targets`. - Removes: `EnvReference`, dispatch inputs, schedule keys, `DispatchedRun`, and `select_tasks`/`resolve_task_inputs`. - [ ] **Step 1: Replace configuration fixtures with watcher fields and write failing validation tests** ```python CONFIG = """ [github] owner = "tidbcloud" repo = "tidb-management-service" token_env = "AI_TRIAGE_GITHUB_TOKEN" [monitor] poll_interval_seconds = 60 lookback_seconds = 86400 max_pages_per_target = 10 state_path = "ai-triage-state/state.sqlite3" lease_seconds = 600 retry_base_seconds = 30 retry_max_seconds = 1800 max_attempts = 10 retention_days = 30 [feishu] webhook_env = "AI_TRIAGE_FEISHU_WEBHOOK" [[targets]] id = "dedicated-gcp" enabled = true workflow = "e2e-openapi-dedicated-scenario.yaml" branch = "main" allowed_scenario_steps = ["Run scenario cases"] """ def test_loads_monitor_and_targets(self) -> None: config = self.load(CONFIG) self.assertEqual(60, config.monitor.poll_interval_seconds) self.assertEqual("dedicated-gcp", config.targets[0].id) self.assertEqual("main", config.targets[0].branch) def test_rejects_retry_max_below_retry_base(self) -> None: with self.assertRaisesRegex(ConfigError, "retry_max_seconds"): self.load(CONFIG.replace("retry_max_seconds = 1800", "retry_max_seconds = 10")) ``` - [ ] **Step 2: Run the focused tests and confirm the old schema fails** Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_config -v` Expected: FAIL because `[monitor]` and `[[targets]]` are unknown and `AppConfig.monitor` does not exist. - [ ] **Step 3: Implement the immutable models and strict parser** ```python @dataclass(frozen=True, slots=True) class MonitorConfig: poll_interval_seconds: float = 60.0 lookback_seconds: float = 86_400.0 max_pages_per_target: int = 10 state_path: str = "ai-triage-state/state.sqlite3" lease_seconds: float = 600.0 retry_base_seconds: float = 30.0 retry_max_seconds: float = 1_800.0 max_attempts: int = 10 retention_days: int = 30 @dataclass(frozen=True, slots=True) class MonitorTarget: id: str enabled: bool workflow: str branch: str | None = None allowed_scenario_steps: tuple[str, ...] = ("Run scenario cases",) @dataclass(frozen=True, slots=True) class FailureEvent: event_id: str target_id: str workflow: str run: WorkflowRun job: WorkflowJob | None ``` Parse positive numeric values, require `max_pages_per_target`, `max_attempts`, and `retention_days` to be integers, enforce `retry_max_seconds >= retry_base_seconds`, reject duplicate target IDs, and require at least one target. Keep disabled targets valid but omit them from `enabled_targets(config)`. - [ ] **Step 4: Update the example TOML files and rerun configuration tests** ```toml [monitor] poll_interval_seconds = 60 lookback_seconds = 86400 max_pages_per_target = 10 state_path = "ai-triage-state/state.sqlite3" lease_seconds = 600 retry_base_seconds = 30 retry_max_seconds = 1800 max_attempts = 10 retention_days = 30 [[targets]] id = "dedicated-gcp-staging-smoke" enabled = false workflow = "e2e-openapi-dedicated-scenario.yaml" branch = "main" allowed_scenario_steps = ["Run scenario cases"] ``` Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_config -v` Expected: all configuration tests PASS. - [ ] **Step 5: Commit the configuration slice** ```bash git add ai_triage/src/ai_triage/models.py ai_triage/src/ai_triage/config.py ai_triage/tests/test_config.py ai_triage/config.example.toml git commit -m "refactor: configure workflow monitoring targets" ``` ### Task 2: Read-only GitHub workflow discovery **Files:** - Modify: `ai_triage/src/ai_triage/github.py` - Modify: `ai_triage/tests/test_github.py` **Interfaces:** - Consumes: `WorkflowRun` and `WorkflowJob` from Task 1. - Produces: `GitHubClient.list_completed_runs(workflow, *, branch, created_since, max_pages) -> tuple[WorkflowRun, ...]`. - Retains: `list_jobs(run_id, *, attempt_number)` and `download_job_log(job_id, *, max_bytes)`. - Removes: `dispatch_workflow`, legacy dispatch discovery, `poll_run`, and their mutation-oriented errors. - [ ] **Step 1: Write failing tests for filters, pagination, timestamps, and page bounds** ```python def test_lists_completed_runs_with_branch_and_created_filter(self) -> None: transport = FakeTransport([response(200, { "workflow_runs": [{ "id": 101, "status": "completed", "conclusion": "failure", "run_attempt": 2, "created_at": "2026-07-18T01:00:00Z", "updated_at": "2026-07-18T02:00:00Z", }] })]) runs = self.client(transport).list_completed_runs( "scenario.yaml", branch="main", created_since=datetime(2026, 7, 17, tzinfo=timezone.utc), max_pages=3, ) self.assertEqual(101, runs[0].id) self.assertIn("status=completed", transport.requests[0].url) self.assertIn("branch=main", transport.requests[0].url) self.assertIn("created=%3E%3D2026-07-17T00%3A00%3A00Z", transport.requests[0].url) def test_raises_when_page_cap_cannot_prove_exhaustion(self) -> None: with self.assertRaises(GitHubPageLimitError): self.client(FakeTransport([response(200, {"workflow_runs": [run(i) for i in range(100)]})])).list_completed_runs( "scenario.yaml", branch=None, created_since=NOW, max_pages=1 ) ``` - [ ] **Step 2: Run the GitHub tests and confirm discovery is missing** Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_github -v` Expected: FAIL because `list_completed_runs` and `GitHubPageLimitError` do not exist. - [ ] **Step 3: Implement bounded read-only discovery and delete dispatch code** ```python def list_completed_runs( self, workflow: str, *, branch: str | None, created_since: datetime, max_pages: int, ) -> tuple[WorkflowRun, ...]: workflow_id = urllib.parse.quote(workflow, safe="") runs: list[WorkflowRun] = [] for page in range(1, max_pages + 1): query_values = { "status": "completed", "created": f">={_format_timestamp(created_since)}", "per_page": 100, "page": page, } if branch: query_values["branch"] = branch payload = self._expect(self._request( "GET", f"/actions/workflows/{workflow_id}/runs?{urllib.parse.urlencode(query_values)}", ), {200}) raw_runs = payload.get("workflow_runs", []) runs.extend(_workflow_run(item) for item in raw_runs if isinstance(item, dict)) if len(raw_runs) < 100: return tuple(runs) raise GitHubPageLimitError(f"workflow {workflow} exceeded {max_pages} pages") ``` Delete the POST dispatch path, legacy snapshot/discovery methods, exact-run polling method, and tests that assert dispatch behavior. Keep transient retries restricted to GET. - [ ] **Step 4: Run all GitHub client tests** Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_github -v` Expected: all read-only discovery, job pagination, safe log redirect, response limits, and transient retry tests PASS. - [ ] **Step 5: Commit the GitHub slice** ```bash git add ai_triage/src/ai_triage/github.py ai_triage/tests/test_github.py git commit -m "refactor: discover completed workflow runs" ``` ### Task 3: SQLite event state machine **Files:** - Create: `ai_triage/src/ai_triage/state.py` - Create: `ai_triage/tests/test_state.py` **Interfaces:** - Consumes: `FailureEvent` from Task 1. - Produces: `StateStore.initialize_target`, `discover`, `claim_next`, `complete_triage`, `complete_report`, `complete_notification`, `fail_stage`, and `prune`. - Produces immutable `ClaimedEvent(event, stage, report, notification, report_directory, attempt_count)`. - [ ] **Step 1: Write failing tests for baseline, uniqueness, leases, retries, and payload resumption** ```python def test_discovery_is_unique_per_run_attempt_and_job(self) -> None: store = self.store() event = failure_event(run_id=10, attempt=2, job_id=30) self.assertTrue(store.discover(event, NOW)) self.assertFalse(store.discover(event, NOW)) self.assertTrue(store.discover(failure_event(run_id=10, attempt=3, job_id=31), NOW)) def test_notification_retry_keeps_triage_payload(self) -> None: store = self.store() store.discover(failure_event(), NOW) claim = store.claim_next("worker-a", NOW, lease_seconds=60) store.complete_triage(claim.event.event_id, "worker-a", {"incidents": []}, {"event_id": claim.event.event_id}, "target/run/job", NOW) report_claim = store.claim_next("worker-a", NOW, lease_seconds=60) store.complete_report(report_claim.event.event_id, "worker-a", NOW) notify_claim = store.claim_next("worker-a", NOW, lease_seconds=60) store.fail_stage(notify_claim.event.event_id, "worker-a", "timeout token=secret", NOW, retry_base_seconds=10, retry_max_seconds=60, max_attempts=3) resumed = store.claim_next("worker-b", NOW + timedelta(seconds=10), lease_seconds=60) self.assertEqual("notify", resumed.stage) self.assertEqual({"incidents": []}, resumed.report) self.assertNotIn("secret", store.last_error(resumed.event.event_id)) ``` - [ ] **Step 2: Run state tests and confirm the module is absent** Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_state -v` Expected: ERROR importing `ai_triage.state`. - [ ] **Step 3: Implement schema creation and transaction boundaries** ```sql CREATE TABLE IF NOT EXISTS targets ( target_id TEXT PRIMARY KEY, initialized_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS events ( event_id TEXT PRIMARY KEY, target_id TEXT NOT NULL, event_json TEXT NOT NULL, status TEXT NOT NULL, resume_stage TEXT NOT NULL, attempt_count INTEGER NOT NULL DEFAULT 0, next_retry_at TEXT, lease_owner TEXT, lease_until TEXT, report_json TEXT, notification_json TEXT, report_directory TEXT, last_error TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, notified_at TEXT ); CREATE INDEX IF NOT EXISTS events_due ON events(status, next_retry_at, lease_until); ``` Use `BEGIN IMMEDIATE` around compare-and-set claims. Never hold a transaction across caller work. Serialize only redacted dataclass metadata and bounded JSON payloads. - [ ] **Step 4: Implement leased stage transitions and bounded retry** ```python _NEXT_STAGE = {"triage": "report", "report": "notify"} def _retry_delay(attempt: int, base: float, maximum: float) -> float: return min(maximum, base * (2 ** max(0, attempt - 1))) def fail_stage(..., max_attempts: int) -> str: attempts = current_attempt_count + 1 status = "abandoned" if attempts >= max_attempts else "retry_wait" next_retry = None if status == "abandoned" else now + timedelta( seconds=_retry_delay(attempts, retry_base_seconds, retry_max_seconds) ) # Persist redact_text(error), resume_stage, attempts, and next_retry; clear lease. return status ``` Expired `working` leases are claimable at their recorded `resume_stage`. `complete_triage` stores report and notification JSON before releasing the lease to the `report` stage. `complete_report` advances to `notify`. `complete_notification` is terminal. - [ ] **Step 5: Run state tests including two-connection contention** Run: `cd ai_triage && PYTHONPATH=src python3.11 -W error::ResourceWarning -m unittest tests.test_state -v` Expected: all state tests PASS with no leaked connections. - [ ] **Step 6: Commit the state slice** ```bash git add ai_triage/src/ai_triage/state.py ai_triage/tests/test_state.py git commit -m "feat: persist triage monitor events" ``` ### Task 4: Job-level triage and unique reports **Files:** - Modify: `ai_triage/src/ai_triage/orchestrator.py` - Modify: `ai_triage/src/ai_triage/feishu.py` - Modify: `ai_triage/src/ai_triage/report.py` - Modify: `ai_triage/tests/test_orchestrator.py` - Modify: `ai_triage/tests/test_feishu.py` - Modify: `ai_triage/tests/test_report.py` **Interfaces:** - Consumes: `FailureEvent`, existing `TriageEngine`, and GitHub job-log reads. - Produces: `Orchestrator.triage_event(event, target) -> EventTriage(report, notification, relative_directory)`. - Produces: `FeishuWebhookNotifier.send_failure(notification) -> Mapping[str, Any]`. - [ ] **Step 1: Write failing tests for one job-level result and stable event context** ```python def test_triage_event_returns_redacted_job_report_without_notifying(self) -> None: event = failure_event(event_id="dedicated:101:2:201") result = Orchestrator(FakeGitHub(), FakeTriage()).triage_event(event, TARGET) self.assertEqual(event.event_id, result.report["event_id"]) self.assertEqual(201, result.report["job_id"]) 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_feishu_sends_one_aggregate_failure_message(self) -> None: notifier = FeishuWebhookNotifier("https://secret.invalid/hook", transport=transport) notifier.send_failure({ "event_id": "dedicated:101:2:201", "target_id": "dedicated", "workflow": "scenario.yaml", "run_id": 101, "run_attempt": 2, "job_name": "gcp (case.yaml)", "job_url": "https://github.example/jobs/201", "run_url": "https://github.example/runs/101", "incidents": [{"triage": {"category": "product_error", "owner": "owner", "summary": "bad response"}}], }) self.assertIn("dedicated:101:2:201", transport.body_text) ``` - [ ] **Step 2: Run focused tests and confirm the old dispatch orchestration fails** Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_orchestrator tests.test_feishu tests.test_report -v` Expected: FAIL because `triage_event`, `EventTriage`, and `send_failure` do not exist. - [ ] **Step 3: Refactor orchestration to pure event triage** ```python @dataclass(frozen=True, slots=True) class EventTriage: report: dict[str, Any] notification: dict[str, Any] relative_directory: str def triage_event(self, event: FailureEvent, target: MonitorTarget) -> EventTriage: job = event.job incidents = self._triage_failed_steps(job, target.allowed_scenario_steps) if job else [ self._run_level_infrastructure_incident(event.run) ] report = redact_data({ "event_id": event.event_id, "target_id": event.target_id, "workflow": event.workflow, "run_id": event.run.id, "run_attempt": event.run.run_attempt, "run_url": event.run.html_url, "job_id": None if job is None else job.id, "job_name": None if job is None else job.name, "job_url": None if job is None else job.html_url, "conclusion": event.run.conclusion, "incidents": incidents, }) notification = dict(report) return EventTriage(report, notification, event_relative_directory(event)) ``` Delete dispatch, poll, lifecycle notification, controller-level dispatch error, and global fixed-output behavior. Keep log-unavailable fallback and deterministic classification enforcement. - [ ] **Step 4: Implement aggregate Feishu rendering and unique report directory safety** ```python def send_failure(self, value: Mapping[str, Any]) -> dict[str, Any]: event_id = redact_text(str(value.get("event_id", "unknown")), max_length=300) title = f"AI Triage failure: {value.get('target_id', 'unknown')}" text = _failure_event_summary(value, event_id=event_id) return self._send_post(title, text, str(value.get("job_url") or value.get("run_url") or "")) ``` `event_relative_directory` must accept only numeric GitHub IDs and sanitize the target ID to `[A-Za-z0-9._-]`, preventing traversal. Continue writing `triage-report.json` and `triage-summary.md` inside the caller-provided event directory. - [ ] **Step 5: Run focused tests** Run: `cd ai_triage && PYTHONPATH=src python3.11 -W error::ResourceWarning -m unittest tests.test_orchestrator tests.test_feishu tests.test_report -v` Expected: all job-level triage, aggregation, redaction, truncation, escaping, and report tests PASS. - [ ] **Step 6: Commit the event triage slice** ```bash git add ai_triage/src/ai_triage/orchestrator.py ai_triage/src/ai_triage/feishu.py ai_triage/src/ai_triage/report.py ai_triage/tests/test_orchestrator.py ai_triage/tests/test_feishu.py ai_triage/tests/test_report.py git commit -m "refactor: triage one discovered failed job" ``` ### Task 5: Polling monitor and retry loop **Files:** - Create: `ai_triage/src/ai_triage/monitor.py` - Create: `ai_triage/tests/test_monitor.py` **Interfaces:** - Consumes: `GitHubClient.list_completed_runs`, `StateStore`, `Orchestrator.triage_event`, `write_report`, and `FeishuWebhookNotifier.send_failure`. - Produces: `Monitor.scan_once(now=None) -> CycleResult` and `Monitor.run(stop_event) -> None`. - [ ] **Step 1: Write failing bootstrap, rerun, dedup, and target-isolation tests** ```python def test_fresh_store_ignores_old_failure_but_detects_later_failure(self) -> None: github = FakeGitHub(runs=[failed_run(updated_at=NOW - timedelta(minutes=1))]) monitor = self.monitor(github, now=NOW) first = monitor.scan_once(NOW) self.assertEqual(0, first.discovered) github.runs = [failed_run(run_id=102, updated_at=NOW + timedelta(minutes=1))] second = monitor.scan_once(NOW + timedelta(minutes=1)) self.assertEqual(1, second.discovered) self.assertEqual(1, second.notified) def test_restart_does_not_repeat_ai_or_notification(self) -> None: first = self.monitor(state_path=self.state_path) first.scan_once(NOW) second = self.monitor(state_path=self.state_path) second.scan_once(NOW + timedelta(minutes=1)) self.assertEqual(1, self.triage.calls) self.assertEqual(1, self.notifier.calls) def test_failed_rerun_is_a_new_event(self) -> None: self.scan(failed_run(run_id=101, attempt=1, job_id=201)) result = self.scan(failed_run(run_id=101, attempt=2, job_id=301)) self.assertEqual(1, result.discovered) ``` - [ ] **Step 2: Run monitor tests and confirm the module is absent** Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_monitor -v` Expected: ERROR importing `ai_triage.monitor`. - [ ] **Step 3: Implement scan discovery with explicit abnormal conclusions** ```python _ABNORMAL = frozenset({"failure", "cancelled", "timed_out", "action_required", "stale"}) _BENIGN_JOB = frozenset({None, "success", "neutral", "skipped"}) def scan_target(self, target: MonitorTarget, now: datetime) -> int: initialized_at = self.state.initialize_target(target.id, now) runs = self.github.list_completed_runs( target.workflow, branch=target.branch, created_since=now - timedelta(seconds=self.config.lookback_seconds), max_pages=self.config.max_pages_per_target, ) discovered = 0 for run in runs: if run.updated_at < initialized_at or run.conclusion not in _ABNORMAL: continue jobs = self.github.list_jobs(run.id, attempt_number=run.run_attempt) abnormal_jobs = [job for job in jobs if job.conclusion not in _BENIGN_JOB] events = [failure_event(target, run, job) for job in abnormal_jobs] if not events: events = [failure_event(target, run, None)] discovered += sum(self.state.discover(event, now) for event in events) return discovered ``` Target exceptions append sanitized `CycleError` values and do not stop other targets. - [ ] **Step 4: Implement staged work draining without repeated AI** ```python def _drain_due(self, now: datetime) -> tuple[int, int]: triaged = notified = 0 while claim := self.state.claim_next(self.worker_id, now, self.config.lease_seconds): try: if claim.stage == "triage": target = self.targets_by_id[claim.event.target_id] result = self.orchestrator.triage_event(claim.event, target) self.state.complete_triage(claim.event.event_id, self.worker_id, result.report, result.notification, result.relative_directory, now) triaged += 1 elif claim.stage == "report": write_report(self.output_root / claim.report_directory, claim.report, redactor=redact_data) self.state.complete_report(claim.event.event_id, self.worker_id, now) else: self.notifier.send_failure(claim.notification) self.state.complete_notification(claim.event.event_id, self.worker_id, now) notified += 1 except Exception as error: self.state.fail_stage(claim.event.event_id, self.worker_id, str(error), now, retry_base_seconds=self.config.retry_base_seconds, retry_max_seconds=self.config.retry_max_seconds, max_attempts=self.config.max_attempts) return triaged, notified ``` - [ ] **Step 5: Implement continuous wait and graceful stop** ```python def run(self, stop_event: threading.Event) -> None: while not stop_event.is_set(): self.scan_once() jitter = self.random.uniform(0.0, min(5.0, self.config.poll_interval_seconds * 0.1)) stop_event.wait(self.config.poll_interval_seconds + jitter) ``` Tests inject the clock and random source, set the stop event during `wait`, and assert there is no second scan. - [ ] **Step 6: Run all monitor and state tests** Run: `cd ai_triage && PYTHONPATH=src python3.11 -W error::ResourceWarning -m unittest tests.test_monitor tests.test_state -v` Expected: bootstrap, new-failure detection, rerun, restart dedup, retry resumption, page-limit isolation, multi-target isolation, retention, and shutdown tests PASS. - [ ] **Step 7: Commit the polling loop** ```bash git add ai_triage/src/ai_triage/monitor.py ai_triage/tests/test_monitor.py git commit -m "feat: continuously monitor failed workflow jobs" ``` ### Task 6: CLI composition and signal handling **Files:** - Modify: `ai_triage/src/ai_triage/cli.py` - Modify: `ai_triage/tests/test_cli.py` - Modify: `ai_triage/src/ai_triage/__main__.py` **Interfaces:** - Consumes: configuration, `StateStore`, `GitHubClient`, `TriageEngine`, `Orchestrator`, `FeishuWebhookNotifier`, and `Monitor`. - Produces: `ai-triage watch --config PATH [--once] [--dry-run] [--output-dir PATH] [--state-path PATH]`. - [ ] **Step 1: Write failing dry-run, one-cycle, and signal tests** ```python def test_watch_dry_run_has_no_secret_or_external_requirements(self) -> None: code, stdout = self.run_cli(["watch", "--config", str(config), "--dry-run"]) self.assertEqual(0, code) self.assertIn('"targets": ["safe-example"]', stdout) def test_watch_once_returns_nonzero_for_cycle_errors(self) -> None: with patch("ai_triage.cli.Monitor") as monitor_type: monitor_type.return_value.scan_once.return_value = CycleResult(errors=(CycleError("target", "api unavailable"),)) self.assertEqual(1, main(["watch", "--config", str(config), "--once"])) ``` - [ ] **Step 2: Run CLI tests and confirm the old `run --schedule-key` surface fails expectations** Run: `cd ai_triage && PYTHONPATH=src python3.11 -m unittest tests.test_cli -v` Expected: FAIL because the `watch` command is missing. - [ ] **Step 3: Build watcher dependencies only after validation and dry-run exit** ```python watch = subparsers.add_parser("watch", help="poll workflows and triage new failed jobs") watch.add_argument("--config", required=True, type=Path) watch.add_argument("--once", action="store_true") watch.add_argument("--dry-run", action="store_true") watch.add_argument("--output-dir", type=Path) watch.add_argument("--state-path", type=Path) ``` Dry-run prints enabled target IDs, workflows, branches, effective state path, poll interval, and lookback without reading GitHub or Feishu environment variables. Live mode resolves only the GitHub token and Feishu webhook; downstream test secrets no longer exist. - [ ] **Step 4: Install temporary SIGINT/SIGTERM handlers around continuous mode** ```python stop_event = threading.Event() previous = {signal_number: signal.getsignal(signal_number) for signal_number in (signal.SIGINT, signal.SIGTERM)} try: for signal_number in previous: signal.signal(signal_number, lambda *_: stop_event.set()) monitor.run(stop_event) finally: for signal_number, handler in previous.items(): signal.signal(signal_number, handler) ``` `--once` calls `scan_once` directly. Continuous mode exits zero after a graceful signal. Configuration/authentication errors exit 2; a one-cycle result with target or event errors exits 1. - [ ] **Step 5: Run CLI and complete unit tests** Run: `cd ai_triage && PYTHONPATH=src python3.11 -W error::ResourceWarning -m unittest discover -s tests -v` Expected: all tests PASS and no dispatch-oriented tests remain. - [ ] **Step 6: Commit the CLI slice** ```bash git add ai_triage/src/ai_triage/cli.py ai_triage/src/ai_triage/__main__.py ai_triage/tests/test_cli.py git commit -m "feat: expose continuous ai triage watcher" ``` ### Task 7: Operator documentation **Files:** - Modify: `ai_triage/README.md` - Modify: `ai_triage/docs/knowledge/gha-triage-evidence.md` - Modify: `ai_triage/docs/specs/ai-triage/impl_details/task_progress.md` - Create: `ai_triage/docs/specs/ai-triage-monitor/impl_details/task_progress.md` - Create: `ai_triage/docs/specs/ai-triage-monitor/impl_details/implementation_effect.md` **Interfaces:** - Documents: process-manager command, durable SQLite requirement, topic-group webhook behavior, retry semantics, bootstrap, reruns, permissions, and validation. - Removes: the repository-hosted controller/check workflow, scheduled downstream dispatch, schedule keys, downstream inputs/secrets, dispatched/completed lifecycle messages, and fixed report artifacts. - [ ] **Step 1: Keep monitor execution outside GitHub Actions** Do not add a repository GitHub Actions workflow for validation or live execution. Run validation from the repository root and operate the continuous watcher under a process manager with protected host-local configuration. - [ ] **Step 2: Rewrite README operation and security sections in English** Document this repository-root-relative production command: ```bash PYTHONPATH=ai_triage/src python3.11 -m ai_triage watch \ --config ai_triage/config.toml \ --state-path ai_triage/ai-triage-state/state.sqlite3 \ --output-dir ai_triage/ai-triage-results ``` State that a process manager owns restart, the state/report paths require durable local storage, only one logical deployment should share a database, the custom webhook is bound to the destination topic group, first startup ignores historical failures, reruns have independent event IDs, and delivery is at least once. - [ ] **Step 3: Update progress and evidence documents** Create a task table with rows for configuration, GitHub discovery, SQLite state, job triage, monitor loop, CLI, docs, and verification. Mark rows as they pass. Record that no live workflow, ACP provider, Feishu webhook, or EC2 operation was invoked during development. - [ ] **Step 4: Run configuration and documentation checks** Run: ```bash PYTHONPATH=ai_triage/src python3.11 -m ai_triage validate --config ai_triage/config.example.toml PYTHONPATH=ai_triage/src python3.11 -m ai_triage watch --config ai_triage/config.example.toml --dry-run ``` Expected: config is valid; dry-run lists zero enabled targets and performs no external calls. - [ ] **Step 5: Commit integration and docs** ```bash git add ai_triage/README.md ai_triage/docs ai_triage/config.example.toml git commit -m "docs: describe continuous workflow failure monitoring" ``` ### Task 8: Full verification and final cleanup **Files:** - Modify only files required by failing verification. - Update: `ai_triage/docs/specs/ai-triage-monitor/impl_details/implementation_effect.md` **Interfaces:** - Produces fresh evidence for every acceptance criterion. - [ ] **Step 1: Run the complete unit suite with resource warnings promoted** Run: `cd ai_triage && PYTHONPATH=src python3.11 -W error::ResourceWarning -m unittest discover -s tests -v` Expected: all tests PASS. - [ ] **Step 2: Run compile and package checks** Run: ```bash python3.11 -m compileall -q ai_triage/src ai_triage/tests python3.11 -m pip wheel --no-deps --no-build-isolation ./ai_triage -w ai_triage/build/wheels ``` Expected: both commands exit 0 and produce one wheel under the ignored `ai_triage/build/` directory. - [ ] **Step 3: Run whitespace and secret-pattern checks** Run: ```bash git diff --check rg -n --hidden -g '!*.pyc' -g '!ai_triage/tests/fixtures/**' '(https://open\.feishu\.cn/open-apis/bot/v2/hook/[A-Za-z0-9_-]+|gh[pousr]_[A-Za-z0-9_]{20,}|Authorization: Bearer [^$])' ai_triage ``` Expected: diff check exits 0; secret scan prints no real credential or webhook value. - [ ] **Step 4: Prove dispatch code and fixed report overwrite paths are gone** Run: ```bash rg -n 'dispatch_workflow|/dispatches|schedule_key|resolve_task_inputs|DispatchedRun|notify_states.*dispatched' ai_triage/src ai_triage/tests ai_triage/*.toml ``` Expected: no matches. - [ ] **Step 5: Review the diff against every requirement and update evidence** Record exact test counts and command results in `implementation_effect.md`. Explicitly list any validation gap; do not claim a live GitHub or Feishu test unless one was actually run. - [ ] **Step 6: Commit final verification fixes and evidence** ```bash git add ai_triage git commit -m "test: verify ai triage monitor delivery" ``` ### Task 9: Partial AI resilience and compact batch notifications **Files:** - Modify: `ai_triage/src/ai_triage/triage.py` - Modify: `ai_triage/src/ai_triage/oncall.py` - Modify: `ai_triage/src/ai_triage/feishu.py` - Modify: `ai_triage/prompts/triage_knowledge.md` - Modify: focused unit tests and operating documentation - [x] Preserve validated fields from non-empty partial AI JSON and track missing fields without accepting an empty object as analysis. - [x] Default insufficient scenario evidence and invalid AI output to `case_error`; reserve infrastructure classification for the deterministic non-scenario gate and confirmed GitHub Actions faults. - [x] Route unknown components or unresolved product ownership to the Case Owner, while still attempting live oncall resolution for known components. - [x] Render two detailed failures and at most three additional job links, omit empty evidence/recommendation lists, and retain total failure scope. - [x] Run full local verification, independent review, and EC2 deployment verification.