diff --git a/.changeset/per-shape-disk-usage-metric.md b/.changeset/per-shape-disk-usage-metric.md
new file mode 100644
index 0000000000..015f9fc34c
--- /dev/null
+++ b/.changeset/per-shape-disk-usage-metric.md
@@ -0,0 +1,6 @@
+---
+"@core/electric-telemetry": patch
+"@core/sync-service": patch
+---
+
+Export in-app BEAM allocator (`vm.alloc.*`), cgroup v2 (`cgroup.*`), and per-process (`host.proc.beam.*`) metrics for reconciling VM-reported memory/CPU/disk against what the kernel charges. Also emit a new `electric.storage.dir.bytes` stack-level metric reporting on-disk size for the top-N largest shapes (tagged by shape handle), computed during the existing periodic disk-usage walk so it adds no extra filesystem traversal.
diff --git a/integration-tests/tests/otel-export.lux b/integration-tests/tests/otel-export.lux
index b325c34912..978e9c34df 100644
--- a/integration-tests/tests/otel-export.lux
+++ b/integration-tests/tests/otel-export.lux
@@ -101,6 +101,21 @@
   # Verify the presence of process.memory.total metric
   [invoke grep_otel_collector_output "Metric process\.memory\.total .*process_type=A_memory_hog.*Value=[0-9]+"]
 
+  # Verify BEAM allocator metrics (vm.alloc.*) from SystemMetrics. These come from
+  # :recon_alloc and are cross-platform; they are sampled every poll tick.
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.allocated .*Value=[0-9]+"]
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.used .*Value=[0-9]+"]
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.unused .*Value=[0-9]+"]
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.carrier_usage .*Value=[.0-9]+"]
+
+  # Verify per-process /proc metrics (host.proc.beam.*) from SystemMetrics.
+  # These read /proc and are populated on Linux (the CI runner).
+  [invoke grep_otel_collector_output "Metric host\.proc\.beam\.vm_rss .*Value=[0-9]+"]
+
+  # Verify cgroup accounting metrics (cgroup.*) from SystemMetrics. Populated when a
+  # cgroup is detected — cgroup v2 on the CI runner.
+  [invoke grep_otel_collector_output "Metric cgroup\.memory\.current .*Value=[0-9]+"]
+
   # Verify that LSN metrics are exported
   [invoke grep_otel_collector_output "Metric electric\.postgres\.replication\.pg_wal_offset .*stack_id=single_stack"]
   [invoke grep_otel_collector_output "Metric electric\.postgres\.replication\.slot_confirmed_flush_lsn_lag .*stack_id=single_stack"]
diff --git a/packages/electric-telemetry/lib/electric/telemetry/application_telemetry.ex b/packages/electric-telemetry/lib/electric/telemetry/application_telemetry.ex
index d2479ec687..708bcc7526 100644
--- a/packages/electric-telemetry/lib/electric/telemetry/application_telemetry.ex
+++ b/packages/electric-telemetry/lib/electric/telemetry/application_telemetry.ex
@@ -86,7 +86,16 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
           :get_system_memory_usage
         ],
         &{__MODULE__, &1, [telemetry_opts]}
-      )
+      ) ++
+      [
+        # System metrics (vm.alloc.*, cgroup.*, host.*); each measurement no-ops
+        # where unsupported (no cgroup v2, non-Linux). The expensive
+        # vm.alloc.fragmentation.* breakdown is sampled by SystemMonitor on a
+        # one-minute timer instead of this poller.
+        {ElectricTelemetry.SystemMetrics, :recon_alloc_measurement, [telemetry_opts]},
+        {ElectricTelemetry.SystemMetrics.Cgroup, :measurement, [telemetry_opts]},
+        {ElectricTelemetry.SystemMetrics.Proc, :measurement, [telemetry_opts]}
+      ]
   end
 
   def metrics(telemetry_opts) do
@@ -117,6 +126,35 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
       last_value("vm.memory.processes_used", unit: :byte),
       last_value("vm.memory.system", unit: :byte),
       last_value("vm.memory.total", unit: :byte),
+      # BEAM allocator metrics (emitted by ElectricTelemetry.SystemMetrics).
+      last_value("vm.alloc.allocated", unit: :byte),
+      last_value("vm.alloc.used", unit: :byte),
+      last_value("vm.alloc.unused", unit: :byte),
+      last_value("vm.alloc.carrier_usage"),
+      last_value("vm.alloc.fragmentation.unused", tags: [:allocator], unit: :byte),
+      # cgroup v2 accounting metrics (emitted by
+      # ElectricTelemetry.SystemMetrics.Cgroup).
+      last_value("cgroup.memory.current", unit: :byte),
+      last_value("cgroup.memory.anon", unit: :byte),
+      last_value("cgroup.memory.file", unit: :byte),
+      last_value("cgroup.memory.working_set", unit: :byte),
+      last_value("cgroup.memory.max", unit: :byte),
+      # PSI "full avg10" stall percentage (v2 only).
+      last_value("cgroup.memory.pressure.full.avg10"),
+      last_value("cgroup.cpu.usage_usec", unit: :microsecond),
+      last_value("cgroup.cpu.nr_throttled"),
+      last_value("cgroup.cpu.throttled_usec", unit: :microsecond),
+      last_value("cgroup.cpu.pressure.full.avg10"),
+      last_value("cgroup.io.rbytes", unit: :byte),
+      last_value("cgroup.io.wbytes", unit: :byte),
+      # per-process /proc metrics (emitted by
+      # ElectricTelemetry.SystemMetrics.Proc).
+      last_value("host.proc.beam.rss_anon", unit: :byte),
+      last_value("host.proc.beam.rss_file", unit: :byte),
+      last_value("host.proc.beam.rss_shmem", unit: :byte),
+      last_value("host.proc.beam.vm_rss", unit: :byte),
+      last_value("host.proc.beam.io.read_bytes", unit: :byte),
+      last_value("host.proc.beam.io.write_bytes", unit: :byte),
       sum("vm.monitor.long_message_queue.length", tags: [:process_type]),
       distribution("vm.monitor.long_schedule.timeout",
         tags: [:process_type],
diff --git a/packages/electric-telemetry/lib/electric/telemetry/disk_usage.ex b/packages/electric-telemetry/lib/electric/telemetry/disk_usage.ex
index 08167c42f9..143d53b812 100644
--- a/packages/electric-telemetry/lib/electric/telemetry/disk_usage.ex
+++ b/packages/electric-telemetry/lib/electric/telemetry/disk_usage.ex
@@ -4,6 +4,7 @@ defmodule ElectricTelemetry.DiskUsage do
   alias ElectricTelemetry.DiskUsage.Disk
 
   @default_update_period 60_000
+  @default_top_n 10
 
   def start_link(args) do
     {:ok, stack_id} = Keyword.fetch(args, :stack_id)
@@ -15,15 +16,29 @@ defmodule ElectricTelemetry.DiskUsage do
   end
 
   def current(stack_id) do
-    table = name(stack_id)
-
-    case :ets.lookup(table, :usage_bytes) do
-      [] -> :pending
-      [{:usage_bytes, _bytes, nil, _duration}] -> :pending
+    case lookup(stack_id, :usage_bytes) do
       [{:usage_bytes, bytes, %DateTime{}, duration}] -> {:ok, bytes, duration}
+      _ -> :pending
+    end
+  end
+
+  @doc """
+  Returns the top-N largest per-directory subtotals as a list of
+  `{dir_name, bytes}` tuples sorted descending by size, or `:pending` if no
+  measurement with grouping enabled has completed yet.
+  """
+  def current_dirs(stack_id) do
+    case lookup(stack_id, :top_dirs) do
+      [{:top_dirs, dirs}] -> {:ok, dirs}
+      _ -> :pending
     end
+  end
+
+  # The table doesn't exist until this stack's DiskUsage server has started.
+  defp lookup(stack_id, key) do
+    :ets.lookup(name(stack_id), key)
   rescue
-    ArgumentError -> :pending
+    ArgumentError -> []
   end
 
   @impl GenServer
@@ -47,9 +62,12 @@ defmodule ElectricTelemetry.DiskUsage do
         storage_dir: storage_dir,
         manual_refresh: Keyword.get(args, :manual_refresh, false),
         update_period: update_period,
+        group_depth: Keyword.get(args, :group_depth),
+        top_n: Keyword.get(args, :top_n, @default_top_n),
         usage_bytes: usage_bytes,
         updated_at: updated_at,
         measurement_duration: 0,
+        top_dirs: nil,
         timer: nil
       }
       |> ets_write()
@@ -77,32 +95,45 @@ defmodule ElectricTelemetry.DiskUsage do
   end
 
   defp read_disk_usage(state) do
-    {duration, bytes} =
+    exclude = [usage_cache_file(state.storage_dir)]
+
+    {duration, {bytes, buckets}} =
       :timer.tc(
-        fn ->
-          Disk.recursive_usage(state.storage_dir, [usage_cache_file(state.storage_dir)])
-        end,
+        fn -> Disk.recursive_usage_grouped(state.storage_dir, exclude, state.group_depth) end,
         :millisecond
       )
 
-    updated_at = DateTime.utc_now()
-
-    %{state | usage_bytes: bytes, updated_at: updated_at, measurement_duration: duration}
+    %{
+      state
+      | usage_bytes: bytes,
+        updated_at: DateTime.utc_now(),
+        measurement_duration: duration,
+        top_dirs: state.group_depth && top_n(buckets, state.top_n)
+    }
     |> ets_write()
     |> save_usage!()
     |> cancel_timer()
     |> schedule_update()
   end
 
+  # The `n` largest `{name, bytes}` buckets, sorted descending by size. The
+  # full bucket map is materialized and sorted once per (~60s) walk before
+  # being trimmed; fine at ~10k shapes per stack — replace with a bounded
+  # min-heap if a stack ever holds hundreds of thousands.
+  defp top_n(buckets, n) do
+    buckets
+    |> Enum.sort_by(fn {_name, bytes} -> bytes end, :desc)
+    |> Enum.take(n)
+  end
+
   defp ets_write(state) do
-    %{
-      usage_bytes: usage_bytes,
-      updated_at: updated_at,
-      table: table,
-      measurement_duration: duration
-    } = state
+    :ets.insert(
+      state.table,
+      {:usage_bytes, state.usage_bytes, state.updated_at, state.measurement_duration}
+    )
+
+    if dirs = state.top_dirs, do: :ets.insert(state.table, {:top_dirs, dirs})
 
-    :ets.insert(table, {:usage_bytes, usage_bytes, updated_at, duration})
     state
   end
 
diff --git a/packages/electric-telemetry/lib/electric/telemetry/disk_usage/disk.ex b/packages/electric-telemetry/lib/electric/telemetry/disk_usage/disk.ex
index 49daab9289..6cad150f3b 100644
--- a/packages/electric-telemetry/lib/electric/telemetry/disk_usage/disk.ex
+++ b/packages/electric-telemetry/lib/electric/telemetry/disk_usage/disk.ex
@@ -3,36 +3,89 @@ defmodule ElectricTelemetry.DiskUsage.Disk do
 
   Record.defrecord(:file_info, Record.extract(:file_info, from_lib: "kernel/include/file.hrl"))
 
+  @doc """
+  Recursively sum the size of all regular files under `path`, excluding any
+  paths listed in `exclude`.
+  """
   def recursive_usage(path, exclude) do
-    do_recursive_usage(path, MapSet.new(exclude), 0)
+    {total, _buckets} = recursive_usage_grouped(path, exclude, nil)
+    total
   end
 
-  def do_recursive_usage(path, exclude, acc) do
+  @doc """
+  Like `recursive_usage/2`, but in the same single traversal also returns a map
+  of per-directory subtotals bucketed at `group_depth` (`nil` disables
+  bucketing and yields an empty map).
+
+  The bucket map is keyed by the name (not full path) of each directory found
+  at exactly `group_depth` levels below `path` (the root `path` itself is depth
+  0), with the value being the recursive size of every regular file under that
+  directory. Excluded paths contribute to neither the total nor the buckets.
+  The buckets are a best-effort tally that only ever grows, while the total
+  keeps the legacy behaviour of resetting to 0 on an unreadable entry.
+
+  Returns `{total_bytes, %{dir_name => bytes}}`.
+  """
+  def recursive_usage_grouped(path, exclude, group_depth)
+      when is_nil(group_depth) or (is_integer(group_depth) and group_depth >= 0) do
+    # When grouping at the root (depth 0), the root's own basename is the bucket.
+    initial_key = if group_depth == 0, do: Path.basename(path), else: nil
+    walk(path, initial_key, MapSet.new(exclude), group_depth, 0, {0, %{}})
+  end
+
+  # `bucket_key` is the name of the ancestor directory sitting at `group_depth`,
+  # or `nil` until that depth is reached.
+  defp walk(path, bucket_key, exclude, group_depth, depth, {acc, buckets}) do
     case stat(path) do
       {:ok, file_info(size: size, type: :regular)} ->
         if MapSet.member?(exclude, path) do
-          acc
+          {acc, buckets}
         else
-          size + acc
+          {size + acc, add_to_bucket(buckets, bucket_key, size)}
         end
 
       {:ok, file_info(type: :directory)} ->
         case ls(path) do
           {:ok, files} ->
-            Enum.reduce(files, acc, &do_recursive_usage(Path.join(path, &1), exclude, &2))
+            Enum.reduce(files, {acc, buckets}, fn name, {acc, bucks} ->
+              # `name` is a charlist from :prim_file.list_dir/1 and sits at
+              # `depth + 1`. Once established, the bucket key propagates to all
+              # descendants.
+              child_bucket_key =
+                cond do
+                  not is_nil(bucket_key) -> bucket_key
+                  depth + 1 == group_depth -> List.to_string(name)
+                  true -> nil
+                end
+
+              walk(
+                Path.join(path, name),
+                child_bucket_key,
+                exclude,
+                group_depth,
+                depth + 1,
+                {acc, bucks}
+              )
+            end)
 
           {:error, _} ->
-            0
+            {0, buckets}
         end
 
       {:ok, _} ->
-        0
+        {0, buckets}
 
       {:error, _} ->
-        0
+        {0, buckets}
     end
   end
 
+  defp add_to_bucket(buckets, nil, _size), do: buckets
+
+  defp add_to_bucket(buckets, key, size) do
+    Map.update(buckets, key, size, &(&1 + size))
+  end
+
   defdelegate stat(path), to: :prim_file, as: :read_file_info
 
   defp ls(path) do
diff --git a/packages/electric-telemetry/lib/electric/telemetry/opts.ex b/packages/electric-telemetry/lib/electric/telemetry/opts.ex
index fedd966578..7082c1bfa0 100644
--- a/packages/electric-telemetry/lib/electric/telemetry/opts.ex
+++ b/packages/electric-telemetry/lib/electric/telemetry/opts.ex
@@ -5,6 +5,9 @@ defmodule ElectricTelemetry.Opts do
       installation_id: [type: :string],
       stack_id: [type: :string],
       storage_dir: [type: :string],
+      # Directory depth (relative to the disk-usage walk root) at which to
+      # bucket per-directory disk usage subtotals; `nil` disables bucketing.
+      disk_usage_group_depth: [type: {:or, [:pos_integer, nil]}, default: nil],
       version: [type: :string, required: true],
       reporters: [
         type: :keyword_list,
diff --git a/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex b/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex
index e8d6086110..91884bceb1 100644
--- a/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex
+++ b/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex
@@ -67,9 +67,20 @@ defmodule ElectricTelemetry.StackTelemetry do
     ]
   end
 
+  # `:disk_usage_group_depth` is supplied by the caller since it depends on the
+  # deployment's storage layout (sync-service derives it from the real shape
+  # storage path, keeping this package layout-agnostic). When set, the disk walk
+  # additionally buckets per-directory subtotals at that depth — for Electric
+  # these are shape dirs, emitted as `electric.storage.dir.bytes` for the top-N
+  # largest. When `nil`, only the aggregate total is measured.
   defp disk_usage_child_specs(%{stack_id: stack_id} = opts) do
     if storage_dir = Map.get(opts, :storage_dir) do
-      [{ElectricTelemetry.DiskUsage, stack_id: stack_id, storage_dir: storage_dir}]
+      [
+        {ElectricTelemetry.DiskUsage,
+         stack_id: stack_id,
+         storage_dir: storage_dir,
+         group_depth: Map.get(opts, :disk_usage_group_depth)}
+      ]
     else
       []
     end
@@ -113,6 +124,7 @@ defmodule ElectricTelemetry.StackTelemetry do
       distribution("electric.storage.transaction_stored.replication_lag", unit: :millisecond),
       last_value("electric.storage.used.bytes", unit: :byte),
       distribution("electric.storage.used.measurement_duration", unit: :millisecond),
+      last_value("electric.storage.dir.bytes", unit: :byte, tags: [:stack_id, :shape]),
       counter("electric.postgres.replication.transaction_received.count"),
       sum("electric.postgres.replication.transaction_received.bytes", unit: :byte),
       sum("electric.storage.transaction_stored.bytes", unit: :byte),
diff --git a/packages/electric-telemetry/lib/electric/telemetry/system_metrics.ex b/packages/electric-telemetry/lib/electric/telemetry/system_metrics.ex
new file mode 100644
index 0000000000..7489a4bfe9
--- /dev/null
+++ b/packages/electric-telemetry/lib/electric/telemetry/system_metrics.ex
@@ -0,0 +1,130 @@
+defmodule ElectricTelemetry.SystemMetrics do
+  @moduledoc """
+  Periodic measurement functions for low-level system/runtime metrics that
+  complement the application-level VM stats in `ElectricTelemetry.ApplicationTelemetry`.
+
+  These functions are not a process; `ElectricTelemetry.Poller` invokes them
+  (and the `SystemMetrics.Cgroup` / `SystemMetrics.Proc` readers) as MFA tuples
+  on its ~5s tick, wrapping every invocation in `safe_invoke/3` so a crash here
+  is logged and swallowed rather than removing the measurement. The one
+  exception is `allocator_fragmentation_measurement/0`, which is too expensive
+  for that tick and is instead sampled by `ElectricTelemetry.SystemMonitor` on
+  a one-minute timer.
+
+  This module holds the BEAM allocator metrics (`vm.alloc.*`, derived from
+  `:recon_alloc`), the cached platform/cgroup detection (`system_info/0`), and
+  the shared `emit/2` helper used by the cgroup//proc readers.
+  """
+
+  @system_info_key {__MODULE__, :system_info}
+
+  @doc """
+  Emit a telemetry event carrying the non-nil `measurements` in a single
+  `:telemetry.execute/2` call, or nothing when every value is nil (file
+  missing / unreadable / metric intentionally skipped).
+  """
+  @spec emit(:telemetry.event_name(), %{optional(atom()) => number() | nil}) :: :ok
+  def emit(event, measurements) do
+    measurements = Map.reject(measurements, fn {_key, value} -> is_nil(value) end)
+    if map_size(measurements) > 0, do: :telemetry.execute(event, measurements)
+    :ok
+  end
+
+  @doc """
+  Boot-time platform/cgroup detection, computed once and cached in `:persistent_term`.
+
+  Returns a map of the form `%{os: {family, name}, cgroup_version: :v2 | :none}`.
+  Cgroup v1 hosts report `:none` — the `cgroup.*` metrics are v2-only.
+  """
+  @spec system_info() :: %{os: {atom(), atom()}, cgroup_version: :v2 | :none}
+  def system_info do
+    memoized(@system_info_key, fn ->
+      os = :os.type()
+      %{os: os, cgroup_version: detect_cgroup_version(os)}
+    end)
+  end
+
+  # The controllers list at the cgroup fs root is the canonical v2 marker.
+  defp detect_cgroup_version({:unix, :linux}) do
+    if File.exists?("/sys/fs/cgroup/cgroup.controllers"), do: :v2, else: :none
+  end
+
+  defp detect_cgroup_version(_non_linux), do: :none
+
+  @doc """
+  Cheap aggregate BEAM allocator metrics, sampled every poll tick.
+
+  Emits the `[:vm, :alloc]` telemetry event carrying:
+
+    * `:allocated` — bytes the allocators have requested from the OS (carriers)
+    * `:used`      — bytes actually in use by blocks
+    * `:unused`    — `allocated - used`, i.e. fragmentation/headroom held in carriers
+    * `:carrier_usage` — `used / allocated`, the carrier usage ratio in `0..1`
+  """
+  @spec recon_alloc_measurement(map()) :: :ok
+  def recon_alloc_measurement(_telemetry_opts) do
+    # Each :recon_alloc.memory/1 call sweeps every allocator instance, so read
+    # the two base numbers once and derive the rest locally.
+    allocated = :recon_alloc.memory(:allocated)
+    used = :recon_alloc.memory(:used)
+
+    emit([:vm, :alloc], %{
+      allocated: allocated,
+      used: used,
+      unused: max(allocated - used, 0),
+      carrier_usage: if(allocated > 0, do: used / allocated)
+    })
+  end
+
+  @doc """
+  Per-allocator fragmentation breakdown.
+
+  Sums unused carrier bytes from `:recon_alloc.fragmentation(:current)` per
+  allocator *type* and emits one `[:vm, :alloc, :fragmentation]` event per
+  type, tagged by `allocator`. This is O(carriers) — too expensive for the
+  poller's ~5s tick — so `ElectricTelemetry.SystemMonitor` samples it on its
+  own one-minute timer instead.
+  """
+  @spec allocator_fragmentation_measurement() :: :ok
+  def allocator_fragmentation_measurement do
+    :recon_alloc.fragmentation(:current)
+    |> Enum.reduce(%{}, fn {{type, _instance}, info}, acc ->
+      unused = unused_bytes(info)
+      Map.update(acc, type, unused, &(&1 + unused))
+    end)
+    |> Enum.each(fn {type, unused} ->
+      :telemetry.execute(
+        [:vm, :alloc, :fragmentation],
+        %{unused: unused},
+        %{allocator: to_string(type)}
+      )
+    end)
+
+    :ok
+  end
+
+  @doc """
+  Get-or-compute a `:persistent_term` entry. The put happens at most once per
+  key (on first call), never on a per-tick hot path.
+  """
+  @spec memoized(term(), (-> value)) :: value when value: term()
+  def memoized(key, fun) do
+    with :undefined <- :persistent_term.get(key, :undefined) do
+      tap(fun.(), &:persistent_term.put(key, &1))
+    end
+  end
+
+  # `:recon_alloc.fragmentation/1` returns a proplist per allocator with carrier
+  # and block sizes for both single-block (sbcs) and multi-block (mbcs)
+  # carriers. Unused bytes are the carrier bytes the OS gave us minus the bytes
+  # actually filled by blocks.
+  defp unused_bytes(info) do
+    carriers =
+      Keyword.get(info, :sbcs_carriers_size, 0) + Keyword.get(info, :mbcs_carriers_size, 0)
+
+    blocks =
+      Keyword.get(info, :sbcs_block_size, 0) + Keyword.get(info, :mbcs_block_size, 0)
+
+    max(carriers - blocks, 0)
+  end
+end
diff --git a/packages/electric-telemetry/lib/electric/telemetry/system_metrics/cgroup.ex b/packages/electric-telemetry/lib/electric/telemetry/system_metrics/cgroup.ex
new file mode 100644
index 0000000000..5e8d6e6813
--- /dev/null
+++ b/packages/electric-telemetry/lib/electric/telemetry/system_metrics/cgroup.ex
@@ -0,0 +1,145 @@
+defmodule ElectricTelemetry.SystemMetrics.Cgroup do
+  @moduledoc """
+  Reads cgroup v2 accounting files from `/sys/fs/cgroup/*` and emits `cgroup.*`
+  telemetry.
+
+  All reads are defensive: a missing file, a permission error, or unexpected
+  content results in the corresponding metric being skipped, never a crash. The
+  cgroup version is taken from `ElectricTelemetry.SystemMetrics.system_info/0`
+  (cached at boot) so we never stat the filesystem on the hot path. Anything
+  other than v2 (cgroup v1 hosts, non-Linux platforms) is a clean no-op.
+
+  Units: memory and io metrics are bytes, cpu usage / throttled time are
+  microseconds, PSI `full avg10` is a stall percentage (float).
+  """
+
+  alias ElectricTelemetry.SystemMetrics
+
+  import ElectricTelemetry.SystemMetrics, only: [emit: 2]
+
+  import ElectricTelemetry.SystemMetrics.ProcfsParse,
+    only: [read_kv_file: 2, read_int_file: 1, read_raw_file: 1, parse_int: 1, parse_float: 1]
+
+  @default_root "/sys/fs/cgroup"
+
+  @files %{
+    memory_current: "memory.current",
+    memory_stat: "memory.stat",
+    memory_max: "memory.max",
+    memory_pressure: "memory.pressure",
+    cpu_stat: "cpu.stat",
+    cpu_pressure: "cpu.pressure",
+    io_stat: "io.stat"
+  }
+  @default_paths Map.new(@files, fn {key, file} -> {key, Path.join(@default_root, file)} end)
+
+  @memory_stat_keys MapSet.new(~w(anon file inactive_file))
+  @cpu_stat_keys MapSet.new(~w(usage_usec nr_throttled throttled_usec))
+
+  @doc """
+  Read the cgroup accounting files and emit `cgroup.*` telemetry events.
+
+  Reads every tick (no slow-tier gating). Options:
+
+    * `:cgroup_root` — override the cgroup filesystem root (default
+      `#{@default_root}`); used by tests to point at fixture trees.
+    * `:cgroup_version` — override the detected cgroup version (`:v2`/`:none`);
+      used by tests. Defaults to the cached `SystemMetrics.system_info/0` value.
+  """
+  @spec measurement(map(), keyword()) :: :ok
+  def measurement(_telemetry_opts, opts \\ []) do
+    version =
+      Keyword.get_lazy(opts, :cgroup_version, fn -> SystemMetrics.system_info().cgroup_version end)
+
+    if version == :v2 do
+      paths = opts |> Keyword.get(:cgroup_root, @default_root) |> paths()
+      emit_memory(paths)
+      emit_cpu(paths)
+      emit([:cgroup, :io], io_totals(read_raw_file(paths.io_stat)))
+    end
+
+    :ok
+  end
+
+  # The production file paths are compile-time constants; only tests pass a
+  # different root.
+  defp paths(@default_root), do: @default_paths
+  defp paths(root), do: Map.new(@files, fn {key, file} -> {key, Path.join(root, file)} end)
+
+  defp emit_memory(paths) do
+    current = read_int_file(paths.memory_current)
+    stat = read_kv_file(paths.memory_stat, @memory_stat_keys)
+    inactive_file = stat["inactive_file"]
+
+    emit([:cgroup, :memory], %{
+      current: current,
+      anon: stat["anon"],
+      file: stat["file"],
+      working_set: current && inactive_file && max(current - inactive_file, 0),
+      # An unlimited memory.max is the literal "max", which parses to nil and is
+      # skipped (the host-RAM ceiling comes from the /proc reader).
+      max: read_int_file(paths.memory_max)
+    })
+
+    emit_pressure(paths.memory_pressure, :memory)
+  end
+
+  defp emit_cpu(paths) do
+    stat = read_kv_file(paths.cpu_stat, @cpu_stat_keys)
+
+    emit([:cgroup, :cpu], %{
+      usage_usec: stat["usage_usec"],
+      nr_throttled: stat["nr_throttled"],
+      throttled_usec: stat["throttled_usec"]
+    })
+
+    emit_pressure(paths.cpu_pressure, :cpu)
+  end
+
+  # PSI files may be absent even on v2 (kernel built without PSI); a nil avg10
+  # is dropped by emit/2, skipping the event.
+  defp emit_pressure(path, plane) do
+    emit([:cgroup, plane, :pressure, :full], %{avg10: psi_full_avg10(read_raw_file(path))})
+  end
+
+  # io.stat is one line per device: "MAJ:MIN rbytes=… wbytes=… rios=… …".
+  # Sum rbytes/wbytes across all devices; a key never seen stays absent.
+  defp io_totals(nil), do: %{}
+
+  defp io_totals(content) do
+    for field <- String.split(content), reduce: %{} do
+      acc ->
+        case field do
+          "rbytes=" <> value -> add_field(acc, :rbytes, parse_int(value))
+          "wbytes=" <> value -> add_field(acc, :wbytes, parse_int(value))
+          _ -> acc
+        end
+    end
+  end
+
+  defp add_field(acc, _key, nil), do: acc
+  defp add_field(acc, key, value), do: Map.update(acc, key, value, &(&1 + value))
+
+  # PSI file format (one "some"/"full" line each):
+  #   some avg10=0.00 avg60=0.00 avg300=0.00 total=12345
+  #   full avg10=0.00 avg60=0.00 avg300=0.05 total=6789
+  # Extract avg10 from the "full" line as a float.
+  defp psi_full_avg10(nil), do: nil
+
+  defp psi_full_avg10(content) do
+    content
+    |> String.split("\n", trim: true)
+    |> Enum.find_value(fn line ->
+      case String.split(line, " ", trim: true) do
+        ["full" | fields] ->
+          Enum.find_value(fields, fn
+            "avg10=" <> rest -> parse_float(rest)
+            _ -> nil
+          end)
+
+        _ ->
+          nil
+      end
+    end)
+  end
+end
diff --git a/packages/electric-telemetry/lib/electric/telemetry/system_metrics/proc.ex b/packages/electric-telemetry/lib/electric/telemetry/system_metrics/proc.ex
new file mode 100644
index 0000000000..727e709b82
--- /dev/null
+++ b/packages/electric-telemetry/lib/electric/telemetry/system_metrics/proc.ex
@@ -0,0 +1,92 @@
+defmodule ElectricTelemetry.SystemMetrics.Proc do
+  @moduledoc """
+  Reads the BEAM's own `/proc/<pid>/status` and `/proc/<pid>/io` and emits
+  `host.proc.beam.*` telemetry: the per-process RSS breakdown (anon vs file vs
+  shmem) shows cgroup `anon` tracking the BEAM's anonymous footprint, and the
+  IO byte counters give the kernel's view of this process's disk traffic.
+
+  All reads are defensive: a missing file, a permission error (`/proc/<pid>/io`
+  can be EACCES-restricted even for self in some sandboxes), or unexpected
+  content results in the corresponding metric being skipped, never a crash.
+  `/proc` only exists on Linux, so the measurement is a clean no-op everywhere
+  else.
+
+  Status `Rss*`/`VmRSS` values are in kB and converted to bytes;
+  `/proc/<pid>/io` `read_bytes`/`write_bytes` are already bytes and emitted
+  as-is.
+  """
+
+  alias ElectricTelemetry.SystemMetrics
+
+  import ElectricTelemetry.SystemMetrics, only: [emit: 2]
+  import ElectricTelemetry.SystemMetrics.ProcfsParse, only: [read_kv_file: 2]
+
+  @default_root "/proc"
+
+  @status_keys MapSet.new(~w(RssAnon RssFile RssShmem VmRSS))
+  @io_keys MapSet.new(~w(read_bytes write_bytes))
+
+  @doc """
+  Read the BEAM's `/proc/<pid>` accounting files and emit `host.proc.beam.*`
+  telemetry events.
+
+  Reads every tick (no slow-tier gating). Options:
+
+    * `:proc_root` — override the `/proc` filesystem root (default
+      `#{@default_root}`); used by tests to point at fixture trees.
+    * `:pid` — override the OS pid of the BEAM whose per-process files are read
+      (default `:os.getpid()`); used by tests.
+    * `:os` — override the detected OS (`{family, name}` tuple); used by tests.
+      Defaults to the cached `SystemMetrics.system_info/0` value.
+  """
+  @spec measurement(map(), keyword()) :: :ok
+  def measurement(_telemetry_opts, opts \\ []) do
+    os = Keyword.get_lazy(opts, :os, fn -> SystemMetrics.system_info().os end)
+
+    if os == {:unix, :linux} do
+      %{status: status_path, io: io_path} = paths(opts)
+
+      status = read_kv_file(status_path, @status_keys)
+
+      emit([:host, :proc, :beam], %{
+        rss_anon: kb_to_bytes(status["RssAnon"]),
+        rss_file: kb_to_bytes(status["RssFile"]),
+        rss_shmem: kb_to_bytes(status["RssShmem"]),
+        vm_rss: kb_to_bytes(status["VmRSS"])
+      })
+
+      io = read_kv_file(io_path, @io_keys)
+
+      emit([:host, :proc, :beam, :io], %{
+        read_bytes: io["read_bytes"],
+        write_bytes: io["write_bytes"]
+      })
+    end
+
+    :ok
+  end
+
+  # In production (no opts) the BEAM pid and the file paths never change for
+  # the life of the VM, so resolve them once and cache in :persistent_term;
+  # test overrides recompute.
+  defp paths([]) do
+    SystemMetrics.memoized({__MODULE__, :paths}, fn ->
+      build_paths(@default_root, :os.getpid())
+    end)
+  end
+
+  defp paths(opts) do
+    build_paths(
+      Keyword.get(opts, :proc_root, @default_root),
+      Keyword.get_lazy(opts, :pid, fn -> :os.getpid() end)
+    )
+  end
+
+  defp build_paths(root, pid) do
+    dir = Path.join(root, to_string(pid))
+    %{status: Path.join(dir, "status"), io: Path.join(dir, "io")}
+  end
+
+  defp kb_to_bytes(nil), do: nil
+  defp kb_to_bytes(kb), do: kb * 1024
+end
diff --git a/packages/electric-telemetry/lib/electric/telemetry/system_metrics/procfs_parse.ex b/packages/electric-telemetry/lib/electric/telemetry/system_metrics/procfs_parse.ex
new file mode 100644
index 0000000000..c03542b256
--- /dev/null
+++ b/packages/electric-telemetry/lib/electric/telemetry/system_metrics/procfs_parse.ex
@@ -0,0 +1,84 @@
+defmodule ElectricTelemetry.SystemMetrics.ProcfsParse do
+  @moduledoc """
+  Small, exhaustively defensive parsers for the flat text files exposed by
+  procfs and cgroupfs.
+
+  Every function degrades to an empty map (or `nil` for scalars) on a missing
+  file, a permission error, or malformed content — it never raises. Parsing
+  `/proc` is Electric's #1 historical crash cause, so the contract here is
+  "skip the bad value, keep the good ones".
+
+  No file content is ever turned into an atom (`String.to_atom/1`); keys stay
+  binaries and the caller looks them up against compile-time literals.
+  """
+
+  @doc """
+  Parse a flat key/value file — `/proc/<pid>/status`, `/proc/<pid>/io`, cgroup
+  `memory.stat`/`cpu.stat` — into a map of `binary key => integer value`,
+  keeping only the keys in the `keys` set. These files carry dozens of lines of
+  which a caller consumes a handful; non-wanted lines skip value parsing.
+
+  Handles both the `key value` and `Key:\twhitespace-padded value [unit]` line
+  layouts: the key is the first whitespace-delimited token with any trailing
+  `:` stripped, the value is the second token, and any trailing unit (e.g.
+  `kB`) is ignored. Lines whose value doesn't parse as an integer are dropped.
+  Values keep their native unit (the caller converts, e.g. kB -> bytes).
+  """
+  @spec read_kv_file(Path.t(), MapSet.t(binary())) :: %{optional(binary()) => integer()}
+  def read_kv_file(path, keys) do
+    for line <- String.split(read_raw_file(path) || "", "\n", trim: true),
+        [key, value | _] <- [String.split(line)],
+        key = String.trim_trailing(key, ":"),
+        MapSet.member?(keys, key),
+        int = parse_int(value),
+        into: %{},
+        do: {key, int}
+  end
+
+  @doc """
+  Read a single-integer file (e.g. cgroup `memory.current`), returning `nil`
+  on any read or parse error.
+  """
+  @spec read_int_file(Path.t()) :: integer() | nil
+  def read_int_file(path), do: parse_int(read_raw_file(path))
+
+  @doc """
+  Read a file, returning its trimmed content or `nil` on any error (missing
+  file, permission denied, …).
+  """
+  @spec read_raw_file(Path.t()) :: binary() | nil
+  def read_raw_file(path) do
+    case File.read(path) do
+      {:ok, content} -> String.trim(content)
+      {:error, _reason} -> nil
+    end
+  end
+
+  @doc """
+  Parse a string as an integer, returning `nil` on `nil` input or any
+  non-integer content.
+  """
+  @spec parse_int(binary() | nil) :: integer() | nil
+  def parse_int(nil), do: nil
+
+  def parse_int(str) do
+    case Integer.parse(String.trim(str)) do
+      {int, ""} -> int
+      _ -> nil
+    end
+  end
+
+  @doc """
+  Parse a string as a float, returning `nil` on `nil` input or any non-float
+  content.
+  """
+  @spec parse_float(binary() | nil) :: float() | nil
+  def parse_float(nil), do: nil
+
+  def parse_float(str) do
+    case Float.parse(String.trim(str)) do
+      {float, ""} -> float
+      _ -> nil
+    end
+  end
+end
diff --git a/packages/electric-telemetry/lib/electric/telemetry/system_monitor.ex b/packages/electric-telemetry/lib/electric/telemetry/system_monitor.ex
index a1b3d82160..34a956da05 100644
--- a/packages/electric-telemetry/lib/electric/telemetry/system_monitor.ex
+++ b/packages/electric-telemetry/lib/electric/telemetry/system_monitor.ex
@@ -8,6 +8,9 @@ defmodule ElectricTelemetry.SystemMonitor do
     - long_schedule
     - long_message_queue
 
+  It also hosts slow periodic work that doesn't fit the poller's single ~5s
+  period: the per-allocator fragmentation sampling (`vm.alloc.fragmentation.*`)
+  runs on its own one-minute timer here.
   """
 
   use GenServer
@@ -21,6 +24,8 @@ defmodule ElectricTelemetry.SystemMonitor do
   @vm_monitor_long_message_queue [:vm, :monitor, :long_message_queue]
   @garbage_collect_interval :timer.hours(1)
   @garbage_collect_message :periodic_garbage_collect
+  @allocator_fragmentation_interval :timer.minutes(1)
+  @allocator_fragmentation_message :sample_allocator_fragmentation
 
   def start_link(opts) do
     GenServer.start_link(__MODULE__, opts.intervals_and_thresholds, name: __MODULE__)
@@ -40,6 +45,8 @@ defmodule ElectricTelemetry.SystemMonitor do
       garbage_collect_timer: nil
     }
 
+    schedule_allocator_fragmentation()
+
     {:ok, schedule_garbage_collect(state)}
   end
 
@@ -142,6 +149,19 @@ defmodule ElectricTelemetry.SystemMonitor do
     {:noreply, %{state | garbage_collect_timer: nil} |> schedule_garbage_collect()}
   end
 
+  def handle_info(@allocator_fragmentation_message, state) do
+    # Don't let a metrics hiccup take down the system monitor.
+    try do
+      ElectricTelemetry.SystemMetrics.allocator_fragmentation_measurement()
+    rescue
+      error -> Logger.warning("Allocator fragmentation sampling failed: #{inspect(error)}")
+    end
+
+    schedule_allocator_fragmentation()
+
+    {:noreply, state}
+  end
+
   defp log_long_message_queue_event(pid, type) do
     with {:message_queue_len, queue_len} <- Process.info(pid, :message_queue_len) do
       :telemetry.execute(@vm_monitor_long_message_queue, %{length: queue_len}, %{
@@ -163,6 +183,14 @@ defmodule ElectricTelemetry.SystemMonitor do
 
   defp maybe_start_long_message_queue_timer(state), do: state
 
+  defp schedule_allocator_fragmentation do
+    Process.send_after(
+      self(),
+      @allocator_fragmentation_message,
+      @allocator_fragmentation_interval
+    )
+  end
+
   defp schedule_garbage_collect(%{garbage_collect_timer: nil} = state) do
     %{
       state
diff --git a/packages/electric-telemetry/mix.exs b/packages/electric-telemetry/mix.exs
index 5ad3b79f28..f8fe494745 100644
--- a/packages/electric-telemetry/mix.exs
+++ b/packages/electric-telemetry/mix.exs
@@ -20,6 +20,7 @@ defmodule ElectricTelemetry.MixProject do
     List.flatten(
       [
         {:otel_metric_exporter, "~> 0.4.1"},
+        {:recon, "~> 2.5"},
         {:req, "~> 0.5"},
         {:telemetry, "~> 1.3"},
         {:telemetry_metrics, "~> 1.1"},
diff --git a/packages/electric-telemetry/mix.lock b/packages/electric-telemetry/mix.lock
index 5f7cd4e908..4aa7b357b0 100644
--- a/packages/electric-telemetry/mix.lock
+++ b/packages/electric-telemetry/mix.lock
@@ -28,6 +28,7 @@
   "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"},
   "protobuf": {:hex, :protobuf, "0.17.0", "39e24e43c9648e148feba16ed51100b5b2028ea900b55460377b0476f6e10613", [:mix], [{:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "ca6c91f6f63e2c147b47f03eefd10b80538aa6fc55ff4b12b795efb786b0152f"},
   "ranch": {:hex, :ranch, "1.8.1", "208169e65292ac5d333d6cdbad49388c1ae198136e4697ae2f474697140f201c", [:make, :rebar3], [], "hexpm", "aed58910f4e21deea992a67bf51632b6d60114895eb03bb392bb733064594dd0"},
+  "recon": {:hex, :recon, "2.5.6", "9052588e83bfedfd9b72e1034532aee2a5369d9d9343b61aeb7fbce761010741", [:mix, :rebar3], [], "hexpm", "96c6799792d735cc0f0fd0f86267e9d351e63339cbe03df9d162010cefc26bb0"},
   "req": {:hex, :req, "0.6.1", "7b904c8b42d0e08136a5c6aba024fd12fc79a1ed8856e7a3522b0917f7e75113", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.21.0 or ~> 0.22.0", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "aaf11c9c80f2df2364630b3594e1857fe610d8ea7cb994e1ce3dcb55f204ff1c"},
   "retry": {:hex, :retry, "0.19.0", "aeb326d87f62295d950f41e1255fe6f43280a1b390d36e280b7c9b00601ccbc2", [:mix], [], "hexpm", "85ef376aa60007e7bff565c366310966ec1bd38078765a0e7f20ec8a220d02ca"},
   "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"},
diff --git a/packages/electric-telemetry/test/electric/telemetry/disk_usage/disk_test.exs b/packages/electric-telemetry/test/electric/telemetry/disk_usage/disk_test.exs
new file mode 100644
index 0000000000..87268600fc
--- /dev/null
+++ b/packages/electric-telemetry/test/electric/telemetry/disk_usage/disk_test.exs
@@ -0,0 +1,116 @@
+defmodule ElectricTelemetry.DiskUsage.DiskTest do
+  use ExUnit.Case, async: true
+
+  alias ElectricTelemetry.DiskUsage.Disk
+
+  @moduletag :tmp_dir
+
+  defp write_file(path, size) do
+    File.mkdir_p!(Path.dirname(path))
+    File.write!(path, :binary.copy("0", size), [:raw, :binary])
+    size
+  end
+
+  describe "recursive_usage/2" do
+    test "sums all regular files recursively", %{tmp_dir: dir} do
+      write_file(Path.join(dir, "a/b/c/1.data"), 100)
+      write_file(Path.join(dir, "a/b/c/2.data"), 50)
+      write_file(Path.join(dir, "a/x/y/z/1.data"), 25)
+
+      assert Disk.recursive_usage(dir, []) == 175
+    end
+
+    test "excludes listed files", %{tmp_dir: dir} do
+      write_file(Path.join(dir, "keep.data"), 100)
+      excluded = Path.join(dir, "skip.data")
+      write_file(excluded, 999)
+
+      assert Disk.recursive_usage(dir, [excluded]) == 100
+    end
+  end
+
+  describe "recursive_usage_grouped/3" do
+    test "total is byte-identical to recursive_usage/2 from a single walk", %{tmp_dir: dir} do
+      write_file(Path.join(dir, "s1/p/q/shape-a/1.data"), 100)
+      write_file(Path.join(dir, "s1/p/q/shape-a/2.data"), 200)
+      write_file(Path.join(dir, "s1/p/q/shape-b/1.data"), 50)
+      write_file(Path.join(dir, "loose.data"), 7)
+
+      {total, _buckets} = Disk.recursive_usage_grouped(dir, [], 4)
+      assert total == Disk.recursive_usage(dir, [])
+      assert total == 357
+    end
+
+    test "buckets by directory name at the configured depth", %{tmp_dir: dir} do
+      # Layout: <dir>/<stack>/<p1>/<p2>/<shape_handle>/...  => shape at depth 4
+      write_file(Path.join(dir, "stack/aa/bb/shape-a/log/1.data"), 100)
+      write_file(Path.join(dir, "stack/aa/bb/shape-a/log/2.data"), 200)
+      write_file(Path.join(dir, "stack/aa/bb/shape-b/snap/1.data"), 50)
+      write_file(Path.join(dir, "stack/cc/dd/shape-c/1.data"), 10)
+
+      {total, buckets} = Disk.recursive_usage_grouped(dir, [], 4)
+
+      assert total == 360
+      assert buckets == %{"shape-a" => 300, "shape-b" => 50, "shape-c" => 10}
+    end
+
+    test "depth 0 buckets the whole tree under the root's basename", %{tmp_dir: dir} do
+      write_file(Path.join(dir, "x/1.data"), 5)
+      write_file(Path.join(dir, "y/1.data"), 5)
+
+      {total, buckets} = Disk.recursive_usage_grouped(dir, [], 0)
+      assert total == 10
+      assert buckets == %{Path.basename(dir) => 10}
+    end
+
+    test "excluded files contribute to neither total nor buckets", %{tmp_dir: dir} do
+      write_file(Path.join(dir, "stack/aa/bb/shape-a/1.data"), 100)
+      excluded = Path.join(dir, "stack/aa/bb/shape-a/skip.data")
+      write_file(excluded, 999)
+
+      {total, buckets} = Disk.recursive_usage_grouped(dir, [excluded], 4)
+      assert total == 100
+      assert buckets == %{"shape-a" => 100}
+    end
+
+    test "unreadable / missing directory is skipped and contributes 0", %{tmp_dir: dir} do
+      write_file(Path.join(dir, "stack/aa/bb/shape-a/1.data"), 100)
+      missing = Path.join(dir, "stack/aa/bb/does-not-exist")
+
+      # A nonexistent path stat fails -> contributes 0, no crash.
+      {total, buckets} = Disk.recursive_usage_grouped(missing, [], 4)
+      assert total == 0
+      assert buckets == %{}
+
+      # And the rest of a valid tree is still counted.
+      {total, buckets} = Disk.recursive_usage_grouped(dir, [], 4)
+      assert total == 100
+      assert buckets == %{"shape-a" => 100}
+    end
+
+    test "total stays byte-identical to recursive_usage/2 when a dir is unreadable",
+         %{tmp_dir: dir} do
+      write_file(Path.join(dir, "stack/aa/bb/shape-a/1.data"), 100)
+      bad = Path.join(dir, "stack/aa/bb/shape-b")
+      write_file(Path.join(bad, "1.data"), 50)
+      File.chmod!(bad, 0o000)
+
+      on_exit(fn -> File.chmod(bad, 0o755) end)
+
+      legacy = Disk.recursive_usage(dir, [])
+      {grouped, _buckets} = Disk.recursive_usage_grouped(dir, [], 4)
+
+      # The grouped total must match the legacy total bit-for-bit, including the
+      # legacy reset-to-0 behaviour on an unreadable directory.
+      assert grouped == legacy
+    end
+
+    test "directory with no files at the bucket depth yields no bucket", %{tmp_dir: dir} do
+      File.mkdir_p!(Path.join(dir, "stack/aa/bb/empty-shape"))
+
+      {total, buckets} = Disk.recursive_usage_grouped(dir, [], 4)
+      assert total == 0
+      assert buckets == %{}
+    end
+  end
+end
diff --git a/packages/electric-telemetry/test/electric/telemetry/disk_usage_test.exs b/packages/electric-telemetry/test/electric/telemetry/disk_usage_test.exs
index c712342bf4..89c81ba0b9 100644
--- a/packages/electric-telemetry/test/electric/telemetry/disk_usage_test.exs
+++ b/packages/electric-telemetry/test/electric/telemetry/disk_usage_test.exs
@@ -70,6 +70,112 @@ defmodule ElectricTelemetry.DiskUsageTest do
     :ok = DiskUsage.update(ctx.usage)
   end
 
+  describe "per-directory grouping (top-N)" do
+    # Mirror the real electric shape storage layout, where DiskUsage walks the
+    # per-stack `shapes` root and a shape lives at
+    # `<walk_root>/<stack_id>/<p1>/<p2>/<shape_handle>` (the two-level shard is
+    # the first two char-pairs of the handle; see
+    # Electric.ShapeCache.PureFileStorage.shape_data_dir/3). The shape handle
+    # therefore sits at depth 4 below the walk root.
+    @shape_dir_depth 4
+
+    defp make_shape(walk_root, stack_id, handle, bytes) do
+      <<p1::binary-2, p2::binary-2, _::binary>> = handle
+      base = Path.join([walk_root, stack_id, p1, p2, handle])
+      File.mkdir_p!(base)
+      File.write!(Path.join(base, "1.data"), :binary.copy("0", bytes), [:raw, :binary])
+      bytes
+    end
+
+    defp make_shape(ctx, handle, bytes) when is_map(ctx) do
+      make_shape(ctx.tmp_dir, ctx.stack_id, handle, bytes)
+    end
+
+    @tag start_usage: false
+    test "emits only the top-N largest shapes", ctx do
+      # 5 shapes with distinct handles and sizes, ask for top 3.
+      make_shape(ctx, "aaaa1111", 100)
+      make_shape(ctx, "bbbb2222", 500)
+      make_shape(ctx, "cccc3333", 300)
+      make_shape(ctx, "dddd4444", 50)
+      make_shape(ctx, "eeee5555", 400)
+
+      ctx = start_usage_grouped(ctx, group_depth: @shape_dir_depth, top_n: 3)
+      :ok = DiskUsage.update(ctx.usage)
+
+      assert {:ok, 1350, _} = DiskUsage.current(ctx.stack_id)
+
+      assert {:ok, top} = DiskUsage.current_dirs(ctx.stack_id)
+      # Sorted desc by size, only the 3 largest, smaller ones dropped.
+      assert top == [{"bbbb2222", 500}, {"eeee5555", 400}, {"cccc3333", 300}]
+      handles = Enum.map(top, &elem(&1, 0))
+      refute "aaaa1111" in handles
+      refute "dddd4444" in handles
+    end
+
+    @tag start_usage: false
+    test "tag values are the shape_handle dir names", ctx do
+      make_shape(ctx, "abcd1234handle", 10)
+
+      ctx = start_usage_grouped(ctx, group_depth: @shape_dir_depth, top_n: 10)
+      :ok = DiskUsage.update(ctx.usage)
+
+      assert {:ok, [{"abcd1234handle", 10}]} = DiskUsage.current_dirs(ctx.stack_id)
+    end
+
+    @tag start_usage: false
+    test "wrong (too shallow) group_depth buckets by hash-prefix, not handle", ctx do
+      # Regression guard for the off-by-one bug: bucketing one level too shallow
+      # keys by the 2-char shard prefix `<p2>`, aggregating many shapes into one
+      # bucket instead of reporting per shape. Two shapes that share a shard
+      # prefix must collapse together at the wrong depth.
+      make_shape(ctx, "zzaa1111", 10)
+      make_shape(ctx, "zzaa2222", 20)
+
+      # @shape_dir_depth - 1 == bucket by <p2> ("aa"), which both shapes share.
+      ctx = start_usage_grouped(ctx, group_depth: @shape_dir_depth - 1, top_n: 10)
+      :ok = DiskUsage.update(ctx.usage)
+
+      assert {:ok, [{"aa", 30}]} = DiskUsage.current_dirs(ctx.stack_id)
+    end
+
+    @tag start_usage: false
+    test "grouping disabled by default leaves current_dirs pending", ctx do
+      make_shape(ctx, "aaaa1111", 10)
+      ctx = start_usage(ctx)
+      :ok = DiskUsage.update(ctx.usage)
+
+      assert {:ok, 10, _} = DiskUsage.current(ctx.stack_id)
+      assert :pending = DiskUsage.current_dirs(ctx.stack_id)
+    end
+
+    @tag start_usage: false
+    test "total stays correct alongside grouping", ctx do
+      total =
+        make_shape(ctx, "aaaa1111", 100) +
+          make_shape(ctx, "bbbb2222", 250)
+
+      ctx = start_usage_grouped(ctx, group_depth: @shape_dir_depth)
+      :ok = DiskUsage.update(ctx.usage)
+
+      assert {:ok, ^total, _} = DiskUsage.current(ctx.stack_id)
+    end
+  end
+
+  defp start_usage_grouped(ctx, opts) do
+    {:ok, usage_pid} =
+      DiskUsage.start_link(
+        [
+          stack_id: ctx.stack_id,
+          storage_dir: ctx.tmp_dir,
+          manual_refresh: true,
+          update_period: 1_000
+        ] ++ opts
+      )
+
+    Map.put(ctx, :usage, usage_pid)
+  end
+
   defp stop_usage(ctx) do
     ref = Process.monitor(ctx.usage)
     Process.unlink(ctx.usage)
diff --git a/packages/electric-telemetry/test/electric/telemetry/system_metrics/cgroup_test.exs b/packages/electric-telemetry/test/electric/telemetry/system_metrics/cgroup_test.exs
new file mode 100644
index 0000000000..344697822a
--- /dev/null
+++ b/packages/electric-telemetry/test/electric/telemetry/system_metrics/cgroup_test.exs
@@ -0,0 +1,269 @@
+defmodule ElectricTelemetry.SystemMetrics.CgroupTest do
+  use ExUnit.Case, async: true
+
+  @moduletag :tmp_dir
+
+  alias ElectricTelemetry.SystemMetrics.Cgroup
+
+  # Attach handlers for every cgroup.* event and collect emitted measurements
+  # into a flat map of metric-name (dotted string) => value. Single-key
+  # measurement maps make this unambiguous.
+  defp collect_cgroup_metrics(fun) do
+    test_pid = self()
+    ref = make_ref()
+
+    events = [
+      [:cgroup, :memory],
+      [:cgroup, :cpu],
+      [:cgroup, :io],
+      [:cgroup, :memory, :pressure, :full],
+      [:cgroup, :cpu, :pressure, :full]
+    ]
+
+    handler_id = {__MODULE__, ref}
+
+    :telemetry.attach_many(
+      handler_id,
+      events,
+      fn event, measurements, _meta, _ ->
+        send(test_pid, {ref, event, measurements})
+      end,
+      nil
+    )
+
+    try do
+      fun.()
+    after
+      :telemetry.detach(handler_id)
+    end
+
+    drain(ref, %{})
+  end
+
+  defp drain(ref, acc) do
+    receive do
+      {^ref, event, measurements} ->
+        name = event |> Enum.map(&Atom.to_string/1) |> Enum.join(".")
+
+        acc =
+          Enum.reduce(measurements, acc, fn {k, v}, acc ->
+            Map.put(acc, "#{name}.#{k}", v)
+          end)
+
+        drain(ref, acc)
+    after
+      0 -> acc
+    end
+  end
+
+  defp write_file(root, rel, content) do
+    path = Path.join(root, rel)
+    File.mkdir_p!(Path.dirname(path))
+    File.write!(path, content)
+  end
+
+  describe "cgroup v2" do
+    setup %{tmp_dir: tmp_dir} do
+      root = Path.join(tmp_dir, "v2")
+
+      write_file(root, "memory.current", "1048576\n")
+
+      write_file(root, "memory.stat", """
+      anon 524288
+      file 262144
+      inactive_file 131072
+      slab 4096
+      """)
+
+      write_file(root, "memory.max", "max\n")
+
+      write_file(root, "memory.pressure", """
+      some avg10=1.50 avg60=0.30 avg300=0.10 total=12345
+      full avg10=0.75 avg60=0.20 avg300=0.05 total=6789
+      """)
+
+      write_file(root, "cpu.stat", """
+      usage_usec 9000000
+      user_usec 6000000
+      system_usec 3000000
+      nr_periods 100
+      nr_throttled 7
+      throttled_usec 250000
+      """)
+
+      write_file(root, "cpu.pressure", """
+      some avg10=2.00 avg60=0.50 avg300=0.10 total=99999
+      full avg10=1.25 avg60=0.40 avg300=0.08 total=55555
+      """)
+
+      write_file(root, "io.stat", """
+      259:0 rbytes=1000 wbytes=2000 rios=10 wios=20
+      259:1 rbytes=500 wbytes=1500 rios=5 wios=15
+      """)
+
+      %{root: root}
+    end
+
+    test "emits parsed memory metrics", %{root: root} do
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      assert m["cgroup.memory.current"] == 1_048_576
+      assert m["cgroup.memory.anon"] == 524_288
+      assert m["cgroup.memory.file"] == 262_144
+      # working_set = current - inactive_file = 1048576 - 131072
+      assert m["cgroup.memory.working_set"] == 1_048_576 - 131_072
+    end
+
+    test "skips memory.max when value is \"max\"", %{root: root} do
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      refute Map.has_key?(m, "cgroup.memory.max")
+    end
+
+    test "emits a real numeric memory.max", %{root: root} do
+      write_file(root, "memory.max", "2147483648\n")
+
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      assert m["cgroup.memory.max"] == 2_147_483_648
+    end
+
+    test "parses PSI full avg10 for memory and cpu", %{root: root} do
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      assert m["cgroup.memory.pressure.full.avg10"] == 0.75
+      assert m["cgroup.cpu.pressure.full.avg10"] == 1.25
+    end
+
+    test "emits cpu usage / throttling in microseconds", %{root: root} do
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      assert m["cgroup.cpu.usage_usec"] == 9_000_000
+      assert m["cgroup.cpu.nr_throttled"] == 7
+      assert m["cgroup.cpu.throttled_usec"] == 250_000
+    end
+
+    test "sums io rbytes/wbytes across devices", %{root: root} do
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      assert m["cgroup.io.rbytes"] == 1500
+      assert m["cgroup.io.wbytes"] == 3500
+    end
+
+    test "skips working_set when inactive_file is absent", %{root: root} do
+      write_file(root, "memory.stat", "anon 524288\nfile 262144\n")
+
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      refute Map.has_key?(m, "cgroup.memory.working_set")
+      # other metrics still emitted
+      assert m["cgroup.memory.anon"] == 524_288
+    end
+
+    test "skips PSI when pressure files are absent (kernel without PSI)", %{root: root} do
+      File.rm!(Path.join(root, "memory.pressure"))
+      File.rm!(Path.join(root, "cpu.pressure"))
+
+      m =
+        collect_cgroup_metrics(fn ->
+          Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      refute Map.has_key?(m, "cgroup.memory.pressure.full.avg10")
+      refute Map.has_key?(m, "cgroup.cpu.pressure.full.avg10")
+      # rest still works
+      assert m["cgroup.memory.current"] == 1_048_576
+    end
+  end
+
+  describe ":none / non-Linux" do
+    test "no-ops and emits nothing", %{tmp_dir: tmp_dir} do
+      m =
+        collect_cgroup_metrics(fn ->
+          assert :ok = Cgroup.measurement(%{}, cgroup_version: :none, cgroup_root: tmp_dir)
+        end)
+
+      assert m == %{}
+    end
+  end
+
+  describe "malformed / missing files" do
+    test "skips missing files without crashing, still emits available metrics", %{
+      tmp_dir: tmp_dir
+    } do
+      root = Path.join(tmp_dir, "partial")
+      # Only a memory.current and a malformed memory.stat; everything else absent.
+      write_file(root, "memory.current", "4096\n")
+      write_file(root, "memory.stat", "this is not valid\nanon notanumber\nfile 8192\n")
+
+      m =
+        collect_cgroup_metrics(fn ->
+          assert :ok = Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      assert m["cgroup.memory.current"] == 4096
+      # malformed/absent keys are skipped
+      refute Map.has_key?(m, "cgroup.memory.anon")
+      assert m["cgroup.memory.file"] == 8192
+      refute Map.has_key?(m, "cgroup.memory.working_set")
+      # entirely-absent planes simply emit nothing
+      refute Map.has_key?(m, "cgroup.cpu.usage_usec")
+      refute Map.has_key?(m, "cgroup.io.rbytes")
+    end
+
+    test "empty io.stat emits nothing for io", %{tmp_dir: tmp_dir} do
+      root = Path.join(tmp_dir, "emptyio")
+      write_file(root, "io.stat", "")
+
+      m =
+        collect_cgroup_metrics(fn ->
+          assert :ok = Cgroup.measurement(%{}, cgroup_version: :v2, cgroup_root: root)
+        end)
+
+      refute Map.has_key?(m, "cgroup.io.rbytes")
+      refute Map.has_key?(m, "cgroup.io.wbytes")
+    end
+  end
+
+  describe "metric definitions" do
+    test "cgroup.* metrics are present in ApplicationTelemetry.metrics/1" do
+      names =
+        ElectricTelemetry.ApplicationTelemetry.metrics(%{})
+        |> Enum.map(& &1.name)
+
+      assert [:cgroup, :memory, :current] in names
+      assert [:cgroup, :memory, :anon] in names
+      assert [:cgroup, :memory, :file] in names
+      assert [:cgroup, :memory, :working_set] in names
+      assert [:cgroup, :memory, :max] in names
+      assert [:cgroup, :memory, :pressure, :full, :avg10] in names
+      assert [:cgroup, :cpu, :usage_usec] in names
+      assert [:cgroup, :cpu, :nr_throttled] in names
+      assert [:cgroup, :cpu, :throttled_usec] in names
+      assert [:cgroup, :cpu, :pressure, :full, :avg10] in names
+      assert [:cgroup, :io, :rbytes] in names
+      assert [:cgroup, :io, :wbytes] in names
+    end
+  end
+end
diff --git a/packages/electric-telemetry/test/electric/telemetry/system_metrics/proc_test.exs b/packages/electric-telemetry/test/electric/telemetry/system_metrics/proc_test.exs
new file mode 100644
index 0000000000..f0ec394cb9
--- /dev/null
+++ b/packages/electric-telemetry/test/electric/telemetry/system_metrics/proc_test.exs
@@ -0,0 +1,231 @@
+defmodule ElectricTelemetry.SystemMetrics.ProcTest do
+  use ExUnit.Case, async: true
+
+  @moduletag :tmp_dir
+
+  alias ElectricTelemetry.SystemMetrics.Proc
+
+  @pid "4242"
+
+  # Attach handlers for every host.* event and collect emitted measurements
+  # into a flat map of metric-name (dotted string) => value. Single-key
+  # measurement maps make this unambiguous.
+  defp collect_host_metrics(fun) do
+    test_pid = self()
+    ref = make_ref()
+
+    events = [
+      [:host, :proc, :beam],
+      [:host, :proc, :beam, :io]
+    ]
+
+    handler_id = {__MODULE__, ref}
+
+    :telemetry.attach_many(
+      handler_id,
+      events,
+      fn event, measurements, _meta, _ ->
+        send(test_pid, {ref, event, measurements})
+      end,
+      nil
+    )
+
+    try do
+      fun.()
+    after
+      :telemetry.detach(handler_id)
+    end
+
+    drain(ref, %{})
+  end
+
+  defp drain(ref, acc) do
+    receive do
+      {^ref, event, measurements} ->
+        name = event |> Enum.map(&Atom.to_string/1) |> Enum.join(".")
+
+        acc =
+          Enum.reduce(measurements, acc, fn {k, v}, acc ->
+            Map.put(acc, "#{name}.#{k}", v)
+          end)
+
+        drain(ref, acc)
+    after
+      0 -> acc
+    end
+  end
+
+  defp write_file(root, rel, content) do
+    path = Path.join(root, rel)
+    File.mkdir_p!(Path.dirname(path))
+    File.write!(path, content)
+  end
+
+  defp status do
+    """
+    Name:\tbeam.smp
+    State:\tS (sleeping)
+    Pid:\t4242
+    VmPeak:\t 2000000 kB
+    VmRSS:\t  500000 kB
+    RssAnon:\t  300000 kB
+    RssFile:\t  150000 kB
+    RssShmem:\t  50000 kB
+    Threads:\t42
+    """
+  end
+
+  defp proc_io do
+    """
+    rchar: 123456789
+    wchar: 987654321
+    syscr: 1000
+    syscw: 2000
+    read_bytes: 65536
+    write_bytes: 131072
+    cancelled_write_bytes: 0
+    """
+  end
+
+  describe "linux /proc" do
+    setup %{tmp_dir: tmp_dir} do
+      root = Path.join(tmp_dir, "proc")
+
+      write_file(root, "#{@pid}/status", status())
+      write_file(root, "#{@pid}/io", proc_io())
+
+      %{root: root}
+    end
+
+    test "emits BEAM Rss*/VmRSS in bytes (kB * 1024)", %{root: root} do
+      m =
+        collect_host_metrics(fn ->
+          Proc.measurement(%{}, os: {:unix, :linux}, proc_root: root, pid: @pid)
+        end)
+
+      assert m["host.proc.beam.rss_anon"] == 300_000 * 1024
+      assert m["host.proc.beam.rss_file"] == 150_000 * 1024
+      assert m["host.proc.beam.rss_shmem"] == 50_000 * 1024
+      assert m["host.proc.beam.vm_rss"] == 500_000 * 1024
+    end
+
+    test "emits BEAM io read/write in bytes (no scaling)", %{root: root} do
+      m =
+        collect_host_metrics(fn ->
+          Proc.measurement(%{}, os: {:unix, :linux}, proc_root: root, pid: @pid)
+        end)
+
+      assert m["host.proc.beam.io.read_bytes"] == 65_536
+      assert m["host.proc.beam.io.write_bytes"] == 131_072
+    end
+
+    test "io read_bytes/write_bytes are not multiplied by 1024", %{root: root} do
+      m =
+        collect_host_metrics(fn ->
+          Proc.measurement(%{}, os: {:unix, :linux}, proc_root: root, pid: @pid)
+        end)
+
+      refute m["host.proc.beam.io.read_bytes"] == 65_536 * 1024
+    end
+  end
+
+  describe "non-linux guard" do
+    test "no-ops and emits nothing on macOS", %{tmp_dir: tmp_dir} do
+      root = Path.join(tmp_dir, "proc")
+      write_file(root, "#{@pid}/status", status())
+      write_file(root, "#{@pid}/io", proc_io())
+
+      m =
+        collect_host_metrics(fn ->
+          assert :ok =
+                   Proc.measurement(%{}, os: {:unix, :darwin}, proc_root: root, pid: @pid)
+        end)
+
+      assert m == %{}
+    end
+  end
+
+  describe "missing / malformed files" do
+    test "missing status file skips beam.rss_* but other metrics still emit", %{tmp_dir: tmp_dir} do
+      root = Path.join(tmp_dir, "proc")
+      write_file(root, "#{@pid}/io", proc_io())
+
+      m =
+        collect_host_metrics(fn ->
+          assert :ok = Proc.measurement(%{}, os: {:unix, :linux}, proc_root: root, pid: @pid)
+        end)
+
+      refute Map.has_key?(m, "host.proc.beam.rss_anon")
+      refute Map.has_key?(m, "host.proc.beam.vm_rss")
+      assert m["host.proc.beam.io.read_bytes"] == 65_536
+    end
+
+    test "missing/unreadable io file skips io metrics but memory metrics still emit", %{
+      tmp_dir: tmp_dir
+    } do
+      root = Path.join(tmp_dir, "proc")
+      write_file(root, "#{@pid}/status", status())
+      # no io file written
+
+      m =
+        collect_host_metrics(fn ->
+          assert :ok = Proc.measurement(%{}, os: {:unix, :linux}, proc_root: root, pid: @pid)
+        end)
+
+      refute Map.has_key?(m, "host.proc.beam.io.read_bytes")
+      refute Map.has_key?(m, "host.proc.beam.io.write_bytes")
+      assert m["host.proc.beam.rss_anon"] == 300_000 * 1024
+    end
+
+    test "malformed value skips that metric, others still emitted", %{tmp_dir: tmp_dir} do
+      root = Path.join(tmp_dir, "proc")
+
+      write_file(root, "#{@pid}/status", """
+      VmRSS:\t  garbage kB
+      RssAnon:\t  300000 kB
+      """)
+
+      write_file(root, "#{@pid}/io", """
+      read_bytes: nope
+      write_bytes: 131072
+      """)
+
+      m =
+        collect_host_metrics(fn ->
+          assert :ok = Proc.measurement(%{}, os: {:unix, :linux}, proc_root: root, pid: @pid)
+        end)
+
+      refute Map.has_key?(m, "host.proc.beam.vm_rss")
+      assert m["host.proc.beam.rss_anon"] == 300_000 * 1024
+
+      refute Map.has_key?(m, "host.proc.beam.io.read_bytes")
+      assert m["host.proc.beam.io.write_bytes"] == 131_072
+    end
+
+    test "all files missing emits nothing, no crash", %{tmp_dir: tmp_dir} do
+      root = Path.join(tmp_dir, "empty")
+
+      m =
+        collect_host_metrics(fn ->
+          assert :ok = Proc.measurement(%{}, os: {:unix, :linux}, proc_root: root, pid: @pid)
+        end)
+
+      assert m == %{}
+    end
+  end
+
+  describe "metric definitions" do
+    test "host.* metrics are present in ApplicationTelemetry.metrics/1" do
+      names =
+        ElectricTelemetry.ApplicationTelemetry.metrics(%{})
+        |> Enum.map(& &1.name)
+
+      assert [:host, :proc, :beam, :rss_anon] in names
+      assert [:host, :proc, :beam, :rss_file] in names
+      assert [:host, :proc, :beam, :rss_shmem] in names
+      assert [:host, :proc, :beam, :vm_rss] in names
+      assert [:host, :proc, :beam, :io, :read_bytes] in names
+      assert [:host, :proc, :beam, :io, :write_bytes] in names
+    end
+  end
+end
diff --git a/packages/electric-telemetry/test/electric/telemetry/system_metrics_test.exs b/packages/electric-telemetry/test/electric/telemetry/system_metrics_test.exs
new file mode 100644
index 0000000000..d2553ce162
--- /dev/null
+++ b/packages/electric-telemetry/test/electric/telemetry/system_metrics_test.exs
@@ -0,0 +1,119 @@
+defmodule ElectricTelemetry.SystemMetricsTest do
+  use ExUnit.Case, async: true
+
+  alias ElectricTelemetry.SystemMetrics
+
+  describe "system_info/0" do
+    test "returns a well-formed map" do
+      info = SystemMetrics.system_info()
+
+      assert %{os: os, cgroup_version: cgroup_version} = info
+      assert is_tuple(os)
+      assert cgroup_version in [:v1, :v2, :none]
+    end
+
+    test "is stable across calls (cached)" do
+      assert SystemMetrics.system_info() == SystemMetrics.system_info()
+    end
+
+    test "reports the correct cgroup version for the current platform" do
+      info = SystemMetrics.system_info()
+
+      case :os.type() do
+        {:unix, :linux} ->
+          # On Linux CI/dev this is typically v2, but we don't hard-code it;
+          # we only assert it's a real cgroup detection result.
+          assert info.cgroup_version in [:v1, :v2, :none]
+
+        _ ->
+          assert info.cgroup_version == :none
+      end
+    end
+  end
+
+  describe "recon_alloc_measurement/1" do
+    test "emits vm.alloc.* measurements with sane numeric values" do
+      ref = make_ref()
+      handler_id = {__MODULE__, ref}
+
+      :telemetry.attach(
+        handler_id,
+        [:vm, :alloc],
+        fn _event, measurements, _meta, pid -> send(pid, {ref, measurements}) end,
+        self()
+      )
+
+      on_exit(fn -> :telemetry.detach(handler_id) end)
+
+      assert :ok = SystemMetrics.recon_alloc_measurement(%{})
+
+      assert_received {^ref, measurements}
+
+      assert is_integer(measurements.allocated)
+      assert is_integer(measurements.used)
+      assert is_integer(measurements.unused)
+      assert is_number(measurements.carrier_usage)
+
+      assert measurements.allocated >= measurements.used
+      assert measurements.unused == measurements.allocated - measurements.used
+      assert measurements.unused >= 0
+      # carrier_usage is recon_alloc's used/allocated ratio — nominally 0..1, but an
+      # instantaneous live-VM sample can momentarily exceed 1.0 (observed ~1.003) because
+      # the underlying mbcs/sbcs stats are sampled non-atomically. Use a tolerant sanity
+      # ceiling rather than a hard 1.0 to avoid a flaky test.
+      assert measurements.carrier_usage >= 0.0
+      assert measurements.carrier_usage <= 1.5
+    end
+  end
+
+  describe "allocator_fragmentation_measurement/0" do
+    test "emits unused bytes per allocator type" do
+      ref = make_ref()
+      handler_id = {__MODULE__, ref}
+
+      events = :ets.new(:events, [:public, :duplicate_bag])
+
+      :telemetry.attach(
+        handler_id,
+        [:vm, :alloc, :fragmentation],
+        fn _event, measurements, meta, _ ->
+          :ets.insert(events, {meta.allocator, measurements.unused})
+        end,
+        nil
+      )
+
+      on_exit(fn -> :telemetry.detach(handler_id) end)
+
+      assert :ok = SystemMetrics.allocator_fragmentation_measurement()
+
+      results = :ets.tab2list(events)
+      assert results != []
+
+      # One event per distinct allocator *type*, not per scheduler instance.
+      assert length(results) == length(Enum.uniq_by(results, fn {allocator, _} -> allocator end))
+
+      for {allocator, unused} <- results do
+        assert is_binary(allocator)
+        assert is_integer(unused)
+        assert unused >= 0
+      end
+    end
+  end
+
+  describe "metric definitions" do
+    test "vm.alloc.* metrics are present in ApplicationTelemetry.metrics/1" do
+      metrics = ElectricTelemetry.ApplicationTelemetry.metrics(%{})
+      names = Enum.map(metrics, & &1.name)
+
+      assert [:vm, :alloc, :allocated] in names
+      assert [:vm, :alloc, :unused] in names
+      assert [:vm, :alloc, :carrier_usage] in names
+
+      frag =
+        Enum.find(metrics, &(&1.name == [:vm, :alloc, :fragmentation, :unused]))
+
+      assert frag, "expected per-allocator fragmentation metric to be defined"
+      assert :allocator in frag.tags
+    end
+  end
+end
diff --git a/packages/sync-service/lib/electric/stack_supervisor/telemetry.ex b/packages/sync-service/lib/electric/stack_supervisor/telemetry.ex
index ea35dfe3f3..daf19b6cdc 100644
--- a/packages/sync-service/lib/electric/stack_supervisor/telemetry.ex
+++ b/packages/sync-service/lib/electric/stack_supervisor/telemetry.ex
@@ -161,6 +161,27 @@ defmodule Electric.StackSupervisor.Telemetry do
         :pending ->
           :ok
       end
+
+      report_per_shape_disk_usage(stack_id)
+    end
+
+    # Emit `electric.storage.dir.bytes` for the top-N largest shapes only (the
+    # disk walk bounds cardinality by keeping just the top-N buckets). Each
+    # series is tagged with the shape handle.
+    defp report_per_shape_disk_usage(stack_id) do
+      case ElectricTelemetry.DiskUsage.current_dirs(stack_id) do
+        {:ok, top_dirs} ->
+          Enum.each(top_dirs, fn {shape_handle, bytes} ->
+            Electric.Telemetry.OpenTelemetry.execute(
+              [:electric, :storage, :dir],
+              %{bytes: bytes},
+              %{stack_id: stack_id, shape: shape_handle}
+            )
+          end)
+
+        :pending ->
+          :ok
+      end
     end
   else
     def report_disk_usage(_stack_id, _telemetry_opts) do
@@ -200,6 +221,7 @@ defmodule Electric.StackSupervisor.Telemetry do
         config.telemetry_opts
         |> Keyword.put(:stack_id, config.stack_id)
         |> Keyword.put(:storage_dir, config.storage_dir)
+        |> Keyword.put(:disk_usage_group_depth, shape_dir_group_depth(config))
         |> Keyword.put(:otel_opts, otel_opts)
         # Always enable default periodic measurements in addition to the user-provided ones
         |> Keyword.update(
@@ -218,6 +240,48 @@ defmodule Electric.StackSupervisor.Telemetry do
       {ElectricTelemetry.StackTelemetry, telemetry_opts}
     end
 
+    @doc """
+    Computes the directory depth, relative to the disk-usage walk root
+    (`config.storage_dir`), at which a single shape's directory sits. Used to
+    bucket the per-shape `electric.storage.dir.bytes` metric by shape handle.
+
+    The walk root and the shape storage root differ between deployments:
+
+      * standalone sync-service walks the bare `ELECTRIC_STORAGE_DIR` while
+        PureFileStorage is configured with `<storage_dir>/shapes`, so a shape
+        lives at `<root>/shapes/<stack_id>/<p1>/<p2>/<shape_handle>` (depth 5);
+      * cloud (multi-tenant) walks the per-tenant `<root>/<tenant_id>/shapes`
+        directly, so a shape lives at `<root>/<p1>/<p2>/<shape_handle>` relative
+        to that walk root (depth 4).
+
+    Rather than hard-code either number, the depth is derived from the actual
+    path that `Electric.ShapeCache.PureFileStorage.shape_data_dir/3` produces for
+    a sample shape handle, so it stays correct if the storage sharding scheme
+    changes. Returns `nil` (grouping disabled) for storage backends without an
+    on-disk shape layout.
+    """
+    def shape_dir_group_depth(config) do
+      with {Electric.ShapeCache.PureFileStorage, storage_opts} <- config.storage,
+           shape_storage_dir when is_binary(shape_storage_dir) <-
+             Keyword.get(storage_opts, :storage_dir) do
+        base_path = Path.join(shape_storage_dir, config.stack_id)
+        # Use a syntactically valid sample handle (>= 4 chars) so the real
+        # path-building code determines the depth, then measure it relative to
+        # the walk root.
+        sample_handle = "00000000-0000"
+        shape_dir = Electric.ShapeCache.PureFileStorage.shape_data_dir(base_path, sample_handle)
+        path_depth(shape_dir) - path_depth(config.storage_dir)
+      else
+        _ -> nil
+      end
+    end
+
+    defp path_depth(path) do
+      path
+      |> Path.split()
+      |> length()
+    end
+
     defp default_metrics_from_periodic_measurements do
       [
         Telemetry.Metrics.last_value("electric.shapes.total_shapes.count"),
diff --git a/packages/sync-service/mix.lock b/packages/sync-service/mix.lock
index 91befd7aa7..524db607b5 100644
--- a/packages/sync-service/mix.lock
+++ b/packages/sync-service/mix.lock
@@ -52,6 +52,7 @@
   "postgrex": {:hex, :postgrex, "0.22.2", "4aec14df2a72722aee92492566edbeeb44e233ecb86b1915d03136297ef1385d", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "8946382ddb06294f56026ac4278b3cc212bac8a2c82ed68b4087819ed1abc53b"},
   "protobuf": {:hex, :protobuf, "0.13.0", "7a9d9aeb039f68a81717eb2efd6928fdf44f03d2c0dfdcedc7b560f5f5aae93d", [:mix], [{:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "21092a223e3c6c144c1a291ab082a7ead32821ba77073b72c68515aa51fef570"},
   "protox": {:hex, :protox, "2.0.9", "5d95631413695639166ab82d3596a016cf2e974b45ce60b87fb926d6e1650977", [:mix], [], "hexpm", "d00d861815351b0fae88532f403f9dd40ade1e52d55728f7151a1f79fa61e9f1"},
+  "recon": {:hex, :recon, "2.5.6", "9052588e83bfedfd9b72e1034532aee2a5369d9d9343b61aeb7fbce761010741", [:mix, :rebar3], [], "hexpm", "96c6799792d735cc0f0fd0f86267e9d351e63339cbe03df9d162010cefc26bb0"},
   "remote_ip": {:hex, :remote_ip, "1.2.0", "fb078e12a44414f4cef5a75963c33008fe169b806572ccd17257c208a7bc760f", [:mix], [{:combine, "~> 0.10", [hex: :combine, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "2ff91de19c48149ce19ed230a81d377186e4412552a597d6a5137373e5877cb7"},
   "repatch": {:hex, :repatch, "1.6.1", "e2bc55d7358de5727a7d989886d9aebd6497ccab4232b41dcae54f8fd6539c19", [:mix], [{:ex2ms, "~> 1.7.0", [hex: :ex2ms, repo: "hexpm", optional: false]}], "hexpm", "486f80b503fad097ae9ff33f3c7ad3a24c2bad85e694d76ff214e15d13cf9d25"},
   "req": {:hex, :req, "0.6.1", "7b904c8b42d0e08136a5c6aba024fd12fc79a1ed8856e7a3522b0917f7e75113", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.21.0 or ~> 0.22.0", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "aaf11c9c80f2df2364630b3594e1857fe610d8ea7cb994e1ce3dcb55f204ff1c"},
diff --git a/packages/sync-service/test/electric/stack_supervisor/telemetry_test.exs b/packages/sync-service/test/electric/stack_supervisor/telemetry_test.exs
index 5c543abf87..b2bc6edad1 100644
--- a/packages/sync-service/test/electric/stack_supervisor/telemetry_test.exs
+++ b/packages/sync-service/test/electric/stack_supervisor/telemetry_test.exs
@@ -130,5 +130,70 @@ if Electric.telemetry_enabled?() and Code.ensure_loaded?(ElectricTelemetry.Repor
         assert scrape =~ ~r/electric_admission_control_reject_count\{kind="shape"\} 2/
       end
     end
+
+    describe "shape_dir_group_depth/1" do
+      alias Electric.ShapeCache.PureFileStorage
+
+      # The depth must land exactly on the shape-handle directory produced by the
+      # real storage code, for both deployment layouts. We derive the expected
+      # depth from PureFileStorage.shape_data_dir/3 rather than hard-coding it, so
+      # config/sharding drift can't silently re-introduce the off-by-one bug.
+      defp expected_depth(walk_root, shape_storage_dir, stack_id) do
+        base_path = Path.join(shape_storage_dir, stack_id)
+        shape_dir = PureFileStorage.shape_data_dir(base_path, "00000000-0000")
+        length(Path.split(shape_dir)) - length(Path.split(walk_root))
+      end
+
+      test "standalone layout: walk root is the bare storage dir, shapes nested under shapes/" do
+        # ELECTRIC_STORAGE_DIR walked directly; PureFileStorage at <root>/shapes.
+        walk_root = "/var/electric"
+        shape_storage_dir = Path.join(walk_root, "shapes")
+        stack_id = "single-stack"
+
+        config = %{
+          stack_id: stack_id,
+          storage_dir: walk_root,
+          storage: {PureFileStorage, storage_dir: shape_storage_dir}
+        }
+
+        depth = Telemetry.shape_dir_group_depth(config)
+
+        # <root>/shapes/<stack_id>/<p1>/<p2>/<handle> => depth 5.
+        assert depth == 5
+        assert depth == expected_depth(walk_root, shape_storage_dir, stack_id)
+        # Guard the exact bug: bucketing at 4 would key by the <p2> shard prefix.
+        refute depth == 4
+      end
+
+      test "cloud layout: walk root is the per-tenant shapes dir" do
+        # Cloud passes the per-tenant <root>/<tenant_id>/shapes as BOTH the walk
+        # root and the PureFileStorage storage_dir.
+        tenant_id = "tenant-abc"
+        walk_root = Path.join(["/var/electric", tenant_id, "shapes"])
+        shape_storage_dir = walk_root
+
+        config = %{
+          stack_id: tenant_id,
+          storage_dir: walk_root,
+          storage: {PureFileStorage, storage_dir: shape_storage_dir}
+        }
+
+        depth = Telemetry.shape_dir_group_depth(config)
+
+        # <walk_root>/<tenant_id>/<p1>/<p2>/<handle> => depth 4.
+        assert depth == 4
+        assert depth == expected_depth(walk_root, shape_storage_dir, tenant_id)
+      end
+
+      test "non-file storage backends disable grouping" do
+        config = %{
+          stack_id: "s",
+          storage_dir: "/var/electric",
+          storage: {Electric.ShapeCache.InMemoryStorage, []}
+        }
+
+        assert Telemetry.shape_dir_group_depth(config) == nil
+      end
+    end
   end
 end