Fix code-review correctness and lifecycle issues

apps/demo/e2e/data-search.spec.ts

@@ -230,3 +230,57 @@ test('csv drop: a 50-row edges fixture derives nodes and swaps the dataset', asy
   await expect(page.locator(SEARCH_OPTION).first()).toBeVisible({ timeout: 10_000 });
   await expect(page.locator(SEARCH_LABEL).first()).toHaveText(/^c17/);
 });
+
+test('a newer data-mode choice supersedes an in-flight CSV preparation', async ({
+  page,
+}, testInfo) => {
+  // Make the otherwise-small fixture deterministically asynchronous. Orbit's
+  // CSV lane consumes File.stream(), so delaying its first chunk leaves time
+  // for the user to choose another mode before preparation resolves.
+  await page.addInitScript(() => {
+    const state = window as typeof window & { __orbitCsvStreamFinished?: boolean };
+    state.__orbitCsvStreamFinished = false;
+    const originalStream = Blob.prototype.stream;
+    Blob.prototype.stream = function (this: Blob): ReadableStream<Uint8Array> {
+      const original = originalStream.call(this);
+      // Do not delay unrelated Blob consumers used while the app boots.
+      if (!(this instanceof File) || this.name !== 'superseded.csv') return original;
+      const reader = original.getReader();
+      return new ReadableStream<Uint8Array>({
+        async pull(controller) {
+          await new Promise<void>((resolve) => {
+            setTimeout(resolve, 500);
+          });
+          const next = await reader.read();
+          if (next.done) {
+            state.__orbitCsvStreamFinished = true;
+            controller.close();
+          } else controller.enqueue(next.value);
+        },
+        cancel(reason) {
+          return reader.cancel(reason);
+        },
+      });
+    };
+  });
+
+  const fixture = testInfo.outputPath('superseded.csv');
+  writeFileSync(fixture, 'source,target,weight\na,b,1\nb,c,2\n', 'utf8');
+  await gotoReady(page);
+
+  await page.setInputFiles('[data-testid="csv-file-input"]', fixture);
+  await page.getByTestId('semantic-mode').click();
+  await expect(page.getByTestId('m5-panel')).toBeVisible({ timeout: 15_000 });
+
+  // Wait for EOF rather than a wall-clock approximation, then give the
+  // preparation promise and React commit a chance to drain. Its completion
+  // must not replace the newer semantic mode or publish a CSV summary.
+  await page.waitForFunction(
+    () =>
+      (window as typeof window & { __orbitCsvStreamFinished?: boolean })
+        .__orbitCsvStreamFinished === true,
+  );
+  await page.waitForTimeout(250);
+  await expect(page.getByTestId('m5-panel')).toBeVisible();
+  await expect(page.getByTestId('csv-summary')).toHaveCount(0);
+});

apps/demo/e2e/ingestion.spec.ts

@@ -134,3 +134,23 @@ test('stream feed → live meter → commit → isolate via context menu → res
   await expect(page.locator(SCOPE_STATUS)).toHaveText('full', { timeout: 10_000 });
   await expect(page.locator(NODE_COUNT)).toHaveText(fmt(STREAM_NODES));
 });
+
+test('leaving an active stream run cannot resurrect its cleared progress meter', async ({ page }) => {
+  // Enough batches to guarantee a paint/yield between the first progress
+  // receipt and completion; the test cancels immediately after that paint.
+  await page.goto('/?rows=200000');
+  await page.waitForSelector(READY_DOT, { timeout: 60_000 });
+  await page.getByTestId('stream-feed').click();
+  await expect(page.locator(METER)).toHaveAttribute('data-phase', 'streaming', {
+    timeout: 15_000,
+  });
+
+  await page.getByTestId('semantic-mode').click();
+  await expect(page.getByTestId('m5-panel')).toBeVisible({ timeout: 15_000 });
+  await expect(page.locator(METER)).toHaveCount(0);
+
+  // The abandoned driver emits its terminal `aborted` receipt at the next
+  // batch boundary. It must remain private to that obsolete run.
+  await page.waitForTimeout(500);
+  await expect(page.locator(METER)).toHaveCount(0);
+});

apps/demo/e2e/overlay-a11y.spec.ts

@@ -132,9 +132,19 @@ test('navigator: Tab to toggle, open, arrow through items, Enter focuses a node'
 }) => {
   await gotoReady(page);
 
-  // The collapsible navigator toggle is the page's first tabbable element.
-  await page.keyboard.press('Tab');
   const toggle = page.locator(NAV_TOGGLE);
+  // Labels are legitimate keyboard controls and precede the overlay children
+  // in Graph's DOM. Prove the toggle is reachable by sequential keyboard
+  // navigation without coupling this test to the current number of labels.
+  let reachedToggle = false;
+  for (let tabs = 0; tabs < 64; tabs++) {
+    await page.keyboard.press('Tab');
+    if (await toggle.evaluate((element) => element === document.activeElement)) {
+      reachedToggle = true;
+      break;
+    }
+  }
+  expect(reachedToggle, 'Tab should reach the navigator toggle').toBe(true);
   await expect(toggle).toBeFocused();
   await expect(toggle).toHaveAttribute('aria-expanded', 'false');
 

apps/demo/package.json

@@ -5,7 +5,7 @@
   "type": "module",
   "scripts": {
     "dev": "vite",
-    "build": "vite build",
+    "build": "vite build && node ../../scripts/check-workspace-worker-asset.mjs",
     "preview": "vite preview",
     "typecheck": "tsc --noEmit",
     "test": "echo \"demo has no unit tests\" && exit 0",

apps/demo/src/App.tsx

@@ -614,6 +614,10 @@ export function App() {
   /** Last unavailable activation ("<label>: <reason>") — sticky. */
   const [searchNote, setSearchNote] = useState<string | null>(null);
   const [csvError, setCsvError] = useState<string | null>(null);
+  /** Invalidates an in-flight CSV preparation when a newer data-mode action
+   * wins. Preparation intentionally runs before the graph mode changes. */
+  const csvLoadEpochRef = useRef(0);
+  const csvBusyRef = useRef(false);
 
   // --- M5 semantic-exploration state (see semantic.tsx) ---
   const [m5Grouping, setM5Grouping] = useState<M5Grouping>('none');
@@ -830,6 +834,7 @@ export function App() {
 
   // --- declarative data updates ---
   const regenerate = useCallback(() => {
+    csvLoadEpochRef.current += 1;
     setGen((g) => {
       const seed = nextSeed(g.seed);
       // New datasetKey → full reset (positions, per-dataset state).
@@ -847,6 +852,7 @@ export function App() {
   }, [resetOmnigraphState]);
 
   const addNodes = useCallback(() => {
+    csvLoadEpochRef.current += 1;
     // Same datasetKey, bumped sourceRevision, strict superset snapshot →
     // exercises the incremental diff + position preservation.
     setMode(DECLARATIVE);
@@ -860,6 +866,7 @@ export function App() {
   // purpose:'replace' is rejected while a declarative source drives, so
   // the session begins against a fresh, never-declarative instance.
   const streamFeed = useCallback(() => {
+    csvLoadEpochRef.current += 1;
     setSelection([]);
     setPendingIsolate(false);
     setDragNote(null);
@@ -881,7 +888,11 @@ export function App() {
       nodeShare: readStreamNodeShare(),
       family: readStreamFamily(),
       shouldCancel: () => cancelled,
-      onProgress: setMeter, // driver-throttled — receipts arrive far faster
+      // The driver emits a terminal aborted receipt during cleanup. Never let
+      // that old run repopulate a meter cleared by a mode/run switch.
+      onProgress: (progress) => {
+        if (!cancelled) setMeter(progress);
+      },
     }).catch((err: unknown) => {
       if (!cancelled) setError(err instanceof Error ? err.message : String(err));
     });
@@ -895,6 +906,7 @@ export function App() {
   // `purpose:'replace'` session, which may not race a declarative source,
   // so each load targets a fresh instance (`key` change).
   const loadOmnigraph = useCallback(() => {
+    csvLoadEpochRef.current += 1;
     setSelection([]);
     setPendingIsolate(false);
     setDragNote(null);
@@ -953,14 +965,15 @@ export function App() {
   // mode change remounts <Graph> (`key`) — the previous mode may have been a
   // replace-session stream, and a fresh instance takes the declarative
   // prepared snapshot without racing it.
-  const csvBusyRef = useRef(false);
   const onCsvFile = useCallback(
     (file: File) => {
       if (csvBusyRef.current) return; // one prepare at a time
       csvBusyRef.current = true;
+      const requestEpoch = ++csvLoadEpochRef.current;
       setCsvError(null);
       loadCsvEdgesFile(file)
         .then((result) => {
+          if (csvLoadEpochRef.current !== requestEpoch) return;
           setSelection([]);
           setPendingIsolate(false);
           setDragNote(null);
@@ -972,6 +985,7 @@ export function App() {
           setMode((m) => ({ kind: 'csv', runId: m.kind === 'csv' ? m.runId + 1 : 1, result }));
         })
         .catch((err: unknown) => {
+          if (csvLoadEpochRef.current !== requestEpoch) return;
           setCsvError(err instanceof Error ? err.message : String(err));
         })
         .finally(() => {
@@ -987,6 +1001,7 @@ export function App() {
   // this mode run a FIXED layout while the others stay on force.
   const enterSemantic = useCallback(
     (layout: M5Layout) => {
+      csvLoadEpochRef.current += 1;
       setSelection([]);
       setPendingIsolate(false);
       setDragNote(null);
@@ -1060,6 +1075,7 @@ export function App() {
 
   const m5OnLayoutChange = useCallback(
     (layout: M5Layout) => {
+      csvLoadEpochRef.current += 1;
       setMode((m) =>
         m.kind === 'semantic' && m.layout !== layout
           ? { kind: 'semantic', runId: m.runId + 1, layout }

packages/core/src/columnar.ts

@@ -77,6 +77,16 @@ function checkStringColumn(
     out.push({ where, problem: 'not-a-string-column', detail: 'expected {kind:"string", dictionary, codes}' });
     return;
   }
+  for (let d = 0; d < col.dictionary.length; d++) {
+    if (typeof col.dictionary[d] !== 'string') {
+      out.push({
+        where,
+        problem: 'not-a-string-column',
+        detail: `dictionary entry ${d} is not a string`,
+      });
+      return;
+    }
+  }
   if (col.codes.length !== rows) {
     const detached = col.codes.length === 0 && col.codes.buffer.byteLength === 0;
     out.push({

packages/core/src/columnarValidate.ts

@@ -11,9 +11,11 @@
  * STRING, which survives.
  * - duplicate edge ids drop, first wins ('duplicate-edge-id', warning).
  * - self-loops are RETAINED with 'self-loop-retained' (info).
- * - invalid-node / invalid-edge / dangling-edge cannot occur here: ids come
- * from a structurally validated dictionary column and endpoints are
- * in-bounds indices by prior validation (validateColumnarStructure).
+ * - invalid-node / invalid-edge rows with NUL-reserved ids drop here;
+ * dangling edges can then arise when an endpoint names a dropped invalid
+ * node. Other structural corruption cannot occur because ids come from a
+ * structurally validated dictionary column and endpoints are in-bounds
+ * indices by prior validation (validateColumnarStructure).
  *
  * Duplicates hide in TWO encodings: two rows sharing a code, and two
  * DISTINCT dictionary entries holding equal strings. Both are handled by
@@ -95,6 +97,7 @@ export function acceptColumnar(
   const edgeRows = snapshot.edges.length;
 
   const duplicateNode: Tally = { count: 0, samples: [] };
+  const invalidEdge: Tally = { count: 0, samples: [] };
   const duplicateEdge: Tally = { count: 0, samples: [] };
   const selfLoop: Tally = { count: 0, samples: [] };
   const invalidNode: Tally = { count: 0, samples: [] };
@@ -137,18 +140,21 @@ export function acceptColumnar(
 
   // --- Edges: first occurrence per canonical edge id wins; self-loops kept. -
   const edgeCanonical = canonicalizeDictionary(edgeIds.dictionary);
+  const nulEdgeDict = new Uint8Array(edgeIds.dictionary.length);
+  for (let d = 0; d < edgeIds.dictionary.length; d++) {
+    if (edgeIds.dictionary[d]!.includes('\u0000')) nulEdgeDict[d] = 1;
+  }
   const keepEdges = new Uint8Array(edgeRows);
   const seenEdgeByCanonical = new Uint8Array(edgeIds.dictionary.length);
   const { source, target } = snapshot.edges;
   const linksOut = new Uint32Array(edgeRows * 2); // trimmed after the scan
   let acceptedEdgeCount = 0;
   for (let e = 0; e < edgeRows; e++) {
-    const canonical = edgeCanonical[edgeIds.codes[e]!]!;
-    if (seenEdgeByCanonical[canonical] !== 0) {
-      record(duplicateEdge, edgeIds.dictionary[edgeIds.codes[e]!]!);
+    if (nulEdgeDict[edgeIds.codes[e]!] !== 0) {
+      record(invalidEdge, `[${e}]`);
       continue;
     }
-    seenEdgeByCanonical[canonical] = 1;
+    const canonical = edgeCanonical[edgeIds.codes[e]!]!;
     const s = nodeAcceptedIndex[source[e]!]!;
     const t = nodeAcceptedIndex[target[e]!]!;
     if (s === -1 || t === -1) {
@@ -160,6 +166,16 @@ export function acceptColumnar(
       );
       continue;
     }
+    // Object-lane order is endpoint admission BEFORE final-id dedupe: a
+    // dangling record never occupies its edge id, so a later endpoint-valid
+    // record with the same id remains eligible. NUL-invalid node rows make
+    // that distinction observable in the otherwise structurally sound
+    // columnar lane.
+    if (seenEdgeByCanonical[canonical] !== 0) {
+      record(duplicateEdge, edgeIds.dictionary[edgeIds.codes[e]!]!);
+      continue;
+    }
+    seenEdgeByCanonical[canonical] = 1;
     if (s === t) {
       // Same ACCEPTED node = same id string (the object lane compares
       // source/target strings) — retained, reported.
@@ -188,6 +204,13 @@ export function acceptColumnar(
     duplicateNode,
     `${duplicateNode.count} duplicate node id(s) dropped (first occurrence wins)`,
   );
+  pushDiagnostic(
+    diagnostics,
+    'invalid-edge',
+    'error',
+    invalidEdge,
+    `${invalidEdge.count} edge row(s) dropped: missing or non-string source/target, or NUL-containing explicit id`,
+  );
   pushDiagnostic(
     diagnostics,
     'dangling-edge-endpoint',

packages/core/src/ingestion.ts

@@ -222,7 +222,11 @@ export function stageBatch<N, E>(
   for (let i = 0; i < rawNodes.length; i++) {
     const row = rawNodes[i];
     const id = typeof row === 'object' && row !== null ? (row as { id?: unknown }).id : undefined;
-    if (typeof id !== 'string') {
+    if (typeof id !== 'string' || id.includes('\u0000')) {
+      // U+0000 is reserved by the scene-key codec for collapsed-group
+      // super-nodes. Streaming ingestion must enforce the same namespace
+      // boundary as declarative snapshot validation; otherwise an overlay
+      // row can collide with a synthetic scene row during group rewrite.
       recordRow(tallies.invalidNodes, `[${batch.sequence}:${i}]`);
       continue;
     }
@@ -248,6 +252,12 @@ export function stageBatch<N, E>(
       recordRow(tallies.invalidEdges, typeof edge.id === 'string' ? edge.id : `[${batch.sequence}:${i}]`);
       continue;
     }
+    if (typeof edge.id === 'string' && edge.id.includes('\u0000')) {
+      // Explicit edge ids occupy the same rendered id lane as internal
+      // meta-edge scene keys, whose reserved namespace begins with U+0000.
+      recordRow(tallies.invalidEdges, `[${batch.sequence}:${i}]`);
+      continue;
+    }
     const hasExplicitId = typeof edge.id === 'string';
     let id: string;
     if (hasExplicitId) {
@@ -511,7 +521,7 @@ export function sessionCommitDiagnostics(
     'invalid-node',
     'error',
     tallies.invalidNodes,
-    `${tallies.invalidNodes.count} ingested node row(s) dropped: missing or non-string id`,
+    `${tallies.invalidNodes.count} ingested node row(s) dropped: missing, non-string, or NUL-containing id`,
   );
   push(
     'duplicate-node-id',
@@ -523,7 +533,7 @@ export function sessionCommitDiagnostics(
     'invalid-edge',
     'error',
     tallies.invalidEdges,
-    `${tallies.invalidEdges.count} ingested edge row(s) dropped: missing or non-string source/target`,
+    `${tallies.invalidEdges.count} ingested edge row(s) dropped: missing or non-string source/target, or NUL-containing explicit id`,
   );
   push(
     'duplicate-edge-id',