@spec assembler_spec(
  keyword(),
  Capstan.AssemblerServer.checkpoint_store()
) :: Supervisor.child_spec()

The AssemblerServer child spec, wired to the started checkpoint store {impl, handle}.

@spec checkpoint_store(keyword()) :: {module(), keyword()}

The lib-owned checkpoint store's implementation module and its start_link/1 options, drawn from checkpoint_store: [module: impl, options: keyword()].

@spec connection_spec(
  keyword(),
  pid(),
  Capstan.Position.t() | nil,
  Capstan.AssemblerServer.checkpoint_store() | nil
) :: Supervisor.child_spec()

The Connection child spec, wired so its :receiver is the AssemblerServer PID, its dump resumes from start_position, and — given the started checkpoint store {impl, handle} — it re-reads the durable resume position on every establish (so a reconnect resumes from the current watermark, not the frozen start-up position; design Q7 / F1). checkpoint_store defaults to nil so a Connection wired without a store (a unit test) keeps the injected start_position.

@spec coordinator_spec(
  keyword(),
  coordinator_wiring()
) :: Supervisor.child_spec()

The Capstan.Snapshot.Coordinator child spec (snapshot mode only), mirroring assembler_spec/2.

The real (downstream) sink + the retry budget come from the pipeline opts; wiring carries the bootstrap-produced pieces (see coordinator_wiring/0) — Task 10's supervisor builds it once the assembler + snapshot-store children have started. A :temporary child: a fail-closed snapshot halt never restarts.

@spec lib_mode?(keyword()) :: boolean()

Is this a lib-owned pipeline? True iff a non-nil :checkpoint_store is configured.

@spec resolve_start_position(
  keyword(),
  Capstan.Position.t() | nil
) ::
  {:ok, Capstan.Position.t() | nil}
  | {:error,
     :start_position_override_unsupported
     | :start_position_current_unsupported
     | :config_invalid}

Resolve the public :start_position (default :checkpoint) against the position the checkpoint store resumed with.

• :checkpoint — the resumed position (a %Position{} or nil for a fresh start);
• a %Capstan.Position{} — {:error, :start_position_override_unsupported} in C1 (an explicit override would resume the Connection's dump but NOT the AssemblerServer's watermark, which seeds from the store alone — a silent hole; so it is refused fail-closed, matching Capstan.start_link/1's pre-flight check);
• :current — {:error, :start_position_current_unsupported} (C1 does not implement "start from the server's current position"; it needs a live pre-connect query the spine does not yet wire);
• anything else — {:error, :config_invalid}.

@spec snapshot_mode?(keyword()) :: boolean()

Is this an initial-snapshot pipeline? True iff a non-nil :snapshot block is configured (mirrors lib_mode?/1). An absent (or nil) :snapshot key ⇒ pure C1.

@spec snapshot_store_spec(
  module(),
  keyword()
) :: Supervisor.child_spec()

The durable Capstan.SnapshotStore child spec (snapshot mode only), mirroring store_spec/2.

A :temporary child — a snapshot-store fault halts fail-closed, never restarts into a re-scan-from-zero livelock.

@spec store_spec(
  module(),
  keyword()
) :: Supervisor.child_spec()

The checkpoint-store child spec (a :temporary child — a store fault halts, never restarts).

@spec validate_sink(keyword()) :: :ok | {:error, sink_error()}

Validate the configured :sink against its checkpoint mode's required callbacks.

Returns :ok, or a distinct value-free refusal naming the missing callback. Every check is a runtime function_exported?/3 (the callbacks are @optional_callbacks).

@spec validate_snapshot_tables(
  keyword(),
  Capstan.Config.snapshot_config() | nil
) :: :ok | {:error, :snapshot_table_not_captured}

Refuse a snapshot table outside the capture allowlist (:snapshot_table_not_captured).

Every snapshot table MUST be captured by the stream: the stream is what delivers the gtid > G changes the cursor-gate suppresses from the chunk, so a snapshot table the stream does not carry would silently gap. :all capture trivially includes every snapshot table. snapshot is the normalised Capstan.Config.snapshot_config() (or nil in pure C1 — a no-op :ok).