fix(mcp): release inactive monitor quotas before admission

Reclaim expired and stopped subscriptions while serializing admission. Include checkout/release status changes and suppress unchanged status writes. Cover retained quota, retry limits, late commits and native transitions.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/public-mcp.md

@@ -221,8 +221,8 @@ at least once, with up to six attempts and exponential backoff. IDs stay stable
 across retries while signatures receive fresh timestamps. HTTP 410 stops a
 monitor; 413 and other permanent failures are not retried. Secret rotation signs
 with both keys for five minutes. Expired subscriptions and their receipts are
-removed after seven days; retained quotas are 20 per grant, 100 per company and
-1,000 per instance. Unsubscribe frees the subscription and its receipts.
+removed on the next admission or after seven days; active quotas are 20 per
+grant, 100 per company and 1,000 per instance. Unsubscribe frees the subscription and its receipts.
 
 This release returns `cursor: null`: it does not offer protocol replay after an
 expired/stopped subscription. Use task history and document tools to recover

server/src/__tests__/public-mcp.test.ts

@@ -456,6 +456,33 @@ describe.skipIf(!support.supported)("public MCP OAuth and tool boundary", () =>
     await f.service.unsubscribe(f.principal, f.input);
   });
 
+  it("reclaims stopped and expired monitors before admitting another subscription", async () => {
+    const f = await eventFixture();
+    const makeInput = (index: number) => ({ ...f.input, delivery: { ...f.input.delivery, url: f.input.delivery.url + "/" + index } });
+    for (let i = 0; i < 20; i++) await f.service.subscribe(f.principal, makeInput(i));
+    await expect(f.service.subscribe(f.principal, makeInput(20))).rejects.toMatchObject({ code: -32602 });
+    await db.update(mcpEventSubscriptions).set({ stoppedAt: new Date(f.now()) }).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id));
+    await expect(f.service.subscribe(f.principal, makeInput(20))).resolves.toHaveProperty("id");
+    expect(await db.select().from(mcpEventSubscriptions).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id))).toHaveLength(1);
+    await db.update(mcpEventSubscriptions).set({ expiresAt: new Date(f.now() - 1) }).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id));
+    await expect(f.service.subscribe(f.principal, makeInput(21))).resolves.toHaveProperty("id");
+    expect(await db.select().from(mcpEventSubscriptions).where(eq(mcpEventSubscriptions.grantId, f.principal.grant.id))).toHaveLength(1);
+    await f.service.unsubscribe(f.principal, makeInput(21));
+  });
+
+  it("filters unchanged statuses and includes native checkout and release transitions", async () => {
+    const f = await eventFixture();
+    const input = { ...f.input, arguments: { companyId: f.company.id, taskId: f.task.id } };
+    await f.service.subscribe(f.principal, input);
+    await f.activity("issue.updated", { status: "done", changes: { title: { from: "a", to: "b" } } });
+    await f.activity("issue.checked_out", { status: "in_progress", _previous: { status: "in_progress" } });
+    await f.activity("issue.checked_out", { status: "in_progress", _previous: { status: "todo" } });
+    await f.activity("issue.released", { status: "todo", _previous: { status: "in_progress" } });
+    await f.service.tick();
+    expect(f.received.filter(r => r.body.eventId).map(r => r.body.data.status)).toEqual(["in_progress", "todo"]);
+    await f.service.unsubscribe(f.principal, input);
+  });
+
   it("bounds lifetimes and retries without claiming unsupported replay", async () => {
     const f = await eventFixture();
     const short = await f.service.subscribe(f.principal, { ...f.input, ttlMs: 1 });

server/src/routes/issues.ts

@@ -15083,7 +15083,7 @@ export function issueRoutes(
         action: "issue.checked_out",
         entityType: "issue",
         entityId: issue.id,
-        details: { agentId: req.body.agentId },
+        details: { agentId: req.body.agentId, status: updated.status, _previous: { status: issue.status } },
       });
 
       if (
@@ -15151,6 +15151,7 @@ export function issueRoutes(
       action: "issue.released",
       entityType: "issue",
       entityId: released.id,
+      details: { status: released.status, _previous: { status: existing.status } },
     });
 
     res.json(released);

server/src/services/public-mcp/events.ts

@@ -1,5 +1,5 @@
 import { createHash } from "node:crypto";
-import { and, asc, count, eq, gt, isNull, lt, lte, sql } from "drizzle-orm";
+import { and, asc, count, eq, gt, isNotNull, isNull, lt, lte, or, sql } from "drizzle-orm";
 import { z } from "zod";
 import { companies, activityLog, mcpEventDeliveries as deliveries, mcpEventSubscriptions as subscriptions, type Db } from "@paperclipai/db";
 import { ISSUE_STATUSES } from "@paperclipai/shared";
@@ -96,14 +96,18 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis
       // Network verification holds only this subscription's lock. The global
       // quota lock covers short admission writes, never a remote callback.
       await tx.execute(sql`select pg_advisory_xact_lock(736721043)`);
-      if (!existing) {
+      const active = existing && !existing.stoppedAt && existing.expiresAt.getTime() > now();
+      // No replay is promised for expired/stopped subscriptions. Reclaim these
+      // rows (and their receipts) before admission so historical monitors cannot
+      // exhaust active capacity or grow storage without bound.
+      await tx.delete(subscriptions).where(or(lte(subscriptions.expiresAt, new Date(now())), isNotNull(subscriptions.stoppedAt)));
+      if (!active) {
         const [total] = await tx.select({ n: count() }).from(subscriptions);
         const [company] = await tx.select({ n: count() }).from(subscriptions).where(eq(subscriptions.companyId, principal.grant.companyId));
         const [grant] = await tx.select({ n: count() }).from(subscriptions).where(eq(subscriptions.grantId, principal.grant.id));
         if (total!.n >= 1000 || company!.n >= 100 || grant!.n >= 20) throw new McpEventError(-32602, "Subscription limit reached. Stop an existing monitor first.");
       }
       const expiresAt = new Date(Math.min(now() + Math.min(Math.max(input.ttlMs ?? lifetime, 30_000), lifetime), cloudOrigin ? Math.min(now() + rotationMs, cloud!.expiresAt) : Infinity));
-      const active = existing && !existing.stoppedAt && existing.expiresAt.getTime() > now();
       const destination: Destination = { url, secret, ...(cloudOrigin ? { cloud } : {}),
         ...(previous && previous.secret !== secret ? { previousSecret: previous.secret, previousUntil: now() + rotationMs }
           : previous?.previousUntil && previous.previousUntil > now() ? { previousSecret: previous.previousSecret, previousUntil: previous.previousUntil } : {}) };
@@ -141,7 +145,10 @@ export function createPublicMcpEvents(db: Db, oauth: PublicMcpOAuth, api: ApiDis
       .orderBy(asc(activityLog.createdAt), asc(activityLog.id)).limit(100);
     for (const { activity } of rows) {
       const details = activity.details ?? {};
-      const wanted = s.name === names[0] ? activity.action === "issue.updated" && ISSUE_STATUSES.includes(details.status as typeof ISSUE_STATUSES[number]) && (!Array.isArray(s.arguments.statuses) || s.arguments.statuses.includes(details.status))
+      const changes = details.changes && typeof details.changes === "object" ? details.changes as Record<string, unknown> : null;
+      const previous = details._previous && typeof details._previous === "object" ? details._previous as Record<string, unknown> : null;
+      const changedStatus = changes ? Object.hasOwn(changes, "status") : previous?.status !== details.status;
+      const wanted = s.name === names[0] ? ["issue.updated", "issue.checked_out", "issue.released"].includes(activity.action) && changedStatus && ISSUE_STATUSES.includes(details.status as typeof ISSUE_STATUSES[number]) && (!Array.isArray(s.arguments.statuses) || s.arguments.statuses.includes(details.status))
         : s.name === names[1] ? activity.action === "issue.comment_added" && z.uuid().safeParse(details.commentId).success
         : ["issue.document_created", "issue.document_updated"].includes(activity.action) && typeof details.key === "string" && typeof details.revisionNumber === "number";
       const data = { companyId: s.companyId, taskId: s.taskId, url: oauth.config.origin + "/" + encodeURIComponent(company.prefix) + "/issues/" + s.taskId,