Serialize deferred tombstone confirmations through the publish worker

Author: cursoragent

apps/server/src/relay/AgentAwarenessRelay.ts

@@ -272,6 +272,7 @@ export const make = Effect.gen(function* () {
   const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets);
   const activeSnapshotPublishedRef = yield* Ref.make(false);
   const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());
+  const pendingTombstoneConfirmsRef = yield* Ref.make(new Set<ThreadId>());
 
   const readSecretString = (name: string) =>
     secrets
@@ -306,7 +307,7 @@ export const make = Effect.gen(function* () {
       transformClient: relayEnvironmentClient(relayConfig.environmentCredential),
     }).pipe(Effect.provide(FetchHttpClient.layer));
 
-  // Assigned after publishThreadUnsafe below; indirection keeps the deferred
+  // Assigned after the publish worker below; indirection keeps the deferred
   // tombstone confirmation from making the definition self-referential.
   let confirmTombstoneLater: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
 
@@ -408,6 +409,20 @@ export const make = Effect.gen(function* () {
       // immediately deletes the thread from the lock-screen card mid-
       // conversation. Defer, re-resolve, and only tombstone if it holds;
       // genuinely deleted threads still propagate a few seconds later.
+      const alreadyPending = yield* Ref.modify(pendingTombstoneConfirmsRef, (pending) => {
+        if (pending.has(threadId)) {
+          return [true, pending] as const;
+        }
+        return [false, new Set(pending).add(threadId)] as const;
+      });
+      if (alreadyPending) {
+        yield* Effect.logDebug("agent activity tombstone confirmation already pending", {
+          environmentId,
+          threadId,
+          reason: snapshot.reason,
+        });
+        return;
+      }
       yield* Effect.logInfo("agent activity tombstone deferred pending confirmation", {
         environmentId,
         threadId,
@@ -442,30 +457,25 @@ export const make = Effect.gen(function* () {
     });
   });
 
-  confirmTombstoneLater = (threadId) =>
-    publishThreadUnsafe(threadId, { confirmTombstone: true }).pipe(
-      Effect.delay("5 seconds"),
-      Effect.catchCause((cause) =>
-        Effect.logWarning("deferred agent activity tombstone failed", {
-          threadId,
-          cause: Cause.pretty(cause),
-        }),
-      ),
-      Effect.asVoid,
-    );
-
-  const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) =>
-    publishThreadUnsafe(threadId).pipe(
+  const runPublishTask = (task: {
+    readonly threadId: ThreadId;
+    readonly confirmTombstone?: boolean;
+  }) =>
+    publishThreadUnsafe(task.threadId, { confirmTombstone: task.confirmTombstone === true }).pipe(
       Effect.catchCause((cause) => {
         return Effect.logWarning("agent activity publish failed", {
-          threadId,
+          threadId: task.threadId,
+          confirmTombstone: task.confirmTombstone === true,
           cause: Cause.pretty(cause),
         });
       }),
       Effect.withSpan("AgentAwarenessRelay.publishThread"),
       withRelayClientTracing,
     );
 
+  const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) =>
+    runPublishTask({ threadId });
+
   const publishActiveThreadsUnsafe = Effect.gen(function* () {
     const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe(
       Effect.orElseSucceed(() => false),
@@ -515,7 +525,23 @@ export const make = Effect.gen(function* () {
       }
     });
 
-  const worker = yield* makeDrainableWorker(publishThread);
+  const worker = yield* makeDrainableWorker(runPublishTask);
+
+  // Deferred confirmations re-enter the same publish worker so they serialize
+  // against event-driven publishes instead of racing them from a detached
+  // fiber; only the delay itself runs outside the queue.
+  confirmTombstoneLater = (threadId) =>
+    Effect.sleep("5 seconds").pipe(
+      Effect.andThen(
+        Ref.update(pendingTombstoneConfirmsRef, (pending) => {
+          const next = new Set(pending);
+          next.delete(threadId);
+          return next;
+        }),
+      ),
+      Effect.andThen(worker.enqueue({ threadId, confirmTombstone: true })),
+      Effect.asVoid,
+    );
 
   const start: AgentAwarenessRelay["Service"]["start"] = Effect.fn("AgentAwarenessRelay.start")(
     function* () {
@@ -567,7 +593,7 @@ export const make = Effect.gen(function* () {
           return Effect.logDebug("agent activity publishing queued thread publish", {
             eventType: event.type,
             threadId,
-          }).pipe(Effect.andThen(worker.enqueue(threadId)));
+          }).pipe(Effect.andThen(worker.enqueue({ threadId })));
         }),
       );
     },