diff --git a/src/store/postgres.ts b/src/store/postgres.ts
index 4ba0ca7..20eccf0 100644
--- a/src/store/postgres.ts
+++ b/src/store/postgres.ts
@@ -1,7 +1,7 @@
 import { randomUUID } from "node:crypto";
 import { Pool, type PoolClient, type QueryResultRow } from "pg";
 import { type AppSlug, type HeimdallAuthAttemptStatus, type LinkedIdentityInput, type Provider } from "../contracts.js";
-import { CREATE_SCHEMA_SQL, REQUIRED_COLUMNS, REQUIRED_KEYS, REQUIRED_RELATIONS } from "./schema.js";
+import { CREATE_SCHEMA_SQL, REQUIRED_COLUMNS, REQUIRED_KEYS, REQUIRED_TABLES } from "./schema.js";
 import {
   type CreateAccountInput,
   type CreateAuthAttemptInput,
@@ -284,6 +284,85 @@ function mapAuthAttemptRow(row: AuthAttemptRow): StoredAuthAttempt {
   return attempt;
 }
 
+/**
+ * The AND-ed conditions of a predicate as Postgres prints it, each without a
+ * pair of parentheses that encloses all of it. Postgres flattens a chain of
+ * ANDs into one list, so `((a) AND (b) AND (c))` yields `a`, `b`, `c`. An OR,
+ * or an AND nested under one, stays whole: `((a OR b) AND c)` yields `a OR b`
+ * and `c`, and `(a OR (b AND c))` yields itself. Quoted text is never split.
+ */
+export function predicateConditions(printed: string): string[] {
+  const unwrap = (text: string): string => {
+    let current = text.trim();
+    while (current.startsWith("(") && closingParenthesis(current, 0) === current.length - 1) {
+      current = current.slice(1, -1).trim();
+    }
+    return current;
+  };
+  const whole = unwrap(printed);
+  if (whole === "") return [];
+  const parts: string[] = [];
+  let depth = 0;
+  let quote: string | undefined;
+  let start = 0;
+  for (let position = 0; position < whole.length; position += 1) {
+    const character = whole[position]!;
+    if (quote) {
+      if (character === quote) quote = undefined;
+    } else if (character === "'" || character === '"') {
+      quote = character;
+    } else if (character === "(") {
+      depth += 1;
+    } else if (character === ")") {
+      depth -= 1;
+    } else if (depth === 0 && whole.startsWith(" AND ", position)) {
+      parts.push(whole.slice(start, position));
+      start = position + " AND ".length;
+      position = start - 1;
+    }
+  }
+  parts.push(whole.slice(start));
+  return parts.map(unwrap);
+}
+
+/** The index of the parenthesis closing the one at `open`, outside quoted text, or -1. */
+function closingParenthesis(text: string, open: number): number {
+  let depth = 0;
+  let quote: string | undefined;
+  for (let position = open; position < text.length; position += 1) {
+    const character = text[position]!;
+    if (quote) {
+      if (character === quote) quote = undefined;
+    } else if (character === "'" || character === '"') {
+      quote = character;
+    } else if (character === "(") {
+      depth += 1;
+    } else if (character === ")") {
+      depth -= 1;
+      if (depth === 0) return position;
+    }
+  }
+  return -1;
+}
+
+/**
+ * Whether an index predicate, as Postgres prints it, keeps exactly the rows
+ * the declared one (printed the same way) does: the same set of AND-ed
+ * conditions, ignoring `key_column IS NOT NULL` unless the index treats NULLs
+ * as equal, since otherwise rows with a NULL key never collide anyway.
+ */
+export function samePredicate(declared: string, actual: string, keyColumns: string[], nullsNotDistinct: boolean): boolean {
+  // A key without a predicate must have none: ON CONFLICT cannot infer a
+  // partial index as its arbiter.
+  if (predicateConditions(declared).length === 0) return actual.trim() === "";
+  const redundant = new Set(nullsNotDistinct ? [] : keyColumns.map((column) => `${column} IS NOT NULL`));
+  const conditions = (printed: string) =>
+    [...new Set(predicateConditions(printed).filter((condition) => !redundant.has(condition)))].sort();
+  const left = conditions(declared);
+  const right = conditions(actual);
+  return left.length === right.length && left.every((condition, index) => condition === right[index]);
+}
+
 /**
  * What the store's statements do to each table, and so the privileges its
  * database role needs: it reads what it returns or looks up, inserts what it
@@ -311,21 +390,43 @@ export class PostgresStore implements HeimdallStore {
 
   /**
    * Proves, before Heimdall serves, that the database can take its requests:
-   * every relation the schema makes, every column with its declared type,
-   * every unique key as the store relies on it, and every privilege in
-   * STORE_TABLE_PRIVILEGES. A missing or altered piece fails with code
-   * SCHEMA_MISSING and a missing privilege with SCHEMA_PRIVILEGES; neither
-   * error names the piece.
+   * every table the schema makes; every column with its declared type, and
+   * nullable where the schema leaves it nullable; every unique key; and every
+   * privilege in STORE_TABLE_PRIVILEGES, held on the table or on each of its
+   * columns. A missing or altered piece fails with code SCHEMA_MISSING and a
+   * missing privilege with SCHEMA_PRIVILEGES; neither error names the piece.
+   *
+   * A key is met by a unique index on its table that is valid and ready (a
+   * failed CREATE INDEX CONCURRENTLY leaves one that is neither), immediate
+   * (ON CONFLICT refuses a deferrable arbiter), has exactly the key's columns
+   * as its key columns (INCLUDE columns aside), compares them under
+   * deterministic collations (a case-insensitive one would merge two
+   * providers' user ids into one row), and has the key's predicate. A
+   * predicate matches when both, as Postgres prints them, are the same set of
+   * AND-ed conditions; a `key_column IS NOT NULL` condition may be absent on
+   * either side, because a unique index never makes NULL keys collide unless
+   * it is NULLS NOT DISTINCT.
+   *
+   * Postgres prints the declared predicates itself: they are indexed on a
+   * temporary copy of the table in a transaction that is rolled back.
    */
   async checkSchema(): Promise<void> {
+    const declaredPredicates = await this.printedDeclaredPredicates();
     const privileges = Object.entries(STORE_TABLE_PRIVILEGES).flatMap(([table, wanted]) =>
       wanted.map((privilege) => ({ table, privilege }))
     );
-    const result = await this.pool.query<{ kind: "missing" | "privilege" }>(
+    const result = await this.pool.query<{
+      kind: "missing" | "privilege" | "candidate";
+      key: number | null;
+      predicate: string | null;
+      nulls_not_distinct: boolean | null;
+    }>(
       `
-      SELECT 'missing' AS kind FROM unnest($1::text[]) AS name WHERE to_regclass(name) IS NULL
+      SELECT 'missing' AS kind, NULL::int AS key, NULL::text AS predicate, NULL::boolean AS nulls_not_distinct
+      FROM unnest($1::text[]) AS name WHERE to_regclass(name) IS NULL
       UNION ALL
-      SELECT 'missing' FROM unnest($2::text[], $3::text[], $4::text[]) AS required(table_name, column_name, type_name)
+      SELECT 'missing', NULL, NULL, NULL
+      FROM unnest($2::text[], $3::text[], $4::text[], $5::boolean[]) AS required(table_name, column_name, type_name, nullable)
       WHERE NOT EXISTS (
         SELECT 1 FROM pg_attribute a
         WHERE a.attrelid = to_regclass(required.table_name)
@@ -333,50 +434,99 @@ export class PostgresStore implements HeimdallStore {
           AND a.attnum > 0
           AND NOT a.attisdropped
           AND a.atttypid = to_regtype(required.type_name)
+          AND (NOT required.nullable OR NOT a.attnotnull)
       )
       UNION ALL
-      SELECT 'missing' FROM unnest($5::text[], $6::text[], $7::text[]) AS required(table_name, columns, predicate)
-      WHERE NOT EXISTS (
-        SELECT 1 FROM pg_index i
-        WHERE i.indrelid = to_regclass(required.table_name)
-          AND i.indisunique
-          AND i.indimmediate
-          AND coalesce(regexp_replace(lower(pg_get_expr(i.indpred, i.indrelid)), '[()[:space:]]', '', 'g'), '') = required.predicate
-          AND i.indnkeyatts = cardinality(string_to_array(required.columns, ','))
-          AND (
-            SELECT string_agg(a.attname, ',' ORDER BY a.attname)
-            FROM unnest(i.indkey::int2[]) WITH ORDINALITY AS key(attnum, position)
-            JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = key.attnum
-            WHERE key.position <= i.indnkeyatts
-          ) = required.columns
-      )
+      SELECT 'candidate', required.key::int, pg_get_expr(i.indpred, i.indrelid),
+        coalesce((to_jsonb(i) ->> 'indnullsnotdistinct')::boolean, false)
+      FROM unnest($6::text[], $7::text[]) WITH ORDINALITY AS required(table_name, columns, key)
+      JOIN pg_index i ON i.indrelid = to_regclass(required.table_name)
+      WHERE i.indisunique
+        AND i.indisvalid
+        AND i.indisready
+        AND i.indimmediate
+        AND i.indnkeyatts = cardinality(string_to_array(required.columns, ','))
+        AND (
+          SELECT string_agg(a.attname, ',' ORDER BY a.attname)
+          FROM unnest(i.indkey::int2[]) WITH ORDINALITY AS key(attnum, position)
+          JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = key.attnum
+          WHERE key.position <= i.indnkeyatts
+        ) = required.columns
+        AND NOT EXISTS (
+          SELECT 1 FROM unnest(i.indcollation::oid[]) WITH ORDINALITY AS collation(oid, position)
+          JOIN pg_collation c ON c.oid = collation.oid
+          WHERE collation.position <= i.indnkeyatts AND NOT c.collisdeterministic
+        )
       UNION ALL
-      SELECT 'privilege' FROM unnest($8::text[], $9::text[]) AS required(table_name, privilege)
-      WHERE to_regclass(required.table_name) IS NOT NULL
-        AND NOT has_table_privilege(to_regclass(required.table_name), required.privilege)
+      SELECT 'privilege', NULL, NULL, NULL
+      FROM unnest($8::text[], $9::text[]) AS required(table_name, privilege)
+      WHERE EXISTS (
+        SELECT 1 FROM unnest($2::text[], $3::text[]) AS used(table_name, column_name)
+        JOIN pg_attribute a ON a.attrelid = to_regclass(used.table_name) AND a.attname = used.column_name AND NOT a.attisdropped
+        WHERE used.table_name = required.table_name
+          AND NOT has_column_privilege(a.attrelid, a.attnum, required.privilege)
+      )
       `,
       [
-        REQUIRED_RELATIONS,
+        REQUIRED_TABLES,
         REQUIRED_COLUMNS.map((required) => required.table),
         REQUIRED_COLUMNS.map((required) => required.column),
         REQUIRED_COLUMNS.map((required) => required.type),
+        REQUIRED_COLUMNS.map((required) => required.nullable),
         REQUIRED_KEYS.map((required) => required.table),
         REQUIRED_KEYS.map((required) => required.columns),
-        REQUIRED_KEYS.map((required) => required.predicate),
         privileges.map((required) => required.table),
         privileges.map((required) => required.privilege),
       ]
     );
-    if (result.rows.some((row) => row.kind === "missing")) {
+    const keyMet = REQUIRED_KEYS.map((required, index) =>
+      result.rows.some(
+        (row) =>
+          row.kind === "candidate" &&
+          row.key === index + 1 &&
+          samePredicate(declaredPredicates[index]!, row.predicate ?? "", required.columns.split(","), row.nulls_not_distinct === true)
+      )
+    );
+    if (result.rows.some((row) => row.kind === "missing") || keyMet.includes(false)) {
       throw Object.assign(new Error("The Heimdall schema is not applied."), { code: "SCHEMA_MISSING" });
     }
-    if (result.rows.length > 0) {
+    if (result.rows.some((row) => row.kind === "privilege")) {
       throw Object.assign(new Error("The Heimdall database role lacks a privilege the store needs."), {
         code: "SCHEMA_PRIVILEGES",
       });
     }
   }
 
+  /**
+   * Each REQUIRED_KEYS predicate as Postgres prints it back (empty for none).
+   * One simple-protocol round trip on one connection: a transaction that
+   * builds a temporary copy of each keyed table's declared columns, indexes
+   * it with the declared predicate, reads the printed form and rolls back.
+   * The names and SQL are the schema's own constants, never input.
+   */
+  private async printedDeclaredPredicates(): Promise<string[]> {
+    const partial = REQUIRED_KEYS.flatMap((required, index) => (required.predicate ? [{ ...required, index }] : []));
+    if (partial.length === 0) return REQUIRED_KEYS.map(() => "");
+    const statements = partial.flatMap(({ table, columns, predicate, index }) => {
+      const probe = `heimdall_key_probe_${index}`;
+      const declared = REQUIRED_COLUMNS.filter((column) => column.table === table)
+        .map((column) => `${column.column} ${column.type}`)
+        .join(", ");
+      return [
+        `CREATE TEMP TABLE ${probe} (${declared}) ON COMMIT DROP`,
+        `CREATE UNIQUE INDEX ${probe}_idx ON ${probe} (${columns}) WHERE ${predicate}`,
+      ];
+    });
+    const select = partial
+      .map(({ index }) => `SELECT ${index} AS key, pg_get_expr(indpred, indrelid) AS predicate FROM pg_index WHERE indexrelid = 'pg_temp.heimdall_key_probe_${index}_idx'::regclass`)
+      .join(" UNION ALL ");
+    const results = (await this.pool.query(["BEGIN", ...statements, select, "ROLLBACK"].join(";\n"))) as unknown as Array<{
+      rows: Array<{ key: number; predicate: string }>;
+    }>;
+    const printed = results[results.length - 2]!.rows;
+    return REQUIRED_KEYS.map((_, index) => printed.find((row) => row.key === index)?.predicate ?? "");
+  }
+
   async close(): Promise<void> {
     await this.pool.end();
   }
diff --git a/src/store/schema.ts b/src/store/schema.ts
index 91db7fc..f3588c0 100644
--- a/src/store/schema.ts
+++ b/src/store/schema.ts
@@ -180,30 +180,25 @@ function keyColumns(line: string, keyword: "PRIMARY KEY" | "UNIQUE"): string | u
   return match ? columnList(match[1]!) : undefined;
 }
 
-/**
- * An index predicate in the form the schema check compares: lower case, with
- * no parentheses or white space. Postgres stores `a IS NOT NULL AND b IS NULL`
- * back as `((a IS NOT NULL) AND (b IS NULL))`; both reduce to the same text.
- */
-export function comparablePredicate(predicate: string): string {
-  return predicate.toLowerCase().replace(/[()\s]/g, "");
-}
-
 /*
  * What every start must find before it serves, read from the SQL above so the
  * two cannot drift. A start that applies the schema checks as well, because
- * CREATE ... IF NOT EXISTS leaves a damaged table as it is.
+ * CREATE ... IF NOT EXISTS leaves a damaged table as it is. Indexes that only
+ * speed lookups are not required; the unique ones are, by shape, in
+ * REQUIRED_KEYS.
+ *
+ * Out of scope: CHECK constraints, triggers and row-level security. Each can
+ * make a write fail on a database that passes this check.
  */
 
-/** Every table and index the schema creates. */
-export const REQUIRED_RELATIONS = [
-  ...schemaText.matchAll(/^CREATE (?:UNIQUE )?(?:TABLE|INDEX) IF NOT EXISTS (\w+)/gm),
-].map((match) => match[1]!);
+/** Every table the schema creates. */
+export const REQUIRED_TABLES = createdTables.map(({ table }) => table);
 
 /**
  * Every column a created table declares and every column the schema adds to
- * an existing table, each with its declared type: the store reads and writes
- * the values as that type, and a column retyped under it fails its queries.
+ * an existing table, each with its declared type, and whether it is nullable:
+ * the store reads and writes values as that type, and leaves a nullable
+ * column null, so a column retyped or made NOT NULL under it fails its writes.
  */
 export const REQUIRED_COLUMNS = [
   ...createdTables.flatMap(({ table, lines }) =>
@@ -211,24 +206,23 @@ export const REQUIRED_COLUMNS = [
       .filter((line) => !/^(?:PRIMARY KEY|UNIQUE)\b/.test(line))
       .map((line) => {
         const [, column, type] = /^(\w+) (\w+)/.exec(line)!;
-        return { table, column: column!, type: type! };
+        return { table, column: column!, type: type!, nullable: !/\b(?:NOT NULL|PRIMARY KEY)\b/.test(line) };
       })
   ),
-  ...[...schemaText.matchAll(/^ALTER TABLE (\w+) ADD COLUMN IF NOT EXISTS (\w+) (\w+)/gm)].map((match) => ({
+  ...[...schemaText.matchAll(/^ALTER TABLE (\w+) ADD COLUMN IF NOT EXISTS (\w+) (\w+)([^;]*);/gm)].map((match) => ({
     table: match[1]!,
     column: match[2]!,
     type: match[3]!,
+    nullable: !/\b(?:NOT NULL|PRIMARY KEY)\b/.test(match[4]!),
   })),
 ];
 
 /**
  * Every unique key the schema makes: the primary keys and unique constraints
  * of the created tables, and each unique index it creates, with that index's
- * predicate. The store relies on each as an immediate unique index on its
- * table whose key columns are exactly these, with exactly this predicate: ON
- * CONFLICT refuses a deferrable arbiter and cannot infer a partial or wider
- * one, and the partial attempt index is what keeps one unconsumed completion
- * per attempt (R21.1). Any index with that shape serves, whatever its name.
+ * predicate as the schema writes it (empty for none). Any index of the right
+ * shape serves, whatever its name; PostgresStore.checkSchema says what the
+ * right shape is.
  */
 export const REQUIRED_KEYS = [
   ...createdTables.flatMap(({ table, lines }) =>
@@ -241,6 +235,6 @@ export const REQUIRED_KEYS = [
   ...[...schemaText.matchAll(/^CREATE UNIQUE INDEX IF NOT EXISTS \w+\s+ON (\w+)\(([^)]*)\)(?:\s+WHERE ([^;]+))?;/gm)].map((match) => ({
     table: match[1]!,
     columns: columnList(match[2]!),
-    predicate: comparablePredicate(match[3] ?? ""),
+    predicate: (match[3] ?? "").replace(/\s+/g, " ").trim(),
   })),
 ];
diff --git a/tests/store.test.ts b/tests/store.test.ts
index 25500c7..22a694b 100644
--- a/tests/store.test.ts
+++ b/tests/store.test.ts
@@ -4,8 +4,14 @@ import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
 import { newDb } from "pg-mem";
 import { type HeimdallConfig } from "../src/config.js";
 import { createStore } from "../src/store/index.js";
-import { createPostgresStore, PostgresStore, STORE_TABLE_PRIVILEGES } from "../src/store/postgres.js";
-import { comparablePredicate, REQUIRED_COLUMNS, REQUIRED_KEYS, REQUIRED_RELATIONS } from "../src/store/schema.js";
+import {
+  createPostgresStore,
+  PostgresStore,
+  predicateConditions,
+  samePredicate,
+  STORE_TABLE_PRIVILEGES,
+} from "../src/store/postgres.js";
+import { REQUIRED_COLUMNS, REQUIRED_KEYS, REQUIRED_TABLES } from "../src/store/schema.js";
 
 describe("PostgresStore", () => {
   it("round-trips core auth records through postgres", async () => {
@@ -216,19 +222,27 @@ describe("createPostgresStore", () => {
 });
 
 describe("PostgresStore.checkSchema", () => {
-  it("derives every relation, typed column and unique key from the schema", () => {
-    expect(REQUIRED_RELATIONS).toEqual(
-      expect.arrayContaining(["accounts", "audit_events", "auth_completions_attempt_unconsumed_unique_idx", "audit_events_lookup_idx"])
-    );
-    expect(REQUIRED_RELATIONS).toHaveLength(17);
+  it("derives every table, typed column and unique key from the schema", () => {
+    expect(REQUIRED_TABLES).toEqual([
+      "accounts",
+      "linked_identities",
+      "sessions",
+      "auth_attempts",
+      "private_command_receipts",
+      "auth_completions",
+      "capability_grants",
+      "entitlement_snapshots",
+      "audit_events",
+    ]);
     expect(REQUIRED_COLUMNS).toHaveLength(85);
     expect(REQUIRED_COLUMNS).toEqual(
       expect.arrayContaining([
-        { table: "auth_completions", column: "code", type: "TEXT" },
-        { table: "sessions", column: "access_revision", type: "INTEGER" },
-        { table: "linked_identities", column: "profile_json", type: "JSONB" },
-        { table: "entitlement_snapshots", column: "is_allowed", type: "BOOLEAN" },
-        { table: "auth_completions", column: "attempt_id", type: "TEXT" },
+        { table: "auth_completions", column: "code", type: "TEXT", nullable: false },
+        { table: "sessions", column: "access_revision", type: "INTEGER", nullable: false },
+        { table: "linked_identities", column: "profile_json", type: "JSONB", nullable: false },
+        { table: "linked_identities", column: "username", type: "TEXT", nullable: true },
+        { table: "entitlement_snapshots", column: "is_allowed", type: "BOOLEAN", nullable: false },
+        { table: "auth_completions", column: "attempt_id", type: "TEXT", nullable: true },
       ])
     );
     expect(REQUIRED_KEYS).toHaveLength(12);
@@ -237,38 +251,43 @@ describe("PostgresStore.checkSchema", () => {
         { table: "linked_identities", columns: "provider,provider_user_id", predicate: "" },
         { table: "private_command_receipts", columns: "app_slug,idempotency_key", predicate: "" },
         { table: "entitlement_snapshots", columns: "account_id,provider,scope", predicate: "" },
-        {
-          table: "auth_completions",
-          columns: "app_slug,attempt_id",
-          predicate: comparablePredicate("attempt_id IS NOT NULL AND consumed_at IS NULL"),
-        },
+        { table: "auth_completions", columns: "app_slug,attempt_id", predicate: "attempt_id IS NOT NULL AND consumed_at IS NULL" },
       ])
     );
   });
 
-  it("compares a predicate in the form Postgres stores it back", () => {
-    expect(comparablePredicate("((attempt_id IS NOT NULL) AND (consumed_at IS NULL))")).toBe(
-      comparablePredicate("attempt_id IS NOT NULL AND consumed_at IS NULL")
-    );
+  // Soul pass 6: stripping parentheses and case merged predicates of
+  // different meaning. These are Postgres's own printed forms.
+  it("splits a printed predicate only at its top-level ANDs", () => {
+    expect(predicateConditions("((a OR b) AND c)")).toEqual(["a OR b", "c"]);
+    expect(predicateConditions("(a OR (b AND c))")).toEqual(["a OR (b AND c)"]);
+    expect(predicateConditions("((attempt_id IS NOT NULL) AND (consumed_at IS NULL))")).toEqual([
+      "attempt_id IS NOT NULL",
+      "consumed_at IS NULL",
+    ]);
+    expect(predicateConditions("(s = 'x AND (y'::text)")).toEqual(["s = 'x AND (y'::text"]);
+    expect(predicateConditions("")).toEqual([]);
   });
 
-  function storeAnswering(kinds: string[]) {
-    return new PostgresStore({
-      query: (async () => ({ rows: kinds.map((kind) => ({ kind })) })) as never,
-      end: async () => undefined,
-    });
-  }
-
-  it("refuses with SCHEMA_MISSING when anything is missing, whatever else is", async () => {
-    await expect(storeAnswering(["privilege", "missing"]).checkSchema()).rejects.toMatchObject({ code: "SCHEMA_MISSING" });
+  it("matches predicates by their conditions, never by case or by dropping structure", () => {
+    const keys = ["app_slug", "attempt_id"];
+    expect(samePredicate("((a OR b) AND c)", "(a OR (b AND c))", keys, false)).toBe(false);
+    expect(samePredicate("(s = 'Pending'::text)", "(s = 'pending'::text)", keys, false)).toBe(false);
+    expect(samePredicate("((a) AND (b))", "((b) AND (a))", keys, false)).toBe(true);
+    expect(samePredicate("((a) AND (b))", "((a) AND (b) AND (c))", keys, false)).toBe(false);
   });
 
-  it("refuses with SCHEMA_PRIVILEGES when only a privilege is missing", async () => {
-    await expect(storeAnswering(["privilege"]).checkSchema()).rejects.toMatchObject({ code: "SCHEMA_PRIVILEGES" });
+  it("lets a key column's IS NOT NULL be absent only where NULL keys cannot collide", () => {
+    const keys = ["app_slug", "attempt_id"];
+    const declared = "((attempt_id IS NOT NULL) AND (consumed_at IS NULL))";
+    expect(samePredicate(declared, "(consumed_at IS NULL)", keys, false)).toBe(true);
+    expect(samePredicate(declared, "(consumed_at IS NULL)", keys, true)).toBe(false);
+    expect(samePredicate(declared, "((consumed_at IS NULL) AND (other IS NOT NULL))", keys, false)).toBe(false);
   });
 
-  it("passes when nothing is missing", async () => {
-    await expect(storeAnswering([]).checkSchema()).resolves.toBeUndefined();
+  it("requires a key without a predicate to have none", () => {
+    expect(samePredicate("", "", ["id"], false)).toBe(true);
+    expect(samePredicate("", "(id IS NOT NULL)", ["id"], false)).toBe(false);
   });
 });
 
@@ -470,9 +489,7 @@ describe.skipIf(!adminUrl)("startup against a live Postgres (HEIMDALL_TEST_PG_AD
 
   it("refuses each privilege taken away, and the store needs each one", async () => {
     const pairs = Object.entries(STORE_TABLE_PRIVILEGES).flatMap(([table, wanted]) => wanted.map((privilege) => ({ table, privilege })));
-    expect(Object.keys(STORE_TABLE_PRIVILEGES).sort()).toEqual(
-      REQUIRED_RELATIONS.filter((relation) => REQUIRED_COLUMNS.some((column) => column.table === relation)).sort()
-    );
+    expect(Object.keys(STORE_TABLE_PRIVILEGES).sort()).toEqual([...REQUIRED_TABLES].sort());
     for (const { table, privilege } of pairs) {
       const { name, owner } = await appliedDatabase();
       await grant(owner, { ...STORE_TABLE_PRIVILEGES, [table]: STORE_TABLE_PRIVILEGES[table]!.filter((p) => p !== privilege) });
@@ -507,41 +524,106 @@ describe.skipIf(!adminUrl)("startup against a live Postgres (HEIMDALL_TEST_PG_AD
     expect(await start(urlFor(name), true)).toBe(refusal("SCHEMA_MISSING"));
   });
 
+  it("serves every store operation with each privilege granted column by column", async () => {
+    const { name, owner } = await appliedDatabase();
+    await owner.query(`GRANT USAGE ON SCHEMA public TO ${role}`);
+    for (const [table, wanted] of Object.entries(STORE_TABLE_PRIVILEGES)) {
+      const columns = REQUIRED_COLUMNS.filter((column) => column.table === table).map((column) => column.column).join(", ");
+      for (const privilege of wanted) await owner.query(`GRANT ${privilege} (${columns}) ON ${table} TO ${role}`);
+    }
+
+    expect(await start(urlFor(name, role, rolePassword), false)).toBe("started");
+    const pool = fixturePool(urlFor(name, role, rolePassword));
+    cleanup.push(async () => {
+      await pool.end();
+    });
+    await exerciseStore(new PostgresStore(pool));
+  });
+
+  it("refuses a privilege held on only some of a table's columns", async () => {
+    const { name, owner } = await appliedDatabase();
+    await grant(owner, { ...STORE_TABLE_PRIVILEGES, audit_events: [] });
+    await owner.query(`GRANT INSERT (id) ON audit_events TO ${role}`);
+
+    expect(await start(urlFor(name, role, rolePassword), false)).toBe(refusal("SCHEMA_PRIVILEGES"));
+  });
+
   // Each damage below leaves a database whose shape still resembles the
-  // schema but on which a store write fails or R21.1 no longer holds.
+  // schema but on which a store write fails or R21.1 no longer holds. A step
+  // starting with "!" is expected to fail and leave its mark (a failed
+  // CREATE INDEX CONCURRENTLY leaves an invalid index behind).
+  const seedDuplicates = [
+    "INSERT INTO accounts VALUES ('acc', now(), now(), null, null)",
+    "INSERT INTO sessions VALUES ('s', 'acc', 'ghostlight', now(), now(), now() + interval '1 hour', '{}', 1)",
+  ];
   it.each([
     ["a key made DEFERRABLE (ON CONFLICT refuses it as an arbiter)",
-      "ALTER TABLE private_command_receipts DROP CONSTRAINT private_command_receipts_pkey; ALTER TABLE private_command_receipts ADD PRIMARY KEY (app_slug, idempotency_key) DEFERRABLE INITIALLY IMMEDIATE"],
+      ["ALTER TABLE private_command_receipts DROP CONSTRAINT private_command_receipts_pkey", "ALTER TABLE private_command_receipts ADD PRIMARY KEY (app_slug, idempotency_key) DEFERRABLE INITIALLY IMMEDIATE"]],
     ["a key narrowed to one column, the other only INCLUDEd",
-      "ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key; CREATE UNIQUE INDEX li_include ON linked_identities(provider) INCLUDE (provider_user_id)"],
+      ["ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key", "CREATE UNIQUE INDEX li_include ON linked_identities(provider) INCLUDE (provider_user_id)"]],
     ["a key widened by an expression column",
-      "ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key; CREATE UNIQUE INDEX li_expr ON linked_identities(provider, provider_user_id, lower(username))"],
+      ["ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key", "CREATE UNIQUE INDEX li_expr ON linked_identities(provider, provider_user_id, lower(username))"]],
     ["a key replaced by a partial one",
-      "ALTER TABLE entitlement_snapshots DROP CONSTRAINT entitlement_snapshots_account_id_provider_scope_key; CREATE UNIQUE INDEX es_partial ON entitlement_snapshots(account_id, provider, scope) WHERE is_allowed"],
+      ["ALTER TABLE entitlement_snapshots DROP CONSTRAINT entitlement_snapshots_account_id_provider_scope_key", "CREATE UNIQUE INDEX es_partial ON entitlement_snapshots(account_id, provider, scope) WHERE is_allowed"]],
     ["a key replaced by a plain index on the same columns",
-      "ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key; CREATE INDEX li_plain ON linked_identities(provider, provider_user_id)"],
+      ["ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key", "CREATE INDEX li_plain ON linked_identities(provider, provider_user_id)"]],
     ["a key replaced by one on other columns of the same count",
-      "ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key; CREATE UNIQUE INDEX li_other ON linked_identities(provider, username)"],
+      ["ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key", "CREATE UNIQUE INDEX li_other ON linked_identities(provider, username)"]],
+    ["a key left only as an invalid index by a failed concurrent build",
+      [...seedDuplicates,
+        "ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key",
+        "INSERT INTO linked_identities (id, account_id, provider, provider_user_id, created_at, updated_at) VALUES ('l1', 'acc', 'youtube', 'dup', now(), now()), ('l2', 'acc', 'youtube', 'dup', now(), now())",
+        "!CREATE UNIQUE INDEX CONCURRENTLY li_ccnew ON linked_identities(provider, provider_user_id)",
+        "DELETE FROM linked_identities WHERE id = 'l2'"]],
+    ["the attempt index left invalid over duplicate attempts by a failed concurrent build",
+      [...seedDuplicates,
+        "DROP INDEX auth_completions_attempt_unconsumed_unique_idx",
+        "INSERT INTO auth_completions (code, attempt_id, app_slug, provider, mode, account_id, session_id, return_to, payload_json, created_at, expires_at) VALUES ('c1', 'att', 'ghostlight', 'discord', 'sign_in', 'acc', 's', 'https://x/', '{}', now(), now() + interval '1 hour'), ('c2', 'att', 'ghostlight', 'discord', 'sign_in', 'acc', 's', 'https://x/', '{}', now(), now() + interval '1 hour')",
+        "!CREATE UNIQUE INDEX CONCURRENTLY auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id) WHERE attempt_id IS NOT NULL AND consumed_at IS NULL"]],
+    ["a key compared under a case-insensitive collation (two users' ids become one row)",
+      ["CREATE COLLATION heimdall_ci (provider = icu, locale = 'und-u-ks-level2', deterministic = false)",
+        "ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key",
+        "CREATE UNIQUE INDEX li_ci ON linked_identities(provider, provider_user_id COLLATE heimdall_ci)"]],
     ["the attempt index replaced by a plain one of the same name",
-      "DROP INDEX auth_completions_attempt_unconsumed_unique_idx; CREATE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id)"],
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id)"]],
     ["the attempt index made unique on every row (no predicate)",
-      "DROP INDEX auth_completions_attempt_unconsumed_unique_idx; CREATE UNIQUE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id)"],
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE UNIQUE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id)"]],
     ["the attempt index given another predicate",
-      "DROP INDEX auth_completions_attempt_unconsumed_unique_idx; CREATE UNIQUE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id) WHERE attempt_id IS NOT NULL"],
-    ["a column retyped", "ALTER TABLE sessions ALTER COLUMN access_revision TYPE TEXT"],
-    ["a created table's column dropped", "ALTER TABLE auth_completions DROP COLUMN mode"],
-  ])("refuses %s", async (_damage, sql) => {
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE UNIQUE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id) WHERE attempt_id IS NOT NULL"]],
+    ["the attempt index narrowed by an extra condition",
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE UNIQUE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id) WHERE attempt_id IS NOT NULL AND consumed_at IS NULL AND app_slug <> 'ghostlight'"]],
+    ["the attempt index with a condition ORed into its predicate",
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE UNIQUE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id) WHERE (attempt_id IS NOT NULL OR app_slug = 'x') AND consumed_at IS NULL"]],
+    ["the attempt index made NULLS NOT DISTINCT without excluding NULL attempts",
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE UNIQUE INDEX auth_completions_attempt_unconsumed_unique_idx ON auth_completions(app_slug, attempt_id) NULLS NOT DISTINCT WHERE consumed_at IS NULL"]],
+    ["a column retyped", ["ALTER TABLE sessions ALTER COLUMN access_revision TYPE TEXT"]],
+    ["a created table's column dropped", ["ALTER TABLE auth_completions DROP COLUMN mode"]],
+    ["NOT NULL added to a column the store leaves null", ["ALTER TABLE auth_completions ALTER COLUMN attempt_id SET NOT NULL"]],
+  ])("refuses %s", async (_damage, steps) => {
     const { name, owner } = await appliedDatabase();
-    await owner.query(sql);
+    for (const step of steps) {
+      if (step.startsWith("!")) await expect(owner.query(step.slice(1))).rejects.toBeDefined();
+      else await owner.query(step);
+    }
 
     expect(await start(urlFor(name), false)).toBe(refusal("SCHEMA_MISSING"));
   });
 
-  it("accepts a key index that carries extra INCLUDE columns", async () => {
+  // Each change below leaves a database the store serves exactly as before.
+  it.each([
+    ["a key index that carries extra INCLUDE columns",
+      ["ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key", "CREATE UNIQUE INDEX li_covering ON linked_identities(provider, provider_user_id) INCLUDE (username)"]],
+    ["the attempt index under another name", ["ALTER INDEX auth_completions_attempt_unconsumed_unique_idx RENAME TO attempt_idx_renamed"]],
+    ["the attempt index's conditions in the other order",
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE UNIQUE INDEX attempt_reordered ON auth_completions(app_slug, attempt_id) WHERE consumed_at IS NULL AND attempt_id IS NOT NULL"]],
+    // NULL attempt ids never collide in a unique index that treats NULLs as
+    // distinct (the default), so dropping that condition keeps the same rows unique.
+    ["the attempt index without its redundant attempt_id IS NOT NULL",
+      ["DROP INDEX auth_completions_attempt_unconsumed_unique_idx", "CREATE UNIQUE INDEX attempt_unconsumed ON auth_completions(app_slug, attempt_id) WHERE consumed_at IS NULL"]],
+    ["a lookup index dropped", ["DROP INDEX audit_events_lookup_idx", "DROP INDEX auth_completions_lookup_idx"]],
+  ])("accepts %s", async (_change, steps) => {
     const { name, owner } = await appliedDatabase();
-    await owner.query(
-      "ALTER TABLE linked_identities DROP CONSTRAINT linked_identities_provider_provider_user_id_key; CREATE UNIQUE INDEX li_covering ON linked_identities(provider, provider_user_id) INCLUDE (username)"
-    );
+    for (const step of steps) await owner.query(step);
 
     expect(await start(urlFor(name), false)).toBe("started");
   });