feat(automation): expose no-write typed route preview

This commit is contained in:
ogt
2026-07-15 12:23:30 +08:00
parent 3df1772a7e
commit 15f3a1f9a3
3 changed files with 100 additions and 2 deletions

View File

@@ -383,6 +383,7 @@ from src.services.backup_notification_policy import (
from src.services.backup_restore_drill_approval_package_template import (
load_latest_backup_restore_drill_approval_package_template,
)
from src.services.controlled_alert_target_router import resolve_typed_alert_target
from src.services.credential_escrow_evidence_intake_readiness import (
load_latest_credential_escrow_evidence_intake_readiness,
validate_credential_escrow_evidence_owner_response,
@@ -552,6 +553,17 @@ _PRIORITY_WORK_ORDER_LIVE_READBACK_TIMEOUT_SECONDS = 12.0
# Request/Response Models
# =============================================================================
class ControlledAlertTargetPreviewRequest(BaseModel):
"""只讀解析一筆告警的 canonical typed target。"""
alertname: str = Field(min_length=1, max_length=160)
target_resource: str = Field(default="", max_length=240)
namespace: str = Field(default="", max_length=160)
alert_category: str = Field(default="", max_length=120)
labels: dict[str, str] = Field(default_factory=dict)
class AnalyzeRequest(BaseModel):
"""分析請求"""
incident_id: str | None = Field(
@@ -1231,6 +1243,39 @@ async def get_sre_k3s_controlled_automation_work_items() -> dict[str, Any]:
) from exc
@router.post(
"/controlled-alert-target-preview",
response_model=dict[str, Any],
summary="只讀預覽告警的 canonical typed route",
description=(
"以 production service registry 執行 deterministic target resolution"
"只回傳經 LAN topology redaction 的 route。此端點不呼叫模型、不建立事件、"
"不 dispatch Agent99、不執行 Ansible/Kubernetes/DB 動作,也不更新完成狀態。"
),
)
async def preview_controlled_alert_target(
payload: ControlledAlertTargetPreviewRequest,
) -> dict[str, Any]:
"""提供可重播、零 side effect 的 production typed-router verifier。"""
route = resolve_typed_alert_target(
alertname=payload.alertname,
target_resource=payload.target_resource,
namespace=payload.namespace,
labels=payload.labels,
alert_category=payload.alert_category,
)
return redact_public_lan_topology(
{
"schema_version": "controlled_alert_target_preview_v1",
"preview_only": True,
"dispatch_performed": False,
"runtime_closure_verified": False,
"route": route,
}
)
@router.post(
"/agent99/telegram-lifecycle",
response_model=dict[str, Any],

View File

@@ -122,3 +122,54 @@ def test_endpoint_returns_full_work_ledger() -> None:
assert payload["agent99_host_operations_bridge"][
"completion_callback"
] == "/api/v1/agents/agent99/completion-callback"
def test_typed_target_preview_is_no_write_and_fail_closed() -> None:
app = FastAPI()
app.include_router(router, prefix="/api/v1")
client = TestClient(app)
unresolved = client.post(
"/api/v1/agents/controlled-alert-target-preview",
json={
"alertname": "VMwareGuestHealthDegraded",
"target_resource": "unregistered-vmware-guest",
"namespace": "virtualization",
"labels": {},
"alert_category": "infrastructure",
},
)
agent99 = client.post(
"/api/v1/agents/controlled-alert-target-preview",
json={
"alertname": "Agent99ServiceUnhealthy",
"target_resource": "agent99",
"namespace": "control-plane",
"labels": {},
"alert_category": "infrastructure",
},
)
assert unresolved.status_code == 200
unresolved_payload = unresolved.json()
assert unresolved_payload["preview_only"] is True
assert unresolved_payload["dispatch_performed"] is False
assert unresolved_payload["runtime_closure_verified"] is False
assert unresolved_payload["route"]["resolution_status"] == (
"asset_identity_unresolved"
)
assert unresolved_payload["route"]["executor"] is None
assert unresolved_payload["route"]["agent99_bridge"][
"dispatch_allowed"
] is False
assert agent99.status_code == 200
agent99_payload = agent99.json()
assert agent99_payload["dispatch_performed"] is False
assert agent99_payload["route"]["canonical_asset_id"] == (
"windows-vmware:host_99"
)
assert agent99_payload["route"]["executor"] == "Agent99"
assert agent99_payload["route"]["agent99_bridge"][
"dispatch_allowed"
] is True