A lost rebuild claim reports the persisted catalog only when it is whole

Author: RhysSullivan

packages/core/sdk/src/executor.test.ts

@@ -1049,9 +1049,10 @@ describe("createExecutor", () => {
         expect(rowsAfter.map((row) => row.generation).sort(), "A did not replace the rows").toEqual(
           rowsBefore.map((row) => row.generation).sort(),
         );
-        // And A reported the persisted rows, not the ones it discovered and
-        // failed to write — so its caller cannot disagree with the next list.
-        // A discovered `extra`; persisted has no `extra`; the report must not.
+        // And A reported the persisted rows, VALIDATED against the manifest,
+        // not the ones it discovered and failed to write — so its caller
+        // cannot disagree with the next list. A discovered `extra`; persisted
+        // has no `extra`; the report must not.
         expect(
           reported.map((tool) => String(tool.name)).sort(),
           "refresh reports what is persisted",
@@ -1086,6 +1087,77 @@ describe("createExecutor", () => {
       }),
   );
 
+  // D1 commits the winner's manifest before its row batch. A loser that
+  // reads the rows in that window must NOT report the previous build's rows
+  // as if they were current: `describeAll` refuses that state (manifest names
+  // the new generation, rows carry the old), so the loser reports nothing.
+  it.effect("a lost claim reports nothing while the winner's rows have not landed", () =>
+    Effect.gen(function* () {
+      const config = makeTestConfig({ plugins: [demoPlugin] as const });
+      const raceState: { armed: boolean } = { armed: false };
+      const wrap = (inner: FumaDb): FumaDb =>
+        new Proxy(inner, {
+          get(target, prop) {
+            if (prop === "withContext") {
+              return (context: unknown) =>
+                wrap((target.withContext as (c: unknown) => FumaDb)(context));
+            }
+            if (prop === "transaction") {
+              return (run: (tx: FumaDb) => Promise<unknown>) =>
+                (target.transaction as (r: (tx: FumaDb) => Promise<unknown>) => Promise<unknown>)(
+                  (tx) => run(wrap(tx)),
+                );
+            }
+            if (prop === "replaceMany") {
+              return async (plan: unknown) => {
+                if (raceState.armed) {
+                  raceState.armed = false;
+                  // B claims AND has already committed its manifest for a
+                  // generation whose rows are not in the table yet — the D1
+                  // guard-then-batch window.
+                  await (target.updateMany as (t: unknown, q: unknown) => Promise<unknown>)(
+                    "connection",
+                    {
+                      where: (b: { (c: string, op: string, v: unknown): unknown }) =>
+                        b("integration", "=", String(INTEG)),
+                      set: {
+                        tools_synced_at: null,
+                        tools_rebuild: "build-B",
+                        tools_manifest: { generation: "gen-B", tools: 2, definitions: 6 },
+                      },
+                    },
+                  );
+                }
+                return (target.replaceMany as (p: unknown) => Promise<{ applied: boolean }>)(plan);
+              };
+            }
+            return Reflect.get(target, prop);
+          },
+        });
+      const executor = yield* createExecutor({ ...config, db: wrap(config.db) });
+      yield* executor.demo.seed();
+      yield* executor.connections.create({
+        owner: "org",
+        name: CONN,
+        integration: INTEG,
+        template: TEMPLATE,
+        from: { provider: ProviderKey.make("memory"), id: ProviderItemId.make("v") },
+      });
+      expect((yield* executor.tools.describeAll()).length).toBeGreaterThan(0);
+
+      raceState.armed = true;
+      const reported = yield* executor.connections.refresh({
+        owner: "org",
+        integration: INTEG,
+        name: CONN,
+      });
+      expect(raceState.armed, "B interleaved").toBe(false);
+      // The rows in the table are the OLD build's; the manifest is B's. That
+      // is not a servable catalog, and the loser must not pretend it is.
+      expect(reported, "nothing is reported while the winner lands").toEqual([]);
+    }),
+  );
+
   it.effect("execute dispatches a connection-produced tool to the owning plugin", () =>
     Effect.gen(function* () {
       const executor = yield* makeTestExecutor({

packages/core/sdk/src/executor.ts

@@ -3813,12 +3813,36 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           }),
         );
         // A build that lost its claim persisted nothing; what it discovered
-        // is not the catalog. Report what IS persisted — the winner's rows, or
-        // nothing if the winner is still landing — so a caller never sees
-        // tools the next list will contradict.
+        // is not the catalog. Report what IS persisted — but only if it is a
+        // whole build, checked the way every list checks it (manifest vs
+        // rows). A raw row scan could, on D1, catch the winner between its
+        // committed manifest and its row batch and hand back a catalog the
+        // next list refuses. While the winner is still landing the loser
+        // reports an EMPTY catalog rather than a wrong one; the caller's next
+        // list (or the stale scan) picks up the winner's finished build. This
+        // cannot go through `describeAll`: that runs the stale sync, which
+        // would re-enter this very connection's in-flight production.
         if (!applied) {
-          const persisted = yield* core.findMany("tool", { where });
-          return persisted.map((row) => rowToTool(row as ConnectionToolRow));
+          const [rows, definitionRows, connectionRow] = yield* Effect.all([
+            core.findMany("tool", { where, select: [...TOOL_INVOCATION_COLUMNS, "generation"] }),
+            core.findMany("definition", { where }),
+            findConnectionRow(ref),
+          ]);
+          const manifest = connectionRow
+            ? Option.getOrNull(
+                decodeCatalogManifest(decodeJsonColumn(connectionRow.tools_manifest)),
+              )
+            : null;
+          if (
+            !manifest ||
+            rows.length !== manifest.tools ||
+            definitionRows.length !== manifest.definitions ||
+            rows.some((row) => row.generation !== manifest.generation) ||
+            definitionRows.some((row) => row.generation !== manifest.generation)
+          ) {
+            return [];
+          }
+          return rows.map((row) => rowToTool(row));
         }
 
         return result.tools.map((tool: ToolDef) =>