feat(db): V6 creation schema 地基 + 补回 posts.generation_job_id 外键（T4-01）

creation schema 三表（generation_models / generation_styles /
generation_jobs）逐列取自已评审目标模型 patbond_postgresql.sql
（schema 60/68 + 对象 556-717），含 14 索引与 3 个 updated_at 触发器。

- 零外键需要裁剪：这是第一个不用砍跨 schema 外键的 baseline 迁移，
  identity/media（V1）与 pet_health（V3）目标 schema 均已存在
- 按 V5 文件头指令「M4 补回」补回 community.posts.generation_job_id
  → creation.generation_jobs(id) ON DELETE SET NULL；约束名显式写为
  posts_generation_job_id_fkey，即目标模型内联声明时 PostgreSQL 会
  自动生成的名字，使 V1..V6 建库与照目标模型建库约束名一致
- posts.region_id 保持裸列（区域系统属 M5，ADR-018 裁剪，M5 补回）
- 两条偏索引对齐 worker 语句：ix_generation_jobs_queue 的列序与出队
  ORDER BY 完全一致（SKIP LOCKED 领取走索引扫描不排序）；
  ix_generation_jobs_running 驱动租约回收与待轮询扫描

两处刻意偏离目标模型，文件头已逐条记录原因：
1. input_asset_id 改可空（目标模型 NOT NULL）——参考图选填拍板；原
   NOT NULL 会使文生图无法表达。未发明 requires_input 标记（不超出
   已评审模型），「某模型是否必须有输入」由应用层查目录判断
2. 新增 first_attempt_at（目标模型无此列）——iteration-4/06 §4.2 指标
   口径要求。目标模型无法表达：其 ck_generation_jobs_state 强制
   queued 分支 started_at IS NULL，故重排队必然擦除首次尝试时间，
   导致排队时长与纯执行耗时永久无法分离。该列仅首次进入 running 时
   写入、不覆盖，并刻意排除在 queued 分支的置空要求之外；
   ck_generation_jobs_first_attempt 钉住其与 attempt_count 的不变式

测试：
- 新增 CreationMigrationIntegrationTest（12 项）。重点覆盖「日后同步回
  目标模型会静默改回去」的行为：复合外键真实拦截跨 media_kind 引用、
  queued→running→failed→queued 全程后 first_attempt_at 保留而
  started_at 被清空、不变式双向拒绝、provider_request_id 部分唯一索引
  （NULL 可共存、同值冲突）、idempotency_key 的按用户作用域
- CommunityMigrationIntegrationTest 原断言 %generation% 与 %region%
  外键合计为 0，V6 后必然变红（iteration-4/02 已预警）。拆为两条独立
  断言（generation=1、region=0）而非改成合计 1，使任一侧回退可单独定位

全量 381 → 393，零失败零跳过（auth 40 / common 3 / community 107 /
pet 100 / user 131→143）。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

patbond-user/src/main/resources/db/migration/V6__creation_baseline.sql

@@ -0,0 +1,263 @@
+/*
+ V6 baseline: creation schema (M4 AI 创作), extracted from
+ patbond-doc/docs/database/patbond_postgresql.sql (reviewed target model,
+ lines 60/68 schema + 556-717 objects) per iteration-4 T4-01.
+
+ Scope notes:
+ - All 3 creation objects are created here: generation_models,
+   generation_styles, generation_jobs. Both media kinds ('image', 'video')
+   are structurally supported from the start, as modeled — the target model
+   keys the catalog by (id, media_kind) composite unique so a job can only
+   reference a model/style of its own kind. Video feature rollout is a
+   contract/client concern, not a schema concern (see ADR-018 supersede).
+ - Zero cross-schema FKs need stripping, unlike V3 and V5: every target
+   schema already exists — identity (V1), media (V1), pet_health (V3).
+   This is the first baseline migration that lands its FKs complete.
+ - V5's header registered two deferred constraints. This migration settles
+   the first one, as instructed there ("M4 补回"):
+     * community.posts.generation_job_id -> creation.generation_jobs(id)
+       ON DELETE SET NULL is re-added at the end of this file. The constraint
+       is named explicitly as posts_generation_job_id_fkey so that a database
+       built by V1..V6 is name-identical to one built from the target model
+       (where the FK is declared inline and auto-named by PostgreSQL).
+     * community.posts.region_id -> platform.regions(id) stays STRIPPED;
+       the region system belongs to M5 (ADR-018 scope cut) — M5 补回.
+ - Primary keys keep DEFAULT gen_random_uuid() as modeled (V1/V3/V5 同规);
+   application code supplies explicit UUIDv7 ids, the default is a fallback.
+ - updated_at triggers reuse platform.set_updated_at() (function from V1).
+
+ Two deliberate deviations from the target model, both decided in iteration-4
+ and recorded here so the divergence is never silent:
+
+ 1. generation_jobs.input_asset_id is NULLABLE (target model: NOT NULL).
+    Decision A2: the reference image is optional — mainstream image and video
+    providers support both text-to-* and reference-guided generation, so the
+    presence of an input asset is what selects between them at request time.
+    The target model's NOT NULL would have made text-to-image unrepresentable.
+    Whether a specific model *requires* an input is deliberately NOT encoded
+    here (no requires_input flag invented beyond the reviewed model); it is
+    enforced by the application against the catalog.
+
+ 2. generation_jobs.first_attempt_at is ADDED (absent from the target model).
+    Required by iteration-4/06 §4.2: queue wait and pure execution time can
+    only be separated if the first attempt's start survives retries. The
+    target model cannot express this — its ck_generation_jobs_state forces
+    started_at IS NULL on the 'queued' branch, so re-queueing a failed job
+    necessarily erases the original start. first_attempt_at is therefore
+    written once (on the first transition into 'running') and never
+    overwritten, and is deliberately EXCLUDED from the 'queued' branch's
+    null-out requirements. ck_generation_jobs_first_attempt pins the
+    invariant against attempt_count.
+*/
+
+CREATE SCHEMA creation;
+
+COMMENT ON SCHEMA creation IS 'Asynchronous AI image/video generation';
+
+CREATE TABLE creation.generation_models (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    code varchar(64) NOT NULL,
+    display_name varchar(128) NOT NULL,
+    provider_code varchar(64) NOT NULL,
+    provider_model_name varchar(128) NOT NULL,
+    media_kind varchar(16) NOT NULL,
+    enabled boolean NOT NULL DEFAULT true,
+    sort_order integer NOT NULL DEFAULT 0,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now(),
+    UNIQUE (code, media_kind),
+    -- (id, media_kind) is redundant given the PK, but is the target of the
+    -- composite FK from generation_jobs: it is what makes "a job's model must
+    -- be of the job's own media_kind" enforceable by the database.
+    UNIQUE (id, media_kind),
+    CONSTRAINT ck_generation_models_kind CHECK (media_kind IN ('image', 'video')),
+    CONSTRAINT ck_generation_models_names CHECK (
+        code = btrim(code) AND char_length(code) BETWEEN 2 AND 64
+        AND char_length(btrim(display_name)) BETWEEN 1 AND 128
+    )
+);
+
+CREATE INDEX ix_generation_models_kind_order
+    ON creation.generation_models (media_kind, sort_order, id)
+    WHERE enabled;
+
+CREATE TABLE creation.generation_styles (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    code varchar(64) NOT NULL,
+    title varchar(64) NOT NULL,
+    subtitle varchar(128),
+    media_kind varchar(16) NOT NULL,
+    preview_asset_id uuid REFERENCES media.assets(id) ON DELETE SET NULL,
+    enabled boolean NOT NULL DEFAULT true,
+    sort_order integer NOT NULL DEFAULT 0,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now(),
+    UNIQUE (code, media_kind),
+    UNIQUE (id, media_kind),
+    CONSTRAINT ck_generation_styles_kind CHECK (media_kind IN ('image', 'video')),
+    CONSTRAINT ck_generation_styles_names CHECK (
+        code = btrim(code) AND char_length(code) BETWEEN 2 AND 64
+        AND char_length(btrim(title)) BETWEEN 1 AND 64
+    )
+);
+
+CREATE INDEX ix_generation_styles_preview ON creation.generation_styles (preview_asset_id);
+CREATE INDEX ix_generation_styles_kind_order
+    ON creation.generation_styles (media_kind, sort_order, id)
+    WHERE enabled;
+
+CREATE TABLE creation.generation_jobs (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    user_id uuid NOT NULL REFERENCES identity.users(id) ON DELETE RESTRICT,
+    pet_id uuid REFERENCES pet_health.pets(id) ON DELETE SET NULL,
+    media_kind varchar(16) NOT NULL,
+    model_id uuid NOT NULL,
+    style_id uuid,
+    -- DEVIATION 1 (see header): nullable, not NOT NULL. NULL = text-to-*,
+    -- present = reference-guided generation.
+    input_asset_id uuid REFERENCES media.assets(id) ON DELETE RESTRICT,
+    output_asset_id uuid REFERENCES media.assets(id) ON DELETE SET NULL,
+    prompt varchar(10000),
+    negative_prompt varchar(3000),
+    width_px integer,
+    height_px integer,
+    duration_ms bigint,
+    upscale boolean NOT NULL DEFAULT false,
+    parameters jsonb NOT NULL DEFAULT '{}'::jsonb,
+    -- Snapshots freeze what was actually asked of the provider: the catalog
+    -- may be re-pointed or disabled later, historical jobs must stay readable.
+    provider_code_snapshot varchar(64) NOT NULL,
+    provider_model_snapshot varchar(128) NOT NULL,
+    model_version_snapshot varchar(64) NOT NULL,
+    style_code_snapshot varchar(64),
+    status varchar(16) NOT NULL DEFAULT 'queued',
+    progress smallint NOT NULL DEFAULT 0,
+    priority smallint NOT NULL DEFAULT 0,
+    attempt_count smallint NOT NULL DEFAULT 0,
+    max_attempts smallint NOT NULL DEFAULT 3,
+    provider_request_id varchar(128),
+    error_code varchar(64),
+    error_message varchar(1000),
+    idempotency_key varchar(128) NOT NULL,
+    request_hash bytea NOT NULL,
+    next_attempt_at timestamptz NOT NULL DEFAULT now(),
+    lease_owner varchar(128),
+    lease_expires_at timestamptz,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now(),
+    -- DEVIATION 2 (see header): written once on the first transition into
+    -- 'running', never overwritten, survives re-queueing.
+    first_attempt_at timestamptz,
+    started_at timestamptz,
+    completed_at timestamptz,
+    version integer NOT NULL DEFAULT 0,
+    UNIQUE (user_id, idempotency_key),
+    FOREIGN KEY (model_id, media_kind)
+        REFERENCES creation.generation_models(id, media_kind) ON DELETE RESTRICT,
+    FOREIGN KEY (style_id, media_kind)
+        REFERENCES creation.generation_styles(id, media_kind) ON DELETE RESTRICT,
+    CONSTRAINT ck_generation_jobs_kind CHECK (media_kind IN ('image', 'video')),
+    CONSTRAINT ck_generation_jobs_dimensions CHECK (
+        (width_px IS NULL OR width_px BETWEEN 64 AND 8192)
+        AND (height_px IS NULL OR height_px BETWEEN 64 AND 8192)
+        AND (duration_ms IS NULL OR duration_ms > 0)
+    ),
+    CONSTRAINT ck_generation_jobs_parameters CHECK (jsonb_typeof(parameters) = 'object'),
+    CONSTRAINT ck_generation_jobs_idempotency CHECK (
+        idempotency_key = btrim(idempotency_key)
+        AND char_length(idempotency_key) BETWEEN 1 AND 128
+        AND octet_length(request_hash) = 32
+    ),
+    CONSTRAINT ck_generation_jobs_status CHECK (
+        status IN ('queued', 'running', 'succeeded', 'failed', 'cancelled')
+    ),
+    CONSTRAINT ck_generation_jobs_progress CHECK (progress BETWEEN 0 AND 100),
+    CONSTRAINT ck_generation_jobs_attempts CHECK (
+        attempt_count >= 0 AND max_attempts > 0 AND attempt_count <= max_attempts
+    ),
+    CONSTRAINT ck_generation_jobs_state CHECK (
+        (
+            status = 'queued' AND progress = 0 AND started_at IS NULL AND completed_at IS NULL
+            AND output_asset_id IS NULL AND error_code IS NULL
+            AND lease_owner IS NULL AND lease_expires_at IS NULL
+        )
+        OR (
+            status = 'running' AND started_at IS NOT NULL AND completed_at IS NULL
+            AND output_asset_id IS NULL AND error_code IS NULL
+            AND lease_owner IS NOT NULL AND lease_expires_at IS NOT NULL
+        )
+        OR (
+            status = 'succeeded' AND progress = 100 AND started_at IS NOT NULL
+            AND completed_at IS NOT NULL AND output_asset_id IS NOT NULL AND error_code IS NULL
+            AND lease_owner IS NULL AND lease_expires_at IS NULL
+        )
+        OR (
+            status = 'failed' AND completed_at IS NOT NULL AND error_code IS NOT NULL
+            AND output_asset_id IS NULL AND lease_owner IS NULL AND lease_expires_at IS NULL
+        )
+        OR (
+            status = 'cancelled' AND completed_at IS NOT NULL
+            AND lease_owner IS NULL AND lease_expires_at IS NULL
+        )
+    ),
+    CONSTRAINT ck_generation_jobs_lease CHECK (
+        (lease_owner IS NULL) = (lease_expires_at IS NULL)
+        AND (lease_expires_at IS NULL OR started_at IS NULL OR lease_expires_at > started_at)
+    ),
+    -- DEVIATION 2's invariant: first_attempt_at exists exactly when the job
+    -- has been attempted at least once, and never postdates the current
+    -- attempt's start. Deliberately absent from ck_generation_jobs_state's
+    -- 'queued' branch: a re-queued job keeps it.
+    CONSTRAINT ck_generation_jobs_first_attempt CHECK (
+        (attempt_count = 0) = (first_attempt_at IS NULL)
+        AND (
+            first_attempt_at IS NULL OR started_at IS NULL
+            OR first_attempt_at <= started_at
+        )
+    ),
+    CONSTRAINT ck_generation_jobs_version CHECK (version >= 0)
+);
+
+-- Provider-side de-duplication: the pair (provider, provider_request_id) is
+-- what makes "did I already submit this upstream?" answerable after a timeout
+-- or a worker restart — i.e. the basis of not paying twice.
+CREATE UNIQUE INDEX uq_generation_jobs_provider_request
+    ON creation.generation_jobs (provider_code_snapshot, provider_request_id)
+    WHERE provider_request_id IS NOT NULL;
+CREATE INDEX ix_generation_jobs_user_created
+    ON creation.generation_jobs (user_id, created_at DESC, id DESC);
+CREATE INDEX ix_generation_jobs_pet ON creation.generation_jobs (pet_id);
+CREATE INDEX ix_generation_jobs_model ON creation.generation_jobs (model_id);
+CREATE INDEX ix_generation_jobs_style ON creation.generation_jobs (style_id);
+CREATE INDEX ix_generation_jobs_model_kind ON creation.generation_jobs (model_id, media_kind);
+CREATE INDEX ix_generation_jobs_style_kind ON creation.generation_jobs (style_id, media_kind);
+CREATE INDEX ix_generation_jobs_input ON creation.generation_jobs (input_asset_id);
+CREATE INDEX ix_generation_jobs_output ON creation.generation_jobs (output_asset_id);
+-- Column order matches the dequeue statement's ORDER BY exactly, so the
+-- FOR UPDATE SKIP LOCKED claim is an index scan, not a sort.
+CREATE INDEX ix_generation_jobs_queue
+    ON creation.generation_jobs (priority DESC, next_attempt_at, created_at, id)
+    WHERE status = 'queued';
+-- Drives the lease reaper and the poll-due scan over in-flight jobs.
+CREATE INDEX ix_generation_jobs_running
+    ON creation.generation_jobs (lease_expires_at, id)
+    WHERE status = 'running';
+
+-- Automatic updated_at maintenance (function created in V1). Business version
+-- increments remain explicit so optimistic locking stays visible in
+-- repository update statements.
+CREATE TRIGGER trg_generation_models_updated_at BEFORE UPDATE ON creation.generation_models
+    FOR EACH ROW EXECUTE FUNCTION platform.set_updated_at();
+CREATE TRIGGER trg_generation_styles_updated_at BEFORE UPDATE ON creation.generation_styles
+    FOR EACH ROW EXECUTE FUNCTION platform.set_updated_at();
+CREATE TRIGGER trg_generation_jobs_updated_at BEFORE UPDATE ON creation.generation_jobs
+    FOR EACH ROW EXECUTE FUNCTION platform.set_updated_at();
+
+-- V5's deferred constraint, settled per its header instruction ("M4 补回").
+-- Named explicitly to match the auto-generated name of the target model's
+-- inline declaration, so V1..V6 and the target model stay name-identical.
+-- community.posts.region_id stays bare — that one is M5's to settle.
+ALTER TABLE community.posts
+    ADD CONSTRAINT posts_generation_job_id_fkey
+    FOREIGN KEY (generation_job_id)
+    REFERENCES creation.generation_jobs(id) ON DELETE SET NULL;

patbond-user/src/test/java/com/patbond/patbond/user/persistence/CommunityMigrationIntegrationTest.java

@@ -12,10 +12,14 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
  * Flyway V5 (community schema baseline) applies cleanly on a postgres:18
  * container after V1..V4 and produces the expected structure. Validates the
- * mandatory cross-schema FK cuts (ADR-017/ADR-018, T3-01):
- * generation_job_id (creation belongs to M4) and region_id (regions belong
- * to the M5 region system) exist as bare nullable uuid columns without FKs,
+ * cross-schema FK cuts declared in V5's header (ADR-017/ADR-018, T3-01):
+ * both generation_job_id and region_id exist as bare nullable uuid columns,
  * and the pg_trgm extension missing from V1 is now installed.
+ *
+ * <p>Note on the FK assertions: V5 stripped both FKs, but V6 settled the
+ * creation one ("M4 补回"). Since these tests run the full migration chain,
+ * generation_job_id is asserted to HAVE its FK and region_id to still lack
+ * one — see {@code CreationMigrationIntegrationTest} for the V6 side.
  */
 @SpringBootTest
 @Import(TestcontainersConfiguration.class)
@@ -65,7 +69,7 @@ class CommunityMigrationIntegrationTest {
     }
 
     @Test
-    void v5PostsColumnsExistWithoutCreationAndRegionFKs() {
+    void v5PostsColumnsExistAndOnlyRegionFKStaysStripped() {
         // generation_job_id and region_id columns exist (bare nullable uuid)
         int colCount = jdbcClient.sql(
                         "SELECT COUNT(*) FROM information_schema.columns " +
@@ -76,16 +80,29 @@ class CommunityMigrationIntegrationTest {
                 .single();
         assertThat(colCount).isEqualTo(2);
 
-        // No FK constraints targeting creation.generation_jobs (M4) or
-        // platform.regions (M5 region system)
-        int fkCount = jdbcClient.sql(
+        // V6 settled the creation FK (V5 header: "M4 补回"), so after the full
+        // V1..V6 chain generation_job_id DOES carry its FK. Asserted per
+        // constraint so a future regression on either side is distinguishable
+        // rather than hidden in a combined count.
+        int generationFk = jdbcClient.sql(
                         "SELECT COUNT(*) FROM information_schema.table_constraints " +
                                 "WHERE table_schema = 'community' AND table_name = 'posts' " +
                                 "AND constraint_type = 'FOREIGN KEY' " +
-                                "AND (constraint_name LIKE '%generation%' OR constraint_name LIKE '%region%')")
+                                "AND constraint_name LIKE '%generation%'")
                 .query(Integer.class)
                 .single();
-        assertThat(fkCount).isEqualTo(0);
+        assertThat(generationFk).isEqualTo(1);
+
+        // region_id stays bare: the region system belongs to M5 (ADR-018) —
+        // M5 补回. This must remain 0 until that migration lands.
+        int regionFk = jdbcClient.sql(
+                        "SELECT COUNT(*) FROM information_schema.table_constraints " +
+                                "WHERE table_schema = 'community' AND table_name = 'posts' " +
+                                "AND constraint_type = 'FOREIGN KEY' " +
+                                "AND constraint_name LIKE '%region%'")
+                .query(Integer.class)
+                .single();
+        assertThat(regionFk).isEqualTo(0);
     }
 
     @Test

patbond-user/src/test/java/com/patbond/patbond/user/persistence/CreationMigrationIntegrationTest.java

@@ -0,0 +1,313 @@
+package com.patbond.patbond.user.persistence;
+
+import com.patbond.patbond.user.TestcontainersConfiguration;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+import org.springframework.jdbc.core.simple.JdbcClient;
+import org.springframework.dao.DataIntegrityViolationException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * Flyway V6 (creation schema baseline) applies cleanly on a postgres:18
+ * container after V1..V5 and produces the expected structure.
+ *
+ * <p>Beyond structural checks this pins the two deliberate deviations from the
+ * reviewed target model, because both are the kind of thing a later "sync back
+ * to the target model" pass would silently revert:
+ * <ul>
+ *   <li>{@code input_asset_id} is nullable (decision A2: text-to-* must be
+ *       representable);</li>
+ *   <li>{@code first_attempt_at} exists and survives re-queueing (iteration-4/06
+ *       §4.2: queue wait and execution time must be separable).</li>
+ * </ul>
+ *
+ * <p>The retry-path test is the important one: the target model's state CHECK
+ * forces {@code started_at IS NULL} while queued, so without a dedicated
+ * column the first attempt's timestamp is destroyed on every retry.
+ */
+@SpringBootTest
+@Import(TestcontainersConfiguration.class)
+class CreationMigrationIntegrationTest {
+
+    @Autowired
+    private JdbcClient jdbcClient;
+
+    @Test
+    void v6CreatesCreationSchema() {
+        String exists = jdbcClient.sql(
+                        "SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'creation'")
+                .query(String.class)
+                .optional()
+                .orElse(null);
+        assertThat(exists).isEqualTo("creation");
+    }
+
+    @Test
+    void v6CreatesAllCreationTables() {
+        int count = jdbcClient.sql(
+                        "SELECT COUNT(*) FROM information_schema.tables " +
+                                "WHERE table_schema = 'creation' AND table_type = 'BASE TABLE'")
+                .query(Integer.class)
+                .single();
+        // 3 tables: generation_models, generation_styles, generation_jobs
+        assertThat(count).isEqualTo(3);
+    }
+
+    @Test
+    void v6LandsEveryCrossSchemaFkWithNothingStripped() {
+        // Unlike V3 and V5, V6 needs no FK cuts: identity and media exist
+        // since V1, pet_health since V3. Counts the FKs out of creation.
+        int outbound = jdbcClient.sql(
+                        "SELECT COUNT(*) FROM information_schema.table_constraints tc " +
+                                "JOIN information_schema.constraint_column_usage ccu " +
+                                "  ON tc.constraint_name = ccu.constraint_name " +
+                                " AND tc.constraint_schema = ccu.constraint_schema " +
+                                "WHERE tc.table_schema = 'creation' " +
+                                "AND tc.constraint_type = 'FOREIGN KEY' " +
+                                "AND ccu.table_schema IN ('identity', 'media', 'pet_health')")
+                .query(Integer.class)
+                .single();
+        // identity.users (jobs.user_id), pet_health.pets (jobs.pet_id),
+        // media.assets x3 (styles.preview_asset_id, jobs.input/output_asset_id)
+        assertThat(outbound).isEqualTo(5);
+    }
+
+    @Test
+    void v6ReAddsPostsGenerationJobFkUnderTheTargetModelName() {
+        // V5 header instruction "M4 补回", and the explicit constraint name
+        // keeps a V1..V6 database name-identical to a target-model build.
+        int named = jdbcClient.sql(
+                        "SELECT COUNT(*) FROM information_schema.table_constraints " +
+                                "WHERE table_schema = 'community' AND table_name = 'posts' " +
+                                "AND constraint_type = 'FOREIGN KEY' " +
+                                "AND constraint_name = 'posts_generation_job_id_fkey'")
+                .query(Integer.class)
+                .single();
+        assertThat(named).isEqualTo(1);
+    }
+
+    @Test
+    void v6CompositeFkForcesModelToMatchJobMediaKind() {
+        // A video job must not reference an image model. Enforced by the
+        // (id, media_kind) composite FK, not by application code.
+        String userId = insertUser("kindguard@example.com");
+        String imageModelId = insertModel("kg-image-model", "image");
+
+        assertThatThrownBy(() -> insertJob(userId, imageModelId, "video", null))
+                .isInstanceOf(DataIntegrityViolationException.class);
+
+        assertThatCode(() -> insertJob(userId, imageModelId, "image", null))
+                .doesNotThrowAnyException();
+    }
+
+    // --- deviation 1: input_asset_id nullable (text-to-*) ---
+
+    @Test
+    void v6AllowsJobWithoutInputAssetSoTextToImageIsRepresentable() {
+        String isNullable = jdbcClient.sql(
+                        "SELECT is_nullable FROM information_schema.columns " +
+                                "WHERE table_schema = 'creation' AND table_name = 'generation_jobs' " +
+                                "AND column_name = 'input_asset_id'")
+                .query(String.class)
+                .single();
+        assertThat(isNullable).isEqualTo("YES");
+
+        String userId = insertUser("texttoimage@example.com");
+        String modelId = insertModel("t2i-model", "image");
+        assertThatCode(() -> insertJob(userId, modelId, "image", null))
+                .doesNotThrowAnyException();
+    }
+
+    // --- deviation 2: first_attempt_at survives retries ---
+
+    @Test
+    void v6FirstAttemptAtSurvivesRequeueWhileStartedAtIsCleared() {
+        String userId = insertUser("retrypath@example.com");
+        String modelId = insertModel("retry-model", "image");
+        String jobId = insertJob(userId, modelId, "image", null);
+
+        // queued -> running (first attempt): both timestamps set together
+        jdbcClient.sql(
+                        "UPDATE creation.generation_jobs SET status = 'running', attempt_count = 1, " +
+                                "first_attempt_at = now(), started_at = now(), " +
+                                "lease_owner = 'w1', lease_expires_at = now() + interval '60 seconds' " +
+                                "WHERE id = ?::uuid")
+                .param(jobId)
+                .update();
+
+        // running -> failed
+        jdbcClient.sql(
+                        "UPDATE creation.generation_jobs SET status = 'failed', completed_at = now(), " +
+                                "error_code = 'PROVIDER_TIMEOUT', lease_owner = NULL, lease_expires_at = NULL " +
+                                "WHERE id = ?::uuid")
+                .param(jobId)
+                .update();
+
+        // failed -> queued (retry): the state CHECK forces started_at back to
+        // NULL. first_attempt_at must NOT be cleared — that is the whole point
+        // of the added column.
+        jdbcClient.sql(
+                        "UPDATE creation.generation_jobs SET status = 'queued', progress = 0, " +
+                                "started_at = NULL, completed_at = NULL, error_code = NULL, " +
+                                "next_attempt_at = now() + interval '30 seconds' " +
+                                "WHERE id = ?::uuid")
+                .param(jobId)
+                .update();
+
+        Boolean firstAttemptRetained = jdbcClient.sql(
+                        "SELECT first_attempt_at IS NOT NULL AND started_at IS NULL " +
+                                "FROM creation.generation_jobs WHERE id = ?::uuid")
+                .param(jobId)
+                .query(Boolean.class)
+                .single();
+        assertThat(firstAttemptRetained).isTrue();
+    }
+
+    @Test
+    void v6RejectsFirstAttemptAtInconsistentWithAttemptCount() {
+        String userId = insertUser("invariant@example.com");
+        String modelId = insertModel("invariant-model", "image");
+        String jobId = insertJob(userId, modelId, "image", null);
+
+        // attempt_count > 0 without first_attempt_at is exactly the state the
+        // target model would have left behind after a retry.
+        assertThatThrownBy(() -> jdbcClient.sql(
+                        "UPDATE creation.generation_jobs SET attempt_count = 1 WHERE id = ?::uuid")
+                .param(jobId)
+                .update())
+                .isInstanceOf(DataIntegrityViolationException.class);
+
+        // ...and the mirror case: a timestamp with no attempt behind it.
+        assertThatThrownBy(() -> jdbcClient.sql(
+                        "UPDATE creation.generation_jobs SET first_attempt_at = now() WHERE id = ?::uuid")
+                .param(jobId)
+                .update())
+                .isInstanceOf(DataIntegrityViolationException.class);
+    }
+
+    // --- queue mechanics the worker depends on ---
+
+    @Test
+    void v6CreatesPartialIndexesBackingTheLeaseQueue() {
+        int queue = indexCount("ix_generation_jobs_queue");
+        int running = indexCount("ix_generation_jobs_running");
+        assertThat(queue).isEqualTo(1);
+        assertThat(running).isEqualTo(1);
+
+        // The dequeue index must be partial on 'queued', otherwise the
+        // SKIP LOCKED claim scans terminal rows forever as history grows.
+        String queueDef = jdbcClient.sql(
+                        "SELECT indexdef FROM pg_indexes WHERE schemaname = 'creation' " +
+                                "AND indexname = 'ix_generation_jobs_queue'")
+                .query(String.class)
+                .single();
+        assertThat(queueDef).contains("WHERE").contains("queued");
+    }
+
+    @Test
+    void v6EnforcesProviderRequestUniquenessOnlyWhenPresent() {
+        // Two un-submitted jobs (NULL provider_request_id) must coexist; the
+        // partial unique index is what makes "don't pay twice" enforceable
+        // without blocking the pre-submit state.
+        String userId = insertUser("providerdedup@example.com");
+        String modelId = insertModel("dedup-model", "image");
+        assertThatCode(() -> {
+            insertJob(userId, modelId, "image", null);
+            insertJob(userId, modelId, "image", null);
+        }).doesNotThrowAnyException();
+
+        String firstId = insertJob(userId, modelId, "image", null);
+        String secondId = insertJob(userId, modelId, "image", null);
+        jdbcClient.sql("UPDATE creation.generation_jobs SET provider_code_snapshot = 'p1', " +
+                        "provider_request_id = 'req-1' WHERE id = ?::uuid")
+                .param(firstId)
+                .update();
+
+        assertThatThrownBy(() -> jdbcClient.sql(
+                        "UPDATE creation.generation_jobs SET provider_code_snapshot = 'p1', " +
+                                "provider_request_id = 'req-1' WHERE id = ?::uuid")
+                .param(secondId)
+                .update())
+                .isInstanceOf(DataIntegrityViolationException.class);
+    }
+
+    @Test
+    void v6EnforcesPerUserIdempotencyKey() {
+        String userId = insertUser("idem@example.com");
+        String modelId = insertModel("idem-model", "image");
+        insertJob(userId, modelId, "image", "shared-key");
+
+        assertThatThrownBy(() -> insertJob(userId, modelId, "image", "shared-key"))
+                .isInstanceOf(DataIntegrityViolationException.class);
+
+        // Same key under a different user is fine — the constraint is scoped.
+        String otherUser = insertUser("idem-other@example.com");
+        assertThatCode(() -> insertJob(otherUser, modelId, "image", "shared-key"))
+                .doesNotThrowAnyException();
+    }
+
+    @Test
+    void v6MaintainsUpdatedAtByTrigger() {
+        String modelId = insertModel("touch-model", "image");
+        jdbcClient.sql("UPDATE creation.generation_models SET sort_order = 7 WHERE id = ?::uuid")
+                .param(modelId)
+                .update();
+        Boolean touched = jdbcClient.sql(
+                        "SELECT updated_at > created_at FROM creation.generation_models WHERE id = ?::uuid")
+                .param(modelId)
+                .query(Boolean.class)
+                .single();
+        assertThat(touched).isTrue();
+    }
+
+    // --- helpers ---
+
+    private int indexCount(String indexName) {
+        return jdbcClient.sql(
+                        "SELECT COUNT(*) FROM pg_indexes WHERE schemaname = 'creation' AND indexname = ?")
+                .param(indexName)
+                .query(Integer.class)
+                .single();
+    }
+
+    private String insertUser(String email) {
+        // identity.users holds no credential columns (V1: those live outside
+        // this table); username is the only required field and is constrained
+        // to 3..32 trimmed characters.
+        return jdbcClient.sql(
+                        "INSERT INTO identity.users (id, username, email) " +
+                                "VALUES (gen_random_uuid(), ?, ?) RETURNING id::text")
+                .params(email.substring(0, email.indexOf('@')), email)
+                .query(String.class)
+                .single();
+    }
+
+    private String insertModel(String code, String mediaKind) {
+        return jdbcClient.sql(
+                        "INSERT INTO creation.generation_models " +
+                                "(id, code, display_name, provider_code, provider_model_name, media_kind) " +
+                                "VALUES (gen_random_uuid(), ?, ?, 'fixture', 'fixture-model', ?) " +
+                                "RETURNING id::text")
+                .params(code, code, mediaKind)
+                .query(String.class)
+                .single();
+    }
+
+    private String insertJob(String userId, String modelId, String mediaKind, String idempotencyKey) {
+        String key = idempotencyKey != null ? idempotencyKey : "key-" + java.util.UUID.randomUUID();
+        return jdbcClient.sql(
+                        "INSERT INTO creation.generation_jobs " +
+                                "(id, user_id, media_kind, model_id, provider_code_snapshot, " +
+                                " provider_model_snapshot, model_version_snapshot, idempotency_key, request_hash) " +
+                                "VALUES (gen_random_uuid(), ?::uuid, ?, ?::uuid, 'fixture', 'fixture-model', " +
+                                " 'v1', ?, decode(repeat('ab', 32), 'hex')) RETURNING id::text")
+                .params(userId, mediaKind, modelId, key)
+                .query(String.class)
+                .single();
+    }
+}