🤖 fix: self-heal corrupt analytics rows that wallpaper the dashboard

README.md

@@ -13,13 +13,12 @@
 [![Discord](https://img.shields.io/discord/1446553342699507907?logo=discord&label=Discord)](https://cdr.co/mux-discord)
 [![X (formerly Twitter)](https://img.shields.io/badge/Follow-%40codermux-black?logo=x)](https://x.com/codermux)
 
-
 </div>
 
 > [!IMPORTANT]
 > This project was renamed from Mux to Shux after Mux.com raised a trademark concern. “Mux” is a common technical abbreviation of “multiplexer” and we do not expect confusion between the projects, but chose to rename ours rather than spend more time on the dispute. “Shux” captures our reaction to the process and bears no other significance.
-> 
-Shux is a desktop & browser application for parallel agentic development. It enables developers to plan and execute tasks with multiple AI agents on local or remote compute.
+>
+> Shux is a desktop & browser application for parallel agentic development. It enables developers to plan and execute tasks with multiple AI agents on local or remote compute.
 
 <p><img src="./docs/img/mux-demo.gif" alt="Shux product demo" width="100%" /></p>
 

flake.nix

@@ -84,7 +84,7 @@
 
             outputHashMode = "recursive";
             # Marker used by scripts/update_flake_hash.sh to update this hash in place.
-            outputHash = "sha256-Ci2q4ZCIymKhf4rinh6VKdzaGVBCRBsMUgjQOXTqotM="; # shux-offline-cache-hash
+            outputHash = "sha256-ri3Q1gY4ifnjT9FMg3wNqoIT5OJbfQRKAWw5Zt8DH9k="; # shux-offline-cache-hash
           };
 
           configurePhase = ''

package.json

@@ -61,7 +61,7 @@
     "@dnd-kit/core": "^6.3.1",
     "@dnd-kit/sortable": "^10.0.0",
     "@dnd-kit/utilities": "^3.2.2",
-    "@duckdb/node-api": "^1.4.4-r.1",
+    "@duckdb/node-api": "^1.5.5-r.4",
     "@homebridge/ciao": "^1.3.4",
     "@jitl/quickjs-wasmfile-release-asyncify": "^0.31.0",
     "@lydell/node-pty": "1.1.0",

src/node/services/analytics/analyticsWorker.ts

@@ -6,6 +6,7 @@ import { decideSyncPlan, type SyncAction } from "./backfillDecision";
 import { shouldCheckpointAfterSync } from "./checkpointDecision";
 import {
   clearWorkspaceAnalyticsState,
+  deleteCorruptAnalyticsRows,
   getCurrentPricingFingerprint,
   ingestWorkspace,
   readStoredPricingFingerprint,
@@ -147,13 +148,37 @@ async function handleInit(data: InitData): Promise<void> {
   for (const migrationSql of DELEGATION_ROLLUPS_COLUMN_MIGRATIONS_SQL) {
     await activeConn.run(migrationSql);
   }
+
+  await sweepCorruptRows("init");
+}
+
+/**
+ * Delete corruption-class rows and log when anything was actually removed.
+ * Best-effort: a failed sweep must never reject init (which would cache a
+ * worker error and disable analytics until restart) or fail an
+ * otherwise-successful ingest, so errors are logged and swallowed.
+ */
+async function sweepCorruptRows(context: string): Promise<void> {
+  try {
+    const deleted = await deleteCorruptAnalyticsRows(getConn());
+    if (deleted > 0) {
+      process.stderr.write(
+        `[analytics-worker] Deleted ${deleted} corrupt analytics row(s) (${context})\n`
+      );
+    }
+  } catch (error) {
+    process.stderr.write(
+      `[analytics-worker] Corrupt-row sweep failed (${context}): ${getErrorMessage(error)}\n`
+    );
+  }
 }
 
 async function handleIngest(data: IngestData): Promise<void> {
   assert(data.workspaceId.trim().length > 0, "ingest requires workspaceId");
   assert(data.sessionDir.trim().length > 0, "ingest requires sessionDir");
 
   await ingestWorkspace(getConn(), data.workspaceId, data.sessionDir, data.meta ?? {});
+  await sweepCorruptRows("ingest");
 }
 
 async function handleRebuildAll(data: RebuildAllData): Promise<{ workspacesIngested: number }> {
@@ -169,6 +194,7 @@ async function handleRebuildAll(data: RebuildAllData): Promise<{ workspacesInges
   // A completed rebuild priced everything with the current tables; refresh the
   // fingerprint so the next sync check does not schedule a redundant rebuild.
   await storePricingFingerprint(getConn());
+  await sweepCorruptRows("rebuildAll");
   return result;
 }
 
@@ -390,6 +416,7 @@ async function handleSyncCheck(data: SyncCheckData): Promise<SyncCheckResult> {
     if (pricingFingerprintChanged) {
       await storePricingFingerprint(getConn());
     }
+    await sweepCorruptRows("syncCheck full_rebuild");
     await checkpointIfNeeded(plan.action, workspacesIngested, 0);
 
     const elapsedMs = Math.round(performance.now() - syncStartMs);
@@ -448,6 +475,7 @@ async function handleSyncCheck(data: SyncCheckData): Promise<SyncCheckResult> {
     }
   }
 
+  await sweepCorruptRows("syncCheck incremental");
   await checkpointIfNeeded(plan.action, workspacesIngested, workspacesPurged);
 
   const elapsedMs = Math.round(performance.now() - syncStartMs);

src/node/services/analytics/etl.test.ts

@@ -9,6 +9,7 @@ import {
   appendEvents,
   CHAT_FILE_NAME,
   clearWorkspaceAnalyticsState,
+  deleteCorruptAnalyticsRows,
   getCurrentPricingFingerprint,
   ingestWorkspace,
   parseWorkspaceFromDisk,
@@ -1634,3 +1635,83 @@ describe("pricing fingerprint", () => {
     expect(await readStoredPricingFingerprint(conn)).toBe(getCurrentPricingFingerprint());
   });
 });
+
+describe("deleteCorruptAnalyticsRows", () => {
+  async function seedWatermark(conn: DuckDBConnection, workspaceId: string): Promise<void> {
+    await conn.run(
+      "INSERT INTO ingest_watermarks (workspace_id, last_sequence, last_modified) VALUES (?, ?, ?)",
+      [workspaceId, 1, 1]
+    );
+  }
+
+  test("deletes corrupt rows while keeping healthy rows", async () => {
+    const conn = await createTestConn();
+
+    // Migrated legacy IDs are `${projectBasename}-${workspaceBasename}` with
+    // no length limit (up to 2x NAME_MAX + 1 = 511 chars) and must survive.
+    const legacyId = `${"p".repeat(255)}-${"w".repeat(255)}`;
+    // Custom-provider model IDs have no schema max length; an extremely long
+    // model on an otherwise-healthy row must never be deletion evidence.
+    const longModel = `custom:${"m".repeat(2000)}`;
+
+    for (const workspaceId of ["ws-healthy", legacyId, "ws-long-model", "parent-healthy"]) {
+      await seedWatermark(conn, workspaceId);
+    }
+
+    for (const [workspaceId, model, cost] of [
+      ["ws-healthy", "anthropic:claude-haiku-4-5", 1.0],
+      [legacyId, "anthropic:claude-haiku-4-5", 2.0],
+      ["ws-long-model", longModel, 3.0],
+      // Large-batch phantom: concatenated identifiers exceed the length caps.
+      ["x".repeat(2000), "anthropic:claude-haiku-4-5".repeat(100), 0.05],
+      // Small-batch phantom: two concatenated 10-char workspace IDs stay far
+      // under the length caps but can never match a real watermark.
+      ["aaaaabbbbbcccccddddd", "openai:gpt-5.6-solopenai:gpt-5.6-sol", 0.05],
+    ] as const) {
+      await conn.run("INSERT INTO events (workspace_id, model, total_cost_usd) VALUES (?, ?, ?)", [
+        workspaceId,
+        model,
+        cost,
+      ]);
+    }
+
+    for (const [parent, child] of [
+      ["parent-healthy", "child-healthy"],
+      // A rollup may outlive its removed child workspace; only the parent
+      // must be a known workspace.
+      ["parent-healthy", "child-removed"],
+      ["p".repeat(2000), "child-corrupt"],
+      // Small-batch phantom parent: unknown to watermarks.
+      ["par-aaaaapar-bbbbb", "child-x"],
+    ] as const) {
+      await conn.run(
+        `INSERT INTO delegation_rollups (parent_workspace_id, child_workspace_id, model)
+         VALUES (?, ?, ?)`,
+        [parent, child, "openai:gpt-5.6-sol"]
+      );
+    }
+
+    expect(await deleteCorruptAnalyticsRows(conn)).toBe(4);
+
+    const eventRows = await queryRows(
+      conn,
+      "SELECT workspace_id FROM events ORDER BY LENGTH(workspace_id)"
+    );
+    expect(eventRows).toEqual([
+      { workspace_id: "ws-healthy" },
+      { workspace_id: "ws-long-model" },
+      { workspace_id: legacyId },
+    ]);
+    const rollupRows = await queryRows(
+      conn,
+      "SELECT child_workspace_id FROM delegation_rollups ORDER BY child_workspace_id"
+    );
+    expect(rollupRows).toEqual([
+      { child_workspace_id: "child-healthy" },
+      { child_workspace_id: "child-removed" },
+    ]);
+
+    // Idempotent: nothing left to delete.
+    expect(await deleteCorruptAnalyticsRows(conn)).toBe(0);
+  });
+});

src/node/services/analytics/etl.ts

@@ -865,6 +865,60 @@ export async function clearWorkspaceAnalyticsState(
   }
 }
 
+/**
+ * Self-healing sweep for a rare native-layer corruption class: a phantom row
+ * can materialize whose every VARCHAR column is the concatenation of that
+ * column's non-null values across an entire batch of inserted rows (observed
+ * once in the wild: a 17KB "model" string spanning ~670 events, which then
+ * wallpapered the Analytics dashboard as one giant legend entry). The donor
+ * rows are written correctly, so deleting rows with impossible string lengths
+ * loses no real data.
+ *
+ * Two evidence classes, both structural (unbounded columns like model,
+ * paths, and workspace names are never deletion evidence on their own, since
+ * custom-provider model IDs etc. have no schema max length):
+ *
+ * 1. Identifier length beyond the legal construction maximum. New workspace
+ *    IDs are short hex; migrated legacy IDs are
+ *    `${projectBasename}-${workspaceBasename}` (config.generateLegacyId),
+ *    each basename bounded by the filesystem's NAME_MAX (255 bytes), so 511
+ *    is the ceiling. Agent IDs/types also derive from basenames. Cap at 1024.
+ *
+ * 2. Workspace identity unknown to ingest_watermarks. A concatenation of two
+ *    or more workspace IDs can never equal a real workspace ID, no matter how
+ *    small the corrupted batch, while every legitimate row's workspace gets a
+ *    watermark by the end of the ingest/rebuild pass that wrote it (sweeps
+ *    run after those passes complete). If a crash lands between the event
+ *    write and the watermark write, deleting the orphans is still safe: the
+ *    missing watermark makes the next syncCheck re-ingest that workspace from
+ *    disk in full. delegation_rollups joins on parent_workspace_id only;
+ *    child_workspace_id may legitimately reference a removed child workspace.
+ */
+export async function deleteCorruptAnalyticsRows(conn: DuckDBConnection): Promise<number> {
+  const eventsResult = await conn.run(`
+    DELETE FROM events
+    WHERE LENGTH(workspace_id) > 1024
+       OR LENGTH(parent_workspace_id) > 1024
+       OR LENGTH(agent_id) > 1024
+       OR NOT EXISTS (
+         SELECT 1 FROM ingest_watermarks w WHERE w.workspace_id = events.workspace_id
+       )
+  `);
+
+  const rollupsResult = await conn.run(`
+    DELETE FROM delegation_rollups
+    WHERE LENGTH(parent_workspace_id) > 1024
+       OR LENGTH(child_workspace_id) > 1024
+       OR LENGTH(agent_type) > 1024
+       OR NOT EXISTS (
+         SELECT 1 FROM ingest_watermarks w
+         WHERE w.workspace_id = delegation_rollups.parent_workspace_id
+       )
+  `);
+
+  return eventsResult.rowsChanged + rollupsResult.rowsChanged;
+}
+
 function serializeHeadSignatureValue(value: string | number | null): string {
   if (value === null) {
     return "null";