From e96055eef95669d1b120655be6be0117973a43da Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 26 Apr 2026 20:19:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(p0.4):=20Playbook=20=E5=AD=B8=E7=BF=92?= =?UTF-8?q?=E9=8F=88=E4=B8=89=E9=81=93=E4=BF=AE=E5=BE=A9=20=E2=80=94=20par?= =?UTF-8?q?tial=20index=20+=20race=E9=98=B2=E8=AD=B7=20+=20=E6=89=8B?= =?UTF-8?q?=E5=8B=95=E8=B7=AF=E5=BE=91=E6=8E=A5=E7=B7=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR-092 P0.4 Playbook EWMA 學習閉環的 DB / Repository / Service 三層修補。 DB 層 (db-expert-fix by Engineer-B): - ApprovalRecord.matched_playbook_id 移除 index=True，改 __table_args__ partial index (WHERE matched_playbook_id IS NOT NULL) — 多數列 NULL，full index 浪費空間 - adr092_p1_learning_chain_rollback.sql: 純 ROLLBACK SQL（DBA 手動執行） Repository 層: - playbook_repository.py: SELECT FOR UPDATE 防 lost update 避免並發 EWMA 更新覆蓋彼此 Service 層 (P0.4 修復): - proposal_service.py: 手動審核路徑補 _try_playbook_match_id 呼叫 decision_manager auto_execute 路徑已有此邏輯（行 2035）， 此處補手動路徑缺口，使 matched_playbook_id 可寫入 DB → EWMA 才能演化 測試: - test_playbook_repository_race_condition.py: 3 cases SELECT FOR UPDATE 防 race 正確阻擋並發 EWMA 更新（pass） Note: migration SQL 待 DBA 手動執行（feedback_dev_prod_separation.md）， 不執行 alembic upgrade（statu 文件禁忌條款）。 Co-Authored-By: Claude Opus 4.7 (1M context) --- .../adr092_p1_learning_chain_rollback.sql | 18 ++ apps/api/src/db/models.py | 12 +- .../src/repositories/playbook_repository.py | 50 +++-- apps/api/src/services/proposal_service.py | 56 +++++ ...test_playbook_repository_race_condition.py | 193 ++++++++++++++++++ 5 files changed, 310 insertions(+), 19 deletions(-) create mode 100644 apps/api/migrations/adr092_p1_learning_chain_rollback.sql create mode 100644 apps/api/tests/integration/test_playbook_repository_race_condition.py diff --git a/apps/api/migrations/adr092_p1_learning_chain_rollback.sql b/apps/api/migrations/adr092_p1_learning_chain_rollback.sql new file mode 100644 index 000000000..934887dd6 --- /dev/null +++ b/apps/api/migrations/adr092_p1_learning_chain_rollback.sql @@ -0,0 +1,18 @@ +-- ADR-092 P1 Learning Chain Rollback +-- 撤銷 adr092_p1_learning_chain_fix.sql 的所有變更 +-- 僅在 schema 誤套 / 緊急回滾時使用；資料不可復原 +-- +-- 執行方式（需人工執行一次）： +-- psql $DATABASE_URL -f apps/api/migrations/adr092_p1_learning_chain_rollback.sql +-- +-- 2026-04-25 db-expert-fix by Claude Engineer-B + +BEGIN; + +DROP INDEX IF EXISTS ix_approval_matched_playbook; +ALTER TABLE approval_records DROP COLUMN IF EXISTS matched_playbook_id; + +DROP INDEX IF EXISTS ix_timeline_incident_id; +ALTER TABLE timeline_events DROP COLUMN IF EXISTS incident_id; + +COMMIT; diff --git a/apps/api/src/db/models.py b/apps/api/src/db/models.py index 0d853e0fe..c1b27acf4 100644 --- a/apps/api/src/db/models.py +++ b/apps/api/src/db/models.py @@ -22,6 +22,7 @@ from sqlalchemy import ( Integer, String, Text, + text, ) from sqlalchemy import ( Enum as SQLEnum, @@ -170,10 +171,11 @@ class ApprovalRecord(Base): # B2 fix 2026-04-24 ogt + Claude Sonnet 4.6: Playbook 學習閉環斷鏈修復 # 原欄位缺失 → 人工審核後 matched_playbook_id 永遠 NULL → EWMA 無法更新 + # 2026-04-25 db-expert-fix by Claude Engineer-B: 移除 index=True 避免自動生成 full index + # Partial index 改在 __table_args__ 宣告（WHERE matched_playbook_id IS NOT NULL） matched_playbook_id: Mapped[str | None] = mapped_column( String(36), nullable=True, - index=True, comment="匹配的 Playbook ID，學習服務用以更新 EWMA trust score", ) @@ -203,7 +205,13 @@ class ApprovalRecord(Base): Index("ix_approval_created_at", "created_at"), Index("ix_approval_requested_by", "requested_by"), Index("ix_approval_fingerprint", "fingerprint"), # 戰略 B: 指紋查詢優化 - Index("ix_approval_matched_playbook", "matched_playbook_id"), # B2 fix + # 2026-04-25 db-expert-fix by Claude Engineer-B: 改為 partial index，只索引非 NULL 值 + # 原 full index 與 index=True 三重宣告衝突已修復（一個來源真相：此處） + Index( + "ix_approval_matched_playbook", + "matched_playbook_id", + postgresql_where=text("matched_playbook_id IS NOT NULL"), + ), ) diff --git a/apps/api/src/repositories/playbook_repository.py b/apps/api/src/repositories/playbook_repository.py index 650aa215d..40d9c7de9 100644 --- a/apps/api/src/repositories/playbook_repository.py +++ b/apps/api/src/repositories/playbook_repository.py @@ -342,30 +342,46 @@ class PlaybookRepository: 失敗: trust_new = 0.8 * trust_old + 0.2 * 0.0（衰減速度 2x） trust < 0.1 → 記錄警告，由 Evolver Agent 封存 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3.5 EWMA 雙寫 PG + 2026-04-25 db-expert-fix by Claude Engineer-B: SELECT FOR UPDATE 防 race condition（Lost Update） """ try: - playbook = await self.get_by_id(playbook_id) - if not playbook: - return False + # 使用 SELECT FOR UPDATE 確保並行 update_stats 不會 lost update + factory = get_session_factory() + async with factory() as session: + async with session.begin(): + stmt = ( + select(PlaybookRecord) + .where(PlaybookRecord.playbook_id == playbook_id) + .with_for_update() + ) + result = await session.execute(stmt) + record = result.scalar_one_or_none() + if record is None: + return False - if success: - playbook.success_count += 1 - playbook.trust_score = 0.9 * playbook.trust_score + 0.1 * 1.0 - else: - playbook.failure_count += 1 - playbook.trust_score = 0.8 * playbook.trust_score + 0.2 * 0.0 + if success: + record.success_count += 1 + record.trust_score = 0.9 * record.trust_score + 0.1 * 1.0 + else: + record.failure_count += 1 + record.trust_score = 0.8 * record.trust_score + 0.2 * 0.0 - playbook.trust_score = max(0.0, min(1.0, playbook.trust_score)) - playbook.last_used_at = now_taipei() + record.trust_score = max(0.0, min(1.0, record.trust_score)) + record.last_used_at = now_taipei() + # session.begin() context manager 結束時自動 commit - # 雙寫（PG + Redis） - await self.update(playbook) + # 讀取最新值供 Redis 雙寫 + 日誌（PG 已 commit，此時 Redis 同步） + updated = await self._pg_get(playbook_id) + if updated is None: + return True # PG 已更新，Redis 同步失敗不影響結果 - if playbook.trust_score < 0.1: + await self._redis_set(updated) + + if updated.trust_score < 0.1: logger.warning( "playbook_trust_low_auto_archive_candidate", playbook_id=playbook_id, - trust_score=playbook.trust_score, + trust_score=updated.trust_score, hint="Evolver Agent 應將此 Playbook 封存", ) @@ -373,8 +389,8 @@ class PlaybookRepository: "playbook_stats_updated", playbook_id=playbook_id, success=success, - success_rate=playbook.success_rate, - trust_score=playbook.trust_score, + success_rate=updated.success_rate, + trust_score=updated.trust_score, ) return True diff --git a/apps/api/src/services/proposal_service.py b/apps/api/src/services/proposal_service.py index bc661d834..e20610113 100644 --- a/apps/api/src/services/proposal_service.py +++ b/apps/api/src/services/proposal_service.py @@ -246,6 +246,12 @@ class ProposalService: metadata["signoz_correlation"] = llm_proposal.get("signoz_correlation", "") metadata["optimization_suggestions"] = llm_proposal.get("optimization_suggestions", []) + # 2026-04-25 P0.4 修復 by Claude Engineer-B: + # 手動路徑（API 呼叫 generate_proposal）補 Playbook RAG 匹配， + # 讓 matched_playbook_id 得以寫入 DB，學習服務 EWMA 才能更新 trust score。 + # decision_manager auto_execute 路徑已有此邏輯（行 2035），此處補手動路徑缺口。 + matched_pb_id: str | None = await self._try_playbook_match_id(incident) + approval_create = ApprovalRequestCreate( action=action, description=description, @@ -255,6 +261,7 @@ class ProposalService: requested_by="OpenClaw AI", incident_id=incident_id, metadata=metadata, + matched_playbook_id=matched_pb_id, ) approval = await self._approval_service.create_approval(approval_create) @@ -354,6 +361,55 @@ class ProposalService: return action_type, action, description + # ========================================================================= + # 輔助方法: Playbook RAG 匹配（P0.4 2026-04-25 by Claude Engineer-B） + # ========================================================================= + + async def _try_playbook_match_id(self, incident: Incident) -> str | None: + """ + 嘗試 Playbook RAG 匹配，回傳 matched_playbook_id（相似度 >= 0.85 才填）。 + + 設計動機：手動路徑（generate_proposal）補 matched_playbook_id， + 讓學習服務 EWMA 能在人工審核後更新 Playbook trust score。 + 邏輯與 decision_manager._try_playbook_match 相同，但只回傳 ID 不改 action。 + 失敗時靜默返回 None（不阻塞主流程）。 + """ + PLAYBOOK_SIMILARITY_THRESHOLD = 0.85 + try: + from src.models.playbook import SymptomPattern + from src.services.playbook_service import get_playbook_service + + alert_names = [s.alert_name for s in incident.signals] if incident.signals else [] + symptoms = SymptomPattern( + alert_names=alert_names, + affected_services=incident.affected_services or [], + severity_range=[incident.severity.value] if incident.severity else ["P2"], + ) + recommendations = await get_playbook_service().get_recommendations( + symptoms=symptoms, + top_k=1, + ) + if not recommendations: + return None + best_match = recommendations[0] + if best_match.similarity_score < PLAYBOOK_SIMILARITY_THRESHOLD: + return None + pb_id = best_match.playbook.playbook_id + logger.info( + "proposal_playbook_matched", + incident_id=incident.incident_id, + playbook_id=pb_id, + similarity=best_match.similarity_score, + ) + return pb_id + except Exception as e: + logger.debug( + "proposal_playbook_match_skipped", + incident_id=getattr(incident, "incident_id", "?"), + error=str(e), + ) + return None + # ========================================================================= # 輔助方法: 建立 BlastRadius # ========================================================================= diff --git a/apps/api/tests/integration/test_playbook_repository_race_condition.py b/apps/api/tests/integration/test_playbook_repository_race_condition.py new file mode 100644 index 000000000..05d35821a --- /dev/null +++ b/apps/api/tests/integration/test_playbook_repository_race_condition.py @@ -0,0 +1,193 @@ +""" +Playbook Repository Race Condition 整合測試 +============================================= +驗證 update_stats SELECT FOR UPDATE 防止 Lost Update + +2026-04-25 db-expert-fix by Claude Engineer-B +""" + +import asyncio +import os +import uuid + +import pytest +import pytest_asyncio +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from src.db.models import PlaybookRecord + +# ============================================================================= +# DB URL（與 integration conftest 一致） +# ============================================================================= + +DEV_DB_URL = os.environ.get( + "TEST_DATABASE_URL", + "postgresql+asyncpg://awoooi:awoooi_prod_2026@192.168.0.188:5432/awoooi_dev?ssl=disable", +) + +# 禁止誤打 prod +assert "awoooi_prod" not in DEV_DB_URL.split("/")[-1], ( + "TEST_DATABASE_URL 不可指向 prod DB（awoooi_prod）" +) + + +# ============================================================================= +# Helpers +# ============================================================================= + +def _make_factory(engine): + return async_sessionmaker( + bind=engine, + class_=AsyncSession, + expire_on_commit=False, + autoflush=False, + ) + + +def _test_playbook_id() -> str: + """每次測試使用唯一 ID，避免跨測試污染""" + return f"PB-RACE-TEST-{uuid.uuid4().hex[:8].upper()}" + + +async def _insert_test_playbook(engine, playbook_id: str) -> None: + """直接 INSERT 測試用 Playbook（不走 Repository，避免 Redis 依賴）""" + from sqlalchemy import insert + factory = _make_factory(engine) + async with factory() as session: + async with session.begin(): + stmt = insert(PlaybookRecord).values( + playbook_id=playbook_id, + name="Race Test Playbook", + description="用於 race condition 測試", + status="active", + source="manual", + symptom_pattern={"alert_names": ["TestAlert"], "affected_services": [], "keywords": []}, + repair_steps=[], + estimated_duration_minutes=5, + source_incident_ids=[], + ai_confidence=0.8, + success_count=0, + failure_count=0, + trust_score=0.5, + requires_approval_level="auto", + stateful_targets=[], + requires_pre_backup=False, + tags=[], + ) + await session.execute(stmt) + + +async def _delete_test_playbook(engine, playbook_id: str) -> None: + """清理測試資料""" + from sqlalchemy import delete + factory = _make_factory(engine) + async with factory() as session: + async with session.begin(): + stmt = delete(PlaybookRecord).where(PlaybookRecord.playbook_id == playbook_id) + await session.execute(stmt) + + +async def _get_success_count(engine, playbook_id: str) -> int: + """直接從 PG 讀取 success_count""" + from sqlalchemy import select + factory = _make_factory(engine) + async with factory() as session: + result = await session.get(PlaybookRecord, playbook_id) + return result.success_count if result else -1 + + +# ============================================================================= +# 單一 update_stats via SELECT FOR UPDATE（不走 Redis，直接操作 ORM） +# ============================================================================= + +async def _update_stats_direct(engine, playbook_id: str, success: bool) -> bool: + """ + 模擬 PlaybookRepository.update_stats 的 PG 部分（SELECT FOR UPDATE） + 不走 Redis，僅測試 PG transaction 正確性 + """ + from sqlalchemy import select + factory = _make_factory(engine) + async with factory() as session: + async with session.begin(): + stmt = ( + select(PlaybookRecord) + .where(PlaybookRecord.playbook_id == playbook_id) + .with_for_update() + ) + result = await session.execute(stmt) + record = result.scalar_one_or_none() + if record is None: + return False + + if success: + record.success_count += 1 + record.trust_score = 0.9 * record.trust_score + 0.1 * 1.0 + else: + record.failure_count += 1 + record.trust_score = 0.8 * record.trust_score + 0.2 * 0.0 + + record.trust_score = max(0.0, min(1.0, record.trust_score)) + return True + + +# ============================================================================= +# Race Condition Test +# ============================================================================= + +@pytest.mark.integration +class TestUpdateStatsRaceCondition: + """ + 驗證 SELECT FOR UPDATE 在並行情況下不會發生 Lost Update + + 設計： + - INSERT 1 筆 PlaybookRecord（success_count=0） + - 5 個 coroutine 同時呼叫 _update_stats_direct(success=True) + - 預期：success_count == 5（每個 update 都被序列化，無 lost update） + """ + + @pytest_asyncio.fixture(autouse=True) + async def setup_and_teardown(self): + """每個測試建立獨立 engine + 測試資料，結束後清理""" + self.engine = create_async_engine(DEV_DB_URL, echo=False) + self.playbook_id = _test_playbook_id() + await _insert_test_playbook(self.engine, self.playbook_id) + yield + await _delete_test_playbook(self.engine, self.playbook_id) + await self.engine.dispose() + + async def test_update_stats_race_condition_no_lost_update(self): + """ + 同時 5 個 successful update_stats 對同一 playbook， + success_count 必須最終 == 5（無 lost update） + + 2026-04-25 db-expert-fix by Claude Engineer-B + """ + results = await asyncio.gather(*[ + _update_stats_direct(self.engine, self.playbook_id, success=True) + for _ in range(5) + ]) + + # 所有 5 個必須成功 + assert all(results), f"部分 update_stats 回傳 False: {results}" + + # PG 中 success_count 必須精確 == 5（無 lost update） + final_count = await _get_success_count(self.engine, self.playbook_id) + assert final_count == 5, ( + f"Lost update 偵測：success_count={final_count}，預期 5。" + "SELECT FOR UPDATE 未正確序列化並行寫入。" + ) + + async def test_update_stats_not_found_returns_false(self): + """不存在的 playbook_id 應回傳 False""" + result = await _update_stats_direct(self.engine, "PB-NOT-EXIST-9999", success=True) + assert result is False + + async def test_update_stats_failure_increments_failure_count(self): + """失敗的 update_stats 應遞增 failure_count，不影響 success_count""" + await _update_stats_direct(self.engine, self.playbook_id, success=False) + + factory = _make_factory(self.engine) + async with factory() as session: + record = await session.get(PlaybookRecord, self.playbook_id) + assert record.failure_count == 1 + assert record.success_count == 0