feat(sharing): reconcile recipient policy access

packages/cli/src/commands/coordinator.ts

@@ -561,6 +561,7 @@ export function buildCoordinatorCommand(): Command {
 		.argument("<group>", "group id")
 		.argument("<scope-id>", "Sharing domain scope_id")
 		.argument("<device-id>", "device id")
+		.requiredOption("--effect-id <id>", "deterministic mutation effect id")
 		.option("--role <role>", "membership role")
 		.option("--membership-epoch <epoch>", "membership epoch")
 		.option("--manifest-hash <hash>", "membership manifest hash")
@@ -574,6 +575,7 @@ export function buildCoordinatorCommand(): Command {
 			scopeId: string,
 			deviceId: string,
 			opts: {
+				effectId: string;
 				role?: string;
 				membershipEpoch?: string;
 				manifestHash?: string;
@@ -586,6 +588,7 @@ export function buildCoordinatorCommand(): Command {
 		) => {
 			try {
 				const membership = await coordinatorGrantScopeMembershipAction({
+					effectId: opts.effectId,
 					groupId,
 					scopeId,
 					deviceId,
@@ -626,6 +629,7 @@ export function buildCoordinatorCommand(): Command {
 		.argument("<group>", "group id")
 		.argument("<scope-id>", "Sharing domain scope_id")
 		.argument("<device-id>", "device id")
+		.requiredOption("--effect-id <id>", "deterministic mutation effect id")
 		.option("--membership-epoch <epoch>", "membership epoch")
 		.option("--manifest-hash <hash>", "membership manifest hash")
 		.option("--remote-url <url>", "remote coordinator URL override")
@@ -638,6 +642,7 @@ export function buildCoordinatorCommand(): Command {
 			scopeId: string,
 			deviceId: string,
 			opts: {
+				effectId: string;
 				membershipEpoch?: string;
 				manifestHash?: string;
 				remoteUrl?: string;
@@ -649,6 +654,7 @@ export function buildCoordinatorCommand(): Command {
 		) => {
 			try {
 				const ok = await coordinatorRevokeScopeMembershipAction({
+					effectId: opts.effectId,
 					groupId,
 					scopeId,
 					deviceId,

packages/cli/src/commands/serve.test.ts

@@ -16,6 +16,7 @@ import {
 	maintenanceWorkerPidFilePath,
 	pickViewerPidCandidate,
 	prepareViewerDatabase,
+	runServeCoordinatorMaintenance,
 	sqliteVecFailureDiagnostics,
 	terminateTrustedMaintenanceWorker,
 	terminateTrustedViewerPid,
@@ -186,6 +187,39 @@ describe("serve command option resolution", () => {
 		]);
 	});
 
+	it("runs recipient-policy maintenance only after existing share maintenance", async () => {
+		const calls: string[] = [];
+		const store = {} as MemoryStore;
+		const result = await runServeCoordinatorMaintenance(store, {
+			advancePendingProjectShares: vi.fn(async (_store, options) => {
+				calls.push(`shares:${options.limit}`);
+				return { processed: 1, failed: 0 };
+			}),
+			reconcileRecipientPolicyProjects: vi.fn(async (_store, options) => {
+				calls.push(`policies:${options.limit}`);
+				return { processed: 2, failed: 1 };
+			}),
+		});
+
+		expect(calls).toEqual(["shares:3", "policies:3"]);
+		expect(result).toEqual({
+			projectShares: { processed: 1, failed: 0 },
+			recipientPolicies: { processed: 2, failed: 1 },
+		});
+	});
+
+	it("does not start policy reconciliation when old share maintenance fails", async () => {
+		const reconcileRecipientPolicyProjects = vi.fn(async () => ({ processed: 0, failed: 0 }));
+
+		await expect(
+			runServeCoordinatorMaintenance({} as MemoryStore, {
+				advancePendingProjectShares: vi.fn(async () => ({ processed: 2, failed: 1 })),
+				reconcileRecipientPolicyProjects,
+			}),
+		).rejects.toThrow("share operation maintenance failed for 1 of 2 operations");
+		expect(reconcileRecipientPolicyProjects).not.toHaveBeenCalled();
+	});
+
 	it("detects sqlite-vec load errors for viewer startup fallback", () => {
 		expect(isSqliteVecLoadFailure(new Error("sqlite-vec loaded but version check failed"))).toBe(
 			true,

packages/cli/src/commands/serve.ts

@@ -515,6 +515,36 @@ export function sqliteVecFailureDiagnostics(error: unknown, dbPath: string): str
 	];
 }
 
+export interface ServeCoordinatorMaintenanceResult {
+	projectShares: { processed: number; failed: number };
+	recipientPolicies: { processed: number; failed: number };
+}
+
+export async function runServeCoordinatorMaintenance(
+	store: MemoryStore,
+	dependencies: {
+		advancePendingProjectShares: (
+			store: MemoryStore,
+			options: { limit: number },
+		) => Promise<{ processed: number; failed: number }>;
+		reconcileRecipientPolicyProjects: (
+			store: MemoryStore,
+			options: { limit: number },
+		) => Promise<{ processed: number; failed: number }>;
+	},
+): Promise<ServeCoordinatorMaintenanceResult> {
+	const projectShares = await dependencies.advancePendingProjectShares(store, { limit: 3 });
+	if (projectShares.failed > 0) {
+		throw new Error(
+			`share operation maintenance failed for ${projectShares.failed} of ${projectShares.processed} operations`,
+		);
+	}
+	const recipientPolicies = await dependencies.reconcileRecipientPolicyProjects(store, {
+		limit: 3,
+	});
+	return { projectShares, recipientPolicies };
+}
+
 async function startBackgroundViewer(invocation: ResolvedServeInvocation): Promise<void> {
 	warnIfViewerExposed(invocation.host, invocation.port);
 	if (await isPortOpen(invocation.host, invocation.port)) {
@@ -552,8 +582,14 @@ async function startBackgroundViewer(invocation: ResolvedServeInvocation): Promi
 }
 
 async function startForegroundViewer(invocation: ResolvedServeInvocation): Promise<void> {
-	const { advancePendingProjectShares, createApp, createSyncApp, closeStore, getStore } =
-		await import("@codemem/server");
+	const {
+		advancePendingProjectShares,
+		createApp,
+		createSyncApp,
+		closeStore,
+		getStore,
+		reconcileRecipientPolicyProjects,
+	} = await import("@codemem/server");
 	const { serve } = await import("@hono/node-server");
 
 	if (invocation.dbPath) process.env.CODEMEM_DB = invocation.dbPath;
@@ -677,12 +713,10 @@ async function startForegroundViewer(invocation: ResolvedServeInvocation): Promi
 						// silently falling back to the built-in default ruleset.
 						scanner: store.scanner,
 						onAfterCoordinatorRefresh: async () => {
-							const result = await advancePendingProjectShares(store, { limit: 3 });
-							if (result.failed > 0) {
-								throw new Error(
-									`share operation maintenance failed for ${result.failed} of ${result.processed} operations`,
-								);
-							}
+							await runServeCoordinatorMaintenance(store, {
+								advancePendingProjectShares,
+								reconcileRecipientPolicyProjects,
+							});
 						},
 						onPhaseChange: (phase) => {
 							if (phase === "running") {

packages/cli/src/commands/sync.test.ts

@@ -265,6 +265,8 @@ describe("formatSyncAttempt", () => {
 					"team-a",
 					"scope-acme",
 					"device-a",
+					"--effect-id",
+					"cli:scope-acme:device-a:grant",
 					"--role",
 					"admin",
 					"--db-path",
@@ -315,6 +317,8 @@ describe("formatSyncAttempt", () => {
 					"team-a",
 					"missing-scope",
 					"device-a",
+					"--effect-id",
+					"cli:missing-scope:device-a:grant",
 					"--db-path",
 					dbPath,
 					"--json",

packages/cloudflare-coordinator-worker/migrations/0010_add_scope_membership_effect_receipts.sql

@@ -0,0 +1,23 @@
+ALTER TABLE coordinator_scope_membership_audit_log ADD COLUMN effect_id TEXT;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_coordinator_scope_membership_audit_effect
+ON coordinator_scope_membership_audit_log(effect_id) WHERE effect_id IS NOT NULL;
+
+CREATE TABLE IF NOT EXISTS coordinator_scope_membership_effect_receipts (
+  effect_id TEXT PRIMARY KEY,
+  action TEXT NOT NULL CHECK (action IN ('grant', 'revoke')),
+  request_json TEXT NOT NULL,
+  outcome_applied INTEGER NOT NULL CHECK (outcome_applied IN (0, 1)),
+  scope_id TEXT NOT NULL,
+  device_id TEXT NOT NULL,
+  role TEXT,
+  status TEXT,
+  membership_epoch INTEGER,
+  coordinator_id TEXT,
+  group_id TEXT,
+  manifest_issuer_device_id TEXT,
+  manifest_hash TEXT,
+  signed_manifest_json TEXT,
+  updated_at TEXT,
+  created_at TEXT NOT NULL
+);

packages/cloudflare-coordinator-worker/schema.sql

@@ -157,6 +157,7 @@ ON coordinator_scope_memberships(coordinator_id, group_id);
 
 CREATE TABLE IF NOT EXISTS coordinator_scope_membership_audit_log (
   event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+  effect_id TEXT,
   action TEXT NOT NULL,
   scope_id TEXT NOT NULL,
   device_id TEXT NOT NULL,
@@ -180,6 +181,28 @@ ON coordinator_scope_membership_audit_log(scope_id, created_at, event_id);
 CREATE INDEX IF NOT EXISTS idx_coordinator_scope_membership_audit_device_created
 ON coordinator_scope_membership_audit_log(device_id, created_at, event_id);
 
+CREATE UNIQUE INDEX IF NOT EXISTS idx_coordinator_scope_membership_audit_effect
+ON coordinator_scope_membership_audit_log(effect_id) WHERE effect_id IS NOT NULL;
+
+CREATE TABLE IF NOT EXISTS coordinator_scope_membership_effect_receipts (
+  effect_id TEXT PRIMARY KEY,
+  action TEXT NOT NULL CHECK (action IN ('grant', 'revoke')),
+  request_json TEXT NOT NULL,
+  outcome_applied INTEGER NOT NULL CHECK (outcome_applied IN (0, 1)),
+  scope_id TEXT NOT NULL,
+  device_id TEXT NOT NULL,
+  role TEXT,
+  status TEXT,
+  membership_epoch INTEGER,
+  coordinator_id TEXT,
+  group_id TEXT,
+  manifest_issuer_device_id TEXT,
+  manifest_hash TEXT,
+  signed_manifest_json TEXT,
+  updated_at TEXT,
+  created_at TEXT NOT NULL
+);
+
 CREATE UNIQUE INDEX IF NOT EXISTS idx_coordinator_reciprocal_pending_pair
 ON coordinator_reciprocal_approvals(group_id, pending_pair_low_device_id, pending_pair_high_device_id)
 WHERE status = 'pending';

packages/cloudflare-coordinator-worker/src/index.test.ts

@@ -89,6 +89,8 @@ describe("createCloudflareCoordinatorWorker", () => {
 		tmpDir = mkdtempSync(join(tmpdir(), "cloudflare-coord-worker-test-"));
 		db = connectCoordinator(join(tmpDir, "coordinator.sqlite"));
 		db.exec(`
+			DROP TABLE IF EXISTS coordinator_scope_membership_effect_receipts;
+			DROP TABLE IF EXISTS coordinator_scope_membership_audit_log;
 			DROP TABLE IF EXISTS coordinator_scope_memberships;
 			DROP TABLE IF EXISTS coordinator_scopes;
 			DROP TABLE IF EXISTS coordinator_reciprocal_approvals;
@@ -104,6 +106,54 @@ describe("createCloudflareCoordinatorWorker", () => {
 		d1db = new SqliteD1Database(db);
 	});
 
+	it("migration 0010 preserves audit history and adds immutable effect receipts", () => {
+		db.exec("DROP TABLE coordinator_scope_membership_effect_receipts");
+		db.exec("DROP INDEX idx_coordinator_scope_membership_audit_effect");
+		db.exec("ALTER TABLE coordinator_scope_membership_audit_log RENAME TO audit_with_effect");
+		db.exec(`
+			CREATE TABLE coordinator_scope_membership_audit_log (
+				event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+				action TEXT NOT NULL,
+				scope_id TEXT NOT NULL,
+				device_id TEXT NOT NULL,
+				role TEXT,
+				status TEXT NOT NULL,
+				membership_epoch INTEGER NOT NULL,
+				previous_role TEXT,
+				previous_status TEXT,
+				previous_membership_epoch INTEGER,
+				coordinator_id TEXT,
+				group_id TEXT,
+				actor_type TEXT,
+				actor_id TEXT,
+				manifest_hash TEXT,
+				created_at TEXT NOT NULL
+			);
+			INSERT INTO coordinator_scope_membership_audit_log(
+				action, scope_id, device_id, status, membership_epoch, created_at
+			) VALUES ('grant', 'scope-a', 'device-a', 'active', 1, '2026-07-21T00:00:00Z');
+			DROP TABLE audit_with_effect;
+		`);
+		const migration = readFileSync(
+			join(import.meta.dirname, "../migrations/0010_add_scope_membership_effect_receipts.sql"),
+			"utf8",
+		);
+
+		db.exec(migration);
+
+		expect(
+			db.prepare("SELECT action, effect_id FROM coordinator_scope_membership_audit_log").get(),
+		).toEqual({ action: "grant", effect_id: null });
+		expect(
+			db
+				.prepare(
+					"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'coordinator_scope_membership_effect_receipts'",
+				)
+				.pluck()
+				.get(),
+		).toBe("coordinator_scope_membership_effect_receipts");
+	});
+
 	afterEach(() => {
 		db.close();
 		rmSync(tmpDir, { recursive: true, force: true });