chore: sync latest Gitea main

apps/api/tests/test_awooop_truth_chain_service.py

@@ -39,6 +39,7 @@ from src.services.awooop_ansible_check_mode_service import (
     build_ansible_post_apply_runtime_stage_receipts,
     build_ansible_pre_apply_runtime_stage_receipts,
     build_ansible_timeline_runtime_stage_receipt,
+    claim_catalog_drift_failed_check_modes,
     claim_pending_check_modes,
     claim_stale_pending_check_modes,
     detect_ansible_transport_blockers,
@@ -1692,6 +1693,28 @@ def test_stale_check_mode_row_is_revalidated_with_canonical_run_id() -> None:
     assert claim.playbook_path == "infra/ansible/playbooks/188-momo-backup-user.yml"
 
 
+def test_failed_nginx_check_mode_revalidates_to_readonly_catalog_path() -> None:
+    source_candidate_op_id = "00000000-0000-0000-0000-000000000093"
+    claim = _claim_from_stale_check_mode_row({
+        "op_id": "00000000-0000-0000-0000-000000000094",
+        "parent_op_id": source_candidate_op_id,
+        "incident_id": "INC-20260710-F87642",
+        "input": {
+            "incident_id": "INC-20260710-F87642",
+            "source_candidate_op_id": source_candidate_op_id,
+            "catalog_id": "ansible:nginx-sync",
+            "catalog_playbook_path": "infra/ansible/playbooks/nginx-sync.yml",
+            "check_mode_playbook_path": "infra/ansible/playbooks/nginx-sync.yml",
+            "inventory_hosts": ["host_110", "host_188"],
+            "risk_level": "high",
+        },
+    })
+
+    assert claim is not None
+    assert claim.playbook_path == "infra/ansible/playbooks/nginx-sync-readonly.yml"
+    assert claim.apply_playbook_path == "infra/ansible/playbooks/nginx-sync.yml"
+
+
 def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None:
     source = inspect.getsource(claim_stale_pending_check_modes)
     run_source = inspect.getsource(run_pending_check_modes_once)
@@ -1705,6 +1728,16 @@ def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None:
     assert "effective_timeout_seconds + 120" in run_source
 
 
+def test_failed_check_mode_catalog_drift_replays_once() -> None:
+    source = inspect.getsource(claim_catalog_drift_failed_check_modes)
+    run_source = inspect.getsource(run_pending_check_modes_once)
+
+    assert "replay_of_check_mode_op_id" in source
+    assert "catalog_drift_replay" in source
+    assert "FOR UPDATE SKIP LOCKED" in source
+    assert "claim_catalog_drift_failed_check_modes" in run_source
+
+
 def test_ansible_apply_receipt_backfill_queries_existing_apply_rows() -> None:
     source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once)
 
@@ -1902,9 +1935,12 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
     assert "FOR UPDATE SKIP LOCKED" in source
     assert "controlled_retry_check_mode_replay" in source
     assert "build_ansible_check_mode_command" in source
-    assert "controlled_apply_allowed=False" in source
+    assert "controlled_apply_allowed=True" in source
+    assert '"approval_required_before_apply": False' in source
+    assert '"owner_review_required": False' in source
     assert '"runtime_apply_executed": False' in source
-    assert "retry_requires_repair_and_new_apply_gate" in source
+    assert "queue_ai_playbook_or_transport_repair_candidate" in source
+    assert "retry_requires_repair_and_new_apply_gate" not in source
     assert "_record_retry_runtime_stage_receipt" in source
     assert "NOT EXISTS" in source
     assert "run_controlled_apply_for_claim" not in source

apps/api/tests/test_telegram_message_templates.py

@@ -1235,6 +1235,15 @@ async def test_controlled_apply_result_receipt_marks_callback_reply_evidence(mon
     assert source_extra["source_refs"]["automation_run_ids"] == [
         "00000000-0000-0000-0000-000000000001"
     ]
+    assert source_extra["outbound_message_type"] == "final"
+    assert (
+        telegram_gateway_module._infer_outbound_message_type(
+            payload["text"],
+            payload,
+            source_extra,
+        )
+        == "final"
+    )
     km_snapshot = source_extra["km_stale_completion_summary"]
     assert (
         km_snapshot["source_schema_version"]
@@ -2393,6 +2402,14 @@ def test_outbound_message_type_inference():
         )
         == "approval_request"
     )
+    assert (
+        telegram_gateway_module._infer_outbound_message_type(
+            "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復",
+            {"reply_markup": {"inline_keyboard": []}},
+            {"outbound_message_type": "error"},
+        )
+        == "error"
+    )
     assert (
         telegram_gateway_module._infer_outbound_message_type(
             "🤖❌ [AUTO] AI 自動修復失敗",