feat(subscriptions): live transfer of subscriptions from non-deleted accounts

.env.example

@@ -186,10 +186,17 @@ POSTHOG_API_HOST=
 POSTHOG_PERSONAL_API_KEY=
 POSTHOG_PROJECT_ID=
 
-# --- Subscription restoration ---
-# Claims of deleted accounts' Apple subscriptions.
+# --- Subscription claim (reclaim) launch flags ---
+# Tombstone restoration tier (claims of deleted accounts' subscriptions).
 SUBSCRIPTION_CLAIM_TOMBSTONE_ENABLED=true
-# Live ownership transfers and Google claim proof are deferred to a follow-up.
+# Live bearer-transfer tier. OFF until security sign-off.
+SUBSCRIPTION_CLAIM_LIVE_TRANSFER_ENABLED=false
+# Google claim/restoration surface. OFF (Apple-only product today); verify /
+# RTDN ingest and the Google money accounting stay on regardless.
+SUBSCRIPTION_CLAIM_GOOGLE_ENABLED=false
+# Contest window (hours, whole numbers only) for live-tier claims. 0 =
+# instant transfer, which requires explicit security acceptance.
+CLAIM_CONTEST_WINDOW_HOURS=72
 
 # --- Payments / Credits ---
 # REQUIRED — All five PAYMENTS_* knobs below are hard-required. Backend

prisma/migrations/20260722120000_add_deletion_final_balance/migration.sql

@@ -0,0 +1,3 @@
+-- Display-only snapshot of the wallet balance at deletion time (pre-escrow).
+-- Nullable: records written before this field existed stay null.
+ALTER TABLE "DeletionRecord" ADD COLUMN "finalBalanceCredits" BIGINT;

prisma/schema.prisma

@@ -345,6 +345,12 @@ model DeletionRecord {
   /// Plain string column, validated in app code (see the AuthMethod.type
   /// comment for why Postgres enums are avoided).
   status      String    @default("purging")
+  /// Display-only snapshot of the wallet balance at deletion time, read
+  /// inside the teardown transaction BEFORE any escrow debit or wallet
+  /// teardown (what the user had). Null on records written before the
+  /// field existed; 0 when the account never had a wallet row
+  /// (getBalance semantics).
+  finalBalanceCredits BigInt?
   requestedAt DateTime  @default(now())
   completedAt DateTime?
   /// When the retention job may delete this record (set once the drain

src/accounts/auth-activity.ts

@@ -0,0 +1,56 @@
+import { prisma } from "@/utils/prisma";
+
+/**
+ * Record "any authenticated act" on the account. The live-transfer contest
+ * window uses lastAuthAt strictly as a veto - an old owner who touches any
+ * authenticated route during the window cancels the pending transfer - so
+ * the stamp must be reliable exactly when it matters:
+ *
+ * - The write is awaited before the request proceeds (a fire-and-forget
+ *   stamp could land after settlement locked and read the row).
+ * - The timestamp is database now(), the same clock that stamps the pending
+ *   row's createdAt, so app/DB clock skew can never make a later act
+ *   compare as older.
+ * - Throttling applies only while the account has no pending outgoing
+ *   transfer. With one pending, every authenticated act is stamped
+ *   unconditionally - a suppressed write inside the throttle window would
+ *   otherwise leave lastAuthAt before the pending row and the transfer
+ *   would settle despite real victim activity.
+ * - Failures propagate (fail closed): a failed stamp must never silently
+ *   cost a veto. Callers fail the request with a 5xx so the client retries;
+ *   the alternative - swallowing the error and proceeding - lets a
+ *   transient DB blip during a contest window hand the subscription to the
+ *   claimant despite real owner activity. An UPDATE matching zero rows
+ *   (account deleted mid-request) is not a failure: there is no veto left
+ *   to preserve.
+ */
+const STAMP_INTERVAL_MS = 5 * 60 * 1000;
+
+let stampFailureForTests: Error | null = null;
+
+/** Test seam: make stamp writes fail with the given error (null clears). */
+export const __setAuthActivityStampFailureForTests = (
+  err: Error | null,
+): void => {
+  stampFailureForTests = err;
+};
+
+export const stampAuthActivity = async (
+  accountId: string,
+  knownLastAuthAt: Date | null,
+): Promise<void> => {
+  const withinThrottle =
+    knownLastAuthAt !== null &&
+    Date.now() - knownLastAuthAt.getTime() < STAMP_INTERVAL_MS;
+  if (withinThrottle) {
+    const pending = await prisma.subscriptionTransfer.findFirst({
+      where: { status: "pending", fromAccountId: accountId },
+      select: { id: true },
+    });
+    if (!pending) return;
+  }
+  if (stampFailureForTests) throw stampFailureForTests;
+  await prisma.$executeRaw`
+    UPDATE "Account" SET "lastAuthAt" = now() WHERE id = ${accountId}::uuid
+  `;
+};

src/accounts/deletion/executors.ts

@@ -1,4 +1,5 @@
 import { DeleteObjectCommand, S3Client } from "@aws-sdk/client-s3";
+import { Code, ConnectError } from "@connectrpc/connect";
 import { z } from "zod";
 import type { DeletionTaskKind } from "@/accounts/deletion/service";
 import { createComposioService } from "@/api/v2/connections/composio.service";
@@ -138,6 +139,19 @@ export const __setDeletionNotificationClientForTests = (
 };
 const POSTHOG_FETCH_TIMEOUT_MS = 10_000;
 
+/**
+ * NotFound and Unimplemented from the notification server both mean the
+ * installation is already gone: NotFound is the explicit answer, and the
+ * Connect protocol maps a bare HTTP 404 (route no longer served) to
+ * `unimplemented`. Either way there is nothing left to purge, so the task
+ * must complete instead of retrying forever. Genuinely transient failures
+ * (unavailable, 5xx, timeouts, connection refused) carry other codes and
+ * keep their retry semantics.
+ */
+const isInstallationAlreadyAbsent = (err: unknown): err is ConnectError =>
+  err instanceof ConnectError &&
+  (err.code === Code.NotFound || err.code === Code.Unimplemented);
+
 /** Remove one notification-server installation (per ClientIdentifier). */
 const executeNotificationInstallation: DeletionExecutor = async (payload) => {
   const parsed = installationPayloadSchema.parse(payload);
@@ -147,11 +161,20 @@ const executeNotificationInstallation: DeletionExecutor = async (payload) => {
   const result = await withInstallationMutationFence({
     installationId: parsed.installationId,
     expectation: { state: "absent" },
-    mutate: () =>
-      notificationClient.deleteInstallation(
-        { installationId: parsed.installationId },
-        notificationMutationCallOptions(),
-      ),
+    mutate: async () => {
+      try {
+        await notificationClient.deleteInstallation(
+          { installationId: parsed.installationId },
+          notificationMutationCallOptions(),
+        );
+      } catch (err) {
+        if (!isInstallationAlreadyAbsent(err)) throw err;
+        logger.info(
+          { installationId: parsed.installationId, code: Code[err.code] },
+          "deletion.notification_installation.already_absent",
+        );
+      }
+    },
   });
   if (!result.applied) {
     logger.info(

src/accounts/deletion/outbox.ts

@@ -1,5 +1,6 @@
 import { getDeletionExecutor } from "@/accounts/deletion/executors";
 import { PURGE_WINDOW_HOURS } from "@/accounts/deletion/service";
+import { settlePendingTransfers } from "@/subscriptions/claim";
 import { runReclaimReconciliationSweep } from "@/subscriptions/reconciliation";
 import logger from "@/utils/logger";
 import { prisma } from "@/utils/prisma";
@@ -346,7 +347,13 @@ export const runDeletionOutboxSweep = async (): Promise<void> => {
     logger.error({ err }, "deletion.outbox.expiry_pass_failed");
   }
   try {
-    // Reclaim reconciliation (quarantine drain + lineage drift)
+    // Live-tier claim contest windows settle on the same tick.
+    await settlePendingTransfers();
+  } catch (err) {
+    logger.error({ err }, "deletion.outbox.pending_transfer_pass_failed");
+  }
+  try {
+    // Reclaim reconciliation (quarantine drain + post-transfer drift)
     // rides the same tick, self-throttled: it makes provider calls, so it
     // runs at most once per interval rather than every minute.
     if (

src/accounts/deletion/service.ts

@@ -227,6 +227,17 @@ const runDeleteAccountTransaction = async (args: {
         select: { inputs: true },
       });
 
+      // Display-only balance snapshot for the DeletionRecord: what the wallet
+      // held BEFORE the escrow debits and the wallet teardown below. A plain
+      // read (no ledger mutation) — stable because the Account FOR UPDATE
+      // lock above fences every concurrent ledger writer. Missing wallet row
+      // reads as 0, matching getBalance semantics.
+      const walletRow = await tx.userCredits.findUnique({
+        where: { accountId },
+        select: { balance: true },
+      });
+      const finalBalanceCredits = walletRow?.balance ?? 0n;
+
       // Money bookkeeping before the wallet goes: escrow the conservative
       // remainder of each held custody period (the tombstone snapshot,
       // released to a future claimant), journal the move, and flip the
@@ -331,7 +342,12 @@ const runDeleteAccountTransaction = async (args: {
       const record = await tx.deletionRecord.upsert({
         where: { operationId },
         update: {},
-        create: { operationId, accountRef, status: "purging" },
+        create: {
+          operationId,
+          accountRef,
+          status: "purging",
+          finalBalanceCredits,
+        },
       });
 
       const tasks: Prisma.DeletionTaskCreateManyInput[] = [];