diff --git a/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md b/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md
new file mode 100644
index 00000000000..fe56e609510
--- /dev/null
+++ b/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fixed a rare case where a single run or span carrying data that could not be ingested would make other runs or trace events in the same batch go missing from the runs list, traces, and logs. Now the rest of the batch is always kept: an affected run still appears with its status (only its un-ingestable output is dropped), and an affected trace event or payload is skipped instead of taking down everything around it.
diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts
index 53b5edf0a46..1d8ab1a4dd1 100644
--- a/apps/webapp/app/env.server.ts
+++ b/apps/webapp/app/env.server.ts
@@ -1672,6 +1672,7 @@ const EnvironmentSchema = z
     RUN_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(2),
     RUN_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     RUN_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
+    RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH: z.coerce.number().int().default(1),
     RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
     RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
     RUN_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
diff --git a/apps/webapp/app/services/runsReplicationInstance.server.ts b/apps/webapp/app/services/runsReplicationInstance.server.ts
index 15950aa9008..164ce07fb92 100644
--- a/apps/webapp/app/services/runsReplicationInstance.server.ts
+++ b/apps/webapp/app/services/runsReplicationInstance.server.ts
@@ -117,6 +117,7 @@ function initializeRunsReplicationInstance() {
     maxFlushConcurrency: env.RUN_REPLICATION_MAX_FLUSH_CONCURRENCY,
     flushIntervalMs: env.RUN_REPLICATION_FLUSH_INTERVAL_MS,
     flushBatchSize: env.RUN_REPLICATION_FLUSH_BATCH_SIZE,
+    maxPoisonStripsPerBatch: env.RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH,
     leaderLockTimeoutMs: env.RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS,
     leaderLockExtendIntervalMs: env.RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS,
     leaderLockAcquireAdditionalTimeMs: env.RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS,
diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts
index 732665461dd..555bca49a6b 100644
--- a/apps/webapp/app/services/runsReplicationService.server.ts
+++ b/apps/webapp/app/services/runsReplicationService.server.ts
@@ -1,11 +1,13 @@
 import type { ClickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server";
 import {
   type ClickHouse,
+  type ClickHouseSettings,
   type PayloadInsertArray,
   type TaskRunInsertArray,
   composeTaskRunVersion,
   getPayloadField,
   getTaskRunField,
+  TASK_RUN_INDEX,
 } from "@internal/clickhouse";
 import { type RedisOptions } from "@internal/redis";
 import {
@@ -41,9 +43,9 @@ import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
 import { calculateErrorFingerprint } from "~/utils/errorFingerprinting";
 import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
 import {
-  isClickHouseJsonParseError,
-  parseRowNumberFromError,
-  sanitizeRows,
+  insertWithBadRowSkip,
+  insertWithLimitedStrip,
+  type JsonParseRecoveryOutcome,
 } from "~/v3/eventRepository/sanitizeRowsOnParseError.server";
 
 interface TransactionEvent<T = any> {
@@ -112,6 +114,7 @@ export type RunsReplicationServiceOptions = {
   insertMaxDelayMs?: number;
   disablePayloadInsert?: boolean;
   disableErrorFingerprinting?: boolean;
+  maxPoisonStripsPerBatch?: number;
 };
 
 type PostgresTaskRun = TaskRun & { masterQueue: string };
@@ -175,14 +178,41 @@ export class RunsReplicationService {
   private _disableErrorFingerprinting: boolean;
 
   /**
-   * Counts batches that hit a ClickHouse `Cannot parse JSON object` failure
-   * that survived one sanitize-retry. These batches are dropped on the floor
-   * (returning success-ish to the caller so the retry layer doesn't spin on
-   * the same deterministic failure), and we track the drop count for
-   * observability. Counter only — does not gate behaviour.
+   * Counts batches where every row was un-ingestable even with its JSON
+   * stripped, so nothing landed. Row isolation lands anything strippable, so
+   * this should stay at zero; a non-zero value means the recovery itself is
+   * failing. Counter only, does not gate behaviour.
    */
   private _permanentlyDroppedBatches = 0;
 
+  /**
+   * Counts batches that took the row-isolation recovery path: a
+   * `Cannot parse JSON object` failure the sanitizer could not repair, where we
+   * bisected to the poison rows and landed the batch with their JSON stripped.
+   * Reliable per-event signal that user data is hitting the ceiling.
+   */
+  private _rowIsolationRecoveries = 0;
+
+  /**
+   * Counts batches whose isolation blew the per-batch insert budget (a poison
+   * flood), so the remaining rows were stripped in one insert instead of
+   * bisected further. A burst signal; clean rows in those batches also lose
+   * their JSON.
+   */
+  private _recoveryCapHits = 0;
+
+  /**
+   * Counts rows that landed with their un-ingestable JSON column(s) stripped
+   * (the run kept its status, only the output/payload content was lost).
+   */
+  private _rowsStripped = 0;
+
+  /**
+   * Counts rows dropped entirely because they could not be parsed even with
+   * their JSON stripped. The true data-loss signal; expected to stay near zero.
+   */
+  private _permanentlyDroppedRows = 0;
+
   // Metrics
   private _replicationLagHistogram: Histogram;
   private _batchesFlushedCounter: Counter;
@@ -191,6 +221,11 @@ export class RunsReplicationService {
   private _payloadsInsertedCounter: Counter;
   private _insertRetriesCounter: Counter;
   private _eventsProcessedCounter: Counter;
+  private _rowIsolatedBatchesCounter: Counter;
+  private _recoveryCapHitsCounter: Counter;
+  private _rowsStrippedCounter: Counter;
+  private _rowsDroppedCounter: Counter;
+  private _droppedBatchesCounter: Counter;
   private _flushDurationHistogram: Histogram;
 
   public readonly events: EventEmitter<RunsReplicationServiceEvents>;
@@ -244,6 +279,38 @@ export class RunsReplicationService {
       description: "Replication events processed (inserts, updates, deletes)",
     });
 
+    this._rowIsolatedBatchesCounter = this._meter.createCounter(
+      "runs_replication.batches_row_isolated",
+      {
+        description:
+          "Batches recovered by isolating un-ingestable rows (landed the rest) after a ClickHouse JSON parse error",
+        unit: "batches",
+      }
+    );
+
+    this._recoveryCapHitsCounter = this._meter.createCounter("runs_replication.recovery_cap_hits", {
+      description:
+        "Batches whose row isolation hit the per-batch insert budget and stripped the remainder in one insert (poison-flood signal)",
+      unit: "batches",
+    });
+
+    this._rowsStrippedCounter = this._meter.createCounter("runs_replication.rows_stripped", {
+      description:
+        "Rows landed with their un-ingestable JSON stripped (kept the row, lost only the JSON content)",
+      unit: "rows",
+    });
+
+    this._rowsDroppedCounter = this._meter.createCounter("runs_replication.rows_dropped", {
+      description:
+        "Rows dropped entirely because they could not parse even with their JSON stripped",
+      unit: "rows",
+    });
+
+    this._droppedBatchesCounter = this._meter.createCounter("runs_replication.batches_dropped", {
+      description: "Batches where every row was un-ingestable even with its JSON stripped",
+      unit: "batches",
+    });
+
     this._flushDurationHistogram = this._meter.createHistogram(
       "runs_replication.flush_duration_ms",
       {
@@ -411,11 +478,31 @@ export class RunsReplicationService {
     });
   }
 
-  /** Exposed for tests and metrics — total batches lost to unrecoverable parse errors. */
+  /** Exposed for tests and metrics — batches where nothing landed even after stripping JSON. */
   get permanentlyDroppedBatches() {
     return this._permanentlyDroppedBatches;
   }
 
+  /** Exposed for tests and metrics — batches that took the row-isolation recovery path. */
+  get rowIsolationRecoveries() {
+    return this._rowIsolationRecoveries;
+  }
+
+  /** Exposed for tests and metrics — batches whose isolation hit the per-batch insert budget. */
+  get recoveryCapHits() {
+    return this._recoveryCapHits;
+  }
+
+  /** Exposed for tests and metrics — rows that landed with their un-ingestable JSON stripped. */
+  get rowsStripped() {
+    return this._rowsStripped;
+  }
+
+  /** Exposed for tests and metrics — rows dropped entirely (could not parse even stripped). */
+  get permanentlyDroppedRows() {
+    return this._permanentlyDroppedRows;
+  }
+
   public async shutdown() {
     if (this._isShuttingDown) return;
 
@@ -870,15 +957,11 @@ export class RunsReplicationService {
           payloadError = plErr;
         }
 
-        // Only count rows that actually landed in ClickHouse. `kind: "dropped"`
-        // means the recovery wrapper bailed (sanitizer no-op or sanitize-retry
-        // still failed) — those rows never made it, so they must not show up
-        // as successful inserts in the per-batch counter.
-        if (!trErr && trOutcome?.kind !== "dropped") {
-          this._taskRunsInsertedCounter.add(group.taskRunInserts.length);
+        if (!trErr && trOutcome) {
+          this._taskRunsInsertedCounter.add(landedRowCount(group.taskRunInserts.length, trOutcome));
         }
-        if (!plErr && plOutcome?.kind !== "dropped") {
-          this._payloadsInsertedCounter.add(group.payloadInserts.length);
+        if (!plErr && plOutcome) {
+          this._payloadsInsertedCounter.add(landedRowCount(group.payloadInserts.length, plOutcome));
         }
       }
 
@@ -1034,11 +1117,12 @@ export class RunsReplicationService {
       return;
     }
     return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => {
-      const doInsert = async () => {
-        const [insertError, insertResult] = await clickhouse.taskRuns.insertCompactArrays(
-          taskRunInserts,
-          { params: { clickhouse_settings: this.#getClickhouseInsertSettings() } }
-        );
+      const rawInsert = async (rows: TaskRunInsertArray[], extraSettings?: ClickHouseSettings) => {
+        const [insertError, insertResult] = await clickhouse.taskRuns.insertCompactArrays(rows, {
+          params: {
+            clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings },
+          },
+        });
         if (insertError) {
           this.logger.error("Error inserting task run inserts attempt", {
             error: insertError,
@@ -1050,12 +1134,27 @@ export class RunsReplicationService {
         return insertResult;
       };
 
-      return await this.#insertWithJsonParseRecovery(
-        taskRunInserts,
-        doInsert,
-        "task_runs_v2",
-        attempt
-      );
+      const outcome = await insertWithLimitedStrip({
+        rows: taskRunInserts,
+        contextLabel: "task_runs_v2",
+        logger: this.logger,
+        logContext: { attempt },
+        insert: (rows) => rawInsert(rows),
+        insertSync: (rows) =>
+          rawInsert(rows, { async_insert: 0, input_format_parallel_parsing: 0 }),
+        insertAllowingBadRows: (rows) =>
+          rawInsert(rows, {
+            async_insert: 0,
+            input_format_parallel_parsing: 0,
+            input_format_allow_errors_num: String(rows.length),
+            input_format_allow_errors_ratio: 1,
+          }),
+        stripJsonColumns: stripTaskRunJsonColumns,
+        maxPoisonStrips: this.options.maxPoisonStripsPerBatch,
+        hasMaterializedViews: true,
+      });
+      this.#recordRecoveryOutcome(outcome, "task_runs_v2", taskRunInserts.length);
+      return outcome;
     });
   }
 
@@ -1068,10 +1167,14 @@ export class RunsReplicationService {
       return;
     }
     return await startSpan(this._tracer, "insertPayloadInserts", async (span) => {
-      const doInsert = async () => {
+      const rawInsert = async (rows: PayloadInsertArray[], extraSettings?: ClickHouseSettings) => {
         const [insertError, insertResult] = await clickhouse.taskRuns.insertPayloadsCompactArrays(
-          payloadInserts,
-          { params: { clickhouse_settings: this.#getClickhouseInsertSettings() } }
+          rows,
+          {
+            params: {
+              clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings },
+            },
+          }
         );
         if (insertError) {
           this.logger.error("Error inserting payload inserts attempt", {
@@ -1084,111 +1187,54 @@ export class RunsReplicationService {
         return insertResult;
       };
 
-      return await this.#insertWithJsonParseRecovery(
-        payloadInserts,
-        doInsert,
-        "raw_task_runs_payload_v1",
-        attempt
-      );
+      const outcome = await insertWithBadRowSkip({
+        rows: payloadInserts,
+        contextLabel: "raw_task_runs_payload_v1",
+        logger: this.logger,
+        logContext: { attempt },
+        insert: (rows) => rawInsert(rows),
+        insertAllowingBadRows: (rows) =>
+          rawInsert(rows, {
+            async_insert: 0,
+            input_format_parallel_parsing: 0,
+            input_format_allow_errors_num: String(rows.length),
+            input_format_allow_errors_ratio: 1,
+          }),
+        hasMaterializedViews: false,
+      });
+      this.#recordRecoveryOutcome(outcome, "raw_task_runs_payload_v1", payloadInserts.length);
+      return outcome;
     });
   }
 
-  /**
-   * Wraps a ClickHouse insert with reactive UTF-16 sanitization for
-   * `Cannot parse JSON object` rejections. Mirrors the pattern from
-   * `ClickhouseEventRepository.#insertWithJsonParseRecovery` introduced
-   * in #3659 — same root cause (lone UTF-16 surrogates in user-provided
-   * JSON), same recovery shape:
-   *
-   *   1. Try the insert. Healthy batches pay zero scan cost.
-   *   2. On parse error, walk the whole batch via `sanitizeRows` and
-   *      replace any lone-surrogate string with `"[invalid-utf16]"`.
-   *   3. Retry once. If the sanitizer found nothing or the retry also
-   *      fails with the same error class, drop the batch loudly and
-   *      return — do NOT rethrow, otherwise the surrounding
-   *      `#insertWithRetry` layer would spin three more times on the
-   *      same deterministic failure.
-   *   4. Non-parse errors propagate unchanged so the existing
-   *      transient-retry path still handles them.
-   *
-   * The whole-batch scan (rather than slicing on the `at row N` hint) is
-   * deliberate: `at row N` semantics under `input_format_parallel_parsing`
-   * aren't stable enough to safely skip rows. The cost is bounded because
-   * `detectBadJsonStrings` exits in O(1) for clean strings.
-   */
-  async #insertWithJsonParseRecovery<T extends object>(
-    rows: T[],
-    doInsert: () => Promise<unknown>,
+  #recordRecoveryOutcome(
+    outcome: JsonParseRecoveryOutcome,
     contextLabel: string,
-    attempt: number
-  ): Promise<
-    | { kind: "inserted"; insertResult: unknown }
-    | { kind: "sanitized"; insertResult: unknown }
-    | { kind: "dropped" }
-  > {
-    try {
-      return { kind: "inserted", insertResult: await doInsert() };
-    } catch (firstError) {
-      if (!isClickHouseJsonParseError(firstError)) throw firstError;
-
-      const firstMessage =
-        typeof firstError === "object" && firstError !== null && "message" in firstError
-          ? String((firstError as { message?: unknown }).message ?? "")
-          : String(firstError);
-
-      const rowHint = parseRowNumberFromError(firstMessage);
-      const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
-
-      if (fieldsSanitized === 0) {
-        this._permanentlyDroppedBatches += 1;
-        this.logger.error(
-          "Dropped batch — ClickHouse JSON parse error but sanitizer found nothing to fix",
-          {
-            contextLabel,
-            attempt,
-            batchSize: rows.length,
-            clickhouseRowHint: rowHint,
-            permanentlyDroppedBatches: this._permanentlyDroppedBatches,
-            sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024),
-            clickhouseError: firstMessage.split("\n")[0],
-          }
-        );
-        return { kind: "dropped" };
-      }
+    batchSize: number
+  ) {
+    if (outcome.kind !== "recovered") {
+      return;
+    }
 
-      this.logger.warn("Sanitizing batch after ClickHouse JSON parse error", {
-        contextLabel,
-        attempt,
-        batchSize: rows.length,
-        clickhouseRowHint: rowHint,
-        rowsTouched,
-        fieldsSanitized,
-        clickhouseError: firstMessage.split("\n")[0],
-      });
+    this._rowIsolationRecoveries += 1;
+    this._rowIsolatedBatchesCounter.add(1, { table: contextLabel });
 
-      try {
-        return { kind: "sanitized", insertResult: await doInsert() };
-      } catch (retryError) {
-        if (!isClickHouseJsonParseError(retryError)) throw retryError;
+    if (outcome.capped) {
+      this._recoveryCapHits += 1;
+      this._recoveryCapHitsCounter.add(1, { table: contextLabel });
+    }
 
+    if (outcome.rowsStripped > 0) {
+      this._rowsStripped += outcome.rowsStripped;
+      this._rowsStrippedCounter.add(outcome.rowsStripped, { table: contextLabel });
+    }
+
+    if (outcome.rowsDropped > 0) {
+      this._permanentlyDroppedRows += outcome.rowsDropped;
+      this._rowsDroppedCounter.add(outcome.rowsDropped, { table: contextLabel });
+      if (outcome.rowsDropped === batchSize) {
         this._permanentlyDroppedBatches += 1;
-        const retryMessage =
-          typeof retryError === "object" && retryError !== null && "message" in retryError
-            ? String((retryError as { message?: unknown }).message ?? "")
-            : String(retryError);
-        this.logger.error(
-          "Dropped batch after sanitize-retry still hit ClickHouse JSON parse error",
-          {
-            contextLabel,
-            attempt,
-            batchSize: rows.length,
-            permanentlyDroppedBatches: this._permanentlyDroppedBatches,
-            sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024),
-            firstError: firstMessage.split("\n")[0],
-            retryError: retryMessage.split("\n")[0],
-          }
-        );
-        return { kind: "dropped" };
+        this._droppedBatchesCounter.add(1, { table: contextLabel });
       }
     }
   }
@@ -1542,3 +1588,19 @@ function lsnToUInt64(lsn: string): bigint {
   const [seg, off] = lsn.split("/");
   return (BigInt("0x" + seg) << 32n) | BigInt("0x" + off);
 }
+
+function landedRowCount(groupSize: number, outcome: JsonParseRecoveryOutcome): number {
+  if (outcome.kind === "recovered") {
+    return Math.max(0, groupSize - outcome.rowsDropped);
+  }
+  return groupSize;
+}
+
+const STRIPPED_JSON: { data: unknown } = { data: undefined };
+
+function stripTaskRunJsonColumns(row: TaskRunInsertArray): TaskRunInsertArray {
+  const stripped = [...row] as TaskRunInsertArray;
+  stripped[TASK_RUN_INDEX.output] = STRIPPED_JSON;
+  stripped[TASK_RUN_INDEX.error] = STRIPPED_JSON;
+  return stripped;
+}
diff --git a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts
index 573097005ed..3a2c34f0dfc 100644
--- a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts
+++ b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts
@@ -1,5 +1,6 @@
 import type {
   ClickHouse,
+  ClickHouseSettings,
   LlmMetricsV1Input,
   MetricsV1Input,
   TaskEventDetailedSummaryV1Result,
@@ -66,9 +67,8 @@ import type {
   TraceSummary,
 } from "./eventRepository.types";
 import {
-  isClickHouseJsonParseError,
-  parseRowNumberFromError,
-  sanitizeRows,
+  insertWithBadRowSkip,
+  type JsonParseRecoveryOutcome,
 } from "./sanitizeRowsOnParseError.server";
 
 export type ClickhouseEventRepositoryConfig = {
@@ -121,14 +121,43 @@ export class ClickhouseEventRepository implements IEventRepository {
   private _tracer: Tracer;
   private _version: "v1" | "v2";
   /**
-   * Counts batches that hit a ClickHouse JSON parse failure that survived
-   * one sanitize-retry. These batches are dropped on the floor (the scheduler
-   * is told the flush "succeeded" so its queue counter doesn't leak), and we
-   * track the drop count for observability.
+   * Counts batches where every row was un-ingestable even with its JSON
+   * stripped, so nothing landed. Row isolation lands anything strippable, so
+   * this should stay at zero; a non-zero value means the recovery is failing.
    */
   private _permanentlyDroppedBatches = 0;
   private readonly _droppedBatchesCounter: Counter;
 
+  /**
+   * Counts batches that took the row-isolation recovery path: a
+   * `Cannot parse JSON object` failure the sanitizer could not repair, where we
+   * bisected to the poison rows and landed the batch with their JSON stripped.
+   * Reliable per-event signal.
+   */
+  private _rowIsolationRecoveries = 0;
+  private readonly _rowIsolatedBatchesCounter: Counter;
+
+  /**
+   * Counts batches whose isolation blew the per-batch insert budget (a poison
+   * flood), so the remaining rows were stripped in one insert. A burst signal.
+   */
+  private _recoveryCapHits = 0;
+  private readonly _recoveryCapHitsCounter: Counter;
+
+  /**
+   * Counts rows that landed with their un-ingestable JSON stripped (the span
+   * kept its place in the trace, only the attributes content was lost).
+   */
+  private _rowsStripped = 0;
+  private readonly _rowsStrippedCounter: Counter;
+
+  /**
+   * Counts rows dropped entirely because they could not parse even with their
+   * JSON stripped. The true data-loss signal; expected to stay near zero.
+   */
+  private _permanentlyDroppedRows = 0;
+  private readonly _rowsDroppedCounter: Counter;
+
   constructor(config: ClickhouseEventRepositoryConfig) {
     this._clickhouse = config.clickhouse;
     this._config = config;
@@ -140,6 +169,26 @@ export class ClickhouseEventRepository implements IEventRepository {
       description: "Batches permanently dropped after an unrecoverable ClickHouse JSON parse error",
       unit: "batches",
     });
+    this._rowIsolatedBatchesCounter = meter.createCounter("ingest.flush.batches_row_isolated", {
+      description:
+        "Batches recovered by isolating un-ingestable rows (landed the rest) after a ClickHouse JSON parse error",
+      unit: "batches",
+    });
+    this._recoveryCapHitsCounter = meter.createCounter("ingest.flush.recovery_cap_hits", {
+      description:
+        "Batches whose row isolation hit the per-batch insert budget and stripped the remainder in one insert (poison-flood signal)",
+      unit: "batches",
+    });
+    this._rowsStrippedCounter = meter.createCounter("ingest.flush.rows_stripped", {
+      description:
+        "Rows landed with their un-ingestable JSON stripped (kept the row, lost only the JSON content)",
+      unit: "rows",
+    });
+    this._rowsDroppedCounter = meter.createCounter("ingest.flush.rows_dropped", {
+      description:
+        "Rows dropped entirely because they could not parse even with their JSON stripped",
+      unit: "rows",
+    });
 
     this._flushScheduler = new DynamicFlushScheduler({
       name: `task_events_${this._version}`,
@@ -189,11 +238,31 @@ export class ClickhouseEventRepository implements IEventRepository {
     return this._config.maximumLiveReloadingSetting ?? 1000;
   }
 
-  /** Exposed for tests and metrics — total batches lost to unrecoverable parse errors. */
+  /** Exposed for tests and metrics — batches where nothing landed even after stripping JSON. */
   get permanentlyDroppedBatches() {
     return this._permanentlyDroppedBatches;
   }
 
+  /** Exposed for tests and metrics — batches that took the row-isolation recovery path. */
+  get rowIsolationRecoveries() {
+    return this._rowIsolationRecoveries;
+  }
+
+  /** Exposed for tests and metrics — batches whose isolation hit the per-batch insert budget. */
+  get recoveryCapHits() {
+    return this._recoveryCapHits;
+  }
+
+  /** Exposed for tests and metrics — rows that landed with their un-ingestable JSON stripped. */
+  get rowsStripped() {
+    return this._rowsStripped;
+  }
+
+  /** Exposed for tests and metrics — rows dropped entirely (could not parse even stripped). */
+  get permanentlyDroppedRows() {
+    return this._permanentlyDroppedRows;
+  }
+
   /**
    * Clamps a start time (in nanoseconds) to now if it's too far in the past.
    * Returns the clamped value as a bigint.
@@ -262,32 +331,39 @@ export class ClickhouseEventRepository implements IEventRepository {
           ? this._clickhouse.taskEventsV2.insert
           : this._clickhouse.taskEvents.insert;
 
-      const doInsert = async () => {
-        const [insertError, insertResult] = await insertFn(events, {
+      const contextLabel = `task_events_${this._version}`;
+      const rawInsert = async (
+        rows: (TaskEventV1Input | TaskEventV2Input)[],
+        extraSettings?: ClickHouseSettings
+      ) => {
+        const [insertError, insertResult] = await insertFn(rows, {
           params: {
-            clickhouse_settings: this.#getClickhouseInsertSettings(),
+            clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings },
           },
         });
         if (insertError) throw insertError;
         return insertResult;
       };
 
-      const outcome = await this.#insertWithJsonParseRecovery(
-        flushId,
-        events,
-        doInsert,
-        `task_events_${this._version}`
-      );
-
-      if (outcome.kind === "dropped") {
-        // Loud log already emitted; nothing landed in ClickHouse — don't publish to Redis.
-        return;
-      }
+      const outcome = await insertWithBadRowSkip({
+        rows: events,
+        contextLabel,
+        logger,
+        logContext: { flushId, version: this._version },
+        insert: (rows) => rawInsert(rows),
+        insertAllowingBadRows: (rows) =>
+          rawInsert(rows, {
+            async_insert: 0,
+            input_format_parallel_parsing: 0,
+            input_format_allow_errors_num: String(rows.length),
+            input_format_allow_errors_ratio: 1,
+          }),
+      });
+      this.#recordRecoveryOutcome(outcome, contextLabel, events.length);
 
       logger.debug("ClickhouseEventRepository.flushBatch Inserted batch into clickhouse", {
         events: events.length,
-        insertResult: outcome.insertResult,
-        sanitized: outcome.kind === "sanitized",
+        outcome: outcome.kind,
         version: this._version,
       });
 
@@ -296,129 +372,66 @@ export class ClickhouseEventRepository implements IEventRepository {
   }
 
   async #flushLlmMetricsBatch(flushId: string, rows: LlmMetricsV1Input[]) {
-    const doInsert = async () => {
-      const [insertError, insertResult] = await this._clickhouse.llmMetrics.insert(rows, {
+    const rawInsert = async (batch: LlmMetricsV1Input[], extraSettings?: ClickHouseSettings) => {
+      const [insertError, insertResult] = await this._clickhouse.llmMetrics.insert(batch, {
         params: {
-          clickhouse_settings: this.#getClickhouseInsertSettings(),
+          clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings },
         },
       });
       if (insertError) throw insertError;
       return insertResult;
     };
 
-    const outcome = await this.#insertWithJsonParseRecovery(
-      flushId,
+    const outcome = await insertWithBadRowSkip({
       rows,
-      doInsert,
-      "llm_metrics_v1"
-    );
-
-    if (outcome.kind === "dropped") {
-      return;
-    }
+      contextLabel: "llm_metrics_v1",
+      logger,
+      logContext: { flushId },
+      insert: (batch) => rawInsert(batch),
+      insertAllowingBadRows: (batch) =>
+        rawInsert(batch, {
+          async_insert: 0,
+          input_format_parallel_parsing: 0,
+          input_format_allow_errors_num: String(batch.length),
+          input_format_allow_errors_ratio: 1,
+        }),
+    });
+    this.#recordRecoveryOutcome(outcome, "llm_metrics_v1", rows.length);
 
     logger.debug("ClickhouseEventRepository.flushLlmMetricsBatch Inserted LLM metrics batch", {
       rows: rows.length,
-      sanitized: outcome.kind === "sanitized",
+      outcome: outcome.kind,
     });
   }
 
-  /**
-   * Wraps a ClickHouse insert callable with reactive UTF-16 sanitization.
-   *
-   * On a `Cannot parse JSON object` failure:
-   *   1. Sanitize the batch from `max(0, parsedRowN - 1)` onwards (rows
-   *      before the failing one parsed fine — known good).
-   *   2. Retry the insert once with the sanitized batch.
-   *   3. If the retry still fails with the same error class, log loudly,
-   *      increment `permanentlyDroppedBatches`, and return without
-   *      throwing — the scheduler's transient-retry path would just repeat
-   *      the same deterministic failure.
-   *
-   * Non-parse errors propagate unchanged so the scheduler's existing
-   * backoff/retry behaviour still handles transient network or CH issues.
-   */
-  async #insertWithJsonParseRecovery<T extends object>(
-    flushId: string,
-    rows: T[],
-    doInsert: () => Promise<unknown>,
-    contextLabel: string
-  ): Promise<
-    | { kind: "inserted"; insertResult: unknown }
-    | { kind: "sanitized"; insertResult: unknown }
-    | { kind: "dropped" }
-  > {
-    try {
-      return { kind: "inserted", insertResult: await doInsert() };
-    } catch (firstError) {
-      if (!isClickHouseJsonParseError(firstError)) throw firstError;
-
-      const firstMessage =
-        typeof firstError === "object" && firstError !== null && "message" in firstError
-          ? String((firstError as { message?: unknown }).message ?? "")
-          : String(firstError);
-
-      // Sanitize the whole batch. ClickHouse's `at row N` index is logged
-      // for observability but not used to slice — its semantics under
-      // parallel parsing are not stable enough to safely skip rows.
-      const rowHint = parseRowNumberFromError(firstMessage);
-      const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
-
-      // Sanitizer found nothing to fix → retrying the exact same batch is
-      // guaranteed to hit the same deterministic parse failure. Skip the
-      // wasted ClickHouse round-trip and drop loudly. Throwing instead would
-      // hand the failure back to the scheduler's 3× transient-retry loop —
-      // exactly the retry storm this wrapper is designed to avoid.
-      if (fieldsSanitized === 0) {
-        this._permanentlyDroppedBatches += 1;
-        this._droppedBatchesCounter.add(1, { table: contextLabel });
-        logger.error(
-          "Dropped batch — ClickHouse JSON parse error but sanitizer found nothing to fix",
-          {
-            flushId,
-            contextLabel,
-            batchSize: rows.length,
-            clickhouseRowHint: rowHint,
-            permanentlyDroppedBatches: this._permanentlyDroppedBatches,
-            sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024),
-            clickhouseError: firstMessage.split("\n")[0],
-          }
-        );
-        return { kind: "dropped" };
-      }
+  #recordRecoveryOutcome(
+    outcome: JsonParseRecoveryOutcome,
+    contextLabel: string,
+    batchSize: number
+  ) {
+    if (outcome.kind !== "recovered") {
+      return;
+    }
 
-      logger.warn("Sanitizing batch after ClickHouse JSON parse error", {
-        flushId,
-        contextLabel,
-        batchSize: rows.length,
-        clickhouseRowHint: rowHint,
-        rowsTouched,
-        fieldsSanitized,
-        clickhouseError: firstMessage.split("\n")[0],
-      });
+    this._rowIsolationRecoveries += 1;
+    this._rowIsolatedBatchesCounter.add(1, { table: contextLabel });
+
+    if (outcome.capped) {
+      this._recoveryCapHits += 1;
+      this._recoveryCapHitsCounter.add(1, { table: contextLabel });
+    }
 
-      try {
-        return { kind: "sanitized", insertResult: await doInsert() };
-      } catch (retryError) {
-        if (!isClickHouseJsonParseError(retryError)) throw retryError;
+    if (outcome.rowsStripped > 0) {
+      this._rowsStripped += outcome.rowsStripped;
+      this._rowsStrippedCounter.add(outcome.rowsStripped, { table: contextLabel });
+    }
 
+    if (outcome.rowsDropped > 0) {
+      this._permanentlyDroppedRows += outcome.rowsDropped;
+      this._rowsDroppedCounter.add(outcome.rowsDropped, { table: contextLabel });
+      if (outcome.rowsDropped === batchSize) {
         this._permanentlyDroppedBatches += 1;
         this._droppedBatchesCounter.add(1, { table: contextLabel });
-        const retryMessage =
-          typeof retryError === "object" && retryError !== null && "message" in retryError
-            ? String((retryError as { message?: unknown }).message ?? "")
-            : String(retryError);
-        logger.error("Dropped batch after sanitize-retry still hit ClickHouse JSON parse error", {
-          flushId,
-          contextLabel,
-          batchSize: rows.length,
-          permanentlyDroppedBatches: this._permanentlyDroppedBatches,
-          sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024),
-          firstError: firstMessage.split("\n")[0],
-          retryError: retryMessage.split("\n")[0],
-        });
-
-        return { kind: "dropped" };
       }
     }
   }
diff --git a/apps/webapp/app/v3/eventRepository/sanitizeRowsOnParseError.server.ts b/apps/webapp/app/v3/eventRepository/sanitizeRowsOnParseError.server.ts
index 0374442200a..d2e87230f72 100644
--- a/apps/webapp/app/v3/eventRepository/sanitizeRowsOnParseError.server.ts
+++ b/apps/webapp/app/v3/eventRepository/sanitizeRowsOnParseError.server.ts
@@ -162,3 +162,270 @@ export function sanitizeRows<T extends object>(rows: T[]): SanitizeResult {
 
   return result;
 }
+
+export function errorMessage(err: unknown): string {
+  return typeof err === "object" && err !== null && "message" in err
+    ? String((err as { message?: unknown }).message ?? "")
+    : String(err);
+}
+
+export function rawErrorMessage(err: unknown): string {
+  if (typeof err === "object" && err !== null) {
+    const raw = (err as { rawMessage?: unknown }).rawMessage;
+    if (typeof raw === "string" && raw.length > 0) return raw;
+  }
+  return errorMessage(err);
+}
+
+export type JsonParseRecoveryLogger = {
+  info: (message: string, meta?: Record<string, unknown>) => void;
+  warn: (message: string, meta?: Record<string, unknown>) => void;
+};
+
+export type JsonParseRecoveryOutcome =
+  | { kind: "inserted"; insertResult: unknown }
+  | { kind: "sanitized"; insertResult: unknown }
+  | { kind: "recovered"; rowsStripped: number; rowsDropped: number; capped: boolean };
+
+/**
+ * Default number of poison rows to isolate-and-strip precisely before bailing
+ * to a single `allow_errors` skip insert. One covers the common case (a single
+ * un-ingestable run in a flush) exactly, keeping that run's status. The bound
+ * matters because run-replication flushes are large (thousands of rows in prod)
+ * and stripping re-sends the whole batch once per poison row: without a small
+ * limit, a burst of un-ingestable runs in one flush would re-parse a large
+ * batch many times on the shared ClickHouse server.
+ */
+export const DEFAULT_MAX_POISON_STRIPS = 1;
+
+/**
+ * ClickHouse insert recovery for `Cannot parse JSON object` rejections on the
+ * runs table, where the poison run should KEEP its status (its row lands with
+ * its JSON column emptied) rather than be dropped.
+ *
+ *   1. Try the insert. Healthy batches pay zero recovery cost.
+ *   2. On a parse error, `sanitizeRows` losslessly repairs what it can in place
+ *      (lone UTF-16 surrogates, out-of-range integers) and retries once.
+ *   3. If the sanitizer can't help, follow ClickHouse's `at row N` hint to the
+ *      un-ingestable row and re-insert with that row's JSON column(s) emptied
+ *      via `stripJsonColumns`, up to `maxPoisonStrips` rows. Each stripped run
+ *      still lands (keeps its terminal status); only its un-ingestable JSON is
+ *      lost. `insertSync` disables parallel parsing so `at row N` is reliable.
+ *   4. Cost bound: once `maxPoisonStrips` rows have been stripped and the batch
+ *      STILL fails (or the failing row can't be located), stop stripping and
+ *      land the batch with one `allow_errors` insert — the stripped rows and
+ *      every clean row land in a single pass and the remaining un-ingestable
+ *      rows are skipped. Recovery stays a fixed handful of inserts no matter how
+ *      large or poisoned the batch is (`capped` marks that the bail was taken).
+ *   5. Non-parse errors propagate unchanged.
+ */
+export async function insertWithLimitedStrip<T extends object>(params: {
+  rows: T[];
+  contextLabel: string;
+  logger: JsonParseRecoveryLogger;
+  logContext?: Record<string, unknown>;
+  insert: (rows: T[]) => Promise<unknown>;
+  insertSync: (rows: T[]) => Promise<unknown>;
+  insertAllowingBadRows: (rows: T[]) => Promise<unknown>;
+  stripJsonColumns: (row: T) => T;
+  maxPoisonStrips?: number;
+  hasMaterializedViews?: boolean;
+}): Promise<JsonParseRecoveryOutcome> {
+  const { rows, contextLabel, logger, logContext, insert, insertSync, insertAllowingBadRows } =
+    params;
+  const stripJsonColumns = params.stripJsonColumns;
+  const maxPoisonStrips = params.maxPoisonStrips ?? DEFAULT_MAX_POISON_STRIPS;
+  const hasMaterializedViews = params.hasMaterializedViews ?? true;
+
+  try {
+    return { kind: "inserted", insertResult: await insert(rows) };
+  } catch (firstError) {
+    if (!isClickHouseJsonParseError(firstError)) throw firstError;
+
+    const firstMessage = errorMessage(firstError);
+    const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
+
+    if (fieldsSanitized > 0) {
+      logger.warn("Sanitizing batch after ClickHouse JSON parse error", {
+        ...logContext,
+        contextLabel,
+        batchSize: rows.length,
+        rowsTouched,
+        fieldsSanitized,
+        clickhouseError: firstMessage.split("\n")[0],
+      });
+
+      try {
+        return { kind: "sanitized", insertResult: await insert(rows) };
+      } catch (retryError) {
+        if (!isClickHouseJsonParseError(retryError)) throw retryError;
+      }
+    }
+
+    const working = rows.slice();
+    const stripped = new Array(working.length).fill(false);
+    let rowsStripped = 0;
+
+    let guard = maxPoisonStrips + 2;
+    while (guard-- > 0) {
+      let parseError: unknown;
+      try {
+        await insertSync(working);
+        if (rowsStripped > 0) {
+          logger.info(
+            "Stripped un-ingestable rows after ClickHouse JSON parse error — batch landed with their JSON emptied",
+            {
+              ...logContext,
+              contextLabel,
+              batchSize: rows.length,
+              rowsStripped,
+              clickhouseError: firstMessage.split("\n")[0],
+            }
+          );
+        }
+        return { kind: "recovered", rowsStripped, rowsDropped: 0, capped: false };
+      } catch (error) {
+        if (!isClickHouseJsonParseError(error)) throw error;
+        parseError = error;
+      }
+
+      const hint = parseRowNumberFromError(rawErrorMessage(parseError));
+      const index = hint === null ? -1 : hint - 1;
+      const canStrip =
+        rowsStripped < maxPoisonStrips && index >= 0 && index < working.length && !stripped[index];
+
+      if (!canStrip) break;
+
+      working[index] = stripJsonColumns(working[index]);
+      stripped[index] = true;
+      rowsStripped += 1;
+    }
+
+    const insertResult = await insertAllowingBadRows(working);
+    const rowsDropped = droppedRowCount(insertResult, working.length, hasMaterializedViews);
+
+    logger.warn(
+      "Hit the poison-row strip limit — landed the batch via allow_errors and skipped the remaining un-ingestable rows",
+      {
+        ...logContext,
+        contextLabel,
+        batchSize: rows.length,
+        rowsStripped,
+        rowsDropped,
+        landedRows: writtenRowCount(insertResult),
+        clickhouseError: firstMessage.split("\n")[0],
+      }
+    );
+
+    return { kind: "recovered", rowsStripped, rowsDropped, capped: true };
+  }
+}
+
+function writtenRowCount(insertResult: unknown): number | null {
+  if (typeof insertResult === "object" && insertResult !== null) {
+    const summary = (insertResult as { summary?: { written_rows?: unknown } }).summary;
+    const written = summary?.written_rows;
+    if (typeof written === "number" && Number.isFinite(written)) return written;
+    if (typeof written === "string" && written.length > 0) {
+      const parsed = Number.parseInt(written, 10);
+      if (Number.isFinite(parsed)) return parsed;
+    }
+  }
+  return null;
+}
+
+/**
+ * Derives how many rows a `allow_errors` insert dropped, from the insert
+ * summary's `written_rows`. This is exact only when `written_rows` is a clean
+ * count of the target-table rows. On tables with row-multiplying materialized
+ * views, ClickHouse folds the MV-written rows into `written_rows` too, so a
+ * partial-drop count isn't derivable; the only reliable signal there is
+ * `written_rows === 0`, which means the whole batch was dropped (no base rows,
+ * so no MV rows either).
+ */
+function droppedRowCount(
+  insertResult: unknown,
+  batchSize: number,
+  hasMaterializedViews: boolean
+): number {
+  const written = writtenRowCount(insertResult);
+  if (written === null) return 0;
+  if (written === 0) return batchSize;
+  if (hasMaterializedViews) return 0;
+  return Math.max(0, batchSize - written);
+}
+
+/**
+ * Shared ClickHouse insert recovery that SKIPS un-ingestable rows, for the
+ * high-volume append-only tables (trace events, run payloads) where dropping a
+ * single un-ingestable row is acceptable and precise row isolation isn't worth
+ * its re-parse cost on the shared ClickHouse server.
+ *
+ *   1. Try the insert. Healthy batches pay zero recovery cost.
+ *   2. On a parse error, `sanitizeRows` losslessly repairs what it can in place
+ *      and retries once, so a repairable row still lands in full.
+ *   3. If the sanitizer can't help, re-insert once with ClickHouse's
+ *      `input_format_allow_errors_*` so the good rows land in a single pass and
+ *      only the un-ingestable rows are skipped. A poison flood costs one extra
+ *      insert regardless of how many rows are bad.
+ *   4. Non-parse errors propagate unchanged.
+ *
+ * The batch-level recovery is always counted by the caller; the per-row dropped
+ * count is exact on tables without row-multiplying materialized views and
+ * whole-batch-only on tables that have them (see `droppedRowCount`).
+ */
+export async function insertWithBadRowSkip<T extends object>(params: {
+  rows: T[];
+  contextLabel: string;
+  logger: JsonParseRecoveryLogger;
+  logContext?: Record<string, unknown>;
+  insert: (rows: T[]) => Promise<unknown>;
+  insertAllowingBadRows: (rows: T[]) => Promise<unknown>;
+  hasMaterializedViews?: boolean;
+}): Promise<JsonParseRecoveryOutcome> {
+  const { rows, contextLabel, logger, logContext, insert, insertAllowingBadRows } = params;
+  const hasMaterializedViews = params.hasMaterializedViews ?? true;
+
+  try {
+    return { kind: "inserted", insertResult: await insert(rows) };
+  } catch (firstError) {
+    if (!isClickHouseJsonParseError(firstError)) throw firstError;
+
+    const firstMessage = errorMessage(firstError);
+    const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
+
+    if (fieldsSanitized > 0) {
+      logger.warn("Sanitizing batch after ClickHouse JSON parse error", {
+        ...logContext,
+        contextLabel,
+        batchSize: rows.length,
+        rowsTouched,
+        fieldsSanitized,
+        clickhouseError: firstMessage.split("\n")[0],
+      });
+
+      try {
+        return { kind: "sanitized", insertResult: await insert(rows) };
+      } catch (retryError) {
+        if (!isClickHouseJsonParseError(retryError)) throw retryError;
+      }
+    }
+
+    const insertResult = await insertAllowingBadRows(rows);
+    const rowsDropped = droppedRowCount(insertResult, rows.length, hasMaterializedViews);
+
+    logger.info(
+      "Skipped un-ingestable rows after ClickHouse JSON parse error — landed the rest of the batch",
+      {
+        ...logContext,
+        contextLabel,
+        batchSize: rows.length,
+        rowsDropped,
+        landedRows: writtenRowCount(insertResult),
+        clickhouseError: firstMessage.split("\n")[0],
+      }
+    );
+
+    return { kind: "recovered", rowsStripped: 0, rowsDropped, capped: false };
+  }
+}
diff --git a/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts b/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts
new file mode 100644
index 00000000000..695046965ae
--- /dev/null
+++ b/apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts
@@ -0,0 +1,126 @@
+import { ClickHouse, type TaskEventV2Input } from "@internal/clickhouse";
+import { clickhouseTest } from "@internal/testcontainers";
+import { describe, expect } from "vitest";
+import { z } from "zod";
+import {
+  ClickhouseEventRepository,
+  convertDateToClickhouseDateTime,
+} from "~/v3/eventRepository/clickhouseEventRepository.server";
+
+const TIMEOUT_MS = 60_000;
+
+function deeplyNested(depth: number): Record<string, unknown> {
+  let node: Record<string, unknown> = { leaf: 1 };
+  for (let i = 0; i < depth; i++) {
+    node = { [`k${i}`]: node };
+  }
+  return node;
+}
+
+function startTime(baseMs: number, offsetMs: number): string {
+  const ns = ((BigInt(baseMs) + BigInt(offsetMs)) * 1_000_000n).toString();
+  return `${ns.substring(0, 10)}.${ns.substring(10)}`;
+}
+
+describe("ClickhouseEventRepository JSON parse recovery", () => {
+  clickhouseTest(
+    "lands the good events and skips the poison event when one event has ClickHouse-unparseable attributes",
+    async ({ clickhouseContainer }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        logLevel: "error",
+      });
+
+      const repository = new ClickhouseEventRepository({
+        clickhouse,
+        version: "v2",
+        insertStrategy: "insert",
+        batchSize: 100,
+        flushInterval: 200,
+      });
+
+      const environmentId = "env_json_recovery_test";
+      const organizationId = "org_json_recovery_test";
+      const projectId = "proj_json_recovery_test";
+      const traceId = "b".repeat(32);
+      const baseMs = Date.now();
+      const expiresAt = convertDateToClickhouseDateTime(
+        new Date(baseMs + 365 * 24 * 60 * 60 * 1000)
+      );
+
+      function makeRow(i: number, attributes: unknown): TaskEventV2Input {
+        return {
+          environment_id: environmentId,
+          organization_id: organizationId,
+          project_id: projectId,
+          task_identifier: "json-recovery-task",
+          run_id: `run_${i}`,
+          start_time: startTime(baseMs, i),
+          duration: "1000000",
+          trace_id: traceId,
+          span_id: `span_recovery_${String(i).padStart(6, "0")}`,
+          parent_span_id: "",
+          message: `event ${i}`,
+          kind: "SPAN",
+          status: "OK",
+          attributes,
+          metadata: "{}",
+          expires_at: expiresAt,
+        };
+      }
+
+      const goodSpanIds: string[] = [];
+      let poisonSpanId = "";
+      const rows: TaskEventV2Input[] = [];
+      for (let i = 0; i < 5; i++) {
+        const isPoison = i === 2;
+        const row = makeRow(i, isPoison ? deeplyNested(1500) : { ok: true, i });
+        rows.push(row);
+        if (isPoison) {
+          poisonSpanId = row.span_id;
+        } else {
+          goodSpanIds.push(row.span_id);
+        }
+      }
+
+      try {
+        (repository as any).addToBatch(rows);
+
+        const queryEvents = clickhouse.reader.query({
+          name: "event-recovery-check",
+          query:
+            "SELECT span_id, toJSONString(attributes) AS attributes_json FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}",
+          schema: z.object({ span_id: z.string(), attributes_json: z.string() }),
+          params: z.object({ env_id: z.string() }),
+        });
+
+        const rowsById = await vi.waitFor(
+          async () => {
+            const [queryError, resultRows] = await queryEvents({ env_id: environmentId });
+            expect(queryError).toBeNull();
+            const byId = new Map((resultRows ?? []).map((r) => [r.span_id, r]));
+            for (const id of goodSpanIds) {
+              expect(byId.has(id)).toBe(true);
+            }
+            return byId;
+          },
+          { timeout: 30_000, interval: 250 }
+        );
+
+        for (const id of goodSpanIds) {
+          expect(rowsById.get(id)!.attributes_json).toContain('"ok":true');
+        }
+
+        expect(rowsById.has(poisonSpanId)).toBe(false);
+
+        expect(repository.permanentlyDroppedBatches).toBe(0);
+        expect(repository.permanentlyDroppedRows).toBe(0);
+        expect(repository.rowIsolationRecoveries).toBeGreaterThanOrEqual(1);
+        expect(repository.rowsStripped).toBe(0);
+      } finally {
+        await (repository as any)._flushScheduler?.shutdown?.();
+      }
+    },
+    TIMEOUT_MS
+  );
+});
diff --git a/apps/webapp/test/runsReplicationBenchmark.producer.ts b/apps/webapp/test/runsReplicationBenchmark.producer.ts
index 547d1318c46..3f34d5c9f80 100644
--- a/apps/webapp/test/runsReplicationBenchmark.producer.ts
+++ b/apps/webapp/test/runsReplicationBenchmark.producer.ts
@@ -15,6 +15,17 @@ interface ProducerConfig {
   numRuns: number;
   errorRate: number; // 0.07 = 7%
   batchSize: number;
+  poisonRate?: number;
+  poisonDepth?: number;
+}
+
+function deeplyNestedJson(depth: number): string {
+  const safeDepth = Number.isSafeInteger(depth) && depth >= 0 ? depth : 0;
+  const open: string[] = [];
+  for (let i = 0; i < safeDepth; i++) {
+    open.push(`{"k${i}":`);
+  }
+  return open.join("") + '{"leaf":1}' + "}".repeat(safeDepth);
 }
 
 // Error templates for realistic variety
@@ -108,6 +119,9 @@ async function runProducer(config: ProducerConfig) {
     const startTime = performance.now();
     let created = 0;
     let withErrors = 0;
+    let poisoned = 0;
+    const poisonRate = config.poisonRate ?? 0;
+    const poisonDepth = config.poisonDepth ?? 1500;
 
     // Process in batches to avoid overwhelming the database
     for (let batch = 0; batch < Math.ceil(config.numRuns / config.batchSize); batch++) {
@@ -120,6 +134,8 @@ async function runProducer(config: ProducerConfig) {
         const hasError = Math.random() < config.errorRate;
         const status = hasError ? "COMPLETED_WITH_ERRORS" : "COMPLETED_SUCCESSFULLY";
 
+        const isPoison = Math.random() < poisonRate;
+
         const runData: any = {
           friendlyId: `run_bench_${Date.now()}_${i}`,
           taskIdentifier: `benchmark-task-${i % 10}`, // Vary task identifiers
@@ -142,6 +158,12 @@ async function runProducer(config: ProducerConfig) {
           withErrors++;
         }
 
+        if (isPoison) {
+          runData.output = deeplyNestedJson(poisonDepth);
+          runData.outputType = "application/json";
+          poisoned++;
+        }
+
         runs.push(runData);
       }
 
@@ -168,6 +190,7 @@ async function runProducer(config: ProducerConfig) {
     console.log(`[Producer] Completed:`);
     console.log(`  - Total runs: ${created}`);
     console.log(`  - With errors: ${withErrors} (${((withErrors / created) * 100).toFixed(1)}%)`);
+    console.log(`  - Poisoned: ${poisoned} (${((poisoned / created) * 100).toFixed(1)}%)`);
     console.log(`  - Duration: ${duration.toFixed(0)}ms`);
     console.log(`  - Throughput: ${throughput.toFixed(0)} runs/sec`);
 
@@ -178,6 +201,7 @@ async function runProducer(config: ProducerConfig) {
         stats: {
           created,
           withErrors,
+          poisoned,
           duration,
           throughput,
         },
diff --git a/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts b/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts
new file mode 100644
index 00000000000..7e97dba5d14
--- /dev/null
+++ b/apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts
@@ -0,0 +1,287 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { replicationContainerTest } from "@internal/testcontainers";
+import { fork } from "node:child_process";
+import path from "node:path";
+import { performance } from "node:perf_hooks";
+import { setTimeout } from "node:timers/promises";
+import { z } from "zod";
+import { RunsReplicationService } from "~/services/runsReplicationService.server";
+import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
+import { createInMemoryMetrics, createInMemoryTracing } from "./utils/tracing";
+
+vi.setConfig({ testTimeout: 300_000 });
+
+const CONFIG = {
+  NUM_RUNS: parseInt(process.env.BENCHMARK_NUM_RUNS || "5000", 10),
+  PRODUCER_BATCH_SIZE: 100,
+  FLUSH_BATCH_SIZE: 50,
+  FLUSH_INTERVAL_MS: 100,
+  MAX_FLUSH_CONCURRENCY: 4,
+  REPLICATION_TIMEOUT_MS: 180_000,
+  POISON_RATE: parseFloat(process.env.BENCHMARK_POISON_RATE || "0.05"),
+};
+
+class ELUMonitor {
+  private samples: number[] = [];
+  private interval: NodeJS.Timeout | null = null;
+
+  start(intervalMs = 100) {
+    this.samples = [];
+    let last = performance.eventLoopUtilization();
+    this.interval = setInterval(() => {
+      const current = performance.eventLoopUtilization();
+      this.samples.push(performance.eventLoopUtilization(current, last).utilization * 100);
+      last = current;
+    }, intervalMs);
+  }
+
+  stop() {
+    if (this.interval) {
+      clearInterval(this.interval);
+      this.interval = null;
+    }
+    if (this.samples.length === 0) return { mean: 0, p50: 0, p95: 0, p99: 0, samples: 0 };
+    const sorted = [...this.samples].sort((a, b) => a - b);
+    const mean = sorted.reduce((s, v) => s + v, 0) / sorted.length;
+    return {
+      mean,
+      p50: sorted[Math.floor(sorted.length * 0.5)],
+      p95: sorted[Math.floor(sorted.length * 0.95)],
+      p99: sorted[Math.floor(sorted.length * 0.99)],
+      samples: sorted.length,
+    };
+  }
+}
+
+function runProducer(config: {
+  postgresUrl: string;
+  organizationId: string;
+  projectId: string;
+  environmentId: string;
+  numRuns: number;
+  errorRate: number;
+  batchSize: number;
+  poisonRate: number;
+  poisonDepth: number;
+}): Promise<{ created: number; withErrors: number; poisoned: number; duration: number }> {
+  return new Promise((resolve, reject) => {
+    const producerPath = path.join(__dirname, "runsReplicationBenchmark.producer.ts");
+    const child = fork(producerPath, [JSON.stringify(config)], {
+      stdio: ["ignore", "pipe", "pipe", "ipc"],
+      execArgv: ["-r", "tsx/cjs"],
+    });
+    child.stdout?.on("data", (d) => console.log(d.toString().trim()));
+    child.stderr?.on("data", (d) => console.error(d.toString().trim()));
+    child.on("message", (m: any) => {
+      if (m.type === "complete") resolve(m.stats);
+      else if (m.type === "error") reject(new Error(m.error));
+    });
+    child.on("error", reject);
+    child.on("exit", (code) => {
+      if (code !== 0) reject(new Error(`Producer exited with code ${code}`));
+    });
+  });
+}
+
+async function waitForCount(
+  clickhouse: ClickHouse,
+  organizationId: string,
+  expectedCount: number,
+  timeoutMs: number
+): Promise<{ duration: number; count: number }> {
+  const startTime = performance.now();
+  const deadline = startTime + timeoutMs;
+  const queryRuns = clickhouse.reader.query({
+    name: "bench-count",
+    query:
+      "SELECT count(*) as count FROM trigger_dev.task_runs_v2 WHERE organization_id = {org_id:String}",
+    schema: z.object({ count: z.number() }),
+    params: z.object({ org_id: z.string() }),
+  });
+  while (performance.now() < deadline) {
+    const [error, result] = await queryRuns({ org_id: organizationId });
+    if (error) throw new Error(`Failed to query ClickHouse: ${error.message}`);
+    const count = result?.[0]?.count || 0;
+    if (count >= expectedCount) return { duration: performance.now() - startTime, count };
+    await setTimeout(500);
+  }
+  const [, result] = await queryRuns({ org_id: organizationId });
+  return { duration: performance.now() - startTime, count: result?.[0]?.count || 0 };
+}
+
+async function runScenario(
+  name: string,
+  poisonRate: number,
+  ctx: { clickhouseContainer: any; redisOptions: any; postgresContainer: any; prisma: any }
+) {
+  const { clickhouseContainer, redisOptions, postgresContainer, prisma } = ctx;
+
+  const organization = await prisma.organization.create({
+    data: { title: `bench-${name}`, slug: `bench-${name}` },
+  });
+  const project = await prisma.project.create({
+    data: {
+      name: `bench-${name}`,
+      slug: `bench-${name}`,
+      organizationId: organization.id,
+      externalRef: `bench-${name}`,
+    },
+  });
+  const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+    data: {
+      slug: `bench-${name}`,
+      type: "DEVELOPMENT",
+      projectId: project.id,
+      organizationId: organization.id,
+      apiKey: `bench-${name}`,
+      pkApiKey: `bench-${name}`,
+      shortcode: `bench-${name}`,
+    },
+  });
+
+  const clickhouse = new ClickHouse({
+    url: clickhouseContainer.getConnectionUrl(),
+    name: `bench-${name}`,
+    compression: { request: true },
+    logLevel: "error",
+  });
+
+  const { tracer } = createInMemoryTracing();
+  const metricsHelper = createInMemoryMetrics();
+
+  const service = new RunsReplicationService({
+    clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse),
+    pgConnectionUrl: postgresContainer.getConnectionUri(),
+    serviceName: `bench-${name}`,
+    slotName: `bench_${name.replace(/-/g, "_")}`,
+    publicationName: `bench_${name.replace(/-/g, "_")}_pub`,
+    redisOptions,
+    maxFlushConcurrency: CONFIG.MAX_FLUSH_CONCURRENCY,
+    flushIntervalMs: CONFIG.FLUSH_INTERVAL_MS,
+    flushBatchSize: CONFIG.FLUSH_BATCH_SIZE,
+    maxPoisonStripsPerBatch: CONFIG.NUM_RUNS,
+    leaderLockTimeoutMs: 10000,
+    leaderLockExtendIntervalMs: 2000,
+    ackIntervalSeconds: 10,
+    tracer,
+    meter: metricsHelper.meter,
+    logLevel: "error",
+  });
+
+  await service.start();
+
+  const elu = new ELUMonitor();
+  elu.start(100);
+
+  let producerStats!: { created: number; withErrors: number; poisoned: number; duration: number };
+  let replication!: { duration: number; count: number };
+  let eluStats!: ReturnType<ELUMonitor["stop"]>;
+
+  try {
+    producerStats = await runProducer({
+      postgresUrl: postgresContainer.getConnectionUri(),
+      organizationId: organization.id,
+      projectId: project.id,
+      environmentId: runtimeEnvironment.id,
+      numRuns: CONFIG.NUM_RUNS,
+      errorRate: 0.07,
+      batchSize: CONFIG.PRODUCER_BATCH_SIZE,
+      poisonRate,
+      poisonDepth: 1500,
+    });
+
+    const expectedLanded = producerStats.created;
+    replication = await waitForCount(
+      clickhouse,
+      organization.id,
+      expectedLanded,
+      CONFIG.REPLICATION_TIMEOUT_MS
+    );
+  } finally {
+    eluStats = elu.stop();
+    await service.stop();
+    await metricsHelper.shutdown();
+  }
+
+  const throughput = (replication.count / replication.duration) * 1000;
+  const expectedLanded = producerStats.created;
+
+  console.log(`\n${"=".repeat(72)}`);
+  console.log(`SCENARIO: ${name}  (poison rate ${(poisonRate * 100).toFixed(1)}%)`);
+  console.log(`${"=".repeat(72)}`);
+  console.log(`Produced:     ${producerStats.created} runs (${producerStats.poisoned} poisoned)`);
+  console.log(`Landed in CH: ${replication.count} / expected ${expectedLanded}`);
+  console.log(`Repl dur:     ${replication.duration.toFixed(0)}ms`);
+  console.log(`Throughput:   ${throughput.toFixed(0)} runs/sec`);
+  console.log(`Row-isolation recoveries: ${service.rowIsolationRecoveries}`);
+  console.log(`Recovery cap hits: ${service.recoveryCapHits}`);
+  console.log(`Rows stripped (kept row, lost JSON): ${service.rowsStripped}`);
+  console.log(`Rows dropped (unrecoverable): ${service.permanentlyDroppedRows}`);
+  console.log(`Dropped batches (whole): ${service.permanentlyDroppedBatches}`);
+  console.log(
+    `ELU mean=${eluStats.mean.toFixed(2)}% p50=${eluStats.p50.toFixed(2)}% p95=${eluStats.p95.toFixed(2)}% p99=${eluStats.p99.toFixed(2)}% (${eluStats.samples} samples)`
+  );
+  console.log(`${"=".repeat(72)}\n`);
+
+  return {
+    name,
+    poisonRate,
+    producerStats,
+    replication,
+    eluStats,
+    throughput,
+    expectedLanded,
+    service,
+  };
+}
+
+describe("RunsReplicationService JSON-recovery ELU benchmark", () => {
+  replicationContainerTest.skipIf(process.env.BENCHMARKS_ENABLED !== "1")(
+    "measures ELU on the healthy hot path vs the row-isolation recovery path",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const healthy = await runScenario("healthy-0pct", 0, {
+        clickhouseContainer,
+        redisOptions,
+        postgresContainer,
+        prisma,
+      });
+
+      const poisoned = await runScenario("poison", CONFIG.POISON_RATE, {
+        clickhouseContainer,
+        redisOptions,
+        postgresContainer,
+        prisma,
+      });
+
+      const eluMeanDelta =
+        ((poisoned.eluStats.mean - healthy.eluStats.mean) / healthy.eluStats.mean) * 100;
+      const eluP99Delta =
+        ((poisoned.eluStats.p99 - healthy.eluStats.p99) / healthy.eluStats.p99) * 100;
+
+      console.log(`\n${"=".repeat(72)}`);
+      console.log("COMPARISON  (healthy 0% -> poison)");
+      console.log(`${"=".repeat(72)}`);
+      console.log(
+        `ELU mean: ${healthy.eluStats.mean.toFixed(2)}% -> ${poisoned.eluStats.mean.toFixed(2)}% (${eluMeanDelta > 0 ? "+" : ""}${eluMeanDelta.toFixed(1)}%)`
+      );
+      console.log(
+        `ELU p99:  ${healthy.eluStats.p99.toFixed(2)}% -> ${poisoned.eluStats.p99.toFixed(2)}% (${eluP99Delta > 0 ? "+" : ""}${eluP99Delta.toFixed(1)}%)`
+      );
+      console.log(`${"=".repeat(72)}\n`);
+
+      expect(healthy.replication.count).toBe(healthy.expectedLanded);
+      expect(healthy.service.rowIsolationRecoveries).toBe(0);
+      expect(healthy.service.rowsStripped).toBe(0);
+      expect(healthy.service.permanentlyDroppedBatches).toBe(0);
+
+      expect(poisoned.replication.count).toBe(poisoned.expectedLanded);
+      expect(poisoned.service.permanentlyDroppedBatches).toBe(0);
+      expect(poisoned.service.permanentlyDroppedRows).toBe(0);
+      expect(poisoned.service.recoveryCapHits).toBe(0);
+      expect(poisoned.service.rowIsolationRecoveries).toBeGreaterThanOrEqual(1);
+      expect(poisoned.service.rowsStripped).toBe(poisoned.producerStats.poisoned);
+    }
+  );
+});
diff --git a/apps/webapp/test/runsReplicationService.part10.test.ts b/apps/webapp/test/runsReplicationService.part10.test.ts
new file mode 100644
index 00000000000..13f20097025
--- /dev/null
+++ b/apps/webapp/test/runsReplicationService.part10.test.ts
@@ -0,0 +1,229 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { replicationContainerTest } from "@internal/testcontainers";
+import { z } from "zod";
+import { RunsReplicationService } from "~/services/runsReplicationService.server";
+import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
+import { createInMemoryTracing } from "./utils/tracing";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+function deeplyNested(depth: number): Record<string, unknown> {
+  let node: Record<string, unknown> = { leaf: 1 };
+  for (let i = 0; i < depth; i++) {
+    node = { [`k${i}`]: node };
+  }
+  return node;
+}
+
+function createService(
+  clickhouse: ClickHouse,
+  postgresUrl: string,
+  redisOptions: any,
+  flushIntervalMs = 500
+) {
+  const { tracer } = createInMemoryTracing();
+  return new RunsReplicationService({
+    clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse),
+    pgConnectionUrl: postgresUrl,
+    serviceName: "runs-replication",
+    slotName: "task_runs_to_clickhouse_v1",
+    publicationName: "task_runs_to_clickhouse_v1_publication",
+    redisOptions,
+    maxFlushConcurrency: 1,
+    flushIntervalMs,
+    flushBatchSize: 50,
+    leaderLockTimeoutMs: 5000,
+    leaderLockExtendIntervalMs: 1000,
+    ackIntervalSeconds: 5,
+    tracer,
+    logLevel: "warn",
+  });
+}
+
+async function setupProject(prisma: any) {
+  const organization = await prisma.organization.create({ data: { title: "test", slug: "test" } });
+  const project = await prisma.project.create({
+    data: { name: "test", slug: "test", organizationId: organization.id, externalRef: "test" },
+  });
+  const runtimeEnvironment = await prisma.runtimeEnvironment.create({
+    data: {
+      slug: "test",
+      type: "DEVELOPMENT",
+      projectId: project.id,
+      organizationId: organization.id,
+      apiKey: "test",
+      pkApiKey: "test",
+      shortcode: "test",
+    },
+  });
+  return { organization, project, runtimeEnvironment };
+}
+
+async function createRun(
+  prisma: any,
+  ctx: { organization: any; project: any; runtimeEnvironment: any },
+  i: number,
+  isPoison: boolean
+) {
+  return prisma.taskRun.create({
+    data: {
+      friendlyId: `run_batchdrop_${i}`,
+      taskIdentifier: "my-task",
+      payload: JSON.stringify({ i }),
+      payloadType: "application/json",
+      output: isPoison ? JSON.stringify(deeplyNested(1500)) : JSON.stringify({ ok: true, i }),
+      outputType: "application/json",
+      traceId: `trace_${i}`,
+      spanId: `span_${i}`,
+      queue: "test",
+      status: "COMPLETED_SUCCESSFULLY",
+      runtimeEnvironmentId: ctx.runtimeEnvironment.id,
+      projectId: ctx.project.id,
+      organizationId: ctx.organization.id,
+      environmentType: "DEVELOPMENT",
+      engine: "V2",
+    },
+  });
+}
+
+describe("RunsReplicationService (part 10/10) — JSON parse recovery", () => {
+  replicationContainerTest(
+    "strips a single poison run and lands every run (poison run keeps its status, output stripped)",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-replication",
+        compression: { request: true },
+        logLevel: "warn",
+      });
+
+      const runsReplicationService = createService(
+        clickhouse,
+        postgresContainer.getConnectionUri(),
+        redisOptions
+      );
+      await runsReplicationService.start();
+
+      const ctx = await setupProject(prisma);
+
+      const goodRunIds: string[] = [];
+      let poisonRunId = "";
+      for (let i = 0; i < 5; i++) {
+        const isPoison = i === 2;
+        const run = await createRun(prisma, ctx, i, isPoison);
+        if (isPoison) poisonRunId = run.id;
+        else goodRunIds.push(run.id);
+      }
+
+      const queryRuns = clickhouse.reader.query({
+        name: "runs-replication-batchdrop",
+        query:
+          "SELECT run_id, status, toJSONString(output) AS output_json FROM trigger_dev.task_runs_v2 FINAL WHERE organization_id = {org_id:String}",
+        schema: z.object({ run_id: z.string(), status: z.string(), output_json: z.string() }),
+        params: z.object({ org_id: z.string() }),
+      });
+
+      const rowsById = await vi.waitFor(
+        async () => {
+          const [queryError, rows] = await queryRuns({ org_id: ctx.organization.id });
+          expect(queryError).toBeNull();
+          const byId = new Map((rows ?? []).map((r) => [r.run_id, r]));
+          for (const id of [...goodRunIds, poisonRunId]) {
+            expect(byId.has(id)).toBe(true);
+          }
+          return byId;
+        },
+        { timeout: 30_000, interval: 250 }
+      );
+
+      for (const id of goodRunIds) {
+        expect(rowsById.get(id)!.output_json).toContain('"ok":true');
+      }
+
+      const poison = rowsById.get(poisonRunId)!;
+      expect(poison.status).toBe("COMPLETED_SUCCESSFULLY");
+      expect(poison.output_json).toBe("{}");
+
+      expect(runsReplicationService.permanentlyDroppedBatches).toBe(0);
+      expect(runsReplicationService.permanentlyDroppedRows).toBe(0);
+      expect(runsReplicationService.recoveryCapHits).toBe(0);
+      expect(runsReplicationService.rowIsolationRecoveries).toBeGreaterThanOrEqual(1);
+      expect(runsReplicationService.rowsStripped).toBeGreaterThanOrEqual(1);
+
+      await runsReplicationService.stop();
+    }
+  );
+
+  replicationContainerTest(
+    "strips up to the limit then skips the excess poison via allow_errors (2 poison, limit 1)",
+    async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
+      await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
+
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-replication",
+        compression: { request: true },
+        logLevel: "warn",
+      });
+
+      const runsReplicationService = createService(
+        clickhouse,
+        postgresContainer.getConnectionUri(),
+        redisOptions,
+        3000
+      );
+      await runsReplicationService.start();
+
+      const ctx = await setupProject(prisma);
+
+      const goodRunIds: string[] = [];
+      const poisonRunIds: string[] = [];
+      for (let i = 0; i < 5; i++) {
+        const isPoison = i === 1 || i === 3;
+        const run = await createRun(prisma, ctx, i, isPoison);
+        if (isPoison) poisonRunIds.push(run.id);
+        else goodRunIds.push(run.id);
+      }
+
+      const queryRuns = clickhouse.reader.query({
+        name: "runs-replication-bail",
+        query:
+          "SELECT run_id, status, toJSONString(output) AS output_json FROM trigger_dev.task_runs_v2 FINAL WHERE organization_id = {org_id:String}",
+        schema: z.object({ run_id: z.string(), status: z.string(), output_json: z.string() }),
+        params: z.object({ org_id: z.string() }),
+      });
+
+      const rowsById = await vi.waitFor(
+        async () => {
+          const [queryError, rows] = await queryRuns({ org_id: ctx.organization.id });
+          expect(queryError).toBeNull();
+          const byId = new Map((rows ?? []).map((r) => [r.run_id, r]));
+          for (const id of goodRunIds) {
+            expect(byId.has(id)).toBe(true);
+          }
+          expect(runsReplicationService.recoveryCapHits).toBeGreaterThanOrEqual(1);
+          return byId;
+        },
+        { timeout: 30_000, interval: 250 }
+      );
+
+      for (const id of goodRunIds) {
+        expect(rowsById.get(id)!.output_json).toContain('"ok":true');
+      }
+
+      const landedPoison = poisonRunIds.filter((id) => rowsById.has(id));
+      expect(landedPoison).toHaveLength(1);
+      const strippedPoison = rowsById.get(landedPoison[0]!)!;
+      expect(strippedPoison.status).toBe("COMPLETED_SUCCESSFULLY");
+      expect(strippedPoison.output_json).toBe("{}");
+
+      expect(runsReplicationService.permanentlyDroppedBatches).toBe(0);
+      expect(runsReplicationService.rowsStripped).toBeGreaterThanOrEqual(1);
+      expect(runsReplicationService.recoveryCapHits).toBeGreaterThanOrEqual(1);
+
+      await runsReplicationService.stop();
+    }
+  );
+});
diff --git a/apps/webapp/test/sanitizeRowsOnParseError.test.ts b/apps/webapp/test/sanitizeRowsOnParseError.test.ts
index e3353176624..5d3bd4e7cfa 100644
--- a/apps/webapp/test/sanitizeRowsOnParseError.test.ts
+++ b/apps/webapp/test/sanitizeRowsOnParseError.test.ts
@@ -1,6 +1,8 @@
 import { describe, it, expect } from "vitest";
 import {
   INVALID_UTF16_SENTINEL,
+  insertWithBadRowSkip,
+  insertWithLimitedStrip,
   isClickHouseJsonParseError,
   parseRowNumberFromError,
   sanitizeRows,
@@ -10,6 +12,58 @@ import {
 const HIGH_SURROGATE = "\uD800";
 const LOW_SURROGATE = "\uDC00";
 
+type FakeRow = {
+  id: number;
+  poison?: boolean;
+  unstrippable?: boolean;
+  stripped?: boolean;
+  msg?: string;
+};
+
+const silentLogger = { info: () => {}, warn: () => {} };
+
+function parseErrorAtRow(oneBasedRow: number) {
+  return new Error(
+    `Cannot parse JSON object here: {...}: (at row ${oneBasedRow})\n: While executing ParallelParsingBlockInputFormat.`
+  );
+}
+
+/**
+ * Builds insert doubles for `insertWithLimitedStrip`: an insert that fails at
+ * the first still-poison row (reporting ClickHouse's 1-based `at row N`), an
+ * allow-bad-rows insert that lands only the good rows and reports the landed
+ * count in a summary, and a strip that clears the poison marker (but cannot fix
+ * an `unstrippable` row).
+ */
+function makeHarness() {
+  const landed: FakeRow[] = [];
+  let insertCalls = 0;
+  let allowCalls = 0;
+  const insertSync = async (rows: FakeRow[]) => {
+    insertCalls += 1;
+    const badIndex = rows.findIndex((r) => r.poison || r.unstrippable);
+    if (badIndex >= 0) throw parseErrorAtRow(badIndex + 1);
+    landed.push(...rows);
+    return { summary: { written_rows: String(rows.length) } };
+  };
+  const insertAllowingBadRows = async (rows: FakeRow[]) => {
+    allowCalls += 1;
+    const good = rows.filter((r) => !(r.poison || r.unstrippable));
+    landed.push(...good);
+    return { summary: { written_rows: String(good.length) } };
+  };
+  const stripJsonColumns = (row: FakeRow): FakeRow =>
+    row.unstrippable ? row : { ...row, poison: false, stripped: true };
+  return {
+    landed,
+    insertSync,
+    insertAllowingBadRows,
+    stripJsonColumns,
+    insertCalls: () => insertCalls,
+    allowCalls: () => allowCalls,
+  };
+}
+
 describe("isClickHouseJsonParseError", () => {
   it("recognises ClickHouse's parse-error string", () => {
     const err = new Error(
@@ -294,3 +348,324 @@ describe("sanitizeRows", () => {
     expect(rows[1].attributes.bigint).toBe("-100000000000000000000");
   });
 });
+
+describe("insertWithLimitedStrip", () => {
+  const clean = (id: number): FakeRow => ({ id });
+
+  it("inserts a healthy batch with no recovery", async () => {
+    const { landed, insertSync, insertAllowingBadRows, stripJsonColumns, allowCalls } =
+      makeHarness();
+    const rows = [clean(0), clean(1), clean(2)];
+
+    const outcome = await insertWithLimitedStrip({
+      rows,
+      contextLabel: "test",
+      logger: silentLogger,
+      insert: insertSync,
+      insertSync,
+      insertAllowingBadRows,
+      stripJsonColumns,
+    });
+
+    expect(outcome.kind).toBe("inserted");
+    expect(landed).toHaveLength(3);
+    expect(allowCalls()).toBe(0);
+  });
+
+  it("strips a single poison row and lands the rest in full without bailing", async () => {
+    const { landed, insertSync, insertAllowingBadRows, stripJsonColumns, allowCalls } =
+      makeHarness();
+    const rows = [clean(0), clean(1), { id: 2, poison: true }, clean(3), clean(4)];
+
+    const outcome = await insertWithLimitedStrip({
+      rows,
+      contextLabel: "test",
+      logger: silentLogger,
+      insert: insertSync,
+      insertSync,
+      insertAllowingBadRows,
+      stripJsonColumns,
+    });
+
+    expect(outcome).toEqual({ kind: "recovered", rowsStripped: 1, rowsDropped: 0, capped: false });
+    expect(landed.map((r) => r.id).sort((a, b) => a - b)).toEqual([0, 1, 2, 3, 4]);
+    expect(landed.find((r) => r.id === 2)?.stripped).toBe(true);
+    expect(landed.filter((r) => r.id !== 2).every((r) => !r.stripped)).toBe(true);
+    expect(allowCalls()).toBe(0);
+  });
+
+  it("strips up to the limit, then bails to allow_errors and skips the excess poison", async () => {
+    const { landed, insertSync, insertAllowingBadRows, stripJsonColumns, allowCalls } =
+      makeHarness();
+    const rows = [clean(0), { id: 1, poison: true }, clean(2), { id: 3, poison: true }, clean(4)];
+
+    const outcome = await insertWithLimitedStrip({
+      rows,
+      contextLabel: "test",
+      logger: silentLogger,
+      insert: insertSync,
+      insertSync,
+      insertAllowingBadRows,
+      stripJsonColumns,
+      maxPoisonStrips: 1,
+      hasMaterializedViews: false,
+    });
+
+    expect(outcome).toEqual({ kind: "recovered", rowsStripped: 1, rowsDropped: 1, capped: true });
+    expect(allowCalls()).toBe(1);
+    expect(landed.map((r) => r.id).sort((a, b) => a - b)).toEqual([0, 1, 2, 4]);
+    expect(landed.find((r) => r.id === 1)?.stripped).toBe(true);
+    expect(landed.some((r) => r.id === 3)).toBe(false);
+  });
+
+  it("strips every poison row when the limit is high enough", async () => {
+    const { landed, insertSync, insertAllowingBadRows, stripJsonColumns, allowCalls } =
+      makeHarness();
+    const rows = [clean(0), { id: 1, poison: true }, { id: 2, poison: true }, clean(3)];
+
+    const outcome = await insertWithLimitedStrip({
+      rows,
+      contextLabel: "test",
+      logger: silentLogger,
+      insert: insertSync,
+      insertSync,
+      insertAllowingBadRows,
+      stripJsonColumns,
+      maxPoisonStrips: 3,
+    });
+
+    expect(outcome).toEqual({ kind: "recovered", rowsStripped: 2, rowsDropped: 0, capped: false });
+    expect(allowCalls()).toBe(0);
+    expect(landed.map((r) => r.id).sort((a, b) => a - b)).toEqual([0, 1, 2, 3]);
+  });
+
+  it("bails to allow_errors when the failing row cannot be located", async () => {
+    const landed: FakeRow[] = [];
+    let allowCalls = 0;
+    const insertSync = async (rows: FakeRow[]) => {
+      if (rows.some((r) => r.poison)) {
+        throw new Error("Cannot parse JSON object here: {...} (no row hint here)");
+      }
+      landed.push(...rows);
+      return { summary: { written_rows: String(rows.length) } };
+    };
+    const insertAllowingBadRows = async (rows: FakeRow[]) => {
+      allowCalls += 1;
+      const good = rows.filter((r) => !r.poison);
+      landed.push(...good);
+      return { summary: { written_rows: String(good.length) } };
+    };
+
+    const outcome = await insertWithLimitedStrip({
+      rows: [clean(0), { id: 1, poison: true }, clean(2)],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert: insertSync,
+      insertSync,
+      insertAllowingBadRows,
+      stripJsonColumns: (row) => row,
+      hasMaterializedViews: false,
+    });
+
+    expect(outcome).toEqual({ kind: "recovered", rowsStripped: 0, rowsDropped: 1, capped: true });
+    expect(allowCalls).toBe(1);
+    expect(landed.map((r) => r.id).sort((a, b) => a - b)).toEqual([0, 2]);
+  });
+
+  it("rethrows non-parse errors so the caller's transient-retry path handles them", async () => {
+    const insert = async () => {
+      throw new Error("Connection refused");
+    };
+
+    await expect(
+      insertWithLimitedStrip({
+        rows: [clean(0)],
+        contextLabel: "test",
+        logger: silentLogger,
+        insert,
+        insertSync: insert,
+        insertAllowingBadRows: insert,
+        stripJsonColumns: (row) => row,
+      })
+    ).rejects.toThrow("Connection refused");
+  });
+});
+
+/**
+ * Builds insert doubles for `insertWithBadRowSkip`: a normal insert that fails
+ * whole when any row is poison, plus an allow-bad-rows insert that lands only
+ * the good rows and reports the landed count in a ClickHouse-style summary
+ * (`written_rows`) so the recovery can derive how many rows were skipped.
+ */
+function makeSkipHarness() {
+  const landed: FakeRow[] = [];
+  let insertCalls = 0;
+  let allowCalls = 0;
+  const insert = async (rows: FakeRow[]) => {
+    insertCalls += 1;
+    const badIndex = rows.findIndex((r) => r.poison || r.unstrippable);
+    if (badIndex >= 0) throw parseErrorAtRow(badIndex + 1);
+    landed.push(...rows);
+    return { summary: { written_rows: String(rows.length) } };
+  };
+  const insertAllowingBadRows = async (rows: FakeRow[]) => {
+    allowCalls += 1;
+    const good = rows.filter((r) => !(r.poison || r.unstrippable));
+    landed.push(...good);
+    return { summary: { written_rows: String(good.length) } };
+  };
+  return {
+    landed,
+    insert,
+    insertAllowingBadRows,
+    insertCalls: () => insertCalls,
+    allowCalls: () => allowCalls,
+  };
+}
+
+describe("insertWithBadRowSkip", () => {
+  const clean = (id: number): FakeRow => ({ id });
+
+  it("inserts a healthy batch with no recovery", async () => {
+    const { landed, insert, insertAllowingBadRows, allowCalls } = makeSkipHarness();
+
+    const outcome = await insertWithBadRowSkip({
+      rows: [clean(0), clean(1), clean(2)],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert,
+      insertAllowingBadRows,
+    });
+
+    expect(outcome.kind).toBe("inserted");
+    expect(landed).toHaveLength(3);
+    expect(allowCalls()).toBe(0);
+  });
+
+  it("skips the poison rows in one extra insert and counts drops exactly on a table with no materialized views", async () => {
+    const { landed, insert, insertAllowingBadRows, insertCalls, allowCalls } = makeSkipHarness();
+    const rows = [clean(0), { id: 1, poison: true }, clean(2), { id: 3, poison: true }, clean(4)];
+
+    const outcome = await insertWithBadRowSkip({
+      rows,
+      contextLabel: "test",
+      logger: silentLogger,
+      insert,
+      insertAllowingBadRows,
+      hasMaterializedViews: false,
+    });
+
+    expect(outcome).toEqual({ kind: "recovered", rowsStripped: 0, rowsDropped: 2, capped: false });
+    expect(landed.map((r) => r.id).sort((a, b) => a - b)).toEqual([0, 2, 4]);
+    expect(insertCalls()).toBe(1);
+    expect(allowCalls()).toBe(1);
+  });
+
+  it("counts only whole-batch drops on a table with materialized views (partial drop uncountable)", async () => {
+    const inflatedSummary = (goodCount: number) => ({
+      summary: { written_rows: String(goodCount * 3) },
+    });
+    const insert = async (rows: FakeRow[]) => {
+      if (rows.some((r) => r.poison)) throw parseErrorAtRow(1);
+      return inflatedSummary(rows.length);
+    };
+    const insertAllowingBadRows = async (rows: FakeRow[]) =>
+      inflatedSummary(rows.filter((r) => !r.poison).length);
+
+    const partial = await insertWithBadRowSkip({
+      rows: [clean(0), { id: 1, poison: true }, clean(2)],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert,
+      insertAllowingBadRows,
+      hasMaterializedViews: true,
+    });
+    expect(partial).toEqual({ kind: "recovered", rowsStripped: 0, rowsDropped: 0, capped: false });
+
+    const wholeBatch = await insertWithBadRowSkip({
+      rows: [
+        { id: 0, poison: true },
+        { id: 1, poison: true },
+      ],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert,
+      insertAllowingBadRows,
+      hasMaterializedViews: true,
+    });
+    expect(wholeBatch).toEqual({
+      kind: "recovered",
+      rowsStripped: 0,
+      rowsDropped: 2,
+      capped: false,
+    });
+  });
+
+  it("sanitizes a repairable batch and lands it in full without skipping any row", async () => {
+    const landed: FakeRow[] = [];
+    let allowCalls = 0;
+    const insert = async (rows: FakeRow[]) => {
+      if (rows.some((r) => typeof r.msg === "string" && r.msg.includes(HIGH_SURROGATE))) {
+        throw parseErrorAtRow(1);
+      }
+      landed.push(...rows);
+      return { summary: { written_rows: String(rows.length) } };
+    };
+    const insertAllowingBadRows = async (rows: FakeRow[]) => {
+      allowCalls += 1;
+      landed.push(...rows);
+      return { summary: { written_rows: String(rows.length) } };
+    };
+    const rows: FakeRow[] = [
+      { id: 0, msg: `bad ${HIGH_SURROGATE}` },
+      { id: 1, msg: "clean" },
+    ];
+
+    const outcome = await insertWithBadRowSkip({
+      rows,
+      contextLabel: "test",
+      logger: silentLogger,
+      insert,
+      insertAllowingBadRows,
+    });
+
+    expect(outcome.kind).toBe("sanitized");
+    expect(allowCalls).toBe(0);
+    expect(landed).toHaveLength(2);
+    expect(rows[0].msg).toBe(INVALID_UTF16_SENTINEL);
+  });
+
+  it("reports zero dropped when the summary has no written_rows count", async () => {
+    const insert = async (rows: FakeRow[]) => {
+      if (rows.some((r) => r.poison)) throw parseErrorAtRow(1);
+      return undefined;
+    };
+    const insertAllowingBadRows = async () => ({ summary: {} });
+
+    const outcome = await insertWithBadRowSkip({
+      rows: [{ id: 0, poison: true }, clean(1)],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert,
+      insertAllowingBadRows,
+    });
+
+    expect(outcome).toEqual({ kind: "recovered", rowsStripped: 0, rowsDropped: 0, capped: false });
+  });
+
+  it("rethrows non-parse errors so the caller's transient-retry path handles them", async () => {
+    const insert = async () => {
+      throw new Error("Connection refused");
+    };
+
+    await expect(
+      insertWithBadRowSkip({
+        rows: [clean(0)],
+        contextLabel: "test",
+        logger: silentLogger,
+        insert,
+        insertAllowingBadRows: insert,
+      })
+    ).rejects.toThrow("Connection refused");
+  });
+});
diff --git a/internal-packages/clickhouse/src/client/client.ts b/internal-packages/clickhouse/src/client/client.ts
index d91f86b428f..33ff2d3db9e 100644
--- a/internal-packages/clickhouse/src/client/client.ts
+++ b/internal-packages/clickhouse/src/client/client.ts
@@ -719,7 +719,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
 
           recordClickhouseError(span, clickhouseError);
 
-          return [new InsertError(clickhouseError.message), null];
+          return [toInsertError(clickhouseError), null];
         }
 
         this.logger.debug("Inserted into clickhouse", {
@@ -810,7 +810,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
           });
 
           recordClickhouseError(span, clickhouseError);
-          return [new InsertError(clickhouseError.message), null];
+          return [toInsertError(clickhouseError), null];
         }
 
         return [null, result];
@@ -872,7 +872,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
 
           recordClickhouseError(span, clickhouseError);
 
-          return [new InsertError(clickhouseError.message), null];
+          return [toInsertError(clickhouseError), null];
         }
 
         this.logger.debug("Inserted into clickhouse", {
@@ -971,7 +971,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
           });
 
           recordClickhouseError(span, clickhouseError);
-          return [new InsertError(clickhouseError.message), null];
+          return [toInsertError(clickhouseError), null];
         }
 
         this.logger.debug("Inserted into clickhouse", {
@@ -1001,6 +1001,11 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
   }
 }
 
+function toInsertError(error: Error): InsertError {
+  const rawMessage = error instanceof ClickHouseError ? error.rawMessage : undefined;
+  return new InsertError(error.message, { rawMessage });
+}
+
 function recordClickhouseError(span: Span, error: Error): void {
   if (error instanceof ClickHouseError) {
     span.setAttributes({
diff --git a/internal-packages/clickhouse/src/client/errors.ts b/internal-packages/clickhouse/src/client/errors.ts
index a620fb54649..c0ceacb830f 100644
--- a/internal-packages/clickhouse/src/client/errors.ts
+++ b/internal-packages/clickhouse/src/client/errors.ts
@@ -24,10 +24,12 @@ export abstract class BaseError<TContext extends ErrorContext = ErrorContext> ex
 export class InsertError extends BaseError {
   public readonly retry = true;
   public readonly name = InsertError.name;
-  constructor(message: string) {
+  public readonly rawMessage?: string;
+  constructor(message: string, options?: { rawMessage?: string }) {
     super({
       message,
     });
+    this.rawMessage = options?.rawMessage;
   }
 }
 export class QueryError extends BaseError<{ query: string }> {
diff --git a/package.json b/package.json
index 3b00102ac83..19b1a6ed765 100644
--- a/package.json
+++ b/package.json
@@ -91,7 +91,8 @@
       "@window-splitter/state@1.1.3": "patches/@window-splitter__state@1.1.3.patch",
       "streamdown@2.5.0": "patches/streamdown@2.5.0.patch",
       "tsup@8.4.0": "patches/tsup@8.4.0.patch",
-      "@remix-run/router@1.23.3": "patches/@remix-run__router@1.23.3.patch"
+      "@remix-run/router@1.23.3": "patches/@remix-run__router@1.23.3.patch",
+      "@clickhouse/client-common@1.12.1": "patches/@clickhouse__client-common@1.12.1.patch"
     },
     "overrides": {
       "typescript": "catalog:",
diff --git a/patches/@clickhouse__client-common@1.12.1.patch b/patches/@clickhouse__client-common@1.12.1.patch
new file mode 100644
index 00000000000..eb4dc08a000
--- /dev/null
+++ b/patches/@clickhouse__client-common@1.12.1.patch
@@ -0,0 +1,29 @@
+diff --git a/dist/error/error.d.ts b/dist/error/error.d.ts
+index fb04ebb65a8c7115c465093462edaafa14a814de..30b370bf2eb8ba7491c249db96d565a67fe367c6 100644
+--- a/dist/error/error.d.ts
++++ b/dist/error/error.d.ts
+@@ -7,6 +7,7 @@ interface ParsedClickHouseError {
+ export declare class ClickHouseError extends Error {
+     readonly code: string;
+     readonly type: string | undefined;
++    rawMessage?: string;
+     constructor({ message, code, type }: ParsedClickHouseError);
+ }
+ export declare function parseError(input: string | Error): ClickHouseError | Error;
+diff --git a/dist/error/error.js b/dist/error/error.js
+index e06fa4ab36537c2a448857c1b001e70cd771bfe5..9750bd7e1285ac8cf2c8095b592ed344126176b6 100644
+--- a/dist/error/error.js
++++ b/dist/error/error.js
+@@ -35,7 +35,11 @@ function parseError(input) {
+     const match = message.match(errorRe);
+     const groups = match?.groups;
+     if (groups) {
+-        return new ClickHouseError(groups);
++        const error = new ClickHouseError(groups);
++        if (!inputIsError && groups.message.startsWith("Cannot parse JSON object")) {
++            error.rawMessage = message;
++        }
++        return error;
+     }
+     else {
+         return inputIsError ? input : new Error(input);
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index b09dc718a6f..1183051b975 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -56,6 +56,9 @@ patchedDependencies:
   '@changesets/assemble-release-plan@5.2.4':
     hash: a7a12643e8d89a00d5322f750c292e8567b5ee0fc9c613a238a1184c25533e4b
     path: patches/@changesets__assemble-release-plan@5.2.4.patch
+  '@clickhouse/client-common@1.12.1':
+    hash: 870d9f887c4057af8fa50fb52936fb7e8463f62f5e88e70b804c3a7445c98b4d
+    path: patches/@clickhouse__client-common@1.12.1.patch
   '@kubernetes/client-node@1.0.0':
     hash: ba1a06f46256cdb8d6faf7167246692c0de2e7cd846a9dc0f13be0137e1c3745
     path: patches/@kubernetes__client-node@1.0.0.patch
@@ -17703,7 +17706,7 @@ snapshots:
 
   '@clickhouse/client-common@1.11.1': {}
 
-  '@clickhouse/client-common@1.12.1': {}
+  '@clickhouse/client-common@1.12.1(patch_hash=870d9f887c4057af8fa50fb52936fb7e8463f62f5e88e70b804c3a7445c98b4d)': {}
 
   '@clickhouse/client@1.11.1':
     dependencies:
@@ -17711,7 +17714,7 @@ snapshots:
 
   '@clickhouse/client@1.12.1':
     dependencies:
-      '@clickhouse/client-common': 1.12.1
+      '@clickhouse/client-common': 1.12.1(patch_hash=870d9f887c4057af8fa50fb52936fb7e8463f62f5e88e70b804c3a7445c98b4d)
 
   '@codemirror/autocomplete@6.4.0(@codemirror/language@6.3.2)(@codemirror/state@6.2.0)(@codemirror/view@6.7.2)(@lezer/common@1.3.0)':
     dependencies: