diff --git a/README.md b/README.md
index 0201748..488c5ba 100644
--- a/README.md
+++ b/README.md
@@ -1,12 +1,12 @@
-# GovernsAI Precheck
+# GovernsAI Precheck — The Enforcement Engine
 
 [![npm](https://img.shields.io/npm/v/%40governs-ai%2Fsdk?label=npm%20%40governs-ai%2Fsdk)](https://www.npmjs.com/package/@governs-ai/sdk)
 [![PyPI](https://img.shields.io/pypi/v/governs-ai-sdk?label=PyPI%20governs-ai-sdk)](https://pypi.org/project/governs-ai-sdk/)
 [![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)
 
-**Fully Open Source (MIT)** - PII detection and policy evaluation service for AI applications.
+**GovernsAI is the AI governance layer where policy actually enforces.** This service is the enforcement engine: it reads the per-org policy from the dashboard, runs PII detection (Presidio + regex), applies network-scope and tool-deny rules, and returns an `allow`, `transform`, or `deny` decision on every prompt, tool call, and response — without a redeploy.
 
-This service provides real-time policy evaluation and PII detection/redaction for AI tool usage. You can use it, modify it, and even offer it as a hosted service - no restrictions.
+**Fully Open Source (MIT)** — use it, modify it, host it. No restrictions.
 
 ## Features
 
diff --git a/app/api.py b/app/api.py
index 3d8ef47..f1a5a06 100644
--- a/app/api.py
+++ b/app/api.py
@@ -8,7 +8,7 @@
 from datetime import datetime
 from typing import List, Optional, Tuple
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Response
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response
 from sqlalchemy.orm import Session
 
 from .auth import AuthContext, require_api_key
@@ -155,6 +155,37 @@ async def health():
     return {"ok": True, "service": "governsai-precheck", "version": "0.1.0"}
 
 
+@router.post("/v1/internal/policy/invalidate")
+async def invalidate_policy_cache(request: Request, payload: dict):
+    """Drop the cached active policy for a single org.
+
+    Called by the dashboard after a policy create/update/delete (ADR-005).
+    Authenticated via HMAC over `org_id` with `KEY_HMAC_SECRET` — the same
+    shared secret used by api-key sync, so no new secret material to manage.
+
+    Body:   {"org_id": "<id>"}
+    Header: X-Govs-Invalidate-HMAC: hex(hmac_sha256(KEY_HMAC_SECRET, org_id))
+    """
+    import hashlib as _hashlib
+    import hmac as _hmac
+
+    from .policy_source import invalidate
+    from .settings import settings as _settings
+
+    org_id = (payload or {}).get("org_id")
+    if not isinstance(org_id, str) or not org_id:
+        raise HTTPException(status_code=400, detail="org_id required")
+
+    secret = _settings.key_hmac_secret.encode()
+    expected = _hmac.new(secret, org_id.encode(), _hashlib.sha256).hexdigest()
+    provided = request.headers.get("x-govs-invalidate-hmac", "")
+    if not _hmac.compare_digest(expected, provided):
+        raise HTTPException(status_code=401, detail="invalid signature")
+
+    invalidate(org_id)
+    return {"invalidated": org_id}
+
+
 @router.get("/v1/ready")
 async def ready():
     """
@@ -343,6 +374,7 @@ async def precheck(
                 tool_config=tool_config,
                 user_id=user_id,
                 budget_context=budget_context,
+                org_id=org_id,
             )
 
             # Add budget info to result if not already present
@@ -496,6 +528,7 @@ async def postcheck(
             tool_config=tool_config,
             user_id=user_id,
             budget_context=budget_context,
+            org_id=org_id,
         )
 
         # Add budget info to result if not already present
diff --git a/app/policies.py b/app/policies.py
index 9f9e5cc..73093c2 100644
--- a/app/policies.py
+++ b/app/policies.py
@@ -910,15 +910,35 @@ def evaluate_with_payload_policy(
     tool_config: Optional[Dict] = None,
     user_id: Optional[str] = None,
     budget_context: Optional[Dict] = None,
+    org_id: Optional[str] = None,
 ) -> Dict:
     """
-    Evaluate policy using payload-provided configuration
-    Falls back to the loaded static YAML policy if no policy_config is provided.
+    Evaluate policy with the following precedence (ADR-005):
+      1. Payload `policy_config` if provided (backwards-compatible)
+      2. Dashboard-managed policy fetched by `org_id` from the shared DB
+      3. Static YAML policy (legacy fallback)
     """
 
-    resolved_policy_config = (
-        deepcopy(policy_config) if policy_config else deepcopy(get_policy())
-    )
+    if policy_config:
+        resolved_policy_config = deepcopy(policy_config)
+    else:
+        org_policy = None
+        if org_id:
+            try:
+                from .policy_source import get_active_policy
+
+                org_policy = get_active_policy(org_id)
+            except Exception as exc:  # never break the request path on a fetcher bug
+                org_policy = None
+                # Use module logger; import is local to avoid circular load risk.
+                import logging as _logging
+
+                _logging.getLogger(__name__).warning(
+                    "policy_source fetch failed for org=%s err=%s", org_id, exc
+                )
+        resolved_policy_config = (
+            deepcopy(org_policy) if org_policy else deepcopy(get_policy())
+        )
     resolved_policy_config["tool"] = tool
     resolved_policy_config["scope"] = scope or ""
 
diff --git a/app/policy_source.py b/app/policy_source.py
new file mode 100644
index 0000000..f3a0952
--- /dev/null
+++ b/app/policy_source.py
@@ -0,0 +1,187 @@
+"""Org-scoped policy source — reads from the dashboard's `Policy` table.
+
+See ADR-005 (knowledge/ADR/005-policy-source-of-truth.md) for the design.
+
+Public surface:
+    get_active_policy(org_id: str) -> dict | None
+    invalidate(org_id: str) -> None
+    cache_stats() -> dict   # for diagnostics / metrics
+
+The returned policy dict has the shape that `evaluate_with_payload_policy`
+already understands (precheck/app/policies.py), so the evaluator does not
+need to learn a new format — we just translate at the edge.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import threading
+import time
+from dataclasses import dataclass
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from .storage import DashboardPolicy, SessionLocal
+
+logger = logging.getLogger(__name__)
+
+
+def _ttl_seconds() -> int:
+    raw = os.environ.get("POLICY_CACHE_TTL_S", "60")
+    try:
+        return max(1, int(raw))
+    except ValueError:
+        return 60
+
+
+@dataclass(frozen=True)
+class _CacheEntry:
+    policy: Optional[dict]
+    fetched_at: float
+
+
+_cache: dict[str, _CacheEntry] = {}
+_cache_lock = threading.RLock()
+_metrics = {"hit": 0, "miss": 0, "invalidate": 0}
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Shape translation: dashboard Policy row → precheck PolicyConfig dict
+# ──────────────────────────────────────────────────────────────────────
+# The dashboard stores `defaults` as a free-form JSON. The convention we
+# enforce in v1 is `{"pii": "<action>"}` where action ∈
+# {"redact", "block", "tokenize", "allow"}. The evaluator wants the precheck
+# shape `defaults[direction]["action"]` with action ∈
+# {"deny", "redact", "tokenize", "pass_through"}.
+_PII_ACTION_MAP = {
+    "redact":       "redact",
+    "block":        "deny",
+    "deny":         "deny",
+    "tokenize":     "tokenize",
+    "pass":         "pass_through",
+    "pass_through": "pass_through",
+    "allow":        "pass_through",
+}
+
+
+def _map_row_to_policy_config(row: DashboardPolicy) -> dict:
+    """Translate a DashboardPolicy row into a precheck PolicyConfig dict.
+
+    The output shape is consumed verbatim by `evaluate_with_payload_policy`
+    (precheck/app/policies.py). Unknown keys in `row.defaults` are preserved
+    verbatim so future extensions don't need to touch this function.
+    """
+    raw_defaults = row.defaults or {}
+
+    # Translate the v1 convention; default to "redact" if the field is absent.
+    pii_action_in = str(raw_defaults.get("pii", "redact")).lower()
+    pii_action = _PII_ACTION_MAP.get(pii_action_in, "redact")
+
+    return {
+        "version": row.version or "v1",
+        "defaults": {
+            "ingress": {"action": pii_action},
+            "egress":  {"action": pii_action},
+            # Forward any non-pii defaults verbatim for forward-compat.
+            **{k: v for k, v in raw_defaults.items() if k != "pii"},
+        },
+        "tool_access":    row.tool_access or {},
+        "deny_tools":     row.deny_tools or [],
+        "allow_tools":    row.allow_tools or [],
+        "network_scopes": row.network_scopes or [],
+        "network_tools":  row.network_tools or [],
+        "on_error":       row.on_error or "block",
+        # Provenance — useful in logs and audit, ignored by the evaluator.
+        "_policy_id":     row.id,
+        "_policy_name":   row.name,
+        "_priority":      row.priority,
+    }
+
+
+# ──────────────────────────────────────────────────────────────────────
+# DB read
+# ──────────────────────────────────────────────────────────────────────
+def _fetch_from_db(org_id: str) -> Optional[dict]:
+    db: Session = SessionLocal()
+    try:
+        row = (
+            db.query(DashboardPolicy)
+            .filter(
+                DashboardPolicy.org_id == org_id,
+                DashboardPolicy.is_active.is_(True),
+            )
+            .order_by(DashboardPolicy.priority.desc(), DashboardPolicy.updated_at.desc())
+            .first()
+        )
+        if row is None:
+            return None
+        return _map_row_to_policy_config(row)
+    except Exception as exc:
+        # Don't blow up the request path on a DB hiccup — let the caller fall
+        # back to the YAML policy. Logged loudly so it's visible in audits.
+        logger.warning(
+            "policy_source: db fetch failed for org=%s err=%s", org_id, exc
+        )
+        return None
+    finally:
+        db.close()
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Public API
+# ──────────────────────────────────────────────────────────────────────
+def get_active_policy(org_id: Optional[str]) -> Optional[dict]:
+    """Return the highest-priority active policy for `org_id`, or None.
+
+    None means "no row in DB" — the caller should fall back to the YAML
+    default policy (preserving today's behavior for orgs without a
+    dashboard-managed policy).
+    """
+    if not org_id:
+        return None
+
+    now = time.monotonic()
+    ttl = _ttl_seconds()
+
+    with _cache_lock:
+        entry = _cache.get(org_id)
+        if entry is not None and (now - entry.fetched_at) < ttl:
+            _metrics["hit"] += 1
+            return entry.policy
+
+    # Cache miss — fetch outside the lock to avoid holding it across IO.
+    fetched = _fetch_from_db(org_id)
+
+    with _cache_lock:
+        _cache[org_id] = _CacheEntry(policy=fetched, fetched_at=time.monotonic())
+        _metrics["miss"] += 1
+
+    return fetched
+
+
+def invalidate(org_id: Optional[str]) -> None:
+    """Drop the cached entry for org_id. No-op if missing or org_id is empty.
+
+    Called from the dashboard's invalidation webhook on policy writes
+    (see ADR-005 §Write/invalidate path).
+    """
+    if not org_id:
+        return
+    with _cache_lock:
+        _cache.pop(org_id, None)
+        _metrics["invalidate"] += 1
+
+
+def cache_stats() -> dict:
+    """Return cache hit/miss/invalidate counts and current entry count."""
+    with _cache_lock:
+        return {**_metrics, "entries": len(_cache)}
+
+
+def _clear_for_tests() -> None:
+    """Test-only helper — clears the cache and resets counters."""
+    with _cache_lock:
+        _cache.clear()
+        for k in _metrics:
+            _metrics[k] = 0
diff --git a/app/storage.py b/app/storage.py
index 9696704..f34dc1a 100644
--- a/app/storage.py
+++ b/app/storage.py
@@ -2,6 +2,7 @@
 from typing import Optional
 
 from sqlalchemy import (
+    JSON,
     Boolean,
     Column,
     DateTime,
@@ -42,6 +43,10 @@ class APIKey(Base):
 
 
 class Policy(Base):
+    """Legacy precheck-local policy table — kept for backward compat with the
+    YAML-import flow (Phase 2.3). Org-scoped policies live in DashboardPolicy
+    below, which mirrors the dashboard's Prisma Policy model. See ADR-005."""
+
     __tablename__ = "policies"
 
     id = Column(String, primary_key=True)
@@ -51,6 +56,37 @@ class Policy(Base):
     is_active = Column(Boolean, default=True)
 
 
+class DashboardPolicy(Base):
+    """Read-only mirror of the dashboard's `Policy` table (Prisma model).
+
+    Precheck reads this table directly via the shared Postgres connection;
+    the dashboard owns writes. See ADR-005 (policy source of truth).
+
+    The table name is `Policy` (Prisma default — model name unchanged).
+    Column names use the Prisma @map snake_case form where applicable.
+    """
+
+    __tablename__ = "Policy"
+
+    id = Column(String, primary_key=True)
+    org_id = Column("org_id", String, nullable=False, index=True)
+    user_id = Column("user_id", String, nullable=True)
+    name = Column(String, nullable=False)
+    description = Column(String, nullable=True)
+    version = Column(String, nullable=False, default="v1")
+    defaults = Column(JSON, nullable=False)
+    tool_access = Column("tool_access", JSON, nullable=False, default=dict)
+    deny_tools = Column("deny_tools", JSON, nullable=False, default=list)
+    allow_tools = Column("allow_tools", JSON, nullable=False, default=list)
+    network_scopes = Column("network_scopes", JSON, nullable=False, default=list)
+    network_tools = Column("network_tools", JSON, nullable=False, default=list)
+    on_error = Column("on_error", String, nullable=False, default="block")
+    is_active = Column("isActive", Boolean, nullable=False, default=True)
+    priority = Column(Integer, nullable=False, default=0)
+    created_at = Column("createdAt", DateTime, default=datetime.utcnow)
+    updated_at = Column("updatedAt", DateTime, default=datetime.utcnow)
+
+
 class UsageEvent(Base):
     __tablename__ = "usage_events"
 
@@ -108,8 +144,15 @@ class BudgetTransaction(Base):
 
 
 def create_tables():
-    """Create all tables"""
-    Base.metadata.create_all(bind=engine)
+    """Create all precheck-owned tables.
+
+    `DashboardPolicy` is intentionally excluded — that table is owned and
+    migrated by the dashboard (Prisma). Precheck only reads it. See ADR-005.
+    """
+    owned_tables = [
+        t for t in Base.metadata.sorted_tables if t.name != "Policy"
+    ]
+    Base.metadata.create_all(bind=engine, tables=owned_tables)
 
 
 def get_db():
diff --git a/tests/test_policy_invalidate_endpoint.py b/tests/test_policy_invalidate_endpoint.py
new file mode 100644
index 0000000..255853b
--- /dev/null
+++ b/tests/test_policy_invalidate_endpoint.py
@@ -0,0 +1,65 @@
+# SPDX-License-Identifier: MIT
+"""HMAC-gated /api/v1/internal/policy/invalidate endpoint tests (ADR-005)."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+
+from app import policy_source
+from app.settings import settings
+
+
+def _hmac(org_id: str) -> str:
+    return hmac.new(
+        settings.key_hmac_secret.encode(), org_id.encode(), hashlib.sha256
+    ).hexdigest()
+
+
+def test_invalidate_with_valid_hmac_drops_cache(test_client, monkeypatch):
+    # seed a cache entry
+    monkeypatch.setattr(policy_source, "_fetch_from_db", lambda _oid: {"x": 1})
+    policy_source.get_active_policy("org-z")
+    assert policy_source.cache_stats()["entries"] >= 1
+
+    r = test_client.post(
+        "/api/v1/internal/policy/invalidate",
+        json={"org_id": "org-z"},
+        headers={"x-govs-invalidate-hmac": _hmac("org-z")},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json() == {"invalidated": "org-z"}
+
+
+def test_invalidate_rejects_bad_hmac(test_client):
+    r = test_client.post(
+        "/api/v1/internal/policy/invalidate",
+        json={"org_id": "org-z"},
+        headers={"x-govs-invalidate-hmac": "not-a-real-hmac"},
+    )
+    assert r.status_code == 401
+
+
+def test_invalidate_rejects_missing_hmac(test_client):
+    r = test_client.post(
+        "/api/v1/internal/policy/invalidate",
+        json={"org_id": "org-z"},
+    )
+    assert r.status_code == 401
+
+
+def test_invalidate_requires_org_id(test_client):
+    r = test_client.post(
+        "/api/v1/internal/policy/invalidate",
+        json={},
+        headers={"x-govs-invalidate-hmac": _hmac("")},
+    )
+    assert r.status_code == 400
+
+
+def test_invalidate_rejects_hmac_signed_for_different_org(test_client):
+    r = test_client.post(
+        "/api/v1/internal/policy/invalidate",
+        json={"org_id": "org-a"},
+        headers={"x-govs-invalidate-hmac": _hmac("org-b")},
+    )
+    assert r.status_code == 401
diff --git a/tests/test_policy_source.py b/tests/test_policy_source.py
new file mode 100644
index 0000000..d0ad0e9
--- /dev/null
+++ b/tests/test_policy_source.py
@@ -0,0 +1,237 @@
+# SPDX-License-Identifier: MIT
+"""Unit tests for app.policy_source — the TTL-cached org-scoped policy fetcher.
+
+See ADR-005 for design. Tests cover cache hit/miss/expiry, missing-org, priority
+ordering, the dashboard-shape → precheck-shape translation, and invalidate().
+"""
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from app import policy_source
+from app.storage import DashboardPolicy
+
+
+@pytest.fixture(autouse=True)
+def _patch_session_local(monkeypatch, db_session):
+    """Make policy_source.SessionLocal yield the in-memory test session.
+
+    The fetcher opens its own session each call (and closes it in `finally`),
+    so we hand it a callable that returns the shared test session and stub
+    out .close() to avoid killing the suite-wide session.
+    """
+
+    class _SessionAdapter:
+        def __init__(self, sess):
+            self._sess = sess
+
+        def __getattr__(self, name):
+            return getattr(self._sess, name)
+
+        def close(self):
+            # The autouse `_reset_db` fixture in conftest handles teardown.
+            return None
+
+    monkeypatch.setattr(
+        policy_source, "SessionLocal", lambda: _SessionAdapter(db_session)
+    )
+    policy_source._clear_for_tests()
+    yield
+    policy_source._clear_for_tests()
+
+
+def _make_policy(
+    db_session,
+    *,
+    org_id: str,
+    name: str = "p",
+    defaults: dict | None = None,
+    priority: int = 0,
+    is_active: bool = True,
+    on_error: str = "block",
+    deny_tools=None,
+    allow_tools=None,
+):
+    row = DashboardPolicy(
+        id=f"{org_id}-{name}",
+        org_id=org_id,
+        name=name,
+        version="v1",
+        defaults=defaults if defaults is not None else {"pii": "redact"},
+        tool_access={},
+        deny_tools=deny_tools or [],
+        allow_tools=allow_tools or [],
+        network_scopes=[],
+        network_tools=[],
+        on_error=on_error,
+        is_active=is_active,
+        priority=priority,
+    )
+    db_session.add(row)
+    db_session.commit()
+    return row
+
+
+# ─── cache miss + DB fetch ────────────────────────────────────────────────
+def test_get_active_policy_returns_none_for_missing_org():
+    assert policy_source.get_active_policy("does-not-exist") is None
+
+
+def test_get_active_policy_fetches_from_db(db_session):
+    _make_policy(db_session, org_id="org-1", defaults={"pii": "redact"})
+    cfg = policy_source.get_active_policy("org-1")
+
+    assert cfg is not None
+    assert cfg["defaults"]["ingress"]["action"] == "redact"
+    assert cfg["defaults"]["egress"]["action"] == "redact"
+    assert cfg["on_error"] == "block"
+    assert cfg["_policy_id"] == "org-1-p"
+
+
+def test_pii_block_maps_to_deny(db_session):
+    _make_policy(db_session, org_id="org-2", defaults={"pii": "block"})
+    cfg = policy_source.get_active_policy("org-2")
+    assert cfg["defaults"]["ingress"]["action"] == "deny"
+    assert cfg["defaults"]["egress"]["action"] == "deny"
+
+
+def test_pii_tokenize_maps_through(db_session):
+    _make_policy(db_session, org_id="org-3", defaults={"pii": "tokenize"})
+    cfg = policy_source.get_active_policy("org-3")
+    assert cfg["defaults"]["ingress"]["action"] == "tokenize"
+
+
+def test_pii_allow_maps_to_pass_through(db_session):
+    _make_policy(db_session, org_id="org-4", defaults={"pii": "allow"})
+    cfg = policy_source.get_active_policy("org-4")
+    assert cfg["defaults"]["ingress"]["action"] == "pass_through"
+
+
+def test_unknown_pii_action_falls_back_to_redact(db_session):
+    _make_policy(db_session, org_id="org-5", defaults={"pii": "shrug"})
+    cfg = policy_source.get_active_policy("org-5")
+    assert cfg["defaults"]["ingress"]["action"] == "redact"
+
+
+# ─── priority order ───────────────────────────────────────────────────────
+def test_highest_priority_active_policy_wins(db_session):
+    _make_policy(db_session, org_id="org-9", name="lo", defaults={"pii": "redact"}, priority=1)
+    _make_policy(db_session, org_id="org-9", name="hi", defaults={"pii": "block"}, priority=10)
+    cfg = policy_source.get_active_policy("org-9")
+    # priority=10 ("block") must beat priority=1 ("redact")
+    assert cfg["defaults"]["ingress"]["action"] == "deny"
+    assert cfg["_policy_name"] == "hi"
+
+
+def test_inactive_policy_ignored(db_session):
+    _make_policy(db_session, org_id="org-10", name="dead", defaults={"pii": "block"}, priority=999, is_active=False)
+    _make_policy(db_session, org_id="org-10", name="live", defaults={"pii": "redact"}, priority=1, is_active=True)
+    cfg = policy_source.get_active_policy("org-10")
+    assert cfg["_policy_name"] == "live"
+
+
+# ─── cache behavior ───────────────────────────────────────────────────────
+def test_cache_hit_does_not_requery_db(db_session, monkeypatch):
+    _make_policy(db_session, org_id="org-h", defaults={"pii": "redact"})
+    # first call → miss
+    policy_source.get_active_policy("org-h")
+
+    calls = {"n": 0}
+    real_fetch = policy_source._fetch_from_db
+
+    def spy(org_id):
+        calls["n"] += 1
+        return real_fetch(org_id)
+
+    monkeypatch.setattr(policy_source, "_fetch_from_db", spy)
+
+    # second + third calls within TTL → must NOT hit the spy
+    policy_source.get_active_policy("org-h")
+    policy_source.get_active_policy("org-h")
+    assert calls["n"] == 0
+
+
+def test_invalidate_forces_refetch(db_session):
+    _make_policy(db_session, org_id="org-i", defaults={"pii": "redact"})
+    first = policy_source.get_active_policy("org-i")
+    assert first["defaults"]["ingress"]["action"] == "redact"
+
+    # mutate the underlying row
+    row = db_session.query(DashboardPolicy).filter(DashboardPolicy.org_id == "org-i").one()
+    row.defaults = {"pii": "block"}
+    db_session.commit()
+
+    # without invalidation, the cache should still serve the old value
+    cached = policy_source.get_active_policy("org-i")
+    assert cached["defaults"]["ingress"]["action"] == "redact", "cache should still hold stale entry"
+
+    # invalidate → next call refetches and reflects the change
+    policy_source.invalidate("org-i")
+    refreshed = policy_source.get_active_policy("org-i")
+    assert refreshed["defaults"]["ingress"]["action"] == "deny"
+
+
+def test_invalidate_unknown_org_is_noop():
+    # Must not raise.
+    policy_source.invalidate("never-cached")
+    policy_source.invalidate(None)
+
+
+def test_ttl_expiry_forces_refetch(db_session, monkeypatch):
+    monkeypatch.setenv("POLICY_CACHE_TTL_S", "1")
+    _make_policy(db_session, org_id="org-t", defaults={"pii": "redact"})
+
+    policy_source.get_active_policy("org-t")
+    # advance the clock past TTL
+    base = time.monotonic()
+    monkeypatch.setattr(time, "monotonic", lambda: base + 5)
+
+    # mutate the row; without a hit-spy, the new value proves a refetch happened
+    row = db_session.query(DashboardPolicy).filter(DashboardPolicy.org_id == "org-t").one()
+    row.defaults = {"pii": "block"}
+    db_session.commit()
+
+    refreshed = policy_source.get_active_policy("org-t")
+    assert refreshed["defaults"]["ingress"]["action"] == "deny"
+
+
+# ─── translation surface ──────────────────────────────────────────────────
+def test_translation_preserves_arrays_and_on_error(db_session):
+    _make_policy(
+        db_session,
+        org_id="org-x",
+        deny_tools=["python.exec"],
+        allow_tools=["chat"],
+        defaults={"pii": "redact"},
+        on_error="pass",
+    )
+    cfg = policy_source.get_active_policy("org-x")
+    assert cfg["deny_tools"] == ["python.exec"]
+    assert cfg["allow_tools"] == ["chat"]
+    assert cfg["on_error"] == "pass"
+
+
+def test_translation_forwards_unknown_default_keys(db_session):
+    _make_policy(
+        db_session,
+        org_id="org-y",
+        defaults={"pii": "redact", "future_knob": "value"},
+    )
+    cfg = policy_source.get_active_policy("org-y")
+    assert cfg["defaults"]["future_knob"] == "value"
+
+
+# ─── observability ────────────────────────────────────────────────────────
+def test_cache_stats_increments(db_session):
+    _make_policy(db_session, org_id="org-s", defaults={"pii": "redact"})
+    s0 = policy_source.cache_stats()
+    policy_source.get_active_policy("org-s")  # miss
+    policy_source.get_active_policy("org-s")  # hit
+    policy_source.invalidate("org-s")
+    s1 = policy_source.cache_stats()
+
+    assert s1["miss"] >= s0["miss"] + 1
+    assert s1["hit"] >= s0["hit"] + 1
+    assert s1["invalidate"] >= s0["invalidate"] + 1