diff --git a/src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts b/src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts index 541d2cb920..9b1e5a56dd 100644 --- a/src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts +++ b/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() { + let resolve!: (value: T) => void; + const promise = new Promise(resolvePromise => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + describe('DispatchJobObserver', () => { beforeEach(() => { vi.useFakeTimers(); @@ -409,6 +441,7 @@ describe('DispatchJobObserver', () => { ...dispatchJobStore.getState().jobs['job-1'], state: 'submitting', }); + mocks.listJobs.mockResolvedValue([]); const cleanup = installDispatchJobObserver(createContext()); await vi.advanceTimersByTimeAsync(0); @@ -416,6 +449,41 @@ describe('DispatchJobObserver', () => { cleanup(); }); + it('does not recreate a dismissed projection from an in-flight status response', async () => { + registerRunningJob(); + const deferred = createDeferred(); + 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', diff --git a/src/web-ui/src/features/dispatch/DispatchJobObserver.ts b/src/web-ui/src/features/dispatch/DispatchJobObserver.ts index fb5ea57754..3622a9f9d5 100644 --- a/src/web-ui/src/features/dispatch/DispatchJobObserver.ts +++ b/src/web-ui/src/features/dispatch/DispatchJobObserver.ts @@ -27,10 +27,24 @@ const log = createLogger('DispatchJobObserver'); 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 = { @@ -149,6 +163,15 @@ function transportErrorMessage(error: unknown): string { 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 @@ function ensureProjection(context: FlowChatContext, job: DispatchObserverJob): b 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 @@ function ensureProjection(context: FlowChatContext, job: DispatchObserverJob): b // 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 @@ function reconcileDispatchTerminalRuntime( }); } -async function refreshJob(context: FlowChatContext, requestedJobId: string): Promise { +async function refreshJob( + context: FlowChatContext, + requestedJobId: string, + isObserverCurrent: () => boolean, +): Promise { + if (!isObserverCurrent()) { + return; + } let job = dispatchJobStore.getState().jobs[requestedJobId]; if (!job) { return; @@ -348,6 +393,9 @@ async function refreshJob(context: FlowChatContext, requestedJobId: string): Pro try { response = await dispatchApi.status(job.jobId, requestCursor); } catch (error) { + if (!isObserverCurrent() || !isJobStillObserved(job)) { + return; + } dispatchJobStore.getState().setTransportState( job.jobId, 'unreachable', @@ -355,6 +403,12 @@ async function refreshJob(context: FlowChatContext, requestedJobId: string): Pro ); 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 @@ async function refreshJob(context: FlowChatContext, requestedJobId: string): Pro 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 @@ async function refreshJob(context: FlowChatContext, requestedJobId: string): Pro appliedEventIds: [eventId], }); } + if (!isObserverCurrent() || !isJobStillObserved(job)) { + return; + } const terminalDrained = isDispatchJobTerminal(response.state) && @@ -511,13 +571,28 @@ async function refreshJob(context: FlowChatContext, requestedJobId: string): Pro } 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 | null = null; + let interval: ReturnType | null = null; + let handleVisibilityChanged: (() => void) | null = null; + let lease: DispatchObserverLease; + + const ownsLease = (): boolean => ( + !disposed + && observerGlobal.__bitfunDispatchJobObserverLease__ === lease + ); async function run(requestedJobId?: string): Promise { - if (disposed || isPeerDeviceModeActive()) return; + if (!ownsLease() || isPeerDeviceModeActive()) return; if (inFlight) { queuedJobId = requestedJobId; return; @@ -526,17 +601,29 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void 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 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void 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 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void } function schedule(jobId?: string): void { - if (disposed) return; + if (!ownsLease()) return; if (immediateTimer !== null) { clearTimeout(immediateTimer); } @@ -573,11 +660,36 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void }, 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 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void } 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; } diff --git a/src/web-ui/src/features/dispatch/dispatchJobStore.test.ts b/src/web-ui/src/features/dispatch/dispatchJobStore.test.ts index 2f2a291156..358a067464 100644 --- a/src/web-ui/src/features/dispatch/dispatchJobStore.test.ts +++ b/src/web-ui/src/features/dispatch/dispatchJobStore.test.ts @@ -240,6 +240,7 @@ describe('dispatchJobStore', () => { workspacePath: '/repo', displayName: 'build-host', }, + sourceWorkspacePath: '/source', workspacePath: '/repo', promptPreview: 'Dispatch test', lastCursor: 10, @@ -250,6 +251,31 @@ describe('dispatchJobStore', () => { expect(dispatchJobStore.getState().jobs['job-1']).toBeUndefined(); expect(dispatchJobStore.getState().dismissedJobIds).toContain('job-1'); + expect(dispatchJobStore.getState().dismissedSessionIds).toContain('session-1'); + }); + + it('uses a session tombstone when deletion happens before the job id is known', () => { + dispatchJobStore.getState().dismissSession('session-late'); + dispatchJobStore.getState().mergeOutboundRecords([{ + jobId: 'job-late', + sessionId: 'session-late', + target: { + kind: 'ssh', + connectionId: 'ssh-1', + workspacePath: '/repo', + displayName: 'build-host', + }, + sourceWorkspacePath: '/source', + workspacePath: '/repo', + promptPreview: 'Dispatch test', + lastCursor: 0, + lastState: 'running', + createdAt: '2026-07-28T00:00:00Z', + updatedAt: '2026-07-28T00:00:01Z', + }]); + + expect(dispatchJobStore.getState().jobs['job-late']).toBeUndefined(); + expect(dispatchJobStore.getState().dismissedSessionIds).toContain('session-late'); }); it('keeps transport reachability transient and separate from authoritative job state', () => { @@ -271,5 +297,6 @@ describe('dispatchJobStore', () => { dispatchJobStore.getState(), ) as Record | undefined; expect(persistedState?.transportByJobId).toBeUndefined(); + expect(persistedState?.dismissedSessionIds).toEqual([]); }); }); diff --git a/src/web-ui/src/features/dispatch/dispatchJobStore.ts b/src/web-ui/src/features/dispatch/dispatchJobStore.ts index ce707412c9..fa82a1d5f2 100644 --- a/src/web-ui/src/features/dispatch/dispatchJobStore.ts +++ b/src/web-ui/src/features/dispatch/dispatchJobStore.ts @@ -2,7 +2,6 @@ import { create } from 'zustand'; import { createJSONStorage, persist, - type StateStorage, } from 'zustand/middleware'; import type { DispatchApprovalPolicy, @@ -14,11 +13,32 @@ import type { OutboundDispatchRecord, } from './types'; import { isDispatchJobTerminal } from './types'; +import { createLogger } from '@/shared/utils/logger'; +const log = createLogger('DispatchJobStore'); const MAX_APPLIED_EVENT_IDS = 2048; const MAX_DISMISSED_JOB_IDS = 2048; +const MAX_DISMISSED_SESSION_IDS = 2048; +const DISPATCH_JOB_STORAGE_KEY = 'bitfun-dispatch-jobs-v1'; +// Keep deletion authority outside the general Zustand snapshot. A stale HMR +// renderer may still persist its old job cache, but it must never erase a +// dismissal recorded by the current renderer. +const DISPATCH_DISMISSAL_LEDGER_KEY = 'bitfun-dispatch-dismissals-v1'; +const reportedSuppressedProjectionKeys = new Set(); const fallbackStorageValues = new Map(); -const fallbackStorage: StateStorage = { + +interface SyncStringStorage { + getItem: (name: string) => string | null; + setItem: (name: string, value: string) => void; + removeItem: (name: string) => void; +} + +interface DispatchDismissalLedger { + dismissedJobIds: string[]; + dismissedSessionIds: string[]; +} + +const fallbackStorage: SyncStringStorage = { getItem: (name) => fallbackStorageValues.get(name) ?? null, setItem: (name, value) => { fallbackStorageValues.set(name, value); @@ -28,6 +48,84 @@ const fallbackStorage: StateStorage = { }, }; +function getDispatchStorage(): SyncStringStorage { + return typeof localStorage === 'undefined' ? fallbackStorage : localStorage; +} + +function mergeDismissedIds( + current: unknown[], + additions: unknown[], + limit: number, +): string[] { + return Array.from(new Set( + [...current, ...additions] + .filter((value): value is string => typeof value === 'string') + .map(value => value.trim()) + .filter(Boolean), + )).slice(-limit); +} + +function readDismissalLedger(): DispatchDismissalLedger { + try { + const raw = getDispatchStorage().getItem(DISPATCH_DISMISSAL_LEDGER_KEY); + if (!raw) { + return { dismissedJobIds: [], dismissedSessionIds: [] }; + } + const parsed = JSON.parse(raw) as Partial; + return { + dismissedJobIds: mergeDismissedIds( + [], + Array.isArray(parsed.dismissedJobIds) ? parsed.dismissedJobIds : [], + MAX_DISMISSED_JOB_IDS, + ), + dismissedSessionIds: mergeDismissedIds( + [], + Array.isArray(parsed.dismissedSessionIds) ? parsed.dismissedSessionIds : [], + MAX_DISMISSED_SESSION_IDS, + ), + }; + } catch (error) { + log.warn('Failed to read dispatch dismissal ledger', { error }); + return { dismissedJobIds: [], dismissedSessionIds: [] }; + } +} + +function recordDismissals( + jobIds: string[], + sessionIds: string[], +): DispatchDismissalLedger { + const current = readDismissalLedger(); + const next = { + dismissedJobIds: mergeDismissedIds( + current.dismissedJobIds, + jobIds, + MAX_DISMISSED_JOB_IDS, + ), + dismissedSessionIds: mergeDismissedIds( + current.dismissedSessionIds, + sessionIds, + MAX_DISMISSED_SESSION_IDS, + ), + }; + try { + getDispatchStorage().setItem( + DISPATCH_DISMISSAL_LEDGER_KEY, + JSON.stringify(next), + ); + } catch (error) { + log.error('Failed to persist dispatch dismissal ledger', { error }); + } + return next; +} + +function clearDismissalLedger(): void { + try { + getDispatchStorage().removeItem(DISPATCH_DISMISSAL_LEDGER_KEY); + } catch (error) { + log.warn('Failed to clear dispatch dismissal ledger', { error }); + } +} + export interface DispatchObserverJob { jobId: string; sessionId: string; @@ -69,6 +167,11 @@ interface DispatchJobStoreState { transportByJobId: Record; /** Local projection tombstones. The target job remains durable, but must not reopen in navigation. */ dismissedJobIds: string[]; + /** + * Session-level tombstones cover incomplete projections where the job id is + * temporarily missing when the user deletes the navigation row. + */ + dismissedSessionIds: string[]; registerJob: (job: DispatchObserverJob) => void; mergeOutboundRecords: (records: OutboundDispatchRecord[]) => void; updateProgress: ( @@ -96,6 +199,7 @@ interface DispatchJobStoreState { updateTitle: (jobId: string, title: string) => void; updateModel: (jobId: string, model: string) => void; updateApprovalPolicy: (jobId: string, policy: DispatchApprovalPolicy) => void; + dismissSession: (sessionId: string, knownJobId?: string) => void; dismissJob: (jobId: string) => void; removeJob: (jobId: string) => void; clear: () => void; @@ -130,15 +234,41 @@ function requestFromTarget(target: DispatchTarget): DispatchTargetRequest { } } +const initialDismissalLedger = readDismissalLedger(); + export const useDispatchJobStore = create()( persist( (set, get) => ({ jobs: {}, transportByJobId: {}, - dismissedJobIds: [], + dismissedJobIds: initialDismissalLedger.dismissedJobIds, + dismissedSessionIds: initialDismissalLedger.dismissedSessionIds, registerJob: (job) => { set(state => { + const ledger = readDismissalLedger(); + const dismissedJobIds = mergeDismissedIds( + state.dismissedJobIds, + ledger.dismissedJobIds, + MAX_DISMISSED_JOB_IDS, + ); + const dismissedSessionIds = mergeDismissedIds( + state.dismissedSessionIds, + ledger.dismissedSessionIds, + MAX_DISMISSED_SESSION_IDS, + ); + if ( + dismissedJobIds.includes(job.jobId) + || dismissedSessionIds.includes(job.sessionId) + ) { + log.info('Dispatch diagnostic: job registration suppressed by tombstone', { + jobId: job.jobId, + sessionId: job.sessionId, + jobTombstoned: dismissedJobIds.includes(job.jobId), + sessionTombstoned: dismissedSessionIds.includes(job.sessionId), + }); + return { dismissedJobIds, dismissedSessionIds }; + } const transportByJobId = { ...state.transportByJobId, [job.jobId]: state.transportByJobId[job.jobId] ?? { @@ -155,31 +285,63 @@ export const useDispatchJobStore = create()( }, }, transportByJobId, - dismissedJobIds: state.dismissedJobIds.filter(id => id !== job.jobId), + dismissedJobIds, + dismissedSessionIds, }; }); }, mergeOutboundRecords: (records) => { set(state => { + const ledger = readDismissalLedger(); + const dismissedJobIds = mergeDismissedIds( + state.dismissedJobIds, + ledger.dismissedJobIds, + MAX_DISMISSED_JOB_IDS, + ); + const dismissedSessionIds = mergeDismissedIds( + state.dismissedSessionIds, + ledger.dismissedSessionIds, + MAX_DISMISSED_SESSION_IDS, + ); const jobs = { ...state.jobs }; const authoritativeJobIds = new Set(records.map(record => record.jobId)); const prunedJobIds = new Set(); for (const [jobId, job] of Object.entries(jobs)) { if ( - !authoritativeJobIds.has(jobId) - && job.state !== 'submitting' - && job.state !== 'submission_unknown' + dismissedJobIds.includes(jobId) + || dismissedSessionIds.includes(job.sessionId) + || ( + !authoritativeJobIds.has(jobId) + && job.state !== 'submitting' + && job.state !== 'submission_unknown' + ) ) { - // The controller index is authoritative after acknowledgement. - // Remove renderer cache left behind by retention, manual cleanup, - // or an older build instead of restoring a ghost projection. + // The controller index is authoritative after acknowledgement, + // while a tombstone also wins over cache rehydrated by an older + // renderer build. delete jobs[jobId]; prunedJobIds.add(jobId); } } for (const record of records) { - if (state.dismissedJobIds.includes(record.jobId)) { + if ( + dismissedJobIds.includes(record.jobId) + || dismissedSessionIds.includes(record.sessionId) + ) { + const projectionKey = `${record.jobId}:${record.sessionId}`; + if (!reportedSuppressedProjectionKeys.has(projectionKey)) { + if (reportedSuppressedProjectionKeys.size >= MAX_DISMISSED_JOB_IDS) { + reportedSuppressedProjectionKeys.clear(); + } + reportedSuppressedProjectionKeys.add(projectionKey); + log.info('Dispatch diagnostic: outbound record suppressed by tombstone', { + jobId: record.jobId, + sessionId: record.sessionId, + jobTombstoned: dismissedJobIds.includes(record.jobId), + sessionTombstoned: dismissedSessionIds.includes(record.sessionId), + }); + } continue; } const sourceWorkspacePath = record.sourceWorkspacePath?.trim() || undefined; @@ -220,6 +382,12 @@ export const useDispatchJobStore = create()( }; continue; } + log.info('Dispatch diagnostic: outbound record restored into renderer cache', { + jobId: record.jobId, + sessionId: record.sessionId, + sourceWorkspaceId: record.sourceWorkspaceId, + state: record.lastState, + }); jobs[record.jobId] = { jobId: record.jobId, sessionId: record.sessionId, @@ -253,7 +421,12 @@ export const useDispatchJobStore = create()( for (const jobId of Object.keys(jobs)) { transportByJobId[jobId] ??= { reachability: 'unknown' }; } - return { jobs, transportByJobId }; + return { + jobs, + transportByJobId, + dismissedJobIds, + dismissedSessionIds, + }; }); }, @@ -401,19 +574,72 @@ export const useDispatchJobStore = create()( }); }, - dismissJob: (jobId) => { + dismissSession: (rawSessionId, knownJobId) => { + const sessionId = rawSessionId.trim(); + const normalizedKnownJobId = knownJobId?.trim(); + const matchingJobIds = Object.values(get().jobs) + .filter(job => job.sessionId === sessionId) + .map(job => job.jobId); + const ledger = recordDismissals( + [ + ...matchingJobIds, + ...(normalizedKnownJobId ? [normalizedKnownJobId] : []), + ], + sessionId ? [sessionId] : [], + ); set(state => { + const dismissedJobIds = new Set(mergeDismissedIds( + state.dismissedJobIds, + ledger.dismissedJobIds, + MAX_DISMISSED_JOB_IDS, + )); + for (const job of Object.values(state.jobs)) { + if (job.sessionId === sessionId) { + dismissedJobIds.add(job.jobId); + } + } + const jobs = { ...state.jobs }; const transportByJobId = { ...state.transportByJobId }; - delete jobs[jobId]; - delete transportByJobId[jobId]; + for (const jobId of dismissedJobIds) { + delete jobs[jobId]; + delete transportByJobId[jobId]; + } + return { jobs, transportByJobId, - dismissedJobIds: Array.from(new Set([...state.dismissedJobIds, jobId])) + dismissedJobIds: Array.from(dismissedJobIds) .slice(-MAX_DISMISSED_JOB_IDS), + dismissedSessionIds: mergeDismissedIds( + state.dismissedSessionIds, + ledger.dismissedSessionIds, + MAX_DISMISSED_SESSION_IDS, + ), }; }); + const state = get(); + log.info('Dispatch diagnostic: projection dismissed', { + sessionId, + knownJobId: normalizedKnownJobId, + matchingJobIds, + persistedJobTombstone: normalizedKnownJobId + ? state.dismissedJobIds.includes(normalizedKnownJobId) + : false, + persistedSessionTombstone: state.dismissedSessionIds.includes(sessionId), + ledgerJobTombstone: normalizedKnownJobId + ? ledger.dismissedJobIds.includes(normalizedKnownJobId) + : matchingJobIds.length > 0 + && matchingJobIds.every(jobId => ledger.dismissedJobIds.includes(jobId)), + ledgerSessionTombstone: ledger.dismissedSessionIds.includes(sessionId), + dismissedJobCount: state.dismissedJobIds.length, + dismissedSessionCount: state.dismissedSessionIds.length, + }); + }, + + dismissJob: (jobId) => { + const sessionId = get().jobs[jobId]?.sessionId ?? ''; + get().dismissSession(sessionId, jobId); }, removeJob: (jobId) => { @@ -427,18 +653,67 @@ export const useDispatchJobStore = create()( }); }, - clear: () => set({ jobs: {}, transportByJobId: {}, dismissedJobIds: [] }), + clear: () => { + clearDismissalLedger(); + set({ + jobs: {}, + transportByJobId: {}, + dismissedJobIds: [], + dismissedSessionIds: [], + }); + }, }), { - name: 'bitfun-dispatch-jobs-v1', + name: DISPATCH_JOB_STORAGE_KEY, version: 1, - storage: createJSONStorage(() => ( - typeof localStorage === 'undefined' ? fallbackStorage : localStorage - )), + storage: createJSONStorage(getDispatchStorage), partialize: state => ({ jobs: state.jobs, dismissedJobIds: state.dismissedJobIds, + dismissedSessionIds: state.dismissedSessionIds, }), + merge: (persistedState, currentState) => { + const persisted = (persistedState ?? {}) as Partial; + const ledger = recordDismissals( + persisted.dismissedJobIds ?? [], + persisted.dismissedSessionIds ?? [], + ); + const dismissedJobIds = mergeDismissedIds( + persisted.dismissedJobIds ?? [], + ledger.dismissedJobIds, + MAX_DISMISSED_JOB_IDS, + ); + const dismissedSessionIds = mergeDismissedIds( + persisted.dismissedSessionIds ?? [], + ledger.dismissedSessionIds, + MAX_DISMISSED_SESSION_IDS, + ); + const jobs = Object.fromEntries( + Object.entries(persisted.jobs ?? {}).filter(([, job]) => ( + !dismissedJobIds.includes(job.jobId) + && !dismissedSessionIds.includes(job.sessionId) + )), + ); + return { + ...currentState, + ...persisted, + jobs, + transportByJobId: {}, + dismissedJobIds, + dismissedSessionIds, + }; + }, + onRehydrateStorage: () => (state, error) => { + if (error) { + log.error('Dispatch diagnostic: projection state rehydration failed', { error }); + return; + } + log.info('Dispatch diagnostic: projection state rehydrated', { + jobIds: Object.keys(state?.jobs ?? {}), + dismissedJobIds: state?.dismissedJobIds ?? [], + dismissedSessionIds: state?.dismissedSessionIds ?? [], + }); + }, }, ), ); diff --git a/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStrip.tsx b/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStrip.tsx index 6eaaabb3a7..f380144914 100644 --- a/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStrip.tsx +++ b/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStrip.tsx @@ -22,7 +22,6 @@ import { Tooltip, IconButton } from '@/component-library'; import { useGitState } from '@/tools/git/hooks/useGitState'; import type { SessionExecutionTarget } from '@/infrastructure/api/service-api/WorktreeAPI'; import { useI18n } from '@/infrastructure/i18n'; -import { DispatchTargetPicker } from '@/features/dispatch/DispatchTargetPicker'; import { DispatchResultDialog } from '@/features/dispatch/DispatchResultDialog'; import type { DispatchSelection, DispatchTarget } from '@/features/dispatch/types'; import './ChatInputWorkspaceStrip.scss'; @@ -136,8 +135,8 @@ export const ChatInputWorkspaceStrip: React.FC = ( const showUsage = usageReport?.visible && !!usageReport.onOpen; const showGoal = threadGoal?.visible && !!threadGoal.onOpen; const showPermission = !!permissionControl; - const showDispatch = !!dispatchControl; - const showRightActions = showDispatch || showPermission || showUsage || showGoal; + const showDispatchResult = !!dispatchControl?.completedSnapshotJobId; + const showRightActions = showDispatchResult || showPermission || showUsage || showGoal; const isWorktree = !!executionTarget?.worktreeId; const worktreeEnabled = worktreeControl?.enabled ?? isWorktree; const worktreeEnabledRef = useRef(worktreeEnabled); @@ -326,15 +325,11 @@ export const ChatInputWorkspaceStrip: React.FC = ( {showRightActions ? (
- {dispatchControl ? ( - - ) : null} + {/* + * 0.2.15 release gate: dispatch session creation stays hidden while + * its lifecycle semantics stabilize. Restore DispatchTargetPicker + * here in a later release; existing result review remains available. + */} {dispatchControl?.completedSnapshotJobId ? ( <> diff --git a/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStripLayout.test.ts b/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStripLayout.test.ts index e73163423a..23ec349003 100644 --- a/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStripLayout.test.ts +++ b/src/web-ui/src/flow_chat/components/ChatInputWorkspaceStripLayout.test.ts @@ -17,13 +17,6 @@ function readWorkspaceStripComponent(): string { ).replace(/\r\n/g, '\n'); } -function readDispatchPickerStylesheet(): string { - return readFileSync( - fileURLToPath(new URL('../../features/dispatch/DispatchTargetPicker.scss', import.meta.url)), - 'utf8', - ).replace(/\r\n/g, '\n'); -} - describe('ChatInputWorkspaceStrip layout styles', () => { it('keeps the session usage action visible without overpowering the strip', () => { const stylesheet = readWorkspaceStripStylesheet(); @@ -52,21 +45,14 @@ describe('ChatInputWorkspaceStrip layout styles', () => { expect(stylesheet).toContain('display: none;'); }); - it('places dispatch first in right actions and protects the narrow layout', () => { + it('keeps dispatch session creation hidden for the 0.2.15 release', () => { const component = readWorkspaceStripComponent(); - const pickerStylesheet = readDispatchPickerStylesheet(); - const actionsStart = component.indexOf( - '
', - ); - const dispatchIndex = component.indexOf(' .dispatch-target-picker__chevron'); + expect(component).toContain('0.2.15 release gate'); + expect(component).toContain('Restore DispatchTargetPicker'); + expect(component).not.toContain(' ({ })); const dispatchStoreMocks = vi.hoisted(() => ({ + jobs: {} as Record, registerJob: vi.fn(), - dismissJob: vi.fn(), + dismissSession: vi.fn(), updateTitle: vi.fn(), })); @@ -513,6 +514,7 @@ describe('reloadSessionTitle', () => { describe('SessionModule historical session coordination', () => { beforeEach(() => { vi.useFakeTimers(); + dispatchStoreMocks.jobs = {}; }); afterEach(async () => { @@ -1055,7 +1057,36 @@ describe('SessionModule historical session coordination', () => { await deleteChatSession(context, session.sessionId); - expect(dispatchStoreMocks.dismissJob).toHaveBeenCalledWith('job-1'); + expect(dispatchStoreMocks.dismissSession).toHaveBeenCalledWith( + session.sessionId, + 'job-1', + ); + expect(flowChatStore.removeSession).toHaveBeenCalledWith( + session.sessionId, + { nextActiveSessionId: null }, + ); + expect(flowChatStore.deleteSession).not.toHaveBeenCalled(); + }); + + it('deletes a dispatch projection found only through the observer job index', async () => { + const session = createSession({ + sessionId: 'dispatch-session', + isHistorical: false, + config: { agentType: 'agentic' }, + }); + dispatchStoreMocks.jobs = { + 'job-1': { sessionId: session.sessionId }, + }; + const { context, flowChatStore } = createContext(session, { + activeSessionId: session.sessionId, + }); + + await deleteChatSession(context, session.sessionId); + + expect(dispatchStoreMocks.dismissSession).toHaveBeenCalledWith( + session.sessionId, + undefined, + ); expect(flowChatStore.removeSession).toHaveBeenCalledWith( session.sessionId, { nextActiveSessionId: null }, diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts index 097563d35a..2f47b7a954 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts @@ -60,6 +60,31 @@ const log = createLogger('SessionModule'); const pendingSessionCreations = new Map>(); const DISPATCH_OBSERVER_MAX_CONTEXT_TOKENS = 128128; +function isDispatchObserverProjection( + sessionId: string, + session: Session | undefined, +): boolean { + if ( + isNonLocalDispatchTarget(session?.config.dispatchTarget) + || Boolean(session?.config.dispatchJobId?.trim()) + ) { + return true; + } + + return Object.values(dispatchJobStore.getState().jobs) + .some(job => job.sessionId === sessionId); +} + +function dismissDispatchObserverProjection( + sessionId: string, + session: Session | undefined, +): void { + dispatchJobStore.getState().dismissSession( + sessionId, + session?.config.dispatchJobId, + ); +} + const getHydrationLocationKey = ( location: SessionHistoryHydrationLocation | undefined, ): string => location?.workspacePath @@ -993,11 +1018,22 @@ export async function deleteChatSession( && removedSessionIdSet.has(stateBeforeDelete.activeSessionId) ); const session = stateBeforeDelete.sessions.get(sessionId); - if (isNonLocalDispatchTarget(session?.config.dispatchTarget)) { - if (session?.config.dispatchJobId) { - dispatchJobStore.getState().dismissJob(session.config.dispatchJobId); - } - context.flowChatStore.removeSession( + const observerJobIds = Object.values(dispatchJobStore.getState().jobs) + .filter(job => job.sessionId === sessionId) + .map(job => job.jobId); + const deleteAsDispatchProjection = isDispatchObserverProjection(sessionId, session); + log.info('Dispatch diagnostic: session delete evaluated', { + sessionId, + sessionFound: Boolean(session), + dispatchTargetKind: session?.config.dispatchTarget?.kind, + dispatchJobId: session?.config.dispatchJobId, + observerJobIds, + deleteAsDispatchProjection, + cascadeSessionIds: removedSessionIds, + }); + if (deleteAsDispatchProjection) { + dismissDispatchObserverProjection(sessionId, session); + const locallyRemovedSessionIds = context.flowChatStore.removeSession( sessionId, removedActiveSession ? { nextActiveSessionId: null } : undefined, ); @@ -1006,8 +1042,17 @@ export async function deleteChatSession( cleanupSaveState(context, id); cleanupSessionBuffers(context, id); }); + log.info('Dispatch diagnostic: projection removed from flow chat store', { + sessionId, + locallyRemovedSessionIds, + activeSessionId: context.flowChatStore.getState().activeSessionId, + }); return; } + log.info('Dispatch diagnostic: delete routed to persisted backend session', { + sessionId, + hasWorkspacePath: Boolean(session && sessionProjectWorkspacePath(session)), + }); await context.flowChatStore.deleteSession( sessionId, removedActiveSession ? { nextActiveSessionId: null } : undefined, @@ -1044,10 +1089,8 @@ export async function archiveChatSession( && removedSessionIdSet.has(stateBeforeArchive.activeSessionId) ); - if (isNonLocalDispatchTarget(session.config.dispatchTarget)) { - if (session.config.dispatchJobId) { - dispatchJobStore.getState().dismissJob(session.config.dispatchJobId); - } + if (isDispatchObserverProjection(sessionId, session)) { + dismissDispatchObserverProjection(sessionId, session); context.flowChatStore.removeSession( sessionId, removedActiveSession ? { nextActiveSessionId: null } : undefined, diff --git a/src/web-ui/src/flow_chat/store/FlowChatStore.ts b/src/web-ui/src/flow_chat/store/FlowChatStore.ts index e2a436f17d..7211142d96 100644 --- a/src/web-ui/src/flow_chat/store/FlowChatStore.ts +++ b/src/web-ui/src/flow_chat/store/FlowChatStore.ts @@ -72,9 +72,43 @@ import { isDispatchJobTerminal, isNonLocalDispatchTarget, } from '@/features/dispatch/types'; +import { dispatchJobStore } from '@/features/dispatch/dispatchJobStore'; const log = createLogger('FlowChatStore'); +function logPersistedDispatchMetadataOverlap( + metadata: Record, + source: 'metadata-page' | 'metadata-list', +): void { + const sessionId = + typeof metadata.sessionId === 'string' ? metadata.sessionId : undefined; + if (!sessionId) return; + + const dispatchState = dispatchJobStore.getState(); + const observerJobIds = Object.values(dispatchState.jobs) + .filter(job => job.sessionId === sessionId) + .map(job => job.jobId); + const sessionTombstoned = dispatchState.dismissedSessionIds.includes(sessionId); + const metadataDispatchJobId = + typeof metadata.dispatchJobId === 'string' + ? metadata.dispatchJobId + : typeof metadata.dispatch_job_id === 'string' + ? metadata.dispatch_job_id + : undefined; + if (!sessionTombstoned && observerJobIds.length === 0 && !metadataDispatchJobId) { + return; + } + + log.info('Dispatch diagnostic: persisted backend metadata overlaps observer state', { + source, + sessionId, + metadataDispatchJobId, + observerJobIds, + sessionTombstoned, + dismissedSessionCount: dispatchState.dismissedSessionIds.length, + }); +} + function sameDispatchTargetIdentity( left: NonNullable, right: NonNullable, @@ -4070,6 +4104,7 @@ export class FlowChatStore { const processSession = async (metadata: any) => { try { + logPersistedDispatchMetadataOverlap(metadata, 'metadata-page'); if (surfaceGeneration !== this.surfaceGeneration) { return; } @@ -4445,6 +4480,7 @@ export class FlowChatStore { const processSession = async (metadata: any) => { try { + logPersistedDispatchMetadataOverlap(metadata, 'metadata-list'); const existingSession = this.state.sessions.get(metadata.sessionId); if (existingSession) { return;