diff --git a/.changeset/restore-shutdown-shape-removal.md b/.changeset/restore-shutdown-shape-removal.md
new file mode 100644
index 0000000000..1b2021a2f2
--- /dev/null
+++ b/.changeset/restore-shutdown-shape-removal.md
@@ -0,0 +1,9 @@
+---
+'@core/sync-service': patch
+---
+
+Stop subquery shapes from being spuriously removed during a server restart. When
+a dependency consumer's inline call to its materializer raced the materializer's
+shutdown, the resulting `:noproc` exit crashed the consumer and removed the shape
+from disk, causing a `409 must-refetch` after the restart. The consumer now
+absorbs that exit and lets the monitored `:DOWN` drive a clean stop.
diff --git a/.changeset/subquery-move-replay-on-restart.md b/.changeset/subquery-move-replay-on-restart.md
new file mode 100644
index 0000000000..2c4156c0a9
--- /dev/null
+++ b/.changeset/subquery-move-replay-on-restart.md
@@ -0,0 +1,11 @@
+---
+'@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. Replay uses the authoritative persisted shape-log
+offset for control messages and spliced move-in rows, including nested
+subqueries whose generated JSON records do not contain source LSNs.
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..dbfc023839 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()}
 
@@ -155,7 +170,7 @@ defmodule Electric.ShapeCache.InMemoryStorage do
   defp snapshot_end(),
     do: snapshot_chunk_end(storage_offset(LogOffset.last_before_real_offsets()))
 
-  defp get_offset_indexed_stream(offset, max_offset, offset_indexed_table) do
+  defp get_offset_indexed_stream(offset, max_offset, offset_indexed_table, project_item) do
     offset = storage_offset(offset)
     max_offset = storage_offset(max_offset)
 
@@ -168,11 +183,15 @@ defmodule Electric.ShapeCache.InMemoryStorage do
           nil
 
         {{:offset, position}, [{_, item}]} ->
-          {item, position}
+          {project_item.(LogOffset.new(position), item), position}
       end
     end)
   end
 
+  defp get_offset_indexed_stream(offset, max_offset, offset_indexed_table) do
+    get_offset_indexed_stream(offset, max_offset, offset_indexed_table, fn _, item -> item end)
+  end
+
   @snapshot_boundary_offset LogOffset.last_before_real_offsets()
   @impl Electric.ShapeCache.Storage
   def get_log_stream(offset, max_offset, %MS{} = opts)
@@ -188,6 +207,37 @@ 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
+
+  defp get_offset_indexed_stream_with_offsets(offset, max_offset, offset_indexed_table) do
+    get_offset_indexed_stream(offset, max_offset, offset_indexed_table, fn offset, item ->
+      {offset, item}
+    end)
+  end
+
+  defp stream_from_snapshot_with_offsets(offset, max_offset, opts) do
+    offset
+    |> stream_from_snapshot(max_offset, opts)
+    |> Stream.map(&{nil, &1})
+  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..5487202023 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,71 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
+  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_with_offsets(min_offset, max_offset, opts)
+  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
+
+    {stream_kind, stream} =
+      case {last_snapshot_chunk, min_offset} do
+        {_, x} when is_min_offset(x) ->
+          {:snapshot, Snapshot.stream_chunk_lines(opts, 0)}
+
+        {%LogOffset{} = latest, min_offset} when is_log_offset_lt(min_offset, latest) ->
+          {:snapshot, Snapshot.stream_chunk_lines(opts, op_offset + 1)}
+
+        {nil, _offset} ->
+          {:snapshot, wait_for_chunk_file_or_snapshot_end(opts, op_offset + 1)}
+
+        {%LogOffset{}, offset} ->
+          {:main_log, stream_main_log_with_offsets(offset, max_offset, opts)}
+      end
+
+    case stream_kind do
+      :snapshot -> Stream.map(stream, &{nil, &1})
+      :main_log -> stream
+    end
+  end
+
+  defp stream_main_log_with_offsets(min_offset, max_offset, opts) do
+    stream_main_log(
+      min_offset,
+      max_offset,
+      opts,
+      fn offset, item -> {offset, item} end,
+      true
+    )
+  end
+
   defp stream_main_log(min_offset, max_offset, %__MODULE__{} = opts) do
+    stream_main_log(min_offset, max_offset, opts, fn _, item -> item end, false)
+  end
+
+  defp stream_main_log(
+         min_offset,
+         max_offset,
+         %__MODULE__{} = opts,
+         project_item,
+         exact_upper_bound?
+       ) do
     storage_meta(
       ets_table: ets,
       last_persisted_offset: last_persisted,
@@ -1045,43 +1122,66 @@ defmodule Electric.ShapeCache.PureFileStorage do
 
     upper_read_bound = LogOffset.min(max_offset, last_seen)
 
-    # Convert upper_read_bound to tuple for comparison with ETS offsets
-    upper_read_bound_tuple = LogOffset.to_tuple(upper_read_bound)
-
     cond do
       is_log_offset_lte(last_persisted, min_offset) and is_nil(ets) ->
         []
 
       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
-          {_data, last_offset}
-          when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
+        case read_range_from_ets_cache(ets, min_offset, upper_read_bound, project_item) do
+          {_data, :incomplete} ->
             # 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)
-
-          {data, _last_offset} ->
+            stream_from_disk(
+              opts,
+              min_offset,
+              upper_read_bound,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
+
+          {data, :complete} ->
             data
         end
 
       is_log_offset_lte(upper_read_bound, last_persisted) ->
-        stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+        stream_from_disk(
+          opts,
+          min_offset,
+          upper_read_bound,
+          boundary_info,
+          project_item,
+          exact_upper_bound?
+        )
 
       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
-          {_upper_range, last_offset}
-          when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
+        case read_range_from_ets_cache(ets, last_persisted, upper_read_bound, project_item) do
+          {_upper_range, :incomplete} ->
             # 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)
-
-          {upper_range, _last_offset} ->
-            stream_from_disk(opts, min_offset, last_persisted, boundary_info)
+            stream_from_disk(
+              opts,
+              min_offset,
+              upper_read_bound,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
+
+          {upper_range, :complete} ->
+            stream_from_disk(
+              opts,
+              min_offset,
+              last_persisted,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
             |> Stream.concat(upper_range)
         end
     end
@@ -1126,15 +1226,27 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
-  # Returns {data, last_offset_read} where last_offset_read is the offset tuple of the
-  # last entry read, or nil if no entries were read. This allows callers to detect
-  # partial reads due to concurrent ETS clearing.
-  @spec read_range_from_ets_cache(:ets.tid() | nil, LogOffset.t(), LogOffset.t()) ::
-          {list(), LogOffset.t_tuple() | nil}
-  defp read_range_from_ets_cache(nil, _min, _max), do: {[], nil}
-
-  defp read_range_from_ets_cache(ets, %LogOffset{} = min, %LogOffset{} = max) do
-    read_range_from_ets_cache(ets, LogOffset.to_tuple(min), LogOffset.to_tuple(max), [], nil)
+  # Returns whether the requested range was read completely. A range can be
+  # complete even when no item exists at its upper bound because log offsets are
+  # sparse. `:incomplete` means ETS disappeared or ended before the known live
+  # tail, so callers must retry from disk after a concurrent flush.
+  @spec read_range_from_ets_cache(
+          :ets.tid() | nil,
+          LogOffset.t(),
+          LogOffset.t(),
+          (LogOffset.t(), binary() -> term())
+        ) ::
+          {list(), :complete | :incomplete}
+  defp read_range_from_ets_cache(nil, _min, _max, _project_item), do: {[], :incomplete}
+
+  defp read_range_from_ets_cache(ets, %LogOffset{} = min, %LogOffset{} = max, project_item) do
+    read_range_from_ets_cache(
+      ets,
+      LogOffset.to_tuple(min),
+      LogOffset.to_tuple(max),
+      [],
+      project_item
+    )
   end
 
   @spec read_range_from_ets_cache(
@@ -1142,21 +1254,33 @@ defmodule Electric.ShapeCache.PureFileStorage do
           LogOffset.t_tuple(),
           LogOffset.t_tuple(),
           list(),
-          LogOffset.t_tuple() | nil
-        ) :: {list(), LogOffset.t_tuple() | nil}
-  defp read_range_from_ets_cache(ets, min, {max_tx, max_op} = max, acc, last_offset) do
+          (LogOffset.t(), binary() -> term())
+        ) :: {list(), :complete | :incomplete}
+  defp read_range_from_ets_cache(
+         ets,
+         min,
+         {max_tx, max_op} = max,
+         acc,
+         project_item
+       ) do
     case safe_next_lookup(ets, min) do
       :ets_dead ->
-        {Enum.reverse(acc), last_offset}
+        {Enum.reverse(acc), :incomplete}
 
       :"$end_of_table" ->
-        {Enum.reverse(acc), last_offset}
+        {Enum.reverse(acc), :incomplete}
 
       {{min_tx, min_op}, _} when min_tx > max_tx or (min_tx == max_tx and min_op > max_op) ->
-        {Enum.reverse(acc), last_offset}
+        {Enum.reverse(acc), :complete}
 
       {new_min, [{_, item}]} ->
-        read_range_from_ets_cache(ets, new_min, max, [item | acc], new_min)
+        projected_item = project_item.(LogOffset.new(new_min), item)
+
+        if new_min == max do
+          {Enum.reverse([projected_item | acc]), :complete}
+        else
+          read_range_from_ets_cache(ets, new_min, max, [projected_item | acc], project_item)
+        end
     end
   end
 
@@ -1168,7 +1292,14 @@ defmodule Electric.ShapeCache.PureFileStorage do
     ArgumentError -> :ets_dead
   end
 
-  defp stream_from_disk(%__MODULE__{}, min_offset, max_offset, _)
+  defp stream_from_disk(
+         %__MODULE__{},
+         min_offset,
+         max_offset,
+         _,
+         _project_item,
+         _exact_upper_bound?
+       )
        when is_log_offset_lte(max_offset, min_offset),
        do: []
 
@@ -1176,20 +1307,43 @@ defmodule Electric.ShapeCache.PureFileStorage do
          %__MODULE__{} = opts,
          min_offset,
          max_offset,
-         boundary_info
+         boundary_info,
+         project_item,
+         exact_upper_bound?
        ) 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_jsons(
-          opts,
-          json_file(opts, suffix),
-          start_pos,
-          end_pos,
-          min_offset
-        )
-        |> Stream.concat(stream_from_disk(opts, chunk_end_offset, max_offset, boundary_info))
+        if exact_upper_bound? and is_log_offset_lt(max_offset, chunk_end_offset) do
+          LogFile.stream_jsons_until_offset(
+            opts,
+            json_file(opts, suffix),
+            start_pos,
+            min_offset,
+            max_offset,
+            project_item
+          )
+        else
+          LogFile.stream_jsons(
+            opts,
+            json_file(opts, suffix),
+            start_pos,
+            end_pos,
+            min_offset,
+            project_item
+          )
+          |> Stream.concat(
+            stream_from_disk(
+              opts,
+              chunk_end_offset,
+              max_offset,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
+          )
+        end
 
       {:ok, nil, {start_pos, nil}} ->
         LogFile.stream_jsons_until_offset(
@@ -1197,7 +1351,8 @@ defmodule Electric.ShapeCache.PureFileStorage do
           json_file(opts, suffix),
           start_pos,
           min_offset,
-          max_offset
+          max_offset,
+          project_item
         )
 
       :error ->
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 06761c8043..f7c1c0b317 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
@@ -179,7 +179,8 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
         log_file_path,
         start_position,
         end_position,
-        exclusive_min_offset
+        exclusive_min_offset,
+        project_item \\ &json_only/2
       ) do
     # We can read ahead entire chunk into memory since chunk sizes are expected to be ~10MB by default,
     case safely_open_file!(opts, log_file_path, [:read, :raw]) do
@@ -189,7 +190,9 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
       {:ok, file} ->
         try do
           with {:ok, data} <- :file.pread(file, start_position, end_position - start_position) do
-            {jsons, _} = extract_jsons_from_binary(data, exclusive_min_offset, nil)
+            {jsons, _} =
+              extract_jsons_from_binary(data, exclusive_min_offset, nil, project_item)
+
             jsons
           else
             :eof ->
@@ -212,7 +215,9 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
         log_file_path,
         start_position,
         exclusive_min_offset,
-        inclusive_max_offset
+        inclusive_max_offset,
+        project_item \\ &json_only/2,
+        read_fun \\ &:file.read/2
       ) do
     Stream.resource(
       fn ->
@@ -229,14 +234,18 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
         :halt ->
           {:halt, []}
 
+        {file, :halt} ->
+          {:halt, {file, ""}}
+
         {file, binary_rest} ->
-          case :file.read(file, 4096) do
+          case read_fun.(file, 4096) do
             {:ok, data} ->
               {jsons, rest} =
                 extract_jsons_from_binary(
                   binary_rest <> data,
                   exclusive_min_offset,
-                  inclusive_max_offset
+                  inclusive_max_offset,
+                  project_item
                 )
 
               {jsons, {file, rest}}
@@ -252,10 +261,23 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
     )
   end
 
-  @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 \\ [])
-  defp extract_jsons_from_binary(<<>>, _, _, acc), do: {Enum.reverse(acc), ""}
+  defp json_only(_offset, json), do: json
+
+  @spec extract_jsons_from_binary(
+          binary(),
+          LogOffset.t(),
+          LogOffset.t() | nil,
+          (LogOffset.t(), String.t() -> term())
+        ) :: {list(), binary() | :halt}
+  defp extract_jsons_from_binary(
+         binary,
+         exclusive_min_offset,
+         inclusive_max_offset,
+         project_item,
+         acc \\ []
+       )
+
+  defp extract_jsons_from_binary(<<>>, _, _, _, acc), do: {Enum.reverse(acc), ""}
 
   defp extract_jsons_from_binary(
          <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
@@ -265,41 +287,53 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
            op_offset: op_offset2
          } = log_offset,
          inclusive_max_offset,
+         project_item,
          acc
        )
        when tx_offset1 < tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 <= op_offset2),
-       do: extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, acc)
+       do: extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, project_item, acc)
 
   defp extract_jsons_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,
+         _log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2},
+         project_item,
          acc
        )
        when tx_offset1 == tx_offset2 and op_offset1 == op_offset2,
-       do: extract_jsons_from_binary("", log_offset, inclusive_max_offset, [json | acc])
+       do: {
+         Enum.reverse([
+           project_item.(LogOffset.new(tx_offset1, op_offset1), json) | acc
+         ]),
+         :halt
+       }
 
   defp extract_jsons_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,
+         _log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2},
+         _project_item,
          acc
        )
        when tx_offset1 > tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 > op_offset2),
-       do: extract_jsons_from_binary("", log_offset, inclusive_max_offset, acc)
+       do: {Enum.reverse(acc), :halt}
 
   defp extract_jsons_from_binary(
-         <<_::128, key_size::32, _::binary-size(key_size), _::8, _flag::8, json_size::64,
-           json::binary-size(json_size), rest::binary>>,
+         <<tx_offset::64, op_offset::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,
+         project_item,
          acc
        ),
-       do: extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, [json | acc])
+       do:
+         extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, project_item, [
+           project_item.(LogOffset.new(tx_offset, op_offset), json) | acc
+         ])
 
-  defp extract_jsons_from_binary(rest, _, _, acc),
+  defp extract_jsons_from_binary(rest, _, _, _, acc),
     do: {Enum.reverse(acc), rest}
 
   defp get_op_type(:insert), do: ?i
diff --git a/packages/sync-service/lib/electric/shape_cache/storage.ex b/packages/sync-service/lib/electric/shape_cache/storage.ex
index eededf50fb..f0433814db 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()
@@ -31,6 +36,7 @@ defmodule Electric.ShapeCache.Storage do
           {LogOffset.t(), key :: String.t(), operation_type :: operation_type(),
            Querying.json_iodata()}
   @type log :: Enumerable.t(Querying.json_iodata())
+  @type offset_log :: Enumerable.t({LogOffset.t() | nil, Querying.json_iodata()})
 
   @typedoc """
   A move-in snapshot row, represented as a 3-element list `[key, tags, json]`:
@@ -77,6 +83,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 +189,21 @@ defmodule Electric.ShapeCache.Storage do
   @callback get_log_stream(offset :: LogOffset.t(), max_offset :: LogOffset.t(), shape_opts()) ::
               log()
 
+  @doc """
+  Get a stream of the log together with the authoritative offsets assigned by
+  storage.
+
+  Main-log entries are returned as `{offset, item}`. Initial snapshot entries
+  have no main-log offset and are returned as `{nil, item}`.
+  """
+  @callback get_log_stream_with_offsets(
+              offset :: LogOffset.t(),
+              max_offset :: LogOffset.t(),
+              shape_opts()
+            ) :: offset_log()
+
+  @optional_callbacks get_log_stream_with_offsets: 3
+
   @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,31 @@ defmodule Electric.ShapeCache.Storage do
     []
   end
 
+  @doc """
+  Get a shape log stream with the authoritative storage offset for each item.
+
+  The lower bound is exclusive and the upper bound is inclusive, matching
+  `get_log_stream/3`. Snapshot entries are paired with `nil` because they do
+  not have main-log offsets.
+  """
+  @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
+    if Code.ensure_loaded?(mod) and function_exported?(mod, :get_log_stream_with_offsets, 3) do
+      mod.get_log_stream_with_offsets(offset, max_offset, shape_opts)
+    else
+      raise Error,
+        message: "Storage adapter #{inspect(mod)} does not support offset-preserving log streams"
+    end
+  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 4991b8cdb9..419be52ff6 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,130 @@ 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 subquery move pipeline is fully drained (Steady with an empty move
+  # queue), every move received so far has been applied to storage, so the
+  # per-dependency moves-positions can be safely advanced to the latest received
+  # source LSNs and persisted. This is the dedup key used to replay only the
+  # missed tail after a restart.
+  # 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)
+
+      try do
+        ShapeCache.Storage.set_move_positions!(positions, storage)
+      rescue
+        _ -> :ok
+      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
@@ -1063,6 +1193,18 @@ defmodule Electric.Shapes.Consumer do
          opts
        ) do
     Materializer.new_changes(Map.take(state, [:stack_id, :shape_handle]), changes_or_bounds, opts)
+  catch
+    # The consumer monitors the materializer; if the materializer died the
+    # :DOWN message is already in our mailbox and handle_materializer_down/2
+    # will run after the current handle_event/handle_call completes.
+    # Treat a `:noproc` (or transient `:normal`/`:shutdown` exit) here as
+    # the same condition: don't crash the consumer (which would route into
+    # the abnormal-shutdown path of handle_writer_termination and remove
+    # the shape from disk).
+    :exit, {:noproc, _} -> :ok
+    :exit, :noproc -> :ok
+    :exit, {:normal, _} -> :ok
+    :exit, {:shutdown, _} -> :ok
   end
 
   defp notify_materializer_of_new_changes(_state, _changes_or_bounds, _opts), do: :ok
@@ -1154,7 +1296,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,
@@ -1187,41 +1329,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),
@@ -1230,9 +1399,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(
@@ -1241,7 +1420,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 30405459c0..8c81cc362f 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -103,11 +103,29 @@ defmodule Electric.Shapes.Consumer.Materializer do
     end)
   end
 
-  def subscribe(pid) when is_pid(pid), do: GenServer.call(pid, :subscribe)
+  @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.
+  """
+  def subscribe(pid, from_lsn \\ nil)
 
-  def subscribe(opts) when is_map(opts), do: GenServer.call(name(opts), :subscribe)
+  def subscribe(pid, from_lsn) when is_pid(pid),
+    do: GenServer.call(pid, {:subscribe, from_lsn})
 
-  def subscribe(stack_id, shape_handle),
+  def subscribe(opts, from_lsn) when is_map(opts),
+    do: GenServer.call(name(opts), {:subscribe, from_lsn})
+
+  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
@@ -132,6 +150,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()
@@ -172,6 +194,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
@@ -195,7 +225,7 @@ 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) do
     cond do
       is_nil(state.subscribed_offset) ->
         state
@@ -205,7 +235,7 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
       true ->
         stream = Storage.get_log_stream(state.offset, state.subscribed_offset, storage)
-        {state, _} = stream |> decode_json_stream() |> apply_changes(state)
+        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
@@ -246,12 +276,203 @@ 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)
           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)
+
+    seed_state = replay_seed_at(state, from_lsn, storage)
+    seed = link_values_from_counts(seed_state.value_counts)
+
+    # Replay only the missing tail using storage's authoritative offsets.
+    # Move controls, snapshot-end controls, and spliced move-in rows do not
+    # encode Postgres offsets in their JSON, so JSON headers are not a safe
+    # replay cursor.
+    replay0 = %{
+      seed_state
+      | offset: from_lsn,
+        subscribed_offset: state.applied_offset,
+        pending_events: %{pid: pid}
+    }
+
+    Storage.get_log_stream_with_offsets(from_lsn, state.applied_offset, storage)
+    |> apply_replay_stream(replay0)
+
+    seed
+  end
+
+  # Reconstruct the exact dependency view at the persisted subscriber cursor.
+  # Initial snapshot entries have virtual chunk offsets rather than per-row
+  # offsets, so replay that section with the existing snapshot iterator. Read
+  # the main-log section separately through the strictly bounded offset-aware
+  # storage API; ordinary get_log_stream/3 reads whole storage chunks and may
+  # intentionally include entries past its requested max offset.
+  defp replay_seed_at(state, from_lsn, storage) do
+    snapshot_end = LogOffset.min(from_lsn, LogOffset.last_before_real_offsets())
+
+    seed0 = %{
+      state
+      | index: %{},
+        tag_indices: %{},
+        value_counts: %{},
+        offset: LogOffset.before_all(),
+        subscribed_offset: snapshot_end,
+        pending_events: %{}
+    }
+
+    seed_state = read_history_up_to_subscribed(seed0, storage)
+
+    if is_log_offset_lt(snapshot_end, from_lsn) do
+      {seed_state, _events} =
+        Storage.get_log_stream_with_offsets(snapshot_end, from_lsn, storage)
+        |> Stream.map(fn {_offset, item} -> item end)
+        |> decode_json_stream()
+        |> apply_changes(seed_state)
+
+      %{seed_state | offset: from_lsn, subscribed_offset: from_lsn}
+    else
+      seed_state
+    end
+  end
+
+  # Apply function used during move replay. The supplied stream is the missing
+  # tail `(from_lsn, applied_offset]`, paired with authoritative storage
+  # offsets. Group each transaction, apply it to the throwaway seed state, and
+  # emit its net moves at the persisted offset.
+  defp apply_replay_stream(stream, acc) do
+    %{pid: pid} = acc.pending_events
+    handle = acc.shape_handle
+
+    stream
+    |> decode_json_stream_with_offsets()
+    |> chunk_by_txn()
+    |> Enum.reduce(acc, fn {txn_offset, changes, txids}, acc ->
+      {acc, events} = apply_changes(changes, acc)
+
+      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
+
+      acc
+    end)
+  end
+
+  # Decode a raw JSON log stream into `{log_offset, txids, change}` tuples,
+  # preserving the authoritative per-item storage offset so replay can delimit
+  # transactions and tag emitted moves.
+  defp decode_json_stream_with_offsets(stream) do
+    stream
+    |> Stream.map(fn {offset, item} -> {offset, Jason.decode!(item)} end)
+    |> Stream.filter(fn {_offset, decoded} ->
+      Map.has_key?(decoded, "key") || Map.has_key?(decoded["headers"], "event")
+    end)
+    |> Stream.map(fn {offset, decoded} ->
+      headers = decoded["headers"]
+      {offset, decode_txids(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 per-transaction batches. Items sharing a
+  # `tx_offset` belong to the same source transaction; each batch is tagged with
+  # the batch's last (largest) offset as its source LSN.
+  defp chunk_by_txn(items) do
+    items
+    |> Stream.chunk_by(fn {offset, _txids, _change} -> offset.tx_offset end)
+    |> Stream.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
@@ -260,10 +481,28 @@ defmodule Electric.Shapes.Consumer.Materializer do
     {:reply, :ok, state}
   end
 
+  def handle_call(
+        {:new_changes, {_range_start, range_end}, _xid, _commit?},
+        _from,
+        %{applied_offset: applied_offset} = state
+      )
+      when is_log_offset_lte(range_end, applied_offset) do
+    # This range has already been applied — either during the startup history
+    # replay (`read_history_up_to_subscribed`) or a previous `new_changes` call.
+    # This happens on restart when the persistent replication slot re-delivers
+    # already-persisted transactions. Re-applying them would raise
+    # "Key already exists" in `apply_changes/2`, so skip the range entirely.
+    {:reply, :ok, state}
+  end
+
   def handle_call({:new_changes, {range_start, range_end}, xid, commit?}, _from, state) 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()
@@ -282,10 +521,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
@@ -465,7 +712,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/pure_file_storage/log_file_test.exs b/packages/sync-service/test/electric/shape_cache/pure_file_storage/log_file_test.exs
new file mode 100644
index 0000000000..d017620957
--- /dev/null
+++ b/packages/sync-service/test/electric/shape_cache/pure_file_storage/log_file_test.exs
@@ -0,0 +1,50 @@
+defmodule Electric.ShapeCache.PureFileStorage.LogFileTest do
+  use ExUnit.Case, async: false
+
+  alias Electric.Replication.LogOffset
+  alias Electric.ShapeCache.PureFileStorage
+  alias Electric.ShapeCache.PureFileStorage.LogFile
+
+  @moduletag :tmp_dir
+
+  test "bounded streaming stops reading the file after the inclusive max", %{tmp_dir: tmp_dir} do
+    path = Path.join(tmp_dir, "bounded-log.bin")
+    max_offset = LogOffset.new(2, 0)
+
+    entries = [
+      {LogOffset.new(1, 0), "first", :insert, Jason.encode!(%{id: "first"})},
+      {max_offset, "max", :insert, Jason.encode!(%{id: "max"})}
+    ]
+
+    tail =
+      for tx <- 3..8 do
+        json = Jason.encode!(%{id: "tail-#{tx}", padding: String.duplicate("x", 4_096)})
+        {LogOffset.new(tx, 0), "tail-#{tx}", :insert, json}
+      end
+
+    LogFile.write_log_file(entries ++ tail, path, 1_000_000)
+
+    test_pid = self()
+
+    read_fun = fn file, size ->
+      send(test_pid, :file_read)
+      :file.read(file, size)
+    end
+
+    opts = %PureFileStorage{stack_id: "test", shape_handle: "test"}
+
+    assert LogFile.stream_jsons_until_offset(
+             opts,
+             path,
+             0,
+             LogOffset.first(),
+             max_offset,
+             fn _offset, json -> json end,
+             read_fun
+           )
+           |> Enum.map(&Jason.decode!/1) == [%{"id" => "first"}, %{"id" => "max"}]
+
+    assert_receive :file_read
+    refute_receive :file_read
+  end
+end
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..fd3ccd8499 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
@@ -14,6 +14,9 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
   import Support.ComponentSetup
   import Support.TestUtils
 
+  defmodule StorageWithoutOffsetStream do
+  end
+
   @moduletag :tmp_dir
 
   @shape_handle "the-shape-handle"
@@ -49,6 +52,18 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
 
   setup [:with_stack_id_from_test, :with_async_deleter]
 
+  test "offset-preserving reads fail clearly for adapters without the optional capability" do
+    assert_raise Storage.Error,
+                 "Storage adapter #{inspect(StorageWithoutOffsetStream)} does not support offset-preserving log streams",
+                 fn ->
+                   Storage.get_log_stream_with_offsets(
+                     LogOffset.first(),
+                     LogOffset.last(),
+                     {StorageWithoutOffsetStream, :shape_opts}
+                   )
+                 end
+  end
+
   for module <- [InMemoryStorage, PureFileStorage] do
     module_name = module |> Module.split() |> List.last()
 
@@ -124,6 +139,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)}}
@@ -412,6 +454,158 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
       end
     end
 
+    describe "#{module_name}.get_log_stream_with_offsets/3" do
+      setup do
+        {:ok, %{module: unquote(module)}}
+      end
+
+      setup :start_storage
+
+      test "pairs snapshot entries with nil offsets", %{storage: storage} do
+        Storage.mark_snapshot_as_started(storage)
+        Storage.make_new_snapshot!(@data_stream, storage)
+
+        entries =
+          Storage.get_log_stream_with_offsets(
+            LogOffset.before_all(),
+            LogOffset.last_before_real_offsets(),
+            storage
+          )
+          |> Enum.to_list()
+
+        assert [{nil, _}, {nil, _}] = entries
+      end
+
+      @tag chunk_size: 1
+      test "preserves exact main-log offsets and bounds across gaps, generated entries, and hibernate",
+           %{
+             storage: storage,
+             writer: writer
+           } do
+        Storage.mark_snapshot_as_started(storage)
+        Storage.make_new_snapshot!([], storage)
+
+        base_offset = LogOffset.new(1_000, 0)
+
+        writer =
+          Storage.append_to_log!(
+            [
+              {base_offset, "regular", :insert,
+               Jason.encode!(%{
+                 key: "regular",
+                 value: %{id: "regular"},
+                 headers: %{operation: "insert"}
+               })}
+            ],
+            writer
+          )
+
+        {{^base_offset, control_offset}, writer} =
+          Storage.append_control_message!(
+            %{headers: %{event: "move-out", patterns: [%{pos: 0, value: "tag"}]}},
+            writer
+          )
+
+        Storage.write_move_in_snapshot!(
+          [
+            [
+              "move-in",
+              ["tag"],
+              Jason.encode!(%{
+                key: "move-in",
+                value: %{id: "move-in"},
+                headers: %{operation: "insert", is_move_in: true}
+              })
+            ]
+          ],
+          "move-in-snapshot",
+          storage
+        )
+
+        {{^control_offset, move_in_offset}, writer} =
+          Storage.append_move_in_snapshot_to_log!("move-in-snapshot", writer)
+
+        gapped_offset = LogOffset.new(2_000, 5)
+
+        writer =
+          Storage.append_to_log!(
+            [
+              {gapped_offset, "after-gap", :insert,
+               Jason.encode!(%{
+                 key: "after-gap",
+                 value: %{id: "after-gap"},
+                 headers: %{operation: "insert"}
+               })}
+            ],
+            writer
+          )
+
+        {{^gapped_offset, snapshot_end_offset}, writer} =
+          Storage.append_control_message!(
+            %{headers: %{control: "snapshot-end"}},
+            writer
+          )
+
+        expected_offsets = [
+          base_offset,
+          control_offset,
+          move_in_offset,
+          gapped_offset,
+          snapshot_end_offset
+        ]
+
+        read_entries = fn min_offset, max_offset ->
+          storage
+          |> then(
+            &Storage.get_log_stream_with_offsets(
+              min_offset,
+              max_offset,
+              &1
+            )
+          )
+          |> Enum.map(fn {offset, json} ->
+            label =
+              case Jason.decode!(json) do
+                %{"key" => key} -> key
+                %{"headers" => %{"event" => event}} -> event
+                %{"headers" => %{"control" => control}} -> control
+              end
+
+            {offset, label}
+          end)
+        end
+
+        expected_entries =
+          Enum.zip(expected_offsets, [
+            "regular",
+            "move-out",
+            "move-in",
+            "after-gap",
+            "snapshot-end"
+          ])
+
+        assert read_entries.(LogOffset.last_before_real_offsets(), LogOffset.last()) ==
+                 expected_entries
+
+        assert read_entries.(base_offset, LogOffset.new(1_500, 0)) ==
+                 Enum.slice(expected_entries, 1, 2)
+
+        assert read_entries.(control_offset, gapped_offset) ==
+                 Enum.slice(expected_entries, 2, 2)
+
+        _writer = Storage.hibernate(writer)
+
+        assert read_entries.(LogOffset.last_before_real_offsets(), LogOffset.last()) ==
+                 expected_entries
+
+        assert read_entries.(base_offset, LogOffset.new(1_500, 0)) ==
+                 Enum.slice(expected_entries, 1, 2)
+
+        assert read_entries.(control_offset, gapped_offset) ==
+                 Enum.slice(expected_entries, 2, 2)
+      end
+    end
+
     describe "#{module_name}.append_control_message!/2" do
       setup do
         {:ok, %{module: unquote(module)}}
@@ -931,6 +1125,75 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
     # FS is occasionally slow so we set a high timeout for compaction
     @default_compaction_timeout 2000
 
+    describe "#{module_name}.get_log_stream_with_offsets/3 disk and live tail bounds" do
+      @tag chunk_size: 1
+      setup :start_storage
+      setup :start_empty_snapshot
+
+      test "keeps sparse lower-exclusive and upper-inclusive bounds across disk and ETS", %{
+        storage: storage,
+        writer: writer
+      } do
+        entry = fn offset, label ->
+          {offset, label, :insert,
+           Jason.encode!(%{
+             key: label,
+             value: %{id: label},
+             headers: %{operation: "insert"}
+           })}
+        end
+
+        disk_first = LogOffset.new(100, 0)
+        disk_second = LogOffset.new(100, 10)
+        disk_last = LogOffset.new(200, 0)
+
+        writer =
+          Storage.append_to_log!(
+            [entry.(disk_first, "disk-first"), entry.(disk_second, "disk-second")],
+            writer
+          )
+
+        writer = Storage.append_to_log!([entry.(disk_last, "disk-last")], writer)
+        writer = Storage.hibernate(writer)
+
+        live_first = LogOffset.new(1_000, 0)
+        live_last = LogOffset.new(2_000, 0)
+        writer = Storage.append_to_log!([entry.(live_first, "live-first")], writer)
+        writer = Storage.append_to_log!([entry.(live_last, "live-last")], writer)
+
+        read = fn min_offset, max_offset ->
+          Storage.get_log_stream_with_offsets(min_offset, max_offset, storage)
+          |> Enum.map(fn {offset, json} -> {offset, Jason.decode!(json)["key"]} end)
+        end
+
+        # The upper bound falls in a deliberate offset gap inside a complete
+        # on-disk chunk. The later item from that chunk must not leak through.
+        assert read.(LogOffset.last_before_real_offsets(), LogOffset.new(100, 5)) == [
+                 {disk_first, "disk-first"}
+               ]
+
+        # The requested range crosses the persisted/live boundary and ends in
+        # another deliberate gap. Disk and ETS entries are joined once each.
+        assert read.(disk_first, LogOffset.new(1_500, 0)) == [
+                 {disk_second, "disk-second"},
+                 {disk_last, "disk-last"},
+                 {live_first, "live-first"}
+               ]
+
+        _writer = Storage.hibernate(writer)
+
+        assert read.(LogOffset.last_before_real_offsets(), LogOffset.new(100, 5)) == [
+                 {disk_first, "disk-first"}
+               ]
+
+        assert read.(disk_first, LogOffset.new(1_500, 0)) == [
+                 {disk_second, "disk-second"},
+                 {disk_last, "disk-last"},
+                 {live_first, "live-first"}
+               ]
+      end
+    end
+
     describe "#{module_name}.compact/1" do
       setup :start_storage
 
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..c4626275b2 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,27 @@ 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 (headers carry the
+  # `lsn`/`op_position` used to reconstruct the offset during replay).
+  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
+
   defp prep_changes(changes, opts \\ []) do
     pk_cols = Keyword.get(opts, :pk_cols, ["id"])
     relation = Keyword.get(opts, :relation, {"public", "test_table"})
@@ -1526,6 +1547,239 @@ 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
+
+    @tag chunk_size: 1
+    test "replays persisted move control messages at their authoritative log offsets", ctx do
+      shape_handle = "nested-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)
+
+      Storage.make_new_snapshot!(
+        make_snapshot_data([
+          %Changes.NewRecord{
+            record: %{"id" => "1", "value" => "10"},
+            move_tags: ["nested-tag"],
+            active_conditions: [true]
+          }
+        ]),
+        storage
+      )
+
+      cursor = LogOffset.new(100, 0)
+      writer = Storage.append_to_log!(main_log_insert(cursor, "2", "20"), writer)
+
+      {{^cursor, ignored_offset}, writer} =
+        Storage.append_control_message!(
+          Jason.encode!(%{headers: %{control: "up_to_date"}}),
+          writer
+        )
+
+      {{^ignored_offset, control_offset}, writer} =
+        Storage.append_control_message!(
+          Jason.encode!(%{
+            headers: %{
+              event: "move-out",
+              patterns: [%{pos: 0, value: "nested-tag"}],
+              txids: [42]
+            }
+          }),
+          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, control_offset})
+
+      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([20])
+
+      assert {:ok, seed, ^control_offset} = Materializer.subscribe(mat_ctx, cursor)
+      assert seed == MapSet.new([10, 20])
+
+      assert_receive {:materializer_changes, ^shape_handle,
+                      %{
+                        move_out: [{10, "10"}],
+                        lsn: ^control_offset,
+                        txids: [42]
+                      }}
+    end
+
+    @tag chunk_size: 1
+    test "replays persisted move-in rows at their authoritative log offsets", ctx do
+      shape_handle = "move-in-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)
+
+      Storage.make_new_snapshot!(
+        make_snapshot_data([%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}}]),
+        storage
+      )
+
+      cursor = LogOffset.new(100, 0)
+      writer = Storage.append_to_log!(main_log_insert(cursor, "2", "20"), writer)
+
+      move_in_item =
+        Jason.encode!(%{
+          key: ~s|"public"."test_table"/"3"|,
+          value: %{"id" => "3", "value" => "30"},
+          headers: %{operation: "insert", tags: ["nested-tag"], active_conditions: [true]}
+        })
+
+      Storage.write_move_in_snapshot!(
+        [[~s|"public"."test_table"/"3"|, ["nested-tag"], move_in_item]],
+        "nested-move-in",
+        storage
+      )
+
+      {{^cursor, move_in_offset}, writer} =
+        Storage.append_move_in_snapshot_to_log!("nested-move-in", 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, move_in_offset})
+
+      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])
+
+      assert {:ok, seed, ^move_in_offset} = Materializer.subscribe(mat_ctx, cursor)
+      assert seed == MapSet.new([10, 20])
+
+      assert_receive {:materializer_changes, ^shape_handle,
+                      %{
+                        move_in: [{30, "30"}],
+                        lsn: ^move_in_offset,
+                        txids: []
+                      }}
+    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 4fe0bb44d9..0bbe988123 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
 
@@ -2760,6 +2842,64 @@ defmodule Electric.Shapes.ConsumerTest do
       # After cleanup, the shape's rows should be removed from the index
       refute SubqueryIndex.has_positions?(index, shape_handle)
     end
+
+    test "dependency consumer survives a :noproc from its materializer without removing the shape",
+         ctx do
+      # Bug 6 cascade route: during a stack restart's shutdown, the dependency
+      # consumer's inline call into its materializer can race the
+      # materializer's death and exit with :noproc. Without the catch in
+      # notify_materializer_of_new_changes/3, that crashes the consumer with a
+      # non-shutdown reason, which routes through handle_writer_termination and
+      # removes the shape from disk — mid stack-shutdown that leaves the shape
+      # half-removed and 409s on the next poll after restart. The catch must
+      # absorb the exit so the pending :DOWN can drive a clean stop instead.
+
+      # Make the dependency consumer's notification call into the materializer
+      # exit exactly as a GenServer.call to an already-dead process would.
+      Repatch.patch(Consumer.Materializer, :new_changes, [mode: :shared], fn _, _, _ ->
+        exit({:noproc, {GenServer, :call, [:materializer, :new_changes, 5000]}})
+      end)
+
+      Support.TestUtils.activate_mocks_for_descendant_procs(Consumer)
+
+      # If the bug were present the consumer would crash and remove the shape;
+      # assert remove_shape is never called.
+      patch_shape_status(
+        remove_shape: fn _, handle ->
+          raise "Unexpected remove_shape for #{handle}"
+        end
+      )
+
+      {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
+
+      dep_consumer = Consumer.whereis(ctx.stack_id, dep_handle)
+      assert is_pid(dep_consumer)
+      ref = Process.monitor(dep_consumer)
+
+      # A change to the dependency table makes the dependency consumer notify
+      # its materializer — hitting the patched, exiting call.
+      ShapeLogCollector.handle_event(
+        complete_txn_fragment(100, Lsn.from_integer(50), [
+          %Changes.NewRecord{
+            relation: {"public", "other_table"},
+            record: %{"id" => "1"},
+            log_offset: LogOffset.new(Lsn.from_integer(50), 0)
+          }
+        ]),
+        ctx.stack_id
+      )
+
+      # With the catch, the dependency consumer absorbs the :noproc and stays
+      # alive; the shape is not removed.
+      refute_receive {:DOWN, ^ref, :process, _, _}, 500
+      assert Consumer.whereis(ctx.stack_id, dep_handle) == dep_consumer
+    end
   end
 
   defp refute_storage_calls_for_txn_fragment(shape_handle) do
diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs
index 6de429df03..9a0671746a 100644
--- a/packages/sync-service/test/integration/oracle_restore_test.exs
+++ b/packages/sync-service/test/integration/oracle_restore_test.exs
@@ -7,7 +7,8 @@ defmodule Electric.Integration.OracleRestoreTest do
   These tests use a small, readable "issue tracker" domain schema rather than
   the abstract `level_N` hierarchy from `Support.OracleHarness.StandardSchema`:
 
-      projects (id, active)
+      teams (id, active)
+          └── projects (id, team_id, active)
           └── issues (id, project_id, title)
 
   The shape under test is "issues belonging to an active project", expressed
@@ -35,7 +36,12 @@ 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) trips a
+        # separate post-restart long-poll readiness race (bug 3): the poll times
+        # out before replication has caught up after the 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
       )
 
@@ -50,17 +56,26 @@ defmodule Electric.Integration.OracleRestoreTest do
     %{replication_opts_overrides: [slot_temporary?: false]}
   end
 
-  # A two-table "issue tracker": projects own issues. `projects.active` drives
-  # the subquery shape under test. Seeded so that p1, p3, p5 start active and
-  # p2, p4 start inactive; issues are spread round-robin across the projects
-  # so each project owns four (e.g. p1 owns i1, i6, i11, i16).
+  # A three-table "issue tracker": teams own projects, which own issues.
+  # `projects.active` drives the existing subquery shapes, while `teams.active`
+  # drives the nested-subquery regression below. Issues are spread round-robin
+  # across the projects so each project owns four (e.g. p1 owns i1, i6, i11,
+  # i16).
   defp setup_issue_tracker_schema(ctx) do
     OracleHarness.apply_sql(ctx, [
       "DROP TABLE IF EXISTS issues CASCADE",
       "DROP TABLE IF EXISTS projects CASCADE",
+      "DROP TABLE IF EXISTS teams CASCADE",
+      """
+      CREATE TABLE teams (
+        id TEXT PRIMARY KEY,
+        active BOOLEAN NOT NULL DEFAULT true
+      )
+      """,
       """
       CREATE TABLE projects (
         id TEXT PRIMARY KEY,
+        team_id TEXT NOT NULL REFERENCES teams(id) ON DELETE CASCADE,
         active BOOLEAN NOT NULL DEFAULT true
       )
       """,
@@ -73,12 +88,16 @@ defmodule Electric.Integration.OracleRestoreTest do
       """
     ])
 
+    team_values = "('t1', true), ('t2', false)"
     project_ids = for n <- 1..5, do: "p#{n}"
 
     project_values =
       project_ids
       |> Enum.with_index()
-      |> Enum.map_join(", ", fn {id, idx} -> "('#{id}', #{rem(idx, 2) == 0})" end)
+      |> Enum.map_join(", ", fn {id, idx} ->
+        team_id = if rem(idx, 2) == 0, do: "t1", else: "t2"
+        "('#{id}', '#{team_id}', #{rem(idx, 2) == 0})"
+      end)
 
     issue_values =
       for n <- 1..20 do
@@ -88,7 +107,8 @@ defmodule Electric.Integration.OracleRestoreTest do
       |> Enum.join(", ")
 
     OracleHarness.apply_sql(ctx, [
-      "INSERT INTO projects (id, active) VALUES #{project_values}",
+      "INSERT INTO teams (id, active) VALUES #{team_values}",
+      "INSERT INTO projects (id, team_id, active) VALUES #{project_values}",
       "INSERT INTO issues (id, project_id, title) VALUES #{issue_values}"
     ])
 
@@ -129,4 +149,99 @@ defmodule Electric.Integration.OracleRestoreTest do
 
     OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
   end
+
+  @tag :oracle_restore_nested_subquery
+  test "nested subquery dependency moves replay across repeated server restarts", ctx do
+    shapes = [
+      %{
+        name: "issues_of_projects_in_active_teams",
+        table: "issues",
+        where:
+          "project_id IN (SELECT id FROM projects WHERE team_id IN " <>
+            "(SELECT id FROM teams WHERE active = true))",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      }
+    ]
+
+    # Only dependency rows change: the projects and issues remain untouched.
+    # Restart after every batch so both move-out and move-in must be restored
+    # from the persisted nested dependency logs, twice, without duplication or
+    # shape rotation.
+    batches = [
+      [[%{name: "deactivate_t1", sql: "UPDATE teams SET active = false WHERE id = 't1'"}]],
+      [[%{name: "reactivate_t1", sql: "UPDATE teams SET active = true WHERE id = 't1'"}]],
+      [[%{name: "deactivate_t1_again", sql: "UPDATE teams SET active = false WHERE id = 't1'"}]],
+      [[%{name: "reactivate_t1_again", sql: "UPDATE teams SET active = true WHERE id = 't1'"}]]
+    ]
+
+    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 shape must-refetches after restart during slot catch-up replay",
+       ctx do
+    # Two `optimized: true` subquery shapes over the same `projects` source.
+    #
+    # Regression test. Before the fix: after the restart, the `projects` source
+    # consumer replays batch_1 from the persistent slot and re-delivers those
+    # already-applied changes to the subquery materializer via `new_changes`. The
+    # materializer re-applied them and crashed ("Key ... already exists"), which
+    # cascaded — handle_materializer_down -> stop_and_clean ->
+    # handle_writer_termination({:shutdown, :cleanup}) -> remove_shape_async ->
+    # notify_shape_rotation — removing the (healthy) shapes and sending the
+    # polling client a 409 must-refetch.
+    #
+    # The fix makes the materializer ignore `new_changes` ranges it already
+    # applied during its startup history replay, so the crash (and the whole
+    # removal cascade) no longer happens. NB the underlying cascade — a
+    # materializer crash tearing down and *removing* healthy dependent shapes —
+    # is a separate hardening concern (the "bug 6" cascade) this fix leaves open.
+    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
 end
diff --git a/packages/sync-service/test/support/test_storage.ex b/packages/sync-service/test/support/test_storage.ex
index 32283ef824..ec79a88149 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,19 @@ 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})