diff --git a/doc/public-mcp.md b/doc/public-mcp.md index 21ac11f2a7e..d5c618a046d 100644 --- a/doc/public-mcp.md +++ b/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 diff --git a/server/src/__tests__/public-mcp.test.ts b/server/src/__tests__/public-mcp.test.ts index f777d6fdeb2..6dafae8fc40 100644 --- a/server/src/__tests__/public-mcp.test.ts +++ b/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 }); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 35c467d91e3..0dcfb800084 100644 --- a/server/src/routes/issues.ts +++ b/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); diff --git a/server/src/services/public-mcp/events.ts b/server/src/services/public-mcp/events.ts index 048cae01d8b..1f03cb6359f 100644 --- a/server/src/services/public-mcp/events.ts +++ b/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 : null; + const previous = details._previous && typeof details._previous === "object" ? details._previous as Record : 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,