Fix(datasets): never take this process's own publish for a foreign change

The watcher told its own publish apart by the fingerprint reloadRow
records, but that lands only after the reload finishes. A settle tick in
between — the state write and a slow reload outlast the settle window, as
on a loaded CI runner — saw new files under an old fingerprint and reloaded
them a second time as a "file change". Rows being published by this
process are now skipped by the watcher until that reload has recorded them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

common/dataset-updater.js

@@ -120,6 +120,10 @@ const writeState = (row, state) =>
 // The published files as this process last loaded them (mtime + size per
 // file), so the watcher can tell someone else's publish from its own.
 const loadedFingerprints = new Map();
+// Rows this process is publishing right now: from the first file move until
+// the reload has recorded the new fingerprints, the watcher must not take
+// the changing files for someone else's.
+const ownPublishes = new Set();
 const fingerprint = (row) => row.files.map((file) => {
     try {
         const stat = fs.statSync(path.join(row.dir, file));
@@ -253,6 +257,7 @@ export const updateDataset = async (row, { signal, reason = 'auto update', wait
         // publishes a whole set, and the next boot does so before trusting
         // the files.
         work.throwIfAborted();
+        ownPublishes.add(row.id);
         await writeState(row, { ...state, identifier: null, publishing: true });
         await publish(row, staged, { signal: work, lockLost: lock.lost });
         // A whole set is in place: record it unless the lock went meanwhile.
@@ -262,6 +267,7 @@ export const updateDataset = async (row, { signal, reason = 'auto update', wait
         await reloadRow(row, reason);
         return { updated: true, identifier: remote.identifier };
     } finally {
+        ownPublishes.delete(row.id);
         await fsp.rm(tempDir, { recursive: true, force: true });
         await lock.release().catch(() => {});
     }
@@ -443,6 +449,8 @@ export const watchDatasets = (rows, { intervalMs = 5000, settleMs = 1000 } = {})
                 timer.unref?.();
                 return;
             }
+            // This process's own publish records its files once reloaded.
+            if (ownPublishes.has(row.id)) return;
             if (current === loadedFingerprints.get(row.id) || !isPresent(row)) return;
             const published = Object.fromEntries(row.files.map((file) => [file, path.join(row.dir, file)]));
             try {

tests/dataset-updater.test.js

@@ -493,6 +493,23 @@ describe('watchDatasets', () => {
         }
     });
 
+    it('does not take its own publish for a foreign change while its reload is still running', async () => {
+        // A reload slower than the settle window: the watcher sees the new
+        // files before this process has recorded them as its own.
+        const { row, log } = makeRow();
+        row.reload = async (reason) => { await sleep(120); log.reloads.push(reason); };
+        await updateDataset(row);
+        const stop = watchDatasets([row], fast);
+        try {
+            row.version = 'v2';
+            await updateDataset(row);
+            await sleep(150);
+            assert.deepEqual(log.reloads, ['auto update', 'auto update']);
+        } finally {
+            stop();
+        }
+    });
+
     it('picks up files dropped into an empty dataset, once all are there', async () => {
         const { row, log } = makeRow({ enabled: () => false }); // no credentials: hand-placed only
         const stop = watchDatasets([row], fast);