spike.py

#!/usr/bin/env python3
"""
EdgeTable -- Phase 0 spike.

The plan (notes/plan.md) lists six risks and says: kill them all before any UI work
starts. This script is that proof. It builds the real schema, runs the real
enrichment, searches it semantically and syncs it to a second database file, then
prints a verdict against each risk.

It is not a demo. Nothing here is narrated for an audience -- it exists so that
Phase 1 starts from measured facts instead of hopes. It is still worth running
after an extension version bump: it is the only thing that scores category
accuracy, and a model or prompt change can quietly wreck that.

    make spike
    make spike SPIKE_ARGS="--rows 10"    # shorter enrichment pass
    make spike SPIKE_ARGS="--cloud"      # add a live SQLite Cloud round trip

Run from anywhere; paths resolve relative to the repository root.
"""

from __future__ import annotations

import argparse
import os
import sqlite3
import statistics
import sys
import time
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
VENDOR = REPO / "vendor"

# The same paths the Rust engine uses, and the same environment overrides, so the
# spike and the app can never disagree about which build they measured.
# `make prereqs` fills vendor/.
AI_EXT = Path(os.environ.get("EDGETABLE_AI_EXT", VENDOR / "extensions/ai"))
VECTOR_EXT = Path(os.environ.get("EDGETABLE_VECTOR_EXT", VENDOR / "extensions/vector"))
SYNC_EXT = Path(os.environ.get("EDGETABLE_SYNC_EXT", VENDOR / "extensions/cloudsync"))

# Qwen3-1.7B, not 0.6B. The smaller model is no faster in practice (it writes longer
# summaries, and tokens dominate) and cannot follow the ordered category rules below:
# 92% vs 58% category accuracy on the seed set. See notes/FINDINGS.md #4.
GEN_MODEL = Path(os.environ.get(
    "EDGETABLE_GEN_MODEL", VENDOR / "models/Qwen3-1.7B-Q4_K_M.gguf"))
EMBED_MODEL = Path(os.environ.get(
    "EDGETABLE_EMBED_MODEL", VENDOR / "models/all-MiniLM-L6-v2-q8_0.gguf"))

SEED = Path(__file__).parent / "seed.sql"
DB_A = REPO / "vendor/spike-a.db"
DB_B = REPO / "vendor/spike-b.db"

SENTIMENTS = ("positive", "neutral", "negative")
CATEGORIES = ("billing", "bugs", "feature_request", "ux", "performance", "other")

# Hand-labelled expected category for each row of seed.sql, in file order. A set means
# more than one label is genuinely defensible and any of them counts as correct.
#
# The grammar guarantees the label is IN the enum; it says nothing about it being the
# RIGHT one, and these pills are on screen during the demo. This is the regression
# test for prompt and model changes -- change either, re-run, compare the percentage.
EXPECTED_CATEGORY = [
    {"billing"}, {"ux", "other"}, {"performance"}, {"feature_request"}, {"billing"},
    {"ux", "other"}, {"bugs"}, {"other", "ux"}, {"billing"}, {"performance"},
    {"bugs"}, {"feature_request"}, {"ux"}, {"billing"}, {"bugs"},
    {"ux", "other"}, {"ux", "performance"}, {"billing"}, {"feature_request"},
    {"other", "ux"}, {"billing"}, {"bugs"}, {"ux", "other"}, {"performance"},
]

# One generation produces all three AI fields. The plan called for three separate
# llm_text_generate() calls; that is three times the latency for no extra quality,
# and it cannot guarantee the labels agree with the summary. See notes/FINDINGS.md.
GRAMMAR = """
root      ::= "{" ws "\\"summary\\":" ws summary "," ws "\\"sentiment\\":" ws sentiment "," ws "\\"category\\":" ws category ws "}"
summary   ::= "\\"" char{10,110} "\\""
sentiment ::= "\\"positive\\"" | "\\"neutral\\"" | "\\"negative\\""
category  ::= "\\"billing\\"" | "\\"bugs\\"" | "\\"feature_request\\"" | "\\"ux\\"" | "\\"performance\\"" | "\\"other\\""
char      ::= [^"\\\\\\x00-\\x1F] | "\\\\" ["\\\\/bfnrt]
ws        ::= [ \\t\\n]*
"""

# llm_text_generate() applies no chat template, so the prompt carries Qwen's own
# ChatML markers. Under the grammar the model cannot emit a <think> block either,
# which is what makes a reasoning-first model like Qwen3 usable for this at all.
PROMPT_SQL = """
  '<|im_start|>system' || char(10) ||
  'You label incoming customer feedback for a SaaS product. Reply with one JSON object.' || char(10) ||
  'summary: one plain sentence under 15 words, restating only what the message says.' || char(10) ||
  'Do not invent numbers, causes or details that are not in the message.' || char(10) ||
  'sentiment: positive | neutral | negative' || char(10) ||
  'category: what the message is ABOUT. Test the rules in order and stop at the first match:' || char(10) ||
  '  1. billing - money or plans: a charge, refund, invoice, price, seat count, upgrade or downgrade.' || char(10) ||
  '  2. feature_request - the user asks for something that does not exist yet, however small.' || char(10) ||
  '  3. bugs - something is broken: it crashes, errors, fails silently, or does the wrong thing.' || char(10) ||
  '  4. performance - it works correctly but is too slow.' || char(10) ||
  '  5. ux - how the existing product looks or feels to use, including praise for it.' || char(10) ||
  '  6. other - anything else, such as documentation or general praise.' || char(10) ||
  'Asking for a dark theme or a shortcut is feature_request, not ux.' || char(10) ||
  'Logging in to the wrong place is bugs, not ux.<|im_end|>' || char(10) ||
  '<|im_start|>user' || char(10) || NEW.body || '<|im_end|>' || char(10) ||
  '<|im_start|>assistant' || char(10)
"""


# Every NOT NULL column carries a DEFAULT because cloudsync_init() refuses the table
# otherwise ("All non-primary key columns declared as NOT NULL must have a DEFAULT
# value"). CRDT merges arrive column by column, so a NOT NULL column with no default
# has nothing to hold the row together in between. The plan's schema omitted these.
SCHEMA = """
CREATE TABLE IF NOT EXISTS feedback (
  id          TEXT PRIMARY KEY,
  author      TEXT NOT NULL DEFAULT '',
  channel     TEXT NOT NULL DEFAULT 'email',
  body        TEXT NOT NULL DEFAULT '',
  created_at  TEXT NOT NULL DEFAULT '',
  summary     TEXT,
  sentiment   TEXT,
  category    TEXT,
  ai_json     TEXT,
  embedding   BLOB
);
"""

# The guard is load-bearing: rows arriving from another device already carry their
# AI values, and re-running the model on them would be both wasteful and wrong.
TRIGGER = f"""
CREATE TRIGGER feedback_ai AFTER INSERT ON feedback
WHEN NEW.summary IS NULL
BEGIN
  UPDATE feedback SET ai_json = llm_text_generate({PROMPT_SQL}, 'n_predict=140')
   WHERE id = NEW.id;
  UPDATE feedback
     SET summary   = json_extract(ai_json, '$.summary'),
         sentiment = json_extract(ai_json, '$.sentiment'),
         category  = json_extract(ai_json, '$.category')
   WHERE id = NEW.id;
END;
"""


# ---------------------------------------------------------------- infrastructure

def ext(stem: Path) -> str:
    for suffix in (".dylib", ".so", ".dll"):
        if stem.with_suffix(suffix).exists():
            return str(stem.with_suffix(suffix))
    raise SystemExit(f"ERROR: no built extension at {stem}.[dylib|so|dll]")


# Teardown order is not a detail here -- see shutdown().
OPEN: list[sqlite3.Connection] = []


def connect(db: Path) -> sqlite3.Connection:
    """A connection with all three extensions and its own private model slot."""
    conn = sqlite3.connect(db, isolation_level=None)
    conn.enable_load_extension(True)
    for stem in (AI_EXT, VECTOR_EXT, SYNC_EXT):
        conn.load_extension(ext(stem))
    conn.enable_load_extension(False)
    conn.execute("PRAGMA journal_mode = WAL")     # so the reader can work while the writer generates
    conn.execute("PRAGMA trusted_schema = ON")    # or the trigger may not call the model
    OPEN.append(conn)
    return conn


def one(conn: sqlite3.Connection, sql: str, *params):
    row = conn.execute(sql, params).fetchone()
    return row[0] if row else None


def shutdown() -> None:
    """Close every connection in the order they were opened -- oldest first.

    This is load-bearing, not tidiness. sqlite3_ai_init() calls
    llama_log_set(ai_logger, ctx) (src/sqlite-ai.c:3806), and llama.cpp's log
    callback is process-global while `ctx` belongs to one connection. Whichever
    connection loaded the extension LAST owns the callback for the whole process.
    Close that one while others are still using llama and the next log line --
    llama_context's destructor emits one -- runs through a freed pointer and
    segfaults.

    So the youngest connection must be closed last. See notes/FINDINGS.md #1.
    """
    for conn in OPEN:
        try:
            conn.execute("SELECT llm_context_free()")
            conn.execute("SELECT llm_model_free()")
            conn.execute("SELECT cloudsync_terminate()")
        except sqlite3.Error:
            pass
        conn.close()
    OPEN.clear()


def head(n: int, title: str) -> None:
    print(f"\n\033[1m{n}. {title}\033[0m")
    print("─" * 78)


VERDICTS: list[tuple[str, str, str]] = []


def verdict(risk: str, status: str, detail: str) -> None:
    VERDICTS.append((risk, status, detail))


# ------------------------------------------------------------------------ steps

def step_extensions(conn: sqlite3.Connection) -> None:
    head(1, "Three extensions, one connection")
    versions = conn.execute(
        "SELECT ai_version(), vector_version(), cloudsync_version()"
    ).fetchone()
    print(f"   sqlite-ai {versions[0]}   sqlite-vector {versions[1]}   sqlite-sync {versions[2]}")
    verdict("Three extensions co-resident", "OK", f"ai {versions[0]} / vector {versions[1]} / sync {versions[2]}")


def step_two_models(gen: sqlite3.Connection, emb: sqlite3.Connection) -> int:
    head(2, "Two models, two connections")

    t0 = time.perf_counter()
    gen.execute("SELECT llm_model_load(?, 'gpu_layers=99')", (str(GEN_MODEL),))
    gen.execute("SELECT llm_context_create_textgen('context_size=2048')")
    gen_load = time.perf_counter() - t0

    gen.execute("SELECT llm_sampler_create()")
    gen.execute("SELECT llm_sampler_init_grammar(?, 'root')", (GRAMMAR,))
    gen.execute("SELECT llm_sampler_init_greedy()")   # greedy/dist must be registered LAST

    t0 = time.perf_counter()
    emb.execute("SELECT llm_model_load(?)", (str(EMBED_MODEL),))
    emb.execute("SELECT llm_context_create_embedding('embedding_type=FLOAT32')")
    emb_load = time.perf_counter() - t0

    dim = one(emb, "SELECT llm_model_n_embd()")
    print(f"   generation  {GEN_MODEL.name}  loaded in {gen_load:.2f}s")
    print(f"   embedding   {EMBED_MODEL.name}  loaded in {emb_load:.2f}s, {dim} dims")

    # The plan assumed one connection could hold both. It cannot, and the failure
    # is silent -- so prove the split actually works rather than trusting it.
    still_generating = one(gen, "SELECT length(llm_text_generate('ping', 'n_predict=4'))")
    still_embedding = one(emb, "SELECT length(llm_embed_generate('ping'))")
    ok = bool(still_generating) and still_embedding == dim * 4
    print(f"   both live simultaneously: {'yes' if ok else 'NO'} "
          f"(textgen {still_generating} chars, embed {still_embedding} bytes)")
    verdict("Two models resident at once", "OK" if ok else "FAIL",
            f"separate connections, {dim}-dim FLOAT32 embeddings")
    return dim


def step_schema(gen: sqlite3.Connection, dim: int) -> None:
    head(3, "Schema, sync metadata and the enrichment trigger")
    gen.execute("DROP TRIGGER IF EXISTS feedback_ai")
    gen.execute("DROP TABLE IF EXISTS feedback")
    gen.executescript(SCHEMA)
    gen.execute("SELECT cloudsync_init('feedback')")
    gen.execute(f"SELECT vector_init('feedback', 'embedding', 'type=FLOAT32,dimension={dim},distance=cosine')")
    gen.execute(TRIGGER)
    print("   feedback table created, cloudsync_init done, vector column declared")
    print("   trigger feedback_ai installed, guarded WHEN NEW.summary IS NULL")


def step_enrich(gen: sqlite3.Connection, limit: int) -> list[float]:
    head(4, f"Enrichment: {limit} rows through the trigger")

    # seed.sql holds exactly one row per line, so each row can be inserted (and
    # therefore timed) on its own rather than as one bulk statement.
    body = SEED.read_text().split("VALUES", 1)[1]
    values = [ln.strip().rstrip(";").rstrip(",")
              for ln in body.splitlines() if ln.strip().startswith("(")][:limit]

    timings: list[float] = []
    print(f"   {'':3} {'sec':>6}  {'sentiment':<9} {'category':<16} summary")
    for i, value in enumerate(values, 1):
        stmt = f"INSERT INTO feedback (id, author, channel, body, created_at) VALUES {value}"
        t0 = time.perf_counter()
        gen.execute(stmt)
        elapsed = time.perf_counter() - t0
        timings.append(elapsed)
        row = gen.execute(
            "SELECT sentiment, category, summary FROM feedback ORDER BY rowid DESC LIMIT 1"
        ).fetchone()
        print(f"   {i:>3} {elapsed:6.2f}  {row[0] or '-':<9} {row[1] or '-':<16} {(row[2] or '-')[:44]}")

    total, mean = sum(timings), statistics.mean(timings)
    print(f"\n   {len(timings)} rows in {total:.1f}s -- mean {mean:.2f}s/row, "
          f"median {statistics.median(timings):.2f}s, max {max(timings):.2f}s")

    checks = gen.execute("""
        SELECT count(*), sum(json_valid(ai_json)),
               sum(sentiment IN ('positive','neutral','negative')),
               sum(category IN ('billing','bugs','feature_request','ux','performance','other'))
          FROM feedback
    """).fetchone()
    n, valid, sent_ok, cat_ok = checks
    print(f"   valid JSON {valid}/{n}   sentiment in enum {sent_ok}/{n}   category in enum {cat_ok}/{n}")

    # In-enum is not the same as correct. Score against the hand-labelled truth.
    labelled = gen.execute("SELECT category, substr(body, 1, 46) FROM feedback ORDER BY rowid").fetchall()
    wrong = [(i, got, txt) for i, (got, txt) in enumerate(labelled)
             if i < len(EXPECTED_CATEGORY) and got not in EXPECTED_CATEGORY[i]]
    scored = min(len(labelled), len(EXPECTED_CATEGORY))
    right = scored - len(wrong)
    pct = right / scored * 100 if scored else 0
    print(f"   category CORRECT (vs hand labels) {right}/{scored} ({pct:.0f}%)")
    for i, got, txt in wrong:
        print(f"     row {i + 1:>2} got {got:<16} want {'/'.join(sorted(EXPECTED_CATEGORY[i])):<20} {txt}")

    verdict("Trigger latency", "OK" if mean < 2.0 else "SLOW",
            f"{mean:.2f}s/row mean for all three AI fields (one generation)")
    verdict("LLM output off-enum", "OK" if (sent_ok == n and cat_ok == n) else "FAIL",
            f"GBNF grammar: {valid}/{n} valid JSON, {sent_ok + cat_ok}/{2 * n} labels in enum -- no normalizer needed")
    verdict("Label accuracy (demo-visible)", "OK" if pct >= 85 else "WEAK",
            f"{right}/{scored} categories correct ({pct:.0f}%) against hand labels")
    return timings


def step_embed(gen: sqlite3.Connection, emb: sqlite3.Connection, dim: int) -> None:
    head(5, "Embeddings on the second connection")
    # Both init calls are per-connection, and this connection was opened before the
    # table existed. Without vector_init it cannot scan the column; without
    # cloudsync_init every write fails with "Unable to retrieve table name feedback
    # in cloudsync_update". Both are idempotent, so a worker can just always call them.
    emb.execute("SELECT cloudsync_init('feedback')")
    emb.execute(f"SELECT vector_init('feedback', 'embedding', 'type=FLOAT32,dimension={dim},distance=cosine')")
    pending = emb.execute("SELECT id, body FROM feedback WHERE embedding IS NULL").fetchall()
    t0 = time.perf_counter()
    for row_id, body in pending:
        emb.execute("UPDATE feedback SET embedding = llm_embed_generate(?) WHERE id = ?", (body, row_id))
    elapsed = time.perf_counter() - t0
    done = one(emb, "SELECT count(*) FROM feedback WHERE embedding IS NOT NULL")
    print(f"   embedded {len(pending)} rows in {elapsed:.2f}s ({elapsed / max(len(pending), 1) * 1000:.0f}ms/row)")
    print(f"   rows with an embedding: {done}")
    print("   note: embedding cannot be a trigger -- the trigger would fire on the")
    print("         writer connection, which holds the generation model, not this one.")


def step_search(emb: sqlite3.Connection) -> None:
    head(6, "Semantic search vs keyword search")
    query = "complaints about billing"

    qv = one(emb, "SELECT llm_embed_generate(?)", query)
    t0 = time.perf_counter()
    hits = emb.execute("""
        SELECT f.body, f.category, v.distance
          FROM vector_full_scan('feedback', 'embedding', ?, 6) AS v
          JOIN feedback f ON f.rowid = v.rowid
         ORDER BY v.distance
    """, (qv,)).fetchall()
    elapsed = time.perf_counter() - t0

    print(f'   semantic: "{query}"  ({elapsed * 1000:.0f}ms)')
    without_word = 0
    for body, category, distance in hits:
        has_word = "billing" in body.lower()
        without_word += 0 if has_word else 1
        print(f"     {distance:.3f} [{category or '-':<15}] {'   ' if has_word else ' * '}{body[:62]}")

    keyword = emb.execute(
        "SELECT count(*) FROM feedback WHERE body LIKE '%billing%'"
    ).fetchone()[0]
    print(f'\n   keyword:  body LIKE \'%billing%\'  ->  {keyword} rows')
    print(f"   {without_word} of the {len(hits)} semantic hits contain no form of the word (marked *)")

    verdict("Semantic search", "OK" if without_word >= 3 else "WEAK",
            f"{len(hits)} hits in {elapsed * 1000:.0f}ms, {without_word} invisible to LIKE (keyword finds {keyword})")


def step_sync(gen: sqlite3.Connection, dim: int) -> None:
    head(7, "Sync round trip to a second device (offline, no server)")

    DB_B.unlink(missing_ok=True)
    for suffix in ("-wal", "-shm"):
        Path(str(DB_B) + suffix).unlink(missing_ok=True)

    # Device B is deliberately NOT closed in this function. It loaded the extension
    # last, so it owns the global llama log callback and must be closed last of all;
    # shutdown() does that. See notes/FINDINGS.md #1.
    b = connect(DB_B)
    # Device B is a full peer: same schema, same trigger, same model. It has to
    # be, or the "unenriched row gets enriched on arrival" branch proves nothing.
    b.executescript(SCHEMA)
    b.execute("SELECT cloudsync_init('feedback')")
    b.execute(f"SELECT vector_init('feedback', 'embedding', 'type=FLOAT32,dimension={dim},distance=cosine')")
    b.execute("SELECT llm_model_load(?, 'gpu_layers=99')", (str(GEN_MODEL),))
    b.execute("SELECT llm_context_create_textgen('context_size=2048')")
    b.execute("SELECT llm_sampler_create()")
    b.execute("SELECT llm_sampler_init_grammar(?, 'root')", (GRAMMAR,))
    b.execute("SELECT llm_sampler_init_greedy()")
    b.execute(TRIGGER)

    payload = one(gen, """
        SELECT cloudsync_payload_encode(tbl, pk, col_name, col_value,
                                        col_version, db_version, site_id, cl, seq)
          FROM cloudsync_changes
    """)
    print(f"   device A -> payload of {len(payload)} bytes")

    t0 = time.perf_counter()
    applied = one(b, "SELECT cloudsync_payload_apply(?)", payload)
    elapsed = time.perf_counter() - t0

    landed = b.execute(
        "SELECT count(*), sum(summary IS NOT NULL), sum(category IS NOT NULL) FROM feedback"
    ).fetchone()
    print(f"   device B <- applied in {elapsed:.2f}s ({applied} changes)")
    print(f"   rows on B: {landed[0]}, of which {landed[1]} arrived with a summary already computed")

    # Zero inference on B is the whole claim. If the trigger had fired, applying
    # N rows would have taken N seconds, not milliseconds.
    recomputed = elapsed > 1.0
    print(f"   inference re-run on arrival: {'YES -- guard failed' if recomputed else 'no'}")

    # The other branch. A device that cannot run the model at all -- the phone
    # companion in Phase 4 -- sends raw text. Simulate one by dropping the
    # trigger on A, so the row leaves without AI values, and check that B
    # enriches it on arrival rather than storing a half-empty row forever.
    watermark = one(gen, "SELECT max(db_version) FROM cloudsync_changes") or 0
    gen.execute("DROP TRIGGER feedback_ai")
    gen.execute(
        "INSERT INTO feedback (id, author, channel, body, created_at) "
        "VALUES (cloudsync_uuid(), 'raw@example.com', 'email', "
        "'The invoice you sent lists a seat count we never agreed to.', '2026-07-15T09:00:00Z')"
    )
    gen.execute(TRIGGER)
    raw = gen.execute("SELECT summary FROM feedback WHERE author = 'raw@example.com'").fetchone()
    print(f"   device A -> raw row created with no AI values (summary is {raw[0]!r})")

    raw_payload = one(gen, """
        SELECT cloudsync_payload_encode(tbl, pk, col_name, col_value,
                                        col_version, db_version, site_id, cl, seq)
          FROM cloudsync_changes WHERE db_version > ?
    """, watermark)

    t0 = time.perf_counter()
    b.execute("SELECT cloudsync_payload_apply(?)", (raw_payload,))
    enrich_time = time.perf_counter() - t0
    landed_raw = b.execute(
        "SELECT summary, sentiment, category FROM feedback WHERE author = 'raw@example.com'"
    ).fetchone()
    enriched_on_b = bool(landed_raw and landed_raw[0])
    print(f"   device B <- enriched it locally in {enrich_time:.2f}s: "
          f"[{(landed_raw or [None, None, '-'])[2] or '-'}] "
          f"{((landed_raw or [None])[0] or 'NOT ENRICHED')[:46]}")

    guard_ok = not recomputed and enriched_on_b
    # Do not read more into this than it says. It measures ONE payload_apply, which
    # writes each row in a single statement -- and there the guard does hold. Real
    # network sync applies a row column by column and re-enters the trigger on each
    # write, where the guard does NOT hold: 101s vs 4.6s for a 24-row first sync.
    # notes/FINDINGS.md #13. The engine drops the trigger while applying.
    verdict("Sync-applied rows re-firing triggers", "OK" if guard_ok else "FAIL",
            "guard holds for a single-statement apply -- NOT for network sync, see FINDINGS #13")
    verdict("AI values carried by sync", "OK",
            f"{landed[1]}/{landed[0]} rows arrived with AI columns intact -- no second inference")


def step_cloud(gen: sqlite3.Connection, *, attempt: bool) -> None:
    head(8, "Live SQLite Cloud round trip")
    db_id = os.environ.get("EDGETABLE_DB_ID")
    api_key = os.environ.get("EDGETABLE_API_KEY")
    if not attempt:
        print("   not attempted -- pass --cloud to include this step")
        verdict("Live sync round trip", "UNTESTED", "re-run with --cloud once a SQLite Cloud project exists")
        return
    if not (db_id and api_key):
        print("   skipped -- set EDGETABLE_DB_ID and EDGETABLE_API_KEY to run this step")
        verdict("Live sync round trip", "UNTESTED", "needs a SQLite Cloud project (EDGETABLE_DB_ID/_API_KEY)")
        return
    gen.execute("SELECT cloudsync_network_init(?)", (db_id,))
    gen.execute("SELECT cloudsync_network_set_apikey(?)", (api_key,))
    t0 = time.perf_counter()
    result = one(gen, "SELECT cloudsync_network_sync()")
    print(f"   sync in {time.perf_counter() - t0:.2f}s: {result}")
    verdict("Live sync round trip", "OK", f"{result}")


def main() -> int:
    ap = argparse.ArgumentParser(description="EdgeTable Phase 0 spike")
    ap.add_argument("--rows", type=int, default=24, help="how many seed rows to enrich")
    ap.add_argument("--cloud", action="store_true", help="also attempt a live SQLite Cloud sync")
    args = ap.parse_args()

    for path in (GEN_MODEL, EMBED_MODEL):
        if not path.exists():
            print(f"ERROR: missing model {path}", file=sys.stderr)
            return 1

    DB_A.unlink(missing_ok=True)
    for suffix in ("-wal", "-shm"):
        Path(str(DB_A) + suffix).unlink(missing_ok=True)

    print("\n\033[1mEdgeTable -- Phase 0 spike\033[0m")
    print("Killing the unknowns in notes/plan.md before Phase 1 starts.")

    gen = connect(DB_A)
    emb = connect(DB_A)
    try:
        step_extensions(gen)
        dim = step_two_models(gen, emb)
        step_schema(gen, dim)
        step_enrich(gen, args.rows)
        step_embed(gen, emb, dim)
        step_search(emb)
        step_sync(gen, dim)
        step_cloud(gen, attempt=args.cloud)

        head(9, "Verdict against the plan's risk table")
        width = max(len(r) for r, _, _ in VERDICTS)
        for risk, status, detail in VERDICTS:
            colour = {"OK": "32", "FAIL": "31", "SLOW": "33", "WEAK": "33"}.get(status, "90")
            print(f"   {risk:<{width}}  \033[{colour}m{status:<9}\033[0m {detail}")
        # WEAK counts as a failure: these are thresholds the demo depends on, so a
        # prompt or model change that erodes one should break the run, not just
        # print a yellow word nobody reads. UNTESTED does not fail.
        bad = [r for r, st, _ in VERDICTS if st in ("FAIL", "WEAK")]
        if bad:
            print(f"   {len(bad)} below threshold: {', '.join(bad)}")
        print()
        return 1 if bad else 0
    finally:
        shutdown()


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