fix(governance): dedupe km degradation owner review

apps/api/tests/test_governance_agent.py

@@ -25,7 +25,6 @@ from src.services.governance_agent import (
     GovernanceAgent,
 )
 
-
 # =============================================================================
 # Helpers
 # =============================================================================
@@ -78,9 +77,9 @@ class TestCheckTrustDrift:
     @pytest.mark.asyncio
     async def test_drifted_playbooks_trigger_alert(self):
         """有 playbook trust_score < 0.2 + 最近用過 → 觸發告警，不 auto-deprecate"""
-        from datetime import datetime, timezone
+        from datetime import UTC, datetime
 
-        recent = datetime.now(timezone.utc)
+        recent = datetime.now(UTC)
         low_record = MagicMock()
         low_record.trust_score = 0.05
         low_record.playbook_id = "PB-LOW"
@@ -125,10 +124,10 @@ class TestCheckTrustDrift:
 
         2026-05-02 ogt + Claude Sonnet 4.6: 飛輪自治新路徑
         """
-        from datetime import datetime, timedelta, timezone
+        from datetime import UTC, datetime, timedelta
 
-        old = datetime.now(timezone.utc) - timedelta(days=45)
-        recent = datetime.now(timezone.utc)
+        old = datetime.now(UTC) - timedelta(days=45)
+        recent = datetime.now(UTC)
 
         stale_low = MagicMock()
         stale_low.trust_score = 0.1
@@ -214,13 +213,20 @@ class TestCheckKnowledgeDegradation:
         alerter.alert_governance = AsyncMock()
         agent = _make_agent(alerter=alerter)
 
-        with patch("src.services.governance_agent.get_db_context") as mock_ctx:
+        with (
+            patch("src.services.governance_agent.get_db_context") as mock_ctx,
+            patch(
+                "src.services.governance_agent._resolve_open_knowledge_degradation_events",
+                new=AsyncMock(return_value=0),
+            ) as mock_resolve,
+        ):
             mock_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
             mock_ctx.return_value.__aexit__ = AsyncMock(return_value=False)
 
             result = await agent.check_knowledge_degradation()
 
         alerter.alert_governance.assert_not_called()
+        mock_resolve.assert_awaited_once()
         assert result["stale"] == 1
         assert result["total"] == 10
         assert result["ratio"] == 0.1
@@ -249,12 +255,17 @@ class TestCheckKnowledgeDegradation:
                 "src.services.governance_agent.create_dispatch",
                 new=AsyncMock(),
             ) as mock_create_dispatch,
+            patch(
+                "src.services.governance_agent._has_open_knowledge_degradation_review",
+                new=AsyncMock(return_value=False),
+            ) as mock_has_open_review,
         ):
             mock_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
             mock_ctx.return_value.__aexit__ = AsyncMock(return_value=False)
 
             result = await agent.check_knowledge_degradation()
 
+        mock_has_open_review.assert_awaited_once()
         alerter.alert_governance.assert_called_once()
         call_args = alerter.alert_governance.call_args
         assert call_args[0][0] == "knowledge_degradation"
@@ -274,9 +285,49 @@ class TestCheckKnowledgeDegradation:
         assert result["stale"] == 3
         assert result["ratio"] == 0.3
 
+    @pytest.mark.asyncio
+    async def test_stale_ratio_above_threshold_suppresses_when_owner_review_open(self):
+        """已有 Hermes owner-review 時，同一 KM stale 狀態不重複打 Telegram / 建草稿。"""
+        mock_db = AsyncMock()
+
+        total_mock = MagicMock()
+        total_mock.scalar.return_value = 10
+        stale_mock = MagicMock()
+        stale_mock.scalar.return_value = 3
+
+        mock_db.execute = AsyncMock(side_effect=[total_mock, stale_mock])
+
+        alerter = AsyncMock()
+        alerter.alert_governance = AsyncMock()
+        agent = _make_agent(alerter=alerter)
+
+        with (
+            patch("src.services.governance_agent.get_db_context") as mock_ctx,
+            patch(
+                "src.services.governance_agent.create_dispatch",
+                new=AsyncMock(),
+            ) as mock_create_dispatch,
+            patch(
+                "src.services.governance_agent._has_open_knowledge_degradation_review",
+                new=AsyncMock(return_value=True),
+            ) as mock_has_open_review,
+        ):
+            mock_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            mock_ctx.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            result = await agent.check_knowledge_degradation()
+
+        mock_has_open_review.assert_awaited_once()
+        alerter.alert_governance.assert_not_called()
+        mock_create_dispatch.assert_not_awaited()
+        assert result["alert_suppressed"] is True
+        assert result["suppress_reason"] == "open_owner_review_exists"
+
     def test_knowledge_degradation_dispatch_context(self):
         """intake dispatch context 必須能被 Work Items 直接讀出 owner / stage / next_action."""
-        from src.services.governance_agent import _build_knowledge_degradation_dispatch_context
+        from src.services.governance_agent import (
+            _build_knowledge_degradation_dispatch_context,
+        )
 
         ctx = _build_knowledge_degradation_dispatch_context(
             "evt-km-001",
@@ -542,6 +593,60 @@ class TestRunSelfCheck:
             assert "error" in results[key]
 
 
+# =============================================================================
+# Governance self-check cycle lease
+# =============================================================================
+
+class TestGovernanceSelfCheckLease:
+    """多 API Pod 只能有一個 Pod 寫入本輪治理事件。"""
+
+    @pytest.mark.asyncio
+    async def test_cycle_lease_acquired(self):
+        from src.services.governance_agent import (
+            GOVERNANCE_SELF_CHECK_LEASE_KEY,
+            _try_acquire_governance_self_check_lease,
+        )
+
+        mock_redis = AsyncMock()
+        mock_redis.set = AsyncMock(return_value=True)
+
+        with patch("src.core.redis_client.get_redis", return_value=mock_redis):
+            acquired = await _try_acquire_governance_self_check_lease(3600)
+
+        assert acquired is True
+        mock_redis.set.assert_awaited_once_with(
+            GOVERNANCE_SELF_CHECK_LEASE_KEY,
+            "1",
+            ex=3600,
+            nx=True,
+        )
+
+    @pytest.mark.asyncio
+    async def test_cycle_lease_blocks_second_pod(self):
+        from src.services.governance_agent import (
+            _try_acquire_governance_self_check_lease,
+        )
+
+        mock_redis = AsyncMock()
+        mock_redis.set = AsyncMock(return_value=None)
+
+        with patch("src.core.redis_client.get_redis", return_value=mock_redis):
+            acquired = await _try_acquire_governance_self_check_lease(3600)
+
+        assert acquired is False
+
+    @pytest.mark.asyncio
+    async def test_cycle_lease_fail_open_when_redis_unavailable(self):
+        from src.services.governance_agent import (
+            _try_acquire_governance_self_check_lease,
+        )
+
+        with patch("src.core.redis_client.get_redis", side_effect=RuntimeError("redis down")):
+            acquired = await _try_acquire_governance_self_check_lease(3600)
+
+        assert acquired is True
+
+
 # =============================================================================
 # FailoverAlerter.alert_governance — dedup 邏輯
 # =============================================================================