From a5430a7a512f495f9a733d08e634f59199a76431 Mon Sep 17 00:00:00 2001 From: watashiwaaniket Date: Wed, 28 Oct 2026 02:28:37 +0530 Subject: [PATCH 01/14] fix(server): deliver Claude push notifications through T3 Claude Code treats the SDK session as an active terminal, so PushNotification never reaches the user. Intercept the tool and deliver the message through T3 clients and phone push. Co-authored-by: Cursor --- apps/server/src/auth/RpcAuthorization.ts | 1 + .../src/notifications/AgentAlerts.test.ts | 41 ++++++ apps/server/src/notifications/AgentAlerts.ts | 69 ++++++++++ .../src/observability/RpcInstrumentation.ts | 1 + .../Adapters/ClaudeAdapterV2.ts | 21 +++ .../Adapters/claudePushNotification.test.ts | 78 +++++++++++ .../Adapters/claudePushNotification.ts | 86 ++++++++++++ apps/server/src/relay/AgentAwarenessRelay.ts | 97 ++++++++++++++ apps/server/src/server.ts | 7 +- apps/server/src/ws.ts | 9 ++ .../ThreadNotificationCoordinator.tsx | 85 ++++++++++++ apps/web/src/connection/runtime.ts | 11 +- apps/web/src/lib/agentAlertNotifications.ts | 60 +++++++++ apps/web/src/lib/agentAlertPresenter.ts | 19 +++ apps/web/src/rpc/requestLatencyState.ts | 1 + .../AgentActivityPublisher.test.ts | 122 ++++++++++++++++++ .../agentActivity/AgentActivityPublisher.ts | 108 +++++++++++++++- .../relay/src/agentActivity/FcmDeliveries.ts | 72 +++++++++++ .../agentActivity/MobileRegistrations.test.ts | 1 + .../EnvironmentPublishSignatures.ts | 115 +++++++++++++++++ infra/relay/src/http/Api.test.ts | 2 + infra/relay/src/http/Api.ts | 52 ++++++++ packages/client-runtime/src/rpc/client.ts | 1 + packages/contracts/src/agentAlert.ts | 9 ++ packages/contracts/src/index.ts | 1 + packages/contracts/src/relay.ts | 38 ++++++ packages/contracts/src/rpc.ts | 10 ++ packages/shared/src/relayJwt.ts | 1 + 28 files changed, 1107 insertions(+), 11 deletions(-) create mode 100644 apps/server/src/notifications/AgentAlerts.test.ts create mode 100644 apps/server/src/notifications/AgentAlerts.ts create mode 100644 apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts create mode 100644 apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts create mode 100644 apps/web/src/lib/agentAlertNotifications.ts create mode 100644 apps/web/src/lib/agentAlertPresenter.ts create mode 100644 packages/contracts/src/agentAlert.ts diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index f127f94fdba35..01d268d361c16 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -108,6 +108,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..2f62c0170c694 --- /dev/null +++ b/apps/server/src/notifications/AgentAlerts.test.ts @@ -0,0 +1,41 @@ +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 }); + }).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({ + armed: Effect.succeed(true), + publish: (item) => Deferred.succeed(published, item).pipe(Effect.asVoid), + }); + expect(yield* alerts.offer(alert)).toEqual({ local: true, phone: true }); + 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..faa4707012ff9 --- /dev/null +++ b/apps/server/src/notifications/AgentAlerts.ts @@ -0,0 +1,69 @@ +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 armed: Effect.Effect; + 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") {} + +export const make = Effect.gen(function* () { + const scope = yield* Effect.scope; + const subscribers = yield* Ref.make(0); + const alerts = yield* PubSub.unbounded(); + const phone = yield* Ref.make(null); + + const stream = Stream.unwrap( + Effect.gen(function* () { + yield* Ref.update(subscribers, (count) => count + 1); + yield* Effect.addFinalizer(() => + Ref.update(subscribers, (count) => (count > 0 ? count - 1 : 0)), + ); + return Stream.fromPubSub(alerts); + }), + ); + + const offer: AgentAlerts["Service"]["offer"] = Effect.fn("AgentAlerts.offer")(function* (alert) { + const delivery = yield* Ref.get(phone); + const [local, phoneArmed] = yield* Effect.all( + [ + Ref.get(subscribers).pipe(Effect.map((count) => count > 0)), + delivery === null ? Effect.succeed(false) : delivery.armed, + ], + { concurrency: 2 }, + ); + if (local) yield* PubSub.publish(alerts, alert); + if (phoneArmed && delivery !== null) { + yield* delivery.publish(alert).pipe( + Effect.catchCause((cause) => + Effect.logWarning("agent alert phone delivery failed", { cause }), + ), + Effect.forkIn(scope), + ); + } + return { local, phone: phoneArmed }; + }); + + 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 504e533b37e05..f8698096535a9 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -85,6 +85,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.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 5c676d684f686..deaf606adc6cc 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1,6 +1,8 @@ import * as DocumentText from "../../provider/DocumentText.ts"; 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, @@ -1453,6 +1455,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); const { @@ -1551,6 +1557,9 @@ 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 })), + }), systemPrompt: { type: "preset" as const, preset: "claude_code" as const, @@ -4302,6 +4311,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, crypto, idAllocator, queryRunner } = adapterOptions; @@ -9837,6 +9847,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 querySession = yield* queryRunner .open({ 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..64584dc52adf7 --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts @@ -0,0 +1,78 @@ +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 is connected, and phone push is off.", + 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, + }, + }, + }); + }); +}); 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..9bf7a8ed979c6 --- /dev/null +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts @@ -0,0 +1,86 @@ +import type { HookCallback, SyncHookJSONOutput } from "@anthropic-ai/claude-agent-sdk"; + +export 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 is connected, and phone push is off.", + 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", + }, + }, +}; + +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 {}; + } + 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.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index cbe7b35ef409e..c72a558dd9cc5 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -6,10 +6,13 @@ import type { Project, 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"; @@ -18,6 +21,7 @@ import { withRelayClientTracing } from "@t3tools/shared/relayTracing"; import { normalizeRelayIssuer, RELAY_ACTIVITY_PUBLISH_TYP, + RELAY_AGENT_ALERT_TYP, signRelayJwt, } from "@t3tools/shared/relayJwt"; import * as Cause from "effect/Cause"; @@ -47,6 +51,7 @@ import { } from "../cloud/config.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"; @@ -269,6 +274,34 @@ const makePublishProof = Effect.fn("makePublishProof")(function* (input: { 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, ): Record { @@ -360,6 +393,7 @@ 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; @@ -837,6 +871,69 @@ export const make = Effect.gen(function* () { }, ); + const phoneArmed = Effect.gen(function* () { + const [relayConfig, publishEnabled] = yield* Effect.all([ + readRelayConfig.pipe(Effect.orElseSucceed(() => null)), + readPublishAgentActivityEnabled.pipe(Effect.orElseSucceed(() => false)), + ]); + return relayConfig !== null && publishEnabled; + }); + const publishPhone = Effect.fn("AgentAwarenessRelay.publishAgentAlert")(function* ( + alert: AgentAlert, + ) { + const relayConfig = yield* readRelayConfig; + const publishEnabled = yield* readPublishAgentActivityEnabled; + if (!relayConfig || !publishEnabled) return; + if ( + publishedRelayConfig?.url !== relayConfig.url || + publishedRelayConfig.issuer !== relayConfig.issuer || + publishedRelayConfig.environmentCredential !== relayConfig.environmentCredential + ) { + cachedRelayClient = undefined; + publishedRelayConfig = relayConfig; + } + if (cachedRelayClient === undefined) { + cachedRelayClient = yield* makeRelayClient(relayConfig); + } + 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* crypto.randomUUIDv4(), + }); + yield* cachedRelayClient.server + .publishAgentAlert({ + params: { environmentId, threadId: alert.threadId }, + payload: { alert: phoneAlert, proof }, + }) + .pipe(Effect.timeout("10 seconds")); + }); + if (Option.isSome(agentAlerts)) { + yield* agentAlerts.value.registerPhone({ + armed: phoneArmed, + publish: (alert) => + publishPhone(alert).pipe( + Effect.catchCause((cause) => + Effect.logWarning("agent alert phone push failed", { + threadId: alert.threadId, + cause, + }), + ), + ), + }); + } + return AgentAwarenessRelay.of({ publishThread, drain: worker.drain, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 741fc73103b7d..73818f3bbfce4 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -91,6 +91,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 ProjectEnrichmentService from "./project/ProjectEnrichmentService.ts"; @@ -580,8 +581,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, @@ -735,7 +738,7 @@ const layerMakeRoutes = Layer.mergeAll( DeviceHubProxy.layer, ServerBrowserStream.routeLayer, ServerHttp.layerStaticAndDevRoute, - Ws.layer, + Ws.layer.pipe(Layer.provide(layerAgentAlerts)), ), McpHttpServer.layer.pipe( Layer.provide(ProviderAdapterRegistry.layerFromProviderInstanceRegistry), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index e35aaf1bb93cd..dd7351ea59176 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 Crypto from "effect/Crypto"; +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"; @@ -1338,6 +1339,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( @@ -2541,6 +2543,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) => @@ -3246,6 +3253,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 pullRequests = yield* PullRequestService.PullRequestService; @@ -3319,6 +3327,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(ServerSelfUpdate.ServerSelfUpdate, serverSelfUpdate)), diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index fa4fdbb672faa..052ccddfc53dc 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -10,6 +10,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, @@ -17,6 +18,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"; @@ -335,6 +337,88 @@ export function startThreadNotifications( }; }; + const presentAgentAlert = (alert: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly body: string; + }) => { + 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 = toastManager.add({ + type: "warning", + title, + description: alert.body, + data: { + hideCopyButton: true, + leadingIcon: , + }, + actionProps: { + children: "Open thread", + onClick: () => { + toastManager.close(toastId); + openThread(); + }, + }, + }); + return; + } + if ( + !hasDesktopNotifications(mode) || + (document.visibilityState === "visible" && document.hasFocus()) || + 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, 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(); @@ -350,6 +434,7 @@ export function startThreadNotifications( onSettings(); return () => { + setAgentAlertPresenter(null); unsubscribeSettings(); unsubscribeEnvironmentIds(); for (const entry of watched.values()) entry.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..ae6232c3dd484 --- /dev/null +++ b/apps/web/src/lib/agentAlertNotifications.ts @@ -0,0 +1,60 @@ +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 Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import { emitAgentAlert } from "./agentAlertPresenter"; + +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + if (typeof window === "undefined") return; + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const running = yield* Ref.make(new Map>()); + const sync = Effect.fnUntraced(function* (entries: ReadonlyMap) { + const current = yield* Ref.get(running); + for (const [environmentId, fiber] of current) { + if (entries.has(environmentId)) continue; + yield* Fiber.interrupt(fiber); + yield* Ref.update(running, (next) => { + const updated = new Map(next); + updated.delete(environmentId); + return updated; + }); + } + for (const environmentId of entries.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* Stream.concat( + Stream.fromEffect(SubscriptionRef.get(registry.entries)), + SubscriptionRef.changes(registry.entries), + ).pipe( + Stream.runForEach((entries) => sync(entries)), + 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..d3e4b4a3e6126 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -521,4 +521,126 @@ 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( + 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..29ad77df3dcc6 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -2,18 +2,21 @@ 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 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 +25,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 +47,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; @@ -112,6 +127,80 @@ 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, + 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 +332,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.ts b/infra/relay/src/agentActivity/FcmDeliveries.ts index 615840fe27aeb..2c60e1798b7d7 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.ts @@ -39,6 +39,15 @@ 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, + }), + ), }); export type FcmDeliveryJob = typeof FcmDeliveryJob.Type; const decodeJob = Schema.decodeUnknownEffect(FcmDeliveryJob); @@ -120,6 +129,7 @@ export class FcmDeliveries extends Context.Service< readonly target: LiveActivities.TargetRow; readonly state: RelayAgentActivityState | null; readonly replay?: boolean; + readonly directAlert?: FcmDeliveryJob["directAlert"]; }) => Effect.Effect; readonly process: ( body: unknown, @@ -169,6 +179,7 @@ const make = Effect.gen(function* () { token: input.target.push_token, state: input.state, ...(input.replay ? { replay: true } : {}), + ...(input.directAlert === undefined ? {} : { directAlert: input.directAlert }), queuedAt: now.epochMilliseconds, }) .pipe(Effect.mapError((cause) => new FcmDeliveryError({ operation: "enqueue", cause }))); @@ -188,6 +199,67 @@ 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; + } + const alert = { + alert_id: [ + job.directAlert.environmentId, + job.directAlert.threadId, + job.directAlert.body, + ].join("\n"), + alert_title: job.directAlert.title, + alert_body: job.directAlert.body, + alert_path: job.directAlert.deepLink, + }; + const data: Record = { + t3_kind: "agent_activity", + device_id: job.deviceId, + user_id: job.userId, + updated_at: String(now.epochMilliseconds), + ...alert, + }; + const digest = yield* crypto.digest("SHA-256", 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 6f5997c931bdb..3f105755c4c40 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -191,6 +191,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 4c89bff5342b0..3738843562b6a 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.ts @@ -1,12 +1,15 @@ import { RelayAgentActivityPublishProofPayload, + RelayAgentAlertPublishProofPayload, RelayAgentActivityPublishProofInvalidReason, type RelayAgentActivityPublishRequest, + 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"; @@ -70,10 +73,17 @@ 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") {} const decodeProof = Schema.decodeUnknownEffect(RelayAgentActivityPublishProofPayload); +const decodeAlertProof = Schema.decodeUnknownEffect(RelayAgentAlertPublishProofPayload); function environmentPublishReplayThumbprintData(input: { readonly environmentId: string; @@ -204,6 +214,111 @@ const make = Effect.gen(function* () { }); } }), + verifyAlert: Effect.fn("relay.environment_publish_signatures.verify_alert")(function* (input) { + yield* Effect.annotateCurrentSpan({ + "relay.environment_id": input.environmentId, + "relay.thread_id": input.threadId, + }); + const now = yield* DateTime.now; + const decoded = yield* Effect.try({ + try: () => decodeRelayJwt(input.request.proof), + catch: (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({ + publicKey: input.environmentPublicKey, + token: input.request.proof, + typ: RELAY_AGENT_ALERT_TYP, + issuer: `t3-env:${input.environmentId}`, + audience: normalizeRelayIssuer(config.relayIssuer), + nowEpochSeconds: Math.floor(now.epochMilliseconds / 1_000), + }).pipe( + Effect.flatMap(decodeAlertProof), + 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", + }); + } + const expiresAt = DateTime.make(proof.exp * 1_000); + if (expiresAt._tag === "None") { + return yield* new EnvironmentPublishSignatureInvalid({ + environmentId: input.environmentId, + threadId: input.threadId, + reason: "invalid_signature_or_payload", + stage: "validate_expiration", + }); + } + const thumbprint = yield* crypto + .digest( + "SHA-256", + environmentPublishReplayThumbprintData({ + environmentId: input.environmentId, + environmentPublicKey: input.environmentPublicKey, + }), + ) + .pipe( + Effect.map(formatEnvironmentPublishReplayThumbprint), + Effect.mapError( + (cause) => + new EnvironmentPublishSignatureInvalid({ + environmentId: input.environmentId, + threadId: input.threadId, + reason: "invalid_signature_or_payload", + stage: "generate_replay_thumbprint", + cause, + }), + ), + ); + const consumedNonce = yield* proofReplay.consume({ + thumbprint, + jti: proof.jti, + iat: proof.iat, + expiresAt: expiresAt.value, + }); + if (!consumedNonce) { + return yield* new EnvironmentPublishSignatureInvalid({ + environmentId: input.environmentId, + threadId: input.threadId, + reason: "replayed_nonce", + stage: "consume_nonce", + }); + } + }), }); }); diff --git a/infra/relay/src/http/Api.test.ts b/infra/relay/src/http/Api.test.ts index 39cdd619febe1..04803da2d3fdd 100644 --- a/infra/relay/src/http/Api.test.ts +++ b/infra/relay/src/http/Api.test.ts @@ -1376,6 +1376,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, @@ -1384,6 +1385,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 53551d6d344f1..34ba20f5f604a 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -1307,6 +1307,58 @@ 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({ + EnvironmentPublishPublicKeyMissing: (_error, traceId) => + new RelayAuthInvalidError({ + code: "auth_invalid", + reason: "not_authorized", + traceId, + }), + 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 2832a4e231556..7e8377c9e7f6d 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 e61597e52ecdb..8fef085af73ee 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -64,6 +64,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 b11b7cca5ef55..d0d3cfc4cdfab 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", @@ -1209,6 +1234,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 e4823f30272bd..5e3b00cd4eb6f 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, @@ -518,6 +519,7 @@ export const WS_METHODS = { serverPrepareTriage: "server.prepareTriage", serverSignalProcess: "server.signalProcess", serverReportClientActivity: "server.reportClientActivity", + serverSubscribeAgentAlerts: "server.subscribeAgentAlerts", serverGetUsageSummary: "server.getUsageSummary", serverRefreshUsageRates: "server.refreshUsageRates", @@ -967,6 +969,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, @@ -1979,6 +1988,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 ba309b028e313..a0406f2fde457 100644 --- a/packages/shared/src/relayJwt.ts +++ b/packages/shared/src/relayJwt.ts @@ -10,6 +10,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"; From f48fe54b94348c59bbcab69b591fdd9242eeb026 Mon Sep 17 00:00:00 2001 From: watashiwaaniket Date: Wed, 28 Oct 2026 02:50:16 +0530 Subject: [PATCH 02/14] fix(server): keep Claude alerts on one shared path Claude only sees AgentAlerts when that layer is provided into provider hydration. Phone alerts use their own relay client, and Android alerts keep a stable id without clearing the activity card or outliving a revoked link. --- .../AgentNotifications.kt | 6 +- .../AgentNotificationsTest.kt | 19 +++++++ apps/server/src/relay/AgentAwarenessRelay.ts | 23 +++++--- apps/server/src/server.ts | 5 +- .../src/agentActivity/FcmDeliveries.test.ts | 55 +++++++++++++++++++ .../relay/src/agentActivity/FcmDeliveries.ts | 42 +++++++++++++- 6 files changed, 137 insertions(+), 13 deletions(-) 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 f9b69a7117741..ab937aa105164 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" showAlert(context, prefs, scheme, data) - updateActivity(context, prefs, scheme, data, updatedAt) + // A one-shot alert carries no activity transition. Missing activity + // fields would otherwise end the card. + if (data["alert_only"] != "true") { + updateActivity(context, prefs, scheme, data, updatedAt) + } } } 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 7ca6edbc2133e..c2d4f7634c425 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 @@ -102,6 +102,25 @@ class AgentNotificationsTest { 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/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index c72a558dd9cc5..09c0c5d1483bb 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -441,6 +441,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(); @@ -885,16 +889,19 @@ export const make = Effect.gen(function* () { const publishEnabled = yield* readPublishAgentActivityEnabled; if (!relayConfig || !publishEnabled) return; if ( - publishedRelayConfig?.url !== relayConfig.url || - publishedRelayConfig.issuer !== relayConfig.issuer || - publishedRelayConfig.environmentCredential !== relayConfig.environmentCredential + phoneRelayConfig?.url !== relayConfig.url || + phoneRelayConfig.issuer !== relayConfig.issuer || + phoneRelayConfig.environmentCredential !== relayConfig.environmentCredential ) { - cachedRelayClient = undefined; - publishedRelayConfig = relayConfig; + phoneRelayClient = undefined; + phoneRelayConfig = relayConfig; } - if (cachedRelayClient === undefined) { - cachedRelayClient = yield* makeRelayClient(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"; @@ -912,7 +919,7 @@ export const make = Effect.gen(function* () { alert: phoneAlert, jti: yield* crypto.randomUUIDv4(), }); - yield* cachedRelayClient.server + yield* relayClient.server .publishAgentAlert({ params: { environmentId, threadId: alert.threadId }, payload: { alert: phoneAlert, proof }, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 73818f3bbfce4..cd4366708879c 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -641,7 +641,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), 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 2c60e1798b7d7..52e2683d38a08 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.ts @@ -46,6 +46,9 @@ export const FcmDeliveryJob = Schema.Struct({ 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, }), ), }); @@ -129,7 +132,9 @@ export class FcmDeliveries extends Context.Service< readonly target: LiveActivities.TargetRow; readonly state: RelayAgentActivityState | null; readonly replay?: boolean; - readonly directAlert?: FcmDeliveryJob["directAlert"]; + readonly directAlert?: Omit, "eventId"> & { + readonly eventId?: string; + }; }) => Effect.Effect; readonly process: ( body: unknown, @@ -172,6 +177,13 @@ const make = Effect.gen(function* () { }; } const now = yield* DateTime.now; + const directAlert = + input.directAlert === undefined + ? undefined + : { + ...input.directAlert, + eventId: input.directAlert.eventId ?? (yield* crypto.randomUUIDv4()), + }; yield* sender .send({ userId: input.target.user_id, @@ -179,7 +191,7 @@ const make = Effect.gen(function* () { token: input.target.push_token, state: input.state, ...(input.replay ? { replay: true } : {}), - ...(input.directAlert === undefined ? {} : { directAlert: input.directAlert }), + ...(directAlert === undefined ? {} : { directAlert }), queuedAt: now.epochMilliseconds, }) .pipe(Effect.mapError((cause) => new FcmDeliveryError({ operation: "enqueue", cause }))); @@ -216,11 +228,31 @@ const make = Effect.gen(function* () { ) { 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.body, + job.directAlert.eventId, ].join("\n"), alert_title: job.directAlert.title, alert_body: job.directAlert.body, @@ -228,9 +260,13 @@ const make = Effect.gen(function* () { }; 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("SHA-256", new TextEncoder().encode(alert.alert_id)); From 8d52d31ba3d9951af3a03b90de9efe60d90ea29e Mon Sep 17 00:00:00 2001 From: watashiwaaniket Date: Wed, 28 Oct 2026 03:04:27 +0530 Subject: [PATCH 03/14] fix(server): deliver Claude alerts on the instance clients read HttpRouter.serve builds routes on a forked memo map, so a second AgentAlerts inside the websocket route never saw Claude's publish. A client with notifications off also no longer counts as a local delivery. Co-authored-by: Cursor --- apps/server/src/notifications/AgentAlerts.ts | 7 +++-- apps/server/src/server.ts | 4 ++- apps/web/src/lib/agentAlertNotifications.ts | 28 +++++++++++++++++--- 3 files changed, 32 insertions(+), 7 deletions(-) diff --git a/apps/server/src/notifications/AgentAlerts.ts b/apps/server/src/notifications/AgentAlerts.ts index faa4707012ff9..f39267aea35ae 100644 --- a/apps/server/src/notifications/AgentAlerts.ts +++ b/apps/server/src/notifications/AgentAlerts.ts @@ -22,7 +22,7 @@ export class AgentAlerts extends Context.Service< } >()("t3/notifications/AgentAlerts") {} -export const make = Effect.gen(function* () { +const make = Effect.gen(function* () { const scope = yield* Effect.scope; const subscribers = yield* Ref.make(0); const alerts = yield* PubSub.unbounded(); @@ -30,11 +30,14 @@ export const make = Effect.gen(function* () { 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.fromPubSub(alerts); + return Stream.fromSubscription(subscription); }), ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index cd4366708879c..ea21e6404474c 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -741,7 +741,9 @@ const layerMakeRoutes = Layer.mergeAll( DeviceHubProxy.layer, ServerBrowserStream.routeLayer, ServerHttp.layerStaticAndDevRoute, - Ws.layer.pipe(Layer.provide(layerAgentAlerts)), + // 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, ), McpHttpServer.layer.pipe( Layer.provide(ProviderAdapterRegistry.layerFromProviderInstanceRegistry), diff --git a/apps/web/src/lib/agentAlertNotifications.ts b/apps/web/src/lib/agentAlertNotifications.ts index ae6232c3dd484..d341cdc85b512 100644 --- a/apps/web/src/lib/agentAlertNotifications.ts +++ b/apps/web/src/lib/agentAlertNotifications.ts @@ -4,21 +4,32 @@ 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 { getClientSettings, subscribeClientSettings } from "../hooks/useSettings"; import { emitAgentAlert } from "./agentAlertPresenter"; +function alertsPresentable(): boolean { + const { notificationMode, inAppNotificationsEnabled } = getClientSettings(); + return notificationMode !== "off" || inAppNotificationsEnabled; +} + export const layer = Layer.effectDiscard( Effect.gen(function* () { if (typeof window === "undefined") return; const registry = yield* EnvironmentRegistry.EnvironmentRegistry; const running = yield* Ref.make(new Map>()); - const sync = Effect.fnUntraced(function* (entries: ReadonlyMap) { + const entries = yield* Ref.make(new Map()); + const refresh = yield* Queue.sliding(1); + const sync = Effect.fnUntraced(function* () { + const presentable = alertsPresentable(); + const active = presentable ? yield* Ref.get(entries) : new Map(); const current = yield* Ref.get(running); for (const [environmentId, fiber] of current) { - if (entries.has(environmentId)) continue; + if (active.has(environmentId)) continue; yield* Fiber.interrupt(fiber); yield* Ref.update(running, (next) => { const updated = new Map(next); @@ -26,7 +37,8 @@ export const layer = Layer.effectDiscard( return updated; }); } - for (const environmentId of entries.keys()) { + 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, {})) @@ -49,11 +61,19 @@ export const layer = Layer.effectDiscard( }); } }); + 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((entries) => sync(entries)), + Stream.runForEach((next) => Ref.set(entries, new Map(next)).pipe(Effect.andThen(sync()))), + Effect.forkScoped, + ); + yield* Stream.fromQueue(refresh).pipe( + Stream.runForEach(() => sync()), Effect.forkScoped, ); }), From d79bc68a1e01922f28a13364b55c8f54054b3752 Mon Sep 17 00:00:00 2001 From: watashiwaaniket Date: Wed, 28 Oct 2026 03:13:10 +0530 Subject: [PATCH 04/14] fix(server): keep a successful Claude push result The PostToolUse hook rewrote every PushNotification result, so a real send became "not sent" when T3 had no client. T3 now delivers only when Claude suppressed the push because the session looks like a terminal. Co-authored-by: Cursor --- .../Adapters/claudePushNotification.test.ts | 35 +++++++++++++++++++ .../Adapters/claudePushNotification.ts | 14 ++++++++ 2 files changed, 49 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts index 64584dc52adf7..63ce68cb5485f 100644 --- a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts @@ -75,4 +75,39 @@ describe("claudePushNotification", () => { }, }); }); + + 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 index 9bf7a8ed979c6..52cbf42bd0463 100644 --- a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts @@ -66,6 +66,15 @@ const missingMessageOutput: SyncHookJSONOutput = { }, }; +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; }) { @@ -76,6 +85,11 @@ export function claudePushNotificationHooks(input: { ) { 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) }); From 970166e4e059882f17decedfc68af3213d85f1b8 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 05:40:23 +0000 Subject: [PATCH 05/14] fix(server): confirm Claude phone alert acceptance --- .../src/notifications/AgentAlerts.test.ts | 10 ++- apps/server/src/notifications/AgentAlerts.ts | 34 +++++----- .../Adapters/claudePushNotification.ts | 2 +- .../src/relay/AgentAwarenessRelay.test.ts | 64 +++++++++++++++++-- apps/server/src/relay/AgentAwarenessRelay.ts | 23 ++----- 5 files changed, 88 insertions(+), 45 deletions(-) diff --git a/apps/server/src/notifications/AgentAlerts.test.ts b/apps/server/src/notifications/AgentAlerts.test.ts index 2f62c0170c694..f0936e9884ba0 100644 --- a/apps/server/src/notifications/AgentAlerts.test.ts +++ b/apps/server/src/notifications/AgentAlerts.test.ts @@ -17,6 +17,10 @@ describe("AgentAlerts", () => { 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)), ); @@ -30,11 +34,13 @@ describe("AgentAlerts", () => { Effect.forkScoped({ startImmediately: true }), ); yield* alerts.registerPhone({ - armed: Effect.succeed(true), - publish: (item) => Deferred.succeed(published, item).pipe(Effect.asVoid), + 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 index f39267aea35ae..ac8851e3e29c6 100644 --- a/apps/server/src/notifications/AgentAlerts.ts +++ b/apps/server/src/notifications/AgentAlerts.ts @@ -9,8 +9,7 @@ import * as Stream from "effect/Stream"; import type { ClaudePushChannels } from "../orchestration-v2/Adapters/claudePushNotification.ts"; export interface AgentAlertPhoneDelivery { - readonly armed: Effect.Effect; - readonly publish: (alert: AgentAlert) => Effect.Effect; + readonly publish: (alert: AgentAlert) => Effect.Effect; } export class AgentAlerts extends Context.Service< @@ -23,7 +22,6 @@ export class AgentAlerts extends Context.Service< >()("t3/notifications/AgentAlerts") {} const make = Effect.gen(function* () { - const scope = yield* Effect.scope; const subscribers = yield* Ref.make(0); const alerts = yield* PubSub.unbounded(); const phone = yield* Ref.make(null); @@ -43,23 +41,21 @@ const make = Effect.gen(function* () { const offer: AgentAlerts["Service"]["offer"] = Effect.fn("AgentAlerts.offer")(function* (alert) { const delivery = yield* Ref.get(phone); - const [local, phoneArmed] = yield* Effect.all( - [ - Ref.get(subscribers).pipe(Effect.map((count) => count > 0)), - delivery === null ? Effect.succeed(false) : delivery.armed, - ], - { concurrency: 2 }, - ); + const local = (yield* Ref.get(subscribers)) > 0; if (local) yield* PubSub.publish(alerts, alert); - if (phoneArmed && delivery !== null) { - yield* delivery.publish(alert).pipe( - Effect.catchCause((cause) => - Effect.logWarning("agent alert phone delivery failed", { cause }), - ), - Effect.forkIn(scope), - ); - } - return { local, phone: phoneArmed }; + 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({ diff --git a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts index 52cbf42bd0463..74e0b9a10f6d2 100644 --- a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts @@ -1,6 +1,6 @@ import type { HookCallback, SyncHookJSONOutput } from "@anthropic-ai/claude-agent-sdk"; -export const CLAUDE_PUSH_NOTIFICATION_TOOL = "PushNotification"; +const CLAUDE_PUSH_NOTIFICATION_TOOL = "PushNotification"; const MAX_PUSH_NOTIFICATION_LENGTH = 200; export interface ClaudePushChannels { diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 48c4a5c240ea9..a465d4a55dfef 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -41,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"); @@ -142,6 +143,7 @@ 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; } = {}, @@ -242,9 +244,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"), @@ -257,6 +259,10 @@ const makeTestRelay = Effect.fnUntraced(function* ( { preconnect: () => {} }, ); const relay = yield* AgentAwarenessRelay.make.pipe( + (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, { @@ -288,10 +294,60 @@ const makeTestRelay = Effect.fnUntraced(function* ( 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(); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index c331a7d89533a..2e118bebd122b 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -880,19 +880,12 @@ export const make = Effect.gen(function* () { }, ); - const phoneArmed = Effect.gen(function* () { - const [relayConfig, publishEnabled] = yield* Effect.all([ - readRelayConfig.pipe(Effect.orElseSucceed(() => null)), - readPublishAgentActivityEnabled.pipe(Effect.orElseSucceed(() => false)), - ]); - return relayConfig !== null && publishEnabled; - }); const publishPhone = Effect.fn("AgentAwarenessRelay.publishAgentAlert")(function* ( alert: AgentAlert, ) { const relayConfig = yield* readRelayConfig; const publishEnabled = yield* readPublishAgentActivityEnabled; - if (!relayConfig || !publishEnabled) return; + if (!relayConfig || !publishEnabled) return false; if ( phoneRelayConfig?.url !== relayConfig.url || phoneRelayConfig.issuer !== relayConfig.issuer || @@ -924,25 +917,17 @@ export const make = Effect.gen(function* () { alert: phoneAlert, jti: yield* crypto.randomUUIDv4(), }); - yield* relayClient.server + 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({ - armed: phoneArmed, - publish: (alert) => - publishPhone(alert).pipe( - Effect.catchCause((cause) => - Effect.logWarning("agent alert phone push failed", { - threadId: alert.threadId, - cause, - }), - ), - ), + publish: (alert) => publishPhone(alert).pipe(Effect.orElseSucceed(() => false)), }); } From ef10b141e79ed5ea297aa7932b10327cae04198d Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 05:41:48 +0000 Subject: [PATCH 06/14] fix(server): describe rejected Claude alerts accurately --- .../orchestration-v2/Adapters/claudePushNotification.test.ts | 2 +- .../src/orchestration-v2/Adapters/claudePushNotification.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts index 63ce68cb5485f..f7ccc6a1caefa 100644 --- a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts @@ -29,7 +29,7 @@ describe("claudePushNotification", () => { hookSpecificOutput: { hookEventName: "PostToolUse", updatedToolOutput: { - message: "Not sent. No T3 client is connected, and phone push is off.", + message: "Not sent. No T3 client or phone accepted the notification.", pushSent: false, localSent: false, disabledReason: "no_transport", diff --git a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts index 74e0b9a10f6d2..e3061553f2db4 100644 --- a/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts +++ b/apps/server/src/orchestration-v2/Adapters/claudePushNotification.ts @@ -34,7 +34,7 @@ export function claudePushNotificationOutput(input: { hookSpecificOutput: { hookEventName: "PostToolUse", updatedToolOutput: { - message: "Not sent. No T3 client is connected, and phone push is off.", + message: "Not sent. No T3 client or phone accepted the notification.", pushSent: false, localSent: false, disabledReason: "no_transport", From 5cdf9987aa14009c19340a7854b1702e3fc04f73 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 05:45:23 +0000 Subject: [PATCH 07/14] fix(web): initialize Claude alert settings after the runtime --- apps/web/src/lib/agentAlertNotifications.ts | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/apps/web/src/lib/agentAlertNotifications.ts b/apps/web/src/lib/agentAlertNotifications.ts index d341cdc85b512..cc52b4bb051e2 100644 --- a/apps/web/src/lib/agentAlertNotifications.ts +++ b/apps/web/src/lib/agentAlertNotifications.ts @@ -9,23 +9,22 @@ import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; -import { getClientSettings, subscribeClientSettings } from "../hooks/useSettings"; import { emitAgentAlert } from "./agentAlertPresenter"; -function alertsPresentable(): boolean { - const { notificationMode, inAppNotificationsEnabled } = getClientSettings(); - return notificationMode !== "off" || inAppNotificationsEnabled; -} - 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 presentable = alertsPresentable(); + const { notificationMode, inAppNotificationsEnabled } = getClientSettings(); + const presentable = notificationMode !== "off" || inAppNotificationsEnabled; const active = presentable ? yield* Ref.get(entries) : new Map(); const current = yield* Ref.get(running); for (const [environmentId, fiber] of current) { From 29d4fddcdc4f12b59adaac7fdf07fa6e245a2dbf Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 05:57:09 +0000 Subject: [PATCH 08/14] fix(relay): align Claude alerts with delivery contracts --- .../src/agentActivity/AgentActivityPublisher.test.ts | 4 ++++ .../relay/src/agentActivity/AgentActivityPublisher.ts | 3 +++ infra/relay/src/agentActivity/FcmDeliveries.ts | 10 +++++++++- .../src/agentActivity/MobileRegistrations.test.ts | 1 + infra/relay/src/http/Api.ts | 6 ------ 5 files changed, 17 insertions(+), 7 deletions(-) diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index d3e4b4a3e6126..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()), @@ -588,6 +591,7 @@ describe("makeAggregateState", () => { }).pipe( Effect.provide( AgentActivityPublisher.layer.pipe( + Layer.provide(NodeCrypto.layer), Layer.provide( Layer.mergeAll( Layer.succeed( diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index 29ad77df3dcc6..c8a92e99d848d 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -10,6 +10,7 @@ import { } from "@t3tools/contracts/relay"; import * as Arr from "effect/Array"; import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -67,6 +68,7 @@ export class AgentActivityPublisher extends Context.Service< >()("t3code-relay/agentActivity/AgentActivityPublisher") {} const make = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; const rows = yield* AgentActivityRows.AgentActivityRows; const links = yield* EnvironmentLinks.EnvironmentLinks; const liveActivities = yield* LiveActivities.LiveActivities; @@ -185,6 +187,7 @@ const make = Effect.gen(function* () { return yield* apnsDeliveries.sendPushNotification({ target, token: target.push_token, + sourceJobId: yield* crypto.randomUUIDv4(), notification, }); }), diff --git a/infra/relay/src/agentActivity/FcmDeliveries.ts b/infra/relay/src/agentActivity/FcmDeliveries.ts index 52e2683d38a08..615ab9396681a 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.ts @@ -182,7 +182,15 @@ const make = Effect.gen(function* () { ? undefined : { ...input.directAlert, - eventId: input.directAlert.eventId ?? (yield* crypto.randomUUIDv4()), + eventId: + input.directAlert.eventId ?? + (yield* crypto + .randomUUIDv4() + .pipe( + Effect.mapError( + (cause) => new FcmDeliveryError({ operation: "enqueue", cause }), + ), + )), }; yield* sender .send({ diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 3f105755c4c40..c36ab50ca9137 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), diff --git a/infra/relay/src/http/Api.ts b/infra/relay/src/http/Api.ts index d008e7e116ecd..6fba7207e9cdf 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -1369,12 +1369,6 @@ export const layerServerApi = HttpApiBuilder.group( }); }, mapErrorTags({ - EnvironmentPublishPublicKeyMissing: (_error, traceId) => - new RelayAuthInvalidError({ - code: "auth_invalid", - reason: "not_authorized", - traceId, - }), EnvironmentPublishSignatureExpired: (_error, traceId) => new RelayAgentActivityPublishProofExpiredError({ code: "agent_activity_publish_proof_expired", From ae959b95c798e5ce0ab4a0934eda40f113410c60 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 01:39:50 +0000 Subject: [PATCH 09/14] fix(web): keep Claude alert delivery on available channels --- .../web/src/components/ThreadNotificationCoordinator.tsx | 1 - apps/web/src/lib/agentAlertNotifications.ts | 9 ++++++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index 775992a1481ba..76de621a7133c 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -448,7 +448,6 @@ export function startThreadNotifications( } if ( !hasDesktopNotifications(mode) || - (document.visibilityState === "visible" && document.hasFocus()) || typeof Notification === "undefined" || Notification.permission !== "granted" ) { diff --git a/apps/web/src/lib/agentAlertNotifications.ts b/apps/web/src/lib/agentAlertNotifications.ts index cc52b4bb051e2..804d608312fcb 100644 --- a/apps/web/src/lib/agentAlertNotifications.ts +++ b/apps/web/src/lib/agentAlertNotifications.ts @@ -9,6 +9,8 @@ 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( @@ -24,7 +26,12 @@ export const layer = Layer.effectDiscard( const refresh = yield* Queue.sliding(1); const sync = Effect.fnUntraced(function* () { const { notificationMode, inAppNotificationsEnabled } = getClientSettings(); - const presentable = notificationMode !== "off" || inAppNotificationsEnabled; + 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) { From 442f6e4ca410ede7301844a18890a8918cf55e11 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 01:41:49 +0000 Subject: [PATCH 10/14] fix: preserve notification behavior after main merge --- apps/server/src/relay/AgentAwarenessRelay.ts | 2 +- .../src/components/ThreadNotificationCoordinator.tsx | 8 ++++++-- .../relay/src/agentActivity/AgentActivityPublisher.ts | 6 +++--- infra/relay/src/agentActivity/FcmDeliveries.ts | 11 ++++++++--- .../src/environments/EnvironmentPublishSignatures.ts | 2 +- 5 files changed, 19 insertions(+), 10 deletions(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 0b45feda746de..9e31273de2a3f 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -925,7 +925,7 @@ export const make = Effect.gen(function* () { environmentId, threadId: alert.threadId, alert: phoneAlert, - jti: yield* crypto.randomUUIDv4(), + jti: yield* random.uuidV4(), }); const response = yield* relayClient.server .publishAgentAlert({ diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index 76de621a7133c..a2d7727b3a0d2 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -396,6 +396,8 @@ export function startThreadNotifications( 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 { @@ -428,7 +430,8 @@ export function startThreadNotifications( document.visibilityState === "visible" && document.hasFocus() ) { - const toastId = toastManager.add({ + const toastId: ToastId = toastManager.add({ + onRemove: (): void => forgetToast(toastKey, toastId), type: "warning", title, description: alert.body, @@ -444,6 +447,7 @@ export function startThreadNotifications( }, }, }); + openToasts.set(toastKey, [...(openToasts.get(toastKey) ?? []), toastId]); return; } if ( @@ -459,7 +463,7 @@ export function startThreadNotifications( tag: `${threadNotificationTag(alert.environmentId, alert.threadId)}:alert`, silent: true, }); - trackNotification(alert.environmentId, notification); + trackNotification(alert.environmentId, alert.threadId, notification); notification.addEventListener("click", () => { notification.close(); window.focus(); diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index c8a92e99d848d..a4435e65f2526 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -10,7 +10,7 @@ import { } from "@t3tools/contracts/relay"; import * as Arr from "effect/Array"; import * as Context from "effect/Context"; -import * as Crypto from "effect/Crypto"; +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"; @@ -68,7 +68,7 @@ export class AgentActivityPublisher extends Context.Service< >()("t3code-relay/agentActivity/AgentActivityPublisher") {} const make = Effect.gen(function* () { - const crypto = yield* Crypto.Crypto; + const random = yield* SecureRandom.SecureRandom; const rows = yield* AgentActivityRows.AgentActivityRows; const links = yield* EnvironmentLinks.EnvironmentLinks; const liveActivities = yield* LiveActivities.LiveActivities; @@ -187,7 +187,7 @@ const make = Effect.gen(function* () { return yield* apnsDeliveries.sendPushNotification({ target, token: target.push_token, - sourceJobId: yield* crypto.randomUUIDv4(), + sourceJobId: yield* random.uuidV4(), notification, }); }), diff --git a/infra/relay/src/agentActivity/FcmDeliveries.ts b/infra/relay/src/agentActivity/FcmDeliveries.ts index 86ea7ab003c0f..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"; @@ -155,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; @@ -184,8 +186,8 @@ const make = Effect.gen(function* () { ...input.directAlert, eventId: input.directAlert.eventId ?? - (yield* crypto - .randomUUIDv4() + (yield* random + .uuidV4() .pipe( Effect.mapError( (cause) => new FcmDeliveryError({ operation: "enqueue", cause }), @@ -277,7 +279,10 @@ const make = Effect.gen(function* () { ...androidActivityData(preferences.value.liveActivitiesEnabled ? aggregate : null), ...alert, }; - const digest = yield* crypto.digest("SHA-256", new TextEncoder().encode(alert.alert_id)); + 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, diff --git a/infra/relay/src/environments/EnvironmentPublishSignatures.ts b/infra/relay/src/environments/EnvironmentPublishSignatures.ts index e7c5988b14b18..9565ea8f4aa77 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.ts @@ -285,7 +285,7 @@ const make = Effect.gen(function* () { } const thumbprint = yield* crypto .digest( - "SHA-256", + Digest.sha256, environmentPublishReplayThumbprintData({ environmentId: input.environmentId, environmentPublicKey: input.environmentPublicKey, From 6a675378588c6b9c86ec1f4dd8b7b0f894559ccb Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 01:43:54 +0000 Subject: [PATCH 11/14] fix(web): serialize Claude alert subscriptions --- apps/web/src/lib/agentAlertNotifications.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/web/src/lib/agentAlertNotifications.ts b/apps/web/src/lib/agentAlertNotifications.ts index 804d608312fcb..44cde9d4b0a42 100644 --- a/apps/web/src/lib/agentAlertNotifications.ts +++ b/apps/web/src/lib/agentAlertNotifications.ts @@ -75,7 +75,9 @@ export const layer = Layer.effectDiscard( Stream.fromEffect(SubscriptionRef.get(registry.entries)), SubscriptionRef.changes(registry.entries), ).pipe( - Stream.runForEach((next) => Ref.set(entries, new Map(next)).pipe(Effect.andThen(sync()))), + Stream.runForEach((next) => + Ref.set(entries, new Map(next)).pipe(Effect.andThen(Queue.offer(refresh, undefined))), + ), Effect.forkScoped, ); yield* Stream.fromQueue(refresh).pipe( From 3fecd3a012f2ec3d0a55564a06df766d36b8c3bb Mon Sep 17 00:00:00 2001 From: watashiwaaniket Date: Tue, 3 Nov 2026 08:22:17 +0530 Subject: [PATCH 12/14] fix(server): keep Claude push hooks beside device subagent hooks A session with device tools spread a second hooks object over the query options and replaced PostToolUse. PushNotification and SubagentStart now share one hooks object. --- .../Adapters/ClaudeAdapterV2.test.ts | 2 ++ .../src/orchestration-v2/Adapters/ClaudeAdapterV2.ts | 12 ++++++------ 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 796e3655de7fb..715b6bcb0d9cc 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -727,6 +727,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 5156c75ce442e..de0a0eed8f9a2 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1599,9 +1599,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 })), - }), + hooks: { + ...claudePushNotificationHooks({ + offer: input.pushNotificationOffer ?? (async () => ({ local: false, phone: false })), + }), + ...(deviceInstructions ? { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } : {}), + }, systemPrompt: { type: "preset" as const, preset: "claude_code" as const, @@ -1617,9 +1620,6 @@ export function makeClaudeQueryOptions(input: { ...launchAppendSystemPrompt, ].join("\n\n"), }, - ...(deviceInstructions - ? { hooks: { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } } - : {}), ...(Object.keys(extraArgs).length === 0 ? {} : { extraArgs }), }; const additionalDirectories = [ From 9912d82ad577f317b217604051386ef67ddbae86 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 4 Nov 2026 20:47:04 +0000 Subject: [PATCH 13/14] fix(relay): adapt alert signature verification to current main --- .../EnvironmentPublishSignatures.ts | 143 +++++++----------- 1 file changed, 51 insertions(+), 92 deletions(-) diff --git a/infra/relay/src/environments/EnvironmentPublishSignatures.ts b/infra/relay/src/environments/EnvironmentPublishSignatures.ts index 9565ea8f4aa77..f89feef315881 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.ts @@ -1,8 +1,8 @@ import { RelayAgentActivityPublishProofPayload, - RelayAgentAlertPublishProofPayload, RelayAgentActivityPublishProofInvalidReason, type RelayAgentActivityPublishRequest, + RelayAgentAlertPublishProofPayload, type RelayAgentAlertPublishRequest, } from "@t3tools/contracts/relay"; import { @@ -15,6 +15,7 @@ import { import { stableStringify } from "@t3tools/shared/relaySigning"; import * as Context from "effect/Context"; import * as Digest from "effect/crypto/Digest"; +import * as Signature from "effect/crypto/Signature"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Base64Url from "effect/encoding/Base64Url"; @@ -82,9 +83,6 @@ export class EnvironmentPublishSignatures extends Context.Service< } >()("t3code-relay/environments/EnvironmentPublishSignatures") {} -const decodeProof = Schema.decodeUnknownEffect(RelayAgentActivityPublishProofPayload); -const decodeAlertProof = Schema.decodeUnknownEffect(RelayAgentAlertPublishProofPayload); - function environmentPublishReplayThumbprintData(input: { readonly environmentId: string; readonly environmentPublicKey: string; @@ -104,25 +102,27 @@ const make = Effect.gen(function* () { const proofReplay = yield* DpopProofs.DpopProofReplay; const config = yield* RelayConfiguration.RelayConfiguration; const crypto = yield* Digest.Digest; - - return EnvironmentPublishSignatures.of({ - verify: Effect.fn("relay.environment_publish_signatures.verify")(function* (input) { - yield* Effect.annotateCurrentSpan({ - "relay.environment_id": input.environmentId, - "relay.thread_id": input.threadId, - }); + 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* Effect.try({ - try: () => decodeRelayJwt(input.request.proof), - catch: (cause) => - new EnvironmentPublishSignatureInvalid({ - environmentId: input.environmentId, - threadId: input.threadId, - reason: "invalid_signature_or_payload", - stage: "decode_token", - cause, - }), - }); + 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) @@ -133,15 +133,14 @@ const make = Effect.gen(function* () { expiresAt: DateTime.formatIso(DateTime.makeUnsafe(decoded.exp * 1_000)), }); } - const proof = yield* verifyRelayJwt({ + const proof = yield* verifyRelayJwt(RelayAgentAlertPublishProofPayload, { publicKey: input.environmentPublicKey, token: input.request.proof, - typ: RELAY_ACTIVITY_PUBLISH_TYP, + typ: RELAY_AGENT_ALERT_TYP, issuer: `t3-env:${input.environmentId}`, audience: normalizeRelayIssuer(config.relayIssuer), - nowEpochSeconds: Math.floor(now.epochMilliseconds / 1_000), }).pipe( - Effect.flatMap(decodeProof), + Effect.provideService(Signature.Signature, signature), Effect.mapError( (cause) => new EnvironmentPublishSignatureInvalid({ @@ -157,10 +156,7 @@ const make = Effect.gen(function* () { proof.environmentId !== input.environmentId || proof.threadId !== input.threadId || proof.sub !== input.environmentId || - stableStringify(proof.state) !== stableStringify(input.request.state) || - (input.request.state !== null && - (input.request.state.environmentId !== input.environmentId || - input.request.state.threadId !== input.threadId)) + stableStringify(proof.alert) !== stableStringify(input.request.alert) ) { return yield* new EnvironmentPublishSignatureInvalid({ environmentId: input.environmentId, @@ -169,68 +165,29 @@ const make = Effect.gen(function* () { stage: "validate_claims", }); } - const expiresAt = DateTime.make(proof.exp * 1_000); - if (expiresAt._tag === "None") { - return yield* new EnvironmentPublishSignatureInvalid({ - environmentId: input.environmentId, - threadId: input.threadId, - reason: "invalid_signature_or_payload", - stage: "validate_expiration", - }); - } - const thumbprint = yield* crypto - .digest( - Digest.sha256, - environmentPublishReplayThumbprintData({ - environmentId: input.environmentId, - environmentPublicKey: input.environmentPublicKey, - }), - ) - .pipe( - Effect.map(formatEnvironmentPublishReplayThumbprint), - Effect.mapError( - (cause) => - new EnvironmentPublishSignatureInvalid({ - environmentId: input.environmentId, - threadId: input.threadId, - reason: "invalid_signature_or_payload", - stage: "generate_replay_thumbprint", - cause, - }), - ), - ); - const consumedNonce = yield* proofReplay.consume({ - thumbprint, - jti: proof.jti, - iat: proof.iat, - expiresAt: expiresAt.value, - }); - if (!consumedNonce) { - return yield* new EnvironmentPublishSignatureInvalid({ - environmentId: input.environmentId, - threadId: input.threadId, - reason: "replayed_nonce", - stage: "consume_nonce", - }); - } - }), - verifyAlert: Effect.fn("relay.environment_publish_signatures.verify_alert")(function* (input) { + }, + ); + + return EnvironmentPublishSignatures.of({ + verifyAlert, + verify: Effect.fn("relay.environment_publish_signatures.verify")(function* (input) { yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId, "relay.thread_id": input.threadId, }); const now = yield* DateTime.now; - const decoded = yield* Effect.try({ - try: () => decodeRelayJwt(input.request.proof), - catch: (cause) => - new EnvironmentPublishSignatureInvalid({ - environmentId: input.environmentId, - threadId: input.threadId, - reason: "invalid_signature_or_payload", - stage: "decode_token", - cause, - }), - }); + 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) @@ -241,15 +198,14 @@ const make = Effect.gen(function* () { expiresAt: DateTime.formatIso(DateTime.makeUnsafe(decoded.exp * 1_000)), }); } - const proof = yield* verifyRelayJwt({ + const proof = yield* verifyRelayJwt(RelayAgentActivityPublishProofPayload, { publicKey: input.environmentPublicKey, token: input.request.proof, - typ: RELAY_AGENT_ALERT_TYP, + typ: RELAY_ACTIVITY_PUBLISH_TYP, issuer: `t3-env:${input.environmentId}`, audience: normalizeRelayIssuer(config.relayIssuer), - nowEpochSeconds: Math.floor(now.epochMilliseconds / 1_000), }).pipe( - Effect.flatMap(decodeAlertProof), + Effect.provideService(Signature.Signature, signature), Effect.mapError( (cause) => new EnvironmentPublishSignatureInvalid({ @@ -265,7 +221,10 @@ const make = Effect.gen(function* () { proof.environmentId !== input.environmentId || proof.threadId !== input.threadId || proof.sub !== input.environmentId || - stableStringify(proof.alert) !== stableStringify(input.request.alert) + stableStringify(proof.state) !== stableStringify(input.request.state) || + (input.request.state !== null && + (input.request.state.environmentId !== input.environmentId || + input.request.state.threadId !== input.threadId)) ) { return yield* new EnvironmentPublishSignatureInvalid({ environmentId: input.environmentId, From 7b3fad8c2c4ffda1e00d05c6305925d3a205a5f6 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 4 Nov 2026 21:11:27 +0000 Subject: [PATCH 14/14] fix(server): provide relay signing service --- apps/server/src/relay/AgentAwarenessRelay.test.ts | 1 + apps/server/src/relay/AgentAwarenessRelay.ts | 8 +++++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 6444a6ed66684..08a630872f3cb 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -207,6 +207,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, diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 9e31273de2a3f..b34fb3b8b4d8f 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -28,6 +28,7 @@ import { 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"; @@ -412,6 +413,7 @@ export const make = Effect.gen(function* () { const threads = yield* ThreadManagement.ThreadManagementService; const projects = yield* ProjectService.ProjectService; const random = yield* SecureRandom.SecureRandom; + const signature = yield* Signature.Signature; const scope = yield* Effect.scope; const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets); const startedAt = (yield* DateTime.now).epochMilliseconds; @@ -937,7 +939,11 @@ export const make = Effect.gen(function* () { }); if (Option.isSome(agentAlerts)) { yield* agentAlerts.value.registerPhone({ - publish: (alert) => publishPhone(alert).pipe(Effect.orElseSucceed(() => false)), + publish: (alert) => + publishPhone(alert).pipe( + Effect.provideService(Signature.Signature, signature), + Effect.orElseSucceed(() => false), + ), }); }