fix(sre): render backup checks as no-write receipts

This commit is contained in:
Your Name
2026-07-19 01:06:53 +08:00
parent e127c8b56d
commit ca47b23b64
2 changed files with 151 additions and 0 deletions

View File

@@ -17,6 +17,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
from src.db.awooop_models import AwoooPMcpGatewayAudit
from src.db.base import get_db_context
from src.db.models import PlaybookRecord
from src.models.agent99_completion import Agent99TelegramLifecycleRequest
from src.models.knowledge import (
EntrySource,
EntryStatus,
@@ -44,6 +45,9 @@ from src.services.agent99_sre_bridge import (
bridge_alertmanager_to_agent99,
read_agent99_sre_outcome,
)
from src.services.agent99_telegram_lifecycle import (
deliver_agent99_telegram_lifecycle,
)
from src.services.knowledge_service import get_knowledge_service
from src.services.telegram_gateway import get_telegram_gateway
from src.utils.timezone import now_taipei
@@ -582,6 +586,75 @@ async def _ensure_telegram_receipt(
mode: str,
) -> str | None:
try:
if "backup_health" in identity.route_id:
# BackupCheck is an evidence-only investigation. Never render it
# through the controlled-apply success card: that would falsely
# claim an apply/return code even though backup, restore, escrow,
# retention and remote-delete writes are all prohibited.
response = await deliver_agent99_telegram_lifecycle(
Agent99TelegramLifecycleRequest(
schema_version="agent99_telegram_lifecycle_v1",
delivery_id=(
f"agent99-backup-lifecycle:{identity.run_id}"
),
project_id="awoooi",
incident_id=identity.incident_id,
fingerprint=(
"sha256:"
+ hashlib.sha256(
identity.source_fingerprint.encode()
).hexdigest()
),
event_type="backup_restore_readback_verified",
severity="warning",
lifecycle="verifying",
state_key=f"backup_restore:{identity.run_id}",
run_id=str(identity.run_id),
trace_id=identity.trace_id,
work_item_id=identity.work_item_id,
executor_name="Agent99 BackupCheck (read-only)",
verifier_name="backup_restore_readback_verifier",
received_status="passed",
check_status="passed",
apply_status="not_applicable",
verifier_status="passed",
closure_status="pending",
title="Backup / restore evidence verified",
target="backup_restore",
impact=(
"資料保護證據已完成唯讀驗證;沒有執行備份、還原、"
"刪除、retention 或 escrow 寫入。"
),
action=(
"Agent99 完成同一 run 的唯讀 BackupCheck"
"LLM 未取得 runtime authority。"
),
verification=(
"Independent verifier 已核對 backup status、freshness、"
"offsite、escrow、restore drill 與 source resolution。"
),
duration_text="same-run verified",
occurrences=1,
next_update=(
"等待 KM/RAG/MCP/PlayBook 與 DR scorecard durable ack"
"之後才關閉事件。"
),
occurred_at=now_taipei(),
)
)
if (
not isinstance(response, dict)
or response.get("ok") is not True
or response.get("durable_outbound_acknowledged") is not True
or response.get("destination_binding_verified") is not True
or not str(response.get("provider_message_id") or "")
):
return None
return (
"telegram_outbound:"
f"{response['provider_message_id']}:durable_ack"
)
response = await get_telegram_gateway().send_controlled_apply_result_receipt(
automation_run_id=str(identity.run_id),
incident_id=identity.incident_id,

View File

@@ -147,6 +147,84 @@ async def test_agent99_mcp_ack_is_same_run_durable_and_idempotent(
assert second == expected
@pytest.mark.asyncio
async def test_backup_telegram_ack_is_no_write_lifecycle_not_apply(
monkeypatch,
) -> None:
identity = build_agent99_dispatch_identity(
project_id="awoooi",
incident_id="INC-20260719-BACKUP",
source_fingerprint="backup-fingerprint-1",
route_id="agent99:backup_health:BackupCheck",
)
delivered = []
async def deliver(payload): # type: ignore[no-untyped-def]
delivered.append(payload)
return {
"ok": True,
"provider_message_id": "9919",
"durable_outbound_acknowledged": True,
"destination_binding_verified": True,
}
class ForbiddenGateway:
async def send_controlled_apply_result_receipt(self, **_kwargs):
raise AssertionError("BackupCheck must not use controlled-apply receipt")
monkeypatch.setattr(job, "deliver_agent99_telegram_lifecycle", deliver)
monkeypatch.setattr(job, "get_telegram_gateway", lambda: ForbiddenGateway())
receipt = await job._ensure_telegram_receipt(
identity,
mode="BackupCheck",
)
assert receipt == "telegram_outbound:9919:durable_ack"
assert len(delivered) == 1
payload = delivered[0]
assert payload.run_id == str(identity.run_id)
assert payload.trace_id == identity.trace_id
assert payload.work_item_id == identity.work_item_id
assert payload.lifecycle == "verifying"
assert payload.executor_name == "Agent99 BackupCheck (read-only)"
assert payload.verifier_name == "backup_restore_readback_verifier"
assert payload.apply_status == "not_applicable"
assert payload.verifier_status == "passed"
assert payload.closure_status == "pending"
assert "沒有執行備份" in payload.impact
assert "LLM 未取得 runtime authority" in payload.action
@pytest.mark.asyncio
async def test_backup_telegram_ack_requires_durable_destination_proof(
monkeypatch,
) -> None:
identity = build_agent99_dispatch_identity(
project_id="awoooi",
incident_id="INC-20260719-BACKUP-FAIL",
source_fingerprint="backup-fingerprint-2",
route_id="agent99:backup_health:BackupCheck",
)
async def deliver(_payload): # type: ignore[no-untyped-def]
return {
"ok": False,
"provider_message_id": "9920",
"durable_outbound_acknowledged": True,
"destination_binding_verified": False,
}
monkeypatch.setattr(job, "deliver_agent99_telegram_lifecycle", deliver)
receipt = await job._ensure_telegram_receipt(
identity,
mode="BackupCheck",
)
assert receipt is None
@pytest.mark.asyncio
async def test_reconciler_consumes_authenticated_outcome_and_closes_same_run(
monkeypatch,