diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java
index 99026c4cd53..a3e92dd596d 100644
--- a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java
+++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java
@@ -1973,22 +1973,57 @@ ORDER BY (workspace_id, project_id, trace_id, id) DESC, last_updated_at DESC
      * project_id}, so no delete is ever project-less - also required once {@code traces} is a Distributed table
      * (OPIK-7455).
      * <p>
-     * {@code <if(partitions)>} adds the table's own weekly partition expression, bound as the set of partitions the
-     * batch's ids resolve to. It is emitted whenever every id in the batch is one whose partition can be derived
-     * exactly ({@link WeeklyPartitions#of}); otherwise the predicate is omitted and the statement is byte-identical to
-     * the previous unbounded form. That is what preserves the original guarantee — a row whose {@code id_at} cannot be
-     * trusted is still deleted, because no id in such a batch is used to derive a partition.
+     * The unbounded form: no partition scoping at all, so it is the correct statement whenever
+     * {@link WeeklyPartitions#groupByPartition} cannot derive an exact partition for every id in the batch (a non-UUIDv7
+     * id, or one past the {@code DateTime64} ceiling — see that method's Javadoc). {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS}
+     * is preferred whenever it can be used; this is the fallback that stays correct on any batch, at the cost of the
+     * mutation visiting every part of the table (OPIK-8230).
      * <p>
-     * No schema flag gates it. {@link WeeklyPartitions} derives a value per {@code id_at} type the mutation may meet
-     * — the legacy 32-bit {@code DateTime} of {@code traces} as well as the {@code DateTime64(0)} of the partitioned
-     * successor — so one rendered statement is correct on both sides of the cutover EXCHANGE, in either direction, with
-     * nothing to flip and nothing to revert on rollback.
+     * The pairs are bound (never inlined) as two positional string arrays and zipped back into {@code (project_id, id)}
+     * tuples with {@code arrayZip}, so the query text is constant regardless of batch size and no value reaches the SQL
+     * as a literal. {@code arrayZip} is a deterministic function, not a subquery - ClickHouse rejects subqueries in
+     * delete mutations. Callers batch to keep each array within the driver's reliable bind size ({@link
+     * com.comet.opik.infrastructure.FilterUtils#ANALYTICS_DELETE_BATCH_SIZE}).
+     */
+    /**
+     * Deletes by the full {@code (workspace_id, project_id, id)} sort key, matching on {@code (project_id, id)} tuples
+     * so a single statement can span several projects (e.g. a reused id resolved to all its owning projects, or a
+     * cross-project batch) instead of one delete per project (OPIK-7483). Every deleted row carries its {@code
+     * project_id}, so no delete is ever project-less - also required once {@code traces} is a Distributed table
+     * (OPIK-7455).
+     * <p>
+     * {@code <if(partition)>} adds {@code IN PARTITION <partition>} (OPIK-8230), which scopes the <b>mutation itself</b>
+     * — which parts it is registered against — rather than which rows its {@code WHERE} selects. That distinction is
+     * the entire fix. A ClickHouse mutation selects parts at the <b>partition</b> stage, before the {@code WHERE}
+     * clause is considered at all — a {@code (workspace_id, project_id, id)} predicate, however exact, prunes zero
+     * parts, only rows within whichever parts were already selected. Post-cutover {@code traces_local} is
+     * weekly-partitioned into ~1,900 partitions / ~3,650 parts; a delete matching a handful of rows was registered
+     * against every one of them regardless, at ~19 ms/part fixed overhead, which is what pushed ordinary deletes past
+     * the {@code max_execution_time} ceiling (`Code: 159 TIMEOUT_EXCEEDED`) that pre-cutover deletes (1.5–2.2 s) never
+     * approached. Omitted for the unbounded fallback ({@link TraceDAOImpl#deleteBatch}), whose statement is then
+     * byte-identical to the pre-OPIK-8230 form.
+     * <p>
+     * One statement per partition the batch's ids resolve to ({@link WeeklyPartitions#groupByPartition}), each binding
+     * only the {@code (project_id, id)} pairs whose id belongs to that partition — {@code IN PARTITION} accepts exactly
+     * one partition per statement; ClickHouse's {@code IN PARTITION p1, p2, …} form is for <em>composite</em> partition
+     * keys and rejects a list of distinct partitions outright. Naming a partition that does not exist is a silent
+     * no-op, not an error, which is what makes it safe for a far-future id to be bound into two statements (see
+     * {@link WeeklyPartitions#groupByPartition}'s Javadoc) — the one naming the partition it is not actually in simply
+     * matches nothing there.
      * <p>
-     * Why it matters: a mutation selects parts at the <b>partition</b> stage, where the (workspace_id, project_id, id)
-     * predicate prunes nothing, so deleting a handful of rows rewrote every part of the table. Measured on prod-test
-     * (271.6 M rows, 3,928 parts): 12 ids rewrote <b>3,928 parts / 5.40 TiB</b>. With this predicate the same batch
-     * selects <b>5</b> parts. An {@code id_at} <em>range</em> is not a substitute: on a batch spanning 1996 and 2200 a
-     * range still selected 2,644 parts, where the exact set selected 4.
+     * {@code <partition>} is interpolated by StringTemplate, exactly like {@code <traces_mutation_table>} and
+     * {@code <log_comment>} above it — {@code IN PARTITION {p:UInt32}} is a ClickHouse syntax error, so the value
+     * cannot be bound. Unlike the {@code <log_comment>} interpolation this carries no escaping question: the value is
+     * always a Java {@code long} produced by {@link WeeklyPartitions}, and a {@code long} cannot carry a quote or any
+     * other SQL-meaningful character.
+     * <p>
+     * Kept as a single {@code DELETE FROM} statement rather than a hand-written {@code ALTER TABLE ... UPDATE ... IN
+     * PARTITION}: both compile to the same mutation, but the hand-written form would step outside
+     * {@link #tracesMutationTable()}'s routing guarantee ({@code TraceMutationRoutingArchTest}), outside the project's
+     * one {@code lightweight_deletes_sync} convention, and — decisively — outside {@code lightweight_delete_mode}: a
+     * {@code DELETE FROM} takes the patch-part path when that setting is enabled and the table's data-model columns
+     * allow it; a hand-written {@code ALTER ... UPDATE} does not; so staying on {@code DELETE FROM} means any future
+     * enabling of patch parts benefits this path for free, with no further code change here.
      * <p>
      * The pairs are bound (never inlined) as two positional string arrays and zipped back into {@code (project_id, id)}
      * tuples with {@code arrayZip}, so the query text is constant regardless of batch size and no value reaches the SQL
@@ -1998,9 +2033,9 @@ ORDER BY (workspace_id, project_id, trace_id, id) DESC, last_updated_at DESC
      */
     private static final String DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS = """
             DELETE FROM <traces_mutation_table>
+            <if(partition)>IN PARTITION <partition><endif>
             WHERE workspace_id = :workspace_id
             AND (project_id, id) IN arrayZip(:project_ids, :trace_ids)
-            <if(partitions)>AND toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1))) IN :partitions<endif>
             SETTINGS log_comment = '<log_comment>'
             ;
             """;
@@ -2328,7 +2363,7 @@ ORDER BY (workspace_id, project_id, id) DESC, last_updated_at DESC
      * {@code :min_id}'s week and inverted the window, so the fast pass matched nothing and every id fell through to
      * the unbounded pass. It now resolves them.
      * <p>
-     * The fallback remains load-bearing for what {@link com.comet.opik.utils.WeeklyPartitions#of} still cannot derive
+     * The fallback remains load-bearing for what {@link com.comet.opik.utils.WeeklyPartitions#groupByPartition} still cannot derive
      * exactly: an id at or past the end of {@code DateTime64}'s range, where {@code id_at} saturates to
      * {@code 2299-12-31} whatever the real week, so every such id collapses into one partition. Real data contains
      * them, so the bounded query is never a delete's sole resolver.
@@ -3649,38 +3684,103 @@ public Mono<Void> delete(Set<Pair<UUID, UUID>> projectIdTraceIdPairs, @NonNull C
 
         return makeMonoContextAware((userName, workspaceId) -> Flux
                 .fromIterable(Lists.partition(List.copyOf(projectIdTraceIdPairs), ANALYTICS_DELETE_BATCH_SIZE))
-                .concatMap(batch -> {
-                    var template = getSTWithLogComment(DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS, "delete_traces",
-                            workspaceId,
-                            userName, "pairs_size=%s".formatted(batch.size()));
-                    selectTracesMutationTable(template);
-
-                    var projectIds = batch.stream().map(pair -> pair.getLeft().toString()).toArray(String[]::new);
-                    var traceIds = batch.stream().map(pair -> pair.getRight().toString()).toArray(String[]::new);
-
-                    // Prune to the batch's own partitions when every id in the batch allows it; otherwise emit the
-                    // unbounded form. Needs no schema flag: WeeklyPartitions derives a value per id_at type the
-                    // mutation may meet, so the set is correct on the legacy traces and on the partitioned successor.
-                    var partitions = WeeklyPartitions.of(batch.stream().map(Pair::getRight).toList());
-                    // Flag only, exactly like distributed_wrap: the values reach ClickHouse via the bind below,
-                    // never through the template, so the rendered SQL is constant regardless of batch contents.
-                    partitions.ifPresent(_ -> template.add("partitions", true));
-
-                    var statement = connection.createStatement(template.render())
-                            .bind("workspace_id", workspaceId)
-                            .bind("project_ids", projectIds)
-                            .bind("trace_ids", traceIds);
+                .concatMap(batch -> deleteBatch(batch, workspaceId, userName, connection))
+                .then());
+    }
 
-                    if (partitions.isPresent()) {
-                        statement = statement.bind("partitions", partitions.get().toArray(Long[]::new));
-                    }
+    /**
+     * Deletes one ANALYTICS_DELETE_BATCH_SIZE-sized batch, scoped to its own partitions when every id in it resolves
+     * exactly (OPIK-8230): one {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS} statement per partition the
+     * batch's ids resolve to ({@link WeeklyPartitions#groupByPartition}), each binding only the pairs whose id belongs
+     * to that partition. Falls back to a single {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS} statement over the whole
+     * batch either when {@code groupByPartition} cannot derive every id's partition exactly, or when the mutation's
+     * own target table is not actually partitioned — always correct, merely unscoped, and identical to what this
+     * method emitted before OPIK-8230.
+     * <p>
+     * The second condition is load-bearing and easy to miss: {@link WeeklyPartitions} is deliberately schema-agnostic
+     * (see its own Javadoc) and derives a partition value regardless of which table {@code id_at} actually lives in —
+     * including the legacy, UNPARTITIONED {@code traces} that {@link #tracesMutationTable()} still targets before the
+     * sharding-readiness wrap is applied. {@code IN PARTITION <value>} against a table with no partition key is not a
+     * no-op; ClickHouse rejects it outright (Code 248 {@code INVALID_PARTITION_VALUE}, "Wrong number of fields in the
+     * partition expression: 1, must be: 0"). So the scoped path is gated on {@link #tracesMutationTable()}'s
+     * OWN resolved name, not on whether a partition value merely COULD be derived: this method calls the
+     * resolver itself, exactly as {@link #selectTracesMutationTable} does for the template - never the wrap
+     * flag directly ({@code TraceMutationRoutingArchTest} enforces that {@link #tracesDistributedWrapEnabled()}
+     * is called only from {@link #tracesMutationTable()}, so no second place can get the read/mutate split
+     * wrong). Resolved once per batch and reused across every partition-scoped statement it emits: the table
+     * a mutation targets does not vary per partition, only per topology.
+     * <p>
+     * The gate is a NAME check — {@code TRACES_LOCAL_TABLE.equals(table)} — not a runtime read of
+     * {@code traces_local}'s actual engine or partition key. That is an accepted operational precondition, not an
+     * oversight: it relies on the wrap being flipped only after the EXCHANGE that makes {@code traces_local} the
+     * partitioned successor has already run, which is the documented ordering the migration follows and the same
+     * assumption {@link #tracesMutationTable()} already makes for every other mutation in this class (a premature
+     * flip breaks inserts and reads the same way, table-missing rather than Code 248). Verifying the live schema
+     * before every batch would trade a real per-request cost for a misordering that procedure, not this code, is
+     * relied on to prevent.
+     * <p>
+     * Sequential ({@code concatMap}), deliberately not concurrent: bounded concurrency was measured and rejected for
+     * this change (OPIK-8230) — a real gain in the rare many-partition tail, but a knob to hold in reserve rather than
+     * ship alongside the fix itself.
+     */
+    private Mono<Void> deleteBatch(List<Pair<UUID, UUID>> batch, String workspaceId, String userName,
+            Connection connection) {
+        // Resolved once, via the resolver itself - never the flag directly (see the method Javadoc for why).
+        var table = tracesMutationTable();
+        var grouped = TRACES_LOCAL_TABLE.equals(table)
+                ? WeeklyPartitions.groupByPartition(batch.stream().map(Pair::getRight).toList())
+                : Optional.<Map<Long, Set<UUID>>>empty();
+
+        if (grouped.isEmpty()) {
+            return executeDelete(table, batch, null, workspaceId, userName, connection);
+        }
 
-                    var segment = startSegment("traces", "Clickhouse", "delete");
-                    return Mono.from(statement.execute())
-                            .doFinally(_ -> endSegment(segment))
-                            .then();
+        // id -> its pairs, built once per batch rather than rescanning the whole batch once per partition: an id can
+        // map to more than one pair (the same trace id reused across projects, OPIK-7483), so this is a
+        // Collectors.groupingBy, not a plain lookup map.
+        var pairsById = batch.stream().collect(Collectors.groupingBy(Pair::getRight));
+
+        return Flux.fromIterable(grouped.get().entrySet())
+                .concatMap(entry -> {
+                    var partitionPairs = entry.getValue().stream()
+                            .flatMap(id -> pairsById.get(id).stream())
+                            .toList();
+                    return executeDelete(table, partitionPairs, entry.getKey(), workspaceId, userName, connection);
                 })
-                .then());
+                .then();
+    }
+
+    /**
+     * Renders and executes one {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS} statement — unbounded when
+     * {@code partition} is null, scoped to it otherwise (the weekly partition value the caller has already resolved
+     * every pair in {@code pairs} to belong to). Shares the bind/segment/logging shape in exactly one place rather
+     * than duplicated per call site in {@link #deleteBatch}.
+     * <p>
+     * Takes {@code table} as a parameter rather than resolving it itself via {@link #selectTracesMutationTable}: it
+     * is already resolved once, up front, by {@link #deleteBatch} — the physical table does not vary per statement
+     * within one batch, only per topology — so re-resolving it here per call would be redundant, not incorrect.
+     */
+    private Mono<Void> executeDelete(String table, List<Pair<UUID, UUID>> pairs, Long partition,
+            String workspaceId, String userName, Connection connection) {
+        var template = getSTWithLogComment(DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS, "delete_traces", workspaceId,
+                userName, "pairs_size=%s".formatted(pairs.size()));
+        template.add("traces_mutation_table", table);
+        if (partition != null) {
+            template.add("partition", partition);
+        }
+
+        var projectIds = pairs.stream().map(pair -> pair.getLeft().toString()).toArray(String[]::new);
+        var traceIds = pairs.stream().map(pair -> pair.getRight().toString()).toArray(String[]::new);
+
+        var statement = connection.createStatement(template.render())
+                .bind("workspace_id", workspaceId)
+                .bind("project_ids", projectIds)
+                .bind("trace_ids", traceIds);
+
+        var segment = startSegment("traces", "Clickhouse", "delete");
+        return Mono.from(statement.execute())
+                .doFinally(_ -> endSegment(segment))
+                .then();
     }
 
     /**
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java
index 3ab8e754f06..b6f83cded45 100644
--- a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java
+++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java
@@ -537,7 +537,7 @@ public Mono<Void> delete(@NonNull Set<UUID> ids, UUID projectId) {
      * Resolves every owning project for each id: a bounded fast pass, then an unbounded pass over only the ids the
      * bounded one leaves unresolved. Returns id -> owning projects; ids absent from the result have no live row.
      * <p>
-     * The bounded pass's week window can miss a row whose week {@link com.comet.opik.utils.WeeklyPartitions#of}
+     * The bounded pass's week window can miss a row whose week {@link com.comet.opik.utils.WeeklyPartitions#groupByPartition}
      * cannot derive exactly — an id at or past the end of {@code DateTime64}'s range, where {@code id_at} saturates
      * to {@code 2299-12-31} whatever the real week — so the unbounded pass re-resolves the miss set and the bounded
      * query is never a delete's sole resolver. A far-future timestamp short of that ceiling is no longer such a case:
@@ -566,6 +566,20 @@ private Mono<Map<UUID, Set<UUID>>> resolveOwningProjects(Set<UUID> ids) {
                 });
     }
 
+    /**
+     * All-or-nothing over {@code projectIdTraceIdPairs}: {@code TracesDeleted} publishes and the deletion-events
+     * bridge captures only {@code .doOnSuccess}/{@code .then} of {@link TraceDAO#delete}, so any error anywhere in
+     * the batch skips both for the WHOLE batch, not just the pairs that did not complete.
+     * <p>
+     * OPIK-8230 widens the window this can bite in, worth being explicit about here rather than only in that
+     * ticket: {@link TraceDAO#delete} can now emit several sequential statements per batch, one per partition
+     * (via {@code deleteBatch}), so an error partway through leaves the EARLIER statements' rows genuinely,
+     * synchronously deleted while the cascade still does not fire for any pair in the batch - not only for the one
+     * that errored. Restructuring this coupling to be per-statement aware is explicitly out of scope for that
+     * ticket ("this ticket removes the condition that trips it; it does not restructure the coupling") - deletes
+     * are idempotent, so a caller that retries converges, and the ticket's own fix (scoping each mutation so it
+     * completes well inside the statement timeout) is what makes hitting this at all much rarer than it is today.
+     */
     private Mono<Void> delete(Set<Pair<UUID, UUID>> projectIdTraceIdPairs, Connection connection) {
         return Mono.deferContextual(ctx -> {
             String workspaceId = ctx.get(RequestContext.WORKSPACE_ID);
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java b/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java
index 06ba82c0bba..0bafa6bd07e 100644
--- a/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java
+++ b/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java
@@ -9,14 +9,19 @@
 import java.time.ZoneOffset;
 import java.time.temporal.TemporalAdjusters;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
+import java.util.stream.Collectors;
 
 /**
- * Single derivation point for the {@code id_at} weekly partition values a batch of ids resolves to, so a mutation can
- * name its own partitions instead of being planned against every part of the table.
+ * Single derivation point for the {@code id_at} weekly partition value(s) a batch of ids resolves to, so a mutation
+ * can name its own partitions instead of being planned against every part of the table. {@link #groupByPartition}
+ * groups ids by the partition each belongs to, for a caller emitting one statement per partition (OPIK-8230) that
+ * must bind only the ids belonging to it.
  *
  * <p>Mirrors the partition expression of {@code traces_local_v2} / {@code spans_local_v2} exactly —
  * {@code PARTITION BY toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))}, where {@code id_at} is
@@ -55,9 +60,9 @@
  * <h2>Why the caller must treat an empty result as "no predicate", never as "no partitions"</h2>
  *
  * <p>The whole value of the derivation is that the partitions a batch resolves to are the <em>only</em> places its rows
- * can be; a set that is merely close is a silently skipped delete, not a slower one. So this returns a set only when
- * every id in the batch is one it can derive exactly, and empty otherwise — leaving the caller to emit its unbounded
- * form, which is always correct and merely slower. The two rejections are:</p>
+ * can be; a grouping that is merely close is a silently skipped delete, not a slower one. So this returns non-empty
+ * only when every id in the batch is one it can derive exactly, and empty otherwise — leaving the caller to emit its
+ * unbounded form, which is always correct and merely slower. The two rejections are:</p>
  * <ul>
  *     <li><b>Any id that is not a UUIDv7.</b> Its high 48 bits are not a timestamp, and {@code UUIDv7ToDateTime}
  *     returns {@code 1970-01-01} for it rather than throwing, so the row sits in the epoch partition while the bits
@@ -101,10 +106,17 @@ public class WeeklyPartitions {
     private static final long ID_AT_LEGACY_MODULUS = 1L << 32;
 
     /**
-     * The weekly partition values the ids resolve to — under each {@code id_at} type the mutation may run against (see
-     * the class javadoc) — or empty if the batch contains an id whose partition cannot be derived exactly. In the empty
-     * case the caller must omit its partition predicate entirely. An empty batch yields empty for the same reason:
-     * there is nothing to bound the mutation to.
+     * Groups the ids by the weekly partition value(s) each resolves to — under each {@code id_at} type the mutation
+     * may run against (see the class javadoc) — or empty if the batch contains an id whose partition cannot be
+     * derived exactly. In the empty case the caller must omit its partition predicate entirely and emit the
+     * unbounded form. An empty batch yields empty for the same reason: there is nothing to bound the mutation to.
+     * {@code IN PARTITION} accepts exactly one partition per statement (OPIK-8230), so a caller emitting one
+     * statement per partition needs to know which ids go in each, not merely their union.
+     * <p>
+     * An id past {@link #ID_AT_LEGACY_MODULUS} appears in <b>two</b> groups — one per {@code id_at} representation
+     * it resolves to. That is safe, not a widening bug: a statement scoped to the group that does not actually
+     * contain the id's row is a no-op there, never a wrong deletion, since the row-matching {@code (project_id, id)}
+     * predicate inside each statement is unchanged regardless of which partition the statement targets.
      * <p>
      * A {@code null} batch throws rather than reading as empty, which is the one place this class is deliberately
      * intolerant. Empty is a <em>documented answer</em> — "this batch cannot be pruned, emit the unbounded form" — and a
@@ -114,36 +126,61 @@ public class WeeklyPartitions {
      *
      * @throws NullPointerException if {@code ids} is null.
      */
-    public static Optional<Set<Long>> of(@NonNull Collection<UUID> ids) {
-        var partitions = new HashSet<Long>();
+    public static Optional<Map<Long, Set<UUID>>> groupByPartition(@NonNull Collection<UUID> ids) {
+        var grouped = new HashMap<Long, Set<UUID>>();
 
         for (UUID id : ids) {
-            if (id == null || id.version() != 7) {
-                return Optional.empty();
-            }
-            // UUIDv7: the high 48 bits are the unix epoch in milliseconds. `>>> 16` reads them unsigned, so the value
-            // is in [0, 2^48) — never negative, and never large enough for Instant.ofEpochSecond to overflow. That is
-            // also why neither derivation checks a floor: the smallest id_at any UUIDv7 can carry is the epoch, and
-            // even its Monday (1969-12-29) is comfortably inside Date32 on both schemas.
-            long epochMilli = id.getMostSignificantBits() >>> 16;
-            if (epochMilli >= ID_AT_CEILING) {
+            var idPartitions = partitionsOf(id);
+            if (idPartitions.isEmpty()) {
                 return Optional.empty();
             }
-            // Truncating to whole seconds is the column's own conversion (DateTime64(0) / DateTime both store seconds)
-            // and cannot move a value into an earlier day, so it never changes the week.
-            long epochSecond = epochMilli / 1_000L;
-            partitions.add(weeklyPartitionOf(epochSecond)); // DateTime64(0, 'UTC') — the partitioned successor
-            // Only past the 32-bit range do the two columns disagree; below it the modulo is the identity, so the
-            // branch is the invariant stated as code rather than an optimisation: an ordinary id CANNOT widen the set.
-            if (epochSecond >= ID_AT_LEGACY_MODULUS) {
-                partitions.add(weeklyPartitionOf(epochSecond % ID_AT_LEGACY_MODULUS)); // DateTime('UTC') — legacy
+            for (Long partition : idPartitions.get()) {
+                grouped.computeIfAbsent(partition, _ -> new HashSet<>()).add(id);
             }
         }
 
-        // Set.copyOf, not the working HashSet: what escapes here decides which partitions a DELETE mutation touches, so
-        // a caller holding a mutable reference could narrow the set after it was derived and turn a correct delete into
-        // a silent no-op. Immutable by default per apps/opik-backend/AGENTS.md, and the accumulator stays local.
-        return partitions.isEmpty() ? Optional.empty() : Optional.of(Set.copyOf(partitions));
+        if (grouped.isEmpty()) {
+            return Optional.empty();
+        }
+        // Immutable outer map AND immutable per-partition sets: what escapes here decides which partitions a DELETE
+        // mutation touches, so a caller holding a mutable reference could narrow it after derivation and turn a
+        // correct delete into a silent no-op. Collectors.toUnmodifiableMap's values must themselves be made
+        // unmodifiable explicitly; it does not do so for you.
+        return Optional.of(grouped.entrySet().stream()
+                .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, entry -> Set.copyOf(entry.getValue()))));
+    }
+
+    /**
+     * The weekly partition value(s) a single id resolves to under each {@code id_at} type the mutation may meet (see
+     * class javadoc) — extracted from {@link #groupByPartition} so the per-id {@code id_at} math reads separately
+     * from the grouping it feeds. Empty exactly when the id cannot be derived exactly: not a UUIDv7, or an embedded
+     * timestamp at or past {@link #ID_AT_CEILING}.
+     */
+    private static Optional<Set<Long>> partitionsOf(UUID id) {
+        if (id == null || id.version() != 7) {
+            return Optional.empty();
+        }
+        // UUIDv7: the high 48 bits are the unix epoch in milliseconds. `>>> 16` reads them unsigned, so the value
+        // is in [0, 2^48) — never negative, and never large enough for Instant.ofEpochSecond to overflow. That is
+        // also why neither derivation checks a floor: the smallest id_at any UUIDv7 can carry is the epoch, and
+        // even its Monday (1969-12-29) is comfortably inside Date32 on both schemas.
+        long epochMilli = id.getMostSignificantBits() >>> 16;
+        if (epochMilli >= ID_AT_CEILING) {
+            return Optional.empty();
+        }
+        // Truncating to whole seconds is the column's own conversion (DateTime64(0) / DateTime both store seconds)
+        // and cannot move a value into an earlier day, so it never changes the week.
+        long epochSecond = epochMilli / 1_000L;
+        // DateTime64(0, 'UTC') — the partitioned successor. Only past the 32-bit range do the two columns disagree;
+        // below it the modulo is the identity, so the branch is the invariant stated as code rather than an
+        // optimisation: an ordinary id CANNOT widen the set. Immutable on the way out, for the same reason
+        // groupByPartition copies its own result — see there. Set.of throws on a duplicate, which is safe here
+        // rather than merely untested: the two arguments are weeks 2^32 seconds (~136 years) apart, so they cannot
+        // be the same week.
+        return Optional.of(epochSecond >= ID_AT_LEGACY_MODULUS
+                ? Set.of(weeklyPartitionOf(epochSecond),
+                        weeklyPartitionOf(epochSecond % ID_AT_LEGACY_MODULUS)) // DateTime('UTC') — legacy
+                : Set.of(weeklyPartitionOf(epochSecond)));
     }
 
     /**
diff --git a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java
index fe67ab37cb5..ccc4dea6b41 100644
--- a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java
+++ b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java
@@ -41,8 +41,6 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.function.Consumer;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 
 import static com.comet.opik.api.resources.utils.AuthTestUtils.mockTargetWorkspace;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -84,40 +82,9 @@ class TracesLegacyTablePruningMutationTest {
     /** A UUIDv7 carrying id_at 2200-01-01 — the litellm shape, and the id the legacy 32-bit column wraps. */
     private static final UUID FAR_FUTURE_ID = UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2");
 
-    /**
-     * The week that id partitions into under a {@code DateTime64} {@code id_at} — its honest one, and the only value the
-     * flag-gated predicate used to carry. Stated as a literal because it is what ClickHouse returned for
-     * {@code toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))} on that id, not something re-derived
-     * here.
-     */
-    private static final long HONEST_WEEK = 21991230L;
-
-    /**
-     * The week the same id partitions into on <b>this</b> table: {@code CAST(UUIDv7ToDateTime(...) AS DateTime('UTC'))}
-     * wraps 2200-01-01 to 2063-11-25, a Wednesday, whose Monday is 2063-11-19. This is the value that makes the delete
-     * below land, and the reason the predicate needs no flag.
-     */
-    private static final long LEGACY_WEEK = 20631119L;
-
     /** {@link UUID#randomUUID()} is a v4 by definition: no embedded timestamp to derive a partition from. */
     private static final UUID NON_V7_ID = UUID.randomUUID();
 
-    /**
-     * The partition-key fragment the DAO's template emits. Only ever <b>compared against</b> the statement read back
-     * from {@code system.query_log} — never spliced into a query this suite runs.
-     */
-    private static final String PARTITION_PREDICATE = "toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))";
-
-    /**
-     * The {@code IN} clause the DAO emitted, captured to end of line: the predicate sits on its own line in the
-     * template with {@code SETTINGS log_comment} on the next, so the line boundary delimits it exactly.
-     */
-    private static final Pattern EMITTED_IN_CLAUSE = Pattern.compile(
-            Pattern.quote(PARTITION_PREDICATE) + "\\s+IN\\s+([^\\n]*)");
-
-    /** A weekly partition value as it appears in SQL — {@code yyyyMMdd}, so always eight digits. */
-    private static final Pattern PARTITION_VALUE = Pattern.compile("\\d{8}");
-
     private static final String INSERT_RAW_TRACE = """
             INSERT INTO traces (workspace_id, project_id, id)
             VALUES (:workspace_id, :project_id, :id)
@@ -213,45 +180,46 @@ void afterAll() {
     }
 
     @Test
-    @DisplayName("a far-future row on the legacy table is deleted by the same pruned statement")
-    void farFutureRowOnLegacyTableIsDeletedByAPrunedStatement() {
-        // The one test that shows the union is load-bearing rather than decorative. With only the honest week in the
-        // set, this delete matches nothing on this table and reports success - which is exactly the failure the
-        // configuration flag used to exist to avoid, now avoided by the predicate itself.
+    @DisplayName("a far-future row on the legacy table is still deleted, by the unbounded form")
+    void farFutureRowOnLegacyTableIsStillDeleted() {
+        // OPIK-8230 changes what this test proves. IN PARTITION cannot be used against a table with no partition key
+        // at all - ClickHouse rejects it outright (Code 248 INVALID_PARTITION_VALUE, "Wrong number of fields in the
+        // partition expression: 1, must be: 0") - so TraceDAO#deleteBatch now gates the scoped path on the mutation's
+        // OWN resolved table, not merely on whether WeeklyPartitions can derive a value. WeeklyPartitions itself stays
+        // schema-agnostic (it still derives BOTH representations for a far-future id, honest and legacy-wrapped, on
+        // purpose - see its Javadoc), but the DAO never spends that derivation here: on the legacy table every delete
+        // takes the unbounded form, correctly, and this is that guarantee restated for the topology it actually
+        // guards.
         //
-        // The project and its id come from the real ingestion path - create a trace through the endpoint, then read the
-        // project id back off it - so the delete runs against a project that exists and a genuine UUIDv7 project id,
-        // not a fabricated one.
+        // The project and its id come from the real ingestion path - create a trace through the endpoint, then read
+        // the project id back off it - so the delete runs against a project that exists and a genuine UUIDv7 project
+        // id, not a fabricated one.
         var projectId = projectIdOf(createTrace());
 
-        // Only the far-future row is raw, because ingestion rejects it by design (24h window). A recent id would prove
-        // nothing here: the legacy id_at is accurate for one, so even an honest-week-only predicate would match it.
+        // Only the far-future row is raw, because ingestion rejects it by design (24h window).
         insertRawTrace(projectId, FAR_FUTURE_ID);
         assertThat(liveRowCount(projectId, FAR_FUTURE_ID)).as("the far-future row is seeded").isEqualTo("1");
 
         delete(Set.of(Pair.of(projectId, FAR_FUTURE_ID)));
 
         assertThat(liveRowCount(projectId, FAR_FUTURE_ID))
-                .as("it is deleted - so the predicate named the week THIS table filed it under, not only the honest one")
+                .as("it is deleted - the unbounded form is always correct, merely unscoped")
                 .isEqualTo("0");
 
         var sql = lastTraceDeleteSql(FAR_FUTURE_ID);
         assertThat(sql)
-                .as("the predicate is emitted here too - there is no schema flag holding it back any more")
-                .contains(PARTITION_PREDICATE);
-        // Asserted as an exact set, both ways round. Only the legacy week can match a row on this table, so a set
-        // missing it would fail the delete above; and a set missing the honest week would pass here while breaking the
-        // post-EXCHANGE suite, which is the pair this has to stay consistent with.
-        assertThat(boundPartitionsOf(sql))
-                .as("both representations of the same id: the honest week and the one the 32-bit column wraps it to")
-                .containsExactlyInAnyOrder(HONEST_WEEK, LEGACY_WEEK);
+                .as("no id_at predicate of any kind, and no IN PARTITION clause: the legacy table has no partition"
+                        + " key for either to name")
+                .doesNotContain("id_at")
+                .doesNotContain("IN PARTITION");
     }
 
     @Test
-    @DisplayName("an ordinary row is deleted by a single-week statement, since both id_at types agree below 2106")
-    void ordinaryRowIsBoundedToOneWeek() {
-        // The counterweight: the union widens only where the two representations differ, so real traffic binds exactly
-        // what it bound before. A regression that added the wrapped week unconditionally would show up here.
+    @DisplayName("an ordinary row is also deleted by the unbounded form, same as a far-future one")
+    void ordinaryRowIsAlsoDeletedUnbounded() {
+        // The counterweight to the far-future case above: on the legacy table there is no "recent id gets the scoped
+        // form, far-future gets the unbounded one" split. Every delete here is unbounded, regardless of era - proving
+        // that ONLY with a far-future row would leave open whether an ordinary one still worked.
         var target = createTrace();
         var projectId = projectIdOf(target);
         assertThat(liveRowCount(projectId, target.id())).as("the row is there").isEqualTo("1");
@@ -260,10 +228,10 @@ void ordinaryRowIsBoundedToOneWeek() {
 
         assertThat(liveRowCount(projectId, target.id())).as("and it is deleted").isEqualTo("0");
         var sql = lastTraceDeleteSql(target.id());
-        assertThat(sql).as("the mutation carries the partition predicate").contains(PARTITION_PREDICATE);
-        assertThat(boundPartitionsOf(sql))
-                .as("one week, not two: a recent id_at is inside the 32-bit range, so the two derivations coincide")
-                .hasSize(1);
+        assertThat(sql)
+                .as("no id_at predicate of any kind, and no IN PARTITION clause")
+                .doesNotContain("id_at")
+                .doesNotContain("IN PARTITION");
     }
 
     @Test
@@ -291,9 +259,9 @@ void nonV7IdDisablesPruning() {
         assertThat(sql)
                 .as("the unbounded form carries no id_at predicate of any kind")
                 .doesNotContain("id_at");
-        assertThat(EMITTED_IN_CLAUSE.matcher(sql).find())
-                .as("and no partition IN clause: %s", sql)
-                .isFalse();
+        assertThat(sql)
+                .as("and no IN PARTITION clause: %s", sql)
+                .doesNotContain("IN PARTITION");
     }
 
     /** A trace through the real ingestion path, with the fields podam cannot fill sensibly cleared. */
@@ -306,26 +274,6 @@ private Trace createTrace() {
         return trace;
     }
 
-    /**
-     * The partition values actually bound into the emitted {@code IN} clause, so a test can assert the set is exact
-     * rather than merely inclusive. The driver substitutes bound values into the query text client-side, which is why
-     * they are readable here at all; the two assertions below are what make a change in that behaviour say so plainly
-     * instead of quietly turning every set assertion into a tautology on an empty set.
-     */
-    private static Set<Long> boundPartitionsOf(String sql) {
-        var clause = EMITTED_IN_CLAUSE.matcher(sql);
-        assertThat(clause.find())
-                .as("the delete SQL carries the partition predicate followed by an IN clause:%n%s", sql)
-                .isTrue();
-        var bound = PARTITION_VALUE.matcher(clause.group(1)).results()
-                .map(match -> Long.parseLong(match.group()))
-                .collect(Collectors.toUnmodifiableSet());
-        assertThat(bound)
-                .as("the IN clause carries inlined partition values — got '%s'", clause.group(1))
-                .isNotEmpty();
-        return bound;
-    }
-
     /** Invokes the DAO under a workspace/user context, as {@code TraceService} does for the live delete path. */
     private void delete(Set<Pair<UUID, UUID>> projectIdTraceIdPairs) {
         template.nonTransaction(connection -> traceDAO.delete(projectIdTraceIdPairs, connection))
diff --git a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java
index cde05fff7e6..84af1247613 100644
--- a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java
+++ b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java
@@ -20,17 +20,13 @@
 import com.comet.opik.infrastructure.auth.RequestContext;
 import com.comet.opik.infrastructure.db.TransactionTemplateAsync;
 import com.comet.opik.podam.PodamFactoryUtils;
-import com.comet.opik.utils.JsonUtils;
 import com.comet.opik.utils.WeeklyPartitions;
 import com.comet.opik.utils.template.TemplateUtils;
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.databind.JsonNode;
 import com.redis.testcontainers.RedisContainer;
 import io.r2dbc.spi.Statement;
-import lombok.Builder;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.commons.lang3.tuple.Pair;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.DisplayName;
@@ -51,6 +47,8 @@
 import ru.vyarus.dropwizard.guice.test.jupiter.ext.TestDropwizardAppExtension;
 import uk.co.jemos.podam.api.PodamFactory;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.time.LocalDate;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
@@ -85,7 +83,7 @@
  * the DAO's predicate resolved to any partition other than the one ClickHouse filed a row under, the mutation would
  * select the wrong parts and that row would survive, so
  * {@link #deleteClearsEveryEraAndBindsExactlyThosePartitions} passing <em>is</em> the agreement between the migration's
- * {@code PARTITION BY} as installed, the DAO's predicate, and {@link WeeklyPartitions#of}.
+ * {@code PARTITION BY} as installed, the DAO's predicate, and {@link WeeklyPartitions#groupByPartition}.
  *
  * <p>Each test then pairs that with the SQL ClickHouse actually received, because rows alone cannot see pruning
  * silently stop — a delete that stopped bounding itself is still correct, just slow, and that is the regression this
@@ -144,15 +142,15 @@ class TracesPartitionPruningMutationTest {
     private static final String USER = "user-" + RandomStringUtils.secure().nextAlphanumeric(32);
 
     /**
-     * The partition-key fragment the DAO's template emits. Only ever <b>compared against</b> the statement read back
-     * from {@code system.query_log} — never spliced into a query this suite runs. Verbatim comparison is safe because
-     * {@code query_log} stores the query as submitted, so both sides are the DAO's own template text.
+     * Matches the {@code IN PARTITION <value>} clause the scoped delete template emits (OPIK-8230), capturing the
+     * eight-digit {@code yyyyMMdd} partition value. Only ever <b>compared against</b> the statement read back from
+     * {@code system.query_log} - never spliced into a query this suite runs.
      */
-    private static final String PARTITION_PREDICATE = "toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))";
+    private static final Pattern IN_PARTITION_CLAUSE = Pattern.compile("IN\\s+PARTITION\\s+(\\d{8})");
 
     // The suite's whole SQL surface, per .agents/skills/opik-backend/SKILL.md "SQL Query Construction": one text block
     // per query, every varying value a :placeholder. There are no StringTemplate fragments and no interpolation at all,
-    // because nothing here re-implements the DAO's predicate - PARTITION_PREDICATE is only ever compared against the
+    // because nothing here re-implements the DAO's predicate - IN_PARTITION_CLAUSE is only ever compared against the
     // statement the DAO emitted, never spliced into a query of ours.
     //
     // The EXCHANGE/RENAME pair in installPartitionedSuccessorUnderTraces() stays as inline literals on purpose: they
@@ -196,18 +194,26 @@ AND query LIKE concat('%', :trace_id, '%')
             """;
 
     /**
-     * The newest {@code delete_traces} statement carrying exactly {@code pairs_size} pairs. A request larger than
-     * {@link com.comet.opik.infrastructure.FilterUtils#ANALYTICS_DELETE_BATCH_SIZE} is chunked by the DAO into one
-     * statement per chunk, and pruning is derived <b>per chunk</b> — so a test that reads only one statement cannot see
-     * the second chunk at all.
+     * Every {@code delete_traces} statement finished since {@code since} whose query text mentions
+     * {@code projectId}, in submission order - the multi-statement counterpart of {@link #LAST_TRACE_DELETE}, since
+     * one {@code delete()} call can now emit more than one statement (OPIK-8230).
+     * <p>
+     * Scoped by project id, not just by time: {@code WORKSPACE_ID} is one constant shared by every test in this
+     * class, so a time window alone is not exclusive to one test's own statements - a delete finished by a
+     * DIFFERENT test can still surface inside this window if system.query_log's buffered writes become visible only
+     * once a LATER test's own {@code SYSTEM FLUSH LOGS} call flushes them, even though its recorded
+     * {@code event_time_microseconds} is genuinely earlier. Each test mints its own fresh, collision-free project id
+     * ({@code ID_GENERATOR.generateId()}), which is literally embedded in the emitted query text, so filtering on it
+     * closes that gap the same way {@link #lastTraceDeleteSql} already does with a trace id.
      */
-    private static final String DELETE_BY_PAIR_COUNT = """
+    private static final String ALL_TRACE_DELETES_SINCE = """
             SELECT query
             FROM system.query_log
-            WHERE log_comment LIKE concat('delete_traces:%pairs_size=', :pairs_size)
+            WHERE log_comment LIKE 'delete_traces:%'
             AND type = 'QueryFinish'
-            ORDER BY event_time_microseconds DESC
-            LIMIT 1
+            AND event_time_microseconds >= :since
+            AND query LIKE concat('%', :project_id, '%')
+            ORDER BY event_time_microseconds
             """;
 
     /**
@@ -224,46 +230,47 @@ SELECT toString(uniqExact(id))
             """;
 
     /**
-     * The {@code IN} clause the DAO emitted, captured to end of line: the predicate sits on its own line in the
-     * template with {@code SETTINGS log_comment} on the next, so the line boundary delimits it exactly. Read this way
-     * rather than by matching a bracket style, because the driver's rendering of a {@code Long[]} is its own choice —
-     * what matters is which partition values are in the clause, not how it punctuates them.
+     * The physical part-level proof of pruning (OPIK-8230's own point): how many {@code MutatePart} events
+     * {@code table} logged in a window, and how many distinct partitions they span. {@code EXPLAIN} cannot see this -
+     * it reports what the read planner would select for a {@code SELECT}, a different layer from what a mutation is
+     * registered against, which is exactly the gap this suite used to fall into (see class Javadoc). Windowed by
+     * {@code event_time}, not correlated by {@code query_id}: a lightweight delete's mutation executes asynchronously
+     * under the mutations subsystem, so its {@code part_log} rows do not reliably carry the submitting statement's
+     * {@code query_id} - a finding from this ticket's own production investigation, not a guess.
      */
-    private static final Pattern EMITTED_IN_CLAUSE = Pattern.compile(
-            Pattern.quote(PARTITION_PREDICATE) + "\\s+IN\\s+([^\\n]*)");
-
-    /**
-     * The emitted statement's shape, so its {@code WHERE} clause can be lifted verbatim and re-asked as a
-     * {@code SELECT}: {@code EXPLAIN} does not accept a mutation. Captures the target table too, since the DAO picks
-     * {@code traces} or {@code traces_local} depending on the wrap flag.
-     */
-    private static final Pattern DELETE_SHAPE = Pattern.compile(
-            "DELETE\\s+FROM\\s+(\\S+)\\s+(WHERE\\b.*?)\\s+SETTINGS\\b", Pattern.DOTALL);
+    private static final String MUTATE_PART_EVENTS_SINCE = """
+            SELECT toString(count()), toString(uniqExact(partition_id))
+            FROM system.part_log
+            WHERE database = currentDatabase()
+            AND table = :table
+            AND event_type = 'MutatePart'
+            AND event_time >= :since
+            """;
 
-    /** {@code EXPLAIN} index entries that reflect <b>partition</b>-level part selection. */
-    private static final Set<String> PARTITION_INDEX_TYPES = Set.of("MinMax", "Partition");
+    /** How many active parts {@code table} currently holds - the "everything" a fallback delete should visit. */
+    private static final String ACTIVE_PART_COUNT = """
+            SELECT toString(count())
+            FROM system.parts
+            WHERE database = currentDatabase()
+            AND table = :table
+            AND active
+            """;
 
     /**
-     * Asks the planner how many parts the DAO's partition predicate selects. Declared once as a text block, per
-     * {@code .agents/skills/opik-backend/SKILL.md}: the table {@code <table>} and the predicate
-     * {@code <partition_expression>} are <b>fragments</b> and go through {@link TemplateUtils#newST}, the partition
-     * values are <b>values</b> and are bound — nothing is spliced with {@code .formatted(...)}.
-     * <p>
-     * The predicate fragment is {@link #PARTITION_PREDICATE}, and using the constant does not re-author what is under
-     * test: every caller has already asserted the emitted statement <em>contains</em> that exact text, so the constant
-     * is pinned to the DAO's own SQL by assertion rather than by string surgery on it. The partition values come from
-     * the emitted statement too, parsed by {@link #boundPartitionsOf} and bound here.
-     * <p>
-     * The DAO's {@code workspace_id} and {@code (project_id, id)} predicates are deliberately not reproduced. They are
-     * sort-key filters, not partition filters, so they cannot change partition selection — and leaving them out makes
-     * the unbounded case a full scan, which is the conservative direction for an assertion that the fallback prunes
-     * nothing.
+     * Whether {@code table} has any mutation still in flight. Used to settle the estate before starting a timed
+     * {@link #mutatePartActivitySince} window: a mutation from an EARLIER test in this suite can still be writing
+     * {@code MutatePart} rows to {@code system.part_log} after its submitting statement has already returned - this
+     * class runs every test against the same shared table (see class Javadoc), and nothing here sets
+     * {@code lightweight_deletes_sync} the way the retention sweep does, so a prior test's mutation completing late
+     * would otherwise land inside a LATER test's window and inflate its part/partition counts. Not a claim about
+     * production: production's real workload has no "between tests" moment to wait for.
      */
-    private static final String EXPLAIN_SELECTED_PARTS = """
-            EXPLAIN indexes = 1, json = 1
-            SELECT id
-            FROM <table>
-            <if(partition_expression)>WHERE <partition_expression> IN :partitions<endif>
+    private static final String PENDING_MUTATIONS_COUNT = """
+            SELECT toString(count())
+            FROM system.mutations
+            WHERE database = currentDatabase()
+            AND table = :table
+            AND NOT is_done
             """;
 
     /**
@@ -277,9 +284,6 @@ SELECT toString(uniqExact(id))
             ENGINE = Distributed('{cluster}', '<database>', 'traces_local', sipHash64(project_id))
             """;
 
-    /** A weekly partition value as it appears in SQL — {@code yyyyMMdd}, so always eight digits. */
-    private static final Pattern PARTITION_VALUE = Pattern.compile("\\d{8}");
-
     /**
      * One id per era the derivation has to get right, with the partition each resolves to. Not interchangeable samples:
      * {@code toMonday} agrees with the {@code Date32} expression across the ordinary calendar and diverges only
@@ -449,23 +453,60 @@ private static long partitionNameOf(LocalDate monday) {
     }
 
     /**
-     * The partition values actually bound into the emitted {@code IN} clause, so a test can assert the set is exact
-     * rather than merely inclusive. The driver substitutes bound values into the query text client-side, which is why
-     * they are readable here at all; the two assertions below are what make a change in that behaviour say so plainly
-     * instead of quietly turning every set assertion into a tautology on an empty set.
+     * The single partition value an {@code IN PARTITION} statement names. {@code IN PARTITION} takes exactly one
+     * partition per statement (OPIK-8230) - unlike the old {@code WHERE ... IN (...)} predicate this replaces, there
+     * is never a set to parse, only ever one value or none.
      */
-    private static Set<Long> boundPartitionsOf(String sql) {
-        var clause = EMITTED_IN_CLAUSE.matcher(sql);
+    private static long boundPartitionOf(String sql) {
+        var clause = IN_PARTITION_CLAUSE.matcher(sql);
         assertThat(clause.find())
-                .as("the delete SQL carries the partition predicate followed by an IN clause:%n%s", sql)
+                .as("the delete SQL carries an IN PARTITION clause:%n%s", sql)
                 .isTrue();
-        var bound = PARTITION_VALUE.matcher(clause.group(1)).results()
-                .map(match -> Long.parseLong(match.group()))
-                .collect(Collectors.toUnmodifiableSet());
-        assertThat(bound)
-                .as("the IN clause carries inlined partition values — got '%s'", clause.group(1))
-                .isNotEmpty();
-        return bound;
+        return Long.parseLong(clause.group(1));
+    }
+
+    /**
+     * How many {@code MutatePart} events {@code table} logged since {@code since}, and how many distinct partitions
+     * they span - the part-level proof that a delete actually scoped its mutation, not merely that its {@code WHERE}
+     * clause looks right. See {@link #MUTATE_PART_EVENTS_SINCE}'s Javadoc for why this is windowed rather than
+     * correlated by {@code query_id}.
+     * <p>
+     * {@code SYSTEM FLUSH LOGS} first: {@code system.part_log} is buffered and flushes on its own schedule, so a read
+     * immediately after a delete can race the write - the same reason {@link #lastTraceDeleteSql} flushes before
+     * reading {@code system.query_log}.
+     */
+    private MutatePartActivity mutatePartActivitySince(String table, Instant since) {
+        execute("SYSTEM FLUSH LOGS", _ -> {
+        });
+        return template.nonTransaction(connection -> {
+            var statement = connection.createStatement(MUTATE_PART_EVENTS_SINCE)
+                    .bind("table", table)
+                    .bind("since", since.atOffset(ZoneOffset.UTC).toLocalDateTime());
+            return Mono.from(statement.execute())
+                    .flatMap(result -> Mono.from(result.map((row, _) -> new MutatePartActivity(
+                            Integer.parseInt(row.get(0, String.class)),
+                            Integer.parseInt(row.get(1, String.class))))));
+        }).block();
+    }
+
+    private int activePartCountOf(String table) {
+        return Integer.parseInt(queryOneString(ACTIVE_PART_COUNT, statement -> statement.bind("table", table)));
+    }
+
+    /**
+     * Blocks until {@code table} has no in-flight mutation, polling {@link #PENDING_MUTATIONS_COUNT}. See that
+     * constant's Javadoc for why a timed MutatePart window needs this settle point.
+     */
+    private void waitForMutationsToSettle(String table) {
+        Awaitility.await()
+                .atMost(Duration.ofSeconds(30))
+                .pollInterval(Duration.ofMillis(200))
+                .until(() -> "0".equals(queryOneString(PENDING_MUTATIONS_COUNT,
+                        statement -> statement.bind("table", table))));
+    }
+
+    /** {@code MutatePart} events observed in a window: how many parts were touched, and how many partitions. */
+    private record MutatePartActivity(int parts, int partitions) {
     }
 
     /** Invokes the DAO under a workspace/user context, as {@code TraceService} does for the live delete path. */
@@ -505,24 +546,22 @@ private String lastTraceDeleteSql(UUID traceId) {
     }
 
     /**
-     * The newest {@code delete_traces} statement that carried exactly {@code pairsSize} pairs. Chunks are identified by
-     * their pair count rather than by a contained id, because the point is to inspect a <em>specific chunk</em> of one
-     * request — and the DAO stamps each chunk's size into its {@code log_comment}.
-     * <p>
-     * <b>The returned text is truncated for large statements.</b> ClickHouse caps {@code query_log.query} at
-     * {@code log_queries_cut_to_length} (100,000 bytes by default), and a full 10,000-pair chunk inlines to ~762 KiB,
-     * so anything at the tail of such a statement — the partition predicate included — is simply absent. Only assert on
-     * the text of statements small enough to be recorded whole.
+     * Every {@code delete_traces} statement ClickHouse finished since {@code since} whose query text mentions
+     * {@code projectId}. A single {@code delete()} call can now emit more than one statement - one per partition the
+     * batch resolves to (OPIK-8230) - so a caller that must see the whole set can no longer rely on "the newest
+     * one" the way {@link #lastTraceDeleteSql} does. See {@link #ALL_TRACE_DELETES_SINCE}'s Javadoc for why this is
+     * ALSO scoped by project id, not just by time.
      */
-    private String deleteSqlForChunkOf(int pairsSize) {
+    private List<String> deleteSqlsSince(Instant since, UUID projectId) {
         execute("SYSTEM FLUSH LOGS", _ -> {
         });
-        var sql = queryOneString(DELETE_BY_PAIR_COUNT,
-                statement -> statement.bind("pairs_size", String.valueOf(pairsSize)));
-        assertThat(sql)
-                .as("query_log holds a delete_traces statement with pairs_size=%s", pairsSize)
-                .isNotBlank();
-        return sql;
+        return template.stream(connection -> {
+            var statement = connection.createStatement(ALL_TRACE_DELETES_SINCE)
+                    .bind("since", since.atOffset(ZoneOffset.UTC).toLocalDateTime())
+                    .bind("project_id", projectId.toString());
+            return Flux.from(statement.execute())
+                    .flatMap(result -> result.map((row, _) -> row.get("query", String.class)));
+        }).collectList().block();
     }
 
     /** {@code "1"} while a live (non-lightweight-deleted) row exists for the id, {@code "0"} once it is gone. */
@@ -670,70 +709,6 @@ private void insertRawTrace(UUID projectId, UUID id) {
                         .bind("id", id.toString()));
     }
 
-    /**
-     * Parts the planner selects for the DAO's own statement, or empty when it reports no partition index at all.
-     * <p>
-     * {@code SELECT id} rather than {@code count()}, so no trivial-count optimisation can answer from metadata without
-     * selecting parts at all.
-     * <p>
-     * Only {@code MinMax} and {@code Partition} entries are considered, and {@code PrimaryKey} is deliberately
-     * excluded: the DAO's {@code WHERE} also filters {@code workspace_id} and {@code (project_id, id)}, which are the
-     * sort key, so {@code PrimaryKey} prunes parts for the <b>unbounded</b> statement too — counting it would make the
-     * fallback look pruned and destroy the discrimination this test rests on. Across the entries that do qualify it
-     * takes the smallest selected count and the largest initial count, so it does not depend on which of the two
-     * reports the pruning.
-     * <p>
-     * Empty is a meaningful answer rather than a failure: the {@code Indexes} block carries a partition entry only when
-     * the query filters on the partition key, so its absence is exactly what the fallback should produce.
-     */
-    private Optional<SelectedParts> partsSelectedBy(String daoDeleteSql) {
-        var shape = DELETE_SHAPE.matcher(daoDeleteSql);
-        assertThat(shape.find())
-                .as("the emitted statement has the expected DELETE shape:%n%s", daoDeleteSql)
-                .isTrue();
-        Set<Long> bound = EMITTED_IN_CLAUSE.matcher(daoDeleteSql).find()
-                ? boundPartitionsOf(daoDeleteSql)
-                : Set.of();
-
-        var explainSql = TemplateUtils.newST(EXPLAIN_SELECTED_PARTS)
-                .add("table", shape.group(1));
-        if (!bound.isEmpty()) {
-            explainSql.add("partition_expression", PARTITION_PREDICATE);
-        }
-        var sql = explainSql.render();
-
-        var explainRows = template.stream(connection -> {
-            var statement = connection.createStatement(sql);
-            if (!bound.isEmpty()) {
-                statement.bind("partitions", bound.toArray(Long[]::new));
-            }
-            return Flux.from(statement.execute())
-                    .flatMap(result -> result.map((row, _) -> row.get("explain", String.class)));
-        })
-                .collectList()
-                .block();
-        var explain = String.join("\n", explainRows);
-
-        var indexes = JsonUtils.getJsonNodeFromString(explain).findValue("Indexes");
-        if (indexes == null) {
-            return Optional.empty();
-        }
-        SelectedParts partition = null;
-        for (JsonNode index : indexes) {
-            if (!PARTITION_INDEX_TYPES.contains(index.path("Type").asText()) || !index.has("Selected Parts")) {
-                continue;
-            }
-            var entry = JsonUtils.treeToValue(index, SelectedParts.class);
-            partition = partition == null
-                    ? entry
-                    : partition.toBuilder()
-                            .selected(Math.min(partition.selected(), entry.selected()))
-                            .total(Math.max(partition.total(), entry.total()))
-                            .build();
-        }
-        return Optional.ofNullable(partition);
-    }
-
     /**
      * First column of the first row, as a string. Every read in this suite is a single scalar, so this is the only
      * mapper needed; values go in as binds.
@@ -755,17 +730,6 @@ private void execute(String sql, Consumer<Statement> binder) {
         }).block();
     }
 
-    /**
-     * The part counts {@code EXPLAIN indexes = 1, json = 1} reports for one index entry: how many parts the query
-     * started from, and how many survived pruning.
-     */
-    @Builder(toBuilder = true)
-    @JsonIgnoreProperties(ignoreUnknown = true)
-    private record SelectedParts(
-            @JsonProperty("Selected Parts") int selected,
-            @JsonProperty("Initial Parts") int total) {
-    }
-
     @Test
     @DisplayName("traces is a mutation-rejecting Distributed wrapper, so these deletes ran on traces_local")
     void distributedTracesRejectsDirectMutation() {
@@ -796,23 +760,19 @@ void allUuidV7DeletePrunesAndRemovesTheTargetRow() {
                 .doesNotContain(target.id())
                 .contains(bystander.id());
         var sql = lastTraceDeleteSql(target.id());
-        assertThat(sql).as("the mutation carries the partition predicate").contains(PARTITION_PREDICATE);
-        // Asserted as the SIZE, not as a value read back from WeeklyPartitions: the DAO derives its set from the same
-        // method, so an expectation taken from it would move with any regression and pass regardless. The value itself
-        // is already pinned two independent ways - the row above had to go away, which it only does if the predicate
-        // named the partition ClickHouse filed it under, and deleteClearsEveryEraAndBindsExactlyThosePartitions states
-        // its expectations as literals. What is left to say here is the property those cannot: an id the endpoint just
-        // minted is inside the 32-bit range, where the two id_at types agree, so it must widen the set to nothing.
-        assertThat(boundPartitionsOf(sql))
-                .as("bounded to one partition, not two: a recent id_at is a week both id_at types agree on")
-                .hasSize(1);
+        // A recent id_at is inside the 32-bit range, where the two id_at types agree, so WeeklyPartitions.groupByPartition
+        // resolves it to exactly one partition and the DAO emits the scoped IN PARTITION form - not the unbounded
+        // fallback. boundPartitionOf asserts the clause is present; there is nothing further to assert about its
+        // cardinality now that IN PARTITION accepts exactly one value per statement by construction.
+        assertThat(sql).as("the mutation carries an IN PARTITION clause").contains("IN PARTITION");
+        boundPartitionOf(sql);
     }
 
     @Test
     @DisplayName("the DAO's own delete clears every era and binds exactly those partitions")
     void deleteClearsEveryEraAndBindsExactlyThosePartitions() {
         // The three-way agreement - the migration's PARTITION BY as installed, the DAO's predicate, and
-        // WeeklyPartitions.of - asserted through the DAO's own delete rather than by re-evaluating the expression in
+        // WeeklyPartitions.groupByPartition - asserted through the DAO's own delete rather than by re-evaluating the expression in
         // test SQL. If the predicate resolved to any partition other than the one ClickHouse filed a row under, the
         // mutation would select the wrong parts and that row would SURVIVE. So "every row is gone" IS the agreement.
         //
@@ -826,17 +786,24 @@ void deleteClearsEveryEraAndBindsExactlyThosePartitions() {
         assertThat(ids.stream().map(id -> liveRowCount(projectId, id)))
                 .as("every era is seeded").containsOnly("1");
 
+        var since = Instant.now();
         delete(ids.stream().map(id -> Pair.of(projectId, id)).collect(Collectors.toUnmodifiableSet()));
 
         assertThat(ids.stream().map(id -> liveRowCount(projectId, id)))
                 .as("every era's row is gone, so the predicate named the partition each was actually filed under")
                 .containsOnly("0");
-        var sql = lastTraceDeleteSql(ids.getFirst());
-        assertThat(sql).as("the mutation carries the partition predicate").contains(PARTITION_PREDICATE);
-        // Four values for three eras: 1996 and 2025 are inside the 32-bit range and name one week each, and the 2199
-        // id names its legacy week too. Still an exact set, and still not a range across two centuries - which is the
-        // property under test; the extra value is the batch's own second representation, not a widening of its span.
-        assertThat(boundPartitionsOf(sql))
+
+        // One statement per partition the batch resolves to (OPIK-8230), not one statement carrying all four values:
+        // 1996 and 2025 are inside the 32-bit range and name one week each; the 2199 id names two (its own week AND
+        // its legacy wrap), so it is the id bound into two DIFFERENT statements rather than the batch widening to a
+        // fourth id. Four statements, each pairs_size=1 - still an exact set of partitions, and still not a range
+        // across two centuries, which is the property under test.
+        var sqls = deleteSqlsSince(since, projectId);
+        assertThat(sqls)
+                .as("one statement per partition the batch resolves to: %s", sqls)
+                .hasSize(4)
+                .allSatisfy(sql -> assertThat(sql).contains("pairs_size=1"));
+        assertThat(sqls.stream().map(TracesPartitionPruningMutationTest::boundPartitionOf))
                 .as("exactly the partitions the batch resolves to, not a range across two centuries")
                 .containsExactlyInAnyOrder(
                         partitionNameOf(ERA_MONDAYS.get(0)),
@@ -846,48 +813,59 @@ void deleteClearsEveryEraAndBindsExactlyThosePartitions() {
     }
 
     @Test
-    @DisplayName("the planner actually prunes, and the fallback provably does not")
+    @DisplayName("a scoped delete touches far fewer parts than the table holds, and the fallback touches them all")
     void pruningReachesThePlannerAndTheFallbackDoesNot() {
         // Correctness and pruning are different claims, and this is the only test that makes the second one. Deletes
-        // were already correct before OPIK-6901 - what the change buys is parts touched (3,928/3,928 -> 5/3,928 on
-        // prod-test), so a suite that cannot see pruning stop does not test what this change exists to do.
+        // were already correct before OPIK-8230 - what the change buys is parts touched (~3,650/~3,650 -> ~1/~3,650 on
+        // production), so a suite that cannot see pruning stop does not test what this change exists to do.
         //
-        // The regression it guards is specific: a migration rewrites the partition expression to something semantically
-        // identical but textually different, the planner stops recognising the DAO's predicate as the partition key,
-        // pruning silently stops - and values still agree, so every row is still deleted and every other assertion in
-        // this suite stays green. That is the property the removed AST pin covered; this asks the planner directly
-        // instead of inferring it from text.
-        //
-        // EXPLAIN does not accept a mutation, so the WHERE clause is lifted verbatim out of the DAO's own emitted
-        // DELETE and put behind a SELECT - predicate and bound partition values included. Only the verb changes; the
-        // statement being explained is still the DAO's. Same instrument and record shape as
-        // TracesLocalV2PartitioningTest.
-        // One row per era, so the table holds several partitions for the planner to prune between.
+        // Asked of system.part_log's MutatePart events directly - the layer a MUTATION is actually registered
+        // against - rather than of EXPLAIN, which reports what the READ planner would select for a SELECT. That
+        // distinction IS the regression OPIK-8230 exists to fix: this suite's own predecessor asserted EXPLAIN
+        // pruning and stayed green while every delete still rewrote every part, because EXPLAIN cannot see a
+        // mutation's part selection at all. See the class Javadoc.
+        var table = "traces_local";
         var projectId = ID_GENERATOR.generateId();
         var ids = ERA_MONDAYS.stream().map(TracesPartitionPruningMutationTest::idInWeekOf).toList();
         ids.forEach(id -> insertRawTrace(projectId, id));
 
-        // Bounded: one derivable id, so the predicate names a single one of those partitions.
+        // Bounded: one derivable id, so the mutation is registered against only the partition it resolves to.
+        waitForMutationsToSettle(table);
+        var boundedSince = Instant.now();
         delete(Set.of(Pair.of(projectId, ids.getFirst())));
-        var bounded = partsSelectedBy(lastTraceDeleteSql(ids.getFirst()))
-                .orElseThrow(() -> new AssertionError(
-                        "EXPLAIN reported no partition index for the bounded delete"));
-
-        // Unbounded: a non-v7 id in the batch, so no predicate at all. Its partner is a different era, so the query_log
-        // lookup finds this statement rather than the one above.
-        var partner = ids.get(1);
-        delete(Set.of(Pair.of(projectId, partner), Pair.of(projectId, NON_V7_ID)));
-        var unbounded = partsSelectedBy(lastTraceDeleteSql(partner));
-
-        assertThat(bounded.selected())
-                .as("the bounded delete selects fewer parts than the table holds: %s", bounded)
-                .isLessThan(bounded.total());
-        // Shown to discriminate, or it proves nothing - the same trap as `.contains(partition)` and
-        // `doesNotContain("toDayOfWeek")`. The fallback must not prune: either the planner reports no partition index at
-        // all, because nothing filters on the key, or it reports every part still selected.
-        assertThat(unbounded.map(parts -> parts.selected() == parts.total()).orElse(true))
-                .as("the fallback prunes nothing: %s", unbounded)
-                .isTrue();
+        var bounded = mutatePartActivitySince(table, boundedSince);
+
+        assertThat(bounded.partitions())
+                .as("the bounded delete touched at least the one partition it resolves to: %s", bounded)
+                .isGreaterThanOrEqualTo(1);
+
+        // Unbounded: a non-v7 id in the batch, so no IN PARTITION at all - the mutation must fall back to visiting
+        // every part, exactly as it did before OPIK-8230. Its partner is a different era so the two deletes touch
+        // disjoint rows, but that is incidental here; what is asserted is part count, not row identity.
+        waitForMutationsToSettle(table);
+        var totalPartsBeforeUnbounded = activePartCountOf(table);
+        var unboundedSince = Instant.now();
+        delete(Set.of(Pair.of(projectId, ids.get(1)), Pair.of(projectId, NON_V7_ID)));
+        var unbounded = mutatePartActivitySince(table, unboundedSince);
+
+        assertThat(unbounded.parts())
+                .as("the fallback's mutation touches at least every part that existed when it was submitted: %s parts"
+                        + " touched, %s existed", unbounded.parts(), totalPartsBeforeUnbounded)
+                .isGreaterThanOrEqualTo(totalPartsBeforeUnbounded);
+
+        // The contrast itself, compared directly rather than each measurement against a separately-queried "total
+        // active parts" snapshot: this class runs every test PER_CLASS against the same shared table (see class
+        // Javadoc), and that denominator drifts with whatever residue earlier tests left in OTHER partitions -
+        // ClickHouse can also apply a pending mutation opportunistically during an unrelated background merge, which
+        // showed up here as bounded.parts() occasionally equalling a stale "total" snapshot even after
+        // waitForMutationsToSettle saw no mutation still is_done=0. Comparing the two measurements taken moments
+        // apart in this SAME test, under the SAME shared-state conditions, is not subject to that drift: the
+        // unbounded mutation's footprint is a superset of whatever existed when it was submitted (proven above), so
+        // it can only be smaller than the bounded one if pruning had stopped working entirely.
+        assertThat(bounded.parts())
+                .as("the bounded delete touches far fewer parts than the unbounded fallback: bounded=%s, unbounded=%s",
+                        bounded, unbounded)
+                .isLessThan(unbounded.parts());
     }
 
     @Test
@@ -948,7 +926,7 @@ void topologySetupIsANoOpOnceTheEstateProvidesIt() {
                 .isEqualTo("0");
         assertThat(lastTraceDeleteSql(id))
                 .as("and the delete is still pruned")
-                .contains(PARTITION_PREDICATE);
+                .contains("IN PARTITION");
     }
 
     @Test
@@ -988,6 +966,7 @@ void requestSpanningTwoChunksPrunesEachChunkIndependently() {
         ordered.add(secondChunkRow);
         ordered.add(secondChunkCompanion);
 
+        var since = Instant.now();
         delete(ordered.stream().map(id -> Pair.of(projectId, id)).collect(Collectors.toCollection(
                 LinkedHashSet::new)));
 
@@ -998,15 +977,31 @@ void requestSpanningTwoChunksPrunesEachChunkIndependently() {
                 .as("and so is the row in the chunk that pruned")
                 .isEqualTo("0");
 
-        // The full chunk's statement is only checked to exist - that is what shows the request was split at all.
-        // Nothing about its text can be asserted, for the truncation reason above.
-        deleteSqlForChunkOf(ANALYTICS_DELETE_BATCH_SIZE);
+        var sqls = deleteSqlsSince(since, projectId);
+        // Chunk one fell back to exactly one unbounded statement: the non-v7 id disables pruning for the whole
+        // chunk, so it never fans out per-partition the way chunk two does below. Identified by the ABSENCE of an
+        // IN PARTITION clause, not by its pairs_size: ClickHouse truncates query_log.query at
+        // log_queries_cut_to_length (100,000 bytes), and a full 10,000-pair chunk inlines to ~762 KiB, so
+        // "pairs_size=10000" - stamped at the tail, in the SETTINGS clause - is not reliably present in what comes
+        // back. "IN PARTITION" sits right after "DELETE FROM <table>", at the very front of the statement, so its
+        // absence is readable regardless of truncation.
+        var chunkOneSqls = sqls.stream().filter(sql -> !sql.contains("IN PARTITION")).toList();
+        assertThat(chunkOneSqls)
+                .as("the full chunk fell back to exactly one unbounded statement: %s", sqls)
+                .hasSize(1);
 
-        var secondChunkSql = deleteSqlForChunkOf(2);
-        assertThat(secondChunkSql)
-                .as("the all-derivable chunk prunes even though an earlier chunk could not")
-                .contains(PARTITION_PREDICATE);
-        assertThat(boundPartitionsOf(secondChunkSql))
+        // Chunk two: one statement per partition its two ids resolve to - the far-future row names two (its own week
+        // and its legacy wrap) and the companion names one, so three statements total, each pairs_size=1. That makes
+        // the test bite on the refactor it exists to catch: hoisting the derivation out of the per-chunk concatMap,
+        // so the whole request is derived once, would let the non-v7 id in chunk one strip pruning from chunk two as
+        // well - collapsing it back to one unbounded statement, which the size assertion below would catch.
+        var chunkTwoSqls = sqls.stream().filter(sql -> !chunkOneSqls.contains(sql)).toList();
+        assertThat(chunkTwoSqls)
+                .as("the all-derivable chunk prunes even though an earlier chunk could not - one statement per"
+                        + " partition: %s", chunkTwoSqls)
+                .hasSize(3)
+                .allSatisfy(sql -> assertThat(sql).contains("pairs_size=1"));
+        assertThat(chunkTwoSqls.stream().map(TracesPartitionPruningMutationTest::boundPartitionOf))
                 .as("bounded to exactly its own two weeks, plus the legacy representation of the far-future one")
                 .containsExactlyInAnyOrder(partitionNameOf(ERA_MONDAYS.getLast()),
                         partitionNameOf(ERA_MONDAYS.get(1)),
@@ -1046,9 +1041,9 @@ void underivableIdDisablesPruning(String cause, UUID underivableId) {
         assertThat(sql)
                 .as("the unbounded form for a %s batch carries no id_at predicate of any kind", cause)
                 .doesNotContain("id_at");
-        assertThat(EMITTED_IN_CLAUSE.matcher(sql).find())
-                .as("and no partition IN clause: %s", sql)
-                .isFalse();
+        assertThat(sql)
+                .as("and no IN PARTITION clause: %s", sql)
+                .doesNotContain("IN PARTITION");
     }
 
     private static Stream<Arguments> underivableIdDisablesPruning() {
diff --git a/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java b/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java
index d95bcba84ae..c47eb5a0927 100644
--- a/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java
+++ b/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java
@@ -7,6 +7,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Stream;
@@ -16,8 +17,8 @@
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 /**
- * Covers {@link WeeklyPartitions#of}, which derives the weekly partition values a delete batch resolves to so the
- * mutation can prune instead of rewriting every part.
+ * Covers {@link WeeklyPartitions#groupByPartition}, which groups a delete batch's ids by the weekly partition each
+ * resolves to, so the mutation can prune instead of rewriting every part (OPIK-8230).
  * <p>
  * The expected values are not hand-computed: each is what ClickHouse itself returned for
  * {@code toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))} for that id, evaluated once with
@@ -32,9 +33,9 @@ class WeeklyPartitionsTest {
 
     @ParameterizedTest(name = "{0}")
     @MethodSource
-    @DisplayName("matches the partitions ClickHouse computed, under each id_at type")
+    @DisplayName("groupByPartition matches the partitions ClickHouse computed, under each id_at type")
     void matchesThePartitionsClickHouseComputed(String era, UUID id, Set<Long> expectedPartitions) {
-        assertThat(WeeklyPartitions.of(List.of(id))).contains(expectedPartitions);
+        assertThat(WeeklyPartitions.groupByPartition(List.of(id)).map(Map::keySet)).contains(expectedPartitions);
     }
 
     /**
@@ -47,8 +48,9 @@ private static Stream<Arguments> matchesThePartitionsClickHouseComputed() {
         return Stream.of(
                 // The ordinary case: id_at 2026-08-19 (a Wednesday) -> Monday 2026-08-17. Inside the 32-bit DateTime
                 // range, so both column types store the same instant and the id contributes a SINGLE value. That is
-                // every id real traffic produces, and it is why naming both representations costs nothing: the set a
-                // normal delete binds is exactly the set it bound before the legacy one was added.
+                // every id real traffic produces, and it is why naming both representations costs nothing: the
+                // partitions a normal delete resolves to are exactly the ones it resolved to before the legacy one
+                // was added.
                 arguments("ordinary id", UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), Set.of(20260817L)),
                 // Long before Opik existed but well after the Unix epoch, and well inside Date32's 1900 floor: an id
                 // this old prunes like any other, and likewise fits both column types. id_at 1996-02-09 -> 1996-02-05.
@@ -56,85 +58,20 @@ private static Stream<Arguments> matchesThePartitionsClickHouseComputed() {
                 // Far-future ids are supported, not excluded, and are the only ones that contribute two values: past
                 // 2106 the two column types disagree, so the batch has to name both weeks or be wrong on one schema -
                 // which is what removed the cutover flag. DateTime64 stores 2200-01-01 -> Monday 2199-12-30; the legacy
-                // 32-bit DateTime wraps the same id to 2063-11-25 -> Monday 2063-11-19. A set carrying only one of the
-                // two is a delete that reports success and removes nothing on the other schema. 4.1% of rows on
+                // 32-bit DateTime wraps the same id to 2063-11-25 -> Monday 2063-11-19. A grouping carrying only one of
+                // the two is a delete that reports success and removes nothing on the other schema. 4.1% of rows on
                 // prod-test look like this.
                 arguments("far-future id", UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2"),
                         Set.of(21991230L, 20631119L)));
     }
 
-    @Test
-    @DisplayName("a scattered batch yields the exact set, not a range")
-    void scatteredBatch() {
-        // 1996 and 2026 in one batch. An id_at RANGE over this span selected 2,644 of 3,928 parts on prod-test;
-        // the exact set selected 4. This is the reason the predicate is a set.
-        assertThat(WeeklyPartitions.of(List.of(
-                UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"),
-                UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8"))))
-                .contains(Set.of(20260817L, 19960205L));
-    }
-
-    @Test
-    @DisplayName("a non-v7 id disables pruning for the whole batch")
-    void nonV7DisablesPruning() {
-        // All-or-nothing on purpose: deriving a partition from a non-v7 id reads whatever sits in the timestamp
-        // field, and a wrong partition is a SILENTLY skipped delete. Omitting the predicate keeps the old behaviour.
-        assertThat(WeeklyPartitions.of(List.of(
-                UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"),
-                UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22")))) // v4
-                .isEmpty();
-    }
-
     @Test
     @DisplayName("a single non-v7 id yields no partitions")
     void singleNonV7() {
-        assertThat(WeeklyPartitions.of(List.of(UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22"))))
+        assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22"))))
                 .isEmpty();
     }
 
-    @Test
-    @DisplayName("duplicate ids in the same week collapse to one partition")
-    void duplicatesCollapse() {
-        assertThat(WeeklyPartitions.of(List.of(
-                UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"),
-                UUID.fromString("01a01a75-6f8e-7f22-9279-ee4f7ca7810d"),
-                UUID.fromString("01a01a75-609d-7935-8d22-2dd8dfeb2454"))))
-                .contains(Set.of(20260817L));
-    }
-
-    @Test
-    @DisplayName("an empty batch yields no partitions")
-    void emptyBatch() {
-        assertThat(WeeklyPartitions.of(List.of())).isEmpty();
-    }
-
-    @Test
-    @DisplayName("the returned set is immutable, so a delete's partitions cannot be narrowed after derivation")
-    void returnedSetIsImmutable() {
-        // Not a general hygiene assertion: this set IS the partition list a DELETE binds, so a caller that removed an
-        // entry would turn a correct delete into one that matches nothing and reports success.
-        var partitions = WeeklyPartitions.of(List.of(
-                UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"),
-                UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8")))
-                .orElseThrow();
-
-        assertThatThrownBy(() -> partitions.remove(20260817L))
-                .isInstanceOf(UnsupportedOperationException.class);
-        assertThat(partitions).containsExactlyInAnyOrder(20260817L, 19960205L);
-    }
-
-    @Test
-    @DisplayName("a null batch throws rather than reading as an unprunable one")
-    void nullBatchThrows() {
-        // The one intolerant case, and deliberately not folded into the empty result above: empty is a documented
-        // answer ("emit the unbounded form"), so a caller that lost its batch would get a valid-looking answer, issue a
-        // correct-but-unbounded mutation, and never learn it had a bug. Matches every other collection-taking method in
-        // this package, all of which are @NonNull.
-        assertThatThrownBy(() -> WeeklyPartitions.of(null))
-                .isInstanceOf(NullPointerException.class)
-                .hasMessageContaining("ids");
-    }
-
     @Test
     @DisplayName("the last id_at the column can store still prunes")
     void lastRepresentableIdStillPrunes() {
@@ -142,7 +79,8 @@ void lastRepresentableIdStillPrunes() {
         // truncates to 23:59:59, whose Date32 is 2299-12-31 (a Sunday) -> Monday 2299-12-25. The legacy DateTime wraps
         // the same id to 2027-10-18, a Monday. The ceiling check must be exclusive at exactly this point, hence a case
         // sitting on it.
-        assertThat(WeeklyPartitions.of(List.of(UUID.fromString("0978a65f-77ff-7abc-8000-000000000001"))))
+        assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("0978a65f-77ff-7abc-8000-000000000001")))
+                .map(Map::keySet))
                 .contains(Set.of(22991225L, 20271018L));
     }
 
@@ -154,7 +92,7 @@ void firstUnrepresentableIdDisablesPruning() {
         // itself a Monday, giving 23000101) is a partition the row is NOT in. Pruning off rather than clamped: unlike
         // the legacy 32-bit wrap, which is a modulo this reproduces exactly, matching ClickHouse here would mean
         // reproducing its saturation semantics, for ids that should not exist.
-        assertThat(WeeklyPartitions.of(List.of(UUID.fromString("0978a65f-7800-7abc-8000-000000000001"))))
+        assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("0978a65f-7800-7abc-8000-000000000001"))))
                 .isEmpty();
     }
 
@@ -164,29 +102,121 @@ void largestUuidV7TimestampDisablesPruning() {
         // All 48 timestamp bits set: 10889-08-02, the furthest future any UUIDv7 can encode. Read unsigned it is still
         // only ~2.8e14 ms, far inside Instant's range — so the guard is what excludes it, not an exception, and there
         // is no input on which this can throw.
-        assertThat(WeeklyPartitions.of(List.of(UUID.fromString("ffffffff-ffff-7abc-8000-000000000001"))))
+        assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("ffffffff-ffff-7abc-8000-000000000001"))))
+                .isEmpty();
+    }
+
+    @Test
+    @DisplayName("the earliest id a UUIDv7 can carry is inside Date32, so there is no floor to guard")
+    void earliestUuidV7TimestampIsInRange() {
+        // All 48 timestamp bits clear: id_at 1970-01-01, the earliest any UUIDv7 can encode (the field is unsigned).
+        // Its Monday, 1969-12-29, is 70 years above Date32's 1900 floor, so a below-1900 id_at is unreachable by
+        // construction rather than merely untested — which is why the derivation guards only the ceiling. It is also
+        // the floor of the legacy 32-bit DateTime, so the wrap leaves it untouched and it contributes one value.
+        assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("00000000-0000-7abc-8000-000000000001")))
+                .map(Map::keySet))
+                .contains(Set.of(19691229L));
+    }
+
+    @Test
+    @DisplayName("groupByPartition puts each ordinary id under its own single partition")
+    void groupByPartitionOrdinaryBatch() {
+        var ordinary = UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"); // -> 20260817L
+        var from1996 = UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8"); // -> 19960205L
+
+        assertThat(WeeklyPartitions.groupByPartition(List.of(ordinary, from1996)))
+                .contains(Map.of(
+                        20260817L, Set.of(ordinary),
+                        19960205L, Set.of(from1996)));
+    }
+
+    @Test
+    @DisplayName("groupByPartition puts a far-future id under both of its partitions")
+    void groupByPartitionFarFutureIdAppearsTwice() {
+        // Same id as matchesThePartitionsClickHouseComputed's far-future case: id_at 2200-01-01 partitions as
+        // 21991230 on the successor and as the legacy 32-bit wrap's 20631119. of() unions these into one set; here
+        // the id must appear as a MEMBER of both groups, since a statement scoped to 21991230 and one scoped to
+        // 20631119 each need this id in their own bound (project_id, id) pairs to actually delete the row wherever
+        // it physically lives.
+        var farFuture = UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2");
+
+        assertThat(WeeklyPartitions.groupByPartition(List.of(farFuture)))
+                .contains(Map.of(
+                        21991230L, Set.of(farFuture),
+                        20631119L, Set.of(farFuture)));
+    }
+
+    @Test
+    @DisplayName("groupByPartition mixes a far-future id into an ordinary id's group when they share a partition")
+    void groupByPartitionSharedPartitionMergesIntoOneGroup() {
+        // Two DIFFERENT ordinary ids whose legacy 1996 id lands in the same week (19960205) as the far-future id's
+        // legacy-wrapped value — this is the case that would break a naive "one group per id" implementation: the
+        // far-future id's SECOND value must merge into an existing group, not create a fresh singleton group that
+        // happens to collide.
+        var from1996 = UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8"); // -> 19960205L only
+        var farFuture = UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2"); // -> 21991230L, 20631119L
+
+        var grouped = WeeklyPartitions.groupByPartition(List.of(from1996, farFuture)).orElseThrow();
+
+        assertThat(grouped.keySet()).containsExactlyInAnyOrder(19960205L, 21991230L, 20631119L);
+        assertThat(grouped.get(19960205L)).containsExactlyInAnyOrder(from1996);
+        assertThat(grouped.get(21991230L)).containsExactlyInAnyOrder(farFuture);
+        assertThat(grouped.get(20631119L)).containsExactlyInAnyOrder(farFuture);
+    }
+
+    @Test
+    @DisplayName("groupByPartition collapses duplicate ids in the same partition into one entry via the Set")
+    void groupByPartitionDuplicatesCollapseWithinAGroup() {
+        var id = UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3");
+
+        assertThat(WeeklyPartitions.groupByPartition(List.of(id, id)))
+                .contains(Map.of(20260817L, Set.of(id)));
+    }
+
+    @Test
+    @DisplayName("groupByPartition disables pruning for the whole batch on a non-v7 id, same as of()")
+    void groupByPartitionNonV7DisablesPruning() {
+        assertThat(WeeklyPartitions.groupByPartition(List.of(
+                UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"),
+                UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22")))) // v4
                 .isEmpty();
     }
 
     @Test
-    @DisplayName("one out-of-range id disables pruning for the whole batch")
-    void outOfRangeIdDisablesPruningForTheWholeBatch() {
-        // Same all-or-nothing rule as a non-v7 id, for the same reason: a set derived from the rest of the batch is a
-        // set this row is not in.
-        assertThat(WeeklyPartitions.of(List.of(
+    @DisplayName("groupByPartition disables pruning for the whole batch on an out-of-range id, same as of()")
+    void groupByPartitionOutOfRangeIdDisablesPruning() {
+        assertThat(WeeklyPartitions.groupByPartition(List.of(
                 UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"),
                 UUID.fromString("ffffffff-ffff-7abc-8000-000000000001"))))
                 .isEmpty();
     }
 
     @Test
-    @DisplayName("the earliest id a UUIDv7 can carry is inside Date32, so there is no floor to guard")
-    void earliestUuidV7TimestampIsInRange() {
-        // All 48 timestamp bits clear: id_at 1970-01-01, the earliest any UUIDv7 can encode (the field is unsigned).
-        // Its Monday, 1969-12-29, is 70 years above Date32's 1900 floor, so a below-1900 id_at is unreachable by
-        // construction rather than merely untested — which is why `of` guards only the ceiling. It is also the floor of
-        // the legacy 32-bit DateTime, so the wrap leaves it untouched and it contributes one value.
-        assertThat(WeeklyPartitions.of(List.of(UUID.fromString("00000000-0000-7abc-8000-000000000001"))))
-                .contains(Set.of(19691229L));
+    @DisplayName("groupByPartition on an empty batch yields no groups")
+    void groupByPartitionEmptyBatch() {
+        assertThat(WeeklyPartitions.groupByPartition(List.of())).isEmpty();
+    }
+
+    @Test
+    @DisplayName("groupByPartition on a null batch throws rather than reading as unprunable")
+    void groupByPartitionNullBatchThrows() {
+        assertThatThrownBy(() -> WeeklyPartitions.groupByPartition(null))
+                .isInstanceOf(NullPointerException.class)
+                .hasMessageContaining("ids");
+    }
+
+    @Test
+    @DisplayName("groupByPartition's map and its per-partition sets are both immutable")
+    void groupByPartitionResultIsImmutable() {
+        // The map AND its values are load-bearing here: a caller that narrowed either would emit a DELETE ... IN
+        // PARTITION statement missing one of its own bound ids — a correct-looking statement that deletes less than
+        // it should, silently.
+        var ordinary = UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3");
+        var grouped = WeeklyPartitions.groupByPartition(List.of(ordinary)).orElseThrow();
+
+        assertThatThrownBy(() -> grouped.remove(20260817L))
+                .isInstanceOf(UnsupportedOperationException.class);
+        assertThatThrownBy(() -> grouped.get(20260817L).remove(ordinary))
+                .isInstanceOf(UnsupportedOperationException.class);
     }
 }