awoooi/apps/api/src/jobs/knowledge_decay_job.py

"""
AWOOOI AIOps Phase 3 — 知識遺忘 Job
=====================================
職責：每日掃描知識庫（knowledge_entries）中 30 天未被引用（view_count = 0
且 updated_at < now-30d）的草稿/審核條目，標記為 archived（知識遺忘）。

為什麼需要知識遺忘？
  短期學習偏差：AI 早期案例學習的修復模式可能已過時（K8s 版本、服務名稱改變）。
  若不遺忘，舊的 zero-evidence 條目會持續污染 RAG 檢索，
  拉低 Playbook 匹配精度，增加誤診率。

遺忘策略：
  - 對象：status in (draft, review) 且 view_count = 0 且 updated_at < now-30d
  - 動作：status → archived，tags 追加 'kb_decay_30d'
  - 豁免：status = approved（需人工封存）；status = archived（已封存）
  - 每次執行記錄摘要到 structlog（不寫 governance event，避免雜訊）

設計原則：
  1. 只標記，不刪除（符合 archive_not_delete 鐵律）
  2. 批次操作，每次最多 200 筆（避免長事務）
  3. Job 失敗只記錄 error，不影響主路徑

ADR-083 Phase 3: 知識遺忘（L7×D4）
2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from datetime import timedelta

import structlog
from sqlalchemy import and_, select, update

from src.db.base import get_session_factory
from src.db.models import KnowledgeEntryRecord
from src.models.knowledge import EntryStatus
from src.utils.timezone import now_taipei

logger = structlog.get_logger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# 常數
# ─────────────────────────────────────────────────────────────────────────────

DECAY_AGE_DAYS = 30
DECAY_TAG = "kb_decay_30d"
BATCH_LIMIT = 200
DAILY_INTERVAL_SEC = 86_400  # 24h


# ─────────────────────────────────────────────────────────────────────────────
# Data Types
# ─────────────────────────────────────────────────────────────────────────────

@dataclass
class DecayScanResult:
    """知識遺忘掃描結果"""
    total_scanned: int
    decayed_ids: list[str] = field(default_factory=list)
    scanned_at: str = field(default_factory=lambda: now_taipei().isoformat())

    @property
    def decayed_count(self) -> int:
        return len(self.decayed_ids)

    def to_dict(self) -> dict:
        return {
            "total_scanned": self.total_scanned,
            "decayed_count": self.decayed_count,
            "decayed_ids_sample": self.decayed_ids[:20],
            "scanned_at": self.scanned_at,
        }


# ─────────────────────────────────────────────────────────────────────────────
# Main Job
# ─────────────────────────────────────────────────────────────────────────────

class KnowledgeDecayJob:
    """
    知識遺忘 Job（每日執行）

    Usage:
        job = KnowledgeDecayJob()
        result = await job.run()
    """

    async def run(self) -> DecayScanResult:
        """
        完整執行：掃描 → 標記 archived（知識遺忘）。

        Returns:
            DecayScanResult
        """
        from src.core.feature_flags import aiops_flags
        if not aiops_flags.AIOPS_P3_ENABLED:
            logger.debug("knowledge_decay_job_skipped_feature_flag")
            return DecayScanResult(total_scanned=0)

        try:
            return await self._run_scan()
        except Exception as e:
            logger.error(
                "knowledge_decay_job_error",
                error=str(e),
            )
            return DecayScanResult(total_scanned=0)

    async def _run_scan(self) -> DecayScanResult:
        cutoff = now_taipei() - timedelta(days=DECAY_AGE_DAYS)
        decayable_statuses = [EntryStatus.DRAFT.value, EntryStatus.REVIEW.value]

        session_factory = get_session_factory()
        async with session_factory() as db:
            # 查：30 天未引用（view_count=0）且 updated_at < cutoff 的 draft/review 條目
            stmt = select(KnowledgeEntryRecord).where(
                and_(
                    KnowledgeEntryRecord.status.in_(decayable_statuses),
                    KnowledgeEntryRecord.view_count == 0,
                    KnowledgeEntryRecord.updated_at < cutoff,
                )
            ).limit(BATCH_LIMIT)

            result = await db.execute(stmt)
            entries = result.scalars().all()
            total_scanned = len(entries)

            if not entries:
                logger.debug("knowledge_decay_nothing_to_decay")
                return DecayScanResult(total_scanned=0)

            decayed_ids = []
            for entry in entries:
                # 追加 decay tag（不重複）
                current_tags: list[str] = list(entry.tags or [])
                if DECAY_TAG not in current_tags:
                    current_tags.append(DECAY_TAG)

                entry.status = EntryStatus.ARCHIVED.value
                entry.tags = current_tags
                decayed_ids.append(entry.id)

            await db.commit()

        result = DecayScanResult(
            total_scanned=total_scanned,
            decayed_ids=decayed_ids,
        )

        logger.info(
            "knowledge_decay_job_done",
            **result.to_dict(),
        )
        return result


# ─────────────────────────────────────────────────────────────────────────────
# Loop（掛載到 main.py）
# ─────────────────────────────────────────────────────────────────────────────

async def run_knowledge_decay_loop() -> None:
    """
    無限迴圈：每 24h 執行一次知識遺忘掃描。
    在 main.py startup 以 asyncio.create_task 掛載。
    """
    job = KnowledgeDecayJob()
    while True:
        try:
            result = await job.run()
            if result.decayed_count > 0:
                logger.info(
                    "knowledge_decay_loop_tick",
                    decayed=result.decayed_count,
                )
        except Exception as e:
            logger.error("knowledge_decay_loop_error", error=str(e))

        await asyncio.sleep(DAILY_INTERVAL_SEC)