fix(consumer): skip config replay for enrichment batches

.xtap-deployment.json

@@ -1,18 +1,18 @@
 {
-  "source_revision": "35f8318026a3ad56868fd0fc57a42142a50bef3b",
+  "source_revision": "cfd46b156aeec2c30225e63f9873c295fb247731",
   "enrichment_revision_handoff": {
     "active_generation": 87,
     "activation_sha256": "0d09e98a8073db48faf0ea56d7600fc581c1106ae98312d2ef8fd2273a48f404",
     "run_id": "xtap-b0e01d0c7828c30def3948bcb4251e68",
     "plan_sha256": "b0e01d0c7828c30def3948bcb4251e68ddc4151366cb31d3d96ea01ba86b2d85",
     "plan_worker_revision": "6a3c4855dd21f469fab63705f77cd7b2f4095726",
-    "target_worker_revision": "35f8318026a3ad56868fd0fc57a42142a50bef3b",
+    "target_worker_revision": "cfd46b156aeec2c30225e63f9873c295fb247731",
     "contract_sha256": "2a2814ad4162457c0b8fe6de9cc07e0813f9fffadd3bde21ca46c731cc58de69",
     "source_snapshot_revision": "5bcd681d22132a5009c5f3e72d33ffb6fb1ac1e1269a23dd3b84091f570a8dce",
-    "checkpoint_pointer_sha256": "da49a267e516137c0eda63d197847ac307782f4cf31a8ea6624c04a1d4375c91",
-    "checkpoint_sequence": 1,
-    "checkpoint_key": "operations/enrichment/runs/xtap-b0e01d0c7828c30def3948bcb4251e68/checkpoints/sha256-2fce6497a6f99ca29439e57178122c9692a10c5cc01ad61f07a2ef7c265e4527/checkpoint.hfjob",
-    "checkpoint_sha256": "2fce6497a6f99ca29439e57178122c9692a10c5cc01ad61f07a2ef7c265e4527",
-    "checkpoint_bytes": 40241
+    "checkpoint_pointer_sha256": "55bfdde220bcf25eb008481e587d8292a13aa3ed956e9b92ebc51a6f32dab696",
+    "checkpoint_sequence": 237,
+    "checkpoint_key": "operations/enrichment/runs/xtap-b0e01d0c7828c30def3948bcb4251e68/checkpoints/sha256-30d20fbcf0ce6629c95fa0419e9e117cfcb7b4411a7414611b5df3dcf449d38c/checkpoint.hfjob",
+    "checkpoint_sha256": "30d20fbcf0ce6629c95fa0419e9e117cfcb7b4411a7414611b5df3dcf449d38c",
+    "checkpoint_bytes": 41668
   }
 }

space/src/configuration-tail.ts

@@ -0,0 +1,17 @@
+import type { BucketLog, BucketSnapshot } from "./bucket-log.js";
+
+/** Inspect only newly verified segments. Mixed append batches are not config writes. */
+export async function hasConfigurationWrites(
+  log: Pick<BucketLog, "loadSegment">,
+  base: BucketSnapshot,
+  target: BucketSnapshot,
+): Promise<boolean> {
+  const known = new Set(base.files.map((file) => file.key));
+  for (const file of target.files) {
+    if (known.has(file.key)) continue;
+    if (!file.key.includes("/config/") && !file.key.includes("/mixed/")) continue;
+    const segment = await log.loadSegment(file);
+    if (segment.operations.some((operation) => operation.mode === "write")) return true;
+  }
+  return false;
+}

space/src/server.ts

@@ -12,6 +12,7 @@ import { consumerFetch, withConsumerDeadline } from "./consumer-deadline.js";
 import { createApp } from "./app.js";
 import type { AppReadiness } from "./app.js";
 import { loadConfig } from "./config.js";
+import { hasConfigurationWrites } from "./configuration-tail.js";
 import { BucketLog, createRawBucketClient } from "./bucket-log.js";
 import type { BucketSnapshot } from "./bucket-log.js";
 import type { StorageState } from "./storage-state.js";
@@ -263,14 +264,6 @@ async function primeStorageTextCache(snapshot: BucketSnapshot, stage: string): P
   });
 }
 
-function hasConfigurationTail(base: BucketSnapshot, final: BucketSnapshot): boolean {
-  const baseKeys = new Set(base.files.map((file) => file.key));
-  return final.files.some(
-    (file) =>
-      !baseKeys.has(file.key) && (file.key.includes("/config/") || file.key.includes("/mixed/")),
-  );
-}
-
 function applyIndexStats(): void {
   const stats = index.stats();
   rebuilt = { files: stats.tweetFiles, tweets: stats.tweetRows };
@@ -304,7 +297,7 @@ async function refreshExternalEnrichment(): Promise<void> {
     if (!storageCredentialOk(storageCredential)) return;
     const baseSnapshot = index.sourceSnapshot();
     const discovered = await log.discoverSnapshot(baseSnapshot.files);
-    const configChanged = hasConfigurationTail(baseSnapshot, discovered.snapshot);
+    const configChanged = await hasConfigurationWrites(log, baseSnapshot, discovered.snapshot);
     let nextTaxonomy = taxonomy;
     if (configChanged) {
       await primeStorageTextCache(discovered.snapshot, "configuration-refresh");

space/tests/configuration-tail.test.ts

@@ -0,0 +1,80 @@
+import { describe, expect, it, vi } from "vitest";
+import type {
+  BucketOperation,
+  BucketSegment,
+  BucketSnapshot,
+  BucketSnapshotFile,
+} from "../src/bucket-log.js";
+import { hasConfigurationWrites } from "../src/configuration-tail.js";
+
+function file(kind: "config" | "mixed" | "tweet", n: number): BucketSnapshotFile {
+  const hash = String(n).padStart(64, "0");
+  return {
+    key: `v1/segments/${kind}/2026/09/08/1788868800000-00000000-0000-4000-8000-000000000000-${hash}.json.gz`,
+    oid: hash,
+    size: 100,
+    content_sha256: hash,
+  };
+}
+function snapshot(...files: BucketSnapshotFile[]): BucketSnapshot {
+  return {
+    schema_version: 1,
+    bucket: "owner/raw",
+    files: files.sort((a, b) => a.key.localeCompare(b.key)),
+  };
+}
+function reader(operations: BucketOperation[]) {
+  const segment: BucketSegment = {
+    schema_version: 1,
+    transaction_id: "00000000-0000-4000-8000-000000000000",
+    created_at: "2026-09-08T12:00:00.000Z",
+    operations,
+  };
+  return { loadSegment: vi.fn(() => Promise.resolve(segment)) };
+}
+const append: BucketOperation = {
+  mode: "append",
+  path: "enrichment/2026/09/enrichment-2026-09-08.jsonl",
+  lines: ["{}"],
+};
+const write: BucketOperation = { mode: "write", path: "config/labels.json", content: "{}" };
+
+describe("configuration changes in verified source tails", () => {
+  it("does not reread old mixed/config segments or new tweet-only segments", async () => {
+    const base = snapshot(file("config", 1), file("mixed", 2));
+    const log = reader([write]);
+    expect(await hasConfigurationWrites(log, base, snapshot(...base.files, file("tweet", 3)))).toBe(
+      false,
+    );
+    expect(log.loadSegment).not.toHaveBeenCalled();
+  });
+
+  it("keeps mixed enrichment append batches out of the full configuration refresh", async () => {
+    const old = file("mixed", 1),
+      added = file("mixed", 2);
+    const log = reader([
+      append,
+      { ...append, path: "enrichment/attempts/2026/09/attempts-2026-09-08.jsonl" },
+    ]);
+    expect(await hasConfigurationWrites(log, snapshot(old), snapshot(old, added))).toBe(false);
+    expect(log.loadSegment).toHaveBeenCalledExactlyOnceWith(added);
+  });
+
+  it.each(["config", "mixed"] as const)(
+    "detects actual writes in new %s segments, including late arrivals",
+    async (kind) => {
+      const old = file(kind, 2),
+        added = file(kind, 1);
+      const log = reader(kind === "mixed" ? [append, write] : [write]);
+      expect(await hasConfigurationWrites(log, snapshot(old), snapshot(added, old))).toBe(true);
+      expect(log.loadSegment).toHaveBeenCalledExactlyOnceWith(added);
+    },
+  );
+
+  it("does not mistake an unreadable new segment for unchanged configuration", async () => {
+    const log = { loadSegment: vi.fn(() => Promise.reject(new Error("checksum mismatch"))) };
+    await expect(
+      hasConfigurationWrites(log, snapshot(), snapshot(file("mixed", 1))),
+    ).rejects.toThrow("checksum mismatch");
+  });
+});