diff --git a/doc/openai-dot-runner.md b/doc/openai-dot-runner.md index 23b42d8511a..1443bb3f7db 100644 --- a/doc/openai-dot-runner.md +++ b/doc/openai-dot-runner.md @@ -73,6 +73,10 @@ enqueue another run or move the same request to another task. A Dot runs one assignment at a time; competing tasks stay queued regardless of the agent's configured concurrency. During setup the connection panel continues checking for a verified event subscription without requiring a manual refresh. +Mailbox writers and cursor reads serialize through the binding row so a cursor +cannot skip a tool result whose transaction commits later. A paused Dot can +receive fence notices and acknowledge them, but cannot read tasks or execute +tools; those narrow inbox reads leave its task cursor unchanged. ## Limits and recovery diff --git a/server/src/__tests__/dot-runner.test.ts b/server/src/__tests__/dot-runner.test.ts index d455f4d8dd5..e60cf5ef56c 100644 --- a/server/src/__tests__/dot-runner.test.ts +++ b/server/src/__tests__/dot-runner.test.ts @@ -9,7 +9,7 @@ import { and, eq, sql } from "drizzle-orm"; import express, { type Request } from "express"; import request from "supertest"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, authUsers, companies, companyMemberships, createDb, dotMailboxItems, dotRunnerAssignments, dotRunnerOperations, +import { agents, authUsers, companies, companyMemberships, createDb, dotAgentBindings, dotMailboxItems, dotRunnerAssignments, dotRunnerOperations, heartbeatRuns, agentWakeupRequests, issues, nativeRunResults, nativeRunFinalizations, completionContracts, mcpEventDeliveries, workspaceOperations } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { createPublicMcpOAuth, DEVICE_GRANT } from "../services/public-mcp/oauth.js"; @@ -18,7 +18,7 @@ import { publicMcpIngressRoutes, publicMcpManagementRoutes } from "../routes/pub import { createPublicMcpEvents, type EventFetch } from "../services/public-mcp/events.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { canonicalNativeRuntimeContextDigest, type StrictCompletionContractInput } from "../vendor/paperclip-runner/index.js"; -import { dotRunnerBroker } from "../services/dot-runner-broker.js"; +import { createDotRunnerMcpTools, dotRunnerBroker } from "../services/dot-runner-broker.js"; import { prepareNativeHeartbeatRun } from "../services/native-runtime/prepare-native-run.js"; import { buildNativeExecutionInput } from "../services/native-runtime/native-execution-input.js"; import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; @@ -169,6 +169,61 @@ describe("durable Dot Runner integration", () => { } }, 30000); + it("serializes tool-result writes and cursor reads with the binding mailbox lock", async () => { + const f = await fixture(); + const { run, assignment } = await offeredWork(f); + const requestId = randomUUID(); + await db.insert(dotRunnerOperations).values({ companyId: f.company.id, assignmentId: assignment.id, requestId, + digest: "test", command: { action: "tool" }, status: "admitted" }); + let release!: () => void; + let held!: () => void; + const locked = new Promise(resolve => { held = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const firstEvent = randomUUID(); + const transaction = db.transaction(async tx => { + await tx.select().from(dotAgentBindings).where(eq(dotAgentBindings.id, f.snapshot.bindingId)).for("update"); + await tx.insert(dotMailboxItems).values({ companyId: f.company.id, bindingId: f.snapshot.bindingId, + bindingGeneration: f.snapshot.bindingGeneration, kind: "operation_result", sourceEventId: firstEvent, references: {} }); + held(); await gate; + }); + await locked; + const port = f.broker.port({ binding: { companyId: f.company.id, agentId: f.agent.id, runId: run.id }, provider: { binding: f.snapshot } }); + let written = false, read = false; + const secondEvent = randomUUID(); + const writer = port.settle({ sourceEventId: secondEvent, payload: { requestId, binding: { ...f.snapshot, runId: run.id, + normalizedSessionId: assignment.normalizedSessionId, turnId: assignment.turnId, assignmentRevision: assignment.revision }, outcome: { status: "completed", result: {} } } } as Parameters[0]).then(() => { written = true; }); + const reader = f.broker.mailbox(f.principal).then(value => { read = true; return value; }); + try { + await new Promise(resolve => setTimeout(resolve, 30)); + expect(written).toBe(false); expect(read).toBe(false); + } finally { release(); } + await Promise.all([transaction, writer]); + const initial = await reader; + const next = await f.broker.mailbox(f.principal, initial.nextCursor); + expect([...initial.items, ...next.items].map(i => i.sourceEventId)).toEqual(expect.arrayContaining([firstEvent, secondEvent])); + await f.events.unsubscribe(f.principal, f.subscription); await f.events.stop(); + }, 30000); + + it("allows paused Dot fence delivery and acknowledgements without task authority", async () => { + const f = await fixture(); + const { assignment } = await offeredWork(f); + await db.update(dotRunnerAssignments).set({ status: "fenced" }).where(eq(dotRunnerAssignments.id, assignment.id)); + await db.insert(dotMailboxItems).values({ companyId: f.company.id, bindingId: f.snapshot.bindingId, + bindingGeneration: f.snapshot.bindingGeneration, assignmentId: assignment.id, kind: "authority_revoked", sourceEventId: randomUUID(), references: { assignmentId: assignment.id } }); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agent.id)); + const principal = await f.oauth.authenticate(f.tokens.access_token); + const tools = createDotRunnerMcpTools(db); + await f.events.tick(); + expect(f.received.at(-1)?.data.kind).toBe("authority_revoked"); + const inbox = await tools.callTool(principal, "paperclip_dot_inbox", { after: 0 }); + expect(inbox).toMatchObject({ result: { nextCursor: 0, items: [{ kind: "authority_revoked" }] } }); + await expect(tools.callTool(principal, "paperclip_dot_tasks", {})).rejects.toThrow("authority"); + await expect(tools.callTool(principal, "paperclip_dot_read", { assignmentId: assignment.id })).rejects.toThrow("authority"); + await expect(tools.callTool(principal, "paperclip_dot_accept", { assignmentId: assignment.id, requestId: randomUUID() })).rejects.toThrow("authority"); + expect(await tools.callTool(principal, "paperclip_dot_control_ack", { assignmentId: assignment.id, requestId: randomUUID() })).toMatchObject({ result: { status: "acknowledged", externalStopConfirmed: false } }); + await f.events.unsubscribe(principal, f.subscription); await f.events.stop(); + }, 30000); + it("verifies reconnects and wakes the same outstanding assignment after exhausted delivery", async () => { const f = await fixture(); const { assignment } = await offeredWork(f); diff --git a/server/src/services/dot-runner-broker.ts b/server/src/services/dot-runner-broker.ts index fd62fdc320b..efdd2279834 100644 --- a/server/src/services/dot-runner-broker.ts +++ b/server/src/services/dot-runner-broker.ts @@ -148,15 +148,26 @@ function createBroker(db: Db) { acceptByUnixMs: Date.now() + 10 * 60_000, expiresAtUnixMs: Date.now() + 2 * 60 * 60_000 }; }, async authorizeBinding(principal: McpPrincipal, companyId: string, bindingId: string) { - const b = await principalBinding(principal, false); + const b = await principalBinding(principal, false, true); if (b.companyId !== companyId || b.id !== bindingId) throw fail("Binding is outside this connection."); return b; }, async mailbox(principal: McpPrincipal, after = 0) { - const b = await principalBinding(principal, false); - const items = await db.select().from(mailbox).where(and(eq(mailbox.companyId, b.companyId), eq(mailbox.bindingId, b.id), - eq(mailbox.bindingGeneration, b.generation), gt(mailbox.id, after))).orderBy(asc(mailbox.id)).limit(50); - return { bindingId: b.id, generation: b.generation, items, nextCursor: items.at(-1)?.id ?? after }; + const binding = await principalBinding(principal, false, true); + return db.transaction(async tx => { + // Every writer takes this lock before allocating a mailbox ID. Reading + // under it cannot advance a cursor past an earlier uncommitted item. + const [b] = await tx.select().from(bindings).where(and(eq(bindings.id, binding.id), + eq(bindings.generation, binding.generation), isNull(bindings.revokedAt))).for("update"); + if (!b) throw fail("Binding authority is unavailable."); + const [agent] = await tx.select().from(agents).where(eq(agents.id, b.agentId)); + if (!agent || ["terminated", "pending_approval"].includes(agent.status)) throw fail("Agent connection authority is unavailable."); + const paused = agent.status === "paused"; + const items = await tx.select().from(mailbox).where(and(eq(mailbox.companyId, b.companyId), eq(mailbox.bindingId, b.id), + eq(mailbox.bindingGeneration, b.generation), gt(mailbox.id, after), paused ? eq(mailbox.kind, "authority_revoked") : undefined)).orderBy(asc(mailbox.id)).limit(50); + // Paused reads expose fences only and never consume hidden task items. + return { bindingId: b.id, generation: b.generation, items, nextCursor: paused ? after : items.at(-1)?.id ?? after }; + }); }, async read(principal: McpPrincipal, assignmentId: string) { const { assignment } = await authorizeAssignment(principal, assignmentId); @@ -363,6 +374,9 @@ function createBroker(db: Db) { const p = z.object({ requestId: z.uuid(), binding: z.object({ runId: z.literal(runId), bindingId: z.literal(execution.provider.binding.bindingId), bindingGeneration: z.literal(execution.provider.binding.bindingGeneration) }).passthrough(), outcome: z.record(z.string(), z.unknown()) }).parse(event.payload); await db.transaction(async tx => { + // Binding precedes assignment everywhere: serialize mailbox IDs + // with dispatch, readiness and reconnect without reversing locks. + await tx.select({ id: bindings.id }).from(bindings).where(eq(bindings.id, p.binding.bindingId)).for("update"); const [a] = await tx.select().from(assignments).where(and(eq(assignments.runId, runId), eq(assignments.bindingGeneration, p.binding.bindingGeneration))).for("update"); if (!a) throw fail("Runner operation has no assignment."); if (p.binding.companyId !== a.companyId || p.binding.agentId !== a.agentId || p.binding.normalizedSessionId !== a.normalizedSessionId diff --git a/server/src/services/public-mcp/oauth.ts b/server/src/services/public-mcp/oauth.ts index 079bf295577..2c5c8fd90d9 100644 --- a/server/src/services/public-mcp/oauth.ts +++ b/server/src/services/public-mcp/oauth.ts @@ -94,7 +94,9 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { if (membership.membershipRole === "viewer") throw invalidGrant(); if (!grant.agentId) return { type: "none", source: "mcp_oauth", companyId: grant.companyId }; const [agent] = await queryDb.select().from(agents).where(and(eq(agents.id, grant.agentId), eq(agents.companyId, grant.companyId))); - if (!agent || ["paused", "terminated", "pending_approval"].includes(agent.status)) throw invalidGrant(); + // Paused Dot connections retain only the broker's fence inbox and ack + // authority. Each task/tool method independently rejects paused agents. + if (!agent || ["terminated", "pending_approval"].includes(agent.status)) throw invalidGrant(); return { type: "agent", source: "mcp_oauth", agentId: agent.id, companyId: grant.companyId }; } if (grant.purpose !== "personal") throw invalidGrant();