The initial-snapshot orchestration core (C2 Task 8): a GenServer that interposes as the Capstan.AssemblerServer's sink and its :watermark_observer, runs the cursor-gate over the live stream, drives the Capstan.Snapshot.ChunkReader, and emits each backfill chunk through the real sink's Capstan.Sink.handle_snapshot/2 once the stream's processed watermark covers the chunk's exact GTID position G.

Two roles, two message paths

• As the AssemblerServer's sink. The coordinator implements the Capstan.Sink behaviour and is name-registered under Capstan.Snapshot.Coordinator. The AssemblerServer's state.sink is this MODULE, so its handle_transaction/1 and handle_schema_change/2 dispatch here; each is a thin module function that GenServer.calls the registered coordinator (resolved at call time, not start time — this is what breaks the start-order cycle, design § Pinned decisions #4). The coordinator does NOT edit assembler_server.ex; Task 7 supplies the attach_coordinator/2 injection + the silent-death monitor.

• As the :watermark_observer. After a deferred attach_coordinator/2, the AssemblerServer sends {:capstan_watermark, gtid_set_string} on EVERY checkpoint advance (delivered, filtered, AND self-committing DDL — the Ch6 choke-point guarantee). The coordinator feeds that string into the advance gate.

The cursor-gate (handle_transaction/1)

For each streamed %Capstan.Change{} on a snapshot-active table (in the snapshot set and not yet done), the coordinator applies Capstan.Snapshot.CursorGate.classify/3 per row-image with the per-table cursor + PK shape, and forwards the surviving subset to the real sink as a transaction carrying only those changes. A change on a NON-snapshot table — or on a table whose backfill is already done (the stream is authoritative) — passes through unchanged.

When ALL changes are suppressed, the coordinator returns {:ok, position} WITHOUT calling the real sink: a fully-suppressed transaction is exactly a filtered transaction from the real sink's view (ADR-0003), and the AssemblerServer still advances the watermark for it (assembler_server.ex:246/:257). Returning the txn's position keeps a fully-suppressed run from stalling the advance gate → deadlock.

The advance gate

The coordinator holds AT MOST ONE buffered chunk (its rows + its exact G). When the processed watermark covers G (CursorGate.advance?/2), the emit runs in THREE ordered steps:

1. Advance delivered_pk to chunk.max_pk and persist — BEFORE the sink emit.
2. Emit the chunk as a list of %Change{op: :snapshot} through the real sink's handle_snapshot/2 with a value-free %Capstan.Snapshot.Meta{}.
3. Advance pk_cursor to chunk.max_pk, persist, telemetry, drive the next chunk.

The pk_cursor (re-read floor) still advances AFTER the emit — the at-least-once boundary (tripwire 16): a crash between the sink's {:ok} and the pk_cursor persist re-emits the one chunk (bounded dup, C1's posture; an upsert-by-PK sink converges). The NEW step 1 persists the DELIVERED high-water first, so in that same crash window delivered_pk survives AHEAD of the rolled-back pk_cursor; on restart the cursor-gate forwards a streamed delete of an already-delivered key (k ≤ delivered_pk) instead of suppressing it, sweeping what would otherwise be a permanent crash-window phantom (closeout F1, Capstan.Snapshot.CursorGate and Capstan.Snapshot.State). A step-1 persist fault halts fail-closed before any emit.

Fail-closed halts (symmetric with C1)

A coordinator fault sends {:capstan_halt, reason} to the AssemblerServer (the LOUD path) and stops with {:shutdown, {:halt, reason}} (restart: :temporary, never restarted). The SILENT-death path — a coordinator that dies without messaging — is the AssemblerServer's Process.monitor (:snapshot_coordinator_down, Task 7). A raise in the emit/reconcile path is scrubbed value-free to {:snapshot_processing_crashed, Capstan.Error.from(exc)} exactly as the AssemblerServer scrubs a delivery-path raise (assembler_server.ex:148-158); a handle_snapshot/2 {:error, _} halts {:snapshot_sink_error, _} (the outer atom only in telemetry). A DDL on a snapshot-active table halts :snapshot_schema_drifted.

Rule 1

Row values (chunk rows, canonical PKs, the per-table cursor) travel ONLY in the delivered %Change{op: :snapshot} list and in in-memory state; they are never logged or telemetered. Telemetry ([:capstan, :snapshot, :started | :chunk_completed | :completed | :halt]) carries counts in MEASUREMENTS and only structural identity (schema/table/reason) in METADATA, gated by Capstan.Telemetry's value-free allowlist (a stray PK/cursor raises). The coordinator's own struct holds %Capstan.Snapshot.Chunk{} / %Capstan.Snapshot.State{}, each of which derives a value-eliding Inspect, so even an incidental inspect of the state never surfaces a value; and every halt is a graceful {:shutdown, _} exit (never an abnormal crash), so the in-flight message is never dumped to a crash report.