diff --git a/api/policy/__init__.py b/api/policy/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/api/policy/projection_config.yaml b/api/policy/projection_config.yaml
new file mode 100644
index 00000000..8020f546
--- /dev/null
+++ b/api/policy/projection_config.yaml
@@ -0,0 +1,28 @@
+# Visibility / projection config for the read-path audience gate (policy/visibility.py).
+#
+# LOCAL COPY (do NOT cross-import the IndigenomicsAI otter_notion copy). Kept in
+# sync by hand and reviewed like code. Seeded from the KG→Notion projection spec
+# §6.4 denylist + Phase-2 concept allowlist.
+#
+# entity_denylist          — entities that must NEVER reach a team/public surface,
+#                            matched case-insensitively against entity_text + aliases.
+# concept_projection_allowlist — the only Concepts that may ever be visible (a Concept
+#                            not on this list is fail-closed to invisible).
+#
+# Override the path with PROJECTION_CONFIG_PATH; a missing/empty file → empty sets
+# (which, for concepts, means NO concept is ever visible — fail-closed).
+
+entity_denylist:
+  - "Hydro One"
+  - "Layer B"
+  - "Layer B Expansion Opportunity"
+
+concept_projection_allowlist:
+  - "Living Library Model"
+  - "regenerative bonds"
+  - "relational infrastructure"
+  - "bioregional nervous system"
+  - "IP for nature"
+  - "commons pool"
+  - "reserved compute credits"
+  - "data sovereignty"
diff --git a/api/policy/visibility.py b/api/policy/visibility.py
new file mode 100644
index 00000000..c2c8507f
--- /dev/null
+++ b/api/policy/visibility.py
@@ -0,0 +1,316 @@
+"""Audience-scoped, fail-closed visibility kernel for the read path.
+
+This is the async, asyncpg-facing sibling of the ``is_projectable`` kernel in
+``IndigenomicsAI/scripts/otter_notion/entity_projector.py``. It answers one
+question — *may this ``audience`` see this entity?* — over the NEW 4-value
+``entity_registry.visibility_scope`` axis introduced by migration 107.
+
+It exists so the ~50 ``AND NOT node_private`` read-path sites can migrate, one at
+a time, from a single boolean privacy flag to an audience-scoped gate WITHOUT
+changing behaviour until entities are actually classified (today every row is
+``'unclassified'``).
+
+Two exports:
+
+  * ``async visible_at(conn, uri, audience) -> bool`` — the per-entity decision,
+    fail-closed on every ambiguity and degrade-closed on any exception.
+  * ``visibility_predicate(alias, scopes) -> str`` — the reusable, injection-safe
+    SQL fragment that the read-path query sites splice in place of
+    ``AND NOT node_private``. Because it is a DROP-IN replacement for that clause it
+    KEEPS the ``node_private`` hard-deny AND adds the audience-scope gate, emitting
+    ``(NOT COALESCE(<alias>.node_private, false) AND <alias>.visibility_scope IN (…))``.
+
+Convenience: ``scopes_for(audience)`` (alias of ``scopes_for_audience``) turns an
+audience into its allowed-scope set, so a call site can write
+``visibility_predicate("er", scopes_for("team"))``.
+
+Config caching: the denylist / concept-allowlist YAML is read ONCE and cached for the
+life of the process (operator-owned, reviewed like code). To pick up an edit, restart
+the process or call ``_reset_config_cache_for_tests()`` to force a reload.
+
+TWO SEPARATE AXES — never conflate:
+
+  * ``entity_registry.visibility_scope`` — NEW, 4-value AUDIENCE gate
+    (``public`` | ``team`` | ``confidential`` | ``unclassified``). THIS module reads
+    ONLY this column.
+  * ``entity_rid_mappings.visibility_scope`` — PRE-EXISTING, 2-value projection-privacy
+    (``public`` | ``node_private``) consumed by the Notion projector. NOT read here.
+
+Fail-closed / degrade-closed contract:
+
+  * unknown uri (no registry row)                → False
+  * ``node_private`` true (hard override)         → False
+  * name/alias on the confidential denylist       → False
+  * scope not in the audience's allowed set       → False
+  * ``'unclassified'`` scope                       → visible ONLY to the
+    ``confidential`` (full-trust internal) audience; NEVER to ``team`` / ``public``
+  * a Concept not in the ``Concepts/`` vault folder AND on the allowlist → False
+  * ANY exception anywhere                         → False
+
+No hard dependency on ``asyncpg`` (the connection is duck-typed: it only needs
+``await conn.fetchrow(sql, *args)`` / ``await conn.fetch(sql, *args)``), so the
+unit tests can drive it with a fake connection and no live DB.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Audience → allowed-scope mapping (the whole visibility policy, in one table)
+# ---------------------------------------------------------------------------
+
+# The 4 legal values of entity_registry.visibility_scope (migration 107 CHECK).
+VALID_SCOPES: frozenset[str] = frozenset(
+    {"public", "team", "confidential", "unclassified"}
+)
+
+# The audiences a caller may ask about. NOTE: 'unclassified' is a SCOPE, not an
+# audience — it is deliberately absent here, so asking for it fails closed.
+VALID_AUDIENCES: frozenset[str] = frozenset({"public", "team", "confidential"})
+
+# What each audience is allowed to see. Monotone: public ⊆ team ⊆ confidential.
+#   * public       → only genuinely-public entities.
+#   * team         → public + team (NOT unclassified — fail-closed for unclassified).
+#   * confidential  → EVERYTHING, including still-unclassified rows. This is the
+#     full-trust internal/admin viewer; during the migration-107 rollout every row
+#     is 'unclassified', so excluding it here would black out the internal app.
+_AUDIENCE_SCOPES: dict[str, tuple[str, ...]] = {
+    "public": ("public",),
+    "team": ("public", "team"),
+    "confidential": ("public", "team", "confidential", "unclassified"),
+}
+
+_ALIAS_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+# Bundled config lives beside this module (LOCAL copy — never cross-import the
+# IndigenomicsAI otter_notion projection_config.yaml). Override with an env var.
+_DEFAULT_CONFIG_PATH = str(Path(__file__).with_name("projection_config.yaml"))
+
+
+def scopes_for_audience(audience: str) -> tuple[str, ...]:
+    """Return the allowed-scope tuple for ``audience``; unknown audience → ``()``.
+
+    Mapping: ``public → {public}``; ``team → {public, team}``; ``confidential → all
+    four scopes (incl. 'unclassified')``.
+
+    ROLLOUT NOTE: ``'unclassified'`` rows stay INVISIBLE to the ``team`` and ``public``
+    audiences during the migration-107 rollout (their scope sets exclude it). Only the
+    full-trust ``confidential`` audience sees unclassified rows. Callers on the
+    internal / live read paths therefore pass the ``'confidential'`` (or ``'team'``,
+    once rows are classified) audience DELIBERATELY, so that the graph does not go dark
+    while every row is still ``'unclassified'``.
+
+    An empty tuple is the fail-closed answer: it makes ``visibility_predicate``
+    emit a match-nothing predicate and ``visible_at`` deny.
+    """
+    if not isinstance(audience, str):
+        return ()
+    return _AUDIENCE_SCOPES.get(audience, ())
+
+
+# Short convenience alias so call sites read ``scopes_for("team")``.
+scopes_for = scopes_for_audience
+
+
+# ---------------------------------------------------------------------------
+# Reusable SQL fragment (the ~50 read-path sites adopt this)
+# ---------------------------------------------------------------------------
+
+
+def visibility_predicate(alias: str, scopes) -> str:
+    """Return an injection-safe SQL boolean gating ``<alias>``'s ``entity_registry`` row.
+
+    This is a DROP-IN replacement for the read path's ``AND NOT node_private`` clause,
+    so it RETAINS the ``node_private`` hard-deny and ADDS the audience-scope gate:
+
+        ``(NOT COALESCE(<alias>.node_private, false) AND <alias>.visibility_scope IN (…))``
+
+    ``NOT COALESCE(node_private, false)`` keeps a NULL ``node_private`` visible (matching
+    the original ``NOT node_private`` when the column is non-NULL, and failing safe by
+    treating NULL as "not private" exactly as the boolean ``NOT`` would after coalesce).
+
+    ``alias`` is the table alias/name the ``entity_registry`` row is exposed under at
+    the call site (e.g. ``"er"`` or ``"entity_registry"``). ``scopes`` is the set of
+    scope values that audience may see — pass ``scopes_for(audience)``.
+
+    Injection-safety comes from strict whitelisting, NOT string escaping:
+
+      * ``alias`` must match ``^[A-Za-z_][A-Za-z0-9_]*$`` or ``ValueError`` is raised;
+      * every value in ``scopes`` must be one of ``VALID_SCOPES`` or ``ValueError``
+        is raised.
+
+    Because both inputs are validated against fixed allowlists, the literal values
+    spliced into the returned fragment can never carry attacker-controlled text —
+    this matches the codebase's existing ``privacy_filter`` idiom (interpolated,
+    not ``$n``-parametrized) while staying safe, and sidesteps the fact that each
+    call site has a different asyncpg positional-parameter offset.
+
+    Empty ``scopes`` (e.g. an unknown audience) → ``"(false)"`` — matches nothing.
+    """
+    if not isinstance(alias, str) or not _ALIAS_RE.match(alias):
+        raise ValueError(f"unsafe SQL alias: {alias!r}")
+    scope_list = list(scopes or [])
+    if not scope_list:
+        return "(false)"
+    bad = [s for s in scope_list if s not in VALID_SCOPES]
+    if bad:
+        raise ValueError(f"unknown visibility scope(s): {bad!r}")
+    # De-dup while preserving a stable, deterministic order.
+    ordered = [s for s in ("public", "team", "confidential", "unclassified") if s in set(scope_list)]
+    quoted = ", ".join(f"'{s}'" for s in ordered)
+    return (
+        f"(NOT COALESCE({alias}.node_private, false) "
+        f"AND {alias}.visibility_scope IN ({quoted}))"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Type normalization + config loading (copied local — no cross-repo import)
+# ---------------------------------------------------------------------------
+
+
+def normalize_type(raw: str | None) -> str:
+    """Normalize a KG ``entity_type`` to a canonical label (``schema:Concept`` → ``Concept``).
+
+    Copied from entity_projector.normalize_type to keep this module free of any
+    cross-repo import.
+    """
+    if not raw:
+        return "Misc"
+    t = str(raw).strip()
+    low = t.lower()
+    for pref in ("schema:", "bkc:"):
+        if low.startswith(pref):
+            t = t.split(":", 1)[1]
+            low = t.lower()
+            break
+    if low == "place":
+        return "Location"
+    return (t[:1].upper() + t[1:]) if t else "Misc"
+
+
+def _load_config(path: str | None = None) -> tuple[set[str], set[str]]:
+    """Load ``(denylist, concept_allowlist)`` — both lowercased. Missing yaml → empty sets.
+
+    LOCAL copy of entity_projector._load_config (do NOT cross-import). Path resolves
+    to ``PROJECTION_CONFIG_PATH`` env, then the arg, then the bundled default.
+    """
+    resolved = os.environ.get("PROJECTION_CONFIG_PATH") or path or _DEFAULT_CONFIG_PATH
+    denylist: set[str] = set()
+    allow: set[str] = set()
+    try:
+        import yaml  # lazy
+
+        with open(resolved, encoding="utf-8") as f:
+            data = yaml.safe_load(f) or {}
+        for x in data.get("entity_denylist") or []:
+            if str(x).strip():
+                denylist.add(str(x).strip().lower())
+        for x in data.get("concept_projection_allowlist") or []:
+            if str(x).strip():
+                allow.add(str(x).strip().lower())
+    except FileNotFoundError:
+        pass
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.warning("could not load visibility config %s: %s", resolved, exc)
+    return denylist, allow
+
+
+# Config is process-static (operator-owned, reviewed like code). Load once; a test
+# can force a reload by clearing the cache via ``_reset_config_cache_for_tests``.
+_CONFIG_CACHE: tuple[set[str], set[str]] | None = None
+
+
+def _get_config() -> tuple[set[str], set[str]]:
+    global _CONFIG_CACHE
+    if _CONFIG_CACHE is None:
+        _CONFIG_CACHE = _load_config()
+    return _CONFIG_CACHE
+
+
+def _reset_config_cache_for_tests() -> None:  # pragma: no cover - test hook
+    global _CONFIG_CACHE
+    _CONFIG_CACHE = None
+
+
+# ---------------------------------------------------------------------------
+# The per-entity gate
+# ---------------------------------------------------------------------------
+
+
+async def visible_at(conn, uri: str, audience: str) -> bool:
+    """Return True iff ``audience`` may see the entity identified by ``uri``.
+
+    ``conn`` is an asyncpg connection (or any object exposing ``await conn.fetchrow``
+    / ``await conn.fetch``). Reads ONLY ``entity_registry.visibility_scope`` for the
+    audience axis; ``node_private`` remains a hard deny; concepts are additionally
+    folder-gated via a JOIN to ``entity_rid_mappings.vault_path`` (``entity_registry``
+    carries only ``vault_rid``).
+
+    Fail-closed and degrade-closed: any ambiguity or exception → False.
+    """
+    try:
+        allowed = scopes_for_audience(audience)
+        if not allowed:  # unknown / 'unclassified' audience → deny everything
+            return False
+        if not uri:
+            return False
+
+        reg = await conn.fetchrow(
+            "SELECT visibility_scope, node_private, entity_text, entity_type, aliases "
+            "FROM entity_registry WHERE fuseki_uri = $1",
+            uri,
+        )
+        if reg is None:  # unknown uri
+            return False
+
+        scope = reg["visibility_scope"]
+        node_private = reg["node_private"]
+        entity_text = reg["entity_text"]
+        entity_type = reg["entity_type"]
+        aliases = reg["aliases"]
+
+        # 1. node_private is an unconditional hard deny (independent of scope).
+        if node_private:
+            return False
+
+        # 2. Confidential denylist on name + aliases (case-insensitive).
+        denylist, concept_allowlist = _get_config()
+        names_lower = {
+            n.strip().lower()
+            for n in ({entity_text} | set(aliases or []))
+            if n and str(n).strip()
+        }
+        if names_lower & denylist:
+            return False
+
+        # 3. Audience-scope gate. 'unclassified' only clears for the confidential
+        #    audience (its allowed set is the only one that contains 'unclassified').
+        if scope not in allowed:
+            return False
+
+        # 4. Concept folder-gate: a Concept must live in the Concepts/ vault folder
+        #    (JOIN to entity_rid_mappings for vault_path) AND be on the allowlist.
+        if normalize_type(entity_type) == "Concept":
+            if not (names_lower & concept_allowlist):
+                return False
+            rows = await conn.fetch(
+                "SELECT vault_path FROM entity_rid_mappings WHERE canonical_uri = $1",
+                uri,
+            )
+            in_concepts_folder = any(
+                (r["vault_path"] or "").startswith("Concepts/") for r in (rows or [])
+            )
+            if not in_concepts_folder:
+                return False
+
+        return True
+    except Exception as exc:  # degrade-closed on ANY error
+        logger.warning("visible_at(%r, %r) failed: %s", uri, audience, exc)
+        return False
diff --git a/docs/phase1-readpath-swap-map.md b/docs/phase1-readpath-swap-map.md
new file mode 100644
index 00000000..9107bfae
--- /dev/null
+++ b/docs/phase1-readpath-swap-map.md
@@ -0,0 +1,80 @@
+# Phase 1 — read-path swap map (applyable)
+
+Audience model (operator-confirmed 2026-07-14): **koi-processor `:8351` is the TEAM backend** (the public Living Library is a separate app). So:
+- **team-facing *surfacing* reads → `team` audience** (hide confidential + unclassified after backfill)
+- **internal machinery (resolvers, matchers, `COUNT(*)` stats) → LEAVE raw `NOT node_private`** (they must see everything; team-filtering a resolver causes duplicate-entity corruption)
+- **stats/counts → true totals** (leave raw) — operator chose true totals over team-masked
+
+Predicate: `visibility_predicate('entity_registry', scopes_for('team'))` from `api/policy/visibility.py`, which emits `(NOT COALESCE(entity_registry.node_private,false) AND entity_registry.visibility_scope IN ('public','team'))` — a true drop-in for `AND NOT node_private` plus the audience gate.
+
+**Import to add** (top of `api/personal_ingest_api.py` and `api/retrieval_executors.py`):
+```python
+from api.policy.visibility import visibility_predicate, scopes_for
+```
+
+## SWAP → team (10 sites)
+
+Each replaces `AND NOT node_private` with `AND {visibility_predicate('entity_registry', scopes_for('team'))}`. Note the string style per site (some are plain strings that must become f-strings).
+
+| # | File:line | Function | String style | Edit |
+|---|---|---|---|---|
+| 1 | personal_ingest_api.py:2204 | `_semantic_entity_search` (embedding) | f-string block | `WHERE embedding_3072 IS NOT NULL AND {visibility_predicate('entity_registry', scopes_for('team'))}` |
+| 2 | personal_ingest_api.py:2707 | `list_entities` | f-string block | `WHERE entity_type = $1 AND {…team…}` |
+| 3 | personal_ingest_api.py:2715 | `list_entities` | f-string block | `WHERE {…team…}` (drop the bare `NOT node_private`) |
+| 4 | personal_ingest_api.py:2760 | `entity_search` | f-string block | `WHERE normalized_text ILIKE $2 AND entity_type = $3 AND {…team…}` |
+| 5 | personal_ingest_api.py:2773 | `entity_search` | f-string block | `WHERE normalized_text ILIKE $2 AND {…team…}` |
+| 6 | personal_ingest_api.py:3296 | `get_entity_evidence` | **plain string → make f-string** | `f"SELECT entity_text FROM entity_registry WHERE fuseki_uri = $1 AND {visibility_predicate('entity_registry', scopes_for('team'))}"` |
+| 7 | personal_ingest_api.py:3402 | `get_entity` | f-string block | `WHERE fuseki_uri = $1 AND {…team…}` |
+| 8 | personal_ingest_api.py:5455 | `_graph_guided_retrieval` (embedding) | f-string block | `WHERE embedding_3072 IS NOT NULL AND {…team…}` |
+| 9 | personal_ingest_api.py:5477 | `_graph_guided_retrieval` (conditions) | f-string block | `WHERE ({conditions}) AND {…team…}` |
+| 10 | retrieval_executors.py:51 | `entity_lookup` (the PolicyScope seam) | **needs param** | see below |
+
+**Site 10 (the seam)** — not a text swap. `entity_lookup(...)` already has `include_node_private: bool = False`. Add an `audience: str = 'team'` param and change:
+```python
+privacy_filter = "" if include_node_private else "AND NOT node_private"
+```
+to:
+```python
+privacy_filter = "" if include_node_private else f"AND {visibility_predicate('entity_registry', scopes_for(audience))}"
+```
+The chat/RAG caller (chat_endpoint) passes `audience='team'` (or threads the request's audience when per-user audiences exist).
+
+## LEAVE RAW `NOT node_private` (~12 sites — do NOT swap)
+
+Internal machinery that must see everything:
+- `_resolve_entity_uri` (personal_ingest_api.py:5732, 5739) — name→uri resolver
+- `_resolve_extra_label_candidates` (web_router.py:449, 473) — label resolver
+- `web_evaluate` entity-matching context (web_router.py:663, 757) — extractor dedup context
+- `get_stats` counts (personal_ingest_api.py:3484, 3489, 3497) — true-total metrics
+- `graph_version` entity_count (personal_ingest_api.py:2355) — true-total metric
+- `COUNT(*)` (personal_ingest_api.py:5850) — internal count
+- `knowledge_health.py:316`, `b8a_enrich_entities.py:116` — batch script metrics/enrichment
+
+## Deploy sequence (ONE coordinated, gated operation — NOT split across time)
+
+1. **`pg_dump` verified** (have: `personal_koi_clean_20260714T180257.dump`).
+2. **Backfill `--apply`** — classifies 17,422 → team, 5 → confidential, 4,144 stay unclassified. *(Gate: the confidential 5 are the obvious denylist; the 4 Eve-candidates stay held; safe over-restrictive default.)* **Must run BEFORE the swap** — otherwise the team predicate returns nothing (all rows currently unclassified → team-audience = empty graph).
+3. **Apply the 10 swap edits + the import** (this branch).
+4. **Test read-only via psql**: each swapped query now returns the 17,422 team rows (not empty); a confidential entity 404s for team; a resolver still finds any entity.
+5. **Restart the koi-processor launchd backend** (`~/.config/personal-koi/restart.sh`) — the risky moment; rollback = `git checkout` the swap + restart, or the DOWN migration + dump.
+6. **Verify live**: MCP/team reads return team entities; `visible_at('hydro-one','team')`→False; backend healthy.
+
+## ⚠ Deploy-time decision: the 4,144 hidden-from-team entities
+
+After the swap, the **4,144 conservatively-`unclassified`** entities (email-sourced third-party correspondence + ambiguous) **disappear from team search/listing** until triaged. For a 5-person trusted team that may be over-restrictive. Options:
+- **(a)** Accept it — they're triage-pending; the conservative default is deliberate (third-party consent). Triage `Meta/Indigenomics-Classify-Triage.md` over time.
+- **(b)** Broaden the backfill Rule 5 to also team-stamp the non-email internal-ambiguous subset (keep only email-sensor/proton-email + explicit third-party as unclassified), shrinking the hidden set.
+Decide before step 2.
+
+---
+
+## ⛔ BACKFILL OBSTACLE (found 2026-07-14 during first --apply attempt)
+
+**Bulk-updating `entity_registry.visibility_scope` is expensive (~1s/row → the 17k-row backfill is multi-hour).** Root cause: `entity_registry` has **two HNSW vector indexes** (`idx_entity_registry_embedding_3072_hnsw`, `idx_entity_vector`) and **very large rows** (two `embedding_3072` halfvec columns, ~6–12 KB/row → no page room for a second tuple version). So *every* `visibility_scope` UPDATE is **non-HOT** and re-indexes both HNSW indexes. Dropping the small btree `idx_entity_registry_visibility_scope` only helped 5s→1s/row; the HNSW cost is unavoidable for in-place updates.
+
+**This makes the backfill (and any future re-classification — Eve narrowing, triage) a maintenance-window operation, not a quick script run.** Two paths for a focused session to decide:
+
+1. **Maintenance-window drop/rebuild:** `DROP INDEX` the 2 HNSW indexes → bulk `UPDATE` (fast, no vector maintenance) → `CREATE INDEX ... USING hnsw` (rebuild). Cost: **live semantic search is degraded while the HNSW indexes rebuild** (potentially 10–30 min on 21k×3072 vectors). Schedule off-hours.
+2. **Off-table visibility (architecture reconsideration — recommended to evaluate):** store scope in a small `entity_visibility(fuseki_uri PK, visibility_scope, source_context, …)` table instead of a column on `entity_registry`. Then classification/re-classification never touches the vector-indexed table (fast, repeatable), and the read-path predicate becomes a JOIN to `entity_visibility` instead of a column filter. Supersedes migration 107's column-on-entity_registry choice. Slightly more complex read SQL; far cheaper writes forever.
+
+**Until decided, the backfill is NOT run.** DB state is clean: migration 107 columns exist, all 21,592 rows `unclassified`, nothing classified, backend healthy. The `--apply` attempt rolled back cleanly (transactional).
diff --git a/scripts/backfill_visibility_107.py b/scripts/backfill_visibility_107.py
new file mode 100644
index 00000000..ea2e7f13
--- /dev/null
+++ b/scripts/backfill_visibility_107.py
@@ -0,0 +1,557 @@
+#!/usr/bin/env python3
+"""Standalone backfill for entity_registry.visibility_scope (migration 107).
+
+Stamps the NEW 4-value audience axis (public|team|confidential|unclassified) on
+``entity_registry.visibility_scope`` per the corrected classifier rules. This is a
+ONE-OFF operator tool — it is NOT wired into the running backend and touches only the
+new registry column (never ``entity_rid_mappings.visibility_scope``, the pre-existing
+2-value projection-privacy axis, which is a different concept and left untouched).
+
+SAFETY MODEL
+------------
+  * ``--dry-run`` is the DEFAULT. With no flags the script only SELECTs, prints the
+    per-scope plan + the exact rows that would change, and writes NOTHING.
+  * ``--apply`` performs the UPDATEs, but ONLY after two hard preconditions pass:
+      1. a fresh ``pg_dump`` of ``entity_registry`` exists (``--dump-path``), and
+      2. Eve's sign-off token is supplied (``--eve-signoff``) — the Eve-candidate
+         hydro utilities are NEVER auto-stamped; her review governs them separately.
+    ``--apply`` is intended for LATER human execution, not for this authoring session.
+  * A pre-state CSV snapshot of every row the script considers is written BEFORE any
+    write, in BOTH modes.
+
+CLASSIFIER RULES (corrected)
+----------------------------
+  Rule 1  CONFIDENTIAL — a fixed denylist of fuseki_uris (Hydro One + the Layer A/B
+          family). Matched by exact fuseki_uri, never by fuzzy name, so an unrelated
+          "two-layer architecture" concept can never be swept in.
+  Rule 5  TEAM — rows whose ``source`` is an internal/team-authored pipeline
+          (knowledge-add, personal-vault, extract-session-entities, obsidian-vault)
+          AND currently ``visibility_scope='unclassified'``, EXCLUDING any row that
+          is on the confidential denylist or the Eve-candidate list.
+  EVE-CANDIDATES — BC Hydro / Bchydro (Org+Person dup) / Hydro-Québec / Manitoba
+          Hydro. NEVER stamped by this script. Emitted to a review list only. They
+          are held out of Rule 5 even though their source would otherwise qualify.
+  CONSERVATIVE-UNCLASSIFIED — email-sensor / proton-email sources and any
+          third-party-attendee transcript are left 'unclassified'. This falls out
+          naturally: those sources are not in the Rule-5 team-source set, so they are
+          never stamped. (Rules 2/3/4 from the spec are un-driveable from ``source``
+          alone and are deliberately NOT wired.)
+
+Precedence: confidential > eve-hold > team > unclassified. A row that appears in
+both a team source and the confidential/eve set resolves to confidential/eve.
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+# --------------------------------------------------------------------------- #
+# DSN resolution — mirrors otter_notion/entity_projector.py exactly.
+# --------------------------------------------------------------------------- #
+DEFAULT_PG_DSN = "postgresql://darrenzal@localhost:5432/personal_koi"
+
+
+def _dotenv_value(key: str) -> str | None:
+    """Best-effort read of KEY from a nearby .env, without importing dotenv."""
+    for candidate in (
+        Path.cwd() / ".env",
+        Path(__file__).resolve().parent.parent / ".env",
+    ):
+        try:
+            with open(candidate, encoding="utf-8") as fh:
+                for line in fh:
+                    line = line.strip()
+                    if not line or line.startswith("#") or "=" not in line:
+                        continue
+                    k, _, v = line.partition("=")
+                    if k.strip() == key:
+                        return v.strip().strip("'").strip('"')
+        except FileNotFoundError:
+            continue
+    return None
+
+
+def resolve_dsn(explicit: str | None) -> str:
+    return (
+        explicit
+        or os.environ.get("PERSONAL_KOI_PG")
+        or _dotenv_value("PERSONAL_KOI_PG")
+        or DEFAULT_PG_DSN
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Classifier sets — matched by EXACT fuseki_uri (never fuzzy name).
+#
+# These uris were resolved read-only from the live registry on 2026-07-14. If the
+# registry changes, re-derive them; a missing uri is simply a no-op (0 rows).
+# --------------------------------------------------------------------------- #
+
+# Rule 1: confidential denylist (Hydro One + Layer A/B family).
+CONFIDENTIAL_URIS: tuple[str, ...] = (
+    "orn:personal-koi.entity:organization-hydro-one-06e352ca9fb3",   # Hydro One
+    "orn:personal-koi.entity:concept-layer-a-ac555b67fb31",          # Layer A
+    "orn:personal-koi.entity:concept-layer-b-1e09d73a7855",          # Layer B
+    "orn:personal-koi.entity:organization-layer-b-ci-d7cbcfc264ec",  # Layer B CI
+    "orn:personal-koi.entity:project-layer-b-pipeline-04ac0a419fee",  # Layer B Pipeline
+    # "Layer B Expansion Opportunity" is on the projection_config denylist but has no
+    # registry row today; kept here as documentation. Add its uri if one appears.
+)
+
+# Eve-review candidates — NEVER auto-stamped. Held out of Rule 5 and emitted to a
+# review list. Carol Anne's consulting dataset touches Indigenous-nation hydro
+# participation; whether these utilities are confidential is Eve's call, not ours.
+EVE_CANDIDATE_URIS: tuple[str, ...] = (
+    "orn:personal-koi.entity:person-bc-hydro-d17114da6cf9",          # BC Hydro (Person dup)
+    "orn:personal-koi.entity:organization-bchydro-917b7deb1451",     # Bchydro (Org dup)
+    "orn:personal-koi.entity:organization-hydro-qu-bec-50d91d3692cf",  # Hydro-Québec
+    "orn:personal-koi.entity:organization-manitoba-hydro-7ea8181f2605",  # Manitoba Hydro
+)
+
+# Rule 5: team-authored source pipelines.
+TEAM_SOURCES: tuple[str, ...] = (
+    "knowledge-add",
+    "personal-vault",
+    "extract-session-entities",
+    "obsidian-vault",
+)
+
+# --------------------------------------------------------------------------- #
+# Static-URI DRIFT GUARD — re-derive the confidential + Eve sets BY NAME.
+#
+# The URI tuples above were frozen on 2026-07-14. An entity ingested AFTER that
+# date (e.g. a fresh "Hydro One" mention that resolves to a NEW fuseki_uri) would
+# be invisible to a URI-only match and could leak. To close that gap, --apply
+# RE-DERIVES the sets by EXACT (case-insensitive) name and UNIONs them with the
+# frozen tuples.
+#
+# Matching is EXACT name/alias EQUALITY — never a substring/fuzzy — so an unrelated
+# "two-layer architecture" concept is NEVER swept in. This preserves the original
+# safety property (the reason the sets were URI-pinned) while catching drift.
+# --------------------------------------------------------------------------- #
+CONFIDENTIAL_NAMES: tuple[str, ...] = (
+    "Hydro One",
+    "Layer A",
+    "Layer B",
+    "Layer B CI",
+    "Layer B Pipeline",
+    "Layer B Expansion Opportunity",
+)
+
+EVE_CANDIDATE_NAMES: tuple[str, ...] = (
+    "BC Hydro",
+    "Bchydro",
+    "Hydro-Québec",
+    "Hydro-Quebec",
+    "Manitoba Hydro",
+)
+
+
+def _derive_uris_by_name(cur, names: tuple[str, ...]) -> set[str]:
+    """Return fuseki_uris whose entity_text OR an alias EXACTLY (case-insensitively)
+    equals one of ``names``. Exact equality only — no substring/fuzzy match."""
+    if not names:
+        return set()
+    lowered = [n.strip().lower() for n in names if n and n.strip()]
+    if not lowered:
+        return set()
+    cur.execute(
+        """
+        SELECT DISTINCT fuseki_uri
+          FROM entity_registry
+         WHERE lower(entity_text) = ANY(%s)
+            OR EXISTS (
+                 SELECT 1
+                   FROM unnest(COALESCE(aliases, ARRAY[]::text[])) AS a
+                  WHERE lower(a) = ANY(%s)
+               )
+        """,
+        (lowered, lowered),
+    )
+    return {r[0] for r in cur.fetchall()}
+
+
+def derive_effective_sets(cur) -> tuple[set[str], set[str], dict]:
+    """RE-DERIVE (effective_confidential, effective_eve, drift_report) at run time.
+
+    Effective sets = frozen tuples ∪ name-derived URIs. Precedence confidential >
+    eve-hold, so any URI that is confidential-by-name is removed from the eve set.
+    ``drift_report`` lists URIs newly caught by name that are NOT in the frozen
+    tuples — i.e. entities ingested after the 2026-07-14 freeze.
+    """
+    conf_by_name = _derive_uris_by_name(cur, CONFIDENTIAL_NAMES)
+    eve_by_name = _derive_uris_by_name(cur, EVE_CANDIDATE_NAMES)
+
+    conf_uris = set(CONFIDENTIAL_URIS) | conf_by_name
+    eve_uris = (set(EVE_CANDIDATE_URIS) | eve_by_name) - conf_uris
+
+    drift = {
+        "confidential_new": sorted(conf_by_name - set(CONFIDENTIAL_URIS)),
+        "eve_new": sorted(eve_by_name - set(EVE_CANDIDATE_URIS) - conf_uris),
+    }
+    return conf_uris, eve_uris, drift
+
+
+def _existing_uris(cur, uris: tuple[str, ...]) -> set[str]:
+    """Subset of ``uris`` that still exist in entity_registry today."""
+    if not uris:
+        return set()
+    cur.execute(
+        "SELECT fuseki_uri FROM entity_registry WHERE fuseki_uri = ANY(%s)",
+        (list(uris),),
+    )
+    return {r[0] for r in cur.fetchall()}
+
+
+# --------------------------------------------------------------------------- #
+# DB helpers (read-only unless --apply, and even then guarded).
+# --------------------------------------------------------------------------- #
+def _connect(dsn: str):
+    import psycopg2  # lazy import
+
+    conn = psycopg2.connect(dsn)
+    return conn
+
+
+def _fetch_rows(cur, uris: tuple[str, ...]) -> list[tuple]:
+    """Return (fuseki_uri, entity_text, entity_type, source, visibility_scope) for the
+    given uris that are CURRENTLY 'unclassified' (the only rows we would ever change)."""
+    if not uris:
+        return []
+    cur.execute(
+        """
+        SELECT fuseki_uri, entity_text, entity_type, source, visibility_scope
+          FROM entity_registry
+         WHERE fuseki_uri = ANY(%s)
+           AND visibility_scope = 'unclassified'
+         ORDER BY entity_text
+        """,
+        (list(uris),),
+    )
+    return cur.fetchall()
+
+
+def _fetch_eve(cur, uris: tuple[str, ...]) -> list[tuple]:
+    """Eve candidates — report their CURRENT scope regardless (should be unclassified)."""
+    if not uris:
+        return []
+    cur.execute(
+        """
+        SELECT fuseki_uri, entity_text, entity_type, source, visibility_scope
+          FROM entity_registry
+         WHERE fuseki_uri = ANY(%s)
+         ORDER BY entity_text
+        """,
+        (list(uris),),
+    )
+    return cur.fetchall()
+
+
+def _fetch_team_rows(cur, conf_uris, eve_uris, limit: int | None) -> list[tuple]:
+    """Rule-5 team rows: unclassified + team source, minus confidential + eve holds."""
+    sql = """
+        SELECT fuseki_uri, entity_text, entity_type, source, visibility_scope
+          FROM entity_registry
+         WHERE visibility_scope = 'unclassified'
+           AND source = ANY(%s)
+           AND fuseki_uri <> ALL(%s)
+           AND fuseki_uri <> ALL(%s)
+         ORDER BY source, entity_text
+    """
+    params = [
+        list(TEAM_SOURCES),
+        list(conf_uris),
+        list(eve_uris),
+    ]
+    if limit is not None:
+        sql += " LIMIT %s"
+        params.append(limit)
+    cur.execute(sql, params)
+    return cur.fetchall()
+
+
+def _count_team(cur, conf_uris, eve_uris) -> int:
+    cur.execute(
+        """
+        SELECT count(*)
+          FROM entity_registry
+         WHERE visibility_scope = 'unclassified'
+           AND source = ANY(%s)
+           AND fuseki_uri <> ALL(%s)
+           AND fuseki_uri <> ALL(%s)
+        """,
+        (list(TEAM_SOURCES), list(conf_uris), list(eve_uris)),
+    )
+    return cur.fetchone()[0]
+
+
+def _count_scope(cur, scope: str) -> int:
+    cur.execute(
+        "SELECT count(*) FROM entity_registry WHERE visibility_scope = %s", (scope,)
+    )
+    return cur.fetchone()[0]
+
+
+def _total(cur) -> int:
+    cur.execute("SELECT count(*) FROM entity_registry")
+    return cur.fetchone()[0]
+
+
+# --------------------------------------------------------------------------- #
+# Snapshot
+# --------------------------------------------------------------------------- #
+def write_snapshot(cur, path: Path, conf_uris, eve_uris) -> int:
+    """Write a pre-state CSV of every row the script considers (confidential + eve +
+    all team-source-unclassified rows). Returns the row count written."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    considered: dict[str, tuple] = {}
+    for row in _fetch_rows(cur, tuple(conf_uris)):
+        considered[row[0]] = row + ("rule1_confidential",)
+    for row in _fetch_eve(cur, tuple(eve_uris)):
+        considered.setdefault(row[0], row + ("eve_hold",))
+    for row in _fetch_team_rows(cur, conf_uris, eve_uris, limit=None):
+        considered.setdefault(row[0], row + ("rule5_team",))
+    with open(path, "w", newline="", encoding="utf-8") as fh:
+        w = csv.writer(fh)
+        w.writerow(
+            [
+                "fuseki_uri",
+                "entity_text",
+                "entity_type",
+                "source",
+                "visibility_scope_before",
+                "planned_disposition",
+            ]
+        )
+        for uri in sorted(considered):
+            w.writerow(considered[uri])
+    return len(considered)
+
+
+# --------------------------------------------------------------------------- #
+# Reporting
+# --------------------------------------------------------------------------- #
+def print_plan(cur, conf_uris, eve_uris, drift=None) -> None:
+    conf = _fetch_rows(cur, tuple(conf_uris))
+    eve = _fetch_eve(cur, tuple(eve_uris))
+    team_n = _count_team(cur, conf_uris, eve_uris)
+    total = _total(cur)
+    unclassified_now = _count_scope(cur, "unclassified")
+    # Post-plan unclassified = current unclassified - team - confidential-being-changed.
+    stays_unclassified = unclassified_now - team_n - len(conf)
+
+    print("=" * 72)
+    print("VISIBILITY BACKFILL PLAN (migration 107)  — resulting scope counts")
+    print("=" * 72)
+    print(f"  total entity_registry rows : {total}")
+    print(f"  currently 'unclassified'   : {unclassified_now}")
+    print("  ---")
+    print(f"  → confidential (Rule 1)    : {len(conf)}")
+    print(f"  → team         (Rule 5)    : {team_n}")
+    print(f"  → stays unclassified       : {stays_unclassified}")
+    print(f"     (incl. {len(eve)} Eve-candidates held out of team)")
+    print()
+
+    print("CONFIDENTIAL (Rule 1) — exact rows to be stamped:")
+    if not conf:
+        print("    (none currently unclassified)")
+    for uri, text, etype, source, scope in conf:
+        print(f"    [{scope}→confidential] {text}  <{etype}>  src={source}")
+        print(f"        {uri}")
+    print()
+
+    print("EVE-REVIEW CANDIDATES — NOT stamped, routed to Eve:")
+    if not eve:
+        print("    (none found)")
+    for uri, text, etype, source, scope in eve:
+        print(f"    [held @ {scope}] {text}  <{etype}>  src={source}")
+        print(f"        {uri}")
+    print()
+
+    print(f"TEAM (Rule 5) — {team_n} rows; first 25 shown:")
+    for uri, text, etype, source, scope in _fetch_team_rows(cur, conf_uris, eve_uris, limit=25):
+        print(f"    [{scope}→team] {text}  <{etype}>  src={source}")
+    if team_n > 25:
+        print(f"    … and {team_n - 25} more (see snapshot CSV for the full list)")
+
+    if drift:
+        new_conf = drift.get("confidential_new") or []
+        new_eve = drift.get("eve_new") or []
+        print()
+        print("NAME-DERIVED DRIFT (caught by name, NOT in the 2026-07-14 frozen URIs):")
+        if not new_conf and not new_eve:
+            print("    (none — frozen URI lists still cover every name-matched row)")
+        for uri in new_conf:
+            print(f"    [+confidential] {uri}")
+        for uri in new_eve:
+            print(f"    [+eve-hold]     {uri}")
+    print("=" * 72)
+
+
+# --------------------------------------------------------------------------- #
+# Apply (guarded; for LATER human execution)
+# --------------------------------------------------------------------------- #
+def apply_backfill(conn, cur, args, conf_uris, eve_uris, drift) -> None:
+    # --- Precondition 1: fresh pg_dump must exist. -------------------------- #
+    if not args.dump_path:
+        sys.exit("REFUSED: --apply requires --dump-path pointing at a fresh "
+                 "pg_dump of entity_registry (rollback safety).")
+    dump = Path(args.dump_path)
+    if not dump.is_file() or dump.stat().st_size == 0:
+        sys.exit(f"REFUSED: --dump-path {dump} is missing or empty. Run e.g.\n"
+                 f"    pg_dump -t entity_registry <db> > {dump}")
+
+    # --- Precondition 2: Eve governance on the hydro-utility candidates. ----- #
+    # Two HONEST ways past this gate — never fake a sign-off:
+    #   (a) --eve-signoff EVE-SIGNED-OFF : Eve reviewed the candidate list (recorded).
+    #   (b) --defer-eve-candidates       : HOLD mode — every candidate stays
+    #       'unclassified' pending Eve; no candidate is stamped, so no decision that
+    #       is Eve's to make is being made. The over-restrictive-safe path when Eve
+    #       is unavailable. (Precondition 3 below still refuses if a candidate is
+    #       confidential-by-name — that genuinely needs Eve, defer or not.)
+    if args.eve_signoff != "EVE-SIGNED-OFF" and not args.defer_eve_candidates:
+        sys.exit("REFUSED: --apply requires EITHER --eve-signoff EVE-SIGNED-OFF (Eve "
+                 "reviewed the hydro-utility candidates) OR --defer-eve-candidates (hold "
+                 "all candidates as unclassified, pending Eve). Do NOT fabricate a sign-off.")
+
+    # --- Precondition 3: name-based drift verification. --------------------- #
+    # The confidential/eve sets used below were RE-DERIVED by name at run time (see
+    # derive_effective_sets). This guard catches the two ways name-derivation can be
+    # unsafe at apply time, and forces a human to reconcile rather than silently
+    # applying a stale plan:
+    #   (a) any frozen confidential URI that STILL exists in the registry but is NOT
+    #       re-discovered by name → the CONFIDENTIAL_NAMES list has drifted out of
+    #       sync with the data and can no longer be trusted to catch new rows;
+    #   (b) any Eve-hold candidate the name pass now classifies as confidential →
+    #       precedence collision that Eve must adjudicate before we stamp.
+    conf_by_name = _derive_uris_by_name(cur, CONFIDENTIAL_NAMES)
+    still_present_frozen = _existing_uris(cur, CONFIDENTIAL_URIS)
+    unmatched_frozen = still_present_frozen - conf_by_name
+    if unmatched_frozen:
+        sys.exit(
+            "REFUSED: name-based verification failed. These frozen confidential URIs "
+            "still exist but were NOT re-discovered by CONFIDENTIAL_NAMES — the name "
+            "list has drifted and can no longer be trusted to catch newly-ingested "
+            "rows. Reconcile CONFIDENTIAL_NAMES before applying:\n    "
+            + "\n    ".join(sorted(unmatched_frozen))
+        )
+    eve_now_confidential = (set(EVE_CANDIDATE_URIS) | _derive_uris_by_name(cur, EVE_CANDIDATE_NAMES)) & conf_by_name
+    if eve_now_confidential:
+        sys.exit(
+            "REFUSED: an Eve-hold candidate is now classified confidential by name "
+            "(precedence collision). Eve must adjudicate before applying:\n    "
+            + "\n    ".join(sorted(eve_now_confidential))
+        )
+
+    new_conf = (drift or {}).get("confidential_new") or []
+    new_eve = (drift or {}).get("eve_new") or []
+    print(f"[apply] preconditions OK (dump={dump}, eve sign-off present, "
+          f"name-verification passed).")
+    print(f"[apply] effective sets (frozen ∪ name-derived): "
+          f"{len(conf_uris)} confidential, {len(eve_uris)} eve-hold.")
+    if new_conf or new_eve:
+        print(f"[apply] DRIFT caught since 2026-07-14 freeze — will be stamped: "
+              f"{len(new_conf)} new confidential, {len(new_eve)} new eve-hold (held).")
+    print("[apply] running UPDATEs inside a single transaction …")
+
+    # Confidential first (precedence), then team. Both scoped to 'unclassified' so a
+    # re-run is idempotent and can never downgrade an already-classified row.
+    # Uses the RE-DERIVED effective sets, so rows ingested after the freeze are caught.
+    cur.execute(
+        """
+        UPDATE entity_registry
+           SET visibility_scope = 'confidential', updated_at = now()
+         WHERE fuseki_uri = ANY(%s)
+           AND visibility_scope = 'unclassified'
+        """,
+        (list(conf_uris),),
+    )
+    conf_n = cur.rowcount
+    cur.execute(
+        """
+        UPDATE entity_registry
+           SET visibility_scope = 'team', updated_at = now()
+         WHERE visibility_scope = 'unclassified'
+           AND source = ANY(%s)
+           AND fuseki_uri <> ALL(%s)
+           AND fuseki_uri <> ALL(%s)
+        """,
+        (list(TEAM_SOURCES), list(conf_uris), list(eve_uris)),
+    )
+    team_n = cur.rowcount
+    conn.commit()
+    print(f"[apply] committed: {conf_n} → confidential, {team_n} → team. "
+          f"Eve candidates untouched.")
+
+
+# --------------------------------------------------------------------------- #
+# main
+# --------------------------------------------------------------------------- #
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(
+        description="Backfill entity_registry.visibility_scope (migration 107). "
+        "Dry-run by default; --apply is guarded for later human execution.",
+    )
+    mode = ap.add_mutually_exclusive_group()
+    mode.add_argument("--dry-run", action="store_true", default=True,
+                      help="(default) SELECT-only; print the plan, write nothing.")
+    mode.add_argument("--apply", action="store_true",
+                      help="Perform the UPDATEs. Requires --dump-path + --eve-signoff.")
+    ap.add_argument("--dsn", default=None,
+                    help="Postgres DSN (default: PERSONAL_KOI_PG env / .env / localhost).")
+    ap.add_argument("--snapshot",
+                    default=f"visibility_107_prestate_"
+                            f"{datetime.now(timezone.utc):%Y%m%dT%H%M%SZ}.csv",
+                    help="Path for the pre-state CSV snapshot (written in both modes).")
+    ap.add_argument("--dump-path", default=None,
+                    help="[--apply only] path to a fresh pg_dump of entity_registry.")
+    ap.add_argument("--eve-signoff", default=None,
+                    help="[--apply only] pass EVE-SIGNED-OFF to confirm Eve reviewed "
+                         "the hydro-utility candidate list.")
+    ap.add_argument("--defer-eve-candidates", action="store_true",
+                    help="[--apply only] HONEST alternative to --eve-signoff when Eve is "
+                         "unavailable: run in HOLD mode — every hydro-utility candidate "
+                         "stays 'unclassified' (invisible to team), no candidate stamped, "
+                         "no decision made that is Eve's to make. Never fakes her review.")
+    args = ap.parse_args(argv)
+
+    applying = bool(args.apply)
+    dsn = resolve_dsn(args.dsn)
+    print(f"[backfill_visibility_107] DSN={dsn}  mode={'APPLY' if applying else 'DRY-RUN'}")
+
+    conn = _connect(dsn)
+    try:
+        conn.autocommit = False
+        with conn.cursor() as cur:
+            # RE-DERIVE the confidential + eve sets by name (read-only SELECTs), so
+            # both the plan/snapshot AND the apply reflect rows ingested after the
+            # 2026-07-14 freeze — not just the frozen URI tuples.
+            conf_uris, eve_uris, drift = derive_effective_sets(cur)
+            if drift.get("confidential_new") or drift.get("eve_new"):
+                print(f"[derive] name-drift caught since freeze: "
+                      f"{len(drift['confidential_new'])} confidential, "
+                      f"{len(drift['eve_new'])} eve-hold (not in frozen URIs).")
+
+            # Snapshot pre-state BEFORE any possible write.
+            n = write_snapshot(cur, Path(args.snapshot), conf_uris, eve_uris)
+            print(f"[snapshot] wrote {n} considered rows → {args.snapshot}")
+
+            print_plan(cur, conf_uris, eve_uris, drift)
+
+            if applying:
+                apply_backfill(conn, cur, args, conf_uris, eve_uris, drift)
+            else:
+                print("\nDRY-RUN: no rows changed. Re-run with --apply "
+                      "(plus --dump-path and --eve-signoff) to execute.")
+    finally:
+        conn.rollback()  # discard anything uncommitted in dry-run
+        conn.close()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_visibility.py b/tests/test_visibility.py
new file mode 100644
index 00000000..e39b0136
--- /dev/null
+++ b/tests/test_visibility.py
@@ -0,0 +1,277 @@
+"""Fail-closed semantics for the audience-scoped visibility kernel.
+
+READ-ONLY: these never touch Postgres. ``visible_at`` is driven with a fake async
+connection so the tests assert the POLICY (fail-closed / degrade-closed), not any
+data-specific outcome — correct, because every live row is currently
+``'unclassified'`` and stamping real scopes is a separate, un-run step.
+
+Tests are plain (non-async) functions that drive the coroutine via ``asyncio.run``,
+so they need no pytest-asyncio marker under the repo's ``asyncio_mode = strict``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from api.policy.visibility import (
+    scopes_for_audience,
+    visibility_predicate,
+    visible_at,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fake async connection (duck-typed asyncpg: fetchrow / fetch)
+# ---------------------------------------------------------------------------
+
+
+class FakeConn:
+    """Minimal async stand-in for an asyncpg connection.
+
+    ``reg`` is the single ``entity_registry`` row returned by ``fetchrow`` (or None
+    to simulate an unknown uri). ``rid_rows`` is the ``entity_rid_mappings`` result
+    for the concept folder-gate. If ``raises`` is set, every query raises it
+    (drives the degrade-closed path).
+    """
+
+    def __init__(self, reg=None, rid_rows=None, raises: Exception | None = None):
+        self._reg = reg
+        self._rid_rows = rid_rows or []
+        self._raises = raises
+
+    async def fetchrow(self, sql, *args):
+        if self._raises is not None:
+            raise self._raises
+        return self._reg
+
+    async def fetch(self, sql, *args):
+        if self._raises is not None:
+            raise self._raises
+        return self._rid_rows
+
+
+def _reg(scope="unclassified", node_private=False, name="Acme Corp",
+         etype="Organization", aliases=None):
+    return {
+        "visibility_scope": scope,
+        "node_private": node_private,
+        "entity_text": name,
+        "entity_type": etype,
+        "aliases": aliases or [],
+    }
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+URI = "https://example.org/entity/acme"
+
+
+# ---------------------------------------------------------------------------
+# visible_at — fail-closed core
+# ---------------------------------------------------------------------------
+
+
+def test_unknown_uri_denied_for_all_audiences():
+    conn = FakeConn(reg=None)  # no registry row
+    for audience in ("public", "team", "confidential"):
+        assert _run(visible_at(conn, URI, audience)) is False
+
+
+def test_empty_uri_denied():
+    conn = FakeConn(reg=_reg(scope="public"))
+    assert _run(visible_at(conn, "", "confidential")) is False
+
+
+def test_unclassified_denied_for_team_and_public():
+    # Every live row is currently 'unclassified' — it must be invisible to
+    # team + public (fail-closed for the not-yet-classified bucket).
+    conn = FakeConn(reg=_reg(scope="unclassified"))
+    assert _run(visible_at(conn, URI, "team")) is False
+    assert _run(visible_at(conn, URI, "public")) is False
+
+
+def test_exception_path_degrades_closed():
+    conn = FakeConn(raises=RuntimeError("db exploded"))
+    assert _run(visible_at(conn, URI, "confidential")) is False
+
+
+def test_unknown_audience_denied():
+    # 'unclassified' is a scope, never a valid audience; anything unrecognized denies.
+    conn = FakeConn(reg=_reg(scope="public"))
+    assert _run(visible_at(conn, URI, "unclassified")) is False
+    assert _run(visible_at(conn, URI, "admin")) is False
+    assert _run(visible_at(conn, URI, "")) is False
+
+
+def test_node_private_hard_deny_even_when_scope_public():
+    conn = FakeConn(reg=_reg(scope="public", node_private=True))
+    for audience in ("public", "team", "confidential"):
+        assert _run(visible_at(conn, URI, audience)) is False
+
+
+# ---------------------------------------------------------------------------
+# visible_at — the affirmative paths that MUST hold (policy, not live data)
+# ---------------------------------------------------------------------------
+
+
+def test_confidential_audience_sees_confidential_scope():
+    conn = FakeConn(reg=_reg(scope="confidential"))
+    assert _run(visible_at(conn, URI, "confidential")) is True
+
+
+def test_confidential_audience_sees_unclassified_rollout_path():
+    # The full-trust internal viewer must still see rows during the rollout, when
+    # everything is unclassified — otherwise the internal app goes dark.
+    conn = FakeConn(reg=_reg(scope="unclassified"))
+    assert _run(visible_at(conn, URI, "confidential")) is True
+
+
+def test_team_audience_sees_public_and_team_not_confidential():
+    assert _run(visible_at(FakeConn(reg=_reg(scope="public")), URI, "team")) is True
+    assert _run(visible_at(FakeConn(reg=_reg(scope="team")), URI, "team")) is True
+    assert _run(visible_at(FakeConn(reg=_reg(scope="confidential")), URI, "team")) is False
+
+
+def test_public_audience_sees_only_public():
+    assert _run(visible_at(FakeConn(reg=_reg(scope="public")), URI, "public")) is True
+    assert _run(visible_at(FakeConn(reg=_reg(scope="team")), URI, "public")) is False
+
+
+def test_denylisted_name_denied_even_when_scope_public():
+    conn = FakeConn(reg=_reg(scope="public", name="Hydro One"))
+    assert _run(visible_at(conn, URI, "confidential")) is False
+
+
+def test_denylist_matches_alias_case_insensitively():
+    conn = FakeConn(reg=_reg(scope="public", name="A Utility", aliases=["layer b"]))
+    assert _run(visible_at(conn, URI, "confidential")) is False
+
+
+# ---------------------------------------------------------------------------
+# visible_at — Concept folder gate
+# ---------------------------------------------------------------------------
+
+
+def test_concept_in_folder_and_allowlist_visible():
+    conn = FakeConn(
+        reg=_reg(scope="public", name="data sovereignty", etype="schema:Concept"),
+        rid_rows=[{"vault_path": "Concepts/data sovereignty.md"}],
+    )
+    assert _run(visible_at(conn, URI, "team")) is True
+
+
+def test_concept_not_in_concepts_folder_denied():
+    conn = FakeConn(
+        reg=_reg(scope="public", name="data sovereignty", etype="schema:Concept"),
+        rid_rows=[{"vault_path": "Notes/data sovereignty.md"}],
+    )
+    assert _run(visible_at(conn, URI, "team")) is False
+
+
+def test_concept_not_on_allowlist_denied():
+    conn = FakeConn(
+        reg=_reg(scope="public", name="random idea", etype="Concept"),
+        rid_rows=[{"vault_path": "Concepts/random idea.md"}],
+    )
+    assert _run(visible_at(conn, URI, "team")) is False
+
+
+def test_concept_with_no_vault_mapping_denied():
+    conn = FakeConn(
+        reg=_reg(scope="public", name="data sovereignty", etype="Concept"),
+        rid_rows=[],  # no vault_path row → cannot confirm folder → deny
+    )
+    assert _run(visible_at(conn, URI, "team")) is False
+
+
+# ---------------------------------------------------------------------------
+# scopes_for_audience
+# ---------------------------------------------------------------------------
+
+
+def test_scopes_for_audience_mapping():
+    assert scopes_for_audience("public") == ("public",)
+    assert scopes_for_audience("team") == ("public", "team")
+    assert scopes_for_audience("confidential") == (
+        "public", "team", "confidential", "unclassified",
+    )
+
+
+def test_scopes_for_audience_unknown_is_empty():
+    assert scopes_for_audience("unclassified") == ()
+    assert scopes_for_audience("nope") == ()
+    assert scopes_for_audience(None) == ()  # type: ignore[arg-type]
+
+
+# ---------------------------------------------------------------------------
+# visibility_predicate — reusable, injection-safe SQL fragment
+# ---------------------------------------------------------------------------
+
+
+def test_visibility_predicate_team():
+    frag = visibility_predicate("er", scopes_for_audience("team"))
+    assert frag == (
+        "(NOT COALESCE(er.node_private, false) "
+        "AND er.visibility_scope IN ('public', 'team'))"
+    )
+
+
+def test_visibility_predicate_public():
+    assert visibility_predicate("er", scopes_for_audience("public")) == (
+        "(NOT COALESCE(er.node_private, false) "
+        "AND er.visibility_scope IN ('public'))"
+    )
+
+
+def test_visibility_predicate_confidential_all_four():
+    frag = visibility_predicate("entity_registry", scopes_for_audience("confidential"))
+    assert frag == (
+        "(NOT COALESCE(entity_registry.node_private, false) "
+        "AND entity_registry.visibility_scope IN "
+        "('public', 'team', 'confidential', 'unclassified'))"
+    )
+
+
+def test_visibility_predicate_empty_scopes_matches_nothing():
+    assert visibility_predicate("er", ()) == "(false)"
+    assert visibility_predicate("er", scopes_for_audience("bogus")) == "(false)"
+
+
+def test_visibility_predicate_rejects_unsafe_alias():
+    for bad in ("er; DROP TABLE entity_registry", "1abc", "er er", "", "e-r"):
+        with pytest.raises(ValueError):
+            visibility_predicate(bad, ("public",))
+
+
+def test_visibility_predicate_rejects_unknown_scope():
+    with pytest.raises(ValueError):
+        visibility_predicate("er", ("public", "secret'; DROP TABLE x;--"))
+    with pytest.raises(ValueError):
+        visibility_predicate("er", ("public", "node_private"))
+
+
+def test_visibility_predicate_dedups_and_orders():
+    # Duplicate + out-of-order input yields the canonical ordered, de-duped form.
+    frag = visibility_predicate("er", ("team", "public", "team"))
+    assert frag == (
+        "(NOT COALESCE(er.node_private, false) "
+        "AND er.visibility_scope IN ('public', 'team'))"
+    )
+
+
+def test_visibility_predicate_retains_node_private_hard_deny():
+    # Drop-in replacement for `AND NOT node_private`: the node_private hard-deny
+    # MUST survive alongside the audience-scope gate.
+    frag = visibility_predicate("er", scopes_for_audience("confidential"))
+    assert "NOT COALESCE(er.node_private, false)" in frag
+    assert "er.visibility_scope IN" in frag
+
+
+def test_scopes_for_alias_matches_scopes_for_audience():
+    from api.policy.visibility import scopes_for
+    for aud in ("public", "team", "confidential", "nope", ""):
+        assert scopes_for(aud) == scopes_for_audience(aud)