diff --git a/.changeset/subquery-move-replay-on-restart.md b/.changeset/subquery-move-replay-on-restart.md
new file mode 100644
index 0000000000..8a9b26302f
--- /dev/null
+++ b/.changeset/subquery-move-replay-on-restart.md
@@ -0,0 +1,9 @@
+---
+'@core/sync-service': patch
+---
+
+Fix optimized streaming subquery shapes losing dependency move-ins/move-outs
+across a graceful server restart. On restart the dependency materializer now
+replays the moves each outer consumer missed (deduplicated by a persisted
+per-dependency source-LSN position), so the outer shape catches up instead of
+diverging from Postgres.
diff --git a/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex b/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
index a2223f8c60..04e371b050 100644
--- a/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
@@ -15,6 +15,7 @@ defmodule Electric.ShapeCache.InMemoryStorage do
   @snapshot_start_index 0
   @snapshot_end_index :end
   @pg_snapshot_key :pg_snapshot
+  @move_positions_key :move_positions
   @latest_offset_key :latest_offset
 
   defstruct [
@@ -127,6 +128,20 @@ defmodule Electric.ShapeCache.InMemoryStorage do
     :ok
   end
 
+  @impl Electric.ShapeCache.Storage
+  def set_move_positions!(move_positions, %MS{} = opts) do
+    :ets.insert(opts.snapshot_table, {@move_positions_key, move_positions})
+    :ok
+  end
+
+  @impl Electric.ShapeCache.Storage
+  def fetch_move_positions(%MS{} = opts) do
+    case :ets.lookup(opts.snapshot_table, @move_positions_key) do
+      [{@move_positions_key, move_positions}] -> {:ok, move_positions}
+      [] -> {:ok, %{}}
+    end
+  end
+
   @impl Electric.ShapeCache.Storage
   def get_all_stored_shape_handles(_opts), do: {:ok, MapSet.new()}
 
@@ -188,6 +203,50 @@ defmodule Electric.ShapeCache.InMemoryStorage do
     get_offset_indexed_stream(offset, max_offset, opts.log_table)
   end
 
+  @impl Electric.ShapeCache.Storage
+  def get_log_stream_with_offsets(offset, max_offset, %MS{} = opts)
+      when is_log_offset_lt(offset, @snapshot_boundary_offset) do
+    case :ets.lookup_element(opts.snapshot_table, snapshot_end(), 2, nil) do
+      nil ->
+        stream_from_snapshot_with_offsets(offset, max_offset, opts)
+
+      max when is_log_offset_lt(offset, max) ->
+        stream_from_snapshot_with_offsets(offset, max_offset, opts)
+
+      _ ->
+        get_offset_indexed_stream_with_offsets(offset, max_offset, opts.log_table)
+    end
+  end
+
+  def get_log_stream_with_offsets(offset, max_offset, %MS{} = opts) do
+    get_offset_indexed_stream_with_offsets(offset, max_offset, opts.log_table)
+  end
+
+  # Snapshot lines precede all real offsets, so tag them with `before_all/0`.
+  defp stream_from_snapshot_with_offsets(offset, max_offset, %MS{} = opts) do
+    offset
+    |> stream_from_snapshot(max_offset, opts)
+    |> Stream.map(&{LogOffset.before_all(), &1})
+  end
+
+  defp get_offset_indexed_stream_with_offsets(offset, max_offset, offset_indexed_table) do
+    offset = storage_offset(offset)
+    max_offset = storage_offset(max_offset)
+
+    Stream.unfold(offset, fn offset ->
+      case :ets.next_lookup(offset_indexed_table, {:offset, offset}) do
+        :"$end_of_table" ->
+          nil
+
+        {{:offset, position}, _} when position > max_offset ->
+          nil
+
+        {{:offset, position}, [{_, item}]} ->
+          {{LogOffset.new(position), item}, position}
+      end
+    end)
+  end
+
   defp stream_from_snapshot(offset, max_offset, %MS{} = opts) do
     ConcurrentStream.stream_to_end(
       excluded_start_key: snapshot_chunk_end(storage_offset(offset)),
diff --git a/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex b/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
index 137146010c..a2e2db9eaf 100644
--- a/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
@@ -69,6 +69,7 @@ defmodule Electric.ShapeCache.PureFileStorage do
     :last_persisted_txn_offset,
     :snapshot_started?,
     :pg_snapshot,
+    :move_positions,
     :last_snapshot_chunk,
     :compaction_started?,
     :compaction_boundary
@@ -464,6 +465,18 @@ defmodule Electric.ShapeCache.PureFileStorage do
     {:ok, read_cached_metadata(opts, :pg_snapshot)}
   end
 
+  # move_positions is written only when an outer subquery consumer applies a
+  # dependency move and read only at consumer startup, so it is persisted
+  # directly to disk (term-encoded) rather than through the ETS metadata cache.
+  def set_move_positions!(move_positions, %__MODULE__{} = opts) do
+    write_metadata!(opts, :move_positions, move_positions)
+    :ok
+  end
+
+  def fetch_move_positions(%__MODULE__{} = opts) do
+    {:ok, read_metadata!(opts, :move_positions) || %{}}
+  end
+
   defp read_latest_offset(%__MODULE__{} = opts) do
     read_multiple_cached_metadata(opts, [
       :last_seen_txn_offset,
@@ -1022,7 +1035,59 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
-  defp stream_main_log(min_offset, max_offset, %__MODULE__{} = opts) do
+  def get_log_stream_with_offsets(%LogOffset{} = min_offset, %LogOffset{} = max_offset, opts)
+      when is_last_virtual_offset(min_offset) or is_real_offset(min_offset) do
+    stream_main_log(min_offset, max_offset, opts, with_offsets?: true)
+  end
+
+  def get_log_stream_with_offsets(
+        %LogOffset{op_offset: op_offset} = min_offset,
+        %LogOffset{} = max_offset,
+        %__MODULE__{} = opts
+      ) do
+    metadata = read_multiple_cached_metadata(opts, [:snapshot_started?, :last_snapshot_chunk])
+
+    snapshot_started? = Keyword.get(metadata, :snapshot_started?) || false
+    last_snapshot_chunk = Keyword.get(metadata, :last_snapshot_chunk)
+
+    if not snapshot_started? and not shape_gone?(opts) do
+      raise(Storage.Error, message: "Snapshot not started")
+    end
+
+    case {last_snapshot_chunk, min_offset} do
+      {_, x} when is_min_offset(x) ->
+        snapshot_lines_with_offsets(Snapshot.stream_chunk_lines(opts, 0))
+
+      {%LogOffset{} = latest, min_offset} when is_log_offset_lt(min_offset, latest) ->
+        snapshot_lines_with_offsets(Snapshot.stream_chunk_lines(opts, op_offset + 1))
+
+      {nil, _offset} ->
+        snapshot_lines_with_offsets(wait_for_chunk_file_or_snapshot_end(opts, op_offset + 1))
+
+      {%LogOffset{}, offset} ->
+        stream_main_log(offset, max_offset, opts, with_offsets?: true)
+    end
+  end
+
+  # Snapshot lines precede all real offsets, so tag them with `before_all/0`.
+  defp snapshot_lines_with_offsets(stream),
+    do: Stream.map(stream, &{LogOffset.before_all(), &1})
+
+  defp stream_main_log(min_offset, max_offset, opts, gen_opts \\ [])
+
+  defp stream_main_log(min_offset, max_offset, %__MODULE__{} = opts, gen_opts) do
+    with_offsets? = Keyword.get(gen_opts, :with_offsets?, false)
+
+    ets_reader =
+      if with_offsets?,
+        do: &read_range_from_ets_cache_with_offsets/3,
+        else: &read_range_from_ets_cache/3
+
+    disk_streamer =
+      if with_offsets?,
+        do: &stream_from_disk_with_offsets/4,
+        else: &stream_from_disk/4
+
     storage_meta(
       ets_table: ets,
       last_persisted_offset: last_persisted,
@@ -1054,34 +1119,34 @@ defmodule Electric.ShapeCache.PureFileStorage do
 
       is_log_offset_lte(last_persisted, min_offset) ->
         # Pure ETS read case
-        case read_range_from_ets_cache(ets, min_offset, upper_read_bound) do
+        case ets_reader.(ets, min_offset, upper_read_bound) do
           {_data, last_offset}
           when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
             # Empty or partial read - ETS was cleared by a concurrent flush.
             # Data is now on disk (flush writes to disk before clearing ETS),
             # so read directly from there using existing boundary info.
-            stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+            disk_streamer.(opts, min_offset, upper_read_bound, boundary_info)
 
           {data, _last_offset} ->
             data
         end
 
       is_log_offset_lte(upper_read_bound, last_persisted) ->
-        stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+        disk_streamer.(opts, min_offset, upper_read_bound, boundary_info)
 
       true ->
         # Mixed disk + ETS case
         # Because ETS may be cleared by a flush in a parallel process, we're reading it out into memory.
         # It's expected to be fairly small in the worst case, up 64KB
-        case read_range_from_ets_cache(ets, last_persisted, upper_read_bound) do
+        case ets_reader.(ets, last_persisted, upper_read_bound) do
           {_upper_range, last_offset}
           when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
             # Empty or partial read - ETS was cleared by a concurrent flush.
             # Data is now on disk, so read the full range from there.
-            stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+            disk_streamer.(opts, min_offset, upper_read_bound, boundary_info)
 
           {upper_range, _last_offset} ->
-            stream_from_disk(opts, min_offset, last_persisted, boundary_info)
+            disk_streamer.(opts, min_offset, last_persisted, boundary_info)
             |> Stream.concat(upper_range)
         end
     end
@@ -1160,6 +1225,43 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
+  # Offset-preserving counterpart to `read_range_from_ets_cache/3`, yielding
+  # `{LogOffset.t(), item}` pairs. `last_offset` stays a tuple for comparison with
+  # `upper_read_bound_tuple` in `stream_main_log/4`.
+  defp read_range_from_ets_cache_with_offsets(nil, _min, _max), do: {[], nil}
+
+  defp read_range_from_ets_cache_with_offsets(ets, %LogOffset{} = min, %LogOffset{} = max) do
+    read_range_from_ets_cache_with_offsets(
+      ets,
+      LogOffset.to_tuple(min),
+      LogOffset.to_tuple(max),
+      [],
+      nil
+    )
+  end
+
+  defp read_range_from_ets_cache_with_offsets(ets, min, {max_tx, max_op} = max, acc, last_offset) do
+    case safe_next_lookup(ets, min) do
+      :ets_dead ->
+        {Enum.reverse(acc), last_offset}
+
+      :"$end_of_table" ->
+        {Enum.reverse(acc), last_offset}
+
+      {{min_tx, min_op}, _} when min_tx > max_tx or (min_tx == max_tx and min_op > max_op) ->
+        {Enum.reverse(acc), last_offset}
+
+      {new_min, [{_, item}]} ->
+        read_range_from_ets_cache_with_offsets(
+          ets,
+          new_min,
+          max,
+          [{LogOffset.new(new_min), item} | acc],
+          new_min
+        )
+    end
+  end
+
   # The owning Consumer's buffer ETS table may be destroyed during terminate.
   # Falling back is safe because terminate flushes to disk before deleting.
   defp safe_next_lookup(ets, min) do
@@ -1205,6 +1307,47 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
+  # Offset-preserving counterpart to `stream_from_disk/4`, yielding
+  # `{LogOffset.t(), json}` pairs.
+  defp stream_from_disk_with_offsets(%__MODULE__{}, min_offset, max_offset, _)
+       when is_log_offset_lte(max_offset, min_offset),
+       do: []
+
+  defp stream_from_disk_with_offsets(
+         %__MODULE__{} = opts,
+         min_offset,
+         max_offset,
+         boundary_info
+       ) do
+    suffix = get_suffix(min_offset, boundary_info)
+
+    case fetch_chunk(min_offset, opts, boundary_info) do
+      {:ok, chunk_end_offset, {start_pos, end_pos}} when not is_nil(end_pos) ->
+        LogFile.stream_entries_with_offsets(
+          opts,
+          json_file(opts, suffix),
+          start_pos,
+          end_pos,
+          min_offset
+        )
+        |> Stream.concat(
+          stream_from_disk_with_offsets(opts, chunk_end_offset, max_offset, boundary_info)
+        )
+
+      {:ok, nil, {start_pos, nil}} ->
+        LogFile.stream_entries_until_offset_with_offsets(
+          opts,
+          json_file(opts, suffix),
+          start_pos,
+          min_offset,
+          max_offset
+        )
+
+      :error ->
+        []
+    end
+  end
+
   defp get_suffix(min_offset, {_, {compaction_boundary, compacted_name}, _})
        when is_log_offset_lt(min_offset, compaction_boundary),
        do: compacted_name
diff --git a/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex b/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex
index 65533706fb..bc7bb9a316 100644
--- a/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex
+++ b/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex
@@ -282,6 +282,144 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
     )
   end
 
+  @doc """
+  Like `stream_jsons/5` but yields `{LogOffset.t(), json}` pairs, preserving each
+  entry's on-disk offset (used to position control messages, which carry no
+  offset in their JSON headers).
+  """
+  def stream_entries_with_offsets(
+        %PFS{} = opts,
+        log_file_path,
+        start_position,
+        end_position,
+        exclusive_min_offset
+      ) do
+    case safely_open_file!(opts, log_file_path, [:read, :raw]) do
+      {:halt, :data_removed} ->
+        []
+
+      {:ok, file} ->
+        try do
+          with {:ok, data} <- :file.pread(file, start_position, end_position - start_position) do
+            {entries, _} = extract_entries_from_binary(data, exclusive_min_offset, nil)
+            entries
+          else
+            :eof ->
+              raise "unexpected end of file"
+
+            {:error, reason} ->
+              raise File.Error,
+                path: log_file_path,
+                reason: reason,
+                action: "pread(#{start_position}, #{end_position - start_position})"
+          end
+        after
+          File.close(file)
+        end
+    end
+  end
+
+  @doc "Like `stream_jsons_until_offset/5` but yields `{LogOffset.t(), json}` pairs."
+  def stream_entries_until_offset_with_offsets(
+        %PFS{} = opts,
+        log_file_path,
+        start_position,
+        exclusive_min_offset,
+        inclusive_max_offset
+      ) do
+    Stream.resource(
+      fn ->
+        case safely_open_file!(opts, log_file_path, [:read, :raw]) do
+          {:ok, file} ->
+            {:ok, ^start_position} = :file.position(file, start_position)
+            {file, ""}
+
+          {:halt, :data_removed} ->
+            :halt
+        end
+      end,
+      fn
+        :halt ->
+          {:halt, []}
+
+        {file, binary_rest} ->
+          case :file.read(file, 4096) do
+            {:ok, data} ->
+              {entries, rest} =
+                extract_entries_from_binary(
+                  binary_rest <> data,
+                  exclusive_min_offset,
+                  inclusive_max_offset
+                )
+
+              {entries, {file, rest}}
+
+            :eof ->
+              {:halt, {file, binary_rest}}
+          end
+      end,
+      fn
+        [] -> :ok
+        {file, _} -> File.close(file)
+      end
+    )
+  end
+
+  # Offset-preserving counterpart to `extract_jsons_from_binary/4`. Kept separate
+  # so the json-only hot path avoids the per-entry `LogOffset` allocation.
+  @spec extract_entries_from_binary(binary(), LogOffset.t(), LogOffset.t() | nil) ::
+          {[{LogOffset.t(), String.t()}], binary()}
+  defp extract_entries_from_binary(binary, exclusive_min_offset, inclusive_max_offset, acc \\ [])
+  defp extract_entries_from_binary(<<>>, _, _, acc), do: {Enum.reverse(acc), ""}
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, _::binary-size(json_size), rest::binary>>,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2} = log_offset,
+         inclusive_max_offset,
+         acc
+       )
+       when tx_offset1 < tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 <= op_offset2),
+       do: extract_entries_from_binary(rest, log_offset, inclusive_max_offset, acc)
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, json::binary-size(json_size), _::binary>>,
+         log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2} = inclusive_max_offset,
+         acc
+       )
+       when tx_offset1 == tx_offset2 and op_offset1 == op_offset2,
+       do:
+         extract_entries_from_binary("", log_offset, inclusive_max_offset, [
+           {LogOffset.new(tx_offset1, op_offset1), json} | acc
+         ])
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, _json::binary-size(json_size), _::binary>>,
+         log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2} = inclusive_max_offset,
+         acc
+       )
+       when tx_offset1 > tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 > op_offset2),
+       do: extract_entries_from_binary("", log_offset, inclusive_max_offset, acc)
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, json::binary-size(json_size), rest::binary>>,
+         log_offset,
+         inclusive_max_offset,
+         acc
+       ),
+       do:
+         extract_entries_from_binary(rest, log_offset, inclusive_max_offset, [
+           {LogOffset.new(tx_offset1, op_offset1), json} | acc
+         ])
+
+  defp extract_entries_from_binary(rest, _, _, acc),
+    do: {Enum.reverse(acc), rest}
+
   @spec extract_jsons_from_binary(binary(), LogOffset.t(), LogOffset.t() | nil) ::
           Enumerable.t(String.t())
   defp extract_jsons_from_binary(binary, exclusive_min_offset, inclusive_max_offset, acc \\ [])
diff --git a/packages/sync-service/lib/electric/shape_cache/storage.ex b/packages/sync-service/lib/electric/shape_cache/storage.ex
index eededf50fb..d62c8605d2 100644
--- a/packages/sync-service/lib/electric/shape_cache/storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/storage.ex
@@ -18,6 +18,11 @@ defmodule Electric.ShapeCache.Storage do
           filter_txns?: boolean()
         }
   @type offset :: LogOffset.t()
+  @typedoc """
+  Per-dependency "moves-applied-up-to" source LSN positions for an outer
+  subquery consumer, keyed by the dependency's shape handle.
+  """
+  @type move_positions :: %{shape_handle() => LogOffset.t()}
 
   @type compiled_opts :: term()
   @type shape_opts :: term()
@@ -77,6 +82,18 @@ defmodule Electric.ShapeCache.Storage do
 
   @callback set_pg_snapshot(pg_snapshot(), shape_opts()) :: :ok
 
+  @doc """
+  Persist the per-dependency moves-applied-up-to positions for an outer
+  subquery consumer.
+  """
+  @callback set_move_positions!(move_positions(), shape_opts()) :: :ok
+
+  @doc """
+  Fetch the per-dependency moves-applied-up-to positions for an outer subquery
+  consumer. Returns `{:ok, %{}}` when none have been persisted yet.
+  """
+  @callback fetch_move_positions(shape_opts()) :: {:ok, move_positions()} | {:error, term()}
+
   @doc "Check if snapshot for a given shape handle already exists"
   @callback snapshot_started?(shape_opts()) :: boolean()
 
@@ -171,6 +188,22 @@ defmodule Electric.ShapeCache.Storage do
   @callback get_log_stream(offset :: LogOffset.t(), max_offset :: LogOffset.t(), shape_opts()) ::
               log()
 
+  @doc """
+  Like `get_log_stream/3` but yields `{LogOffset.t(), log_item_json}` pairs so the
+  caller gets each item's authoritative storage offset.
+
+  Data-change log items carry their offset in their headers, but control messages
+  (subquery move-in/move-out events) do not, so they cannot be positioned by
+  re-parsing the JSON alone. Materializer replay uses this to place control
+  messages in the correct transaction. Snapshot lines (which precede all real
+  offsets) are yielded with `LogOffset.before_all/0`.
+  """
+  @callback get_log_stream_with_offsets(
+              offset :: LogOffset.t(),
+              max_offset :: LogOffset.t(),
+              shape_opts()
+            ) :: Enumerable.t({LogOffset.t(), binary()})
+
   @doc """
   Get the last exclusive offset of the chunk starting from the given offset.
 
@@ -320,6 +353,16 @@ defmodule Electric.ShapeCache.Storage do
     mod.set_pg_snapshot(pg_snapshot, shape_opts)
   end
 
+  @impl __MODULE__
+  def set_move_positions!(move_positions, {mod, shape_opts}) do
+    mod.set_move_positions!(move_positions, shape_opts)
+  end
+
+  @impl __MODULE__
+  def fetch_move_positions({mod, shape_opts}) do
+    mod.fetch_move_positions(shape_opts)
+  end
+
   @impl __MODULE__
   def snapshot_started?({mod, shape_opts}) do
     mod.snapshot_started?(shape_opts)
@@ -409,6 +452,19 @@ defmodule Electric.ShapeCache.Storage do
     []
   end
 
+  @impl __MODULE__
+  def get_log_stream_with_offsets(offset, max_offset \\ @last_log_offset, storage)
+
+  def get_log_stream_with_offsets(offset, max_offset, {mod, shape_opts})
+      when max_offset == @last_log_offset or not is_log_offset_lt(max_offset, offset) do
+    mod.get_log_stream_with_offsets(offset, max_offset, shape_opts)
+  end
+
+  def get_log_stream_with_offsets(offset, max_offset, _storage)
+      when is_log_offset_lt(max_offset, offset) do
+    []
+  end
+
   @impl __MODULE__
   def get_chunk_end_log_offset(offset, {mod, shape_opts}) do
     mod.get_chunk_end_log_offset(offset, shape_opts)
diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex
index 0935f7c9ea..fb746a3501 100644
--- a/packages/sync-service/lib/electric/shapes/consumer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer.ex
@@ -8,6 +8,7 @@ defmodule Electric.Shapes.Consumer do
   alias Electric.Shapes.Consumer.PendingTxn
   alias Electric.Shapes.Consumer.SetupEffects
   alias Electric.Shapes.Consumer.State
+  alias Electric.Shapes.Consumer.Subqueries.MoveQueue
 
   import Electric.Shapes.Consumer.State, only: :macros
   require Electric.Replication.LogOffset
@@ -352,6 +353,10 @@ defmodule Electric.Shapes.Consumer do
       "Consumer reacting to #{length(move_in)} move ins and #{length(move_out)} move outs from its #{dep_handle} dependency"
     end)
 
+    # Remember the source LSN of this move so that the per-dependency
+    # moves-position can be advanced once the move pipeline is fully drained.
+    state = record_pending_move_lsn(state, dep_handle, payload)
+
     handle_apply_event_result(
       state,
       apply_event(state, {:materializer_changes, dep_handle, payload})
@@ -478,6 +483,11 @@ defmodule Electric.Shapes.Consumer do
     # shape but the alternative is leaking ets tables.
     state = terminate_writer(state)
 
+    # `terminate_writer/1` flushes the writer, so any staged move positions are
+    # now durable — commit them so the persisted position matches storage across
+    # a graceful restart.
+    state = commit_all_move_positions(state)
+
     ShapeCleaner.handle_writer_termination(state.stack_id, state.shape_handle, reason)
 
     State.reply_to_snapshot_waiters(state, {:error, "Shape terminated before snapshot was ready"})
@@ -986,10 +996,133 @@ defmodule Electric.Shapes.Consumer do
             {{previous_offset, result.state.latest_offset}, result.state.latest_offset}
           end
 
-        {result.state, notification, result.num_changes, result.total_size}
+        final_state = maybe_stage_move_positions(result.state)
+
+        {final_state, notification, result.num_changes, result.total_size}
+    end
+  end
+
+  # Stash the source LSN carried by a materializer move payload, keyed by
+  # dependency handle. It is staged once the move pipeline is fully drained (see
+  # `maybe_stage_move_positions/1`) and only persisted once the writer confirms
+  # the flush (see `commit_flushed_move_positions/2`).
+  defp record_pending_move_lsn(state, dep_handle, payload) do
+    case Map.get(payload, :lsn) do
+      nil ->
+        state
+
+      lsn ->
+        %{state | pending_move_lsns: Map.put(state.pending_move_lsns, dep_handle, lsn)}
     end
   end
 
+  # Once the move pipeline is fully drained (Steady, empty queue) every received
+  # move has been applied to the writer buffer. Stage the received source LSNs,
+  # tagged with the current outer `latest_offset` as a flush threshold: the move's
+  # splice rows are at/below `latest_offset`, so they are durable once the writer
+  # has flushed to that offset. Staged entries are only advanced into (and
+  # persisted as) `move_positions` by `commit_flushed_move_positions/2` — so the
+  # persisted position never runs ahead of durable storage across a restart.
+  defp maybe_stage_move_positions(%State{pending_move_lsns: pending} = state)
+       when pending == %{},
+       do: state
+
+  defp maybe_stage_move_positions(%State{} = state) do
+    if move_pipeline_fully_drained?(state.event_handler) do
+      threshold = state.latest_offset
+
+      staged =
+        Enum.reduce(state.pending_move_lsns, state.staged_move_positions, fn {handle, lsn}, acc ->
+          Map.update(acc, handle, [{threshold, lsn}], &(&1 ++ [{threshold, lsn}]))
+        end)
+
+      %{state | staged_move_positions: staged, pending_move_lsns: %{}}
+    else
+      state
+    end
+  end
+
+  # Commit staged move positions whose splice rows are now durable (flush
+  # threshold at/below `flushed_offset`), advancing and persisting
+  # `move_positions`.
+  defp commit_flushed_move_positions(
+         %State{staged_move_positions: staged} = state,
+         _flushed_offset
+       )
+       when staged == %{},
+       do: state
+
+  defp commit_flushed_move_positions(%State{} = state, flushed_offset) do
+    {staged, positions, changed?} =
+      Enum.reduce(state.staged_move_positions, {%{}, state.move_positions, false}, fn
+        {handle, entries}, {staged_acc, positions_acc, changed} ->
+          {committed, remaining} =
+            Enum.split_while(entries, fn {threshold, _lsn} ->
+              LogOffset.is_log_offset_lte(threshold, flushed_offset)
+            end)
+
+          positions_acc =
+            case List.last(committed) do
+              nil -> positions_acc
+              {_threshold, lsn} -> Map.update(positions_acc, handle, lsn, &LogOffset.max(&1, lsn))
+            end
+
+          staged_acc =
+            if remaining == [], do: staged_acc, else: Map.put(staged_acc, handle, remaining)
+
+          {staged_acc, positions_acc, changed or committed != []}
+      end)
+
+    if changed? do
+      ShapeCache.Storage.set_move_positions!(positions, state.storage)
+      %{state | staged_move_positions: staged, move_positions: positions}
+    else
+      state
+    end
+  end
+
+  # Commit all staged move positions unconditionally. Called from `terminate/2`
+  # after `terminate_writer/1` has flushed the writer, so every staged move's
+  # splice rows are durable by then. Note this runs after `terminate_writer/1`
+  # has popped `:writer` off the state, so we match a bare map rather than
+  # `%State{}`.
+  defp commit_all_move_positions(state) do
+    staged = Map.get(state, :staged_move_positions, %{})
+    storage = Map.get(state, :storage)
+
+    if staged == %{} or is_nil(storage) do
+      state
+    else
+      positions =
+        Enum.reduce(staged, state.move_positions, fn {handle, entries}, acc ->
+          {_threshold, lsn} = List.last(entries)
+          Map.update(acc, handle, lsn, &LogOffset.max(&1, lsn))
+        end)
+
+      # A persistence failure here must not crash `terminate/2` (that would mask
+      # the shutdown reason, and storage may already be torn down), but it should
+      # not pass silently: a dropped position just means more moves are replayed
+      # on restart. Log it so a genuine failure is observable.
+      try do
+        ShapeCache.Storage.set_move_positions!(positions, storage)
+      rescue
+        error ->
+          Logger.warning(
+            "Failed to persist move positions for shape " <>
+              "#{Map.get(state, :shape_handle)} during terminate: #{Exception.message(error)}"
+          )
+      end
+
+      %{state | move_positions: positions, staged_move_positions: %{}}
+    end
+  end
+
+  defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Steady{queue: queue}),
+    do: MoveQueue.length(queue) == 0
+
+  defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Buffering{}), do: false
+  defp move_pipeline_fully_drained?(_handler), do: true
+
   defp handle_event_error(state, {:truncate, xid}) do
     handle_txn_with_truncate(xid, state)
   end
@@ -1166,7 +1299,7 @@ defmodule Electric.Shapes.Consumer do
   defp confirm_flushed_and_notify(state, flushed_offset) do
     {state, txn_offset} = State.align_offset_to_txn_boundary(state, flushed_offset)
     ShapeLogCollector.notify_flushed(state.stack_id, state.shape_handle, txn_offset)
-    state
+    commit_flushed_move_positions(state, flushed_offset)
   end
 
   # After a pending transaction completes and txn_offset_mapping is populated,
@@ -1199,41 +1332,68 @@ defmodule Electric.Shapes.Consumer do
   end
 
   defp finish_initialization(%State{} = state, action, otel_ctx) do
-    if all_materializers_alive?(state) do
-      case initialize_event_handler(state, action) do
-        {:ok, state} ->
-          Logger.debug("Writer for #{state.shape_handle} initialized")
-
-          # We start the snapshotter even if there's a snapshot because it also performs the call
-          # to PublicationManager.add_shape/3. We *could* do that call here and avoid spawning a
-          # process if the shape already has a snapshot but the current semantics rely on being able
-          # to wait for the snapshot asynchronously and if we called publication manager here it would
-          # block and prevent await_snapshot_start calls from adding snapshot subscribers.
-
-          {:ok, _pid} =
-            Shapes.DynamicConsumerSupervisor.start_snapshotter(
-              state.stack_id,
-              %{
-                stack_id: state.stack_id,
-                shape: state.shape,
-                shape_handle: state.shape_handle,
-                storage: state.storage,
-                otel_ctx: otel_ctx
-              }
-            )
-
-          {:noreply, state}
-
-        {:error, state} ->
-          stop_and_clean(state)
-      end
-    else
-      stop_and_clean(state)
+    case subscribe_to_materializers(state) do
+      {:ok, state} ->
+        case initialize_event_handler(state, action) do
+          {:ok, state} ->
+            Logger.debug("Writer for #{state.shape_handle} initialized")
+
+            # We start the snapshotter even if there's a snapshot because it also performs the call
+            # to PublicationManager.add_shape/3. We *could* do that call here and avoid spawning a
+            # process if the shape already has a snapshot but the current semantics rely on being able
+            # to wait for the snapshot asynchronously and if we called publication manager here it would
+            # block and prevent await_snapshot_start calls from adding snapshot subscribers.
+
+            {:ok, _pid} =
+              Shapes.DynamicConsumerSupervisor.start_snapshotter(
+                state.stack_id,
+                %{
+                  stack_id: state.stack_id,
+                  shape: state.shape,
+                  shape_handle: state.shape_handle,
+                  storage: state.storage,
+                  otel_ctx: otel_ctx
+                }
+              )
+
+            {:noreply, state}
+
+          {:error, state} ->
+            stop_and_clean(state)
+        end
+
+      :error ->
+        stop_and_clean(state)
     end
   end
 
-  defp all_materializers_alive?(state) do
-    Enum.all?(state.shape.shape_dependencies_handles, fn shape_handle ->
+  # Subscribe to each dependency materializer, passing the persisted per-dep
+  # moves-position so the materializer replays any moves this consumer missed
+  # across a restart. Captures the returned seed views (as-of the position) for
+  # seeding the event handler's dependency views, and baselines a position for
+  # dependencies that don't have one yet so a first missed move can be replayed.
+  #
+  # Returns `{:ok, state}` with `dep_seed_views`/`move_positions` populated, or
+  # `:error` if any dependency materializer is not alive.
+  defp subscribe_to_materializers(state) do
+    case do_subscribe_to_materializers(state) do
+      {:ok, %State{move_positions: move_positions} = state} ->
+        # Persist baselined positions so a restart before the first move can
+        # still replay it (no-op writes if nothing changed are cheap at startup).
+        if state.shape.shape_dependencies_handles != [] do
+          ShapeCache.Storage.set_move_positions!(move_positions, state.storage)
+        end
+
+        {:ok, state}
+
+      :error ->
+        :error
+    end
+  end
+
+  defp do_subscribe_to_materializers(state) do
+    Enum.reduce_while(state.shape.shape_dependencies_handles, {:ok, state}, fn shape_handle,
+                                                                               {:ok, state} ->
       name = Materializer.name(state.stack_id, shape_handle)
 
       with pid when is_pid(pid) <- GenServer.whereis(name),
@@ -1242,9 +1402,19 @@ defmodule Electric.Shapes.Consumer do
           tag: {:dependency_materializer_down, shape_handle}
         )
 
-        Materializer.subscribe(pid)
+        from_lsn = Map.get(state.move_positions, shape_handle)
+        {:ok, seed_view, applied_offset} = Materializer.subscribe(pid, from_lsn)
+
+        move_positions =
+          Map.put_new(state.move_positions, shape_handle, applied_offset)
+
+        state = %{
+          state
+          | dep_seed_views: Map.put(state.dep_seed_views, shape_handle, seed_view),
+            move_positions: move_positions
+        }
 
-        true
+        {:cont, {:ok, state}}
       else
         _ ->
           Logger.warning(
@@ -1253,7 +1423,7 @@ defmodule Electric.Shapes.Consumer do
             state_shape_handle: state.shape_handle
           )
 
-          false
+          {:halt, :error}
       end
     end)
   end
diff --git a/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex b/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex
index feb251e914..db2e7d3316 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex
@@ -22,7 +22,18 @@ defmodule Electric.Shapes.Consumer.EventHandlerBuilder do
                                          {views, handle_mapping, index_mapping} ->
         materializer_opts = %{stack_id: state.stack_id, shape_handle: handle}
         :ok = Materializer.wait_until_ready(materializer_opts)
-        view = Materializer.get_link_values(materializer_opts)
+
+        # Seed the dependency view from the value captured at subscribe time
+        # (as-of this consumer's persisted moves-position), so that any moves
+        # the materializer replays are not eliminated as redundant against a
+        # view that already reflects them. Falls back to the materializer's
+        # current link values if no seed was captured (non-restart paths).
+        view =
+          case Map.fetch(state.dep_seed_views, handle) do
+            {:ok, seed_view} -> seed_view
+            :error -> Materializer.get_link_values(materializer_opts)
+          end
+
         ref = ["$sublink", Integer.to_string(index)]
 
         {Map.put(views, ref, view), Map.put(handle_mapping, handle, {index, ref}),
diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
index 1315c277b0..d62f065cad 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -103,17 +103,37 @@ defmodule Electric.Shapes.Consumer.Materializer do
     end)
   end
 
-  # A materializer cannot answer any call until `handle_continue(:start_materializer)`
-  # returns, and that blocks on `await_snapshot_start(:infinity)` until its own snapshot
-  # starts. A cold dependency snapshot can take longer than the default 5s `GenServer.call`
-  # timeout, so we wait with `:infinity` (consistent with `wait_until_ready/1` and
-  # `new_changes/3`). Liveness is handled by the caller monitoring the materializer, so a
-  # dead materializer surfaces as a call exit rather than being masked by a short timeout.
-  def subscribe(pid) when is_pid(pid), do: GenServer.call(pid, :subscribe, :infinity)
+  @doc """
+  Subscribe `pid` to this materializer's move events.
+
+  `from_lsn` is the source LSN (LogOffset) up to which the subscribing outer
+  consumer has already applied moves from this dependency, or `nil` for a fresh
+  subscription. When `from_lsn` is behind the materializer's applied position,
+  the moves in `(from_lsn, applied_offset]` are replayed to `pid` so it can
+  catch up after a restart.
+
+  Returns `{:ok, seed_link_values, applied_offset}` where `seed_link_values` is
+  the set of link values as of `from_lsn` (used to seed the outer consumer's
+  dependency view so replayed moves are not redundancy-eliminated), and
+  `applied_offset` is the materializer's current applied source LSN.
+
+  Calls use an `:infinity` timeout: a materializer cannot answer until
+  `handle_continue(:start_materializer)` returns, which blocks on
+  `await_snapshot_start(:infinity)`, and a cold dependency snapshot can take
+  longer than the default 5s `GenServer.call` timeout (consistent with
+  `wait_until_ready/1` and `new_changes/3`). Liveness is handled by the caller
+  monitoring the materializer, so a dead materializer surfaces as a call exit
+  rather than being masked by a short timeout.
+  """
+  def subscribe(pid, from_lsn \\ nil)
 
-  def subscribe(opts) when is_map(opts), do: GenServer.call(name(opts), :subscribe, :infinity)
+  def subscribe(pid, from_lsn) when is_pid(pid),
+    do: GenServer.call(pid, {:subscribe, from_lsn}, :infinity)
 
-  def subscribe(stack_id, shape_handle),
+  def subscribe(opts, from_lsn) when is_map(opts),
+    do: GenServer.call(name(opts), {:subscribe, from_lsn}, :infinity)
+
+  def subscribe(stack_id, shape_handle) when is_stack_id(stack_id),
     do: subscribe(%{stack_id: stack_id, shape_handle: shape_handle})
 
   def start_link(opts) do
@@ -138,6 +158,10 @@ defmodule Electric.Shapes.Consumer.Materializer do
         value_counts: %{},
         pending_events: %{},
         offset: LogOffset.before_all(),
+        # The highest source LSN (LogOffset) up to which changes have been
+        # applied to `value_counts`. Used to tag emitted moves with their
+        # source LSN and to bound move replay on subscribe.
+        applied_offset: LogOffset.before_all(),
         subscribed_offset: nil,
         ref: nil,
         subscribers: MapSet.new()
@@ -178,6 +202,14 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
   def handle_continue({:read_stream, storage}, state) do
     state = read_history_up_to_subscribed(state, storage)
+    # After the startup replay, everything up to `subscribed_offset` has been
+    # applied; seed `applied_offset` accordingly so live moves and replay are
+    # tagged/bounded from the right position.
+    state =
+      if is_nil(state.subscribed_offset),
+        do: state,
+        else: %{state | applied_offset: state.subscribed_offset}
+
     write_link_values(state)
     {:noreply, state}
   end
@@ -201,7 +233,12 @@ defmodule Electric.Shapes.Consumer.Materializer do
   changes after this offset will be delivered via new_changes messages
   from the Consumer.
   """
-  def read_history_up_to_subscribed(state, storage) do
+  def read_history_up_to_subscribed(
+        state,
+        storage,
+        apply_fun \\ &default_history_apply/2,
+        with_offsets? \\ false
+      ) do
     cond do
       is_nil(state.subscribed_offset) ->
         state
@@ -210,8 +247,14 @@ defmodule Electric.Shapes.Consumer.Materializer do
         state
 
       true ->
-        stream = Storage.get_log_stream(state.offset, state.subscribed_offset, storage)
-        {state, _} = stream |> decode_json_stream() |> apply_changes(state)
+        stream =
+          if with_offsets? do
+            Storage.get_log_stream_with_offsets(state.offset, state.subscribed_offset, storage)
+          else
+            Storage.get_log_stream(state.offset, state.subscribed_offset, storage)
+          end
+
+        state = apply_fun.(stream, state)
 
         # If the read just covered the main log (because either the
         # current offset is already past the snapshot or the next chunk
@@ -252,12 +295,204 @@ defmodule Electric.Shapes.Consumer.Materializer do
               %{state | offset: state.subscribed_offset}
 
             true ->
-              read_history_up_to_subscribed(%{state | offset: next_offset}, storage)
+              read_history_up_to_subscribed(
+                %{state | offset: next_offset},
+                storage,
+                apply_fun,
+                with_offsets?
+              )
+          end
+        end
+    end
+  end
+
+  # Default apply function for `read_history_up_to_subscribed/3`: apply the
+  # decoded stream to `value_counts`, discarding the emitted move events (the
+  # startup replay only needs the resulting state).
+  defp default_history_apply(stream, state) do
+    {state, _events} = stream |> decode_json_stream() |> apply_changes(state)
+    state
+  end
+
+  # Replay the moves in `(from_lsn, applied_offset]` to `pid`, returning the set
+  # of link values as of `from_lsn` (the seed view for the outer consumer).
+  #
+  # When `from_lsn` is nil (fresh subscription) or is already at/after the
+  # materializer's applied position, there is nothing to replay and the current
+  # link values are returned as the seed.
+  defp maybe_replay_moves(state, _pid, nil), do: link_values_from_counts(state.value_counts)
+
+  defp maybe_replay_moves(state, pid, from_lsn) do
+    if is_log_offset_lte(state.applied_offset, from_lsn) do
+      link_values_from_counts(state.value_counts)
+    else
+      replay_moves(state, pid, from_lsn)
+    end
+  end
+
+  defp replay_moves(state, pid, from_lsn) do
+    stack_storage = Storage.for_stack(state.stack_id)
+    storage = Storage.for_shape(state.shape_handle, stack_storage)
+
+    # A throwaway accumulator shaped like the materializer state so we can reuse
+    # `apply_changes/2` and `cast!/2` without touching the live value_counts.
+    replay0 = %{
+      state
+      | index: %{},
+        tag_indices: %{},
+        value_counts: %{},
+        offset: LogOffset.before_all(),
+        subscribed_offset: state.applied_offset,
+        # replay bookkeeping (consumed only by the replay apply fun below)
+        pending_events: %{from_lsn: from_lsn, pid: pid, seed: nil}
+    }
+
+    replay =
+      read_history_up_to_subscribed(
+        replay0,
+        storage,
+        fn stream, acc -> apply_replay_stream(stream, acc) end,
+        true
+      )
+
+    case replay.pending_events.seed do
+      nil -> link_values_from_counts(replay.value_counts)
+      seed -> seed
+    end
+  end
+
+  # Apply function used during move replay. Decodes the stream with per-item
+  # source offsets, groups items into per-transaction batches, and for each
+  # batch:
+  #   * captures the seed view (value counts as of `from_lsn`) on the first
+  #     batch strictly after `from_lsn`;
+  #   * applies the batch to the throwaway value_counts;
+  #   * emits the resulting moves to `pid`, tagged with the batch's source LSN,
+  #     for batches strictly after `from_lsn` (batches at or before `from_lsn`
+  #     only build state and are not emitted).
+  defp apply_replay_stream(stream, acc) do
+    %{from_lsn: from_lsn, pid: pid} = acc.pending_events
+    handle = acc.shape_handle
+
+    stream
+    |> decode_json_stream_with_offsets()
+    |> chunk_by_offset()
+    |> Enum.reduce(acc, fn {txn_offset, changes, txids}, acc ->
+      emit? = is_log_offset_lt(from_lsn, txn_offset)
+
+      acc =
+        if emit? and is_nil(acc.pending_events.seed) do
+          seed = link_values_from_counts(acc.value_counts)
+          put_in(acc.pending_events.seed, seed)
+        else
+          acc
+        end
+
+      {acc, events} = apply_changes(changes, acc)
+
+      if emit? do
+        events =
+          case events do
+            empty when empty == %{} -> %{}
+            events -> Map.put(events, :txids, MapSet.new(txids))
           end
+          |> cancel_matching_move_events()
+
+        if events != %{} do
+          payload =
+            events
+            |> finalize_txids()
+            |> Map.put(:lsn, txn_offset)
+
+          send(pid, {:materializer_changes, handle, payload})
         end
+      end
+
+      acc
+    end)
+  end
+
+  # Decode a `{log_offset, json}` log stream (from `get_log_stream_with_offsets/3`)
+  # into `{log_offset, txids, change}` tuples so replay can delimit transactions
+  # and tag emitted moves. The storage offset is used rather than the JSON
+  # headers because control messages carry no `lsn`/`op_position` there.
+  defp decode_json_stream_with_offsets(stream) do
+    stream
+    |> Stream.map(fn {offset, json} -> {offset, Jason.decode!(json)} end)
+    |> Stream.filter(fn {_offset, decoded} ->
+      Map.has_key?(decoded, "key") || Map.has_key?(decoded["headers"], "event")
+    end)
+    |> Stream.map(fn {offset, decoded} ->
+      {offset, decode_txids(decoded["headers"]), decode_change(decoded)}
+    end)
+  end
+
+  defp decode_txids(%{"txids" => txids}) when is_list(txids), do: txids
+  defp decode_txids(_headers), do: []
+
+  defp decode_change(%{
+         "key" => key,
+         "value" => value,
+         "headers" => %{"operation" => operation} = headers
+       }) do
+    case operation do
+      "insert" ->
+        %Changes.NewRecord{
+          key: key,
+          record: value,
+          move_tags: Map.get(headers, "tags", []),
+          active_conditions: Map.get(headers, "active_conditions", [])
+        }
+
+      "update" ->
+        %Changes.UpdatedRecord{
+          key: key,
+          record: value,
+          move_tags: Map.get(headers, "tags", []),
+          removed_move_tags: Map.get(headers, "removed_tags", []),
+          active_conditions: Map.get(headers, "active_conditions", [])
+        }
+
+      "delete" ->
+        %Changes.DeletedRecord{
+          key: key,
+          old_record: value,
+          move_tags: Map.get(headers, "tags", []),
+          active_conditions: Map.get(headers, "active_conditions", [])
+        }
     end
   end
 
+  defp decode_change(%{"headers" => %{"event" => event, "patterns" => patterns} = headers})
+       when event in ["move-out", "move-in"] do
+    patterns =
+      Enum.map(patterns, fn %{"pos" => pos, "value" => value} ->
+        %{pos: pos, value: value}
+      end)
+
+    %{headers: %{event: event, patterns: patterns, txids: Map.get(headers, "txids", [])}}
+  end
+
+  # Group consecutive decoded items into replay batches keyed by their full
+  # `(tx_offset, op_offset)` offset — the same space `from_lsn`/`move_positions`
+  # live in, so the seed cut and emit boundary can fall between any two items.
+  # This matters because dependency-driven moves (subquery move-in/out) have no
+  # WAL LSN of their own and are appended as successive `op_offset`s within one
+  # `tx_offset`, so grouping by `tx_offset` alone cannot separate them. Each
+  # batch is tagged with its last (largest) offset as its source LSN. Snapshot
+  # lines all share `before_all/0`, so they group into a single pre-real-offsets
+  # batch.
+  defp chunk_by_offset(items) do
+    items
+    |> Enum.chunk_by(fn {offset, _txids, _change} -> {offset.tx_offset, offset.op_offset} end)
+    |> Enum.map(fn batch ->
+      {offset, _, _} = List.last(batch)
+      changes = Enum.map(batch, fn {_o, _t, change} -> change end)
+      txids = batch |> Enum.flat_map(fn {_o, t, _c} -> t end) |> Enum.uniq()
+      {offset, changes, txids}
+    end)
+  end
+
   def handle_call(:get_link_values, _from, %{value_counts: value_counts} = state) do
     {:reply, link_values_from_counts(value_counts), state}
   end
@@ -270,6 +505,10 @@ defmodule Electric.Shapes.Consumer.Materializer do
     stack_storage = Storage.for_stack(state.stack_id)
     storage = Storage.for_shape(state.shape_handle, stack_storage)
 
+    # Track the source LSN of this batch so emitted moves can be tagged with it
+    # (used by outer consumers to dedup/replay moves across a restart).
+    state = %{state | applied_offset: range_end}
+
     state =
       Storage.get_log_stream(range_start, range_end, storage)
       |> decode_json_stream()
@@ -288,10 +527,18 @@ defmodule Electric.Shapes.Consumer.Materializer do
     {:reply, :ok, state}
   end
 
-  def handle_call(:subscribe, {pid, _ref} = _from, state) do
+  def handle_call({:subscribe, from_lsn}, {pid, _ref} = _from, state) do
     Process.monitor(pid)
 
-    {:reply, :ok, %{state | subscribers: MapSet.put(state.subscribers, pid)}}
+    # Register the subscriber *before* replaying so that any live move flushed
+    # after this call is delivered to `pid` and interleaves correctly after the
+    # replayed tail (replay covers up to `applied_offset`; live moves are
+    # strictly beyond it).
+    state = %{state | subscribers: MapSet.put(state.subscribers, pid)}
+
+    seed_link_values = maybe_replay_moves(state, pid, from_lsn)
+
+    {:reply, {:ok, seed_link_values, state.applied_offset}, state}
   end
 
   # if the supervisor is going down then this process will also be taken down
@@ -471,7 +718,10 @@ defmodule Electric.Shapes.Consumer.Materializer do
       cancel_matching_move_events(state.pending_events)
 
     if events != %{} do
-      events = finalize_txids(events)
+      events =
+        events
+        |> finalize_txids()
+        |> Map.put(:lsn, state.applied_offset)
 
       for pid <- state.subscribers do
         send(pid, {:materializer_changes, state.shape_handle, events})
diff --git a/packages/sync-service/lib/electric/shapes/consumer/state.ex b/packages/sync-service/lib/electric/shapes/consumer/state.ex
index e23edee4dc..26429aa8d2 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/state.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/state.ex
@@ -26,6 +26,27 @@ defmodule Electric.Shapes.Consumer.State do
     buffer: [],
     txn_offset_mapping: [],
     materializer_subscribed?: false,
+    # Per-dependency "moves-applied-up-to" source LSNs (`%{dep_handle => LogOffset}`),
+    # persisted so that after a restart the outer subquery consumer can ask each
+    # dependency materializer to replay the moves it missed and dedup by position.
+    move_positions: %{},
+    # Per-dependency seed views (`%{dep_handle => MapSet}`) captured from the
+    # materializer at subscribe time (as-of `move_positions`), used to seed the
+    # event handler's dependency views so replayed moves are not
+    # redundancy-eliminated.
+    dep_seed_views: %{},
+    # Per-dependency source LSN of the most recently received (not yet applied)
+    # materializer move (`%{dep_handle => LogOffset}`); moved into
+    # `staged_move_positions` once the move pipeline is fully drained (applied to
+    # the writer buffer).
+    pending_move_lsns: %{},
+    # Per-dependency source LSNs that have been applied to the writer buffer but
+    # are not yet known to be durably flushed
+    # (`%{dep_handle => [{flush_threshold_offset, source_lsn}]}`, ascending).
+    # An entry is committed to `move_positions` (advanced + persisted) only once
+    # the writer confirms a flush at/after its threshold, so the persisted
+    # position never runs ahead of durable storage.
+    staged_move_positions: %{},
     terminating?: false,
     buffering?: false,
     # Based on the write unit value, consumer will either buffer txn fragments in memory until
@@ -146,6 +167,7 @@ defmodule Electric.Shapes.Consumer.State do
 
     {:ok, latest_offset} = Storage.fetch_latest_offset(storage)
     {:ok, pg_snapshot} = Storage.fetch_pg_snapshot(storage)
+    {:ok, move_positions} = Storage.fetch_move_positions(storage)
 
     initial_snapshot_state = InitialSnapshot.new(pg_snapshot)
 
@@ -154,6 +176,7 @@ defmodule Electric.Shapes.Consumer.State do
       | latest_offset: latest_offset,
         storage: storage,
         writer: writer,
+        move_positions: move_positions,
         initial_snapshot_state: initial_snapshot_state,
         buffering?: InitialSnapshot.needs_buffering?(initial_snapshot_state)
     }
diff --git a/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs b/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs
index d5cbeb674e..a65ff3b32b 100644
--- a/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs
+++ b/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs
@@ -124,6 +124,33 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
       end
     end
 
+    describe "#{module_name}.fetch_move_positions/1" do
+      setup :start_storage
+
+      test "returns an empty map on startup", %{storage: opts} do
+        assert Storage.fetch_move_positions(opts) == {:ok, %{}}
+      end
+
+      test "round-trips a per-dependency positions map", %{storage: opts} do
+        positions = %{
+          "dep-a" => LogOffset.new(10, 2),
+          "dep-b" => LogOffset.new(42, 0)
+        }
+
+        assert :ok = Storage.set_move_positions!(positions, opts)
+        assert Storage.fetch_move_positions(opts) == {:ok, positions}
+      end
+
+      test "overwrites previously persisted positions", %{storage: opts} do
+        assert :ok = Storage.set_move_positions!(%{"dep-a" => LogOffset.new(1, 0)}, opts)
+
+        updated = %{"dep-a" => LogOffset.new(5, 0), "dep-b" => LogOffset.new(7, 1)}
+        assert :ok = Storage.set_move_positions!(updated, opts)
+
+        assert Storage.fetch_move_positions(opts) == {:ok, updated}
+      end
+    end
+
     describe "#{module_name}.append_to_log!/3" do
       setup do
         {:ok, %{module: unquote(module)}}
diff --git a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
index 5cee0c2428..59b98e81a2 100644
--- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
@@ -1308,6 +1308,66 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     |> Enum.map(fn {_offset, item} -> Jason.encode!(item) end)
   end
 
+  # Build a single main-log insert log item at `offset` introducing `value`,
+  # encoded the same way the source consumer would write it.
+  defp main_log_insert(offset, id, value) do
+    change =
+      %Changes.NewRecord{
+        relation: {"public", "test_table"},
+        key: ~s|"public"."test_table"/"#{id}"|,
+        record: %{"id" => id, "value" => value},
+        log_offset: offset,
+        move_tags: []
+      }
+      |> Changes.fill_key(["id"])
+
+    change
+    |> then(&LogItems.from_change(&1, 1, ["id"], :default))
+    |> Enum.map(fn {item_offset, item} ->
+      {item_offset, change.key, :insert, Jason.encode!(item)}
+    end)
+  end
+
+  # Build a single main-log delete log item at `offset` for row `id` (value only
+  # needs to match the corresponding insert so the materializer decrements it).
+  defp main_log_delete(offset, id, value) do
+    change =
+      %Changes.DeletedRecord{
+        relation: {"public", "test_table"},
+        key: ~s|"public"."test_table"/"#{id}"|,
+        old_record: %{"id" => id, "value" => value},
+        log_offset: offset,
+        move_tags: []
+      }
+      |> Changes.fill_key(["id"])
+
+    change
+    |> then(&LogItems.from_change(&1, 1, ["id"], :default))
+    |> Enum.map(fn {item_offset, item} ->
+      {item_offset, change.key, :delete, Jason.encode!(item)}
+    end)
+  end
+
+  # Like `main_log_insert/3` but attaches move tags so the row is indexed under
+  # those tags (and can later be moved out by a control message targeting them).
+  defp tagged_main_log_insert(offset, id, value, tags) do
+    change =
+      %Changes.NewRecord{
+        relation: {"public", "test_table"},
+        key: ~s|"public"."test_table"/"#{id}"|,
+        record: %{"id" => id, "value" => value},
+        log_offset: offset,
+        move_tags: tags
+      }
+      |> Changes.fill_key(["id"])
+
+    change
+    |> then(&LogItems.from_change(&1, 1, ["id"], :default))
+    |> Enum.map(fn {item_offset, item} ->
+      {item_offset, change.key, :insert, Jason.encode!(item)}
+    end)
+  end
+
   defp prep_changes(changes, opts \\ []) do
     pk_cols = Keyword.get(opts, :pk_cols, ["id"])
     relation = Keyword.get(opts, :relation, {"public", "test_table"})
@@ -1526,6 +1586,264 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     end
   end
 
+  describe "move replay on subscribe" do
+    # A subscriber that is behind (`from_lsn` < the materializer's applied
+    # position) is caught up by replaying only the moves it missed, each tagged
+    # with its source LSN, and is handed the link values as of `from_lsn` to seed
+    # its dependency view.
+    setup ctx do
+      shape_handle = "replay-test-#{System.unique_integer([:positive])}"
+
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+
+      # Snapshot establishes value 10.
+      Storage.make_new_snapshot!(
+        make_snapshot_data([%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}}]),
+        storage
+      )
+
+      # Two main-log inserts at distinct offsets, each introducing a new value
+      # (a move-in): value 20 at (100,0), value 30 at (200,0).
+      writer =
+        Storage.append_to_log!(
+          main_log_insert(LogOffset.new(100, 0), "2", "20"),
+          writer
+        )
+
+      writer =
+        Storage.append_to_log!(
+          main_log_insert(LogOffset.new(200, 0), "3", "30"),
+          writer
+        )
+
+      Storage.hibernate(writer)
+
+      ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id)
+
+      {:ok, _pid} =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      respond_to_call(:await_snapshot_start, :started)
+      # Subscribed offset past both main-log entries so the materializer applies
+      # the full history at startup.
+      respond_to_call(:subscribe_materializer, {:ok, LogOffset.new(200, 0)})
+
+      mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle}
+      assert Materializer.wait_until_ready(mat_ctx) == :ok
+      assert Materializer.get_link_values(mat_ctx) == MapSet.new([10, 20, 30])
+
+      Map.put(ctx, :mat_ctx, mat_ctx)
+    end
+
+    test "replays only moves after from_lsn, tagged with per-range source LSNs",
+         %{mat_ctx: mat_ctx} do
+      # Behind at (100,0): the move-in for value 20 (at (100,0)) is already
+      # applied, only value 30 (at (200,0)) must be replayed.
+      assert {:ok, seed, applied_offset} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(100, 0))
+
+      # Seed view is the link values as of (100,0): value 30 not yet included.
+      assert seed == MapSet.new([10, 20])
+      assert applied_offset == LogOffset.new(200, 0)
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_in: [{30, "30"}], lsn: %LogOffset{tx_offset: 200, op_offset: 0}}}
+
+      # The already-applied move-in for value 20 is NOT replayed.
+      refute_received {:materializer_changes, _handle, %{move_in: [{20, "20"}]}}
+    end
+
+    test "replays every move when from_lsn is before all main-log moves",
+         %{mat_ctx: mat_ctx} do
+      assert {:ok, _seed, _applied} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(50, 0))
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_in: [{20, "20"}], lsn: %LogOffset{tx_offset: 100, op_offset: 0}}}
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_in: [{30, "30"}], lsn: %LogOffset{tx_offset: 200, op_offset: 0}}}
+    end
+
+    test "does not replay when from_lsn is at or past the applied position",
+         %{mat_ctx: mat_ctx} do
+      assert {:ok, seed, _applied} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(200, 0))
+
+      # Caught up: seed is the current link values and nothing is replayed.
+      assert seed == MapSet.new([10, 20, 30])
+      refute_received {:materializer_changes, _handle, _payload}
+    end
+  end
+
+  describe "move replay on subscribe with control messages" do
+    # A dependency shape that is itself an optimized subquery has move-in/move-out
+    # *control messages* in its own log (rows moving in/out as its nested
+    # dependency toggles). Control messages carry no `lsn`/`op_position` in their
+    # headers, so their position comes from the storage offset rather than the
+    # JSON. On restart the materializer replays that log to catch up a behind
+    # subscriber, and a control-message move after `from_lsn` must be re-emitted
+    # just like a data-change move.
+    setup ctx do
+      shape_handle = "control-replay-#{System.unique_integer([:positive])}"
+
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+      Storage.make_new_snapshot!([], storage)
+
+      # tx 100: value 10 enters the shape, tagged "tag1".
+      writer =
+        Storage.append_to_log!(
+          tagged_main_log_insert(LogOffset.new(100, 0), "1", "10", ["tag1"]),
+          writer
+        )
+
+      # tx 200: a second "tag1" row in a *later* transaction. Its only role is to
+      # open transaction 200 (so the control message below joins that transaction
+      # rather than tx 100); it shares value 10, so it emits no move of its own.
+      writer =
+        Storage.append_to_log!(
+          tagged_main_log_insert(LogOffset.new(200, 0), "2", "10", ["tag1"]),
+          writer
+        )
+
+      # (200,1): a move-out control message removing "tag1" — value 10 leaves the
+      # shape. Its source offset is only recoverable from storage, not the JSON.
+      {_range, writer} =
+        Storage.append_control_message!(
+          Jason.encode!(%{
+            headers: %{event: "move-out", patterns: [%{pos: 0, value: "tag1"}]}
+          }),
+          writer
+        )
+
+      Storage.hibernate(writer)
+
+      ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id)
+
+      {:ok, _pid} =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      respond_to_call(:await_snapshot_start, :started)
+      respond_to_call(:subscribe_materializer, {:ok, LogOffset.new(200, 1)})
+
+      mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle}
+      assert Materializer.wait_until_ready(mat_ctx) == :ok
+      # Startup applied the whole history: value 10 entered, then left via the
+      # move-out control message, so nothing remains materialized.
+      assert Materializer.get_link_values(mat_ctx) == MapSet.new()
+
+      Map.put(ctx, :mat_ctx, mat_ctx)
+    end
+
+    test "replays a move-out carried by a control message after from_lsn",
+         %{mat_ctx: mat_ctx} do
+      # Behind at (150,0): value 10 was still present then, and the move-out at
+      # (200,1) is after from_lsn, so replay must re-emit it.
+      assert {:ok, seed, applied_offset} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(150, 0))
+
+      # Seed view is the link values as of (150,0): value 10 still present.
+      assert seed == MapSet.new([10])
+      assert applied_offset == LogOffset.new(200, 1)
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_out: [{10, "10"}], lsn: %LogOffset{tx_offset: 200}}}
+    end
+  end
+
+  describe "move replay of a value toggled multiple times within one transaction" do
+    # A value can cross the 0↔1 boundary several times inside a single source
+    # transaction (op_offsets sharing one tx_offset). Replay batches each op
+    # separately, so these are re-emitted as separate sequential payloads in
+    # offset order — never a single payload carrying both a move_in and a
+    # move_out for the same value (the case `cancel_matching_move_events/1`
+    # guards against, where the consumer's move-in query would race its own
+    # moved-out tag). Sequential single-value payloads are just the normal
+    # per-transaction flow applied in order.
+    setup ctx do
+      shape_handle = "toggle-replay-#{System.unique_integer([:positive])}"
+
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+      Storage.make_new_snapshot!([], storage)
+
+      # One transaction (tx 100) toggles value 10: in (row "a"), out (row "a"),
+      # in (row "b"). It ends present on "b", so startup materializes {10}.
+      writer = Storage.append_to_log!(main_log_insert(LogOffset.new(100, 0), "a", "10"), writer)
+      writer = Storage.append_to_log!(main_log_delete(LogOffset.new(100, 1), "a", "10"), writer)
+      writer = Storage.append_to_log!(main_log_insert(LogOffset.new(100, 2), "b", "10"), writer)
+      Storage.hibernate(writer)
+
+      ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id)
+
+      {:ok, _pid} =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      respond_to_call(:await_snapshot_start, :started)
+      respond_to_call(:subscribe_materializer, {:ok, LogOffset.new(100, 2)})
+
+      mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle}
+      assert Materializer.wait_until_ready(mat_ctx) == :ok
+      assert Materializer.get_link_values(mat_ctx) == MapSet.new([10])
+
+      Map.put(ctx, :mat_ctx, mat_ctx)
+    end
+
+    test "re-emits each toggle as its own payload, in offset order", %{mat_ctx: mat_ctx} do
+      # Behind before the transaction: the whole toggle sequence is replayed.
+      assert {:ok, seed, applied_offset} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(50, 0))
+
+      # As of (50,0) value 10 is not yet present.
+      assert seed == MapSet.new([])
+      assert applied_offset == LogOffset.new(100, 2)
+
+      # Collect the payloads in the order they arrive (replay sends are
+      # sequential from the subscribe call, so mailbox order is emit order) and
+      # assert the full sequence: three distinct payloads, one per op, in offset
+      # order — not a single netted payload.
+      payloads =
+        for _ <- 1..3 do
+          assert_receive {:materializer_changes, _h, payload}
+          payload
+        end
+
+      assert [
+               %{move_in: [{10, "10"}], lsn: %LogOffset{tx_offset: 100, op_offset: 0}},
+               %{move_out: [{10, "10"}], lsn: %LogOffset{tx_offset: 100, op_offset: 1}},
+               %{move_in: [{10, "10"}], lsn: %LogOffset{tx_offset: 100, op_offset: 2}}
+             ] = payloads
+
+      refute_received {:materializer_changes, _h, _payload}
+    end
+  end
+
   describe "startup race condition handling" do
     # Tests for the race condition where Consumer dies between await_snapshot_start
     # and subscribe_materializer. See concurrency_analysis/MATERIALIZER_RACE_ANALYSIS.md
diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs
index 2e180f657f..62a38df45b 100644
--- a/packages/sync-service/test/electric/shapes/consumer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer_test.exs
@@ -2619,6 +2619,88 @@ defmodule Electric.Shapes.ConsumerTest do
              ] = get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage)
     end
 
+    test "consumer advances and persists the per-dependency moves-position on move application",
+         ctx do
+      parent = self()
+
+      Repatch.patch(
+        Electric.Shapes.Consumer.Effects,
+        :query_move_in_async,
+        [mode: :shared],
+        fn _task_sup, _consumer_state, _buffering_state, consumer_pid ->
+          send(parent, {:query_requested, consumer_pid})
+          :ok
+        end
+      )
+
+      Support.TestUtils.activate_mocks_for_descendant_procs(Consumer)
+
+      {shape_handle, _} =
+        ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id)
+
+      :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id)
+
+      {:ok, shape} = Electric.Shapes.fetch_shape_by_handle(ctx.stack_id, shape_handle)
+      [dep_handle] = shape.shape_dependencies_handles
+
+      consumer_pid = Consumer.whereis(ctx.stack_id, shape_handle)
+      ref = Shapes.Consumer.register_for_changes(ctx.stack_id, shape_handle)
+      shape_storage = Storage.for_shape(shape_handle, ctx.storage)
+
+      move_lsn = LogOffset.new(777, 0)
+
+      assert :ok = LsnTracker.broadcast_last_seen_lsn(ctx.stack_id, 100)
+
+      send(
+        consumer_pid,
+        {:materializer_changes, dep_handle, %{move_in: [{1, "1"}], move_out: [], lsn: move_lsn}}
+      )
+
+      assert_receive {:query_requested, ^consumer_pid}
+
+      # While the move-in is still buffering the position has NOT advanced to the
+      # move's LSN — it must only advance once the move is applied.
+      {:ok, buffering_positions} = Storage.fetch_move_positions(shape_storage)
+      refute Map.get(buffering_positions, dep_handle) == move_lsn
+
+      send(consumer_pid, {:pg_snapshot_known, {100, 300, []}})
+
+      send_stored_move_in_complete(
+        consumer_pid,
+        shape_storage,
+        [
+          [
+            ~s'"public"."test_table"/"1"',
+            [],
+            Jason.encode!(%{
+              "key" => ~s'"public"."test_table"/"1"',
+              "value" => %{"id" => "1", "value" => "val"},
+              "headers" => %{"operation" => "insert", "relation" => ["public", "test_table"]}
+            })
+          ]
+        ],
+        Lsn.from_integer(100)
+      )
+
+      assert_receive {^ref, :new_changes, _offset}, @receive_timeout
+
+      # The splice has been applied to the writer buffer, but the moves-position
+      # is only *staged* — it must not be persisted ahead of a durable flush, or
+      # a restart could leave it pointing past storage. So until the writer
+      # confirms the flush, the persisted position has not advanced.
+      {:ok, staged_positions} = Storage.fetch_move_positions(shape_storage)
+      refute Map.get(staged_positions, dep_handle) == move_lsn
+
+      # Once the writer confirms a flush at/after the move's splice, the
+      # per-dependency moves-position is advanced to the move's source LSN and
+      # persisted to storage.
+      send(consumer_pid, {Storage, :flushed, LogOffset.new(1_000_000_000, 0)})
+      :sys.get_state(consumer_pid)
+
+      {:ok, applied_positions} = Storage.fetch_move_positions(shape_storage)
+      assert Map.get(applied_positions, dep_handle) == move_lsn
+    end
+
     test "consumer startup seeds the stack-scoped subquery index", ctx do
       alias Electric.Shapes.Filter.Indexes.SubqueryIndex
 
diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs
index 6de429df03..8736e7dd22 100644
--- a/packages/sync-service/test/integration/oracle_restore_test.exs
+++ b/packages/sync-service/test/integration/oracle_restore_test.exs
@@ -1,8 +1,8 @@
 defmodule Electric.Integration.OracleRestoreTest do
   @moduledoc """
-  Targeted regression tests for restore-from-file. Each test exercises a
-  scenario from `bugs.md` with a deterministic, minimal mutation sequence,
-  reusing `Support.OracleHarness.test_against_oracle/4`.
+  Targeted regression tests for subquery-shape restore across a server restart,
+  each with a deterministic, minimal mutation sequence, reusing
+  `Support.OracleHarness.test_against_oracle/4`.
 
   These tests use a small, readable "issue tracker" domain schema rather than
   the abstract `level_N` hierarchy from `Support.OracleHarness.StandardSchema`:
@@ -12,9 +12,6 @@ defmodule Electric.Integration.OracleRestoreTest do
 
   The shape under test is "issues belonging to an active project", expressed
   as a subquery over `projects.active`.
-
-  These tests are expected to fail until the underlying Electric bugs are
-  fixed.
   """
 
   use ExUnit.Case, async: false
@@ -35,7 +32,11 @@ defmodule Electric.Integration.OracleRestoreTest do
   setup ctx do
     ctx =
       with_electric_client(ctx,
-        router_opts: [long_poll_timeout: 100],
+        # A realistic long-poll timeout. A very short one (e.g. 100ms) can time
+        # out before replication has caught up after a restart, yielding a
+        # spurious 409 that is independent of the subquery-restore behaviour
+        # under test here.
+        router_opts: [long_poll_timeout: 5000],
         num_clients: 1
       )
 
@@ -96,12 +97,11 @@ defmodule Electric.Integration.OracleRestoreTest do
   end
 
   @tag :oracle_restore_bug_1
-  test "bug 1: subquery shape diverges from oracle after server restart", ctx do
-    # Shape on `issues` with a subquery predicate over `projects.active`.
-    # After the server is restarted, the subquery materializer state is not
-    # restored from disk, so toggling `projects.active` on either side of the
-    # restart produces a divergence between the client view and the oracle
-    # (or a 409 must-refetch on this `optimized: true` shape).
+  test "subquery shape stays consistent with oracle across a server restart", ctx do
+    # A single `optimized: true` shape on `issues` with a subquery predicate over
+    # `projects.active`. Toggling `projects.active` on either side of the restart
+    # moves issue rows in and out of the shape via the subquery materializer; the
+    # client view must stay consistent with the oracle across the restart.
     shapes = [
       %{
         name: "issues_of_active_projects",
@@ -129,4 +129,159 @@ defmodule Electric.Integration.OracleRestoreTest do
 
     OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
   end
+
+  @tag :oracle_restore_optimized_refetch
+  # Build a large persisted backlog on the `projects` source shape: 200 toggles
+  # under a small `chunk_bytes_threshold` so its log spans many chunks. After the
+  # restart the persistent replication slot has to replay that backlog.
+  @tag chunk_bytes_threshold: 200
+  test "optimized subquery shapes stay consistent when the slot replays a backlog after restart",
+       ctx do
+    # Two `optimized: true` subquery shapes over the same `projects` source. After
+    # the restart the persistent slot replays batch_1's already-applied
+    # transactions; the source consumer skips those (at/below its restored
+    # `latest_offset`) before re-notifying the subquery materializer, so the shapes
+    # stay consistent and the polling client is not sent a 409 must-refetch.
+    shapes = [
+      %{
+        name: "issues_of_active_projects",
+        table: "issues",
+        where: "project_id IN (SELECT id FROM projects WHERE active = true)",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      },
+      %{
+        name: "issues_of_inactive_projects",
+        table: "issues",
+        where: "project_id IN (SELECT id FROM projects WHERE active = false)",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      }
+    ]
+
+    # batch_1: 200 toggles of p5's `active` flag — the backlog. Under the small
+    # `chunk_bytes_threshold` above this makes the `projects` source log span many
+    # chunks. p5 ends active, so pre-restart both shapes match the oracle; the
+    # restart then replays this backlog from the slot.
+    toggles =
+      Enum.flat_map(1..100, fn _ ->
+        [
+          [%{name: "deactivate_p5", sql: "UPDATE projects SET active = false WHERE id = 'p5'"}],
+          [%{name: "reactivate_p5", sql: "UPDATE projects SET active = true WHERE id = 'p5'"}]
+        ]
+      end)
+
+    # batch_2: a single dependency move applied after the restart. In practice the
+    # test fails during the batch_1 replay before this is reached; it's kept so the
+    # harness runs a post-restart batch/check.
+    batch_2 = [
+      [%{name: "deactivate_p3", sql: "UPDATE projects SET active = false WHERE id = 'p3'"}]
+    ]
+
+    batches = [toggles, batch_2]
+
+    OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
+  end
+
+  # A three-level "issue tracker": regions own projects own issues.
+  #
+  #     regions (id, active)
+  #         └── projects (id, region_id)
+  #                 └── issues (id, project_id, title)
+  #
+  # r1, r3 start active; r2 starts inactive. Two projects per region, two issues
+  # per project.
+  defp setup_nested_schema(ctx) do
+    OracleHarness.apply_sql(ctx, [
+      "DROP TABLE IF EXISTS issues CASCADE",
+      "DROP TABLE IF EXISTS projects CASCADE",
+      "DROP TABLE IF EXISTS regions CASCADE",
+      """
+      CREATE TABLE regions (
+        id TEXT PRIMARY KEY,
+        active BOOLEAN NOT NULL DEFAULT true
+      )
+      """,
+      """
+      CREATE TABLE projects (
+        id TEXT PRIMARY KEY,
+        region_id TEXT NOT NULL REFERENCES regions(id) ON DELETE CASCADE
+      )
+      """,
+      """
+      CREATE TABLE issues (
+        id TEXT PRIMARY KEY,
+        project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+        title TEXT NOT NULL
+      )
+      """
+    ])
+
+    OracleHarness.apply_sql(ctx, [
+      "INSERT INTO regions (id, active) VALUES ('r1', true), ('r2', false), ('r3', true)",
+      """
+      INSERT INTO projects (id, region_id) VALUES
+        ('p1', 'r1'), ('p2', 'r1'),
+        ('p3', 'r2'), ('p4', 'r2'),
+        ('p5', 'r3'), ('p6', 'r3')
+      """,
+      "INSERT INTO issues (id, project_id, title) VALUES " <>
+        (for(n <- 1..12, do: "('i#{n}', 'p#{div(n - 1, 2) + 1}', 'Issue #{n}')")
+         |> Enum.join(", "))
+    ])
+
+    :ok
+  end
+
+  @tag :oracle_restore_nested_subquery
+  # As with the single-level case, force the source shape's log to span multiple
+  # chunks so the post-restart replay path is exercised.
+  @tag chunk_bytes_threshold: 200
+  test "nested optimized subquery shape stays consistent with oracle across a restart", ctx do
+    setup_nested_schema(ctx)
+
+    # The shape under test is a *nested* subquery: issues whose project belongs to
+    # an active region. Its dependency shape — `projects WHERE region_id IN (SELECT
+    # id FROM regions WHERE active)` — is itself an optimized subquery shape, so
+    # its log contains move-in/move-out *control messages* (projects moving in and
+    # out as regions toggle). On restart the dependency materializer replays that
+    # log to catch the outer consumer up; the control-message moves must be
+    # re-emitted so the outer shape isn't left missing the issues of projects that
+    # moved via a control message.
+    shapes = [
+      %{
+        name: "issues_of_active_regions",
+        table: "issues",
+        where:
+          "project_id IN (SELECT id FROM projects WHERE region_id IN " <>
+            "(SELECT id FROM regions WHERE active = true))",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      }
+    ]
+
+    # batch_1: 200 toggles of r3's `active` flag — each toggle moves p5/p6 in and
+    # out of the inner subquery shape, writing control messages to its log. r3 ends
+    # active, so pre-restart the shape matches the oracle.
+    toggles =
+      Enum.flat_map(1..100, fn _ ->
+        [
+          [%{name: "deactivate_r3", sql: "UPDATE regions SET active = false WHERE id = 'r3'"}],
+          [%{name: "reactivate_r3", sql: "UPDATE regions SET active = true WHERE id = 'r3'"}]
+        ]
+      end)
+
+    # batch_2 (after the restart): a single region deactivate that moves a project
+    # out of the inner subquery via a control message.
+    batch_2 = [
+      [%{name: "deactivate_r1", sql: "UPDATE regions SET active = false WHERE id = 'r1'"}]
+    ]
+
+    batches = [toggles, batch_2]
+
+    OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
+  end
 end
diff --git a/packages/sync-service/test/support/test_storage.ex b/packages/sync-service/test/support/test_storage.ex
index 32283ef824..358cf13641 100644
--- a/packages/sync-service/test/support/test_storage.ex
+++ b/packages/sync-service/test/support/test_storage.ex
@@ -101,6 +101,18 @@ defmodule Support.TestStorage do
     Storage.set_pg_snapshot(pg_snapshot, storage)
   end
 
+  @impl Electric.ShapeCache.Storage
+  def set_move_positions!(move_positions, {parent, shape_handle, _, storage}) do
+    send(parent, {__MODULE__, :set_move_positions!, shape_handle, move_positions})
+    Storage.set_move_positions!(move_positions, storage)
+  end
+
+  @impl Electric.ShapeCache.Storage
+  def fetch_move_positions({parent, shape_handle, _, storage}) do
+    send(parent, {__MODULE__, :fetch_move_positions, shape_handle})
+    Storage.fetch_move_positions(storage)
+  end
+
   @impl Electric.ShapeCache.Storage
   def snapshot_started?({parent, shape_handle, _, storage}) do
     send(parent, {__MODULE__, :snapshot_started?, shape_handle})
@@ -113,6 +125,12 @@ defmodule Support.TestStorage do
     Storage.get_log_stream(offset, max_offset, storage)
   end
 
+  @impl Electric.ShapeCache.Storage
+  def get_log_stream_with_offsets(offset, max_offset, {parent, shape_handle, _, storage}) do
+    send(parent, {__MODULE__, :get_log_stream_with_offsets, shape_handle, offset, max_offset})
+    Storage.get_log_stream_with_offsets(offset, max_offset, storage)
+  end
+
   @impl Electric.ShapeCache.Storage
   def get_chunk_end_log_offset(offset, {parent, shape_handle, _, storage}) do
     send(parent, {__MODULE__, :get_chunk_end_log_offset, shape_handle, offset})