fix(playground,test): withdraw a failed catalog publication before releasing its staging link; bind custom observeProgress

- #persistSnapshot rolls the sidecar back while the staging link still exists
  when the post-link directory fsync fails, so a concurrent reader keeps seeing
  an in-progress publication until the path is withdrawn instead of adopting a
  briefly singly linked file
- contractProgressObserver invokes a client's observeProgress method with the
  client as receiver

Author: ScriptedAlchemy

packages/agent-bundle/src/dev/playground/native-playground-service.ts

@@ -1149,10 +1149,19 @@ export class NativePlaygroundService {
         try { await handle.close(); }
         catch (error) { cleanupFailures.push(error); }
       }
+      // A failed publication withdraws its sidecar while the staging link still
+      // exists: concurrent readers keep seeing an in-progress (doubly linked)
+      // publication until the path is gone, never a settled singly linked file
+      // that is about to be rolled back.
+      if (primary !== undefined && created && publicationIdentity !== undefined) {
+        try { await this.#publicationReceipt(path, publicationIdentity, true, true).rollback(); }
+        catch (error) { cleanupFailures.push(error); }
+        created = false;
+      }
       try { await this.#catalogStorage.remove(temporary, { force: true }); }
       catch (error) { cleanupFailures.push(error); }
     }
-    if ((primary !== undefined || cleanupFailures.length > 0) && created && publicationIdentity !== undefined) {
+    if (primary === undefined && cleanupFailures.length > 0 && created && publicationIdentity !== undefined) {
       try { await this.#publicationReceipt(path, publicationIdentity, true, true).rollback(); }
       catch (error) { cleanupFailures.push(error); }
     }

packages/agent-bundle/src/test/contract.ts

@@ -1149,7 +1149,8 @@ const sdkProgressObserver = (
 export const contractProgressObserver = (
   client: ContractMatrixClient,
 ): ContractMatrixProgressSource['observeProgress'] => {
-  if (typeof client.observeProgress === 'function') return client.observeProgress;
+  const { observeProgress } = client;
+  if (typeof observeProgress === 'function') return (listener) => observeProgress.call(client, listener);
   const handlers = (client as { readonly _notificationHandlers?: unknown })._notificationHandlers;
   if (handlers instanceof Map) return sdkProgressObserver(handlers as Map<string, ClientNotificationHandler>);
   throw new Error(

packages/agent-bundle/tests/dev-contract-runner.test.ts

@@ -118,6 +118,24 @@ it('exposes live progress notifications through the session trace for lifecycle
   expect(seen).toEqual(['token-1']);
 });
 
+it('invokes a custom observeProgress method with the client as its receiver', () => {
+  class InstanceClient {
+    readonly listeners = new Set<(notification: { readonly params?: { readonly progressToken?: string | number } }) => void>();
+    observeProgress(listener: (notification: { readonly params?: { readonly progressToken?: string | number } }) => void): () => void {
+      this.listeners.add(listener);
+      return () => this.listeners.delete(listener);
+    }
+  }
+  const client = new InstanceClient();
+  const seen: unknown[] = [];
+  const stop = contractProgressObserver(client as unknown as ContractMatrixClient)((notification) =>
+    seen.push(notification.params?.progressToken));
+  for (const listener of client.listeners) listener({ params: { progressToken: 'bound' } });
+  stop();
+  expect(seen).toEqual(['bound']);
+  expect(client.listeners.size).toBe(0);
+});
+
 it('names the missing notification path for a client that is neither an SDK Client nor exposes observeProgress', () => {
   const bare = {} as ContractMatrixClient;
   expect(() => contractProgressObserver(bare))

packages/agent-bundle/tests/native-playground-service.test.ts

@@ -1599,6 +1599,76 @@ it('never adopts a staged sidecar that its publisher rolls back, and republishes
   }
 });
 
+it('withdraws a sidecar whose directory fsync fails before releasing its staging link, so a waiting reader never adopts it', async () => {
+  const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-fsync-withdrawn-'));
+  const catalogDirectory = join(root, 'catalog');
+  const reference = epoch('epoch-fsync-withdrawn', join(root, 'artifact'));
+  const sidecar = join(catalogDirectory, `${reference.epoch.id}.json`);
+  const directoryFailure = new Error('directory sync failed');
+  let signalLinked!: () => void;
+  const linked = new Promise<void>((resolvePromise) => { signalLinked = resolvePromise; });
+  let releaseDirectorySync!: () => void;
+  const directorySync = new Promise<void>((resolvePromise) => { releaseDirectorySync = resolvePromise; });
+  const winnerStorage: NativePlaygroundCatalogStorage = {
+    link: async (source, destination) => {
+      await link(source, destination);
+      signalLinked();
+    },
+    mkdir,
+    open: async (path, flags, mode) => {
+      const handle = await open(path, flags, mode);
+      return new Proxy(handle, {
+        get(target, property) {
+          if (property === 'sync' && String(path) === catalogDirectory) {
+            return async () => {
+              // The publisher stalls on its directory fsync after the link, then fails it.
+              await directorySync;
+              throw directoryFailure;
+            };
+          }
+          const value = Reflect.get(target, property, target);
+          return typeof value === 'function' ? value.bind(target) : value;
+        },
+      });
+    },
+    remove: rm,
+  } as NativePlaygroundCatalogStorage;
+  const serviceFor = (storage?: NativePlaygroundCatalogStorage): NativePlaygroundService => new NativePlaygroundService({
+    catalogDirectory,
+    ...(storage === undefined ? {} : { catalogStorage: storage }),
+    discover: async () => suite(),
+    inspectArtifact: async (candidate) => Object.freeze({
+      binding: Object.freeze({ manifestPath: 'agent-bundle.manifest.json', source: 'explicit' as const, targetDigests: candidate.epoch.targetDigests }),
+      root: candidate.root,
+    }),
+    planFixture: async () => fixturePlan,
+    projectRoot: '/project',
+  });
+  const winner = serviceFor(winnerStorage);
+  const loser = serviceFor();
+  try {
+    const winning = winner.catalog(reference);
+    await linked;
+    expect((await stat(sidecar)).nlink).toBe(2);
+    const losing = loser.catalog(reference);
+    await expect(Promise.race([
+      losing.then(() => 'adopted' as const),
+      new Promise<'pending'>((resolvePromise) => { setTimeout(() => resolvePromise('pending'), 150); }),
+    ])).resolves.toBe('pending');
+    releaseDirectorySync();
+    // The rollback's own directory fsync fails the same way, so both are retained.
+    await expect(winning).rejects.toMatchObject({ errors: [directoryFailure, directoryFailure] });
+    const adopted = await losing;
+    expect((await stat(sidecar)).nlink).toBe(1);
+    expect((await readdir(catalogDirectory)).filter((name) => name.includes('.stage-'))).toEqual([]);
+    expect(await loser.catalog(reference)).toEqual(adopted);
+    await Promise.all([winner.close(), loser.close()]);
+  } finally {
+    releaseDirectorySync();
+    await rm(root, { force: true, recursive: true });
+  }
+});
+
 it('still rejects a persisted catalog aliased by a hard link that is not an epoch staging file', async () => {
   const root = await mkdtemp(join(tmpdir(), 'agent-bundle-native-playground-aliased-catalog-'));
   const catalogDirectory = join(root, 'catalog');