fix(dispatch): keep deleted session projections dismissed

src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts

@@ -48,7 +48,31 @@ vi.mock('@/flow_chat/services/AgenticEventListener', () => ({
   },
 }));
 
+function runningOutboundRecord() {
+  return {
+    jobId: 'job-1',
+    sessionId: 'session-1',
+    target: {
+      kind: 'ssh' as const,
+      connectionId: 'ssh-1',
+      workspacePath: '/repo',
+      displayName: 'build-host',
+    },
+    sourceWorkspacePath: '/source',
+    workspacePath: '/repo',
+    promptPreview: 'Dispatch test',
+    title: 'Dispatch test',
+    agentType: 'agentic',
+    approvalPolicy: 'reject-and-report' as const,
+    lastCursor: 0,
+    lastState: 'running' as const,
+    createdAt: '2026-07-28T00:00:00Z',
+    updatedAt: '2026-07-28T00:00:01Z',
+  };
+}
+
 function registerRunningJob(): void {
+  mocks.listJobs.mockResolvedValue([runningOutboundRecord()]);
   dispatchJobStore.getState().registerJob({
     jobId: 'job-1',
     sessionId: 'session-1',
@@ -247,6 +271,14 @@ function status(
   };
 }
 
+function createDeferred<T>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>(resolvePromise => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
 describe('DispatchJobObserver', () => {
   beforeEach(() => {
     vi.useFakeTimers();
@@ -409,13 +441,49 @@ describe('DispatchJobObserver', () => {
       ...dispatchJobStore.getState().jobs['job-1'],
       state: 'submitting',
     });
+    mocks.listJobs.mockResolvedValue([]);
     const cleanup = installDispatchJobObserver(createContext());
 
     await vi.advanceTimersByTimeAsync(0);
     expect(mocks.status).not.toHaveBeenCalled();
     cleanup();
   });
 
+  it('does not recreate a dismissed projection from an in-flight status response', async () => {
+    registerRunningJob();
+    const deferred = createDeferred<DispatchStatusResponse>();
+    mocks.status.mockReturnValue(deferred.promise);
+    const context = createContext();
+    const cleanup = installDispatchJobObserver(context);
+
+    await vi.advanceTimersByTimeAsync(0);
+    expect(mocks.status).toHaveBeenCalledWith('job-1', 0);
+
+    dispatchJobStore.getState().dismissJob('job-1');
+    deferred.resolve(status({
+      cursor: 1,
+      events: [{
+        type: 'agentEvent',
+        timestamp: '2026-07-28T00:00:01Z',
+        event: {
+          id: 'event-after-delete',
+          frontendEventName: 'agentic://session-created',
+          frontendPayload: {
+            sessionId: 'session-1',
+            sessionName: 'Dispatch test',
+          },
+        },
+      }],
+    }));
+    await vi.advanceTimersByTimeAsync(0);
+
+    expect(mocks.dispatchExternal).not.toHaveBeenCalled();
+    expect(context.flowChatStore.applyDispatchSnapshot).not.toHaveBeenCalled();
+    expect(dispatchJobStore.getState().jobs['job-1']).toBeUndefined();
+    expect(dispatchJobStore.getState().dismissedJobIds).toContain('job-1');
+    cleanup();
+  });
+
   it('never restores an unowned legacy job into the current workspace', async () => {
     const record = {
       jobId: 'job-restored',

src/web-ui/src/features/dispatch/DispatchJobObserver.ts

@@ -27,10 +27,24 @@
 export const DISPATCH_JOB_POLL_INTERVAL_MS = 1800;
 
 type RefreshRequester = (jobId?: string) => void;
-let installedRefreshRequester: RefreshRequester | null = null;
+
+interface DispatchObserverLease {
+  requestRefresh: RefreshRequester;
+  dispose: () => void;
+}
+
+type DispatchObserverGlobal = typeof globalThis & {
+  __bitfunDispatchJobObserverLease__?: DispatchObserverLease;
+};
+
+function getDispatchObserverGlobal(): DispatchObserverGlobal {
+  return globalThis as DispatchObserverGlobal;
+}
 
 export function requestDispatchJobRefresh(jobId?: string): void {
-  installedRefreshRequester?.(jobId);
+  getDispatchObserverGlobal()
+    .__bitfunDispatchJobObserverLease__
+    ?.requestRefresh(jobId);
 }
 
 const RAW_EVENT_NAMES: Record<string, string> = {
@@ -149,6 +163,15 @@
   return error instanceof Error ? error.message : String(error);
 }
 
+function isJobStillObserved(job: DispatchObserverJob): boolean {
+  const state = dispatchJobStore.getState();
+  return (
+    state.jobs[job.jobId]?.sessionId === job.sessionId
+    && !state.dismissedJobIds.includes(job.jobId)
+    && !state.dismissedSessionIds.includes(job.sessionId)
+  );
+}
+
 export function dispatchEventId(event: DispatchEvent): string {
   if (event.type === 'agentEvent') {
     const envelope = event.event as DispatchAgentEventEnvelope;
@@ -163,6 +186,15 @@
   const sourceWorkspacePath = job.sourceWorkspacePath?.trim() || undefined;
   const existing = context.flowChatStore.getState().sessions.get(job.sessionId);
   if (existing) {
+    if (existing.config.dispatchJobId !== job.jobId) {
+      log.info('Dispatch diagnostic: observer adopted an existing flow chat session', {
+        jobId: job.jobId,
+        sessionId: job.sessionId,
+        previousDispatchJobId: existing.config.dispatchJobId,
+        wasHistorical: existing.isHistorical,
+        historyState: existing.historyState,
+      });
+    }
     // Reconcile both immutable target identity and controller-side ownership.
     // The observer can start before FlowChat knows its workspace, so a legacy
     // outbound record may only gain its source path on a later poll.
@@ -193,6 +225,12 @@
   // renderer process. Rebuild a fresh in-memory projection by replaying from
   // byte zero; never skip straight to that cursor.
   dispatchJobStore.getState().resetReplay(job.jobId);
+  log.info('Dispatch diagnostic: observer created a flow chat projection', {
+    jobId: job.jobId,
+    sessionId: job.sessionId,
+    sourceWorkspaceId: job.sourceWorkspaceId,
+    state: job.state,
+  });
   context.flowChatStore.addExternalSession(
     job.sessionId,
     job.title,
@@ -322,7 +360,14 @@
   });
 }
 
-async function refreshJob(context: FlowChatContext, requestedJobId: string): Promise<void> {
+async function refreshJob(
+  context: FlowChatContext,
+  requestedJobId: string,
+  isObserverCurrent: () => boolean,
+): Promise<void> {
+  if (!isObserverCurrent()) {
+    return;
+  }
   let job = dispatchJobStore.getState().jobs[requestedJobId];
   if (!job) {
     return;
@@ -348,13 +393,22 @@
   try {
     response = await dispatchApi.status(job.jobId, requestCursor);
   } catch (error) {
+    if (!isObserverCurrent() || !isJobStillObserved(job)) {
+      return;
+    }
     dispatchJobStore.getState().setTransportState(
       job.jobId,
       'unreachable',
       transportErrorMessage(error),
     );
     throw error;
   }
+  // Deleting a dispatch session writes a projection tombstone while an
+  // already-issued target poll may still be in flight. Never let that stale
+  // response project SessionCreated/DialogTurnStarted and recreate the row.
+  if (!isObserverCurrent() || !isJobStillObserved(job)) {
+    return;
+  }
   // A successful target status request is the only authoritative signal that
   // clears a transient transport failure. It does not alter the durable job
   // state beyond the snapshot applied below.
@@ -365,6 +419,9 @@
   const userCancelledBeforeRefresh =
     context.userCancelledSessionIds?.has(job.sessionId) ?? false;
   for (const event of response.events) {
+    if (!isObserverCurrent() || !isJobStillObserved(job)) {
+      return;
+    }
     const eventId = dispatchEventId(event);
     if (dispatchJobStore.getState().hasAppliedEvent(job.jobId, eventId)) {
       continue;
@@ -378,6 +435,9 @@
       appliedEventIds: [eventId],
     });
   }
+  if (!isObserverCurrent() || !isJobStillObserved(job)) {
+    return;
+  }
 
   const terminalDrained =
     isDispatchJobTerminal(response.state) &&
@@ -511,13 +571,28 @@
 }
 
 export function installDispatchJobObserver(context: FlowChatContext): () => void {
+  const observerGlobal = getDispatchObserverGlobal();
+  const previousLease = observerGlobal.__bitfunDispatchJobObserverLease__;
+  if (previousLease) {
+    log.info('Replacing an existing dispatch job observer');
+    previousLease.dispose();
+  }
+
   let disposed = false;
   let inFlight = false;
   let queuedJobId: string | undefined;
   let immediateTimer: ReturnType<typeof setTimeout> | null = null;
+  let interval: ReturnType<typeof setInterval> | null = null;
+  let handleVisibilityChanged: (() => void) | null = null;
+  let lease: DispatchObserverLease;
+
+  const ownsLease = (): boolean => (
+    !disposed
+    && observerGlobal.__bitfunDispatchJobObserverLease__ === lease
+  );
 
   async function run(requestedJobId?: string): Promise<void> {
-    if (disposed || isPeerDeviceModeActive()) return;
+    if (!ownsLease() || isPeerDeviceModeActive()) return;
     if (inFlight) {
       queuedJobId = requestedJobId;
       return;
@@ -526,17 +601,29 @@
     inFlight = true;
     try {
       const records = await dispatchApi.listJobs();
+      if (!ownsLease()) {
+        return;
+      }
       dispatchJobStore.getState().mergeOutboundRecords(records);
       const jobs = Object.values(dispatchJobStore.getState().jobs)
         .filter(job => !requestedJobId || job.jobId === requestedJobId);
       for (const job of jobs) {
+        if (!ownsLease()) {
+          return;
+        }
         try {
-          await refreshJob(context, job.jobId);
+          await refreshJob(context, job.jobId, ownsLease);
         } catch (error) {
+          if (!ownsLease()) {
+            return;
+          }
           log.warn('Dispatch job refresh failed', { jobId: job.jobId, error });
         }
       }
     } catch (error) {
+      if (!ownsLease()) {
+        return;
+      }
       const message = transportErrorMessage(error);
       const jobs = Object.values(dispatchJobStore.getState().jobs)
         .filter(job => (
@@ -554,7 +641,7 @@
       log.warn('Failed to reconcile outbound dispatch jobs', { error });
     } finally {
       inFlight = false;
-      if (queuedJobId !== undefined && !disposed) {
+      if (queuedJobId !== undefined && ownsLease()) {
         const next = queuedJobId;
         queuedJobId = undefined;
         schedule(next);
@@ -563,7 +650,7 @@
   }
 
   function schedule(jobId?: string): void {
-    if (disposed) return;
+    if (!ownsLease()) return;
     if (immediateTimer !== null) {
       clearTimeout(immediateTimer);
     }
@@ -573,11 +660,36 @@
     }, 0);
   }
 
-  installedRefreshRequester = schedule;
-  const interval = setInterval(() => {
+  const dispose = (): void => {
+    if (disposed) {
+      return;
+    }
+    disposed = true;
+    if (observerGlobal.__bitfunDispatchJobObserverLease__ === lease) {
+      delete observerGlobal.__bitfunDispatchJobObserverLease__;
+    }
+    if (immediateTimer !== null) {
+      clearTimeout(immediateTimer);
+      immediateTimer = null;
+    }
+    if (interval !== null) {
+      clearInterval(interval);
+      interval = null;
+    }
+    if (typeof document !== 'undefined' && handleVisibilityChanged) {
+      document.removeEventListener('visibilitychange', handleVisibilityChanged);
+    }
+  };
+  lease = {
+    requestRefresh: schedule,
+    dispose,
+  };
+  observerGlobal.__bitfunDispatchJobObserverLease__ = lease;
+
+  interval = setInterval(() => {
     void run();
   }, DISPATCH_JOB_POLL_INTERVAL_MS);
-  const handleVisibilityChanged = () => {
+  handleVisibilityChanged = () => {
     if (typeof document === 'undefined' || document.visibilityState === 'visible') {
       schedule();
     }
@@ -587,17 +699,5 @@
   }
   schedule();
 
-  return () => {
-    disposed = true;
-    if (installedRefreshRequester === schedule) {
-      installedRefreshRequester = null;
-    }
-    if (immediateTimer !== null) {
-      clearTimeout(immediateTimer);
-    }
-    clearInterval(interval);
-    if (typeof document !== 'undefined') {
-      document.removeEventListener('visibilitychange', handleVisibilityChanged);
-    }
-  };
+  return dispose;
 }