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 19d14f450f526..cbac6c76806f0 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 @@ -128,7 +128,11 @@ object AgentNotifications { channels(context) val scheme = prefs.getString("scheme", "t3code") ?: "t3code" val alerted = showAlert(context, prefs, scheme, data) - updateActivity(context, prefs, data, updatedAt, alerted) + // A one-shot alert carries no activity transition. Missing activity + // fields would otherwise end the card. + if (data["alert_only"] != "true") { + updateActivity(context, prefs, data, updatedAt, alerted) + } } } 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 59ab4d6651e1a..f6671179a4c15 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 @@ -104,6 +104,25 @@ class AgentNotificationsTest : AgentNotificationsFixture() { assertEquals("t3-agent-alert", manager.activeNotifications.single().tag) } + @Test + fun alertOnlyLeavesTheActivityCardInPlace() { + AgentNotifications.receive(context, update("activity", true)) + val card = manager.activeNotifications.single { it.tag == "t3-agent-activity" } + + val alert = update("ping", true).toMutableMap() + alert["alert_only"] = "true" + alert.remove("active") + alert.remove("activity_expires_at") + AgentNotifications.receive(context, alert) + + val activity = manager.activeNotifications.single { it.tag == "t3-agent-activity" } + assertEquals(card.id, activity.id) + assertEquals( + "t3-agent-alert", + manager.activeNotifications.single { it.tag == "t3-agent-alert" }.tag + ) + } + @Test fun foregroundSuppressesAlertsWhileOngoingActivityStillUpdatesAndClears() { lifecycle.currentState = Lifecycle.State.RESUMED diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 3ba7af66de98a..b7da283579874 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -112,6 +112,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverRefreshUsageRates]: AuthDiagnosticsReadScope, [WS_METHODS.serverSignalProcess]: AuthEnvironmentMaintainScope, [WS_METHODS.serverReportClientActivity]: AuthOrchestrationReadScope, + [WS_METHODS.serverSubscribeAgentAlerts]: AuthOrchestrationReadScope, [WS_METHODS.scheduledTasksSubscribe]: AuthOrchestrationReadScope, [WS_METHODS.secretsAnswerRequest]: AuthOrchestrationOperateScope, // Delivery logs hold request bodies, so they need the same scope as the URL. diff --git a/apps/server/src/notifications/AgentAlerts.test.ts b/apps/server/src/notifications/AgentAlerts.test.ts new file mode 100644 index 0000000000000..f0936e9884ba0 --- /dev/null +++ b/apps/server/src/notifications/AgentAlerts.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "@effect/vitest"; +import { type AgentAlert, ThreadId } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; + +import * as AgentAlerts from "./AgentAlerts.ts"; + +const alert = { + threadId: ThreadId.make("thread-1"), + body: "fleet.lock stuck", +} satisfies AgentAlert; + +describe("AgentAlerts", () => { + it.effect("reports no channel when nobody is listening", () => + Effect.gen(function* () { + const alerts = yield* AgentAlerts.AgentAlerts; + expect(yield* alerts.offer(alert)).toEqual({ local: false, phone: false }); + yield* alerts.registerPhone({ + publish: () => Effect.succeed(false), + }); + expect(yield* alerts.offer(alert)).toEqual({ local: false, phone: false }); + }).pipe(Effect.provide(AgentAlerts.layer)), + ); + + it.effect("delivers to a connected client and a registered phone", () => + Effect.gen(function* () { + const alerts = yield* AgentAlerts.AgentAlerts; + const received = yield* Queue.unbounded(); + const published = yield* Deferred.make(); + yield* alerts.stream.pipe( + Stream.runForEach((item) => Queue.offer(received, item)), + Effect.forkScoped({ startImmediately: true }), + ); + yield* alerts.registerPhone({ + publish: (item) => Deferred.succeed(published, item).pipe(Effect.as(true)), + }); + expect(yield* alerts.offer(alert)).toEqual({ local: true, phone: true }); + expect(yield* Queue.take(received)).toEqual(alert); + yield* alerts.registerPhone({ publish: () => Effect.die("relay unavailable") }); + expect(yield* alerts.offer(alert)).toEqual({ local: true, phone: false }); + expect(yield* Queue.take(received)).toEqual(alert); + expect(yield* Deferred.await(published)).toEqual(alert); + }).pipe(Effect.provide(AgentAlerts.layer), Effect.scoped), + ); +}); diff --git a/apps/server/src/notifications/AgentAlerts.ts b/apps/server/src/notifications/AgentAlerts.ts new file mode 100644 index 0000000000000..ac8851e3e29c6 --- /dev/null +++ b/apps/server/src/notifications/AgentAlerts.ts @@ -0,0 +1,68 @@ +import type { AgentAlert } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; + +import type { ClaudePushChannels } from "../orchestration-v2/Adapters/claudePushNotification.ts"; + +export interface AgentAlertPhoneDelivery { + readonly publish: (alert: AgentAlert) => Effect.Effect; +} + +export class AgentAlerts extends Context.Service< + AgentAlerts, + { + readonly offer: (alert: AgentAlert) => Effect.Effect; + readonly stream: Stream.Stream; + readonly registerPhone: (delivery: AgentAlertPhoneDelivery) => Effect.Effect; + } +>()("t3/notifications/AgentAlerts") {} + +const make = Effect.gen(function* () { + const subscribers = yield* Ref.make(0); + const alerts = yield* PubSub.unbounded(); + const phone = yield* Ref.make(null); + + const stream = Stream.unwrap( + Effect.gen(function* () { + // Subscribe first. Counting before the subscription exists tells Claude the + // alert was sent in a window where publish has nobody to deliver to. + const subscription = yield* PubSub.subscribe(alerts); + yield* Ref.update(subscribers, (count) => count + 1); + yield* Effect.addFinalizer(() => + Ref.update(subscribers, (count) => (count > 0 ? count - 1 : 0)), + ); + return Stream.fromSubscription(subscription); + }), + ); + + const offer: AgentAlerts["Service"]["offer"] = Effect.fn("AgentAlerts.offer")(function* (alert) { + const delivery = yield* Ref.get(phone); + const local = (yield* Ref.get(subscribers)) > 0; + if (local) yield* PubSub.publish(alerts, alert); + const phoneAccepted = + delivery === null + ? false + : yield* delivery + .publish(alert) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("agent alert phone delivery failed", { cause }).pipe( + Effect.as(false), + ), + ), + ); + return { local, phone: phoneAccepted }; + }); + + return AgentAlerts.of({ + offer, + stream, + registerPhone: Effect.fn("AgentAlerts.registerPhone")((delivery) => Ref.set(phone, delivery)), + }); +}); + +export const layer = Layer.effect(AgentAlerts, make); diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 89f93a9df2c81..a408db64896ff 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -89,6 +89,7 @@ const RPC_AGGREGATES = { [WS_METHODS.serverRefreshUsageRates]: "server", [WS_METHODS.serverSignalProcess]: "server", [WS_METHODS.serverReportClientActivity]: "server", + [WS_METHODS.serverSubscribeAgentAlerts]: "server", [WS_METHODS.scheduledTasksSubscribe]: "scheduledTasks", [WS_METHODS.scheduledTasksUpsert]: "scheduledTasks", [WS_METHODS.scheduledTasksSetEnabled]: "scheduledTasks", diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 70f1471ef54a7..34c5bb9200ebb 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -753,6 +753,8 @@ describe("ClaudeAdapterV2 device instructions", () => { ]); assert.deepEqual(await subagentContext({ mcp: true, deviceToolsAvailable: false }), []); assert.deepEqual(await subagentContext({ mcp: false, deviceToolsAvailable: true }), []); + assert.isArray(options({ mcp: true, deviceToolsAvailable: true }).hooks?.PostToolUse); + assert.isArray(options({ mcp: false, deviceToolsAvailable: false }).hooks?.PostToolUse); }); }); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 331408a21c4a0..dd354c695c095 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6,6 +6,8 @@ import { } from "@t3tools/shared/toolOutput"; import { makeProviderTextDeltaCoalescer } from "./ProviderTextDeltaCoalescer.ts"; import { claudeTaskOutputRoot, tailClaudeTaskOutput } from "./ClaudeTaskOutputTail.ts"; +import { claudePushNotificationHooks } from "./claudePushNotification.ts"; +import * as AgentAlerts from "../../notifications/AgentAlerts.ts"; import { dynamicToolTitle, formatReadToolLabel, @@ -1775,6 +1777,10 @@ export function makeClaudeQueryOptions(input: { readonly appendSystemPromptFileText?: string; // Flag settings, because a user's settings.json env beats the process env in Claude Code. readonly toolSearchOverride?: "true"; + readonly pushNotificationOffer?: (message: string) => Promise<{ + readonly local: boolean; + readonly phone: boolean; + }>; }): ClaudeAgentSdkQueryOptions { const compiledSelection = compileClaudeModelSelection(input.modelSelection, input.modelCatalog); const { @@ -1879,6 +1885,12 @@ export function makeClaudeQueryOptions(input: { : {}), ...(input.environment === undefined ? {} : { env: input.environment }), ...(input.mcpServers === undefined ? {} : { mcpServers: input.mcpServers }), + hooks: { + ...claudePushNotificationHooks({ + offer: input.pushNotificationOffer ?? (async () => ({ local: false, phone: false })), + }), + ...(deviceInstructions ? { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } : {}), + }, systemPrompt: { type: "preset" as const, preset: "claude_code" as const, @@ -1894,9 +1906,6 @@ export function makeClaudeQueryOptions(input: { ...launchAppendSystemPrompt, ].join("\n\n"), }, - ...(deviceInstructions - ? { hooks: { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } } - : {}), ...(Object.keys(extraArgs).length === 0 ? {} : { extraArgs }), }; const additionalDirectories = [ @@ -4851,6 +4860,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( adapterOptions: ClaudeAdapterV2Options, ) { const mcpSessions = yield* McpProviderSessions.McpProviderSessions; + const agentAlerts = yield* Effect.serviceOption(AgentAlerts.AgentAlerts); const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const serverSettings = yield* Effect.serviceOption(ServerSettings.ServerSettingsService); const { attachmentsDir, fileSystem, path, random, idAllocator, queryRunner } = adapterOptions; @@ -10901,6 +10911,17 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( supportedDialogKinds: ["resume_return"], ...(appendSystemPromptFileText === undefined ? {} : { appendSystemPromptFileText }), ...(toolSearchOverride === undefined ? {} : { toolSearchOverride }), + pushNotificationOffer: (message) => + runPromise( + Option.match(agentAlerts, { + onNone: () => Effect.succeed({ local: false, phone: false }), + onSome: (alerts) => + alerts.offer({ + threadId: turnInput.threadId, + body: message, + }), + }), + ), }); const resolvedTurnCwd = yield* fileSystem .realPath(turnInput.runtimePolicy.cwd ?? ".") diff --git a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts new file mode 100644 index 0000000000000..f7ccc6a1caefa --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts @@ -0,0 +1,113 @@ +import { describe, it } from "@effect/vitest"; +import * as NodeAssert from "node:assert/strict"; + +import { + claudePushNotificationHooks, + claudePushNotificationOutput, + readClaudePushNotificationMessage, +} from "./claudePushNotification.ts"; + +describe("claudePushNotification", () => { + it("keeps a one-line message and caps it at 200 characters", () => { + NodeAssert.equal( + readClaudePushNotificationMessage({ message: " fleet.lock stuck " }), + "fleet.lock stuck", + ); + NodeAssert.equal(readClaudePushNotificationMessage({ message: "line\n\nbreak" }), "line break"); + NodeAssert.equal(readClaudePushNotificationMessage({ message: " " }), null); + NodeAssert.equal(readClaudePushNotificationMessage({}), null); + const long = "x".repeat(240); + const read = readClaudePushNotificationMessage({ message: long }); + NodeAssert.equal(read?.length, 200); + NodeAssert.equal(read?.endsWith("…"), true); + }); + + it("does not tell the agent the terminal is active when T3 cannot deliver", () => { + NodeAssert.deepEqual( + claudePushNotificationOutput({ channels: { local: false, phone: false } }), + { + hookSpecificOutput: { + hookEventName: "PostToolUse", + updatedToolOutput: { + message: "Not sent. No T3 client or phone accepted the notification.", + pushSent: false, + localSent: false, + disabledReason: "no_transport", + }, + }, + }, + ); + }); + + it("rewrites a user_present result when a T3 channel can deliver", async () => { + const hooks = claudePushNotificationHooks({ + offer: async () => ({ local: true, phone: false }), + }); + const hook = hooks.PostToolUse[0]?.hooks[0]; + NodeAssert.ok(hook); + const result = await hook( + { + hook_event_name: "PostToolUse", + tool_name: "PushNotification", + tool_input: { message: "fleet.lock stuck", status: "proactive" }, + tool_response: { + message: "Not sent — this terminal is active.", + pushSent: false, + localSent: false, + disabledReason: "user_present", + }, + tool_use_id: "tool-1", + session_id: "session-1", + transcript_path: "", + cwd: "/workspace", + }, + "tool-1", + { signal: new AbortController().signal }, + ); + NodeAssert.deepEqual(result, { + hookSpecificOutput: { + hookEventName: "PostToolUse", + updatedToolOutput: { + message: "Sent through T3.", + pushSent: false, + localSent: true, + }, + }, + }); + }); + + it("leaves a push Claude already resolved unchanged", async () => { + const offered: Array = []; + const hooks = claudePushNotificationHooks({ + offer: async (message) => { + offered.push(message); + return { local: true, phone: true }; + }, + }); + const hook = hooks.PostToolUse[0]?.hooks[0]; + NodeAssert.ok(hook); + const call = (toolResponse: unknown) => + hook( + { + hook_event_name: "PostToolUse", + tool_name: "PushNotification", + tool_input: { message: "fleet.lock stuck", status: "proactive" }, + tool_response: toolResponse, + tool_use_id: "tool-1", + session_id: "session-1", + transcript_path: "", + cwd: "/workspace", + }, + "tool-1", + { signal: new AbortController().signal }, + ); + for (const toolResponse of [ + { message: "Sent", pushSent: true, localSent: true }, + { message: "Notifications are off.", pushSent: false, disabledReason: "config_off" }, + { message: "No push transport.", pushSent: false, disabledReason: "no_transport" }, + ]) { + NodeAssert.deepEqual(await call(toolResponse), {}); + } + NodeAssert.deepEqual(offered, []); + }); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts new file mode 100644 index 0000000000000..e3061553f2db4 --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts @@ -0,0 +1,100 @@ +import type { HookCallback, SyncHookJSONOutput } from "@anthropic-ai/claude-agent-sdk"; + +const CLAUDE_PUSH_NOTIFICATION_TOOL = "PushNotification"; +const MAX_PUSH_NOTIFICATION_LENGTH = 200; + +export interface ClaudePushChannels { + readonly local: boolean; + readonly phone: boolean; +} + +export interface ClaudePushNotificationOffer { + (message: string): Promise; +} + +/** One line, capped the way Claude's own tool asks the model to write it. */ +export function readClaudePushNotificationMessage(toolInput: unknown): string | null { + if (typeof toolInput !== "object" || toolInput === null || !("message" in toolInput)) { + return null; + } + const message = toolInput.message; + if (typeof message !== "string") return null; + const trimmed = message.trim().replace(/\s+/g, " "); + if (trimmed.length === 0) return null; + return trimmed.length > MAX_PUSH_NOTIFICATION_LENGTH + ? `${trimmed.slice(0, MAX_PUSH_NOTIFICATION_LENGTH - 1)}…` + : trimmed; +} + +export function claudePushNotificationOutput(input: { + readonly channels: ClaudePushChannels; +}): SyncHookJSONOutput { + if (!input.channels.local && !input.channels.phone) { + return { + hookSpecificOutput: { + hookEventName: "PostToolUse", + updatedToolOutput: { + message: "Not sent. No T3 client or phone accepted the notification.", + pushSent: false, + localSent: false, + disabledReason: "no_transport", + }, + }, + }; + } + return { + hookSpecificOutput: { + hookEventName: "PostToolUse", + updatedToolOutput: { + message: "Sent through T3.", + pushSent: input.channels.phone, + localSent: input.channels.local, + }, + }, + }; +} + +const missingMessageOutput: SyncHookJSONOutput = { + hookSpecificOutput: { + hookEventName: "PostToolUse", + updatedToolOutput: { + message: "Not sent. The notification had no message.", + pushSent: false, + localSent: false, + disabledReason: "no_transport", + }, + }, +}; + +function claudeSuppressedPushBecauseUserPresent(toolResponse: unknown): boolean { + return ( + typeof toolResponse === "object" && + toolResponse !== null && + "disabledReason" in toolResponse && + toolResponse.disabledReason === "user_present" + ); +} + +export function claudePushNotificationHooks(input: { + readonly offer: ClaudePushNotificationOffer; +}) { + const hook: HookCallback = async (hookInput) => { + if ( + hookInput.hook_event_name !== "PostToolUse" || + hookInput.tool_name !== CLAUDE_PUSH_NOTIFICATION_TOOL + ) { + return {}; + } + // user_present is the SDK session looking like an open terminal. A real send + // or another refusal has to reach the model unchanged. + if (!claudeSuppressedPushBecauseUserPresent(hookInput.tool_response)) { + return {}; + } + const message = readClaudePushNotificationMessage(hookInput.tool_input); + if (message === null) return missingMessageOutput; + return claudePushNotificationOutput({ channels: await input.offer(message) }); + }; + return { + PostToolUse: [{ matcher: CLAUDE_PUSH_NOTIFICATION_TOOL, hooks: [hook] }], + }; +} diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index e76e2a6b64b05..4be95cbadae9b 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -28,7 +28,6 @@ import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as FetchHttpClient from "effect/http/FetchHttpClient"; -import * as EnvironmentKeys from "../cloud/environmentKeys.ts"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import { PUBLISH_AGENT_ACTIVITY_SECRET, @@ -42,6 +41,7 @@ import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as AgentAwarenessRelay from "./AgentAwarenessRelay.ts"; +import * as AgentAlerts from "../notifications/AgentAlerts.ts"; const THREAD_ID = ThreadId.make("relay-thread"); const SECOND_THREAD_ID = ThreadId.make("relay-thread-2"); @@ -143,9 +143,9 @@ const makeTestRelay = Effect.fnUntraced(function* ( readonly respond?: (attempt: number) => Response; readonly failSecretRead?: (name: string) => boolean; readonly linked?: boolean; + readonly alerts?: AgentAlerts.AgentAlerts["Service"]; readonly readShell?: (threadId: ThreadId) => Effect.Effect; readonly domainEvents?: Stream.Stream; - readonly catchUpThreads?: ReadonlyArray; } = {}, ) { const values = new Map( @@ -180,23 +180,17 @@ const makeTestRelay = Effect.fnUntraced(function* ( }); const currentShell = yield* Ref.make(shell()); const shellReads: ThreadId[] = []; - const catchUp = { activitySnapshotReads: 0 }; + const catchUp = { shellSnapshotReads: 0 }; const threads = ThreadManagementService.ThreadManagementService.of({ - getThreadShell: unused, - getShellSnapshot: unused, - getAgentAwarenessSnapshot: (threadId) => - threadId === undefined - ? Effect.sync(() => { - catchUp.activitySnapshotReads += 1; - return { snapshotSequence: 1, threads: options.catchUpThreads ?? [] }; - }) - : Effect.sync(() => shellReads.push(threadId)).pipe( - Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)), - Effect.map((thread) => ({ - snapshotSequence: 1, - threads: thread === null ? [] : [thread], - })), - ), + getThreadShell: (threadId) => + Effect.sync(() => shellReads.push(threadId)).pipe( + Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)), + ), + getShellSnapshot: () => + Effect.sync(() => { + catchUp.shellSnapshotReads += 1; + return { schemaVersion: 2, snapshotSequence: 1, threads: [], archivedThreads: [] }; + }), listActiveSubagentThreadIds: unused, listAppOwnedChildThreadIds: unused, countThreads: unused, @@ -214,6 +208,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( getTurnItem: () => Effect.die("unused turn item read"), getThreadRecords: () => Effect.die("unused record read"), getThreadProjection: unused, + getAgentAwarenessSnapshot: unused, getCheckpointContext: unused, getThreadSnapshot: unused, getThreadTranscript: unused, @@ -253,9 +248,9 @@ const makeTestRelay = Effect.fnUntraced(function* ( if (typeof body !== "string" && !(body instanceof Uint8Array)) { return Promise.reject(new Error("Expected a serialized activity publish payload")); } - const payload = decodePublishPayload( - typeof body === "string" ? body : new TextDecoder().decode(body), - ); + const payload = String(input).endsWith("/agent-alert") + ? { state: null } + : decodePublishPayload(typeof body === "string" ? body : new TextDecoder().decode(body)); publications.push({ url: String(input), authorization: new Headers(init?.headers).get("authorization"), @@ -267,18 +262,11 @@ const makeTestRelay = Effect.fnUntraced(function* ( }, { preconnect: () => {} }, ); - const project = { - id: PROJECT_ID, - title: "Project", - workspaceRoot: "/workspace", - defaultModelSelection: null, - scripts: [], - createdAt: NOW, - updatedAt: NOW, - deletedAt: null, - }; const relay = yield* AgentAwarenessRelay.make.pipe( - Effect.provide(EnvironmentKeys.layer), + (program) => + options.alerts + ? Effect.provideService(program, AgentAlerts.AgentAlerts, options.alerts) + : program, Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), Effect.provideService(ThreadManagementService.ThreadManagementService, threads), Effect.provideService(ServerEnvironment.ServerEnvironment, { @@ -291,19 +279,79 @@ const makeTestRelay = Effect.fnUntraced(function* ( update: unused, delete: unused, getByWorkspaceRoot: unused, - snapshot: Effect.succeed({ projects: [project], updatedAt: NOW }), + snapshot: Effect.succeed({ projects: [] } as never), listPage: unused, getShell: unused, listShells: unused, - getById: () => Effect.succeedSome(project), + getById: () => + Effect.succeedSome({ + id: PROJECT_ID, + title: "Project", + workspaceRoot: "/workspace", + defaultModelSelection: null, + scripts: [], + createdAt: NOW, + updatedAt: NOW, + deletedAt: null, + }), }), Effect.provideService(FetchHttpClient.Fetch, fetch), Effect.provide(NodeCrypto.layer), ); - return { relay, secrets, secretReads, currentShell, shellReads, publications, catchUp }; + return { relay, secrets, secretReads, currentShell, shellReads, publications, catchUp, fetch }; }); describe("AgentAwarenessRelay", () => { + it.effect("reports phone acceptance only for a successful eligible delivery", () => + Effect.gen(function* () { + const alerts = yield* AgentAlerts.AgentAlerts; + const responses = [ + { ok: true, deliveries: [] }, + { ok: false, deliveries: [] }, + { + ok: true, + deliveries: [ + { + deviceId: "phone", + kind: "push_notification", + ok: false, + apnsStatus: 400, + apnsReason: "BadDeviceToken", + apnsId: null, + }, + ], + }, + { + ok: true, + deliveries: [ + { + deviceId: "phone", + kind: "push_notification", + ok: true, + queued: true, + apnsStatus: null, + apnsReason: null, + apnsId: null, + }, + ], + }, + ]; + const { fetch } = yield* makeTestRelay({ + alerts, + respond: (count) => Response.json(responses[count - 1]), + }); + for (const expected of [false, false, false, true]) { + const result = yield* alerts + .offer({ threadId: THREAD_ID, body: "Build finished" }) + .pipe(Effect.provideService(FetchHttpClient.Fetch, fetch)); + assert.deepEqual(result, { + local: false, + phone: expected, + }); + } + }).pipe(Effect.provide(AgentAlerts.layer)), + ); + it.effect("coalesces queued updates and reruns a thread dirtied during publishing", () => Effect.gen(function* () { const started = yield* Deferred.make(); @@ -841,34 +889,24 @@ describe("AgentAwarenessRelay", () => { it.effect("catches up after startup without polling after success", () => Effect.gen(function* () { - const { relay, secrets, secretReads, catchUp, publications } = yield* makeTestRelay({ - catchUpThreads: [shell()], - }); + const { relay, secrets, secretReads, catchUp } = yield* makeTestRelay(); yield* relay.start(); yield* TestClock.adjust("1 second"); - yield* relay.drain; - assert.equal(catchUp.activitySnapshotReads, 1); - assert.equal(publications.length, 1); - assert.include(publications[0]?.state, { - threadId: THREAD_ID, - projectTitle: "Project", - threadTitle: "Thread", - phase: "running", - }); + assert.equal(catchUp.shellSnapshotReads, 1); yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false")); yield* relay.requestCatchUp(); yield* TestClock.adjust("1 second"); - assert.equal(catchUp.activitySnapshotReads, 1); + assert.equal(catchUp.shellSnapshotReads, 1); yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true")); yield* relay.requestCatchUp(); yield* TestClock.adjust("1 second"); - assert.equal(catchUp.activitySnapshotReads, 2); + assert.equal(catchUp.shellSnapshotReads, 2); const readsBeforeIdle = secretReads.length; yield* TestClock.adjust("10 minutes"); - assert.equal(catchUp.activitySnapshotReads, 2); + assert.equal(catchUp.shellSnapshotReads, 2); assert.equal(secretReads.length, readsBeforeIdle); yield* secrets.set( @@ -877,9 +915,6 @@ it.effect("catches up after startup without polling after success", () => ); yield* relay.requestCatchUp(); yield* TestClock.adjust("1 second"); - yield* relay.drain; - assert.equal(catchUp.activitySnapshotReads, 3); - assert.equal(publications.length, 2); - assert.equal(publications[1]?.state?.threadId, THREAD_ID); + assert.equal(catchUp.shellSnapshotReads, 3); }).pipe(Effect.scoped), ); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 2523fea9f2cf0..b34fb3b8b4d8f 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -1,25 +1,34 @@ -import type { AgentAwarenessThread } from "../orchestration-v2/ProjectionStore.ts"; import { isListedThreadLineage, type EnvironmentId, type OrchestrationV2DomainEvent, + type OrchestrationV2ThreadShell, type OrchestrationV2TurnItem, type Project, type ThreadId, } from "@t3tools/contracts"; +import type { AgentAlert } from "@t3tools/contracts"; import { RelayApi, type RelayAgentActivityPublishProofPayload, type RelayAgentActivityState, + type RelayAgentAlert, + type RelayAgentAlertPublishProofPayload, } from "@t3tools/contracts/relay"; import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness"; import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { withRelayClientTracing } from "@t3tools/shared/relayTracing"; -import { normalizeRelayIssuer, RELAY_ACTIVITY_PUBLISH_TYP } from "@t3tools/shared/relayJwt"; +import { + normalizeRelayIssuer, + RELAY_ACTIVITY_PUBLISH_TYP, + RELAY_AGENT_ALERT_TYP, + signRelayJwt, +} from "@t3tools/shared/relayJwt"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Signature from "effect/crypto/Signature"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -42,8 +51,9 @@ import { RELAY_ISSUER_SECRET, RELAY_URL_SECRET, } from "../cloud/config.ts"; -import { EnvironmentSigningKey } from "../cloud/environmentKeys.ts"; +import { getOrCreateEnvironmentKeyPairFromSecretStore } from "../cloud/environmentKeys.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as AgentAlerts from "../notifications/AgentAlerts.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import { forkParked } from "../serverActivation.ts"; @@ -243,7 +253,19 @@ function deliveryStats( }; } +function signRelayAgentActivityPublishProof(input: { + readonly privateKey: string; + readonly payload: RelayAgentActivityPublishProofPayload; +}) { + return signRelayJwt({ + privateKey: input.privateKey, + typ: RELAY_ACTIVITY_PUBLISH_TYP, + payload: input.payload, + }); +} + const makePublishProof = Effect.fn("makePublishProof")(function* (input: { + readonly privateKey: string; readonly relayIssuer: string; readonly environmentId: string; readonly threadId: ThreadId; @@ -263,12 +285,39 @@ const makePublishProof = Effect.fn("makePublishProof")(function* (input: { threadId: input.threadId, state: input.state, } satisfies RelayAgentActivityPublishProofPayload; - const signingKey = yield* EnvironmentSigningKey; - return yield* signingKey.sign({ typ: RELAY_ACTIVITY_PUBLISH_TYP, payload }); + return yield* signRelayAgentActivityPublishProof({ privateKey: input.privateKey, payload }); +}); + +const makeAlertProof = Effect.fn("makeAlertProof")(function* (input: { + readonly privateKey: string; + readonly relayIssuer: string; + readonly environmentId: string; + readonly threadId: ThreadId; + readonly alert: RelayAgentAlert; + readonly jti: string; +}) { + const now = yield* DateTime.now; + const expiresAt = DateTime.add(now, { minutes: 5 }); + const payload = { + iss: `t3-env:${input.environmentId}`, + aud: normalizeRelayIssuer(input.relayIssuer), + sub: input.environmentId, + jti: input.jti, + iat: Math.floor(now.epochMilliseconds / 1_000), + exp: Math.floor(expiresAt.epochMilliseconds / 1_000), + environmentId: input.environmentId as RelayAgentAlertPublishProofPayload["environmentId"], + threadId: input.threadId, + alert: input.alert, + } satisfies RelayAgentAlertPublishProofPayload; + return yield* signRelayJwt({ + privateKey: input.privateKey, + typ: RELAY_AGENT_ALERT_TYP, + payload, + }); }); function describeThreadShellForAwareness( - thread: Option.Option, + thread: Option.Option, ): Record { if (Option.isNone(thread)) { return { found: false }; @@ -287,7 +336,7 @@ function describeThreadShellForAwareness( function resolveAgentAwarenessRelayPublishSnapshot(input: { readonly environmentId: EnvironmentId; - readonly thread: Option.Option; + readonly thread: Option.Option; readonly project: Option.Option; }): { readonly projectId: string | null; @@ -321,7 +370,7 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: { }; } -function terminalWorkSinceStart(thread: AgentAwarenessThread, startedAt: number): boolean { +function terminalWorkSinceStart(thread: OrchestrationV2ThreadShell, startedAt: number): boolean { return ( thread.latestRunCompletedAt != null && DateTime.toEpochMillis(thread.latestRunCompletedAt) > startedAt @@ -332,7 +381,7 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: { readonly environmentId: EnvironmentId; readonly startedAt: number; readonly projects: ReadonlyArray>; - readonly threads: ReadonlyArray; + readonly threads: ReadonlyArray; }): ReadonlyArray { const projectById = new Map(input.projects.map((project) => [project.id, project])); return input.threads @@ -358,13 +407,15 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: { /** @public */ export const make = Effect.gen(function* () { + const agentAlerts = yield* Effect.serviceOption(AgentAlerts.AgentAlerts); const secrets = yield* ServerSecretStore.ServerSecretStore; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const threads = yield* ThreadManagement.ThreadManagementService; const projects = yield* ProjectService.ProjectService; const random = yield* SecureRandom.SecureRandom; - const signingKey = yield* EnvironmentSigningKey; + const signature = yield* Signature.Signature; const scope = yield* Effect.scope; + const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets); const startedAt = (yield* DateTime.now).epochMilliseconds; const catchUpRequests = yield* Queue.dropping(1); const publishedStateByThreadRef = yield* Ref.make(new Map()); @@ -404,6 +455,10 @@ export const make = Effect.gen(function* () { let cachedRelayClient: Effect.Success> | undefined; let publishedRelayConfig: NonNullable> | undefined; + // Phone alerts must not share this cache. Updating it without clearing + // publishedStateByThreadRef makes the next relay look already published. + let phoneRelayClient: Effect.Success> | undefined; + let phoneRelayConfig: NonNullable> | undefined; // Confirm through the same worker so tombstones cannot race live updates; recovery clears the deadline. const publishConfirmDeadlines = new Map(); @@ -470,12 +525,13 @@ export const make = Effect.gen(function* () { } const relayClient = cachedRelayClient; const proof = yield* makePublishProof({ + privateKey: cloudLinkKeyPair.privateKey, relayIssuer: relayConfig.issuer, environmentId, threadId, state: input.state, jti: yield* random.uuidV4(), - }).pipe(Effect.provideService(EnvironmentSigningKey, signingKey)); + }); yield* Effect.logInfo("publishing agent activity for thread", { environmentId, @@ -508,7 +564,7 @@ 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* threads.getAgentAwarenessSnapshot(threadId)).threads[0] ?? null; + const threadShell = yield* threads.getThreadShell(threadId); if ( threadShell !== null && !isListedThreadLineage(threadShell.lineage) && @@ -519,7 +575,7 @@ export const make = Effect.gen(function* () { } const thread = threadShell === null || threadShell.archivedAt !== null - ? Option.none() + ? Option.none() : Option.some(threadShell); const project = Option.isSome(thread) ? yield* projects.getById(thread.value.projectId) @@ -719,12 +775,19 @@ export const make = Effect.gen(function* () { } const environmentId = yield* serverEnvironment.getEnvironmentId; const projectSnapshot = yield* projects.snapshot; - const activitySnapshot = yield* threads.getAgentAwarenessSnapshot(); + // Subagents and side chats never publish awareness, and subagents are most threads. + const topLevelThreadIds = (yield* threads.listActiveShellIndex()).flatMap((thread) => + isListedThreadLineage(thread.lineage) ? [thread.id] : [], + ); + const shellSnapshot = yield* threads.getShellSnapshot({ + location: "active", + threadIds: topLevelThreadIds, + }); const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({ environmentId, startedAt, projects: projectSnapshot.projects, - threads: activitySnapshot.threads, + threads: shellSnapshot.threads, }); if (activeThreadIds.length === 0) { yield* Effect.logDebug("agent activity snapshot has no publishable threads"); @@ -829,6 +892,61 @@ export const make = Effect.gen(function* () { }, ); + const publishPhone = Effect.fn("AgentAwarenessRelay.publishAgentAlert")(function* ( + alert: AgentAlert, + ) { + const relayConfig = yield* readRelayConfig; + const publishEnabled = yield* readPublishAgentActivityEnabled; + if (!relayConfig || !publishEnabled) return false; + if ( + phoneRelayConfig?.url !== relayConfig.url || + phoneRelayConfig.issuer !== relayConfig.issuer || + phoneRelayConfig.environmentCredential !== relayConfig.environmentCredential + ) { + phoneRelayClient = undefined; + phoneRelayConfig = relayConfig; + } + if (phoneRelayClient === undefined) { + phoneRelayClient = yield* makeRelayClient(relayConfig); + } + // Hold the client locally. A later config change can clear the cache while + // the shell lookup or the proof is still in flight. + const relayClient = phoneRelayClient; + const environmentId = yield* serverEnvironment.getEnvironmentId; + const shell = yield* threads.getThreadShell(alert.threadId); + const title = shell?.title.trim() || "T3"; + const deepLink = `/threads/${encodeURIComponent(environmentId)}/${encodeURIComponent(alert.threadId)}`; + const phoneAlert = { + title, + body: alert.body, + deepLink, + } as RelayAgentAlert; + const proof = yield* makeAlertProof({ + privateKey: cloudLinkKeyPair.privateKey, + relayIssuer: relayConfig.issuer, + environmentId, + threadId: alert.threadId, + alert: phoneAlert, + jti: yield* random.uuidV4(), + }); + const response = yield* relayClient.server + .publishAgentAlert({ + params: { environmentId, threadId: alert.threadId }, + payload: { alert: phoneAlert, proof }, + }) + .pipe(Effect.timeout("10 seconds")); + return response.ok && response.deliveries.some((delivery) => delivery.ok); + }); + if (Option.isSome(agentAlerts)) { + yield* agentAlerts.value.registerPhone({ + publish: (alert) => + publishPhone(alert).pipe( + Effect.provideService(Signature.Signature, signature), + Effect.orElseSucceed(() => false), + ), + }); + } + return AgentAwarenessRelay.of({ publishThread, drain: worker.drain, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 5c9ab9c433ffc..966c717a38cba 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -100,6 +100,7 @@ import * as Keybindings from "./keybindings.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as ServerOwnership from "./serverOwnership.ts"; import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts"; +import * as AgentAlerts from "./notifications/AgentAlerts.ts"; import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as ManagedLimits from "./policy/managedLimits.ts"; @@ -628,8 +629,10 @@ const layerProviderInstallationRefresh = Layer.effectDiscard( }), ); +const layerAgentAlerts = AgentAlerts.layer; const layerRuntimeCoreDependenciesBase = Layer.mergeAll( - AgentAwarenessRelay.layer, + AgentAwarenessRelay.layer.pipe(Layer.provide(layerAgentAlerts)), + layerAgentAlerts, HeldHooksWaker.layer, layerThreadSettlementWorker, layerIdleCompactionWorker, @@ -682,7 +685,10 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( Layer.mergeAll(Keybindings.layer, EnvironmentTheme.layer, UsageLimitSources.layer), ), Layer.provideMerge(ProviderRegistry.layer), - Layer.provideMerge(ProviderInstanceRegistryHydration.layer), + // Hydration is built before this merge, so Claude's serviceOption(AgentAlerts) + // would otherwise stay empty. The same layer reference is memoized with the + // websocket and the phone publisher. + Layer.provideMerge(ProviderInstanceRegistryHydration.layer.pipe(Layer.provide(layerAgentAlerts))), // Provider hydration builds the Claude runner, which reports agents to the same scanner the // terminal and preview layers share. Layer.provideMerge(layerPortScanner), @@ -782,6 +788,8 @@ const layerMakeRoutes = Layer.mergeAll( DeviceHubProxy.layer, ServerBrowserStream.routeLayer, ServerHttp.layerStaticAndDevRoute, + // serve forks its memo map. Building AgentAlerts inside the route would be a + // second instance, and Claude would publish to a pubsub no client reads. Ws.layer, speechStreamRouteLayer, ), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index e3c7fdaa47476..7c213c725f64d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,6 +1,7 @@ import { OrchestrationDispatchCommandError } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as AgentAlerts from "./notifications/AgentAlerts.ts"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as PeerLinkRequests from "./peer/PeerLinkRequests.ts"; import * as PeerLinks from "./peer/PeerLinks.ts"; @@ -1515,6 +1516,7 @@ export const layerWsRpc = ( const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; + const agentAlerts = yield* Effect.serviceOption(AgentAlerts.AgentAlerts); const rpcClientIds = yield* Ref.make(new Set()); yield* Effect.addFinalizer(() => Ref.get(rpcClientIds).pipe( @@ -2730,6 +2732,11 @@ export const layerWsRpc = ( ), ), ), + [WS_METHODS.serverSubscribeAgentAlerts]: (_input) => + Option.match(agentAlerts, { + onNone: () => Stream.empty, + onSome: (alerts) => alerts.stream, + }), [WS_METHODS.cloudGetRelayClientStatus]: (_input) => relayClient.resolve, [WS_METHODS.cloudInstallRelayClient]: (_input) => Stream.callback((queue) => @@ -3493,6 +3500,7 @@ export const WS_RPC_SERVER_OPTIONS = { export const layer = Layer.unwrap( Effect.gen(function* () { + const agentAlerts = yield* AgentAlerts.AgentAlerts; const serverBrowser = yield* ServerBrowser.ServerBrowser; const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const npmRelease = yield* NpmRelease.NpmRelease; @@ -3574,6 +3582,7 @@ export const layer = Layer.unwrap( // Handler-context reporters see only request defects, not server-wide defects. Layer.provide(DefectReporter.layer), Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)), + Layer.provide(Layer.succeed(AgentAlerts.AgentAlerts, agentAlerts)), Layer.provide(AgentSessionScanner.layer), Layer.provide(ProviderMaintenanceRunner.layer), Layer.provide(Layer.succeed(NpmRelease.NpmRelease, npmRelease)), diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index a7134b481d131..a2d7727b3a0d2 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -11,6 +11,7 @@ import type { ClientSettings } from "@t3tools/contracts/settings"; import * as Option from "effect/Option"; import type { AtomRegistry } from "effect/reactivity"; import { + BellIcon, CircleAlertIcon, CircleCheckIcon, MessageCircleQuestionIcon, @@ -18,6 +19,7 @@ import { } from "lucide-react"; import { getClientSettings, subscribeClientSettings } from "../hooks/useSettings"; +import { setAgentAlertPresenter } from "../lib/agentAlertPresenter"; import { isWindowsPlatform } from "../lib/utils"; import { environmentSummaries } from "../state/presentation"; import { environmentShell } from "../state/shell"; @@ -389,6 +391,91 @@ export function startThreadNotifications( }; }; + const presentAgentAlert = (alert: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly body: string; + }) => { + const toastKey = threadKey(alert); + if (registry.get(environmentThreadShells.hiddenThreadKeysAtom).has(toastKey)) return; + const { notificationMode: mode, inAppNotificationsEnabled } = getClientSettings(); + if (mode === "off" && !inAppNotificationsEnabled) return; + const active = (router.state.matches.at(-1)?.params ?? {}) as { + environmentId?: string; + threadId?: string; + }; + const looking = + document.visibilityState === "visible" && + document.hasFocus() && + active.environmentId === alert.environmentId && + active.threadId === alert.threadId; + if (looking) return; + const shell = registry.get(environmentShell.stateValueAtom(alert.environmentId)); + const threads = + shell.status === "live" && Option.isSome(shell.snapshot) ? shell.snapshot.value.threads : []; + const title = threads.find((thread) => thread.id === alert.threadId)?.title || "T3"; + if (hasNotificationSound(mode)) { + void playNotificationSound("input", () => + hasNotificationSound(getClientSettings().notificationMode), + ); + } + const openThread = () => { + void router.navigate({ + to: "/$environmentId/$threadId", + params: { environmentId: alert.environmentId, threadId: alert.threadId }, + }); + }; + if ( + inAppNotificationsEnabled && + document.visibilityState === "visible" && + document.hasFocus() + ) { + const toastId: ToastId = toastManager.add({ + onRemove: (): void => forgetToast(toastKey, toastId), + type: "warning", + title, + description: alert.body, + data: { + hideCopyButton: true, + leadingIcon: , + }, + actionProps: { + children: "Open thread", + onClick: () => { + toastManager.close(toastId); + openThread(); + }, + }, + }); + openToasts.set(toastKey, [...(openToasts.get(toastKey) ?? []), toastId]); + return; + } + if ( + !hasDesktopNotifications(mode) || + typeof Notification === "undefined" || + Notification.permission !== "granted" + ) { + return; + } + try { + const notification = new Notification(title, { + body: alert.body, + tag: `${threadNotificationTag(alert.environmentId, alert.threadId)}:alert`, + silent: true, + }); + trackNotification(alert.environmentId, alert.threadId, notification); + notification.addEventListener("click", () => { + notification.close(); + window.focus(); + void window.desktopBridge?.revealWindow?.().catch(() => undefined); + openThread(); + }); + } catch { + // Some browsers expose Notification but reject desktop presentation. + } + }; + setAgentAlertPresenter(presentAgentAlert); + const onSettings = () => { applyMode(); syncEnvironments(); @@ -409,6 +496,7 @@ export function startThreadNotifications( onSettings(); return () => { + setAgentAlertPresenter(null); unsubscribeSettings(); unsubscribeEnvironmentIds(); for (const unsubscribe of unsubscribeHidden) unsubscribe(); diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index 6a3cc4acb43f3..4895ba17fa6ac 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -12,6 +12,7 @@ import { Atom } from "effect/reactivity"; import { isGuestPage } from "../guestPage"; import * as Runtime from "../lib/runtime"; +import * as AgentAlertNotifications from "../lib/agentAlertNotifications"; import * as BackgroundActivityReporter from "../lib/backgroundActivityReporter"; import * as ConnectionPlatform from "./platform"; @@ -30,7 +31,8 @@ type ConnectionLayerSource = | typeof Runtime.layer | typeof ConnectionPlatform.layer | typeof BackgroundActivityReporter.layerObserver - | typeof BackgroundActivityReporter.layer; + | typeof BackgroundActivityReporter.layer + | typeof AgentAlertNotifications.layer; const layerProvidedClientConnection = layerSnapshotLoader.pipe( Layer.provideMerge( @@ -49,9 +51,10 @@ const layerProvidedClientConnection = layerSnapshotLoader.pipe( ), ); -const layerConnection = BackgroundActivityReporter.layer.pipe( - Layer.provideMerge(layerProvidedClientConnection), -); +const layerConnection = Layer.mergeAll( + BackgroundActivityReporter.layer, + AgentAlertNotifications.layer, +).pipe(Layer.provideMerge(layerProvidedClientConnection)); /** * A guest has no session here. Building the connection would open the page origin's environment diff --git a/apps/web/src/lib/agentAlertNotifications.ts b/apps/web/src/lib/agentAlertNotifications.ts new file mode 100644 index 0000000000000..44cde9d4b0a42 --- /dev/null +++ b/apps/web/src/lib/agentAlertNotifications.ts @@ -0,0 +1,88 @@ +import { EnvironmentRegistry } from "@t3tools/client-runtime/connection"; +import { subscribe } from "@t3tools/client-runtime/rpc"; +import { type EnvironmentId, WS_METHODS } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import { hasDesktopNotifications, hasNotificationSound } from "../threadNotifications"; + +import { emitAgentAlert } from "./agentAlertPresenter"; + +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + if (typeof window === "undefined") return; + // Settings imports atoms on this runtime, so load it after runtime initialization. + const { getClientSettings, subscribeClientSettings } = yield* Effect.promise( + () => import("../hooks/useSettings"), + ); + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const running = yield* Ref.make(new Map>()); + const entries = yield* Ref.make(new Map()); + const refresh = yield* Queue.sliding(1); + const sync = Effect.fnUntraced(function* () { + const { notificationMode, inAppNotificationsEnabled } = getClientSettings(); + const presentable = + inAppNotificationsEnabled || + hasNotificationSound(notificationMode) || + (hasDesktopNotifications(notificationMode) && + typeof Notification !== "undefined" && + Notification.permission === "granted"); + const active = presentable ? yield* Ref.get(entries) : new Map(); + const current = yield* Ref.get(running); + for (const [environmentId, fiber] of current) { + if (active.has(environmentId)) continue; + yield* Fiber.interrupt(fiber); + yield* Ref.update(running, (next) => { + const updated = new Map(next); + updated.delete(environmentId); + return updated; + }); + } + if (!presentable) return; + for (const environmentId of active.keys()) { + if ((yield* Ref.get(running)).has(environmentId)) continue; + const fiber = yield* registry + .followStream(environmentId, subscribe(WS_METHODS.serverSubscribeAgentAlerts, {})) + .pipe( + Stream.runForEach((alert) => + Effect.sync(() => + emitAgentAlert({ + environmentId, + threadId: alert.threadId, + body: alert.body, + }), + ), + ), + Effect.forkScoped, + ); + yield* Ref.update(running, (next) => { + const updated = new Map(next); + updated.set(environmentId, fiber); + return updated; + }); + } + }); + yield* Effect.acquireRelease( + Effect.sync(() => subscribeClientSettings(() => Queue.offerUnsafe(refresh, undefined))), + (unsubscribe) => Effect.sync(unsubscribe), + ); + yield* Stream.concat( + Stream.fromEffect(SubscriptionRef.get(registry.entries)), + SubscriptionRef.changes(registry.entries), + ).pipe( + Stream.runForEach((next) => + Ref.set(entries, new Map(next)).pipe(Effect.andThen(Queue.offer(refresh, undefined))), + ), + Effect.forkScoped, + ); + yield* Stream.fromQueue(refresh).pipe( + Stream.runForEach(() => sync()), + Effect.forkScoped, + ); + }), +); diff --git a/apps/web/src/lib/agentAlertPresenter.ts b/apps/web/src/lib/agentAlertPresenter.ts new file mode 100644 index 0000000000000..6284a9e0d57bb --- /dev/null +++ b/apps/web/src/lib/agentAlertPresenter.ts @@ -0,0 +1,19 @@ +import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; + +export interface PresentedAgentAlert { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly body: string; +} + +type PresentAgentAlert = (alert: PresentedAgentAlert) => void; + +let present: PresentAgentAlert | null = null; + +export function setAgentAlertPresenter(next: PresentAgentAlert | null): void { + present = next; +} + +export function emitAgentAlert(alert: PresentedAgentAlert): void { + present?.(alert); +} diff --git a/apps/web/src/rpc/requestLatencyState.ts b/apps/web/src/rpc/requestLatencyState.ts index c95ddae2afcde..853bae75690cb 100644 --- a/apps/web/src/rpc/requestLatencyState.ts +++ b/apps/web/src/rpc/requestLatencyState.ts @@ -33,6 +33,7 @@ const untrackedRpcAckMethods = new Set([ WS_METHODS.serverGetUsageSummary, WS_METHODS.serverReportClientActivity, WS_METHODS.sourceControlSignInWait, + WS_METHODS.serverSubscribeAgentAlerts, ]); const longRunningRpcAckMethods = new Set([ WS_METHODS.serverUpdateProvider, diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 33f1518ec30ae..a935586d1a477 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -1,3 +1,4 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { makeAggregateState } from "./agentActivityAggregate.ts"; import type { RelayAgentActivityState, RelayDeliveryResult } from "@t3tools/contracts/relay"; import { describe, expect, it } from "@effect/vitest"; @@ -12,6 +13,7 @@ import * as FcmDeliveries from "./FcmDeliveries.ts"; import * as ApnsDeliveries from "./ApnsDeliveries.ts"; const layerPublisher = AgentActivityPublisher.layer.pipe( + Layer.provide(NodeCrypto.layer), Layer.provide( Layer.succeed(FcmDeliveries.FcmDeliveries, { enqueue: () => Effect.succeed(null), @@ -165,6 +167,7 @@ describe("AgentActivityPublisher", () => { }).pipe( Effect.provide( AgentActivityPublisher.layer.pipe( + Layer.provide(NodeCrypto.layer), Layer.provide( Layer.mergeAll( Layer.succeed(AgentActivityRows.AgentActivityRows, makeAgentActivityRows()), @@ -521,4 +524,127 @@ describe("makeAggregateState", () => { { threadId: "a-5" }, ]); }); + + it.effect("sends a one-shot alert without writing activity rows", () => { + const preferences = JSON.stringify({ + liveActivitiesEnabled: false, + notificationsEnabled: true, + notifyOnApproval: true, + notifyOnInput: true, + notifyOnCompletion: true, + notifyOnFailure: true, + }); + const android = { + ...target("android"), + platform: "android" as const, + ios_major_version: null, + push_token: "android-token", + preferences_json: preferences, + }; + const ios = { + ...target("ios"), + push_token: "ios-token", + preferences_json: preferences, + }; + const fcmCalls: Array[0]> = []; + const appleCalls: Array< + Parameters[0] + > = []; + let upserts = 0; + return Effect.gen(function* () { + const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; + const result = yield* publisher.publishAlert({ + environmentId: "env", + environmentPublicKey: "key", + threadId: "thread", + title: "Thread", + body: "Build finished", + deepLink: "/threads/env/thread", + }); + expect(upserts).toBe(0); + expect(result.ok).toBe(true); + expect(fcmCalls).toEqual([ + { + target: android, + state: null, + directAlert: { + title: "Thread", + body: "Build finished", + deepLink: "/threads/env/thread", + environmentId: "env", + threadId: "thread", + }, + }, + ]); + expect(appleCalls).toMatchObject([ + { + token: "ios-token", + notification: { + title: "Thread", + body: "Build finished", + deepLink: "/threads/env/thread", + environmentId: "env", + threadId: "thread", + }, + }, + ]); + }).pipe( + Effect.provide( + AgentActivityPublisher.layer.pipe( + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mergeAll( + Layer.succeed( + AgentActivityRows.AgentActivityRows, + makeAgentActivityRows({ + upsert: () => + Effect.sync(() => { + upserts += 1; + }), + }), + ), + Layer.succeed(EnvironmentLinks.EnvironmentLinks, makeEnvironmentLinks()), + Layer.succeed( + LiveActivities.LiveActivities, + makeLiveActivities({ listTargets: () => Effect.succeed([android, ios]) }), + ), + Layer.succeed( + ApnsDeliveries.ApnsDeliveries, + makeApnsDeliveries({ + sendPushNotification: (input) => + Effect.sync(() => { + appleCalls.push(input); + return { + deviceId: input.target.device_id, + kind: "push_notification", + ok: true, + apnsStatus: 200, + apnsReason: null, + apnsId: "apns-id", + }; + }), + }), + ), + Layer.succeed(FcmDeliveries.FcmDeliveries, { + enqueue: (input) => + Effect.sync(() => { + fcmCalls.push(input); + return { + deviceId: input.target.device_id, + kind: "push_notification", + ok: true, + queued: true, + apnsStatus: null, + apnsReason: null, + apnsId: null, + }; + }), + process: () => Effect.void, + }), + ), + ), + ), + ), + ); + }); }); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index 6ef258acdb6d3..a4435e65f2526 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -2,18 +2,22 @@ import { makeAggregateState, TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS, } from "./agentActivityAggregate.ts"; -import type { - RelayAgentActivityState, - RelayDeliveryResult, - RelayPublishResponse, +import { + RelayAgentAwarenessPreferences, + type RelayAgentActivityState, + type RelayDeliveryResult, + type RelayPublishResponse, } from "@t3tools/contracts/relay"; import * as Arr from "effect/Array"; import * as Context from "effect/Context"; +import * as SecureRandom from "effect/crypto/SecureRandom"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; -import { isTerminalPhase } from "./agentActivityPayloads.ts"; +import { isTerminalPhase, sanitizeApnsNotificationPayload } from "./agentActivityPayloads.ts"; export { isExpiredAgentActivityState } from "./agentActivityPayloads.ts"; import * as AgentActivityRows from "./AgentActivityRows.ts"; @@ -22,6 +26,10 @@ import * as LiveActivities from "./LiveActivities.ts"; import * as ApnsDeliveries from "./ApnsDeliveries.ts"; import * as FcmDeliveries from "./FcmDeliveries.ts"; +const decodeAlertPreferences = Schema.decodeUnknownOption( + Schema.fromJsonString(RelayAgentAwarenessPreferences), +); + export type AgentActivityPublishError = | FcmDeliveries.FcmDeliveryError | AgentActivityRows.AgentActivityRowUpsertPersistenceError @@ -40,6 +48,14 @@ export class AgentActivityPublisher extends Context.Service< readonly threadId: string; readonly state: RelayAgentActivityState | null; }) => Effect.Effect; + readonly publishAlert: (input: { + readonly environmentId: string; + readonly environmentPublicKey: string; + readonly threadId: string; + readonly title: string; + readonly body: string; + readonly deepLink: string; + }) => Effect.Effect; readonly replayForLiveActivityRegistration: (input: { readonly userId: string; readonly deviceId: string; @@ -52,6 +68,7 @@ export class AgentActivityPublisher extends Context.Service< >()("t3code-relay/agentActivity/AgentActivityPublisher") {} const make = Effect.gen(function* () { + const random = yield* SecureRandom.SecureRandom; const rows = yield* AgentActivityRows.AgentActivityRows; const links = yield* EnvironmentLinks.EnvironmentLinks; const liveActivities = yield* LiveActivities.LiveActivities; @@ -112,6 +129,81 @@ const make = Effect.gen(function* () { return deliveriesByTarget.flat(); }); + const publishAlertDeliveries = Effect.fn( + "relay.agent_activity_publisher.publish_alert_deliveries", + )(function* (input: { + readonly environmentId: string; + readonly environmentPublicKey: string; + readonly threadId: string; + readonly title: string; + readonly body: string; + readonly deepLink: string; + }) { + const notification = sanitizeApnsNotificationPayload({ + title: input.title, + body: input.body, + environmentId: input.environmentId, + threadId: input.threadId, + deepLink: input.deepLink, + }); + yield* Effect.annotateCurrentSpan({ + "relay.environment_id": input.environmentId, + "relay.thread_id": input.threadId, + }); + const deliveryUsers = yield* links.listDeliveryUsersForEnvironment({ + environmentId: input.environmentId, + environmentPublicKey: input.environmentPublicKey, + }); + const deliveriesByUser = yield* Effect.forEach( + deliveryUsers.filter((user) => user.notificationsEnabled), + Effect.fnUntraced(function* (deliveryUser) { + const targets = yield* liveActivities.listTargets({ userId: deliveryUser.userId }); + return yield* Effect.forEach( + targets, + Effect.fnUntraced(function* (target) { + if (!target.push_token) return null; + const preferences = decodeAlertPreferences(target.preferences_json); + // A proactive alert is the same kind of "come look" as needing input. + if ( + Option.isNone(preferences) || + !preferences.value.notificationsEnabled || + !preferences.value.notifyOnInput + ) { + return null; + } + if (target.platform === "android") { + return yield* fcmDeliveries.enqueue({ + target, + state: null, + directAlert: { + title: notification.title, + body: notification.body, + deepLink: notification.deepLink, + environmentId: notification.environmentId, + threadId: notification.threadId, + }, + }); + } + return yield* apnsDeliveries.sendPushNotification({ + target, + token: target.push_token, + sourceJobId: yield* random.uuidV4(), + notification, + }); + }), + { concurrency: 4 }, + ); + }), + { concurrency: 4 }, + ); + return { + ok: true as const, + deliveries: deliveriesByUser + .flat() + .filter((delivery): delivery is RelayDeliveryResult => delivery !== null), + }; + }); + return AgentActivityPublisher.of({ replayForLiveActivityRegistration: Effect.fn( "relay.agent_activity_publisher.replay_for_live_activity_registration", @@ -243,6 +335,15 @@ const make = Effect.gen(function* () { ), }; }), + publishAlert: Effect.fn("AgentActivityPublisher.publishAlert")((input) => + publishAlertDeliveries(input).pipe( + Effect.catchCause((cause) => + Effect.logWarning("agent alert delivery failed", { cause }).pipe( + Effect.as({ ok: false as const, deliveries: [] }), + ), + ), + ), + ), }); }); diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index 21df615900f30..72cba237291a1 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -208,6 +208,61 @@ function harness() { }; } +describe("direct alerts", () => { + const directAlert = { + title: "Thread", + body: "Build finished", + deepLink: "/threads/env/thread", + environmentId: "env", + threadId: "thread", + eventId: "event-1", + }; + + it.effect("keeps the live card and a stable id across retries", () => { + const h = harness(); + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + const now = yield* DateTime.now; + h.current.state = { ...state, updatedAt: DateTime.formatIso(now) }; + const job = { ...h.job, state: null, queuedAt: now.epochMilliseconds, directAlert }; + yield* delivery.process(job); + yield* delivery.process(job); + yield* delivery.process({ + ...job, + directAlert: { ...directAlert, eventId: "event-2", title: "Later" }, + }); + expect(h.sent).toHaveLength(3); + expect(h.sent[0]?.data.alert_only).toBe("true"); + expect(h.sent[0]?.data.active).toBe("true"); + expect(h.sent[0]?.data.alert_title).toBe("Thread"); + expect(h.sent[0]?.data.alert_id).toBe(h.sent[1]?.data.alert_id); + expect(h.sent[0]?.data.alert_id).not.toBe(h.sent[2]?.data.alert_id); + expect(h.marked).toHaveLength(0); + }).pipe(Effect.provide(h.layer)); + }); + + it.effect("drops a queued alert after the environment is unlinked or muted", () => { + const h = harness(); + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries.FcmDeliveries; + const now = yield* DateTime.now; + const job = { ...h.job, state: null, queuedAt: now.epochMilliseconds, directAlert }; + h.current.linked = false; + yield* delivery.process(job); + h.current.linked = true; + h.current.mutedEnvironments = ["env"]; + yield* delivery.process(job); + h.current.mutedEnvironments = []; + h.current.target.preferences_json = encodeJson({ + ...preferences, + notificationsEnabled: false, + }); + yield* delivery.process(job); + expect(h.sent).toHaveLength(0); + }).pipe(Effect.provide(h.layer)); + }); +}); + describe("Android delivery routing", () => { const secondState: RelayAgentActivityState = { ...state, diff --git a/infra/relay/src/agentActivity/FcmDeliveries.ts b/infra/relay/src/agentActivity/FcmDeliveries.ts index 1c6cc0a6b71ff..f792d04201b0d 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.ts @@ -6,6 +6,7 @@ import { type RelayDeliveryResult, } from "@t3tools/contracts/relay"; import * as Digest from "effect/crypto/Digest"; +import * as SecureRandom from "effect/crypto/SecureRandom"; import type * as PlatformError from "effect/PlatformError"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -39,6 +40,18 @@ export const FcmDeliveryJob = Schema.Struct({ state: Schema.NullOr(RelayAgentActivityState), queuedAt: Schema.Number, replay: Schema.optional(Schema.Boolean), + directAlert: Schema.optional( + Schema.Struct({ + title: Schema.String, + body: Schema.String, + deepLink: Schema.String, + environmentId: Schema.String, + threadId: Schema.String, + // Set once at enqueue so a queue retry stays one alert and a later + // message with the same body does not. + eventId: Schema.String, + }), + ), }); export type FcmDeliveryJob = typeof FcmDeliveryJob.Type; const decodeJob = Schema.decodeUnknownEffect(FcmDeliveryJob); @@ -120,6 +133,9 @@ export class FcmDeliveries extends Context.Service< readonly target: LiveActivities.TargetRow; readonly state: RelayAgentActivityState | null; readonly replay?: boolean; + readonly directAlert?: Omit, "eventId"> & { + readonly eventId?: string; + }; }) => Effect.Effect; readonly process: ( body: unknown, @@ -140,6 +156,7 @@ export class FcmDeliveries extends Context.Service< const make = Effect.gen(function* () { const config = yield* RelayConfiguration.RelayConfiguration; const crypto = yield* Digest.Digest; + const random = yield* SecureRandom.SecureRandom; const sender = yield* FcmDeliveryQueueSender.FcmDeliveryQueueSender; const client = yield* FcmClient.FcmClient; const devices = yield* LiveActivities.LiveActivities; @@ -162,6 +179,21 @@ const make = Effect.gen(function* () { }; } const now = yield* DateTime.now; + const directAlert = + input.directAlert === undefined + ? undefined + : { + ...input.directAlert, + eventId: + input.directAlert.eventId ?? + (yield* random + .uuidV4() + .pipe( + Effect.mapError( + (cause) => new FcmDeliveryError({ operation: "enqueue", cause }), + ), + )), + }; yield* sender .send({ userId: input.target.user_id, @@ -169,6 +201,7 @@ const make = Effect.gen(function* () { token: input.target.push_token, state: input.state, ...(input.replay ? { replay: true } : {}), + ...(directAlert === undefined ? {} : { directAlert }), queuedAt: now.epochMilliseconds, }) .pipe(Effect.mapError((cause) => new FcmDeliveryError({ operation: "enqueue", cause }))); @@ -188,6 +221,94 @@ const make = Effect.gen(function* () { ); const now = yield* DateTime.now; if (now.epochMilliseconds - job.queuedAt > 5 * 60_000) return; + if (job.directAlert) { + const targets = yield* devices.listTargets({ userId: job.userId }); + const target = targets.find( + (device) => + device.device_id === job.deviceId && + device.platform === "android" && + device.push_token === job.token, + ); + if (!target) return; + const preferences = decodePreferences(target.preferences_json); + if ( + Option.isNone(preferences) || + !preferences.value.notificationsEnabled || + !preferences.value.notifyOnInput + ) { + return; + } + // A queued alert can outlive the link or the environment's permission. + const link = yield* links.getForUser({ + userId: job.userId, + environmentId: job.directAlert.environmentId, + }); + if (!link) return; + const deliveryUsers = yield* links.listDeliveryUsersForEnvironment({ + environmentId: job.directAlert.environmentId, + environmentPublicKey: link.environmentPublicKey, + }); + const deliveryUser = deliveryUsers.find((user) => user.userId === job.userId); + if (!deliveryUser?.notificationsEnabled) return; + const states = preferences.value.liveActivitiesEnabled + ? yield* rows.listForUser({ userId: job.userId }) + : []; + const aggregate = makeAggregateState({ + activeStates: states, + terminalState: null, + nowMs: now.epochMilliseconds, + }); + const alert = { + alert_id: [ + job.directAlert.environmentId, + job.directAlert.threadId, + job.directAlert.eventId, + ].join("\n"), + alert_title: job.directAlert.title, + alert_body: job.directAlert.body, + alert_path: job.directAlert.deepLink, + }; + const data: Record = { + t3_kind: "agent_activity", + // Current clients skip the activity update. Older ones still need the + // live card fields, or a missing `active` clears the card. + alert_only: "true", + device_id: job.deviceId, + user_id: job.userId, + updated_at: String(now.epochMilliseconds), + ...androidActivityData(preferences.value.liveActivitiesEnabled ? aggregate : null), + ...alert, + }; + const digest = yield* crypto.digest( + Digest.sha256, + new TextEncoder().encode(alert.alert_id), + ); + data.alert_id = Array.from(digest, (byte) => byte.toString(16).padStart(2, "0")).join(""); + const result = yield* client.send({ + token: job.token, + packageName: target.bundle_id, + alert: true, + data: fitFcmData(data), + }); + if (result.unregistered) { + yield* db + .update(relayMobileDevices) + .set({ pushToken: null }) + .where( + and( + eq(relayMobileDevices.userId, job.userId), + eq(relayMobileDevices.deviceId, job.deviceId), + eq(relayMobileDevices.pushToken, job.token), + ), + ) + .pipe( + Effect.mapError( + (cause) => new FcmDeliveryError({ operation: "invalidate-token", cause }), + ), + ); + } + return; + } const targets = yield* devices.listTargets({ userId: job.userId }); const target = targets.find( (device) => diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 1865a633d5715..56f21b366bae7 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -21,6 +21,7 @@ import * as AgentActivityPublisher from "./AgentActivityPublisher.ts"; import * as FcmDeliveries from "./FcmDeliveries.ts"; const layerPublisher = AgentActivityPublisher.layer.pipe( + Layer.provide(NodeCryptoLayer.layer), Layer.provide( Layer.succeed(FcmDeliveries.FcmDeliveries, { enqueue: () => Effect.succeed(null), @@ -192,6 +193,7 @@ function makeAgentActivityPublisher( ): AgentActivityPublisher.AgentActivityPublisher["Service"] { return { publish: () => Effect.succeed({ ok: true, deliveries: [] }), + publishAlert: () => Effect.succeed({ ok: true, deliveries: [] }), replayForLiveActivityRegistration: () => Effect.succeed(null), endIdleLiveActivities: Effect.void, ...overrides, diff --git a/infra/relay/src/environments/EnvironmentPublishSignatures.ts b/infra/relay/src/environments/EnvironmentPublishSignatures.ts index 625ce46ce4d43..f89feef315881 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.ts @@ -2,11 +2,14 @@ import { RelayAgentActivityPublishProofPayload, RelayAgentActivityPublishProofInvalidReason, type RelayAgentActivityPublishRequest, + RelayAgentAlertPublishProofPayload, + type RelayAgentAlertPublishRequest, } from "@t3tools/contracts/relay"; import { decodeRelayJwt, normalizeRelayIssuer, RELAY_ACTIVITY_PUBLISH_TYP, + RELAY_AGENT_ALERT_TYP, verifyRelayJwt, } from "@t3tools/shared/relayJwt"; import { stableStringify } from "@t3tools/shared/relaySigning"; @@ -71,6 +74,12 @@ export class EnvironmentPublishSignatures extends Context.Service< readonly threadId: string; readonly request: RelayAgentActivityPublishRequest; }) => Effect.Effect; + readonly verifyAlert: (input: { + readonly environmentId: string; + readonly environmentPublicKey: string; + readonly threadId: string; + readonly request: RelayAgentAlertPublishRequest; + }) => Effect.Effect; } >()("t3code-relay/environments/EnvironmentPublishSignatures") {} @@ -94,8 +103,73 @@ const make = Effect.gen(function* () { const config = yield* RelayConfiguration.RelayConfiguration; const crypto = yield* Digest.Digest; const signature = yield* Signature.Signature; + const verifyAlert = Effect.fn("relay.environment_publish_signatures.verify_alert")( + function* (input: { + readonly environmentId: string; + readonly environmentPublicKey: string; + readonly threadId: string; + readonly request: RelayAgentAlertPublishRequest; + }) { + const now = yield* DateTime.now; + const decoded = yield* decodeRelayJwt(input.request.proof).pipe( + Effect.mapError( + (cause) => + new EnvironmentPublishSignatureInvalid({ + environmentId: input.environmentId, + threadId: input.threadId, + reason: "invalid_signature_or_payload", + stage: "decode_token", + cause, + }), + ), + ); + if ( + typeof decoded.exp === "number" && + decoded.exp <= Math.floor(now.epochMilliseconds / 1_000) + ) { + return yield* new EnvironmentPublishSignatureExpired({ + environmentId: input.environmentId, + threadId: input.threadId, + expiresAt: DateTime.formatIso(DateTime.makeUnsafe(decoded.exp * 1_000)), + }); + } + const proof = yield* verifyRelayJwt(RelayAgentAlertPublishProofPayload, { + publicKey: input.environmentPublicKey, + token: input.request.proof, + typ: RELAY_AGENT_ALERT_TYP, + issuer: `t3-env:${input.environmentId}`, + audience: normalizeRelayIssuer(config.relayIssuer), + }).pipe( + Effect.provideService(Signature.Signature, signature), + Effect.mapError( + (cause) => + new EnvironmentPublishSignatureInvalid({ + environmentId: input.environmentId, + threadId: input.threadId, + reason: "invalid_signature_or_payload", + stage: "verify_proof", + cause, + }), + ), + ); + if ( + proof.environmentId !== input.environmentId || + proof.threadId !== input.threadId || + proof.sub !== input.environmentId || + stableStringify(proof.alert) !== stableStringify(input.request.alert) + ) { + return yield* new EnvironmentPublishSignatureInvalid({ + environmentId: input.environmentId, + threadId: input.threadId, + reason: "invalid_signature_or_payload", + stage: "validate_claims", + }); + } + }, + ); return EnvironmentPublishSignatures.of({ + verifyAlert, verify: Effect.fn("relay.environment_publish_signatures.verify")(function* (input) { yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId, diff --git a/infra/relay/src/http/Api.test.ts b/infra/relay/src/http/Api.test.ts index 3abb8b28f4d6a..bff861fa121bb 100644 --- a/infra/relay/src/http/Api.test.ts +++ b/infra/relay/src/http/Api.test.ts @@ -1389,6 +1389,7 @@ describe("relay routing fallback", () => { }), replayForLiveActivityRegistration: () => Effect.succeed(null), endIdleLiveActivities: Effect.void, + publishAlert: () => Effect.succeed({ ok: true as const, deliveries: [] }), }); const layerSignatures = Layer.succeed( EnvironmentPublishSignatures.EnvironmentPublishSignatures, @@ -1397,6 +1398,7 @@ describe("relay routing fallback", () => { Effect.sync(() => { verified.push(input); }), + verifyAlert: () => Effect.void, }, ); const layerAuth = Layer.succeed(RelayEnvironmentAuth, { diff --git a/infra/relay/src/http/Api.ts b/infra/relay/src/http/Api.ts index f06d441c12c48..752754819e9b6 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -1337,6 +1337,52 @@ export const layerServerApi = HttpApiBuilder.group( ); return activityHandlers + .handle( + "publishAgentAlert", + Effect.fn("relay.api.server.publishAgentAlert")( + function* (args) { + const { params, payload } = args; + const principal = yield* RelayEnvironmentPrincipal; + if (principal.environmentId !== params.environmentId) { + return yield* new HttpApiError.Unauthorized({}); + } + yield* publishSignatures.verifyAlert({ + environmentId: params.environmentId, + environmentPublicKey: principal.environmentPublicKey, + threadId: params.threadId, + request: payload, + }); + return yield* publisher.publishAlert({ + environmentId: params.environmentId, + environmentPublicKey: principal.environmentPublicKey, + threadId: params.threadId, + title: payload.alert.title, + body: payload.alert.body, + deepLink: payload.alert.deepLink, + }); + }, + mapErrorTags({ + EnvironmentPublishSignatureExpired: (_error, traceId) => + new RelayAgentActivityPublishProofExpiredError({ + code: "agent_activity_publish_proof_expired", + traceId, + }), + EnvironmentPublishSignatureInvalid: (_error, traceId) => + new RelayAgentActivityPublishProofInvalidError({ + code: "agent_activity_publish_proof_invalid", + reason: "invalid_signature_or_payload", + traceId, + }), + DpopProofReplayPersistenceError: (_error, traceId) => + new RelayInternalError({ + code: "internal_error", + reason: "persistence_failed", + traceId, + }), + }), + mapRelayCommonApiErrors("not_authorized"), + ), + ) .handle( "registerManagedEndpointRecovery", Effect.fn("relay.api.server.registerManagedEndpointRecovery")( diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index e74abde6a2c59..19dfc64065d56 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -58,6 +58,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.subscribeAuthAccess | typeof WS_METHODS.subscribeServerConfig | typeof WS_METHODS.subscribeServerLifecycle + | typeof WS_METHODS.serverSubscribeAgentAlerts | typeof WS_METHODS.scheduledTasksSubscribe | typeof WS_METHODS.serverGetStorageCleanupReport | typeof WS_METHODS.serverGetDatabaseStorage diff --git a/packages/contracts/src/agentAlert.ts b/packages/contracts/src/agentAlert.ts new file mode 100644 index 0000000000000..8c01085b54644 --- /dev/null +++ b/packages/contracts/src/agentAlert.ts @@ -0,0 +1,9 @@ +import * as Schema from "effect/Schema"; + +import { ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +export const AgentAlert = Schema.Struct({ + threadId: ThreadId, + body: TrimmedNonEmptyString, +}); +export type AgentAlert = typeof AgentAlert.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 8dd47df33a21f..29420269634e9 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -67,6 +67,7 @@ export * from "./usage.ts"; export * from "./scheduledTask.ts"; export * from "./worktreeMcp.ts"; export * from "./resourceTelemetry.ts"; +export * from "./agentAlert.ts"; export * from "./rpc.ts"; export * from "./worktreeSetup.ts"; export * from "./secretRequest.ts"; diff --git a/packages/contracts/src/relay.ts b/packages/contracts/src/relay.ts index 8ebc9df23a0bb..e672288fbebe0 100644 --- a/packages/contracts/src/relay.ts +++ b/packages/contracts/src/relay.ts @@ -277,6 +277,31 @@ export const RelayAgentActivityPublishRequest = Schema.Struct({ }).annotate({ description: "Publishes a signed agent-awareness update from an environment." }); export type RelayAgentActivityPublishRequest = typeof RelayAgentActivityPublishRequest.Type; +export const RelayAgentAlert = Schema.Struct({ + title: TrimmedNonEmptyString, + body: TrimmedNonEmptyString, + deepLink: TrimmedNonEmptyString, +}); +export type RelayAgentAlert = typeof RelayAgentAlert.Type; + +export const RelayAgentAlertPublishProofPayload = Schema.Struct({ + ...RelaySignedJwtRegisteredClaims, + environmentId: EnvironmentId, + threadId: ThreadId, + alert: RelayAgentAlert, +}); +export type RelayAgentAlertPublishProofPayload = typeof RelayAgentAlertPublishProofPayload.Type; + +export const RelayAgentAlertPublishRequest = Schema.Struct({ + alert: RelayAgentAlert, + proof: TrimmedNonEmptyString.annotate({ + description: "Environment-signed JWT covering this alert.", + }), +}).annotate({ + description: "Publishes a one-shot agent alert without changing the thread's activity phase.", +}); +export type RelayAgentAlertPublishRequest = typeof RelayAgentAlertPublishRequest.Type; + export const RelayEnvironmentLinkScope = Schema.Literals([ "agent_activity_notifications", "managed_tunnels", @@ -1252,6 +1277,19 @@ const RelayServerGroup = HttpApiGroup.make("server") error: RelayAgentActivityPublishErrors, }, ).annotate(OpenApi.Summary, "Publish agent activity"), + HttpApiEndpoint.post( + "publishAgentAlert", + "/v1/environments/:environmentId/threads/:threadId/agent-alert", + { + params: Schema.Struct({ + environmentId: EnvironmentId, + threadId: ThreadId, + }), + payload: RelayAgentAlertPublishRequest, + success: RelayPublishResponse, + error: RelayAgentActivityPublishErrors, + }, + ).annotate(OpenApi.Summary, "Publish a one-shot agent alert"), HttpApiEndpoint.post( "updateLinkPreferences", "/v1/environments/:environmentId/link-preferences", diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 6465558d5fe03..d788c198a715a 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -37,6 +37,7 @@ import * as Schema from "effect/Schema"; import * as Rpc from "effect/rpc/Rpc"; import * as RpcGroup from "effect/rpc/RpcGroup"; import * as RpcMiddleware from "effect/rpc/RpcMiddleware"; +import { AgentAlert } from "./agentAlert.ts"; import { NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; import { ProviderAuthCancelInput, @@ -536,6 +537,7 @@ export const WS_METHODS = { serverPrepareTriage: "server.prepareTriage", serverSignalProcess: "server.signalProcess", serverReportClientActivity: "server.reportClientActivity", + serverSubscribeAgentAlerts: "server.subscribeAgentAlerts", serverGetUsageSummary: "server.getUsageSummary", serverRefreshUsageRates: "server.refreshUsageRates", @@ -997,6 +999,13 @@ const WsServerReportClientActivityRpc = Rpc.make(WS_METHODS.serverReportClientAc error: EnvironmentAuthorizationError, }); +const WsServerSubscribeAgentAlertsRpc = Rpc.make(WS_METHODS.serverSubscribeAgentAlerts, { + payload: Schema.Struct({}), + success: AgentAlert, + error: EnvironmentAuthorizationError, + stream: true, +}); + const PullRequestRpcError = Schema.Union([ PullRequestUnavailableError, PullRequestOperationError, @@ -2066,6 +2075,7 @@ export const WsRpcGroup = RpcGroup.make( WsScheduledTasksListWebhookDeliveriesRpc, WsScheduledTasksGetWebhookDeliveryRpc, WsServerReportClientActivityRpc, + WsServerSubscribeAgentAlertsRpc, WsCloudGetRelayClientStatusRpc, WsCloudInstallRelayClientRpc, WsPullRequestsListRpc, diff --git a/packages/shared/src/relayJwt.ts b/packages/shared/src/relayJwt.ts index c85356421ade3..908e6c5f3085c 100644 --- a/packages/shared/src/relayJwt.ts +++ b/packages/shared/src/relayJwt.ts @@ -14,6 +14,7 @@ export const RELAY_HEALTH_REQUEST_TYP = "t3-cloud-health+jwt"; export const RELAY_MINT_RESPONSE_TYP = "t3-env-mint+jwt"; export const RELAY_HEALTH_RESPONSE_TYP = "t3-env-health+jwt"; export const RELAY_ACTIVITY_PUBLISH_TYP = "t3-env-activity+jwt"; +export const RELAY_AGENT_ALERT_TYP = "t3-env-alert+jwt"; export const RELAY_MANAGED_TUNNEL_RECOVERY_TYP = "t3-env-managed-tunnel-recovery+jwt"; export const RELAY_HOOK_DELIVERY_TYP = "t3-relay-hook-delivery+jwt"; export const RELAY_HOOK_DELIVERY_HEADER = "x-t3-relay-delivery";