V2 P2: Abstention + KnowledgeGap (#76)

api/gaps.py

@@ -0,0 +1,50 @@
+"""Gaps and abstention API."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Query
+from pydantic import BaseModel, Field
+
+from api.deps import RolesDep
+from gaps.detector import coverage_score, get_gaps, maybe_abstain, record_gap
+
+router = APIRouter(prefix="/gaps", tags=["gaps"])
+
+
+class AbstainRequest(BaseModel):
+    workspace_id: str
+    domain: str
+    memory_confidence: float = Field(ge=0.0, le=1.0)
+    claim_count: int = Field(default=0, ge=0)
+    missing: list[str] = Field(default_factory=list)
+
+
+@router.post("/abstain")
+def post_abstain(body: AbstainRequest, _roles: RolesDep) -> dict:
+    cov = coverage_score(claim_count=body.claim_count)
+    abstain = maybe_abstain(
+        memory_confidence=body.memory_confidence,
+        coverage=cov,
+        missing=body.missing,
+    )
+    if abstain:
+        record_gap(
+            workspace_id=body.workspace_id,
+            domain=body.domain,
+            gap_type="MISSING_KNOWLEDGE",
+            description=f"Abstention on {body.domain}",
+            coverage=cov,
+        )
+    return {
+        "coverage_score": cov,
+        "abstention": abstain,
+    }
+
+
+@router.get("")
+def list_gaps(
+    _roles: RolesDep,
+    workspace_id: str = Query(...),
+) -> dict:
+    gaps = [g.model_dump(mode="json") for g in get_gaps(workspace_id)]
+    return {"workspace_id": workspace_id, "gaps": gaps, "total": len(gaps)}

api/main.py

@@ -31,6 +31,7 @@
 from api.memory_state import router as memory_state_router
 from api.reliability import router as reliability_router
 from api.outcomes import router as outcomes_router
+from api.gaps import router as gaps_router
 from api.metrics import record_http_request, record_query, render_metrics
 from api.decisions import router as decisions_router
 from api.deps import RolesDep, memory, set_memory_service
@@ -113,6 +114,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 app.include_router(memory_state_router)
 app.include_router(reliability_router)
 app.include_router(outcomes_router)
+app.include_router(gaps_router)
 
 
 @app.middleware("http")

gaps/__init__.py

@@ -0,0 +1,5 @@
+"""Abstention + knowledge gap package."""
+
+from gaps.detector import coverage_score, get_gaps, maybe_abstain, record_gap
+
+__all__ = ["coverage_score", "get_gaps", "maybe_abstain", "record_gap"]

gaps/detector.py

@@ -0,0 +1,70 @@
+"""Knowledge gap detection and coverage estimates."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from shared.v2_models import KnowledgeGap
+
+_GAPS: dict[str, KnowledgeGap] = {}
+
+
+def coverage_score(*, claim_count: int, expected_min: int = 3) -> float:
+    """Simple domain coverage estimate in [0, 1]."""
+    if expected_min <= 0:
+        return 1.0
+    return max(0.0, min(1.0, claim_count / float(expected_min)))
+
+
+def maybe_abstain(
+    *,
+    memory_confidence: float,
+    coverage: float,
+    missing: list[str] | None = None,
+) -> dict[str, Any] | None:
+    """Return abstention payload when evidence is insufficient."""
+    if memory_confidence >= 0.45 and coverage >= 0.5:
+        return None
+    return {
+        "status": "INSUFFICIENT_EVIDENCE",
+        "memory_confidence": memory_confidence,
+        "coverage": coverage,
+        "missing": missing or [],
+        "next_action": "ASK",
+    }
+
+
+def record_gap(
+    *,
+    workspace_id: str,
+    domain: str,
+    gap_type: str,
+    description: str,
+    coverage: float,
+) -> KnowledgeGap:
+    """Create or bump a knowledge gap for repeated abstentions."""
+    key = f"{workspace_id}:{domain}:{gap_type}"
+    existing = _GAPS.get(key)
+    now = datetime.now(UTC)
+    if existing:
+        existing.frequency += 1
+        existing.last_seen = now
+        existing.coverage_score = coverage
+        existing.description = description
+        return existing
+    gap = KnowledgeGap(
+        workspace_id=workspace_id,
+        domain=domain,
+        gap_type=gap_type,
+        description=description,
+        coverage_score=coverage,
+        first_seen=now,
+        last_seen=now,
+    )
+    _GAPS[key] = gap
+    return gap
+
+
+def get_gaps(workspace_id: str) -> list[KnowledgeGap]:
+    return [g for g in _GAPS.values() if g.workspace_id == workspace_id]

tests/gaps/test_detector.py

@@ -0,0 +1,32 @@
+from gaps.detector import coverage_score, maybe_abstain, record_gap, get_gaps
+
+
+def test_coverage_scales() -> None:
+    assert coverage_score(claim_count=0) == 0.0
+    assert coverage_score(claim_count=3) == 1.0
+
+
+def test_abstain_when_thin() -> None:
+    out = maybe_abstain(memory_confidence=0.2, coverage=0.1, missing=["rollback"])
+    assert out is not None
+    assert out["next_action"] == "ASK"
+
+
+def test_gap_aggregates() -> None:
+    record_gap(
+        workspace_id="ws",
+        domain="payments",
+        gap_type="MISSING_KNOWLEDGE",
+        description="no rollback",
+        coverage=0.2,
+    )
+    record_gap(
+        workspace_id="ws",
+        domain="payments",
+        gap_type="MISSING_KNOWLEDGE",
+        description="still missing",
+        coverage=0.2,
+    )
+    gaps = get_gaps("ws")
+    assert len(gaps) == 1
+    assert gaps[0].frequency == 2