fix(agent): keep failed apply recovery AI controlled
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m50s
CD Pipeline / build-and-deploy (push) Successful in 5m8s
CD Pipeline / post-deploy-checks (push) Successful in 4m5s

This commit is contained in:
ogt
2026-07-10 22:48:32 +08:00
parent 918c72bcd3
commit 682cbcf1ae
4 changed files with 63 additions and 10 deletions

View File

@@ -1935,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

View File

@@ -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 RESULTAI Agent 受控執行待修復",
{"reply_markup": {"inline_keyboard": []}},
{"outbound_message_type": "error"},
)
== "error"
)
assert (
telegram_gateway_module._infer_outbound_message_type(
"🤖❌ [AUTO] AI 自動修復失敗",