feat(governance): queue stale km owner review

apps/api/src/api/v1/ai_governance.py

@@ -32,11 +32,17 @@ from src.models.governance import (
     KnowledgeReviewDraftArchiveResponse,
     KnowledgeReviewDraftDedupeResponse,
     KnowledgeStaleCandidatesResponse,
+    KnowledgeStaleOwnerReviewRequest,
+    KnowledgeStaleOwnerReviewResponse,
 )
 from src.services.governance_km_review_service import (
     KmReviewDraftArchiveError,
     archive_km_review_draft_duplicates,
 )
+from src.services.governance_km_stale_review_service import (
+    KmStaleOwnerReviewError,
+    queue_km_stale_owner_review,
+)
 from src.services.governance_query_service import (
     query_governance_events,
     query_governance_queue,
@@ -221,6 +227,36 @@ async def get_km_stale_candidates(
     return await query_km_stale_candidates(project_id=project_id, limit=limit)
 
 
+# =============================================================================
+# POST /api/v1/ai/governance/km-stale-candidates/{entry_id}/queue-review
+# =============================================================================
+
+@router.post(
+    "/ai/governance/km-stale-candidates/{entry_id}/queue-review",
+    response_model=KnowledgeStaleOwnerReviewResponse,
+)
+async def post_km_stale_candidate_queue_review(
+    entry_id: str,
+    request: KnowledgeStaleOwnerReviewRequest,
+) -> KnowledgeStaleOwnerReviewResponse:
+    """
+    將單筆 stale KM candidate 排入 owner review。
+
+    這個 endpoint 只建立治理事件與 dispatch work item，不修改 KM 內容。
+    實際 refresh / archive / supersede 仍需 owner 在後續流程確認。
+    """
+    logger.info(
+        "km_stale_candidate_queue_review_request",
+        entry_id=entry_id,
+        owner=request.owner,
+        dry_run=request.dry_run,
+    )
+    try:
+        return await queue_km_stale_owner_review(entry_id=entry_id, request=request)
+    except KmStaleOwnerReviewError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
+
+
 # =============================================================================
 # GET /api/v1/ai/governance/summary
 # =============================================================================

apps/api/src/models/governance.py

@@ -241,6 +241,33 @@ class KnowledgeStaleCandidatesResponse(BaseModel):
     generated_at: datetime
 
 
+class KnowledgeStaleOwnerReviewRequest(BaseModel):
+    owner: str = Field(default="operator_console", min_length=1, max_length=100)
+    owner_note: str | None = Field(default=None, max_length=240)
+    dry_run: bool = False
+
+
+class KnowledgeStaleOwnerReviewResponse(BaseModel):
+    schema_version: str = "km_stale_owner_review_v1"
+    entry_id: str
+    project_id: str
+    status: Literal["dry_run", "queued", "already_queued"]
+    governance_event_id: str | None = None
+    dispatch_id: str | None = None
+    workflow_stage: str
+    recommended_action: Literal[
+        "refresh_with_evidence",
+        "owner_review",
+        "archive_or_supersede",
+    ]
+    owner: str
+    owner_note: str | None = None
+    writes_km: bool = False
+    writes_governance_audit: bool
+    next_action: str = "owner_review_stale_km_candidate"
+    generated_at: datetime
+
+
 # =============================================================================
 # Endpoint 3: summary
 # =============================================================================

apps/api/src/services/governance_km_stale_review_service.py

@@ -0,0 +1,343 @@
+"""
+Governance KM Stale Review Service
+==================================
+
+Owner-review intake for stale KM priority candidates.
+
+這層只把 stale KM 候選排入治理工作項與 audit trail，不改寫 KM 內容。
+真正 refresh / archive / supersede 仍需 owner 後續審核。
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import Any, Literal
+
+import structlog
+from sqlalchemy import select, text
+
+from src.db.base import get_db_context
+from src.db.models import (
+    AiGovernanceEvent,
+    GovernanceRemediationDispatch,
+    KnowledgeEntryRecord,
+    generate_uuid,
+    taipei_now,
+)
+from src.models.governance import (
+    KnowledgeStaleOwnerReviewRequest,
+    KnowledgeStaleOwnerReviewResponse,
+)
+from src.models.knowledge import EntryStatus
+from src.services.governance_agent import KM_STALE_DAYS, KM_STALE_RATIO
+from src.services.governance_query_service import _build_km_stale_candidate
+from src.utils.timezone import now_taipei
+
+logger = structlog.get_logger(__name__)
+
+_EXECUTOR_TYPE = "hermes_km_stale_owner_review"
+
+
+class KmStaleOwnerReviewError(Exception):
+    """KM stale owner-review request failed validation."""
+
+    def __init__(self, status_code: int, detail: str) -> None:
+        super().__init__(detail)
+        self.status_code = status_code
+        self.detail = detail
+
+
+async def queue_km_stale_owner_review(
+    *,
+    entry_id: str,
+    request: KnowledgeStaleOwnerReviewRequest,
+) -> KnowledgeStaleOwnerReviewResponse:
+    """Queue a stale KM candidate for owner review without modifying KM content."""
+    record = await _load_stale_candidate_record(entry_id)
+    candidate = _build_km_stale_candidate(
+        record,
+        now=now_taipei(),
+        threshold_days=KM_STALE_DAYS,
+    )
+
+    existing = await _load_active_owner_review_dispatch(entry_id)
+    if existing is not None:
+        return _build_response(
+            entry_id=entry_id,
+            project_id=candidate.project_id,
+            status="already_queued",
+            governance_event_id=str(existing.governance_event_id),
+            dispatch_id=str(existing.id),
+            recommended_action=candidate.recommended_action,
+            owner=request.owner,
+            owner_note=request.owner_note,
+            writes_governance_audit=False,
+        )
+
+    if request.dry_run:
+        return _build_response(
+            entry_id=entry_id,
+            project_id=candidate.project_id,
+            status="dry_run",
+            recommended_action=candidate.recommended_action,
+            owner=request.owner,
+            owner_note=request.owner_note,
+            writes_governance_audit=False,
+        )
+
+    now = taipei_now()
+    event_id = generate_uuid()
+    dispatch_id = generate_uuid()
+    event_details = _build_stale_owner_review_event_details(
+        entry_id=entry_id,
+        candidate=candidate.model_dump(mode="json"),
+        owner=request.owner,
+        owner_note=request.owner_note,
+    )
+    decision_context = _build_stale_owner_review_decision_context(
+        governance_event_id=event_id,
+        entry_id=entry_id,
+        candidate=candidate.model_dump(mode="json"),
+        owner=request.owner,
+        owner_note=request.owner_note,
+    )
+
+    async with get_db_context() as db:
+        event = AiGovernanceEvent(
+            id=event_id,
+            event_type="knowledge_degradation",
+            triggered_at=now,
+            details=event_details,
+            resolved=False,
+        )
+        dispatch = GovernanceRemediationDispatch(
+            id=dispatch_id,
+            governance_event_id=event_id,
+            event_type="knowledge_degradation",
+            dispatch_status="pending",
+            playbook_id=None,
+            incident_id=None,
+            approval_id=None,
+            decision_context=decision_context,
+            executor_type=_EXECUTOR_TYPE,
+            attempt_count=0,
+            max_attempts=1,
+            dispatched_at=now,
+            created_by=request.owner[:100],
+        )
+        db.add(event)
+        db.add(dispatch)
+        await db.flush()
+
+    logger.info(
+        "km_stale_owner_review_queued",
+        entry_id=entry_id,
+        project_id=candidate.project_id,
+        governance_event_id=event_id,
+        dispatch_id=dispatch_id,
+        recommended_action=candidate.recommended_action,
+    )
+    return _build_response(
+        entry_id=entry_id,
+        project_id=candidate.project_id,
+        status="queued",
+        governance_event_id=event_id,
+        dispatch_id=dispatch_id,
+        recommended_action=candidate.recommended_action,
+        owner=request.owner,
+        owner_note=request.owner_note,
+        writes_governance_audit=True,
+    )
+
+
+async def _load_stale_candidate_record(entry_id: str) -> KnowledgeEntryRecord:
+    cutoff = now_taipei() - timedelta(days=KM_STALE_DAYS)
+    async with get_db_context() as db:
+        result = await db.execute(
+            select(KnowledgeEntryRecord).where(KnowledgeEntryRecord.id == entry_id)
+        )
+        record = result.scalar_one_or_none()
+
+    if record is None:
+        raise KmStaleOwnerReviewError(404, "KM entry not found")
+    if _enum_value(record.status) == EntryStatus.ARCHIVED.value:
+        raise KmStaleOwnerReviewError(409, "archived KM entries cannot be queued for stale review")
+    updated_at = record.updated_at
+    if updated_at is not None and updated_at.tzinfo is None:
+        updated_at = updated_at.replace(tzinfo=cutoff.tzinfo)
+    if updated_at is None or updated_at >= cutoff:
+        raise KmStaleOwnerReviewError(409, "KM entry is no longer past the stale threshold")
+    return record
+
+
+async def _load_active_owner_review_dispatch(
+    entry_id: str,
+) -> GovernanceRemediationDispatch | None:
+    sql = text("""
+        SELECT *
+        FROM governance_remediation_dispatch d
+        WHERE d.executor_type = :executor_type
+          AND d.dispatch_status::text IN ('pending', 'dispatched', 'executing')
+          AND (
+            d.decision_context -> 'workflow' ->> 'entry_id' = :entry_id
+            OR d.decision_context ->> 'entry_id' = :entry_id
+          )
+        ORDER BY d.dispatched_at DESC
+        LIMIT 1
+    """)
+    async with get_db_context() as db:
+        result = await db.execute(
+            select(GovernanceRemediationDispatch).from_statement(sql),
+            {"executor_type": _EXECUTOR_TYPE, "entry_id": entry_id},
+        )
+        return result.scalar_one_or_none()
+
+
+def _build_stale_owner_review_event_details(
+    *,
+    entry_id: str,
+    candidate: dict[str, Any],
+    owner: str,
+    owner_note: str | None,
+) -> dict[str, Any]:
+    return {
+        "schema_version": "km_stale_owner_review_event_v1",
+        "trigger_source": "stale_km_priority_queue",
+        "next_action": "owner_review_stale_km_candidate",
+        "impact": {
+            "status": "waiting_owner_review",
+            "metric": "stale_days",
+            "value": candidate.get("stale_days"),
+            "entry_id": entry_id,
+            "priority_tier": candidate.get("priority_tier"),
+            "priority_score": candidate.get("priority_score"),
+            "stale_ratio_threshold": KM_STALE_RATIO,
+            "stale_days_threshold": KM_STALE_DAYS,
+        },
+        "remediation": {
+            "next_action": "owner_review_stale_km_candidate",
+            "items": [
+                "review_current_incident_sentry_signoz_playbook_evidence",
+                "refresh_archive_or_supersede_after_owner_approval",
+                "run_stale_ratio_recheck_after_writeback",
+            ],
+        },
+        "ownership": _stale_owner_review_ownership(),
+        "candidate": candidate,
+        "owner": owner,
+        "owner_note": owner_note,
+    }
+
+
+def _build_stale_owner_review_decision_context(
+    *,
+    governance_event_id: str,
+    entry_id: str,
+    candidate: dict[str, Any],
+    owner: str,
+    owner_note: str | None,
+) -> dict[str, Any]:
+    recommended_action = str(candidate.get("recommended_action") or "owner_review")
+    return {
+        "schema_version": "km_stale_owner_review_dispatch_v1",
+        "version": "v1",
+        "trigger_source": "stale_km_priority_queue",
+        "triggered_metric": "knowledge_degradation",
+        "metric_value": candidate.get("stale_days"),
+        "threshold": KM_STALE_DAYS,
+        "suggested_action": recommended_action,
+        "next_action": "owner_review_stale_km_candidate",
+        "decision_path": "pending_owner_review",
+        "ownership": _stale_owner_review_ownership(),
+        "entry_id": entry_id,
+        "candidate": candidate,
+        "workflow": {
+            "work_item_id": (
+                "governance:knowledge_degradation:"
+                f"{governance_event_id}:km_stale_owner_review:{entry_id}"
+            ),
+            "work_kind": "km_stale_owner_review",
+            "current_stage": "waiting_owner_review",
+            "entry_id": entry_id,
+            "project_id": candidate.get("project_id"),
+            "priority_tier": candidate.get("priority_tier"),
+            "priority_score": candidate.get("priority_score"),
+            "recommended_action": recommended_action,
+            "steps": [
+                "detected",
+                "prioritized_stale_candidate",
+                "waiting_owner_review",
+                "owner_updates_or_archives_km",
+                "stale_ratio_recheck",
+            ],
+            "stage_by_dispatch_status": {
+                "pending": "waiting_owner_review",
+                "dispatched": "waiting_owner_review",
+                "executing": "owner_review_in_progress",
+                "succeeded": "km_candidate_reviewed",
+                "failed": "needs_manual_km_triage",
+                "skipped": "waiting_owner_review",
+                "cancelled": "cancelled",
+            },
+            "next_action": "owner_review_stale_km_candidate",
+            "needs_human_review": True,
+            "writes_km_without_approval": False,
+            "writes_km": False,
+            "stale_ratio_recheck_after_writeback": True,
+        },
+        "worker_result": {
+            "status": "queued_owner_review",
+            "entry_id": entry_id,
+            "recommended_action": recommended_action,
+            "writes_km": False,
+        },
+        "owner": owner,
+        "owner_note": owner_note,
+    }
+
+
+def _stale_owner_review_ownership() -> dict[str, Any]:
+    return {
+        "lead_agent": "Hermes",
+        "support_agents": [
+            "OpenClaw：補 Incident / 規則 / PlayBook 脈絡，不直接批量改寫 KM。",
+            "ElephantAlpha：read-only 稽核 owner review 草稿與風險。",
+        ],
+        "human_owner": "KM owner / SRE owner",
+    }
+
+
+def _build_response(
+    *,
+    entry_id: str,
+    project_id: str,
+    status: Literal["dry_run", "queued", "already_queued"],
+    recommended_action: Literal[
+        "refresh_with_evidence",
+        "owner_review",
+        "archive_or_supersede",
+    ],
+    owner: str,
+    writes_governance_audit: bool,
+    owner_note: str | None = None,
+    governance_event_id: str | None = None,
+    dispatch_id: str | None = None,
+) -> KnowledgeStaleOwnerReviewResponse:
+    return KnowledgeStaleOwnerReviewResponse(
+        entry_id=entry_id,
+        project_id=project_id,
+        status=status,
+        governance_event_id=governance_event_id,
+        dispatch_id=dispatch_id,
+        workflow_stage="waiting_owner_review",
+        recommended_action=recommended_action,
+        owner=owner,
+        owner_note=owner_note,
+        writes_km=False,
+        writes_governance_audit=writes_governance_audit,
+        generated_at=now_taipei(),
+    )
+
+
+def _enum_value(value: Any) -> str:
+    return str(value.value if hasattr(value, "value") else value)