codex/cli.py

@@ -29,6 +29,37 @@ app.add_typer(graph_app, name="graph")
 app.add_typer(search_app, name="search")
 
 
+@app.command()
+def migrate() -> None:
+    """Apply infra/schema.sql to bring the database schema up to date (idempotent).
+
+    Must connect as a role that can run DDL (owns / can CREATE + ALTER the tables).
+    The application's DATABASE_URL is usually a least-privilege DML role and will
+    fail with InsufficientPrivilege; set MIGRATION_DATABASE_URL to a privileged
+    (owner/superuser) connection (audit M-1).
+    """
+    import psycopg
+    import psycopg.rows
+
+    from codex.config import get_settings
+    from codex.db import apply_schema
+
+    settings = get_settings()
+    url = settings.migration_database_url or settings.database_url
+    try:
+        with psycopg.connect(url, row_factory=psycopg.rows.dict_row) as conn:
+            apply_schema(conn)
+    except psycopg.errors.InsufficientPrivilege as exc:
+        typer.echo(
+            "ERROR: schema migration needs a role that owns the tables "
+            "(CREATE/ALTER). Set MIGRATION_DATABASE_URL to a privileged connection "
+            f"and retry. ({exc})",
+            err=True,
+        )
+        raise typer.Exit(1) from exc
+    typer.echo("Schema applied — database is up to date.")
+
+
 @app.command()
 def ingest(
     paper_id: str = typer.Argument(..., help="arXiv ID, DOI, or OpenAlex W-ID"),

codex/config.py

@@ -34,6 +34,17 @@ class Settings(BaseSettings):
         ),
     )
 
+    migration_database_url: str | None = Field(
+        default=None,
+        description=(
+            "Optional privileged connection string used by `codex migrate` to apply "
+            "schema DDL. The app's database_url is typically a least-privilege DML "
+            "role that cannot CREATE/ALTER owner-held tables (raises "
+            "InsufficientPrivilege); point this at an owner/superuser connection. "
+            "Falls back to database_url when unset."
+        ),
+    )
+
     # ------------------------------------------------------------------
     # External services
     # ------------------------------------------------------------------

codex/db.py

@@ -45,11 +45,16 @@ def get_conn() -> Generator[psycopg.Connection[psycopg.rows.DictRow], None, None
 def apply_schema(conn: psycopg.Connection[psycopg.rows.DictRow]) -> None:
     """Idempotently apply ``infra/schema.sql`` to the connected database.
 
-    Safe to call on every startup — all DDL statements use
+    Safe to call repeatedly — every statement is ``CREATE … IF NOT EXISTS`` or
-    ``CREATE … IF NOT EXISTS``.
+    ``ADD COLUMN IF NOT EXISTS``, so re-application is a no-op.
+
+    Requires a connection whose role can run DDL (owns / can CREATE + ALTER the
+    tables). The least-privilege application role is typically DML-only and will
+    raise ``InsufficientPrivilege``; use a privileged connection — see
+    ``codex migrate`` and ``MIGRATION_DATABASE_URL`` (audit M-1).
 
     Args:
-        conn: An open psycopg connection (obtained via :func:`get_conn`).
+        conn: An open psycopg connection with DDL privileges.
     """
     schema_path = Path(__file__).parent.parent / "infra" / "schema.sql"
     sql = schema_path.read_text(encoding="utf-8")

infra/schema.sql

@@ -13,7 +13,7 @@ CREATE EXTENSION IF NOT EXISTS vector;
 -- ---------------------------------------------------------------------
 -- Layer 1: Papers + full-text chunks
 -- ---------------------------------------------------------------------
-CREATE TABLE papers (
+CREATE TABLE IF NOT EXISTS papers (
     id            TEXT PRIMARY KEY,        -- canonical ID: arXiv ID or DOI
     openalex_id   TEXT UNIQUE,             -- e.g. W2741809807
     bibkey        TEXT UNIQUE,             -- BibTeX key -> .bib + Doxygen @cite
@@ -27,10 +27,10 @@ CREATE TABLE papers (
 );
 
 -- HNSW index for fast approximate nearest-neighbour search on abstracts.
-CREATE INDEX papers_abstract_emb_idx
+CREATE INDEX IF NOT EXISTS papers_abstract_emb_idx
     ON papers USING hnsw (abstract_emb vector_cosine_ops);
 
-CREATE TABLE chunks (
+CREATE TABLE IF NOT EXISTS chunks (
     id        BIGSERIAL PRIMARY KEY,
     paper_id  TEXT REFERENCES papers(id) ON DELETE CASCADE,
     ord       INT NOT NULL,               -- position within the paper
@@ -38,13 +38,13 @@ CREATE TABLE chunks (
     embedding vector(1024)
 );
 
-CREATE INDEX chunks_emb_idx
+CREATE INDEX IF NOT EXISTS chunks_emb_idx
     ON chunks USING hnsw (embedding vector_cosine_ops);
-CREATE INDEX chunks_paper_idx ON chunks (paper_id);
+CREATE INDEX IF NOT EXISTS chunks_paper_idx ON chunks (paper_id);
 
 -- Sparse / keyword hits for exact mathematical terminology (hybrid search).
 -- Dense (above) + full-text (below) combined = robust against math terms.
-CREATE INDEX chunks_fts_idx
+CREATE INDEX IF NOT EXISTS chunks_fts_idx
     ON chunks USING gin (to_tsvector('english', content));
 
 -- ---------------------------------------------------------------------
@@ -53,19 +53,19 @@ CREATE INDEX chunks_fts_idx
 --   edges to not-yet-ingested papers are intentionally preserved —
 --   those are your discovery leads.
 -- ---------------------------------------------------------------------
-CREATE TABLE citations (
+CREATE TABLE IF NOT EXISTS citations (
     citing_id TEXT REFERENCES papers(id) ON DELETE CASCADE,
     cited_id  TEXT NOT NULL,               -- arXiv/DOI/OpenAlex ID of the target
     context   TEXT,                        -- optional: citation context (S2)
     PRIMARY KEY (citing_id, cited_id)
 );
 
-CREATE INDEX citations_cited_idx ON citations (cited_id);
+CREATE INDEX IF NOT EXISTS citations_cited_idx ON citations (cited_id);
 
 -- ---------------------------------------------------------------------
 -- Layer 3: Provenance  (the "cleanly couple" goal)
 -- ---------------------------------------------------------------------
-CREATE TABLE code_links (
+CREATE TABLE IF NOT EXISTS code_links (
     id        BIGSERIAL PRIMARY KEY,
     symbol    TEXT NOT NULL,               -- e.g. 'dec::hodge_star' or 'file.cpp:120'
     paper_id  TEXT REFERENCES papers(id) ON DELETE SET NULL,
@@ -74,8 +74,8 @@ CREATE TABLE code_links (
     added_at  TIMESTAMPTZ DEFAULT now()
 );
 
-CREATE INDEX code_links_symbol_idx ON code_links (symbol);
+CREATE INDEX IF NOT EXISTS code_links_symbol_idx ON code_links (symbol);
-CREATE INDEX code_links_paper_idx  ON code_links (paper_id);
+CREATE INDEX IF NOT EXISTS code_links_paper_idx  ON code_links (paper_id);
 
 -- ---------------------------------------------------------------------
 -- Example query: discovery leads

tests/scaffold/test_db.py

@@ -33,3 +33,40 @@ def test_apply_schema_executes_sql() -> None:
 
     mock_conn.execute.assert_called_once()
     mock_conn.commit.assert_called_once()
+
+
+def test_schema_sql_is_idempotent() -> None:
+    """Every CREATE TABLE/INDEX in infra/schema.sql uses IF NOT EXISTS (audit M-1, T-2).
+
+    Re-applying schema.sql must be a safe no-op so the migration can run
+    repeatedly against an existing database.
+    """
+    import re
+    from pathlib import Path
+
+    schema = (Path(__file__).resolve().parents[2] / "infra" / "schema.sql").read_text()
+    # Drop line comments so commented-out example queries don't count.
+    code = "\n".join(line.split("--", 1)[0] for line in schema.splitlines())
+    non_idempotent = re.findall(
+        r"CREATE\s+(?:UNIQUE\s+)?(?:TABLE|INDEX)\s+(?!IF\s+NOT\s+EXISTS)(\w+)",
+        code,
+        re.IGNORECASE,
+    )
+    assert not non_idempotent, f"schema.sql has non-idempotent CREATE statements: {non_idempotent}"
+
+
+def test_migrate_command_reports_privilege_error() -> None:
+    """`codex migrate` surfaces InsufficientPrivilege with guidance and exits 1 (M-1)."""
+    import psycopg
+    from typer.testing import CliRunner
+
+    from codex.cli import app
+
+    def boom(*args: object, **kwargs: object) -> object:
+        raise psycopg.errors.InsufficientPrivilege("must be owner of table chunks")
+
+    with patch("psycopg.connect", boom):
+        result = CliRunner().invoke(app, ["migrate"])
+
+    assert result.exit_code == 1
+    assert "MIGRATION_DATABASE_URL" in result.output