fix(webapp,clickhouse): keep the rest of a ClickHouse batch when one run or span has un-ingestable JSON (#4358)

## Summary

A single run output, trace span, or payload carrying JSON that
ClickHouse can't ingest (for example nesting past its depth limit) used
to fail the whole insert batch, so unrelated runs and spans silently
disappeared from the runs list, traces, and logs. This keeps the rest of
the batch and handles the offending row instead of dropping everything
around it.

## Fix

Recovery is per-table, matched to what each table needs:

- **Runs** (`task_runs_v2`) keep their status. We follow ClickHouse's
failing-row hint to strip just the un-ingestable JSON column(s) so the
run still lands (its output reads from Postgres on the detail page), up
to a configurable limit (`RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH`,
default `1`). Past the limit we stop and land the batch with
`allow_errors` in a single pass, skipping the remainder. Cost stays a
fixed handful of inserts no matter how large or poisoned a flush is.
- **Trace events and payloads** (high volume, append-only) recover with
a single `allow_errors` insert: the good rows land in one pass and only
the un-ingestable rows are skipped.

Before falling back, a lightweight sanitizer still repairs what it can
losslessly (lone UTF-16 surrogates, out-of-range integers) so a
repairable row lands in full.

To read the failing-row hint we patch `@clickhouse/client-common`: its
error parser truncates the server response and discards the `(at row N)`
position, so the patch preserves the full text for the recovery path to
read.

Author: ericallam

.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.

apps/webapp/app/env.server.ts

@@ -1723,6 +1723,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),

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,