feat: a fork server can be the Android push host for itself and linked environments

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`.

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() {
     <>
       <SplashScreenCoordinator />
       <SubscriptionUsageCoordinator />
+      <ForkPushRegistrar />
       <GestureHandlerRootView className="flex-1">
         <KeyboardProvider statusBarTranslucent>
           <SafeAreaProvider>

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<string, string>):
 }
 
 export function clearAndroidAgentNotifications(): void {
+  if (forkOwnsAndroidAgentNotifications()) return;
   native?.clear?.();
 }
 

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;
+}

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 (
+    <SettingsSection
+      title="Push host"
+      footer="The environment started with an FCM key sends this phone's notifications. Others reach it through a link to it in Settings → Environments → Linked environments; links last 30 days."
+    >
+      {host === undefined ? (
+        <SettingsRow
+          icon="slash.circle"
+          label="No push host"
+          value="None of your connected environments holds an FCM key."
+          valuePosition="below"
+        />
+      ) : (
+        <HostRows
+          environmentId={host.environmentId}
+          label={host.environmentLabel}
+          registration={registration}
+        />
+      )}
+      {environments
+        .filter((environment) => environment.environmentId !== hostEnvironmentId)
+        .map((environment) => (
+          <SenderRow
+            key={environment.environmentId}
+            environmentId={environment.environmentId}
+            label={environment.environmentLabel}
+            hostEnvironmentId={hostEnvironmentId}
+            hostLabel={host?.environmentLabel ?? null}
+          />
+        ))}
+    </SettingsSection>
+  );
+}
+
+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 (
+    <>
+      <SettingsRow
+        icon="bell.badge"
+        label={props.label}
+        value={`${registrationText(registration)}${publishingOff ? ` ${PUBLISHING_OFF}` : ""}`}
+        valuePosition="below"
+        {...(registration.kind === "needs-permission" || registration.kind === "failed"
+          ? { onPress: () => void turnOn() }
+          : {})}
+      />
+      {registration.kind === "registered" ? (
+        <SettingsRow
+          icon="bolt.circle"
+          label="Send a test notification"
+          onPress={() =>
+            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" ? (
+        <SettingsRow
+          icon="bell.badge"
+          label="Turn on notifications on this phone"
+          disabled={!canRegister}
+          onPress={() => void turnOn()}
+        />
+      ) : (
+        <SettingsRow
+          icon="slash.circle"
+          label="Stop notifications on this phone"
+          disabled={!canRegister}
+          onPress={() => 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 (
+    <SettingsRow
+      icon="server.rack"
+      label={props.label}
+      value={senderText(
+        status.data,
+        status.failure !== null,
+        props.hostEnvironmentId,
+        props.hostLabel,
+      )}
+      valuePosition="below"
+    />
+  );
+}

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,
+  }),
+};