feat(core): add reliable incremental media usage indexing

.changeset/steady-cats-retry.md

@@ -0,0 +1,5 @@
+---
+"emdash": patch
+---
+
+Adds bounded controls for listing and retrying durable Media Usage indexing work.

docs/src/content/docs/reference/rest-api.mdx

@@ -195,6 +195,7 @@ GET /_emdash/api/media?includeUsage=1
 
 ```json
 {
+	"success": true,
 	"data": {
 		"items": [
 			{
@@ -363,6 +364,74 @@ Content-Type: application/json
 DELETE /_emdash/api/media/:id
 ```
 
+### List Media Usage Work
+
+```http
+GET /_emdash/api/admin/media-usage/work?collection=posts&state=failed&limit=50&cursor=...
+```
+
+Returns a bounded page of durable entry-indexing work for one current collection. The endpoint
+requires `schema:manage`; bearer tokens also require the `admin` scope.
+
+`collection` is required. `state` optionally filters `pending`, `retry`, `leased`, or `failed`
+work. `limit` defaults to 50 and is capped at 100. `cursor` is opaque and comes from the previous
+page's `nextCursor`. The endpoint does not calculate an exact backlog count.
+
+```json
+{
+	"success": true,
+	"data": {
+		"items": [
+			{
+				"collectionId": "01COLLECTION...",
+				"collectionSlug": "posts",
+				"contentId": "01CONTENT...",
+				"state": "failed",
+				"attemptCount": 5,
+				"nextAttemptAt": "2026-08-07T12:00:00.000Z",
+				"leaseExpiresAt": null,
+				"lastAttemptedAt": "2026-08-07T11:45:00.000Z",
+				"lastErrorCode": "MEDIA_USAGE_PROCESSING_FAILED",
+				"updatedAt": "2026-08-07T11:45:00.000Z"
+			}
+		],
+		"nextCursor": "eyJvcmRlclZhbHVlIjoiLi4uIn0"
+	}
+}
+```
+
+Responses omit work versions, lease tokens, raw database errors, indexed content, media
+references, and exact counts.
+
+### Retry Media Usage Work
+
+```http
+POST /_emdash/api/admin/media-usage/work/retry
+Content-Type: application/json
+X-EmDash-Request: 1
+```
+
+Idempotently reopens or creates one durable entry job. It has the same authorization requirements
+as the list endpoint.
+
+```json
+{
+	"collectionId": "01COLLECTION...",
+	"contentId": "01CONTENT..."
+}
+```
+
+A successful response returns `changed` and the current pending item. `changed: false` means the
+job was already pending. A non-expired worker lease returns `409 WORK_LEASE_ACTIVE` with
+`details.leaseExpiresAt`; a concurrent mutation returns `409 WORK_CHANGED`. Neither conflict
+replaces newer work or exposes its lease token.
+
+The list returns only known durable work. Retry can create work for the supplied identity in an
+active collection even when no work row exists, but it does not scan for historical gaps. Use
+collection-scoped Media Usage repair after imports or direct database writes.
+When scheduled maintenance is disabled, failed jobs remain visible and manually retryable, but no
+automatic freshness deadline is promised.
+
 ### Repair Media Usage
 
 ```http

packages/cloudflare/src/sandbox/bridge.ts

@@ -10,14 +10,20 @@
 import type { D1Database } from "@cloudflare/workers-types";
 import { WorkerEntrypoint } from "cloudflare:workers";
 import type { SandboxEmailSendCallback } from "emdash";
-import { ulid, PluginStorageRepository } from "emdash";
+import {
+	createSandboxRouteError,
+	getSandboxRouteErrorDetails,
+	ulid,
+	PluginStorageRepository,
+} from "emdash";
 import { Kysely } from "kysely";
 import { D1Dialect } from "kysely-d1";
 
 import { sandboxHttpFetch } from "./bridge-http.js";
 
 /** Regex to validate collection names (prevent SQL injection) */
 const COLLECTION_NAME_REGEX = /^[a-z][a-z0-9_]*$/;
+const MISSING_MEDIA_USAGE_ACTIVATION_TABLE_REGEX = /no such table.*_emdash_media_usage_activation/i;
 
 /** Regex to validate file extensions (simple alphanumeric, 1-10 chars) */
 const FILE_EXT_REGEX = /^\.[a-z0-9]{1,10}$/i;
@@ -214,6 +220,25 @@ export interface PluginBridgeProps {
  * 3. Plugins call bridge methods which validate and proxy to the database
  */
 export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridgeProps> {
+	private async assertMediaUsageActivationWriteAllowed(): Promise<void> {
+		try {
+			const activation = await this.env.DB.prepare(
+				"SELECT state FROM _emdash_media_usage_activation WHERE task_key = ? LIMIT 1",
+			)
+				.bind("incremental_capture")
+				.first<{ state: string }>();
+			if (activation?.state === "activating") {
+				throw createSandboxRouteError("MEDIA_USAGE_ACTIVATION_IN_PROGRESS");
+			}
+		} catch (error) {
+			const message = error instanceof Error ? error.message : String(error);
+			if (MISSING_MEDIA_USAGE_ACTIVATION_TABLE_REGEX.test(message)) return;
+			if (getSandboxRouteErrorDetails(error)) throw error;
+			console.error("[media-usage] Failed to check the sandbox write fence:", error);
+			throw createSandboxRouteError("MEDIA_USAGE_ACTIVATION_CHECK_FAILED");
+		}
+	}
+
 	/**
 	 * Construct a PluginStorageRepository for the requested collection.
 	 * Uses the indexes from the plugin's storage config (if provided) so
@@ -549,6 +574,7 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 		if (!COLLECTION_NAME_REGEX.test(collection)) {
 			throw new Error(`Invalid collection name: ${collection}`);
 		}
+		await this.assertMediaUsageActivationWriteAllowed();
 
 		const id = ulid();
 		const now = new Date().toISOString();
@@ -621,6 +647,7 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 		if (!COLLECTION_NAME_REGEX.test(collection)) {
 			throw new Error(`Invalid collection name: ${collection}`);
 		}
+		await this.assertMediaUsageActivationWriteAllowed();
 
 		const now = new Date().toISOString();
 		// Quote identifiers to avoid SQL keyword collisions
@@ -679,6 +706,7 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 		if (!COLLECTION_NAME_REGEX.test(collection)) {
 			throw new Error(`Invalid collection name: ${collection}`);
 		}
+		await this.assertMediaUsageActivationWriteAllowed();
 
 		// Soft-delete: set deleted_at timestamp
 		const now = new Date().toISOString();

packages/cloudflare/src/sandbox/runner.ts

@@ -13,6 +13,8 @@
 
 import { env, exports } from "cloudflare:workers";
 import {
+	createSandboxRouteError,
+	getSandboxRouteErrorEnvelope,
 	normalizeCapabilities,
 	type SandboxRunner,
 	type SandboxedPluginInstance,
@@ -356,10 +358,13 @@ class CloudflareSandboxedPlugin implements SandboxedPluginInstance {
 		input: unknown,
 		request: SerializedRequest,
 	): Promise<unknown> {
-		return this.withWallTimeLimit(`route:${routeName}`, () => {
+		return this.withWallTimeLimit(`route:${routeName}`, async () => {
 			const worker = this.createWorker();
 			const entrypoint = worker.getEntrypoint<PluginEntrypoint>("default");
-			return entrypoint.invokeRoute(routeName, input, request);
+			const result = await entrypoint.invokeRoute(routeName, input, request);
+			const envelope = getSandboxRouteErrorEnvelope(result);
+			if (envelope) throw createSandboxRouteError(envelope.error.code);
+			return result;
 		});
 	}
 

packages/cloudflare/src/sandbox/wrapper.ts

@@ -60,6 +60,27 @@ import pluginModule from "sandbox-plugin.js";
 const hooks = pluginModule?.hooks || pluginModule?.default?.hooks || {};
 const routes = pluginModule?.routes || pluginModule?.default?.routes || {};
 
+function sandboxRouteErrorDetails(value) {
+	if (!value || typeof value !== "object") return null;
+	const code =
+		value.code === "MEDIA_USAGE_ACTIVATION_IN_PROGRESS" ||
+		value.code === "MEDIA_USAGE_ACTIVATION_CHECK_FAILED"
+			? value.code
+			: value.name === "MEDIA_USAGE_ACTIVATION_IN_PROGRESS" ||
+				  value.name === "MEDIA_USAGE_ACTIVATION_CHECK_FAILED"
+				? value.name
+				: null;
+	if (!code || (value.status !== undefined && value.status !== 503)) return null;
+	return {
+		code,
+		message:
+			code === "MEDIA_USAGE_ACTIVATION_IN_PROGRESS"
+				? "Media usage activation is in progress"
+				: "Unable to verify media usage activation state",
+		status: 503,
+	};
+}
+
 // -----------------------------------------------------------------------------
 // Context Factory - creates ctx that proxies to BRIDGE
 // -----------------------------------------------------------------------------
@@ -238,7 +259,18 @@ export default class PluginEntrypoint extends WorkerEntrypoint {
 		}
 
 		// Execute the route handler with input, request metadata, and context
-		return handler({ input, request: serializedRequest, requestMeta: serializedRequest.meta }, ctx);
+		try {
+			return await handler(
+				{ input, request: serializedRequest, requestMeta: serializedRequest.meta },
+				ctx,
+			);
+		} catch (error) {
+			const details = sandboxRouteErrorDetails(error);
+			if (details) {
+				return { __emdashSandboxRouteError: true, error: details };
+			}
+			throw error;
+		}
 	}
 }
 `;

packages/cloudflare/tests/sandbox/bridge-content-write-fence.test.ts

@@ -0,0 +1,120 @@
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("cloudflare:workers", () => ({
+	WorkerEntrypoint: class {
+		ctx: unknown;
+		env: unknown;
+		constructor(ctx: unknown, env: unknown) {
+			this.ctx = ctx;
+			this.env = env;
+		}
+	},
+}));
+
+import { PluginBridge } from "../../src/sandbox/bridge.js";
+
+const bridgeContext = {
+	props: {
+		pluginId: "test-plugin",
+		pluginVersion: "1.0.0",
+		capabilities: ["content:write"],
+		allowedHosts: [],
+		storageCollections: [],
+	},
+};
+
+function makeBridge(db: unknown) {
+	return new PluginBridge(bridgeContext as never, { DB: db } as never);
+}
+
+describe("PluginBridge content write fence", () => {
+	it("rejects content mutations while media usage activation is incomplete", async () => {
+		const queries: string[] = [];
+		const db = {
+			prepare(sql: string) {
+				queries.push(sql);
+				return {
+					bind() {
+						return this;
+					},
+					async first() {
+						return { state: "activating" };
+					},
+					async run() {
+						return { meta: { changes: 1 } };
+					},
+				};
+			},
+		};
+		const bridge = makeBridge(db);
+
+		await expect(bridge.contentCreate("posts", { slug: "blocked" })).rejects.toMatchObject({
+			code: "MEDIA_USAGE_ACTIVATION_IN_PROGRESS",
+			message: "Media usage activation is in progress",
+			status: 503,
+		});
+		expect(queries).toHaveLength(1);
+		expect(queries[0]).toContain("_emdash_media_usage_activation");
+	});
+
+	it("preserves content writes before the activation table is migrated", async () => {
+		const queries: string[] = [];
+		const db = {
+			prepare(sql: string) {
+				queries.push(sql);
+				const statement = {
+					bind() {
+						return statement;
+					},
+					async first() {
+						if (sql.includes("_emdash_media_usage_activation")) {
+							throw new Error("D1_ERROR: no such table: _emdash_media_usage_activation");
+						}
+						return {
+							id: "created-id",
+							created_at: "2026-08-09T00:00:00.000Z",
+							updated_at: "2026-08-09T00:00:00.000Z",
+						};
+					},
+					async run() {
+						return { meta: { changes: 1 } };
+					},
+				};
+				return statement;
+			},
+		};
+
+		await expect(makeBridge(db).contentCreate("posts", { slug: "created" })).resolves.toEqual(
+			expect.objectContaining({ id: "created-id", type: "posts" }),
+		);
+		expect(queries).toHaveLength(3);
+	});
+
+	it("fails closed without exposing unexpected database errors", async () => {
+		const db = {
+			prepare() {
+				return {
+					bind() {
+						return this;
+					},
+					async first() {
+						throw new Error("private database failure");
+					},
+				};
+			},
+		};
+		const consoleError = vi.spyOn(console, "error").mockImplementation(() => {});
+
+		try {
+			await expect(
+				makeBridge(db).contentCreate("posts", { slug: "blocked" }),
+			).rejects.toMatchObject({
+				code: "MEDIA_USAGE_ACTIVATION_CHECK_FAILED",
+				message: "Unable to verify media usage activation state",
+				status: 503,
+			});
+		} finally {
+			consoleError.mockRestore();
+		}
+	});
+});