diff --git a/services/ai_automation_smoke_service.py b/services/ai_automation_smoke_service.py index 29dc8c9..6cef2b8 100644 --- a/services/ai_automation_smoke_service.py +++ b/services/ai_automation_smoke_service.py @@ -214,6 +214,9 @@ def build_scheduled_automation_health_summary( sitewide_details = sitewide_readback.get("details") or {} visual_qa_readback = _find_check(source_result, "Sitewide visual QA readback") visual_qa_details = visual_qa_readback.get("details") or {} + if not visual_qa_readback or not visual_qa_details: + visual_qa_readback = _latest_sitewide_visual_qa_readback_check() + visual_qa_details = visual_qa_readback.get("details") or {} smoke_status = source_result.get("status") or ("warning" if latest_history else "warning") freshness_family = _history_freshness_family( latest_history, @@ -1019,6 +1022,60 @@ def _sitewide_visual_qa_check() -> Dict[str, Any]: ) +def _latest_sitewide_visual_qa_readback_check() -> Dict[str, Any]: + """Read the latest visual QA artifact for scheduled summaries without running smoke.""" + try: + from services.ai_surface_html_readback_service import build_sitewide_visual_qa_readback + + readback = build_sitewide_visual_qa_readback() + summary = readback.get("summary") or {} + status = readback.get("status") or "warning" + route_count = int(summary.get("route_count") or 0) + viewport_count = int(summary.get("viewport_count") or 0) + failed_count = int(summary.get("failed_count") or 0) + return _check( + "Sitewide visual QA readback", + status, + ( + f"Sitewide visual QA ok,{route_count} routes x {viewport_count} viewports 無 overflow 退化" + if status == "ok" + else ( + f"Sitewide visual QA 需更新或修復:failed={failed_count}、" + f"routes={route_count}、stale={bool(summary.get('stale'))}" + ) + ), + { + "policy": readback.get("policy"), + "result_count": int(summary.get("result_count") or 0), + "route_count": route_count, + "viewport_count": viewport_count, + "pass_count": int(summary.get("pass_count") or 0), + "failed_count": failed_count, + "overflow_issue_count": int(summary.get("overflow_issue_count") or 0), + "visual_offender_count": int(summary.get("visual_offender_count") or 0), + "artifact_path": readback.get("artifact_path"), + "artifact_exists": bool(readback.get("artifact_exists")), + "artifact_generated_at": readback.get("artifact_generated_at"), + "age_hours": summary.get("age_hours"), + "stale": bool(summary.get("stale")), + "next_machine_action": readback.get("next_machine_action"), + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + ) + except Exception as exc: + return _check( + "Sitewide visual QA readback", + "warning", + f"Sitewide visual QA artifact fallback 無法讀取:{exc}", + { + "writes_database": False, + "primary_human_gate_count": 0, + }, + ) + + def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: int = 20) -> Dict[str, Any]: checks: List[Dict[str, Any]] = [ _event_router_check(), diff --git a/tests/test_ai_automation_smoke_service.py b/tests/test_ai_automation_smoke_service.py index 59c7482..82b45b7 100644 --- a/tests/test_ai_automation_smoke_service.py +++ b/tests/test_ai_automation_smoke_service.py @@ -335,6 +335,59 @@ def test_scheduled_automation_health_summary_can_use_current_smoke_without_recor assert not history_path.exists() +def test_scheduled_automation_health_summary_falls_back_to_visual_qa_artifact(tmp_path, monkeypatch): + from datetime import datetime + from services import ai_automation_smoke_service as smoke + from services import ai_surface_html_readback_service as surface_service + + history_path = tmp_path / "smoke_history.jsonl" + history_path.write_text( + json.dumps({ + "generated_at": datetime.now().isoformat(timespec="seconds"), + "status": "ok", + "summary": {"ok": 9, "warning": 0, "critical": 0, "total": 9}, + "checks": [], + }, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path)) + monkeypatch.setattr( + surface_service, + "build_sitewide_visual_qa_readback", + lambda **_kwargs: { + "policy": "read_only_sitewide_visual_qa_readback_v1", + "status": "ok", + "artifact_path": "data/ai_automation/sitewide_visual_qa_latest.json", + "artifact_exists": True, + "artifact_generated_at": "2026-07-03T07:40:00", + "next_machine_action": "keep_sitewide_visual_qa_monitoring", + "summary": { + "result_count": 147, + "route_count": 49, + "viewport_count": 3, + "pass_count": 147, + "failed_count": 0, + "overflow_issue_count": 0, + "visual_offender_count": 0, + "stale": False, + }, + }, + ) + + summary = smoke.build_scheduled_automation_health_summary( + history_limit=5, + freshness_max_age_hours=9999, + ) + + visual_family = next( + item for item in summary["families"] + if item["key"] == "sitewide_visual_qa" + ) + assert visual_family["status"] == "ok" + assert visual_family["details"]["route_count"] == 49 + assert visual_family["details"]["failed_count"] == 0 + + def test_scheduled_automation_health_summary_route_returns_compact_payload(tmp_path, monkeypatch): from datetime import datetime from flask import Flask