diff --git a/deploy/configs/civitai-index.yaml b/deploy/configs/civitai-index.yaml
index 02cd97b..61f0ded 100644
--- a/deploy/configs/civitai-index.yaml
+++ b/deploy/configs/civitai-index.yaml
@@ -9,6 +9,7 @@ config:
     - { name: availability, field_type: single_value, eager_load: true }
     - { name: postId, field_type: single_value, per_value_lazy: true }
     - { name: postedToId, field_type: single_value, per_value_lazy: true }
+    - { name: model3dId, field_type: single_value, per_value_lazy: true }
     - { name: remixOfId, field_type: single_value }
     - { name: hasMeta, field_type: boolean, eager_load: true }
     - { name: onSite, field_type: boolean, eager_load: true }
@@ -25,12 +26,14 @@ config:
 
   sort_fields:
     - { name: reactionCount, bits: 32, eager_load: true }
-    - name: sortAt
-      bits: 32
-      eager_load: true
-      computed:
-        op: greatest
-        source_fields: [existedAt, publishedAt]
+    # [PR-M5] sortAt is now INGESTED, not engine-computed. The old
+    # `computed: { op: greatest, source_fields: [existedAt, publishedAt] }` block
+    # is removed in the SAME deploy that activates the sortAtUnix->sortAt emission
+    # (sync-config-civitai.yaml) — computed and ingested must NEVER coexist, or a
+    # stale GREATEST would fight the ingested value. Values now arrive via the
+    # data_schema sortAtUnix->sortAt mapping (steady-state ops, ms ÷1000 seconds)
+    # or the dump's `sortAt` seconds column (fallback: sortAt). Layers hold SECONDS.
+    - { name: sortAt, bits: 32, eager_load: true }
     - { name: commentCount, bits: 32, eager_load: true }
     - { name: collectedCount, bits: 32, eager_load: true }
     - { name: existedAt, bits: 32 }
@@ -69,6 +72,7 @@ data_schema:
     - { source: availability, target: availability, value_type: low_cardinality_string, nullable: true }
     - { source: postId, target: postId, value_type: integer, nullable: true }
     - { source: postedToId, target: postedToId, value_type: integer, nullable: true }
+    - { source: model3dId, target: model3dId, value_type: integer, nullable: true }
     - { source: remixOfId, target: remixOfId, value_type: integer, nullable: true }
     - { source: publishedAtUnix, target: isPublished, value_type: exists_boolean }
     - { source: remixOfId, target: isRemix, value_type: exists_boolean }
@@ -83,6 +87,17 @@ data_schema:
     - { source: toolIds, target: toolIds, value_type: integer_array, default: [], filter_only: true }
     - { source: techniqueIds, target: techniqueIds, value_type: integer_array, default: [], filter_only: true }
     - { source: reactionCount, target: reactionCount, value_type: integer, default: 0 }
+    # [PR-B4] PINNED UNIT STATEMENT. Steady-state ops carry `sortAtUnix` in
+    # MILLISECONDS; ms_to_seconds:true divides by 1000 → the sortAt SORT LAYERS
+    # hold SECONDS. The dump carries a `sortAt` column already in SECONDS, hit via
+    # `fallback: sortAt` (a fallback source is NOT ms-converted — it is assumed to
+    # already be in target units, see config.rs resolve). Both paths → SECONDS.
+    # QUERY SIDE: the Civitai builder's `sortAtUnix Gte <seconds>` clause resolves
+    # ONLY through time_buckets (filter_field: sortAtUnix → sort_field: sortAt);
+    # ts_secs_u64 (query.rs:316) leaves the seconds threshold unchanged, and it
+    # snaps to a bucket built from the sortAt seconds layers (executor.rs Gt/Gte).
+    # No ingested `sortAtUnix` filter field exists. Net: SECONDS everywhere except
+    # the wire ms encoding; the builder's seconds-Gte is correct.
     - { source: sortAtUnix, target: sortAt, value_type: integer, fallback: sortAt, ms_to_seconds: true }
     - { source: commentCount, target: commentCount, value_type: integer, default: 0 }
     - { source: collectedCount, target: collectedCount, value_type: integer, default: 0 }
diff --git a/deploy/configs/prod-sync-config-civitai.yaml b/deploy/configs/prod-sync-config-civitai.yaml
deleted file mode 100644
index 2c13052..0000000
--- a/deploy/configs/prod-sync-config-civitai.yaml
+++ /dev/null
@@ -1,296 +0,0 @@
-# Sync V2 — Civitai Sync Config (Draft v2)
-#
-# Read by bitdex-sync (sidecar). Defines PG tables, dump phases, and triggers.
-# BitDex server does NOT read this file — it receives dump instructions
-# via PUT /dumps request body, populated by bitdex-sync from this config.
-#
-# Conventions:
-# - Fields reference column names from COPY query (not indices)
-# - When source column name == target field name, use shorthand: `fields: [fieldName]`
-# - sets_alive defaults to false (omit when false)
-# - computed_fields: any derived/computed value (from expressions, lookups, bitfields)
-# - Trigger expressions use {column} templates: {publishedAt} resolves to
-#   OLD."publishedAt" / NEW."publishedAt" for remove/set ops respectively
-
-index: civitai
-
-postgres:
-  connection_string: "${DATABASE_URL}"
-  stage_dir: /data/load_stage
-
-clickhouse:
-  connection_string: "${CLICKHOUSE_URL}"
-
-bitdex:
-  url: "http://localhost:3000"
-  index: civitai
-
-# -----------------------------------------------------------------------
-# Dump phases — processed in order during boot
-# -----------------------------------------------------------------------
-
-dump_phases:
-
-  # Phase 1: Images (14GB, primary entity — loaded first for earliest query readiness)
-  # NOTE: Images before tags so the server can serve real queries while
-  # the 63GB tags phase is still downloading/processing.
-  - name: images
-    table: Image
-    copy_query: >
-      COPY (SELECT id, url, "nsfwLevel", hash, flags, type::text,
-                   "userId", "blockedFor",
-                   extract(epoch from "scannedAt")::bigint as "scannedAtSecs",
-                   extract(epoch from "createdAt")::bigint as "createdAtSecs",
-                   "postId", width, height
-            FROM "Image")
-      TO STDOUT WITH (FORMAT csv)
-    columns: [id, url, nsfwLevel, hash, flags, type, userId, blockedFor, scannedAtSecs, createdAtSecs, postId, width, height]
-    slot_field: id
-    sets_alive: true
-    fields:
-      - nsfwLevel
-      - { column: type, target: type }
-      - userId
-      - postId
-      - blockedFor
-      - { column: url, target: url }          # doc-only
-      - { column: hash, target: hash }        # doc-only
-      - width                                  # doc-only
-      - height                                 # doc-only
-    computed_fields:
-      # From flags bitfield:
-      - target: hasMeta
-        expression: "(flags >> 13) & 1 == 1 && (flags >> 2) & 1 == 0"
-      - target: onSite
-        expression: "(flags >> 14) & 1 == 1"
-      - target: minor
-        expression: "(flags >> 3) & 1 == 1"
-      - target: poi
-        expression: "(flags >> 4) & 1 == 1"
-      # Sort fields computed from CSV columns:
-      - target: existedAt
-        expression: "max(scannedAtSecs, createdAtSecs)"
-      - target: id
-        expression: "id"  # slot ID as sort value
-    enrichment:
-      - lookup: posts.csv
-        table: Post
-        copy_query: >
-          COPY (SELECT id,
-                       extract(epoch from "publishedAt")::bigint as "publishedAtSecs",
-                       availability::text,
-                       "modelVersionId"
-                FROM "Post")
-          TO STDOUT WITH (FORMAT csv)
-        columns: [id, publishedAtSecs, availability, modelVersionId]
-        key: id
-        join_on: postId
-        fields:
-          - { column: publishedAtSecs, target: publishedAt }
-          - { column: availability, target: availability }
-          - { column: modelVersionId, target: postedToId }
-        computed_fields:
-          - target: isPublished
-            expression: "publishedAtSecs != null"
-    # sortAt = GREATEST(existedAt, publishedAt) is defined in index config
-    # as a computed sort field — BitDex resolves it automatically.
-
-  # Phase 2: Tags (63GB, largest — loaded after images so server can serve queries)
-  - name: tags
-    table: TagsOnImageNew
-    copy_query: >
-      COPY (SELECT "tagId", "imageId", "attributes"
-            FROM "TagsOnImageNew")
-      TO STDOUT WITH (FORMAT csv)
-    columns: [tagId, imageId, attributes]
-    slot_field: imageId
-    fields:
-      - { column: tagId, target: tagIds }
-    filter: "(attributes >> 10) & 1 = 0"  # skip disabled tags
-
-  # Phase 3: Resources (820MB)
-  - name: resources
-    table: ImageResourceNew
-    copy_query: >
-      COPY (SELECT "imageId", "modelVersionId", detected
-            FROM "ImageResourceNew")
-      TO STDOUT WITH (FORMAT csv)
-    columns: [imageId, modelVersionId, detected]
-    slot_field: imageId
-    fields:
-      - { column: modelVersionId, target: modelVersionIds }
-    computed_fields:
-      - target: modelVersionIdsManual
-        expression: "detected == false"
-        value: modelVersionId
-    enrichment:
-      - lookup: model_versions.csv
-        table: ModelVersion
-        copy_query: >
-          COPY (SELECT id, "baseModel", "modelId"
-                FROM "ModelVersion")
-          TO STDOUT WITH (FORMAT csv)
-        columns: [id, baseModel, modelId]
-        key: id
-        join_on: modelVersionId
-        fields:
-          - { column: baseModel, target: baseModel }
-        enrichment:
-          - lookup: models.csv
-            table: Model
-            copy_query: >
-              COPY (SELECT id, poi, type::text
-                    FROM "Model")
-              TO STDOUT WITH (FORMAT csv)
-            columns: [id, poi, type]
-            key: id
-            join_on: modelId
-            fields:
-              - { column: poi, target: poi }
-            filter: "type = 'Checkpoint'"
-
-  # Phase 4: Tools (50MB)
-  - name: tools
-    table: ImageTool
-    copy_query: >
-      COPY (SELECT "toolId", "imageId" FROM "ImageTool")
-      TO STDOUT WITH (FORMAT csv)
-    columns: [toolId, imageId]
-    slot_field: imageId
-    fields:
-      - { column: toolId, target: toolIds }
-
-  # Phase 5: Techniques (71MB)
-  - name: techniques
-    table: ImageTechnique
-    copy_query: >
-      COPY (SELECT "techniqueId", "imageId" FROM "ImageTechnique")
-      TO STDOUT WITH (FORMAT csv)
-    columns: [techniqueId, imageId]
-    slot_field: imageId
-    fields:
-      - { column: techniqueId, target: techniqueIds }
-
-  # Phase 6: Metrics (1.4GB, from ClickHouse, TAB-separated)
-  - name: metrics
-    source: clickhouse
-    query: >
-      SELECT imageId, reactionCount, commentCount, collectedCount
-      FROM image_metrics
-    format: tsv
-    slot_field: imageId
-    fields: [reactionCount, commentCount, collectedCount]
-
-# -----------------------------------------------------------------------
-# Steady-state triggers — PG triggers that write to BitdexOps
-# -----------------------------------------------------------------------
-# Trigger name: bitdex_{table}_{config_hash8}
-# Expressions use {column} templates that resolve to OLD/NEW for remove/set ops.
-
-triggers:
-
-  - table: Image
-    slot_field: id
-    sets_alive: true
-    on_delete: true
-    track_fields:
-      - nsfwLevel
-      - { column: type, expression: "{type}::text" }
-      - userId
-      - postId
-      - blockedFor
-      - url
-      - hash
-      - width
-      - height
-    computed_fields:
-      - target: hasMeta
-        expression: "({flags} >> 13) & 1 = 1 AND ({flags} >> 2) & 1 = 0"
-      - target: onSite
-        expression: "({flags} >> 14) & 1 = 1"
-      - target: minor
-        expression: "({flags} >> 3) & 1 = 1"
-      - target: poi
-        expression: "({flags} >> 4) & 1 = 1"
-
-  - table: TagsOnImageNew
-    slot_field: imageId
-    field: tagIds
-    value_field: tagId
-    filter: '({attributes} >> 10) & 1 = 0'
-
-  - table: ImageTool
-    slot_field: imageId
-    field: toolIds
-    value_field: toolId
-
-  - table: ImageTechnique
-    slot_field: imageId
-    field: techniqueIds
-    value_field: techniqueId
-
-  - table: ImageResourceNew
-    slot_field: imageId
-    field: modelVersionIds
-    value_field: modelVersionId
-    computed_fields:
-      - target: modelVersionIdsManual
-        expression: "{detected} = false"
-        value: "{modelVersionId}"
-
-  # Fan-out triggers (queryOpSet):
-  - table: Post
-    type: fan_out
-    query: "postId eq {id}"
-    track_fields:
-      - { column: publishedAt, target: publishedAt, expression: "extract(epoch from {publishedAt})::bigint" }
-      # Null handling: if {publishedAt} is NULL, trigger emits remove op (clears the field).
-      # The {field} template resolves to OLD.field for remove ops, NEW.field for set ops.
-      # When OLD is not null and NEW is null → remove. When NEW is not null → set.
-      - { column: availability, target: availability, expression: "{availability}::text" }
-      - { column: modelVersionId, target: postedToId }
-
-  - table: ModelVersion
-    type: fan_out
-    query: "modelVersionIds eq {id}"
-    track_fields:
-      - { column: baseModel, target: baseModel }
-    join: 'INNER JOIN "Model" ON "Model".id = "ModelVersion"."modelId"'
-    filter: '"Model".type = ''Checkpoint'''
-
-  - table: Model
-    type: fan_out
-    expression: 'SELECT json_agg(mv.id) as ids FROM "ModelVersion" mv WHERE mv."modelId" = {id}' # @justin: added the missing cast to the correct name to be used inside of the query. I guess we'll have to correctly parse JSON or whatever in our query handler before handing it to the BitdexOps endpoint.
-    # @adam:* Right — the trigger function parses the json_agg result array and constructs the queryOpSet with `modelVersionIds in [id1, id2, ...]`. The trigger_gen code handles this: it detects the expression returns JSON, parses it to i64 array, formats into the query template. Straightforward since we control the SQL.
-    # expression replaces the default slot resolution. {id} → NEW.id.
-    # Result is an array of MV ids, used in query below.
-    query: "modelVersionIds in [{ids}]"
-    track_fields:
-      - poi
-
-# -----------------------------------------------------------------------
-# Dump request body schema (what bitdex-sync sends to BitDex)
-# -----------------------------------------------------------------------
-# PUT /api/indexes/{index}/dumps
-# {
-#   "name": "tags-a1b2c3d4",
-#   "csv_path": "/data/load_stage/tags.csv",
-#   "format": "csv",
-#   "slot_field": "imageId",
-#   "sets_alive": false,
-#   "fields": [
-#     { "column": "tagId", "target": "tagIds" }
-#   ],
-#   "filter": "(attributes >> 10) & 1 = 0",
-#   "computed_fields": [],
-#   "enrichment": [
-#     {
-#       "csv_path": "/data/load_stage/posts.csv",
-#       "key": "id",
-#       "join_on": "postId",
-#       "fields": [...],
-#       "computed_fields": [...],
-#       "enrichment": [...]
-#     }
-#   ]
-# }
diff --git a/deploy/configs/sync-config-civitai.yaml b/deploy/configs/sync-config-civitai.yaml
index 20713df..1056658 100644
--- a/deploy/configs/sync-config-civitai.yaml
+++ b/deploy/configs/sync-config-civitai.yaml
@@ -36,15 +36,28 @@ dump_phases:
   # the 63GB tags phase is still downloading/processing.
   - name: images
     table: Image
+    # [PR-B3] sortAt: emitted directly as epoch SECONDS (bigint) so it lands in
+    # the sortAt sort layers with no ms conversion (data_schema fallback: sortAt,
+    # which does NOT apply ms_to_seconds). TYPE-CORRECT belt — every COALESCE/
+    # GREATEST operand is a TIMESTAMP, and extract(epoch ...)::bigint wraps the
+    # whole thing (never COALESCE(timestamp, bigint)). i."sortAt" is the value the
+    # model-share trigger maintains as GREATEST(post.publishedAt, scannedAt,
+    # createdAt); the GREATEST here is only the fallback for rows the backfill
+    # hasn't populated yet (COALESCE short-circuits, so the Post subselect runs
+    # only when i."sortAt" IS NULL). Matches the steady-state trigger's sortAtUnix
+    # value ÷ 1000 (W3 asserts dump == trigger on the same rows).
     copy_query: >
-      COPY (SELECT id, url, "nsfwLevel", hash, flags, type::text,
-                   "userId", "blockedFor",
-                   extract(epoch from "scannedAt")::bigint as "scannedAtSecs",
-                   extract(epoch from "createdAt")::bigint as "createdAtSecs",
-                   "postId", width, height
-            FROM "Image")
+      COPY (SELECT i.id, i.url, i."nsfwLevel", i.hash, i.flags, i.type::text,
+                   i."userId", i."blockedFor",
+                   extract(epoch from i."scannedAt")::bigint as "scannedAtSecs",
+                   extract(epoch from i."createdAt")::bigint as "createdAtSecs",
+                   extract(epoch from COALESCE(i."sortAt",
+                     GREATEST((SELECT p."publishedAt" FROM "Post" p WHERE p.id = i."postId"),
+                              i."scannedAt", i."createdAt")))::bigint as "sortAt",
+                   i."postId", i.width, i.height
+            FROM "Image" i)
       TO STDOUT WITH (FORMAT csv)
-    columns: [id, url, nsfwLevel, hash, flags, type, userId, blockedFor, scannedAtSecs, createdAtSecs, postId, width, height]
+    columns: [id, url, nsfwLevel, hash, flags, type, userId, blockedFor, scannedAtSecs, createdAtSecs, sortAt, postId, width, height]
     slot_field: id
     sets_alive: true
     fields:
@@ -57,6 +70,11 @@ dump_phases:
       - { column: hash, target: hash }        # doc-only
       - width                                  # doc-only
       - height                                 # doc-only
+      # sortAt sort layers (SECONDS). Direct mapping — the dump sort path reads
+      # this i64 as-is (data_schema ms_to_seconds is an ops-path concern, not the
+      # dump). Replaces the former `computed: greatest(existedAt, publishedAt)`
+      # sort field in civitai-index.yaml (removed same-deploy, [PR-M5]).
+      - { column: sortAt, target: sortAt }
     computed_fields:
       # From flags bitfield:
       - target: hasMeta
@@ -79,21 +97,28 @@ dump_phases:
           COPY (SELECT id,
                        extract(epoch from "publishedAt")::bigint as "publishedAtSecs",
                        availability::text,
-                       "modelVersionId"
+                       "modelVersionId",
+                       "model3dId"
                 FROM "Post")
           TO STDOUT WITH (FORMAT csv)
-        columns: [id, publishedAtSecs, availability, modelVersionId]
+        columns: [id, publishedAtSecs, availability, modelVersionId, model3dId]
         key: id
         join_on: postId
         fields:
           - { column: publishedAtSecs, target: publishedAt }
           - { column: availability, target: availability }
           - { column: modelVersionId, target: postedToId }
+          # model3dId (Post column, Meili parity). Enriched onto each image via
+          # the postId join, mirroring the steady-state Image trigger's
+          # model3dId INSERT-read subselect.
+          - { column: model3dId, target: model3dId }
         computed_fields:
           - target: isPublished
             expression: "publishedAtSecs != null"
-    # sortAt = GREATEST(existedAt, publishedAt) is defined in index config
-    # as a computed sort field — BitDex resolves it automatically.
+    # sortAt is now INGESTED (copy_query column above), not engine-computed.
+    # The old `computed: greatest(existedAt, publishedAt)` sort field is removed
+    # from civitai-index.yaml in the SAME deploy as this emission flip ([PR-M5]:
+    # computed and ingested must never coexist).
 
   # Phase 2: Tags (63GB, largest — loaded after images so server can serve queries)
   - name: tags
@@ -193,6 +218,32 @@ triggers:
     slot_field: id
     sets_alive: true
     on_delete: true
+    # [PR-B4] PINNED UNIT STATEMENT (emission + query chains — see also civitai-index.yaml):
+    #   EMISSION: sortAtUnix is emitted in MILLISECONDS (`* 1000`). The engine
+    #   data_schema mapping `sortAtUnix -> sortAt` (civitai-index.yaml,
+    #   ms_to_seconds:true) divides by 1000, so the sortAt SORT LAYERS store
+    #   SECONDS. The dump path instead emits a `sortAt` column already in SECONDS
+    #   (data_schema fallback: sortAt — ms_to_seconds is NOT applied to a
+    #   fallback source). Both wire encodings converge on SECONDS in the layers.
+    #   QUERY: the Civitai builder sends `sortAtUnix Gte <seconds>`
+    #   (image.service.ts:3940). That clause is resolved by the time-bucket
+    #   manager (civitai-index.yaml time_buckets.filter_field: sortAtUnix,
+    #   sort_field: sortAt): the threshold is unit-normalized by ts_secs_u64
+    #   (src/query.rs:316 — value already seconds, so passes through unchanged)
+    #   and snapped to a bucket built from the sortAt SECONDS layers
+    #   (src/executor.rs Gt/Gte ~L668, Lt/Lte ~L702). No ingested `sortAtUnix`
+    #   filter field exists — the clause only ever hits the sort layers via the
+    #   bucket. Net: SECONDS on the query side; MILLISECONDS only on the wire,
+    #   divided back on ingest. The builder's seconds-Gte is CORRECT both before
+    #   this change (sortAt computed) and after (sortAt ingested).
+    #
+    # sortAtUnix is emitted via the SHARED bitdex_image_sortat_ops() function so
+    # the W1-3 re-emitter re-asserts the identical value (shape parity). MS units
+    # (`* 1000`) with a COALESCE(GREATEST...) belt for rows the PG backfill hasn't
+    # populated Image.sortAt on yet. Expression is byte-identical to the W1-1
+    # codegen pin (trigger_gen.rs SORTATUNIX_EXPR); COALESCE short-circuits so the
+    # Post subselect only evaluates when Image.sortAt is still NULL.
+    shared_ops_fields: [sortAtUnix]
     track_fields:
       - nsfwLevel
       - { column: type, expression: "{type}::text" }
@@ -203,6 +254,7 @@ triggers:
       - hash
       - width
       - height
+      - 'COALESCE(extract(epoch from {sortAt})::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = {postId}))::bigint, extract(epoch from {scannedAt})::bigint, extract(epoch from {createdAt})::bigint)) * 1000 as sortAtUnix'
     computed_fields:
       - target: hasMeta
         expression: "({flags} >> 13) & 1 = 1 AND ({flags} >> 2) & 1 = 0"
@@ -234,6 +286,15 @@ triggers:
         expression: '(SELECT p."availability"::text FROM "Post" p WHERE p.id = {postId})'
       - target: postedToId
         expression: '(SELECT p."modelVersionId" FROM "Post" p WHERE p.id = {postId})'
+      # model3dId (Post column, indexed for Meili parity). INSERT-reads the
+      # parent Post via subselect, exactly like publishedAt's INSERT branch, and
+      # re-resolves on UPDATE-when-changed (e.g. if the image moves posts).
+      # NOT carried on the Post per-image fan-out below: model3dId is set at
+      # post-CREATE, before any image rows exist, and does not change over a
+      # post's lifetime — so the Image INSERT-read fully covers it and a Post
+      # fan-out entry would only add a redundant (disjointness-breaking) writer.
+      - target: model3dId
+        expression: '(SELECT p."model3dId" FROM "Post" p WHERE p.id = {postId})'
 
   - table: TagsOnImageNew
     slot_field: imageId
@@ -260,15 +321,33 @@ triggers:
         expression: "{detected} = false"
         value: "{modelVersionId}"
 
-  # Fan-out triggers (queryOpSet):
+  # Post publish fan-out — PER-IMAGE MATERIALIZED (W1-1 / PR #325).
+  # Replaces the old `type: fan_out` queryOpSet ("postId eq {id}") that BitDex
+  # resolved against its own moving index at apply time (FD #69397: never checked
+  # what it matched). `fan_out_per_row` instead enumerates the Post's Image rows
+  # TRANSACTIONALLY in PG — INSERT INTO "BitdexOps" SELECT i.id, <ops> FROM
+  # "Image" i WHERE i."postId" = NEW.id — so the match set is fixed inside the
+  # publishing txn and cannot be partial/early/re-resolved on WAL replay.
+  #
+  # [PR-M1] RETAINS the full payload {publishedAt, availability, postedToId}.
+  # publishedAt drives BOTH deferred-alive activation stamping AND the isPublished
+  # shadow — do NOT drop it (design §2.2 Options A/B that delete publishedAt are
+  # PARKED). Only Phase 2 ever removes availability/postedToId.
+  #
+  # [PR-m2] These three fields are DISJOINT from sortAtUnix (the only new field
+  # the Image trigger emits): the Post fan-out never emits sortAtUnix, so the two
+  # writers never collide on one image → op_dedup has nothing to LIFO-resolve
+  # nondeterministically. (The Image trigger DOES also INSERT-read
+  # publishedAt/availability/postedToId for the "image added to an already-
+  # published post" case, but that is temporally separated and same-valued — see
+  # test_post_fanout_disjoint_from_sortatunix in sync_config.rs.)
   - table: Post
-    type: fan_out
-    query: "postId eq {id}"
+    type: fan_out_per_row
+    rows_from: { table: Image, fk: postId, slot: id, parent_key: id }
     track_fields:
       - { column: publishedAt, target: publishedAt, expression: "extract(epoch from {publishedAt})::bigint" }
-      # Null handling: if {publishedAt} is NULL, trigger emits remove op (clears the field).
-      # The {field} template resolves to OLD.field for remove ops, NEW.field for set ops.
-      # When OLD is not null and NEW is null → remove. When NEW is not null → set.
+      # Null handling: if {publishedAt} is NULL, the UPDATE branch emits a remove op
+      # (OLD non-null, NEW null). NEW non-null → set. Fan-out reaches every child image.
       - { column: availability, target: availability, expression: "{availability}::text" }
       - { column: modelVersionId, target: postedToId }
 
diff --git a/docs/design/trigger-sql-review.sql b/docs/design/trigger-sql-review.sql
index 593e59d..dabcc68 100644
--- a/docs/design/trigger-sql-review.sql
+++ b/docs/design/trigger-sql-review.sql
@@ -48,11 +48,17 @@ CREATE TRIGGER trg_cleanup_bitdex_ops
 -- Part 2: Per-table triggers (generated from sync config YAML)
 -- -----------------------------------------------------------------------
 
--- [1/8] Table: Image → Trigger: bitdex_image_a6ea374e
+-- [1/8] Table: Image → Trigger: bitdex_image_ee936694
 -- Sets alive: yes
 -- On delete: emit delete op
 
-CREATE OR REPLACE FUNCTION bitdex_image_ops_a6ea374e() RETURNS trigger AS $$
+CREATE OR REPLACE FUNCTION bitdex_image_sortat_ops(_i "Image") RETURNS jsonb AS $$
+  SELECT jsonb_build_array(
+    jsonb_build_object('op', 'set', 'field', 'sortAtUnix', 'value', to_jsonb(COALESCE(extract(epoch from _i."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = _i."postId"))::bigint, extract(epoch from _i."scannedAt")::bigint, extract(epoch from _i."createdAt")::bigint)) * 1000))
+  );
+$$ LANGUAGE sql STABLE;
+
+CREATE OR REPLACE FUNCTION bitdex_image_ops_ee936694() RETURNS trigger AS $$
 DECLARE
   _ops jsonb;
 BEGIN
@@ -75,8 +81,10 @@ BEGIN
       jsonb_build_object('op', 'set', 'field', 'existedAt', 'value', to_jsonb(GREATEST(extract(epoch from NEW."scannedAt")::bigint, extract(epoch from NEW."createdAt")::bigint))),
       jsonb_build_object('op', 'set', 'field', 'publishedAt', 'value', to_jsonb((SELECT extract(epoch from p."publishedAt")::bigint FROM "Post" p WHERE p.id = NEW."postId"))),
       jsonb_build_object('op', 'set', 'field', 'availability', 'value', to_jsonb((SELECT p."availability"::text FROM "Post" p WHERE p.id = NEW."postId"))),
-      jsonb_build_object('op', 'set', 'field', 'postedToId', 'value', to_jsonb((SELECT p."modelVersionId" FROM "Post" p WHERE p.id = NEW."postId")))
+      jsonb_build_object('op', 'set', 'field', 'postedToId', 'value', to_jsonb((SELECT p."modelVersionId" FROM "Post" p WHERE p.id = NEW."postId"))),
+      jsonb_build_object('op', 'set', 'field', 'model3dId', 'value', to_jsonb((SELECT p."model3dId" FROM "Post" p WHERE p.id = NEW."postId")))
     );
+    _ops := _ops || bitdex_image_sortat_ops(NEW);
     INSERT INTO "BitdexOps" (entity_id, ops) VALUES (NEW."id", _ops);
     RETURN NEW;
   ELSIF TG_OP = 'DELETE' THEN
@@ -138,6 +146,12 @@ BEGIN
         jsonb_build_object('op', 'set', 'field', 'height', 'value', to_jsonb(NEW."height"))
       );
     END IF;
+    IF (COALESCE(extract(epoch from OLD."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = OLD."postId"))::bigint, extract(epoch from OLD."scannedAt")::bigint, extract(epoch from OLD."createdAt")::bigint)) * 1000) IS DISTINCT FROM (COALESCE(extract(epoch from NEW."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = NEW."postId"))::bigint, extract(epoch from NEW."scannedAt")::bigint, extract(epoch from NEW."createdAt")::bigint)) * 1000) THEN
+      _ops := _ops || jsonb_build_array(
+        jsonb_build_object('op', 'remove', 'field', 'sortAtUnix', 'value', to_jsonb(COALESCE(extract(epoch from OLD."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = OLD."postId"))::bigint, extract(epoch from OLD."scannedAt")::bigint, extract(epoch from OLD."createdAt")::bigint)) * 1000)),
+        jsonb_build_object('op', 'set', 'field', 'sortAtUnix', 'value', to_jsonb(COALESCE(extract(epoch from NEW."sortAt")::bigint, GREATEST(extract(epoch from (SELECT p."publishedAt" FROM "Post" p WHERE p.id = NEW."postId"))::bigint, extract(epoch from NEW."scannedAt")::bigint, extract(epoch from NEW."createdAt")::bigint)) * 1000))
+      );
+    END IF;
     IF ((OLD."flags" >> 13) & 1 = 1 AND (OLD."flags" >> 2) & 1 = 0) IS DISTINCT FROM ((NEW."flags" >> 13) & 1 = 1 AND (NEW."flags" >> 2) & 1 = 0) THEN
       _ops := _ops || jsonb_build_array(
         jsonb_build_object('op', 'remove', 'field', 'hasMeta', 'value', to_jsonb((OLD."flags" >> 13) & 1 = 1 AND (OLD."flags" >> 2) & 1 = 0)),
@@ -186,6 +200,12 @@ BEGIN
         jsonb_build_object('op', 'set', 'field', 'postedToId', 'value', to_jsonb((SELECT p."modelVersionId" FROM "Post" p WHERE p.id = NEW."postId")))
       );
     END IF;
+    IF ((SELECT p."model3dId" FROM "Post" p WHERE p.id = OLD."postId")) IS DISTINCT FROM ((SELECT p."model3dId" FROM "Post" p WHERE p.id = NEW."postId")) THEN
+      _ops := _ops || jsonb_build_array(
+        jsonb_build_object('op', 'remove', 'field', 'model3dId', 'value', to_jsonb((SELECT p."model3dId" FROM "Post" p WHERE p.id = OLD."postId"))),
+        jsonb_build_object('op', 'set', 'field', 'model3dId', 'value', to_jsonb((SELECT p."model3dId" FROM "Post" p WHERE p.id = NEW."postId")))
+      );
+    END IF;
     IF jsonb_array_length(_ops) > 0 THEN
       INSERT INTO "BitdexOps" (entity_id, ops) VALUES (NEW."id", _ops);
     END IF;
@@ -194,10 +214,10 @@ BEGIN
 END;
 $$ LANGUAGE plpgsql;
 
-DROP TRIGGER IF EXISTS bitdex_image_a6ea374e ON "Image";
-CREATE TRIGGER bitdex_image_a6ea374e AFTER INSERT OR UPDATE OR DELETE ON "Image"
-  FOR EACH ROW EXECUTE FUNCTION bitdex_image_ops_a6ea374e();
-ALTER TABLE "Image" ENABLE ALWAYS TRIGGER bitdex_image_a6ea374e;
+DROP TRIGGER IF EXISTS bitdex_image_ee936694 ON "Image";
+CREATE TRIGGER bitdex_image_ee936694 AFTER INSERT OR UPDATE OR DELETE ON "Image"
+  FOR EACH ROW EXECUTE FUNCTION bitdex_image_ops_ee936694();
+ALTER TABLE "Image" ENABLE ALWAYS TRIGGER bitdex_image_ee936694;
 
 
 -- [2/8] Table: TagsOnImageNew → Trigger: bitdex_tagsonimagenew_bcbef3c3
@@ -324,27 +344,27 @@ CREATE TRIGGER bitdex_imageresourcenew_d84d15a8 AFTER INSERT OR DELETE ON "Image
 ALTER TABLE "ImageResourceNew" ENABLE ALWAYS TRIGGER bitdex_imageresourcenew_d84d15a8;
 
 
--- [6/8] Table: Post → Trigger: bitdex_post_519ce657
--- Type: fan_out
+-- [6/8] Table: Post → Trigger: bitdex_post_8511462a
+-- Type: fan_out_per_row
 
-CREATE OR REPLACE FUNCTION bitdex_post_ops_519ce657() RETURNS trigger AS $$
+CREATE OR REPLACE FUNCTION bitdex_post_fanout_ops(_p "Post") RETURNS jsonb AS $$
+  SELECT jsonb_build_array(
+    jsonb_build_object('op', 'set', 'field', 'publishedAt', 'value', to_jsonb(extract(epoch from _p."publishedAt")::bigint)),
+    jsonb_build_object('op', 'set', 'field', 'availability', 'value', to_jsonb(_p."availability"::text)),
+    jsonb_build_object('op', 'set', 'field', 'postedToId', 'value', to_jsonb(_p."modelVersionId"))
+  );
+$$ LANGUAGE sql STABLE;
+
+CREATE OR REPLACE FUNCTION bitdex_post_ops_8511462a() RETURNS trigger AS $$
 DECLARE
   _ops jsonb;
-  _query text;
 BEGIN
   IF TG_OP = 'INSERT' THEN
-    _query := 'postId eq ' || NEW."id"::text;
-    _ops := jsonb_build_array(
-      jsonb_build_object('op', 'set', 'field', 'publishedAt', 'value', to_jsonb(extract(epoch from NEW."publishedAt")::bigint)),
-      jsonb_build_object('op', 'set', 'field', 'availability', 'value', to_jsonb(NEW."availability"::text)),
-      jsonb_build_object('op', 'set', 'field', 'postedToId', 'value', to_jsonb(NEW."modelVersionId"))
-    );
-    INSERT INTO "BitdexOps" (entity_id, ops) VALUES (NEW.id, jsonb_build_array(
-        jsonb_build_object('op', 'queryOpSet', 'query', _query, 'ops', _ops)
-      ));
+    _ops := bitdex_post_fanout_ops(NEW);
+    INSERT INTO "BitdexOps" (entity_id, ops)
+      SELECT c."id", _ops FROM "Image" c WHERE c."postId" = NEW."id";
     RETURN NEW;
   ELSIF TG_OP = 'UPDATE' THEN
-    _query := 'postId eq ' || NEW."id"::text;
     _ops := '[]'::jsonb;
     IF (extract(epoch from OLD."publishedAt")::bigint) IS DISTINCT FROM (extract(epoch from NEW."publishedAt")::bigint) THEN
       _ops := _ops || jsonb_build_array(
@@ -365,9 +385,8 @@ BEGIN
       );
     END IF;
     IF jsonb_array_length(_ops) > 0 THEN
-      INSERT INTO "BitdexOps" (entity_id, ops) VALUES (NEW.id, jsonb_build_array(
-        jsonb_build_object('op', 'queryOpSet', 'query', _query, 'ops', _ops)
-      ));
+      INSERT INTO "BitdexOps" (entity_id, ops)
+        SELECT c."id", _ops FROM "Image" c WHERE c."postId" = NEW."id";
     END IF;
     RETURN NEW;
   END IF;
@@ -375,10 +394,10 @@ BEGIN
 END;
 $$ LANGUAGE plpgsql;
 
-DROP TRIGGER IF EXISTS bitdex_post_519ce657 ON "Post";
-CREATE TRIGGER bitdex_post_519ce657 AFTER INSERT OR UPDATE ON "Post"
-  FOR EACH ROW EXECUTE FUNCTION bitdex_post_ops_519ce657();
-ALTER TABLE "Post" ENABLE ALWAYS TRIGGER bitdex_post_519ce657;
+DROP TRIGGER IF EXISTS bitdex_post_8511462a ON "Post";
+CREATE TRIGGER bitdex_post_8511462a AFTER INSERT OR UPDATE ON "Post"
+  FOR EACH ROW EXECUTE FUNCTION bitdex_post_ops_8511462a();
+ALTER TABLE "Post" ENABLE ALWAYS TRIGGER bitdex_post_8511462a;
 
 
 -- [7/8] Table: ModelVersion → Trigger: bitdex_modelversion_22dd59b3
@@ -475,12 +494,12 @@ ALTER TABLE "Model" ENABLE ALWAYS TRIGGER bitdex_model_a13d0fe3;
 -- -----------------------------------------------------------------------
 -- Tables created: BitdexOps, bitdex_cursors
 -- Triggers: 8
---   bitdex_image_a6ea374e on "Image"
+--   bitdex_image_ee936694 on "Image"
 --   bitdex_tagsonimagenew_bcbef3c3 on "TagsOnImageNew"
 --   bitdex_imagetool_f87e1fc4 on "ImageTool"
 --   bitdex_imagetechnique_ee2b2860 on "ImageTechnique"
 --   bitdex_imageresourcenew_d84d15a8 on "ImageResourceNew"
---   bitdex_post_519ce657 on "Post"
+--   bitdex_post_8511462a on "Post"
 --   bitdex_modelversion_22dd59b3 on "ModelVersion"
 --   bitdex_model_a13d0fe3 on "Model"
 --
diff --git a/src/pg_sync/sync_config.rs b/src/pg_sync/sync_config.rs
index 14fe23c..35bfcff 100644
--- a/src/pg_sync/sync_config.rs
+++ b/src/pg_sync/sync_config.rs
@@ -804,4 +804,166 @@ triggers: []
             assert!(!enrich.columns.is_empty(), "posts enrichment must have explicit columns");
         }
     }
+
+    // -----------------------------------------------------------------------
+    // W2-1 config-level guards (scheduled-publish plan)
+    // -----------------------------------------------------------------------
+
+    /// Load the production deploy sync config.
+    fn load_deploy_sync_config() -> FullSyncConfig {
+        let yaml = std::fs::read_to_string(
+            concat!(env!("CARGO_MANIFEST_DIR"), "/deploy/configs/sync-config-civitai.yaml")
+        ).expect("deploy/configs/sync-config-civitai.yaml must exist");
+        // from_yaml now runs SyncSource::validate() on every trigger (c4a4aa7),
+        // so a green parse here also proves the fan_out_per_row/shared_ops_fields
+        // structural rules hold on the real config.
+        FullSyncConfig::from_yaml(&yaml).expect("deploy sync config must parse + validate")
+    }
+
+    /// Extract the set of BitDex field names a generated trigger emits, by
+    /// scanning the generated SQL for `'field', '<name>'` op markers (covers
+    /// both inlined ops and the shared `bitdex_*_ops` functions embedded in the
+    /// trigger SQL).
+    fn emitted_trigger_fields(sql: &str) -> std::collections::HashSet<String> {
+        let mut out = std::collections::HashSet::new();
+        let marker = "'field', '";
+        let mut rest = sql;
+        while let Some(pos) = rest.find(marker) {
+            let after = &rest[pos + marker.len()..];
+            if let Some(end) = after.find('\'') {
+                out.insert(after[..end].to_string());
+            }
+            rest = after;
+        }
+        out
+    }
+
+    /// [W1-1 review minor 3 / PR-m2] Disjointness on the REAL deploy config.
+    ///
+    /// The load-bearing invariant is that the new `sortAtUnix` writer (the Image
+    /// trigger's shared function) and the Post per-image fan-out never emit the
+    /// SAME field for one image — an overlap would let op_dedup LIFO-resolve the
+    /// two writers nondeterministically.
+    ///
+    /// The Image trigger DOES also emit {publishedAt, availability, postedToId}
+    /// (the 2026-07-07 "Mode B" INSERT-read for images added to an already-
+    /// published post). That overlap with the Post fan-out is INTENTIONAL and
+    /// safe: the two writers fire for temporally-separated events and resolve to
+    /// the SAME value (both read Post.publishedAt), so LIFO order is immaterial.
+    /// This test pins that the overlap is EXACTLY that documented set and that
+    /// `sortAtUnix` (and `model3dId`) are strictly disjoint.
+    #[test]
+    fn test_post_fanout_disjoint_from_sortatunix() {
+        use crate::pg_sync::trigger_gen::generate_trigger_sql;
+
+        let config = load_deploy_sync_config();
+        let post = config.triggers.iter().find(|t| t.table == "Post")
+            .expect("Post trigger must exist");
+        let image = config.triggers.iter().find(|t| t.table == "Image")
+            .expect("Image trigger must exist");
+
+        // The Post trigger is the per-image materialized fan-out (not queryOpSet).
+        assert_eq!(post.table_type.as_deref(), Some("fan_out_per_row"),
+            "Post must be fan_out_per_row");
+        assert!(post.rows_from.is_some(), "Post fan_out_per_row must set rows_from");
+        assert!(post.query.is_none(), "Post must not carry a query fan-out anymore");
+
+        let post_fields = emitted_trigger_fields(&generate_trigger_sql(post));
+        let image_fields = emitted_trigger_fields(&generate_trigger_sql(image));
+        assert!(!post_fields.is_empty() && !image_fields.is_empty());
+
+        // Post fan-out emits exactly the retained payload (PR-M1) — and crucially
+        // NOT sortAtUnix.
+        let expected_post: std::collections::HashSet<String> =
+            ["publishedAt", "availability", "postedToId"].iter().map(|s| s.to_string()).collect();
+        assert_eq!(post_fields, expected_post,
+            "Post fan-out must emit exactly {{publishedAt, availability, postedToId}}, got {post_fields:?}");
+
+        // sortAtUnix is emitted by the Image trigger and is DISJOINT from Post.
+        assert!(image_fields.contains("sortAtUnix"), "Image trigger must emit sortAtUnix");
+        assert!(!post_fields.contains("sortAtUnix"),
+            "Post fan-out must NOT emit sortAtUnix — it is the Image trigger's field");
+        // model3dId likewise Image-only (INSERT-read, not carried on the fan-out).
+        assert!(image_fields.contains("model3dId"), "Image trigger must emit model3dId");
+        assert!(!post_fields.contains("model3dId"), "Post fan-out must NOT emit model3dId");
+
+        // The ONLY overlap between the two writers is the documented, safe Mode-B set.
+        let overlap: std::collections::HashSet<String> =
+            post_fields.intersection(&image_fields).cloned().collect();
+        assert_eq!(overlap, expected_post,
+            "the only Post∩Image overlap must be the intentional Mode-B set \
+             {{publishedAt, availability, postedToId}}; any other overlap (esp. \
+             sortAtUnix) would be op_dedup-nondeterministic. Got {overlap:?}");
+    }
+
+    /// [PR-B4 emission side / AR-4] The Image trigger emits sortAtUnix in
+    /// MILLISECONDS via the shared function, on the REAL deploy config. Seconds
+    /// here would shrink every value 1000× once the engine's ms_to_seconds
+    /// mapping divides by 1000.
+    #[test]
+    fn test_image_sortatunix_emitted_ms_on_deploy_config() {
+        use crate::pg_sync::trigger_gen::generate_trigger_sql;
+
+        let config = load_deploy_sync_config();
+        let image = config.triggers.iter().find(|t| t.table == "Image")
+            .expect("Image trigger must exist");
+        assert_eq!(image.shared_ops_fields.as_deref(), Some(&["sortAtUnix".to_string()][..]),
+            "Image must factor sortAtUnix into the shared re-emitter function");
+
+        let sql = generate_trigger_sql(image);
+        assert!(sql.contains("* 1000"), "sortAtUnix must be milliseconds:\n{sql}");
+        assert!(sql.contains("_ops := _ops || bitdex_image_sortat_ops(NEW);"),
+            "INSERT must call the shared sortat ops function (re-emitter parity):\n{sql}");
+        assert!(emitted_trigger_fields(&sql).contains("sortAtUnix"));
+    }
+
+    /// [PR-B4 mapping side + PR-M5] Index config: sortAt is INGESTED not computed,
+    /// the sortAtUnix→sortAt mapping keeps ms_to_seconds + fallback semantics, and
+    /// model3dId is indexed. Parsed generically (the `serde_yaml` FEATURE flag
+    /// that gates Config::from_yaml is not active under --features pg-sync, but the
+    /// crate is available, so we walk the YAML value tree directly).
+    #[test]
+    fn test_index_config_sortat_ingested_and_model3did() {
+        let yaml = std::fs::read_to_string(
+            concat!(env!("CARGO_MANIFEST_DIR"), "/deploy/configs/civitai-index.yaml")
+        ).expect("deploy/configs/civitai-index.yaml must exist");
+        let root: serde_yaml::Value =
+            serde_yaml::from_str(&yaml).expect("civitai-index.yaml must be valid YAML");
+
+        let config = &root["config"];
+
+        // sortAt sort field: present, and NO `computed:` block ([PR-M5]).
+        let sort_fields = config["sort_fields"].as_sequence().expect("sort_fields seq");
+        let sort_at = sort_fields.iter()
+            .find(|f| f["name"].as_str() == Some("sortAt"))
+            .expect("sortAt sort field must exist");
+        assert!(sort_at.get("computed").is_none(),
+            "sortAt must NOT be computed — ingested and computed must never coexist ([PR-M5])");
+
+        // model3dId filter field: single_value, per_value_lazy.
+        let filter_fields = config["filter_fields"].as_sequence().expect("filter_fields seq");
+        let model3did = filter_fields.iter()
+            .find(|f| f["name"].as_str() == Some("model3dId"))
+            .expect("model3dId filter field must exist");
+        assert_eq!(model3did["field_type"].as_str(), Some("single_value"));
+        assert_eq!(model3did["per_value_lazy"].as_bool(), Some(true));
+
+        // time_buckets pins the query-side field/units.
+        let tb = &config["time_buckets"];
+        assert_eq!(tb["filter_field"].as_str(), Some("sortAtUnix"));
+        assert_eq!(tb["sort_field"].as_str(), Some("sortAt"));
+
+        // data_schema: sortAtUnix→sortAt mapping intact (ms_to_seconds + fallback),
+        // and model3dId ingested.
+        let ds_fields = root["data_schema"]["fields"].as_sequence().expect("data_schema.fields seq");
+        let sortat_map = ds_fields.iter()
+            .find(|f| f["source"].as_str() == Some("sortAtUnix") && f["target"].as_str() == Some("sortAt"))
+            .expect("sortAtUnix->sortAt data_schema mapping must exist");
+        assert_eq!(sortat_map["ms_to_seconds"].as_bool(), Some(true),
+            "sortAtUnix->sortAt must divide ms→s (emission is ms)");
+        assert_eq!(sortat_map["fallback"].as_str(), Some("sortAt"),
+            "dump path emits `sortAt` seconds via this fallback (no ms division)");
+        assert!(ds_fields.iter().any(|f| f["target"].as_str() == Some("model3dId")),
+            "model3dId must be in data_schema");
+    }
 }