From 2a81cd9d45142bac8e42406564da0a1e106e9772 Mon Sep 17 00:00:00 2001
From: Sinity <ezo.dev@gmail.com>
Date: Mon, 3 Aug 2026 22:57:42 +0200
Subject: [PATCH 1/4] feat(storage): rename table_existence to introspection,
 add column/index checks

Ref polylogue-48h, polylogue-a7xr.9

Problem: polylogue-48h found ~25 independently maintained
_table_exists/_column_exists/_index_exists copies (plus sync/async and
schema-qualified variants) scattered across cli/, daemon/, storage/,
sources/, insights/, and operations/ -- each trivially small and subtly
different. A prior PR (#2912) had already consolidated table_exists/
table_exists_async into polylogue/storage/table_existence.py and redirected
most call sites, but column_exists/index_exists were never added and ~25
call sites still carried their own local definition.

Solution: rename polylogue/storage/table_existence.py to
polylogue/storage/introspection.py (kept at storage/ root, not
storage/sqlite/, for the same documented circular-import reason as before:
storage/sqlite/__init__.py eagerly imports the heavy async backend stack),
and add column_exists/column_exists_async/index_exists/index_exists_async
alongside the existing table_exists/table_exists_async. Schema-qualification
(the schema= kwarg) is omitted for the default "main" schema so every
pre-consolidation bare `PRAGMA table_info(...)`/`sqlite_master` query stays
byte-identical -- several call sites redirected in the next commits are
covered by mock-based tests that pattern-match exact query text.
---
 polylogue/storage/introspection.py            | 154 ++++++++++++++++++
 polylogue/storage/table_existence.py          |  69 --------
 ...ble_existence.py => test_introspection.py} |  84 +++++++++-
 3 files changed, 236 insertions(+), 71 deletions(-)
 create mode 100644 polylogue/storage/introspection.py
 delete mode 100644 polylogue/storage/table_existence.py
 rename tests/unit/storage/{test_table_existence.py => test_introspection.py} (51%)

diff --git a/polylogue/storage/introspection.py b/polylogue/storage/introspection.py
new file mode 100644
index 0000000000..45a5af7c48
--- /dev/null
+++ b/polylogue/storage/introspection.py
@@ -0,0 +1,154 @@
+"""Centralized SQLite schema-introspection primitives (table/column/index existence).
+
+Lives directly under `storage/` (not `storage/sqlite/`) deliberately:
+`polylogue/storage/sqlite/__init__.py` eagerly imports the heavy async
+backend stack, so a low-level module like `storage/blob_publication.py`
+importing a sibling under `storage.sqlite` triggers that whole chain and
+circles back through `insights/` -- a real circular import, not a
+hypothetical one. `storage/__init__.py` itself only exposes lazy
+`__getattr__`-based exports, so importing this module doesn't pull in that
+chain. Raw SQL belongs in `storage/` per the package-placement rules in
+`docs/architecture.md`; `core/` is reserved for no-I/O shared primitives.
+
+This module consolidates what were ~25 near-identical, independently
+maintained `_table_exists`/`_column_exists`/`_index_exists` helpers scattered
+across `cli/`, `daemon/`, `storage/`, `sources/`, `insights/`, and
+`operations/` (polylogue-48h). Several of those copies checked
+`type IN ('table', 'virtual table')` or `type IN ('table', 'shadow')` --
+neither `'virtual table'` nor `'shadow'` is ever an actual `sqlite_master`
+`type` value (virtual tables, including FTS5 shadow tables, register with
+`type='table'`), so those extra alternatives were dead and the plain
+`type='table'` check here is behavior-preserving.
+
+@owner storage-root
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import aiosqlite
+
+
+def _schema_prefix(*, schema: str) -> str:
+    # Deliberately empty for the default "main" schema: an unqualified
+    # `sqlite_master`/`PRAGMA table_info(...)` already resolves to `main`
+    # only (SQLite never merges attached-database schemas into a bare
+    # `sqlite_master` read), and every pre-consolidation call site issued
+    # exactly that bare form -- some with test doubles that pattern-match
+    # the query text. Qualifying only for a genuinely non-default schema
+    # keeps 100% of existing call sites byte-identical while still letting
+    # `schema=` target an attached database explicitly.
+    return "" if schema == "main" else f"{schema}."
+
+
+def table_exists(conn: sqlite3.Connection, name: str, *, schema: str = "main") -> bool:
+    """Check if a table (or virtual table) exists in the given schema (sync SQLite).
+
+    `schema` selects an attached database by name (e.g. a source.db attached
+    to an index.db connection) and must be a trusted internal literal, never
+    user input -- SQLite has no way to bind a schema/database name as a query
+    parameter, so it is interpolated into the query text.
+
+    Args:
+        conn: SQLite connection
+        name: Table name to check
+        schema: Schema name (default: "main")
+
+    Returns:
+        True if the table exists, False otherwise
+    """
+    cursor = conn.execute(
+        f"SELECT 1 FROM {_schema_prefix(schema=schema)}sqlite_master WHERE type='table' AND name=? LIMIT 1",
+        (name,),
+    )
+    return cursor.fetchone() is not None
+
+
+async def table_exists_async(conn: aiosqlite.Connection, name: str, *, schema: str = "main") -> bool:
+    """Check if a table exists in the given schema (async SQLite).
+
+    See `table_exists` for the `schema` trust requirement.
+
+    Args:
+        conn: aiosqlite connection
+        name: Table name to check
+        schema: Schema name (default: "main")
+
+    Returns:
+        True if the table exists, False otherwise
+    """
+    cursor = await conn.execute(
+        f"SELECT 1 FROM {_schema_prefix(schema=schema)}sqlite_master WHERE type='table' AND name=? LIMIT 1",
+        (name,),
+    )
+    row = await cursor.fetchone()
+    return row is not None
+
+
+def index_exists(conn: sqlite3.Connection, name: str, *, schema: str = "main") -> bool:
+    """Check if an index exists in the given schema (sync SQLite).
+
+    See `table_exists` for the `schema` trust requirement.
+    """
+    cursor = conn.execute(
+        f"SELECT 1 FROM {_schema_prefix(schema=schema)}sqlite_master WHERE type='index' AND name=? LIMIT 1",
+        (name,),
+    )
+    return cursor.fetchone() is not None
+
+
+async def index_exists_async(conn: aiosqlite.Connection, name: str, *, schema: str = "main") -> bool:
+    """Check if an index exists in the given schema (async SQLite).
+
+    See `table_exists` for the `schema` trust requirement.
+    """
+    cursor = await conn.execute(
+        f"SELECT 1 FROM {_schema_prefix(schema=schema)}sqlite_master WHERE type='index' AND name=? LIMIT 1",
+        (name,),
+    )
+    row = await cursor.fetchone()
+    return row is not None
+
+
+def _table_info_pragma(table: str, *, schema: str) -> str:
+    return f"PRAGMA {_schema_prefix(schema=schema)}table_info({table})"
+
+
+def column_exists(conn: sqlite3.Connection, table: str, column: str, *, schema: str = "main") -> bool:
+    """Check if `column` exists on `table` in the given schema (sync SQLite).
+
+    Returns False (rather than raising) when `table` itself does not exist,
+    matching `PRAGMA table_info` on a missing table returning zero rows.
+    See `table_exists` for the `schema`/`table` trust requirement -- both are
+    interpolated into `PRAGMA` text since SQLite cannot bind pragma targets.
+    """
+    if not table_exists(conn, table, schema=schema):
+        return False
+    rows = conn.execute(_table_info_pragma(table, schema=schema)).fetchall()
+    return any(str(row[1]) == column for row in rows)
+
+
+async def column_exists_async(conn: aiosqlite.Connection, table: str, column: str, *, schema: str = "main") -> bool:
+    """Check if `column` exists on `table` in the given schema (async SQLite).
+
+    See `column_exists` for behavior on a missing table and the trust
+    requirement on `schema`/`table`.
+    """
+    if not await table_exists_async(conn, table, schema=schema):
+        return False
+    cursor = await conn.execute(_table_info_pragma(table, schema=schema))
+    rows = await cursor.fetchall()
+    return any(str(row[1]) == column for row in rows)
+
+
+__all__ = [
+    "table_exists",
+    "table_exists_async",
+    "index_exists",
+    "index_exists_async",
+    "column_exists",
+    "column_exists_async",
+]
diff --git a/polylogue/storage/table_existence.py b/polylogue/storage/table_existence.py
deleted file mode 100644
index 7112fdf1da..0000000000
--- a/polylogue/storage/table_existence.py
+++ /dev/null
@@ -1,69 +0,0 @@
-"""Centralized table existence checks for SQLite connections.
-
-Lives directly under `storage/` (not `storage/sqlite/`) deliberately:
-`polylogue/storage/sqlite/__init__.py` eagerly imports the heavy async
-backend stack, so a low-level module like `storage/blob_publication.py`
-importing a sibling under `storage.sqlite` triggers that whole chain and
-circles back through `insights/` -- a real circular import, not a
-hypothetical one. `storage/__init__.py` itself only exposes lazy
-`__getattr__`-based exports, so importing this module doesn't pull in that
-chain. Raw SQL belongs in `storage/` per the package-placement rules in
-`docs/architecture.md`; `core/` is reserved for no-I/O shared primitives.
-
-@owner storage-root
-"""
-
-from __future__ import annotations
-
-import sqlite3
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    import aiosqlite
-
-
-def table_exists(conn: sqlite3.Connection, name: str, *, schema: str = "main") -> bool:
-    """Check if a table exists in the given schema (sync SQLite).
-
-    `schema` selects an attached database by name (e.g. a source.db attached
-    to an index.db connection) and must be a trusted internal literal, never
-    user input -- SQLite has no way to bind a schema/database name as a query
-    parameter, so it is interpolated into the query text.
-
-    Args:
-        conn: SQLite connection
-        name: Table name to check
-        schema: Schema name (default: "main")
-
-    Returns:
-        True if the table exists, False otherwise
-    """
-    cursor = conn.execute(
-        f"SELECT 1 FROM {schema}.sqlite_master WHERE type='table' AND name=? LIMIT 1",
-        (name,),
-    )
-    return cursor.fetchone() is not None
-
-
-async def table_exists_async(conn: aiosqlite.Connection, name: str, *, schema: str = "main") -> bool:
-    """Check if a table exists in the given schema (async SQLite).
-
-    See `table_exists` for the `schema` trust requirement.
-
-    Args:
-        conn: aiosqlite connection
-        name: Table name to check
-        schema: Schema name (default: "main")
-
-    Returns:
-        True if the table exists, False otherwise
-    """
-    cursor = await conn.execute(
-        f"SELECT 1 FROM {schema}.sqlite_master WHERE type='table' AND name=? LIMIT 1",
-        (name,),
-    )
-    row = await cursor.fetchone()
-    return row is not None
-
-
-__all__ = ["table_exists", "table_exists_async"]
diff --git a/tests/unit/storage/test_table_existence.py b/tests/unit/storage/test_introspection.py
similarity index 51%
rename from tests/unit/storage/test_table_existence.py
rename to tests/unit/storage/test_introspection.py
index fc0f44aa24..6a758f8506 100644
--- a/tests/unit/storage/test_table_existence.py
+++ b/tests/unit/storage/test_introspection.py
@@ -1,4 +1,4 @@
-"""tests for polylogue.storage.table_existence.
+"""tests for polylogue.storage.introspection.
 
 polylogue-a7xr.9: the canonical table_exists()/table_exists_async() this
 module exports had never been exercised against a real connection --
@@ -6,6 +6,10 @@
 `db` as a query parameter, but sqlite_master has no `db` column, so every
 call raised sqlite3.OperationalError. Regression-tested here so a future
 edit can't silently reintroduce a query that only "works" against a mock.
+
+polylogue-48h: column_exists()/index_exists() (+ async variants) were added
+alongside table_exists() when ~25 duplicated `_table_exists`/`_column_exists`/
+`_index_exists` copies across the codebase were consolidated into this module.
 """
 
 from __future__ import annotations
@@ -15,7 +19,14 @@
 import aiosqlite
 import pytest
 
-from polylogue.storage.table_existence import table_exists, table_exists_async
+from polylogue.storage.introspection import (
+    column_exists,
+    column_exists_async,
+    index_exists,
+    index_exists_async,
+    table_exists,
+    table_exists_async,
+)
 
 
 def test_table_exists_true_for_a_real_table() -> None:
@@ -84,3 +95,72 @@ async def test_table_exists_async_checks_the_named_attached_schema(tmp_path: obj
         assert await table_exists_async(conn, "raw_sessions", schema="source") is True
     finally:
         await conn.close()
+
+
+def test_column_exists_true_for_a_real_column() -> None:
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.execute("CREATE TABLE sessions (id INTEGER, title TEXT)")
+        assert column_exists(conn, "sessions", "title") is True
+    finally:
+        conn.close()
+
+
+def test_column_exists_false_for_a_missing_column() -> None:
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.execute("CREATE TABLE sessions (id INTEGER)")
+        assert column_exists(conn, "sessions", "nonexistent_column") is False
+    finally:
+        conn.close()
+
+
+def test_column_exists_false_for_a_missing_table() -> None:
+    conn = sqlite3.connect(":memory:")
+    try:
+        assert column_exists(conn, "nonexistent_table", "id") is False
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_column_exists_async_true_for_a_real_column() -> None:
+    conn = await aiosqlite.connect(":memory:")
+    try:
+        await conn.execute("CREATE TABLE sessions (id INTEGER, title TEXT)")
+        assert await column_exists_async(conn, "sessions", "title") is True
+        assert await column_exists_async(conn, "sessions", "nonexistent_column") is False
+        assert await column_exists_async(conn, "nonexistent_table", "id") is False
+    finally:
+        await conn.close()
+
+
+def test_index_exists_true_for_a_real_index() -> None:
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.execute("CREATE TABLE sessions (id INTEGER)")
+        conn.execute("CREATE INDEX idx_sessions_id ON sessions (id)")
+        assert index_exists(conn, "idx_sessions_id") is True
+    finally:
+        conn.close()
+
+
+def test_index_exists_false_for_a_missing_index() -> None:
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.execute("CREATE TABLE sessions (id INTEGER)")
+        assert index_exists(conn, "nonexistent_index") is False
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_index_exists_async_true_for_a_real_index() -> None:
+    conn = await aiosqlite.connect(":memory:")
+    try:
+        await conn.execute("CREATE TABLE sessions (id INTEGER)")
+        await conn.execute("CREATE INDEX idx_sessions_id ON sessions (id)")
+        assert await index_exists_async(conn, "idx_sessions_id") is True
+        assert await index_exists_async(conn, "nonexistent_index") is False
+    finally:
+        await conn.close()

From 1b3838ad3ce0d34534d1316bdf52c2f88df70523 Mon Sep 17 00:00:00 2001
From: Sinity <ezo.dev@gmail.com>
Date: Mon, 3 Aug 2026 22:58:02 +0200
Subject: [PATCH 2/4] refactor(storage): repoint existing table_existence
 importers to introspection

Ref polylogue-48h

Mechanical import-path update for the 25 modules that already imported
table_exists/table_exists_async from the (now renamed) storage.table_existence
module, following the rename in the previous commit. No behavior change.
---
 polylogue/cli/commands/tutorial.py                        | 2 +-
 polylogue/cli/read_views/streaming_markdown.py            | 2 +-
 polylogue/daemon/convergence_stages.py                    | 2 +-
 polylogue/daemon/fts_orphan_audit.py                      | 2 +-
 polylogue/daemon/metrics.py                               | 8 ++------
 polylogue/hooks/__init__.py                               | 2 +-
 polylogue/insights/capture_coverage.py                    | 2 +-
 polylogue/insights/readiness.py                           | 2 +-
 polylogue/maintenance/archive_verification.py             | 2 +-
 polylogue/maintenance/rebuild_index.py                    | 2 +-
 polylogue/sources/live/convergence_debt_retry.py          | 2 +-
 polylogue/sources/live/hook_paste_enrichment.py           | 2 +-
 polylogue/storage/blob_gc.py                              | 2 +-
 polylogue/storage/blob_publication.py                     | 2 +-
 polylogue/storage/blob_repair.py                          | 2 +-
 polylogue/storage/fts/dangling_repair.py                  | 2 +-
 polylogue/storage/fts/session_repair.py                   | 2 +-
 polylogue/storage/hook_payload_ref_reconciliation.py      | 2 +-
 polylogue/storage/search/runtime.py                       | 2 +-
 polylogue/storage/source_sessions.py                      | 2 +-
 polylogue/storage/sqlite/archive_tiers/archive.py         | 2 +-
 .../sqlite/archive_tiers/session_annotations_write.py     | 2 +-
 polylogue/storage/sqlite/archive_tiers/source_write.py    | 2 +-
 polylogue/storage/sqlite/archive_tiers/user_audit.py      | 2 +-
 polylogue/storage/sqlite/archive_tiers/user_overlay.py    | 2 +-
 polylogue/storage/sqlite/archive_tiers/user_write.py      | 2 +-
 26 files changed, 27 insertions(+), 31 deletions(-)

diff --git a/polylogue/cli/commands/tutorial.py b/polylogue/cli/commands/tutorial.py
index 1c64144240..309d2f1b8f 100644
--- a/polylogue/cli/commands/tutorial.py
+++ b/polylogue/cli/commands/tutorial.py
@@ -20,7 +20,7 @@
 import click
 
 from polylogue.cli.shared.types import AppEnv
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 
 @dataclass(frozen=True, slots=True)
diff --git a/polylogue/cli/read_views/streaming_markdown.py b/polylogue/cli/read_views/streaming_markdown.py
index 6636bf19ff..64e18bf527 100644
--- a/polylogue/cli/read_views/streaming_markdown.py
+++ b/polylogue/cli/read_views/streaming_markdown.py
@@ -12,7 +12,7 @@
 from polylogue.rendering.blocks import has_structured_blocks, render_blocks_markdown
 from polylogue.rendering.core_markdown import format_message_text
 from polylogue.rendering.core_messages import normalize_render_timestamp
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 
 def stream_exact_session_markdown(
diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py
index b4c88a2b5a..d84d56d164 100644
--- a/polylogue/daemon/convergence_stages.py
+++ b/polylogue/daemon/convergence_stages.py
@@ -27,13 +27,13 @@
 from polylogue.logging import get_logger
 from polylogue.sources.origin_specs import artifact_rule_for_path
 from polylogue.storage.insights.session.runtime import session_profile_stale_predicate
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.runtime import SESSION_INSIGHT_MATERIALIZER_VERSION
 from polylogue.storage.source_sessions import (
     session_ids_for_source_path,
     session_ids_for_source_paths,
 )
 from polylogue.storage.sqlite.connection_profile import open_daemon_connection
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 if TYPE_CHECKING:
     from polylogue.sinex.service import PublicationService
diff --git a/polylogue/daemon/fts_orphan_audit.py b/polylogue/daemon/fts_orphan_audit.py
index adb54cc350..8568b93155 100644
--- a/polylogue/daemon/fts_orphan_audit.py
+++ b/polylogue/daemon/fts_orphan_audit.py
@@ -87,7 +87,7 @@ def find_orphaned_fts_sessions_sync(
     ready yet.
     """
     from polylogue.storage.archive_identity import ArchiveLocation
-    from polylogue.storage.table_existence import table_exists
+    from polylogue.storage.introspection import table_exists
 
     resolved = ArchiveLocation.resolve(db_path.parent).active_index_path
     target = resolved if resolved.exists() else (db_path if db_path.exists() else None)
diff --git a/polylogue/daemon/metrics.py b/polylogue/daemon/metrics.py
index 335c50c594..b54da5562c 100644
--- a/polylogue/daemon/metrics.py
+++ b/polylogue/daemon/metrics.py
@@ -105,8 +105,8 @@
     ARCHIVE_STORAGE_LAYOUTS,
 )
 from polylogue.storage.archive_readiness import active_rebuild_index_attempts
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
@@ -232,14 +232,10 @@ def _attached_table_exists(conn: sqlite3.Connection, schema_name: str, table: st
     if not schema_name.replace("_", "").isalnum() or not table.replace("_", "").isalnum():
         return False
     try:
-        row = conn.execute(
-            f"SELECT 1 FROM {schema_name}.sqlite_master WHERE type='table' AND name=?",
-            (table,),
-        ).fetchone()
+        return _table_exists(conn, table, schema=schema_name)
     except sqlite3.Error as exc:
         logger.warning("metrics: attached-table probe failed for %s.%s: %s", schema_name, table, exc, exc_info=True)
         return False
-    return row is not None
 
 
 def _attached_table_name(conn: sqlite3.Connection, schema_name: str, table: str) -> str:
diff --git a/polylogue/hooks/__init__.py b/polylogue/hooks/__init__.py
index 96b4d1a985..9a60bfc411 100644
--- a/polylogue/hooks/__init__.py
+++ b/polylogue/hooks/__init__.py
@@ -25,8 +25,8 @@
 
 import tomllib
 
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 HookHarness = Literal["claude-code", "codex"]
 HookChangeAction = Literal["install", "uninstall"]
diff --git a/polylogue/insights/capture_coverage.py b/polylogue/insights/capture_coverage.py
index f367a5a2b6..955a0347f9 100644
--- a/polylogue/insights/capture_coverage.py
+++ b/polylogue/insights/capture_coverage.py
@@ -51,7 +51,7 @@
 from typing import Literal
 
 from polylogue.insights.measurement.canon import content_ref
-from polylogue.storage.table_existence import table_exists
+from polylogue.storage.introspection import table_exists
 
 CoverageEvidenceSourceKind = Literal[
     "hook_session_start",
diff --git a/polylogue/insights/readiness.py b/polylogue/insights/readiness.py
index 8abebfd34f..48db124b56 100644
--- a/polylogue/insights/readiness.py
+++ b/polylogue/insights/readiness.py
@@ -13,7 +13,7 @@
 from polylogue.insights.archive_models import ARCHIVE_INSIGHT_CONTRACT_VERSION, ArchiveInsightModel
 from polylogue.maintenance.targets import build_maintenance_target_catalog
 from polylogue.storage.insights.session.runtime import SessionInsightStatusSnapshot
-from polylogue.storage.table_existence import table_exists_async as _table_exists
+from polylogue.storage.introspection import table_exists_async as _table_exists
 
 InsightReadinessVerdict = Literal[
     "ready", "partial", "empty", "missing", "stale", "incompatible", "degraded", "unknown"
diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py
index 4d25fa315b..87e4348469 100644
--- a/polylogue/maintenance/archive_verification.py
+++ b/polylogue/maintenance/archive_verification.py
@@ -44,10 +44,10 @@
 from polylogue.core.json import JSONDocument, json_document
 from polylogue.core.outcomes import OutcomeCheck, OutcomeReport, OutcomeStatus
 from polylogue.logging import get_logger
+from polylogue.storage.introspection import table_exists
 from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
-from polylogue.storage.table_existence import table_exists
 
 logger = get_logger(__name__)
 
diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py
index 534651e9e1..8379956656 100644
--- a/polylogue/maintenance/rebuild_index.py
+++ b/polylogue/maintenance/rebuild_index.py
@@ -23,10 +23,10 @@
 from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation, assert_owns_archive_location
 from polylogue.storage.fts.fts_lifecycle import rebuild_command_trigram_index_sync, rebuild_fts_index_sync
 from polylogue.storage.fts.sql import FTS_REBUILD_SQL, TRIGRAM_REBUILD_DELETE_ALL_SQL
+from polylogue.storage.introspection import table_exists
 from polylogue.storage.sqlite.action_pairs import rebuild_all_action_pairs_sync
 from polylogue.storage.sqlite.connection_profile import BULK_BUILD_WRITE_CONNECTION_PRAGMA_STATEMENTS
 from polylogue.storage.sqlite.delegation_facts import rebuild_all_delegation_facts_sync
-from polylogue.storage.table_existence import table_exists
 
 if TYPE_CHECKING:
     from polylogue.sources.revision_backfill import RawParsePrefetchCache
diff --git a/polylogue/sources/live/convergence_debt_retry.py b/polylogue/sources/live/convergence_debt_retry.py
index 45c7e00642..4eed311a2a 100644
--- a/polylogue/sources/live/convergence_debt_retry.py
+++ b/polylogue/sources/live/convergence_debt_retry.py
@@ -6,7 +6,7 @@
 from datetime import UTC, datetime
 from pathlib import Path
 
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 _HOT_INSIGHT_DEFERRED = "insights deferred until source quiet"
 
diff --git a/polylogue/sources/live/hook_paste_enrichment.py b/polylogue/sources/live/hook_paste_enrichment.py
index c8be623b5c..a7aa3d5c67 100644
--- a/polylogue/sources/live/hook_paste_enrichment.py
+++ b/polylogue/sources/live/hook_paste_enrichment.py
@@ -21,7 +21,7 @@
 from polylogue.archive.message.paste_detection import has_paste_indicator
 from polylogue.core.enums import PasteBoundary
 from polylogue.logging import get_logger
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
diff --git a/polylogue/storage/blob_gc.py b/polylogue/storage/blob_gc.py
index 8de621250c..2f6304dfb8 100644
--- a/polylogue/storage/blob_gc.py
+++ b/polylogue/storage/blob_gc.py
@@ -50,8 +50,8 @@
 from pathlib import Path
 from uuid import uuid4
 
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.connection_profile import open_connection
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 logger = logging.getLogger(__name__)
 
diff --git a/polylogue/storage/blob_publication.py b/polylogue/storage/blob_publication.py
index 15316af647..666c825c4c 100644
--- a/polylogue/storage/blob_publication.py
+++ b/polylogue/storage/blob_publication.py
@@ -13,7 +13,7 @@
 from uuid import uuid4
 
 from polylogue.storage.blob_store import BlobStore, Heartbeat, PreparedBlob
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 
 @dataclass(frozen=True, slots=True)
diff --git a/polylogue/storage/blob_repair.py b/polylogue/storage/blob_repair.py
index d1eb9a78ee..4e44631818 100644
--- a/polylogue/storage/blob_repair.py
+++ b/polylogue/storage/blob_repair.py
@@ -8,7 +8,7 @@
 
 from polylogue.config import Config
 from polylogue.logging import get_logger
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
diff --git a/polylogue/storage/fts/dangling_repair.py b/polylogue/storage/fts/dangling_repair.py
index 5a986cbf7d..187acb1fdd 100644
--- a/polylogue/storage/fts/dangling_repair.py
+++ b/polylogue/storage/fts/dangling_repair.py
@@ -17,7 +17,7 @@
     FTS_INDEX_DOC_COUNT_SQL,
     FTS_INDEXABLE_MESSAGE_COUNT_SQL,
 )
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 BOUNDED_REPAIR_PRAGMAS = (
     "PRAGMA temp_store = FILE",
diff --git a/polylogue/storage/fts/session_repair.py b/polylogue/storage/fts/session_repair.py
index ce251dbcbe..986462ade0 100644
--- a/polylogue/storage/fts/session_repair.py
+++ b/polylogue/storage/fts/session_repair.py
@@ -5,7 +5,7 @@
 import sqlite3
 from typing import Any, cast
 
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 
 def _row_int(row: sqlite3.Row | tuple[object, ...] | None, key: int | str) -> int:
diff --git a/polylogue/storage/hook_payload_ref_reconciliation.py b/polylogue/storage/hook_payload_ref_reconciliation.py
index 5c8116f8d2..069ba5a34f 100644
--- a/polylogue/storage/hook_payload_ref_reconciliation.py
+++ b/polylogue/storage/hook_payload_ref_reconciliation.py
@@ -35,8 +35,8 @@
 import sqlite3
 from dataclasses import dataclass
 
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.archive_tiers.source_write import deterministic_raw_session_id
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 
 @dataclass(frozen=True, slots=True)
diff --git a/polylogue/storage/search/runtime.py b/polylogue/storage/search/runtime.py
index 94b985a897..2bdfc8f9dc 100644
--- a/polylogue/storage/search/runtime.py
+++ b/polylogue/storage/search/runtime.py
@@ -9,12 +9,12 @@
 
 from polylogue.core.errors import DatabaseError
 from polylogue.storage.fts.fts_lifecycle import check_fts_readiness, message_fts_search_readiness_sync
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.search.cache import SearchCacheKey
 from polylogue.storage.search.models import SearchHit, SearchResult
 from polylogue.storage.search.query_builders import build_ranked_session_search_query, session_web_url
 from polylogue.storage.search.query_support import normalize_fts5_query, sort_key_to_iso
 from polylogue.storage.sqlite.connection import open_read_connection
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 
 @lru_cache(maxsize=128)
diff --git a/polylogue/storage/source_sessions.py b/polylogue/storage/source_sessions.py
index 397c53dd14..24d2ab11ea 100644
--- a/polylogue/storage/source_sessions.py
+++ b/polylogue/storage/source_sessions.py
@@ -7,7 +7,7 @@
 from pathlib import Path
 
 from polylogue.logging import get_logger
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py
index 034613dce4..32f4f6122d 100644
--- a/polylogue/storage/sqlite/archive_tiers/archive.py
+++ b/polylogue/storage/sqlite/archive_tiers/archive.py
@@ -145,6 +145,7 @@
     SessionInsightStatusSnapshot,
 )
 from polylogue.storage.insights.session.status import session_insight_status_sync
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.raw.models import RawSessionStateUpdate
 from polylogue.storage.runtime.store_constants import SESSION_INSIGHT_MATERIALIZER_VERSION
 from polylogue.storage.search.query_support import normalize_fts5_query
@@ -316,7 +317,6 @@
     run_relation_sql,
 )
 from polylogue.storage.sqlite.runtime_indexes import ensure_runtime_indexes_sync
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 
 @dataclass(slots=True)
diff --git a/polylogue/storage/sqlite/archive_tiers/session_annotations_write.py b/polylogue/storage/sqlite/archive_tiers/session_annotations_write.py
index 5faf2999ef..f673eb9957 100644
--- a/polylogue/storage/sqlite/archive_tiers/session_annotations_write.py
+++ b/polylogue/storage/sqlite/archive_tiers/session_annotations_write.py
@@ -24,7 +24,7 @@
 import sqlite3
 from dataclasses import dataclass
 
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 
 @dataclass(frozen=True, slots=True)
diff --git a/polylogue/storage/sqlite/archive_tiers/source_write.py b/polylogue/storage/sqlite/archive_tiers/source_write.py
index 6e565abe0d..fe8656dde5 100644
--- a/polylogue/storage/sqlite/archive_tiers/source_write.py
+++ b/polylogue/storage/sqlite/archive_tiers/source_write.py
@@ -15,9 +15,9 @@
 
 from polylogue.archive.revision_authority import RawRevisionAuthority, RawRevisionEnvelope
 from polylogue.core.enums import ArtifactSupportStatus, Origin, Provider, ValidationMode, ValidationStatus
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.raw.models import RawSessionStateUpdate
 from polylogue.storage.sqlite.raw_state_update import compile_raw_state_update
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 
 class ContentExcisedError(RuntimeError):
diff --git a/polylogue/storage/sqlite/archive_tiers/user_audit.py b/polylogue/storage/sqlite/archive_tiers/user_audit.py
index 72aeafa368..52beba28d9 100644
--- a/polylogue/storage/sqlite/archive_tiers/user_audit.py
+++ b/polylogue/storage/sqlite/archive_tiers/user_audit.py
@@ -6,8 +6,8 @@
 from dataclasses import dataclass
 
 from polylogue.core.json import JSONDocument, json_document
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.archive_tiers.user_write import AssertionKind
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 _ASSERTION_BACKED_SURFACES: dict[str, AssertionKind] = {
     "marks": AssertionKind.MARK,
diff --git a/polylogue/storage/sqlite/archive_tiers/user_overlay.py b/polylogue/storage/sqlite/archive_tiers/user_overlay.py
index d46f727ac5..ec7514833e 100644
--- a/polylogue/storage/sqlite/archive_tiers/user_overlay.py
+++ b/polylogue/storage/sqlite/archive_tiers/user_overlay.py
@@ -5,7 +5,7 @@
 import sqlite3
 from dataclasses import dataclass
 
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 
 @dataclass(frozen=True, slots=True)
diff --git a/polylogue/storage/sqlite/archive_tiers/user_write.py b/polylogue/storage/sqlite/archive_tiers/user_write.py
index 67d75f78a9..55fde537ef 100644
--- a/polylogue/storage/sqlite/archive_tiers/user_write.py
+++ b/polylogue/storage/sqlite/archive_tiers/user_write.py
@@ -28,8 +28,8 @@
 from polylogue.core.enums import AssertionKind, AssertionStatus, AssertionVisibility
 from polylogue.core.json import JSONValue
 from polylogue.core.refs import ObjectRef, normalize_object_ref_text, normalize_public_ref_text
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.connection_profile import WRITE_CONNECTION_PROFILE
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 if TYPE_CHECKING:
     from polylogue.insights.judgment.types import ComparativeJudgment

From 744a93e070bedba21ad74197dae94717cecbfcfe Mon Sep 17 00:00:00 2001
From: Sinity <ezo.dev@gmail.com>
Date: Mon, 3 Aug 2026 22:58:38 +0200
Subject: [PATCH 3/4] refactor(storage): eliminate remaining duplicated
 table/column/index exists checks

Ref polylogue-48h

Replace the ~25 remaining independently maintained _table_exists/
_column_exists/_index_exists definitions (and sync/async pairs) with imports
from polylogue.storage.introspection. Several copies checked
`type IN ('table', 'virtual table')` or `type IN ('table', 'shadow')`;
neither value is ever a real sqlite_master `type` (virtual tables, including
FTS5 shadow tables, register with type='table'), so redirecting to the
canonical `type='table'` check is behavior-preserving -- confirmed directly
against a live FTS5 vtable (all 6 shadow tables show type='table').

Genuinely distinct behavior is kept as a thin, differently-named wrapper
delegating to the canonical function rather than folded away:
- usage.py's _table_exists_in_schema and daemon/metrics.py's
  _attached_table_exists swallow sqlite3.Error for a schema alias that may
  not be ATTACHed yet.
- embeddings/support.py's table_exists_sync_missing_safe/
  table_exists_async_missing_safe swallow a still-in-flight
  sqlite3.OperationalError ("no such table").
- cli/commands/status.py and storage/archive_readiness.py keep their local
  _schema_object_exists/_view_exists (view support isn't part of this
  bead's scope) and only redirect _table_exists/_column_exists.
- storage/sqlite/run_projection_relations.py's table_exists_sync had no
  callers anywhere in the repo; deleted outright instead of redirected.

Test fixes for query-text-sensitive mocks broken by the consolidation:
- tests/unit/daemon/test_daemon_cli.py: FakeConnection assertions expected
  the old `type IN ('table', 'virtual table')` text; updated to the
  behaviorally-identical `type='table'` text the canonical function emits.
- tests/unit/storage/test_embedding_contracts.py: _VeclessConnection.execute
  dropped bind parameters entirely, which only worked because the old
  support.py table_exists_sync interpolated the table name into the SQL
  string instead of binding it. The canonical function binds `name` as a
  real parameter, so the fake now forwards parameters to the real
  sqlite3.Connection.execute -- the fake was silently wrong before, not the
  production code.
- tests/unit/cli/commands/test_status.py: mypy --strict's no-implicit-
  reexport rule correctly rejects importing _table_exists/_column_exists
  through status.py's own re-import; the test now imports table_exists/
  column_exists directly from polylogue.storage.introspection, matching the
  existing polylogue-ogn1 precedent for _archive_readiness_status.
---
 polylogue/browser_capture/receiver.py         |  9 +---
 polylogue/cli/commands/status.py              | 10 +---
 polylogue/daemon/embedding_backlog.py         |  9 +---
 polylogue/daemon/fts_automerge.py             |  9 +---
 polylogue/daemon/fts_startup.py               | 10 +---
 polylogue/daemon/fts_status.py                |  9 +---
 polylogue/operations/archive_debt.py          |  9 +---
 .../generation/archive_workload_profile.py    | 11 +---
 polylogue/storage/archive_readiness.py        | 10 +---
 polylogue/storage/blob_integrity.py           |  9 +---
 polylogue/storage/derived/derived_status.py   |  9 +---
 .../storage/embeddings/embedding_stats.py     | 12 ++---
 .../storage/embeddings/materialization.py     | 11 +---
 polylogue/storage/embeddings/preflight.py     |  9 +---
 .../storage/embeddings/status_payload.py      | 27 +++-------
 polylogue/storage/embeddings/support.py       | 33 ++++++------
 polylogue/storage/fts/freshness.py            | 53 ++++---------------
 polylogue/storage/fts/fts_lifecycle.py        | 21 +-------
 .../storage/insights/feedback/__init__.py     |  7 +--
 polylogue/storage/raw_retention.py            |  9 +---
 polylogue/storage/session_replacement.py      | 20 +------
 .../sqlite/run_projection_relations.py        | 10 ----
 polylogue/storage/usage.py                    | 11 ++--
 tests/unit/cli/commands/test_status.py        | 11 +++-
 tests/unit/daemon/test_daemon_cli.py          | 20 +++----
 .../unit/storage/test_embedding_contracts.py  |  3 +-
 26 files changed, 88 insertions(+), 273 deletions(-)

diff --git a/polylogue/browser_capture/receiver.py b/polylogue/browser_capture/receiver.py
index 0a71047f63..aa96933acc 100644
--- a/polylogue/browser_capture/receiver.py
+++ b/polylogue/browser_capture/receiver.py
@@ -39,6 +39,7 @@
     browser_capture_receiver_token_path,
     browser_capture_spool_root,
 )
+from polylogue.storage.introspection import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
@@ -335,14 +336,6 @@ def _open_readonly_sqlite(path: Path) -> sqlite3.Connection | None:
     return conn
 
 
-def _table_exists(conn: sqlite3.Connection, table_name: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'view') AND name=? LIMIT 1",
-        (table_name,),
-    ).fetchone()
-    return row is not None
-
-
 def _columns(conn: sqlite3.Connection, table_name: str) -> set[str]:
     return {str(row["name"]) for row in conn.execute(f"PRAGMA table_info({table_name})").fetchall()}
 
diff --git a/polylogue/cli/commands/status.py b/polylogue/cli/commands/status.py
index 6e37e98ac4..b2d95a46e0 100644
--- a/polylogue/cli/commands/status.py
+++ b/polylogue/cli/commands/status.py
@@ -27,6 +27,8 @@
 from polylogue.storage.archive_identity import archive_file_set_root
 from polylogue.storage.archive_readiness import archive_readiness_status as _archive_readiness_status
 from polylogue.storage.archive_readiness import raw_materialization_ready as _raw_materialization_ready_bool
+from polylogue.storage.introspection import column_exists as _column_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 
 logger = get_logger(__name__)
@@ -202,10 +204,6 @@ def _schema_object_exists(conn: Any, name: str, *, types: Sequence[str]) -> bool
     return row is not None
 
 
-def _table_exists(conn: Any, table_name: str) -> bool:
-    return _schema_object_exists(conn, table_name, types=("table",))
-
-
 def _view_exists(conn: Any, view_name: str) -> bool:
     return _schema_object_exists(conn, view_name, types=("view",))
 
@@ -894,10 +892,6 @@ def _archive_source_table_count(conn: Any, *, table: str, sql: str, configured_r
         return 0
 
 
-def _column_exists(conn: Any, table_name: str, column_name: str) -> bool:
-    return any(str(row[1]) == column_name for row in conn.execute(f"PRAGMA table_info({table_name})").fetchall())
-
-
 # Live ingest workload is read directly from ops.db so it is visible even when
 # the daemon runs with --no-api (no HTTP /api/status to query). The data already
 # exists in ingest_attempts/ingest_cursor/convergence_debt; this surface derives
diff --git a/polylogue/daemon/embedding_backlog.py b/polylogue/daemon/embedding_backlog.py
index edd5eeea4f..8ff92af2e9 100644
--- a/polylogue/daemon/embedding_backlog.py
+++ b/polylogue/daemon/embedding_backlog.py
@@ -12,6 +12,7 @@
 from polylogue.config import load_polylogue_config
 from polylogue.logging import get_logger
 from polylogue.sources.live.sqlite_locking import is_transient_sqlite_lock
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
 
 if TYPE_CHECKING:
@@ -392,14 +393,6 @@ def embedding_catchup_estimated_cost_this_month(conn: sqlite3.Connection) -> flo
     return float(row[0] or 0.0) if row is not None else 0.0
 
 
-def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1",
-        (table,),
-    ).fetchone()
-    return row is not None
-
-
 __all__ = [
     "EMBEDDING_BACKLOG_RETRY_INTERVAL_SECONDS",
     "EMBEDDING_ORPHAN_RECONCILE_INTERVAL_SECONDS",
diff --git a/polylogue/daemon/fts_automerge.py b/polylogue/daemon/fts_automerge.py
index 71c1df22e5..8e016e048f 100644
--- a/polylogue/daemon/fts_automerge.py
+++ b/polylogue/daemon/fts_automerge.py
@@ -24,6 +24,7 @@
 from pathlib import Path
 
 from polylogue.logging import get_logger
+from polylogue.storage.introspection import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
@@ -37,14 +38,6 @@
 _PERIODIC_MERGE_WORK_UNITS = 500
 
 
-def _table_exists(conn: sqlite3.Connection, name: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'shadow') AND name = ? LIMIT 1",
-        (name,),
-    ).fetchone()
-    return row is not None
-
-
 def configure_fts_automerge_sync(conn: sqlite3.Connection) -> list[str]:
     """Persist ``automerge=0`` for each present FTS surface.
 
diff --git a/polylogue/daemon/fts_startup.py b/polylogue/daemon/fts_startup.py
index ba6d85cf51..a9e8d5aa4a 100644
--- a/polylogue/daemon/fts_startup.py
+++ b/polylogue/daemon/fts_startup.py
@@ -13,6 +13,7 @@
 from pathlib import Path
 
 from polylogue.logging import get_logger
+from polylogue.storage.introspection import table_exists as table_exists_sync
 
 logger = get_logger(__name__)
 _ARCHIVE_MESSAGE_FTS_TRIGGERS = ("messages_fts_ai", "messages_fts_ad", "messages_fts_au")
@@ -52,15 +53,6 @@ def missing_fts_triggers_sync(conn: sqlite3.Connection) -> list[str]:
     return [name for name in expected if name not in present]
 
 
-def table_exists_sync(conn: sqlite3.Connection, table_name: str) -> bool:
-    """Return whether ``table_name`` is present in ``sqlite_master``."""
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1",
-        (table_name,),
-    ).fetchone()
-    return row is not None
-
-
 def record_fts_freshness_snapshot_sync(conn: sqlite3.Connection) -> None:
     """Write per-surface freshness rows after a successful startup readiness pass.
 
diff --git a/polylogue/daemon/fts_status.py b/polylogue/daemon/fts_status.py
index 219c240600..2d7be48646 100644
--- a/polylogue/daemon/fts_status.py
+++ b/polylogue/daemon/fts_status.py
@@ -12,6 +12,7 @@
 from polylogue.storage.fts.freshness import STALE, UNKNOWN, freshness_ready_record_trusted
 from polylogue.storage.fts.fts_lifecycle import FtsInvariantSnapshot, FtsSurfaceInvariant, fts_invariant_snapshot_sync
 from polylogue.storage.fts.sql import message_identity_mismatch_sql
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
 
 logger = get_logger(__name__)
@@ -53,14 +54,6 @@ class FTSReadiness(BaseModel):
     surfaces: dict[str, dict[str, int | bool | str | None]] = Field(default_factory=dict)
 
 
-def _table_exists(conn: sqlite3.Connection, table_name: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1",
-        (table_name,),
-    ).fetchone()
-    return row is not None
-
-
 def _triggers_present(conn: sqlite3.Connection, trigger_names: tuple[str, ...]) -> bool:
     placeholders = ",".join("?" for _ in trigger_names)
     rows = conn.execute(
diff --git a/polylogue/operations/archive_debt.py b/polylogue/operations/archive_debt.py
index 7c8d001290..7c1eab17b4 100644
--- a/polylogue/operations/archive_debt.py
+++ b/polylogue/operations/archive_debt.py
@@ -22,6 +22,7 @@
 from polylogue.daemon.fts_status import fts_readiness_info
 from polylogue.maintenance.targets import MAINTENANCE_TARGET_NAMES
 from polylogue.sources.dispatch import is_stream_record_provider
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.repair import RAW_MATERIALIZATION_EXECUTE_BLOB_LIMIT_BYTES
 from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS
 from polylogue.storage.sqlite.archive_tiers.user_write import list_assertion_candidates
@@ -908,14 +909,6 @@ def _provider_usage_rows(index_db: Path) -> list[ArchiveDebtRowPayload]:
     return debt_rows
 
 
-def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'view') AND name = ? LIMIT 1",
-        (table,),
-    ).fetchone()
-    return row is not None
-
-
 def _embedding_rows(index_db: Path) -> list[ArchiveDebtRowPayload]:
     info = embedding_readiness_info(index_db, detail=True)
     rows: list[ArchiveDebtRowPayload] = []
diff --git a/polylogue/schemas/generation/archive_workload_profile.py b/polylogue/schemas/generation/archive_workload_profile.py
index 03e4ceeb7a..5a27698dcd 100644
--- a/polylogue/schemas/generation/archive_workload_profile.py
+++ b/polylogue/schemas/generation/archive_workload_profile.py
@@ -18,6 +18,7 @@
     workload_profile_identity,
 )
 from polylogue.schemas.workload_tiers import WorkloadScaleTier, WorkloadSelectivityTier
+from polylogue.storage.introspection import table_exists as _table_exists
 
 ARCHIVE_WORKLOAD_PROFILE_FILE = "archive-workload-profile.json.gz"
 _INFERENCE_VERSION = "archive-composition-v1"
@@ -31,16 +32,6 @@ def _connect_read_only(path: Path) -> sqlite3.Connection:
     return conn
 
 
-def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
-    return (
-        conn.execute(
-            "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'view') AND name = ?",
-            (table,),
-        ).fetchone()
-        is not None
-    )
-
-
 def _columns(conn: sqlite3.Connection, table: str) -> set[str]:
     if not _table_exists(conn, table):
         return set()
diff --git a/polylogue/storage/archive_readiness.py b/polylogue/storage/archive_readiness.py
index 561f86b000..5d628c20e1 100644
--- a/polylogue/storage/archive_readiness.py
+++ b/polylogue/storage/archive_readiness.py
@@ -20,6 +20,8 @@
 from polylogue.core.payload_coercion import row_int as _row_int
 from polylogue.logging import get_logger
 from polylogue.storage.insights.session.status import session_insight_status_sync
+from polylogue.storage.introspection import column_exists as _column_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.raw_authority import raw_authority_detail_query_handle
 from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
@@ -1007,18 +1009,10 @@ def _schema_object_exists(conn: sqlite3.Connection, name: str, *, types: tuple[s
     return row is not None
 
 
-def _table_exists(conn: sqlite3.Connection, table_name: str) -> bool:
-    return _schema_object_exists(conn, table_name, types=("table",))
-
-
 def _view_exists(conn: sqlite3.Connection, view_name: str) -> bool:
     return _schema_object_exists(conn, view_name, types=("view",))
 
 
-def _column_exists(conn: sqlite3.Connection, table_name: str, column_name: str) -> bool:
-    return any(str(row[1]) == column_name for row in conn.execute(f"PRAGMA table_info({table_name})").fetchall())
-
-
 def _action_readiness_counts(conn: sqlite3.Connection) -> dict[str, Any]:
     """Return exact, non-vacuous evidence for the derived ``actions`` view."""
     tool_use_block_count = (
diff --git a/polylogue/storage/blob_integrity.py b/polylogue/storage/blob_integrity.py
index dda030a3f6..1a6c17c797 100644
--- a/polylogue/storage/blob_integrity.py
+++ b/polylogue/storage/blob_integrity.py
@@ -28,8 +28,9 @@
 from polylogue.core.json import loads as json_loads
 from polylogue.logging import get_logger
 from polylogue.storage.blob_store import BlobStore
+from polylogue.storage.introspection import column_exists as _column_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.connection import open_read_connection
-from polylogue.storage.table_existence import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
@@ -466,12 +467,6 @@ def _blob_hash_text(value: object) -> str | None:
     return text if text else None
 
 
-def _column_exists(conn: sqlite3.Connection, table: str, column: str) -> bool:
-    if not _table_exists(conn, table):
-        return False
-    return any(row[1] == column for row in conn.execute(f"PRAGMA table_info({table})").fetchall())
-
-
 def _archive_source_blob_hashes(conn: sqlite3.Connection) -> list[str]:
     hashes_by_table = _archive_source_blob_hashes_by_table(conn)
     hashes: set[str] = set()
diff --git a/polylogue/storage/derived/derived_status.py b/polylogue/storage/derived/derived_status.py
index 00129ae8b6..89542a66e6 100644
--- a/polylogue/storage/derived/derived_status.py
+++ b/polylogue/storage/derived/derived_status.py
@@ -28,6 +28,7 @@
 from polylogue.storage.derived.insights import build_archive_insight_statuses, pending_docs, pending_rows
 from polylogue.storage.insights.session.runtime import SessionInsightStatusSnapshot
 from polylogue.storage.insights.session.status import session_insight_status_sync
+from polylogue.storage.introspection import table_exists as _table_exists
 
 MetricValue: TypeAlias = int | bool
 Metrics: TypeAlias = dict[str, MetricValue]
@@ -41,14 +42,6 @@
 # ---------------------------------------------------------------------------
 
 
-def _table_exists(conn: sqlite3.Connection, name: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table','view') AND name = ? LIMIT 1",
-        (name,),
-    ).fetchone()
-    return row is not None
-
-
 def _count(conn: sqlite3.Connection, sql: str, params: tuple[object, ...] = ()) -> int:
     row = conn.execute(sql, params).fetchone()
     return int(row[0] or 0) if row is not None else 0
diff --git a/polylogue/storage/embeddings/embedding_stats.py b/polylogue/storage/embeddings/embedding_stats.py
index 32a0b91200..466b3f1670 100644
--- a/polylogue/storage/embeddings/embedding_stats.py
+++ b/polylogue/storage/embeddings/embedding_stats.py
@@ -31,8 +31,8 @@
     optional_row_sync,
     optional_rows_async,
     optional_rows_sync,
-    table_exists_async,
-    table_exists_sync,
+    table_exists_async_missing_safe,
+    table_exists_sync_missing_safe,
 )
 from polylogue.storage.insights.session.status import (
     session_insight_status_async,
@@ -141,8 +141,8 @@ def _dimension_counts(rows: list[sqlite3.Row]) -> dict[int, int]:
 
 
 def _base_parts_sync(conn: sqlite3.Connection, *, detail: bool) -> _EmbeddingStatsParts:
-    sessions_exist = table_exists_sync(conn, "sessions")
-    status_exists = table_exists_sync(conn, "embedding_status")
+    sessions_exist = table_exists_sync_missing_safe(conn, "sessions")
+    status_exists = table_exists_sync_missing_safe(conn, "embedding_status")
     if not detail:
         return _EmbeddingStatsParts(
             bounds=None,
@@ -192,8 +192,8 @@ def _base_parts_sync(conn: sqlite3.Connection, *, detail: bool) -> _EmbeddingSta
 
 
 async def _base_parts_async(conn: aiosqlite.Connection, *, detail: bool) -> _EmbeddingStatsParts:
-    sessions_exist = await table_exists_async(conn, "sessions")
-    status_exists = await table_exists_async(conn, "embedding_status")
+    sessions_exist = await table_exists_async_missing_safe(conn, "sessions")
+    status_exists = await table_exists_async_missing_safe(conn, "embedding_status")
     if not detail:
         return _EmbeddingStatsParts(
             bounds=None,
diff --git a/polylogue/storage/embeddings/materialization.py b/polylogue/storage/embeddings/materialization.py
index 6f323937ce..a19de329a8 100644
--- a/polylogue/storage/embeddings/materialization.py
+++ b/polylogue/storage/embeddings/materialization.py
@@ -33,7 +33,8 @@
     register_embedding_identity_sql,
     sql_string_literal,
 )
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import index_exists as _index_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 if TYPE_CHECKING:
     from polylogue.archive.models import Session
@@ -1700,14 +1701,6 @@ def _should_embed_archive_message(material_origin: object, message_type: object,
     return str(material_origin) in _PROSE_MATERIAL_ORIGINS
 
 
-def _index_exists(conn: sqlite3.Connection, index: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type='index' AND name=? LIMIT 1",
-        (index,),
-    ).fetchone()
-    return row is not None
-
-
 def _usable_db_path(db_path: object) -> Path | None:
     if isinstance(db_path, Path):
         return db_path
diff --git a/polylogue/storage/embeddings/preflight.py b/polylogue/storage/embeddings/preflight.py
index bf75527a2b..d5bcab7e38 100644
--- a/polylogue/storage/embeddings/preflight.py
+++ b/polylogue/storage/embeddings/preflight.py
@@ -12,6 +12,7 @@
 from pathlib import Path
 
 from polylogue.storage.embeddings.identity import EmbeddingRecipe
+from polylogue.storage.introspection import table_exists as _table_exists
 
 
 @dataclass(frozen=True, slots=True)
@@ -243,14 +244,6 @@ def _select_archive_pending_window(
     return [(item.session_id, item.message_count) for item in pending]
 
 
-def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1",
-        (table,),
-    ).fetchone()
-    return row is not None
-
-
 def build_preflight_report(
     db_path: Path,
     *,
diff --git a/polylogue/storage/embeddings/status_payload.py b/polylogue/storage/embeddings/status_payload.py
index 74cf06dce1..bf560572a7 100644
--- a/polylogue/storage/embeddings/status_payload.py
+++ b/polylogue/storage/embeddings/status_payload.py
@@ -25,6 +25,7 @@
 )
 from polylogue.storage.embeddings.models import EmbeddingStatsSnapshot
 from polylogue.storage.embeddings.progress import EmbeddingCatchupRunPayload
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.search_providers.sqlite_vec_support import (
     ESTIMATED_TOKENS_PER_MESSAGE,
     VOYAGE_4_COST_PER_1M_TOKENS,
@@ -174,25 +175,9 @@ def _total_sessions(conn: sqlite3.Connection) -> int:
     return optional_count_sync(conn, "SELECT COUNT(*) FROM sessions")
 
 
-def _table_exists(conn: sqlite3.Connection, table_name: str) -> bool:
-    return (
-        conn.execute(
-            "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'view') AND name = ? LIMIT 1",
-            (table_name,),
-        ).fetchone()
-        is not None
-    )
-
-
 def _attached_table_exists(conn: sqlite3.Connection, schema_name: str, table_name: str) -> bool:
     quoted_schema = '"' + schema_name.replace('"', '""') + '"'
-    return (
-        conn.execute(
-            f"SELECT 1 FROM {quoted_schema}.sqlite_master WHERE type IN ('table', 'view') AND name = ? LIMIT 1",
-            (table_name,),
-        ).fetchone()
-        is not None
-    )
+    return _table_exists(conn, table_name, schema=quoted_schema)
 
 
 def _attached_table_name(conn: sqlite3.Connection, schema_name: str, table_name: str) -> str:
@@ -1130,7 +1115,7 @@ def embedding_status_payload(
     from polylogue.storage.archive_identity import archive_file_set_root
     from polylogue.storage.embeddings.embedding_stats import read_embedding_stats_sync
     from polylogue.storage.embeddings.progress import latest_embedding_catchup_run
-    from polylogue.storage.embeddings.support import table_exists_sync
+    from polylogue.storage.embeddings.support import table_exists_sync_missing_safe
 
     cfg = load_polylogue_config()
     db_path = Path(env.config.db_path)
@@ -1172,7 +1157,11 @@ def embedding_status_payload(
             include_retrieval_bands=include_retrieval_bands,
             detail=include_detail,
         )
-        latest_run = latest_embedding_catchup_run(conn) if table_exists_sync(conn, "embedding_catchup_runs") else None
+        latest_run = (
+            latest_embedding_catchup_run(conn)
+            if table_exists_sync_missing_safe(conn, "embedding_catchup_runs")
+            else None
+        )
     finally:
         conn.close()
 
diff --git a/polylogue/storage/embeddings/support.py b/polylogue/storage/embeddings/support.py
index 608cc389d4..9fdff6d98a 100644
--- a/polylogue/storage/embeddings/support.py
+++ b/polylogue/storage/embeddings/support.py
@@ -8,6 +8,8 @@
 import aiosqlite
 
 from polylogue.storage.insights.session.runtime import SessionInsightStatusSnapshot
+from polylogue.storage.introspection import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists_async as _table_exists_async
 
 StatsRow = sqlite3.Row | tuple[object, ...]
 
@@ -157,30 +159,25 @@ def is_missing_table_error(exc: sqlite3.OperationalError) -> bool:
     )
 
 
-def table_exists_sync(conn: sqlite3.Connection, table: str) -> bool:
-    table_name = table.replace("'", "''")
+def table_exists_sync_missing_safe(conn: sqlite3.Connection, table: str) -> bool:
+    # Thin wrapper (not a duplicate): swallows a still-in-flight
+    # sqlite3.OperationalError ("no such table") that `introspection.table_exists`
+    # itself never raises but a caller mid-migration/attach can still hit.
     try:
-        row = conn.execute(
-            f"SELECT 1 FROM sqlite_master WHERE type='table' AND name='{table_name}' LIMIT 1",
-        ).fetchone()
+        return _table_exists(conn, table)
     except sqlite3.OperationalError as exc:
         if is_missing_table_error(exc):
             return False
         raise
-    return row is not None
 
 
-async def table_exists_async(conn: aiosqlite.Connection, table: str) -> bool:
-    table_name = table.replace("'", "''")
+async def table_exists_async_missing_safe(conn: aiosqlite.Connection, table: str) -> bool:
     try:
-        cursor = await conn.execute(
-            f"SELECT 1 FROM sqlite_master WHERE type='table' AND name='{table_name}' LIMIT 1",
-        )
+        return await _table_exists_async(conn, table)
     except sqlite3.OperationalError as exc:
         if is_missing_table_error(exc):
             return False
         raise
-    return await cursor.fetchone() is not None
 
 
 def optional_count_sync(conn: sqlite3.Connection, sql: str) -> int:
@@ -203,11 +200,11 @@ def embedded_message_count_sync(conn: sqlite3.Connection) -> int:
     per-message count and is checked first; older/legacy shapes fall back to
     the pre-v4 behavior.
     """
-    if table_exists_sync(conn, "message_embedding_refs"):
+    if table_exists_sync_missing_safe(conn, "message_embedding_refs"):
         return optional_count_sync(conn, "SELECT COUNT(*) FROM message_embedding_refs")
-    if table_exists_sync(conn, "message_embeddings_meta"):
+    if table_exists_sync_missing_safe(conn, "message_embeddings_meta"):
         return optional_count_sync(conn, "SELECT COUNT(*) FROM message_embeddings_meta")
-    if table_exists_sync(conn, "message_embeddings_rowids"):
+    if table_exists_sync_missing_safe(conn, "message_embeddings_rowids"):
         return optional_count_sync(conn, "SELECT COUNT(*) FROM message_embeddings_rowids")
     return optional_count_sync(conn, "SELECT COUNT(*) FROM message_embeddings")
 
@@ -243,11 +240,11 @@ async def optional_count_async(conn: aiosqlite.Connection, sql: str) -> int:
 
 async def embedded_message_count_async(conn: aiosqlite.Connection) -> int:
     """Count messages that have a current embedding (see sync counterpart)."""
-    if await table_exists_async(conn, "message_embedding_refs"):
+    if await table_exists_async_missing_safe(conn, "message_embedding_refs"):
         return await optional_count_async(conn, "SELECT COUNT(*) FROM message_embedding_refs")
-    if await table_exists_async(conn, "message_embeddings_meta"):
+    if await table_exists_async_missing_safe(conn, "message_embeddings_meta"):
         return await optional_count_async(conn, "SELECT COUNT(*) FROM message_embeddings_meta")
-    if await table_exists_async(conn, "message_embeddings_rowids"):
+    if await table_exists_async_missing_safe(conn, "message_embeddings_rowids"):
         return await optional_count_async(conn, "SELECT COUNT(*) FROM message_embeddings_rowids")
     return await optional_count_async(conn, "SELECT COUNT(*) FROM message_embeddings")
 
diff --git a/polylogue/storage/fts/freshness.py b/polylogue/storage/fts/freshness.py
index 0507e93a89..9f29f3c5ab 100644
--- a/polylogue/storage/fts/freshness.py
+++ b/polylogue/storage/fts/freshness.py
@@ -8,6 +8,7 @@
 
 import aiosqlite
 
+from polylogue.storage.introspection import table_exists, table_exists_async
 from polylogue.storage.sqlite.archive_tiers.index import FTS_FRESHNESS_STATE_DDL
 
 FRESHNESS_TABLE = "fts_freshness_state"
@@ -91,44 +92,8 @@ def freshness_ready_record_trusted(
     return not (source_rows == 0 and indexed_rows == 0 and source_has_rows is not False)
 
 
-def _table_exists_sync(conn: sqlite3.Connection) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
-        (FRESHNESS_TABLE,),
-    ).fetchone()
-    return row is not None
-
-
-def _named_table_exists_sync(conn: sqlite3.Connection, table_name: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1",
-        (table_name,),
-    ).fetchone()
-    return row is not None
-
-
-async def _table_exists_async(conn: aiosqlite.Connection) -> bool:
-    row = await (
-        await conn.execute(
-            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
-            (FRESHNESS_TABLE,),
-        )
-    ).fetchone()
-    return row is not None
-
-
-async def _named_table_exists_async(conn: aiosqlite.Connection, table_name: str) -> bool:
-    row = await (
-        await conn.execute(
-            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1",
-            (table_name,),
-        )
-    ).fetchone()
-    return row is not None
-
-
 def _message_fts_source_has_rows_sync(conn: sqlite3.Connection) -> bool | None:
-    if not _named_table_exists_sync(conn, "blocks"):
+    if not table_exists(conn, "blocks"):
         return None
     row = conn.execute("SELECT 1 FROM blocks WHERE search_text != '' LIMIT 1").fetchone()
     return row is not None
@@ -144,7 +109,7 @@ def _message_fts_triggers_present_sync(conn: sqlite3.Connection) -> bool:
 
 
 async def _message_fts_source_has_rows_async(conn: aiosqlite.Connection) -> bool | None:
-    if not await _named_table_exists_async(conn, "blocks"):
+    if not await table_exists_async(conn, "blocks"):
         return None
     row = await (await conn.execute("SELECT 1 FROM blocks WHERE search_text != '' LIMIT 1")).fetchone()
     return row is not None
@@ -368,7 +333,7 @@ async def message_fts_marked_ready_async(conn: aiosqlite.Connection) -> bool:
 
 
 def message_fts_recorded_state_sync(conn: sqlite3.Connection) -> str | None:
-    if not _table_exists_sync(conn):
+    if not table_exists(conn, FRESHNESS_TABLE):
         return None
     row = conn.execute(
         "SELECT state FROM fts_freshness_state WHERE surface=?",
@@ -383,7 +348,7 @@ def message_fts_recorded_state_sync(conn: sqlite3.Connection) -> str | None:
 
 
 async def message_fts_recorded_state_async(conn: aiosqlite.Connection) -> str | None:
-    if not await _table_exists_async(conn):
+    if not await table_exists_async(conn, FRESHNESS_TABLE):
         return None
     row = await (
         await conn.execute(
@@ -400,7 +365,7 @@ async def message_fts_recorded_state_async(conn: aiosqlite.Connection) -> str |
 
 
 def _message_fts_record_sync(conn: sqlite3.Connection) -> dict[str, object] | None:
-    if not _table_exists_sync(conn):
+    if not table_exists(conn, FRESHNESS_TABLE):
         return None
     columns = {str(row[1]) for row in conn.execute(f"PRAGMA table_info({FRESHNESS_TABLE})").fetchall()}
     selected = ["state"]
@@ -424,7 +389,7 @@ def _message_fts_record_sync(conn: sqlite3.Connection) -> dict[str, object] | No
 
 
 async def _message_fts_record_async(conn: aiosqlite.Connection) -> dict[str, object] | None:
-    if not await _table_exists_async(conn):
+    if not await table_exists_async(conn, FRESHNESS_TABLE):
         return None
     rows = await (await conn.execute(f"PRAGMA table_info({FRESHNESS_TABLE})")).fetchall()
     columns = {str(row[1]) for row in rows}
@@ -499,7 +464,7 @@ def message_fts_recorded_readiness_sync(conn: sqlite3.Connection) -> dict[str, i
     if record is None:
         return None
     state = str(record["state"])
-    exists = _named_table_exists_sync(conn, MESSAGE_SURFACE)
+    exists = table_exists(conn, MESSAGE_SURFACE)
     triggers_present = exists and _message_fts_triggers_present_sync(conn)
     if state == READY:
         if not triggers_present or not _recorded_ready_state_sync(conn, record):
@@ -528,7 +493,7 @@ async def message_fts_recorded_readiness_async(conn: aiosqlite.Connection) -> di
     if record is None:
         return None
     state = str(record["state"])
-    exists = await _named_table_exists_async(conn, MESSAGE_SURFACE)
+    exists = await table_exists_async(conn, MESSAGE_SURFACE)
     triggers_present = exists and await _message_fts_triggers_present_async(conn)
     if state == READY:
         if not triggers_present or not await _recorded_ready_state_async(conn, record):
diff --git a/polylogue/storage/fts/fts_lifecycle.py b/polylogue/storage/fts/fts_lifecycle.py
index 9db5f72350..b169623dc3 100644
--- a/polylogue/storage/fts/fts_lifecycle.py
+++ b/polylogue/storage/fts/fts_lifecycle.py
@@ -42,6 +42,8 @@
     trigram_delete_session_rows_sql,
     trigram_insert_session_rows_sql,
 )
+from polylogue.storage.introspection import table_exists as _table_exists_sync
+from polylogue.storage.introspection import table_exists_async as _table_exists_async
 
 _chunked = chunked
 IndexedMessageLike: TypeAlias = tuple[str, str, str | None] | IndexedMessage
@@ -290,15 +292,6 @@ def _fts_trigger_ddl_for_existing_surfaces_sync(conn: sqlite3.Connection) -> tup
     return tuple(ddl)
 
 
-async def _table_exists_async(conn: aiosqlite.Connection, table_name: str) -> bool:
-    cursor = await conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type='table' AND name = ? LIMIT 1",
-        (table_name,),
-    )
-    row = await cursor.fetchone()
-    return row is not None
-
-
 async def _fts_trigger_ddl_for_existing_surfaces_async(conn: aiosqlite.Connection) -> tuple[str, ...]:
     ddl: list[str] = []
     if await _table_exists_async(conn, "blocks") and await _table_exists_async(conn, "messages_fts"):
@@ -963,16 +956,6 @@ def check_fts_readiness(readiness: Mapping[str, object], repair_hint: str = MESS
     raise DatabaseError(f"Search index is incomplete. {repair_hint}")
 
 
-def _table_exists_sync(conn: sqlite3.Connection, table_name: str) -> bool:
-    return (
-        conn.execute(
-            "SELECT 1 FROM sqlite_master WHERE type='table' AND name = ?",
-            (table_name,),
-        ).fetchone()
-        is not None
-    )
-
-
 def _trigger_invariant_sync(
     conn: sqlite3.Connection,
     *,
diff --git a/polylogue/storage/insights/feedback/__init__.py b/polylogue/storage/insights/feedback/__init__.py
index 8346681637..9e95688b62 100644
--- a/polylogue/storage/insights/feedback/__init__.py
+++ b/polylogue/storage/insights/feedback/__init__.py
@@ -27,6 +27,7 @@
     now_utc,
     parse_correction_kind,
 )
+from polylogue.storage.introspection import table_exists_async
 from polylogue.storage.sqlite.archive_tiers.user_write import (
     ASSERTION_DEFAULT_AUTHOR_KIND,
     ASSERTION_DEFAULT_AUTHOR_REF,
@@ -43,11 +44,7 @@
 
 
 async def _attached_table_exists(conn: aiosqlite.Connection, schema_name: str, table_name: str) -> bool:
-    cursor = await conn.execute(
-        f"SELECT 1 FROM {schema_name}.sqlite_master WHERE type='table' AND name = ? LIMIT 1",
-        (table_name,),
-    )
-    return await cursor.fetchone() is not None
+    return await table_exists_async(conn, table_name, schema=schema_name)
 
 
 async def _attach_user_tier_if_present(conn: aiosqlite.Connection) -> bool:
diff --git a/polylogue/storage/raw_retention.py b/polylogue/storage/raw_retention.py
index 4e8f916e2a..6e680d976a 100644
--- a/polylogue/storage/raw_retention.py
+++ b/polylogue/storage/raw_retention.py
@@ -12,7 +12,8 @@
 
 from polylogue.logging import get_logger
 from polylogue.storage.blob_store import BlobStore, get_blob_store
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import column_exists as _column_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 logger = get_logger(__name__)
 
@@ -121,12 +122,6 @@ class _EligibleRawReceipt:
     predecessor_raw_id: str | None
 
 
-def _column_exists(conn: sqlite3.Connection, table: str, column: str) -> bool:
-    if not _table_exists(conn, table):
-        return False
-    return any(row[1] == column for row in conn.execute(f"PRAGMA table_info({table})").fetchall())
-
-
 def _blob_hash_text(value: object) -> str | None:
     if value is None:
         return None
diff --git a/polylogue/storage/session_replacement.py b/polylogue/storage/session_replacement.py
index f7a33d5424..610cea83b9 100644
--- a/polylogue/storage/session_replacement.py
+++ b/polylogue/storage/session_replacement.py
@@ -6,27 +6,11 @@
 
 import aiosqlite
 
+from polylogue.storage.introspection import table_exists as _table_exists_sync
+from polylogue.storage.introspection import table_exists_async as _table_exists_async
 from polylogue.storage.sqlite.sqlite_vec_extension import try_load_sqlite_vec_async
 
 
-def _table_exists_sync(conn: sqlite3.Connection, table_name: str) -> bool:
-    row = conn.execute(
-        "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ?",
-        (table_name,),
-    ).fetchone()
-    return row is not None
-
-
-async def _table_exists_async(conn: aiosqlite.Connection, table_name: str) -> bool:
-    row = await (
-        await conn.execute(
-            "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ?",
-            (table_name,),
-        )
-    ).fetchone()
-    return row is not None
-
-
 def _trigger_exists_sync(conn: sqlite3.Connection, trigger_name: str) -> bool:
     row = conn.execute(
         "SELECT 1 FROM sqlite_master WHERE type = 'trigger' AND name = ?",
diff --git a/polylogue/storage/sqlite/run_projection_relations.py b/polylogue/storage/sqlite/run_projection_relations.py
index df1884a2a9..b8a5d218f7 100644
--- a/polylogue/storage/sqlite/run_projection_relations.py
+++ b/polylogue/storage/sqlite/run_projection_relations.py
@@ -9,7 +9,6 @@
 from __future__ import annotations
 
 import json
-import sqlite3
 from typing import Protocol
 
 from polylogue.archive.query.predicate import QueryBoolPredicate, QueryFieldPredicate, QueryPredicate
@@ -515,12 +514,3 @@ def row_to_session_context_snapshot_record(row: RowLike) -> SessionContextSnapsh
         snapshot=context_snapshot_from_row(row),
         search_text=str(row["search_text"] or ""),
     )
-
-
-def table_exists_sync(conn: sqlite3.Connection, table_name: str) -> bool:
-    return bool(
-        conn.execute(
-            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1",
-            (table_name,),
-        ).fetchone()
-    )
diff --git a/polylogue/storage/usage.py b/polylogue/storage/usage.py
index 1c6dba92a0..14ae65da20 100644
--- a/polylogue/storage/usage.py
+++ b/polylogue/storage/usage.py
@@ -45,7 +45,7 @@
     sum_evidence_values,
 )
 from polylogue.core.refs import ObjectRef
-from polylogue.storage.table_existence import table_exists as _table_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 
 logger = logging.getLogger(__name__)
 
@@ -1517,14 +1517,13 @@ def _quote_identifier(name: str) -> str:
 
 
 def _table_exists_in_schema(conn: sqlite3.Connection, schema: str, name: str) -> bool:
+    # Thin wrapper (not a duplicate): callers probe candidate schema aliases
+    # that may not be ATTACHed yet, so unlike `introspection.table_exists`
+    # this swallows sqlite3.Error (e.g. "no such database: <alias>") as False.
     try:
-        row = conn.execute(
-            f"SELECT 1 FROM {_quote_identifier(schema)}.sqlite_master WHERE type = 'table' AND name = ?",
-            (name,),
-        ).fetchone()
+        return _table_exists(conn, name, schema=_quote_identifier(schema))
     except sqlite3.Error:
         return False
-    return row is not None
 
 
 def _base_session_stats(conn: sqlite3.Connection, origin: str | None) -> dict[str, dict[str, int]]:
diff --git a/tests/unit/cli/commands/test_status.py b/tests/unit/cli/commands/test_status.py
index f9763c006f..717b7613fa 100644
--- a/tests/unit/cli/commands/test_status.py
+++ b/tests/unit/cli/commands/test_status.py
@@ -23,13 +23,11 @@
     _archive_tier_status,
     _archive_unidentified_artifact_count,
     _candidate_daemon_urls,
-    _column_exists,
     _default_daemon_url,
     _direct_archive_counts,
     _fast_count,
     _fmt_bytes,
     _parse_cmdline_api_port,
-    _table_exists,
     _view_exists,
 )
 
@@ -45,6 +43,15 @@
     _archive_readiness_counts,
     _archive_status_surfaces,
 )
+
+# polylogue-48h: _table_exists/_column_exists moved from private status.py
+# definitions to the shared polylogue.storage.introspection module (which now
+# owns the canonical table_exists/column_exists implementation); status.py
+# only re-imports them, and mypy --strict's no-implicit-reexport rule
+# correctly rejects importing a bare re-import through status.py. Import the
+# canonical names directly, matching the _archive_readiness precedent above.
+from polylogue.storage.introspection import column_exists as _column_exists
+from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER
 from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
diff --git a/tests/unit/daemon/test_daemon_cli.py b/tests/unit/daemon/test_daemon_cli.py
index bef25d8350..bb5f6137e8 100644
--- a/tests/unit/daemon/test_daemon_cli.py
+++ b/tests/unit/daemon/test_daemon_cli.py
@@ -2174,7 +2174,7 @@ def execute(self, sql: str, _params: object = ()) -> FakeCursor:
                 return FakeCursor(("messages_fts",))
             if query == "SELECT name FROM sqlite_master WHERE type='table' AND name='messages'":
                 return FakeCursor(("messages",))
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(_params[0]) if isinstance(_params, tuple) and _params else ""
                 return FakeCursor((1,)) if name in {"messages", "messages_fts"} else FakeCursor(None)
             if query.startswith("SELECT name FROM sqlite_master WHERE type='trigger'"):
@@ -2268,7 +2268,7 @@ def execute(self, sql: str, _params: object = ()) -> FakeCursor:
                 return FakeCursor(("messages_fts",))
             if query == "SELECT name FROM sqlite_master WHERE type='table' AND name='messages'":
                 return FakeCursor(("messages",))
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(_params[0]) if isinstance(_params, tuple) and _params else ""
                 return FakeCursor((1,)) if name in {"messages", "messages_fts"} else FakeCursor(None)
             if query.startswith("SELECT name FROM sqlite_master WHERE type='trigger'"):
@@ -2337,7 +2337,7 @@ def fetchall(self) -> list[tuple[object, ...]]:
     class FakeConnection:
         def execute(self, sql: str, params: object = ()) -> FakeCursor:
             query = " ".join(sql.split())
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(params[0]) if isinstance(params, tuple) and params else ""
                 return (
                     FakeCursor((1,)) if name in {"blocks", "messages_fts", "messages_fts_docsize"} else FakeCursor(None)
@@ -2433,7 +2433,7 @@ def __init__(self) -> None:
         def execute(self, sql: str, params: object = ()) -> FakeCursor:
             query = " ".join(sql.split())
             self.queries.append(query)
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(params[0]) if isinstance(params, tuple) and params else ""
                 return (
                     FakeCursor((1,)) if name in {"blocks", "messages_fts", "messages_fts_docsize"} else FakeCursor(None)
@@ -2503,7 +2503,7 @@ def __init__(self) -> None:
         def execute(self, sql: str, params: object = ()) -> FakeCursor:
             query = " ".join(sql.split())
             self.queries.append(query)
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(params[0]) if isinstance(params, tuple) and params else ""
                 return (
                     FakeCursor((1,)) if name in {"blocks", "messages_fts", "messages_fts_docsize"} else FakeCursor(None)
@@ -2592,7 +2592,7 @@ def __init__(self) -> None:
         def execute(self, sql: str, params: object = ()) -> FakeCursor:
             query = " ".join(sql.split())
             self.queries.append(query)
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(params[0]) if isinstance(params, tuple) and params else ""
                 return (
                     FakeCursor((1,)) if name in {"blocks", "messages_fts", "messages_fts_docsize"} else FakeCursor(None)
@@ -2693,7 +2693,7 @@ def execute(self, sql: str, _params: object = ()) -> FakeCursor:
             self.queries.append(query)
             if query == "PRAGMA busy_timeout = 120000":
                 return FakeCursor(None)
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 return FakeCursor(None)
             raise AssertionError(f"unexpected query: {query}")
 
@@ -2718,7 +2718,7 @@ def close(self) -> None:
 
     asyncio.run(daemon_cli._ensure_fts_startup_readiness())
 
-    assert "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1" in conn.queries
+    assert "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1" in conn.queries
     assert conn.committed is False
     assert conn.closed is True
 
@@ -2775,7 +2775,7 @@ def execute(self, sql: str, params: object = ()) -> FakeCursor:
                         (9, "detail"),
                     ],
                 )
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(params[0]) if isinstance(params, tuple) and params else ""
                 return (
                     FakeCursor((1,)) if name in {"blocks", "messages_fts", "messages_fts_docsize"} else FakeCursor(None)
@@ -2901,7 +2901,7 @@ def execute(self, sql: str, _params: object = ()) -> FakeCursor:
                 return FakeCursor(("messages_fts",))
             if query == "SELECT name FROM sqlite_master WHERE type='table' AND name='messages'":
                 return FakeCursor(("messages",))
-            if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1":
+            if query == "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1":
                 name = str(_params[0]) if isinstance(_params, tuple) and _params else ""
                 return FakeCursor((1,)) if name in {"messages", "messages_fts"} else FakeCursor(None)
             if query.startswith("SELECT name FROM sqlite_master WHERE type='trigger'"):
diff --git a/tests/unit/storage/test_embedding_contracts.py b/tests/unit/storage/test_embedding_contracts.py
index c10f388e7d..1b9215578c 100644
--- a/tests/unit/storage/test_embedding_contracts.py
+++ b/tests/unit/storage/test_embedding_contracts.py
@@ -350,13 +350,12 @@ def __init__(self, database: str = ":memory:", *, timeout: float = 5.0, **kwargs
         self._query_count = 0
 
     def execute(self, sql: str, parameters: object = (), /) -> sqlite3.Cursor:
-        del parameters
         self._query_count += 1
         if "message_embeddings" in sql:
             raise sqlite3.OperationalError("no such module: vec0")
         if self._query_count >= 2:
             raise sqlite3.OperationalError("no such table: embedding_status")
-        return super().execute(sql)
+        return super().execute(sql, parameters)  # type: ignore[arg-type]
 
 
 # ---------------------------------------------------------------------------

From 1aca75075733b8dfe24c8f789a2b8c7fee6ceb65 Mon Sep 17 00:00:00 2001
From: Sinity <ezo.dev@gmail.com>
Date: Mon, 3 Aug 2026 22:58:56 +0200
Subject: [PATCH 4/4] chore(devtools): add table-exists-duplication lint +
 layering baseline updates

Ref polylogue-48h

Add devtools/verify_table_exists_duplication.py: a grep/AST-based tripwire
wired into `devtools verify --quick` (via the "lab policy" static-check
block) that forbids a new top-level def named table_exists/column_exists/
index_exists (or a _-prefixed/_sync/_async variant) outside
polylogue/storage/introspection.py -- the same pattern the bead's
consolidation removed. Registered as `devtools lab policy
table-exists-duplication` in command_catalog.py; docs/devtools.md
regenerated via `devtools render devtools-reference`.

docs/plans/layering-surface-baseline.json: renamed the 5 existing
table_existence baseline entries to introspection, then added 5 new entries
for modules (cli/commands/status.py, daemon/embedding_backlog.py,
daemon/fts_automerge.py, daemon/fts_startup.py, daemon/fts_status.py) that
previously defined their own _table_exists and so were never in the
baseline for this import -- now that they import the shared substrate
module, `devtools verify layering`'s cli/daemon-may-not-import-storage
ratchet correctly requires them to be declared.
---
 devtools/command_catalog.py                 |  16 +++
 devtools/verify.py                          |   9 ++
 devtools/verify_table_exists_duplication.py | 148 ++++++++++++++++++++
 docs/devtools.md                            |   1 +
 docs/plans/layering-surface-baseline.json   |  49 +++++--
 5 files changed, 211 insertions(+), 12 deletions(-)
 create mode 100644 devtools/verify_table_exists_duplication.py

diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py
index 9ea4bfb51e..7177f44944 100644
--- a/devtools/command_catalog.py
+++ b/devtools/command_catalog.py
@@ -1578,6 +1578,22 @@ def to_dict(self) -> dict[str, object]:
         ),
         examples=("devtools lab policy raw-payload-hash-purity", "devtools lab policy raw-payload-hash-purity --json"),
     ),
+    CommandSpec(
+        "lab policy table-exists-duplication",
+        "verification lab",
+        "Verify no module outside storage/introspection.py redefines table_exists/column_exists/index_exists.",
+        "devtools.verify_table_exists_duplication",
+        use_when=(
+            "Keep polylogue-48h's consolidation from silently regrowing: ~25 independently maintained "
+            "_table_exists/_column_exists/_index_exists copies (each trivially small and subtly different) "
+            "were merged into polylogue.storage.introspection. A grep-based tripwire forbidding a new "
+            "top-level def with one of the retired names outside that module."
+        ),
+        examples=(
+            "devtools lab policy table-exists-duplication",
+            "devtools lab policy table-exists-duplication --json",
+        ),
+    ),
     CommandSpec(
         "lab policy position-derived-identity",
         "verification lab",
diff --git a/devtools/verify.py b/devtools/verify.py
index 0c1270739f..dfae510394 100644
--- a/devtools/verify.py
+++ b/devtools/verify.py
@@ -1823,6 +1823,15 @@ def build_verify_steps(
                     "lab policy raw-authority-frontier-executability",
                     _devtools_cmd("lab policy raw-authority-frontier-executability"),
                 ),
+                # Static, archive-independent, sub-second: forbids a NEW
+                # top-level def named table_exists/column_exists/index_exists
+                # (or a _-prefixed/_sync/_async variant) outside
+                # polylogue/storage/introspection.py -- the ~25-copy
+                # duplication polylogue-48h consolidated into that module.
+                (
+                    "lab policy table-exists-duplication",
+                    _devtools_cmd("lab policy table-exists-duplication"),
+                ),
                 # Publication gate. Committed provider schema packages are
                 # public artifacts; this blocks local provenance
                 # (bundle_scopes/representative_paths) and scans for secrets.
diff --git a/devtools/verify_table_exists_duplication.py b/devtools/verify_table_exists_duplication.py
new file mode 100644
index 0000000000..8f417faeb2
--- /dev/null
+++ b/devtools/verify_table_exists_duplication.py
@@ -0,0 +1,148 @@
+"""Forbid a new duplicate SQLite existence-check helper outside the canonical module.
+
+Background
+----------
+
+polylogue-48h found ~25 independently maintained copies of
+``_table_exists``/``table_exists``/``_column_exists``/``_index_exists`` (and
+their async variants) scattered across ``cli/``, ``daemon/``, ``storage/``,
+``sources/``, ``insights/``, and ``operations/`` -- each trivially small and
+subtly different (a ``schema=`` kwarg on some, ``type IN (...)`` alternatives
+that never actually match anything in ``sqlite_master`` on others). They were
+consolidated into ``polylogue.storage.introspection`` (``table_exists``,
+``table_exists_async``, ``column_exists``, ``column_exists_async``,
+``index_exists``, ``index_exists_async``). This grep-based tripwire keeps the
+consolidation from silently regrowing: a module that wants a table/column/
+index existence check should import from ``polylogue.storage.introspection``,
+not redefine its own.
+
+What this lint checks
+----------------------
+
+Every ``polylogue/**/*.py`` file except ``polylogue/storage/introspection.py``
+itself is scanned line-by-line for a top-level (column 0) ``def``/``async def``
+whose name matches the forbidden shape:
+
+* ``_table_exists`` / ``table_exists`` (+ ``_sync``/``_async`` suffix variants)
+* ``_column_exists`` / ``column_exists`` (+ suffix variants)
+* ``_index_exists`` / ``index_exists`` (+ suffix variants)
+
+A thin, behaviorally-distinct wrapper that *delegates* to the canonical
+module (e.g. one that also swallows a specific ``sqlite3.OperationalError``,
+or checks an ATTACHed schema alias that may not exist yet) is not itself
+flagged by name matching alone -- this lint only catches the exact duplicate
+*names*, on the theory that a genuinely new name (``_attached_table_exists``,
+``_named_table_exists_sync``, ``_schema_object_exists``) signals a real design
+choice made under review, while reusing one of the exact retired names is the
+easy way to silently reintroduce the duplication this bead removed.
+
+Wired into ``devtools verify --quick`` (the static/generated-surface gate,
+alongside the other ``lab policy`` checks): archive-independent, sub-second.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from dataclasses import dataclass
+
+from devtools import repo_root as _get_root
+
+ROOT = _get_root()
+
+# The one place these names are allowed to be defined.
+CANONICAL_MODULE = "polylogue/storage/introspection.py"
+
+_FORBIDDEN_BASE_NAMES = ("table_exists", "column_exists", "index_exists")
+_SUFFIXES = ("", "_sync", "_async")
+
+_FORBIDDEN_NAMES = frozenset(
+    f"{prefix}{base}{suffix}" for prefix in ("", "_") for base in _FORBIDDEN_BASE_NAMES for suffix in _SUFFIXES
+)
+
+_DEF_PATTERN = re.compile(r"^(?:async\s+)?def\s+(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*\(")
+
+
+@dataclass(frozen=True, slots=True)
+class DuplicationViolation:
+    path: str
+    lineno: int
+    name: str
+
+
+def scan_source_for_duplicate_definitions(source: str, *, path: str) -> list[DuplicationViolation]:
+    """Return every forbidden-named top-level def in *source*.
+
+    Exposed standalone so a test can feed a synthetic source-string fixture
+    directly, mirroring ``verify_raw_payload_hash_purity.scan_source_for_payload_concatenation``.
+    """
+    violations: list[DuplicationViolation] = []
+    for lineno, line in enumerate(source.splitlines(), start=1):
+        match = _DEF_PATTERN.match(line)
+        if match is None:
+            continue
+        name = match.group("name")
+        if name in _FORBIDDEN_NAMES:
+            violations.append(DuplicationViolation(path=path, lineno=lineno, name=name))
+    return violations
+
+
+def _collect_violations() -> list[DuplicationViolation]:
+    violations: list[DuplicationViolation] = []
+    for full_path in sorted((ROOT / "polylogue").rglob("*.py")):
+        rel = full_path.relative_to(ROOT).as_posix()
+        if rel == CANONICAL_MODULE:
+            continue
+        source = full_path.read_text(encoding="utf-8")
+        violations.extend(scan_source_for_duplicate_definitions(source, path=rel))
+    return violations
+
+
+def _format_report(violations: list[DuplicationViolation]) -> str:
+    if not violations:
+        return (
+            "Table/column/index existence-check consolidation intact: no module outside "
+            f"{CANONICAL_MODULE} redefines table_exists/column_exists/index_exists (polylogue-48h)."
+        )
+    lines = [f"SQLite existence-check duplication violations: {len(violations)}", ""]
+    for violation in violations:
+        lines.append(f"  {violation.path}:{violation.lineno}: def {violation.name}(...)")
+    lines.append("")
+    lines.append(
+        "Policy violation (polylogue-48h): table/column/index existence checks are "
+        f"centralized in {CANONICAL_MODULE} (table_exists, table_exists_async, column_exists, "
+        "column_exists_async, index_exists, index_exists_async). Import from there instead of "
+        "redefining one of these names. If you genuinely need different error-handling or "
+        "schema-quoting behavior, write a differently-named thin wrapper that delegates to the "
+        "canonical function (see polylogue/storage/usage.py's _table_exists_in_schema for the pattern)."
+    )
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--json", action="store_true", help="emit machine-readable JSON")
+    args = parser.parse_args(argv)
+
+    violations = _collect_violations()
+
+    if args.json:
+        payload = {
+            "violations": [{"path": v.path, "lineno": v.lineno, "name": v.name} for v in violations],
+            "canonical_module": CANONICAL_MODULE,
+            "ok": not violations,
+        }
+        print(json.dumps(payload, indent=2))
+    else:
+        print(_format_report(violations))
+
+    return 0 if not violations else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/docs/devtools.md b/docs/devtools.md
index 39223338b9..2d223aab34 100644
--- a/docs/devtools.md
+++ b/docs/devtools.md
@@ -150,6 +150,7 @@ These are the commands worth remembering during normal repo work:
 | `devtools lab policy raw-authority-frontier-executability` | Verify every raw-authority frontier state has a reachable actuator. |
 | `devtools lab policy raw-payload-hash-purity` | Verify no raw-capture write path splices a synthesized literal onto captured bytes before hashing. |
 | `devtools lab policy schema-versioning` | Verify durable-tier migration and derived-tier rebuild boundaries. |
+| `devtools lab policy table-exists-duplication` | Verify no module outside storage/introspection.py redefines table_exists/column_exists/index_exists. |
 | `devtools lab policy timestamp-doctrine` | Verify durable-tier DDL never stores a timestamp column as TEXT. |
 | `devtools lab probe bead-pr-reconciliation` | Surface beads whose referenced PR merged but the bead is still open. |
 | `devtools lab probe capture-regression` | Capture pipeline-probe summaries as durable local regression cases. |
diff --git a/docs/plans/layering-surface-baseline.json b/docs/plans/layering-surface-baseline.json
index 33e4ebac52..7db1b91adc 100644
--- a/docs/plans/layering-surface-baseline.json
+++ b/docs/plans/layering-surface-baseline.json
@@ -469,6 +469,11 @@
     "file": "polylogue/cli/commands/status.py",
     "import": "polylogue.storage.embeddings.status_payload"
   },
+  {
+    "target": "polylogue/cli",
+    "file": "polylogue/cli/commands/status.py",
+    "import": "polylogue.storage.introspection"
+  },
   {
     "target": "polylogue/cli",
     "file": "polylogue/cli/commands/status.py",
@@ -502,7 +507,7 @@
   {
     "target": "polylogue/cli",
     "file": "polylogue/cli/commands/tutorial.py",
-    "import": "polylogue.storage.table_existence"
+    "import": "polylogue.storage.introspection"
   },
   {
     "target": "polylogue/cli",
@@ -537,7 +542,7 @@
   {
     "target": "polylogue/cli",
     "file": "polylogue/cli/read_views/streaming_markdown.py",
-    "import": "polylogue.storage.table_existence"
+    "import": "polylogue.storage.introspection"
   },
   {
     "target": "polylogue/cli",
@@ -889,6 +894,11 @@
     "file": "polylogue/daemon/convergence_stages.py",
     "import": "polylogue.storage.insights.session.status"
   },
+  {
+    "target": "polylogue/daemon",
+    "file": "polylogue/daemon/convergence_stages.py",
+    "import": "polylogue.storage.introspection"
+  },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/convergence_stages.py",
@@ -949,11 +959,6 @@
     "file": "polylogue/daemon/convergence_stages.py",
     "import": "polylogue.storage.sqlite.sqlite_vec_extension"
   },
-  {
-    "target": "polylogue/daemon",
-    "file": "polylogue/daemon/convergence_stages.py",
-    "import": "polylogue.storage.table_existence"
-  },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/convergence_standing_queries.py",
@@ -1024,6 +1029,11 @@
     "file": "polylogue/daemon/embedding_backlog.py",
     "import": "polylogue.storage.embeddings.reconcile"
   },
+  {
+    "target": "polylogue/daemon",
+    "file": "polylogue/daemon/embedding_backlog.py",
+    "import": "polylogue.storage.introspection"
+  },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/embedding_backlog.py",
@@ -1079,6 +1089,11 @@
     "file": "polylogue/daemon/events.py",
     "import": "polylogue.storage.sqlite.connection_profile"
   },
+  {
+    "target": "polylogue/daemon",
+    "file": "polylogue/daemon/fts_automerge.py",
+    "import": "polylogue.storage.introspection"
+  },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/fts_automerge.py",
@@ -1127,7 +1142,7 @@
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/fts_orphan_audit.py",
-    "import": "polylogue.storage.table_existence"
+    "import": "polylogue.storage.introspection"
   },
   {
     "target": "polylogue/daemon",
@@ -1154,6 +1169,11 @@
     "file": "polylogue/daemon/fts_startup.py",
     "import": "polylogue.storage.fts.fts_lifecycle"
   },
+  {
+    "target": "polylogue/daemon",
+    "file": "polylogue/daemon/fts_startup.py",
+    "import": "polylogue.storage.introspection"
+  },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/fts_startup.py",
@@ -1189,6 +1209,11 @@
     "file": "polylogue/daemon/fts_status.py",
     "import": "polylogue.storage.fts.sql"
   },
+  {
+    "target": "polylogue/daemon",
+    "file": "polylogue/daemon/fts_status.py",
+    "import": "polylogue.storage.introspection"
+  },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/fts_status.py",
@@ -1372,22 +1397,22 @@
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/metrics.py",
-    "import": "polylogue.storage.sqlite.archive_tiers.bootstrap"
+    "import": "polylogue.storage.introspection"
   },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/metrics.py",
-    "import": "polylogue.storage.sqlite.archive_tiers.ops_write"
+    "import": "polylogue.storage.sqlite.archive_tiers.bootstrap"
   },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/metrics.py",
-    "import": "polylogue.storage.sqlite.connection_profile"
+    "import": "polylogue.storage.sqlite.archive_tiers.ops_write"
   },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/metrics.py",
-    "import": "polylogue.storage.table_existence"
+    "import": "polylogue.storage.sqlite.connection_profile"
   },
   {
     "target": "polylogue/daemon",