refactor(storage): consolidate table/column/index exists checks into introspection

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",

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.

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())

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. |

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",

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()}
 

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

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)