#!/usr/bin/env python3
"""pgvector embedding feasibility spike — disposable read-only probe.

Answers the M1.5 prerequisite questions against the *live* Immich Postgres and
records them in a findings doc. See the design spec:
    docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md

Questions (pass/fail):
  1. Can we read the CLIP embeddings at all?
  2. Can each embedding join back to a photo we track (embedding key -> assets.id)?
  3. Exact shapes: table, embedding column, vector dimension, distance operator.
  4. Coverage over the *embeddable* (IMAGE) population — raw and image-only ratios.

This is read-only and idempotent. The DB session is opened read-only at the
session layer (not merely by which statements we issue), so even Immich's
write-capable postgres user cannot mutate the source-of-truth DB while we probe.

Usage:
    IMMICH_DB_URL=postgresql://user:pass@host:5432/immich \\
        python scripts/pgvector_spike.py

    # or rely on a .env (searched upward from CWD) that defines IMMICH_DB_URL
    python scripts/pgvector_spike.py [--sqlite PATH] [--findings PATH] [--no-write]
"""
from __future__ import annotations

import argparse
import os
import sqlite3
import sys
import textwrap
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

import psycopg
from psycopg import sql

# pgvector type candidates to probe first (verified against the catalog, not trusted).
KNOWN_TABLE = "smart_search"
KNOWN_COLUMN = "embedding"
# pgvector opclass -> (distance operator, human name).
OPCLASS_OPERATORS = {
    "vector_cosine_ops": ("<=>", "cosine"),
    "vector_l2_ops": ("<->", "L2 / euclidean"),
    "vector_ip_ops": ("<#>", "negative inner product"),
}
DEFAULT_FINDINGS = "docs/superpowers/specs/2026-06-27-pgvector-embedding-findings.md"


# --------------------------------------------------------------------------- env
def find_dsn() -> Optional[str]:
    """IMMICH_DB_URL from the environment, else from the nearest .env walking up."""
    dsn = (os.environ.get("IMMICH_DB_URL") or "").strip()
    if dsn:
        return dsn
    here = Path.cwd().resolve()
    for base in [here, *here.parents, Path(__file__).resolve().parent.parent]:
        env_file = base / ".env"
        if not env_file.is_file():
            continue
        for line in env_file.read_text().splitlines():
            line = line.strip()
            if line.startswith("#") or "=" not in line:
                continue
            key, _, val = line.partition("=")
            if key.strip() == "IMMICH_DB_URL":
                val = val.strip().strip('"').strip("'")
                if val:
                    return val
    return None


def connect_readonly(dsn: str) -> psycopg.Connection:
    """Open a connection and force the whole session read-only.

    Belt-and-suspenders: psycopg's ``read_only`` wraps each transaction READ ONLY,
    and ``default_transaction_read_only`` covers anything that slips outside it.
    """
    conn = psycopg.connect(dsn, autocommit=True)
    conn.execute("SET default_transaction_read_only = on")
    conn.read_only = True
    return conn


# ----------------------------------------------------------------------- discovery
def discover_vector_columns(conn: psycopg.Connection) -> list[tuple[str, str, str]]:
    """All (schema, table, column) holding a pgvector ``vector`` column.

    Catalog-driven so it survives Immich's cross-version schema drift. The known
    ``smart_search.embedding`` pair is sorted first when present.
    """
    rows = conn.execute(
        """
        SELECT n.nspname, c.relname, a.attname
        FROM pg_attribute a
        JOIN pg_class c ON c.oid = a.attrelid
        JOIN pg_namespace n ON n.oid = c.relnamespace
        JOIN pg_type t ON t.oid = a.atttypid
        WHERE t.typname = 'vector'
          AND a.attnum > 0 AND NOT a.attisdropped
          AND c.relkind IN ('r', 'p')          -- ordinary + partitioned tables
          AND n.nspname NOT IN ('pg_catalog', 'information_schema')
        ORDER BY (c.relname = %s AND a.attname = %s) DESC, n.nspname, c.relname
        """,
        (KNOWN_TABLE, KNOWN_COLUMN),
    ).fetchall()
    return [(r[0], r[1], r[2]) for r in rows]


def vector_dimension(conn, schema: str, table: str, column: str) -> tuple[Optional[int], str]:
    """Vector dimension, preferring a live sample over the declared typmod."""
    # atttypmod holds the declared dimension for pgvector (>0 when fixed).
    typmod = conn.execute(
        """
        SELECT a.atttypmod
        FROM pg_attribute a
        JOIN pg_class c ON c.oid = a.attrelid
        JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE n.nspname = %s AND c.relname = %s AND a.attname = %s
        """,
        (schema, table, column),
    ).fetchone()
    declared = typmod[0] if typmod and typmod[0] and typmod[0] > 0 else None

    sampled = None
    try:
        ident = sql.Identifier(schema, table)
        col = sql.Identifier(column)
        row = conn.execute(
            sql.SQL("SELECT vector_dims({col}) FROM {tbl} WHERE {col} IS NOT NULL LIMIT 1")
            .format(col=col, tbl=ident)
        ).fetchone()
        sampled = row[0] if row else None
    except Exception as exc:  # noqa: BLE001 — record, don't abort the probe
        return declared, f"declared typmod={declared}; live sample failed: {exc}"

    if sampled is not None:
        note = f"live sample={sampled}" + (f", declared typmod={declared}" if declared else "")
        return sampled, note
    return declared, f"declared typmod={declared}; table empty (no live vector to sample)"


def distance_operator(conn, schema: str, table: str, column: str) -> tuple[str, str]:
    """The distance operator Immich's index was built for, read from its opclass."""
    row = conn.execute(
        """
        SELECT opc.opcname
        FROM pg_index i
        JOIN pg_class ic ON ic.oid = i.indexrelid
        JOIN pg_class tc ON tc.oid = i.indrelid
        JOIN pg_namespace n ON n.oid = tc.relnamespace
        JOIN pg_opclass opc ON opc.oid = ANY (i.indclass)
        JOIN pg_attribute at ON at.attrelid = i.indrelid AND at.attnum = ANY (i.indkey)
        WHERE n.nspname = %s AND tc.relname = %s AND at.attname = %s
          AND opc.opcname LIKE 'vector_%%'
        LIMIT 1
        """,
        (schema, table, column),
    ).fetchone()
    if row and row[0] in OPCLASS_OPERATORS:
        op, name = OPCLASS_OPERATORS[row[0]]
        return op, f"{op} ({name}) — from index opclass {row[0]}"
    return "<=>", "<=> (cosine) — assumed; no vector index opclass found on this column"


def self_similarity_ok(conn, schema, table, column, key_col, op) -> str:
    """Sanity-check the operator: a vector must be its own nearest neighbour (dist ~0).

    Param-free (seed picked inside a CTE) so it works regardless of how psycopg
    binds vector literals.
    """
    try:
        tbl, col, key = sql.Identifier(schema, table), sql.Identifier(column), sql.Identifier(key_col)
        rows = conn.execute(
            sql.SQL(
                "WITH seed AS (SELECT {key} AS k, {col} AS v FROM {tbl} "
                "  WHERE {col} IS NOT NULL LIMIT 1) "
                "SELECT (e.{key} = seed.k) AS is_seed, (e.{col} {op} seed.v) AS dist, "
                "       (seed.v {op} seed.v) AS self_dist "
                "FROM {tbl} e, seed WHERE e.{col} IS NOT NULL ORDER BY dist ASC LIMIT 5"
            ).format(key=key, col=col, tbl=tbl, op=sql.SQL(op))
        ).fetchall()
        if not rows:
            return "skipped — no embeddings present"
        is_seed, top_dist, self_dist = rows[0]
        # Operator works iff an identical vector sorts first at the operator's own
        # self-distance. That value is operator-dependent (0 for cosine <=> / L2 <->,
        # ~-1 for inner product <#>), so compare the nearest distance to the seed's
        # self-distance rather than to a hardcoded 0. The nearest row may be a duplicate
        # photo rather than the seed itself — still a pass.
        good = abs(float(top_dist) - float(self_dist)) < 1e-6
        who = "the seed itself" if is_seed else "a duplicate (identical-vector) asset"
        return (
            f"{'OK' if good else 'UNEXPECTED'} — nearest neighbour ({who}) "
            f"at distance {float(top_dist):.6g} (seed self-distance {float(self_dist):.6g})"
        )
    except Exception as exc:  # noqa: BLE001
        return f"failed: {exc}"


def detect_asset_key(conn, schema, table) -> tuple[Optional[str], Optional[str], Optional[str], str]:
    """Find the column + table joining the embedding row to the asset record.

    Returns ``(key_col, asset_table, asset_pk, note)``. The asset table name is
    *discovered from the FK*, never assumed — Immich renamed ``assets`` -> ``asset``
    across versions, exactly the drift the spec warns about.
    """
    fks = conn.execute(
        """
        SELECT kcu.column_name, ccu.table_name, ccu.column_name
        FROM information_schema.table_constraints tc
        JOIN information_schema.key_column_usage kcu
          ON tc.constraint_name = kcu.constraint_name AND tc.table_schema = kcu.table_schema
        JOIN information_schema.constraint_column_usage ccu
          ON ccu.constraint_name = tc.constraint_name AND ccu.table_schema = tc.table_schema
        WHERE tc.constraint_type = 'FOREIGN KEY'
          AND tc.table_schema = %s AND tc.table_name = %s
        """,
        (schema, table),
    ).fetchall()
    # Prefer the FK whose target table is the asset record (singular/plural tolerated).
    for col, ftable, fcol in fks:
        if ftable in ("asset", "assets"):
            return col, ftable, fcol, f"FK {table}.{col} -> {ftable}.{fcol}"

    cols = {
        r[0]
        for r in conn.execute(
            "SELECT column_name FROM information_schema.columns "
            "WHERE table_schema = %s AND table_name = %s",
            (schema, table),
        ).fetchall()
    }
    existing_tables = {
        r[0]
        for r in conn.execute(
            "SELECT table_name FROM information_schema.tables "
            "WHERE table_schema = %s AND table_name IN ('asset', 'assets')",
            (schema,),
        ).fetchall()
    }
    asset_table = "asset" if "asset" in existing_tables else ("assets" if "assets" in existing_tables else None)
    for cand in ("assetId", "asset_id", "assetsId"):
        if cand in cols and asset_table:
            return cand, asset_table, "id", f"no FK; matched column {table}.{cand} -> {asset_table}.id (assumed)"
    return None, None, None, "no asset-key column found (FK or assetId/asset_id)"


def join_and_coverage(conn, schema, table, key_col, asset_table, assets_pk):
    """Join validity + coverage (raw and image-only) inside Postgres.

    Returns a dict (the contract its three callers read by key):
      n_embeddings_distinct, n_joined_to_assets, n_orphan, n_assets_total: int
      n_assets_image, n_embedded_image: int | None   (None when no asset 'type' column)
      raw_ratio, image_ratio: float | None            (None when the denominator is 0/None)
    """
    tbl, key = sql.Identifier(schema, table), sql.Identifier(key_col)
    atbl, apk = sql.Identifier(schema, asset_table), sql.Identifier(assets_pk)

    n_emb = conn.execute(
        sql.SQL("SELECT count(DISTINCT {key}) FROM {tbl} WHERE {key} IS NOT NULL")
        .format(key=key, tbl=tbl)
    ).fetchone()[0]
    n_joined = conn.execute(
        sql.SQL(
            "SELECT count(DISTINCT e.{key}) FROM {tbl} e "
            "JOIN {atbl} a ON a.{apk} = e.{key}"
        ).format(key=key, tbl=tbl, atbl=atbl, apk=apk)
    ).fetchone()[0]
    n_assets_total = conn.execute(
        sql.SQL("SELECT count(*) FROM {atbl}").format(atbl=atbl)
    ).fetchone()[0]

    # Image-only population — Immich never embeds video/audio/other with CLIP.
    has_type = bool(
        conn.execute(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = %s AND table_name = %s AND column_name = 'type' LIMIT 1",
            (schema, asset_table),
        ).fetchone()
    )
    n_assets_image = n_emb_image = None
    if has_type:
        n_assets_image = conn.execute(
            sql.SQL("SELECT count(*) FROM {atbl} WHERE type = 'IMAGE'").format(atbl=atbl)
        ).fetchone()[0]
        n_emb_image = conn.execute(
            sql.SQL(
                "SELECT count(DISTINCT a.{apk}) FROM {atbl} a "
                "JOIN {tbl} e ON e.{key} = a.{apk} WHERE a.type = 'IMAGE'"
            ).format(apk=apk, atbl=atbl, tbl=tbl, key=key)
        ).fetchone()[0]

    return {
        "n_embeddings_distinct": n_emb,
        "n_joined_to_assets": n_joined,
        "n_orphan": n_emb - n_joined,
        "n_assets_total": n_assets_total,
        "n_assets_image": n_assets_image,
        "n_embedded_image": n_emb_image,
        "raw_ratio": (n_joined / n_assets_total) if n_assets_total else None,
        "image_ratio": (n_emb_image / n_assets_image)
        if (n_assets_image not in (None, 0)) else None,
    }


def sqlite_crosscheck(conn, schema, table, key_col, sqlite_path: Path) -> str:
    """Optional confirmation that embedding asset IDs match our SQLite store.

    Degrades gracefully: a fresh checkout has no `ingest` run, so an empty/absent
    store is not a failure of the join requirement.

    NOTE: this opens SQLite directly, a deliberate exception for this throwaway probe.
    Production M1.5 code must NOT — it must go through shared/photoflow/core/store.py,
    the only SQLite owner (CLAUDE.md). This raw read-only sample check stays in the spike.
    """
    if not sqlite_path.is_file():
        return f"skipped — no SQLite store at {sqlite_path} (fresh checkout; not a failure)"
    try:
        sample = [
            str(r[0])
            for r in conn.execute(
                sql.SQL("SELECT {key} FROM {tbl} WHERE {key} IS NOT NULL LIMIT 50")
                .format(key=sql.Identifier(key_col), tbl=sql.Identifier(schema, table))
            ).fetchall()
        ]
        if not sample:
            return "skipped — no embeddings to sample"
        sconn = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True)
        try:
            tbl_exists = sconn.execute(
                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='assets'"
            ).fetchone()
            if not tbl_exists:
                return "skipped — SQLite store has no assets table (unpopulated)"
            total = sconn.execute("SELECT count(*) FROM assets").fetchone()[0]
            if total == 0:
                return "skipped — SQLite assets table empty (no ingest run; not a failure)"
            placeholders = ",".join("?" * len(sample))
            hits = sconn.execute(
                f"SELECT count(*) FROM assets WHERE immich_id IN ({placeholders})", sample
            ).fetchone()[0]
        finally:
            sconn.close()
        return f"{hits}/{len(sample)} sampled embedding asset IDs found in SQLite (store has {total} assets)"
    except Exception as exc:  # noqa: BLE001
        return f"failed (non-fatal): {exc}"


def environment_meta(conn) -> dict:
    """Best-effort version context. Immich's *server* version is not in the DB."""
    out = {}
    try:
        out["postgres_version"] = conn.execute("SHOW server_version").fetchone()[0]
    except Exception as exc:  # noqa: BLE001
        out["postgres_version"] = f"unknown: {exc}"
    try:
        row = conn.execute(
            "SELECT extversion FROM pg_extension WHERE extname IN ('vector', 'vectors') LIMIT 1"
        ).fetchone()
        out["pgvector_extension"] = row[0] if row else "not found"
    except Exception as exc:  # noqa: BLE001
        out["pgvector_extension"] = f"unknown: {exc}"
    # Newest applied migration as a proxy for schema age (table name varies by version).
    out["latest_migration"] = "unknown"
    for mig_table in ("migrations", "kysely_migration", "typeorm_metadata"):
        try:
            row = conn.execute(
                sql.SQL("SELECT name FROM {} ORDER BY 1 DESC LIMIT 1")
                .format(sql.Identifier(mig_table))
            ).fetchone()
            if row:
                out["latest_migration"] = f"{mig_table}: {row[0]}"
                break
        except Exception:  # noqa: BLE001 — table absent in this version
            continue
    return out


# -------------------------------------------------------------------------- output
def _pct(ratio: Optional[float]) -> str:
    """Format a coverage ratio as a percentage, or 'n/a' when it could not be computed."""
    return "n/a" if ratio is None else f"{ratio * 100:.1f}%"


def render_findings(facts: dict) -> str:
    cov = facts["coverage"]
    raw_pct = _pct(cov["raw_ratio"])
    img_pct = _pct(cov["image_ratio"])
    meta = facts["meta"]
    return textwrap.dedent(
        f"""\
        # pgvector embedding feasibility spike — findings

        **Generated:** {facts['generated']} by `scripts/pgvector_spike.py`
        **Status:** machine-probed against the live Immich Postgres.
        **This is M1.5's contract.** See the design spec:
        `docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md`.

        > ⚠️ **Immich's undocumented, internal schema — no deprecation contract.**
        > Everything below is valid **only** for the model + Immich version observed and
        > can be renamed/restructured on any Immich upgrade. M1.5 must re-run this probe
        > (or version-guard) on every Immich upgrade. The "contract" is version-pinned,
        > not durable.

        ## Requirement 0 — does M1.5 need raw vectors (vs. a REST query)?

        M1.5 clusters photos by *visual similarity* at trip level — it needs either the
        raw CLIP embedding vectors or an arbitrary asset→asset nearest-neighbour query.
        Immich's REST surface provides neither:

        - `POST /api/search/smart` — **text→image** CLIP search: takes a text query,
          returns assets. It never returns embedding vectors and cannot do asset→asset
          similarity without a text prompt. Insufficient.
        - `POST /api/search/metadata`, `/api/search/random` — metadata/random only; no
          embeddings, no similarity.
        - Duplicate detection (`/api/duplicates`) consumes embeddings *internally* but only
          surfaces near-duplicate groups above Immich's own threshold — not a tunable
          pairwise similarity usable for trip-level clustering. Insufficient.
        - No documented endpoint returns raw CLIP vectors or arbitrary k-NN neighbours.

        **Conclusion:** the load-bearing "REST can't expose embeddings" premise holds for
        Immich's documented API → **read-only Postgres access (below) is the viable path.**
        (Re-confirm against the OpenAPI of the running version on upgrade.)

        ## Requirements 1–4 — probed facts

        | # | Question | Answer |
        |---|----------|--------|
        | 1 | Embeddings readable? | **{facts['readable']}** |
        | 2 | Join to asset table? | **{facts['join_summary']}** |
        | 3 | Table / column | `{facts['schema']}.{facts['table']}` / `{facts['column']}` |
        | 3 | Vector dimension | **{facts['dimension']}** ({facts['dimension_note']}) |
        | 3 | Distance operator | **{facts['operator']}** |
        | 3 | Operator sanity check | {facts['self_similarity']} |
        | 4 | Coverage (raw) | **{raw_pct}** ({cov['n_joined_to_assets']}/{cov['n_assets_total']} assets) |
        | 4 | Coverage (image-only) | **{img_pct}** ({cov['n_embedded_image']}/{cov['n_assets_image']} IMAGE assets) |

        - **Asset-key join:** {facts['asset_key_note']}
        - **Orphan embeddings** (no matching asset): {cov['n_orphan']}
        - **SQLite cross-check (optional):** {facts['sqlite']}

        ## The join M1.5 relies on

        `{facts['table']}.{facts['asset_key']}` → `{facts['asset_table']}.{facts['assets_pk']}` →
        SQLite `assets.immich_id` (`shared/photoflow/core`). The Immich asset UUID is the
        same key our store uses, so embeddings index straight onto tracked photos.

        > ⚠️ Immich's asset table is **`{facts['asset_table']}`** in the observed version
        > (it was `assets` in older versions). The probe discovers this from the FK; M1.5's
        > reader must not hardcode the name.

        ## Environment observed

        - **Postgres:** {meta['postgres_version']}
        - **pgvector extension:** {meta['pgvector_extension']}
        - **Schema age proxy (latest migration):** {meta['latest_migration']}
        - **Immich server version:** _record from the Immich UI / `GET /api/server/version`_
          — not reliably available in the DB.
        - **CLIP model:** _record the model from Immich's Machine-Learning settings_ — the
          user is re-running CLIP with a stronger model, so dimension + coverage above are a
          snapshot of whichever model was live at probe time.

        ## Read-only access (recommended hardening for M1.5)

        The probe enforces read-only at the session layer
        (`SET default_transaction_read_only = on` + psycopg `read_only`), which neutralises
        write capability even with Immich's write/DDL-capable `postgres` user. For M1.5,
        provision a dedicated least-privilege role instead:

        ```sql
        CREATE ROLE photoflow_ro LOGIN PASSWORD '...';
        GRANT CONNECT ON DATABASE immich TO photoflow_ro;
        GRANT USAGE ON SCHEMA {facts['schema']} TO photoflow_ro;
        GRANT SELECT ON {facts['schema']}.{facts['table']}, {facts['schema']}.{facts['asset_table']} TO photoflow_ro;
        ```
        """
    )


def print_report(facts: dict) -> None:
    cov = facts["coverage"]
    print("\n=== pgvector spike — report ===")
    print(f"  table/column     : {facts['schema']}.{facts['table']}.{facts['column']}")
    print(f"  dimension        : {facts['dimension']}  ({facts['dimension_note']})")
    print(f"  distance operator: {facts['operator']}")
    print(f"  operator sanity  : {facts['self_similarity']}")
    print(f"  asset key        : {facts['asset_key_note']}")
    print(f"  join             : {facts['join_summary']}  (orphans: {cov['n_orphan']})")
    print(f"  coverage (raw)   : {_pct(cov['raw_ratio'])}  ({cov['n_joined_to_assets']}/{cov['n_assets_total']})")
    print(f"  coverage (image) : {_pct(cov['image_ratio'])}  ({cov['n_embedded_image']}/{cov['n_assets_image']})")
    print(f"  sqlite x-check   : {facts['sqlite']}")
    print(f"  postgres         : {facts['meta']['postgres_version']}")
    print(f"  pgvector ext     : {facts['meta']['pgvector_extension']}")
    print("===============================\n")


# ---------------------------------------------------------------------------- main
def main() -> int:
    ap = argparse.ArgumentParser(description="Read-only pgvector embedding feasibility probe.")
    ap.add_argument("--sqlite", default="data/trip-cluster.db",
                    help="SQLite store path for the optional cross-check (default: data/trip-cluster.db)")
    ap.add_argument("--findings", default=DEFAULT_FINDINGS, help="findings doc path to write")
    ap.add_argument("--no-write", action="store_true", help="print the report but do not write the findings doc")
    args = ap.parse_args()

    dsn = find_dsn()
    if not dsn:
        print("ERROR: IMMICH_DB_URL not set (env or a .env up-tree). See .env.example.", file=sys.stderr)
        return 2

    try:
        conn = connect_readonly(dsn)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: could not connect read-only: {exc}", file=sys.stderr)
        return 2

    with conn:
        vec_cols = discover_vector_columns(conn)
        if not vec_cols:
            print("FAIL (Req 1): no pgvector `vector` column found in this database.", file=sys.stderr)
            print("  Immich may not have run CLIP yet, or the schema changed beyond the probe's discovery.",
                  file=sys.stderr)
            return 1
        schema, table, column = vec_cols[0]
        if len(vec_cols) > 1:
            others = ", ".join(f"{s}.{t}.{c}" for s, t, c in vec_cols[1:])
            print(f"NOTE: multiple vector columns found; probing {schema}.{table}.{column}. Others: {others}")

        dim, dim_note = vector_dimension(conn, schema, table, column)
        op, op_desc = distance_operator(conn, schema, table, column)
        asset_key, asset_table, assets_pk, asset_key_note = detect_asset_key(conn, schema, table)
        if not asset_key:
            print(f"FAIL (Req 2): {asset_key_note}", file=sys.stderr)
            return 1

        self_sim = self_similarity_ok(conn, schema, table, column, asset_key, op)
        cov = join_and_coverage(conn, schema, table, asset_key, asset_table, assets_pk)
        sqlite_note = sqlite_crosscheck(conn, schema, table, asset_key, Path(args.sqlite))
        meta = environment_meta(conn)

        join_summary = (
            f"{cov['n_joined_to_assets']}/{cov['n_embeddings_distinct']} embeddings join to "
            f"{asset_table} via {asset_key}"
        )
        facts = {
            "generated": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"),
            "schema": schema, "table": table, "column": column,
            "readable": f"yes — read {cov['n_embeddings_distinct']} distinct embedding(s)",
            "dimension": dim if dim is not None else "unknown",
            "dimension_note": dim_note,
            "operator": op_desc,
            "self_similarity": self_sim,
            "asset_key": asset_key, "asset_table": asset_table, "assets_pk": assets_pk,
            "asset_key_note": asset_key_note,
            "join_summary": join_summary,
            "coverage": cov, "sqlite": sqlite_note, "meta": meta,
        }

    print_report(facts)

    join_clean = cov["n_orphan"] == 0 and cov["n_joined_to_assets"] > 0
    if not args.no_write:
        out = Path(args.findings)
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_text(render_findings(facts))
        print(f"Wrote findings -> {out}")

    if cov["n_embeddings_distinct"] == 0:
        print("WARN: zero embeddings — readable+well-shaped but coverage 0 (CLIP re-run unfinished?).")
    if not join_clean:
        print(f"WARN (Req 2): {cov['n_orphan']} orphan embedding(s) did not join to assets.")
    return 0


if __name__ == "__main__":
    sys.exit(main())