diff --git a/src/backend/clients/events/EventsWorkerInvokerClient.ts b/src/backend/clients/events/EventsWorkerInvokerClient.ts
new file mode 100644
index 000000000..4b4aaba0d
--- /dev/null
+++ b/src/backend/clients/events/EventsWorkerInvokerClient.ts
@@ -0,0 +1,197 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+// Claims the global dispatcher slot for Node's own fetch before npm `undici`
+// can — see the guard's own comment.
+import '../../util/nodeFetchDispatcherGuard.js';
+
+import { Agent, fetch as undiciFetch } from 'undici';
+import { PuterClient } from '../types.js';
+
+/**
+ * The call that leaves the platform and runs an app's own code.
+ *
+ * The protocol is fixed and lives here rather than in the events service
+ * because it is a wire format, not a delivery decision: one POST, one header,
+ * and a status code that says whether the handler took the delivery, refused
+ * it, or could not answer.
+ *
+ *     POST <events worker>/__events/invoke
+ *     puter-auth: <subscriber-scoped access token>
+ *     { handler, event, ctx }
+ *
+ * The token is minted above this layer and passed in: what a delivery is
+ * allowed to do is an authorization question, and this only carries the
+ * answer.
+ */
+
+/** The one route an events worker reserves for the platform. */
+export const EVENTS_INVOKE_PATH = '/__events/invoke';
+
+/** How long a handler has to answer before the attempt is abandoned. */
+export const EVENTS_INVOKE_TIMEOUT_MS = 30_000;
+
+/** Connections held open per worker host, so a busy app is not re-handshaking. */
+const INVOKE_CONNECTIONS = 64;
+const INVOKE_KEEP_ALIVE_MS = 30_000;
+
+/**
+ * Where an app's events worker can be reached.
+ *
+ * Building and deploying that worker is a separate concern, so this is a seam:
+ * the default resolves nothing, and the deployment that owns worker addressing
+ * registers an implementation. A `null` answer means the app has no events
+ * worker to invoke.
+ */
+export interface EventsWorkerResolver {
+    resolveInvokeUrl(appUid: string): Promise<string | null>;
+}
+
+/** The stand-in until something can address an app's events worker. */
+export class UnresolvedEventsWorkerResolver implements EventsWorkerResolver {
+    resolveInvokeUrl(): Promise<string | null> {
+        return Promise.resolve(null);
+    }
+}
+
+/**
+ * What the call did.
+ *
+ * - `settled` — the handler took the delivery (2xx).
+ * - `terminal` — it refused it (4xx). Retrying sends the same body to the same
+ *   code, so it is not retried.
+ * - `retriable` — it could not answer: 5xx, 429, a timeout, a transport failure,
+ *   or no worker to address at all.
+ */
+export type WorkerInvokeOutcome = 'settled' | 'terminal' | 'retriable';
+
+export interface WorkerInvokeRequest {
+    appUid: string;
+    handler: string;
+    /** Subscriber-scoped access token, sent in `puter-auth`. */
+    token: string;
+    event: unknown;
+    ctx: unknown;
+}
+
+export interface WorkerInvokeResult {
+    outcome: WorkerInvokeOutcome;
+    /** The status the handler answered with, or `null` when it never did. */
+    status: number | null;
+    /** Why it could not answer, when nothing did. */
+    error?: string;
+}
+
+const RESOLVER_UNAVAILABLE = 'no events worker is deployed for this app';
+
+export class EventsWorkerInvokerClient extends PuterClient {
+    /** Replaced by whatever owns worker addressing on this deployment. */
+    #resolver: EventsWorkerResolver = new UnresolvedEventsWorkerResolver();
+    #agent: Agent | null = null;
+
+    setResolver(resolver: EventsWorkerResolver): void {
+        this.#resolver = resolver;
+    }
+
+    override onServerShutdown(): void {
+        void this.#agent?.close().catch(() => {});
+        this.#agent = null;
+    }
+
+    async invoke(request: WorkerInvokeRequest): Promise<WorkerInvokeResult> {
+        let url: string | null = null;
+        try {
+            url = await this.#resolver.resolveInvokeUrl(request.appUid);
+        } catch (err) {
+            return {
+                outcome: 'retriable',
+                status: null,
+                error: err instanceof Error ? err.message : String(err),
+            };
+        }
+        // Nothing to call yet. Retriable rather than terminal: the address is
+        // the platform's to provide, and an app whose worker is not there has
+        // done nothing wrong — the consecutive-failure rule is what stops this
+        // from retrying forever.
+        if (!url)
+            return {
+                outcome: 'retriable',
+                status: null,
+                error: RESOLVER_UNAVAILABLE,
+            };
+
+        try {
+            const response = await undiciFetch(
+                new URL(EVENTS_INVOKE_PATH, url).toString(),
+                {
+                    method: 'POST',
+                    headers: {
+                        'content-type': 'application/json',
+                        'puter-auth': request.token,
+                    },
+                    body: JSON.stringify({
+                        handler: request.handler,
+                        event: request.event,
+                        ctx: request.ctx,
+                    }),
+                    signal: AbortSignal.timeout(this.#timeoutMs()),
+                    dispatcher: this.#dispatcher(),
+                },
+            );
+            // The body is not read: a handler answers with a status, and
+            // whatever else it writes is its own business.
+            void response.body?.cancel().catch(() => {});
+            return {
+                outcome: outcomeForStatus(response.status),
+                status: response.status,
+            };
+        } catch (err) {
+            return {
+                outcome: 'retriable',
+                status: null,
+                error: err instanceof Error ? err.message : String(err),
+            };
+        }
+    }
+
+    #timeoutMs(): number {
+        const configured = this.config.events?.invokeTimeoutMs;
+        return typeof configured === 'number' && configured > 0
+            ? configured
+            : EVENTS_INVOKE_TIMEOUT_MS;
+    }
+
+    /** One pool per process, built on first use and closed with the server. */
+    #dispatcher(): Agent {
+        this.#agent ??= new Agent({
+            connections: INVOKE_CONNECTIONS,
+            keepAliveTimeout: INVOKE_KEEP_ALIVE_MS,
+            keepAliveMaxTimeout: INVOKE_KEEP_ALIVE_MS,
+        });
+        return this.#agent;
+    }
+}
+
+/** 2xx took it, 429 is "not now", any other 4xx is a refusal. */
+export const outcomeForStatus = (status: number): WorkerInvokeOutcome => {
+    if (status >= 200 && status < 300) return 'settled';
+    if (status === 429) return 'retriable';
+    if (status >= 400 && status < 500) return 'terminal';
+    return 'retriable';
+};
diff --git a/src/backend/clients/index.ts b/src/backend/clients/index.ts
index aa6de932a..2fc257a37 100644
--- a/src/backend/clients/index.ts
+++ b/src/backend/clients/index.ts
@@ -21,6 +21,7 @@ import { AlarmClient } from './alarm/AlarmClient';
 import { DatabaseClientFactory } from './database';
 import { EmailClient } from './email/EmailClient';
 import { EventClient } from './event/EventClient';
+import { EventsWorkerInvokerClient } from './events/EventsWorkerInvokerClient';
 import { DDBClient } from './dynamodb/DDBClient';
 import { RedisClient } from './redis/RedisClient';
 import { S3Client } from './s3/S3Client';
@@ -32,6 +33,7 @@ export const puterClients = {
     db: DatabaseClientFactory,
     email: EmailClient,
     event: EventClient,
+    eventsWorkerInvoker: EventsWorkerInvokerClient,
     dynamo: DDBClient,
     redis: RedisClient,
     s3: S3Client,
diff --git a/src/backend/controllers/events/limits.ts b/src/backend/controllers/events/limits.ts
index 6ec9ba37e..294ba5b25 100644
--- a/src/backend/controllers/events/limits.ts
+++ b/src/backend/controllers/events/limits.ts
@@ -183,6 +183,41 @@ export const EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION = 10_000;
  */
 export const EVENTS_REGION_PENDING_CEILING = 1_000_000;
 
+// -- Handler retries -------------------------------------------------
+//
+// A handler that answers "not now" — a 5xx, a timeout, a 429 — is retried, and
+// a handler that answers "no" is not. Retrying on a fixed cadence turns one
+// broken deploy into a permanent load on whatever is failing, so the wait
+// doubles per attempt up to a ceiling, and a run of failures stops the
+// subscription rather than retrying it forever.
+
+/** Wait before the first retry of a delivery a handler could not take. */
+export const EVENTS_RETRY_BASE_MS = 2_000;
+
+/** Longest wait between retries, however many have failed. */
+export const EVENTS_RETRY_MAX_MS = 5 * 60 * 1000;
+
+/**
+ * Failures in a row before a subscription is suspended. Counted per
+ * subscription and reset by the first delivery a handler takes, so an
+ * occasional failure never accumulates into one.
+ */
+export const EVENTS_CONSECUTIVE_FAILURES = 5;
+
+/**
+ * How long a run of failures is remembered. Five failures at the capped wait
+ * span well under this, so a counter nothing has touched for an hour describes
+ * a run that ended.
+ */
+export const EVENTS_FAILURE_COUNTER_TTL_MS = 60 * 60 * 1000;
+
+/** How long to hold a delivery whose handler has failed `attempts` times. */
+export const deliveryBackoffMs = (attempts: number): number =>
+    Math.min(
+        EVENTS_RETRY_MAX_MS,
+        EVENTS_RETRY_BASE_MS * 2 ** Math.max(0, attempts - 1),
+    );
+
 // -- Suspended backlog -----------------------------------------------
 //
 // A suspended subscription stops metering but keeps what it is owed, and that
diff --git a/src/backend/data/hardcoded-permissions.js b/src/backend/data/hardcoded-permissions.js
index 1f0b81b4a..a0c5d4a8c 100644
--- a/src/backend/data/hardcoded-permissions.js
+++ b/src/backend/data/hardcoded-permissions.js
@@ -100,6 +100,10 @@ const implicit_user_app_permissions = [
 const default_user_permissions = {
     driver: {},
     service: {},
+    // Every account may have events delivered to an app's handler while nobody
+    // is there; which app may is the per-app grant, which is what the user is
+    // actually asked about.
+    'events:background': {},
 };
 
 module.exports = {
diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts
index 308941b99..336b9a388 100644
--- a/src/backend/services/events/EventsService.ts
+++ b/src/backend/services/events/EventsService.ts
@@ -23,6 +23,7 @@ import {
     EVENTS_ACK_LIMIT,
     EVENTS_BROADCAST_DELIVERY_LIMIT,
     EVENTS_COALESCE_WINDOW_MS,
+    EVENTS_CONSECUTIVE_FAILURES,
     EVENTS_HANDLER_PUBLISH_BATCH,
     EVENTS_HANDLER_PUBLISH_LIMIT,
     EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT,
@@ -82,13 +83,17 @@ import {
 import {
     assertCrossAppKvAuthorized,
     assertSubscribeAuthorized,
+    backgroundConsentRequired,
     checkDeliveryAuthorized,
     crossAppKvDenial,
     crossAppKvPermissions,
     deliveryGenerationTag,
+    EVENTS_BACKGROUND_PERMISSION,
+    needsBackgroundConsent,
     nodeDescriptor,
     resolveGrantActor,
     rowInActorScope,
+    subscriptionTokenPermissions,
     SUBSCRIBE_MODE,
     type CrossAppKvDeps,
     type EventAclDeps,
@@ -130,8 +135,10 @@ import {
 } from './subjects.js';
 import { backlogPolicyFor, isResumable } from './suspension.js';
 import {
+    EventsWorkerInvoker,
     RecordingWorkerInvoker,
     type WorkerInvocation,
+    type WorkerInvocationOutcome,
     type WorkerInvokerSeam,
 } from './workerSeam.js';
 
@@ -404,6 +411,13 @@ const PENDING_SWEEP_SUBSCRIPTIONS = 100;
  */
 const PENDING_DRAIN_BATCH = 25;
 
+/**
+ * How long the token an invocation carries is good for. Long enough for a
+ * handler doing real work with the account's own data, short enough that a copy
+ * of it is worth nothing by the time anyone finds it.
+ */
+const DELIVERY_TOKEN_TTL = '5m';
+
 /** The part of a socket this service uses, so tests need not build one. */
 export interface EventSocket {
     id: string;
@@ -624,9 +638,9 @@ const parseTargets = (
         throw badRequest('Unknown delivery target', 'invalid_targets');
 
     const targets = [...new Set(value)];
-    if (!targetsAllowedForDelivery(delivery, targets))
+    if (!targetsAllowedForDelivery(delivery, targets, appUid))
         throw badRequest(
-            'A `single` subscription needs a `worker` target and may not target `push`',
+            'A `single` subscription may not target `push`, and an app`s needs a `worker` target',
             'invalid_targets',
         );
     if (appUid === null && targets.includes('worker'))
@@ -725,15 +739,22 @@ export class EventsService extends PuterService {
     #pendingSweep: ReturnType<typeof setInterval> | null = null;
 
     /**
-     * What runs an app's handler. The default records the intent and settles
-     * nothing, so a delivery handed to it stays owed until there is a real
-     * invoker to take it.
+     * What runs an app's handler. Replaced at start-up by the invoker that
+     * calls the app's events worker; the recorder stands in until then and
+     * settles nothing, so a delivery handed to it stays owed.
      */
     worker: WorkerInvokerSeam = new RecordingWorkerInvoker();
 
     // -- Lifecycle ---------------------------------------------------
 
     override onServerStart(): void {
+        // Left alone when something has already put its own invoker here.
+        if (this.worker instanceof RecordingWorkerInvoker)
+            this.worker = new EventsWorkerInvoker(
+                this.clients.eventsWorkerInvoker,
+                (invocation) => this.#mintSubscriberToken(invocation),
+            );
+
         this.clients.event.on(
             'outer.events.generationBumped',
             (_key, data, meta) => {
@@ -974,6 +995,14 @@ export class EventsService extends PuterService {
         // only one that is always there to take it.
         if (delivery === 'single' && !handlerName) throw handlerRequired();
 
+        // Consent first: a row that would run the app's code with nobody
+        // present is refused before anything about it is resolved or stored.
+        if (
+            needsBackgroundConsent(targets) &&
+            !(await this.#hasBackgroundConsent(actor))
+        )
+            throw backgroundConsentRequired();
+
         if (handlerName)
             await this.#assertHandlerBinding(appUid, handlerName, handlerHash);
 
@@ -1099,7 +1128,9 @@ export class EventsService extends PuterService {
             throw unknownSubscription();
 
         await this.stores.pendingDelivery.settle(subId, entryId);
-        await this.#drain(row);
+        // A suspended row is not delivered to: settling what it already handed
+        // out must not be the trigger that hands out the next one.
+        if (row.suspendedAt === null) await this.#drain(row);
     }
 
     // -- Handlers ----------------------------------------------------
@@ -1511,21 +1542,28 @@ export class EventsService extends PuterService {
         return suspended;
     }
 
-    /** Bring back what was waiting on this name. The other half of a removal. */
+    /**
+     * Bring back what was waiting on this name. The other half of a removal —
+     * and of a handler that kept failing, since new source under the name is
+     * the fix for one that could not take its deliveries.
+     */
     async #resumeHandlerDependents(
         appUid: string,
         name: string,
     ): Promise<number> {
         let resumed = 0;
-        for (;;) {
-            const batch = await this.stores.durableSubscription.listByHandler(
-                appUid,
-                name,
-                { suspendedReason: 'handler_not_found' },
-            );
-            if (batch.length === 0) break;
-            resumed += await this.resumeSubscriptions(batch);
-            if (batch.length < HANDLER_SETTLE_BATCH) break;
+        for (const reason of ['handler_not_found', 'failures'] as const) {
+            for (;;) {
+                const batch =
+                    await this.stores.durableSubscription.listByHandler(
+                        appUid,
+                        name,
+                        { suspendedReason: reason },
+                    );
+                if (batch.length === 0) break;
+                resumed += await this.resumeSubscriptions(batch);
+                if (batch.length < HANDLER_SETTLE_BATCH) break;
+            }
         }
         return resumed;
     }
@@ -1581,6 +1619,25 @@ export class EventsService extends PuterService {
         }
     }
 
+    /**
+     * Whether this actor may have its handler run in the background.
+     *
+     * Asked of the actor rather than of the app, so an access token an app
+     * issued has to carry the consent itself — a delivery that runs code with
+     * nobody present is not something a narrower credential inherits.
+     */
+    async #hasBackgroundConsent(actor: Actor): Promise<boolean> {
+        try {
+            return await this.services.permission.check(
+                actor,
+                EVENTS_BACKGROUND_PERMISSION,
+            );
+        } catch (err) {
+            console.warn('[events] background consent check failed', err);
+            return false;
+        }
+    }
+
     /**
      * Whether a subscription may bind the handler name it asked for.
      *
@@ -2180,14 +2237,7 @@ export class EventsService extends PuterService {
             revocation.holderUserId,
             revocation.appUid,
         );
-        // Withdrawing an app's access wholesale is the user saying the app is
-        // done, so it takes everything the app holds — no per-row question,
-        // and none of the standing exemptions an app enjoys over its own data
-        // keep a background subscription alive past the consent that made it.
-        const settling =
-            revocation.permission === null
-                ? held
-                : await this.#leftUnauthorized(held, revocation.permission);
+        const settling = await this.#leftSettling(held, revocation.permission);
         if (settling.length === 0) return 0;
 
         // The `permission_revoked` arm of the shared policy purges the backlog
@@ -2199,6 +2249,52 @@ export class EventsService extends PuterService {
         return suspended.length;
     }
 
+    /**
+     * Which of a holder's rows one withdrawn grant actually stops.
+     *
+     * Withdrawing an app's access wholesale is the user saying the app is done,
+     * so it takes everything the app holds — no per-row question, and none of
+     * the standing exemptions an app enjoys over its own data keep a
+     * subscription alive past the consent that made it.
+     */
+    async #leftSettling(
+        held: readonly DurableSubscription[],
+        permission: string | null,
+    ): Promise<DurableSubscription[]> {
+        if (permission === null) return [...held];
+        if (permission === EVENTS_BACKGROUND_PERMISSION)
+            return this.#leftWithoutConsent(held);
+        return this.#leftUnauthorized(held, permission);
+    }
+
+    /**
+     * Of a holder's rows, the ones that were running the app's code in the
+     * background on a consent that has just been withdrawn. Rows delivered only
+     * to a connection are untouched: what was withdrawn is the right to run
+     * with nobody there, not the app itself.
+     *
+     * The consent is asked for again rather than assumed gone — a mode change
+     * is recorded as a revoke followed by a grant, and settling on the revoke
+     * alone would end subscriptions whose consent still stands.
+     */
+    async #leftWithoutConsent(
+        rows: readonly DurableSubscription[],
+    ): Promise<DurableSubscription[]> {
+        const background = rows.filter((row) =>
+            needsBackgroundConsent(targetsOf(row)),
+        );
+        if (background.length === 0) return [];
+
+        const deps = this.#aclDeps();
+        const settling: DurableSubscription[] = [];
+        for (const row of background) {
+            const actor = await resolveGrantActor(row, deps);
+            if (actor && (await this.#hasBackgroundConsent(actor))) continue;
+            settling.push(row);
+        }
+        return settling;
+    }
+
     /**
      * Of a holder's rows, the ones a named withdrawn grant has actually left
      * without access.
@@ -2562,10 +2658,59 @@ export class EventsService extends PuterService {
 
         const outcome = await this.worker.invoke(invocation);
         this.onDelivered({ subId: row.subId, event: claimed.event });
-        if (outcome !== 'settled') return false;
+        if (outcome === 'settled') {
+            await this.stores.pendingDelivery.clearFailures(row.subId);
+            await this.stores.pendingDelivery.settle(
+                row.subId,
+                claimed.entryId,
+            );
+            return true;
+        }
+        // Nothing was attempted, so nothing failed: the lease paces the retry.
+        if (outcome === 'deferred') return false;
 
-        await this.stores.pendingDelivery.settle(row.subId, claimed.entryId);
-        return true;
+        await this.#handlerFailed(row, claimed, outcome);
+        return false;
+    }
+
+    /**
+     * What a handler that would not take a delivery costs it, and what a run of
+     * them costs the subscription.
+     *
+     * A refusal is the handler's answer, so the delivery is dropped with a gap
+     * marker rather than sent again to the same answer; anything else is "not
+     * now", and the delivery waits longer each time. Both count: five failures
+     * in a row is a handler that is not working, whichever way it is failing,
+     * and retrying it forever is how one bad deploy becomes a standing load on
+     * whatever it is calling.
+     */
+    async #handlerFailed(
+        row: DispatchSubscription,
+        claimed: ClaimedDelivery,
+        outcome: Exclude<WorkerInvocationOutcome, 'settled' | 'deferred'>,
+    ): Promise<void> {
+        const pending = this.stores.pendingDelivery;
+        try {
+            if (outcome === 'terminal')
+                await pending.discard(
+                    row.subId,
+                    claimed.entryId,
+                    'handler_rejected',
+                );
+            else await pending.deferAfterFailure(row.subId, claimed.entryId);
+
+            const failures = await pending.recordFailure(row.subId);
+            if (failures < EVENTS_CONSECUTIVE_FAILURES) return;
+
+            await pending.clearFailures(row.subId);
+            await this.suspendForFailures(row.subId);
+        } catch (err) {
+            console.warn(
+                '[events] could not record a failed handler attempt',
+                row.subId,
+                err,
+            );
+        }
     }
 
     /** What the handler seam is handed, or null for a row that wants none. */
@@ -2582,9 +2727,51 @@ export class EventsService extends PuterService {
             handlerName: row.handlerName ?? null,
             event,
             context: row.context ?? null,
+            permissions: subscriptionTokenPermissions(row),
         };
     }
 
+    /**
+     * The token one invocation carries: the subscriber's own identity, the app
+     * whose handler runs, and exactly the grant the subscription was made
+     * under. Short-lived, because it leaves the platform — a handler that needs
+     * longer than the delivery it was given is asking for standing access the
+     * subscription never granted.
+     *
+     * Null when the identity cannot be rebuilt (the holder or the app is gone),
+     * which is not a handler failure: there is nothing to invoke.
+     */
+    async #mintSubscriberToken(
+        invocation: WorkerInvocation,
+    ): Promise<string | null> {
+        const actor = await resolveGrantActor(
+            {
+                holderUserId: invocation.holderUserId,
+                appUid: invocation.appUid,
+                permission: SUBSCRIBE_MODE,
+            },
+            this.#aclDeps(),
+        );
+        if (!actor) return null;
+
+        try {
+            return await this.services.auth.createAccessToken(
+                actor,
+                invocation.permissions.map(
+                    (permission) => [permission] as [string],
+                ),
+                { expiresIn: DELIVERY_TOKEN_TTL },
+            );
+        } catch (err) {
+            console.warn(
+                '[events] could not mint a delivery token',
+                invocation.subId,
+                err,
+            );
+            return null;
+        }
+    }
+
     /**
      * Say that deliveries were dropped to stay inside a cap. The subscriptions
      * that lost them are told by the gap marker the store queued in their
@@ -2702,7 +2889,9 @@ export class EventsService extends PuterService {
             const invocation = delivery.worker;
             // At-most-once by construction: a `broadcast` invocation is never
             // retried, which is why the docs ask handlers to be idempotent
-            // rather than promising them each event exactly once.
+            // rather than promising them each event exactly once. It counts
+            // toward nothing either — a row whose socket copies are arriving
+            // must not be stopped by a handler nobody is waiting on.
             try {
                 void this.worker.invoke(invocation).catch((err: unknown) => {
                     console.warn('[events] handler invocation failed', err);
diff --git a/src/backend/services/events/authorization.ts b/src/backend/services/events/authorization.ts
index 4f182d10f..b58484b6b 100644
--- a/src/backend/services/events/authorization.ts
+++ b/src/backend/services/events/authorization.ts
@@ -20,6 +20,7 @@
 import { actorUid, makeActor, type Actor } from '../../core/actor.js';
 import { HttpError } from '../../core/http/HttpError.js';
 import type { UserRow } from '../../stores/user/UserStore.js';
+import type { SubscriptionTarget } from '../../stores/events/types.js';
 import type {
     AclError,
     AclMode,
@@ -29,6 +30,8 @@ import {
     appDataPermission,
     appDataSharingAllowed,
 } from '../permission/appDataScopes.js';
+import { PermissionUtil } from '../permission/permissionUtil.js';
+import { isKvToken } from './subjects.js';
 
 /**
  * Who may subscribe to an anchor, who may still be delivered from it, and which
@@ -202,6 +205,51 @@ export const checkDeliveryAuthorized = async (
     }
 };
 
+// -- Background delivery ----------------------------------------------
+
+/**
+ * Consent to run an app's handler with nobody present.
+ *
+ * Delivering to a connected client is the app doing what the user opened it to
+ * do; invoking its handler hours later, on their account and their bill, is a
+ * different thing to agree to — so it is a per-app grant of its own, asked for
+ * through the same flow as every other one and revocable in the same place.
+ */
+export const EVENTS_BACKGROUND_PERMISSION = 'events:background';
+
+/** Whether a row's transports include one that runs code with nobody there. */
+export const needsBackgroundConsent = (
+    targets: readonly SubscriptionTarget[],
+): boolean => targets.includes('worker');
+
+export const backgroundConsentRequired = (): HttpError =>
+    new HttpError(
+        403,
+        'Background delivery needs this app’s `events:background` permission',
+        { legacyCode: 'events_background_consent_required' },
+    );
+
+/**
+ * The grant a delivery's token carries: exactly the one its subscribe check
+ * passed under, so the token can reach what the subscription watches and
+ * nothing else. The issuer-subset check refuses anything wider.
+ *
+ * A row on the app's own key-value namespace has no grant behind it — the app
+ * has always been able to read its own data — so its token carries none and
+ * stands only for who the delivery is for.
+ */
+export const subscriptionTokenPermissions = (row: {
+    token: string;
+    anchorUid: string;
+    appUid: string | null;
+    permission: AclMode;
+}): string[] => {
+    if (!isKvToken(row.token))
+        return [PermissionUtil.join('fs', row.anchorUid, row.permission)];
+    if (row.appUid === null || row.appUid === row.anchorUid) return [];
+    return [appDataPermission(row.anchorUid, 'kv', CROSS_APP_KV_CLASS)];
+};
+
 // -- Cross-app KV ------------------------------------------------------
 
 /**
diff --git a/src/backend/services/events/durable.integration.test.ts b/src/backend/services/events/durable.integration.test.ts
index c47d3f5b4..e1a375d71 100644
--- a/src/backend/services/events/durable.integration.test.ts
+++ b/src/backend/services/events/durable.integration.test.ts
@@ -33,6 +33,7 @@ import { makeActor } from '../../core/actor.js';
 import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js';
 import type { IConfig } from '../../types.js';
 import { appSocketRoom } from '../socket/SocketService.js';
+import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
 import type { DeliveryEnvelope } from './EventsService.js';
 
 const BOOT_TIMEOUT_MS = 120_000;
@@ -92,8 +93,13 @@ const unsubscribe = (token: string, subId: string): Promise<ApiResponse> =>
 const subIdsOf = (response: ApiResponse): string[] =>
     (response.body.items as Array<{ subId: string }>).map((row) => row.subId);
 
-/** An app the user has granted `list` on the shared anchor. */
-const makeApp = async (): Promise<{ uid: string; token: string }> => {
+/**
+ * An app the user has granted `list` on the shared anchor, and consent to run
+ * its handler in the background — which durable rows target by default.
+ */
+const makeApp = async (
+    options: { background?: boolean } = {},
+): Promise<{ uid: string; token: string }> => {
     const uid = `app-${uuidv4()}`;
     await env.server.clients.db.write(
         'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)',
@@ -108,6 +114,12 @@ const makeApp = async (): Promise<{ uid: string; token: string }> => {
         uid,
         `fs:${entry!.uid}:list`,
     );
+    if (options.background !== false)
+        await env.server.services.permission.grantUserAppPermission(
+            actor.actor!,
+            uid,
+            EVENTS_BACKGROUND_PERMISSION,
+        );
     return {
         uid,
         token: await env.server.services.auth.getUserAppToken(actor.actor!, uid),
@@ -272,6 +284,50 @@ describe('creating a durable subscription over HTTP', () => {
         expect(refused.body.code).toBe('invalid_targets');
     });
 
+    it('refuses background delivery an app has no consent for', async () => {
+        // An app of its own, so nothing it holds includes the consent.
+        const { token } = await makeApp({ background: false });
+
+        const refused = await call('POST', '/events/subscribe', token, {
+            subject: `fs:${anchor}`,
+            delivery: 'single',
+            handlerName: 'onWrite',
+            targets: ['worker'],
+        });
+
+        expect(refused.status).toBe(403);
+        expect(refused.body.code).toBe('events_background_consent_required');
+        // Refused before the handler is even looked up: consent comes first.
+        expect(refused.body.message).toContain('events:background');
+    });
+
+    it('requires consent for the default targets an app row gets, even with none named', async () => {
+        // An app of its own, so nothing it holds includes the consent.
+        const { token } = await makeApp({ background: false });
+
+        // No `targets` at all: an app row defaults to `['socket', 'worker']`,
+        // and that default still needs the consent — the gate runs on the
+        // resolved targets, not only on an explicit ask for `worker`.
+        const refused = await call('POST', '/events/subscribe', token, {
+            subject: `fs:${anchor}`,
+        });
+
+        expect(refused.status).toBe(403);
+        expect(refused.body.code).toBe('events_background_consent_required');
+    });
+
+    it('needs no consent for a subscription only a connection hears', async () => {
+        const { token } = await makeApp({ background: false });
+
+        const created = await call('POST', '/events/subscribe', token, {
+            subject: `fs:${anchor}`,
+            targets: ['socket'],
+        });
+
+        expect(created.status).toBe(200);
+        expect(created.body.targets).toEqual(['socket']);
+    });
+
     it('refuses a target outside the known set', async () => {
         const refused = await subscribe(env.users.user.token, {
             targets: ['socket', 'carrier-pigeon'],
@@ -281,8 +337,8 @@ describe('creating a durable subscription over HTTP', () => {
         expect(refused.body.code).toBe('invalid_targets');
     });
 
-    it('refuses a `single` subscription with no handler to fall back to', async () => {
-        const refused = await subscribe(env.users.user.token, {
+    it('refuses an app`s `single` subscription with no worker to fall back to', async () => {
+        const refused = await subscribe(appOneToken, {
             delivery: 'single',
             handlerName: 'onWrite',
             targets: ['socket'],
diff --git a/src/backend/services/events/handlers.integration.test.ts b/src/backend/services/events/handlers.integration.test.ts
index 383c561c6..2ec2520a2 100644
--- a/src/backend/services/events/handlers.integration.test.ts
+++ b/src/backend/services/events/handlers.integration.test.ts
@@ -35,6 +35,7 @@ import {
 import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js';
 import { hashContent } from '../../stores/events/EventHandlerStore.js';
 import type { IConfig } from '../../types.js';
+import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
 import type { DurableSubscriptionView } from './EventsService.js';
 import { RecordingWorkerInvoker } from './workerSeam.js';
 
@@ -119,6 +120,13 @@ const makeApp = async (
             uid,
             `fs:${entry!.uid}:list`,
         );
+        // Durable rows target the app's worker by default, which takes its own
+        // consent.
+        await env.server.services.permission.grantUserAppPermission(
+            actor!,
+            uid,
+            EVENTS_BACKGROUND_PERMISSION,
+        );
         tokens.push(
             await env.server.services.auth.getUserAppToken(actor!, uid),
         );
@@ -676,6 +684,25 @@ describe('the handler lifecycle', () => {
         // A credit restore does not lift a suspension it did not cause.
         expect(await events().resumeForCredit(userId)).toBe(0);
     });
+
+    it('puts a row a failing handler stopped back in service on a republish', async () => {
+        const subId = await bind();
+        await events().suspendForFailures(subId);
+
+        // New source under the name is the fix for a handler that could not
+        // take its deliveries, so it is what brings its subscriptions back.
+        const republished = await call('POST', '/events/handlers/publish', appToken, {
+            name: 'ingestUpload',
+            source: NEXT_SOURCE,
+            replace: true,
+        });
+
+        expect(republished.body.resumed).toBe(1);
+        expect(await rowOf(subId)).toMatchObject({
+            suspendedAt: null,
+            suspendedReason: null,
+        });
+    });
 });
 
 describe('the context a subscription carries', () => {
diff --git a/src/backend/services/events/kv.integration.test.ts b/src/backend/services/events/kv.integration.test.ts
index 6f8675971..8f6f7e7a4 100644
--- a/src/backend/services/events/kv.integration.test.ts
+++ b/src/backend/services/events/kv.integration.test.ts
@@ -33,6 +33,7 @@ import { runWithContext } from '../../core/context.js';
 import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js';
 import type { IConfig } from '../../types.js';
 import { appDataPermission } from '../permission/appDataScopes.js';
+import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
 import type { DeliveryEnvelope } from './EventsService.js';
 
 const BOOT_TIMEOUT_MS = 120_000;
@@ -136,6 +137,13 @@ beforeAll(async () => {
         userActor,
         ownAppUid,
     );
+    // Durable rows target the app's worker by default, which takes its own
+    // consent.
+    await env.server.services.permission.grantUserAppPermission(
+        userActor,
+        ownAppUid,
+        EVENTS_BACKGROUND_PERMISSION,
+    );
 
     delivered = [];
     events().onDelivered = (envelope) => delivered.push(envelope);
diff --git a/src/backend/services/events/registry.ts b/src/backend/services/events/registry.ts
index aba9c6ec0..a4b928847 100644
--- a/src/backend/services/events/registry.ts
+++ b/src/backend/services/events/registry.ts
@@ -79,7 +79,9 @@ export type GapReason =
     | 'backlog_overflow'
     // A suspension holds what it is owed only for as long as the suspension is
     // plausibly recoverable; past that the backlog goes and this stands in.
-    | 'suspended_backlog_expired';
+    | 'suspended_backlog_expired'
+    // The handler refused the delivery outright, so it is not offered again.
+    | 'handler_rejected';
 
 /**
  * `gap` says an event existed and was not delivered. It rides the delivery
diff --git a/src/backend/services/events/revocationSettle.integration.test.ts b/src/backend/services/events/revocationSettle.integration.test.ts
index b5b9c5d7a..df6db96c7 100644
--- a/src/backend/services/events/revocationSettle.integration.test.ts
+++ b/src/backend/services/events/revocationSettle.integration.test.ts
@@ -40,6 +40,7 @@ import {
 } from '../../testUtil.js';
 import type { IConfig } from '../../types.js';
 import type { AclMode } from '../acl/ACLService.js';
+import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
 import type { DeliveryEnvelope } from './EventsService.js';
 import { fsAnchorToken } from './subjects.js';
 
@@ -197,6 +198,13 @@ const makeApp = async (path: string): Promise<Actor> => {
         uid,
         `fs:${await uidOf(path)}:list`,
     );
+    // Durable rows target the app's worker by default, which takes its own
+    // consent.
+    await env.server.services.permission.grantUserAppPermission(
+        owner.actor,
+        uid,
+        EVENTS_BACKGROUND_PERMISSION,
+    );
     const app = await env.server.stores.app.getByUid(uid);
     return makeActor({
         user: owner.actor.user as never,
@@ -503,6 +511,33 @@ describe('what a revoked grant settles', () => {
         expect(delivered).toEqual([]);
     });
 
+    it('settles what background consent allowed, and leaves the rest running', async () => {
+        await clearRows();
+        const path = await folder(`/${owner.username}/settle-background`);
+        const appActor = await makeApp(path);
+
+        const background = (
+            await events().subscribeDurable(appActor, { subject: `fs:${path}` })
+        ).sub;
+        // Delivered to a connection and nowhere else, so the consent that just
+        // went was never what allowed it.
+        const foreground = (
+            await events().subscribeDurable(appActor, {
+                subject: `fs:${path}`,
+                targets: ['socket'],
+            })
+        ).sub;
+
+        await env.server.services.permission.revokeUserAppPermission(
+            owner.actor,
+            appActor.app!.uid,
+            EVENTS_BACKGROUND_PERMISSION,
+        );
+
+        await suspendedRow(background.subId);
+        expect((await rowOf(foreground.subId)).suspended_at).toBeFalsy();
+    });
+
     it('settles a subscription when one app permission is revoked, not just all of them', async () => {
         await clearRows();
         const path = await folder(`/${owner.username}/settle-app-single`);
diff --git a/src/backend/services/events/singleDelivery.test.ts b/src/backend/services/events/singleDelivery.test.ts
index ae3e9df05..b49ac2bd8 100644
--- a/src/backend/services/events/singleDelivery.test.ts
+++ b/src/backend/services/events/singleDelivery.test.ts
@@ -22,6 +22,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import {
     EVENTS_COALESCE_WINDOW_MS,
     EVENTS_REGION_PENDING_CEILING,
+    deliveryBackoffMs,
 } from '../../controllers/events/limits.js';
 import type { Actor } from '../../core/actor.js';
 import { EventSubscriptionStore } from '../../stores/events/EventSubscriptionStore.js';
@@ -379,6 +380,59 @@ describe('a delivery owed to exactly one consumer', () => {
         expect(sent).toHaveLength(1);
         expect(sent[0].ackRequired).toBe(true);
     });
+
+    it('does not clear a run of handler failures when a socket ack settles a later delivery', async () => {
+        socketConnected = false;
+        workerOutcome = 'retriable';
+        const row = await register();
+
+        // Nothing is connected, so this one goes to the handler and fails.
+        await dispatch();
+        expect(invoked).toHaveLength(1);
+        await expect(redis.get(`ev:qf:{${row.subId}}`)).resolves.toBe('1');
+
+        // The client reconnects before the backoff clears; the retry finds a
+        // socket this time and is acked there rather than run again.
+        socketConnected = true;
+        jump(deliveryBackoffMs(1) + 1);
+        await service.sweepPending();
+        expect(sent).toHaveLength(1);
+
+        await service.ackDelivery(actorFor(), {
+            subId: row.subId,
+            id: sent[0].ackId,
+        });
+
+        await expect(pending.depth(row.subId)).resolves.toBe(0);
+        // A socket taking a delivery is not the handler answering: the strike
+        // the handler earned earlier stands.
+        await expect(redis.get(`ev:qf:{${row.subId}}`)).resolves.toBe('1');
+    });
+
+    it('does not hand out the next delivery when an ack lands on a suspended row', async () => {
+        const row = await register();
+        await dispatch(entry({ uid: `file-a-${seq}` }));
+        await dispatch(entry({ uid: `file-b-${seq}` }));
+        expect(sent).toHaveLength(1);
+
+        // Suspended between the first delivery and its ack — a settle hook
+        // mid-flight, not a fresh dispatch decision.
+        rows.set(row.subId, {
+            ...row,
+            suspendedAt: Math.floor(Date.now() / 1000),
+            suspendedReason: 'failures',
+        });
+
+        await service.ackDelivery(actorFor(), {
+            subId: row.subId,
+            id: sent[0].ackId,
+        });
+
+        // The ack itself still settles; what it must not do is hand the next
+        // owed delivery out to a subscription that is no longer in service.
+        expect(sent).toHaveLength(1);
+        await expect(pending.depth(row.subId)).resolves.toBe(1);
+    });
 });
 
 describe('a delivery everyone connected gets', () => {
diff --git a/src/backend/services/events/workerInvoker.integration.test.ts b/src/backend/services/events/workerInvoker.integration.test.ts
new file mode 100644
index 000000000..62ac2b6c9
--- /dev/null
+++ b/src/backend/services/events/workerInvoker.integration.test.ts
@@ -0,0 +1,486 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/**
+ * The call that leaves the platform, against a stub standing in for an app's
+ * events worker.
+ *
+ * The worker runtime is not built yet, so what is pinned here is everything on
+ * this side of the wire: the body, the token it carries and what that token is
+ * allowed to do, and what each answer does to the delivery — settled, dropped,
+ * or held for longer each time until the subscription stops.
+ */
+
+import http from 'node:http';
+import type { AddressInfo } from 'node:net';
+import { v4 as uuidv4 } from 'uuid';
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+    EVENTS_CONSECUTIVE_FAILURES,
+    deliveryBackoffMs,
+} from '../../controllers/events/limits.js';
+import { runWithContext } from '../../core/context.js';
+import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js';
+import type { IConfig } from '../../types.js';
+import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
+
+const BOOT_TIMEOUT_MS = 120_000;
+/** Short enough that a hung handler is a test case rather than a stall. */
+const INVOKE_TIMEOUT_MS = 300;
+const HANDLER = 'ingestUpload';
+const SOURCE = 'async ({ event, ctx }) => { console.log(event.path, ctx.label); }';
+
+interface StubCall {
+    method: string;
+    path: string;
+    auth: string | undefined;
+    body: {
+        handler?: string;
+        event?: Record<string, unknown>;
+        ctx?: Record<string, unknown>;
+    };
+}
+
+let env: PuterTestEnv;
+let userId: number;
+let anchor: string;
+let anchorUid: string;
+let appUid: string;
+let appToken: string;
+let stub: http.Server;
+let calls: StubCall[];
+/** What the stub answers with, or `hang` for a handler that never does. */
+let answer: number | 'hang' = 200;
+
+const events = () => env.server.services.events;
+const pending = () => env.server.stores.pendingDelivery;
+const durable = () => env.server.stores.durableSubscription;
+
+const readBody = async (req: http.IncomingMessage): Promise<string> => {
+    const chunks: Buffer[] = [];
+    for await (const chunk of req) chunks.push(chunk as Buffer);
+    return Buffer.concat(chunks).toString('utf8');
+};
+
+const startStub = async (): Promise<string> => {
+    stub = http.createServer((req, res) => {
+        void readBody(req).then((raw) => {
+            calls.push({
+                method: req.method ?? '',
+                path: req.url ?? '',
+                auth: req.headers['puter-auth'] as string | undefined,
+                body: JSON.parse(raw || '{}') as StubCall['body'],
+            });
+            if (answer === 'hang') return;
+            res.writeHead(answer).end();
+        });
+    });
+    await new Promise<void>((resolve) =>
+        stub.listen(0, '127.0.0.1', () => resolve()),
+    );
+    return `http://127.0.0.1:${(stub.address() as AddressInfo).port}`;
+};
+
+const subscribe = async (): Promise<string> => {
+    const response = await fetch(new URL('/events/subscribe', env.apiOrigin), {
+        method: 'POST',
+        headers: {
+            'content-type': 'application/json',
+            authorization: `Bearer ${appToken}`,
+        },
+        body: JSON.stringify({
+            subject: `fs:${anchor}`,
+            delivery: 'single',
+            handlerName: HANDLER,
+            targets: ['worker'],
+            context: { label: 'ingest' },
+        }),
+    });
+    const body = (await response.json()) as { subId: string };
+    expect(response.status).toBe(200);
+    return body.subId;
+};
+
+/** A durable KV subscription on the app's own namespace, targeting the worker. */
+const subscribeKv = async (key: string): Promise<string> => {
+    const response = await fetch(new URL('/events/subscribe', env.apiOrigin), {
+        method: 'POST',
+        headers: {
+            'content-type': 'application/json',
+            authorization: `Bearer ${appToken}`,
+        },
+        body: JSON.stringify({
+            subject: `kv:${key}`,
+            delivery: 'single',
+            handlerName: HANDLER,
+            targets: ['worker'],
+        }),
+    });
+    const body = (await response.json()) as { subId: string };
+    expect(response.status).toBe(200);
+    return body.subId;
+};
+
+/** One write under the anchor, which is one delivery owed to the handler. */
+const touch = async (name: string): Promise<void> => {
+    await env.server.services.fs.touch(userId, { path: `${anchor}/${name}` });
+};
+
+const invoked = (count: number): Promise<void> =>
+    vi.waitFor(() => expect(calls.length).toBeGreaterThanOrEqual(count), {
+        timeout: 5_000,
+        interval: 25,
+    });
+
+/** Move time past whatever the failed delivery is being held for. */
+const jump = (ms: number): void => {
+    vi.useFakeTimers({ toFake: ['Date'] });
+    vi.setSystemTime(Date.now() + ms);
+};
+
+/** How much longer the one in-flight delivery is being held. */
+const heldForMs = async (subId: string): Promise<number> => {
+    const [, score] = await env.server.clients.redis.zrange(
+        `ev:ql:{${subId}}`,
+        0,
+        0,
+        'WITHSCORES',
+    );
+    return Number(score) - Date.now();
+};
+
+beforeAll(async () => {
+    env = await setupPuterTestEnv({
+        events: { enabled: true, invokeTimeoutMs: INVOKE_TIMEOUT_MS },
+    } as IConfig);
+
+    const user = await env.server.stores.user.getByUsername(
+        env.users.user.username,
+    );
+    userId = user!.id;
+    anchor = `/${env.users.user.username}/invoker`;
+    await env.server.services.fs.mkdir(userId, {
+        path: anchor,
+        createMissingParents: true,
+    });
+    anchorUid = (await env.server.stores.fsEntry.getEntryByPath(anchor))!.uid;
+
+    appUid = `app-${uuidv4()}`;
+    await env.server.clients.db.write(
+        'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)',
+        [appUid, appUid, appUid, `https://${appUid}.example/`, userId],
+    );
+    const { actor } = await env.server.services.auth.authenticate(
+        env.users.user.token,
+    );
+    await env.server.services.permission.grantUserAppPermission(
+        actor!,
+        appUid,
+        `fs:${anchorUid}:list`,
+    );
+    await env.server.services.permission.grantUserAppPermission(
+        actor!,
+        appUid,
+        EVENTS_BACKGROUND_PERMISSION,
+    );
+    appToken = await env.server.services.auth.getUserAppToken(actor!, appUid);
+    await env.server.stores.eventHandler.publish({
+        appUid,
+        name: HANDLER,
+        source: SOURCE,
+    });
+
+    calls = [];
+    const origin = await startStub();
+    env.server.clients.eventsWorkerInvoker.setResolver({
+        resolveInvokeUrl: () => Promise.resolve(origin),
+    });
+}, BOOT_TIMEOUT_MS);
+
+afterAll(async () => {
+    await new Promise<void>((resolve) => stub?.close(() => resolve()));
+    await env?.shutdown();
+});
+
+beforeEach(async () => {
+    vi.useRealTimers();
+    calls = [];
+    answer = 200;
+    await env.server.clients.db.write(
+        'DELETE FROM `event_subscriptions`',
+        [],
+    );
+    events().invalidateUser(userId);
+    await env.server.stores.eventSubscription.markRegionCold(userId);
+    await durable().warmRegion(userId);
+});
+
+describe('the call an owed delivery makes', () => {
+    it('posts the handler, the event and the context to the app`s worker', async () => {
+        await subscribe();
+
+        await touch('note.txt');
+        await invoked(1);
+
+        const call = calls[0];
+        expect(call.method).toBe('POST');
+        expect(call.path).toBe('/__events/invoke');
+        expect(call.body.handler).toBe(HANDLER);
+        expect(call.body.ctx).toEqual({ label: 'ingest' });
+        expect(Object.keys(call.body.event ?? {}).sort()).toEqual([
+            'id',
+            'op',
+            'path',
+            'self',
+            'seq',
+            'subject',
+            'ts',
+            'uid',
+        ]);
+        expect(call.body.event).toMatchObject({
+            path: `${anchor}/note.txt`,
+            self: true,
+        });
+    });
+
+    it('carries a token that is the subscriber, the app, and nothing wider', async () => {
+        await subscribe();
+
+        await touch('token.txt');
+        await invoked(1);
+
+        const token = calls[0].auth;
+        expect(typeof token).toBe('string');
+
+        const { actor } = await env.server.services.auth.authenticate(token!);
+        expect(actor!.user?.id).toBe(userId);
+        expect(actor!.effectiveApp?.uid).toBe(appUid);
+
+        const permission = env.server.services.permission;
+        await expect(
+            permission.check(actor!, `fs:${anchorUid}:list`),
+        ).resolves.toBe(true);
+        // The subscription was made under `list`, so the token stops there —
+        // even though the account it acts for owns the folder outright.
+        await expect(
+            permission.check(actor!, `fs:${anchorUid}:write`),
+        ).resolves.toBe(false);
+
+        const decoded = env.server.services.token.verify('auth', token!) as {
+            exp: number;
+            iat: number;
+        };
+        expect(decoded.exp - decoded.iat).toBe(5 * 60);
+    });
+
+    it('mints no extra grant for a row on the app`s own kv namespace', async () => {
+        // The app already holds `fs:${anchorUid}:list` at the account level
+        // (see `beforeAll`) — proof, below, that a kv delivery's token cannot
+        // spend a grant the app holds for a different reason entirely.
+        await subscribeKv('widget');
+
+        const actor = (await env.server.services.auth.authenticate(appToken))
+            .actor!;
+        await runWithContext({ actor }, () =>
+            env.server.drivers.kvStore.set({ key: 'widget', value: 'x' }),
+        );
+        await invoked(1);
+
+        const token = calls[0].auth!;
+        const decoded = env.server.services.token.verify('auth', token) as {
+            token_uid: string;
+        };
+        const rows = await env.server.clients.db.read(
+            'SELECT `permission` FROM `access_token_permissions` WHERE `token_uid` = ?',
+            [decoded.token_uid],
+        );
+        // Own-namespace kv has no grant behind it: nothing was minted at all.
+        expect(rows).toEqual([]);
+
+        const tokenActor = (
+            await env.server.services.auth.authenticate(token)
+        ).actor!;
+        expect(tokenActor.effectiveApp?.uid).toBe(appUid);
+        await expect(
+            env.server.services.permission.check(
+                tokenActor,
+                `fs:${anchorUid}:list`,
+            ),
+        ).resolves.toBe(false);
+    });
+
+    it('settles the delivery when the handler takes it', async () => {
+        const subId = await subscribe();
+
+        await touch('settled.txt');
+        await invoked(1);
+
+        await vi.waitFor(async () =>
+            expect(await pending().depth(subId)).toBe(0),
+        );
+    });
+});
+
+describe('what each answer does to the delivery', () => {
+    it('drops one the handler refused, and says so with a gap marker', async () => {
+        answer = 400;
+        const subId = await subscribe();
+
+        await touch('refused.txt');
+        await invoked(1);
+
+        // The event is gone and a marker stands in its place, so the
+        // subscription learns there was one rather than reading silence.
+        await vi.waitFor(async () =>
+            expect(await pending().depth(subId)).toBe(1),
+        );
+        answer = 200;
+        const claimed = await pending().claim(subId, { leaseMs: 0 });
+        expect(claimed?.event).toMatchObject({
+            op: 'gap',
+            reason: 'handler_rejected',
+        });
+        // It counted: a refusal is still a handler that did not work.
+        await expect(
+            env.server.clients.redis.get(`ev:qf:{${subId}}`),
+        ).resolves.toBe('1');
+    });
+
+    it('holds one it could not answer, for longer each time', async () => {
+        answer = 500;
+        const subId = await subscribe();
+
+        await touch('failing.txt');
+        await invoked(1);
+
+        const waits: number[] = [];
+        for (let attempt = 1; attempt < EVENTS_CONSECUTIVE_FAILURES; attempt++) {
+            await vi.waitFor(async () =>
+                expect(await heldForMs(subId)).toBeGreaterThan(0),
+            );
+            waits.push(await heldForMs(subId));
+
+            // Nothing may take it while it is held.
+            await events().sweepPending();
+            expect(calls).toHaveLength(attempt);
+
+            jump(deliveryBackoffMs(attempt) + 50);
+            await events().sweepPending();
+            await invoked(attempt + 1);
+        }
+
+        expect(waits.map((wait) => Math.round(wait / 1000))).toEqual([2, 4, 8, 16]);
+
+        // The fifth failure in a row is the one that stops it.
+        await vi.waitFor(async () =>
+            expect((await durable().getBySubId(subId))?.suspendedReason).toBe(
+                'failures',
+            ),
+        );
+        expect(calls).toHaveLength(EVENTS_CONSECUTIVE_FAILURES);
+    });
+
+    it('tells the developer their handler stopped working', async () => {
+        answer = 503;
+        const notify = vi.spyOn(env.server.services.notification, 'notify');
+        const subId = await subscribe();
+
+        await touch('notified.txt');
+        for (let attempt = 1; attempt < EVENTS_CONSECUTIVE_FAILURES; attempt++) {
+            await invoked(attempt);
+            jump(deliveryBackoffMs(attempt) + 50);
+            await events().sweepPending();
+        }
+
+        await vi.waitFor(() =>
+            expect(notify).toHaveBeenCalledWith(
+                [userId],
+                expect.objectContaining({ handler: HANDLER }),
+                expect.objectContaining({ type: 'app.events.suspended' }),
+            ),
+        );
+        expect((await durable().getBySubId(subId))?.suspendedAt).not.toBeNull();
+        notify.mockRestore();
+    });
+
+    it('takes a handler that never answers as one that could not', async () => {
+        answer = 'hang';
+        const subId = await subscribe();
+
+        await touch('hanging.txt');
+        await invoked(1);
+
+        // Still owed, and held rather than dropped: nobody said no.
+        await vi.waitFor(async () =>
+            expect(await heldForMs(subId)).toBeGreaterThan(0),
+        );
+        expect(await pending().depth(subId)).toBe(1);
+        answer = 200;
+    });
+
+    it('takes a handler that is too busy as one that could not', async () => {
+        answer = 429;
+        const subId = await subscribe();
+
+        await touch('busy.txt');
+        await invoked(1);
+
+        await vi.waitFor(async () =>
+            expect(await heldForMs(subId)).toBeGreaterThan(0),
+        );
+        expect(await pending().depth(subId)).toBe(1);
+    });
+});
+
+describe('with no events worker to address', () => {
+    it('counts the same as a handler that could not answer', async () => {
+        env.server.clients.eventsWorkerInvoker.setResolver({
+            resolveInvokeUrl: () => Promise.resolve(null),
+        });
+        const subId = await subscribe();
+
+        try {
+            await touch('unresolved.txt');
+            for (
+                let attempt = 1;
+                attempt < EVENTS_CONSECUTIVE_FAILURES;
+                attempt++
+            ) {
+                await vi.waitFor(async () =>
+                    expect(await heldForMs(subId)).toBeGreaterThan(0),
+                );
+                jump(deliveryBackoffMs(attempt) + 50);
+                await events().sweepPending();
+            }
+
+            await vi.waitFor(async () =>
+                expect(
+                    (await durable().getBySubId(subId))?.suspendedReason,
+                ).toBe('failures'),
+            );
+            // Nothing was ever called: there was nowhere to call.
+            expect(calls).toEqual([]);
+        } finally {
+            const origin = `http://127.0.0.1:${(stub.address() as AddressInfo).port}`;
+            env.server.clients.eventsWorkerInvoker.setResolver({
+                resolveInvokeUrl: () => Promise.resolve(origin),
+            });
+        }
+    });
+});
diff --git a/src/backend/services/events/workerSeam.ts b/src/backend/services/events/workerSeam.ts
index f0759f4f8..28b4effa7 100644
--- a/src/backend/services/events/workerSeam.ts
+++ b/src/backend/services/events/workerSeam.ts
@@ -17,16 +17,15 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+import type { EventsWorkerInvokerClient } from '../../clients/events/EventsWorkerInvokerClient.js';
 import type { DeliverableEvent } from './registry.js';
 
 /**
  * Where a delivery leaves the event system for the app's own code.
  *
- * The invoker itself — minting the subscriber-scoped token, the call, its
- * retries — is not built yet, and delivery semantics must not wait for it: what
- * runs a handler is one decision, and how many times a `single` may be handed
- * out is another. So the seam is the boundary, and the default records the
- * intent without acting on it.
+ * The seam is the boundary between "this delivery is owed to a handler" and
+ * "something ran it": what runs a handler is one decision, and how many times a
+ * `single` may be handed out is another.
  *
  * An invocation is not an ack. A `single` delivery stays leased until the
  * invoker reports the handler took it, which is what keeps a handler that never
@@ -44,19 +43,92 @@ export interface WorkerInvocation {
     event: DeliverableEvent;
     /** The subscription's stored context, delivered to the handler as `ctx`. */
     context: string | null;
+    /**
+     * The grant the subscription was made under, and the whole of what its
+     * token may carry. Resolved with the row so the invoker needs no lookup of
+     * its own.
+     */
+    permissions: string[];
 }
 
 /**
- * What the invoker did with it. `settled` means the handler took the delivery
- * and its lease may be released; `deferred` means it has not, and the delivery
- * stays owed.
+ * What the invoker did with it.
+ *
+ * - `settled` — the handler took the delivery and its lease may be released.
+ * - `terminal` — the handler refused it. Nothing is gained by sending it again.
+ * - `retriable` — nobody could answer; the delivery is owed and comes back.
+ * - `deferred` — nothing was attempted, so nothing failed either.
+ *
+ * `terminal` and `retriable` are both failures and both count toward the run
+ * that suspends a subscription; they differ only in whether the delivery itself
+ * gets another turn.
  */
-export type WorkerInvocationOutcome = 'settled' | 'deferred';
+export type WorkerInvocationOutcome =
+    | 'settled'
+    | 'terminal'
+    | 'retriable'
+    | 'deferred';
 
 export interface WorkerInvokerSeam {
     invoke(invocation: WorkerInvocation): Promise<WorkerInvocationOutcome>;
 }
 
+/** Mints the token one invocation carries, scoped to the subscriber. */
+export type SubscriberTokenMinter = (
+    invocation: WorkerInvocation,
+) => Promise<string | null>;
+
+/**
+ * The invoker that actually calls an app's events worker: mint the
+ * subscriber-scoped token, hand the call to the protocol client, and report
+ * what the handler said.
+ *
+ * A row with no app, no handler name, or no token that can be minted for it has
+ * nothing to invoke, and says so rather than reporting a failure — there is no
+ * handler to blame for a row that never named one.
+ */
+export class EventsWorkerInvoker implements WorkerInvokerSeam {
+    readonly #client: Pick<EventsWorkerInvokerClient, 'invoke'>;
+    readonly #mintToken: SubscriberTokenMinter;
+
+    constructor(
+        client: Pick<EventsWorkerInvokerClient, 'invoke'>,
+        mintToken: SubscriberTokenMinter,
+    ) {
+        this.#client = client;
+        this.#mintToken = mintToken;
+    }
+
+    async invoke(
+        invocation: WorkerInvocation,
+    ): Promise<WorkerInvocationOutcome> {
+        const { appUid, handlerName } = invocation;
+        if (!appUid || !handlerName) return 'deferred';
+
+        const token = await this.#mintToken(invocation);
+        if (token === null) return 'deferred';
+
+        const result = await this.#client.invoke({
+            appUid,
+            handler: handlerName,
+            token,
+            event: invocation.event,
+            ctx: parseContext(invocation.context),
+        });
+        return result.outcome;
+    }
+}
+
+/** Context is stored as JSON text and delivered as the object it was. */
+const parseContext = (context: string | null): unknown => {
+    if (context === null) return undefined;
+    try {
+        return JSON.parse(context) as unknown;
+    } catch {
+        return undefined;
+    }
+};
+
 /** Invocations one recorder holds, so it cannot grow with event volume. */
 const RECORDED_INVOCATIONS = 100;
 
diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts
index 10641a667..33e1ad2a3 100644
--- a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts
+++ b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts
@@ -195,10 +195,11 @@ describe('validation at the row write', () => {
         await expect(durable().countForHolder(userId)).resolves.toBe(0);
     });
 
-    it('refuses a `single` row with no handler to fall back to', async () => {
+    it('refuses an app`s `single` row with no worker to fall back to', async () => {
         await expect(
             durable().create(
                 input({
+                    appUid: 'app-single-rule',
                     delivery: 'single',
                     handlerName: 'onWrite',
                     targets: ['socket'],
@@ -207,6 +208,17 @@ describe('validation at the row write', () => {
         ).rejects.toSatisfy(codeOf('invalid_targets'));
     });
 
+    it('lets an account`s own `single` row stand on the socket alone', async () => {
+        const { row } = await durable().create(
+            input({
+                delivery: 'single',
+                handlerName: 'onWrite',
+                targets: ['socket'],
+            }),
+        );
+        expect(row.targets).toEqual(['socket']);
+    });
+
     it('keeps the same targets on a `broadcast` row, where push is fine', async () => {
         const { row } = await durable().create(
             input({ targets: ['socket', 'push'] }),
diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts
index a243135b0..a3635f0f8 100644
--- a/src/backend/stores/events/DurableSubscriptionStore.ts
+++ b/src/backend/stores/events/DurableSubscriptionStore.ts
@@ -150,7 +150,7 @@ const invalidTargets = (): HttpError =>
 const pushOnSingle = (): HttpError =>
     new HttpError(
         400,
-        'A `single` subscription needs a `worker` target and may not target `push`',
+        'A `single` subscription may not target `push`, and an app`s needs a `worker` target',
         { legacyCode: 'invalid_targets' },
     );
 
@@ -740,7 +740,8 @@ export class DurableSubscriptionStore extends PuterStore {
         if (!targets.every(isSubscriptionTarget)) throw invalidTargets();
 
         const unique = [...new Set(targets as SubscriptionTarget[])];
-        if (!targetsAllowedForDelivery(delivery, unique)) throw pushOnSingle();
+        if (!targetsAllowedForDelivery(delivery, unique, appUid))
+            throw pushOnSingle();
         if (appUid === null && unique.includes('worker'))
             throw workerNeedsApp();
         return unique;
diff --git a/src/backend/stores/events/PendingDeliveryStore.test.ts b/src/backend/stores/events/PendingDeliveryStore.test.ts
index 26e526df6..5264421c9 100644
--- a/src/backend/stores/events/PendingDeliveryStore.test.ts
+++ b/src/backend/stores/events/PendingDeliveryStore.test.ts
@@ -22,6 +22,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import {
     EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION,
     EVENTS_REGION_PENDING_CEILING,
+    EVENTS_RETRY_BASE_MS,
+    EVENTS_RETRY_MAX_MS,
+    deliveryBackoffMs,
 } from '../../controllers/events/limits.js';
 import type { ProjectedEvent } from '../../services/events/registry.js';
 import type { IConfig } from '../../types.js';
@@ -205,6 +208,72 @@ describe('handing a delivery out', () => {
     });
 });
 
+describe('when a handler could not take it', () => {
+    it('holds the delivery longer after each failed attempt', async () => {
+        const { entryId } = await store.enqueue(subId, event('a'));
+        await store.claim(subId);
+
+        await expect(store.deferAfterFailure(subId, entryId)).resolves.toEqual({
+            attempts: 1,
+            retryInMs: EVENTS_RETRY_BASE_MS,
+        });
+        // Held: nothing may take it while the wait stands.
+        await expect(store.claim(subId)).resolves.toBeNull();
+
+        vi.useFakeTimers({ toFake: ['Date'] });
+        vi.setSystemTime(Date.now() + EVENTS_RETRY_BASE_MS + 1);
+        await expect(store.claim(subId)).resolves.toMatchObject({ entryId });
+
+        await expect(store.deferAfterFailure(subId, entryId)).resolves.toEqual({
+            attempts: 2,
+            retryInMs: EVENTS_RETRY_BASE_MS * 2,
+        });
+    });
+
+    it('never waits longer than the cap, however many have failed', async () => {
+        expect(deliveryBackoffMs(1)).toBe(EVENTS_RETRY_BASE_MS);
+        expect(deliveryBackoffMs(50)).toBe(EVENTS_RETRY_MAX_MS);
+    });
+
+    it('drops a refused delivery and leaves a marker naming why', async () => {
+        const { entryId } = await store.enqueue(subId, event('a'));
+
+        await expect(
+            store.discard(subId, entryId, 'handler_rejected'),
+        ).resolves.toBe(true);
+
+        const held = await pendingEvents();
+        expect(held).toHaveLength(1);
+        expect(held[0]).toMatchObject({
+            op: 'gap',
+            reason: 'handler_rejected',
+            subject: 'fs:/u/Documents',
+        });
+        // A delivery that is already gone is not dropped twice.
+        await expect(
+            store.discard(subId, entryId, 'handler_rejected'),
+        ).resolves.toBe(false);
+    });
+
+    it('counts failures in a row, and forgets them when one lands', async () => {
+        await expect(store.recordFailure(subId)).resolves.toBe(1);
+        await expect(store.recordFailure(subId)).resolves.toBe(2);
+
+        await store.clearFailures(subId);
+
+        await expect(store.recordFailure(subId)).resolves.toBe(1);
+    });
+
+    it('takes the failure count with the subscription it belongs to', async () => {
+        await store.enqueue(subId, event('a'));
+        await store.recordFailure(subId);
+
+        await store.purge(subId);
+
+        await expect(keysOf()).resolves.toEqual([]);
+    });
+});
+
 describe('when there is more than can be held', () => {
     it('drops the oldest of one backlog and leaves one marker', async () => {
         await seedBacklog(subId, EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION);
diff --git a/src/backend/stores/events/PendingDeliveryStore.ts b/src/backend/stores/events/PendingDeliveryStore.ts
index e49e5201e..16559fa3a 100644
--- a/src/backend/stores/events/PendingDeliveryStore.ts
+++ b/src/backend/stores/events/PendingDeliveryStore.ts
@@ -19,8 +19,10 @@
 
 import { randomUUID } from 'node:crypto';
 import {
+    EVENTS_FAILURE_COUNTER_TTL_MS,
     EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION,
     EVENTS_REGION_PENDING_CEILING,
+    deliveryBackoffMs,
 } from '../../controllers/events/limits.js';
 import type {
     DeliverableEvent,
@@ -42,9 +44,10 @@ import { PuterStore } from '../types.js';
  * all deleted the moment the last one settles — a subscription that is keeping
  * up owns nothing:
  *
- *     ev:q:{<subId>}    HASH  entryId -> the delivery and its attempt count
+ *     ev:q:{<subId>}    HASH  entryId -> the delivery and its attempt counts
  *     ev:qp:{<subId>}   ZSET  entryId -> enqueued at; membership means unsettled
  *     ev:ql:{<subId>}   ZSET  entryId -> lease expiry; membership means in flight
+ *     ev:qf:{<subId>}   STR   handler failures in a row, expiring on its own
  *
  * And two the region shares:
  *
@@ -72,6 +75,7 @@ const entriesKey = (subId: string): string => `ev:q:{${subId}}`;
 const pendingKey = (subId: string): string => `ev:qp:{${subId}}`;
 const leaseKey = (subId: string): string => `ev:ql:{${subId}}`;
 const holdKey = (subId: string): string => `ev:qt:{${subId}}`;
+const failureKey = (subId: string): string => `ev:qf:{${subId}}`;
 
 const INDEX_KEY = 'ev:qx';
 const COUNTER_KEY = 'ev:qc';
@@ -171,6 +175,14 @@ export interface ClaimedDelivery {
     socketAttempts: number;
 }
 
+/** What one failed handler attempt left behind. */
+export interface DeferredDelivery {
+    /** Handler attempts this delivery has now had. */
+    attempts: number;
+    /** How long it is held before anything may claim it again. */
+    retryInMs: number;
+}
+
 /** What one shed took, for the marker and the alarm that follow it. */
 export interface PendingShed {
     subId: string;
@@ -187,6 +199,8 @@ export interface PendingHead {
 interface StoredEntry {
     event: DeliverableEvent;
     socketAttempts: number;
+    /** Handler attempts spent, which is what the retry wait is derived from. */
+    handlerAttempts?: number;
 }
 
 const parseEntry = (raw: string | null): StoredEntry | null => {
@@ -342,6 +356,78 @@ export class PendingDeliveryStore extends PuterStore {
         return socketAttempts;
     }
 
+    /**
+     * Count one failed handler attempt and hold the delivery until its wait is
+     * over. The lease is the hold: a score in the future is what `claim` reads
+     * as "in flight", so pushing it out paces the retry with no second
+     * mechanism to keep in step.
+     */
+    async deferAfterFailure(
+        subId: string,
+        entryId: string,
+    ): Promise<DeferredDelivery> {
+        const entry = parseEntry(
+            await this.clients.redis.hget(entriesKey(subId), entryId),
+        );
+        const attempts = (entry?.handlerAttempts ?? 0) + 1;
+        const retryInMs = deliveryBackoffMs(attempts);
+
+        const write = this.clients.redis.pipeline();
+        if (entry)
+            write.hset(
+                entriesKey(subId),
+                entryId,
+                JSON.stringify({ ...entry, handlerAttempts: attempts }),
+            );
+        write.zadd(leaseKey(subId), Date.now() + retryInMs, entryId);
+        await write.exec();
+
+        return { attempts, retryInMs };
+    }
+
+    /**
+     * Drop one delivery nothing will ever take, leaving a gap marker in its
+     * place. A refused delivery is still an event its subscription was
+     * promised, so the marker is what keeps the silence from reading as
+     * "nothing happened".
+     */
+    async discard(
+        subId: string,
+        entryId: string,
+        reason: GapMarker['reason'],
+    ): Promise<boolean> {
+        const subject = await this.#subjectOf(subId, entryId);
+        const removed = await this.clients.redis.zrem(
+            pendingKey(subId),
+            entryId,
+        );
+        if (Number(removed) !== 1) return false;
+
+        await this.#forget(subId, [entryId]);
+        await this.#append(subId, gapMarker(subject, reason));
+        return true;
+    }
+
+    /**
+     * Count one failure against the subscription and answer how many are in a
+     * row. Region-local like the rest of the delivery state: the lease that
+     * produced the failure is this region's, and a row update per 5xx would put
+     * a write on the table for every failed attempt.
+     */
+    async recordFailure(subId: string): Promise<number> {
+        const count = Number(await this.clients.redis.incr(failureKey(subId)));
+        await this.clients.redis.pexpire(
+            failureKey(subId),
+            EVENTS_FAILURE_COUNTER_TTL_MS,
+        );
+        return Number.isFinite(count) ? count : 0;
+    }
+
+    /** Forget a run of failures, for a delivery that landed. */
+    async clearFailures(subId: string): Promise<void> {
+        await this.clients.redis.del(failureKey(subId));
+    }
+
     /**
      * Settle a delivery a consumer took. False for an id this subscription is
      * not holding — a second ack for one already settled, which at-least-once
@@ -367,6 +453,7 @@ export class PendingDeliveryStore extends PuterStore {
             pendingKey(subId),
             leaseKey(subId),
             holdKey(subId),
+            failureKey(subId),
         );
         await this.clients.redis.zrem(INDEX_KEY, subId);
         if (held > 0) await this.clients.redis.decrby(COUNTER_KEY, held);
diff --git a/src/backend/stores/events/types.ts b/src/backend/stores/events/types.ts
index c89fea22b..1d2cc90e1 100644
--- a/src/backend/stores/events/types.ts
+++ b/src/backend/stores/events/types.ts
@@ -51,17 +51,21 @@ export const DEFAULT_DURABLE_TARGETS: SubscriptionTarget[] = [
 /**
  * Whether a delivery class may carry these transports. `single` and `push` are
  * incompatible by construction — a lease can only be settled by a consumer that
- * reports back, and a device notification never does — and `single` needs a
- * `worker` to fall back to once the connected clients have had their turns, or
- * an unacknowledged delivery sits at the head of its queue for good. Both are
- * invariants every writer is held to, not defaults.
+ * reports back, and a device notification never does. An app's `single` row
+ * also needs a `worker` to fall back to once the connected clients have had
+ * their turns, or an unacknowledged delivery sits at the head of its queue for
+ * good. An account's own row has no app to run a worker for, so it is held on
+ * the socket alone and the pending hold is its backstop. Both are invariants
+ * every writer is held to, not defaults.
  */
 export const targetsAllowedForDelivery = (
     delivery: DeliveryClass,
     targets: readonly SubscriptionTarget[],
+    appUid: string | null,
 ): boolean =>
     delivery !== 'single' ||
-    (!targets.includes('push') && targets.includes('worker'));
+    (!targets.includes('push') &&
+        (appUid === null || targets.includes('worker')));
 
 /** What dispatch needs from a subscription, whichever store it came from. */
 export interface DispatchSubscription {
diff --git a/src/backend/types.ts b/src/backend/types.ts
index f1c2e7e3a..fb2fbe542 100644
--- a/src/backend/types.ts
+++ b/src/backend/types.ts
@@ -1139,6 +1139,8 @@ interface IConfigOptional {
     events?: {
         enabled?: boolean;
         crossAppKv?: boolean;
+        /** How long a handler has to answer an invocation. Default 30 s. */
+        invokeTimeoutMs?: number;
     };
 
     /**
diff --git a/src/docs/src/Events.md b/src/docs/src/Events.md
index c13040d54..fb80a7bcc 100644
--- a/src/docs/src/Events.md
+++ b/src/docs/src/Events.md
@@ -166,7 +166,13 @@ Values reach a handler through **`context`**, which is evaluated **once, at subs
 
 See [`puter.events.handlers`](/Events/handlers/) for the deploy side — publishing, replacing, and what removing a name does to the subscriptions bound to it.
 
-A persistent subscription can also stop without you unsubscribing: its handler was removed, its holder ran out of credit, or the share it was made under was withdrawn. It is then *suspended* rather than deleted, and [`list()`](/Events/list/) reports `suspendedAt` and `suspendedReason`. Everything but a withdrawn grant can resume.
+### Running when nobody is there takes consent
+
+A persistent subscription delivers to a connected client when there is one and runs the app's handler in the background when there is not. The background half is a separate thing to agree to — your code running on the user's account with nobody watching — so it takes the per-app permission **`events:background`**, requested with [`puter.perms.request()`](/Perms/request/) and revocable wherever the user manages the app's access. Without it, subscribing with `worker` among its `targets` (the default for an app) fails with `events_background_consent_required`; taking it back suspends every worker-target subscription that app holds for that user. A subscription that only wants deliveries while your app is open asks for `targets: ['socket']` and needs no consent.
+
+Pass `handler` as a **function** and it runs here too, whenever this client is the one the delivery goes to — the same body that runs in the worker, with the same `{ event, ctx, user, fetch, ack }`. See [`onPersistent()`](/Events/onPersistent/) for the acknowledgement rules; the short version is that a `single` delivery is settled by returning from the handler, and a handler that throws sees the event again.
+
+A persistent subscription can also stop without you unsubscribing: its handler was removed, its holder ran out of credit, the handler kept failing, or the share it was made under was withdrawn. It is then *suspended* rather than deleted, and [`list()`](/Events/list/) reports `suspendedAt` and `suspendedReason`. Everything but a withdrawn grant can resume.
 
 ## Limits
 
diff --git a/src/docs/src/Events/onPersistent.md b/src/docs/src/Events/onPersistent.md
index 75d40376f..40c4791bd 100644
--- a/src/docs/src/Events/onPersistent.md
+++ b/src/docs/src/Events/onPersistent.md
@@ -27,6 +27,40 @@ puter.events.onPersistent(options)
 - `context` (Object): Values the handler needs, delivered to it as a frozen `ctx`. **Capped at 4 KB serialized** — see below.
 - `expiresAt` (Number | String): When the subscription ends by itself — unix seconds or an ISO-8601 string, and it has to be in the future.
 
+## Background delivery takes the user's consent
+
+A persistent subscription can run your handler when nobody is there — a different thing from delivering to a page the user has open — so it takes its own per-app permission, **`events:background`**. Subscribing with `worker` among its `targets` without it fails with `events_background_consent_required`, and `['socket', 'worker']` is the default for a subscription an app creates. Ask for it the way you ask for anything else:
+
+```js
+await puter.perms.request(['events:background']);
+```
+
+The user can take it back wherever they manage an app's access; every worker-target subscription that app holds for them is then suspended with `permission_revoked`, and re-granting does not bring one back — subscribe again. A subscription that only wants deliveries while your app is open needs no consent at all: pass `targets: ['socket']`.
+
+## Where the handler runs, and what it is handed
+
+The handler runs **in this client while it is connected**, and in the app's events worker when it is not. It is the same body either way, called with:
+
+| Binding | What it is |
+| --- | --- |
+| `event` | The projected event, or a gap marker. |
+| `ctx` | The frozen `context` this subscription was created with. |
+| `user` | A `puter` bound to the account holding the subscription — the ambient one in a client. |
+| `fetch` | [`puter.net.fetch`](/Networking/fetch/) where it exists, the environment's `fetch` otherwise. |
+| `ack` | On a `single` subscription only — see below. |
+
+Passing `handler` as a **function** is what registers it to run here; a source string or `{ file }` is sent as a hash only, and nothing runs client-side. Either way the hash must match what is published under `handlerName`.
+
+### Acknowledging a `single` delivery
+
+A `single` delivery is owed to exactly one consumer, so it stays owed until it is acknowledged:
+
+- Calling `ack()` takes the delivery.
+- Returning **without** calling it acknowledges it anyway — a handler that finished did the work.
+- **Throwing acknowledges nothing.** The lease lapses after 30 seconds and the delivery is offered again, so a handler that throws sees the same event twice. `event.id` is stable across redeliveries; use it to make the second one a no-op.
+
+In the events worker the same three outcomes are the response status: `2xx` takes the delivery, `4xx` refuses it (it is dropped with a `gap` marker carrying `reason: 'handler_rejected'`), and `5xx`, `429` or no answer within 30 seconds means "not now" — the delivery is retried after 2 seconds, doubling to at most 5 minutes. **Five failures in a row, refusals included, suspend the subscription** with `failures`; the developer is notified and republishing the handler puts it back in service.
+
 ## `context` is evaluated once, and capped at 4 KB
 
 A handler is deployed, not called: it is serialized and run later, somewhere else, so it cannot close over anything. `context` is how values reach it — and it is evaluated **at this call**, serialized, and never re-evaluated. `ctx.endpoint` is whatever `process.env.INGEST_URL` was when you subscribed, forever, until you subscribe again.
@@ -52,6 +86,7 @@ A `Promise` that resolves to the subscription:
 - `contextKeys` (Array | null), `contextHash` (String | null): the shape of the stored context, never its values.
 - `createdAt`, `expiresAt` (Number | null): unix seconds.
 - `suspendedAt` (Number | null), `suspendedReason` (String | null): why it stopped delivering without being removed — see [`puter.events.handlers.remove()`](/Events/handlers/).
+- `off()` (Function): ends the subscription — stops running its handler here and unsubscribes it. The same thing as [`puter.events.unsubscribe(subId)`](/Events/unsubscribe/), with nothing to pass.
 
 The promise rejects with `{ message, code }`:
 
@@ -65,6 +100,7 @@ The promise rejects with `{ message, code }`:
 | `events_handler_not_found` | No handler is published under `handlerName`. The subscription is **not** created. |
 | `events_handler_hash_mismatch` | The published handler is not the source this subscription was written against. |
 | `events_handler_required` | `delivery: 'single'` without a `handlerName`. |
+| `events_background_consent_required` | The subscription targets `worker` and the user has not granted this app `events:background`. |
 | `events_context_too_large` | The serialized `context` is over 4 KB. |
 | `events_context_invalid` | `context` is not JSON-serializable. |
 | `invalid_targets` | A target outside `socket`/`worker`/`push`, `push` on a `single` subscription, or `worker` on a subscription with no app. |
diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md
index 08d1fcfb9..a9acc98c8 100644
--- a/src/docs/src/rate-limits-and-quotas.md
+++ b/src/docs/src/rate-limits-and-quotas.md
@@ -172,6 +172,10 @@ One write can reach many subscriptions, so events are bounded on both halves: ho
 | Undelivered deliveries per subscription      | 10,000       |
 | Undelivered deliveries per *suspended* subscription | 100   |
 | Suspended subscriptions kept for             | 30 days      |
+| Handler invocation timeout                   | 30 seconds   |
+| Wait before retrying a failed handler        | 2 seconds, doubling |
+| Longest wait between retries                 | 5 minutes    |
+| Handler failures in a row before suspension  | 5            |
 | Published handlers per app                   | 100          |
 | Handler source size                          | 64 KB        |
 | Handlers per `publishAll` call               | 50           |
@@ -207,6 +211,8 @@ A `kv:` subject is indexed on the first **6** `:`-segments, or **160 bytes**, of
 
 The three per-event ceilings do not fail your call — they truncate the delivery and send a `gap` marker in its place, an event with `op: 'gap'` and no `uid` or `path`. A gap means something happened that you were not told the details of, so a client that must not miss changes should re-read the anchor when it sees one rather than treat the silence as "nothing changed".
 
+A **background delivery** — one that runs your app's handler with nobody there — takes the user's consent, the per-app permission `events:background`, and a subscription targeting `worker` without it is refused with `events_background_consent_required`. A handler has **30 seconds** to answer each invocation. Answering `2xx` takes the delivery; `4xx` refuses it, and it is dropped with a `gap` marker carrying `reason: 'handler_rejected'` rather than sent again to the same answer; `5xx`, `429` and a timeout are all "not now", and the delivery is held **2 seconds** before the next attempt, doubling each time up to **5 minutes**. **Five failures in a row** — refusals included — suspend the subscription with `failures`, hold what it is owed under the suspended-backlog rules above, and notify the app's developer. Until an events worker is deployed for an app there is nothing to invoke, so a worker-target subscription self-limits along exactly this path.
+
 A `single` subscription is delivered to exactly one consumer, which has **30 seconds** to acknowledge each delivery before it is offered again — twice to a connected client, then to the subscription's handler. Until it is acknowledged it is held for you, so a consumer that is away is a backlog that grows: **10,000** undelivered deliveries per subscription, after which the oldest are dropped and one `gap` marker with `reason: 'backlog_overflow'` takes their place. Each region also holds at most **1,000,000** undelivered deliveries across every subscription it serves, and sheds the oldest first — with the same marker — before it reaches that. A redelivery after a missed acknowledgement is normal and expected: deliveries are at-least-once, `event.id` is stable across them, and a handler that runs twice on the same id should do nothing the second time.
 
 ### Peer connections
diff --git a/src/gui/src/UI/UIPermissionDialog.js b/src/gui/src/UI/UIPermissionDialog.js
index 7aac5e58f..bc4ac639f 100644
--- a/src/gui/src/UI/UIPermissionDialog.js
+++ b/src/gui/src/UI/UIPermissionDialog.js
@@ -646,6 +646,10 @@ async function get_permission_description (permission, options = {}) {
         return await get_app_data_description(parts, options);
     }
 
+    if ( parts[0] === 'events' && parts[1] === 'background' ) {
+        return { html: i18n('perm_events_background'), icon: 'zap' };
+    }
+
     if ( parts[0] === 'app-root-dir' ) {
         // Format: app-root-dir:<app_uid>:<read|write>
         if ( parts[2] === 'read' ) {
diff --git a/src/gui/src/i18n/translations/en.js b/src/gui/src/i18n/translations/en.js
index ede010a43..97ed37c88 100644
--- a/src/gui/src/i18n/translations/en.js
+++ b/src/gui/src/i18n/translations/en.js
@@ -772,6 +772,7 @@ const en = {
         'perm_app_data_delete': 'delete entries from {{subject}}.',
         'perm_app_data_store_all': 'read, change and delete {{subject}}.',
         'perm_app_data_all': "read, change and delete everything {{app}} has saved for you, including any saved logins.",
+        'perm_events_background': 'run in the background when your files or data change, even while it is closed',
         'perm_dialog_wants_to': 'wants permission to',
         'perm_dialog_footnote': 'You can change this anytime in Settings.',
         'perm_dialog_error': 'Something went wrong. Please try again.',
diff --git a/src/puter-js/src/modules/events/events.test.js b/src/puter-js/src/modules/events/events.test.js
index 10433af6e..67d3297a2 100644
--- a/src/puter-js/src/modules/events/events.test.js
+++ b/src/puter-js/src/modules/events/events.test.js
@@ -411,6 +411,123 @@ describe('reconnect', () => {
     });
 });
 
+describe('persistent delivery routing', () => {
+    /** One durable envelope, as the server addresses it at a room. */
+    const durable = (subId, path, over = {}) => ({
+        ...projected(subId, path),
+        ...over,
+    });
+
+    it('runs the handler on a delivery the server sends here', async () => {
+        const events = makeModule();
+        const seen = [];
+        events.channel.registerDurable(
+            'app-1#sub',
+            delivery => seen.push(delivery),
+            { label: 'ingest' },
+        );
+
+        sockets[0].fire('events.delivery', durable('app-1#sub', '/user/a/one.txt'));
+
+        expect(seen).toHaveLength(1);
+        expect(seen[0].event.path).toBe('/user/a/one.txt');
+        expect(seen[0].ctx).toEqual({ label: 'ingest' });
+        expect(Object.isFrozen(seen[0].ctx)).toBe(true);
+        // Nothing to acknowledge: everyone connected gets a copy of this one.
+        expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toEqual([]);
+    });
+
+    it('hands a delivery owed to one consumer the environment its worker gets', async () => {
+        const events = makeModule();
+        let handed;
+        events.channel.registerDurable('app-1#sub', delivery => {
+            handed = delivery;
+            return delivery.ack();
+        });
+
+        sockets[0].fire(
+            'events.delivery',
+            durable('app-1#sub', '/user/a/one.txt', {
+                ackRequired: true,
+                ackId: 'entry-1',
+            }),
+        );
+        await Promise.resolve();
+
+        expect(handed.user).toBe(events.puter);
+        expect(typeof handed.fetch).toBe('function');
+        expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toMatchObject([
+            { payload: { subId: 'app-1#sub', id: 'entry-1' } },
+        ]);
+    });
+
+    it('acknowledges a handler that returns without doing so itself', async () => {
+        const events = makeModule();
+        events.channel.registerDurable('app-1#sub', async () => 'done');
+
+        sockets[0].fire(
+            'events.delivery',
+            durable('app-1#sub', '/user/a/one.txt', {
+                ackRequired: true,
+                ackId: 'entry-2',
+            }),
+        );
+        await Promise.resolve();
+        await Promise.resolve();
+
+        expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toMatchObject([
+            { payload: { subId: 'app-1#sub', id: 'entry-2' } },
+        ]);
+    });
+
+    it('acknowledges nothing when the handler throws, so it is delivered again', async () => {
+        const events = makeModule();
+        const errors = vi.spyOn(console, 'error').mockImplementation(() => {});
+        events.channel.registerDurable('app-1#sub', async () => {
+            throw new Error('handler bug');
+        });
+
+        sockets[0].fire(
+            'events.delivery',
+            durable('app-1#sub', '/user/a/one.txt', {
+                ackRequired: true,
+                ackId: 'entry-3',
+            }),
+        );
+        await Promise.resolve();
+        await Promise.resolve();
+
+        expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toEqual([]);
+        expect(errors).toHaveBeenCalled();
+    });
+
+    it('keeps routing across a reconnect without subscribing again', async () => {
+        const events = makeModule();
+        const seen = [];
+        events.channel.registerDurable('app-1#sub', ({ event }) => seen.push(event));
+
+        sockets[0].fire('disconnect');
+        sockets[0].fire('connect');
+
+        expect(sockets[0].sent.filter(s => s.verb === 'events.subscribe')).toEqual([]);
+
+        sockets[0].fire('events.delivery', durable('app-1#sub', '/user/a/back.txt'));
+        expect(seen).toHaveLength(1);
+    });
+
+    it('stops routing once the subscription is let go, and closes with the last', async () => {
+        const events = makeModule();
+        const seen = [];
+        events.channel.registerDurable('app-1#sub', ({ event }) => seen.push(event));
+
+        events.channel.deregisterDurable('app-1#sub');
+        sockets[0].fire('events.delivery', durable('app-1#sub', '/user/a/late.txt'));
+
+        expect(seen).toEqual([]);
+        expect(sockets[0].disconnected).toBe(true);
+    });
+});
+
 describe('ack timeout', () => {
     afterEach(() => {
         vi.useRealTimers();
diff --git a/src/puter-js/src/modules/events/lib/channel.js b/src/puter-js/src/modules/events/lib/channel.js
index a2f155d25..e1d8be844 100644
--- a/src/puter-js/src/modules/events/lib/channel.js
+++ b/src/puter-js/src/modules/events/lib/channel.js
@@ -22,10 +22,11 @@ import { EventSubscription } from './subscription.js';
 
 /** @typedef {{ ok: true, sub?: SubscriptionView }} VerbAck */
 
-// The wire, fixed by the server: two verbs answered with an ack, one channel
+// The wire, fixed by the server: three verbs answered with an ack, one channel
 // events arrive on.
 const SUBSCRIBE_VERB = 'events.subscribe';
 const UNSUBSCRIBE_VERB = 'events.unsubscribe';
+const ACK_VERB = 'events.ack';
 const DELIVERY_CHANNEL = 'events.delivery';
 
 /** How long a verb waits for its ack before the call is called lost. */
@@ -82,6 +83,11 @@ const handshakeError = (error) => {
  * so a reconnect is not transparent server-side — this re-issues each live
  * subscription and re-points its handle at the new id, which is what keeps
  * `onLocal` a thing you call once.
+ *
+ * A persistent subscription is the other way round: the server holds it, its
+ * id outlives every connection, and what a reconnect has to rebuild is only
+ * this side's routing — so those registrations are kept apart from the session
+ * ones and are never re-subscribed.
  */
 export class EventChannel {
     /** @param {import('../index.js').EventsModule} module */
@@ -94,6 +100,12 @@ export class EventChannel {
         this.subscriptions = new Set();
         /** @internal @type {Map<string, EventSubscription>} */
         this.byId = new Map();
+        /**
+         * @internal Persistent subscriptions this client is running the
+         *   handler for, by their server-side id.
+         * @type {Map<string, DurableRegistration>}
+         */
+        this.durable = new Map();
         /** @internal Rejectors for verbs still waiting on an ack. */
         this.waiters = new Set();
         /** @internal Subscribes that have not resolved yet. */
@@ -154,6 +166,38 @@ export class EventChannel {
         }
     }
 
+    /**
+     * Run a persistent subscription's handler here whenever this client is the
+     * one the server delivers to. The subscription itself already exists and is
+     * not re-registered by any of this — what is registered is only where its
+     * deliveries go while this page is open.
+     *
+     * @internal
+     * @param {string} subId
+     * @param {import('../types.js').EventHandler} handler
+     * @param {Record<string, unknown>} [ctx] The context the subscription was
+     *   created with, delivered frozen alongside every event.
+     * @returns {void}
+     */
+    registerDurable (subId, handler, ctx) {
+        this.durable.set(subId, { subId, handler, ctx: Object.freeze({ ...(ctx ?? {}) }) });
+        this.connect();
+    }
+
+    /**
+     * Stop routing a persistent subscription here. The subscription is
+     * untouched — ending it is `unsubscribe()`, which is a different thing from
+     * this page no longer running its handler.
+     *
+     * @internal
+     * @param {string} subId
+     * @returns {void}
+     */
+    deregisterDurable (subId) {
+        if ( ! this.durable.delete(subId) ) return;
+        this.closeIfIdle();
+    }
+
     /**
      * Rebuild the connection against the current token and origin. Live
      * subscriptions are re-issued once it is up.
@@ -163,7 +207,7 @@ export class EventChannel {
      */
     reset () {
         this.close();
-        if ( this.subscriptions.size > 0 ) this.connect();
+        if ( this.subscriptions.size > 0 || this.durable.size > 0 ) this.connect();
     }
 
     /**
@@ -319,21 +363,86 @@ export class EventChannel {
 
     /**
      * @internal
-     * @param {{ subId?: string, event?: unknown }} envelope
+     * @param {{ subId?: string, event?: unknown, ackRequired?: boolean, ackId?: string }} envelope
      * @returns {void}
      */
     route (envelope) {
-        if ( ! envelope || typeof envelope !== 'object' ) return;
-        const sub = this.byId.get(/** @type {string} */ (envelope.subId));
+        if ( ! envelope || typeof envelope !== 'object' || ! envelope.event ) return;
+        const subId = /** @type {string} */ (envelope.subId);
+        const sub = this.byId.get(subId);
+        if ( sub ) {
+            sub.deliver(
+                /** @type {PuterEvent | PuterKvEvent | EventGapMarker} */ (envelope.event),
+                /** @type {Record<string, unknown> | undefined} */ (envelope.ctx),
+            );
+            return;
+        }
+        const registered = this.durable.get(subId);
         // An event for something this client has already unsubscribed from:
         // in flight when `off()` was called, and no longer anybody's.
-        if ( ! sub || ! envelope.event ) return;
-        sub.deliver(
-            /** @type {PuterEvent | PuterKvEvent | EventGapMarker} */ (envelope.event),
-            /** @type {Record<string, unknown> | undefined} */ (envelope.ctx),
+        if ( registered ) this.runDurable(registered, envelope);
+    }
+
+    /**
+     * Run a persistent subscription's handler on one delivery.
+     *
+     * The handler is handed the same environment its published copy gets in the
+     * app's events worker, so one body runs unchanged in either place. A
+     * delivery owed to exactly one consumer carries `ack`: calling it settles
+     * the delivery, resolving without calling it settles it anyway, and
+     * throwing settles nothing — the lease lapses and it is delivered again.
+     *
+     * @internal
+     * @param {DurableRegistration} registration
+     * @param {{ event?: unknown, ackRequired?: boolean, ackId?: string }} envelope
+     * @returns {void}
+     */
+    runDurable (registration, envelope) {
+        const event = /** @type {PuterEvent | PuterKvEvent | EventGapMarker} */ (envelope.event);
+        const delivery = { event, ctx: registration.ctx };
+
+        if ( ! envelope.ackRequired || typeof envelope.ackId !== 'string' ) {
+            settleHandler(() => registration.handler(delivery));
+            return;
+        }
+
+        let acked = false;
+        const ack = () => {
+            if ( acked ) return Promise.resolve();
+            acked = true;
+            return this.ack(registration.subId, /** @type {string} */ (envelope.ackId));
+        };
+        const { puter } = this.module;
+        settleHandler(
+            () =>
+                registration.handler({
+                    ...delivery,
+                    user: puter,
+                    fetch: puter?.net?.fetch ?? globalThis.fetch,
+                    ack,
+                }),
+            () => ack(),
         );
     }
 
+    /**
+     * Tell the server a delivery was taken. Best effort: a lost ack is a
+     * redelivery, which `single` callers are told to expect, and there is
+     * nothing a failure here leaves for the caller to fix.
+     *
+     * @internal
+     * @param {string} subId
+     * @param {string} ackId
+     * @returns {Promise<void>}
+     */
+    async ack (subId, ackId) {
+        try {
+            await this.request(ACK_VERB, { subId, id: ackId }, DEFAULT_TIMEOUT_MS);
+        } catch (error) {
+            console.warn('[puter.events] could not acknowledge a delivery', error);
+        }
+    }
+
     /**
      * The connection is not coming back: fail what is waiting on it and end
      * every subscription it was carrying.
@@ -408,6 +517,7 @@ export class EventChannel {
      */
     closeIfIdle () {
         if ( this.subscriptions.size > 0 || this.inflight > 0 ) return;
+        if ( this.durable.size > 0 ) return;
         this.close();
     }
 
@@ -433,3 +543,37 @@ export class EventChannel {
  */
 const timeoutFor = (sub) =>
     typeof sub.timeout === 'number' && sub.timeout > 0 ? sub.timeout : DEFAULT_TIMEOUT_MS;
+
+/**
+ * One persistent subscription this client runs the handler for.
+ *
+ * @typedef {Object} DurableRegistration
+ * @property {string} subId
+ * @property {import('../types.js').EventHandler} handler
+ * @property {Readonly<Record<string, unknown>>} ctx
+ */
+
+/**
+ * Run a handler and, if it finishes without throwing, do whatever the delivery
+ * still needs. A handler that throws is the app's bug: it is reported, and
+ * nothing is settled on its behalf.
+ *
+ * @param {() => unknown} run
+ * @param {() => unknown} [onResolved]
+ * @returns {void}
+ */
+const settleHandler = (run, onResolved) => {
+    try {
+        const result = run();
+        if ( result instanceof Promise ) {
+            result.then(
+                () => onResolved?.(),
+                error => console.error('[puter.events] subscription handler failed', error),
+            );
+            return;
+        }
+        onResolved?.();
+    } catch (error) {
+        console.error('[puter.events] subscription handler failed', error);
+    }
+};
diff --git a/src/puter-js/src/modules/events/onPersistent.js b/src/puter-js/src/modules/events/onPersistent.js
index 7b91bb95e..c12a8a2b2 100644
--- a/src/puter-js/src/modules/events/onPersistent.js
+++ b/src/puter-js/src/modules/events/onPersistent.js
@@ -9,10 +9,14 @@ import { assertSubject } from './lib/validate.js';
 /**
  * Subscribes to a subject with a subscription that outlives this connection.
  *
- * Unlike `onLocal()`, nothing about this lives in the page: the subscription is
- * stored against the account, keeps matching while the app is closed, and is
- * ended by `puter.events.unsubscribe()` rather than by navigating away. What
- * runs is the app's published handler, named by `handlerName`.
+ * Unlike `onLocal()`, the subscription is stored against the account, keeps
+ * matching while the app is closed, and is ended by
+ * `puter.events.unsubscribe()` rather than by navigating away. What runs it
+ * while nothing is open is the app's published handler, named by `handlerName`.
+ *
+ * While this client *is* open it runs the handler itself, if one was passed as
+ * a function: the same body, the same `{ event, ctx }`, plus `user`, `fetch`
+ * and — for a subscription owed to one consumer — `ack`.
  *
  * `context` is evaluated **here, now** — serialized once and delivered to every
  * invocation as a frozen `ctx`. It never re-evaluates, so a value read from the
@@ -56,7 +60,20 @@ export async function onPersistent (options = {}) {
     if ( serializeContext(options.context) !== undefined )
         body.context = options.context;
 
-    return /** @type {PersistentSubscription} */ (
+    const sub = /** @type {PersistentSubscription} */ (
         await request(puter, '/events/subscribe', body)
     );
+
+    // Durable ids are the server's and survive every reconnect, so routing is
+    // registered once and never re-subscribed.
+    if ( typeof handler === 'function' && typeof sub?.subId === 'string' ) {
+        this.channel.registerDurable(sub.subId, handler, options.context);
+    }
+
+    const { channel } = this;
+    sub.off = async () => {
+        if ( typeof sub.subId === 'string' ) channel.deregisterDurable(sub.subId);
+        await this.unsubscribe(sub.subId);
+    };
+    return sub;
 }
diff --git a/src/puter-js/src/modules/events/persistent.test.js b/src/puter-js/src/modules/events/persistent.test.js
index 42f1b1754..adae16b19 100644
--- a/src/puter-js/src/modules/events/persistent.test.js
+++ b/src/puter-js/src/modules/events/persistent.test.js
@@ -23,6 +23,16 @@ const makeModule = (fsRead) => {
             APIOrigin: 'https://api.test',
             fs: { read: fsRead ?? vi.fn() },
         },
+        channel: {
+            registered: [],
+            deregistered: [],
+            registerDurable (subId, handler, ctx) {
+                this.registered.push({ subId, handler, ctx });
+            },
+            deregisterDurable (subId) {
+                this.deregistered.push(subId);
+            },
+        },
         onPersistent,
         unsubscribe,
         list,
@@ -188,6 +198,53 @@ describe('onPersistent', () => {
             expect(error.code).toBe('events_context_invalid');
         });
     });
+
+    describe('running the handler here as well', () => {
+        it('routes the subscription`s deliveries to the function it was given', async () => {
+            mockRequest.mockResolvedValue({ subId: 'app-1#a', subject: SUBJECT });
+            const module = makeModule();
+
+            await module.onPersistent({
+                subject: SUBJECT,
+                handlerName: 'ingestUpload',
+                handler: HANDLER,
+                context: { url: 'https://ingest.example' },
+            });
+
+            expect(module.channel.registered).toMatchObject([
+                { subId: 'app-1#a', handler: HANDLER, ctx: { url: 'https://ingest.example' } },
+            ]);
+        });
+
+        it('routes nothing when the handler is source this client cannot run', async () => {
+            mockRequest.mockResolvedValue({ subId: 'app-1#a', subject: SUBJECT });
+            const module = makeModule();
+
+            await module.onPersistent({
+                subject: SUBJECT,
+                handlerName: 'ingestUpload',
+                handler: '({ event }) => console.log(event.path)',
+            });
+
+            expect(module.channel.registered).toEqual([]);
+        });
+
+        it('off() stops routing and ends the subscription', async () => {
+            mockRequest.mockResolvedValue({ subId: 'app-1#a', subject: SUBJECT });
+            const module = makeModule();
+            const sub = await module.onPersistent({
+                subject: SUBJECT,
+                handlerName: 'ingestUpload',
+                handler: HANDLER,
+            });
+
+            await sub.off();
+
+            expect(module.channel.deregistered).toEqual(['app-1#a']);
+            expect(routeOf(1)).toBe('/events/unsubscribe');
+            expect(bodyOf(1)).toEqual({ subId: 'app-1#a' });
+        });
+    });
 });
 
 describe('unsubscribe', () => {
diff --git a/src/puter-js/src/modules/events/types.js b/src/puter-js/src/modules/events/types.js
index 5dae4f9aa..dcafb9744 100644
--- a/src/puter-js/src/modules/events/types.js
+++ b/src/puter-js/src/modules/events/types.js
@@ -64,8 +64,9 @@
  * @property {string} reason Why the delivery was dropped —
  *   `matched_subscription_limit`, `filter_evaluation_limit`,
  *   `delivery_rate_limit`, `backlog_overflow` when undelivered events were
- *   shed to stay inside a backlog cap, or `suspended_backlog_expired` when a
- *   suspended subscription held them past its deadline.
+ *   shed to stay inside a backlog cap, `suspended_backlog_expired` when a
+ *   suspended subscription held them past its deadline, or `handler_rejected`
+ *   when the subscription's handler refused the delivery outright.
  * @property {number} ts Milliseconds since the epoch.
  */
 
@@ -73,12 +74,23 @@
  * What a handler is called with. An object rather than the event itself, so
  * more can be added to the call without breaking existing handlers.
  *
+ * The last three arrive only on a persistent subscription, and are what let one
+ * handler body run unchanged here and in the app's events worker.
+ *
  * @typedef {Object} EventDelivery
  * @property {PuterEvent | PuterKvEvent | EventGapMarker} event The delivered
  *   event, or a gap marker in place of events that were dropped.
  * @property {Readonly<Record<string, unknown>>} [ctx] The `context` the
  *   subscription was created with, frozen. Present only for a persistent
  *   subscription; a session subscription carries none.
+ * @property {unknown} [user] A `puter` bound to the account holding the
+ *   subscription — the ambient one when the handler runs in a client.
+ * @property {(input: unknown, init?: unknown) => Promise<unknown>} [fetch]
+ *   `puter.net.fetch` where it exists, the environment's `fetch` otherwise.
+ * @property {() => Promise<void>} [ack] Settles the delivery. Present only on a
+ *   `single` subscription: calling it hands the delivery back as taken,
+ *   returning without calling it does the same, and throwing does neither — the
+ *   delivery is offered again when its lease lapses.
  */
 
 /**
@@ -161,6 +173,9 @@
  *   being removed, or `null` while it is live.
  * @property {string | null} suspendedReason Why it stopped —
  *   `handler_not_found`, `failures`, `no_credit`, or `permission_revoked`.
+ * @property {() => Promise<void>} [off] Ends the subscription: stops running
+ *   its handler here and unsubscribes it. Present on the subscription
+ *   `onPersistent()` returns, not on one a listing reports.
  */
 
 /**
diff --git a/src/puter-js/tests/api/suites/events.suite.ts b/src/puter-js/tests/api/suites/events.suite.ts
index 2db58390c..961edb5ae 100644
--- a/src/puter-js/tests/api/suites/events.suite.ts
+++ b/src/puter-js/tests/api/suites/events.suite.ts
@@ -30,15 +30,15 @@ const sleep = (ms: number): Promise<void> =>
     new Promise((resolve) => setTimeout(resolve, ms));
 
 const waitFor = async (
-    condition: () => boolean,
+    condition: () => boolean | Promise<boolean>,
     timeoutMs: number,
 ): Promise<boolean> => {
     const deadline = Date.now() + timeoutMs;
     while (Date.now() < deadline) {
-        if (condition()) return true;
+        if (await condition()) return true;
         await sleep(50);
     }
-    return condition();
+    return await condition();
 };
 
 const unique = (prefix: string): string =>
@@ -71,6 +71,61 @@ const makeApp = async (t: TestContext): Promise<string> => {
     return (app as unknown as { uid: string }).uid;
 };
 
+/**
+ * A handler that closes over nothing and reports what it ran on through the
+ * one thing it is allowed to reach. Published and subscribed as the same
+ * function, so the hash the subscribe sends is the hash that was published.
+ */
+const RECORDING_HANDLER = async ({
+    event,
+    ctx,
+}: {
+    event: { path?: string };
+    ctx: Record<string, unknown>;
+}) => {
+    const seen = ((globalThis as Record<string, unknown>).puterEventsSeen ??
+        []) as unknown[];
+    seen.push({ path: event.path, label: ctx.label });
+    (globalThis as Record<string, unknown>).puterEventsSeen = seen;
+};
+
+const recorded = (): Array<{ path?: string; label?: unknown }> =>
+    ((globalThis as Record<string, unknown>).puterEventsSeen ?? []) as Array<{
+        path?: string;
+        label?: unknown;
+    }>;
+
+/** Run as the app rather than as the account, then put the session back. */
+const asApp = async <T>(
+    t: TestContext,
+    appUid: string,
+    run: () => Promise<T>,
+): Promise<T> => {
+    const response = await fetch(`${t.env.apiOrigin}/auth/get-user-app-token`, {
+        method: 'POST',
+        headers: {
+            'content-type': 'application/json',
+            authorization: `Bearer ${t.env.users.user.token}`,
+        },
+        body: JSON.stringify({ app_uid: appUid }),
+    });
+    const { token } = (await response.json()) as { token: string };
+    // setAuthToken adopts the app identity from an app token's claims and never
+    // drops it, so the shared instance would keep resolving relative paths
+    // under ~/AppData/<app> for every later test unless it is put back here.
+    const sdk = t.puter as unknown as { appID?: string; appDataPath?: string };
+    const { appID, appDataPath } = sdk;
+    t.puter.setAuthToken(token);
+    try {
+        return await run();
+    } finally {
+        t.puter.setAuthToken(t.env.users.user.token);
+        sdk.appID = appID;
+        sdk.appDataPath = appDataPath;
+    }
+};
+
+
 export default suite('events', {
     'exposes onLocal': async (t) => {
         t.assert.ok(t.puter.events, 'puter.events is registered');
@@ -438,6 +493,144 @@ export default suite('events', {
         t.assert.equal(codeOf(error), 'subscription_does_not_exist');
     },
 
+    'ends a persistent subscription through the handle it returns': async (t) => {
+        const dir = await makeDir(t, 'events-off');
+
+        const sub = await t.puter.events.onPersistent({ subject: `fs:${dir}` });
+        t.assert.equal(typeof sub.off, 'function');
+        await sub.off!();
+
+        const after = await t.puter.events.list();
+        t.assert.ok(
+            ! after.some((row) => row.subId === sub.subId),
+            'off() ends the subscription it was called on',
+        );
+    },
+
+    'runs the handler here while this client is the one connected': async (t) => {
+        const dir = await makeDir(t, 'events-durable');
+        const appUid = await makeApp(t);
+        await t.puter.events.handlers.publish('ingestUpload', RECORDING_HANDLER, {
+            appUid,
+        });
+        // Enough to watch the folder, write into it, and run in the background.
+        await t.puter.perms.grantApp(appUid, `fs:${dir}:write`);
+        await t.puter.perms.grantApp(appUid, 'events:background');
+
+        const before = recorded().length;
+        await asApp(t, appUid, async () => {
+            // A session subscription first: its answer is proof the connection
+            // is up, and the persistent deliveries ride the same one. Without
+            // it the write can land before the socket registers, and the
+            // delivery goes looking for the app's events worker instead.
+            const live = await open(t, `fs:${dir}`, () => {});
+            if (! live) return;
+
+            const sub = await t.puter.events.onPersistent({
+                subject: `fs:${dir}`,
+                delivery: 'single',
+                handlerName: 'ingestUpload',
+                handler: RECORDING_HANDLER,
+                context: { label: 'ingest' },
+            });
+            try {
+                await t.puter.fs.write(`${dir}/first.txt`, 'one');
+                await waitFor(
+                    () => recorded().length > before,
+                    DELIVERY_TIMEOUT_MS,
+                );
+                const first = recorded()[before];
+                t.assert.ok(first, 'the persistent handler ran in this client');
+                t.assert.equal(first?.path, `${dir}/first.txt`);
+                t.assert.equal(
+                    first?.label,
+                    'ingest',
+                    'the handler is handed the context the subscription carries',
+                );
+
+                // A `single` hands out one delivery at a time, so a second one
+                // arriving is the acknowledgement of the first.
+                await t.puter.fs.write(`${dir}/second.txt`, 'two');
+                const settled = await waitFor(
+                    () => recorded().length > before + 1,
+                    DELIVERY_TIMEOUT_MS,
+                );
+                t.assert.ok(
+                    settled,
+                    'the first delivery was acknowledged, so the next was handed over',
+                );
+            } finally {
+                await sub.off!();
+                await live.off();
+            }
+        });
+    },
+
+    'takes background delivery only once the app is allowed it': async (t) => {
+        const dir = await makeDir(t, 'events-consent');
+        const appUid = await makeApp(t);
+        await t.puter.events.handlers.publish('ingestUpload', RECORDING_HANDLER, {
+            appUid,
+        });
+        await t.puter.perms.grantApp(appUid, `fs:${dir}:list`);
+
+        const refused = await asApp(t, appUid, () =>
+            t.assert.rejects(() =>
+                t.puter.events.onPersistent({
+                    subject: `fs:${dir}`,
+                    delivery: 'single',
+                    handlerName: 'ingestUpload',
+                    handler: RECORDING_HANDLER,
+                }),
+            ),
+        );
+        t.assert.equal(codeOf(refused), 'events_background_consent_required');
+
+        await t.puter.perms.grantApp(appUid, 'events:background');
+        const subId = await asApp(t, appUid, async () => {
+            const sub = await t.puter.events.onPersistent({
+                subject: `fs:${dir}`,
+                delivery: 'single',
+                handlerName: 'ingestUpload',
+                handler: RECORDING_HANDLER,
+            });
+            return sub.subId;
+        });
+
+        // Taking the consent back stops the subscription it allowed.
+        await t.puter.perms.revokeApp(appUid, 'events:background');
+        const settled = await waitFor(async () => {
+            const held = await t.puter.events.list();
+            return (
+                held.find((row) => row.subId === subId)?.suspendedReason ===
+                'permission_revoked'
+            );
+        }, DELIVERY_TIMEOUT_MS);
+        t.assert.ok(settled, 'the subscription settled when consent was taken back');
+
+        await t.puter.events.unsubscribe(subId);
+    },
+
+    'needs no consent for a subscription only this connection hears': async (t) => {
+        const dir = await makeDir(t, 'events-no-consent');
+        const appUid = await makeApp(t);
+        await t.puter.perms.grantApp(appUid, `fs:${dir}:list`);
+
+        await asApp(t, appUid, async () => {
+            // Session subscriptions are socket-only by construction.
+            const sub = await open(t, `fs:${dir}`, () => {});
+            if (sub) await sub.off();
+
+            // So is a durable one that asks for nothing else.
+            const durable = await t.puter.events.onPersistent({
+                subject: `fs:${dir}`,
+                targets: ['socket'],
+            });
+            t.assert.deepEqual(durable.targets, ['socket']);
+            await durable.off!();
+        });
+    },
+
     'refuses to bind an inline handler nothing is published for': async (t) => {
         const dir = await makeDir(t, 'events-unbound');
         const error = await t.assert.rejects(() =>