diff --git a/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentActivityPresentation.kt b/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentActivityPresentation.kt index 7f4d8c0e691ae..19167516b045a 100644 --- a/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentActivityPresentation.kt +++ b/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentActivityPresentation.kt @@ -34,6 +34,7 @@ internal enum class ActivityPhase( "Answer", R.color.agent_activity_input ), + WOKE("Woke", "Snooze ended", "Woke", "Open", R.color.agent_activity_attention), STALE( "Waiting", "Waiting for an update", @@ -56,7 +57,7 @@ internal enum class ActivityPhase( R.color.agent_activity_failed ); - val needsUser get() = this == APPROVAL || this == INPUT + val needsUser get() = this == APPROVAL || this == INPUT || this == WOKE val finished get() = this == COMPLETED || this == FAILED companion object { @@ -80,6 +81,7 @@ internal fun activityPhase(data: Map, rows: List): "running" -> ActivityPhase.RUNNING "waiting_for_approval" -> ActivityPhase.APPROVAL "waiting_for_input" -> ActivityPhase.INPUT + "woke" -> ActivityPhase.WOKE "stale" -> ActivityPhase.STALE "completed" -> ActivityPhase.COMPLETED "failed" -> ActivityPhase.FAILED @@ -94,9 +96,11 @@ internal class ActivityPresentation(data: Map, private val activ ?: rows.count { ActivityPhase.forStatus(it.status)?.finished != true } private val attentionCount = data["activity_attention_count"]?.toIntOrNull()?.coerceAtLeast(0) ?: rows.count { ActivityPhase.forStatus(it.status)?.needsUser == true } + private val workingCount = data["activity_working_count"]?.toIntOrNull()?.coerceAtLeast(0) + ?: (activeCount - attentionCount).coerceAtLeast(0) private val failedCount = rows.count { it.status == ActivityPhase.FAILED.status } - val threadCount = - activeCount + rows.count { ActivityPhase.forStatus(it.status)?.finished == true } + val threadCount = data["activity_total_count"]?.toIntOrNull()?.coerceAtLeast(0) + ?: (activeCount + rows.count { ActivityPhase.forStatus(it.status)?.finished == true }) private val singleProject = rows.map { it.project }.distinct().size == 1 private val legacyBody = (0..4).mapNotNull { data["activity_line_$it"]?.take(300) } .takeIf { it.isNotEmpty() }?.joinToString("\n") @@ -109,7 +113,7 @@ internal class ActivityPresentation(data: Map, private val activ attentionCount == 1 -> "1 needs you" attentionCount > 1 -> "$attentionCount need you" activeCount > 0 && failedCount > 0 -> "$failedCount failed" - activeCount > 0 -> "$activeCount working" + workingCount > 0 -> "$workingCount working" failedCount > 0 -> "Finished, $failedCount failed" else -> "All finished" } @@ -136,7 +140,7 @@ internal class ActivityPresentation(data: Map, private val activ builder.setSubText( listOfNotNull( hero!!.project.takeIf { singleProject && it.isNotBlank() }, - if (activeCount > 0) "$activeCount active" else "$threadCount threads" + if (workingCount > 0) "$workingCount working" else "$threadCount threads" ).joinToString(" · ") ) } diff --git a/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentNotifications.kt b/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentNotifications.kt index fdfd144ebb9cc..12801a05b9e4a 100644 --- a/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentNotifications.kt +++ b/apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentNotifications.kt @@ -242,10 +242,16 @@ object AgentNotifications { // make an abandoned host look active indefinitely. Older relays omit it. val expiresAt = data["activity_expires_at"]?.toLongOrNull() ?: if (active) updatedAt + RUNNING_LIFETIME_MS else 0L - val remainingMs = (expiresAt - System.currentTimeMillis()).coerceAtMost(MAX_LIFETIME_MS) + // A retained unread result has no timeout; zero on an inactive payload still clears the card. + val retained = active && expiresAt == 0L + val remainingMs = if (retained) { + 0L + } else { + (expiresAt - System.currentTimeMillis()).coerceAtMost(MAX_LIFETIME_MS) + } val wasActive = prefs.getBoolean("lastActive", false) prefs.edit().putBoolean("lastActive", active).apply() - if (remainingMs <= 0 || !prefs.getBoolean("ongoing", false)) { + if ((!retained && remainingMs <= 0) || !prefs.getBoolean("ongoing", false)) { cancelActivity(context) prefs.edit().putBoolean("dismissed", false).apply() return @@ -314,6 +320,9 @@ object AgentNotifications { if (ongoing) builder.addAction(0, "Dismiss", dismissIntent) manager(context).notify(ACTIVITY_TAG, ACTIVITY_ID, builder.build()) if (Build.VERSION.SDK_INT < Build.VERSION_CODES.O) { + context.getSystemService(AlarmManager::class.java).cancel(expiryIntent(context)) + context.getSharedPreferences(STORE, Context.MODE_PRIVATE).edit().remove("expiresAt").apply() + if (remainingMs <= 0) return // Notification timeouts were added in API 26. One inexact alarm also // expires cards on Android 7, including when the app process has exited. // No exact-alarm permission, foreground service or periodic work needed. diff --git a/apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentActivityLifecycleTest.kt b/apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentActivityLifecycleTest.kt new file mode 100644 index 0000000000000..68558c8c1a30a --- /dev/null +++ b/apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentActivityLifecycleTest.kt @@ -0,0 +1,110 @@ +package expo.modules.t3agentnotifications + +import android.app.Activity +import android.app.AlarmManager +import android.app.Application +import android.app.Notification +import android.app.NotificationManager +import android.content.ComponentName +import android.content.Intent +import android.content.IntentFilter +import android.os.Build +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleRegistry +import androidx.lifecycle.ProcessLifecycleOwner +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.RuntimeEnvironment +import org.robolectric.Shadows.shadowOf +import org.robolectric.annotation.Config + +@RunWith(RobolectricTestRunner::class) +@Config(sdk = [24, 26, 33, 36], manifest = Config.NONE) +class AgentActivityLifecycleTest { + private lateinit var context: Application + private lateinit var manager: NotificationManager + private lateinit var lifecycle: LifecycleRegistry + + @Before + fun setUp() { + context = RuntimeEnvironment.getApplication() + manager = context.getSystemService(NotificationManager::class.java) + shadowOf(manager).setNotificationsEnabled(true) + lifecycle = ProcessLifecycleOwner.get().lifecycle as LifecycleRegistry + lifecycle.currentState = Lifecycle.State.CREATED + + val launcher = ComponentName(context, Activity::class.java) + shadowOf(context.packageManager).addActivityIfNotPresent(launcher) + shadowOf(context.packageManager).addIntentFilterForActivity( + launcher, + IntentFilter(Intent.ACTION_MAIN).apply { + addCategory(Intent.CATEGORY_LAUNCHER) + } + ) + AgentNotifications.clear(context) + AgentNotifications.configure(context, "device", "user", "t3code-dev", true) + AgentNotifications.setThreadOnScreen("/threads/environment/thread") + } + + private fun update(alertId: String, active: Boolean) = mapOf( + "device_id" to "device", + "user_id" to "user", + "updated_at" to System.currentTimeMillis().toString(), + "active" to active.toString(), + "activity_title" to "1 active agent", + "activity_body" to "Test thread · Working", + "activity_path" to "/threads/environment/thread", + "alert_id" to alertId, + "alert_group" to "environment/thread", + "alert_title" to "Test thread", + "alert_body" to "Done: Test project", + "alert_path" to "/threads/environment/thread", + ) + + @Test + fun retainedUnreadResultHasNoTimeoutAndReadingItClearsTheCard() { + lifecycle.currentState = Lifecycle.State.RESUMED + AgentNotifications.receive( + context, + update("done", true) + mapOf( + "activity_phase" to "completed", + "activity_expires_at" to "0", + ) + ) + val card = manager.activeNotifications.single().notification + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) assertEquals(0L, card.timeoutAfter) + assertTrue( + shadowOf(context.getSystemService(AlarmManager::class.java)).scheduledAlarms.isEmpty() + ) + AgentNotifications.expire(context, System.currentTimeMillis() + 24 * 60 * 60 * 1000L) + assertEquals(1, manager.activeNotifications.size) + AgentNotifications.receive( + context, + update("done-read", false) + mapOf( + "activity_expires_at" to "0", + ) + ) + assertEquals(0, manager.activeNotifications.size) + } + + @Test + fun waitingThreadsShowTheirCountInsteadOfZeroWorking() { + lifecycle.currentState = Lifecycle.State.RESUMED + val data = update("waiting-only", true) + mapOf( + "activity_line_0" to "Input\tChoose a model\tProject", + "activity_line_1" to "Input\tChoose an icon\tProject", + "activity_phase" to "waiting_for_input", + "activity_active_count" to "2", + "activity_attention_count" to "2", + "activity_working_count" to "0", + "activity_total_count" to "2", + ) + AgentNotifications.receive(context, data) + val card = manager.activeNotifications.single().notification + assertEquals("Project · 2 threads", card.extras.getString(Notification.EXTRA_SUB_TEXT)) + } +} diff --git a/apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentNotificationsTest.kt b/apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentNotificationsTest.kt index 41224273d2ec4..db68708dadce9 100644 --- a/apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentNotificationsTest.kt +++ b/apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentNotificationsTest.kt @@ -316,7 +316,7 @@ class AgentNotificationsTest : AgentNotificationsFixture() { AgentNotifications.receive(context, data) val card = manager.activeNotifications.single().notification assertEquals("1 needs you", card.extras.getCharSequence(Notification.EXTRA_TITLE).toString()) - assertEquals("8 active", card.extras.getString(Notification.EXTRA_SUB_TEXT)) + assertEquals("7 working", card.extras.getString(Notification.EXTRA_SUB_TEXT)) assertEquals( "Approval $title · Project\nWorking Another thread · Other project", card.extras.getCharSequence(Notification.EXTRA_BIG_TEXT).toString() @@ -354,8 +354,12 @@ class AgentNotificationsTest : AgentNotificationsFixture() { fun waitingCardsIgnoreMutableThreadTimestampsEvenAfterRename() { lifecycle.currentState = Lifecycle.State.RESUMED val now = System.currentTimeMillis() - for (phase in listOf("waiting_for_approval", "waiting_for_input")) { - val status = if (phase == "waiting_for_approval") "Approval" else "Input" + for (phase in listOf("waiting_for_approval", "waiting_for_input", "woke")) { + val status = when (phase) { + "waiting_for_approval" -> "Approval" + "woke" -> "Woke" + else -> "Input" + } for ((title, updatedAt) in listOf("Original" to now - 1200000L, "Renamed" to now)) { AgentNotifications.receive( context, @@ -396,6 +400,7 @@ class AgentNotificationsTest : AgentNotificationsFixture() { for (phase in listOf( "waiting_for_approval", "waiting_for_input", + "woke", "stale", "failed", "completed" @@ -463,7 +468,7 @@ class AgentNotificationsTest : AgentNotificationsFixture() { val card = manager.activeNotifications.single().notification assertEquals("2 live", card.extras.getString(NotificationCompat.EXTRA_SHORT_CRITICAL_TEXT)) assertEquals("2 working", card.extras.getCharSequence(Notification.EXTRA_TITLE).toString()) - assertEquals("Project · 2 active", card.extras.getString(Notification.EXTRA_SUB_TEXT)) + assertEquals("Project · 2 working", card.extras.getString(Notification.EXTRA_SUB_TEXT)) assertEquals( "Working Build feature\nDone Write tests", card.extras.getCharSequence(Notification.EXTRA_BIG_TEXT).toString() diff --git a/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts b/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts index 9085f98058066..dce8b6e14c11c 100644 --- a/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts +++ b/apps/mobile/src/features/agent-awareness/remoteRegistration.test.ts @@ -282,6 +282,67 @@ describe("makeRelayDeviceRegistrationRequest", () => { environmentConfigsMock.configs.clear(); }); + it.effect.each([true, false])("primes unread Done without live work: unread=%s", (unread) => { + const aggregate = { + title: "T3 Code", + subtitle: "Agent work completed", + activeCount: 0, + workingCount: 0, + attentionCount: 0, + unreadCount: unread ? 1 : 0, + pendingCount: unread ? 1 : 0, + totalCount: 1, + updatedAt: "2026-10-27T12:00:00.000Z", + activities: [ + { + environmentId: "env", + threadId: "thread", + projectTitle: "Project", + threadTitle: "Completed task", + modelTitle: "Test", + phase: "completed", + unread, + status: "Done", + updatedAt: "2026-10-27T12:00:00.000Z", + completedAt: "2026-10-27T12:00:00.000Z", + deepLink: "/", + }, + ], + }; + vi.mocked(loadPreferences).mockResolvedValue({ liveActivitiesEnabled: true }); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : String(input); + if (url.endsWith("/v1/client/dpop-token")) + return Response.json({ + access_token: "relay-dpop-token", + issued_token_type: "urn:ietf:params:oauth:token-type:access_token", + token_type: "DPoP", + expires_in: 300, + scope: "mobile:registration", + }); + if (url.endsWith("/v1/mobile/agent-activity")) return Response.json({ aggregate }); + return Response.json({ ok: true }); + }); + widgetMocks.start.mockReturnValue({ + getId: () => "activity-1", + getPushToken: () => Promise.resolve(null), + addPushTokenListener: vi.fn(), + }); + Constants.expoConfig!.extra = { relay: { url: "https://relay.example.test" } }; + setAgentAwarenessRelayTokenProvider(() => Promise.resolve("clerk-token"), "user-a"); + return Effect.gen(function* () { + yield* runBackgroundOperations(); + if (unread) { + expect(widgetMocks.start).toHaveBeenCalledWith(aggregate, expect.any(Date)); + } else { + expect(widgetMocks.start).not.toHaveBeenCalled(); + } + }).pipe( + Effect.provideService(FetchHttpClient.Fetch, globalThis.fetch), + Effect.provide(layerRelayTest), + ); + }); + it("disables push features in Personal Team relay registrations", () => { Constants.expoConfig!.extra = { iosPersonalTeamBuild: true }; diff --git a/apps/mobile/src/features/agent-awareness/remoteRegistration.ts b/apps/mobile/src/features/agent-awareness/remoteRegistration.ts index ec712608fad94..7c14f98f30655 100644 --- a/apps/mobile/src/features/agent-awareness/remoteRegistration.ts +++ b/apps/mobile/src/features/agent-awareness/remoteRegistration.ts @@ -1075,20 +1075,13 @@ export const refreshActiveLiveActivityRemoteRegistration = Effect.fn( }).pipe(Effect.orElseSucceed(() => [] as ReadonlyArray>)); if (armedMeanwhile.length > 0) { activities = [...armedMeanwhile]; - } else if (snapshot?.aggregate && snapshot.aggregate.activeCount > 0) { + } else if ( + snapshot?.aggregate && + (snapshot.aggregate.pendingCount ?? snapshot.aggregate.activeCount) > 0 + ) { const aggregate = snapshot.aggregate; const primed = yield* Effect.try({ - try: () => - startAgentLiveActivity( - { - title: aggregate.title, - subtitle: aggregate.subtitle, - activeCount: aggregate.activeCount, - updatedAt: aggregate.updatedAt, - activities: aggregate.activities, - }, - liveActivityStaleDate(), - ), + try: () => startAgentLiveActivity(aggregate, liveActivityStaleDate()), catch: (cause) => new AgentAwarenessOperationError({ operation: "prime-live-activity", diff --git a/apps/mobile/src/features/showcase/showcaseAgentActivity.ts b/apps/mobile/src/features/showcase/showcaseAgentActivity.ts index 784875fd6253b..c936c90d5baa6 100644 --- a/apps/mobile/src/features/showcase/showcaseAgentActivity.ts +++ b/apps/mobile/src/features/showcase/showcaseAgentActivity.ts @@ -26,6 +26,7 @@ const STATUS_BY_PHASE: Record = { waiting_for_approval: "Approval", waiting_for_input: "Input", stale: "Waiting", + woke: "Woke", completed: "Done", failed: "Failed", }; diff --git a/apps/mobile/src/widgets/AgentActivity.test.tsx b/apps/mobile/src/widgets/AgentActivity.test.tsx new file mode 100644 index 0000000000000..f6c81b71d10da --- /dev/null +++ b/apps/mobile/src/widgets/AgentActivity.test.tsx @@ -0,0 +1,85 @@ +import type { LiveActivityComponent } from "expo-widgets"; +import { isValidElement } from "react"; +import { expect, it, vi } from "vite-plus/test"; + +import activity, { type AgentActivityProps, type AgentActivityRowProps } from "./AgentActivity.tsx"; + +vi.mock("expo-widgets", () => ({ + createLiveActivity: (_name: string, render: LiveActivityComponent) => render, +})); +vi.mock("@expo/ui/swift-ui", () => ({ + HStack: "HStack", + Image: "Image", + Text: "Text", + Spacer: "Spacer", + VStack: "VStack", + ZStack: "ZStack", +})); +vi.mock("@expo/ui/swift-ui/modifiers", () => + Object.fromEntries( + [ + "accessibilityElement", + "accessibilityLabel", + "activityBackgroundTint", + "aspectRatio", + "font", + "foregroundStyle", + "frame", + "layoutPriority", + "lineLimit", + "minimumScaleFactor", + "multilineTextAlignment", + "padding", + "resizable", + "widgetURL", + ].map((name) => [name, (value: unknown) => ({ name, value })]), + ), +); + +it("targets the first displayed attention thread when incoming rows are unordered", () => { + const render = activity as unknown as LiveActivityComponent; + const older: AgentActivityRowProps = { + environmentId: "environment", + threadId: "older", + projectTitle: "Project", + threadTitle: "Older request", + modelTitle: "Codex", + phase: "waiting_for_approval", + status: "Approval needed", + updatedAt: "2026-10-28T09:00:00Z", + deepLink: "/threads/environment/older", + }; + const newer = { + ...older, + threadId: "newer", + phase: "woke" as const, + updatedAt: "2026-10-28T10:00:00Z", + deepLink: "/threads/environment/newer", + }; + const layout = render( + { + title: "Agents", + subtitle: "", + activeCount: 2, + updatedAt: newer.updatedAt, + activities: [older, newer], + }, + { colorScheme: "dark" }, + ); + const compact = layout.compactTrailing; + expect(isValidElement(compact)).toBe(true); + if (!isValidElement<{ children: string; modifiers: unknown[] }>(compact)) return; + expect(compact.props.children).toBe("Woke"); + expect(compact.props.modifiers).toContainEqual({ + name: "widgetURL", + value: "t3code://threads/environment/newer", + }); + expect(layout.minimal).toMatchObject({ + props: { + children: { props: { systemName: "alarm.fill" } }, + modifiers: expect.arrayContaining([ + { name: "widgetURL", value: "t3code://threads/environment/newer" }, + ]), + }, + }); +}); diff --git a/apps/mobile/src/widgets/AgentActivity.tsx b/apps/mobile/src/widgets/AgentActivity.tsx index 609beceac0cdf..2481691b1c0e5 100644 --- a/apps/mobile/src/widgets/AgentActivity.tsx +++ b/apps/mobile/src/widgets/AgentActivity.tsx @@ -4,12 +4,14 @@ import { accessibilityElement, accessibilityLabel, activityBackgroundTint, + aspectRatio, font, foregroundStyle, frame, layoutPriority, lineLimit, minimumScaleFactor, + multilineTextAlignment, padding, resizable, widgetURL, @@ -29,7 +31,8 @@ export type AgentActivityPhase = | "waiting_for_input" | "completed" | "failed" - | "stale"; + | "stale" + | "woke"; export interface AgentActivityRowProps { readonly environmentId: string; @@ -38,6 +41,8 @@ export interface AgentActivityRowProps { readonly threadTitle: string; readonly modelTitle: string; readonly phase: AgentActivityPhase; + readonly unread?: boolean; + readonly completedAt?: string; readonly status: string; readonly updatedAt: string; readonly deepLink: string; @@ -47,6 +52,11 @@ export interface AgentActivityProps { readonly title: string; readonly subtitle: string; readonly activeCount: number; + readonly workingCount?: number | null; + readonly attentionCount?: number | null; + readonly totalCount?: number | null; + readonly unreadCount?: number | null; + readonly pendingCount?: number | null; readonly updatedAt: string; readonly activities: ReadonlyArray; } @@ -58,6 +68,49 @@ function AgentActivity( ): LiveActivityLayout { "widget"; + // Placeholder / first-paint entries arrive with empty props. Treat missing + // fields as idle rather than throwing inside the widget JS runtime. + const receivedActivities = Array.isArray(props.activities) ? props.activities : []; + const activeCount = + props.activeCount === null || (typeof props.activeCount === "number" && props.activeCount < 0) + ? null + : typeof props.activeCount === "number" + ? props.activeCount + : 0; + const visibleActiveCount = receivedActivities.filter( + (row) => row.phase !== "completed" && row.phase !== "failed", + ).length; + const countsComplete = activeCount !== null && activeCount === visibleActiveCount; + const workingCount = + typeof props.workingCount === "number" && props.workingCount >= 0 + ? props.workingCount + : countsComplete + ? receivedActivities.filter((row) => row.phase === "running" || row.phase === "starting") + .length + : null; + const attentionCount = + typeof props.attentionCount === "number" && props.attentionCount >= 0 + ? props.attentionCount + : countsComplete + ? receivedActivities.filter( + (row) => + row.phase === "waiting_for_input" || + row.phase === "waiting_for_approval" || + row.phase === "woke", + ).length + : null; + const totalCount = + typeof props.totalCount === "number" && props.totalCount >= receivedActivities.length + ? props.totalCount + : null; + const unreadCount = + typeof props.unreadCount === "number" && props.unreadCount >= 0 + ? props.unreadCount + : totalCount === receivedActivities.length + ? receivedActivities.filter((row) => row.phase === "completed" && row.unread !== false) + .length + : null; + // Hierarchical foregrounds inherit system tint and vibrancy. type Foreground = Parameters[0]; const primaryForeground = { type: "hierarchical", style: "primary" } as const; @@ -76,77 +129,121 @@ function AgentActivity( } switch (phase) { case "waiting_for_approval": - return isLightScheme ? "#d97706" : "#fcd34d"; case "waiting_for_input": - return isLightScheme ? "#4f46e5" : "#a5b4fc"; + case "woke": + return isLightScheme ? "#d97706" : "#fcd34d"; // amber-600 / amber-300 case "failed": - return isLightScheme ? "#dc2626" : "#fca5a5"; + return isLightScheme ? "#dc2626" : "#fca5a5"; // red-600 / red-300 case "completed": - return isLightScheme ? "#059669" : "#6ee7b7"; + return isLightScheme ? "#059669" : "#6ee7b7"; // emerald-600 / emerald-300 case "stale": return secondaryForeground; case "starting": case "running": default: - return isLightScheme ? "#0284c7" : "#7dd3fc"; + return isLightScheme ? "#0284c7" : "#7dd3fc"; // sky-600 / sky-300 } }; - const phasePriority = (phase: AgentActivityPhase): number => { - if (phase === "waiting_for_approval" || phase === "waiting_for_input") return 0; - if (phase === "failed") return 1; - if (phase === "running" || phase === "starting") return 2; - return 3; + const phasePriority = (row: AgentActivityRowProps): number => { + if ( + row.phase === "waiting_for_approval" || + row.phase === "waiting_for_input" || + row.phase === "woke" + ) + return 0; + if (row.phase === "failed") return 1; + if (row.phase === "completed" && row.unread !== false) return 2; + if (row.phase === "running" || row.phase === "starting") return 3; + return 4; }; - // Mark in-flight rows stale after expiry so they cannot claim agents still work. - const activities: ReadonlyArray = environment.isStale - ? props.activities.map((row) => - row.phase === "completed" || row.phase === "failed" + // Past a Live Activity's stale date the system stops vouching for the + // content, so every in-flight row degrades to "stale" rather than claiming + // an agent is still working. Terminal phases keep their own state. + const isSystemStale = "isStale" in environment && environment.isStale === true; + const activities: ReadonlyArray = isSystemStale + ? receivedActivities.map((row) => + row.phase === "completed" || row.phase === "failed" || row.phase === "woke" ? row : { ...row, phase: "stale", status: "Out of date" }, ) - : props.activities; - const ordered = [...activities].sort((a, b) => phasePriority(a.phase) - phasePriority(b.phase)); + : receivedActivities; + const ordered = [...activities].sort( + (a, b) => + phasePriority(a) - phasePriority(b) || + b.updatedAt.localeCompare(a.updatedAt) || + a.environmentId.localeCompare(b.environmentId) || + a.threadId.localeCompare(b.threadId), + ); const row0 = ordered[0]; const row1 = ordered[1]; const row2 = ordered[2]; const row3 = ordered[3]; const row4 = ordered[4]; - const attentionRows = activities.filter( - (row) => row.phase === "waiting_for_approval" || row.phase === "waiting_for_input", + const attentionRow = ordered.find( + (row) => + row.phase === "waiting_for_approval" || + row.phase === "waiting_for_input" || + row.phase === "woke", ); - const attentionRow = attentionRows[0]; const failedRow = activities.find((row) => row.phase === "failed"); const heroRow = attentionRow ?? failedRow ?? row0; - const tint = phaseTint(heroRow?.phase); + const tint = + heroRow?.phase === "completed" && heroRow.unread === false + ? secondaryForeground + : phaseTint(heroRow?.phase); + // Headline count leans on the accent when a human is actually blocked. const headerTint = attentionRow ? phaseTint(attentionRow.phase) : failedRow ? phaseTint(failedRow.phase) : tint; - const allDone = props.activeCount === 0; - const doneLabel = failedRow ? "Failed" : "Done"; - const outcomeLabel = failedRow ? "Agent work failed" : "Agent work completed"; + // Derive terminal outcomes from all rows so a failure dominates a newer success in every presentation. + const allDone = + typeof props.pendingCount === "number" && props.pendingCount >= 0 + ? props.pendingCount === 0 + : activeCount === 0 && (unreadCount === 0 || unreadCount === null); + const hasRows = activities.length > 0; + const doneLabel = failedRow ? "Failed" : hasRows ? "Done" : "Idle"; + const outcomeLabel = failedRow + ? "Agent work failed" + : hasRows + ? "All caught up" + : "No active agents"; - const agentWord = props.activeCount === 1 ? "agent" : "agents"; - const agentsLabel = allDone + const countLabel = allDone ? outcomeLabel - : environment.isStale + : isSystemStale ? "Agent status out of date" - : `${props.activeCount} active ${agentWord}`; + : workingCount === null + ? "Working count unavailable" + : workingCount > 0 + ? `${workingCount} working` + : failedRow + ? "Agent work failed" + : ""; + const agentsLabel = countLabel; const attentionSuffix = - attentionRows.length > 0 - ? `${attentionRows.length} need${attentionRows.length === 1 ? "s" : ""} attention` + !isSystemStale && attentionCount !== null && attentionCount > 0 + ? `${attentionCount} need${attentionCount === 1 ? "s" : ""} you` : ""; - const activeLabel = allDone + const doneSuffix = unreadCount !== null && unreadCount > 0 ? `${unreadCount} done` : ""; + const currentLabel = allDone ? doneLabel - : environment.isStale + : isSystemStale ? "Out of date" - : `${props.activeCount} active`; - const showCompactGlyph = !attentionRow && !failedRow && !allDone && !environment.isStale; - const summary = attentionSuffix || activeLabel; + : workingCount === null + ? "Count unavailable" + : workingCount > 0 + ? `${workingCount} working` + : failedRow + ? "Failed" + : ""; + const activeLabel = currentLabel; + const showCompactGlyph = !attentionRow && !failedRow && !doneSuffix && !allDone && !isSystemStale; + const summary = attentionSuffix || doneSuffix || activeLabel; // Widget taps target the containing app, so its production scheme is safe in every build. const deepLinkRow = attentionRow ?? row0; @@ -154,8 +251,11 @@ function AgentActivity( deepLinkRow && deepLinkRow.deepLink.startsWith("/") && !deepLinkRow.deepLink.startsWith("//") ? `t3code://${deepLinkRow.deepLink.slice(1)}` : null; + const linkModifiers = deepLink ? [widgetURL(deepLink)] : []; + // A scannable status glyph per phase — reads faster than colored words and + // ties the compact / expanded / banner / watch presentations together. type SFName = NonNullable["systemName"]>; const phaseSymbol = (phase: AgentActivityPhase): SFName => { switch (phase) { @@ -163,6 +263,8 @@ function AgentActivity( return "exclamationmark.circle.fill"; case "waiting_for_input": return "questionmark.circle.fill"; + case "woke": + return "alarm.fill"; case "failed": return "xmark.octagon.fill"; case "completed": @@ -179,7 +281,8 @@ function AgentActivity( }; // SF Symbols, like the logo, ignore frame/foregroundStyle applied directly to - // the image; size + tint them through a container the resizable symbol fills. + // the image; size + tint them through a container. Preserve the symbol's + // intrinsic aspect ratio when the resizable image fills that frame. const renderGlyph = ( systemName: SFName, size: number, @@ -189,16 +292,56 @@ function AgentActivity( - + ); + const isRead = (row: AgentActivityRowProps) => row.phase === "completed" && row.unread === false; + const isEmphasized = (row: AgentActivityRowProps) => + row.phase === "woke" || + row.phase === "waiting_for_input" || + row.phase === "waiting_for_approval" || + (row.phase === "completed" && row.unread !== false); + const renderRowStatus = (row: AgentActivityRowProps) => + isRead(row) ? ( + + ) : ( + + {row.status} + + ); + + // Single-line rows fit the expanded island height budget; protect project and status from truncation. const renderCompactRow = (row: AgentActivityRowProps) => ( + {row.phase === "completed" && row.unread !== false ? ( + ● + ) : null} @@ -209,22 +352,11 @@ function AgentActivity( {row.projectTitle} - - {row.status} - + {renderRowStatus(row)} ); - // Image views only honor `resizable` directly (frame/foregroundStyle are - // dropped), so we size it via a container frame the resizable image fills and - // tint it through the container's foreground style, which the template image - // inherits. + // Widget Image ignores frame/foregroundStyle; size and tint its resizable template through a 3:2 container. const renderLogo = (height: number, color: Foreground, modifiers: typeof linkModifiers = []) => ( ); - return { - banner: ( - - - - {renderLogo(13, primaryForeground)} - - - - - - {agentsLabel} - - {attentionSuffix ? ( - · - ) : null} - {attentionSuffix ? ( - - {attentionSuffix} - - ) : null} - - - - {row0 ? renderCompactRow(row0) : null} - {row1 ? renderCompactRow(row1) : null} - {row2 ? renderCompactRow(row2) : null} - {row3 ? renderCompactRow(row3) : null} - {row4 ? renderCompactRow(row4) : null} - - ), - bannerSmall: ( - + // Compact card for the watchOS Smart Stack + CarPlay (the `.small` family) + // and lock-screen accessory widgets. + const renderCompactLayout = () => ( + + + {renderLogo(14, primaryForeground)} + + {activeCount === null ? activeLabel : summary} + + + + {row0 ? ( - {renderLogo(14, primaryForeground)} - {attentionRows.length > 0 ? summary : activeLabel} + {row0.threadTitle} + {renderRowStatus(row0)} - {row0 ? ( - + ) : null} + + ); + + const banner = ( + + {/* Keep the status centered independently of the logo. */} + + + {renderLogo(13, primaryForeground)} + + + + + + {agentsLabel} + + {attentionSuffix && agentsLabel ? ( + · + ) : null} + {attentionSuffix ? ( - {row0.threadTitle} + {attentionSuffix} - - - {row0.status} + ) : null} + {doneSuffix ? ( + + {`${agentsLabel || attentionSuffix ? "· " : ""}${doneSuffix}`} - - ) : null} - - ), + ) : null} + + + + {row0 ? renderCompactRow(row0) : null} + {row1 ? renderCompactRow(row1) : null} + {row2 ? renderCompactRow(row2) : null} + {row3 ? renderCompactRow(row3) : null} + {totalCount !== null && totalCount > 5 ? ( + + {`${totalCount - 4} more threads · Open T3`} + + ) : row4 ? ( + renderCompactRow(row4) + ) : null} + + ); + + return { + banner, + bannerSmall: renderCompactLayout(), compactLeading: renderLogo(14, tint, linkModifiers), // While agents are only working, stay quiet: the thread list's Working // glyph marks the activity without a count. Labels appear only for @@ -347,10 +511,12 @@ function AgentActivity( {attentionRow ? attentionRow.phase === "waiting_for_approval" ? "Approval" - : "Input" + : attentionRow.phase === "woke" + ? "Woke" + : "Input" : failedRow ? "Failed" - : activeLabel} + : doneSuffix || activeLabel} ), minimal: heroRow @@ -360,13 +526,23 @@ function AgentActivity( {renderLogo(15, tint)} - {allDone ? doneLabel : `${props.activeCount}`} + {isSystemStale ? "Out of date" : allDone ? doneLabel : summary} ), + // No center content: the phase glyphs + statuses in expandedBottom already + // carry the attention signal, and the expanded island's height budget is + // tight enough that a summary line there pushed the third row off. expandedCenter: null, + // No trailing content: a timestamp is glanceable-lock-screen info, not + // useful in a view the user is actively holding open — and the trailing + // region hugs the island's corner radius, which clipped it anyway. expandedTrailing: null, expandedBottom: ( + // Vertical padding only: the expanded region provides its own horizontal + // content margins, so `all` padding double-indented the rows. + // Horizontal padding keeps both edges clear of the island's corner + // curvature (right edge clipped status labels; titles hugged the left). = projectId: state.thread.projectId, creationSource: state.thread.creationSource, archivedAt: state.thread.archivedAt, + snoozedUntil: state.thread.snoozedUntil ?? null, + snoozedAt: state.thread.snoozedAt ?? null, + lastVisitedAt: state.thread.lastVisitedAt, activeRunId: state.activeRunId, latestRunId: state.latestRunId, hasActionableProposedPlan: state.hasActionableProposedPlan, diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 85b9b44f4daf7..7b4003572b86a 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -119,12 +119,49 @@ function waitingChild( }); } -const PublishPayload = Schema.Struct({ state: Schema.NullOr(RelayAgentActivityState) }); +const PublishPayload = Schema.Struct({ + state: Schema.NullOr(RelayAgentActivityState), + replay: Schema.optional(Schema.Boolean), +}); const decodePublishPayload = Schema.decodeUnknownSync(Schema.fromJsonString(PublishPayload)); const unused = () => Effect.die("Unexpected test dependency call"); describe("startup agent activity", () => { - it("publishes active work and only terminal runs completed after startup", () => { + it.effect("backfills pre-start completions silently", () => + Effect.gen(function* () { + const { relay, publications } = yield* makeTestRelay({ + readShell: () => + Effect.succeed( + shell({ + status: "completed", + latestRunCompletedAt: DateTime.makeUnsafe("1969-12-31T00:00:00.000Z"), + }), + ), + }); + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.length, 1); + assert.equal(publications[0]?.replay, true); + }), + ); + it.each([30_000, 120_000])( + "uses the current time for snoozes in a delayed backfill: elapsed=%s", + (elapsed) => { + const startedAt = DateTime.toEpochMillis(DateTime.makeUnsafe(NOW)); + const input = { + environmentId: EnvironmentId.make("relay-env"), + startedAt, + now: startedAt + elapsed, + projects: [{ id: PROJECT_ID, title: "Project" }], + threads: [shell({ snoozedUntil: DateTime.makeUnsafe(startedAt + 60_000) })], + }; + assert.deepEqual( + AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds(input), + elapsed < 60_000 ? [] : [THREAD_ID], + ); + }, + ); + + it("publishes active work and every result inside the retention window", () => { const startedAt = DateTime.toEpochMillis(DateTime.makeUnsafe(NOW)); const oldCompleted = ThreadId.make("old-completed"); const newCompleted = ThreadId.make("new-completed"); @@ -133,6 +170,7 @@ describe("startup agent activity", () => { const ids = AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds({ environmentId: EnvironmentId.make("relay-env"), startedAt, + now: startedAt + 60_000, projects: [{ id: PROJECT_ID, title: "Project" }], threads: [ shell(), @@ -156,11 +194,37 @@ describe("startup agent activity", () => { status: "failed", latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T12:00:01.000Z"), }), + shell({ + id: ThreadId.make("expired-failed"), + status: "failed", + latestRunCompletedAt: DateTime.makeUnsafe("2026-09-03T11:59:00.000Z"), + }), + shell({ + id: ThreadId.make("unread-completed"), + status: "completed", + latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T11:59:00.000Z"), + lastVisitedAt: DateTime.makeUnsafe("2026-09-04T11:00:00.000Z"), + }), + shell({ + id: ThreadId.make("settled-completed"), + status: "completed", + settledOverride: "settled", + latestRunCompletedAt: DateTime.makeUnsafe("2026-09-04T11:59:00.000Z"), + }), shell({ id: ThreadId.make("idle"), status: "idle" }), shell({ id: ThreadId.make("missing-project"), projectId: ProjectId.make("missing") }), ], }); - assert.deepStrictEqual(ids, [THREAD_ID, newCompleted, newFailed]); + // Read and settled results still replace a row the relay may hold from before. + assert.deepStrictEqual(ids, [ + THREAD_ID, + oldCompleted, + newCompleted, + oldFailed, + newFailed, + ThreadId.make("unread-completed"), + ThreadId.make("settled-completed"), + ]); }); }); @@ -270,6 +334,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( const publications: Array<{ readonly url: string; readonly authorization: string | null; + readonly replay?: boolean; readonly state: RelayAgentActivityState | null; }> = []; const fetch: typeof globalThis.fetch = Object.assign( @@ -288,6 +353,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( url: String(input), authorization: new Headers(init?.headers).get("authorization"), state: payload.state, + replay: payload.replay ?? false, }); return Promise.resolve( options.respond?.(publications.length) ?? Response.json({ ok: true, deliveries: [] }), @@ -382,7 +448,7 @@ describe("AgentAwarenessRelay", () => { const parent = shell({ status: "completed" }); const child = waitingChild(); const { relay, secrets, publications } = yield* makeTestRelay({ - catchUpThreads: [parent, child], + catchUpThreads: [child, parent], readShell: (id) => Effect.succeed(id === child.id ? child : parent), }); yield* relay.start(); @@ -486,6 +552,64 @@ describe("AgentAwarenessRelay", () => { }), ); + it.effect("publishes secure-input transitions without sending the secret prompt", () => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const pulls = yield* Queue.unbounded(); + const { relay, currentShell, publications } = yield* makeTestRelay({ + domainEvents: Stream.fromEffectRepeat( + Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))), + ), + }); + yield* relay.start(); + yield* Queue.take(pulls); + yield* relay.publishThread(THREAD_ID); + const now = yield* DateTime.now; + const request = { + id: RuntimeRequestId.make("secret-input"), + kind: "user_input" as const, + createdAt: now, + blocking: true, + }; + for (const waiting of [true, false]) { + yield* Ref.set(currentShell, shell({ pendingRuntimeRequest: waiting ? request : null })); + yield* Queue.offer(events, { + id: EventId.make(`secret-event-${waiting}`), + type: "turn-item.updated", + threadId: THREAD_ID, + runId: RunId.make("secret-run"), + occurredAt: now, + payload: { + id: TurnItemId.make("secret-input"), + type: "secret_request", + threadId: THREAD_ID, + runId: RunId.make("secret-run"), + nodeId: NodeId.make("secret-node"), + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + title: null, + startedAt: now, + completedAt: waiting ? null : now, + updatedAt: now, + status: waiting ? "waiting" : "completed", + label: "PRIVATE_LABEL", + reason: "PRIVATE_REASON", + secretStatus: waiting ? "pending" : "saved", + }, + }); + yield* Queue.take(pulls); + yield* relay.drain; + assert.equal(publications.at(-1)?.state?.phase, waiting ? "waiting_for_input" : "running"); + } + assert.equal(publications.length, 3); + assert.isFalse(JSON.stringify(publications).includes("PRIVATE_LABEL")); + assert.isFalse(JSON.stringify(publications).includes("PRIVATE_REASON")); + }).pipe(Effect.scoped), + ); + it.effect("deduplicates state and republishes title changes", () => Effect.gen(function* () { const { relay, currentShell, publications } = yield* makeTestRelay(); @@ -568,9 +692,165 @@ describe("AgentAwarenessRelay", () => { yield* relay.drain; assert.equal(publications.length, 3); assert.equal(publications[2]?.state, null); + yield* Ref.set(currentShell, terminal); + yield* Queue.offer(events, { + id: EventId.make("event:pinned"), + type: "thread.pinned", + threadId: THREAD_ID, + occurredAt: now, + payload: { ...terminal, lastVisitedAt: null, settledAt: null }, + }); + yield* Queue.take(pulls); + yield* relay.drain; + assert.equal(publications.at(-1)?.state?.phase, status); }).pipe(Effect.scoped), ); + it.effect( + "publishes reading and marking unread immediately without another completion alert", + () => + Effect.gen(function* () { + const { relay, currentShell, publications } = yield* makeTestRelay(); + yield* relay.publishThread(THREAD_ID); + const completedAt = DateTime.makeUnsafe("1970-01-01T00:00:01.000Z"); + const completed = shell({ + status: "completed", + latestRunCompletedAt: completedAt, + lastVisitedAt: DateTime.makeUnsafe("1970-01-01T00:00:00.000Z"), + }); + yield* Ref.set(currentShell, completed); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.state?.unread, true); + yield* Ref.set(currentShell, { ...completed, lastVisitedAt: completedAt }); + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.length, 3); + assert.equal(publications.at(-1)?.state?.unread, false); + assert.equal(publications.at(-1)?.replay, true); + yield* Ref.set(currentShell, completed); + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.length, 4); + assert.equal(publications.at(-1)?.state?.unread, true); + assert.equal(publications.at(-1)?.replay, true); + }), + ); + + it.effect("restores a snoozed result after startup backfill without an unsnooze event", () => + Effect.gen(function* () { + const snoozed = shell({ status: "completed", snoozedUntil: DateTime.makeUnsafe(60_000) }); + const { relay, currentShell, publications } = yield* makeTestRelay({ + catchUpThreads: [snoozed], + }); + yield* Ref.set(currentShell, snoozed); + yield* relay.start(); + yield* TestClock.adjust("1 second"); + yield* relay.drain; + assert.equal(publications.length, 0); + yield* TestClock.adjust("59 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.state?.phase, "woke"); + assert.equal(publications.at(-1)?.replay, false); + }).pipe(Effect.scoped), + ); + + it.effect("reschedules a changed snooze deadline and ignores the old deadline", () => + Effect.gen(function* () { + const { relay, currentShell, publications } = yield* makeTestRelay(); + const completed = shell({ status: "completed", snoozedUntil: DateTime.makeUnsafe(60_000) }); + yield* Ref.set(currentShell, completed); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + yield* Ref.set(currentShell, { ...completed, snoozedUntil: DateTime.makeUnsafe(120_000) }); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + const before = publications.length; + yield* TestClock.adjust("50 seconds"); + yield* relay.drain; + assert.equal(publications.length, before); + yield* TestClock.adjust("60 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.state?.phase, "woke"); + }), + ); + + it.effect("restores a snoozed result at expiry without another domain event or alert", () => + Effect.gen(function* () { + const { relay, currentShell, publications } = yield* makeTestRelay(); + const completed = shell({ status: "completed" }); + yield* Ref.set(currentShell, completed); + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.at(-1)?.state?.phase, "completed"); + yield* Ref.set(currentShell, { + ...completed, + snoozedUntil: DateTime.makeUnsafe(60_000), + }); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.state, null); + yield* TestClock.adjust("55 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.state?.phase, "woke"); + assert.equal(publications.at(-1)?.replay, false); + }), + ); + + it.effect("does not mistake two untimestamped completions for the same result", () => + Effect.gen(function* () { + const { relay, currentShell, publications } = yield* makeTestRelay(); + for (let run = 0; run < 2; run++) { + yield* Ref.set(currentShell, shell()); + yield* relay.publishThread(THREAD_ID); + yield* Ref.set(currentShell, shell({ status: "completed", latestRunCompletedAt: null })); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + } + const completions = publications.filter( + (publication) => publication.state?.phase === "completed", + ); + assert.equal(completions.length, 2); + assert.isTrue(completions.every((publication) => publication.replay === false)); + }), + ); + + it.effect("does not alert again when a completed thread is settled then reopened", () => + Effect.gen(function* () { + const { relay, currentShell, publications } = yield* makeTestRelay(); + yield* relay.publishThread(THREAD_ID); + const completedAt = DateTime.makeUnsafe("1970-01-01T00:00:01.000Z"); + const completed = shell({ + status: "completed", + latestRunCompletedAt: completedAt, + lastVisitedAt: DateTime.makeUnsafe("1970-01-01T00:00:00.000Z"), + }); + yield* Ref.set(currentShell, completed); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.replay, false); + yield* Ref.set(currentShell, { ...completed, settledOverride: "settled" }); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.state, null); + yield* Ref.set(currentShell, completed); + yield* relay.publishThread(THREAD_ID); + yield* relay.drain; + assert.equal(publications.at(-1)?.state?.phase, "completed"); + assert.equal(publications.at(-1)?.replay, true); + const nextCompletedAt = yield* DateTime.now; + yield* Ref.set(currentShell, shell({ ...completed, latestRunCompletedAt: nextCompletedAt })); + yield* relay.publishThread(THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications.at(-1)?.replay, false); + }), + ); + it.effect("holds a completion through a continuation gap, then publishes it once", () => Effect.gen(function* () { const { relay, currentShell, publications } = yield* makeTestRelay(); @@ -663,7 +943,7 @@ describe("AgentAwarenessRelay", () => { }), ); - it.effect("does not alert an old completion after publishing its thread's tombstone", () => + it.effect("reopens an old completion silently after publishing its thread's tombstone", () => Effect.gen(function* () { const oldCompleted = shell({ status: "completed", @@ -690,8 +970,28 @@ describe("AgentAwarenessRelay", () => { yield* TestClock.adjust("5 seconds"); yield* relay.drain; assert.deepEqual( - publications.map((publication) => publication.state), - [null], + publications.map((publication) => [publication.state?.phase ?? null, publication.replay]), + [ + [null, false], + ["completed", true], + ], + ); + }), + ); + + it.effect("drops a held completion when a queued continuation hides the thread", () => + Effect.gen(function* () { + const { relay, currentShell, publications } = yield* makeTestRelay(); + yield* relay.publishThread(THREAD_ID); + yield* Ref.set(currentShell, shell({ status: "completed" })); + 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( + publications.map((publication) => publication.state?.phase ?? null), + "completed", ); }), ); @@ -735,23 +1035,6 @@ describe("AgentAwarenessRelay", () => { }), ); - it.effect("drops a held completion when a queued continuation hides the thread", () => - Effect.gen(function* () { - const { relay, currentShell, publications } = yield* makeTestRelay(); - yield* relay.publishThread(THREAD_ID); - yield* Ref.set(currentShell, shell({ status: "completed" })); - 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( - publications.map((publication) => publication.state?.phase ?? null), - "completed", - ); - }), - ); - it.effect("retries a failed completion even after the thread settles", () => Effect.gen(function* () { const { relay, currentShell, publications } = yield* makeTestRelay({ diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index c14558762183a..f724216968121 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -96,6 +96,12 @@ function shouldPublishAgentAwarenessEvent( case "thread.pull-request-link-updated": case "thread.pull-request-synced": case "thread.settled": + case "thread.unsettled": + case "thread.snoozed": + case "thread.pinned": + case "thread.unsnoozed": + case "thread.visited": + case "thread.marked-unread": case "thread.model-selection-updated": case "thread.provider-switched": case "run.created": @@ -104,19 +110,12 @@ function shouldPublishAgentAwarenessEvent( case "subagent.updated": case "provider-thread.updated": return true; - // Unsettling must not re-alert a dismissed result. - case "thread.unsettled": case "thread.guest-share-updated": - case "thread.snoozed": - case "thread.unsnoozed": case "thread.auto-settle-set": - case "thread.pinned": case "thread.unpinned": case "thread.pin-reordered": case "thread.active-reordered": case "thread.tags-updated": - case "thread.visited": - case "thread.marked-unread": case "thread.runtime-mode-updated": case "thread.interaction-mode-updated": case "run.background-work-cancelled": @@ -128,7 +127,11 @@ function shouldPublishAgentAwarenessEvent( case "provider-session.detached": case "provider-turn.updated": case "turn-item.updated": - return isTurnItemPayload(event.payload) && turnItemUpdateCanEndBackgroundWork(event.payload); + return ( + isTurnItemPayload(event.payload) && + (event.payload.type === "secret_request" || + turnItemUpdateCanEndBackgroundWork(event.payload)) + ); case "message.updated": case "plan.updated": case "checkpoint-scope.created": @@ -261,6 +264,7 @@ const makePublishProof = Effect.fn("makePublishProof")(function* (input: { readonly threadId: ThreadId; readonly state: RelayAgentActivityState | null; readonly jti: string; + readonly replay?: boolean; }) { const now = yield* DateTime.now; const expiresAt = DateTime.add(now, { minutes: 5 }); @@ -274,6 +278,7 @@ const makePublishProof = Effect.fn("makePublishProof")(function* (input: { environmentId: input.environmentId as RelayAgentActivityPublishProofPayload["environmentId"], threadId: input.threadId, state: input.state, + ...(input.replay ? { replay: true } : {}), } satisfies RelayAgentActivityPublishProofPayload; const signingKey = yield* EnvironmentSigningKey; return yield* signingKey.sign({ typ: RELAY_ACTIVITY_PUBLISH_TYP, payload }); @@ -316,6 +321,7 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: { readonly environmentId: EnvironmentId; readonly thread: Option.Option; readonly project: Option.Option; + readonly now: number; readonly pendingSubagentRequest: SubagentRequestAwareness | null; }): { readonly projectId: string | null; @@ -344,6 +350,7 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: { environmentId: input.environmentId, project: input.project.value, thread: input.thread.value, + now: input.now, }; const state = projectThreadAwarenessV2({ ...awarenessInput, @@ -367,9 +374,13 @@ function terminalWorkSinceStart(thread: AgentAwarenessThread, startedAt: number) ); } +// Matches the relay's retention of results; older rows have already expired there. +const RETAINED_RESULT_REPLAY_WINDOW_MS = 24 * 60 * 60 * 1_000; + export function resolveAgentAwarenessRelayActiveThreadIds(input: { readonly environmentId: EnvironmentId; readonly startedAt: number; + readonly now: number; readonly projects: ReadonlyArray>; readonly threads: ReadonlyArray; }): ReadonlyArray { @@ -386,12 +397,16 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: { environmentId: input.environmentId, project, thread, + now: input.now, }); + if (state !== null && state.phase !== "completed" && state.phase !== "failed") return true; + // The relay may still hold a result from before a restart or while publishing was off: re-send + // it so read results update and settled or hidden ones (null, a tombstone) clear the row. return ( - state !== null && - (state.phase !== "completed" && state.phase !== "failed" - ? true - : terminalWorkSinceStart(thread, input.startedAt)) + thread.latestRunCompletedAt != null && + (terminalWorkSinceStart(thread, input.startedAt) || + input.now - DateTime.toEpochMillis(thread.latestRunCompletedAt) <= + RETAINED_RESULT_REPLAY_WINDOW_MS) ); }) .map((thread) => thread.id); @@ -409,6 +424,11 @@ export const make = Effect.gen(function* () { const startedAt = (yield* DateTime.now).epochMilliseconds; const catchUpRequests = yield* Queue.dropping(1); const publishedStateByThreadRef = yield* Ref.make(new Map()); + // Tombstones remove published rows; reopening must still recognize an already announced result. + const publishedCompletions = new Map< + ThreadId, + Pick + >(); // 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(); @@ -477,6 +497,33 @@ export const make = Effect.gen(function* () { yield* Ref.set(publishedStateByThreadRef, new Map()); }); + const snoozeWakeups = new Map }>(); + let enqueueSnoozeWakeup: (threadId: ThreadId) => Effect.Effect = () => Effect.void; + const scheduleSnoozeWakeup = Effect.fn("AgentAwarenessRelay.scheduleSnoozeWakeup")(function* ( + thread: AgentAwarenessThread, + nowMs: number, + ) { + const deadline = thread.snoozedUntil?.epochMilliseconds; + const previous = snoozeWakeups.get(thread.id); + const pending = deadline !== undefined && deadline > nowMs && thread.archivedAt === null; + if (pending && previous?.deadline === deadline) return; + snoozeWakeups.delete(thread.id); + if (previous) yield* Fiber.interrupt(previous.timer); + if (!pending) return; + // Expiry changes the projection without a domain event; startup must restore these timers too. + const timer = yield* Effect.sleep(deadline - nowMs).pipe( + Effect.andThen( + Effect.suspend(() => { + if (snoozeWakeups.get(thread.id)?.deadline !== deadline) return Effect.void; + snoozeWakeups.delete(thread.id); + return enqueueSnoozeWakeup(thread.id); + }), + ), + Effect.forkIn(scope), + ); + snoozeWakeups.set(thread.id, { deadline, timer }); + }); + const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* (threadId: ThreadId) { const publishAgentActivity = yield* readPublishAgentActivityEnabled; if (!publishAgentActivity) { @@ -510,6 +557,7 @@ export const make = Effect.gen(function* () { readonly projectId: string | null; readonly state: RelayAgentActivityState | null; readonly reason: string; + readonly replay?: boolean; }) { if (cachedRelayClient === undefined) { cachedRelayClient = yield* makeRelayClient(relayConfig); @@ -520,6 +568,7 @@ export const make = Effect.gen(function* () { environmentId, threadId, state: input.state, + replay: input.replay ?? false, jti: yield* random.uuidV4(), }).pipe(Effect.provideService(EnvironmentSigningKey, signingKey)); @@ -540,6 +589,7 @@ export const make = Effect.gen(function* () { }, payload: { state: input.state, + ...(input.replay ? { replay: true } : {}), proof, }, }) @@ -556,6 +606,9 @@ export const make = Effect.gen(function* () { // Read only this thread; full shell reads would make each activity event scale with all threads. const threadShell = yield* readAwarenessThread(threadId); yield* trackSubagentRequest(threadId, threadShell); + if (threadShell !== null) { + yield* scheduleSnoozeWakeup(threadShell, (yield* DateTime.now).epochMilliseconds); + } if ( threadShell !== null && !isListedThreadLineage(threadShell.lineage) && @@ -575,16 +628,22 @@ export const make = Effect.gen(function* () { environmentId, thread, project, + now: (yield* DateTime.now).epochMilliseconds, pendingSubagentRequest: yield* currentSubagentRequestFor(threadId), }); const publishIdentity = agentAwarenessPublishIdentity(snapshot.state, snapshot.subagentWait); const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef); - if ( + const previousCompletion = publishedCompletions.get(threadId); + const sameCompletion = + snapshot.state?.completedAt !== undefined && + previousCompletion?.phase === snapshot.state?.phase && + previousCompletion?.completedAt === snapshot.state?.completedAt; + const replay = (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") && - !hasLivePublishedState(publishedStateByThread, threadId) - ) { - if (Option.isNone(thread) || !terminalWorkSinceStart(thread.value, startedAt)) return; - } + (sameCompletion || + (!hasLivePublishedState(publishedStateByThread, threadId) && + Option.isSome(thread) && + !terminalWorkSinceStart(thread.value, startedAt))); if ( publishedStateByThread.get(threadId) === publishIdentity && // A held completion settled after an earlier clear still has to alert. @@ -604,7 +663,8 @@ export const make = Effect.gen(function* () { // 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"; + const requiresConfirmation = + snapshot.state === null || (snapshot.state?.phase === "completed" && !replay); if (requiresConfirmation) { if (snapshot.state?.phase === "completed") heldCompletions.add(threadId); const nowMs = (yield* DateTime.now).epochMilliseconds; @@ -642,6 +702,7 @@ export const make = Effect.gen(function* () { environmentId, thread: Option.some({ ...thread.value, settledOverride: null }), project, + now: nowMs, pendingSubagentRequest: null, }) : null; @@ -652,6 +713,10 @@ export const make = Effect.gen(function* () { state: completion, reason: unsettled.reason, }).pipe(withRelayClientTracing); + publishedCompletions.set(threadId, { + phase: completion.phase, + completedAt: completion.completedAt, + }); heldCompletions.delete(threadId); publishConfirmDeadlines.set(threadId, nowMs + 5_000); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => @@ -690,7 +755,14 @@ export const make = Effect.gen(function* () { projectId: snapshot.projectId, state, reason: snapshot.reason, + replay, }).pipe(withRelayClientTracing); + if (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") { + publishedCompletions.set(threadId, { + phase: snapshot.state.phase, + completedAt: snapshot.state.completedAt, + }); + } if (state !== null && childRequestId !== undefined) { childAlertByThread.set(threadId, { requestId: childRequestId, updatedAt: state.updatedAt }); } @@ -787,6 +859,7 @@ export const make = Effect.gen(function* () { const worker = yield* makeAgentAwarenessPublishWorker(processThreadPublish); const enqueueThreadPublish = (threadId: ThreadId) => cancelPublishRetry(threadId).pipe(Effect.andThen(worker.enqueue(threadId))); + enqueueSnoozeWakeup = enqueueThreadPublish; enqueueTopLevelPublish = enqueueThreadPublish; const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) => enqueueThreadPublish(threadId).pipe(Effect.andThen(worker.drain)); @@ -851,11 +924,19 @@ export const make = Effect.gen(function* () { ), (thread) => readAwarenessThread(thread.id), ); + const nowMs = (yield* DateTime.now).epochMilliseconds; + yield* Effect.forEach( + activitySnapshot.threads, + (thread) => scheduleSnoozeWakeup(thread, nowMs), + { discard: true }, + ); const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({ environmentId, startedAt, + now: nowMs, projects: projectSnapshot.projects, - threads: [...activitySnapshot.threads, ...liveChildren.filter((thread) => thread !== null)], + // Children first, so a parent's catch-up publish already sees its waiting child. + threads: [...liveChildren.filter((thread) => thread !== null), ...activitySnapshot.threads], }); if (activeThreadIds.length === 0) { yield* Effect.logDebug("agent activity snapshot has no publishable threads"); diff --git a/apps/web/src/components/chat/ThreadVisitWatermark.browser.test.tsx b/apps/web/src/components/chat/ThreadVisitWatermark.browser.test.tsx index a730db051bf0e..ed39ce62114c2 100644 --- a/apps/web/src/components/chat/ThreadVisitWatermark.browser.test.tsx +++ b/apps/web/src/components/chat/ThreadVisitWatermark.browser.test.tsx @@ -37,13 +37,14 @@ const ENVIRONMENT = EnvironmentId.make("environment-visit-watermark"); const threadA: ScopedThreadRef = { environmentId: ENVIRONMENT, threadId: ThreadId.make("a") }; const threadB: ScopedThreadRef = { environmentId: ENVIRONMENT, threadId: ThreadId.make("b") }; -function setShell(ref: ScopedThreadRef, updatedAt: string) { +function setShell(ref: ScopedThreadRef, updatedAt: string, snoozedUntil: string | null = null) { appAtomRegistry.set(shellAtom(shellKey(ref)), { environmentId: ref.environmentId, id: ref.threadId, updatedAt, lastVisitedAt: "2026-01-01T00:00:00.000Z", latestRun: null, + snoozedUntil, }); } @@ -57,7 +58,8 @@ it("records the visit of the next thread at once after switching threads", async mocks.visitRun.mockReset(); mocks.visitRun.mockResolvedValue(AsyncResult.success(undefined)); setShell(threadA, "2026-01-02T00:00:00.000Z"); - setShell(threadB, "2026-01-03T00:00:00.000Z"); + const wokeAt = "2026-01-03T12:00:00.000Z"; + setShell(threadB, "2026-01-03T00:00:00.000Z", wokeAt); using _timers = fakeTimers(); vi.setSystemTime(NOW); // A palette pick already sent B's open with its query, so B's visit is a plain watermark advance. @@ -79,4 +81,7 @@ it("records the visit of the next thread at once after switching threads", async // Thread A's throttle must not hold back B's first watermark: leaving B // within the throttle window would otherwise keep it unread. expect(visitedThreadIds()).toEqual(["a", "b"]); + expect(mocks.visitRun.mock.calls.at(-1)?.[1]).toMatchObject({ + input: { threadId: threadB.threadId, visitedAt: wokeAt }, + }); }); diff --git a/apps/web/src/components/chat/ThreadVisitWatermark.tsx b/apps/web/src/components/chat/ThreadVisitWatermark.tsx index 7737f0dfb733b..aa92e790a1db1 100644 --- a/apps/web/src/components/chat/ThreadVisitWatermark.tsx +++ b/apps/web/src/components/chat/ThreadVisitWatermark.tsx @@ -1,13 +1,36 @@ import { useAtomMount } from "@effect/atom-react"; import { scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { parseThreadKey, threadKey } from "@t3tools/client-runtime/state/entities"; +import { threadWokeAt } from "@t3tools/client-runtime/state/thread-settled"; import { createThreadVisitTracker } from "@t3tools/client-runtime/state/thread-visits"; import { type ScopedThreadRef } from "@t3tools/contracts"; +import { Atom } from "effect/reactivity"; import { useUiStateStore } from "../../uiStateStore"; import { environmentThreadShells, threadEnvironment } from "../../state/threads"; +const wakeAwareVisitShell = Atom.family((key: string) => { + const ref = parseThreadKey(key); + return Atom.make((get) => { + const shell = get(environmentThreadShells.threadShellAtom(ref)); + if (shell === null) return null; + const nowMs = Date.now(); + const wokeAt = threadWokeAt(shell, { now: new Date(nowMs).toISOString() }); + const wakeAtMs = shell.snoozedUntil == null ? NaN : Date.parse(shell.snoozedUntil); + // Snooze expiry advances the visit watermark without a server event. + if (wokeAt === null && wakeAtMs > nowMs) { + const timer = setTimeout( + () => get.refreshSelf(), + Math.min(wakeAtMs - nowMs + 50, 2_147_483_647), + ); + get.addFinalizer(() => clearTimeout(timer)); + } + return { ...shell, wokeAt }; + }).pipe(Atom.withLabel(`web-thread-visit-shell:${key}`)); +}); + const threadVisitTracker = createThreadVisitTracker({ - threadShellAtom: environmentThreadShells.threadShellAtom, + threadShellAtom: (ref) => wakeAwareVisitShell(threadKey(ref)), visit: threadEnvironment.visit, markLocalVisit: (ref, visitedAt) => useUiStateStore.getState().markThreadVisited(scopedThreadKey(ref), visitedAt), diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 33f1518ec30ae..04adb20560b99 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -140,6 +140,120 @@ function makeApnsDeliveries( } describe("AgentActivityPublisher", () => { + it.effect("repaints backfilled inbox rows without a notification", () => { + const replays: boolean[] = []; + return Effect.gen(function* () { + const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; + yield* publisher.publish({ + environmentId: state.environmentId, + environmentPublicKey: "key", + threadId: state.threadId, + state: { ...state, phase: "completed", retained: true }, + replay: true, + }); + expect(replays).toEqual([true]); + }).pipe( + Effect.provide( + layerPublisher.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(AgentActivityRows.AgentActivityRows, makeAgentActivityRows()), + Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()), + Layer.succeed( + LiveActivities.LiveActivities, + makeLiveActivities({ listTargets: () => Effect.succeed([target("ios")]) }), + ), + Layer.succeed( + ApnsDeliveries.ApnsDeliveries, + makeApnsDeliveries({ + sendForTarget: (input) => + Effect.sync(() => { + replays.push(input.replay === true); + return null; + }), + sendPushNotificationForTarget: () => Effect.die("Backfill must not notify"), + }), + ), + ), + ), + ), + ), + ); + }); + it.effect("notifies an off-card completion without changing the five display rows", () => { + const completed: RelayAgentActivityState = { + ...state, + phase: "completed", + unread: true, + retained: true, + }; + const failures = Array.from({ length: 5 }, (_, i) => ({ + ...state, + threadId: `failure-${i}` as RelayAgentActivityState["threadId"], + phase: "failed" as const, + unread: true, + retained: true, + })); + const cards: Array[0]> = + []; + const notifications: Array< + Parameters[0] + > = []; + return Effect.gen(function* () { + const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; + const input = { + environmentId: state.environmentId, + environmentPublicKey: "key", + threadId: state.threadId, + state: completed, + }; + yield* publisher.publish(input); + expect(cards[0]?.aggregate?.activities).toHaveLength(5); + expect(cards[0]?.aggregate?.activities.some((row) => row.threadId === state.threadId)).toBe( + false, + ); + expect(notifications).toHaveLength(1); + expect(notifications[0]?.aggregate?.activities[0]?.threadId).toBe(state.threadId); + yield* publisher.publish({ ...input, replay: true }); + expect(notifications).toHaveLength(1); + }).pipe( + Effect.provide( + layerPublisher.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed( + AgentActivityRows.AgentActivityRows, + makeAgentActivityRows({ + listForUser: () => Effect.succeed([...failures, completed]), + }), + ), + Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()), + Layer.succeed( + LiveActivities.LiveActivities, + makeLiveActivities({ listTargets: () => Effect.succeed([target("ios")]) }), + ), + Layer.succeed( + ApnsDeliveries.ApnsDeliveries, + makeApnsDeliveries({ + sendForTarget: (input) => + Effect.sync(() => { + cards.push(input); + return null; + }), + sendPushNotificationForTarget: (input) => + Effect.sync(() => { + notifications.push(input); + return null; + }), + }), + ), + ), + ), + ), + ), + ); + }); + it.effect("routes Android publication and registration replay to FCM alongside iOS", () => { const android = { ...target("android"), platform: "android" as const, ios_major_version: null }; const ios = target("ios"); @@ -158,7 +272,7 @@ describe("AgentActivityPublisher", () => { deviceId: android.device_id, }); expect(fcmCalls).toEqual([ - { target: android, state }, + { target: android, state, replay: false }, { target: android, state: null, replay: true }, ]); expect(appleDevices).toEqual(["ios"]); @@ -455,8 +569,8 @@ describe("makeAggregateState", () => { expect(aggregate?.activeCount).toBe(1); expect(aggregate?.subtitle).toBe("Agent work in progress"); expect(aggregate?.activities).toMatchObject([ - { threadId: "thread-active", phase: "running" }, { threadId: "thread-done", phase: "completed", status: "Done" }, + { threadId: "thread-active", phase: "running" }, ]); expect(aggregate?.updatedAt).toBe("1970-01-01T00:59:00.000Z"); }); @@ -487,7 +601,44 @@ describe("makeAggregateState", () => { ).toBeNull(); }); - it("gives active agents the display slots before finished ones", () => { + it("keeps older inbox completions and exact counts beyond the row cap", () => { + const oldRows = Array.from({ length: 8 }, (_, index): RelayAgentActivityState => ({ + ...state, + threadId: `old-${index}` as RelayAgentActivityState["threadId"], + phase: "completed", + retained: true, + })); + const aggregate = makeAggregateState({ + activeStates: [ + { ...state, updatedAt: "1970-01-02T00:00:00.000Z" }, + { + ...state, + threadId: "input" as RelayAgentActivityState["threadId"], + phase: "waiting_for_input", + updatedAt: "1970-01-02T00:00:00.000Z", + }, + ...oldRows, + ], + terminalState: null, + nowMs: 24 * hourMs, + }); + expect(aggregate).toMatchObject({ + activeCount: 2, + workingCount: 1, + attentionCount: 1, + totalCount: 10, + }); + expect(aggregate?.activities).toHaveLength(5); + expect(aggregate?.activities.map((row) => row.phase)).toEqual([ + "waiting_for_input", + "completed", + "completed", + "completed", + "completed", + ]); + }); + + it("gives unread completions display slots before working agents", () => { const mkActive = (id: string): RelayAgentActivityState => ({ ...state, threadId: id as RelayAgentActivityState["threadId"], @@ -514,11 +665,11 @@ describe("makeAggregateState", () => { expect(aggregate?.activeCount).toBe(5); expect(aggregate?.activities).toMatchObject([ + { threadId: "thread-done" }, { threadId: "a-1" }, { threadId: "a-2" }, { threadId: "a-3" }, { threadId: "a-4" }, - { threadId: "a-5" }, ]); }); }); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index 6ef258acdb6d3..455e9903460ae 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -39,6 +39,7 @@ export class AgentActivityPublisher extends Context.Service< readonly environmentPublicKey: string; readonly threadId: string; readonly state: RelayAgentActivityState | null; + readonly replay?: boolean; }) => Effect.Effect; readonly replayForLiveActivityRegistration: (input: { readonly userId: string; @@ -62,6 +63,7 @@ const make = Effect.gen(function* () { readonly deliveryUser: EnvironmentLinks.AgentAwarenessDeliveryUserRecord; readonly state: RelayAgentActivityState | null; readonly nowMs: number; + readonly replay?: boolean; }) { const activeStates = input.deliveryUser.liveActivitiesEnabled ? yield* rows.listForUser({ userId: input.deliveryUser.userId }) @@ -74,9 +76,15 @@ const make = Effect.gen(function* () { }) : null; const notificationOnlyAggregate = + !input.replay && input.deliveryUser.notificationsEnabled && - !input.deliveryUser.liveActivitiesEnabled && - input.state !== null + input.state !== null && + (!input.deliveryUser.liveActivitiesEnabled || + !liveActivityAggregate?.activities.some( + (row) => + row.environmentId === input.state?.environmentId && + row.threadId === input.state.threadId, + )) ? makeAggregateState({ activeStates: isTerminalPhase(input.state) ? [] : [input.state], terminalState: isTerminalPhase(input.state) ? input.state : null, @@ -88,14 +96,29 @@ 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, + replay: input.replay ?? false, + }), + ]; } return yield* Effect.all( [ apnsDeliveries.sendForTarget({ target, aggregate: liveActivityAggregate, + ...(input.replay && input.state + ? { + replayThread: { + environmentId: input.state.environmentId, + threadId: input.state.threadId, + }, + } + : {}), nowMs: input.nowMs, + replay: input.replay ?? false, }), notificationOnlyAggregate === null ? Effect.succeed(null) @@ -166,7 +189,7 @@ const make = Effect.gen(function* () { terminalState: null, nowMs: now.epochMilliseconds, }); - if (aggregate !== null) return; + if (aggregate !== null && aggregate.pendingCount !== 0) return; const targets = yield* liveActivities.listTargets({ userId }); const idleDeviceIds = new Set(candidates.map((target) => target.device_id)); yield* Effect.forEach( @@ -207,7 +230,7 @@ const make = Effect.gen(function* () { "relay.agent_activity.phase": input.state?.phase ?? "deleted", }); if (input.state) { - // Retain terminal rows until cron pruning so completed threads remain visible alongside active ones. + // Inbox rows are removed by settlement or archival, not by elapsed time. yield* rows.upsert({ environmentPublicKey: input.environmentPublicKey, state: input.state, @@ -231,6 +254,7 @@ const make = Effect.gen(function* () { publishForDeliveryUser({ deliveryUser, state: input.state, + replay: input.replay ?? false, nowMs: now.epochMilliseconds, }), { concurrency: 4 }, diff --git a/infra/relay/src/agentActivity/AgentActivityRows.ts b/infra/relay/src/agentActivity/AgentActivityRows.ts index 0faf5b9da358e..a5f8263ef0283 100644 --- a/infra/relay/src/agentActivity/AgentActivityRows.ts +++ b/infra/relay/src/agentActivity/AgentActivityRows.ts @@ -7,7 +7,7 @@ import * as Function from "effect/Function"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; -import { and, desc, eq, isNull, lt, sql } from "drizzle-orm"; +import { and, desc, eq, isNull, lt, or, sql } from "drizzle-orm"; import * as RelayDb from "../db.ts"; import { relayAgentActivityRows, relayEnvironmentLinks } from "../persistence/schema.ts"; @@ -71,6 +71,7 @@ export class AgentActivityRows extends Context.Service< }) => Effect.Effect; readonly pruneTerminal: (input: { readonly updatedBefore: string; + readonly retainedBefore: string; }) => Effect.Effect; readonly remove: (input: { readonly environmentId: string; @@ -189,7 +190,13 @@ const make = Effect.gen(function* () { .where( and( sql`${relayAgentActivityRows.stateJson} ->> 'phase' IN ('completed', 'failed')`, - lt(relayAgentActivityRows.updatedAt, input.updatedBefore), + or( + and( + sql`COALESCE(${relayAgentActivityRows.stateJson} ->> 'retained', 'false') != 'true'`, + lt(relayAgentActivityRows.updatedAt, input.updatedBefore), + ), + lt(relayAgentActivityRows.updatedAt, input.retainedBefore), + ), ), ) .pipe( diff --git a/infra/relay/src/agentActivity/ApnsClient.test.ts b/infra/relay/src/agentActivity/ApnsClient.test.ts index 81e76b0391aae..c21aecc58da38 100644 --- a/infra/relay/src/agentActivity/ApnsClient.test.ts +++ b/infra/relay/src/agentActivity/ApnsClient.test.ts @@ -112,6 +112,20 @@ describe("ApnsClient", () => { }).pipe(Effect.provide(layerTest)), ); + it.effect("dismisses a fully read Live Activity immediately", () => + Effect.gen(function* () { + const apns = yield* ApnsClient.ApnsClient; + const request = apns.makeLiveActivityRequest({ + event: "end", + token: "token", + state: { ...state, activeCount: 0, pendingCount: 0 }, + nowEpochSeconds: 123, + nowIso: DateTime.formatIso(now), + }); + expect(request.payload).toMatchObject({ aps: { event: "end", "dismissal-date": 123 } }); + }).pipe(Effect.provide(layerTest)), + ); + it.effect("puts remote routing metadata only in Expo's body field", () => Effect.gen(function* () { const apns = yield* ApnsClient.ApnsClient; diff --git a/infra/relay/src/agentActivity/ApnsClient.ts b/infra/relay/src/agentActivity/ApnsClient.ts index b21953c59eecc..efcd7b7d3a107 100644 --- a/infra/relay/src/agentActivity/ApnsClient.ts +++ b/infra/relay/src/agentActivity/ApnsClient.ts @@ -181,7 +181,12 @@ function buildLiveActivityRequest(input: MakeLiveActivityRequestInput): ApnsLive ...(input.state ? { "content-state": contentState(input.state) } : {}), ...(input.alert ? liveActivityAlertPayload(input.alert) : {}), "dismissal-date": - timestamp + (input.state ? DISMISS_AFTER_SECONDS : CONTENTLESS_DISMISS_AFTER_SECONDS), + timestamp + + (input.state?.pendingCount === 0 + ? 0 + : input.state + ? DISMISS_AFTER_SECONDS + : CONTENTLESS_DISMISS_AFTER_SECONDS), }, }, }; diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index 64823f4973593..1ada40f61464e 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -1111,6 +1111,97 @@ describe("live activity alert decisions", () => { }); describe("fast completion delivery", () => { + it.effect.each([true, false])( + "keeps unread Done live and ends it when read: unread=%s", + (unread) => { + const queuedJobs: SignedApnsDeliveryJob[] = []; + const done = { + ...aggregate, + activeCount: 0, + pendingCount: unread ? 1 : 0, + unreadCount: unread ? 1 : 0, + activities: [{ ...aggregate.activities[0]!, phase: "completed" as const, unread }], + }; + const device = { ...target, last_aggregate_json: JSON.stringify(aggregate) }; + return Effect.gen(function* () { + const d = yield* ApnsDeliveries.ApnsDeliveries; + yield* d.sendForTarget({ target: device, aggregate: done, nowMs: 0 }); + expect(queuedJobs[0]?.payload.kind).toBe( + unread ? "live_activity_update" : "live_activity_end", + ); + if (!unread) expect(queuedJobs[0]?.payload.alert ?? null).toBeNull(); + }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs, currentTargets: [device] }))); + }, + ); + + it.effect.each([ + { replay: true, replayThread: false, ends: false }, + { replay: false, replayThread: false, ends: true }, + // A thread replay is a real read of an already announced result. + { replay: true, replayThread: true, ends: true }, + ])( + "preserves a freshly armed token only for a registration replay: %o", + ({ replay, replayThread, ends }) => { + const queuedJobs: SignedApnsDeliveryJob[] = []; + const read = { + ...aggregate, + activeCount: 0, + pendingCount: 0, + unreadCount: 0, + activities: [{ ...aggregate.activities[0]!, phase: "completed" as const, unread: false }], + }; + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* deliveries.sendForTarget({ + target, + aggregate: read, + nowMs: 0, + replay, + ...(replayThread + ? { replayThread: { environmentId: state.environmentId, threadId: state.threadId } } + : {}), + }); + expect(queuedJobs.map((job) => job.payload.kind)).toEqual( + ends ? ["live_activity_end"] : [], + ); + }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs }))); + }, + ); + + it.effect("does not deliver a last-read end after another thread starts", () => { + const queuedJobs: SignedApnsDeliveryJob[] = []; + const requests: HttpClientRequest.HttpClientRequest[] = []; + const read = { ...state, phase: "completed" as const, unread: false }; + const readAggregate = { + ...aggregate, + activeCount: 0, + pendingCount: 0, + unreadCount: 0, + activities: [{ ...aggregate.activities[0]!, phase: "completed" as const, unread: false }], + }; + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* deliveries.sendForTarget({ target, aggregate: readAggregate, nowMs: 20_000 }); + expect(queuedJobs).toHaveLength(1); + yield* deliveries.processSignedJob(queuedJobs[0]!); + expect(requests).toHaveLength(0); + }).pipe( + Effect.provide( + layerFor({ + attempts: [], + queuedJobs, + config: signingConfig, + activityStates: [read, { ...state, threadId: ThreadId.make("new-work") }], + execute: (request) => + Effect.sync(() => { + requests.push(request); + return HttpClientResponse.fromWeb(request, Response.json({})); + }), + }), + ), + ); + }); + it.effect("keeps a completion alert when work finishes before running delivery", () => { const queuedJobs: SignedApnsDeliveryJob[] = []; const old = { @@ -1139,19 +1230,20 @@ describe("fast completion delivery", () => { ).toBe(true); }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs, currentTargets: [device] }))); }); - it.effect("replays a newly visible completion without alerting", () => { + it.effect("replays a newly visible completion without consuming another thread's alert", () => { const queuedJobs: SignedApnsDeliveryJob[] = []; + const markedDeliveries: Array< + Parameters[0] + > = []; + const other = { ...aggregate.activities[0]!, threadId: ThreadId.make("other") }; const done = { - ...aggregate, - activeCount: 0, - activities: [{ ...aggregate.activities[0]!, phase: "completed" as const }], - }; - const previous = { ...aggregate, activities: [ - { ...aggregate.activities[0]!, threadId: "other" as RelayAgentActivityState["threadId"] }, + { ...aggregate.activities[0]!, phase: "completed" as const }, + { ...other, phase: "completed" as const }, ], }; + const previous = { ...aggregate, activities: [other] }; const device = { ...target, last_aggregate_json: JSON.stringify(previous) }; return Effect.gen(function* () { const deliveries = yield* ApnsDeliveries.ApnsDeliveries; @@ -1160,10 +1252,33 @@ describe("fast completion delivery", () => { aggregate: done, nowMs: 0, replay: true, + replayThread: { environmentId: state.environmentId, threadId: state.threadId }, }); expect(queuedJobs).toHaveLength(1); expect(queuedJobs[0]?.payload.alert).toBeUndefined(); - }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs }))); + const result = yield* deliveries.processSignedJob(queuedJobs[0]!); + expect(result.ok).toBe(true); + expect(markedDeliveries[0]?.aggregate?.activities).toEqual([other, done.activities[0]]); + device.last_aggregate_json = JSON.stringify(markedDeliveries[0]?.aggregate); + yield* deliveries.sendForTarget({ target: device, aggregate: done, nowMs: 0 }); + expect(queuedJobs[1]?.payload.alert).toMatchObject({ title: "Thread" }); + }).pipe( + Effect.provide( + layerFor({ + attempts: [], + queuedJobs, + markedDeliveries, + config: signingConfig, + currentTargets: [device], + activityStates: [ + { ...state, phase: "completed" }, + { ...state, threadId: other.threadId, phase: "completed" }, + ], + execute: (request) => + Effect.succeed(HttpClientResponse.fromWeb(request, new Response("", { status: 200 }))), + }), + ), + ); }); }); diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 8913bdd1f8c5b..a251c7c27609c 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -34,6 +34,7 @@ import { SignedApnsDeliveryJob, verifySignedApnsDeliveryJob, type ApnsDeliveryJobVerificationError, + type ApnsDeliveryJobPayload, } from "./apnsDeliveryJobs.ts"; import * as AgentActivityRows from "./AgentActivityRows.ts"; import * as DeliveryAttempts from "./DeliveryAttempts.ts"; @@ -144,7 +145,10 @@ function parsePreferences(value: string): RelayAgentAwarenessPreferences | null function aggregateNeedsAttention(aggregate: RelayAgentActivityAggregateState): boolean { return aggregate.activities.some( - (row) => row.phase === "waiting_for_approval" || row.phase === "waiting_for_input", + (row) => + row.phase === "waiting_for_approval" || + row.phase === "waiting_for_input" || + row.phase === "woke", ); } @@ -160,7 +164,12 @@ function shouldUpdateLiveActivity(input: { if (JSON.stringify(input.previousAggregate) === JSON.stringify(input.nextAggregate)) { return false; } - if (input.previousAggregate.activeCount !== input.nextAggregate.activeCount) { + if ( + input.previousAggregate.activeCount !== input.nextAggregate.activeCount || + input.previousAggregate.unreadCount !== input.nextAggregate.unreadCount || + input.previousAggregate.pendingCount !== input.nextAggregate.pendingCount || + input.previousAggregate.totalCount !== input.nextAggregate.totalCount + ) { return true; } if (aggregateNeedsAttention(input.nextAggregate)) { @@ -213,6 +222,7 @@ function chooseLiveActivityDelivery(input: { readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; + readonly threadReplay?: boolean; }): ChosenLiveActivityDelivery | "suppressed" | null { const preferences = parsePreferences(input.target.preferences_json); if (preferences?.liveActivitiesEnabled === false) { @@ -228,7 +238,7 @@ function chooseLiveActivityDelivery(input: { if (!input.target.activity_push_token) { return null; } - if (input.aggregate === null) { + if (input.aggregate === null || input.aggregate.pendingCount === 0) { // Registration replay can precede a new thread's first publish; preserve a freshly armed card until state arrives. const armedAtMs = Option.match( input.target.remote_started_at === null @@ -236,7 +246,12 @@ function chooseLiveActivityDelivery(input: { : DateTime.make(input.target.remote_started_at), { onNone: () => null, onSome: (dt) => dt.epochMilliseconds }, ); - if (armedAtMs !== null && input.nowMs - armedAtMs < FRESHLY_ARMED_GRACE_MS) { + // Per-thread replays carry real changes such as a read result, so only registration replays wait. + if ( + (input.aggregate === null || (input.replay && !input.threadReplay)) && + armedAtMs !== null && + input.nowMs - armedAtMs < FRESHLY_ARMED_GRACE_MS + ) { return null; } return { @@ -281,6 +296,7 @@ function chooseDelivery(input: { readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; + readonly threadReplay?: boolean; }): ChosenDelivery | null { const liveActivityDelivery = chooseLiveActivityDelivery(input); if (liveActivityDelivery === "suppressed") { @@ -427,6 +443,7 @@ interface SendLiveActivityDeliveryInputBase { readonly target: LiveActivityDeliveryTarget; readonly token: string; readonly sourceJobId: string; + readonly replayThread?: ApnsDeliveryJobPayload["replayThread"]; } export type SendLiveActivityDeliveryInput = @@ -486,6 +503,7 @@ export class ApnsDeliveries extends Context.Service< readonly aggregate: RelayAgentActivityAggregateState | null; readonly nowMs: number; readonly replay?: boolean; + readonly replayThread?: ApnsDeliveryJobPayload["replayThread"]; }) => Effect.Effect; readonly sendPushNotificationForTarget: (input: { readonly target: LiveActivities.TargetRow; @@ -518,14 +536,14 @@ const make = Effect.gen(function* () { const userHasContentToShow = Effect.fnUntraced(function* (userId: string) { const now = yield* DateTime.now; return yield* activityRows.listForUser({ userId }).pipe( - Effect.map( - (activeStates) => - makeAggregateState({ - activeStates, - terminalState: null, - nowMs: now.epochMilliseconds, - }) !== null, - ), + Effect.map((activeStates) => { + const aggregate = makeAggregateState({ + activeStates, + terminalState: null, + nowMs: now.epochMilliseconds, + }); + return aggregate !== null && aggregate.pendingCount !== 0; + }), // Keep the card on persistence failures; a later sweep can retry. Effect.catch((error) => Effect.logWarning("Content recheck failed; keeping the card", { error }).pipe( @@ -553,6 +571,8 @@ const make = Effect.gen(function* () { (current) => current !== null && current.phase === input.phase && + (!(current.phase === "completed" || current.phase === "failed") || + current.unread !== false) && current.updatedAt === input.updatedAt, ), Effect.catchCause((cause) => @@ -582,6 +602,7 @@ const make = Effect.gen(function* () { return ( current !== undefined && current.phase === row.phase && + current.unread === row.unread && current.updatedAt === row.updatedAt ); }); @@ -752,11 +773,25 @@ const make = Effect.gen(function* () { }), ); if (result.ok) { + let deliveryBaseline = aggregate; + if (input.replayThread && aggregate) { + const matchesReplay = (row: RelayAgentActivityAggregateState["activities"][number]) => + row.environmentId === input.replayThread?.environmentId && + row.threadId === input.replayThread.threadId; + const previous = parseAggregate(currentTarget.last_aggregate_json); + deliveryBaseline = { + ...(previous ?? aggregate), + activities: [ + ...(previous?.activities ?? []).filter((row) => !matchesReplay(row)), + ...aggregate.activities.filter(matchesReplay), + ], + }; + } yield* liveActivities.markDelivery({ userId: input.target.user_id, deviceId: input.target.device_id, kind: input.kind, - aggregate, + aggregate: deliveryBaseline, deliveredAt: iso, }); } else if (isPermanentApnsTokenFailure(result)) { @@ -970,6 +1005,7 @@ const make = Effect.gen(function* () { kind: payload.kind, aggregate: payload.aggregate, alert: payload.alert ?? null, + ...(payload.replayThread ? { replayThread: payload.replayThread } : {}), }); case "live_activity_end": return sendLiveActivity({ @@ -1041,6 +1077,7 @@ const make = Effect.gen(function* () { aggregate: input.aggregate, nowMs: input.nowMs, replay: input.replay ?? false, + threadReplay: input.replayThread !== undefined, }); if (!delivery) { return null; @@ -1082,6 +1119,7 @@ const make = Effect.gen(function* () { bundleId: input.target.bundle_id, apsEnvironment: input.target.aps_environment, aggregate: delivery.aggregate, + ...(input.replayThread ? { replayThread: input.replayThread } : {}), alert, }); if (delivery.kind === "live_activity_end" && notification && input.target.push_token) { diff --git a/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts b/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts index 09cff5184b058..7e9456cdffc16 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveryQueue.ts @@ -65,6 +65,7 @@ export class ApnsDeliveryQueue extends Context.Service< readonly apsEnvironment?: "sandbox" | "production" | null; readonly aggregate: ApnsDeliveryJobPayload["aggregate"]; readonly alert?: ApnsDeliveryJobPayload["alert"]; + readonly replayThread?: ApnsDeliveryJobPayload["replayThread"]; }) => Effect.Effect; readonly enqueuePushNotification: (input: { readonly userId: string; diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index f1ce50d5832b6..a701513b3c77f 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -26,6 +26,7 @@ const aggregateFor = (states: ReadonlyArray) => makeAggregateState({ activeStates: states, terminalState: null, nowMs: 0 })!; const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); +const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const state: RelayAgentActivityState = { environmentId: EnvironmentId.make("env"), threadId: ThreadId.make("thread"), @@ -241,7 +242,7 @@ describe("Android delivery routing", () => { expect(alerts).toHaveLength(1); expect(alerts[0]?.data).toMatchObject({ alert_title: title, - alert_body: "Fix notifications, Second thread", + alert_body: "Second thread, Fix notifications", alert_path: "/", active, }); @@ -250,6 +251,193 @@ describe("Android delivery routing", () => { }, ); + it.effect( + "a per-thread replay does not consume another thread's pending completion alert", + () => { + const h = harness(); + h.current.otherStates = [secondState]; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process(h.job); + h.current.otherStates = [{ ...secondState, phase: "completed", unread: true }]; + const baseline = h.current.target.last_aggregate_json; + yield* delivery.process({ ...h.job, replay: true }); + expect(decodeJson(h.current.target.last_aggregate_json)).toEqual(decodeJson(baseline)); + yield* delivery.process({ ...h.job, state: h.current.otherStates[0] }); + expect(h.sent.filter((message) => message.alert)).toHaveLength(1); + }).pipe(Effect.provide(h.layer)); + }, + ); + + it.effect( + "a replay adds a missing baseline row without acknowledging another pending result", + () => { + const h = harness(); + h.current.target.last_aggregate_json = encodeJson(aggregateFor([secondState])); + h.current.state = { ...state, phase: "completed", unread: true, retained: true }; + h.current.otherStates = [ + { ...secondState, phase: "completed", unread: true, retained: true }, + ]; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process({ ...h.job, state: h.current.state, replay: true }); + yield* delivery.process({ ...h.job, state: h.current.otherStates[0] }); + const alerts = h.sent.filter((message) => message.alert); + expect(alerts).toHaveLength(1); + expect(alerts[0]?.data.alert_title).toBe(secondState.threadTitle); + }).pipe(Effect.provide(h.layer)); + }, + ); + + it.effect.each(["completed", "waiting_for_approval"] as const)( + "a replayed %s row does not alert on the next ordinary job", + (phase) => { + const h = harness(); + h.current.otherStates = [secondState]; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process(h.job); + h.current.state = { ...state, phase, unread: true, retained: true }; + yield* delivery.process({ ...h.job, state: h.current.state, replay: true }); + yield* delivery.process({ ...h.job, state: secondState }); + expect(h.sent.filter((message) => message.alert)).toHaveLength(0); + h.current.otherStates = [{ ...secondState, phase: "completed", unread: true }]; + yield* delivery.process({ ...h.job, state: h.current.otherStates[0] }); + const alerts = h.sent.filter((message) => message.alert); + expect(alerts).toHaveLength(1); + expect(alerts[0]?.data.alert_title).toBe(secondState.threadTitle); + }).pipe(Effect.provide(h.layer)); + }, + ); + + it.effect.each(["running", "completed"] as const)( + "an off-card %s update cannot consume a visible approval alert", + (phase) => { + const h = harness(); + const failures = Array.from({ length: 5 }, (_, i) => ({ + ...state, + threadId: ThreadId.make(`failure-${i}`), + phase: "failed" as const, + unread: false, + retained: true, + })); + h.current.target.preferences_json = encodeJson({ ...preferences, notifyOnCompletion: false }); + h.current.otherStates = [...failures, secondState]; + h.current.target.last_aggregate_json = encodeJson( + aggregateFor([state, ...h.current.otherStates]), + ); + h.current.state = { ...state, phase, unread: true, retained: true }; + const approval = { ...secondState, phase: "waiting_for_approval" as const }; + h.current.otherStates = [...failures, approval]; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process({ ...h.job, state: h.current.state }); + expect(h.sent[0]?.data).toMatchObject({ + alert_title: secondState.threadTitle, + alert_body: "Approval: Project", + }); + yield* delivery.process({ ...h.job, state: approval }); + expect(h.sent.filter((message) => message.alert)).toHaveLength(1); + }).pipe(Effect.provide(h.layer)); + }, + ); + + it.effect.each(["completed", "woke"] as const)( + "keeps a pending %s row on the card for released Android builds", + (phase) => { + const h = harness(); + h.current.state = { ...state, phase, unread: phase === "completed", retained: true }; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process({ ...h.job, state: h.current.state }); + // An unread result must not hold `active`, or native dismissal never sees a new run. + expect(h.sent[0]?.data).toMatchObject({ + active: String(phase === "woke"), + activity_expires_at: String(24 * 60 * 60_000), + activity_working_count: "0", + activity_attention_count: phase === "woke" ? "1" : "0", + }); + expect(h.sent.filter((message) => message.alert)).toHaveLength(phase === "woke" ? 0 : 1); + }).pipe(Effect.provide(h.layer)); + }, + ); + + it.effect("uses running threads outside the five display rows to set Android expiry", () => { + const h = harness(); + h.current.otherStates = Array.from({ length: 5 }, (_, i) => ({ + ...state, + threadId: ThreadId.make(`failure-${i}`), + threadTitle: `Old failure ${i}`, + phase: "failed" as const, + unread: false, + retained: true, + })); + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process(h.job); + expect(h.sent[0]?.data.activity_expires_at).toBe(String(2 * 60 * 60_000)); + }).pipe(Effect.provide(h.layer)); + }); + + it.effect("alerts a new result outside the five visible rows and keeps replays silent", () => { + const h = harness(); + h.current.otherStates = Array.from({ length: 5 }, (_, i) => ({ + ...state, + threadId: ThreadId.make(`failure-${i}`), + threadTitle: `Old failure ${i}`, + phase: "failed" as const, + unread: true, + retained: true, + })); + h.current.target.last_aggregate_json = encodeJson( + aggregateFor([state, ...h.current.otherStates]), + ); + h.current.state = { ...state, phase: "completed", unread: true, retained: true }; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process({ ...h.job, state: h.current.state }); + expect(h.sent[0]?.data.alert_title).toBe(state.threadTitle); + expect(h.sent[0]?.data.activity_line_4).not.toContain(state.threadTitle); + yield* delivery.process({ ...h.job, state: h.current.state, replay: true }); + expect(h.sent.filter((message) => message.alert)).toHaveLength(1); + }).pipe(Effect.provide(h.layer)); + }); + + it.effect("groups an off-card completion with a new on-card failure", () => { + const h = harness(); + const oldFailures = Array.from({ length: 4 }, (_, i) => ({ + ...state, + threadId: ThreadId.make(`failure-${i}`), + threadTitle: `Old failure ${i}`, + phase: "failed" as const, + unread: true, + retained: true, + })); + const newFailure = { + ...oldFailures[0]!, + threadId: ThreadId.make("new-failure"), + threadTitle: "New failure", + }; + h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, ...oldFailures])); + h.current.state = { ...state, phase: "completed", unread: true, retained: true }; + h.current.otherStates = [...oldFailures, newFailure]; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + yield* delivery.process({ ...h.job, state: h.current.state }); + expect(h.sent[0]?.data.alert_title).toBe("2 agents finished"); + expect(h.sent[0]?.data.alert_body).toContain(state.threadTitle); + expect(h.sent[0]?.data.alert_body).toContain("New failure"); + }).pipe(Effect.provide(h.layer)); + }); + + it("expires an Android card immediately after its last result is read", () => { + const read = { ...state, phase: "completed" as const, unread: false, retained: true }; + expect(androidActivityData(aggregateFor([read]), 60_000)).toMatchObject({ + active: "false", + activity_expires_at: "0", + }); + }); + it.effect("filters disabled event types before counting a group", () => { const h = harness(); h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, secondState])); diff --git a/infra/relay/src/agentActivity/FcmDeliveries.ts b/infra/relay/src/agentActivity/FcmDeliveries.ts index d2795a90a88e5..3ea70e2445244 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.ts @@ -213,8 +213,11 @@ const make = Effect.gen(function* () { : null; let alert: ReturnType = null; // Deletions update the card but must not consume another thread's pending completion alert. - let acknowledgeAggregate = job.state !== null || job.replay === true || aggregate === null; - if (job.state && preferences.value.notificationsEnabled) { + // Full registration replays establish a baseline; per-thread replays cannot consume other alerts. + let acknowledgeAggregate = job.replay + ? job.state === null + : job.state !== null || aggregate === null; + if (job.state && !job.replay && preferences.value.notificationsEnabled) { const state = yield* rows.getForUserThread({ userId: job.userId, environmentId: job.state.environmentId, @@ -261,13 +264,32 @@ const make = Effect.gen(function* () { allowedEnvironments.add(environmentId); } } + const hiddenCurrentState = + state?.phase === job.state.phase && + state.updatedAt === job.state.updatedAt && + !aggregate.activities.some( + (row) => row.environmentId === state.environmentId && row.threadId === state.threadId, + ) && + deliveryUser.notificationsEnabled && + !isExpiredAgentActivityState(state, now.epochMilliseconds) + ? state + : null; + // An off-card result joins the group instead of losing to another thread's alert. + const hiddenRow = hiddenCurrentState + ? makeAggregateState({ + activeStates: [hiddenCurrentState], + terminalState: null, + nowMs: now.epochMilliseconds, + })?.activities[0] + : undefined; alert = androidAlertForAggregate({ previousAggregate, nextAggregate: { ...aggregate, - activities: aggregate.activities.filter((row) => - allowedEnvironments.has(row.environmentId), - ), + activities: [ + ...aggregate.activities.filter((row) => allowedEnvironments.has(row.environmentId)), + ...(hiddenRow ? [hiddenRow] : []), + ], }, preferences: preferences.value, nowMs: now.epochMilliseconds, @@ -281,6 +303,38 @@ const make = Effect.gen(function* () { alert = androidAlertForState(state, preferences.value, now.epochMilliseconds); } } + let deliveryBaseline = aggregate; + if (job.replay && job.state) { + const replayedAggregate = makeAggregateState({ + activeStates: states.filter( + (state) => + state.environmentId === job.state?.environmentId && + state.threadId === job.state.threadId && + state.phase === job.state.phase && + state.updatedAt === job.state.updatedAt, + ), + terminalState: null, + nowMs: now.epochMilliseconds, + }); + const replayedRow = replayedAggregate?.activities[0]; + if (replayedAggregate && replayedRow) { + // A silent thread replay acknowledges only itself, preserving other pending transitions. + const previousRows = previousAggregate?.activities ?? []; + const index = previousRows.findIndex( + (row) => + row.environmentId === replayedRow.environmentId && + row.threadId === replayedRow.threadId, + ); + deliveryBaseline = { + ...(previousAggregate ?? replayedAggregate), + activities: + index === -1 + ? [...previousRows, replayedRow] + : previousRows.toSpliced(index, 1, replayedRow), + }; + acknowledgeAggregate = true; + } + } const displayedAggregate = preferences.value.liveActivitiesEnabled ? aggregate : null; const active = (displayedAggregate?.activeCount ?? 0) > 0; if (!displayedAggregate && !alert && !previousAggregate && job.state !== null) return; @@ -289,7 +343,11 @@ const make = Effect.gen(function* () { device_id: job.deviceId, user_id: job.userId, updated_at: String(now.epochMilliseconds), - ...androidActivityData(displayedAggregate), + ...androidActivityData( + displayedAggregate, + now.epochMilliseconds, + displayedAggregate ? states : [], + ), ...alert, }; if (alert) { @@ -328,7 +386,7 @@ const make = Effect.gen(function* () { deviceId: job.deviceId, kind: active ? "live_activity_update" : "live_activity_end", // Retain terminal rows as the baseline so finished-card replays cannot alert again. - aggregate: preferences.value.liveActivitiesEnabled ? aggregate : null, + aggregate: preferences.value.liveActivitiesEnabled ? deliveryBaseline : null, deliveredAt: DateTime.formatIso(now), }); } diff --git a/infra/relay/src/agentActivity/agentActivityAggregate.ts b/infra/relay/src/agentActivity/agentActivityAggregate.ts index 99fdcf71ef90d..4e98d44a1a6aa 100644 --- a/infra/relay/src/agentActivity/agentActivityAggregate.ts +++ b/infra/relay/src/agentActivity/agentActivityAggregate.ts @@ -8,6 +8,7 @@ import { isExpiredAgentActivityState, isTerminalPhase, MAX_ACTIVITY_ROWS, + RETAINED_AGENT_ACTIVITY_ROW_TTL_MS, sanitizeAgentActivityAggregateState, } from "./agentActivityPayloads.ts"; @@ -28,6 +29,8 @@ export function statusForPhase(phase: RelayAgentActivityState["phase"]): string return "Working"; case "stale": return "Waiting"; + case "woke": + return "Woke"; } } @@ -39,22 +42,14 @@ function aggregateRowForState(state: RelayAgentActivityState) { threadTitle: state.threadTitle, modelTitle: state.modelTitle, phase: state.phase, + ...(state.unread === undefined ? {} : { unread: state.unread }), + ...(state.completedAt === undefined ? {} : { completedAt: state.completedAt }), status: statusForPhase(state.phase), updatedAt: state.updatedAt, deepLink: state.deepLink, }; } -function terminalAggregateState(state: RelayAgentActivityState): RelayAgentActivityAggregateState { - return sanitizeAgentActivityAggregateState({ - title: "T3 Code", - subtitle: state.phase === "failed" ? "Agent work failed" : "Agent work completed", - activeCount: 0, - updatedAt: state.updatedAt, - activities: [aggregateRowForState(state)], - }); -} - export const TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS = 15 * 60 * 1_000; function isRecentTerminalState(state: RelayAgentActivityState, nowMs: number): boolean { @@ -68,7 +63,12 @@ function isRecentTerminalState(state: RelayAgentActivityState, nowMs: number): b if (Number.isNaN(updatedAtMs)) { return false; } - return nowMs - updatedAtMs <= TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS; + return ( + nowMs - updatedAtMs <= + (state.retained === true + ? RETAINED_AGENT_ACTIVITY_ROW_TTL_MS + : TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS) + ); } export function makeAggregateState(input: { @@ -76,53 +76,68 @@ export function makeAggregateState(input: { readonly terminalState: RelayAgentActivityState | null; readonly nowMs: number; }): RelayAgentActivityAggregateState | null { - const activeStates = input.activeStates.filter( - (state) => !isTerminalPhase(state) && !isExpiredAgentActivityState(state, input.nowMs), + const states = input.activeStates.filter((state) => + isTerminalPhase(state) + ? isRecentTerminalState(state, input.nowMs) + : !isExpiredAgentActivityState(state, input.nowMs), + ); + if (states.length === 0 && input.terminalState !== null) states.push(input.terminalState); + if (states.length === 0) return null; + const activeCount = states.filter((state) => !isTerminalPhase(state)).length; + const unreadCount = states.filter( + (state) => state.phase === "completed" && state.unread !== false, + ).length; + const pendingCount = + activeCount + states.filter((state) => isTerminalPhase(state) && state.unread !== false).length; + const displayedStates = states + .toSorted( + (a, b) => + activityPhasePriority(a.phase, a.unread) - activityPhasePriority(b.phase, b.unread) || + b.updatedAt.localeCompare(a.updatedAt) || + a.environmentId.localeCompare(b.environmentId) || + a.threadId.localeCompare(b.threadId), + ) + .slice(0, MAX_ACTIVITY_ROWS); + const newest = states.reduce((latest, state) => + state.updatedAt > latest.updatedAt ? state : latest, ); - if (activeStates.length === 0) { - if (input.terminalState !== null) { - return terminalAggregateState(input.terminalState); - } - // Alerts compare with the delivered aggregate; replays must repaint without buzzing. - const recentTerminal = input.activeStates - .filter((state) => isRecentTerminalState(state, input.nowMs)) - .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)); - const newest = recentTerminal[0]; - if (!newest) { - return null; - } - return sanitizeAgentActivityAggregateState({ - title: "T3 Code", - subtitle: newest.phase === "failed" ? "Agent work failed" : "Agent work completed", - activeCount: 0, - updatedAt: newest.updatedAt, - activities: recentTerminal.slice(0, MAX_ACTIVITY_ROWS).map(aggregateRowForState), - }); - } - const recentTerminalStates = input.activeStates - .filter((state) => isRecentTerminalState(state, input.nowMs)) - .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)); - const displayedStates = [ - ...activeStates - .toSorted((a, b) => activityPhasePriority(a.phase) - activityPhasePriority(b.phase)) - .slice(0, MAX_ACTIVITY_ROWS), - ...recentTerminalStates, - ].slice(0, MAX_ACTIVITY_ROWS); - const updatedAt = [...activeStates, ...recentTerminalStates].reduce((latest, state) => - state.updatedAt.localeCompare(latest.updatedAt) > 0 ? state : latest, - ).updatedAt; return sanitizeAgentActivityAggregateState({ title: "T3 Code", - subtitle: "Agent work in progress", - activeCount: activeStates.length, - updatedAt, + subtitle: + activeCount > 0 + ? "Agent work in progress" + : newest.phase === "failed" + ? "Agent work failed" + : "Agent work completed", + activeCount, + workingCount: states.filter((state) => state.phase === "running" || state.phase === "starting") + .length, + attentionCount: states.filter( + (state) => + state.phase === "waiting_for_input" || + state.phase === "waiting_for_approval" || + state.phase === "woke", + ).length, + unreadCount, + // Legacy publishers cannot report reads; keep their existing completion lifecycle. + ...(states.every((state) => state.unread !== undefined) || + states.some((state) => state.unread === true || state.phase === "woke") + ? { pendingCount } + : {}), + totalCount: states.length, + updatedAt: newest.updatedAt, activities: displayedStates.map(aggregateRowForState), }); } -export function activityPhasePriority(phase: RelayAgentActivityState["phase"]): number { - if (phase === "waiting_for_approval" || phase === "waiting_for_input") return 0; +export function activityPhasePriority( + phase: RelayAgentActivityState["phase"], + unread?: boolean, +): number { + if (phase === "waiting_for_approval" || phase === "waiting_for_input" || phase === "woke") + return 0; if (phase === "failed") return 1; - if (phase === "starting" || phase === "running") return 2; - return 3; + if (phase === "completed" && unread !== false) return 2; + if (phase === "starting" || phase === "running") return 3; + return 4; } diff --git a/infra/relay/src/agentActivity/agentActivityAlerts.ts b/infra/relay/src/agentActivity/agentActivityAlerts.ts index c6008950e95e8..19270e9a0deb2 100644 --- a/infra/relay/src/agentActivity/agentActivityAlerts.ts +++ b/infra/relay/src/agentActivity/agentActivityAlerts.ts @@ -36,9 +36,11 @@ export function rowPhasesChanged( ): boolean { if (previousAggregate.activities.length !== nextAggregate.activities.length) return true; const previousPhases = new Map( - previousAggregate.activities.map((row) => [rowKey(row), row.phase]), + previousAggregate.activities.map((row) => [rowKey(row), `${row.phase}:${row.unread}`]), + ); + return nextAggregate.activities.some( + (row) => previousPhases.get(rowKey(row)) !== `${row.phase}:${row.unread}`, ); - return nextAggregate.activities.some((row) => previousPhases.get(rowKey(row)) !== row.phase); } function isAttentionPhase(phase: string): boolean { @@ -88,7 +90,7 @@ function newlyTerminalRows( previousAggregate.activities.map((row) => [rowKey(row), row.phase]), ); return nextAggregate.activities.filter((row) => { - if (row.phase !== "completed" && row.phase !== "failed") return false; + if (row.unread === false || (row.phase !== "completed" && row.phase !== "failed")) return false; const previousPhase = previousPhases.get(rowKey(row)); return ( (includeUnobserved || previousPhase !== undefined) && @@ -142,11 +144,13 @@ export function alertForTerminalAggregate(input: { readonly preferences: RelayAgentAwarenessPreferences | null; }): AgentActivityAlert | null { const row = input.aggregate?.activities[0]; - if (!row || (row.phase !== "completed" && row.phase !== "failed")) return null; + if (!row || row.unread === false || (row.phase !== "completed" && row.phase !== "failed")) + return null; return alertAllowedForPhase(input.preferences, row.phase) ? alertForActivityRows([row]) : null; } export function shouldAlertForActivity(input: { + readonly unread?: boolean | undefined; readonly phase: RelayAgentActivityAggregateRow["phase"]; readonly updatedAt: string; readonly preferences: RelayAgentAwarenessPreferences | null; @@ -154,6 +158,7 @@ export function shouldAlertForActivity(input: { }): boolean { return ( input.preferences?.notificationsEnabled === true && + (!(input.phase === "completed" || input.phase === "failed") || input.unread !== false) && alertAllowedForPhase(input.preferences, input.phase) && ((input.phase !== "completed" && input.phase !== "failed") || isFreshTerminalNotification(input.updatedAt, input.nowMs)) diff --git a/infra/relay/src/agentActivity/agentActivityPayloads.ts b/infra/relay/src/agentActivity/agentActivityPayloads.ts index 4195c81171803..dd71d920df622 100644 --- a/infra/relay/src/agentActivity/agentActivityPayloads.ts +++ b/infra/relay/src/agentActivity/agentActivityPayloads.ts @@ -15,6 +15,8 @@ export function isTerminalPhase(state: RelayAgentActivityState): boolean { // Dead environments never publish terminal states; expire their rows so activeCount cannot remain inflated. const RUNNING_AGENT_ACTIVITY_ROW_TTL_MS = 2 * 60 * 60 * 1_000; const WAITING_AGENT_ACTIVITY_ROW_TTL_MS = 24 * 60 * 60 * 1_000; +// Unread results stay until read, but a missed tombstone must not pin them forever. +export const RETAINED_AGENT_ACTIVITY_ROW_TTL_MS = WAITING_AGENT_ACTIVITY_ROW_TTL_MS; export function agentActivityExpiresAt( state: Pick, diff --git a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts index e73cfd85145a5..ee78b945c2387 100644 --- a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts +++ b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts @@ -31,6 +31,111 @@ const aggregate = (states: RelayAgentActivityState[]) => makeAggregateState({ activeStates: states, terminalState: null, nowMs: 0 })!; describe("shared agent activity policy", () => { + it("prioritizes actionable results, counts capped rows, and preserves read results", () => { + const make = ( + id: string, + phase: RelayAgentActivityState["phase"], + unread: boolean, + updatedAt = state.updatedAt, + ) => ({ ...state, threadId: ThreadId.make(id), phase, unread, updatedAt, retained: true }); + const states = [ + make("working", "running", false), + make("read", "completed", false), + make("done-old", "completed", true, "1969-12-31T23:59:00.000Z"), + make("done-new", "completed", true), + make("failed", "failed", true), + make("input", "waiting_for_input", false), + make("woke", "woke", false), + ]; + expect(aggregate(states)).toMatchObject({ + activeCount: 3, + workingCount: 1, + attentionCount: 2, + unreadCount: 2, + pendingCount: 6, + totalCount: 7, + activities: [ + { threadId: "input" }, + { threadId: "woke", status: "Woke" }, + { threadId: "failed" }, + { threadId: "done-new" }, + { threadId: "done-old" }, + ], + }); + expect( + aggregate(states.filter((s) => s.threadId !== "done-old" && s.threadId !== "woke")), + ).toMatchObject({ + totalCount: 5, + activities: [ + { threadId: "input" }, + { threadId: "failed" }, + { threadId: "done-new" }, + { threadId: "working" }, + { threadId: "read", unread: false }, + ], + }); + expect(aggregate([make("read", "completed", false)])).toMatchObject({ + totalCount: 1, + activeCount: 0, + unreadCount: 0, + pendingCount: 0, + }); + }); + + it.each(["completed", "failed"] as const)( + "keeps an unread modern %s pending alongside legacy results", + (phase) => { + const modern = { + ...state, + threadId: ThreadId.make("modern"), + phase, + unread: true, + retained: true, + }; + const legacy = { ...state, phase: "completed" as const }; + const mixed = aggregate([modern, legacy]); + expect(mixed.pendingCount).toBe(2); + expect(mixed.activeCount).toBe(0); + expect(aggregate([legacy]).pendingCount).toBeUndefined(); + }, + ); + + it("does not notify for already read results or marking a result unread again", () => { + expect(shouldAlertForActivity({ ...state, phase: "woke", preferences, nowMs: 0 })).toBe(false); + const woke = { ...state, phase: "woke" as const, retained: true }; + expect( + makeAggregateState({ activeStates: [woke], terminalState: null, nowMs: 60 * 60_000 }), + ).toMatchObject({ workingCount: 0, attentionCount: 1, pendingCount: 1 }); + // A missed tombstone cannot pin a retained row forever. + for (const retained of [woke, { ...state, phase: "completed" as const, retained: true }]) { + expect( + makeAggregateState({ + activeStates: [retained], + terminalState: null, + nowMs: 2 * 24 * 60 * 60_000, + }), + ).toBeNull(); + } + const done = { ...state, phase: "completed" as const, unread: false }; + expect(shouldAlertForActivity({ ...done, preferences, nowMs: 0 })).toBe(false); + expect( + terminalTransitionRows({ + previousAggregate: aggregate([state]), + nextAggregate: aggregate([done]), + preferences, + nowMs: 0, + }), + ).toEqual([]); + expect( + terminalTransitionRows({ + previousAggregate: aggregate([done]), + nextAggregate: aggregate([{ ...done, unread: true }]), + preferences, + nowMs: 0, + }), + ).toEqual([]); + }); + it.each(["waiting_for_approval", "waiting_for_input"] as const)( "keeps an older %s ahead of five running rows", (phase) => { diff --git a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts index b732507a731a5..afcb1754cc309 100644 --- a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts +++ b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts @@ -60,6 +60,9 @@ export const ApnsDeliveryJobPayload = Schema.Struct({ aggregate: Schema.NullOr(RelayAgentActivityAggregateState), notification: Schema.NullOr(ApnsNotificationPayload), alert: Schema.optional(Schema.NullOr(ApnsLiveActivityAlert)), + replayThread: Schema.optional( + Schema.Struct({ environmentId: Schema.String, threadId: Schema.String }), + ), createdAt: Schema.String, expiresAt: Schema.String, }); @@ -235,6 +238,7 @@ export function makeApnsDeliveryJobPayload(input: { readonly aggregate: ApnsDeliveryJobPayload["aggregate"]; readonly notification?: ApnsNotificationPayload | null; readonly alert?: ApnsLiveActivityAlert | null | undefined; + readonly replayThread?: ApnsDeliveryJobPayload["replayThread"]; readonly createdAt: string; readonly expiresAt: string; readonly jobId: string; @@ -254,6 +258,7 @@ export function makeApnsDeliveryJobPayload(input: { notification: input.notification ?? null, // Omit absent fields to preserve signatures from older relay builds. ...(input.alert ? { alert: input.alert } : {}), + ...(input.replayThread ? { replayThread: input.replayThread } : {}), createdAt: input.createdAt, expiresAt: input.expiresAt, }; diff --git a/infra/relay/src/agentActivity/fcmPayloads.ts b/infra/relay/src/agentActivity/fcmPayloads.ts index 6c813235e4f61..b78cd8944111c 100644 --- a/infra/relay/src/agentActivity/fcmPayloads.ts +++ b/infra/relay/src/agentActivity/fcmPayloads.ts @@ -1,43 +1,92 @@ -import type { RelayAgentActivityAggregateState } from "@t3tools/contracts/relay"; +import type { + RelayAgentActivityAggregateState, + RelayAgentActivityState, +} from "@t3tools/contracts/relay"; import { activityPhasePriority, TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS, } from "./agentActivityAggregate.ts"; -import { agentActivityExpiresAt } from "./agentActivityPayloads.ts"; +import { + agentActivityExpiresAt, + RETAINED_AGENT_ACTIVITY_ROW_TTL_MS, +} from "./agentActivityPayloads.ts"; -export function androidActivityData(aggregate: RelayAgentActivityAggregateState | null) { +export function androidActivityData( + aggregate: RelayAgentActivityAggregateState | null, + nowMs = Date.parse(aggregate?.updatedAt ?? ""), + lifecycleStates?: ReadonlyArray, +) { const rows = [...(aggregate?.activities ?? [])].sort( - (a, b) => activityPhasePriority(a.phase) - activityPhasePriority(b.phase), + (a, b) => + activityPhasePriority(a.phase, a.unread) - activityPhasePriority(b.phase, b.unread) || + b.updatedAt.localeCompare(a.updatedAt), ); const activeCount = aggregate?.activeCount ?? 0; - const attentionCount = rows.filter((row) => activityPhasePriority(row.phase) === 0).length; + const attentionCount = + aggregate?.attentionCount ?? + rows.filter((row) => activityPhasePriority(row.phase) === 0).length; + const workingCount = aggregate?.workingCount ?? Math.max(0, activeCount - attentionCount); + const unreadCount = aggregate?.unreadCount ?? 0; + // Native dismissal resets on an inactive-to-active edge, so unread results must not hold `active`. + const active = activeCount > 0; const failed = rows.some((row) => row.phase === "failed"); const clean = (value: string) => value.replace(/\s+/g, " ").trim(); const lines = rows.map((row) => - [row.status, clean(row.threadTitle), clean(row.projectTitle)].join("\t"), + [ + row.phase === "completed" && row.unread === false + ? elapsedLabel(nowMs - Date.parse(row.completedAt ?? row.updatedAt)) + : row.status, + clean(row.threadTitle), + clean(row.projectTitle), + ].join("\t"), ); const hero = rows[0]; - const title = - activeCount > 0 - ? `${activeCount} active agent${activeCount === 1 ? "" : "s"}${attentionCount ? ` · ${attentionCount} need${attentionCount === 1 ? "s" : ""} attention` : ""}` - : failed - ? "Agent work failed" - : "Agent work completed"; + const title = active + ? [ + workingCount ? `${workingCount} working` : "", + attentionCount ? `${attentionCount} need${attentionCount === 1 ? "s" : ""} attention` : "", + unreadCount ? `${unreadCount} done` : "", + ] + .filter(Boolean) + .join(" · ") + : failed + ? "Agent work failed" + : unreadCount + ? `${unreadCount} done` + : "All caught up"; + // Released Android builds cancel the card on a past or zero expiry, and terminal identity includes it. const expiresAt = Math.max( 0, - ...rows.map((row) => + ...(lifecycleStates ?? rows).map((row) => row.phase === "completed" || row.phase === "failed" - ? Date.parse(row.updatedAt) + TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS + ? row.unread === false + ? 0 + : Date.parse(row.updatedAt) + + // Match the relay's retention so a kept unread result does not time out first. + ("retained" in row && row.retained === true + ? RETAINED_AGENT_ACTIVITY_ROW_TTL_MS + : TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS) : agentActivityExpiresAt(row), ), ); return { - active: String(activeCount > 0), - activity_chip: activeCount > 0 ? (attentionCount > 0 ? "Review" : "Active") : "", + active: String(active), + activity_chip: active + ? attentionCount > 0 + ? "Review" + : unreadCount > 0 + ? "Done" + : "Active" + : "", activity_title: title, activity_phase: hero?.phase ?? "", activity_active_count: String(activeCount), + activity_working_count: String(workingCount), + ...(aggregate?.totalCount === undefined + ? {} + : { activity_total_count: String(aggregate.totalCount) }), activity_attention_count: String(attentionCount), + activity_unread_count: String(aggregate?.unreadCount ?? 0), activity_body: hero ? `${hero.status}: ${clean(hero.threadTitle)} · ${clean(hero.projectTitle)}` : "", @@ -47,6 +96,18 @@ export function androidActivityData(aggregate: RelayAgentActivityAggregateState }; } +function elapsedLabel(ageMs: number): string { + if (!Number.isFinite(ageMs)) return "Read"; + const minutes = Math.max(0, Math.floor(ageMs / 60_000)); + return minutes < 1 + ? "now" + : minutes < 60 + ? `${minutes}m` + : minutes < 1_440 + ? `${Math.floor(minutes / 60)}h` + : `${Math.floor(minutes / 1_440)}d`; +} + /** Keep Unicode, escaping and grouped alerts within FCM's 4 KB data budget. */ export function fitFcmData(input: Readonly>): Record { const data = { ...input }; diff --git a/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts b/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts index e50e61810aad3..f57c2e0ec2436 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts @@ -138,6 +138,22 @@ describe("EnvironmentPublishSignatures", () => { ); }); + it.effect("rejects an unsigned backfill flag", () => + Effect.gen(function* () { + const request = yield* freshRequest; + const signatures = yield* EnvironmentPublishSignatures.EnvironmentPublishSignatures; + const result = yield* Effect.result( + signatures.verify({ + environmentId: state.environmentId, + environmentPublicKey: keyPair.publicKey, + threadId: state.threadId, + request: { ...request, replay: true }, + }), + ); + expect(Result.isFailure(result)).toBe(true); + }).pipe(Effect.provide(layer())), + ); + it.effect("rejects top-level state tampering", () => Effect.gen(function* () { const request = yield* freshRequest; diff --git a/infra/relay/src/environments/EnvironmentPublishSignatures.ts b/infra/relay/src/environments/EnvironmentPublishSignatures.ts index 625ce46ce4d43..bbda36cca35ca 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.ts @@ -148,6 +148,7 @@ const make = Effect.gen(function* () { proof.threadId !== input.threadId || proof.sub !== input.environmentId || stableStringify(proof.state) !== stableStringify(input.request.state) || + (proof.replay ?? false) !== (input.request.replay ?? false) || (input.request.state !== null && (input.request.state.environmentId !== input.environmentId || input.request.state.threadId !== input.threadId)) diff --git a/infra/relay/src/http/Api.test.ts b/infra/relay/src/http/Api.test.ts index 3abb8b28f4d6a..7a23fceb7ff25 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, + replay: 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..bbf0da7dff469 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -1227,6 +1227,7 @@ export const layerServerApi = HttpApiBuilder.group( environmentPublicKey: principal.environmentPublicKey, threadId: params.threadId, state: payload.state, + replay: payload.replay ?? false, }); }, mapErrorTags({ diff --git a/infra/relay/src/worker.ts b/infra/relay/src/worker.ts index e9da9a216726f..3f4c19009d811 100644 --- a/infra/relay/src/worker.ts +++ b/infra/relay/src/worker.ts @@ -487,6 +487,7 @@ export const layer = Api.make( Effect.flatMap(([activityRows, now]) => activityRows.pruneTerminal({ updatedBefore: DateTime.formatIso(DateTime.subtract(now, { minutes: 30 })), + retainedBefore: DateTime.formatIso(DateTime.subtract(now, { hours: 25 })), }), ), ), diff --git a/packages/contracts/src/relay.ts b/packages/contracts/src/relay.ts index 8ebc9df23a0bb..72e845a852eb9 100644 --- a/packages/contracts/src/relay.ts +++ b/packages/contracts/src/relay.ts @@ -27,6 +27,7 @@ export const RelayAgentAwarenessPhase = Schema.Literals([ "completed", "failed", "stale", + "woke", ]); export type RelayAgentAwarenessPhase = typeof RelayAgentAwarenessPhase.Type; @@ -122,7 +123,10 @@ export const RelayAgentActivityState = Schema.Struct({ projectTitle: TrimmedNonEmptyString, threadTitle: TrimmedNonEmptyString, phase: RelayAgentAwarenessPhase, + unread: Schema.optional(Schema.Boolean), + completedAt: Schema.optional(TrimmedNonEmptyString), headline: TrimmedNonEmptyString, + retained: Schema.optional(Schema.Boolean), detail: Schema.optional(TrimmedNonEmptyString), modelTitle: TrimmedNonEmptyString, updatedAt: TrimmedNonEmptyString, @@ -137,6 +141,8 @@ export const RelayAgentActivityAggregateRow = Schema.Struct({ threadTitle: TrimmedNonEmptyString, modelTitle: TrimmedNonEmptyString, phase: RelayAgentAwarenessPhase, + unread: Schema.optional(Schema.Boolean), + completedAt: Schema.optional(TrimmedNonEmptyString), status: TrimmedNonEmptyString, updatedAt: TrimmedNonEmptyString, deepLink: TrimmedNonEmptyString, @@ -147,6 +153,11 @@ export const RelayAgentActivityAggregateState = Schema.Struct({ title: TrimmedNonEmptyString, subtitle: TrimmedNonEmptyString, activeCount: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), + workingCount: Schema.optional(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), + attentionCount: Schema.optional(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), + totalCount: Schema.optional(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), + unreadCount: Schema.optional(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), + pendingCount: Schema.optional(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), updatedAt: TrimmedNonEmptyString, activities: Schema.Array(RelayAgentActivityAggregateRow), }); @@ -263,11 +274,13 @@ export const RelayAgentActivityPublishProofPayload = Schema.Struct({ environmentId: EnvironmentId, threadId: ThreadId, state: Schema.NullOr(RelayAgentActivityState), + replay: Schema.optional(Schema.Boolean), }); export type RelayAgentActivityPublishProofPayload = typeof RelayAgentActivityPublishProofPayload.Type; export const RelayAgentActivityPublishRequest = Schema.Struct({ + replay: Schema.optional(Schema.Boolean), state: Schema.NullOr(RelayAgentActivityState).annotate({ description: "Current agent-awareness state, or null to remove the published state.", }), diff --git a/packages/shared/src/agentAwareness.test.ts b/packages/shared/src/agentAwareness.test.ts index 71f4f21bd1e2a..09bd8062e55fd 100644 --- a/packages/shared/src/agentAwareness.test.ts +++ b/packages/shared/src/agentAwareness.test.ts @@ -49,6 +49,103 @@ describe("projectThreadAwarenessV2", () => { ...overrides, }); + it.each(["completed", "cancelled", "interrupted", "idle", "queued", "rolled_back"] as const)( + "restores a snoozed %s inbox row until its wake is acknowledged", + (status) => { + const snoozedUntil = DateTime.makeUnsafe("2026-05-22T13:00:00.000Z"); + const base = { + environmentId: "env-1" as EnvironmentId, + project, + now: Date.parse("2026-05-22T13:01:00.000Z"), + thread: { ...v2Thread({ status }), snoozedUntil, lastVisitedAt: updatedAt }, + }; + expect(projectThreadAwarenessV2(base)).toMatchObject({ + phase: "woke", + headline: "Snooze ended", + retained: true, + updatedAt: "2026-05-22T13:00:00.000Z", + }); + expect(projectThreadAwarenessV2({ ...base, now: Date.parse(NOW) })).toBeNull(); + expect( + projectThreadAwarenessV2({ ...base, thread: { ...base.thread, archivedAt: updatedAt } }), + ).toBeNull(); + expect( + projectThreadAwarenessV2({ + ...base, + thread: { ...base.thread, settledOverride: "settled" }, + }), + ).toBeNull(); + if (status === "completed") { + expect( + projectThreadAwarenessV2({ + ...base, + thread: { ...base.thread, latestRunCompletedAt: snoozedUntil }, + }), + ).toMatchObject({ phase: "completed", unread: true }); + } + for (const visit of [snoozedUntil, DateTime.makeUnsafe("2026-05-22T13:01:00.000Z")]) { + expect( + projectThreadAwarenessV2({ ...base, thread: { ...base.thread, lastVisitedAt: visit } }) + ?.phase ?? null, + ).toBe(status === "completed" ? "completed" : null); + } + expect( + projectThreadAwarenessV2({ ...base, thread: { ...base.thread, snoozedUntil: null } }) + ?.phase ?? null, + ).toBe(status === "completed" ? "completed" : null); + expect( + projectThreadAwarenessV2({ ...base, thread: { ...base.thread, lastVisitedAt: null } }) + ?.phase, + ).toBe("woke"); + }, + ); + + it.each([ + { visit: null, unread: false }, + { visit: "2026-05-22T11:59:00.000Z", unread: true }, + { visit: NOW, unread: false }, + { visit: "2026-05-22T12:01:00.000Z", unread: false }, + ])("projects the desktop completion read state: $visit", ({ visit, unread }) => { + expect( + projectThreadAwarenessV2({ + environmentId: "env-1" as EnvironmentId, + project, + thread: { + ...v2Thread({ status: "completed" }), + latestRunCompletedAt: updatedAt, + lastVisitedAt: visit === null ? null : DateTime.makeUnsafe(visit), + }, + }), + ).toMatchObject({ phase: "completed", unread, completedAt: NOW, retained: true }); + }); + + it.each([ + { kind: "user_input", createdAt: "2026-05-22T11:59:00.000Z", phase: null }, + { kind: "user_input", createdAt: "2026-05-22T12:01:00.000Z", phase: "waiting_for_input" }, + { kind: "command", createdAt: "2026-05-22T11:59:00.000Z", phase: "waiting_for_approval" }, + ] as const)( + "matches the inbox for a $kind request created $createdAt during a snooze", + ({ kind, createdAt, phase }) => { + const state = projectThreadAwarenessV2({ + environmentId: "env-1" as EnvironmentId, + project, + now: Date.parse("2026-05-22T12:02:00.000Z"), + thread: { + ...v2Thread({ + pendingRuntimeRequest: { + id: RuntimeRequestId.make("request-snoozed"), + kind, + createdAt: DateTime.makeUnsafe(createdAt), + }, + }), + snoozedAt: updatedAt, + snoozedUntil: DateTime.makeUnsafe("2026-05-22T14:00:00.000Z"), + }, + }); + expect(state?.phase ?? null).toBe(phase); + }, + ); + it("does not announce completion after a manual stop", () => { const input = { environmentId: "env-1" as EnvironmentId, diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index 55ad807957d64..87f996928d591 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -21,7 +21,8 @@ export type AgentAwarenessPhase = | "waiting_for_input" | "completed" | "failed" - | "stale"; + | "stale" + | "woke"; export interface SubagentRequestAwareness { readonly requestId: RuntimeRequestId; @@ -55,6 +56,9 @@ export interface AgentAwarenessState { readonly threadTitle: string; readonly phase: AgentAwarenessPhase; readonly headline: string; + readonly retained?: boolean; + readonly unread?: boolean; + readonly completedAt?: string; readonly detail?: string; readonly modelTitle: string; readonly updatedAt: string; @@ -71,21 +75,28 @@ function buildAgentAwarenessDeepLink(input: { export interface ProjectThreadAwarenessV2Input { readonly environmentId: EnvironmentId; readonly project: Pick; - readonly thread: Pick< - OrchestrationV2ThreadShell, - | "activityRunStatus" - | "id" - | "lineage" - | "latestRunBackgroundWake" - | "latestRunStopRequested" - | "modelSelection" - | "pendingBackgroundTasks" - | "pendingRuntimeRequest" - | "settledOverride" - | "status" - | "title" - | "updatedAt" - >; + readonly now?: number; + readonly thread: Partial< + Pick< + OrchestrationV2ThreadShell, + "archivedAt" | "snoozedUntil" | "snoozedAt" | "latestRunCompletedAt" | "lastVisitedAt" + > + > & + Pick< + OrchestrationV2ThreadShell, + | "activityRunStatus" + | "id" + | "lineage" + | "latestRunBackgroundWake" + | "latestRunStopRequested" + | "modelSelection" + | "pendingBackgroundTasks" + | "pendingRuntimeRequest" + | "settledOverride" + | "status" + | "title" + | "updatedAt" + >; /** The approval a live subagent below this thread is waiting on. */ readonly pendingSubagentRequest?: SubagentRequestAwareness | null; } @@ -94,9 +105,39 @@ export function projectThreadAwarenessV2( input: ProjectThreadAwarenessV2Input, ): AgentAwarenessState | null { const { environmentId, project, thread } = input; - if (!isListedThreadLineage(thread.lineage)) return null; + if (!isListedThreadLineage(thread.lineage) || thread.archivedAt != null) return null; const subagentRequest = input.pendingSubagentRequest ?? null; - const phase = resolveThreadAwarenessPhaseV2(thread, subagentRequest !== null); + if ( + input.now !== undefined && + thread.snoozedUntil != null && + DateTime.toEpochMillis(thread.snoozedUntil) > input.now + ) { + const failedAfterSnooze = + (thread.activityRunStatus ?? thread.status) === "failed" && + (thread.snoozedAt == null || + (thread.latestRunCompletedAt != null && + DateTime.toEpochMillis(thread.latestRunCompletedAt) > + DateTime.toEpochMillis(thread.snoozedAt))); + // Like the inbox, snoozing an existing question defers it; new questions and approvals wake. + const request = thread.pendingRuntimeRequest; + const requestWakes = + request != null && + (request.kind !== "user_input" || + thread.snoozedAt == null || + DateTime.toEpochMillis(request.createdAt) > DateTime.toEpochMillis(thread.snoozedAt)); + if (!requestWakes && subagentRequest === null && !failedAfterSnooze) return null; + } + const snoozeWakeAt = thread.snoozedUntil; + const woke = + input.now !== undefined && + snoozeWakeAt != null && + DateTime.toEpochMillis(snoozeWakeAt) <= input.now && + (thread.lastVisitedAt == null || + DateTime.toEpochMillis(thread.lastVisitedAt) < DateTime.toEpochMillis(snoozeWakeAt)) && + (thread.latestRunCompletedAt == null || + DateTime.toEpochMillis(thread.latestRunCompletedAt) < DateTime.toEpochMillis(snoozeWakeAt)); + const runPhase = resolveThreadAwarenessPhaseV2(thread, subagentRequest !== null); + const phase = woke && (runPhase === null || runPhase === "completed") ? "woke" : runPhase; if (phase === null) { return null; } @@ -107,11 +148,20 @@ export function projectThreadAwarenessV2( thread.settledOverride === "settled" && (phase === "completed" || phase === "failed" || + phase === "woke" || (phase === "running" && runStatus === "completed")) && !backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? []) ) { return null; } + const completedAt = + phase === "completed" || phase === "failed" ? thread.latestRunCompletedAt : null; + const unread = + (phase === "completed" || phase === "failed") && + (thread.lastVisitedAt == null + ? phase === "failed" + : completedAt != null && + DateTime.toEpochMillis(completedAt) > DateTime.toEpochMillis(thread.lastVisitedAt)); const detail = phase === "completed" ? "Review the completed task." @@ -125,14 +175,19 @@ export function projectThreadAwarenessV2( threadTitle: thread.title, phase, headline: headlineForPhase(phase), + retained: true, + unread, + ...(completedAt == null ? {} : { completedAt: DateTime.formatIso(completedAt) }), ...(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), + phase === "woke" && snoozeWakeAt != null + ? snoozeWakeAt + : subagentRequest === null + ? thread.updatedAt + : DateTime.max(thread.updatedAt, subagentRequest.updatedAt), ), deepLink: buildAgentAwarenessDeepLink({ environmentId, threadId: thread.id }), }; @@ -191,5 +246,7 @@ function headlineForPhase(phase: AgentAwarenessPhase): string { return "Agent failed"; case "stale": return "Update delayed"; + case "woke": + return "Snooze ended"; } }