diff --git a/FORK.md b/FORK.md index 3c7036d2e508c..27e475542bb4c 100644 --- a/FORK.md +++ b/FORK.md @@ -74,3 +74,12 @@ adds the feature, and remove it in the PR that drops the feature. - #32 The Android APK is the fork's own app: an ember icon, plus the package name, app name, release keystore and Firebase config from repository variables and secrets when they are set: `apps/mobile/app.config.ts` (`T3CODE_ANDROID_PACKAGE`), `.github/workflows/fork-release.yml`. +- #45 A server started with an FCM service account key is the push host: it sends the + fork's Android app its notifications and ongoing activity card, for itself and for environments + linked to it, with upstream's relay payloads; the app registers there and shows its status in + Settings → Notifications: `apps/server/src/fork/push/`, + `apps/server/src/relay/AgentAwarenessRelay.ts` (`ForkPushTarget`), + `packages/contracts/src/forkPush.ts`, `apps/mobile/src/features/fork-push/`, + `apps/mobile/src/features/agent-awareness/androidNotifications.ts` + (`forkOwnsAndroidAgentNotifications`), `apps/server/vite.config.ts` (`resolve.modules`), + `docs/user/push-host.md`. diff --git a/apps/mobile/src/App.tsx b/apps/mobile/src/App.tsx index 6287a05956519..789477c8e99ea 100644 --- a/apps/mobile/src/App.tsx +++ b/apps/mobile/src/App.tsx @@ -75,6 +75,7 @@ import { import { useMobileNavigationTheme } from "./lib/useMobileNavigationTheme"; import { installUiRuntimeMemoryWarningGc } from "./lib/useUiRuntimeMemoryWarningGc"; import { SubscriptionUsageCoordinator } from "./widgets/SubscriptionUsageCoordinator"; +import { ForkPushRegistrar } from "./features/fork-push/ForkPushRegistrar"; import { installSubscriptionUsageRefresh } from "./widgets/useSubscriptionUsage"; import { GlobalVoiceInputControl } from "./features/voice-input/GlobalVoiceInputControl"; import { restoreSettingsLinkState } from "./features/settings/settings-navigation"; @@ -243,6 +244,7 @@ function AppContent() { <> + diff --git a/apps/mobile/src/features/agent-awareness/androidNotifications.ts b/apps/mobile/src/features/agent-awareness/androidNotifications.ts index 31107d9d9ab78..f0f3fbef5d07b 100644 --- a/apps/mobile/src/features/agent-awareness/androidNotifications.ts +++ b/apps/mobile/src/features/agent-awareness/androidNotifications.ts @@ -2,6 +2,8 @@ import Constants from "expo-constants"; import { requireOptionalNativeModule } from "expo"; import { Linking, Platform } from "react-native"; +import { forkOwnsAndroidAgentNotifications } from "../fork-push/nativeOwnership"; + interface AndroidAgentNotifications { configure(deviceId: string, userId: string, scheme: string, ongoingEnabled: boolean): void; clear(): void; @@ -29,6 +31,7 @@ export function configureAndroidAgentNotifications( userId: string, ongoingEnabled: boolean, ): void { + if (forkOwnsAndroidAgentNotifications()) return; native?.configure?.(deviceId, userId, appScheme(), ongoingEnabled); } @@ -39,6 +42,7 @@ export function showAndroidShowcaseAgentActivity(data: Record): } export function clearAndroidAgentNotifications(): void { + if (forkOwnsAndroidAgentNotifications()) return; native?.clear?.(); } diff --git a/apps/mobile/src/features/fork-push/ForkPushRegistrar.tsx b/apps/mobile/src/features/fork-push/ForkPushRegistrar.tsx new file mode 100644 index 0000000000000..7d3d073a71198 --- /dev/null +++ b/apps/mobile/src/features/fork-push/ForkPushRegistrar.tsx @@ -0,0 +1,60 @@ +import { useAtomValue } from "@effect/atom-react"; +import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; +import { EnvironmentId } from "@t3tools/contracts"; +import { AsyncResult, Atom } from "effect/reactivity"; +import { useEffect, useEffectEvent, useMemo } from "react"; + +import { useAtomCommand } from "../../state/use-atom-command"; +import { useWorkspaceEnvironments } from "../../state/workspace"; +import { forkPushEnvironment } from "./forkPushState"; +import { supportsForkPush } from "./nativeOwnership"; +import { syncForkPushHosts } from "./registrar"; + +/** Registers this phone with a connected environment that holds the fork's FCM key. */ +export function ForkPushRegistrar() { + const connectedKey = useWorkspaceEnvironments() + .filter((environment) => environment.connectionState === "connected") + .map((environment) => environment.environmentId) + .sort() + .join("\n"); + const hostsAtom = useMemo(() => { + const connected = connectedKey === "" ? [] : connectedKey.split("\n"); + return Atom.make((get) => + connected + .filter( + (environmentId) => + AsyncResult.getOrElse( + get( + forkPushEnvironment.status({ + environmentId: EnvironmentId.make(environmentId), + input: {}, + }), + ), + () => null, + )?.role === "host", + ) + .join("\n"), + ); + }, [connectedKey]); + const hostsKey = useAtomValue(hostsAtom); + const register = useAtomCommand(forkPushEnvironment.register, { reportFailure: false }); + const sync = useEffectEvent((key: string) => { + if (!supportsForkPush()) return; + syncForkPushHosts( + key === "" ? [] : key.split("\n").map((environmentId) => EnvironmentId.make(environmentId)), + async (environmentId, input) => { + const result = await register({ environmentId, input }); + return result._tag === "Success" + ? { ok: true } + : { ok: false, error: String(squashAtomCommandFailure(result)) }; + }, + ); + }); + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- registration with the push host follows which connected environments are hosts + useEffect(() => { + sync(hostsKey); + }, [hostsKey]); + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- an unmounted registrar leaves no registration running + useEffect(() => () => sync(""), []); + return null; +} diff --git a/apps/mobile/src/features/fork-push/ForkPushSettingsSection.tsx b/apps/mobile/src/features/fork-push/ForkPushSettingsSection.tsx new file mode 100644 index 0000000000000..3abdf611d8057 --- /dev/null +++ b/apps/mobile/src/features/fork-push/ForkPushSettingsSection.tsx @@ -0,0 +1,217 @@ +import type { EnvironmentId, ForkPushStatus } from "@t3tools/contracts"; +import { useAtomValue } from "@effect/atom-react"; +import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; +import { useSyncExternalStore } from "react"; + +import * as Dialog from "../../components/ui/Dialog"; +import { runtime } from "../../lib/runtime"; +import { useEnvironmentQuery } from "../../state/query"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { useWorkspaceEnvironments } from "../../state/workspace"; +import { requestAgentNotificationPermission } from "../agent-awareness/notificationPermissions"; +import { SettingsRow } from "../settings/components/SettingsRow"; +import { SettingsSection } from "../settings/components/SettingsSection"; +import { forkPushEnvironment } from "./forkPushState"; +import { supportsForkPush } from "./nativeOwnership"; +import { + type ForkPushRegistrationState, + getForkPushRegistrationState, + resumeForkPush, + stopForkPush, + subscribeForkPushRegistrationState, +} from "./registrar"; + +const PUBLISHING_OFF = "Publish agent activity is off there, so its threads don't notify."; + +function registrationText(state: ForkPushRegistrationState): string { + switch (state.kind) { + case "idle": + case "registering": + return "Registering this phone…"; + case "registered": + return "This phone gets its notifications."; + case "needs-permission": + return "Notifications are off for T3 Code. Tap to turn them on."; + case "stopped": + return "Stopped on this phone."; + case "failed": + return `Couldn't register: ${state.error}`; + } +} + +/** Notifications through the owner's push host, the fork's replacement for T3 Connect's relay. */ +export function ForkPushSettingsSection() { + const registration = useSyncExternalStore( + subscribeForkPushRegistrationState, + getForkPushRegistrationState, + getForkPushRegistrationState, + ); + const environments = useWorkspaceEnvironments().filter( + (environment) => environment.connectionState === "connected", + ); + if (!supportsForkPush()) return null; + const hostEnvironmentId = registration.kind === "idle" ? null : registration.hostEnvironmentId; + const host = environments.find((environment) => environment.environmentId === hostEnvironmentId); + + return ( + + {host === undefined ? ( + + ) : ( + + )} + {environments + .filter((environment) => environment.environmentId !== hostEnvironmentId) + .map((environment) => ( + + ))} + + ); +} + +function HostRows(props: { + readonly environmentId: EnvironmentId; + readonly label: string; + readonly registration: ForkPushRegistrationState; +}) { + const status = useEnvironmentQuery( + forkPushEnvironment.status({ environmentId: props.environmentId, input: {} }), + ); + const sendTest = useAtomCommand(forkPushEnvironment.sendTest, "send a test notification"); + const unregister = useAtomCommand(forkPushEnvironment.unregister, "stop notifications"); + const canRegister = useAtomValue( + forkPushEnvironment.register.permissionAtom(props.environmentId), + ); + const { registration } = props; + const publishingOff = status.data?.publishingOff === true; + + const turnOn = async () => { + if (registration.kind === "needs-permission") { + const permission = await runtime + .runPromise(requestAgentNotificationPermission) + .catch(() => null); + if (permission?.type !== "granted") { + Dialog.alert( + "Notifications are off", + "Allow notifications for T3 Code in Android Settings, then come back here.", + ); + return; + } + } + await resumeForkPush(); + }; + + const stop = () => + stopForkPush(async (deviceId) => { + await unregister({ environmentId: props.environmentId, input: { deviceId } }); + }); + + return ( + <> + void turnOn() } + : {})} + /> + {registration.kind === "registered" ? ( + + void sendTest({ + environmentId: props.environmentId, + input: { deviceId: registration.deviceId }, + }).then((result) => { + if (result._tag === "Failure") { + Dialog.alert("Test failed", String(squashAtomCommandFailure(result))); + } + }) + } + /> + ) : null} + {registration.kind === "stopped" ? ( + void turnOn()} + /> + ) : ( + void stop()} + /> + )} + + ); +} + +function senderText( + status: ForkPushStatus | null, + failed: boolean, + hostEnvironmentId: EnvironmentId | null, + hostLabel: string | null, +): string { + if (failed) return "This server has no push host support."; + if (status === null) return "Checking…"; + const host = hostLabel ?? status.host?.label ?? "the push host"; + const sendsToHost = + status.host !== null && + (hostEnvironmentId === null || status.host.environmentId === hostEnvironmentId); + const expires = status.host ? new Date(status.host.linkExpiresAt).toLocaleDateString() : ""; + const text = + status.role === "sender" && sendsToHost + ? `Sends to ${host}. Link expires ${expires}.` + : sendsToHost && status.error !== null + ? `Can't reach ${host}: ${status.error} Link it again to renew.` + : `Not linked to ${host}, so its threads don't notify.`; + return status.publishingOff ? `${text} ${PUBLISHING_OFF}` : text; +} + +function SenderRow(props: { + readonly environmentId: EnvironmentId; + readonly label: string; + readonly hostEnvironmentId: EnvironmentId | null; + readonly hostLabel: string | null; +}) { + const status = useEnvironmentQuery( + forkPushEnvironment.status({ environmentId: props.environmentId, input: {} }), + ); + return ( + + ); +} diff --git a/apps/mobile/src/features/fork-push/forkPushState.ts b/apps/mobile/src/features/fork-push/forkPushState.ts new file mode 100644 index 0000000000000..29f63eef5b7cd --- /dev/null +++ b/apps/mobile/src/features/fork-push/forkPushState.ts @@ -0,0 +1,35 @@ +import { + createEnvironmentRpcCommand, + createEnvironmentRpcQueryAtomFamily, +} from "@t3tools/client-runtime/state/runtime"; +import { FORK_PUSH_WS_METHODS } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; + +import { connectionAtomRuntime } from "../../connection/runtime"; + +const status = createEnvironmentRpcQueryAtomFamily(connectionAtomRuntime, { + label: "environment-data:fork-push:status", + tag: FORK_PUSH_WS_METHODS.getStatus, + // A sender re-checks its links every two minutes, so a fresher read shows nothing new. + staleTimeMs: 30_000, +}); + +export const forkPushEnvironment = { + status, + register: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-data:fork-push:register", + tag: FORK_PUSH_WS_METHODS.registerDevice, + onSuccess: ({ environmentId }, registry) => + Effect.sync(() => registry.refresh(status({ environmentId, input: {} }))), + }), + unregister: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-data:fork-push:unregister", + tag: FORK_PUSH_WS_METHODS.unregisterDevice, + onSuccess: ({ environmentId }, registry) => + Effect.sync(() => registry.refresh(status({ environmentId, input: {} }))), + }), + sendTest: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-data:fork-push:send-test", + tag: FORK_PUSH_WS_METHODS.sendTest, + }), +}; diff --git a/apps/mobile/src/features/fork-push/nativeOwnership.ts b/apps/mobile/src/features/fork-push/nativeOwnership.ts new file mode 100644 index 0000000000000..f8055e3a97d04 --- /dev/null +++ b/apps/mobile/src/features/fork-push/nativeOwnership.ts @@ -0,0 +1,46 @@ +import Constants from "expo-constants"; +import { requireOptionalNativeModule } from "expo"; +import { Platform } from "react-native"; + +interface AgentNotificationsModule { + configure(deviceId: string, userId: string, scheme: string, ongoingEnabled: boolean): void; + clear(): void; +} + +const native = + Platform.OS === "android" + ? requireOptionalNativeModule("T3AgentNotifications") + : null; + +let owned = false; + +/** + * While the push host owns the native handler, upstream's T3 Connect registration must not + * reconfigure or clear it: a different user ID makes the handler drop every push-host message. + */ +export function forkOwnsAndroidAgentNotifications(): boolean { + return owned; +} + +export function supportsForkPush(): boolean { + return typeof native?.configure === "function"; +} + +function appScheme(): string { + const scheme = Constants.expoConfig?.scheme; + return (Array.isArray(scheme) ? scheme[0] : scheme) ?? "t3code"; +} + +export function claimAndroidAgentNotifications( + deviceId: string, + userId: string, + ongoingEnabled: boolean, +): void { + owned = true; + native?.configure(deviceId, userId, appScheme(), ongoingEnabled); +} + +export function releaseAndroidAgentNotifications(): void { + owned = false; + native?.clear(); +} diff --git a/apps/mobile/src/features/fork-push/registrar.test.ts b/apps/mobile/src/features/fork-push/registrar.test.ts new file mode 100644 index 0000000000000..6589fbf1cb7e1 --- /dev/null +++ b/apps/mobile/src/features/fork-push/registrar.test.ts @@ -0,0 +1,159 @@ +import { vi } from "vite-plus/test"; +import { describe, expect, it } from "@effect/vitest"; +import { EnvironmentId } from "@t3tools/contracts"; + +const mocks = vi.hoisted(() => ({ + secureStore: new Map(), + tokenListeners: [] as Array<(token: { type: string; data: string }) => void>, + native: { configure: vi.fn(), clear: vi.fn() }, + token: () => Promise.resolve({ type: "android", data: "token-1" }), +})); + +vi.mock("expo-constants", () => ({ + default: { expoConfig: { android: { package: "com.example.t3code" } }, deviceName: "Phone" }, +})); +vi.mock("expo", () => ({ requireOptionalNativeModule: () => mocks.native })); +vi.mock("react-native", () => ({ + Platform: { OS: "android" }, + AppState: { addEventListener: () => ({ remove: () => {} }) }, +})); +vi.mock("expo-secure-store", () => ({ + getItemAsync: (key: string) => Promise.resolve(mocks.secureStore.get(key) ?? null), + setItemAsync: (key: string, value: string) => { + mocks.secureStore.set(key, value); + return Promise.resolve(); + }, +})); +vi.mock("expo-notifications", () => ({ + getPermissionsAsync: () => Promise.resolve({ granted: true }), + getDevicePushTokenAsync: () => mocks.token(), + addPushTokenListener: (listener: (token: { type: string; data: string }) => void) => { + mocks.tokenListeners.push(listener); + return { remove: () => {} }; + }, +})); +vi.mock("../../lib/protectedData", () => ({ whenProtectedDataAvailable: () => Promise.resolve() })); +vi.mock("../../persistence/imperative", () => ({ + loadOrCreateAgentAwarenessDeviceId: () => Promise.resolve("device-1"), + loadPreferences: () => Promise.resolve({ liveActivitiesEnabled: true }), +})); + +import { configureAndroidAgentNotifications } from "../agent-awareness/androidNotifications"; +import { forkOwnsAndroidAgentNotifications } from "./nativeOwnership"; +import { + getForkPushRegistrationState, + resumeForkPush, + startForkPushRegistration, + stopForkPush, + subscribeForkPushRegistrationState, + syncForkPushHosts, +} from "./registrar"; + +const HOST = EnvironmentId.make("env-host"); + +const settled = (kind: string) => + new Promise((resolve) => { + if (getForkPushRegistrationState().kind === kind) return resolve(); + const unsubscribe = subscribeForkPushRegistrationState(() => { + if (getForkPushRegistrationState().kind !== kind) return; + unsubscribe(); + resolve(); + }); + }); + +describe("fork push registrar", () => { + it("registers with the push host's identity, follows token rotation, and stops and resumes", async () => { + const registered: Array<{ token: string; userId: string }> = []; + const unregistered: string[] = []; + let onRegister = () => {}; + const nextRegistration = () => new Promise((resolve) => (onRegister = resolve)); + const stop = startForkPushRegistration(HOST, (input) => { + registered.push({ token: input.token, userId: input.userId }); + onRegister(); + return Promise.resolve({ ok: true }); + }); + await settled("registered"); + expect(registered).toEqual([{ token: "token-1", userId: "t3-fork:env-host" }]); + expect(mocks.native.configure).toHaveBeenLastCalledWith( + "device-1", + "t3-fork:env-host", + "t3code", + true, + ); + + // Upstream's T3 Connect registration must not overwrite the push host's identity. + expect(forkOwnsAndroidAgentNotifications()).toBe(true); + configureAndroidAgentNotifications("device-1", "clerk-user", true); + expect(mocks.native.configure).toHaveBeenCalledTimes(1); + + const rotated = nextRegistration(); + mocks.tokenListeners[0]!({ type: "android", data: "token-2" }); + await rotated; + await settled("registered"); + expect(registered.at(-1)).toEqual({ token: "token-2", userId: "t3-fork:env-host" }); + + await stopForkPush((deviceId) => { + unregistered.push(deviceId); + return Promise.resolve(); + }); + expect(unregistered).toEqual(["device-1"]); + expect(mocks.native.clear).toHaveBeenCalledTimes(1); + expect(forkOwnsAndroidAgentNotifications()).toBe(false); + + await resumeForkPush(); + expect(getForkPushRegistrationState().kind).toBe("registered"); + expect(registered).toHaveLength(3); + stop(); + }); + + it("lets a stop pressed during a pending registration win", async () => { + mocks.native.configure.mockClear(); + let tokenRequested = () => {}; + const requested = new Promise((resolve) => (tokenRequested = resolve)); + let releaseToken = (_token: { type: string; data: string }) => {}; + mocks.token = () => { + tokenRequested(); + return new Promise((resolve) => (releaseToken = resolve)); + }; + const registered: string[] = []; + const stop = startForkPushRegistration(HOST, (input) => { + registered.push(input.token); + return Promise.resolve({ ok: true }); + }); + await requested; + const stopping = stopForkPush(() => Promise.resolve()); + releaseToken({ type: "android", data: "token-3" }); + await stopping; + expect(registered).toEqual([]); + expect(mocks.native.configure).not.toHaveBeenCalled(); + expect(forkOwnsAndroidAgentNotifications()).toBe(false); + expect(getForkPushRegistrationState().kind).toBe("stopped"); + mocks.token = () => Promise.resolve({ type: "android", data: "token-3" }); + await resumeForkPush(); + expect(registered).toEqual(["token-3"]); + stop(); + }); + + it("registers with one push host, the same one each time, and fails over when it leaves", async () => { + mocks.token = () => Promise.resolve({ type: "android", data: "token-1" }); + const first = EnvironmentId.make("env-a"); + const second = EnvironmentId.make("env-b"); + const calls: string[] = []; + let onRegister = () => {}; + const nextRegistration = () => new Promise((resolve) => (onRegister = resolve)); + const register = (environmentId: EnvironmentId, input: { readonly userId: string }) => { + calls.push(`${environmentId} ${input.userId}`); + onRegister(); + return Promise.resolve({ ok: true as const }); + }; + let registered = nextRegistration(); + syncForkPushHosts([second, first], register); + await registered; + registered = nextRegistration(); + syncForkPushHosts([second], register); + await registered; + expect(calls).toEqual(["env-a t3-fork:env-a", "env-b t3-fork:env-b"]); + syncForkPushHosts([], register); + expect(getForkPushRegistrationState().kind).toBe("idle"); + }); +}); diff --git a/apps/mobile/src/features/fork-push/registrar.ts b/apps/mobile/src/features/fork-push/registrar.ts new file mode 100644 index 0000000000000..470fb8ded0253 --- /dev/null +++ b/apps/mobile/src/features/fork-push/registrar.ts @@ -0,0 +1,227 @@ +import type { EnvironmentId, ForkPushDeviceRegistration } from "@t3tools/contracts"; +import Constants from "expo-constants"; +import * as Notifications from "expo-notifications"; +import * as SecureStore from "expo-secure-store"; +import { AppState } from "react-native"; + +import { whenProtectedDataAvailable } from "../../lib/protectedData"; +import { loadOrCreateAgentAwarenessDeviceId, loadPreferences } from "../../persistence/imperative"; +import { + claimAndroidAgentNotifications, + releaseAndroidAgentNotifications, +} from "./nativeOwnership"; + +export type ForkPushRegistrationState = + | { readonly kind: "idle" } + | { readonly kind: "needs-permission"; readonly hostEnvironmentId: EnvironmentId } + | { readonly kind: "registering"; readonly hostEnvironmentId: EnvironmentId } + | { + readonly kind: "registered"; + readonly hostEnvironmentId: EnvironmentId; + readonly deviceId: string; + } + | { readonly kind: "stopped"; readonly hostEnvironmentId: EnvironmentId } + | { readonly kind: "failed"; readonly hostEnvironmentId: EnvironmentId; readonly error: string }; + +type Register = ( + input: ForkPushDeviceRegistration, +) => Promise<{ readonly ok: true } | { readonly ok: false; readonly error: string }>; + +const STORAGE_KEY = "t3code.fork-push.v1"; +/** A registration this old is sent again on foreground, in case the host lost it. */ +const REFRESH_AFTER_MS = 6 * 60 * 60 * 1000; + +let state: ForkPushRegistrationState = { kind: "idle" }; +const listeners = new Set<() => void>(); +let active: { + readonly hostEnvironmentId: EnvironmentId; + readonly run: () => Promise; + readonly settled: () => Promise; +} | null = null; +/** Bumped by Stop; an attempt started before it gives up at its next step. */ +let stopGeneration = 0; + +function setState(next: ForkPushRegistrationState) { + state = next; + for (const listener of listeners) listener(); +} + +export const getForkPushRegistrationState = () => state; +export function subscribeForkPushRegistrationState(listener: () => void): () => void { + listeners.add(listener); + return () => listeners.delete(listener); +} + +async function loadStopped(): Promise { + try { + await whenProtectedDataAvailable(); + const raw = await SecureStore.getItemAsync(STORAGE_KEY); + const parsed: unknown = raw === null ? null : JSON.parse(raw); + return typeof parsed === "object" && parsed !== null && "stopped" in parsed + ? parsed.stopped === true + : false; + } catch { + return false; + } +} + +async function saveStopped(stopped: boolean): Promise { + await whenProtectedDataAvailable(); + await SecureStore.setItemAsync(STORAGE_KEY, JSON.stringify({ stopped })); +} + +/** The native handler drops messages for another user, so the identity names the push host. */ +const forkPushUserId = (hostEnvironmentId: string) => `t3-fork:${hostEnvironmentId}`; + +/** + * Registers this phone with the push host now, when the app comes back, and when FCM rotates + * its token. Returns the cleanup. Only one push host is registered with at a time. + */ +export function startForkPushRegistration( + hostEnvironmentId: EnvironmentId, + register: Register, +): () => void { + if (active !== null) return () => {}; + let disposed = false; + let inFlight: Promise | null = null; + let registered: { readonly key: string; readonly at: number } | null = null; + + const attempt = async (observedToken: string | null) => { + const generation = stopGeneration; + const superseded = () => disposed || generation !== stopGeneration; + if (await loadStopped()) { + registered = null; + setState({ kind: "stopped", hostEnvironmentId }); + return; + } + const permission = await Notifications.getPermissionsAsync(); + if (!permission.granted) { + setState({ kind: "needs-permission", hostEnvironmentId }); + return; + } + const token = + observedToken ?? String((await Notifications.getDevicePushTokenAsync()).data ?? ""); + if (token.trim() === "") { + setState({ kind: "failed", hostEnvironmentId, error: "This phone has no FCM token yet." }); + return; + } + const deviceId = await loadOrCreateAgentAwarenessDeviceId(); + const preferences = await loadPreferences().catch(() => null); + const ongoing = preferences?.liveActivitiesEnabled !== false; + const userId = forkPushUserId(hostEnvironmentId); + const key = JSON.stringify([token, deviceId, ongoing]); + if (registered?.key === key && Date.now() - registered.at < REFRESH_AFTER_MS) return; + if (superseded()) return; + claimAndroidAgentNotifications(deviceId, userId, ongoing); + setState({ kind: "registering", hostEnvironmentId }); + const result = await register({ + deviceId, + userId, + token, + packageName: Constants.expoConfig?.android?.package?.trim() || null, + label: Constants.deviceName?.trim() || "Android device", + preferences: { + notificationsEnabled: true, + liveActivitiesEnabled: ongoing, + notifyOnApproval: true, + notifyOnInput: true, + notifyOnCompletion: true, + notifyOnFailure: true, + }, + }); + if (superseded()) return; + if (result.ok) { + registered = { key, at: Date.now() }; + setState({ kind: "registered", hostEnvironmentId, deviceId }); + } else { + setState({ kind: "failed", hostEnvironmentId, error: result.error }); + } + }; + + // A rotation that lands mid-registration runs once the current one ends, with the new token. + let rerun: { readonly token: string | null } | null = null; + const run = (observedToken: string | null = null): Promise => { + if (inFlight !== null) { + rerun = { token: observedToken ?? rerun?.token ?? null }; + return inFlight; + } + inFlight = attempt(observedToken) + .catch((error: unknown) => { + if (!disposed) setState({ kind: "failed", hostEnvironmentId, error: String(error) }); + }) + .finally(() => { + inFlight = null; + const next = rerun; + rerun = null; + if (next !== null && !disposed) void run(next.token); + }); + return inFlight; + }; + + const tokenSubscription = Notifications.addPushTokenListener((token) => { + if (typeof token.data === "string" && token.data.trim() !== "") void run(token.data.trim()); + }); + const appStateSubscription = AppState.addEventListener("change", (next) => { + if (next === "active") void run(); + }); + active = { hostEnvironmentId, run: () => run(), settled: () => inFlight ?? Promise.resolve() }; + void run(); + + return () => { + disposed = true; + tokenSubscription.remove(); + appStateSubscription.remove(); + if (active?.hostEnvironmentId === hostEnvironmentId) active = null; + }; +} + +let selection: { readonly hostEnvironmentId: EnvironmentId; readonly stop: () => void } | null = + null; + +/** + * Registers with one of the connected push hosts: the same one while it stays, so a second host + * never competes with it, and the next one when it leaves. + */ +export function syncForkPushHosts( + hostEnvironmentIds: ReadonlyArray, + register: ( + hostEnvironmentId: EnvironmentId, + input: ForkPushDeviceRegistration, + ) => ReturnType, +): void { + const current = selection?.hostEnvironmentId; + const chosen = + current !== undefined && hostEnvironmentIds.includes(current) + ? current + : ([...hostEnvironmentIds].sort()[0] ?? null); + if (chosen === (current ?? null)) return; + selection?.stop(); + selection = null; + if (chosen === null) { + setState({ kind: "idle" }); + return; + } + selection = { + hostEnvironmentId: chosen, + stop: startForkPushRegistration(chosen, (input) => register(chosen, input)), + }; +} + +/** Registers again now, after the user turned notifications back on or granted permission. */ +export async function resumeForkPush(): Promise { + await saveStopped(false); + await active?.run(); +} + +/** Stops push-host notifications on this phone until the user turns them back on. */ +export async function stopForkPush(unregister: (deviceId: string) => Promise) { + const hostEnvironmentId = active?.hostEnvironmentId ?? null; + stopGeneration++; + await saveStopped(true); + // A registration already sent finishes first, so the unregistration below comes after it. + await active?.settled(); + const deviceId = await loadOrCreateAgentAwarenessDeviceId(); + releaseAndroidAgentNotifications(); + await unregister(deviceId); + if (hostEnvironmentId !== null) setState({ kind: "stopped", hostEnvironmentId }); +} diff --git a/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx index 2c1f0d1ce226c..99e518e739738 100644 --- a/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsNotificationsRouteScreen.tsx @@ -49,6 +49,7 @@ import { SettingsRow } from "./components/SettingsRow"; import { SettingsSection } from "./components/SettingsSection"; import { SettingsSwitchRow } from "./components/SettingsSwitchRow"; import { SettingsScreen } from "./components/SettingsScreen"; +import { ForkPushSettingsSection } from "../fork-push/ForkPushSettingsSection"; import { resolveAgentAwarenessPlatformPresentation } from "./SettingsRouteScreen.logic"; /** @@ -120,6 +121,7 @@ export function SettingsNotificationsRouteScreen() { Notifications require T3 Connect in this app build. + ); @@ -488,6 +490,7 @@ function ConfiguredSettingsNotificationsRouteScreen() { /> ) : null} + ); diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index f5ef50bc9232f..cb9f3846d5d2a 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -1,5 +1,6 @@ import { CLIENT_GUARDED_RPC_SCOPES, + FORK_PUSH_QUERY_SCOPES, type DeviceListInput, clientRpcRequiredScopes, authScopeRequiredResponse, @@ -39,6 +40,7 @@ type WsRpcMethod = RpcGroup.Rpcs["_tag"]; export const RPC_REQUIRED_SCOPES = { ...CLIENT_GUARDED_RPC_SCOPES, + ...FORK_PUSH_QUERY_SCOPES, [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: AuthOrchestrationOperateScope, [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: AuthOrchestrationReadScope, [ORCHESTRATION_V2_WS_METHODS.getNativeChildWork]: AuthOrchestrationReadScope, diff --git a/apps/server/src/fork/push/ForkPushRpc.ts b/apps/server/src/fork/push/ForkPushRpc.ts new file mode 100644 index 0000000000000..61f39ee45421c --- /dev/null +++ b/apps/server/src/fork/push/ForkPushRpc.ts @@ -0,0 +1,83 @@ +import { FORK_PUSH_WS_METHODS, ForkPushError, type ForkPushStatus } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; + +import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; +import { PUBLISH_AGENT_ACTIVITY_SECRET } from "../../cloud/config.ts"; +import * as ForkPushSender from "./ForkPushSender.ts"; +import * as PushHost from "./PushHost.ts"; + +const NOT_A_HOST = "This environment holds no FCM key, so it is not a push host."; + +/** WebSocket handlers for the fork's push host, spread into ws.ts's handler set. */ +export const makeHandlers = Effect.gen(function* () { + const host = yield* PushHost.PushHost; + const sender = yield* ForkPushSender.ForkPushSender; + const secrets = yield* ServerSecretStore.ServerSecretStore; + + const status = Effect.gen(function* () { + const toggle = yield* secrets.get(PUBLISH_AGENT_ACTIVITY_SECRET).pipe( + Effect.map(Option.map((bytes) => new TextDecoder().decode(bytes))), + Effect.orElseSucceed(() => Option.none()), + ); + const hostStatus = host.enabled ? yield* host.status : { devices: [], peers: [] }; + return { + ...(yield* sender.status), + publishingOff: Option.getOrNull(toggle) === "false", + ...hostStatus, + } satisfies ForkPushStatus; + }).pipe(Effect.withSpan("ForkPushRpc.status")); + + return { + [FORK_PUSH_WS_METHODS.getStatus]: () => status, + [FORK_PUSH_WS_METHODS.registerDevice]: ( + input: Parameters[0], + ) => + host.enabled + ? host.registerDevice(input).pipe(Effect.as({})) + : Effect.fail(new ForkPushError({ message: NOT_A_HOST })), + [FORK_PUSH_WS_METHODS.unregisterDevice]: (input: { readonly deviceId: string }) => + host.unregisterDevice(input.deviceId).pipe(Effect.map((removed) => ({ removed }))), + [FORK_PUSH_WS_METHODS.sendTest]: (input: { readonly deviceId: string }) => + host.enabled + ? host.sendTest(input.deviceId).pipe( + Effect.mapError((message) => new ForkPushError({ message })), + Effect.flatMap((sent) => + sent + ? Effect.succeed({}) + : Effect.fail( + new ForkPushError({ message: "This device is not registered here." }), + ), + ), + ) + : Effect.fail(new ForkPushError({ message: NOT_A_HOST })), + }; +}); + +const unavailable = Effect.fail(new ForkPushError({ message: NOT_A_HOST })); + +/** + * The handlers, captured where ws.ts builds its route and provided to each connection. Without + * them, as in ws.ts's own tests, this server reports no push host. + */ +export const ForkPushHandlers = Context.Reference>( + "t3/fork/push/ForkPushRpc/ForkPushHandlers", + { + defaultValue: () => ({ + [FORK_PUSH_WS_METHODS.getStatus]: () => + Effect.succeed({ + role: "none", + host: null, + checkedAt: null, + error: null, + publishingOff: false, + devices: [], + peers: [], + } satisfies ForkPushStatus), + [FORK_PUSH_WS_METHODS.registerDevice]: () => unavailable, + [FORK_PUSH_WS_METHODS.unregisterDevice]: () => Effect.succeed({ removed: false }), + [FORK_PUSH_WS_METHODS.sendTest]: () => unavailable, + }), + }, +); diff --git a/apps/server/src/fork/push/ForkPushSender.test.ts b/apps/server/src/fork/push/ForkPushSender.test.ts new file mode 100644 index 0000000000000..8592a6edd75ca --- /dev/null +++ b/apps/server/src/fork/push/ForkPushSender.test.ts @@ -0,0 +1,148 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { EnvironmentId, type PeerLinkSummary, PeerLinkError } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; + +import * as PeerLinks from "../../peer/PeerLinks.ts"; +import { PushHostFcm } from "./fcm.ts"; +import * as ForkPushSender from "./ForkPushSender.ts"; +import * as PushHost from "./PushHost.ts"; +import * as PushHostStore from "./PushHostStore.ts"; + +const HOST_ID = EnvironmentId.make("env-host"); +const HOST_URL = "https://host.example.test"; + +const link = (status: PeerLinkSummary["status"]): PeerLinkSummary => ({ + environmentId: HOST_ID, + label: "Host", + urls: [HOST_URL], + access: "read-only", + linkedAt: DateTime.makeUnsafe("2026-11-01T00:00:00.000Z"), + expiresAt: DateTime.makeUnsafe("2026-12-01T00:00:00.000Z"), + lastReachedAt: null, + lastError: null, + status, +}); + +const peerHarness = () => { + const current = { status: "reachable" as PeerLinkSummary["status"], infoAsks: [] as string[] }; + const peerLinks = Layer.mock(PeerLinks.PeerLinks)({ + list: Effect.sync(() => [link(current.status)]), + resolve: () => + current.status === "expired" + ? Effect.fail( + new PeerLinkError({ reason: "expired", message: "The link to Host expired." }), + ) + : Effect.succeed({ link: link(current.status), url: HOST_URL, token: "link-token" }), + }); + const http = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + current.infoAsks.push(`${request.url} ${request.headers.authorization}`); + return HttpClientResponse.fromWeb( + request, + Response.json({ pushHost: true, environmentId: HOST_ID }), + ); + }), + ), + ); + const host = PushHost.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(PushHostFcm, PushHostFcm.of({ send: null, projectId: null })), + PushHostStore.layerMemory(), + NodeCrypto.layer, + ), + ), + ); + return { + current, + layer: ForkPushSender.layer.pipe(Layer.provide(Layer.mergeAll(peerLinks, http, host))), + }; +}; + +describe("ForkPushSender", () => { + it.effect("delivers in process on the push host", () => + Effect.gen(function* () { + const sender = yield* ForkPushSender.ForkPushSender; + const host = yield* PushHost.PushHost; + const route = yield* sender.resolve; + assert.strictEqual(route?.url, "fork-push:local"); + yield* route!.deliver!({ environmentId: "env-host", threadId: "thread", state: null }); + assert.deepStrictEqual( + (yield* host.status).peers.map((peer) => peer.environmentId), + ["env-host"], + ); + assert.strictEqual((yield* sender.status).role, "host"); + }).pipe( + Effect.provide( + ForkPushSender.layer.pipe( + // The push host never looks at its own links. + Layer.provide( + Layer.merge( + Layer.mock(PeerLinks.PeerLinks)({}), + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make(() => Effect.die("unused")), + ), + ), + ), + Layer.provideMerge( + PushHost.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed( + PushHostFcm, + PushHostFcm.of({ + send: () => Effect.succeed({ unregistered: false }), + projectId: "project", + }), + ), + PushHostStore.layerMemory(), + NodeCrypto.layer, + ), + ), + ), + ), + ), + ), + ), + ); + + it.effect("sends to the linked push host, and stops when the link lapses", () => { + const { current, layer } = peerHarness(); + return Effect.gen(function* () { + const sender = yield* ForkPushSender.ForkPushSender; + yield* sender.refresh; + assert.strictEqual( + current.infoAsks[0], + `${HOST_URL}/api/fork/push/v1/info Bearer link-token`, + ); + const route = yield* sender.resolve; + assert.deepStrictEqual(route, { + url: `${HOST_URL}/api/fork/push`, + issuer: HOST_URL, + environmentCredential: "link-token", + }); + const status = yield* sender.status; + assert.strictEqual(status.role, "sender"); + assert.strictEqual(status.host?.linkExpiresAt, "2026-12-01T00:00:00.000Z"); + + const changes = yield* sender.subscribe; + current.status = "expired"; + yield* sender.refresh; + yield* PubSub.take(changes); + assert.isNull(yield* sender.resolve); + const lapsed = yield* sender.status; + assert.strictEqual(lapsed.role, "none"); + assert.strictEqual(lapsed.host?.environmentId, HOST_ID); + assert.strictEqual(lapsed.error, "The link to Host expired."); + }).pipe(Effect.scoped, Effect.provide(layer)); + }); +}); diff --git a/apps/server/src/fork/push/ForkPushSender.ts b/apps/server/src/fork/push/ForkPushSender.ts new file mode 100644 index 0000000000000..dcbc5fb63c337 --- /dev/null +++ b/apps/server/src/fork/push/ForkPushSender.ts @@ -0,0 +1,202 @@ +import type { EnvironmentId } from "@t3tools/contracts"; +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 PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import type * as Scope from "effect/Scope"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; + +import * as PeerLinks from "../../peer/PeerLinks.ts"; +import { type ForkPushRoute, ForkPushTarget } from "./ForkPushTarget.ts"; +import * as PushHost from "./PushHost.ts"; + +export const PUSH_ROUTE_PREFIX = "/api/fork/push/v1"; +const LOCAL_ROUTE_URL = "fork-push:local"; +/** Links are re-checked this often, so a host found, lost or moved shows up within minutes. */ +const REFRESH_INTERVAL = "2 minutes"; + +export const PushHostInfo = Schema.Struct({ + pushHost: Schema.Boolean, + environmentId: Schema.String, +}); + +export interface ForkPushSenderStatus { + readonly role: "host" | "sender" | "none"; + /** The linked environment this server sends to, once one answered as the push host. */ + readonly host: { + readonly environmentId: string; + readonly label: string; + readonly url: string; + readonly linkExpiresAt: string; + } | null; + readonly checkedAt: string | null; + readonly error: string | null; +} + +interface SenderState { + readonly route: ForkPushRoute | null; + readonly status: ForkPushSenderStatus; +} + +/** + * Finds where this server's agent activity goes: into its own PushHost when it holds the FCM key, + * otherwise to the first linked environment that answers as a push host. + */ +export class ForkPushSender extends Context.Service< + ForkPushSender, + { + readonly resolve: Effect.Effect; + readonly status: Effect.Effect; + /** Checks the links again now, as a relink does. */ + readonly refresh: Effect.Effect; + /** Takes one value each time the route changes, so active threads can be published again. */ + readonly subscribe: Effect.Effect, never, Scope.Scope>; + } +>()("t3/fork/push/ForkPushSender") {} + +const routeKey = (route: ForkPushRoute | null) => + route === null ? "" : `${route.url}\n${route.environmentCredential}`; + +const make = Effect.gen(function* () { + const pushHost = yield* PushHost.PushHost; + const changes = yield* PubSub.unbounded(); + + if (pushHost.enabled) { + const route: ForkPushRoute = { + url: LOCAL_ROUTE_URL, + issuer: LOCAL_ROUTE_URL, + environmentCredential: LOCAL_ROUTE_URL, + deliver: (input) => pushHost.ingest(input).pipe(Effect.asVoid), + }; + return ForkPushSender.of({ + resolve: Effect.succeed(route), + status: Effect.succeed({ role: "host", host: null, checkedAt: null, error: null }), + refresh: Effect.void, + subscribe: PubSub.subscribe(changes), + }); + } + + const peerLinks = yield* PeerLinks.PeerLinks; + const httpClient = yield* HttpClient.HttpClient; + const stateRef = yield* Ref.make({ + route: null, + status: { role: "none", host: null, checkedAt: null, error: null }, + }); + + const askInfo = (url: string, token: string) => + HttpClientRequest.get(`${url}${PUSH_ROUTE_PREFIX}/info`).pipe( + HttpClientRequest.bearerToken(token), + httpClient.execute, + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap(HttpClientResponse.schemaBodyJson(PushHostInfo)), + Effect.timeout("10 seconds"), + Effect.option, + ); + + /** The push host behind one link, or why that link cannot reach it. */ + const probe = Effect.fn("ForkPushSender.probe")(function* (environmentId: EnvironmentId) { + const resolved = yield* peerLinks.resolve(environmentId).pipe(Effect.result); + if (resolved._tag === "Failure") return { found: null, error: resolved.failure.message }; + const { link, url, token } = resolved.success; + const info = yield* askInfo(url, token); + if (Option.isNone(info) || !info.value.pushHost) return { found: null, error: null }; + return { + found: { + route: { + url: `${url}/api/fork/push`, + issuer: url, + environmentCredential: token, + } satisfies ForkPushRoute, + host: { + environmentId: link.environmentId, + label: link.label, + url, + linkExpiresAt: DateTime.formatIso(link.expiresAt), + }, + }, + error: null, + }; + }); + + const refresh = Effect.gen(function* () { + const previous = yield* Ref.get(stateRef); + const links = yield* peerLinks.list.pipe(Effect.orElseSucceed(() => [])); + const known = previous.status.host?.environmentId; + // The host found last time is asked first; others only while it is not answering. + const candidates = [ + ...links.filter((link) => link.environmentId === known), + ...links.filter((link) => link.environmentId !== known && link.status === "reachable"), + ]; + let found: Effect.Success>["found"] = null; + let error: string | null = null; + for (const link of candidates) { + const result = yield* probe(link.environmentId); + if (link.environmentId === known) error = result.error; + if (result.found !== null) { + found = result.found; + error = null; + break; + } + } + const now = DateTime.formatIso(yield* DateTime.now); + const knownLink = links.find((link) => link.environmentId === known); + const next: SenderState = + found !== null + ? { + route: found.route, + status: { role: "sender", host: found.host, checkedAt: now, error: null }, + } + : { + route: null, + status: { + role: "none", + // A host that stops answering stays named, with the reason, until it is unlinked. + host: + previous.status.host !== null && knownLink !== undefined + ? { + ...previous.status.host, + linkExpiresAt: DateTime.formatIso(knownLink.expiresAt), + } + : null, + checkedAt: now, + error, + }, + }; + yield* Ref.set(stateRef, next); + if (routeKey(previous.route) !== routeKey(next.route)) { + yield* Effect.logInfo("Push host route changed", { + host: next.status.host?.environmentId ?? null, + url: next.route?.url ?? null, + }); + yield* PubSub.publish(changes, undefined); + } + }).pipe(Effect.withSpan("ForkPushSender.refresh")); + + yield* refresh.pipe( + Effect.repeat(Schedule.spaced(REFRESH_INTERVAL)), + Effect.catchCause((cause) => Effect.logWarning("Push host lookup stopped", { cause })), + Effect.forkScoped, + ); + + return ForkPushSender.of({ + resolve: Ref.get(stateRef).pipe(Effect.map((state) => state.route)), + status: Ref.get(stateRef).pipe(Effect.map((state) => state.status)), + refresh, + subscribe: PubSub.subscribe(changes), + }); +}); + +export const layer = Layer.effect(ForkPushSender, make); + +/** Hands AgentAwarenessRelay the sender's route. */ +export const layerTarget = Layer.effect( + ForkPushTarget, + ForkPushSender.pipe(Effect.map((sender) => ({ resolve: sender.resolve }))), +); diff --git a/apps/server/src/fork/push/ForkPushTarget.ts b/apps/server/src/fork/push/ForkPushTarget.ts new file mode 100644 index 0000000000000..dea834c1c9e06 --- /dev/null +++ b/apps/server/src/fork/push/ForkPushTarget.ts @@ -0,0 +1,30 @@ +import type { RelayAgentActivityState } from "@t3tools/contracts/relay"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; + +import { isAgentActivityPublishingEnabledValue } from "../../cloud/config.ts"; + +/** Where AgentAwarenessRelay sends agent activity instead of the T3 Connect relay. */ +export interface ForkPushRoute { + /** The relay API base URL, or a `fork-push:` name when `deliver` takes the update in process. */ + readonly url: string; + readonly issuer: string; + readonly environmentCredential: string; + readonly deliver?: (input: { + readonly environmentId: string; + readonly threadId: string; + readonly state: RelayAgentActivityState | null; + }) => Effect.Effect; +} + +/** Nothing here by default, so a server without the fork's push host publishes as upstream does. */ +export const ForkPushTarget = Context.Reference<{ + readonly resolve: Effect.Effect; +}>("t3/fork/push/ForkPushTarget", { defaultValue: () => ({ resolve: Effect.succeed(null) }) }); + +/** + * Upstream's toggle publishes only when it is set to on, and T3 Connect sets it. Updates to the + * push host stay on the owner's own servers, so there they flow unless the toggle is explicitly off. + */ +export const publishEnabled = (toggle: string | null, route: ForkPushRoute | null) => + route === null ? isAgentActivityPublishingEnabledValue(toggle) : toggle !== "false"; diff --git a/apps/server/src/fork/push/PushHost.test.ts b/apps/server/src/fork/push/PushHost.test.ts new file mode 100644 index 0000000000000..1f68c30c4c6d5 --- /dev/null +++ b/apps/server/src/fork/push/PushHost.test.ts @@ -0,0 +1,575 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import type { RelayAgentActivityState } from "@t3tools/contracts/relay"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as TestClock from "effect/testing/TestClock"; + +import { FcmClientError, type FcmSend, PushHostFcm } from "./fcm.ts"; +import * as PushHost from "./PushHost.ts"; +import * as PushHostStore from "./PushHostStore.ts"; + +const running: RelayAgentActivityState = { + environmentId: EnvironmentId.make("env-desktop"), + threadId: ThreadId.make("thread-1"), + projectTitle: "Project", + threadTitle: "Fix notifications", + phase: "running", + headline: "Working", + modelTitle: "Claude", + updatedAt: "1970-01-01T00:00:00.000Z", + deepLink: "/threads/env-desktop/thread-1", +}; +const waiting: RelayAgentActivityState = { + ...running, + phase: "waiting_for_approval", + headline: "Needs approval", + updatedAt: "1970-01-01T00:00:01.000Z", +}; + +const registration: PushHost.PushDeviceRegistration = { + deviceId: "device-1", + userId: "t3-fork:env-host", + token: "fcm-token", + packageName: "com.example.t3code", + label: "Phone", + preferences: { + notificationsEnabled: true, + liveActivitiesEnabled: true, + notifyOnApproval: true, + notifyOnInput: true, + notifyOnCompletion: true, + notifyOnFailure: true, + }, +}; + +type Sent = Parameters[0]; + +const harness = (options: { readonly unregistered?: boolean; readonly failWith?: number } = {}) => { + const sent: Sent[] = []; + const send: FcmSend = (input) => + options.failWith !== undefined + ? Effect.fail(new FcmClientError({ operation: "send", status: options.failWith })) + : Effect.sync(() => { + sent.push(input); + return { unregistered: options.unregistered === true }; + }); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + return { sent, fcm }; +}; + +const hostLayer = ( + fcm: Layer.Layer, + store: Layer.Layer, +) => PushHost.layer.pipe(Layer.provide(Layer.mergeAll(fcm, store, NodeCrypto.layer))); + +describe("PushHost", () => { + it.effect("shows the card and alerts once when a thread starts waiting", () => { + const { sent, fcm } = harness(); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.drain; + yield* host.ingest({ + environmentId: running.environmentId, + threadId: running.threadId, + state: running, + }); + yield* host.drain; + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: waiting, + }); + yield* host.drain; + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: waiting, + }); + yield* host.drain; + + assert.deepStrictEqual( + sent.map((message) => [message.alert, message.data.activity_active_count]), + [ + [false, "0"], + [false, "1"], + [true, "1"], + [false, "1"], + ], + ); + const alerted = sent[2]!; + assert.strictEqual(alerted.token, "fcm-token"); + assert.strictEqual(alerted.packageName, "com.example.t3code"); + assert.strictEqual(alerted.data.device_id, "device-1"); + assert.strictEqual(alerted.data.user_id, "t3-fork:env-host"); + assert.strictEqual(alerted.data.alert_path, "/threads/env-desktop/thread-1"); + assert.match(alerted.data.alert_id ?? "", /^[0-9a-f]{64}$/); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("drops a device FCM reports as unregistered", () => { + const { fcm } = harness({ unregistered: true }); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.drain; + assert.deepStrictEqual((yield* host.status).devices, []); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("records a refused send on the device and keeps it", () => { + const { fcm } = harness({ failWith: 400 }); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.drain; + const [device] = (yield* host.status).devices; + assert.strictEqual(device?.lastError, "FCM send failed (400)."); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("keeps devices and rows across a restart", () => { + const first = harness(); + const second = harness(); + const store = PushHostStore.layerMemory(); + return Effect.gen(function* () { + const memory = yield* PushHostStore.PushHostStore; + const storeLayer = Layer.succeed(PushHostStore.PushHostStore, memory); + yield* Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.ingest({ + environmentId: running.environmentId, + threadId: running.threadId, + state: running, + }); + yield* host.drain; + }).pipe(Effect.provide(hostLayer(first.fcm, storeLayer))); + yield* Effect.gen(function* () { + const host = yield* PushHost.PushHost; + const status = yield* host.status; + assert.deepStrictEqual( + status.devices.map((device) => device.deviceId), + ["device-1"], + ); + assert.deepStrictEqual( + status.peers.map((peer) => peer.environmentId), + ["env-desktop"], + ); + // Registering again replays the stored card without alerting. + yield* host.registerDevice(registration); + yield* host.drain; + }).pipe(Effect.provide(hostLayer(second.fcm, storeLayer))); + assert.deepStrictEqual( + second.sent.map((message) => [message.alert, message.data.activity_active_count]), + [[false, "1"]], + ); + }).pipe(Effect.provide(store)); + }); + + it.effect("clears a thread's row when its state is removed", () => { + const { sent, fcm } = harness(); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.drain; + yield* host.ingest({ + environmentId: running.environmentId, + threadId: running.threadId, + state: running, + }); + yield* host.drain; + yield* host.ingest({ + environmentId: running.environmentId, + threadId: running.threadId, + state: null, + }); + yield* host.drain; + assert.deepStrictEqual( + sent.map((message) => message.data.active), + ["false", "true", "false"], + ); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("keeps a token registered while the previous token's send is refused", () => { + const sent: Sent[] = []; + return Effect.gen(function* () { + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + const send: FcmSend = (input) => + input.token === "token-a" + ? Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.as({ unregistered: true }), + ) + : Effect.sync(() => { + sent.push(input); + return { unregistered: false }; + }); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + yield* Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice({ ...registration, token: "token-a" }); + yield* Deferred.await(started); + yield* host.registerDevice({ ...registration, token: "token-b" }); + yield* Deferred.succeed(release, undefined); + yield* host.drain; + assert.deepStrictEqual( + (yield* host.status).devices.map((device) => device.deviceId), + ["device-1"], + ); + assert.deepStrictEqual( + sent.map((message) => message.token), + ["token-b"], + ); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + }); + + it.effect("retries a send FCM could not take, then gives up after a bounded number", () => { + const sent: Sent[] = []; + let failures = 0; + const send: FcmSend = (input) => + failures > 0 + ? Effect.sync(() => void failures--).pipe( + Effect.andThen(Effect.fail(new FcmClientError({ operation: "send", status: 503 }))), + ) + : Effect.sync(() => { + sent.push(input); + return { unregistered: false }; + }); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.drain; + failures = 2; + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: waiting, + }); + const drained = yield* Effect.forkChild(host.drain); + yield* TestClock.adjust("1 minute"); + yield* Fiber.join(drained); + assert.deepStrictEqual( + sent.map((message) => message.alert), + [false, true], + ); + assert.isNull((yield* host.status).devices[0]?.lastError ?? null); + + failures = 100; + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: null, + }); + const gaveUp = yield* Effect.forkChild(host.drain); + yield* TestClock.adjust("10 minutes"); + yield* Fiber.join(gaveUp); + assert.isTrue(failures > 90); + assert.strictEqual((yield* host.status).devices[0]?.lastError, "FCM send failed (503)."); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("retries when FCM's token answer stalls after its headers", () => { + const sent: Sent[] = []; + let stalls = 0; + const send: FcmSend = (input) => + stalls > 0 + ? Effect.sync(() => void stalls--).pipe( + Effect.andThen( + Effect.fail( + new FcmClientError({ + operation: "authorize", + status: 200, + cause: new Cause.TimeoutError(), + }), + ), + ), + ) + : Effect.sync(() => { + sent.push(input); + return { unregistered: false }; + }); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.drain; + stalls = 1; + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: waiting, + }); + const drained = yield* Effect.forkChild(host.drain); + yield* TestClock.adjust("1 minute"); + yield* Fiber.join(drained); + assert.deepStrictEqual( + sent.map((message) => message.alert), + [false, true], + ); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect.each(["completed", "waiting_for_approval", "waiting_for_input", "failed"] as const)( + "deleting one thread preserves another thread's %s alert", + (phase) => { + const { sent, fcm } = harness(); + const second: RelayAgentActivityState = { + ...running, + threadId: ThreadId.make("thread-2"), + threadTitle: "Second", + deepLink: "/threads/env-desktop/thread-2", + }; + const changed = { ...running, phase, updatedAt: "1970-01-01T00:00:02.000Z" }; + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.ingest({ + environmentId: running.environmentId, + threadId: running.threadId, + state: running, + }); + yield* host.ingest({ + environmentId: second.environmentId, + threadId: second.threadId, + state: second, + }); + yield* host.drain; + // The deletion is queued first but already reads the other thread's change. + yield* host.ingest({ + environmentId: second.environmentId, + threadId: second.threadId, + state: null, + }); + yield* host.ingest({ + environmentId: changed.environmentId, + threadId: changed.threadId, + state: changed, + }); + yield* host.drain; + assert.strictEqual(sent.filter((message) => message.alert).length, 1); + yield* host.ingest({ + environmentId: changed.environmentId, + threadId: changed.threadId, + state: changed, + }); + yield* host.drain; + assert.strictEqual(sent.filter((message) => message.alert).length, 1); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }, + ); + + it.effect("a device FCM keeps refusing does not delay another device", () => { + const sent: Sent[] = []; + let stuck = false; + const send: FcmSend = (input) => + stuck && input.token === "token-stuck" + ? Effect.fail(new FcmClientError({ operation: "send", status: 429 })) + : Effect.sync(() => { + sent.push(input); + return { unregistered: false }; + }); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice({ ...registration, deviceId: "stuck", token: "token-stuck" }); + yield* host.registerDevice(registration); + yield* host.drain; + stuck = true; + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: waiting, + }); + yield* TestClock.adjust("2 seconds"); + assert.deepStrictEqual( + sent.filter((message) => message.token === "fcm-token").map((message) => message.alert), + [false, true], + ); + const drained = yield* Effect.forkChild(host.drain); + yield* TestClock.adjust("10 minutes"); + yield* Fiber.join(drained); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("keeps one pending delivery per thread while a device's sends fail", () => { + let attempts = 0; + let failing = true; + const send: FcmSend = () => + Effect.sync(() => void attempts++).pipe( + Effect.andThen( + failing + ? Effect.fail(new FcmClientError({ operation: "send", status: 503 })) + : Effect.succeed({ unregistered: false }), + ), + ); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + const threads = ["thread-1", "thread-2", "thread-3"].map((threadId) => ThreadId.make(threadId)); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + for (let update = 0; update < 100; update++) { + const threadId = threads[update % threads.length]!; + yield* host.ingest({ + environmentId: running.environmentId, + threadId, + state: { + ...running, + threadId, + updatedAt: DateTime.formatIso(DateTime.makeUnsafe(update * 1000)), + }, + }); + } + failing = false; + const drained = yield* Effect.forkChild(host.drain); + yield* TestClock.adjust("2 hours"); + yield* Fiber.join(drained); + // The replay in flight retries, then each thread is delivered once. + assert.isAtMost(attempts, 4 + threads.length); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("loses no thread's alert while a device's deliveries wait", () => { + const sent: Sent[] = []; + return Effect.gen(function* () { + const busy = yield* Deferred.make(); + const release = yield* Deferred.make(); + let blocked = true; + const send: FcmSend = (input) => + Effect.gen(function* () { + if (blocked) { + blocked = false; + yield* Deferred.succeed(busy, undefined); + yield* Deferred.await(release); + } + sent.push(input); + return { unregistered: false }; + }); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + const second: RelayAgentActivityState = { + ...waiting, + threadId: ThreadId.make("thread-2"), + threadTitle: "Second", + deepLink: "/threads/env-desktop/thread-2", + }; + // Without ongoing activity there is no card baseline, so each thread alerts on its own. + const alertsOnly = { + ...registration, + preferences: { ...registration.preferences, liveActivitiesEnabled: false }, + }; + yield* Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(alertsOnly); + yield* Deferred.await(busy); + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: waiting, + }); + yield* host.ingest({ + environmentId: second.environmentId, + threadId: second.threadId, + state: second, + }); + yield* host.ingest({ + environmentId: second.environmentId, + threadId: second.threadId, + state: second, + }); + yield* Deferred.succeed(release, undefined); + yield* host.drain; + assert.deepStrictEqual( + sent.filter((message) => message.alert).map((message) => message.data.alert_path), + ["/threads/env-desktop/thread-1", "/threads/env-desktop/thread-2"], + ); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + }); + + it.effect("a replay that runs before a thread's pending delivery still alerts", () => { + const sent: Sent[] = []; + return Effect.gen(function* () { + const busy = yield* Deferred.make(); + const release = yield* Deferred.make(); + let blocked = false; + const send: FcmSend = (input) => + Effect.gen(function* () { + if (blocked) { + blocked = false; + yield* Deferred.succeed(busy, undefined); + yield* Deferred.await(release); + } + sent.push(input); + return { unregistered: false }; + }); + const fcm = Layer.succeed(PushHostFcm, PushHostFcm.of({ send, projectId: "project" })); + yield* Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.registerDevice(registration); + yield* host.ingest({ + environmentId: running.environmentId, + threadId: running.threadId, + state: running, + }); + yield* host.drain; + blocked = true; + yield* host.registerDevice(registration); + yield* Deferred.await(busy); + yield* host.ingest({ + environmentId: waiting.environmentId, + threadId: waiting.threadId, + state: waiting, + }); + yield* host.registerDevice(registration); + yield* Deferred.succeed(release, undefined); + yield* host.drain; + assert.deepStrictEqual( + sent.filter((message) => message.alert).map((message) => message.data.alert_path), + ["/threads/env-desktop/thread-1"], + ); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + }); + + it.effect("forgets peers not heard from in 30 days", () => { + const { fcm } = harness(); + return Effect.gen(function* () { + const host = yield* PushHost.PushHost; + yield* host.ingest({ environmentId: "env-gone", threadId: "thread", state: null }); + yield* TestClock.adjust("31 days"); + yield* host.ingest({ + environmentId: running.environmentId, + threadId: running.threadId, + state: null, + }); + assert.deepStrictEqual( + (yield* host.status).peers.map((peer) => peer.environmentId), + ["env-desktop"], + ); + }).pipe(Effect.provide(hostLayer(fcm, PushHostStore.layerMemory()))); + }); + + it.effect("is off without an FCM key", () => + Effect.gen(function* () { + const host = yield* PushHost.PushHost; + assert.isFalse(host.enabled); + }).pipe( + Effect.provide( + hostLayer( + Layer.succeed(PushHostFcm, PushHostFcm.of({ send: null, projectId: null })), + PushHostStore.layerMemory(), + ), + ), + ), + ); +}); diff --git a/apps/server/src/fork/push/PushHost.ts b/apps/server/src/fork/push/PushHost.ts new file mode 100644 index 0000000000000..da09929db5449 --- /dev/null +++ b/apps/server/src/fork/push/PushHost.ts @@ -0,0 +1,486 @@ +import type { + RelayAgentActivityState, + RelayAgentAwarenessPreferences, +} from "@t3tools/contracts/relay"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Digest from "effect/crypto/Digest"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Hex from "effect/encoding/Hex"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Semaphore from "effect/Semaphore"; +import * as HttpClientError from "effect/http/HttpClientError"; + +import { + makeAggregateState, + TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS, +} from "../../../../../infra/relay/src/agentActivity/agentActivityAggregate.ts"; +import { + isExpiredAgentActivityState, + isTerminalPhase, +} from "../../../../../infra/relay/src/agentActivity/agentActivityPayloads.ts"; +import { + androidActivityData, + fitFcmData, +} from "../../../../../infra/relay/src/agentActivity/fcmPayloads.ts"; +import { androidAlertForAggregate, androidAlertForState } from "./androidAlerts.ts"; +import { type FcmClientError, PushHostFcm } from "./fcm.ts"; +import * as PushHostStore from "./PushHostStore.ts"; + +export interface PushDeviceRegistration { + readonly deviceId: string; + readonly userId: string; + readonly token: string; + readonly packageName: string | null; + readonly label: string; + readonly preferences: RelayAgentAwarenessPreferences; +} + +export interface PushHostStatus { + readonly devices: ReadonlyArray<{ + readonly deviceId: string; + readonly label: string; + readonly registeredAt: string; + readonly lastSentAt: string | null; + readonly lastError: string | null; + }>; + readonly peers: ReadonlyArray; +} + +/** + * A private relay for one owner: it keeps the latest agent activity of this server and its + * linked peers, and sends Android devices the same FCM payloads upstream's relay sends. + */ +export class PushHost extends Context.Service< + PushHost, + { + /** Whether this server holds an FCM key and acts as the push host. */ + readonly enabled: boolean; + readonly ingest: (input: { + readonly environmentId: string; + readonly threadId: string; + readonly state: RelayAgentActivityState | null; + }) => Effect.Effect>; + readonly registerDevice: (input: PushDeviceRegistration) => Effect.Effect; + readonly unregisterDevice: (deviceId: string) => Effect.Effect; + /** Sends a test alert; fails with FCM's answer when it is refused. */ + readonly sendTest: (deviceId: string) => Effect.Effect; + readonly status: Effect.Effect; + readonly drain: Effect.Effect; + } +>()("t3/fork/push/PushHost") {} + +type Delivery = + | { + readonly kind: "state"; + readonly deviceId: string; + readonly state: RelayAgentActivityState | null; + } + | { readonly kind: "replay"; readonly deviceId: string }; + +/** A peer silent this long is most likely gone, so it leaves the status. */ +const PEER_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; +const SEND_RETRY_DELAYS = ["1 second", "4 seconds", "16 seconds"] as const; + +/** + * FcmClient keeps the response status when reading the body times out or breaks, so a stalled + * 200 counts as transient by its cause, as do transport failures, rate limits and server errors. + */ +const isTransientFcmError = (error: FcmClientError) => + Cause.isTimeoutError(error.cause) || + HttpClientError.isHttpClientError(error.cause) || + error.status === null || + error.status === 429 || + error.status >= 500; + +const rowKey = (environmentId: string, threadId: string) => + JSON.stringify([environmentId, threadId]); + +const keepRow = (row: RelayAgentActivityState, nowMs: number) => + isTerminalPhase(row) + ? nowMs - Date.parse(row.updatedAt) <= TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS + : !isExpiredAgentActivityState(row, nowMs); + +const describeError = (error: unknown) => + typeof error === "object" && error !== null && "message" in error + ? String(error.message) + : String(error); + +const make = Effect.gen(function* () { + const fcm = yield* PushHostFcm; + const store = yield* PushHostStore.PushHostStore; + const digest = yield* Digest.Digest; + const sha256Hex = (value: string) => + digest + .digest(Digest.sha256, new TextEncoder().encode(value)) + .pipe(Effect.map(Hex.encode), Effect.orDie); + const send = fcm.send; + const stateRef = yield* Ref.make(yield* store.load); + const writes = yield* Semaphore.make(1); + + const persist = writes.withPermits(1)(Ref.get(stateRef).pipe(Effect.flatMap(store.save))); + + const updateDevice = ( + deviceId: string, + f: (device: PushHostStore.PushDevice) => PushHostStore.PushDevice | null, + ) => + Ref.update(stateRef, (state) => ({ + ...state, + devices: state.devices.flatMap((device) => { + if (device.deviceId !== deviceId) return [device]; + const next = f(device); + return next === null ? [] : [next]; + }), + })); + + // A send answers for the token it went to; a registration replaced meanwhile keeps its own state. + const updateRegistration = ( + sentTo: PushHostStore.PushDevice, + f: (device: PushHostStore.PushDevice) => PushHostStore.PushDevice | null, + ) => + updateDevice(sentTo.deviceId, (current) => + current.token === sentTo.token ? f(current) : current, + ); + + const sendTo = Effect.fn("PushHost.sendTo")(function* ( + device: PushHostStore.PushDevice, + input: { readonly data: Record; readonly alert: boolean }, + ) { + if (send === null) return { unregistered: false }; + return yield* send({ + token: device.token, + packageName: device.packageName, + alert: input.alert, + data: fitFcmData(input.data), + }); + }); + + // Ingest has already answered its peer, so a send FCM could not take is retried here, and only + // while the registration still holds the token it was meant for. + const sendWithRetries = Effect.fn("PushHost.sendWithRetries")(function* ( + device: PushHostStore.PushDevice, + input: { readonly data: Record; readonly alert: boolean }, + ) { + for (let attempt = 0; ; attempt++) { + const result = yield* sendTo(device, input).pipe(Effect.result); + if (result._tag === "Success") return result.success; + const error = result.failure; + const delay = SEND_RETRY_DELAYS[attempt]; + const retryable = isTransientFcmError(error); + if (!retryable || delay === undefined) return yield* error; + yield* Effect.sleep(delay); + const current = (yield* Ref.get(stateRef)).devices.find( + (candidate) => candidate.deviceId === device.deviceId, + ); + if (current?.token !== device.token) return yield* error; + } + }); + + // Mirrors FcmDeliveries.process in infra/relay, with every environment allowed to alert. + const deliver = Effect.fn("PushHost.deliver")(function* (delivery: Delivery) { + const state = yield* Ref.get(stateRef); + const device = state.devices.find((candidate) => candidate.deviceId === delivery.deviceId); + if (device === undefined) return; + const now = yield* DateTime.now; + const nowMs = now.epochMilliseconds; + const preferences = device.preferences; + const jobState = delivery.kind === "state" ? delivery.state : null; + const aggregate = makeAggregateState({ + activeStates: preferences.liveActivitiesEnabled ? state.rows : [], + terminalState: null, + nowMs, + }); + const previousAggregate = device.lastAggregate; + // A deletion repaints the card but keeps the baseline, or it would swallow another thread's + // pending transition that it happened to read. + const acknowledge = jobState !== null || delivery.kind === "replay" || aggregate === null; + let alert: ReturnType = null; + // A replay moves the baseline too, so it alerts on what it acknowledges: it may run before a + // thread's pending delivery and would otherwise swallow that thread's transition. + if ((jobState !== null || delivery.kind === "replay") && preferences.notificationsEnabled) { + const row = + jobState === null + ? undefined + : state.rows.find( + (candidate) => + candidate.environmentId === jobState.environmentId && + candidate.threadId === jobState.threadId, + ); + const current = + jobState !== null && + row !== undefined && + row.phase === jobState.phase && + row.updatedAt === jobState.updatedAt + ? row + : null; + if (jobState !== null && !previousAggregate && current === null) return; + if (preferences.liveActivitiesEnabled && previousAggregate && aggregate) { + alert = androidAlertForAggregate({ + previousAggregate, + nextAggregate: aggregate, + preferences, + nowMs, + }); + } else if (current !== null && !isExpiredAgentActivityState(current, nowMs)) { + alert = androidAlertForState(current, preferences, nowMs); + } + } + const displayed = preferences.liveActivitiesEnabled ? aggregate : null; + if (!displayed && !alert && !previousAggregate && jobState !== null) return; + const data: Record = { + t3_kind: "agent_activity", + device_id: device.deviceId, + user_id: device.userId, + updated_at: String(nowMs), + ...androidActivityData(displayed), + ...alert, + }; + // Hash the full stable group identity to stay within the payload budget. + if (alert) data.alert_id = yield* sha256Hex(alert.alert_id); + const sentAt = DateTime.formatIso(now); + const result = yield* sendWithRetries(device, { data, alert: alert !== null }).pipe( + Effect.catch((error) => + Effect.logWarning("Push host could not send to a device", { + deviceId: device.deviceId, + status: error.status, + error: error.message, + }).pipe( + Effect.andThen( + updateRegistration(device, (current) => ({ ...current, lastError: error.message })), + ), + Effect.as(null), + ), + ), + ); + if (result === null) return yield* persist; + if (result.unregistered) { + yield* Effect.logInfo("Push host dropped a device FCM no longer knows", { + deviceId: device.deviceId, + }); + yield* updateRegistration(device, () => null); + return yield* persist; + } + yield* Effect.logInfo("Push host sent agent activity", { + deviceId: device.deviceId, + alert: alert !== null, + activeCount: displayed?.activeCount ?? 0, + }); + yield* updateRegistration(device, (current) => ({ + ...current, + lastAggregate: !acknowledge + ? current.lastAggregate + : preferences.liveActivitiesEnabled + ? aggregate + : null, + lastSentAt: sentAt, + lastError: null, + })); + yield* persist; + }); + + // Each device delivers on its own fiber, so one FCM keeps refusing cannot hold up another. + // While a device is busy its pending work keeps one entry per thread: delivery reads the rows + // and compares them with the device's baseline when it runs, as upstream's relay does when its + // queue lags, so a later update to the same thread decides the same alerts as the earlier ones. + const scope = yield* Effect.scope; + const pending = new Map< + string, + { replay: boolean; readonly threads: Map } + >(); + const running = new Map>(); + + const takeNext = (deviceId: string): Delivery | null => { + const queue = pending.get(deviceId); + if (queue === undefined) { + running.delete(deviceId); + return null; + } + if (queue.replay) { + queue.replay = false; + return { kind: "replay", deviceId }; + } + const next = queue.threads.entries().next(); + if (next.done === true) { + // In the same step as the check, so work enqueued from here on starts a new fiber. + pending.delete(deviceId); + running.delete(deviceId); + return null; + } + queue.threads.delete(next.value[0]); + return { kind: "state", deviceId, state: next.value[1] }; + }; + + const deliverPending = Effect.fn("PushHost.deliverPending")(function* (deviceId: string) { + while (true) { + const delivery = takeNext(deviceId); + if (delivery === null) return; + yield* deliver(delivery).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Push host delivery failed", { deviceId, cause }), + ), + ); + } + }); + + const enqueue = ( + deviceId: string, + work: + | { readonly replay: true } + | { readonly key: string; readonly state: RelayAgentActivityState | null }, + ) => + Effect.suspend(() => { + const queue = pending.get(deviceId) ?? { replay: false, threads: new Map() }; + pending.set(deviceId, queue); + if ("replay" in work) { + queue.replay = true; + } else { + // Re-inserting moves the thread behind work that arrived before this update. + queue.threads.delete(work.key); + queue.threads.set(work.key, work.state); + } + if (running.has(deviceId)) return Effect.void; + return deliverPending(deviceId).pipe( + Effect.forkIn(scope), + Effect.map((fiber) => { + if (pending.has(deviceId)) running.set(deviceId, fiber); + }), + ); + }); + + const drain: Effect.Effect = Effect.suspend(() => { + const fibers = [...running.values()]; + return fibers.length === 0 ? Effect.void : Fiber.awaitAll(fibers).pipe(Effect.andThen(drain)); + }); + + const ingest: PushHost["Service"]["ingest"] = Effect.fn("PushHost.ingest")(function* (input) { + const now = yield* DateTime.now; + const key = rowKey(input.environmentId, input.threadId); + const state = yield* Ref.updateAndGet(stateRef, (current) => ({ + ...current, + rows: [ + ...current.rows.filter( + (row) => + rowKey(row.environmentId, row.threadId) !== key && keepRow(row, now.epochMilliseconds), + ), + ...(input.state === null ? [] : [input.state]), + ], + peers: [ + ...current.peers.filter( + (peer) => + peer.environmentId !== input.environmentId && + now.epochMilliseconds - Date.parse(peer.lastHeardAt) <= PEER_RETENTION_MS, + ), + { environmentId: input.environmentId, lastHeardAt: DateTime.formatIso(now) }, + ], + })); + yield* persist; + yield* Effect.forEach( + state.devices, + (device) => enqueue(device.deviceId, { key, state: input.state }), + { discard: true }, + ); + return state.devices.map((device) => device.deviceId); + }); + + const registerDevice: PushHost["Service"]["registerDevice"] = Effect.fn( + "PushHost.registerDevice", + )(function* (input) { + const registeredAt = DateTime.formatIso(yield* DateTime.now); + yield* Ref.update(stateRef, (state) => { + const existing = state.devices.find((device) => device.deviceId === input.deviceId); + // A new token or identity is a fresh native store, so its card baseline starts over. + const sameTarget = + existing !== undefined && + existing.token === input.token && + existing.userId === input.userId; + const device: PushHostStore.PushDevice = { + ...input, + registeredAt: existing?.registeredAt ?? registeredAt, + lastAggregate: sameTarget ? existing.lastAggregate : null, + lastSentAt: existing?.lastSentAt ?? null, + lastError: sameTarget ? existing.lastError : null, + }; + return { + ...state, + // One phone keeps one registration even when its device ID changes with a reinstall. + devices: [ + ...state.devices.filter( + (candidate) => candidate.deviceId !== input.deviceId && candidate.token !== input.token, + ), + device, + ], + }; + }); + yield* persist; + yield* enqueue(input.deviceId, { replay: true }); + }); + + const unregisterDevice: PushHost["Service"]["unregisterDevice"] = Effect.fn( + "PushHost.unregisterDevice", + )(function* (deviceId) { + const before = yield* Ref.get(stateRef); + if (!before.devices.some((device) => device.deviceId === deviceId)) return false; + yield* updateDevice(deviceId, () => null); + yield* persist; + return true; + }); + + const sendTest: PushHost["Service"]["sendTest"] = Effect.fn("PushHost.sendTest")( + function* (deviceId) { + const state = yield* Ref.get(stateRef); + const device = state.devices.find((candidate) => candidate.deviceId === deviceId); + if (device === undefined) return false; + const now = yield* DateTime.now; + const alertId = yield* sha256Hex(`test:${now.epochMilliseconds}`); + const result = yield* sendTo(device, { + alert: true, + data: { + t3_kind: "agent_activity", + device_id: device.deviceId, + user_id: device.userId, + updated_at: String(now.epochMilliseconds), + // The card the device already shows, so the test does not disturb it. + ...androidActivityData(device.lastAggregate), + alert_id: alertId, + alert_title: "T3 Code", + alert_body: "Notifications from your push host work.", + alert_path: "/", + }, + }).pipe(Effect.mapError(describeError)); + if (result.unregistered) { + yield* updateRegistration(device, () => null); + yield* persist; + return yield* Effect.fail("FCM no longer knows this device. Register it again."); + } + return true; + }, + ); + + const status: PushHost["Service"]["status"] = Ref.get(stateRef).pipe( + Effect.map((state) => ({ + devices: state.devices.map(({ deviceId, label, registeredAt, lastSentAt, lastError }) => ({ + deviceId, + label, + registeredAt, + lastSentAt, + lastError, + })), + peers: state.peers, + })), + ); + + return PushHost.of({ + enabled: send !== null, + ingest, + registerDevice, + unregisterDevice, + sendTest, + status, + drain, + }); +}); + +export const layer = Layer.effect(PushHost, make); diff --git a/apps/server/src/fork/push/PushHostStore.ts b/apps/server/src/fork/push/PushHostStore.ts new file mode 100644 index 0000000000000..004a59fa30c71 --- /dev/null +++ b/apps/server/src/fork/push/PushHostStore.ts @@ -0,0 +1,106 @@ +import { + RelayAgentActivityAggregateState, + RelayAgentActivityState, + RelayAgentAwarenessPreferences, +} from "@t3tools/contracts/relay"; +import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; + +import * as ServerConfig from "../../config.ts"; + +export const PushDevice = Schema.Struct({ + deviceId: Schema.String, + userId: Schema.String, + token: Schema.String, + packageName: Schema.NullOr(Schema.String), + label: Schema.String, + preferences: RelayAgentAwarenessPreferences, + registeredAt: Schema.String, + lastAggregate: Schema.NullOr(RelayAgentActivityAggregateState), + lastSentAt: Schema.NullOr(Schema.String), + lastError: Schema.NullOr(Schema.String), +}); +export type PushDevice = typeof PushDevice.Type; + +export const PushPeer = Schema.Struct({ + environmentId: Schema.String, + lastHeardAt: Schema.String, +}); +export type PushPeer = typeof PushPeer.Type; + +export const PushHostState = Schema.Struct({ + devices: Schema.Array(PushDevice), + rows: Schema.Array(RelayAgentActivityState), + peers: Schema.Array(PushPeer), +}); +export type PushHostState = typeof PushHostState.Type; + +const EMPTY_STATE: PushHostState = { devices: [], rows: [], peers: [] }; + +const decodeState = Schema.decodeUnknownOption(Schema.fromJsonString(PushHostState)); +const encodeState = Schema.encodeEffect(Schema.fromJsonString(PushHostState)); + +/** + * The push host's devices, thread rows and peers, kept in one JSON file rather than a + * migration, since a fork migration would collide with upstream's numbering. + */ +export class PushHostStore extends Context.Service< + PushHostStore, + { + readonly load: Effect.Effect; + readonly save: (state: PushHostState) => Effect.Effect; + } +>()("t3/fork/push/PushHostStore") {} + +const STORE_FILE_NAME = "fork-push-host.json"; + +export const layer = Layer.effect( + PushHostStore, + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const platform = yield* HostProcess.Platform; + const filePath = path.join(config.stateDir, STORE_FILE_NAME); + return PushHostStore.of({ + load: fs.readFileString(filePath).pipe( + Effect.map((contents) => Option.getOrElse(decodeState(contents), () => EMPTY_STATE)), + Effect.orElseSucceed(() => EMPTY_STATE), + ), + save: Effect.fn("PushHostStore.save")(function* (state) { + yield* encodeState(state).pipe( + Effect.flatMap((contents) => writeFileStringAtomically({ filePath, contents })), + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + Effect.provideService(HostProcess.Platform, platform), + Effect.catchCause((cause) => + Effect.logWarning("Push host state could not be saved", { filePath, cause }), + ), + ); + }), + }); + }), +); + +/** Keeps state in memory, for tests. */ +export const layerMemory = (initial: PushHostState = EMPTY_STATE) => + Layer.effect( + PushHostStore, + Effect.gen(function* () { + const ref = yield* Ref.make(initial); + return PushHostStore.of({ + load: Ref.get(ref), + save: Effect.fn("PushHostStore.saveInMemory")(function* (next) { + yield* Ref.set(ref, next); + }), + }); + }), + ); diff --git a/apps/server/src/fork/push/androidAlerts.ts b/apps/server/src/fork/push/androidAlerts.ts new file mode 100644 index 0000000000000..2694ffde5c742 --- /dev/null +++ b/apps/server/src/fork/push/androidAlerts.ts @@ -0,0 +1,69 @@ +// Copied from infra/relay/src/agentActivity/FcmDeliveries.ts, whose module also loads the relay's +// database. androidAlerts.test.ts compares these with the originals, so upstream changes surface. +import type { + RelayAgentActivityAggregateState, + RelayAgentActivityState, + RelayAgentAwarenessPreferences, +} from "@t3tools/contracts/relay"; + +import { statusForPhase } from "../../../../../infra/relay/src/agentActivity/agentActivityAggregate.ts"; +import { + alertForActivityRows, + attentionTransitionRows, + shouldAlertForActivity, + terminalTransitionRows, +} from "../../../../../infra/relay/src/agentActivity/agentActivityAlerts.ts"; +import { notificationForActivity } from "../../../../../infra/relay/src/agentActivity/agentActivityPayloads.ts"; + +export function androidAlertForState( + state: RelayAgentActivityState, + preferences: RelayAgentAwarenessPreferences, + nowMs: number, +) { + if (!shouldAlertForActivity({ ...state, preferences, nowMs })) return null; + const notification = notificationForActivity({ ...state, status: statusForPhase(state.phase) }); + return { + alert_id: JSON.stringify([state.environmentId, state.threadId, state.phase, state.updatedAt]), + alert_group: JSON.stringify([state.environmentId, state.threadId]), + alert_title: notification.title, + alert_body: notification.body, + alert_path: notification.deepLink, + }; +} + +export function androidAlertForAggregate(input: { + readonly previousAggregate: RelayAgentActivityAggregateState; + readonly nextAggregate: RelayAgentActivityAggregateState; + readonly preferences: RelayAgentAwarenessPreferences; + readonly nowMs: number; +}) { + if (!input.preferences.notificationsEnabled) return null; + const attention = attentionTransitionRows(input); + const activities = + attention.length > 0 + ? attention + : terminalTransitionRows({ ...input, includeUnobserved: true }); + const first = activities[0]; + const alert = alertForActivityRows(activities); + if (!first || !alert) return null; + if (activities.length === 1) { + const notification = notificationForActivity(first); + return { + alert_id: JSON.stringify([first.environmentId, first.threadId, first.phase, first.updatedAt]), + alert_group: JSON.stringify([first.environmentId, first.threadId]), + alert_title: notification.title, + alert_body: notification.body, + alert_path: notification.deepLink, + }; + } + return { + alert_id: JSON.stringify( + activities + .map((row) => [row.environmentId, row.threadId, row.phase, row.updatedAt]) + .sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b))), + ), + alert_title: alert.title, + alert_body: alert.body, + alert_path: "/", + }; +} diff --git a/apps/server/src/fork/push/fcm.ts b/apps/server/src/fork/push/fcm.ts new file mode 100644 index 0000000000000..dedfab6fd7a94 --- /dev/null +++ b/apps/server/src/fork/push/fcm.ts @@ -0,0 +1,127 @@ +import * as Config from "effect/Config"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Redacted from "effect/Redacted"; +import * as Schema from "effect/Schema"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; +import * as HttpBody from "effect/http/HttpBody"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; + +import * as RelayConfiguration from "../../../../../infra/relay/src/Config.ts"; +import * as WebCrypto from "../../../../../infra/relay/src/WebCrypto.ts"; +import * as FcmAssertionSigner from "../../../../../infra/relay/src/agentActivity/FcmAssertionSigner.ts"; +import * as FcmClient from "../../../../../infra/relay/src/agentActivity/FcmClient.ts"; + +/** The Firebase service account that makes this server the push host. */ +const SERVICE_ACCOUNT_FILE_ENV = "T3CODE_FCM_SERVICE_ACCOUNT_FILE"; +/** Development only: FCM checks each message and the key but delivers nothing. */ +const VALIDATE_ONLY_ENV = "T3CODE_FCM_VALIDATE_ONLY"; + +export type FcmSend = FcmClient.FcmClient["Service"]["send"]; +export { FcmClientError } from "../../../../../infra/relay/src/agentActivity/FcmClient.ts"; + +/** How this server sends to FCM; `send` is null when it holds no key and is not the push host. */ +export class PushHostFcm extends Context.Service< + PushHostFcm, + { readonly send: FcmSend | null; readonly projectId: string | null } +>()("t3/fork/push/fcm/PushHostFcm") {} + +const decodeProjectId = Schema.decodeUnknownOption( + Schema.fromJsonString(Schema.Struct({ project_id: Schema.NonEmptyString })), +); + +const decodeMessage = Schema.decodeUnknownOption( + Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)), +); + +const isFcmSend = (request: HttpClientRequest.HttpClientRequest) => + request.method === "POST" && request.url.startsWith("https://fcm.googleapis.com/"); + +/** Adds FCM's `validate_only` to every send, so a check never reaches a phone. */ +const validateOnly = (request: HttpClientRequest.HttpClientRequest) => { + if (!isFcmSend(request) || request.body._tag !== "Uint8Array") return request; + const message = decodeMessage(new TextDecoder().decode(request.body.body)); + if (Option.isNone(message)) return request; + return HttpClientRequest.setBody( + request, + HttpBody.jsonUnsafe({ ...message.value, validate_only: true }), + ); +}; + +const relayConfiguration = ( + serviceAccount: string, +): RelayConfiguration.RelayConfiguration["Service"] => ({ + // FcmClient reads only the service account; the rest is the relay's own deployment. + relayIssuer: "", + apns: null, + fcmServiceAccount: Redacted.make(serviceAccount), + clerkSecretKey: Redacted.make(""), + clerkPublishableKey: "", + clerkJwtAudience: "", + apnsDeliveryJobSigningSecret: Redacted.make(""), + cloudMintPrivateKey: Redacted.make(""), + cloudMintPublicKey: "", + managedEndpointBaseDomain: undefined, + managedEndpointNamespace: undefined, +}); + +const unconfigured = PushHostFcm.of({ send: null, projectId: null }); + +export const layer = Layer.effect( + PushHostFcm, + Effect.gen(function* () { + const path = yield* Config.String(SERVICE_ACCOUNT_FILE_ENV).pipe(Config.option); + if (Option.isNone(path) || path.value.trim() === "") return unconfigured; + const fs = yield* FileSystem.FileSystem; + const contents = yield* fs.readFileString(path.value.trim()).pipe(Effect.option); + if (Option.isNone(contents)) { + yield* Effect.logError("Push host is off: the FCM service account file could not be read", { + path: path.value, + }); + return unconfigured; + } + const projectId = decodeProjectId(contents.value); + if (Option.isNone(projectId)) { + yield* Effect.logError("Push host is off: the FCM service account file is not a key", { + path: path.value, + }); + return unconfigured; + } + const checkOnly = yield* Config.Boolean(VALIDATE_ONLY_ENV).pipe(Config.withDefault(false)); + const httpClient = yield* HttpClient.HttpClient; + const client = yield* FcmClient.FcmClient.pipe( + Effect.provide( + FcmClient.layer.pipe( + Layer.provide( + FcmAssertionSigner.layer.pipe( + Layer.provide( + Layer.succeed(WebCrypto.WebCrypto, { subtle: globalThis.crypto.subtle }), + ), + ), + ), + Layer.provide( + Layer.succeed( + RelayConfiguration.RelayConfiguration, + relayConfiguration(contents.value), + ), + ), + Layer.provide( + Layer.succeed( + HttpClient.HttpClient, + checkOnly ? HttpClient.mapRequest(httpClient, validateOnly) : httpClient, + ), + ), + ), + ), + ); + yield* Effect.logInfo("Push host is on", { + projectId: projectId.value.project_id, + validateOnly: checkOnly, + }); + return PushHostFcm.of({ send: client.send, projectId: projectId.value.project_id }); + }), +).pipe(Layer.provide(FetchHttpClient.layer)); diff --git a/apps/server/src/fork/push/http.test.ts b/apps/server/src/fork/push/http.test.ts new file mode 100644 index 0000000000000..e106a77f1296b --- /dev/null +++ b/apps/server/src/fork/push/http.test.ts @@ -0,0 +1,140 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { AuthSessionId, EnvironmentId, ThreadId } from "@t3tools/contracts"; +import type { RelayAgentActivityState } from "@t3tools/contracts/relay"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as HttpRouter from "effect/http/HttpRouter"; + +import * as EnvironmentAuth from "../../auth/EnvironmentAuth.ts"; +import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; +import { PushHostFcm } from "./fcm.ts"; +import * as PushHttp from "./http.ts"; +import * as PushHost from "./PushHost.ts"; +import * as PushHostStore from "./PushHostStore.ts"; + +const state: RelayAgentActivityState = { + environmentId: EnvironmentId.make("env-peer"), + threadId: ThreadId.make("thread-1"), + projectTitle: "Project", + threadTitle: "Thread", + phase: "running", + headline: "Working", + modelTitle: "Claude", + updatedAt: "1970-01-01T00:00:00.000Z", + deepLink: "/threads/env-peer/thread-1", +}; + +const session = (peerLink: boolean): EnvironmentAuth.McpClientSession => ({ + sessionId: AuthSessionId.make("session"), + label: "T3 Code · Peer", + access: "read-only", + peerLink, +}); + +const serve = (options: { readonly enabled: boolean }) => + Effect.gen(function* () { + const sessions = new Map([ + ["Bearer peer", session(true)], + ["Bearer agent", session(false)], + ]); + const auth = Layer.mock(EnvironmentAuth.EnvironmentAuth)({ + authenticateMcpClient: (request) => { + const found = sessions.get(request.headers.authorization ?? ""); + return found === undefined + ? Effect.fail(new EnvironmentAuth.ServerAuthMissingCredentialError({})) + : Effect.succeed(found); + }, + }); + const environment = Layer.mock(ServerEnvironment.ServerEnvironment)({ + getEnvironmentId: Effect.succeed(EnvironmentId.make("env-host")), + }); + const fcm = Layer.succeed( + PushHostFcm, + PushHostFcm.of({ + send: options.enabled ? () => Effect.succeed({ unregistered: false }) : null, + projectId: null, + }), + ); + const host = yield* Layer.build( + PushHost.layer.pipe( + Layer.provide(Layer.mergeAll(fcm, PushHostStore.layerMemory(), NodeCrypto.layer)), + ), + ); + const routes = PushHttp.layer.pipe( + Layer.provide(Layer.mergeAll(auth, environment, Layer.succeedContext(host))), + Layer.provide(NodeServices.layer), + ); + const { handler, dispose } = HttpRouter.toWebHandler(routes, { disableLogger: true }); + yield* Effect.addFinalizer(() => Effect.promise(dispose)); + const publish = (input: { + readonly token: string; + readonly environmentId?: string; + readonly body?: unknown; + }) => + Effect.promise(async () => { + const response = await handler( + new Request( + `http://localhost/api/fork/push/v1/environments/${input.environmentId ?? "env-peer"}/threads/thread-1/agent-activity`, + { + method: "POST", + headers: { authorization: input.token, "content-type": "application/json" }, + body: JSON.stringify(input.body ?? { state, proof: "proof" }), + }, + ), + ); + return response.status; + }); + const info = (token: string) => + Effect.promise(async () => { + const response = await handler( + new Request("http://localhost/api/fork/push/v1/info", { + headers: { authorization: token }, + }), + ); + return { status: response.status, body: response.ok ? await response.json() : null }; + }); + return { host: Context.get(host, PushHost.PushHost), publish, info }; + }); + +describe("push host routes", () => { + it.effect("take updates only from a linked environment's session, for its own threads", () => + Effect.gen(function* () { + const { host, publish, info } = yield* serve({ enabled: true }); + assert.strictEqual(yield* publish({ token: "Bearer agent" }), 401); + assert.strictEqual(yield* publish({ token: "Bearer nobody" }), 401); + assert.strictEqual(yield* publish({ token: "Bearer peer", environmentId: "env-host" }), 403); + assert.strictEqual( + yield* publish({ + token: "Bearer peer", + body: { state: { ...state, threadId: "thread-2" }, proof: "proof" }, + }), + 400, + ); + assert.deepStrictEqual((yield* host.status).peers, []); + assert.strictEqual(yield* publish({ token: "Bearer peer" }), 200); + assert.deepStrictEqual( + (yield* host.status).peers.map((peer) => peer.environmentId), + ["env-peer"], + ); + assert.deepStrictEqual(yield* info("Bearer peer"), { + status: 200, + body: { pushHost: true, environmentId: "env-host" }, + }); + assert.strictEqual((yield* info("Bearer agent")).status, 401); + }).pipe(Effect.scoped), + ); + + it.effect("refuse updates when this server holds no key", () => + Effect.gen(function* () { + const { publish, info } = yield* serve({ enabled: false }); + assert.strictEqual(yield* publish({ token: "Bearer peer" }), 404); + assert.deepStrictEqual((yield* info("Bearer peer")).body, { + pushHost: false, + environmentId: "env-host", + }); + }).pipe(Effect.scoped), + ); +}); diff --git a/apps/server/src/fork/push/http.ts b/apps/server/src/fork/push/http.ts new file mode 100644 index 0000000000000..b5622004285cf --- /dev/null +++ b/apps/server/src/fork/push/http.ts @@ -0,0 +1,106 @@ +import { + RelayAgentActivityPublishRequest, + type RelayPublishResponse, +} from "@t3tools/contracts/relay"; +import * as ByteSize from "effect/ByteSize"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/http"; +import * as HttpIncomingMessage from "effect/http/HttpIncomingMessage"; + +import * as EnvironmentAuth from "../../auth/EnvironmentAuth.ts"; +import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; +import { PUSH_ROUTE_PREFIX } from "./ForkPushSender.ts"; +import * as PushHost from "./PushHost.ts"; + +const NO_STORE = { "cache-control": "no-store" }; +const MAX_BODY_BYTES = 64 * 1024; + +const PublishParams = Schema.Struct({ + environmentId: Schema.NonEmptyString, + threadId: Schema.NonEmptyString, +}); +const decodeBody = Schema.decodeUnknownEffect(RelayAgentActivityPublishRequest); + +const refuse = (status: number, message: string) => + HttpServerResponse.jsonUnsafe({ error: message }, { status, headers: NO_STORE }); + +export const layer = Layer.unwrap( + Effect.gen(function* () { + const auth = yield* EnvironmentAuth.EnvironmentAuth; + const host = yield* PushHost.PushHost; + const environment = yield* ServerEnvironment.ServerEnvironment; + + /** A linked environment's session, the credential every push route takes. */ + const peerSession = Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const session = yield* auth.authenticateMcpClient(request).pipe(Effect.option); + return session._tag === "Some" && session.value.peerLink ? session.value : null; + }); + + const infoRoute = HttpRouter.add( + "GET", + `${PUSH_ROUTE_PREFIX}/info`, + Effect.gen(function* () { + if ((yield* peerSession) === null) { + return refuse(401, "A linked environment's session is required."); + } + return HttpServerResponse.jsonUnsafe( + { pushHost: host.enabled, environmentId: yield* environment.getEnvironmentId }, + { headers: NO_STORE }, + ); + }), + ); + + /** Upstream's relay publish endpoint, so a peer's AgentAwarenessRelay posts here unchanged. */ + const publishRoute = HttpRouter.add( + "POST", + `${PUSH_ROUTE_PREFIX}/environments/:environmentId/threads/:threadId/agent-activity`, + Effect.gen(function* () { + // A link's session names only its label, not the environment behind it, so an update is + // not tied to the link that sent it; a link already reads this server's threads anyway. + if ((yield* peerSession) === null) { + return refuse(401, "A linked environment's session is required."); + } + if (!host.enabled) return refuse(404, "This environment is not a push host."); + const params = yield* HttpRouter.schemaPathParams(PublishParams).pipe(Effect.option); + if (params._tag === "None") return refuse(400, "Bad environment or thread."); + const { environmentId, threadId } = params.value; + if (environmentId === (yield* environment.getEnvironmentId)) { + return refuse(403, "The push host publishes its own threads itself."); + } + const request = yield* HttpServerRequest.HttpServerRequest; + const body = yield* request.json.pipe( + Effect.provideService(HttpIncomingMessage.MaxBodySize, ByteSize.bytes(MAX_BODY_BYTES)), + Effect.flatMap(decodeBody), + Effect.option, + ); + if (body._tag === "None") return refuse(400, "Bad agent activity."); + const state = body.value.state; + if ( + state !== null && + (state.environmentId !== environmentId || state.threadId !== threadId) + ) { + return refuse(400, "The state belongs to another thread."); + } + const deviceIds = yield* host.ingest({ environmentId, threadId, state }); + const response: RelayPublishResponse = { + ok: true, + deliveries: deviceIds.map((deviceId) => ({ + deviceId, + kind: "push_notification", + ok: true, + queued: true, + apnsStatus: null, + apnsReason: null, + apnsId: null, + })), + }; + return HttpServerResponse.jsonUnsafe(response, { headers: NO_STORE }); + }), + ); + + return Layer.mergeAll(infoRoute, publishRoute); + }), +); diff --git a/apps/server/src/fork/push/layers.ts b/apps/server/src/fork/push/layers.ts new file mode 100644 index 0000000000000..824da35ef27df --- /dev/null +++ b/apps/server/src/fork/push/layers.ts @@ -0,0 +1,40 @@ +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; + +import * as PeerLinks from "../../peer/PeerLinks.ts"; +import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; +import { forkParked } from "../../serverActivation.ts"; +import * as Fcm from "./fcm.ts"; +import * as ForkPushSender from "./ForkPushSender.ts"; +import * as PushHttp from "./http.ts"; +import * as PushHost from "./PushHost.ts"; +import * as PushHostStore from "./PushHostStore.ts"; + +/** One instance, shared by the routes, the WebSocket handlers and the sender through layer memoization. */ +export const layerPushHost = PushHost.layer.pipe( + Layer.provide(Layer.merge(Fcm.layer, PushHostStore.layer)), +); + +export const layerSender = ForkPushSender.layer.pipe( + Layer.provide(Layer.merge(layerPushHost, PeerLinks.layer)), +); + +/** AgentAwarenessRelay, publishing to the push host when there is one. */ +export const layerAgentAwarenessRelay = Layer.effectDiscard( + Effect.gen(function* () { + const relay = yield* AgentAwarenessRelay.AgentAwarenessRelay; + const sender = yield* ForkPushSender.ForkPushSender; + const changes = yield* sender.subscribe; + // A new route has none of this server's active threads yet. + yield* forkParked( + PubSub.take(changes).pipe(Effect.andThen(relay.requestCatchUp()), Effect.forever), + ); + }), +).pipe( + Layer.provideMerge(AgentAwarenessRelay.layer), + Layer.provide(ForkPushSender.layerTarget), + Layer.provideMerge(layerSender), +); + +export const layerRoutes = PushHttp.layer.pipe(Layer.provide(layerPushHost)); diff --git a/apps/server/src/fork/push/webCrypto.d.ts b/apps/server/src/fork/push/webCrypto.d.ts new file mode 100644 index 0000000000000..3bce5370c84f9 --- /dev/null +++ b/apps/server/src/fork/push/webCrypto.d.ts @@ -0,0 +1,6 @@ +import type * as NodeCrypto from "node:crypto"; + +declare global { + // infra/relay's WebCrypto service names the Workers type; Node's types declare only the value. + interface SubtleCrypto extends NodeCrypto.webcrypto.SubtleCrypto {} +} diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index f9a861cca23fb..c73449254c5a1 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -1,4 +1,9 @@ -import { ORCHESTRATION_V2_WS_METHODS, WS_METHODS, type WsRpcGroup } from "@t3tools/contracts"; +import { + FORK_PUSH_WS_METHODS, + ORCHESTRATION_V2_WS_METHODS, + WS_METHODS, + type WsRpcGroup, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as References from "effect/References"; @@ -11,6 +16,10 @@ type WsRpcMethod = RpcGroup.Rpcs["_tag"]; // Trace queries and dashboards depend on these labels, including deviations from method prefixes. const RPC_AGGREGATES = { + [FORK_PUSH_WS_METHODS.getStatus]: "forkPush", + [FORK_PUSH_WS_METHODS.registerDevice]: "forkPush", + [FORK_PUSH_WS_METHODS.unregisterDevice]: "forkPush", + [FORK_PUSH_WS_METHODS.sendTest]: "forkPush", [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: "orchestrationV2", [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: "orchestration", [ORCHESTRATION_V2_WS_METHODS.getNativeChildWork]: "orchestration", diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 85b9b44f4daf7..aaef2048c9466 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -38,6 +38,7 @@ import { RELAY_URL_SECRET, } from "../cloud/config.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as ForkPushTarget from "../fork/push/ForkPushTarget.ts"; import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; @@ -525,6 +526,40 @@ describe("AgentAwarenessRelay", () => { }), ); + it.effect("publishes to the fork's push host instead of the relay, unless explicitly off", () => + Effect.gen(function* () { + const delivered: Array = []; + const route = yield* Ref.make({ + url: "https://push-host.example.test/api/fork/push", + issuer: "https://push-host.example.test", + environmentCredential: "link-token", + }); + const { relay, secrets, currentShell, publications } = yield* makeTestRelay({ + linked: false, + }).pipe(Effect.provideService(ForkPushTarget.ForkPushTarget, { resolve: Ref.get(route) })); + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.length, 1); + assert.isTrue( + publications[0]?.url.startsWith("https://push-host.example.test/api/fork/push/v1/") ?? + false, + ); + assert.equal(publications[0]?.authorization, "Bearer link-token"); + yield* Ref.set(route, { + url: "fork-push:local", + issuer: "fork-push:local", + environmentCredential: "fork-push:local", + deliver: (input) => Effect.sync(() => void delivered.push(input.state)), + }); + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.length, 1); + assert.equal(delivered[0]?.threadId, THREAD_ID); + yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false")); + yield* Ref.set(currentShell, shell({ title: "Renamed thread" })); + yield* relay.publishThread(THREAD_ID); + assert.equal(delivered.length, 1); + }), + ); + it.effect.each(["completed", "failed"] as const)( "publishes a tombstone when a %s thread is settled through a domain event", (status) => diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index c14558762183a..1d3f4b2b30476 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -41,13 +41,13 @@ import * as HttpApiClient from "effect/http-api/HttpApiClient"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import { - isAgentActivityPublishingEnabledValue, PUBLISH_AGENT_ACTIVITY_SECRET, RELAY_ENVIRONMENT_CREDENTIAL_SECRET, RELAY_ISSUER_SECRET, RELAY_URL_SECRET, } from "../cloud/config.ts"; import { EnvironmentSigningKey } from "../cloud/environmentKeys.ts"; +import * as ForkPush from "../fork/push/ForkPushTarget.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../project/ProjectService.ts"; @@ -413,6 +413,7 @@ export const make = Effect.gen(function* () { // its first alert time through parent renames, relinks and the parent's own requests. const childAlertByThread = new Map(); const subagentRequests = new Map(); + const forkPush = yield* ForkPush.ForkPushTarget; const readSecretString = (name: string) => secrets @@ -424,6 +425,8 @@ export const make = Effect.gen(function* () { ); const readRelayConfig = Effect.gen(function* () { + const forkRoute = yield* forkPush.resolve; + if (forkRoute !== null) return forkRoute; const [url, issuer, environmentCredential] = yield* Effect.all([ readSecretString(RELAY_URL_SECRET), readSecretString(RELAY_ISSUER_SECRET), @@ -434,9 +437,10 @@ export const make = Effect.gen(function* () { : null; }); - const readPublishAgentActivityEnabled = readSecretString(PUBLISH_AGENT_ACTIVITY_SECRET).pipe( - Effect.map(isAgentActivityPublishingEnabledValue), - ); + const readPublishAgentActivityEnabled = Effect.all([ + readSecretString(PUBLISH_AGENT_ACTIVITY_SECRET), + forkPush.resolve, + ]).pipe(Effect.map(([toggle, forkRoute]) => ForkPush.publishEnabled(toggle, forkRoute))); const makeRelayClient = (relayConfig: { readonly url: string; @@ -511,6 +515,9 @@ export const make = Effect.gen(function* () { readonly state: RelayAgentActivityState | null; readonly reason: string; }) { + if ("deliver" in relayConfig && relayConfig.deliver !== undefined) { + return yield* relayConfig.deliver({ environmentId, threadId, state: input.state }); + } if (cachedRelayClient === undefined) { cachedRelayClient = yield* makeRelayClient(relayConfig); } @@ -822,9 +829,9 @@ export const make = Effect.gen(function* () { }); const publishActiveThreadsUnsafe = Effect.gen(function* () { - const relayUrl = yield* readSecretString(RELAY_URL_SECRET).pipe( - Effect.orElseSucceed(() => null), - ); + const relayUrl = + (yield* forkPush.resolve)?.url ?? + (yield* readSecretString(RELAY_URL_SECRET).pipe(Effect.orElseSucceed(() => null))); if (!relayUrl) { yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable"); return "unlinked" as const; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 8587e16d8e947..780aab6b432da 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -100,7 +100,7 @@ import * as EnvironmentTheme from "./environmentTheme.ts"; 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 ForkPush from "./fork/push/layers.ts"; import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as ManagedLimits from "./policy/managedLimits.ts"; @@ -644,7 +644,7 @@ const layerServerRestart = Layer.effect( ); const layerRuntimeCoreDependenciesBase = Layer.mergeAll( - AgentAwarenessRelay.layer, + ForkPush.layerAgentAwarenessRelay, HeldHooksWaker.layer, layerThreadSettlementWorker, layerIdleCompactionWorker, @@ -808,6 +808,7 @@ const layerMakeRoutes = Layer.mergeAll( ServerHttp.layerStaticAndDevRoute, Ws.layer, speechStreamRouteLayer, + ForkPush.layerRoutes, ), McpHttpServer.layer.pipe( Layer.provide(ProviderAdapterRegistry.layerFromProviderInstanceRegistry), @@ -825,6 +826,8 @@ const layerMakeRoutes = Layer.mergeAll( Layer.provide(RemoteDelegation.layer.pipe(Layer.provide(ProjectionStoreV2.layer))), // Links agents ask for: shown in a thread from /mcp, answered over WebSocket. Layer.provide(PeerLinkRequests.layer), + // The fork's push host, for its routes and WebSocket handlers. + Layer.provide(Layer.merge(ForkPush.layerPushHost, ForkPush.layerSender)), // Share the service so HTTP/WS clients single-flight reads and invalidate each other's cached patches. Layer.provide(layerPullRequestService), // The same activity instance RunExecutionService reports to, by layer memoization. diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1992247115620..dea04732a034a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3,6 +3,7 @@ import * as Context from "effect/Context"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as PeerLinkRequests from "./peer/PeerLinkRequests.ts"; +import * as ForkPushRpc from "./fork/push/ForkPushRpc.ts"; import * as PeerLinks from "./peer/PeerLinks.ts"; import * as DateTime from "effect/DateTime"; @@ -1523,6 +1524,7 @@ export const layerWsRpc = ( const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; + const forkPushHandlers = yield* ForkPushRpc.ForkPushHandlers; const rpcClientIds = yield* Ref.make(new Set()); yield* Effect.addFinalizer(() => Ref.get(rpcClientIds).pipe( @@ -1919,6 +1921,7 @@ export const layerWsRpc = ( }); const handlers = ServerWsRpcGroup.of({ + ...forkPushHandlers, [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command) => Effect.annotateCurrentSpan({ "orchestration_v2.command_id": command.commandId, @@ -3454,6 +3457,7 @@ export const layer = Layer.unwrap( const reviewGuides = yield* PullRequestReviewGuideService.PullRequestReviewGuideService; const peerLinks = yield* PeerLinks.PeerLinks; const peerLinkRequests = yield* PeerLinkRequests.PeerLinkRequests; + const forkPushHandlers = yield* ForkPushRpc.makeHandlers; const sql = yield* SqlClient.SqlClient; return HttpRouter.add( "GET", @@ -3543,6 +3547,7 @@ export const layer = Layer.unwrap( ), Layer.provide(Layer.succeed(PeerLinks.PeerLinks, peerLinks)), Layer.provide(Layer.succeed(PeerLinkRequests.PeerLinkRequests, peerLinkRequests)), + Layer.provide(Layer.succeed(ForkPushRpc.ForkPushHandlers, forkPushHandlers)), Layer.provide( SourceControlDiscovery.layer.pipe( Layer.provide( diff --git a/apps/server/vite.config.ts b/apps/server/vite.config.ts index 02fe582953abd..0e798c49e81c0 100644 --- a/apps/server/vite.config.ts +++ b/apps/server/vite.config.ts @@ -99,6 +99,10 @@ export default mergeConfig( clean: true, // SEA can only resolve built-ins; keep dynamic imports inside each emitted module. ...(packExecutable ? { outputOptions: { codeSplitting: false } } : {}), + // Fork: the push host bundles infra/relay sources, whose packages release installs skip. + inputOptions: { + resolve: { modules: ["node_modules", NodePath.join(import.meta.dirname, "node_modules")] }, + }, ...(packMode === "1" ? { exe: { diff --git a/docs/user/mobile-notifications.md b/docs/user/mobile-notifications.md index 992efd5b7b29a..372fdaf4a4ad1 100644 --- a/docs/user/mobile-notifications.md +++ b/docs/user/mobile-notifications.md @@ -13,3 +13,5 @@ Ordinary alerts stay quiet while the mobile app is in the foreground. Ongoing ac Android notifications require Android 7.0 or newer and Google Play services. Android 16 and newer can promote ongoing activity to a Live Update, subject to system settings and device support. Other devices show a regular ongoing notification. If Live Updates don't appear on Android 16 or newer, check **Live Update Settings** in T3 Code's **Settings → Notifications**. Android 7's battery-saving modes can delay removal of expired cards. Notification permission and Android notification channels are controlled in system Settings. Background delivery requires T3 Connect; a direct or Tailscale connection alone does not enable push notifications. The mobile app does not need to maintain a connection to your environment. Force-stopping the Android app in system Settings prevents push delivery until you open it again. + +This fork sends Android notifications through a push host, one of your own servers, instead of T3 Connect. See [Push host](./push-host.md). diff --git a/docs/user/push-host.md b/docs/user/push-host.md new file mode 100644 index 0000000000000..38d1c60076005 --- /dev/null +++ b/docs/user/push-host.md @@ -0,0 +1,27 @@ +# Push host + +This fork's Android app uses its own Firebase project, so T3 Connect's relay cannot send it notifications. Instead, one of your own T3 Code servers, the push host, sends them for itself and for the environments linked to it. Nothing passes through T3 Connect's relay. + +## Set up the push host + +1. Create a service account for your Firebase project with permission to send Cloud Messaging, and download its JSON key. +2. Pick an always-on server as the push host. Start it with `T3CODE_FCM_SERVICE_ACCOUNT_FILE` set to the key's path. Its log then says `Push host is on`. Anyone who can read the key can send notifications to your app, so keep it readable only by the server's user. +3. Open the Android app, connect to that environment, and allow notifications. **Settings → Notifications → Push host** shows the push host and says when this phone gets its notifications. + +**Send a test notification** checks the whole path to this phone. **Stop notifications on this phone** unregisters it from the push host until you turn notifications back on there. + +## Add your other environments + +An environment sends its agent activity to the push host through a link to it. On that environment, turn on **Settings → Labs → Linked environments**, turn it on for the push host too, then link the push host from **Settings → Environments → Linked environments**. Read-only access is enough. Pick the push host from the list rather than typing an address: the list includes its T3 Connect address, which keeps working when the environment is away from your network, such as a server on your phone running on mobile data. + +Each environment finds the push host among its links within a few minutes. The push host's section in **Settings → Notifications** shows each connected environment, whether it sends to the push host, and when its link expires. + +Links last 30 days. When one expires, that environment stops sending until you link it again with a new pairing code. Turning **Linked environments** off later does not stop sending over an existing link. + +## Publish agent activity + +An environment sends to the push host unless **Publish agent activity** is turned off there, in its T3 Connect settings or with `t3 connect publish --disable`. An environment that has never been linked to T3 Connect sends. The push host's section shows when this setting is off. + +## Removing the push host + +Restart the push host without `T3CODE_FCM_SERVICE_ACCOUNT_FILE` to stop sending. Your other environments then go back to publishing through T3 Connect if they are linked to it. The push host keeps its list of registered phones, so adding the key back resumes notifications. diff --git a/infra/relay/src/agentActivity/forkAndroidAlerts.test.ts b/infra/relay/src/agentActivity/forkAndroidAlerts.test.ts new file mode 100644 index 0000000000000..5f3307df40267 --- /dev/null +++ b/infra/relay/src/agentActivity/forkAndroidAlerts.test.ts @@ -0,0 +1,80 @@ +import { describe, expect, it } from "@effect/vitest"; +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import type { RelayAgentActivityState } from "@t3tools/contracts/relay"; + +import * as Upstream from "./FcmDeliveries.ts"; +import { makeAggregateState } from "./agentActivityAggregate.ts"; +import { + androidAlertForAggregate, + androidAlertForState, +} from "../../../../apps/server/src/fork/push/androidAlerts.ts"; + +const base: RelayAgentActivityState = { + environmentId: EnvironmentId.make("env"), + threadId: ThreadId.make("thread"), + projectTitle: "Project", + threadTitle: "Fix notifications", + phase: "running", + headline: "Working", + modelTitle: "Claude", + updatedAt: "2026-01-01T00:00:00.000Z", + deepLink: "/threads/env/thread", +}; +const other: RelayAgentActivityState = { + ...base, + threadId: ThreadId.make("other"), + threadTitle: "Other", + deepLink: "/threads/env/other", +}; +const nowMs = Date.parse("2026-01-01T00:00:30.000Z"); +const preferences = { + notificationsEnabled: true, + liveActivitiesEnabled: true, + notifyOnApproval: true, + notifyOnInput: true, + notifyOnCompletion: true, + notifyOnFailure: true, +}; +const aggregate = (states: ReadonlyArray) => + makeAggregateState({ activeStates: states, terminalState: null, nowMs })!; + +// The copies in androidAlerts.ts must match upstream's relay, which the phone's native code follows. +describe("androidAlerts", () => { + it.each([ + ["waiting_for_approval"], + ["waiting_for_input"], + ["completed"], + ["failed"], + ["running"], + ] as const)("matches upstream for a %s state", (phase) => { + const state = { ...base, phase }; + expect(androidAlertForState(state, preferences, nowMs)).toEqual( + Upstream.androidAlertForState(state, preferences, nowMs), + ); + }); + + it.each([ + [ + "one thread starts waiting", + [base, other], + [{ ...base, phase: "waiting_for_input" as const }, other], + ], + [ + "two threads finish", + [base, other], + [ + { ...base, phase: "completed" as const }, + { ...other, phase: "failed" as const }, + ], + ], + ["nothing changes", [base], [base]], + ])("matches upstream when %s", (_name, previous, next) => { + const input = { + previousAggregate: aggregate(previous), + nextAggregate: aggregate(next), + preferences, + nowMs, + }; + expect(androidAlertForAggregate(input)).toEqual(Upstream.androidAlertForAggregate(input)); + }); +}); diff --git a/packages/contracts/src/clientRpcPermissions.ts b/packages/contracts/src/clientRpcPermissions.ts index 8b4a7d48d279c..8d29e2f03d95f 100644 --- a/packages/contracts/src/clientRpcPermissions.ts +++ b/packages/contracts/src/clientRpcPermissions.ts @@ -12,11 +12,13 @@ import { AuthTerminalOperateScope, type AuthEnvironmentScope, } from "./auth.ts"; +import { FORK_PUSH_RPC_SCOPES } from "./forkPush.ts"; import { ORCHESTRATION_V2_WS_METHODS } from "./orchestrationV2.ts"; import { WS_METHODS } from "./rpc.ts"; /** Incremental client enforcement; the server still authorizes every request. */ export const CLIENT_GUARDED_RPC_SCOPES = { + ...FORK_PUSH_RPC_SCOPES, [WS_METHODS.providerAuthRemoveMethod]: AuthProvidersManageScope, [WS_METHODS.previewClose]: AuthPreviewOperateScope, // Stopping a server signals its processes, as closing a terminal does. diff --git a/packages/contracts/src/forkPush.ts b/packages/contracts/src/forkPush.ts new file mode 100644 index 0000000000000..d29099aad9174 --- /dev/null +++ b/packages/contracts/src/forkPush.ts @@ -0,0 +1,116 @@ +import * as Schema from "effect/Schema"; +import * as Rpc from "effect/rpc/Rpc"; + +import { + AuthOrchestrationReadScope, + AuthSettingsWriteScope, + EnvironmentAuthorizationError, +} from "./auth.ts"; +import { TrimmedNonEmptyString } from "./baseSchemas.ts"; + +/** + * This fork's push host: one of the owner's servers holds an FCM key and sends Android + * notifications for itself and the environments linked to it. + */ +export const FORK_PUSH_WS_METHODS = { + getStatus: "forkPush.getStatus", + registerDevice: "forkPush.registerDevice", + unregisterDevice: "forkPush.unregisterDevice", + sendTest: "forkPush.sendTest", +} as const; + +/** The same fields as relay.ts's RelayAgentAwarenessPreferences, which stays out of client bundles. */ +const ForkPushPreferences = Schema.Struct({ + liveActivitiesEnabled: Schema.Boolean, + notificationsEnabled: Schema.Boolean, + notifyOnApproval: Schema.Boolean, + notifyOnInput: Schema.Boolean, + notifyOnCompletion: Schema.Boolean, + notifyOnFailure: Schema.Boolean, +}); + +export const ForkPushDeviceRegistration = Schema.Struct({ + deviceId: TrimmedNonEmptyString, + /** Echoed in every message; the phone's native handler drops messages for another identity. */ + userId: TrimmedNonEmptyString, + token: TrimmedNonEmptyString, + packageName: Schema.NullOr(TrimmedNonEmptyString), + label: Schema.String, + preferences: ForkPushPreferences, +}); +export type ForkPushDeviceRegistration = typeof ForkPushDeviceRegistration.Type; + +export const ForkPushStatus = Schema.Struct({ + /** `host` holds the key; `sender` sends to a linked push host; `none` publishes as upstream does. */ + role: Schema.Literals(["host", "sender", "none"]), + /** The push host this server sends to, or last sent to. */ + host: Schema.NullOr( + Schema.Struct({ + environmentId: Schema.String, + label: Schema.String, + url: Schema.String, + linkExpiresAt: Schema.String, + }), + ), + checkedAt: Schema.NullOr(Schema.String), + error: Schema.NullOr(Schema.String), + /** Upstream's "Publish agent activity" is explicitly off here, which the push host respects. */ + publishingOff: Schema.Boolean, + devices: Schema.Array( + Schema.Struct({ + deviceId: Schema.String, + label: Schema.String, + registeredAt: Schema.String, + lastSentAt: Schema.NullOr(Schema.String), + lastError: Schema.NullOr(Schema.String), + }), + ), + peers: Schema.Array(Schema.Struct({ environmentId: Schema.String, lastHeardAt: Schema.String })), +}); +export type ForkPushStatus = typeof ForkPushStatus.Type; + +export class ForkPushError extends Schema.TaggedError()("ForkPushError", { + message: Schema.String, +}) {} + +const ForkPushGetStatusRpc = Rpc.make(FORK_PUSH_WS_METHODS.getStatus, { + payload: Schema.Struct({}), + success: ForkPushStatus, + error: EnvironmentAuthorizationError, +}); + +const ForkPushRegisterDeviceRpc = Rpc.make(FORK_PUSH_WS_METHODS.registerDevice, { + payload: ForkPushDeviceRegistration, + success: Schema.Struct({}), + error: Schema.Union([ForkPushError, EnvironmentAuthorizationError]), +}); + +const ForkPushUnregisterDeviceRpc = Rpc.make(FORK_PUSH_WS_METHODS.unregisterDevice, { + payload: Schema.Struct({ deviceId: TrimmedNonEmptyString }), + success: Schema.Struct({ removed: Schema.Boolean }), + error: Schema.Union([ForkPushError, EnvironmentAuthorizationError]), +}); + +const ForkPushSendTestRpc = Rpc.make(FORK_PUSH_WS_METHODS.sendTest, { + payload: Schema.Struct({ deviceId: TrimmedNonEmptyString }), + success: Schema.Struct({}), + error: Schema.Union([ForkPushError, EnvironmentAuthorizationError]), +}); + +export const FORK_PUSH_RPCS = [ + ForkPushGetStatusRpc, + ForkPushRegisterDeviceRpc, + ForkPushUnregisterDeviceRpc, + ForkPushSendTestRpc, +] as const; + +/** Reading status is a query, which clients may not guard; the server still requires this. */ +export const FORK_PUSH_QUERY_SCOPES = { + [FORK_PUSH_WS_METHODS.getStatus]: AuthOrchestrationReadScope, +} as const; + +export const FORK_PUSH_RPC_SCOPES = { + [FORK_PUSH_WS_METHODS.sendTest]: AuthOrchestrationReadScope, + [FORK_PUSH_WS_METHODS.registerDevice]: AuthSettingsWriteScope, + [FORK_PUSH_WS_METHODS.unregisterDevice]: AuthSettingsWriteScope, +} as const; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 8dd47df33a21f..e48bdca36e0c7 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -5,6 +5,7 @@ export * from "./composerContextClipboard.ts"; export * from "./background.ts"; export * from "./acpRegistry.ts"; export * from "./auth.ts"; +export * from "./forkPush.ts"; export * from "./environment.ts"; export * from "./environmentHttp.ts"; export * from "./relayClient.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 0f6af19188e18..6f0a435752add 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -6,6 +6,7 @@ import { OrchestrationV2SearchThreadsResult, } from "./orchestrationV2.ts"; import { OrchestrationDispatchCommandError } from "./orchestrationDispatch.ts"; +import { FORK_PUSH_RPCS } from "./forkPush.ts"; import { McpAppCallToolInput, McpAppCallToolResult, @@ -2067,6 +2068,7 @@ export class RpcScopeAuthorization extends RpcMiddleware.Service