diff --git a/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts b/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts index 9085f98058066..45dea9b2d2dbf 100644 --- a/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts +++ b/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts @@ -645,6 +645,50 @@ describe("makeRelayDeviceRegistrationRequest", () => { }).pipe(Effect.provide(layerRelayTest)); }); + it.effect( + "keeps notifications enabled when permission is granted but the push token is missing", + () => { + vi.mocked(loadPreferences).mockResolvedValue({ liveActivitiesEnabled: true }); + vi.spyOn(Platform, "OS", "get").mockReturnValue("ios"); + vi.spyOn(Platform, "Version", "get").mockReturnValue(18); + vi.mocked(Notifications.getPermissionsAsync).mockResolvedValue({ + granted: true, + } as Awaited>); + vi.mocked(Notifications.getDevicePushTokenAsync).mockRejectedValue(new Error("no token yet")); + const registrations: unknown[] = []; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const request = new Request(input, init); + if (request.url.endsWith("/v1/client/dpop-token")) { + return Response.json({ + access_token: "dpop", + issued_token_type: "urn:ietf:params:oauth:token-type:access_token", + token_type: "DPoP", + expires_in: 300, + scope: "mobile:registration", + }); + } + registrations.push(await request.json()); + return Response.json({ ok: true }); + }); + Constants.expoConfig!.extra = { relay: { url: "https://missing-token-relay.example.test" } }; + setAgentAwarenessRelayTokenProvider(() => Promise.resolve("clerk"), "user-a"); + return Effect.gen(function* () { + yield* runBackgroundOperations(); + expect(registrations.at(-1)).toMatchObject({ + preferences: { notificationsEnabled: true, liveActivitiesEnabled: true }, + }); + expect(registrations.at(-1)).not.toHaveProperty("pushToken"); + }).pipe( + Effect.provideService(FetchHttpClient.Fetch, globalThis.fetch), + Effect.provide( + ManagedRelayLayer.layer("https://missing-token-relay.example.test").pipe( + Layer.provide(Layer.mergeAll(FetchHttpClient.layer, Dpop.layer)), + ), + ), + ); + }, + ); + it.effect.each(["ios", "android"] as const)( "does not enable %s notifications when a token rotates after permission is revoked", (os) => { diff --git a/apps/mobile/src/features/agent-awareness/remoteRegistration.ts b/apps/mobile/src/features/agent-awareness/remoteRegistration.ts index ec712608fad94..0691152643ae5 100644 --- a/apps/mobile/src/features/agent-awareness/remoteRegistration.ts +++ b/apps/mobile/src/features/agent-awareness/remoteRegistration.ts @@ -316,8 +316,11 @@ const nativePushTokenRegistration = Effect.fn("remoteRegistration.nativePushToke token?.type === Platform.OS && typeof token.data === "string" && token.data.trim().length > 0 ? token.data.trim() : null; + // Permission/channel state gates alerts. A missing token must not publish + // notificationsEnabled:false or Live Activity jobs strip their banners while + // Settings still shows Device Notifications on. return { - notificationsEnabled: alertsEnabled && pushToken !== null, + notificationsEnabled: alertsEnabled, pushToken, notificationPermissionGranted: permissions.granted, }; diff --git a/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx index cabe229f9a262..236d09d324128 100644 --- a/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx @@ -19,6 +19,7 @@ import { Linking, Platform } from "react-native"; import { useSafeAreaInsets } from "react-native-safe-area-context"; import { AppText as Text } from "../../components/ui/AppText"; +import { ListChoiceRow } from "../../components/ui/ListChoiceRow"; import { isAtomCommandInterrupted, reportAtomCommandResult, @@ -46,7 +47,6 @@ import { mobilePreferencesAtom, updateMobilePreferencesAtom } from "../../state/ import { useSavedRemoteConnections } from "../../state/use-remote-environment-registry"; import { SettingsRow } from "./components/SettingsRow"; import { SettingsSection } from "./components/SettingsSection"; -import { SettingsSwitchRow } from "./components/SettingsSwitchRow"; import { SettingsScreen } from "./components/SettingsScreen"; import { resolveAgentAwarenessPlatformPresentation } from "./SettingsRouteScreen.logic"; @@ -66,6 +66,7 @@ function canLinkEnvironmentToConnect(environmentId: EnvironmentId) { type NotificationStatus = "checking" | "enabled" | "disabled" | "unsupported"; type LiveActivityStatus = "checking" | "enabled" | "disabled" | "signed-out" | "linking"; +type AgentActivityDeliveryMode = "off" | "banners" | "live"; const notificationStatusAtom = makePermissionStatusAtom({ label: "mobile:notification-permission", @@ -108,6 +109,47 @@ function useDeviceRegistered(): boolean { return status === "registered"; } +function resolveDeliveryMode(input: { + readonly pushAvailable: boolean; + readonly notificationStatus: NotificationStatus; + readonly deviceRegistered: boolean; + readonly liveActivityStatus: LiveActivityStatus; +}): AgentActivityDeliveryMode { + if ( + !input.pushAvailable || + input.notificationStatus === "disabled" || + input.notificationStatus === "unsupported" || + !input.deviceRegistered + ) { + return "off"; + } + if (input.liveActivityStatus === "enabled" || input.liveActivityStatus === "linking") { + return "live"; + } + return "banners"; +} + +function liveActivityOptionCopy(): { readonly label: string; readonly description: string } { + if (Platform.OS === "android") { + return supportsAndroidLiveUpdateSettings() + ? { + label: "Live Updates", + description: + "Ongoing agent work as Live Updates, plus banners when an agent needs you or finishes.", + } + : { + label: "Ongoing activity", + description: + "Ongoing agent work as a persistent notification, plus banners when an agent needs you or finishes.", + }; + } + return { + label: "Live Activities", + description: + "Ongoing work on the Lock Screen and Dynamic Island, plus banners when an agent needs you or finishes.", + }; +} + export function SettingsNotificationsRouteScreen() { if (!hasCloudPublicConfig()) { return ( @@ -132,7 +174,7 @@ function ConfiguredSettingsNotificationsRouteScreen() { const savePreferences = useAtomSet(updateMobilePreferencesAtom); const agentAwarenessPushAvailable = supportsAgentAwarenessPush(); const agentAwarenessPlatform = resolveAgentAwarenessPlatformPresentation(Platform.OS); - const agentAwarenessSubtitle = + const agentAwarenessUnavailableSubtitle = Platform.OS === "android" && !agentAwarenessPushAvailable ? "Install a newer app build to enable notifications" : agentAwarenessPlatform.subtitle; @@ -166,11 +208,86 @@ function ConfiguredSettingsNotificationsRouteScreen() { const canClearLiveActivitiesPreference = AsyncResult.isSuccess(preferencesResult) && preferencesResult.value.liveActivitiesEnabled !== false; + const selectedMode = resolveDeliveryMode({ + pushAvailable: agentAwarenessPushAvailable, + notificationStatus, + deviceRegistered, + liveActivityStatus, + }); + const liveOption = liveActivityOptionCopy(); + const deliveryOptions: ReadonlyArray<{ + readonly mode: AgentActivityDeliveryMode; + readonly label: string; + readonly description: string; + }> = [ + { + mode: "off", + label: "Off", + description: "No agent alerts on this phone.", + }, + { + mode: "banners", + label: "Banners", + description: "Notification banners for approvals, input requests, and finished runs.", + }, + { + mode: "live", + label: liveOption.label, + description: liveOption.description, + }, + ]; const connections = Object.values(savedConnectionsById); const environmentCount = connections.length; + const choicesDisabled = + !agentAwarenessPlatform.supported || + !agentAwarenessPushAvailable || + notificationStatus === "checking" || + notificationStatus === "unsupported" || + !isLoaded || + liveActivityStatus === "checking" || + liveActivityStatus === "linking"; + + const promptSignIn = () => { + Dialog.alert( + "Sign in to T3 Connect", + "Agent activity alerts require T3 Connect so relay can deliver updates to this device.", + [ + { text: "Cancel", style: "cancel" }, + { + text: "Continue", + onPress: () => navigation.navigate("SettingsAuth"), + }, + ], + ); + }; - const requestNotifications = async () => { + const ensureNotificationPermission = async (): Promise => { + if (notificationStatus === "enabled") { + if (getAgentAwarenessRegistrationStatus() === "registered") { + return true; + } + const refreshResult = await settleAsyncResult(() => + runtime.runPromiseExit(refreshAgentAwarenessRegistration()), + ); + if (refreshResult._tag === "Failure") { + if (!isAtomCommandInterrupted(refreshResult)) { + Dialog.alert( + "Couldn't finish enabling notifications", + "Notification access is granted, but this device could not be registered with T3 Connect. Try again once you are online.", + ); + } + return false; + } + if (getAgentAwarenessRegistrationStatus() !== "registered") { + Dialog.alert( + "Couldn't finish enabling notifications", + "Notification access is granted, but this device could not be registered with T3 Connect. Alerts will start once registration succeeds.", + ); + return false; + } + return true; + } const result = await settleAsyncResult(() => runtime.runPromiseExit( requestAgentNotificationPermission.pipe( @@ -188,19 +305,18 @@ function ConfiguredSettingsNotificationsRouteScreen() { error instanceof Error ? error.message : "Could not request notification permission.", ); } - return; + return false; } if (result.value.type === "granted") { setNotificationStatus("enabled"); - if (getAgentAwarenessRegistrationStatus() === "registered") { - Dialog.alert("Notifications enabled", "Agent notifications are enabled for this device."); - } else { + if (getAgentAwarenessRegistrationStatus() !== "registered") { Dialog.alert( "Couldn't finish enabling notifications", - "Notification access was granted, but this device could not be registered with T3 Connect. Notifications will start once registration succeeds.", + "Notification access was granted, but this device could not be registered with T3 Connect. Alerts will start once registration succeeds.", ); + return false; } - return; + return true; } if (result.value.type === "unsupported") { setNotificationStatus("unsupported"); @@ -208,12 +324,12 @@ function ConfiguredSettingsNotificationsRouteScreen() { "Notifications unavailable", "Agent notifications are unavailable on this platform.", ); - return; + return false; } setNotificationStatus("disabled"); if (result.value.canAskAgain) { Dialog.alert("Notifications disabled", "Notifications were not enabled."); - return; + return false; } Dialog.alert( "Notifications disabled", @@ -223,55 +339,60 @@ function ConfiguredSettingsNotificationsRouteScreen() { { text: "Open Settings", onPress: () => void Linking.openSettings() }, ], ); + return false; }; - const promptSignIn = () => { - Dialog.alert( - "Sign in to T3 Connect", - "Live Activity updates require T3 Connect so relay can deliver updates to this device.", - [ - { text: "Cancel", style: "cancel" }, - { - text: "Continue", - onPress: () => navigation.navigate("SettingsAuth"), - }, - ], + async function disableLiveActivities() { + let token: string | null = null; + if (isSignedIn) { + const tokenResult = await settlePromise(() => getToken(resolveRelayClerkTokenOptions())); + if (tokenResult._tag === "Failure") { + setLiveActivityStatus("enabled"); + reportAtomCommandResult(tokenResult, { + label: "live activity disable token lookup", + }); + return false; + } + token = tokenResult.value; + } + + const updateResult = await settleAsyncResult(() => + runtime.runPromiseExit( + setLiveActivityUpdatesEnabled({ + canConfigureEnvironment: canLinkEnvironmentToConnect, + enabled: false, + previousEnabled: liveActivitiesPreferenceEnabled, + clerkToken: token, + connections, + }), + ), ); - }; + if (updateResult._tag === "Failure") { + setLiveActivityStatus(isSignedIn ? "enabled" : "signed-out"); + reportAtomCommandResult(updateResult, { + label: "live activity disable", + }); + return false; + } + savePreferences({ liveActivitiesEnabled: false }); + refreshManagedRelayEnvironments(); + setLiveActivityStatus("disabled"); + return true; + } - const linkEnvironments = async () => { + async function enableLiveActivities(): Promise { if (!isSignedIn) { promptSignIn(); - return; + return false; } setLiveActivityStatus("linking"); if (Platform.OS === "android") { - const permission = await settleAsyncResult(() => - runtime.runPromiseExit(requestAgentNotificationPermission), - ); - if (permission._tag === "Failure") { - setLiveActivityStatus("disabled"); - const error = squashAtomCommandFailure(permission); - Dialog.alert( - "Ongoing activity unavailable", - error instanceof Error ? error.message : "Could not enable agent notifications.", - ); - return; - } - if (permission.value.type !== "granted") { + const permitted = await ensureNotificationPermission(); + if (!permitted) { setLiveActivityStatus("disabled"); - Dialog.alert( - "Notification permission needed", - "Enable notifications in system Settings to show ongoing agent activity.", - [ - { text: "Cancel", style: "cancel" }, - { text: "Open Settings", onPress: () => void Linking.openSettings() }, - ], - ); - return; + return false; } - setNotificationStatus("enabled"); } const tokenResult = await settlePromise(() => getToken(resolveRelayClerkTokenOptions())); if (tokenResult._tag === "Failure") { @@ -281,12 +402,12 @@ function ConfiguredSettingsNotificationsRouteScreen() { Platform.OS === "android" ? "Ongoing activity unavailable" : "Live Activities unavailable", error instanceof Error ? error.message : "Could not enable agent activity updates.", ); - return; + return false; } if (!tokenResult.value) { promptSignIn(); setLiveActivityStatus("signed-out"); - return; + return false; } const updateResult = await settleAsyncResult(() => @@ -311,103 +432,93 @@ function ConfiguredSettingsNotificationsRouteScreen() { error instanceof Error ? error.message : "Could not enable agent activity updates.", ); } - return; + return false; } savePreferences({ liveActivitiesEnabled: true }); refreshManagedRelayEnvironments(); setLiveActivityStatus("enabled"); - // Environment linking can precede device registration; only registered devices can receive Live Activities. - if (getAgentAwarenessRegistrationStatus() === "registered") { - Dialog.alert( - Platform.OS === "android" ? "Ongoing activity enabled" : "Live Activities enabled", - environmentCount > 0 - ? "Agent activity updates are enabled for environments this connection can configure." - : "Agent activity updates are enabled. Add an environment to start receiving updates.", - ); - } else { + if (getAgentAwarenessRegistrationStatus() !== "registered") { Dialog.alert( "Couldn't finish enabling activity updates", "This device could not be registered with T3 Connect, so activity updates won't appear yet. They'll start once registration succeeds.", ); + return false; } - }; + if (environmentCount === 0) { + Dialog.alert( + Platform.OS === "android" ? "Ongoing activity enabled" : "Live Activities enabled", + "Add an environment to start receiving updates.", + ); + } + return true; + } - const handleDeviceNotificationsChange = (enabled: boolean) => { - if (enabled) { - if (!isSignedIn) { - promptSignIn(); - return; - } - void requestNotifications(); + const selectDeliveryMode = async (mode: AgentActivityDeliveryMode) => { + if ( + choicesDisabled || + (mode === selectedMode && !(mode === "off" && canClearLiveActivitiesPreference)) || + liveActivityWriteInFlight.current + ) { return; } - Dialog.alert( - "Disable notifications", - "Open system Settings to disable notifications for T3 Code.", - [ - { text: "Cancel", style: "cancel" }, - { text: "Open Settings", onPress: () => void Linking.openSettings() }, - ], - ); - }; - - async function disableLiveActivities() { - let token: string | null = null; - if (isSignedIn) { - const tokenResult = await settlePromise(() => getToken(resolveRelayClerkTokenOptions())); - if (tokenResult._tag === "Failure") { - setLiveActivityStatus("enabled"); - reportAtomCommandResult(tokenResult, { - label: "live activity disable token lookup", - }); + if (mode === "off") { + liveActivityWriteInFlight.current = true; + setLiveActivityStatus("linking"); + let cleared = true; + if (liveActivitiesPreferenceEnabled) { + cleared = await disableLiveActivities(); + } else { + setLiveActivityStatus(isSignedIn ? "disabled" : "signed-out"); + } + liveActivityWriteInFlight.current = false; + // disableLiveActivities already reports failures; only prompt for OS settings after success. + if (!cleared) { return; } - token = tokenResult.value; + Dialog.alert( + "Turn off agent alerts", + "Open system Settings to disable notifications for T3 Code. Until then, banner delivery stays available if permission is still granted.", + [ + { text: "Cancel", style: "cancel" }, + { text: "Open Settings", onPress: () => void Linking.openSettings() }, + ], + ); + return; } - const updateResult = await settleAsyncResult(() => - runtime.runPromiseExit( - setLiveActivityUpdatesEnabled({ - canConfigureEnvironment: canLinkEnvironmentToConnect, - enabled: false, - previousEnabled: liveActivitiesPreferenceEnabled, - clerkToken: token, - connections, - }), - ), - ); - if (updateResult._tag === "Failure") { - setLiveActivityStatus(isSignedIn ? "enabled" : "signed-out"); - reportAtomCommandResult(updateResult, { - label: "live activity disable", - }); + if (!isSignedIn) { + promptSignIn(); return; } - savePreferences({ liveActivitiesEnabled: false }); - refreshManagedRelayEnvironments(); - setLiveActivityStatus("disabled"); - } - // Both writers settle their own failures, so awaiting them never throws. - const handleLiveActivitiesChange = async (enabled: boolean) => { - if (liveActivityWriteInFlight.current) return; - if (!enabled) { - liveActivityWriteInFlight.current = true; + liveActivityWriteInFlight.current = true; + if (mode === "banners") { setLiveActivityStatus("linking"); - await disableLiveActivities(); + const permitted = await ensureNotificationPermission(); + if (!permitted) { + setLiveActivityStatus( + liveActivitiesPreferenceEnabled ? (isSignedIn ? "enabled" : "signed-out") : "disabled", + ); + liveActivityWriteInFlight.current = false; + return; + } + if (liveActivitiesPreferenceEnabled) { + await disableLiveActivities(); + } else { + setLiveActivityStatus("disabled"); + } liveActivityWriteInFlight.current = false; return; } - if (!isSignedIn) { - promptSignIn(); + const permitted = await ensureNotificationPermission(); + if (!permitted) { + liveActivityWriteInFlight.current = false; return; } - - liveActivityWriteInFlight.current = true; - await linkEnvironments(); + await enableLiveActivities(); liveActivityWriteInFlight.current = false; }; @@ -422,54 +533,36 @@ function ConfiguredSettingsNotificationsRouteScreen() { paddingBottom: Math.max(insets.bottom, 18) + 18, }} > - - - void handleLiveActivitiesChange(enabled)} - /> - {liveActivityStatus === "signed-out" && canClearLiveActivitiesPreference ? ( + + {deliveryOptions.map((option, index) => ( + 0} + disabled={choicesDisabled} + onPress={() => void selectDeliveryMode(option.mode)} + /> + ))} + + {liveActivityStatus === "signed-out" && canClearLiveActivitiesPreference ? ( + void handleLiveActivitiesChange(false)} + onPress={() => void selectDeliveryMode("off")} /> - ) : null} - {supportsAndroidLiveUpdateSettings() ? ( + + ) : null} + {supportsAndroidLiveUpdateSettings() ? ( + - ) : null} - + + ) : null} ); diff --git a/apps/mobile/src/widgets/AgentActivity.tsx b/apps/mobile/src/widgets/AgentActivity.tsx index 609beceac0cdf..4122e603e0231 100644 --- a/apps/mobile/src/widgets/AgentActivity.tsx +++ b/apps/mobile/src/widgets/AgentActivity.tsx @@ -128,25 +128,36 @@ function AgentActivity( const allDone = props.activeCount === 0; const doneLabel = failedRow ? "Failed" : "Done"; - const outcomeLabel = failedRow ? "Agent work failed" : "Agent work completed"; - - const agentWord = props.activeCount === 1 ? "agent" : "agents"; - const agentsLabel = allDone - ? outcomeLabel - : environment.isStale - ? "Agent status out of date" - : `${props.activeCount} active ${agentWord}`; - const attentionSuffix = - attentionRows.length > 0 - ? `${attentionRows.length} need${attentionRows.length === 1 ? "s" : ""} attention` - : ""; + // Inbox header: working / needs reply / caught up — not a live tool transcript. + const inboxHeader = + props.subtitle.trim().length > 0 + ? props.subtitle + : allDone + ? failedRow + ? "Failed" + : "All caught up" + : attentionRows.length > 0 + ? attentionRows.length === 1 + ? "1 needs your reply" + : `${attentionRows.length} need your reply` + : props.activeCount === 1 + ? "1 working" + : `${props.activeCount} working`; + const agentsLabel = environment.isStale ? "Status out of date" : inboxHeader; const activeLabel = allDone ? doneLabel : environment.isStale ? "Out of date" - : `${props.activeCount} active`; + : props.activeCount === 1 + ? "1 working" + : `${props.activeCount} working`; const showCompactGlyph = !attentionRow && !failedRow && !allDone && !environment.isStale; - const summary = attentionSuffix || activeLabel; + const summary = + attentionRows.length > 0 + ? attentionRows.length === 1 + ? "Needs reply" + : `${attentionRows.length} need reply` + : activeLabel; // Widget taps target the containing app, so its production scheme is safe in every build. const deepLinkRow = attentionRow ?? row0; @@ -255,26 +266,14 @@ function AgentActivity( 0 ? headerTint : primaryForeground, + ), lineLimit(1), ]} > {agentsLabel} - {attentionSuffix ? ( - · - ) : null} - {attentionSuffix ? ( - - {attentionSuffix} - - ) : null} diff --git a/apps/server/src/background/BackgroundPolicy.test.ts b/apps/server/src/background/BackgroundPolicy.test.ts index d53fb35f57a0d..c9a7a755742db 100644 --- a/apps/server/src/background/BackgroundPolicy.test.ts +++ b/apps/server/src/background/BackgroundPolicy.test.ts @@ -142,6 +142,92 @@ describe("BackgroundPolicy", () => { }).pipe(Effect.provide(layerFor(nominalHostPower))), ); + it.effect( + "suppresses mobile banners only while web/desktop is actively present and the host is awake", + () => + Effect.gen(function* () { + const policy = yield* BackgroundPolicy.BackgroundPolicy; + assert.equal(yield* policy.shouldSuppressMobileNotificationAlerts, false); + yield* policy.reportClientActivity( + AuthSessionId.make("session-mobile"), + RpcClientId.make(1), + makeReport({ clientId: "phone", clientKind: "mobile" }), + ); + assert.equal(yield* policy.shouldSuppressMobileNotificationAlerts, false); + yield* policy.reportClientActivity( + AuthSessionId.make("session-web"), + RpcClientId.make(2), + makeReport({ clientId: "browser", clientKind: "web" }), + ); + assert.equal(yield* policy.shouldSuppressMobileNotificationAlerts, true); + yield* policy.reportClientActivity( + AuthSessionId.make("session-web"), + RpcClientId.make(2), + makeReport({ + clientId: "browser", + clientKind: "web", + focused: true, + recentlyInteracted: false, + }), + ); + assert.equal(yield* policy.shouldSuppressMobileNotificationAlerts, false); + }).pipe( + Effect.provide( + layerFor({ + ...nominalHostPower, + idle: "false", + locked: "false", + suspended: false, + stale: false, + }), + ), + ), + ); + + it.effect("does not suppress mobile banners when the host is idle", () => + Effect.gen(function* () { + const policy = yield* BackgroundPolicy.BackgroundPolicy; + yield* policy.reportClientActivity( + AuthSessionId.make("session-web"), + RpcClientId.make(1), + makeReport({ clientId: "browser", clientKind: "web" }), + ); + assert.equal(yield* policy.shouldSuppressMobileNotificationAlerts, false); + }).pipe( + Effect.provide( + layerFor({ + ...nominalHostPower, + idle: "true", + locked: "false", + suspended: false, + stale: false, + }), + ), + ), + ); + + it.effect("does not suppress mobile banners when the host is locked", () => + Effect.gen(function* () { + const policy = yield* BackgroundPolicy.BackgroundPolicy; + yield* policy.reportClientActivity( + AuthSessionId.make("session-web"), + RpcClientId.make(1), + makeReport({ clientId: "browser", clientKind: "web" }), + ); + assert.equal(yield* policy.shouldSuppressMobileNotificationAlerts, false); + }).pipe( + Effect.provide( + layerFor({ + ...nominalHostPower, + idle: "false", + locked: "true", + suspended: false, + stale: false, + }), + ), + ), + ); + it.effect("host suspension disables scoped and opportunistic work", () => Effect.gen(function* () { const policy = yield* BackgroundPolicy.BackgroundPolicy; diff --git a/apps/server/src/background/BackgroundPolicy.ts b/apps/server/src/background/BackgroundPolicy.ts index 331b4c4f3365b..d7943408c7108 100644 --- a/apps/server/src/background/BackgroundPolicy.ts +++ b/apps/server/src/background/BackgroundPolicy.ts @@ -34,6 +34,13 @@ export class BackgroundPolicy extends Context.Service< rpcClientId: RpcClientId, ) => Effect.Effect; readonly snapshot: Effect.Effect; + /** + * Someone is actively using web/desktop on this machine — suppress mobile banners. + * Live Activities may still update. Idle, lock, suspend, or stale interaction clears this. + */ + readonly shouldSuppressMobileNotificationAlerts: Effect.Effect; + /** Auth sessions with an actively present web/desktop lease on this machine. */ + readonly presentWebOrDesktopSessionIds: Effect.Effect>; readonly shouldRunScopeWork: (scope: BackgroundScope) => Effect.Effect; readonly shouldRunOpportunisticWork: Effect.Effect; } @@ -113,6 +120,25 @@ function isForegroundLease(lease: ClientActivityLease, now: DateTime.Utc): boole return isLeaseActive(lease, now) && lease.visible && (lease.focused || lease.recentlyInteracted); } +function hostLooksAway(hostPower: HostPowerSnapshot): boolean { + return hostPower.suspended || hostPower.locked === "true" || hostPower.idle === "true"; +} + +/** Present enough to quiet the phone: visible, focused, recent input, and the machine is awake. */ +function isActivelyPresentWebOrDesktopLease( + lease: ClientActivityLease, + hostPower: HostPowerSnapshot, + now: DateTime.Utc, +): boolean { + if (lease.clientKind !== "web" && lease.clientKind !== "desktop-renderer") { + return false; + } + if (!isLeaseActive(lease, now) || !lease.visible || !lease.focused || !lease.recentlyInteracted) { + return false; + } + return !hostLooksAway(hostPower); +} + function leaseHasScope(lease: ClientActivityLease, scope: BackgroundScope): boolean { const key = scopeKey(scope); return lease.scopes.some((leaseScope) => scopeKey(leaseScope) === key); @@ -266,6 +292,26 @@ export const make = Effect.fn("background.policy.make")(function* () { (current) => current.shouldRunOpportunisticWork, ); + const presentWebOrDesktopSessionIds = Effect.gen(function* () { + const [leases, hostPower, now] = yield* Effect.all([ + Ref.get(leasesRef), + hostPowerMonitor.snapshot, + DateTime.now, + ]); + const sessionIds = new Set(); + for (const lease of leases.values()) { + if (isActivelyPresentWebOrDesktopLease(lease, hostPower, now)) { + sessionIds.add(lease.sessionId); + } + } + return [...sessionIds]; + }); + + const shouldSuppressMobileNotificationAlerts = Effect.map( + presentWebOrDesktopSessionIds, + (sessionIds) => sessionIds.length > 0, + ); + yield* Effect.forever( Effect.sleep("15 seconds").pipe( Effect.andThen( @@ -288,6 +334,8 @@ export const make = Effect.fn("background.policy.make")(function* () { return BackgroundPolicy.of({ reportClientActivity, removeRpcClient, + shouldSuppressMobileNotificationAlerts, + presentWebOrDesktopSessionIds, snapshot, shouldRunScopeWork, shouldRunOpportunisticWork, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index ae1640309c133..2dcb555b8a636 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -112,11 +112,19 @@ const assertShellIndexCoversShells = Effect.fnUntraced(function* () { const { tracer, statements, queryTransactions } = traceSqlStatements(); const activity = yield* store.getAgentAwarenessSnapshot().pipe(Effect.withTracer(tracer)); assert.equal(activity.snapshotSequence, snapshot.snapshotSequence); - const awareness = (thread: Parameters[0]["thread"]) => + // Shell snapshots omit side-chat rollup; clear it so this covers shape, not rollup. + const awareness = ( + thread: Parameters[0]["thread"] & { + readonly descendantPendingRequest?: Parameters< + typeof projectThreadAwarenessV2 + >[0]["descendantPendingRequest"]; + }, + ) => projectThreadAwarenessV2({ environmentId: EnvironmentId.make("test"), project: { title: "Test" }, thread, + descendantPendingRequest: null, }); assert.deepEqual( activity.threads.map(awareness), @@ -7093,6 +7101,170 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("rolls pending side-chat requests into listed awareness without family scans", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const later = DateTime.add(now, { minutes: 5 }); + const projectId = ProjectId.make("project:awareness-side-chat-rollup"); + const parentId = ThreadId.make("thread:awareness-side-chat-rollup:parent"); + const pendingSideChatId = ThreadId.make("thread:awareness-side-chat-rollup:pending"); + const idleSideChatId = ThreadId.make("thread:awareness-side-chat-rollup:idle"); + const rootNodeId = NodeId.make("node:awareness-side-chat-rollup"); + const createListed = (threadId: ThreadId, updatedAt: DateTime.Utc) => + projectionStore.apply({ + id: EventId.make(`event:${threadId}:created`), + type: "thread.created", + threadId, + occurredAt: updatedAt, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId, + title: threadId, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: updatedAt, + updatedAt, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + const createSideChat = (threadId: ThreadId, updatedAt: DateTime.Utc) => + projectionStore.apply({ + id: EventId.make(`event:${threadId}:created`), + type: "thread.created", + threadId, + occurredAt: updatedAt, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId, + title: threadId, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: parentId, + relationshipToParent: "fork", + rootThreadId: parentId, + sideChat: true, + }, + forkedFrom: { + type: "run", + threadId: parentId, + runId: RunId.make("run:awareness-side-chat-rollup:parent"), + }, + createdAt: updatedAt, + updatedAt, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + + yield* createListed(parentId, now); + yield* createSideChat(pendingSideChatId, later); + yield* createSideChat(idleSideChatId, now); + yield* projectionStore.apply({ + id: EventId.make("event:awareness-side-chat-rollup:request"), + type: "runtime-request.updated", + threadId: pendingSideChatId, + nodeId: rootNodeId, + driver, + occurredAt: later, + payload: { + id: RuntimeRequestId.make("request:awareness-side-chat-rollup"), + nodeId: rootNodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: later, + resolvedAt: null, + }, + }); + + const { tracer, statements } = traceSqlStatements(); + const full = yield* projectionStore + .getAgentAwarenessSnapshot() + .pipe(Effect.withTracer(tracer)); + const parent = full.threads.find((thread) => thread.id === parentId); + assert.deepEqual(parent?.descendantPendingRequest, { + kind: "user_input", + updatedAt: later, + }); + assert.isUndefined( + full.threads.find( + (thread) => thread.id === pendingSideChatId || thread.id === idleSideChatId, + ), + ); + // Family scans join on rootThreadId via a member subquery; rollup uses json_each parents. + assert.isFalse( + statements.some((statement) => + /FROM orchestration_v2_projection_threads member/.test(statement), + ), + ); + assert.isTrue(statements.some((statement) => /json_each\(/.test(statement))); + assert.isTrue( + statements.some((statement) => + /orchestration_v2_projection_runtime_requests/.test(statement), + ), + ); + + statements.length = 0; + const targetedParent = yield* projectionStore + .getAgentAwarenessSnapshot(parentId) + .pipe(Effect.withTracer(tracer)); + assert.deepEqual(targetedParent.threads[0]?.descendantPendingRequest, { + kind: "user_input", + updatedAt: later, + }); + assert.isFalse( + statements.some((statement) => + /FROM orchestration_v2_projection_threads member/.test(statement), + ), + ); + assert.isTrue(statements.some((statement) => /json_each\(/.test(statement))); + + statements.length = 0; + const targetedSideChat = yield* projectionStore + .getAgentAwarenessSnapshot(pendingSideChatId) + .pipe(Effect.withTracer(tracer)); + assert.equal(targetedSideChat.threads[0]?.id, pendingSideChatId); + assert.isUndefined(targetedSideChat.threads[0]?.descendantPendingRequest); + // Unlisted targets skip the rollup child query (no family member scan). + assert.isFalse( + statements.some((statement) => + /FROM orchestration_v2_projection_threads member/.test(statement), + ), + ); + }).pipe(Effect.provide(layerTest)), + ); + it.effect("counts history inherited by fork shells, but not by side chats", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b49e478fc43c9..c52263b6b3e2d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,5 +1,8 @@ -import type { ProjectThreadAwarenessV2Input } from "@t3tools/shared/agentAwareness"; -import { pendingGuestMessageCount } from "@t3tools/contracts"; +import { + rollupListedThreadAwarenessPending, + type ProjectThreadAwarenessV2Input, +} from "@t3tools/shared/agentAwareness"; +import { isListedThreadLineage, pendingGuestMessageCount } from "@t3tools/contracts"; import { applyThreadPullRequestLinkUpdate } from "../sourceControl/threadPullRequests.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; import * as Stream from "effect/Stream"; @@ -420,6 +423,10 @@ export interface ShellSnapshotOptions { * subagent trees. The thread may be deleted: forks keep a deleted source's root. */ readonly familyOf?: ThreadId; + /** Direct side-chat children of these parents (`lineage.sideChat` + `parentThreadId`). */ + readonly sideChatChildrenOf?: ReadonlyArray; + /** Restrict to threads with a pending non-auth runtime request. */ + readonly pendingRuntimeRequestOnly?: boolean; readonly worktreePath?: string; readonly threadIds?: ReadonlyArray; } @@ -1025,7 +1032,9 @@ export type AgentAwarenessThread = ProjectThreadAwarenessV2Input["thread"] & | "activeRunId" | "latestRunId" | "hasActionableProposedPlan" - >; + > & { + readonly descendantPendingRequest?: ProjectThreadAwarenessV2Input["descendantPendingRequest"]; + }; type ShellThreadRow = { readonly thread_id: string; @@ -6346,6 +6355,8 @@ export const layer: Layer.Layer = projectId, unsettledOnly = false, familyOf, + sideChatChildrenOf, + pendingRuntimeRequestOnly = false, worktreePath, threadIds, listedOnly = false, @@ -6548,6 +6559,22 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_threads member WHERE member.thread_id = ${familyOf} )` + }${ + sideChatChildrenOf === undefined + ? sql`` + : sql` AND json_extract(t.payload_json, '$.lineage.sideChat') = 1 + AND json_extract(t.payload_json, '$.lineage.parentThreadId') IN ( + SELECT value FROM json_each(${encodeIdList(sideChatChildrenOf)}) + )` + }${ + pendingRuntimeRequestOnly + ? sql` AND EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runtime_requests request + WHERE request.thread_id = t.thread_id + AND request.status = 'pending' + AND request.kind <> 'auth_refresh' + )` + : sql`` }${ worktreePath === undefined ? sql`` @@ -7271,14 +7298,42 @@ export const layer: Layer.Layer = const getAgentAwarenessSnapshot = Effect.fn("ProjectionStore.getAgentAwarenessSnapshot")( function* (threadId?: ThreadId) { - const { threadRows, providerThreadRows, pendingTurnItemRows, sequenceRows } = + const { returnedRows, rollupRows, providerThreadRows, pendingTurnItemRows, sequenceRows } = yield* sql.withTransaction( Effect.gen(function* () { - const threadRows = yield* selectShellThreadRows( - threadId, - threadId === undefined ? { location: "active", listedOnly: true } : {}, - ); - const ids = threadRows.map((row) => ThreadId.make(row.thread_id)); + // Returned rows stay listed (or the exact requested thread). Side-chat rollup + // loads only direct pending children by parentThreadId — never the whole family. + const returnedRows = + threadId === undefined + ? yield* selectShellThreadRows(undefined, { + location: "active", + listedOnly: true, + }) + : yield* selectShellThreadRows(threadId); + const rollupParentIds = + threadId === undefined + ? returnedRows.map((row) => ThreadId.make(row.thread_id)) + : yield* Effect.gen(function* () { + const row = returnedRows[0]; + if (row === undefined) return [] as Array; + const thread = yield* decodeThreadPayload(row.payload_json); + return isListedThreadLineage(thread.lineage) + ? [ThreadId.make(row.thread_id)] + : []; + }); + const rollupRows = + rollupParentIds.length === 0 + ? [] + : yield* selectShellThreadRows(undefined, { + location: "active", + sideChatChildrenOf: rollupParentIds, + pendingRuntimeRequestOnly: true, + }); + const ids = [ + ...new Set( + [...returnedRows, ...rollupRows].map((row) => ThreadId.make(row.thread_id)), + ), + ]; const providerThreadRows = ids.length === 0 ? [] : yield* selectShellProviderThreadRows(ids); const pendingTurnItemRows = @@ -7286,7 +7341,7 @@ export const layer: Layer.Layer = ? [] : yield* selectShellPendingTurnItemRows( ids, - threadRows.flatMap((row) => + [...returnedRows, ...rollupRows].flatMap((row) => row.active_run_id !== null && row.active_run_status === "running" ? [row.active_run_id] : [], @@ -7296,12 +7351,18 @@ export const layer: Layer.Layer = SELECT MAX(sequence) AS snapshot_sequence FROM orchestration_events WHERE application_event_version = 2 AND aggregate_kind = 'thread' `; - return { threadRows, providerThreadRows, pendingTurnItemRows, sequenceRows }; + return { + returnedRows, + rollupRows, + providerThreadRows, + pendingTurnItemRows, + sequenceRows, + }; }), ); const { providerThreadsByThreadId, pendingTurnItemsByThreadId } = yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows }); - const threads = yield* Effect.forEach(threadRows, (row) => + const toAwarenessThread = (row: (typeof returnedRows)[number]) => shellThreadStateFromRow({ row, // Awareness never consumes timeline counts or inherited fork history. @@ -7332,6 +7393,19 @@ export const layer: Layer.Layer = pendingRuntimeRequest: state.pendingRuntimeRequest, pendingBackgroundTasks: state.pendingBackgroundTasks, })), + ); + const rollupThreads = yield* Effect.forEach(rollupRows, toAwarenessThread); + const descendantPendingByThread = rollupListedThreadAwarenessPending(rollupThreads); + const threads = yield* Effect.forEach(returnedRows, (row) => + toAwarenessThread(row).pipe( + Effect.map((thread) => + isListedThreadLineage(thread.lineage) + ? { + ...thread, + descendantPendingRequest: descendantPendingByThread.get(thread.id) ?? null, + } + : thread, + ), ), ); return { snapshotSequence: sequenceRows[0]?.snapshot_sequence ?? 0, threads }; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index c3a9efe150e44..e5215023037a4 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -84,6 +84,7 @@ import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationInd import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts"; +import Migration0082 from "./Migrations/082_SideChatAwarenessIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -168,6 +169,7 @@ const migrationEntries = [ [79, "DelegatedRecoveryIndexes", Migration0079], [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionMessageSearchText", Migration0081], + [82, "SideChatAwarenessIndexes", Migration0082], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/082_SideChatAwarenessIndexes.ts b/apps/server/src/persistence/Migrations/082_SideChatAwarenessIndexes.ts new file mode 100644 index 0000000000000..d105f8a149514 --- /dev/null +++ b/apps/server/src/persistence/Migrations/082_SideChatAwarenessIndexes.ts @@ -0,0 +1,19 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Awareness rollups load direct side-chat children by parentThreadId. + yield* sql` + CREATE INDEX orchestration_v2_projection_threads_side_chat_parent_idx + ON orchestration_v2_projection_threads( + json_extract(payload_json, '$.lineage.parentThreadId'), + thread_id + ) + WHERE CASE WHEN json_valid(payload_json) THEN + json_extract(payload_json, '$.lineage.sideChat') = 1 + AND json_extract(payload_json, '$.lineage.parentThreadId') IS NOT NULL + ELSE 0 END + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 8acfd0cb6b9c4..afe8a85f4adb1 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -61,6 +61,7 @@ describe("V2 preview upgrade", () => { [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], + [82, "SideChatAwarenessIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -165,6 +166,7 @@ describe("V2 preview upgrade", () => { [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], + [82, "SideChatAwarenessIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 85b9b44f4daf7..84f697eb1d586 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -1,6 +1,7 @@ import { assert, describe, it } from "@effect/vitest"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { + AuthSessionId, EnvironmentId, EventId, MessageId, @@ -31,7 +32,9 @@ import * as FetchHttpClient from "effect/http/FetchHttpClient"; import * as EnvironmentKeys from "../cloud/environmentKeys.ts"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; import { + CLOUD_LINKED_USER_ID, PUBLISH_AGENT_ACTIVITY_SECRET, RELAY_ENVIRONMENT_CREDENTIAL_SECRET, RELAY_ISSUER_SECRET, @@ -119,7 +122,11 @@ function waitingChild( }); } -const PublishPayload = Schema.Struct({ state: Schema.NullOr(RelayAgentActivityState) }); +const PublishPayload = Schema.Struct({ + state: Schema.NullOr(RelayAgentActivityState), + suppressAlerts: Schema.optionalKey(Schema.Boolean), + suppressAlertUserIds: Schema.optionalKey(Schema.Array(Schema.String)), +}); const decodePublishPayload = Schema.decodeUnknownSync(Schema.fromJsonString(PublishPayload)); const unused = () => Effect.die("Unexpected test dependency call"); @@ -170,6 +177,9 @@ const makeTestRelay = Effect.fnUntraced(function* ( readonly failSecretRead?: (name: string) => boolean; readonly linked?: boolean; readonly readShell?: (threadId: ThreadId) => Effect.Effect; + readonly readFamily?: ( + threadId: ThreadId, + ) => Effect.Effect>; readonly domainEvents?: Stream.Stream; readonly catchUpThreads?: ReadonlyArray; } = {}, @@ -217,10 +227,15 @@ const makeTestRelay = Effect.fnUntraced(function* ( return { snapshotSequence: 1, threads: options.catchUpThreads ?? [] }; }) : Effect.sync(() => shellReads.push(threadId)).pipe( - Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)), - Effect.map((thread) => ({ + Effect.andThen( + options.readFamily?.(threadId) ?? + (options.readShell?.(threadId) ?? Ref.get(currentShell)).pipe( + Effect.map((thread) => (thread === null ? [] : [thread])), + ), + ), + Effect.map((threads) => ({ snapshotSequence: 1, - threads: thread === null ? [] : [thread], + threads, })), ), listActiveSubagentThreadIds: unused, @@ -271,7 +286,10 @@ const makeTestRelay = Effect.fnUntraced(function* ( readonly url: string; readonly authorization: string | null; readonly state: RelayAgentActivityState | null; + readonly suppressAlerts: boolean; + readonly suppressAlertUserIds: ReadonlyArray; }> = []; + const presentWebOrDesktopSessionIds = yield* Ref.make>([]); const fetch: typeof globalThis.fetch = Object.assign( ( input: Parameters[0], @@ -288,6 +306,8 @@ const makeTestRelay = Effect.fnUntraced(function* ( url: String(input), authorization: new Headers(init?.headers).get("authorization"), state: payload.state, + suppressAlerts: payload.suppressAlerts === true, + suppressAlertUserIds: payload.suppressAlertUserIds ?? [], }); return Promise.resolve( options.respond?.(publications.length) ?? Response.json({ ok: true, deliveries: [] }), @@ -326,10 +346,33 @@ const makeTestRelay = Effect.fnUntraced(function* ( listShells: unused, getById: () => Effect.succeedSome(project), }), + Effect.provideService( + BackgroundPolicy.BackgroundPolicy, + BackgroundPolicy.BackgroundPolicy.of({ + reportClientActivity: () => Effect.void, + removeRpcClient: () => Effect.void, + snapshot: Effect.die("unused background snapshot"), + shouldSuppressMobileNotificationAlerts: Ref.get(presentWebOrDesktopSessionIds).pipe( + Effect.map((sessionIds) => sessionIds.length > 0), + ), + presentWebOrDesktopSessionIds: Ref.get(presentWebOrDesktopSessionIds), + shouldRunScopeWork: () => Effect.succeed(false), + shouldRunOpportunisticWork: Effect.succeed(false), + }), + ), Effect.provideService(FetchHttpClient.Fetch, fetch), Effect.provide(NodeCrypto.layer), ); - return { relay, secrets, secretReads, currentShell, shellReads, publications, catchUp }; + return { + relay, + secrets, + secretReads, + presentWebOrDesktopSessionIds, + currentShell, + shellReads, + publications, + catchUp, + }; }); describe("AgentAwarenessRelay", () => { @@ -370,7 +413,6 @@ describe("AgentAwarenessRelay", () => { yield* setShell(waitingChild({ pendingRuntimeRequest: null })); yield* relay.publishThread(CHILD_THREAD_ID); - yield* TestClock.adjust("5 seconds"); yield* relay.drain; assert.equal(publications.length, 4); assert.equal(publications[3]?.state?.phase, "completed"); @@ -503,6 +545,199 @@ describe("AgentAwarenessRelay", () => { }), ); + it.effect("asks relay to suppress banners only for the linked CloudLink user while present", () => + Effect.gen(function* () { + const { relay, currentShell, publications, presentWebOrDesktopSessionIds, secrets } = + yield* makeTestRelay(); + yield* relay.publishThread(THREAD_ID); + assert.deepEqual(publications[0]?.suppressAlertUserIds, []); + yield* secrets.set(CLOUD_LINKED_USER_ID, new TextEncoder().encode("cloud-user-1")); + yield* Ref.set(presentWebOrDesktopSessionIds, [AuthSessionId.make("session-web")]); + yield* Ref.set(currentShell, shell({ title: "Still working" })); + yield* relay.publishThread(THREAD_ID); + assert.deepEqual(publications.at(-1)?.suppressAlertUserIds, ["cloud-user-1"]); + assert.equal(publications.at(-1)?.state?.threadTitle, "Still working"); + }), + ); + + it.effect("publishes parent approval when a nested subagent waits", () => + Effect.gen(function* () { + const childId = ThreadId.make("relay-subagent"); + const grandchildId = ThreadId.make("relay-nested-subagent"); + const parent = shell({ + status: "completed", + pendingBackgroundTasks: [{ taskId: "review", kind: "subagent", childThreadId: childId }], + }); + const child = shell({ + id: childId, + status: "running", + activityRunStatus: "running", + creationSource: "mcp", + lineage: { + rootThreadId: THREAD_ID, + parentThreadId: THREAD_ID, + relationshipToParent: "subagent", + }, + }); + const grandchild = shell({ + id: grandchildId, + status: "waiting", + activityRunStatus: "waiting", + creationSource: "mcp", + lineage: { + rootThreadId: THREAD_ID, + parentThreadId: childId, + relationshipToParent: "subagent", + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("request:approve"), + kind: "command", + createdAt: DateTime.makeUnsafe(NOW), + }, + }); + const byId = new Map([ + [THREAD_ID, parent], + [childId, child], + [grandchildId, grandchild], + ]); + const { relay, publications, shellReads } = yield* makeTestRelay({ + readShell: (threadId) => Effect.succeed(byId.get(threadId) ?? null), + }); + yield* relay.publishThread(grandchildId); + assert.includeMembers(shellReads, [grandchildId, childId, THREAD_ID]); + assert.equal(publications.length, 1); + assert.include(publications[0]?.state, { + threadId: THREAD_ID, + phase: "waiting_for_approval", + headline: "Needs your reply", + }); + }), + ); + + it.effect("republishes the parent when a rolled-up side chat is deleted", () => + Effect.gen(function* () { + const sideChatId = ThreadId.make("relay-side-chat"); + const parentWithAttention = { + ...shell({ status: "completed" }), + descendantPendingRequest: { + kind: "user_input" as const, + updatedAt: DateTime.makeUnsafe(NOW), + }, + }; + const parentClear = shell({ status: "completed" }); + const sideChat = shell({ + id: sideChatId, + status: "waiting", + activityRunStatus: "waiting", + creationSource: "web", + lineage: { + rootThreadId: THREAD_ID, + parentThreadId: THREAD_ID, + relationshipToParent: "fork", + sideChat: true, + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("request:input"), + kind: "user_input", + createdAt: DateTime.makeUnsafe(NOW), + }, + }); + const byId = yield* Ref.make( + new Map | typeof parentWithAttention>([ + [THREAD_ID, parentWithAttention], + [sideChatId, sideChat], + ]), + ); + const { relay, publications } = yield* makeTestRelay({ + readShell: (threadId) => Ref.get(byId).pipe(Effect.map((map) => map.get(threadId) ?? null)), + }); + yield* relay.publishThread(sideChatId); + assert.equal(publications.length, 1); + assert.equal(publications[0]?.state?.phase, "waiting_for_input"); + + yield* Ref.update(byId, (map) => { + const next = new Map(map); + next.delete(sideChatId); + next.set(THREAD_ID, parentClear); + return next; + }); + yield* relay.publishThread(sideChatId); + assert.equal(publications.length, 2); + assert.equal(publications[1]?.state?.phase, "completed"); + assert.equal(publications[1]?.state?.threadId, THREAD_ID); + }), + ); + + it.effect( + "republishes the parent from a side-chat deletion event without a prior map entry", + () => + Effect.gen(function* () { + const sideChatId = ThreadId.make("relay-side-chat-untracked"); + const parentWithAttention = { + ...shell({ status: "completed" }), + descendantPendingRequest: { + kind: "user_input" as const, + updatedAt: DateTime.makeUnsafe(NOW), + }, + }; + const parentClear = shell({ status: "completed" }); + const sideChat = shell({ + id: sideChatId, + status: "waiting", + activityRunStatus: "waiting", + creationSource: "web", + lineage: { + rootThreadId: THREAD_ID, + parentThreadId: THREAD_ID, + relationshipToParent: "fork", + sideChat: true, + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("request:input-untracked"), + kind: "user_input", + createdAt: DateTime.makeUnsafe(NOW), + }, + }); + const byId = yield* Ref.make( + new Map | typeof parentWithAttention>([ + [THREAD_ID, parentWithAttention], + ]), + ); + const events = yield* Queue.unbounded(); + const pulls = yield* Queue.unbounded(); + const { relay, publications } = yield* makeTestRelay({ + readShell: (threadId) => + Ref.get(byId).pipe(Effect.map((map) => map.get(threadId) ?? null)), + domainEvents: Stream.fromEffectRepeat( + Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))), + ), + }); + yield* relay.start(); + yield* Queue.take(pulls); + // Catch-up / reset path: parent already shows rolled-up attention without + // the side chat ever being published through this relay instance. + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.length, 1); + assert.equal(publications[0]?.state?.phase, "waiting_for_input"); + + yield* Ref.set(byId, new Map([[THREAD_ID, parentClear]])); + const now = yield* DateTime.now; + // Domain events carry AppThread payloads; the shell fixture is enough for lineage. + yield* Queue.offer(events, { + id: EventId.make("event:side-chat-deleted"), + type: "thread.deleted", + threadId: sideChatId, + occurredAt: now, + payload: { ...sideChat, deletedAt: now }, + } as OrchestrationV2DomainEvent); + yield* Queue.take(pulls); + yield* relay.drain; + assert.equal(publications.length, 2); + assert.equal(publications[1]?.state?.phase, "completed"); + assert.equal(publications[1]?.state?.threadId, THREAD_ID); + }).pipe(Effect.scoped), + ); + it.effect("republishes unchanged state with fresh credentials after relinking", () => Effect.gen(function* () { const { relay, secrets, publications } = yield* makeTestRelay(); @@ -542,8 +777,6 @@ describe("AgentAwarenessRelay", () => { const terminal = shell({ status }); yield* Ref.set(currentShell, terminal); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; assert.equal(publications[1]?.state?.phase, status); const now = yield* DateTime.now; @@ -564,14 +797,14 @@ describe("AgentAwarenessRelay", () => { yield* Queue.take(pulls); yield* relay.drain; assert.equal(publications.length, 2); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.equal(publications.length, 3); assert.equal(publications[2]?.state, null); }).pipe(Effect.scoped), ); - it.effect("holds a completion through a continuation gap, then publishes it once", () => + it.effect("publishes a brief completion before a continuation resumes", () => Effect.gen(function* () { const { relay, currentShell, publications } = yield* makeTestRelay(); yield* relay.publishThread(THREAD_ID); @@ -579,26 +812,14 @@ describe("AgentAwarenessRelay", () => { yield* relay.publishThread(THREAD_ID); yield* Ref.set(currentShell, shell()); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; - assert.deepEqual( - publications.map((publication) => publication.state?.phase), - ["running"], - ); - - yield* Ref.set(currentShell, shell({ status: "completed" })); - yield* relay.publishThread(THREAD_ID); - assert.equal(publications.length, 1); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state?.phase), - ["running", "completed"], + ["running", "completed", "running"], ); }), ); - it.effect("still publishes a completion when the thread settles during the hold", () => + it.effect("still publishes a completion when the thread settles right after", () => Effect.gen(function* () { const { relay, currentShell, publications } = yield* makeTestRelay(); yield* relay.publishThread(THREAD_ID); @@ -612,14 +833,12 @@ describe("AgentAwarenessRelay", () => { lastVisitedAt: null, }); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state?.phase ?? null), ["running", "completed"], ); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state?.phase ?? null), @@ -635,8 +854,6 @@ describe("AgentAwarenessRelay", () => { const completed = shell({ status: "completed" }); yield* Ref.set(currentShell, completed); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; yield* Ref.set(currentShell, { ...completed, settledOverride: "settled", @@ -644,7 +861,7 @@ describe("AgentAwarenessRelay", () => { lastVisitedAt: null, }); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state?.phase ?? null), @@ -653,7 +870,7 @@ describe("AgentAwarenessRelay", () => { for (let i = 0; i < 3; i++) { yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; } assert.deepEqual( @@ -678,7 +895,7 @@ describe("AgentAwarenessRelay", () => { lastVisitedAt: null, }); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state), @@ -687,7 +904,7 @@ describe("AgentAwarenessRelay", () => { yield* Ref.set(currentShell, oldCompleted); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state), @@ -696,7 +913,7 @@ describe("AgentAwarenessRelay", () => { }), ); - it.effect("alerts a completion settled during its hold after an earlier tombstone", () => + it.effect("alerts a completion that settles after an earlier tombstone", () => Effect.gen(function* () { yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe(NOW))); const { relay, currentShell, publications } = yield* makeTestRelay(); @@ -708,7 +925,7 @@ describe("AgentAwarenessRelay", () => { lastVisitedAt: null, }); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; const completed = shell({ @@ -724,9 +941,11 @@ describe("AgentAwarenessRelay", () => { lastVisitedAt: null, }); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; - yield* TestClock.adjust("5 seconds"); + assert.deepEqual( + publications.map((publication) => publication.state?.phase ?? null), + [null, "completed"], + ); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state?.phase ?? null), @@ -735,7 +954,7 @@ describe("AgentAwarenessRelay", () => { }), ); - it.effect("drops a held completion when a queued continuation hides the thread", () => + it.effect("still publishes a completion when a queued continuation follows", () => Effect.gen(function* () { const { relay, currentShell, publications } = yield* makeTestRelay(); yield* relay.publishThread(THREAD_ID); @@ -743,12 +962,16 @@ describe("AgentAwarenessRelay", () => { yield* relay.publishThread(THREAD_ID); yield* Ref.set(currentShell, shell({ status: "queued" })); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; - assert.notInclude( + assert.include( publications.map((publication) => publication.state?.phase ?? null), "completed", ); + yield* TestClock.adjust("1500 millis"); + yield* relay.drain; + assert.deepEqual( + publications.map((publication) => publication.state?.phase ?? null), + ["running", "completed", null], + ); }), ); @@ -764,24 +987,26 @@ describe("AgentAwarenessRelay", () => { const completed = shell({ status: "completed" }); yield* Ref.set(currentShell, completed); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); - yield* relay.drain; + assert.deepEqual( + publications.map((publication) => publication.state?.phase ?? null), + ["running", "completed"], + ); yield* Ref.set(currentShell, { ...completed, settledOverride: "settled", settledAt: yield* DateTime.now, lastVisitedAt: null, }); + // Failed completion retries, then the settled tombstone confirms after its hold. yield* TestClock.adjust("1 second"); yield* relay.drain; - assert.deepEqual( - publications.map((publication) => publication.state?.phase ?? null), - ["running", "completed", "completed"], - ); - - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.equal(publications.at(-1)?.state ?? null, null); + assert.include( + publications.map((publication) => publication.state?.phase ?? null), + "completed", + ); }), ); @@ -796,14 +1021,14 @@ describe("AgentAwarenessRelay", () => { }); yield* Ref.set(currentShell, held); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.equal(publications.length, 1); assert.equal(publications[0]?.state?.phase, "running"); yield* Ref.set(currentShell, { ...held, pendingBackgroundTasks: [] }); yield* relay.publishThread(THREAD_ID); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.equal(publications.length, 2); assert.equal(publications[1]?.state, null); @@ -961,7 +1186,7 @@ describe("AgentAwarenessRelay", () => { yield* deliver( itemEvent("event:held-item:end", item("item:tool", { ...background, status: "completed" })), ); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("1500 millis"); yield* relay.drain; assert.deepEqual( publications.map((publication) => publication.state?.phase), diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index c14558762183a..8859acff5face 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -40,7 +40,9 @@ import * as HttpClientRequest from "effect/http/HttpClientRequest"; import * as HttpApiClient from "effect/http-api/HttpApiClient"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; import { + CLOUD_LINKED_USER_ID, isAgentActivityPublishingEnabledValue, PUBLISH_AGENT_ACTIVITY_SECRET, RELAY_ENVIRONMENT_CREDENTIAL_SECRET, @@ -53,6 +55,8 @@ import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.t import * as ProjectService from "../project/ProjectService.ts"; import { forkParked } from "../serverActivation.ts"; +const TOMBSTONE_CONFIRM_MS = 1_500; + export class AgentAwarenessRelay extends Context.Service< AgentAwarenessRelay, { @@ -344,6 +348,7 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: { environmentId: input.environmentId, project: input.project.value, thread: input.thread.value, + descendantPendingRequest: input.thread.value.descendantPendingRequest ?? null, }; const state = projectThreadAwarenessV2({ ...awarenessInput, @@ -386,6 +391,7 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: { environmentId: input.environmentId, project, thread, + descendantPendingRequest: thread.descendantPendingRequest ?? null, }); return ( state !== null && @@ -403,12 +409,16 @@ export const make = Effect.gen(function* () { const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const threads = yield* ThreadManagement.ThreadManagementService; const projects = yield* ProjectService.ProjectService; + const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; const random = yield* SecureRandom.SecureRandom; const signingKey = yield* EnvironmentSigningKey; const scope = yield* Effect.scope; const startedAt = (yield* DateTime.now).epochMilliseconds; const catchUpRequests = yield* Queue.dropping(1); const publishedStateByThreadRef = yield* Ref.make(new Map()); + // Unlisted children disappear from awareness snapshots on delete; remember + // the listed parent so side-chat/user_input rollups clear with them. + const childParentByThreadRef = yield* Ref.make(new Map()); // Android keys alerts by phase and time, so a child's unchanged request keeps // its first alert time through parent renames, relinks and the parent's own requests. const childAlertByThread = new Map(); @@ -452,8 +462,6 @@ export const make = Effect.gen(function* () { // Confirm through the same worker so tombstones cannot race live updates; recovery clears the deadline. const publishConfirmDeadlines = new Map(); - // Threads whose confirmation holds a completion, so a thread that settles during the hold still alerts. - const heldCompletions = new Set(); let schedulePublishConfirm: (threadId: ThreadId) => Effect.Effect = () => Effect.void; const publishRetries = new Map< ThreadId, @@ -473,8 +481,8 @@ export const make = Effect.gen(function* () { publishedRelayConfig = undefined; cachedRelayClient = undefined; publishConfirmDeadlines.clear(); - heldCompletions.clear(); yield* Ref.set(publishedStateByThreadRef, new Map()); + yield* Ref.set(childParentByThreadRef, new Map()); }); const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* (threadId: ThreadId) { @@ -532,6 +540,14 @@ export const make = Effect.gen(function* () { reason: input.reason, }); + // Local auth sessions are not mapped to arbitrary paired browsers' cloud + // accounts; presence only suppresses the environment's linked CloudLink user. + const [presentSessionIds, linkedCloudUserId] = yield* Effect.all([ + backgroundPolicy.presentWebOrDesktopSessionIds, + readSecretString(CLOUD_LINKED_USER_ID), + ]); + const suppressAlertUserIds = + presentSessionIds.length > 0 && linkedCloudUserId !== null ? [linkedCloudUserId] : []; const response = yield* relayClient.server .publishAgentActivity({ params: { @@ -541,6 +557,7 @@ export const make = Effect.gen(function* () { payload: { state: input.state, proof, + ...(suppressAlertUserIds.length > 0 ? { suppressAlertUserIds } : {}), }, }) .pipe(Effect.timeout("10 seconds")); @@ -561,9 +578,28 @@ export const make = Effect.gen(function* () { !isListedThreadLineage(threadShell.lineage) && !hasLivePublishedState(yield* Ref.get(publishedStateByThreadRef), threadId) ) { - // Subagents and side chats have no relay rows to clear; per-event tombstones would needlessly redeliver user aggregates. + // Side chats have no relay rows; publish toward the listed parent. Subagents reach it via trackSubagentRequest. + const parentId = threadShell.lineage.parentThreadId; + if (threadShell.lineage.sideChat === true && parentId !== null && parentId !== threadId) { + yield* Ref.update(childParentByThreadRef, (parents) => + new Map(parents).set(threadId, parentId), + ); + yield* enqueueTopLevelPublish(parentId); + } return; } + if (threadShell === null) { + const parentId = (yield* Ref.get(childParentByThreadRef)).get(threadId); + if (parentId !== undefined && parentId !== threadId) { + yield* Ref.update(childParentByThreadRef, (parents) => { + const next = new Map(parents); + next.delete(threadId); + return next; + }); + yield* enqueueTopLevelPublish(parentId); + return; + } + } const thread = threadShell === null || threadShell.archivedAt !== null ? Option.none() @@ -585,14 +621,9 @@ export const make = Effect.gen(function* () { ) { if (Option.isNone(thread) || !terminalWorkSinceStart(thread.value, startedAt)) return; } - if ( - publishedStateByThread.get(threadId) === publishIdentity && - // A held completion settled after an earlier clear still has to alert. - !(snapshot.state === null && heldCompletions.has(threadId)) - ) { + if (publishedStateByThread.get(threadId) === publishIdentity) { // Clear moot deadlines so later transient nulls cannot skip their deferral window. publishConfirmDeadlines.delete(threadId); - heldCompletions.delete(threadId); yield* Effect.logDebug("agent activity publish skipped; projected state unchanged", { environmentId, threadId, @@ -601,21 +632,18 @@ export const make = Effect.gen(function* () { return; } - // Confirm tombstones over live state to avoid removing active cards. - // Confirm completion: a run can end moments before its continuation starts, and the relay - // alerts on every move into completed. - const requiresConfirmation = snapshot.state === null || snapshot.state?.phase === "completed"; - if (requiresConfirmation) { - if (snapshot.state?.phase === "completed") heldCompletions.add(threadId); + // Confirm tombstones over live state to avoid removing active cards. Completions and + // failures publish immediately so mobile banners are not lost to follow-up work. + if (snapshot.state === null) { const nowMs = (yield* DateTime.now).epochMilliseconds; const deadline = publishConfirmDeadlines.get(threadId); if (deadline === undefined) { - publishConfirmDeadlines.set(threadId, nowMs + 5_000); + publishConfirmDeadlines.set(threadId, nowMs + TOMBSTONE_CONFIRM_MS); yield* Effect.logInfo("agent activity publish deferred pending confirmation", { environmentId, threadId, reason: snapshot.reason, - statePhase: snapshot.state?.phase ?? null, + statePhase: null, shell: describeThreadShellForAwareness(thread), }); yield* schedulePublishConfirm(threadId); @@ -628,41 +656,11 @@ export const make = Effect.gen(function* () { environmentId, threadId, reason: snapshot.reason, - statePhase: snapshot.state?.phase ?? null, + statePhase: null, shell: describeThreadShellForAwareness(thread), }); - // Settled during the hold, not resumed: alert the current completion, then clear the card - // on a new window. Only settlement may hide it; queued or silent runs also project to null. - const unsettled = - snapshot.state === null && - heldCompletions.has(threadId) && - Option.isSome(thread) && - thread.value.settledOverride === "settled" - ? resolveAgentAwarenessRelayPublishSnapshot({ - environmentId, - thread: Option.some({ ...thread.value, settledOverride: null }), - project, - pendingSubagentRequest: null, - }) - : null; - if (unsettled?.state?.phase === "completed") { - const completion = unsettled.state; - yield* publishState({ - projectId: unsettled.projectId, - state: completion, - reason: unsettled.reason, - }).pipe(withRelayClientTracing); - heldCompletions.delete(threadId); - publishConfirmDeadlines.set(threadId, nowMs + 5_000); - yield* Ref.update(publishedStateByThreadRef, (publishedStates) => - new Map(publishedStates).set(threadId, agentAwarenessPublishIdentity(completion)), - ); - yield* schedulePublishConfirm(threadId); - return; - } } else { publishConfirmDeadlines.delete(threadId); - heldCompletions.delete(threadId); } if (snapshot.reason === "thread-not-found") { @@ -695,7 +693,6 @@ export const make = Effect.gen(function* () { childAlertByThread.set(threadId, { requestId: childRequestId, updatedAt: state.updatedAt }); } publishConfirmDeadlines.delete(threadId); - heldCompletions.delete(threadId); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => new Map(publishedStates).set(threadId, publishIdentity), ); @@ -903,7 +900,7 @@ export const make = Effect.gen(function* () { schedulePublishConfirm = (threadId) => Effect.forkIn( - Effect.sleep("5 seconds").pipe( + Effect.sleep(`${TOMBSTONE_CONFIRM_MS} millis`).pipe( Effect.andThen(worker.enqueue(threadId)), Effect.catchCause((cause) => Effect.logWarning("deferred agent activity confirmation failed", { @@ -951,10 +948,28 @@ export const make = Effect.gen(function* () { if (!shouldPublishAgentAwarenessEvent(event)) { return Effect.void; } - return Effect.logDebug("agent activity publishing queued thread publish", { - eventType: event.type, - threadId, - }).pipe(Effect.andThen(enqueueThreadPublish(threadId))); + // Awareness snapshots omit deleted threads; remember the side-chat parent + // from the deletion payload so a catch-up or reset that never saw the + // child still republishes the rolled-up parent. + const deletedSideChatParentId = + event.type === "thread.deleted" && event.payload.lineage.sideChat === true + ? event.payload.lineage.parentThreadId + : null; + const rememberDeletedSideChatParent = + deletedSideChatParentId !== null && deletedSideChatParentId !== threadId + ? Ref.update(childParentByThreadRef, (parents) => + new Map(parents).set(threadId, deletedSideChatParentId), + ) + : Effect.void; + return rememberDeletedSideChatParent.pipe( + Effect.andThen( + Effect.logDebug("agent activity publishing queued thread publish", { + eventType: event.type, + threadId, + }), + ), + Effect.andThen(enqueueThreadPublish(threadId)), + ); }), ); }, diff --git a/infra/relay/scripts/android-push-smoke.ts b/infra/relay/scripts/android-push-smoke.ts index d4ee6a7af6b19..7524b8e491a1b 100644 --- a/infra/relay/scripts/android-push-smoke.ts +++ b/infra/relay/scripts/android-push-smoke.ts @@ -71,14 +71,12 @@ const main = Effect.gen(function* () { }).pipe(Effect.flatMap(decodeDevice)); const title = phase === "completed" - ? "Agent finished" + ? "Ready to review" : phase === "failed" ? "Agent failed" - : phase === "approval" - ? "Approval needed" - : phase === "input" - ? "Input needed" - : null; + : phase === "approval" || phase === "input" + ? "Needs your reply" + : null; const active = phase === "running" || phase === "approval" || phase === "input"; const now = yield* Clock.currentTimeMillis; const config: RelayConfiguration.RelayConfiguration["Service"] = { @@ -107,10 +105,12 @@ const main = Effect.gen(function* () { updated_at: String(now), active: String(active), activity_title: active - ? "1 active agent" + ? phase === "approval" || phase === "input" + ? "1 needs your reply" + : "1 working" : phase === "failed" - ? "Agent work failed" - : "Agent work completed", + ? "Failed" + : "All caught up", activity_expires_at: String( phase === "end" ? 0 : now + (active ? 2 * 60 * 60 * 1000 : 15 * 60 * 1000), ), diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 33f1518ec30ae..98d6a801d819e 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -453,7 +453,7 @@ describe("makeAggregateState", () => { }); expect(aggregate?.activeCount).toBe(1); - expect(aggregate?.subtitle).toBe("Agent work in progress"); + expect(aggregate?.subtitle).toBe("1 working"); expect(aggregate?.activities).toMatchObject([ { threadId: "thread-active", phase: "running" }, { threadId: "thread-done", phase: "completed", status: "Done" }, @@ -475,7 +475,7 @@ describe("makeAggregateState", () => { expect(aggregate).toMatchObject({ activeCount: 0, - subtitle: "Agent work completed", + subtitle: "All caught up", activities: [{ phase: "completed", status: "Done" }], }); expect( diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index 6ef258acdb6d3..10b5178ddfe9e 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -39,6 +39,8 @@ export class AgentActivityPublisher extends Context.Service< readonly environmentPublicKey: string; readonly threadId: string; readonly state: RelayAgentActivityState | null; + readonly suppressAlerts?: boolean; + readonly suppressAlertUserIds?: ReadonlyArray; }) => Effect.Effect; readonly replayForLiveActivityRegistration: (input: { readonly userId: string; @@ -62,6 +64,7 @@ const make = Effect.gen(function* () { readonly deliveryUser: EnvironmentLinks.AgentAwarenessDeliveryUserRecord; readonly state: RelayAgentActivityState | null; readonly nowMs: number; + readonly suppressAlerts: boolean; }) { const activeStates = input.deliveryUser.liveActivitiesEnabled ? yield* rows.listForUser({ userId: input.deliveryUser.userId }) @@ -88,7 +91,13 @@ const make = Effect.gen(function* () { targets, Effect.fnUntraced(function* (target) { if (target.platform === "android") { - return [yield* fcmDeliveries.enqueue({ target, state: input.state })]; + return [ + yield* fcmDeliveries.enqueue({ + target, + state: input.state, + ...(input.suppressAlerts ? { suppressAlerts: true } : {}), + }), + ]; } return yield* Effect.all( [ @@ -96,8 +105,9 @@ const make = Effect.gen(function* () { target, aggregate: liveActivityAggregate, nowMs: input.nowMs, + ...(input.suppressAlerts ? { suppressAlerts: true } : {}), }), - notificationOnlyAggregate === null + notificationOnlyAggregate === null || input.suppressAlerts ? Effect.succeed(null) : apnsDeliveries.sendPushNotificationForTarget({ target, @@ -225,6 +235,12 @@ const make = Effect.gen(function* () { environmentPublicKey: input.environmentPublicKey, }); const now = yield* DateTime.now; + const suppressAlertUserIds = new Set(input.suppressAlertUserIds ?? []); + const suppressAllAlerts = input.suppressAlerts === true; + yield* Effect.annotateCurrentSpan({ + "relay.agent_activity.suppress_alerts": suppressAllAlerts, + "relay.agent_activity.suppress_alert_user_count": suppressAlertUserIds.size, + }); const deliveriesByUser = yield* Effect.forEach( deliveryUsers, (deliveryUser) => @@ -232,6 +248,7 @@ const make = Effect.gen(function* () { deliveryUser, state: input.state, nowMs: now.epochMilliseconds, + suppressAlerts: suppressAllAlerts || suppressAlertUserIds.has(deliveryUser.userId), }), { concurrency: 4 }, ); diff --git a/infra/relay/src/agentActivity/ApnsClient.test.ts b/infra/relay/src/agentActivity/ApnsClient.test.ts index 81e76b0391aae..72409ec86f038 100644 --- a/infra/relay/src/agentActivity/ApnsClient.test.ts +++ b/infra/relay/src/agentActivity/ApnsClient.test.ts @@ -33,7 +33,7 @@ describe("ApnsClient", () => { const now = DateTime.makeUnsafe(0); const state: RelayAgentActivityAggregateState = { title: "T3 Code", - subtitle: "Agent work in progress", + subtitle: "1 working", activeCount: 1, updatedAt: DateTime.formatIso(now), activities: [ @@ -58,7 +58,7 @@ describe("ApnsClient", () => { event: "update", token: "token", state, - alert: { title: "Thread", body: "Approval: Project" }, + alert: { title: "Needs your reply", body: "Thread · Project" }, nowEpochSeconds: Math.floor(now.epochMilliseconds / 1_000), nowIso: DateTime.formatIso(now), }); @@ -68,12 +68,34 @@ describe("ApnsClient", () => { aps: { event: "update", alert: { - title: "Thread", - body: "Approval: Project", + title: "Needs your reply", + body: "Thread · Project", sound: "default", }, + "interruption-level": "active", + }, + }); + }).pipe(Effect.provide(layerTest)), + ); + + it.effect("keeps content-only Live Activity updates at priority 5", () => + Effect.gen(function* () { + const apns = yield* ApnsClient.ApnsClient; + const request = apns.makeLiveActivityRequest({ + event: "update", + token: "token", + state, + nowEpochSeconds: Math.floor(now.epochMilliseconds / 1_000), + nowIso: DateTime.formatIso(now), + }); + + expect(request.priority).toBe("5"); + expect(request.payload).toMatchObject({ + aps: { + event: "update", }, }); + expect((request.payload as { aps: Record }).aps.alert).toBeUndefined(); }).pipe(Effect.provide(layerTest)), ); @@ -119,7 +141,7 @@ describe("ApnsClient", () => { token: "push-token", notification: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/threads/env/thread", @@ -131,10 +153,11 @@ describe("ApnsClient", () => { aps: { alert: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", }, sound: "default", "thread-id": "env/thread", + "interruption-level": "active", }, body: { environmentId: "env", @@ -282,7 +305,7 @@ describe("ApnsClient", () => { token: "push-token", notification: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/threads/env/thread", diff --git a/infra/relay/src/agentActivity/ApnsClient.ts b/infra/relay/src/agentActivity/ApnsClient.ts index b21953c59eecc..f28eba36c99ba 100644 --- a/infra/relay/src/agentActivity/ApnsClient.ts +++ b/infra/relay/src/agentActivity/ApnsClient.ts @@ -111,6 +111,8 @@ function liveActivityAlertPayload(alert: ApnsLiveActivityAlert) { body: alert.body, sound: "default", }, + // Respect Focus / notification summary; not time-sensitive. + "interruption-level": "active", }; } @@ -191,6 +193,8 @@ function buildLiveActivityRequest(input: MakeLiveActivityRequestInput): ApnsLive return { token: input.token, event: input.event, + // Content-only updates stay at 5 so they do not burn ActivityKit's hourly + // priority-10 budget; alerts and ends still need immediate delivery. priority: input.event === "update" && !input.alert ? "5" : "10", payload: { aps: { @@ -225,6 +229,8 @@ function makePushNotificationRequest(input: { }, sound: "default", "thread-id": notificationThreadId(input.notification), + // Respect Focus / notification summary; not time-sensitive. + "interruption-level": "active", }, // Expo iOS exposes userInfo.body as notification.content.data for remote pushes. body: { diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 64823f4973593..625d1c009a3ac 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -26,6 +26,7 @@ import * as AgentActivityRows from "./AgentActivityRows.ts"; import * as ApnsDeliveries from "./ApnsDeliveries.ts"; import * as ApnsClient from "./ApnsClient.ts"; import * as ApnsProviderTokens from "./ApnsProviderTokens.ts"; +import { makeAggregateState, statusForPhase } from "./agentActivityAggregate.ts"; const config = RelayConfiguration.RelayConfiguration.of({ relayIssuer: "https://relay.example.test", @@ -74,7 +75,7 @@ const state: RelayAgentActivityState = { const aggregate: RelayAgentActivityAggregateState = { title: "T3 Code", - subtitle: "Agent work in progress", + subtitle: "1 working", activeCount: 1, updatedAt: state.updatedAt, activities: [ @@ -92,6 +93,26 @@ const aggregate: RelayAgentActivityAggregateState = { ], }; +describe("inbox aggregate copy", () => { + it("summarizes working and caught-up threads like an inbox", () => { + expect(statusForPhase("running")).toBe("Working"); + expect( + makeAggregateState({ + activeStates: [state], + terminalState: null, + nowMs: 0, + })?.subtitle, + ).toBe("1 working"); + expect( + makeAggregateState({ + activeStates: [], + terminalState: { ...state, phase: "completed" }, + nowMs: 0, + })?.subtitle, + ).toBe("All caught up"); + }); +}); + const enabledPreferences = JSON.stringify({ liveActivitiesEnabled: true, notificationsEnabled: true, @@ -119,6 +140,15 @@ const notificationsDisabledPreferences = JSON.stringify({ notifyOnFailure: true, }); +const liveActivityWithoutPushPreferences = JSON.stringify({ + liveActivitiesEnabled: true, + notificationsEnabled: false, + notifyOnApproval: true, + notifyOnInput: true, + notifyOnCompletion: true, + notifyOnFailure: true, +}); + const target: LiveActivities.TargetRow = { user_id: "dev:julius", device_id: "device-1", @@ -324,7 +354,7 @@ describe("ApnsDeliveries", () => { it.effect.each([ { label: "inside", nowMs: 5_000, queued: false }, - { label: "after", nowMs: 19_000, queued: true }, + { label: "after", nowMs: 8_000, queued: true }, ])("throttles timestamp-only updates $label the window", ({ nowMs, queued }) => { const attempts: Array = []; const queuedJobs: Array = []; @@ -397,8 +427,8 @@ describe("ApnsDeliveries", () => { token: "apns-device-token", }, notification: { - title: "Thread", - body: "Input: Project", + title: "Needs your reply", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/", @@ -454,6 +484,165 @@ describe("ApnsDeliveries", () => { }).pipe(Effect.provide(layerFor({ attempts, queuedJobs }))); }); + it.effect( + "keeps Live Activity updates but skips companion banners when alerts are suppressed", + () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const previousAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "running", + status: "Working", + }, + ], + }; + const inputAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "waiting_for_approval", + status: "Approval", + }, + ], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + push_token: "apns-device-token", + last_aggregate_json: JSON.stringify(previousAggregate), + }, + aggregate: inputAggregate, + nowMs: 5_000, + suppressAlerts: true, + }); + + expect(result?.kind).toBe("live_activity_update"); + expect(queuedJobs).toHaveLength(1); + expect(queuedJobs[0]?.payload.kind).toBe("live_activity_update"); + expect(queuedJobs[0]?.payload.alert ?? null).toBeNull(); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(layerFor({ attempts, queuedJobs }))); + }, + ); + + it.effect("keeps the Live Activity alert when push notifications are preferred off", () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const previousAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "running", + status: "Working", + }, + ], + }; + const inputAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "waiting_for_approval", + status: "Approval", + }, + ], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + push_token: "apns-device-token", + preferences_json: liveActivityWithoutPushPreferences, + last_aggregate_json: JSON.stringify(previousAggregate), + }, + aggregate: inputAggregate, + nowMs: 5_000, + }); + + expect(result?.kind).toBe("live_activity_update"); + expect(queuedJobs).toHaveLength(1); + expect(queuedJobs[0]?.payload).toMatchObject({ + kind: "live_activity_update", + alert: { title: "Needs your reply" }, + }); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(layerFor({ attempts, queuedJobs }))); + }); + + it.effect( + "queues a companion push with Live Activity updates so stale activity tokens cannot swallow alerts", + () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const previousAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "running", + status: "Working", + }, + ], + }; + const inputAggregate: RelayAgentActivityAggregateState = { + ...aggregate, + activities: [ + { + ...aggregate.activities[0]!, + phase: "waiting_for_approval", + status: "Approval", + }, + ], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendForTarget({ + target: { + ...target, + push_token: "apns-device-token", + last_aggregate_json: JSON.stringify(previousAggregate), + }, + aggregate: inputAggregate, + nowMs: 5_000, + }); + + expect(result?.kind).toBe("live_activity_update"); + expect(queuedJobs[0]?.payload.alert ?? null).toBeNull(); + expect(queuedJobs).toMatchObject([ + { + payload: { + kind: "live_activity_update", + target: { token: "activity-token" }, + }, + }, + { + payload: { + kind: "push_notification", + target: { token: "apns-device-token" }, + notification: { + title: "Needs your reply", + body: "Thread · Project", + phase: "waiting_for_approval", + }, + }, + }, + ]); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(layerFor({ attempts, queuedJobs }))); + }, + ); + it.effect( "queues a push notification for approval and input states when no Live Activity delivery is available", () => { @@ -494,8 +683,8 @@ describe("ApnsDeliveries", () => { token: "apns-device-token", }, notification: { - title: "Thread", - body: "Input: Project", + title: "Needs your reply", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/", @@ -508,6 +697,107 @@ describe("ApnsDeliveries", () => { }, ); + it.effect("queues a push notification for completed states on the notification-only path", () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const completedAggregate: RelayAgentActivityAggregateState = { + title: "T3 Code", + subtitle: "All caught up", + activeCount: 0, + updatedAt: state.updatedAt, + activities: [ + { + ...aggregate.activities[0]!, + phase: "completed", + status: "Done", + }, + ], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendPushNotificationForTarget({ + target: { + ...target, + preferences_json: disabledPreferences, + push_token: "apns-device-token", + push_to_start_token: null, + activity_push_token: null, + remote_started_at: null, + }, + aggregate: completedAggregate, + }); + + expect(result?.kind).toBe("push_notification"); + expect(result?.ok).toBe(true); + expect(queuedJobs).toMatchObject([ + { + payload: { + kind: "push_notification", + notification: { + title: "Ready to review", + body: "Thread · Project", + phase: "completed", + }, + }, + }, + ]); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(layerFor({ attempts, queuedJobs }))); + }); + + it.effect("repairs device prefs that alert on approval but silence completion", () => { + const attempts: Array = []; + const queuedJobs: Array = []; + const inconsistentPreferences = JSON.stringify({ + liveActivitiesEnabled: false, + notificationsEnabled: true, + notifyOnApproval: true, + notifyOnInput: true, + notifyOnCompletion: false, + notifyOnFailure: false, + }); + const completedAggregate: RelayAgentActivityAggregateState = { + title: "T3 Code", + subtitle: "All caught up", + activeCount: 0, + updatedAt: state.updatedAt, + activities: [ + { + ...aggregate.activities[0]!, + phase: "completed", + status: "Done", + }, + ], + }; + + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const result = yield* deliveries.sendPushNotificationForTarget({ + target: { + ...target, + preferences_json: inconsistentPreferences, + push_token: "apns-device-token", + push_to_start_token: null, + activity_push_token: null, + remote_started_at: null, + }, + aggregate: completedAggregate, + }); + + expect(result?.kind).toBe("push_notification"); + expect(queuedJobs).toMatchObject([ + { + payload: { + kind: "push_notification", + notification: { phase: "completed" }, + }, + }, + ]); + expect(attempts).toEqual([]); + }).pipe(Effect.provide(layerFor({ attempts, queuedJobs }))); + }); + it.effect("queues bounded alert notification payloads", () => { const attempts: Array = []; const queuedJobs: Array = []; @@ -616,7 +906,7 @@ describe("ApnsDeliveries", () => { aggregate: null, notification: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/", @@ -678,7 +968,7 @@ describe("ApnsDeliveries", () => { aggregate: null, notification: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/", @@ -776,6 +1066,7 @@ describe("ApnsDeliveries", () => { it.effect("skips a queued Done Live Activity update after the thread resumes working", () => { const attempts: Array = []; + const queuedJobs: Array = []; let executeCount = 0; const completedAt = "1970-01-01T00:00:01.000Z"; const completedAggregate: RelayAgentActivityAggregateState = { @@ -797,7 +1088,7 @@ describe("ApnsDeliveries", () => { deviceId: target.device_id, token: target.activity_push_token ?? "activity-token", aggregate: completedAggregate, - alert: { title: "Thread", body: "Done: Project" }, + alert: { title: "Ready to review", body: "Thread · Project" }, createdAt: completedAt, expiresAt: "1970-01-01T00:10:00.000Z", jobId: "job-update-superseded-by-running", @@ -829,10 +1120,35 @@ describe("ApnsDeliveries", () => { apnsReason: "Stale agent activity state skipped.", }, ]); + expect(queuedJobs).toHaveLength(1); + expect(queuedJobs[0]?.payload).toMatchObject({ + kind: "push_notification", + notification: { + title: "Ready to review", + body: "Thread · Project", + threadId: "thread", + }, + }); + // Superseded completions omit phase/updatedAt so the push survives resume. + const supersededPayload = queuedJobs[0]?.payload; + expect(supersededPayload?.kind).toBe("push_notification"); + expect( + supersededPayload?.kind === "push_notification" + ? supersededPayload.notification?.phase + : "missing", + ).toBeUndefined(); }).pipe( Effect.provide( layerFor({ attempts, + queuedJobs, + currentTargets: [ + { + ...target, + push_token: "apns-device-token", + last_aggregate_json: JSON.stringify(aggregate), + }, + ], activityStates: [ { ...state, @@ -859,7 +1175,7 @@ describe("ApnsDeliveries", () => { aggregate: null, notification: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/", @@ -913,7 +1229,7 @@ describe("ApnsDeliveries", () => { aggregate: null, notification: { title: "Thread", - body: "Failed: Project", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/", @@ -1043,7 +1359,7 @@ describe("live activity alert decisions", () => { }, preferences, }); - expect(alert).toEqual({ title: "Blocked thread", body: "Approval: Project" }); + expect(alert).toEqual({ title: "Needs your reply", body: "Blocked thread · Project" }); }); it("stays silent without a delivered baseline so replays cannot buzz", () => { @@ -1074,7 +1390,7 @@ describe("live activity alert decisions", () => { preferences, nowMs: 0, }), - ).toEqual({ title: "Thread", body: "Done: Project" }); + ).toEqual({ title: "Ready to review", body: "Thread · Project" }); expect( AgentActivityAlerts.alertForNewlyTerminal({ previousAggregate: aggregate, @@ -1188,7 +1504,7 @@ describe("signed APNs registration metadata", () => { ? { notification: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", environmentId: "env", threadId: "thread", deepLink: "/", diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 8913bdd1f8c5b..383b3dc35862a 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -43,13 +43,46 @@ import * as ApnsDeliveryQueue from "./ApnsDeliveryQueue.ts"; import { withSpanAttributes } from "../observability.ts"; import { - alertForAttentionTransition, - alertForNewlyTerminal, + alertForActivityRows, alertForTerminalAggregate, + attentionTransitionRows, + repairAwarenessPreferences, rowPhasesChanged, + rowStatusesChanged, shouldAlertForActivity, + terminalTransitionRows, } from "./agentActivityAlerts.ts"; -const MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS = 15_000; + +type AggregateActivityRow = RelayAgentActivityAggregateState["activities"][number]; + +type LiveActivityAlertChoice = { + readonly alert: ApnsLiveActivityAlert; + readonly row: AggregateActivityRow; +}; + +function liveActivityAlertForTransition(input: { + readonly previousAggregate: RelayAgentActivityAggregateState | null; + readonly nextAggregate: RelayAgentActivityAggregateState; + readonly preferences: RelayAgentAwarenessPreferences | null; + readonly nowMs: number; +}): LiveActivityAlertChoice | null { + const attention = attentionTransitionRows(input); + if (attention[0]) { + const alert = alertForActivityRows(attention); + return alert ? { alert, row: attention[0] } : null; + } + const terminal = terminalTransitionRows({ + ...input, + includeUnobserved: true, + }); + if (terminal[0]) { + const alert = alertForActivityRows(terminal); + return alert ? { alert, row: terminal[0] } : null; + } + return null; +} +// Mid-run status labels coalesce on the host (~3s); keep the relay floor aligned. +const MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS = 3_000; const FRESHLY_ARMED_GRACE_MS = 2 * 60 * 1_000; const PERMANENT_APNS_TOKEN_REASONS = new Set([ "BadDeviceToken", @@ -68,12 +101,14 @@ type ChosenLiveActivityDelivery = readonly token: string; readonly aggregate: RelayAgentActivityAggregateState; readonly alert: ApnsLiveActivityAlert | null; + readonly alertRow: AggregateActivityRow | null; } | { readonly kind: "live_activity_end"; readonly token: string; readonly aggregate: RelayAgentActivityAggregateState | null; readonly alert: ApnsLiveActivityAlert | null; + readonly alertRow: AggregateActivityRow | null; }; type ChosenPushNotificationDelivery = { @@ -139,7 +174,8 @@ function parseAggregate(value: string | null): RelayAgentActivityAggregateState } function parsePreferences(value: string): RelayAgentAwarenessPreferences | null { - return Option.getOrNull(decodeRelayAgentAwarenessPreferencesJson(value)); + const preferences = Option.getOrNull(decodeRelayAgentAwarenessPreferencesJson(value)); + return preferences === null ? null : repairAwarenessPreferences(preferences); } function aggregateNeedsAttention(aggregate: RelayAgentActivityAggregateState): boolean { @@ -166,9 +202,12 @@ function shouldUpdateLiveActivity(input: { if (aggregateNeedsAttention(input.nextAggregate)) { return true; } - // Phase and row changes may be the last update until the next transition. + // Phase and status changes may be the last update until the next transition. // Throttling them would leave the card showing a stale state indefinitely. - if (rowPhasesChanged(input.previousAggregate, input.nextAggregate)) { + if ( + rowPhasesChanged(input.previousAggregate, input.nextAggregate) || + rowStatusesChanged(input.previousAggregate, input.nextAggregate) + ) { return true; } const lastDeliveryAtMs = @@ -207,6 +246,24 @@ function notificationForAggregate(input: { return notificationForActivity(activity); } +function companionNotificationForAlert(input: { + readonly alert: ApnsLiveActivityAlert; + readonly row: AggregateActivityRow; + /** When false, omit phase/updatedAt so a superseded completion still rings. */ + readonly requireCurrentState?: boolean; +}): ApnsNotificationPayload { + return sanitizeApnsNotificationPayload({ + title: input.alert.title, + body: input.alert.body, + environmentId: input.row.environmentId, + threadId: input.row.threadId, + deepLink: input.row.deepLink, + ...(input.requireCurrentState === false + ? {} + : { phase: input.row.phase, updatedAt: input.row.updatedAt }), + }); +} + // Suppressed Live Activity updates must not fall back to alert pushes or unchanged replays would ring. function chooseLiveActivityDelivery(input: { readonly target: LiveActivities.TargetRow; @@ -222,6 +279,7 @@ function chooseLiveActivityDelivery(input: { token: input.target.activity_push_token, aggregate: null, alert: null, + alertRow: null, } : null; } @@ -244,36 +302,36 @@ function chooseLiveActivityDelivery(input: { token: input.target.activity_push_token, aggregate: null, alert: null, + alertRow: null, }; } const nextAggregate = input.aggregate; const previousAggregate = parseAggregate(input.target.last_aggregate_json); - return shouldUpdateLiveActivity({ - previousAggregate, - nextAggregate, - lastDeliveryAt: input.target.last_live_activity_delivery_at, - nowMs: input.nowMs, - }) - ? { - kind: "live_activity_update", - token: input.target.activity_push_token, - aggregate: nextAggregate, - alert: input.replay - ? null - : (alertForAttentionTransition({ - previousAggregate, - nextAggregate, - preferences, - }) ?? - alertForNewlyTerminal({ - previousAggregate, - nextAggregate, - preferences, - nowMs: input.nowMs, - includeUnobserved: true, - })), - } - : "suppressed"; + if ( + !shouldUpdateLiveActivity({ + previousAggregate, + nextAggregate, + lastDeliveryAt: input.target.last_live_activity_delivery_at, + nowMs: input.nowMs, + }) + ) { + return "suppressed"; + } + const transition = input.replay + ? null + : liveActivityAlertForTransition({ + previousAggregate, + nextAggregate, + preferences, + nowMs: input.nowMs, + }); + return { + kind: "live_activity_update", + token: input.target.activity_push_token, + aggregate: nextAggregate, + alert: transition?.alert ?? null, + alertRow: transition?.row ?? null, + }; } function chooseDelivery(input: { @@ -486,6 +544,8 @@ export class ApnsDeliveries extends Context.Service< readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; + /** Keep Live Activity content updates; skip audible banners / companion pushes. */ + readonly suppressAlerts?: boolean; }) => Effect.Effect; readonly sendPushNotificationForTarget: (input: { readonly target: LiveActivities.TargetRow; @@ -687,24 +747,22 @@ const make = Effect.gen(function* () { return staleJobResult({ deviceId: input.target.device_id, kind: input.kind }); } deliveryTarget = currentTarget; - if (alert) { + let transitionAlert: LiveActivityAlertChoice | null = null; + if (alert && aggregate) { const preferences = parsePreferences(currentTarget.preferences_json); const previousAggregate = parseAggregate(currentTarget.last_aggregate_json); - alert = - !preferences?.notificationsEnabled || !aggregate - ? null - : (alertForAttentionTransition({ - previousAggregate, - nextAggregate: aggregate, - preferences, - }) ?? - alertForNewlyTerminal({ - previousAggregate, - nextAggregate: aggregate, - preferences, - nowMs: now.epochMilliseconds, - includeUnobserved: true, - })); + // Gate on notifyOn* flags only. notificationsEnabled was historically set from + // push-token presence, so a granted OS permission with a late token left it + // false and silently stripped every Live Activity banner. + transitionAlert = liveActivityAlertForTransition({ + previousAggregate, + nextAggregate: aggregate, + preferences, + nowMs: now.epochMilliseconds, + }); + alert = transitionAlert?.alert ?? null; + } else { + alert = null; } if ( aggregate !== null && @@ -713,6 +771,22 @@ const make = Effect.gen(function* () { aggregate, })) ) { + // A completion/attention alert on a superseded Live Activity update still needs to ring. + // Omit phase/updatedAt so notificationStateIsCurrent cannot drop the event after resume. + if (transitionAlert && currentTarget.push_token) { + yield* deliveryQueue.enqueuePushNotification({ + userId: currentTarget.user_id, + deviceId: currentTarget.device_id, + token: currentTarget.push_token, + bundleId: currentTarget.bundle_id, + apsEnvironment: currentTarget.aps_environment, + notification: companionNotificationForAlert({ + alert: transitionAlert.alert, + row: transitionAlert.row, + requireCurrentState: false, + }), + }); + } yield* attempts.completeSourceJob({ sourceJobId: input.sourceJobId, apnsReason: "Stale agent activity state skipped.", @@ -1036,6 +1110,7 @@ const make = Effect.gen(function* () { }), sendForTarget: Effect.fnUntraced(function* (input) { if (!config.apns) return null; + const suppressAlerts = input.suppressAlerts === true; const delivery = chooseDelivery({ target: input.target, aggregate: input.aggregate, @@ -1046,6 +1121,9 @@ const make = Effect.gen(function* () { return null; } if (delivery.kind === "push_notification") { + if (suppressAlerts) { + return null; + } const result = yield* deliveryQueue.enqueuePushNotification({ userId: input.target.user_id, deviceId: input.target.device_id, @@ -1056,24 +1134,56 @@ const make = Effect.gen(function* () { }); return result; } - const notification = input.replay - ? null - : notificationForAggregate({ - target: input.target, - aggregate: input.aggregate, - nowMs: input.nowMs, - }); - // Use the end alert only when no companion notification will ring, preserving feedback for Live-Activity-only users. - const alert = input.replay - ? null - : delivery.kind === "live_activity_end" - ? notification && input.target.push_token - ? null - : alertForTerminalAggregate({ - aggregate: delivery.aggregate, - preferences: parsePreferences(input.target.preferences_json), - }) - : delivery.alert; + const preferences = parsePreferences(input.target.preferences_json); + const endAlertRow = + delivery.kind === "live_activity_end" ? (delivery.aggregate?.activities[0] ?? null) : null; + const endAlert = + input.replay || suppressAlerts || delivery.kind !== "live_activity_end" + ? null + : alertForTerminalAggregate({ + aggregate: delivery.aggregate, + preferences, + }); + const liveActivityAlert = + input.replay || suppressAlerts + ? null + : delivery.kind === "live_activity_end" + ? endAlert + : delivery.alert; + const alertRow = + delivery.kind === "live_activity_end" + ? endAlert && endAlertRow + ? endAlertRow + : null + : delivery.alertRow; + // Prefer a companion push when allowed; otherwise keep the Live Activity alert. + // Skip companions when notifications are off so a leftover token cannot mute both. + const transitionCompanion = + input.target.push_token === null || + preferences?.notificationsEnabled !== true || + input.replay || + suppressAlerts || + liveActivityAlert === null || + alertRow === null + ? null + : companionNotificationForAlert({ + alert: liveActivityAlert, + row: alertRow, + }); + const endCompanion = + transitionCompanion !== null || + input.target.push_token === null || + input.replay || + suppressAlerts || + delivery.kind !== "live_activity_end" + ? null + : notificationForAggregate({ + target: input.target, + aggregate: input.aggregate, + nowMs: input.nowMs, + }); + const companionNotification = transitionCompanion ?? endCompanion; + const alert = companionNotification ? null : liveActivityAlert; const result = yield* deliveryQueue.enqueueLiveActivity({ userId: input.target.user_id, deviceId: input.target.device_id, @@ -1084,14 +1194,14 @@ const make = Effect.gen(function* () { aggregate: delivery.aggregate, alert, }); - if (delivery.kind === "live_activity_end" && notification && input.target.push_token) { + if (companionNotification && input.target.push_token) { yield* deliveryQueue.enqueuePushNotification({ userId: input.target.user_id, deviceId: input.target.device_id, token: input.target.push_token, bundleId: input.target.bundle_id, apsEnvironment: input.target.aps_environment, - notification, + notification: companionNotification, }); } return result; diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts index 358698dd197b3..9c87427d71895 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueue.test.ts @@ -53,7 +53,7 @@ describe("ApnsDeliveryQueue", () => { token: "push-token", notification: { title: "Thread", - body: "Input: Project", + body: "Thread · Project", environmentId: "env-1", threadId: "thread-1", deepLink: "/threads/env-1/thread-1", diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index f1ce50d5832b6..78bdafd57f1a3 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -220,10 +220,10 @@ describe("Android delivery routing", () => { { firstPhase: "waiting_for_approval", secondPhase: "waiting_for_input", - title: "2 agents need attention", + title: "2 need your reply", active: "true", }, - { firstPhase: "completed", secondPhase: "failed", title: "2 agents finished", active: "false" }, + { firstPhase: "completed", secondPhase: "failed", title: "2 ready to review", active: "false" }, ] as const)( "routes grouped $firstPhase and $secondPhase to the overview once across their queued jobs", ({ firstPhase, secondPhase, title, active }) => { @@ -260,8 +260,8 @@ describe("Android delivery routing", () => { const delivery = yield* FcmDeliveries.FcmDeliveries; yield* delivery.process({ ...h.job, state: h.current.state }); expect(h.sent[0]?.data).toMatchObject({ - alert_title: "Second thread", - alert_body: "Input: Project", + alert_title: "Needs your reply", + alert_body: "Second thread · Project", alert_path: "/threads/env/second-thread", }); }).pipe(Effect.provide(h.layer)); @@ -290,8 +290,8 @@ describe("Android delivery routing", () => { const delivery = yield* FcmDeliveries.FcmDeliveries; yield* delivery.process({ ...h.job, state: h.current.state }); expect(h.sent[0]?.data).toMatchObject({ - alert_title: "Second thread", - alert_body: "Input: Project", + alert_title: "Needs your reply", + alert_body: "Second thread · Project", }); yield* delivery.process({ ...h.job, state: h.current.otherStates[0] }); expect(h.sent.filter((message) => message.alert)).toHaveLength(1); @@ -346,8 +346,8 @@ describe("Android delivery routing", () => { const delivery = yield* FcmDeliveries.FcmDeliveries; yield* delivery.process({ ...h.job, state: h.current.state }); expect(h.sent[0]?.data).toMatchObject({ - alert_title: "Fix notifications", - alert_body: "Approval: Project", + alert_title: "Needs your reply", + alert_body: "Fix notifications · Project", }); }).pipe(Effect.provide(h.layer)); }, @@ -369,7 +369,7 @@ describe("Android delivery routing", () => { nowMs: 0, }; const alert = FcmDeliveries.androidAlertForAggregate(input); - expect(alert?.alert_title).toBe("2 agents finished"); + expect(alert?.alert_title).toBe("2 ready to review"); expect( FcmDeliveries.androidAlertForAggregate({ ...input, @@ -407,7 +407,10 @@ describe("Android delivery routing", () => { preferences, nowMs: 0, }), - ).toMatchObject({ alert_title: "Second thread", alert_body: "Input: Project" }); + ).toMatchObject({ + alert_title: "Needs your reply", + alert_body: "Second thread · Project", + }); }); it("keeps notification groups distinct when identifiers contain slashes", () => { @@ -447,8 +450,8 @@ describe("Android delivery routing", () => { yield* delivery.process({ ...h.job, state: h.current.state }); expect(h.sent[0]?.data).toMatchObject({ active: "false", - alert_title: "Fix notifications", - alert_body: "Done: Project", + alert_title: "Ready to review", + alert_body: "Fix notifications · Project", alert_path: "/threads/env/thread", }); }).pipe(Effect.provide(h.layer)); @@ -467,8 +470,8 @@ describe("Android delivery routing", () => { yield* delivery.process({ ...h.job, state: h.current.state }); expect(h.sent[0]?.data).toMatchObject({ active: "false", - alert_title: "Fix notifications", - alert_body: "Done: Project", + alert_title: "Ready to review", + alert_body: "Fix notifications · Project", }); expect(h.sent[0]?.data.user_id).toBe("user"); }).pipe(Effect.provide(h.layer)); @@ -562,7 +565,7 @@ describe("Android delivery routing", () => { user_id: "u".repeat(128), updated_at: "1788780000000", alert_id: "a".repeat(64), - alert_title: "5 agents finished", + alert_title: "5 ready to review", alert_body: Array(5).fill(longTitle).join(", "), alert_path: "/threads/env/thread", }); diff --git a/infra/relay/src/agentActivity/FcmDeliveries.ts b/infra/relay/src/agentActivity/FcmDeliveries.ts index d2795a90a88e5..35a4046c12962 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.ts @@ -29,6 +29,7 @@ import { alertForActivityRows, attentionTransitionRows, terminalTransitionRows, + repairAwarenessPreferences, shouldAlertForActivity, } from "./agentActivityAlerts.ts"; @@ -39,12 +40,15 @@ export const FcmDeliveryJob = Schema.Struct({ state: Schema.NullOr(RelayAgentActivityState), queuedAt: Schema.Number, replay: Schema.optional(Schema.Boolean), + suppressAlerts: Schema.optional(Schema.Boolean), }); export type FcmDeliveryJob = typeof FcmDeliveryJob.Type; const decodeJob = Schema.decodeUnknownEffect(FcmDeliveryJob); -const decodePreferences = Schema.decodeUnknownOption( +const decodePreferencesJson = Schema.decodeUnknownOption( Schema.fromJsonString(RelayAgentAwarenessPreferences), ); +const decodePreferences = (value: string) => + Option.map(decodePreferencesJson(value), repairAwarenessPreferences); const decodePreviousActivity = Schema.decodeUnknownOption( Schema.fromJsonString(RelayAgentActivityAggregateState), ); @@ -120,6 +124,7 @@ export class FcmDeliveries extends Context.Service< readonly target: LiveActivities.TargetRow; readonly state: RelayAgentActivityState | null; readonly replay?: boolean; + readonly suppressAlerts?: boolean; }) => Effect.Effect; readonly process: ( body: unknown, @@ -169,6 +174,7 @@ const make = Effect.gen(function* () { token: input.target.push_token, state: input.state, ...(input.replay ? { replay: true } : {}), + ...(input.suppressAlerts ? { suppressAlerts: true } : {}), queuedAt: now.epochMilliseconds, }) .pipe(Effect.mapError((cause) => new FcmDeliveryError({ operation: "enqueue", cause }))); @@ -239,46 +245,50 @@ const make = Effect.gen(function* () { // Notification-only jobs must not acknowledge another environment's pending card transitions. acknowledgeAggregate = deliveryUser?.liveActivitiesEnabled === true || !preferences.value.liveActivitiesEnabled; - if ( - deliveryUser?.liveActivitiesEnabled && - preferences.value.liveActivitiesEnabled && - previousAggregate && - aggregate - ) { - const environmentIds = [...new Set(aggregate.activities.map((row) => row.environmentId))]; - const allowedEnvironments = new Set(); - for (const environmentId of environmentIds) { - const environmentLink = yield* links.getForUser({ - userId: job.userId, - environmentId, - }); - if (!environmentLink) continue; - const users = yield* links.listDeliveryUsersForEnvironment({ - environmentId, - environmentPublicKey: environmentLink.environmentPublicKey, - }); - if (users.some((user) => user.userId === job.userId && user.notificationsEnabled)) { - allowedEnvironments.add(environmentId); + if (job.suppressAlerts !== true) { + if ( + deliveryUser?.liveActivitiesEnabled && + preferences.value.liveActivitiesEnabled && + previousAggregate && + aggregate + ) { + const environmentIds = [ + ...new Set(aggregate.activities.map((row) => row.environmentId)), + ]; + const allowedEnvironments = new Set(); + for (const environmentId of environmentIds) { + const environmentLink = yield* links.getForUser({ + userId: job.userId, + environmentId, + }); + if (!environmentLink) continue; + const users = yield* links.listDeliveryUsersForEnvironment({ + environmentId, + environmentPublicKey: environmentLink.environmentPublicKey, + }); + if (users.some((user) => user.userId === job.userId && user.notificationsEnabled)) { + allowedEnvironments.add(environmentId); + } } + alert = androidAlertForAggregate({ + previousAggregate, + nextAggregate: { + ...aggregate, + activities: aggregate.activities.filter((row) => + allowedEnvironments.has(row.environmentId), + ), + }, + preferences: preferences.value, + nowMs: now.epochMilliseconds, + }); + } else if ( + deliveryUser?.notificationsEnabled && + state?.phase === job.state.phase && + state.updatedAt === job.state.updatedAt && + !isExpiredAgentActivityState(state, now.epochMilliseconds) + ) { + alert = androidAlertForState(state, preferences.value, now.epochMilliseconds); } - alert = androidAlertForAggregate({ - previousAggregate, - nextAggregate: { - ...aggregate, - activities: aggregate.activities.filter((row) => - allowedEnvironments.has(row.environmentId), - ), - }, - preferences: preferences.value, - nowMs: now.epochMilliseconds, - }); - } else if ( - deliveryUser?.notificationsEnabled && - state?.phase === job.state.phase && - state.updatedAt === job.state.updatedAt && - !isExpiredAgentActivityState(state, now.epochMilliseconds) - ) { - alert = androidAlertForState(state, preferences.value, now.epochMilliseconds); } } const displayedAggregate = preferences.value.liveActivitiesEnabled ? aggregate : null; diff --git a/infra/relay/src/agentActivity/LiveActivities.test.ts b/infra/relay/src/agentActivity/LiveActivities.test.ts index 5da94df01f241..68b18f999c221 100644 --- a/infra/relay/src/agentActivity/LiveActivities.test.ts +++ b/infra/relay/src/agentActivity/LiveActivities.test.ts @@ -12,7 +12,7 @@ import * as LiveActivities from "./LiveActivities.ts"; const aggregate: RelayAgentActivityAggregateState = { title: "T3 Code", - subtitle: "Agent work in progress", + subtitle: "1 working", activeCount: 1, updatedAt: "2026-05-25T00:00:00.000Z", activities: [ diff --git a/infra/relay/src/agentActivity/agentActivityAggregate.ts b/infra/relay/src/agentActivity/agentActivityAggregate.ts index 99fdcf71ef90d..2f145dc4d09aa 100644 --- a/infra/relay/src/agentActivity/agentActivityAggregate.ts +++ b/infra/relay/src/agentActivity/agentActivityAggregate.ts @@ -45,13 +45,40 @@ function aggregateRowForState(state: RelayAgentActivityState) { }; } +function inboxSubtitle(input: { + readonly activeCount: number; + readonly activities: ReadonlyArray<{ readonly phase: RelayAgentActivityState["phase"] }>; + readonly terminalOnly: boolean; + readonly failed: boolean; +}): string { + const needsAttention = input.activities.filter( + (row) => row.phase === "waiting_for_approval" || row.phase === "waiting_for_input", + ).length; + if (needsAttention > 0) { + return needsAttention === 1 ? "1 needs your reply" : `${needsAttention} need your reply`; + } + if (input.activeCount > 0) { + return input.activeCount === 1 ? "1 working" : `${input.activeCount} working`; + } + if (input.terminalOnly) { + return input.failed ? "Failed" : "All caught up"; + } + return "All caught up"; +} + function terminalAggregateState(state: RelayAgentActivityState): RelayAgentActivityAggregateState { + const activities = [aggregateRowForState(state)]; return sanitizeAgentActivityAggregateState({ title: "T3 Code", - subtitle: state.phase === "failed" ? "Agent work failed" : "Agent work completed", + subtitle: inboxSubtitle({ + activeCount: 0, + activities, + terminalOnly: true, + failed: state.phase === "failed", + }), activeCount: 0, updatedAt: state.updatedAt, - activities: [aggregateRowForState(state)], + activities, }); } @@ -91,12 +118,18 @@ export function makeAggregateState(input: { if (!newest) { return null; } + const activities = recentTerminal.slice(0, MAX_ACTIVITY_ROWS).map(aggregateRowForState); return sanitizeAgentActivityAggregateState({ title: "T3 Code", - subtitle: newest.phase === "failed" ? "Agent work failed" : "Agent work completed", + subtitle: inboxSubtitle({ + activeCount: 0, + activities, + terminalOnly: true, + failed: recentTerminal.some((state) => state.phase === "failed"), + }), activeCount: 0, updatedAt: newest.updatedAt, - activities: recentTerminal.slice(0, MAX_ACTIVITY_ROWS).map(aggregateRowForState), + activities, }); } const recentTerminalStates = input.activeStates @@ -111,12 +144,19 @@ export function makeAggregateState(input: { const updatedAt = [...activeStates, ...recentTerminalStates].reduce((latest, state) => state.updatedAt.localeCompare(latest.updatedAt) > 0 ? state : latest, ).updatedAt; + const activities = displayedStates.map(aggregateRowForState); return sanitizeAgentActivityAggregateState({ title: "T3 Code", - subtitle: "Agent work in progress", + subtitle: inboxSubtitle({ + activeCount: activeStates.length, + // Count attention across every active thread; the card only shows a capped row list. + activities: activeStates.map(aggregateRowForState), + terminalOnly: false, + failed: false, + }), activeCount: activeStates.length, updatedAt, - activities: displayedStates.map(aggregateRowForState), + activities, }); } diff --git a/infra/relay/src/agentActivity/agentActivityAlerts.ts b/infra/relay/src/agentActivity/agentActivityAlerts.ts index c6008950e95e8..204dba5c96e7a 100644 --- a/infra/relay/src/agentActivity/agentActivityAlerts.ts +++ b/infra/relay/src/agentActivity/agentActivityAlerts.ts @@ -6,6 +6,8 @@ import type { import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; +import { agentActivityNotificationCopy } from "./agentActivityPayloads.ts"; + export interface AgentActivityAlert { readonly title: string; readonly body: string; @@ -15,9 +17,27 @@ const TERMINAL_NOTIFICATION_FRESHNESS_MS = 2 * 60 * 1_000; function isFreshTerminalNotification(updatedAt: string, nowMs: number): boolean { const timestamp = Option.getOrNull(DateTime.make(updatedAt)); - return ( - timestamp !== null && nowMs - timestamp.epochMilliseconds <= TERMINAL_NOTIFICATION_FRESHNESS_MS - ); + const epochMs = timestamp?.epochMilliseconds ?? Date.parse(updatedAt); + return Number.isFinite(epochMs) && nowMs - epochMs <= TERMINAL_NOTIFICATION_FRESHNESS_MS; +} + +/** Registration always enables every notifyOn* flag; repair rows that only ring for attention. */ +export function repairAwarenessPreferences( + preferences: RelayAgentAwarenessPreferences, +): RelayAgentAwarenessPreferences { + if ( + !preferences.notificationsEnabled || + !preferences.notifyOnApproval || + !preferences.notifyOnInput || + (preferences.notifyOnCompletion && preferences.notifyOnFailure) + ) { + return preferences; + } + return { + ...preferences, + notifyOnCompletion: true, + notifyOnFailure: true, + }; } type TransitionInput = { @@ -41,6 +61,17 @@ export function rowPhasesChanged( return nextAggregate.activities.some((row) => previousPhases.get(rowKey(row)) !== row.phase); } +export function rowStatusesChanged( + previousAggregate: RelayAgentActivityAggregateState, + nextAggregate: RelayAgentActivityAggregateState, +): boolean { + if (previousAggregate.activities.length !== nextAggregate.activities.length) return true; + const previousStatuses = new Map( + previousAggregate.activities.map((row) => [rowKey(row), row.status]), + ); + return nextAggregate.activities.some((row) => previousStatuses.get(rowKey(row)) !== row.status); +} + function isAttentionPhase(phase: string): boolean { return phase === "waiting_for_approval" || phase === "waiting_for_input"; } @@ -119,10 +150,12 @@ export function alertForActivityRows( const first = rows[0]; if (!first) return null; if (rows.length === 1) { - return { title: first.threadTitle, body: `${first.status}: ${first.projectTitle}` }; + return agentActivityNotificationCopy(first); } return { - title: `${rows.length} agents ${isAttentionPhase(first.phase) ? "need attention" : "finished"}`, + title: isAttentionPhase(first.phase) + ? `${rows.length} need your reply` + : `${rows.length} ready to review`, body: rows.map((row) => row.threadTitle).join(", "), }; } diff --git a/infra/relay/src/agentActivity/agentActivityPayloads.ts b/infra/relay/src/agentActivity/agentActivityPayloads.ts index 4195c81171803..c9c0009998658 100644 --- a/infra/relay/src/agentActivity/agentActivityPayloads.ts +++ b/infra/relay/src/agentActivity/agentActivityPayloads.ts @@ -100,13 +100,34 @@ export function sanitizeApnsNotificationPayload( }; } +export function agentActivityNotificationCopy( + row: Pick, +): { readonly title: string; readonly body: string } { + const thread = row.threadTitle.trim().length > 0 ? row.threadTitle.trim() : "Thread"; + const project = row.projectTitle.trim().length > 0 ? row.projectTitle.trim() : "T3 Code"; + const where = `${thread} · ${project}`; + switch (row.phase) { + case "waiting_for_approval": + return { title: "Needs your reply", body: where }; + case "waiting_for_input": + return { title: "Needs your reply", body: where }; + case "completed": + return { title: "Ready to review", body: where }; + case "failed": + return { title: "Agent failed", body: where }; + default: + return { title: thread, body: `${row.status}: ${project}` }; + } +} + export function notificationForActivity( row: RelayAgentActivityAggregateRow, ): ApnsNotificationPayload { const activity = sanitizeAgentActivityAggregateRow(row); + const copy = agentActivityNotificationCopy(activity); return sanitizeApnsNotificationPayload({ - title: activity.threadTitle, - body: `${activity.status}: ${activity.projectTitle}`, + title: copy.title, + body: copy.body, environmentId: activity.environmentId, threadId: activity.threadId, deepLink: activity.deepLink, diff --git a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts index e73cfd85145a5..1a9b931080dab 100644 --- a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts +++ b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts @@ -5,6 +5,7 @@ import { makeAggregateState } from "./agentActivityAggregate.ts"; import { attentionTransitionRows, terminalTransitionRows, + repairAwarenessPreferences, shouldAlertForActivity, } from "./agentActivityAlerts.ts"; @@ -102,4 +103,21 @@ describe("shared agent activity policy", () => { ).toBe(false); expect(shouldAlertForActivity({ ...input, nowMs: 180_000 })).toBe(false); }); + + it("repairs prefs that can alert on approval but silence completion", () => { + expect( + repairAwarenessPreferences({ + ...preferences, + notifyOnCompletion: false, + notifyOnFailure: false, + }), + ).toMatchObject({ notifyOnCompletion: true, notifyOnFailure: true }); + expect( + repairAwarenessPreferences({ + ...preferences, + notifyOnApproval: false, + notifyOnCompletion: false, + }), + ).toMatchObject({ notifyOnApproval: false, notifyOnCompletion: false }); + }); }); diff --git a/infra/relay/src/agentActivity/fcmPayloads.ts b/infra/relay/src/agentActivity/fcmPayloads.ts index 6c813235e4f61..da39e0bea3a47 100644 --- a/infra/relay/src/agentActivity/fcmPayloads.ts +++ b/infra/relay/src/agentActivity/fcmPayloads.ts @@ -19,10 +19,16 @@ export function androidActivityData(aggregate: RelayAgentActivityAggregateState const hero = rows[0]; const title = activeCount > 0 - ? `${activeCount} active agent${activeCount === 1 ? "" : "s"}${attentionCount ? ` · ${attentionCount} need${attentionCount === 1 ? "s" : ""} attention` : ""}` + ? attentionCount > 0 + ? attentionCount === 1 + ? "1 needs your reply" + : `${attentionCount} need your reply` + : activeCount === 1 + ? "1 working" + : `${activeCount} working` : failed - ? "Agent work failed" - : "Agent work completed"; + ? "Failed" + : "All caught up"; const expiresAt = Math.max( 0, ...rows.map((row) => diff --git a/infra/relay/src/http/Api.test.ts b/infra/relay/src/http/Api.test.ts index 3abb8b28f4d6a..8e564d4371ff8 100644 --- a/infra/relay/src/http/Api.test.ts +++ b/infra/relay/src/http/Api.test.ts @@ -1460,6 +1460,7 @@ describe("relay routing fallback", () => { environmentPublicKey, threadId, state: null, + suppressAlerts: false, })), ); expect(verified.map((input) => input.threadId)).toEqual(threadIds); diff --git a/infra/relay/src/http/Api.ts b/infra/relay/src/http/Api.ts index f06d441c12c48..c841310c00a83 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -1227,6 +1227,10 @@ export const layerServerApi = HttpApiBuilder.group( environmentPublicKey: principal.environmentPublicKey, threadId: params.threadId, state: payload.state, + suppressAlerts: payload.suppressAlerts === true, + ...(payload.suppressAlertUserIds !== undefined + ? { suppressAlertUserIds: payload.suppressAlertUserIds } + : {}), }); }, mapErrorTags({ diff --git a/packages/contracts/src/relay.ts b/packages/contracts/src/relay.ts index 8ebc9df23a0bb..6a872af991bc7 100644 --- a/packages/contracts/src/relay.ts +++ b/packages/contracts/src/relay.ts @@ -274,6 +274,14 @@ export const RelayAgentActivityPublishRequest = Schema.Struct({ proof: TrimmedNonEmptyString.annotate({ description: "Environment-signed JWT covering this published activity state.", }), + suppressAlerts: Schema.optionalKey(Schema.Boolean).annotate({ + description: + "When true, deliver Live Activity / ongoing card updates without audible banners for every delivery user. Prefer suppressAlertUserIds when suppression should be scoped per user.", + }), + suppressAlertUserIds: Schema.optionalKey(Schema.Array(TrimmedNonEmptyString)).annotate({ + description: + "Cloud user ids that should receive Live Activity / ongoing card updates without audible banners. Other linked delivery users still get alerts.", + }), }).annotate({ description: "Publishes a signed agent-awareness update from an environment." }); export type RelayAgentActivityPublishRequest = typeof RelayAgentActivityPublishRequest.Type; diff --git a/packages/shared/src/agentAwareness.test.ts b/packages/shared/src/agentAwareness.test.ts index 71f4f21bd1e2a..483f85efd2efb 100644 --- a/packages/shared/src/agentAwareness.test.ts +++ b/packages/shared/src/agentAwareness.test.ts @@ -9,7 +9,7 @@ import type { import { ProviderInstanceId, RuntimeRequestId } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; -import { projectThreadAwarenessV2 } from "./agentAwareness.ts"; +import { projectThreadAwarenessV2, rollupListedThreadAwarenessPending } from "./agentAwareness.ts"; const NOW = "2026-05-22T12:00:00.000Z"; @@ -23,6 +23,7 @@ describe("projectThreadAwarenessV2", () => { overrides: Partial< Pick< OrchestrationV2ThreadShell, + | "id" | "activityRunStatus" | "latestRunBackgroundWake" | "latestRunStopRequested" @@ -31,6 +32,7 @@ describe("projectThreadAwarenessV2", () => { | "pendingRuntimeRequest" | "lineage" | "settledOverride" + | "updatedAt" > > = {}, ) => ({ @@ -133,7 +135,7 @@ describe("projectThreadAwarenessV2", () => { project, thread: v2Thread({ status: "cancelled", activityRunStatus: "running" }), }), - ).toMatchObject({ phase: "running", headline: "Agent is working" }); + ).toMatchObject({ phase: "running", headline: "Working" }); }); it.each([ @@ -246,7 +248,7 @@ describe("projectThreadAwarenessV2", () => { }; expect(projectThreadAwarenessV2(input)).toMatchObject({ phase: "waiting_for_input", - headline: "Waiting for input", + headline: "Needs your reply", }); }); @@ -263,6 +265,156 @@ describe("projectThreadAwarenessV2", () => { }, }), }), - ).toMatchObject({ phase: "running", headline: "Agent is working" }); + ).toMatchObject({ phase: "running", headline: "Working" }); + }); + + it("surfaces a tracked subagent approval on the listed parent", () => { + expect( + projectThreadAwarenessV2({ + environmentId: "env-1" as EnvironmentId, + project, + thread: v2Thread({ + status: "completed", + pendingBackgroundTasks: [{ taskId: "review", kind: "subagent" }], + }), + pendingSubagentRequest: { + requestId: RuntimeRequestId.make("approve-rm"), + updatedAt, + }, + }), + ).toMatchObject({ phase: "waiting_for_approval", headline: "Needs your reply" }); + }); + + it("does not roll subagent approvals into descendantPendingRequest", () => { + const root = v2Thread({ + id: "root" as ThreadId, + status: "completed", + pendingBackgroundTasks: [{ taskId: "review", kind: "subagent" }], + }); + const child = v2Thread({ + id: "child" as ThreadId, + status: "running", + activityRunStatus: "running", + lineage: { + rootThreadId: root.id, + parentThreadId: root.id, + relationshipToParent: "subagent", + }, + }); + const grandchild = v2Thread({ + id: "grandchild" as ThreadId, + status: "waiting", + activityRunStatus: "waiting", + lineage: { + rootThreadId: root.id, + parentThreadId: child.id, + relationshipToParent: "subagent", + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("approve-rm"), + kind: "command", + createdAt: updatedAt, + }, + }); + expect(rollupListedThreadAwarenessPending([root, child, grandchild]).get(root.id)).toBe( + undefined, + ); + }); + + it("rolls side-chat approvals and input up to the listed parent", () => { + const root = v2Thread({ id: "root" as ThreadId, status: "completed" }); + const approvalChat = v2Thread({ + id: "side-approval" as ThreadId, + status: "waiting", + lineage: { + rootThreadId: root.id, + parentThreadId: root.id, + relationshipToParent: "fork", + sideChat: true, + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("approve-side"), + kind: "command", + createdAt: updatedAt, + }, + }); + const inputChat = v2Thread({ + id: "side-input" as ThreadId, + status: "waiting", + lineage: { + rootThreadId: root.id, + parentThreadId: root.id, + relationshipToParent: "fork", + sideChat: true, + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("ask-side"), + kind: "user_input", + createdAt: updatedAt, + }, + }); + expect(rollupListedThreadAwarenessPending([root, inputChat]).get(root.id)).toMatchObject({ + kind: "user_input", + updatedAt, + }); + expect( + rollupListedThreadAwarenessPending([root, approvalChat, inputChat]).get(root.id), + ).toMatchObject({ kind: "approval", updatedAt }); + }); + + it("advances parent updatedAt when a second side-chat wait arrives", () => { + const root = v2Thread({ + id: "root" as ThreadId, + status: "completed", + updatedAt: DateTime.makeUnsafe("2026-05-22T11:00:00.000Z"), + }); + const firstWait = DateTime.makeUnsafe("2026-05-22T12:00:00.000Z"); + const secondWait = DateTime.makeUnsafe("2026-05-22T13:00:00.000Z"); + const firstChat = v2Thread({ + id: "side-first" as ThreadId, + status: "waiting", + updatedAt: firstWait, + lineage: { + rootThreadId: root.id, + parentThreadId: root.id, + relationshipToParent: "fork", + sideChat: true, + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("ask-first"), + kind: "user_input", + createdAt: firstWait, + }, + }); + const secondChat = v2Thread({ + id: "side-second" as ThreadId, + status: "waiting", + updatedAt: secondWait, + lineage: { + rootThreadId: root.id, + parentThreadId: root.id, + relationshipToParent: "fork", + sideChat: true, + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("ask-second"), + kind: "user_input", + createdAt: secondWait, + }, + }); + const rolled = rollupListedThreadAwarenessPending([root, firstChat, secondChat]).get(root.id); + expect(rolled).toMatchObject({ kind: "user_input", updatedAt: secondWait }); + expect(rolled).toBeDefined(); + expect( + projectThreadAwarenessV2({ + environmentId: "env-1" as EnvironmentId, + project, + thread: root, + descendantPendingRequest: rolled ?? null, + }), + ).toMatchObject({ + phase: "waiting_for_input", + updatedAt: "2026-05-22T13:00:00.000Z", + }); }); }); diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index 55ad807957d64..8627d6aa97a95 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -23,6 +23,14 @@ export type AgentAwarenessPhase = | "failed" | "stale"; +export type AgentAwarenessDescendantPendingRequestKind = "approval" | "user_input"; + +export interface AgentAwarenessDescendantPendingRequest { + readonly kind: AgentAwarenessDescendantPendingRequestKind; + /** The side-chat child's update time; alert ids key on phase + time. */ + readonly updatedAt: DateTime.Utc; +} + export interface SubagentRequestAwareness { readonly requestId: RuntimeRequestId; /** The child's own update time, which moves whenever its request changes. */ @@ -88,6 +96,8 @@ export interface ProjectThreadAwarenessV2Input { >; /** The approval a live subagent below this thread is waiting on. */ readonly pendingSubagentRequest?: SubagentRequestAwareness | null; + /** Approvals/input rolled up from unlisted side chats under this listed thread. */ + readonly descendantPendingRequest?: AgentAwarenessDescendantPendingRequest | null; } export function projectThreadAwarenessV2( @@ -96,7 +106,12 @@ export function projectThreadAwarenessV2( const { environmentId, project, thread } = input; if (!isListedThreadLineage(thread.lineage)) return null; const subagentRequest = input.pendingSubagentRequest ?? null; - const phase = resolveThreadAwarenessPhaseV2(thread, subagentRequest !== null); + const descendantPendingRequest = input.descendantPendingRequest ?? null; + const phase = resolveThreadAwarenessPhaseV2( + thread, + descendantPendingRequest, + subagentRequest !== null, + ); if (phase === null) { return null; } @@ -118,6 +133,9 @@ export function projectThreadAwarenessV2( : phase === "failed" ? "The agent run failed." : undefined; + // A child's request leaves the parent's own timestamp alone. Alerts are + // keyed by phase and time, so a second wait needs the child's time. + const childUpdatedAt = subagentRequest?.updatedAt ?? descendantPendingRequest?.updatedAt ?? null; return { environmentId, threadId: thread.id, @@ -127,19 +145,61 @@ export function projectThreadAwarenessV2( headline: headlineForPhase(phase), ...(detail === undefined ? {} : { detail }), modelTitle: thread.modelSelection.model, - // A child's request leaves the parent's own timestamp alone. Alerts are - // keyed by phase and time, so a second wait needs the child's time. updatedAt: DateTime.formatIso( - subagentRequest === null - ? thread.updatedAt - : DateTime.max(thread.updatedAt, subagentRequest.updatedAt), + childUpdatedAt === null ? thread.updatedAt : DateTime.max(thread.updatedAt, childUpdatedAt), ), deepLink: buildAgentAwarenessDeepLink({ environmentId, threadId: thread.id }), }; } +function descendantRequest( + thread: Pick< + ProjectThreadAwarenessV2Input["thread"], + "pendingRuntimeRequest" | "lineage" | "updatedAt" + >, +): AgentAwarenessDescendantPendingRequest | null { + const kind = thread.pendingRuntimeRequest?.kind; + if (kind === undefined || kind === "auth_refresh") return null; + // Subagent approvals use pendingSubagentRequest / relay tracking. + if (thread.lineage.sideChat !== true) return null; + return { + kind: kind === "user_input" ? "user_input" : "approval", + updatedAt: thread.updatedAt, + }; +} + +/** + * Maps listed parents to the strongest pending request among their unlisted + * side-chat descendants. Subagent approvals stay on pendingSubagentRequest. + */ +export function rollupListedThreadAwarenessPending( + threads: ReadonlyArray< + Pick< + ProjectThreadAwarenessV2Input["thread"], + "id" | "lineage" | "pendingRuntimeRequest" | "updatedAt" + > + >, +): ReadonlyMap { + const requests = new Map(); + for (const child of threads) { + const request = descendantRequest(child); + if (request === null) continue; + const parentId = child.lineage.parentThreadId; + if (parentId === null) continue; + const current = requests.get(parentId); + const stronger = + current === undefined || + (request.kind === "approval" && current.kind !== "approval") || + (request.kind === current.kind && + DateTime.isGreaterThan(request.updatedAt, current.updatedAt)); + if (stronger) requests.set(parentId, request); + } + return requests; +} + function resolveThreadAwarenessPhaseV2( thread: ProjectThreadAwarenessV2Input["thread"], + descendantPendingRequest: AgentAwarenessDescendantPendingRequest | null, childAwaitsApproval: boolean, ): AgentAwarenessPhase | null { if (thread.pendingRuntimeRequest?.kind === "user_input") return "waiting_for_input"; @@ -149,9 +209,10 @@ function resolveThreadAwarenessPhaseV2( ) { return "waiting_for_approval"; } - // The parent can finish its own turn while a delegated child keeps working, - // so a child's request outranks the parent's run status. - if (childAwaitsApproval) return "waiting_for_approval"; + if (descendantPendingRequest?.kind === "user_input") return "waiting_for_input"; + if (descendantPendingRequest?.kind === "approval" || childAwaitsApproval) { + return "waiting_for_approval"; + } switch (thread.activityRunStatus ?? thread.status) { case "preparing": case "starting": @@ -178,18 +239,17 @@ function resolveThreadAwarenessPhaseV2( function headlineForPhase(phase: AgentAwarenessPhase): string { switch (phase) { case "starting": - return "Starting agent"; + return "Starting"; case "running": - return "Agent is working"; + return "Working"; case "waiting_for_approval": - return "Approval needed"; case "waiting_for_input": - return "Waiting for input"; + return "Needs your reply"; case "completed": - return "Agent finished"; + return "Ready to review"; case "failed": - return "Agent failed"; + return "Failed"; case "stale": - return "Update delayed"; + return "Out of date"; } }