fix(server): deliver Claude push notifications through T3

apps/mobile/modules/t3-agent-notifications/android/src/main/java/expo/modules/t3agentnotifications/AgentNotifications.kt

@@ -128,7 +128,11 @@ object AgentNotifications {
       channels(context)
       val scheme = prefs.getString("scheme", "t3code") ?: "t3code"
       val alerted = showAlert(context, prefs, scheme, data)
-      updateActivity(context, prefs, data, updatedAt, alerted)
+      // A one-shot alert carries no activity transition. Missing activity
+      // fields would otherwise end the card.
+      if (data["alert_only"] != "true") {
+        updateActivity(context, prefs, data, updatedAt, alerted)
+      }
     }
   }
 

apps/mobile/modules/t3-agent-notifications/android/src/test/java/expo/modules/t3agentnotifications/AgentNotificationsTest.kt

@@ -104,6 +104,25 @@ class AgentNotificationsTest : AgentNotificationsFixture() {
     assertEquals("t3-agent-alert", manager.activeNotifications.single().tag)
   }
 
+  @Test
+  fun alertOnlyLeavesTheActivityCardInPlace() {
+    AgentNotifications.receive(context, update("activity", true))
+    val card = manager.activeNotifications.single { it.tag == "t3-agent-activity" }
+
+    val alert = update("ping", true).toMutableMap()
+    alert["alert_only"] = "true"
+    alert.remove("active")
+    alert.remove("activity_expires_at")
+    AgentNotifications.receive(context, alert)
+
+    val activity = manager.activeNotifications.single { it.tag == "t3-agent-activity" }
+    assertEquals(card.id, activity.id)
+    assertEquals(
+      "t3-agent-alert",
+      manager.activeNotifications.single { it.tag == "t3-agent-alert" }.tag
+    )
+  }
+
   @Test
   fun foregroundSuppressesAlertsWhileOngoingActivityStillUpdatesAndClears() {
     lifecycle.currentState = Lifecycle.State.RESUMED

apps/server/src/auth/RpcAuthorization.ts

@@ -112,6 +112,7 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.serverRefreshUsageRates]: AuthDiagnosticsReadScope,
   [WS_METHODS.serverSignalProcess]: AuthEnvironmentMaintainScope,
   [WS_METHODS.serverReportClientActivity]: AuthOrchestrationReadScope,
+  [WS_METHODS.serverSubscribeAgentAlerts]: AuthOrchestrationReadScope,
   [WS_METHODS.scheduledTasksSubscribe]: AuthOrchestrationReadScope,
   [WS_METHODS.secretsAnswerRequest]: AuthOrchestrationOperateScope,
   // Delivery logs hold request bodies, so they need the same scope as the URL.

apps/server/src/notifications/AgentAlerts.test.ts

@@ -0,0 +1,47 @@
+import { describe, expect, it } from "@effect/vitest";
+import { type AgentAlert, ThreadId } from "@t3tools/contracts";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Queue from "effect/Queue";
+import * as Stream from "effect/Stream";
+
+import * as AgentAlerts from "./AgentAlerts.ts";
+
+const alert = {
+  threadId: ThreadId.make("thread-1"),
+  body: "fleet.lock stuck",
+} satisfies AgentAlert;
+
+describe("AgentAlerts", () => {
+  it.effect("reports no channel when nobody is listening", () =>
+    Effect.gen(function* () {
+      const alerts = yield* AgentAlerts.AgentAlerts;
+      expect(yield* alerts.offer(alert)).toEqual({ local: false, phone: false });
+      yield* alerts.registerPhone({
+        publish: () => Effect.succeed(false),
+      });
+      expect(yield* alerts.offer(alert)).toEqual({ local: false, phone: false });
+    }).pipe(Effect.provide(AgentAlerts.layer)),
+  );
+
+  it.effect("delivers to a connected client and a registered phone", () =>
+    Effect.gen(function* () {
+      const alerts = yield* AgentAlerts.AgentAlerts;
+      const received = yield* Queue.unbounded<AgentAlert>();
+      const published = yield* Deferred.make<AgentAlert>();
+      yield* alerts.stream.pipe(
+        Stream.runForEach((item) => Queue.offer(received, item)),
+        Effect.forkScoped({ startImmediately: true }),
+      );
+      yield* alerts.registerPhone({
+        publish: (item) => Deferred.succeed(published, item).pipe(Effect.as(true)),
+      });
+      expect(yield* alerts.offer(alert)).toEqual({ local: true, phone: true });
+      expect(yield* Queue.take(received)).toEqual(alert);
+      yield* alerts.registerPhone({ publish: () => Effect.die("relay unavailable") });
+      expect(yield* alerts.offer(alert)).toEqual({ local: true, phone: false });
+      expect(yield* Queue.take(received)).toEqual(alert);
+      expect(yield* Deferred.await(published)).toEqual(alert);
+    }).pipe(Effect.provide(AgentAlerts.layer), Effect.scoped),
+  );
+});

apps/server/src/notifications/AgentAlerts.ts

@@ -0,0 +1,68 @@
+import type { AgentAlert } from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as PubSub from "effect/PubSub";
+import * as Ref from "effect/Ref";
+import * as Stream from "effect/Stream";
+
+import type { ClaudePushChannels } from "../orchestration-v2/Adapters/claudePushNotification.ts";
+
+export interface AgentAlertPhoneDelivery {
+  readonly publish: (alert: AgentAlert) => Effect.Effect<boolean>;
+}
+
+export class AgentAlerts extends Context.Service<
+  AgentAlerts,
+  {
+    readonly offer: (alert: AgentAlert) => Effect.Effect<ClaudePushChannels>;
+    readonly stream: Stream.Stream<AgentAlert>;
+    readonly registerPhone: (delivery: AgentAlertPhoneDelivery) => Effect.Effect<void>;
+  }
+>()("t3/notifications/AgentAlerts") {}
+
+const make = Effect.gen(function* () {
+  const subscribers = yield* Ref.make(0);
+  const alerts = yield* PubSub.unbounded<AgentAlert>();
+  const phone = yield* Ref.make<AgentAlertPhoneDelivery | null>(null);
+
+  const stream = Stream.unwrap(
+    Effect.gen(function* () {
+      // Subscribe first. Counting before the subscription exists tells Claude the
+      // alert was sent in a window where publish has nobody to deliver to.
+      const subscription = yield* PubSub.subscribe(alerts);
+      yield* Ref.update(subscribers, (count) => count + 1);
+      yield* Effect.addFinalizer(() =>
+        Ref.update(subscribers, (count) => (count > 0 ? count - 1 : 0)),
+      );
+      return Stream.fromSubscription(subscription);
+    }),
+  );
+
+  const offer: AgentAlerts["Service"]["offer"] = Effect.fn("AgentAlerts.offer")(function* (alert) {
+    const delivery = yield* Ref.get(phone);
+    const local = (yield* Ref.get(subscribers)) > 0;
+    if (local) yield* PubSub.publish(alerts, alert);
+    const phoneAccepted =
+      delivery === null
+        ? false
+        : yield* delivery
+            .publish(alert)
+            .pipe(
+              Effect.catchCause((cause) =>
+                Effect.logWarning("agent alert phone delivery failed", { cause }).pipe(
+                  Effect.as(false),
+                ),
+              ),
+            );
+    return { local, phone: phoneAccepted };
+  });
+
+  return AgentAlerts.of({
+    offer,
+    stream,
+    registerPhone: Effect.fn("AgentAlerts.registerPhone")((delivery) => Ref.set(phone, delivery)),
+  });
+});
+
+export const layer = Layer.effect(AgentAlerts, make);

apps/server/src/observability/RpcInstrumentation.ts

@@ -89,6 +89,7 @@ const RPC_AGGREGATES = {
   [WS_METHODS.serverRefreshUsageRates]: "server",
   [WS_METHODS.serverSignalProcess]: "server",
   [WS_METHODS.serverReportClientActivity]: "server",
+  [WS_METHODS.serverSubscribeAgentAlerts]: "server",
   [WS_METHODS.scheduledTasksSubscribe]: "scheduledTasks",
   [WS_METHODS.scheduledTasksUpsert]: "scheduledTasks",
   [WS_METHODS.scheduledTasksSetEnabled]: "scheduledTasks",

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts

@@ -753,6 +753,8 @@ describe("ClaudeAdapterV2 device instructions", () => {
     ]);
     assert.deepEqual(await subagentContext({ mcp: true, deviceToolsAvailable: false }), []);
     assert.deepEqual(await subagentContext({ mcp: false, deviceToolsAvailable: true }), []);
+    assert.isArray(options({ mcp: true, deviceToolsAvailable: true }).hooks?.PostToolUse);
+    assert.isArray(options({ mcp: false, deviceToolsAvailable: false }).hooks?.PostToolUse);
   });
 });
 

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

@@ -6,6 +6,8 @@ import {
 } from "@t3tools/shared/toolOutput";
 import { makeProviderTextDeltaCoalescer } from "./ProviderTextDeltaCoalescer.ts";
 import { claudeTaskOutputRoot, tailClaudeTaskOutput } from "./ClaudeTaskOutputTail.ts";
+import { claudePushNotificationHooks } from "./claudePushNotification.ts";
+import * as AgentAlerts from "../../notifications/AgentAlerts.ts";
 import {
   dynamicToolTitle,
   formatReadToolLabel,
@@ -1775,6 +1777,10 @@ export function makeClaudeQueryOptions(input: {
   readonly appendSystemPromptFileText?: string;
   // Flag settings, because a user's settings.json env beats the process env in Claude Code.
   readonly toolSearchOverride?: "true";
+  readonly pushNotificationOffer?: (message: string) => Promise<{
+    readonly local: boolean;
+    readonly phone: boolean;
+  }>;
 }): ClaudeAgentSdkQueryOptions {
   const compiledSelection = compileClaudeModelSelection(input.modelSelection, input.modelCatalog);
   const {
@@ -1879,6 +1885,12 @@ export function makeClaudeQueryOptions(input: {
       : {}),
     ...(input.environment === undefined ? {} : { env: input.environment }),
     ...(input.mcpServers === undefined ? {} : { mcpServers: input.mcpServers }),
+    hooks: {
+      ...claudePushNotificationHooks({
+        offer: input.pushNotificationOffer ?? (async () => ({ local: false, phone: false })),
+      }),
+      ...(deviceInstructions ? { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } : {}),
+    },
     systemPrompt: {
       type: "preset" as const,
       preset: "claude_code" as const,
@@ -1894,9 +1906,6 @@ export function makeClaudeQueryOptions(input: {
         ...launchAppendSystemPrompt,
       ].join("\n\n"),
     },
-    ...(deviceInstructions
-      ? { hooks: { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } }
-      : {}),
     ...(Object.keys(extraArgs).length === 0 ? {} : { extraArgs }),
   };
   const additionalDirectories = [
@@ -4851,6 +4860,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
   adapterOptions: ClaudeAdapterV2Options,
 ) {
   const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
+  const agentAlerts = yield* Effect.serviceOption(AgentAlerts.AgentAlerts);
   const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
   const serverSettings = yield* Effect.serviceOption(ServerSettings.ServerSettingsService);
   const { attachmentsDir, fileSystem, path, random, idAllocator, queryRunner } = adapterOptions;
@@ -10901,6 +10911,17 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
             supportedDialogKinds: ["resume_return"],
             ...(appendSystemPromptFileText === undefined ? {} : { appendSystemPromptFileText }),
             ...(toolSearchOverride === undefined ? {} : { toolSearchOverride }),
+            pushNotificationOffer: (message) =>
+              runPromise(
+                Option.match(agentAlerts, {
+                  onNone: () => Effect.succeed({ local: false, phone: false }),
+                  onSome: (alerts) =>
+                    alerts.offer({
+                      threadId: turnInput.threadId,
+                      body: message,
+                    }),
+                }),
+              ),
           });
           const resolvedTurnCwd = yield* fileSystem
             .realPath(turnInput.runtimePolicy.cwd ?? ".")

apps/server/src/orchestration-v2/Adapters/claudePushNotification.test.ts

@@ -0,0 +1,113 @@
+import { describe, it } from "@effect/vitest";
+import * as NodeAssert from "node:assert/strict";
+
+import {
+  claudePushNotificationHooks,
+  claudePushNotificationOutput,
+  readClaudePushNotificationMessage,
+} from "./claudePushNotification.ts";
+
+describe("claudePushNotification", () => {
+  it("keeps a one-line message and caps it at 200 characters", () => {
+    NodeAssert.equal(
+      readClaudePushNotificationMessage({ message: "  fleet.lock stuck  " }),
+      "fleet.lock stuck",
+    );
+    NodeAssert.equal(readClaudePushNotificationMessage({ message: "line\n\nbreak" }), "line break");
+    NodeAssert.equal(readClaudePushNotificationMessage({ message: "   " }), null);
+    NodeAssert.equal(readClaudePushNotificationMessage({}), null);
+    const long = "x".repeat(240);
+    const read = readClaudePushNotificationMessage({ message: long });
+    NodeAssert.equal(read?.length, 200);
+    NodeAssert.equal(read?.endsWith("…"), true);
+  });
+
+  it("does not tell the agent the terminal is active when T3 cannot deliver", () => {
+    NodeAssert.deepEqual(
+      claudePushNotificationOutput({ channels: { local: false, phone: false } }),
+      {
+        hookSpecificOutput: {
+          hookEventName: "PostToolUse",
+          updatedToolOutput: {
+            message: "Not sent. No T3 client or phone accepted the notification.",
+            pushSent: false,
+            localSent: false,
+            disabledReason: "no_transport",
+          },
+        },
+      },
+    );
+  });
+
+  it("rewrites a user_present result when a T3 channel can deliver", async () => {
+    const hooks = claudePushNotificationHooks({
+      offer: async () => ({ local: true, phone: false }),
+    });
+    const hook = hooks.PostToolUse[0]?.hooks[0];
+    NodeAssert.ok(hook);
+    const result = await hook(
+      {
+        hook_event_name: "PostToolUse",
+        tool_name: "PushNotification",
+        tool_input: { message: "fleet.lock stuck", status: "proactive" },
+        tool_response: {
+          message: "Not sent — this terminal is active.",
+          pushSent: false,
+          localSent: false,
+          disabledReason: "user_present",
+        },
+        tool_use_id: "tool-1",
+        session_id: "session-1",
+        transcript_path: "",
+        cwd: "/workspace",
+      },
+      "tool-1",
+      { signal: new AbortController().signal },
+    );
+    NodeAssert.deepEqual(result, {
+      hookSpecificOutput: {
+        hookEventName: "PostToolUse",
+        updatedToolOutput: {
+          message: "Sent through T3.",
+          pushSent: false,
+          localSent: true,
+        },
+      },
+    });
+  });
+
+  it("leaves a push Claude already resolved unchanged", async () => {
+    const offered: Array<string> = [];
+    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, []);
+  });
+});