fix(awooop): 僅在 source link 過期時刷新 canary

apps/api/tests/test_awooop_source_correlation_apply_smoke.py

@@ -5,6 +5,7 @@ import io
 import json
 import sys
 import urllib.error
+from datetime import datetime, timezone
 from types import SimpleNamespace
 from pathlib import Path
 
@@ -246,3 +247,75 @@ def test_http_json_does_not_retry_post_502(monkeypatch) -> None:
     else:
         raise AssertionError("POST 502 should fail without retry")
     assert len(calls) == 1
+
+
+def test_run_uses_fresh_existing_apply_and_keeps_current_canary_as_refresh_candidate(
+    monkeypatch,
+) -> None:
+    current_work_item_id = "source-evidence:sentry:upstream_canary:gitea-cd-4299-1"
+    current_provider_event_id = "sentry:upstream_canary:gitea-cd-4299-1"
+    calls: list[tuple[str, str]] = []
+
+    recurrence = {
+        "items": [
+            {
+                "work_item": {
+                    "kind": "source_correlation_review",
+                    "work_item_id": current_work_item_id,
+                },
+                "latest_provider_event_id": current_provider_event_id,
+                "alertname": "AwoooPSourceLinkCanary",
+                "source_correlation_review": None,
+                "source_correlation_apply": None,
+            }
+        ],
+    }
+    status_chain = {
+        "source_refs": {
+            "correlation": {
+                "verification_status": "applied_link_verified",
+                "applied_link_total": 3,
+                "latest_applied_link_at": datetime.now(timezone.utc).isoformat(),
+                "top_candidates": [
+                    {
+                        "link_state": "applied",
+                        "provider_event_id": "sentry:source_correlation_linked:old",
+                    }
+                ],
+            }
+        }
+    }
+
+    def fake_http_json(url: str, *, method: str = "GET", **_: object) -> dict[str, object]:
+        calls.append((method, url))
+        if method != "GET":
+            raise AssertionError("fresh existing apply should not POST review/apply")
+        if "status-chain" in url:
+            return status_chain
+        return recurrence
+
+    monkeypatch.setattr(awooop_source_correlation_apply_smoke, "_http_json", fake_http_json)
+
+    result = awooop_source_correlation_apply_smoke.run(
+        awooop_source_correlation_apply_smoke.parse_args(
+            [
+                "--api-url",
+                "https://awoooi.wooo.work",
+                "--target-incident-id",
+                "INC-20260505-25E744",
+                "--work-item-id",
+                current_work_item_id,
+                "--expected-source-event-provider-event-id",
+                "sentry:source_correlation_linked:gitea-cd-4299-1",
+                "--allow-existing-apply",
+                "--refresh-if-stale-days",
+                "6",
+                "--verify-refresh-candidate",
+            ]
+        )
+    )
+
+    assert result["status"] == "already_applied"
+    assert result["refresh_candidate_work_item_id"] == current_work_item_id
+    assert result["refresh_candidate_latest_provider_event_id"] == current_provider_event_id
+    assert [method for method, _ in calls] == ["GET", "GET"]

scripts/awooop_source_correlation_apply_smoke.py

@@ -272,13 +272,18 @@ def _verify_existing_status_chain(
     *,
     status: str,
     work_item_id: str | None = None,
+    require_expected_event: bool = True,
 ) -> dict[str, Any]:
     chain = _wait_for_status_chain(
         api_url=args.api_url,
         project_id=args.project_id,
         incident_id=args.target_incident_id,
         min_applied=args.min_applied,
-        expected_source_event_provider_event_id=args.expected_source_event_provider_event_id,
+        expected_source_event_provider_event_id=(
+            args.expected_source_event_provider_event_id
+            if require_expected_event
+            else None
+        ),
         attempts=args.status_attempts,
         interval_seconds=args.status_interval_seconds,
     )
@@ -303,14 +308,15 @@ def _select_refresh_work_item(
     *,
     args: argparse.Namespace,
 ) -> dict[str, Any]:
-    if not args.refresh_work_item_id and not args.refresh_from_latest_canary:
+    refresh_work_item_id = args.refresh_work_item_id or args.work_item_id
+    if not refresh_work_item_id and not args.refresh_from_latest_canary:
         raise SmokeError(
             "applied source link is stale, but no refresh source was configured; "
             "pass --refresh-work-item-id or --refresh-from-latest-canary"
         )
     return _find_work_item(
         recurrence,
-        work_item_id=args.refresh_work_item_id,
+        work_item_id=refresh_work_item_id,
         allow_non_canary=args.allow_non_canary,
         allow_existing_apply=False,
     )
@@ -518,6 +524,66 @@ def run(args: argparse.Namespace) -> dict[str, Any]:
         },
     )
     recurrence = _http_json(recurrence_url)
+    if args.allow_existing_apply and args.refresh_if_stale_days is not None:
+        try:
+            existing = _verify_existing_status_chain(
+                args,
+                status="already_applied",
+                require_expected_event=False,
+            )
+        except SmokeError as exc:
+            refresh_reason = f"status_chain_verify_failed:{exc}"
+            previous_latest_applied_link_at = None
+        else:
+            correlation = {
+                "latest_applied_link_at": existing.get("latest_applied_link_at"),
+                "applied_link_total": existing.get("applied_link_total"),
+                "verification_status": existing.get("verification_status"),
+            }
+            refresh_reason = _refresh_reason(
+                correlation,
+                refresh_if_stale_days=args.refresh_if_stale_days,
+            )
+            if not refresh_reason:
+                refresh_candidate: dict[str, Any] = {}
+                if args.verify_refresh_candidate:
+                    refresh_candidate = _refresh_candidate_result(
+                        _select_refresh_work_item(recurrence, args=args)
+                    )
+                return {
+                    **existing,
+                    "apply_status": "applied",
+                    "refresh_if_stale_days": args.refresh_if_stale_days,
+                    "refresh_reason": None,
+                    **refresh_candidate,
+                }
+            previous_latest_applied_link_at = str(
+                existing.get("latest_applied_link_at") or ""
+            ) or None
+
+        refresh_item = _select_refresh_work_item(recurrence, args=args)
+        refresh_work_item_id = _source_review_work_item_id(refresh_item)
+        if args.dry_run:
+            return {
+                "status": "dry_run_refresh",
+                "refresh_work_item_id": refresh_work_item_id,
+                "target_incident_id": args.target_incident_id,
+                "refresh_if_stale_days": args.refresh_if_stale_days,
+                "refresh_reason": refresh_reason,
+                "previous_latest_applied_link_at": previous_latest_applied_link_at,
+                "refresh_latest_provider_event_id": refresh_item.get(
+                    "latest_provider_event_id"
+                ),
+                "refresh_alertname": refresh_item.get("alertname"),
+            }
+        return _apply_source_correlation_item(
+            args=args,
+            work_item_id=refresh_work_item_id,
+            status="refreshed",
+            refresh_reason=refresh_reason,
+            previous_latest_applied_link_at=previous_latest_applied_link_at,
+        )
+
     try:
         item = _find_work_item(
             recurrence,