From e79371870735bca41604c563fd66328c2cbdd88e Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 20:31:30 +0800 Subject: [PATCH] fix(agent): fail closed on incident read drift --- apps/api/src/api/v1/incidents.py | 34 +- apps/api/src/core/redis_client.py | 6 +- .../src/jobs/awooop_ansible_check_mode_job.py | 21 ++ .../api/src/jobs/incident_analysis_sweeper.py | 15 +- .../awooop_ansible_check_mode_service.py | 181 +++++++--- apps/api/src/services/incident_service.py | 318 +++++++++++------- .../src/workers/ansible_executor_broker.py | 20 +- .../tests/test_ansible_verified_closure.py | 182 +++++++++- .../test_incident_readback_source_truth.py | 145 +++++++- 9 files changed, 729 insertions(+), 193 deletions(-) diff --git a/apps/api/src/api/v1/incidents.py b/apps/api/src/api/v1/incidents.py index 2129edcb7..e1c9c8252 100644 --- a/apps/api/src/api/v1/incidents.py +++ b/apps/api/src/api/v1/incidents.py @@ -30,7 +30,10 @@ from src.models.incident import Incident, IncidentStatus, Severity # Phase 22 P1: 移除 Repository 直接存取 (2026-03-31) # Phase 16 R3.3b (2026-03-25 台北時區): Repository 層整合 - 已移至 Service 層 from src.services.decision_manager import get_decision_manager -from src.services.incident_service import get_incident_service +from src.services.incident_service import ( + IncidentDurableReadError, + get_incident_service, +) from src.services.incident_timeline_service import fetch_incident_timeline from src.services.proposal_service import get_proposal_service from src.utils.timezone import now_taipei @@ -276,6 +279,20 @@ async def list_incidents( incidents=responses, ) + except IncidentDurableReadError as e: + logger.warning( + "incident_list_durable_readback_degraded", + project_id=project_id, + ) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail={ + "status": "degraded", + "reason_code": str(e), + "source_of_truth": "postgresql", + "redis_fallback_used": False, + }, + ) from e except Exception as e: logger.exception( "list_incidents_error", @@ -342,6 +359,21 @@ async def get_incident( except HTTPException: raise + except IncidentDurableReadError as e: + logger.warning( + "incident_detail_durable_readback_degraded", + incident_id=incident_id, + project_id=project_id, + ) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail={ + "status": "degraded", + "reason_code": str(e), + "source_of_truth": "postgresql", + "redis_fallback_used": False, + }, + ) from e except Exception as e: logger.exception( "get_incident_error", diff --git a/apps/api/src/core/redis_client.py b/apps/api/src/core/redis_client.py index 0d378079f..9918081df 100644 --- a/apps/api/src/core/redis_client.py +++ b/apps/api/src/core/redis_client.py @@ -66,8 +66,12 @@ async def init_redis_pool() -> redis.Redis: "redis_pool_initialized", url=settings.REDIS_URL.split("@")[-1], # 隱藏密碼 ) - except redis.ConnectionError as e: + except Exception as e: logger.error("redis_connection_failed", error=str(e)) + failed_pool = _redis_pool + _redis_pool = None + if failed_pool is not None: + await failed_pool.close() raise return _redis_pool diff --git a/apps/api/src/jobs/awooop_ansible_check_mode_job.py b/apps/api/src/jobs/awooop_ansible_check_mode_job.py index 95d5509b8..31fd73964 100644 --- a/apps/api/src/jobs/awooop_ansible_check_mode_job.py +++ b/apps/api/src/jobs/awooop_ansible_check_mode_job.py @@ -14,6 +14,7 @@ import random import structlog from src.core.config import settings +from src.core.redis_client import init_redis_pool from src.services.awooop_ansible_check_mode_service import ( reconcile_verified_incident_working_memory_once, run_pending_check_modes_once, @@ -28,6 +29,19 @@ def _error_backoff_seconds() -> float: return random.uniform(_ERROR_BACKOFF_MIN_SECONDS, _ERROR_BACKOFF_MAX_SECONDS) +async def _retry_optional_projection_redis() -> bool: + try: + await init_redis_pool() + return True + except Exception as exc: + logger.warning( + "awooop_working_memory_projection_redis_unavailable", + error_type=type(exc).__name__, + controlled_executor_available=True, + ) + return False + + async def run_awooop_ansible_check_mode_loop() -> None: if not settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER: logger.info("awooop_ansible_check_mode_worker_disabled") @@ -48,6 +62,7 @@ async def run_awooop_ansible_check_mode_loop() -> None: limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT, timeout_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS, ) + projection_redis_ready = await _retry_optional_projection_redis() projection = await reconcile_verified_incident_working_memory_once( limit=max( 1, @@ -71,6 +86,9 @@ async def run_awooop_ansible_check_mode_loop() -> None: "working_memory_projection_error": ( str(projection.get("error") or "")[:500] or None ), + "working_memory_projection_redis_ready": ( + projection_redis_ready + ), } ) if result.get("error") or result.get("catalog_replay_error"): @@ -79,7 +97,10 @@ async def run_awooop_ansible_check_mode_loop() -> None: result.get("claimed") or result.get("blockers") or result.get("working_memory_projection_updated") + or result.get("working_memory_projection_missing") + or result.get("working_memory_projection_failed") or result.get("working_memory_projection_error") + or not result.get("working_memory_projection_redis_ready") ): logger.info("awooop_ansible_check_mode_worker_tick", **result) except Exception as exc: diff --git a/apps/api/src/jobs/incident_analysis_sweeper.py b/apps/api/src/jobs/incident_analysis_sweeper.py index fcb994bf7..35918b874 100644 --- a/apps/api/src/jobs/incident_analysis_sweeper.py +++ b/apps/api/src/jobs/incident_analysis_sweeper.py @@ -24,10 +24,12 @@ Key 格式說明: from __future__ import annotations import asyncio +from datetime import UTC, datetime, timedelta import structlog -from src.models.incident import Incident, IncidentStatus, Severity +from src.core.config import settings +from src.models.incident import Incident, Severity logger = structlog.get_logger(__name__) @@ -68,9 +70,9 @@ async def _sweep_once(sem: asyncio.Semaphore) -> None: Decision token key 格式: decision:DEC-{HEX12} (非 decision:INC-*) 使用 sweeper_done:{incident_id} 輕量標記避免重複觸發。 """ + from src.core.redis_client import get_redis from src.services.decision_manager import get_decision_manager from src.services.incident_service import get_incident_service - from src.core.redis_client import get_redis redis = get_redis() incident_service = get_incident_service() @@ -78,7 +80,9 @@ async def _sweep_once(sem: asyncio.Semaphore) -> None: # 取得所有 INVESTIGATING incidents try: - incidents: list[Incident] = await incident_service.get_active_incidents() + incidents: list[Incident] = await incident_service.get_active_incidents( + project_id=settings.SYSTEM_NAME, + ) except Exception as e: logger.warning("sweeper_get_incidents_failed", error=str(e)) return @@ -87,8 +91,7 @@ async def _sweep_once(sem: asyncio.Semaphore) -> None: return # 過濾：只處理 48h 內的 incident（避免首次啟動把全部歷史舊案洗版 Telegram） - from datetime import datetime, timezone, timedelta - now_utc = datetime.now(timezone.utc) + now_utc = datetime.now(UTC) cutoff = now_utc - timedelta(hours=_MAX_INCIDENT_AGE_HOURS) recent_incidents = [] @@ -97,7 +100,7 @@ async def _sweep_once(sem: asyncio.Semaphore) -> None: if created: # 確保 created_at 有時區資訊 if created.tzinfo is None: - created = created.replace(tzinfo=timezone.utc) + created = created.replace(tzinfo=UTC) if created >= cutoff: recent_incidents.append(incident) else: diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 57a2ab6ea..1fc48aa52 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -53,6 +53,8 @@ _CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS = 30 _EXECUTION_CAPABILITY_MIN_TTL_SECONDS = 300 _EXECUTION_CAPABILITY_MAX_TTL_SECONDS = 1_800 _EXECUTION_CAPABILITY_SAFETY_MARGIN_SECONDS = 5 +_WORKING_MEMORY_PROJECTION_WINDOW_HOURS = 8 * 24 +_WORKING_MEMORY_PROJECTION_CURSOR: dict[str, tuple[datetime, str]] = {} _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE = "remediation_executed" _EXECUTION_CAPABILITY_OPERATION_TYPES = frozenset( { @@ -2288,10 +2290,10 @@ async def _project_incident_terminal_to_working_memory( async def reconcile_verified_incident_working_memory_once( *, project_id: str = "awoooi", - window_hours: int = 24, + window_hours: int = _WORKING_MEMORY_PROJECTION_WINDOW_HOURS, limit: int = 10, ) -> dict[str, Any]: - """Repair terminal DB truth that is still active in Redis working memory.""" + """Project pending terminal DB rows without starving older Redis repairs.""" stats: dict[str, Any] = { "scanned": 0, @@ -2302,42 +2304,104 @@ async def reconcile_verified_incident_working_memory_once( "error": None, } try: + cursor = _WORKING_MEMORY_PROJECTION_CURSOR.get(project_id) + statement = text(""" + SELECT + incident.incident_id, + incident.status::text AS incident_status, + incident.updated_at, + incident.resolved_at + FROM incidents incident + WHERE incident.project_id = :project_id + AND upper(incident.status::text) IN ('RESOLVED', 'CLOSED') + AND incident.resolved_at IS NOT NULL + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,incident_resolved}' = 'true' + AND incident.updated_at >= + NOW() - (:window_hours * INTERVAL '1 hour') + AND ( + coalesce( + cast(incident.outcome AS jsonb) #>> + '{working_memory_projection,ready}', + 'false' + ) <> 'true' + OR cast(incident.outcome AS jsonb) #>> + '{working_memory_projection,incident_status}' + IS DISTINCT FROM incident.status::text + OR cast(incident.outcome AS jsonb) #>> + '{working_memory_projection,source_updated_at_epoch_us}' + IS DISTINCT FROM CAST( + CAST( + round( + extract(epoch FROM incident.updated_at) + * 1000000 + ) AS bigint + ) AS text + ) + ) + AND ( + CAST(:cursor_updated_at AS timestamptz) IS NULL + OR incident.updated_at > + CAST(:cursor_updated_at AS timestamptz) + OR ( + incident.updated_at = + CAST(:cursor_updated_at AS timestamptz) + AND incident.incident_id > :cursor_incident_id + ) + ) + ORDER BY incident.updated_at ASC, incident.incident_id ASC + LIMIT :limit + """) + + def query_params( + active_cursor: tuple[datetime, str] | None, + ) -> dict[str, Any]: + return { + "project_id": project_id, + "window_hours": max(1, int(window_hours)), + "limit": max(1, int(limit)), + "cursor_updated_at": active_cursor[0] if active_cursor else None, + "cursor_incident_id": active_cursor[1] if active_cursor else "", + } + async with get_db_context(project_id) as db: - result = await db.execute( - text(""" - SELECT - incident.incident_id, - incident.status::text AS incident_status, - incident.updated_at, - incident.resolved_at - FROM incidents incident - WHERE incident.project_id = :project_id - AND upper(incident.status::text) IN ('RESOLVED', 'CLOSED') - AND incident.resolved_at IS NOT NULL - AND cast(incident.outcome AS jsonb) #>> - '{automation_terminal,incident_resolved}' = 'true' - AND incident.updated_at >= - NOW() - (:window_hours * INTERVAL '1 hour') - ORDER BY incident.updated_at DESC - LIMIT :limit - """), - { - "project_id": project_id, - "window_hours": max(1, int(window_hours)), - "limit": max(1, int(limit)), - }, - ) + result = await db.execute(statement, query_params(cursor)) rows = [dict(row) for row in result.mappings().all()] + if not rows and cursor is not None: + cursor = None + result = await db.execute(statement, query_params(None)) + rows = [dict(row) for row in result.mappings().all()] + + if rows: + last = rows[-1] + _WORKING_MEMORY_PROJECTION_CURSOR[project_id] = ( + last["updated_at"], + str(last.get("incident_id") or ""), + ) + else: + _WORKING_MEMORY_PROJECTION_CURSOR.pop(project_id, None) stats["scanned"] = len(rows) for row in rows: - projection_status = await _project_incident_terminal_to_working_memory( - incident_id=str(row.get("incident_id") or ""), - project_id=project_id, - incident_status=str(row.get("incident_status") or ""), - updated_at=row["updated_at"], - resolved_at=row.get("resolved_at"), - ) + try: + projection_status = ( + await _project_incident_terminal_to_working_memory( + incident_id=str(row.get("incident_id") or ""), + project_id=project_id, + incident_status=str(row.get("incident_status") or ""), + updated_at=row["updated_at"], + resolved_at=row.get("resolved_at"), + ) + ) + except Exception as exc: + stats["failed"] += 1 + logger.warning( + "ansible_incident_working_memory_projection_failed", + project_id=project_id, + incident_id=str(row.get("incident_id") or ""), + error_type=type(exc).__name__, + ) + continue if projection_status in stats: stats[projection_status] += 1 else: @@ -2737,13 +2801,24 @@ async def _commit_verified_incident_closure( "closure_receipt_written": True, "resolved_lifecycle_written": True, } - projection_status = await _project_incident_terminal_to_working_memory( - incident_id=claim.incident_id, - project_id=project_id, - incident_status=closure_result["incident_status"], - updated_at=row["updated_at"], - resolved_at=row.get("resolved_at"), - ) + try: + projection_status = await _project_incident_terminal_to_working_memory( + incident_id=claim.incident_id, + project_id=project_id, + incident_status=closure_result["incident_status"], + updated_at=row["updated_at"], + resolved_at=row.get("resolved_at"), + ) + except Exception as projection_exc: + projection_status = "failed" + logger.warning( + "ansible_verified_incident_closure_projection_failed", + automation_run_id=automation_run_id, + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + error_type=type(projection_exc).__name__, + durable_closure_preserved=True, + ) closure_result["working_memory_projection_status"] = projection_status closure_result["working_memory_projection_ready"] = ( projection_status in {"updated", "current"} @@ -2802,25 +2877,31 @@ async def _finalize_verified_apply_closure( "resolved_lifecycle", ], } - if terminal.get("working_memory_projection_ready") is not True: - return { - "status": "working_memory_projection_pending", - "closed": False, - "missing": ["working_memory_terminal_projection"], - } - readback = await _read_incident_closure_readback( claim, apply_op_id=apply_op_id, project_id=project_id, ) + projection_ready = terminal.get("working_memory_projection_ready") is True + readback_closed = readback.get("closed") is True + missing = list(readback.get("missing") or []) + if not projection_ready: + missing = sorted({*missing, "working_memory_terminal_projection"}) return { + **readback, "status": ( - "controlled_apply_closed" - if readback.get("closed") is True + ( + "controlled_apply_closed" + if projection_ready + else "controlled_apply_closed_projection_degraded" + ) + if readback_closed else "closure_readback_pending" ), - **readback, + "closed": readback_closed, + "missing": missing, + "working_memory_projection_ready": projection_ready, + "read_model_degraded": not projection_ready, } diff --git a/apps/api/src/services/incident_service.py b/apps/api/src/services/incident_service.py index fc9161294..91a6c86ee 100644 --- a/apps/api/src/services/incident_service.py +++ b/apps/api/src/services/incident_service.py @@ -26,6 +26,7 @@ from typing import Any, Literal from uuid import UUID import structlog +from sqlalchemy import select, text from src.core.redis_client import get_redis from src.db.base import get_db_context @@ -41,6 +42,11 @@ from src.utils.timezone import now_taipei logger = structlog.get_logger(__name__) + +class IncidentDurableReadError(RuntimeError): + """The durable incident source could not be read safely.""" + + # ============================================================================= # C2 修正: 從 webhooks.py 遷入的業務邏輯 # 2026-04-10 Claude Sonnet 4.6 Asia/Taipei @@ -595,10 +601,10 @@ class IncidentService: Returns: bool: 是否成功寫入 """ - redis_client = get_redis() key = f"{INCIDENT_KEY_PREFIX}{incident.incident_id}" try: + redis_client = get_redis() # 序列化為 JSON incident_json = incident.model_dump_json() @@ -634,10 +640,10 @@ class IncidentService: Returns: Incident | None: 事件資料，若不存在則返回 None """ - redis_client = get_redis() key = f"{INCIDENT_KEY_PREFIX}{incident_id}" try: + redis_client = get_redis() data = await redis_client.get(key) if data is None: return None @@ -673,19 +679,26 @@ class IncidentService: updated_at: datetime, resolved_at: datetime | None, ) -> Literal["updated", "current", "missing", "failed"]: - """Project a durable terminal DB state into the Redis read model.""" + """Project the exact durable terminal row into the Redis read model.""" if status not in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED): raise ValueError("working_memory_terminal_projection_requires_terminal_status") - incident = await self.get_from_working_memory(incident_id) - hydrated_from_episodic = incident is None - if incident is None: - incident = await self.get_from_episodic_memory( + try: + durable = await self._get_from_episodic_memory_strict( incident_id, project_id=project_id, ) - if incident is None: + except Exception as exc: + logger.warning( + "working_memory_terminal_projection_durable_read_failed", + incident_id=incident_id, + project_id=project_id, + error_type=type(exc).__name__, + ) + return "failed" + + if durable is None: logger.warning( "working_memory_terminal_projection_missing_incident", incident_id=incident_id, @@ -693,27 +706,35 @@ class IncidentService: ) return "missing" - current_updated_at = incident.updated_at - if current_updated_at.tzinfo is None: - current_updated_at = current_updated_at.replace(tzinfo=UTC) - durable_updated_at = updated_at - if durable_updated_at.tzinfo is None: - durable_updated_at = durable_updated_at.replace(tzinfo=UTC) + if durable.status not in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED): + logger.warning( + "working_memory_terminal_projection_source_not_terminal", + incident_id=incident_id, + project_id=project_id, + source_status=durable.status.value, + ) + return "failed" + if ( + durable.status is not status + or durable.updated_at != updated_at + or durable.resolved_at != resolved_at + ): + logger.info( + "working_memory_terminal_projection_source_advanced", + incident_id=incident_id, + project_id=project_id, + requested_status=status.value, + durable_status=durable.status.value, + ) + + working = await self.get_from_working_memory(incident_id) already_current = bool( - not hydrated_from_episodic - and incident.status in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED) - and current_updated_at >= durable_updated_at - and (resolved_at is None or incident.resolved_at is not None) + working is not None + and working.model_dump(mode="json") + == durable.model_dump(mode="json") ) - if already_current: - return "current" - - if incident.status != IncidentStatus.CLOSED: - incident.status = status - incident.updated_at = updated_at - incident.resolved_at = resolved_at or incident.resolved_at - if not await self.save_to_working_memory(incident): + if not already_current and not await self.save_to_working_memory(durable): logger.warning( "working_memory_terminal_projection_failed", incident_id=incident_id, @@ -721,14 +742,99 @@ class IncidentService: ) return "failed" - logger.info( - "working_memory_terminal_projection_updated", + if not await self._mark_terminal_projection_ready( incident_id=incident_id, project_id=project_id, - status=incident.status.value, - hydrated_from_episodic=hydrated_from_episodic, + status=durable.status, + updated_at=durable.updated_at, + ): + logger.warning( + "working_memory_terminal_projection_marker_failed", + incident_id=incident_id, + project_id=project_id, + ) + return "failed" + + logger.info( + ( + "working_memory_terminal_projection_current" + if already_current + else "working_memory_terminal_projection_updated" + ), + incident_id=incident_id, + project_id=project_id, + status=durable.status.value, ) - return "updated" + return "current" if already_current else "updated" + + async def _mark_terminal_projection_ready( + self, + *, + incident_id: str, + project_id: str, + status: IncidentStatus, + updated_at: datetime, + ) -> bool: + """Persist an ack for the exact durable row projected to Redis.""" + + try: + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + UPDATE incidents incident + SET outcome = CAST( + coalesce( + CASE + WHEN incident.outcome IS NULL + THEN '{}'::jsonb + WHEN jsonb_typeof( + CAST(incident.outcome AS jsonb) + ) = 'object' + THEN CAST(incident.outcome AS jsonb) + ELSE jsonb_build_object( + 'legacy_outcome', + CAST(incident.outcome AS jsonb) + ) + END, + '{}'::jsonb + ) || jsonb_build_object( + 'working_memory_projection', + jsonb_build_object( + 'ready', true, + 'incident_status', incident.status::text, + 'source_updated_at_epoch_us', CAST( + round( + extract( + epoch FROM incident.updated_at + ) * 1000000 + ) AS bigint + ), + 'projected_at', NOW() + ) + ) AS json + ) + WHERE incident.incident_id = :incident_id + AND incident.project_id = :project_id + AND upper(incident.status::text) = :incident_status + AND incident.updated_at = :updated_at + RETURNING incident.incident_id + """), + { + "incident_id": incident_id, + "project_id": project_id, + "incident_status": status.value.upper(), + "updated_at": updated_at, + }, + ) + return result.scalar_one_or_none() == incident_id + except Exception as exc: + logger.warning( + "working_memory_terminal_projection_marker_write_failed", + incident_id=incident_id, + project_id=project_id, + error_type=type(exc).__name__, + ) + return False async def get_for_readback( self, @@ -736,19 +842,23 @@ class IncidentService: *, project_id: str | None = None, ) -> Incident | None: - """Return durable incident truth, with Redis as an availability fallback.""" - incident = await self.get_from_episodic_memory( - incident_id, - project_id=project_id, - ) - if incident is not None: - return incident + """Return durable incident truth without silently trusting stale Redis.""" - logger.warning( - "incident_readback_falling_back_to_working_memory", - incident_id=incident_id, - ) - return await self.get_from_working_memory(incident_id) + try: + return await self._get_from_episodic_memory_strict( + incident_id, + project_id=project_id, + ) + except Exception as exc: + logger.warning( + "incident_durable_readback_unavailable", + incident_id=incident_id, + project_id=project_id, + error_type=type(exc).__name__, + ) + raise IncidentDurableReadError( + "incident_durable_readback_unavailable" + ) from exc async def get_active_incidents( self, @@ -758,8 +868,8 @@ class IncidentService: """ 列出所有活躍的 Incidents。 - PostgreSQL 是 durable source of truth；只有 repository 暫時不可用時， - 才退回 Redis working memory 維持讀取可用性。 + PostgreSQL 是 durable source of truth。Repository 不可用時必須 + fail closed，不可把 stale Redis 事件交給自動化 consumer。 Returns: list[Incident]: 活躍事件列表 (investigating 或 mitigating) @@ -779,67 +889,11 @@ class IncidentService: logger.warning( "active_incidents_episodic_read_failed", error=str(e), - fallback="working_memory", + fallback="disabled_fail_closed", ) - - redis_client = get_redis() - incidents: list[Incident] = [] - - try: - # SCAN 所有 incident:* keys - async for key in redis_client.scan_iter( - match=f"{INCIDENT_KEY_PREFIX}*", - count=100, - ): - # 排除索引 keys - if ":idx:" in key: - continue - - data = await redis_client.get(key) - if data is None: - continue - - try: - # 方案 C: 正規化舊格式 Enum 值 - incident_dict = json.loads(data) - if "status" in incident_dict: - incident_dict["status"] = normalize_status(incident_dict["status"]) - if "severity" in incident_dict: - incident_dict["severity"] = normalize_severity(incident_dict["severity"]) - - # 正規化 signals 內的 severity - for signal in incident_dict.get("signals", []): - if "severity" in signal: - signal["severity"] = normalize_severity(signal["severity"]) - - incident = Incident.model_validate(incident_dict) - - # 只返回活躍狀態的 Incident - if incident.status in ( - IncidentStatus.INVESTIGATING, - IncidentStatus.MITIGATING, - ): - incidents.append(incident) - except Exception as e: - logger.warning( - "incident_parse_error", - key=key, - error=str(e), - ) - continue - - logger.info( - "get_active_incidents", - count=len(incidents), - ) - return incidents - - except Exception as e: - logger.exception( - "get_active_incidents_error", - error=str(e), - ) - return [] + raise IncidentDurableReadError( + "active_incidents_durable_read_unavailable" + ) from e # ========================================================================= # Episodic Memory (PostgreSQL) @@ -884,6 +938,11 @@ class IncidentService: if incident.outcome else None ), + frequency_snapshot=( + incident.frequency_stats.model_dump(mode="json") + if incident.frequency_stats + else None + ), created_at=incident.created_at, updated_at=incident.updated_at, resolved_at=incident.resolved_at, @@ -927,23 +986,10 @@ class IncidentService: Incident | None: 事件資料，若不存在則返回 None """ try: - db_context = ( - get_db_context(project_id) if project_id else get_db_context() + return await self._get_from_episodic_memory_strict( + incident_id, + project_id=project_id, ) - async with db_context as db: - from sqlalchemy import select - - stmt = select(IncidentRecord).where( - IncidentRecord.incident_id == incident_id - ) - result = await db.execute(stmt) - record = result.scalar_one_or_none() - - if record is None: - return None - - # 轉換回 Pydantic model - return self._record_to_incident(record) except Exception as e: logger.exception( @@ -953,6 +999,22 @@ class IncidentService: ) return None + async def _get_from_episodic_memory_strict( + self, + incident_id: str, + *, + project_id: str | None = None, + ) -> Incident | None: + """Read one durable row and preserve DB/schema/context failures.""" + + async with get_db_context(project_id) as db: + stmt = select(IncidentRecord).where( + IncidentRecord.incident_id == incident_id + ) + result = await db.execute(stmt) + record = result.scalar_one_or_none() + return self._record_to_incident(record) if record is not None else None + def _record_to_incident(self, record: IncidentRecord) -> Incident: """ 將 SQLAlchemy record 轉換為 Pydantic Incident @@ -975,6 +1037,19 @@ class IncidentService: record.outcome, incident_id=record.incident_id, ) + frequency_stats = None + frequency_snapshot = getattr(record, "frequency_snapshot", None) + if frequency_snapshot: + try: + frequency_stats = IncidentFrequencyStats( + **frequency_snapshot + ) + except Exception as exc: + logger.warning( + "incident_frequency_snapshot_parse_failed", + incident_id=record.incident_id, + error=str(exc), + ) # 方案 C: 正規化舊格式 Enum 值 normalized_status = normalize_status(record.status) @@ -989,6 +1064,7 @@ class IncidentService: decision_chain=decision_chain, proposal_ids=record.proposal_ids or [], outcome=outcome, + frequency_stats=frequency_stats, created_at=record.created_at, updated_at=record.updated_at, resolved_at=record.resolved_at, diff --git a/apps/api/src/workers/ansible_executor_broker.py b/apps/api/src/workers/ansible_executor_broker.py index 21bc2812d..ada120b48 100644 --- a/apps/api/src/workers/ansible_executor_broker.py +++ b/apps/api/src/workers/ansible_executor_broker.py @@ -31,6 +31,22 @@ async def _heartbeat_loop(shutdown_event: asyncio.Event) -> None: pass +async def _init_optional_projection_redis() -> bool: + """Keep the controlled executor available when the Redis read model is down.""" + + try: + await init_redis_pool() + return True + except Exception as exc: + logger.warning( + "ansible_execution_broker_redis_projection_degraded", + error_type=type(exc).__name__, + controlled_executor_available=True, + durable_source="postgresql", + ) + return False + + async def _main() -> None: if not settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER: raise RuntimeError("ansible_execution_broker_disabled_by_config") @@ -40,7 +56,7 @@ async def _main() -> None: if not ssh_key_path.is_file() or not known_hosts_path.is_file(): raise RuntimeError("ansible_execution_broker_transport_not_mounted") - await init_redis_pool() + redis_projection_ready = await _init_optional_projection_redis() shutdown_event = asyncio.Event() @@ -73,6 +89,8 @@ async def _main() -> None: settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS ), worker_raw_credential_access=False, + redis_projection_ready=redis_projection_ready, + redis_projection_blocks_executor=False, ) try: diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index c75c6858d..fe5eedd99 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -8,9 +8,11 @@ from unittest.mock import AsyncMock import pytest +from src.core import redis_client as redis_client_module from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job from src.services import awooop_ansible_check_mode_service as service from src.services.telegram_gateway import TelegramGateway +from src.workers import ansible_executor_broker as broker class _MappingResult: @@ -269,7 +271,7 @@ async def test_closure_resolves_only_after_durable_readback( @pytest.mark.asyncio -async def test_closure_waits_for_working_memory_projection( +async def test_closure_preserves_durable_terminal_when_projection_is_degraded( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( @@ -282,7 +284,13 @@ async def test_closure_waits_for_working_memory_projection( "_commit_verified_incident_closure", AsyncMock(return_value={"working_memory_projection_ready": False}), ) - readback = AsyncMock() + readback = AsyncMock( + return_value={ + "closed": True, + "missing": [], + "incident_status": "RESOLVED", + } + ) monkeypatch.setattr(service, "_read_incident_closure_readback", readback) result = await service._finalize_verified_apply_closure( @@ -292,17 +300,21 @@ async def test_closure_waits_for_working_memory_projection( ) assert result == { - "status": "working_memory_projection_pending", - "closed": False, + "status": "controlled_apply_closed_projection_degraded", + "closed": True, "missing": ["working_memory_terminal_projection"], + "incident_status": "RESOLVED", + "working_memory_projection_ready": False, + "read_model_degraded": True, } - readback.assert_not_awaited() + readback.assert_awaited_once() @pytest.mark.asyncio async def test_terminal_working_memory_reconciler_projects_durable_rows( monkeypatch: pytest.MonkeyPatch, ) -> None: + service._WORKING_MEMORY_PROJECTION_CURSOR.clear() updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC) db = _SequenceDB( _RowsResult( @@ -354,7 +366,167 @@ async def test_terminal_working_memory_reconciler_projects_durable_rows( "project_id": "awoooi", "window_hours": 24, "limit": 6, + "cursor_updated_at": None, + "cursor_incident_id": "", } + query = db.statements[0] + assert "working_memory_projection,ready" in query + assert "source_updated_at_epoch_us" in query + assert "cursor_updated_at" in query + assert "ORDER BY incident.updated_at ASC" in query + assert service._WORKING_MEMORY_PROJECTION_WINDOW_HOURS >= 7 * 24 + service._WORKING_MEMORY_PROJECTION_CURSOR.clear() + + +@pytest.mark.asyncio +async def test_terminal_reconciler_advances_cursor_after_one_row_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service._WORKING_MEMORY_PROJECTION_CURSOR.clear() + first_at = datetime(2026, 7, 11, 11, 40, tzinfo=UTC) + second_at = datetime(2026, 7, 11, 11, 41, tzinfo=UTC) + db = _SequenceDB( + _RowsResult( + [ + { + "incident_id": "INC-20260711-FAIL01", + "incident_status": "RESOLVED", + "updated_at": first_at, + "resolved_at": first_at, + }, + { + "incident_id": "INC-20260711-NEXT01", + "incident_status": "RESOLVED", + "updated_at": second_at, + "resolved_at": second_at, + }, + ] + ) + ) + + @asynccontextmanager + async def fake_db_context(_project_id: str): + yield db + + project = AsyncMock( + side_effect=[RuntimeError("redis unavailable"), "updated"] + ) + monkeypatch.setattr(service, "get_db_context", fake_db_context) + monkeypatch.setattr( + service, + "_project_incident_terminal_to_working_memory", + project, + ) + + result = await service.reconcile_verified_incident_working_memory_once( + project_id="awoooi", + limit=2, + ) + + assert result["scanned"] == 2 + assert result["failed"] == 1 + assert result["updated"] == 1 + assert project.await_count == 2 + assert service._WORKING_MEMORY_PROJECTION_CURSOR["awoooi"] == ( + second_at, + "INC-20260711-NEXT01", + ) + service._WORKING_MEMORY_PROJECTION_CURSOR.clear() + + +@pytest.mark.asyncio +async def test_projection_exception_does_not_erase_durable_closure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + terminal = { + "automation_run_id": "00000000-0000-0000-0000-000000000102", + "apply_op_id": "00000000-0000-0000-0000-000000000104", + } + updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC) + db = _SequenceDB( + _MappingResult(), + _MappingResult( + { + "incident_status": "RESOLVED", + "updated_at": updated_at, + "resolved_at": updated_at, + "outcome": { + "automation_terminal": terminal, + "proposal_executed": True, + "execution_success": True, + }, + "closure_receipt_written": True, + "resolved_lifecycle_written": True, + } + ), + ) + transaction_completed = False + + @asynccontextmanager + async def fake_db_context(_project_id: str): + nonlocal transaction_completed + yield db + transaction_completed = True + + monkeypatch.setattr(service, "get_db_context", fake_db_context) + monkeypatch.setattr( + service, + "_project_incident_terminal_to_working_memory", + AsyncMock(side_effect=RuntimeError("redis unavailable")), + ) + + result = await service._commit_verified_incident_closure( + _claim(), + apply_op_id="00000000-0000-0000-0000-000000000104", + project_id="awoooi", + closure_prerequisite_count=12, + ) + + assert transaction_completed is True + assert result is not None + assert result["incident_resolved"] is True + assert result["closure_receipt_written"] is True + assert result["working_memory_projection_status"] == "failed" + assert result["working_memory_projection_ready"] is False + + +@pytest.mark.asyncio +async def test_broker_redis_projection_init_is_fail_soft( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + broker, + "init_redis_pool", + AsyncMock(side_effect=RuntimeError("redis unavailable")), + ) + + assert await broker._init_optional_projection_redis() is False + + +@pytest.mark.asyncio +async def test_failed_redis_init_does_not_poison_projection_retry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + failed_pool = type( + "_FailedPool", + (), + { + "ping": AsyncMock(side_effect=RuntimeError("redis unavailable")), + "close": AsyncMock(), + }, + )() + monkeypatch.setattr(redis_client_module, "_redis_pool", None) + monkeypatch.setattr( + redis_client_module.redis, + "from_url", + lambda *_args, **_kwargs: failed_pool, + ) + + with pytest.raises(RuntimeError, match="redis unavailable"): + await redis_client_module.init_redis_pool() + + assert redis_client_module._redis_pool is None + failed_pool.close.assert_awaited_once_with() @pytest.mark.asyncio diff --git a/apps/api/tests/test_incident_readback_source_truth.py b/apps/api/tests/test_incident_readback_source_truth.py index efd896b4f..8844f0338 100644 --- a/apps/api/tests/test_incident_readback_source_truth.py +++ b/apps/api/tests/test_incident_readback_source_truth.py @@ -4,11 +4,15 @@ from datetime import UTC, datetime from unittest.mock import AsyncMock import pytest +from fastapi import HTTPException from src.api.v1 import incidents as incidents_api from src.models.incident import Incident, IncidentStatus, Severity from src.repositories import incident_repository as incident_repository_module -from src.services.incident_service import IncidentService +from src.services.incident_service import ( + IncidentDurableReadError, + IncidentService, +) def _incident(status: IncidentStatus, *, minute: int) -> Incident: @@ -20,7 +24,12 @@ def _incident(status: IncidentStatus, *, minute: int) -> Incident: affected_services=["awoooi-api"], created_at=datetime(2026, 7, 11, 10, 0, tzinfo=UTC), updated_at=timestamp, - resolved_at=timestamp if status is IncidentStatus.RESOLVED else None, + resolved_at=( + timestamp + if status in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED) + else None + ), + closed_at=timestamp if status is IncidentStatus.CLOSED else None, ) @@ -31,7 +40,7 @@ async def test_readback_prefers_durable_resolved_incident_over_stale_cache(monke cached = _incident(IncidentStatus.INVESTIGATING, minute=11) monkeypatch.setattr( service, - "get_from_episodic_memory", + "_get_from_episodic_memory_strict", AsyncMock(return_value=durable), ) working_read = AsyncMock(return_value=cached) @@ -45,27 +54,28 @@ async def test_readback_prefers_durable_resolved_incident_over_stale_cache(monke assert result is durable assert result.status is IncidentStatus.RESOLVED working_read.assert_not_awaited() - service.get_from_episodic_memory.assert_awaited_once_with( + service._get_from_episodic_memory_strict.assert_awaited_once_with( durable.incident_id, project_id="awoooi", ) @pytest.mark.asyncio -async def test_readback_falls_back_to_working_memory_when_durable_record_unavailable( +async def test_readback_returns_missing_without_trusting_working_memory( monkeypatch, ): service = IncidentService() cached = _incident(IncidentStatus.INVESTIGATING, minute=11) monkeypatch.setattr( service, - "get_from_episodic_memory", + "_get_from_episodic_memory_strict", AsyncMock(return_value=None), ) + working_read = AsyncMock(return_value=cached) monkeypatch.setattr( service, "get_from_working_memory", - AsyncMock(return_value=cached), + working_read, ) result = await service.get_for_readback( @@ -73,7 +83,74 @@ async def test_readback_falls_back_to_working_memory_when_durable_record_unavail project_id="awoooi", ) - assert result is cached + assert result is None + working_read.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_readback_db_failure_is_degraded_not_cache_not_found(monkeypatch): + service = IncidentService() + working_read = AsyncMock( + return_value=_incident(IncidentStatus.INVESTIGATING, minute=11) + ) + monkeypatch.setattr( + service, + "_get_from_episodic_memory_strict", + AsyncMock(side_effect=RuntimeError("postgres unavailable")), + ) + monkeypatch.setattr(service, "get_from_working_memory", working_read) + + with pytest.raises( + IncidentDurableReadError, + match="incident_durable_readback_unavailable", + ): + await service.get_for_readback( + "INC-20260711-SOURCE-TRUTH", + project_id="awoooi", + ) + + working_read.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_terminal_projection_replaces_newer_wrong_cache_terminal( + monkeypatch, +): + service = IncidentService() + durable = _incident(IncidentStatus.CLOSED, minute=38) + cached = _incident(IncidentStatus.RESOLVED, minute=59) + save = AsyncMock(return_value=True) + marker = AsyncMock(return_value=True) + monkeypatch.setattr( + service, + "_get_from_episodic_memory_strict", + AsyncMock(return_value=durable), + ) + monkeypatch.setattr( + service, + "get_from_working_memory", + AsyncMock(return_value=cached), + ) + monkeypatch.setattr(service, "save_to_working_memory", save) + monkeypatch.setattr(service, "_mark_terminal_projection_ready", marker) + + result = await service.project_terminal_status_to_working_memory( + durable.incident_id, + project_id="awoooi", + status=IncidentStatus.CLOSED, + updated_at=durable.updated_at, + resolved_at=durable.resolved_at, + ) + + assert result == "updated" + save.assert_awaited_once_with(durable) + assert save.await_args.args[0].status is IncidentStatus.CLOSED + marker.assert_awaited_once_with( + incident_id=durable.incident_id, + project_id="awoooi", + status=IncidentStatus.CLOSED, + updated_at=durable.updated_at, + ) @pytest.mark.asyncio @@ -118,3 +195,55 @@ async def test_active_incidents_come_from_durable_repository(monkeypatch): assert incidents == [durable_active] repository.get_active.assert_awaited_once_with(project_id="awoooi") + + +@pytest.mark.asyncio +async def test_active_incidents_fail_closed_when_postgres_is_unavailable( + monkeypatch, +): + service = IncidentService() + + class _Repository: + get_active = AsyncMock(side_effect=RuntimeError("postgres unavailable")) + + monkeypatch.setattr( + incident_repository_module, + "get_incident_repository", + lambda: _Repository(), + ) + + with pytest.raises( + IncidentDurableReadError, + match="active_incidents_durable_read_unavailable", + ): + await service.get_active_incidents(project_id="awoooi") + + +@pytest.mark.asyncio +async def test_incident_detail_reports_durable_readback_degraded(monkeypatch): + service = type( + "_Service", + (), + { + "get_for_readback": AsyncMock( + side_effect=IncidentDurableReadError( + "incident_durable_readback_unavailable" + ) + ) + }, + )() + monkeypatch.setattr(incidents_api, "get_incident_service", lambda: service) + + with pytest.raises(HTTPException) as exc_info: + await incidents_api.get_incident( + "INC-20260711-SOURCE-TRUTH", + project_id="awoooi", + ) + + assert exc_info.value.status_code == 503 + assert exc_info.value.detail == { + "status": "degraded", + "reason_code": "incident_durable_readback_unavailable", + "source_of_truth": "postgresql", + "redis_fallback_used": False, + }