From 77035c3e3b8bccda0ee04178bd176d2ab4cdad8d Mon Sep 17 00:00:00 2001 From: jln <85513960+jln13x@users.noreply.github.com> Date: Wed, 22 Jul 2026 21:54:52 +0300 Subject: [PATCH] fix reliable thread completion notifications --- .../src/electron/ElectronNotification.test.ts | 36 +- .../src/electron/ElectronNotification.ts | 75 ++-- apps/web/src/interactionSounds.test.ts | 13 - apps/web/src/interactionSounds.ts | 7 - apps/web/src/routes/__root.tsx | 190 ++++++++- .../src/threadCompletionNotifications.test.ts | 382 ++++++++++++++++++ apps/web/src/threadCompletionNotifications.ts | 241 +++++++++++ docs/personal-fork-changes.md | 4 + .../client-runtime/src/state/orchestration.ts | 8 + 9 files changed, 882 insertions(+), 74 deletions(-) create mode 100644 apps/web/src/threadCompletionNotifications.test.ts create mode 100644 apps/web/src/threadCompletionNotifications.ts diff --git a/apps/desktop/src/electron/ElectronNotification.test.ts b/apps/desktop/src/electron/ElectronNotification.test.ts index 1199faef99c..204f9c72129 100644 --- a/apps/desktop/src/electron/ElectronNotification.test.ts +++ b/apps/desktop/src/electron/ElectronNotification.test.ts @@ -4,20 +4,25 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import { vi } from "vite-plus/test"; -const { notificationInstances, notificationIsSupported } = vi.hoisted(() => ({ - notificationInstances: [] as Array<{ - options: unknown; - listeners: Map void>; - show: ReturnType; - }>, - notificationIsSupported: vi.fn(() => true), -})); +const { notificationInstances, notificationIsSupported, notificationShowOutcome } = vi.hoisted( + () => ({ + notificationInstances: [] as Array<{ + options: unknown; + listeners: Map void>; + show: ReturnType; + }>, + notificationIsSupported: vi.fn(() => true), + notificationShowOutcome: { current: "show" as "show" | "failed" }, + }), +); vi.mock("electron", () => ({ Notification: class Notification { static isSupported = notificationIsSupported; readonly listeners = new Map void>(); - readonly show = vi.fn(); + readonly show = vi.fn(() => { + this.listeners.get(notificationShowOutcome.current)?.(); + }); readonly options: unknown; constructor(options: unknown) { @@ -41,6 +46,7 @@ describe("ElectronNotification", () => { it.effect("shows a silent native notification on macOS and handles its click", () => { const onClick = vi.fn(); notificationInstances.length = 0; + notificationShowOutcome.current = "show"; return Effect.gen(function* () { const notifications = yield* ElectronNotification.ElectronNotification; @@ -70,4 +76,16 @@ describe("ElectronNotification", () => { assert.equal(notificationInstances.length, 0); }).pipe(Effect.provide(notificationLayer("linux"))); }); + + it.effect("reports a native failure instead of acknowledging delivery", () => { + notificationInstances.length = 0; + notificationShowOutcome.current = "failed"; + return Effect.gen(function* () { + const notifications = yield* ElectronNotification.ElectronNotification; + assert.isFalse( + yield* notifications.show({ title: "Thread finished", body: "Refactor", onClick: vi.fn() }), + ); + assert.equal(notificationInstances.length, 1); + }).pipe(Effect.provide(notificationLayer("darwin"))); + }); }); diff --git a/apps/desktop/src/electron/ElectronNotification.ts b/apps/desktop/src/electron/ElectronNotification.ts index 26a73e834cc..f9f2bce8bf7 100644 --- a/apps/desktop/src/electron/ElectronNotification.ts +++ b/apps/desktop/src/electron/ElectronNotification.ts @@ -5,6 +5,8 @@ import * as Layer from "effect/Layer"; import * as Electron from "electron"; +const NOTIFICATION_ACKNOWLEDGEMENT_TIMEOUT_MS = 5_000; + export interface ElectronNotificationInput { readonly title: string; readonly body: string; @@ -26,31 +28,56 @@ export const layer = Layer.effect( return ElectronNotification.of({ show: (input) => - Effect.sync(() => { - if (platform !== "darwin" || !Electron.Notification.isSupported()) { - return false; - } + platform !== "darwin" || !Electron.Notification.isSupported() + ? Effect.succeed(false) + : Effect.callback((resume) => { + let notification: Electron.Notification; + try { + notification = new Electron.Notification({ + title: input.title, + body: input.body, + silent: true, + }); + } catch { + resume(Effect.succeed(false)); + return; + } - try { - const notification = new Electron.Notification({ - title: input.title, - body: input.body, - silent: true, - }); - const release = () => activeNotifications.delete(notification); - notification.once("click", () => { - input.onClick(); - release(); - }); - notification.once("close", release); - notification.once("failed", release); - activeNotifications.add(notification); - notification.show(); - return true; - } catch { - return false; - } - }), + let acknowledged = false; + const acknowledge = (shown: boolean) => { + if (acknowledged) return; + acknowledged = true; + resume(Effect.succeed(shown)); + }; + const release = () => activeNotifications.delete(notification); + notification.once("show", () => acknowledge(true)); + notification.once("click", () => { + acknowledge(true); + input.onClick(); + release(); + }); + notification.once("close", () => { + acknowledge(false); + release(); + }); + notification.once("failed", () => { + acknowledge(false); + release(); + }); + activeNotifications.add(notification); + try { + notification.show(); + } catch { + release(); + acknowledge(false); + } + return Effect.sync(release); + }).pipe( + Effect.timeoutOrElse({ + duration: NOTIFICATION_ACKNOWLEDGEMENT_TIMEOUT_MS, + orElse: () => Effect.succeed(false), + }), + ), }); }), ); diff --git a/apps/web/src/interactionSounds.test.ts b/apps/web/src/interactionSounds.test.ts index 409ec00b4e6..b380a24bcd8 100644 --- a/apps/web/src/interactionSounds.test.ts +++ b/apps/web/src/interactionSounds.test.ts @@ -7,7 +7,6 @@ import { COMPLETION_SOUND_VOLUME, deriveInteractionSoundCues, deriveThreadFeedbackEvents, - shouldPostThreadCompletionNotification, } from "./interactionSounds"; function makeThread(overrides: Partial = {}): EnvironmentThreadShell { @@ -93,18 +92,6 @@ describe("interaction sounds", () => { expect(COMPLETION_SOUND_VOLUME).toBe(1.1); }); - it("posts completion notifications only when the flag and desktop bridge are available", () => { - expect( - shouldPostThreadCompletionNotification({ enabled: true, desktopBridgeAvailable: true }), - ).toBe(true); - expect( - shouldPostThreadCompletionNotification({ enabled: false, desktopBridgeAvailable: true }), - ).toBe(false); - expect( - shouldPostThreadCompletionNotification({ enabled: true, desktopBridgeAvailable: false }), - ).toBe(false); - }); - it("plays bloom when a thread starts requesting user input", () => { const thread = makeThread(); diff --git a/apps/web/src/interactionSounds.ts b/apps/web/src/interactionSounds.ts index f000a05010f..4c5296770c2 100644 --- a/apps/web/src/interactionSounds.ts +++ b/apps/web/src/interactionSounds.ts @@ -92,10 +92,3 @@ export function deriveInteractionSoundCues( ): InteractionSoundCue[] { return deriveThreadFeedbackEvents(previous, threads).map((event) => event.cue); } - -export function shouldPostThreadCompletionNotification(input: { - readonly enabled: boolean; - readonly desktopBridgeAvailable: boolean; -}): boolean { - return input.enabled && input.desktopBridgeAvailable; -} diff --git a/apps/web/src/routes/__root.tsx b/apps/web/src/routes/__root.tsx index 305b29d0b17..aa2a89b258c 100644 --- a/apps/web/src/routes/__root.tsx +++ b/apps/web/src/routes/__root.tsx @@ -1,4 +1,4 @@ -import { type ServerLifecycleWelcomePayload } from "@t3tools/contracts"; +import { type EnvironmentId, type ServerLifecycleWelcomePayload } from "@t3tools/contracts"; import { scopedProjectKey, scopeProjectRef } from "@t3tools/client-runtime/environment"; import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; import { @@ -44,7 +44,7 @@ import { syncBrowserChromeTheme } from "../hooks/useTheme"; import { configureClientTracing } from "../observability/clientTracing"; import { resolveInitialServerAuthGateState } from "../environments/primary"; import { hasHostedPairingRequest, isHostedStaticApp } from "../hostedPairing"; -import { shellEnvironment } from "../state/shell"; +import { environmentShell, environmentSnapshotAtom, shellEnvironment } from "../state/shell"; import { useAtomValue } from "@effect/atom-react"; import { useAtomCommand } from "../state/use-atom-command"; import { useEnvironments, usePrimaryEnvironment } from "../state/environments"; @@ -64,9 +64,18 @@ import { captureThreadSoundStateWhileSettingsHydrating, COMPLETION_SOUND_VOLUME, deriveThreadFeedbackEvents, - shouldPostThreadCompletionNotification, type ThreadSoundStateByKey, } from "../interactionSounds"; +import { + clearPendingThreadCompletionNotifications, + deliverPendingThreadCompletionNotifications, + initializeThreadCompletionNotificationState, + readThreadCompletionNotificationState, + reduceThreadCompletionNotificationEvents, + writeThreadCompletionNotificationState, + type ThreadCompletionNotificationState, +} from "../threadCompletionNotifications"; +import { orchestrationEnvironment } from "../state/orchestration"; import { createKeybindingsUpdateToastController, type KeybindingsUpdateToastController, @@ -164,6 +173,7 @@ function RootRouteView() { function ThreadCompletionFeedbackCoordinator() { const threads = useThreadShells(); const navigate = useNavigate(); + const { environments } = useEnvironments(); const completionSoundEnabled = useClientSettings((settings) => settings.enableCompletionSounds); const completionNotificationsEnabled = usePrimarySettings( (settings) => settings.enableMacosCompletionNotifications, @@ -201,27 +211,165 @@ function ThreadCompletionFeedbackCoordinator() { if (completionSoundEnabled) { play(event.cue, event.cue === "success" ? COMPLETION_SOUND_VOLUME : 1); } - const showNotification = window.desktopBridge?.showThreadCompletionNotification; - if ( - event.cue === "success" && - shouldPostThreadCompletionNotification({ - enabled: completionNotificationsEnabled, - desktopBridgeAvailable: typeof showNotification === "function", - }) && - showNotification - ) { - void showNotification({ - threadRef: { - environmentId: event.thread.environmentId, - threadId: event.thread.id, - }, - threadTitle: event.thread.title, - }).catch(() => undefined); - } } } previousStateRef.current = captureThreadSoundState(threads); - }, [completionNotificationsEnabled, completionSoundEnabled, settingsHydrated, threads]); + }, [completionSoundEnabled, settingsHydrated, threads]); + + if (typeof window.desktopBridge?.showThreadCompletionNotification !== "function") { + return null; + } + + return environments.map((environment) => ( + + )); +} + +function EnvironmentThreadCompletionNotificationCoordinator({ + environmentId, + enabled, + settingsHydrated, +}: { + readonly environmentId: EnvironmentId; + readonly enabled: boolean; + readonly settingsHydrated: boolean; +}) { + const snapshot = useAtomValue(environmentSnapshotAtom(environmentId)); + const shellState = useAtomValue(environmentShell.stateValueAtom(environmentId)); + const replayEvents = useAtomCommand(orchestrationEnvironment.replayEvents, { + label: "thread-completion-notifications:replay-events", + reportFailure: false, + reportDefect: false, + }); + const syncChainRef = useRef>(Promise.resolve()); + const memoryStateRef = useRef(null); + const retryAttemptRef = useRef(0); + const retryTimerRef = useRef(null); + const [retryGeneration, setRetryGeneration] = useState(0); + + useEffect( + () => () => { + if (retryTimerRef.current !== null) { + window.clearTimeout(retryTimerRef.current); + } + }, + [], + ); + + useEffect(() => { + const show = window.desktopBridge?.showThreadCompletionNotification; + if ( + !settingsHydrated || + snapshot === null || + shellState.status !== "live" || + typeof show !== "function" + ) { + return; + } + + const synchronize = async () => { + const readState = (): ThreadCompletionNotificationState | null => { + try { + return readThreadCompletionNotificationState(window.localStorage, environmentId); + } catch (cause) { + console.warn("Could not read the thread completion notification cursor.", { + environmentId, + cause, + }); + return memoryStateRef.current; + } + }; + const persistState = (state: ThreadCompletionNotificationState) => { + memoryStateRef.current = state; + try { + writeThreadCompletionNotificationState(window.localStorage, environmentId, state); + } catch (cause) { + console.warn("Could not persist the thread completion notification cursor.", { + environmentId, + cause, + }); + } + }; + + let state = readState(); + if (state === null || snapshot.snapshotSequence < state.cursor) { + state = initializeThreadCompletionNotificationState(snapshot); + persistState(state); + return; + } + + if (!enabled) { + state = clearPendingThreadCompletionNotifications(state); + retryAttemptRef.current = 0; + } + + const replayResult = await replayEvents({ + environmentId, + input: { fromSequenceExclusive: state.cursor }, + }); + if (replayResult._tag === "Failure") { + console.warn("Could not replay events for thread completion notifications.", { + environmentId, + cause: squashAtomCommandFailure(replayResult), + }); + persistState(state); + return; + } + + state = reduceThreadCompletionNotificationEvents({ + state, + events: replayResult.value, + snapshot, + notificationsEnabled: enabled, + }); + persistState(state); + + if (!enabled || state.pending.length === 0) return; + const delivery = await deliverPendingThreadCompletionNotifications({ + state, + environmentId, + show, + onProgress: persistState, + }); + memoryStateRef.current = delivery.state; + if (delivery.failed) { + console.warn("Could not show a thread completion notification; it remains pending.", { + environmentId, + ...(delivery.cause === undefined ? {} : { cause: delivery.cause }), + }); + const retryDelaysMs = [1_000, 5_000, 15_000] as const; + const retryDelayMs = retryDelaysMs[retryAttemptRef.current]; + if (retryDelayMs !== undefined && retryTimerRef.current === null) { + retryAttemptRef.current += 1; + retryTimerRef.current = window.setTimeout(() => { + retryTimerRef.current = null; + setRetryGeneration((generation) => generation + 1); + }, retryDelayMs); + } + } else { + retryAttemptRef.current = 0; + if (retryTimerRef.current !== null) { + window.clearTimeout(retryTimerRef.current); + retryTimerRef.current = null; + } + } + }; + + syncChainRef.current = syncChainRef.current.then(synchronize, synchronize); + }, [ + enabled, + environmentId, + replayEvents, + retryGeneration, + settingsHydrated, + shellState.status, + snapshot, + ]); return null; } diff --git a/apps/web/src/threadCompletionNotifications.test.ts b/apps/web/src/threadCompletionNotifications.test.ts new file mode 100644 index 00000000000..e0117f17fd3 --- /dev/null +++ b/apps/web/src/threadCompletionNotifications.test.ts @@ -0,0 +1,382 @@ +import type { + EnvironmentId, + OrchestrationEvent, + OrchestrationSessionStatus, + OrchestrationShellSnapshot, + ThreadId, + TurnId, +} from "@t3tools/contracts"; +import { describe, expect, it, vi } from "vite-plus/test"; + +import { + deliverPendingThreadCompletionNotifications, + initializeThreadCompletionNotificationState, + parseThreadCompletionNotificationState, + reduceThreadCompletionNotificationEvents, +} from "./threadCompletionNotifications"; + +const ENVIRONMENT_ID = "environment-1" as EnvironmentId; +const PROJECT_THREAD_ID = "project-thread" as ThreadId; +const STANDALONE_THREAD_ID = "standalone-thread" as ThreadId; +const TURN_ID = "turn-1" as TurnId; + +function snapshot(input: { + readonly sequence: number; + readonly sessions?: ReadonlyArray<{ + readonly threadId: ThreadId; + readonly title: string; + readonly status: OrchestrationSessionStatus; + readonly turnId: TurnId | null; + }>; +}): OrchestrationShellSnapshot { + return { + snapshotSequence: input.sequence, + projects: [], + threads: (input.sessions ?? []).map((entry) => ({ + id: entry.threadId, + title: entry.title, + session: { + threadId: entry.threadId, + status: entry.status, + activeTurnId: entry.turnId, + }, + })), + updatedAt: "2026-07-21T12:00:00.000Z", + } as unknown as OrchestrationShellSnapshot; +} + +function sessionEvent(input: { + readonly sequence: number; + readonly threadId: ThreadId; + readonly status: OrchestrationSessionStatus; + readonly turnId: TurnId | null; +}): OrchestrationEvent { + return { + sequence: input.sequence, + eventId: `event-${input.sequence}`, + aggregateKind: "thread", + aggregateId: input.threadId, + occurredAt: "2026-07-21T12:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.session-set", + payload: { + threadId: input.threadId, + session: { + threadId: input.threadId, + status: input.status, + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: input.turnId, + lastError: null, + updatedAt: "2026-07-21T12:00:00.000Z", + }, + }, + } as OrchestrationEvent; +} + +describe("thread completion notification event tracking", () => { + it("seeds a live running turn without notifying for earlier completions", () => { + const state = initializeThreadCompletionNotificationState( + snapshot({ + sequence: 10, + sessions: [ + { + threadId: PROJECT_THREAD_ID, + title: "Project chat", + status: "running", + turnId: TURN_ID, + }, + ], + }), + ); + + expect(state.cursor).toBe(10); + expect(state.activeTurnByThread[PROJECT_THREAD_ID]).toBe(TURN_ID); + expect(state.pending).toEqual([]); + }); + + it("recovers a completion from a collapsed remote reconnect event batch", () => { + const currentSnapshot = snapshot({ + sequence: 3, + sessions: [ + { + threadId: PROJECT_THREAD_ID, + title: "Remote project chat", + status: "ready", + turnId: null, + }, + { + threadId: STANDALONE_THREAD_ID, + title: "Remote standalone chat", + status: "ready", + turnId: null, + }, + ], + }); + const initial = initializeThreadCompletionNotificationState(snapshot({ sequence: 1 })); + const state = reduceThreadCompletionNotificationEvents({ + state: initial, + snapshot: currentSnapshot, + notificationsEnabled: true, + events: [ + sessionEvent({ + sequence: 2, + threadId: PROJECT_THREAD_ID, + status: "running", + turnId: TURN_ID, + }), + sessionEvent({ + sequence: 3, + threadId: PROJECT_THREAD_ID, + status: "ready", + turnId: null, + }), + ], + }); + + expect(state.cursor).toBe(3); + expect(state.pending).toEqual([ + { + id: `${PROJECT_THREAD_ID}:${TURN_ID}`, + threadId: PROJECT_THREAD_ID, + threadTitle: "Remote project chat", + }, + ]); + }); + + it("survives reload between turn start and completion and does not duplicate replayed events", () => { + const initial = initializeThreadCompletionNotificationState(snapshot({ sequence: 1 })); + const running = reduceThreadCompletionNotificationEvents({ + state: initial, + snapshot: snapshot({ sequence: 2 }), + notificationsEnabled: true, + events: [ + sessionEvent({ + sequence: 2, + threadId: STANDALONE_THREAD_ID, + status: "running", + turnId: TURN_ID, + }), + ], + }); + const reloaded = parseThreadCompletionNotificationState(JSON.stringify(running)); + expect(reloaded).not.toBeNull(); + + const completed = reduceThreadCompletionNotificationEvents({ + state: reloaded!, + snapshot: snapshot({ + sequence: 3, + sessions: [ + { + threadId: STANDALONE_THREAD_ID, + title: "Standalone chat", + status: "idle", + turnId: null, + }, + ], + }), + notificationsEnabled: true, + events: [ + sessionEvent({ + sequence: 3, + threadId: STANDALONE_THREAD_ID, + status: "idle", + turnId: null, + }), + ], + }); + const replayed = reduceThreadCompletionNotificationEvents({ + state: completed, + snapshot: snapshot({ sequence: 3 }), + notificationsEnabled: true, + events: [ + sessionEvent({ + sequence: 2, + threadId: STANDALONE_THREAD_ID, + status: "running", + turnId: TURN_ID, + }), + sessionEvent({ + sequence: 3, + threadId: STANDALONE_THREAD_ID, + status: "idle", + turnId: null, + }), + ], + }); + + expect(replayed.pending).toHaveLength(1); + expect(replayed.pending[0]?.threadTitle).toBe("Standalone chat"); + }); + + it("advances the cursor without queuing success notifications while disabled", () => { + const state = reduceThreadCompletionNotificationEvents({ + state: initializeThreadCompletionNotificationState(snapshot({ sequence: 1 })), + snapshot: snapshot({ sequence: 3 }), + notificationsEnabled: false, + events: [ + sessionEvent({ + sequence: 2, + threadId: PROJECT_THREAD_ID, + status: "running", + turnId: TURN_ID, + }), + sessionEvent({ + sequence: 3, + threadId: PROJECT_THREAD_ID, + status: "ready", + turnId: null, + }), + ], + }); + + expect(state.cursor).toBe(3); + expect(state.pending).toEqual([]); + }); + + it("keeps the active turn across a transient starting session", () => { + const state = reduceThreadCompletionNotificationEvents({ + state: initializeThreadCompletionNotificationState(snapshot({ sequence: 1 })), + snapshot: snapshot({ + sequence: 4, + sessions: [ + { + threadId: PROJECT_THREAD_ID, + title: "Restarted provider", + status: "ready", + turnId: null, + }, + ], + }), + notificationsEnabled: true, + events: [ + sessionEvent({ + sequence: 2, + threadId: PROJECT_THREAD_ID, + status: "running", + turnId: TURN_ID, + }), + sessionEvent({ + sequence: 3, + threadId: PROJECT_THREAD_ID, + status: "starting", + turnId: null, + }), + sessionEvent({ + sequence: 4, + threadId: PROJECT_THREAD_ID, + status: "ready", + turnId: null, + }), + ], + }); + + expect(state.pending).toHaveLength(1); + }); + + it.each(["error", "interrupted", "stopped"] as const)( + "does not notify when a running turn ends as %s", + (status) => { + const state = reduceThreadCompletionNotificationEvents({ + state: initializeThreadCompletionNotificationState(snapshot({ sequence: 1 })), + snapshot: snapshot({ sequence: 3 }), + notificationsEnabled: true, + events: [ + sessionEvent({ + sequence: 2, + threadId: PROJECT_THREAD_ID, + status: "running", + turnId: TURN_ID, + }), + sessionEvent({ + sequence: 3, + threadId: PROJECT_THREAD_ID, + status, + turnId: null, + }), + ], + }); + + expect(state.pending).toEqual([]); + }, + ); +}); + +describe("thread completion notification delivery", () => { + const pendingState = reduceThreadCompletionNotificationEvents({ + state: initializeThreadCompletionNotificationState(snapshot({ sequence: 1 })), + snapshot: snapshot({ + sequence: 3, + sessions: [ + { + threadId: PROJECT_THREAD_ID, + title: "Retry me", + status: "ready", + turnId: null, + }, + ], + }), + notificationsEnabled: true, + events: [ + sessionEvent({ + sequence: 2, + threadId: PROJECT_THREAD_ID, + status: "running", + turnId: TURN_ID, + }), + sessionEvent({ + sequence: 3, + threadId: PROJECT_THREAD_ID, + status: "ready", + turnId: null, + }), + ], + }); + + it("keeps a failed native delivery pending for a later retry", async () => { + const progress = vi.fn(); + const first = await deliverPendingThreadCompletionNotifications({ + state: pendingState, + environmentId: ENVIRONMENT_ID, + show: vi.fn().mockResolvedValue(false), + onProgress: progress, + }); + + expect(first.failed).toBe(true); + expect(first.state.pending).toHaveLength(1); + expect(progress).not.toHaveBeenCalled(); + + const show = vi.fn().mockResolvedValue(true); + const second = await deliverPendingThreadCompletionNotifications({ + state: first.state, + environmentId: ENVIRONMENT_ID, + show, + onProgress: progress, + }); + + expect(second.failed).toBe(false); + expect(second.state.pending).toEqual([]); + expect(show).toHaveBeenCalledWith({ + threadRef: { environmentId: ENVIRONMENT_ID, threadId: PROJECT_THREAD_ID }, + threadTitle: "Retry me", + }); + expect(progress).toHaveBeenCalledWith(second.state); + }); + + it("keeps a rejected IPC delivery pending and exposes the cause", async () => { + const cause = new Error("renderer disconnected"); + const result = await deliverPendingThreadCompletionNotifications({ + state: pendingState, + environmentId: ENVIRONMENT_ID, + show: vi.fn().mockRejectedValue(cause), + onProgress: vi.fn(), + }); + + expect(result.failed).toBe(true); + expect(result.cause).toBe(cause); + expect(result.state.pending).toHaveLength(1); + }); +}); diff --git a/apps/web/src/threadCompletionNotifications.ts b/apps/web/src/threadCompletionNotifications.ts new file mode 100644 index 00000000000..9547b426c05 --- /dev/null +++ b/apps/web/src/threadCompletionNotifications.ts @@ -0,0 +1,241 @@ +import type { + EnvironmentId, + OrchestrationEvent, + OrchestrationShellSnapshot, + ThreadId, + TurnId, +} from "@t3tools/contracts"; + +const STORAGE_KEY_PREFIX = "t3code:thread-completion-notifications:v1:"; +const STATE_VERSION = 1; +const MAX_PENDING_NOTIFICATIONS = 100; + +export interface PendingThreadCompletionNotification { + readonly id: string; + readonly threadId: ThreadId; + readonly threadTitle: string; +} + +export interface ThreadCompletionNotificationState { + readonly version: typeof STATE_VERSION; + readonly cursor: number; + readonly activeTurnByThread: Readonly>; + readonly pending: ReadonlyArray; +} + +export interface ThreadCompletionNotificationDeliveryResult { + readonly state: ThreadCompletionNotificationState; + readonly failed: boolean; + readonly cause?: unknown; +} + +function storageKey(environmentId: EnvironmentId): string { + return `${STORAGE_KEY_PREFIX}${environmentId}`; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +export function parseThreadCompletionNotificationState( + raw: string, +): ThreadCompletionNotificationState | null { + try { + const value: unknown = JSON.parse(raw); + if ( + !isRecord(value) || + value.version !== STATE_VERSION || + typeof value.cursor !== "number" || + !Number.isSafeInteger(value.cursor) || + value.cursor < 0 || + !isRecord(value.activeTurnByThread) || + !Array.isArray(value.pending) + ) { + return null; + } + + const activeTurnByThread: Record = {}; + for (const [threadId, turnId] of Object.entries(value.activeTurnByThread)) { + if (typeof turnId !== "string") return null; + activeTurnByThread[threadId] = turnId as TurnId; + } + + const pending: PendingThreadCompletionNotification[] = []; + for (const candidate of value.pending) { + if ( + !isRecord(candidate) || + typeof candidate.id !== "string" || + typeof candidate.threadId !== "string" || + typeof candidate.threadTitle !== "string" + ) { + return null; + } + pending.push({ + id: candidate.id, + threadId: candidate.threadId as ThreadId, + threadTitle: candidate.threadTitle, + }); + } + + return { + version: STATE_VERSION, + cursor: value.cursor, + activeTurnByThread, + pending: pending.slice(-MAX_PENDING_NOTIFICATIONS), + }; + } catch { + return null; + } +} + +export function readThreadCompletionNotificationState( + storage: Storage, + environmentId: EnvironmentId, +): ThreadCompletionNotificationState | null { + const raw = storage.getItem(storageKey(environmentId)); + return raw === null ? null : parseThreadCompletionNotificationState(raw); +} + +export function writeThreadCompletionNotificationState( + storage: Storage, + environmentId: EnvironmentId, + state: ThreadCompletionNotificationState, +): void { + storage.setItem(storageKey(environmentId), JSON.stringify(state)); +} + +export function initializeThreadCompletionNotificationState( + snapshot: OrchestrationShellSnapshot, +): ThreadCompletionNotificationState { + const activeTurnByThread: Record = {}; + for (const thread of snapshot.threads) { + if (thread.session?.status === "running" && thread.session.activeTurnId !== null) { + activeTurnByThread[thread.id] = thread.session.activeTurnId; + } + } + + return { + version: STATE_VERSION, + cursor: snapshot.snapshotSequence, + activeTurnByThread, + pending: [], + }; +} + +export function clearPendingThreadCompletionNotifications( + state: ThreadCompletionNotificationState, +): ThreadCompletionNotificationState { + return state.pending.length === 0 ? state : { ...state, pending: [] }; +} + +export function reduceThreadCompletionNotificationEvents(input: { + readonly state: ThreadCompletionNotificationState; + readonly events: ReadonlyArray; + readonly snapshot: OrchestrationShellSnapshot; + readonly notificationsEnabled: boolean; +}): ThreadCompletionNotificationState { + let cursor = input.state.cursor; + const activeTurnByThread: Record = { + ...input.state.activeTurnByThread, + }; + const pending = [...input.state.pending]; + const pendingIds = new Set(pending.map((notification) => notification.id)); + const titleByThread = new Map( + input.snapshot.threads.map((thread) => [thread.id as string, thread.title]), + ); + + const events = [...input.events].sort((left, right) => left.sequence - right.sequence); + for (const event of events) { + if (event.sequence <= cursor) continue; + cursor = event.sequence; + + if (event.type === "thread.created") { + titleByThread.set(event.payload.threadId, event.payload.title); + continue; + } + if (event.type === "thread.meta-updated" && event.payload.title !== undefined) { + titleByThread.set(event.payload.threadId, event.payload.title); + continue; + } + if (event.type === "thread.deleted") { + delete activeTurnByThread[event.payload.threadId]; + continue; + } + if (event.type !== "thread.session-set") continue; + + const { session, threadId } = event.payload; + if (session.status === "starting") { + continue; + } + if (session.status === "running") { + if (session.activeTurnId !== null) { + activeTurnByThread[threadId] = session.activeTurnId; + } + continue; + } + + const activeTurnId = activeTurnByThread[threadId]; + delete activeTurnByThread[threadId]; + if ( + activeTurnId === undefined || + !input.notificationsEnabled || + (session.status !== "idle" && session.status !== "ready") + ) { + continue; + } + + const id = `${threadId}:${activeTurnId}`; + if (pendingIds.has(id)) continue; + pendingIds.add(id); + pending.push({ + id, + threadId, + threadTitle: titleByThread.get(threadId) ?? "Thread", + }); + } + + return { + version: STATE_VERSION, + cursor, + activeTurnByThread, + pending: pending.slice(-MAX_PENDING_NOTIFICATIONS), + }; +} + +export async function deliverPendingThreadCompletionNotifications(input: { + readonly state: ThreadCompletionNotificationState; + readonly environmentId: EnvironmentId; + readonly show: (input: { + readonly threadRef: { readonly environmentId: EnvironmentId; readonly threadId: ThreadId }; + readonly threadTitle: string; + }) => Promise; + readonly onProgress: (state: ThreadCompletionNotificationState) => void | Promise; +}): Promise { + let state = input.state; + while (state.pending.length > 0) { + const notification = state.pending[0]!; + let shown: boolean; + try { + shown = await input.show({ + threadRef: { + environmentId: input.environmentId, + threadId: notification.threadId, + }, + threadTitle: notification.threadTitle, + }); + } catch (cause) { + return { state, failed: true, cause }; + } + if (!shown) { + return { state, failed: true }; + } + state = { + version: state.version, + cursor: state.cursor, + activeTurnByThread: state.activeTurnByThread, + pending: state.pending.slice(1), + }; + await input.onProgress(state); + } + return { state, failed: false }; +} diff --git a/docs/personal-fork-changes.md b/docs/personal-fork-changes.md index 0e8dcaed51e..1f5c4a6afcc 100644 --- a/docs/personal-fork-changes.md +++ b/docs/personal-fork-changes.md @@ -100,5 +100,9 @@ Upstream PRs integrated into the fork are listed in notification whenever a project thread or standalone chat transitions to completed. Clicking it reveals the app and opens the exact environment-scoped conversation. Web and non-macOS runtimes retain upstream behavior. +- Notification detection uses a persisted per-environment orchestration-event cursor instead of + inferring completion only from adjacent renderer snapshots. Reconnects and renderer reloads replay + any missed running-to-completed transitions, while successful deliveries advance a durable outbox + and failed native/IPC deliveries remain pending for retry without duplicating replayed events. - The synthesized completion cue plays at 110% of its original gain; attention cues retain their original gain. Disabling completion sounds preserves the upstream silent behavior. diff --git a/packages/client-runtime/src/state/orchestration.ts b/packages/client-runtime/src/state/orchestration.ts index a61e42da748..a695a07b3a3 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -17,6 +17,14 @@ export function createOrchestrationEnvironmentAtoms( label: "environment-data:orchestration:thread-activities", tag: ORCHESTRATION_WS_METHODS.getThreadActivities, }), + replayEvents: createEnvironmentRpcCommand(runtime, { + label: "environment-data:orchestration:replay-events", + tag: ORCHESTRATION_WS_METHODS.replayEvents, + concurrency: { + mode: "singleFlight", + key: ({ environmentId }) => environmentId, + }, + }), fullThreadDiff: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:full-thread-diff", tag: ORCHESTRATION_WS_METHODS.getFullThreadDiff,