mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 06:42:06 +01:00
fix(grok): verify file completion and recover failed sessions (#16305)
## Thinking Path > - Paperclip manages work and authority for AI agents. > - Grok Bot runs outside Paperclip and uses authenticated upload grants for files. > - Native completion requires a verified file publication receipt. > - The upload saved the file but did not create that receipt. Real Grok work could not finish. > - A failed run also left an assignment that the Stop action could not fence. > - This pull request connects verified uploads to normal completion and repairs that recovery action. > - The result preserves file integrity and current work-session authority. ## Linked Issues or Issue Description Refs: #16219, #16274. **What happened?** A real Grok Bot saved a task document and CSV. Semantic completion rejected the CSV because its upload had no verified publication receipt. After the managed run failed, Stop rejected its remaining assignment as a changed session. The driver also discarded the remote monitor's known failure reason. **Expected behavior** An authenticated upload of the exact requested bytes can satisfy normal completion. Stop can fence the remaining assignment of the selected failed run. An old Stop action cannot stop newer work. Known remote lease and reconnect errors remain visible without raw diagnostics or credentials. **Steps to reproduce** 1. Connect an experimental Grok Bot agent and verify its routine. 2. Assign a task that requires a document and a file. 3. Upload the file through its accepted assignment and cite the attachment in semantic completion. 4. Observe the missing publication receipt error on the previous implementation. 5. For recovery, select Stop after that run fails but its assignment remains active. **Paperclip version or commit** The live defects were observed after #16274. The current candidate merges master `739f9e40a4` and is `26c5b9601d1cb2b8f5bb7fcf730f019523763f5b`. ## What Changed - Record a server-owned publication receipt in the verified upload transaction. Bind it to the real native Grok run and exact stored file metadata. - Validate external publication against the actual upload and admitted assignment. Keep existing native and person upload rules. - Recover the committed receipt after a lost response. Reverify the same bytes when an older upload lacks proof. - Hold assignment authority through publication with grant-before-assignment and task-before-run lock order. Disconnect does not wait for remote file storage. - Allow Stop to fence the exact lingering assignment of a terminal run. Reject stale Stop calls for successor work. - Preserve only known lease/reconnect failure codes from the remote monitor. Do not expose raw stderr. - Extend real Rust, overlapping document/upload and disconnect/upload, file retry, completion and Stop regressions. Document setup, files and recovery. ## Verification - The real Rust Grok regression requests a file, checks an overlapping native document save, concurrent retries and lost-response recovery, rejects missing delivery evidence, then uses normal semantic completion and finalization. - Focused tests cover person MCP, native file handoff, completion, Stop and both external driver providers. - Local repository typecheck and production build pass on the current candidate. The five focused protocol suites pass (264 tests); all 17 external Runner driver tests pass, with the driver unchanged since the master merge. - Token gates and Storybook build pass. Nine existing Grok fixture journeys pass their browser interaction assertions; the static probe separately records five fixture API routes returning 404. This is UI fixture evidence only. - Full repository coverage passed in CI. The duplicate local `pnpm test:run` was interrupted after CI completed; its partial log is retained and is not reported as a local full-suite pass. The earlier aggregate invocation was interrupted for the second review fix. - All CI gates pass on `26c5b9601d1cb2b8f5bb7fcf730f019523763f5b`: 55 successful checks and two skipped Storybook jobs (no UI diff). This includes the complete sharded repository suites, native Runner, release checks, build and security scans. - Current-head Greptile is 5/5 with zero unresolved findings. The PR has no merge conflicts. - The prior live attempt is failed acceptance evidence. Real Grok requalification, proactive work, controller restart and active-session Stop remain release gates. This PR does not claim those passed. ## Risks - Upload and completion now share a durable run receipt. Lock order and current assignment authority must remain consistent with native mutations. - Publication fails closed if the file, run or grant changes. A retry must keep the same upload identity and bytes. - Person connections gain no agent authority. No database migration is required. - Known exit diagnostics do not prove global Grok termination. Grok spending remains unreported. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository tools, code execution and computer use. The exact deployment model ID and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
ad0bd26bb7
commit
51b56706fb
9 files changed
+392
-29
No files matched your search
@@ -16,7 +16,7 @@ import express, { type Request } from "express";
|
||||
import request from "supertest";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { grokBotRoutines, grokBotDispatches, agents, authUsers, companies, companyMemberships, createDb, dotAgentBindings, dotMailboxItems, dotRunnerAssignments, dotRunnerOperations,
|
||||
heartbeatRuns, agentWakeupRequests, issueComments, issues, nativeRunResults, nativeRunFinalizations, completionContracts, mcpEventDeliveries, mcpEventSubscriptions, workspaceOperations, workAssessments, statusDecisions, issueThreadInteractions } from "@paperclipai/db";
|
||||
heartbeatRuns, mcpOauthGrants, issueAttachments, agentWakeupRequests, issueComments, issues, nativeRunResults, nativeRunFinalizations, completionContracts, mcpEventDeliveries, mcpEventSubscriptions, workspaceOperations, workAssessments, statusDecisions, issueThreadInteractions } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { createPublicMcpOAuth, DEVICE_GRANT } from "../services/public-mcp/oauth.js";
|
||||
import { createPublicMcpExecutor } from "../services/public-mcp/capabilities.js";
|
||||
@@ -25,6 +25,7 @@ import { createPublicMcpEvents, type EventFetch } from "../services/public-mcp/e
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { canonicalNativeRuntimeContextDigest, type StrictCompletionContractInput } from "../vendor/paperclip-runner/index.js";
|
||||
import { createDotRunnerMcpTools, dotRunnerBroker } from "../services/dot-runner-broker.js";
|
||||
import { PaperclipRunnerToolAuthority } from "../services/native-runtime/paperclip-runner-tool-authority.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";
|
||||
@@ -451,6 +452,21 @@ describe("durable Dot Runner integration", () => {
|
||||
} finally { await f.events.stop(); }
|
||||
}, 30000);
|
||||
|
||||
it("fences a failed Grok run's remaining external assignment without stopping a successor", async () => {
|
||||
const f = await fixture({ provider: "grok_bot" });
|
||||
const old = await offeredWork(f);
|
||||
await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, old.run.id));
|
||||
expect(await f.broker.stopGrokWork(f.company.id, f.agent.id, f.userId, old.run.id))
|
||||
.toMatchObject({ stopped: true, remoteTerminationConfirmed: false });
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.runId, old.run.id)))[0]?.status).toBe("fenced");
|
||||
expect(await grokRoutineService(db).recoveryNeeded(f.company.id, f.agent.id)).toBe(true);
|
||||
await expect(f.broker.read(f.principal, old.assignment.id)).rejects.toThrow();
|
||||
const current = await offeredWork(f);
|
||||
await expect(f.broker.stopGrokWork(f.company.id, f.agent.id, f.userId, old.run.id)).rejects.toThrow("changed");
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, current.assignment.id)))[0]?.status).toBe("offered");
|
||||
await f.events.stop();
|
||||
});
|
||||
|
||||
it.each(["queued", "running"] as const)("stops a %s Grok run before an external assignment exists", async status => {
|
||||
const f = await fixture({ provider: "grok_bot" });
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status, invocationSource: "assignment", triggerDetail: "system" }).returning();
|
||||
@@ -752,18 +768,20 @@ describe("durable Dot Runner integration", () => {
|
||||
}, 30000);
|
||||
|
||||
it.each([
|
||||
{ name: "Dot local Rust", managed: false, provider: "openai_dot" as const },
|
||||
{ name: "Grok local Rust", managed: false, provider: "grok_bot" as const },
|
||||
...(process.env.PAPERCLIP_DOT_DAYTONA_LIVE === "1" ? [{ name: "private Daytona Rust", managed: true, provider: "openai_dot" as const }] : []),
|
||||
])("signed readiness, native tools and finalization through $name", async ({ managed, provider }) => {
|
||||
{ name: "Dot local Rust", managed: false, provider: "openai_dot" as const, revokeDuringUpload: false },
|
||||
{ name: "Grok local Rust", managed: false, provider: "grok_bot" as const, revokeDuringUpload: false },
|
||||
{ name: "Grok upload revocation Rust", managed: false, provider: "grok_bot" as const, revokeDuringUpload: true },
|
||||
...(process.env.PAPERCLIP_DOT_DAYTONA_LIVE === "1" ? [{ name: "private Daytona Rust", managed: true, provider: "openai_dot" as const, revokeDuringUpload: false }] : []),
|
||||
])("signed readiness, native tools and finalization through $name", async ({ managed, provider, revokeDuringUpload }) => {
|
||||
const remote = managed ? await (await import("./helpers/dot-daytona.js")).startDotDaytona(root) : null;
|
||||
try {
|
||||
const f = await fixture({ provider });
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Save the arithmetic report", description: "Save 17 + 25 = 42 as a task document.",
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Save the arithmetic report", description: provider === "grok_bot" ? "Save 17 + 25 = 42 as a task document and attach report.txt." : "Save 17 + 25 = 42 as a task document.",
|
||||
status: "in_progress", workMode: "standard", assigneeAgentId: f.agent.id }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status: "running", invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId: issue!.id } }).returning();
|
||||
await db.update(issues).set({ executionRunId: run!.id, checkoutRunId: run!.id }).where(eq(issues.id, issue!.id));
|
||||
const prepared = await prepareNativeHeartbeatRun({ db, run: run!, issue: issue!, environmentLeaseId: randomUUID() });
|
||||
await db.update(heartbeatRuns).set({ driverKind: provider === "grok_bot" ? "grok_bot_mcp" : "openai_dot_mcp" }).where(eq(heartbeatRuns.id, run!.id));
|
||||
await db.insert(nativeRunFinalizations).values({ runId: run!.id, companyId: f.company.id, issueId: issue!.id, phase: "observed" });
|
||||
const [boundRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.id, boundRun!.completionContractId!));
|
||||
@@ -818,12 +836,17 @@ describe("durable Dot Runner integration", () => {
|
||||
expect(await f.broker.operation(f.principal, assignment!.id, randomUUID(), "tool", { name: "write_document", arguments: {} })).toMatchObject({ status: "rejected" });
|
||||
await f.broker.operation(f.principal, assignment!.id, randomUUID(), "accept", {});
|
||||
expect(JSON.stringify(work)).not.toContain("fixture-gateway-secret");
|
||||
let uploadedDeliverableId: string | undefined;
|
||||
if (provider === "grok_bot") {
|
||||
const files = new Map<string, Buffer>();
|
||||
let storingUpload = false;
|
||||
let releaseUploadStorage!: () => void;
|
||||
const uploadStorageRelease = new Promise<void>(resolve => { releaseUploadStorage = resolve; });
|
||||
const transfers = createPublicMcpTransfers(db, f.oauth, async () => { throw new Error("Personal API dispatch forbidden"); }, {
|
||||
provider: "local_disk", putFile: async input => {
|
||||
if (!Buffer.isBuffer(input.body)) throw new Error("Expected bounded bytes");
|
||||
const objectKey = input.objectKey!; files.set(objectKey, input.body);
|
||||
if (revokeDuringUpload) { storingUpload = true; await uploadStorageRelease; }
|
||||
return { provider: "local_disk", objectKey, contentType: input.contentType, byteSize: input.body.length,
|
||||
sha256: createHash("sha256").update(input.body).digest("hex"), originalFilename: input.originalFilename };
|
||||
}, getObject: async (_company, key) => ({ stream: Readable.from(files.get(key)!) }),
|
||||
@@ -834,10 +857,88 @@ describe("durable Dot Runner integration", () => {
|
||||
const args = { companyId: f.company.id, taskId: issue!.id, requestId: uploadId, filename: "report.txt", contentType: "text/plain", byteSize: body.length, sha256: createHash("sha256").update(body).digest("hex") };
|
||||
const grant = await transfers.getUploadUrl(f.principal, args, assignment!.id);
|
||||
const url = new URL(String(grant.url));
|
||||
const uploaded = await request(api).put(url.pathname + url.search).set("Content-Type", "text/plain").send(body);
|
||||
if (revokeDuringUpload) {
|
||||
const uploading = request(api).put(url.pathname + url.search).set("Content-Type", "text/plain").send(body)
|
||||
.then(response => response);
|
||||
let revoked = false;
|
||||
let revoking: Promise<void> | undefined;
|
||||
try {
|
||||
await vi.waitFor(() => expect(storingUpload).toBe(true), { timeout: 10000, interval: 20 });
|
||||
revoking = f.broker.revoke(f.company.id, f.agent.id, f.userId).then(() => { revoked = true; });
|
||||
void revoking.catch(() => {});
|
||||
// Revocation must not wait for remote storage, and late bytes
|
||||
// cannot produce accepted artifacts or completion evidence.
|
||||
await vi.waitFor(() => expect(revoked).toBe(true), { timeout: 3000, interval: 20 });
|
||||
} finally { releaseUploadStorage(); }
|
||||
await revoking;
|
||||
expect((await uploading).status).toBe(403);
|
||||
await resultPromise.catch(() => {});
|
||||
const [revokedGrant] = await db.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, f.principal.grant.id));
|
||||
expect(revokedGrant!.revokedAt).not.toBeNull();
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, assignment!.id)))[0]!.status).toBe("fenced");
|
||||
expect(await db.select().from(issueAttachments).where(eq(issueAttachments.issueId, issue!.id))).toHaveLength(0);
|
||||
const [cancelledRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
expect(Object.keys((cancelledRun?.resultJson?.semanticToolReceipts ?? {}) as Record<string, unknown>)
|
||||
.filter(key => key.startsWith("external-upload:"))).toHaveLength(0);
|
||||
return;
|
||||
}
|
||||
// Hold the document transaction at its task lock, then start the real
|
||||
// HTTP upload. The native document write must still acquire its run
|
||||
// lock and commit; a run-before-task upload deadlocks this overlap.
|
||||
let concurrentUpload: Promise<request.Response> | undefined;
|
||||
let earlyUpload: request.Response | undefined;
|
||||
await db.transaction(async tx => {
|
||||
await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, issue!.id)).for("update");
|
||||
const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`) as unknown as Array<{ pid: number }>;
|
||||
concurrentUpload = request(api).put(url.pathname + url.search).set("Content-Type", "text/plain").send(body)
|
||||
.then(response => { earlyUpload = response; return response; });
|
||||
await vi.waitFor(async () => {
|
||||
expect(earlyUpload?.status, JSON.stringify(earlyUpload?.body)).toBeUndefined();
|
||||
const [state] = await db.execute(sql`select exists (
|
||||
select 1 from pg_stat_activity where ${backend!.pid} = any(pg_blocking_pids(pid))
|
||||
) as waiting`) as unknown as Array<{ waiting: boolean }>;
|
||||
expect(state!.waiting).toBe(true);
|
||||
}, { timeout: 10000, interval: 20 });
|
||||
// Use the same native authority inside this already-open document
|
||||
// transaction, retaining every real identity and mutation check.
|
||||
const documentDb = Object.assign(Object.create(tx), {
|
||||
transaction: async (effect: (database: typeof db) => Promise<unknown>) => effect(tx as unknown as typeof db),
|
||||
}) as typeof db;
|
||||
const authority = new PaperclipRunnerToolAuthority(documentDb, {
|
||||
companyId: f.company.id, issueId: issue!.id, agentId: f.agent.id, runId: run!.id, firstPartyOnly: true,
|
||||
});
|
||||
try {
|
||||
expect(await authority.execute({ tool: "write_document", callId: randomUUID(), arguments: {
|
||||
key: "upload-overlap", title: "Concurrent document", body: "Saved while the file is uploading.",
|
||||
baseRevisionId: null, idempotencyKey: "upload-overlap",
|
||||
} })).toMatchObject({ disposition: "applied" });
|
||||
} finally { authority.close(); }
|
||||
});
|
||||
const uploaded = await concurrentUpload!;
|
||||
expect(uploaded.status).toBe(200);
|
||||
const duplicate = await request(api).put(url.pathname + url.search).set("Content-Type", "text/plain").send(body);
|
||||
expect((await documentService(db).getIssueDocumentByKey(issue!.id, "upload-overlap"))?.body)
|
||||
.toBe("Saved while the file is uploading.");
|
||||
uploadedDeliverableId = uploaded.body.attachment.id;
|
||||
expect(uploaded.body.publication).toMatchObject({ evidenceRef: `deliverable:${uploadedDeliverableId}`,
|
||||
commandId: `deliverable-uploaded:${uploadedDeliverableId}`, source: "authenticated_upload" });
|
||||
const duplicates = await Promise.all([0, 1].map(() => request(api).put(url.pathname + url.search)
|
||||
.set("Content-Type", "text/plain").send(body)));
|
||||
expect(duplicates.every(response => response.status === 200)).toBe(true);
|
||||
expect(duplicates[0]!.body).toEqual(duplicates[1]!.body);
|
||||
const duplicate = duplicates[0]!;
|
||||
expect(duplicate.body.attachment.id).toBe(uploaded.body.attachment.id);
|
||||
expect(duplicate.body.publication).toEqual(uploaded.body.publication);
|
||||
expect(await transfers.getUploadUrl(f.principal, args, assignment!.id)).toMatchObject({
|
||||
status: "completed", publication: uploaded.body.publication, attachment: { id: uploadedDeliverableId },
|
||||
});
|
||||
const [publishedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
expect(Object.values(publishedRun!.resultJson!.semanticToolReceipts as Record<string, unknown>))
|
||||
.toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ operationId: "publish_external_deliverable" }),
|
||||
expect.objectContaining({ operationId: "write_document", input: expect.objectContaining({ key: "upload-overlap" }) }),
|
||||
]));
|
||||
expect(Object.keys(publishedRun!.resultJson!.semanticToolReceipts as Record<string, unknown>)
|
||||
.filter(key => key.startsWith("external-upload:"))).toHaveLength(1);
|
||||
expect(await issueService(db).getAttachmentById(uploaded.body.attachment.id)).toMatchObject({ createdByAgentId: f.agent.id, originatingRunId: run!.id, createdByUserId: null });
|
||||
await expect(transfers.getUploadUrl(f.principal, { ...args, taskId: randomUUID() }, assignment!.id)).rejects.toThrow();
|
||||
const download = new URL(String((await transfers.getDownloadUrl(f.principal, uploaded.body.attachment.id, assignment!.id)).url));
|
||||
@@ -894,7 +995,15 @@ describe("durable Dot Runner integration", () => {
|
||||
const result = { reportedWorkDisposition: "completed", summary: "Saved the arithmetic report.",
|
||||
completionClaim: { contractRevision: execution.completionContract.contract.revision, objectiveSatisfied: true,
|
||||
criteria: execution.completionContract.contract.criteria.map(c => ({ criterionId: c.id, status: "passed", evidenceRefs: [] })), remainingWork: [] },
|
||||
evidence: [], verification: [{ commandOrCheck: "17 + 25", status: "pass" }] };
|
||||
evidence: uploadedDeliverableId ? [{ ref: `deliverable:${uploadedDeliverableId}` }] : [], verification: [{ commandOrCheck: "17 + 25", status: "pass" }] };
|
||||
if (uploadedDeliverableId) {
|
||||
const missingEvidenceId = randomUUID();
|
||||
await f.broker.operation(f.principal, assignment!.id, missingEvidenceId, "tool", {
|
||||
name: "paperclip_finish", arguments: { ...result, evidence: [] },
|
||||
});
|
||||
await vi.waitFor(async () => expect(await f.broker.operationStatus(f.principal, assignment!.id, missingEvidenceId))
|
||||
.toMatchObject({ status: "completed", isError: true, result: { error: expect.stringContaining("requested file") } }), { timeout: 10000 });
|
||||
}
|
||||
const completionId = randomUUID();
|
||||
await f.broker.operation(f.principal, assignment!.id, completionId, "tool", { name: "paperclip_finish", arguments: result });
|
||||
await vi.waitFor(async () => expect(await f.broker.operationStatus(f.principal, assignment!.id, completionId)).toMatchObject({ status: "completed", isError: false }), { timeout: 10000 });
|
||||
|
||||
@@ -220,6 +220,13 @@ function createBroker(db: Db) {
|
||||
acceptByUnixMs: Date.now() + 10 * 60_000, expiresAtUnixMs: Date.now() + 2 * 60 * 60_000 };
|
||||
},
|
||||
async authorizeFileSession(principal: McpPrincipal, assignmentId: string, taskId: string, write: boolean) {
|
||||
if (write && principal.grant.purpose === "agent" && principal.grant.agentId) {
|
||||
// A transfer transaction retains this lock through publication. Stop
|
||||
// and terminal reconciliation cannot fence the assignment midway through
|
||||
// an authorized effect; authority is rechecked after acquiring the lock.
|
||||
await db.select({ id: assignments.id }).from(assignments).where(and(eq(assignments.id, assignmentId),
|
||||
eq(assignments.companyId, principal.grant.companyId), eq(assignments.agentId, principal.grant.agentId))).for("share");
|
||||
}
|
||||
const { binding, assignment } = await authorizeAssignment(principal, assignmentId);
|
||||
if (binding.providerKind !== "grok_bot" || assignment.status !== "accepted") throw fail("Files require a current accepted Grok work session.");
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, assignment.runId), eq(heartbeatRuns.companyId, binding.companyId)));
|
||||
@@ -505,12 +512,17 @@ function createBroker(db: Db) {
|
||||
const rows = await tx.select().from(assignments).where(and(eq(assignments.bindingId, binding.id), inArray(assignments.status, ["offered", "accepted"])));
|
||||
const [requestedRun] = expectedRunId ? await tx.select({ id: heartbeatRuns.id, status: heartbeatRuns.status }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.id, expectedRunId), eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId))) : [];
|
||||
if (expectedRunId && (!requestedRun || !["queued", "running"].includes(requestedRun.status))) throw fail("The work session changed. Reload before stopping it.");
|
||||
// A failed Runner can still leave its external assignment active.
|
||||
// Fence that exact assignment too; an old terminal run with no current
|
||||
// assignment must never stop a successor's work.
|
||||
const targeted = expectedRunId ? rows.filter(row => row.runId === expectedRunId) : rows;
|
||||
if (expectedRunId && (!requestedRun || !["queued", "running"].includes(requestedRun.status) && !targeted.length)) {
|
||||
throw fail("The work session changed. Reload before stopping it.");
|
||||
}
|
||||
const [previousAssignment] = expectedRunId ? await tx.select({ status: assignments.status }).from(assignments)
|
||||
.where(and(eq(assignments.bindingId, binding.id), eq(assignments.runId, expectedRunId))) : [];
|
||||
const otherWork = expectedRunId && rows.some(row => row.runId !== expectedRunId);
|
||||
if (otherWork && previousAssignment && !["offered", "accepted"].includes(previousAssignment.status)) throw fail("The work session changed. Reload before stopping it.");
|
||||
const targeted = expectedRunId ? rows.filter(row => row.runId === expectedRunId) : rows;
|
||||
const runIds = [...new Set([...targeted.map(row => row.runId), ...(requestedRun ? [requestedRun.id] : [])])];
|
||||
if (targeted.length) await tx.update(assignments).set({ status: "fenced" }).where(inArray(assignments.id, targeted.map(row => row.id)));
|
||||
// A queued run has no external authority. Preparing work with no other
|
||||
@@ -709,7 +721,7 @@ export function createDotRunnerMcpTools(db: Db, heartbeat?: DotHeartbeat, transf
|
||||
}));
|
||||
grokDefinitions.push({ name: "paperclip_grok_ack_wake", description: "Acknowledge an exact routine wake through this fixed agent connection. Echo its correlationId, bindingId and generation; for verification also echo its challenge. This records authenticated contact, never work acceptance, completion or remote termination.", schema: grokWakeAcknowledgementSchema });
|
||||
if (transfers) grokDefinitions.push(
|
||||
{ name: "paperclip_grok_get_upload_url", description: "Get a bounded ten-minute grant to attach exact file bytes to this accepted work session's task. Use your own HTTP/file tools, not a Paperclip filesystem. Compute SHA-256 and size, PUT those bytes, and keep requestId stable after a lost response. Revocation, pause, expiry and reassignment invalidate the grant.", schema: uploadSchema.omit({ companyId: true }).extend({ assignmentId: z.uuid() }) },
|
||||
{ name: "paperclip_grok_get_upload_url", description: "Get a bounded ten-minute grant to attach exact file bytes to this accepted work session's task. Use your own HTTP/file tools, not a Paperclip filesystem. Compute SHA-256 and size, PUT those bytes, and cite the returned publication.evidenceRef in your completion report. Keep requestId stable after a lost response. Revocation, pause, expiry and reassignment invalidate the grant.", schema: uploadSchema.omit({ companyId: true }).extend({ assignmentId: z.uuid() }) },
|
||||
{ name: "paperclip_grok_get_download_url", description: "Get a ten-minute download grant for one attachment of this accepted work session's task. Current session authority is rechecked on download; the grant is not a general file credential.", schema: z.object({ assignmentId: z.uuid(), attachmentId: z.uuid() }).strict() },
|
||||
);
|
||||
return {
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
|
||||
/** Lock the real Runner after the task lock and upload body verification.
|
||||
* Publication runs only after the stored attachment matches those bytes. The
|
||||
* caller holds the transfer's current work-session/grant transaction throughout. */
|
||||
export async function prepareExternalDeliverablePublication(db: Db, input: {
|
||||
companyId: string; issueId: string; runId: string; agentId: string;
|
||||
uploadId: string; workSessionId: string;
|
||||
filename: string; byteSize: number; sha256: string;
|
||||
}) {
|
||||
const [run] = await db.select({ resultJson: heartbeatRuns.resultJson }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.driverKind, "grok_bot_mcp"),
|
||||
eq(heartbeatRuns.status, "running"))).for("update");
|
||||
if (!run) throw new Error("external_deliverable_work_session_unavailable");
|
||||
const resultJson = run.resultJson ?? {};
|
||||
const receipts = { ...resultJson.semanticToolReceipts as Record<string, unknown> | undefined };
|
||||
const key = `external-upload:${input.uploadId}`;
|
||||
return async (attachmentId: string) => {
|
||||
const publication = {
|
||||
commandId: `deliverable-uploaded:${attachmentId}`, disposition: "applied",
|
||||
source: "authenticated_upload", attachmentId,
|
||||
entityRefs: [attachmentId], evidenceRef: `deliverable:${attachmentId}`,
|
||||
contentPath: `/api/attachments/${attachmentId}/content`,
|
||||
downloadPath: `/api/attachments/${attachmentId}/content?download=1`,
|
||||
};
|
||||
const receiptInput = { uploadId: input.uploadId, workSessionId: input.workSessionId,
|
||||
filename: input.filename, byteSize: input.byteSize, sha256: input.sha256 };
|
||||
const prior = receipts[key] as { operationId?: string; input?: Record<string, unknown>; result?: Record<string, unknown> } | undefined;
|
||||
if (prior) {
|
||||
if (prior.operationId !== "publish_external_deliverable" || prior.result?.commandId !== publication.commandId
|
||||
|| Object.entries(receiptInput).some(([name, value]) => prior.input?.[name] !== value)) {
|
||||
throw new Error("external_deliverable_publication_conflict");
|
||||
}
|
||||
} else {
|
||||
receipts[key] = { operationId: "publish_external_deliverable", input: receiptInput, result: publication };
|
||||
await db.update(heartbeatRuns).set({ resultJson: { ...resultJson, semanticToolReceipts: receipts }, updatedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, input.runId));
|
||||
}
|
||||
return publication;
|
||||
};
|
||||
}
|
||||
|
||||
/** A lost HTTP response can recover the committed proof without republishing. */
|
||||
export async function readExternalDeliverablePublication(db: Db, input: {
|
||||
companyId: string; issueId: string; runId: string; agentId: string;
|
||||
uploadId: string; workSessionId: string; attachmentId: string;
|
||||
}) {
|
||||
const [run] = await db.select({ resultJson: heartbeatRuns.resultJson }).from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId),
|
||||
eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.driverKind, "grok_bot_mcp"), eq(heartbeatRuns.status, "running")));
|
||||
const receipts = run?.resultJson?.semanticToolReceipts as Record<string, unknown> | undefined;
|
||||
const receipt = receipts?.[`external-upload:${input.uploadId}`] as {
|
||||
operationId?: string; input?: Record<string, unknown>; result?: Record<string, unknown>;
|
||||
} | undefined;
|
||||
if (receipt?.operationId !== "publish_external_deliverable" || receipt.input?.workSessionId !== input.workSessionId
|
||||
|| receipt.result?.commandId !== `deliverable-uploaded:${input.attachmentId}`) return undefined;
|
||||
return receipt.result;
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { assets, documents, documentRevisions, heartbeatRuns, issueDocuments, issueAttachments, issueWorkProducts, type Db } from "@paperclipai/db";
|
||||
import { assets, documents, documentRevisions, heartbeatRuns, issueDocuments, issueAttachments, issueWorkProducts, mcpAttachmentUploads, dotRunnerAssignments, type Db } from "@paperclipai/db";
|
||||
import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js";
|
||||
|
||||
function evidenceRefs(value: unknown): string[] {
|
||||
@@ -17,7 +17,7 @@ function record(value: unknown): Record<string, unknown> {
|
||||
}
|
||||
|
||||
async function hasCurrentPublicationReceipt(db: Db, companyId: string, receipts: unknown, attachment: {
|
||||
id: string; filename: string | null; byteSize: number; sha256: string;
|
||||
id: string; originatingRunId: string | null; filename: string | null; byteSize: number; sha256: string;
|
||||
}): Promise<boolean> {
|
||||
for (const value of Object.values(record(receipts))) {
|
||||
const receipt = record(value);
|
||||
@@ -25,6 +25,22 @@ async function hasCurrentPublicationReceipt(db: Db, companyId: string, receipts:
|
||||
const result = record(receipt.result);
|
||||
if ((result.disposition !== "applied" && result.disposition !== "duplicate") ||
|
||||
!Array.isArray(result.entityRefs) || result.entityRefs[0] !== attachment.id) continue;
|
||||
if (receipt.operationId === "publish_external_deliverable" && result.source === "authenticated_upload"
|
||||
&& result.commandId === `deliverable-uploaded:${attachment.id}` && typeof input.uploadId === "string"
|
||||
&& typeof input.workSessionId === "string" && attachment.originatingRunId && attachment.filename
|
||||
&& /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu.test(input.uploadId)
|
||||
&& /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu.test(input.workSessionId)
|
||||
&& input.filename === attachment.filename
|
||||
&& input.byteSize === attachment.byteSize && input.sha256 === attachment.sha256) {
|
||||
const [upload] = await db.select({ id: mcpAttachmentUploads.id }).from(mcpAttachmentUploads)
|
||||
.innerJoin(dotRunnerAssignments, and(eq(dotRunnerAssignments.id, mcpAttachmentUploads.workSessionId),
|
||||
eq(dotRunnerAssignments.companyId, companyId)))
|
||||
.where(and(eq(mcpAttachmentUploads.id, input.uploadId), eq(mcpAttachmentUploads.companyId, companyId),
|
||||
eq(mcpAttachmentUploads.workSessionId, input.workSessionId), eq(mcpAttachmentUploads.attachmentId, attachment.id),
|
||||
eq(dotRunnerAssignments.runId, attachment.originatingRunId), eq(mcpAttachmentUploads.originalFilename, attachment.filename),
|
||||
eq(mcpAttachmentUploads.byteSize, attachment.byteSize), eq(mcpAttachmentUploads.sha256, attachment.sha256))).limit(1);
|
||||
if (upload) return true;
|
||||
}
|
||||
if (receipt.operationId === "register_deliverable" &&
|
||||
result.commandId === `deliverable-prepared:${attachment.id}` &&
|
||||
typeof input.filename === "string" && input.filename.trim() === attachment.filename &&
|
||||
|
||||
@@ -8,6 +8,7 @@ import { MAX_ATTACHMENT_BYTES, isAllowedContentType } from "../../attachment-typ
|
||||
import type { StorageService } from "../../storage/types.js";
|
||||
import { authorizationService } from "../authorization.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { prepareExternalDeliverablePublication, readExternalDeliverablePublication } from "../native-runtime/external-deliverable-publication.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
import { hashMcpSecret, type PublicMcpOAuth, type McpPrincipal } from "./oauth.js";
|
||||
import { McpCapabilityError, company, task, requestId, pick, object, stable, type ApiDispatch, type Capability } from "./contracts.js";
|
||||
@@ -120,29 +121,44 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api
|
||||
const bytes = req.body;
|
||||
if (!Buffer.isBuffer(bytes)) throw unavailable();
|
||||
const digest = createHash("sha256").update(bytes).digest("hex");
|
||||
const attachment = await db.transaction(async tx => {
|
||||
const published = await db.transaction(async tx => {
|
||||
const [upload] = await tx.select().from(mcpAttachmentUploads).where(eq(mcpAttachmentUploads.id, row.uploadId!)).for("update");
|
||||
if (!upload || upload.companyId !== principal.grant.companyId || upload.userId !== principal.grant.userId || upload.workSessionId !== row.workSessionId || upload.grantId && upload.grantId !== row.grantId || upload.storageProvider !== storage.provider) throw unavailable();
|
||||
if (upload.sha256 !== digest || upload.byteSize !== bytes.length || req.get("Content-Type")?.split(";")[0]?.trim().toLowerCase() !== upload.contentType) throw new McpCapabilityError("Upload bytes, size or Content-Type do not match the requested file.");
|
||||
await taskAccess(principal, upload.taskId, true, tx as unknown as Db, row.workSessionId ?? undefined);
|
||||
if (upload.attachmentId) return attachmentAccess(principal, upload.attachmentId, tx as unknown as Db, row.workSessionId ?? undefined);
|
||||
const stored = await storage.putFile({ companyId: upload.companyId, namespace: "mcp-transfers", objectKey: upload.objectKey, originalFilename: upload.originalFilename, contentType: upload.contentType, body: bytes });
|
||||
// Check access before storage without retaining an agent assignment
|
||||
// lock. Publication locks grant then assignment, matching disconnect;
|
||||
// remote storage must not block revocation. Keep person write checks.
|
||||
await taskAccess(principal, upload.taskId, principal.grant.purpose !== "agent", tx as unknown as Db, row.workSessionId ?? undefined);
|
||||
const stored = upload.attachmentId ? null : await storage.putFile({ companyId: upload.companyId, namespace: "mcp-transfers", objectKey: upload.objectKey, originalFilename: upload.originalFilename, contentType: upload.contentType, body: bytes });
|
||||
// Recheck after receiving/storing bytes. Lock the grant against concurrent revocation.
|
||||
const [grant] = await tx.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, row.grantId)).for("update");
|
||||
if (!grant || grant.revokedAt || row.expiresAt <= new Date()) throw unavailable();
|
||||
const current = await oauth.authorizeGrant(row.grantId, tx as unknown as Db);
|
||||
if (current.grant.purpose !== "agent") requireWrite(current);
|
||||
await taskAccess(current, upload.taskId, true, tx as unknown as Db, row.workSessionId ?? undefined);
|
||||
const session = await workSession(current, row.workSessionId ?? undefined, upload.taskId, true, tx as unknown as Db);
|
||||
// Match native mutations: lock the task before the run, then recheck
|
||||
// assignment and governance against the locked current task.
|
||||
const [issue] = await tx.select().from(issues).where(and(eq(issues.id, upload.taskId), eq(issues.companyId, upload.companyId))).for("update");
|
||||
if (!issue || (current.grant.purpose !== "agent" && issue.conversationAgentId && issue.conversationUserId !== current.grant.userId)) throw unavailable();
|
||||
const session = await workSession(current, row.workSessionId ?? undefined, upload.taskId, true, tx as unknown as Db);
|
||||
const result = await issueService(tx as unknown as Db).createAttachment({ issueId: upload.taskId, ...stored, createdByUserId: session ? null : current.grant.userId, createdByAgentId: session?.agentId, createdByRunId: session?.runId });
|
||||
await workSession(current, row.workSessionId ?? undefined, upload.taskId, true, tx as unknown as Db);
|
||||
const publish = session ? await prepareExternalDeliverablePublication(tx as unknown as Db, {
|
||||
companyId: upload.companyId, issueId: upload.taskId, ...session, uploadId: upload.id,
|
||||
workSessionId: row.workSessionId!, filename: upload.originalFilename, byteSize: bytes.length, sha256: digest,
|
||||
}) : undefined;
|
||||
const result = upload.attachmentId
|
||||
? await attachmentAccess(current, upload.attachmentId, tx as unknown as Db, row.workSessionId ?? undefined)
|
||||
: await issueService(tx as unknown as Db).createAttachment({ issueId: upload.taskId, ...stored!, createdByUserId: session ? null : current.grant.userId, createdByAgentId: session?.agentId, createdByRunId: session?.runId });
|
||||
if (result.originalFilename !== upload.originalFilename || result.byteSize !== bytes.length || result.sha256 !== digest
|
||||
|| session && (result.originatingRunId !== session.runId || result.createdByAgentId !== session.agentId)) throw unavailable();
|
||||
const publication = publish ? await publish(result.id) : undefined;
|
||||
if (upload.attachmentId) return { attachment: result, publication };
|
||||
await tx.update(mcpAttachmentUploads).set({ attachmentId: result.id }).where(eq(mcpAttachmentUploads.id, upload.id));
|
||||
await logActivity(tx as unknown as Db, { companyId: upload.companyId, actorType: session ? "agent" : "user", actorId: session?.agentId ?? current.grant.userId, action: "issue.attachment_added", entityType: "issue", entityId: upload.taskId,
|
||||
details: { attachmentId: result.id, originalFilename: result.originalFilename, contentType: result.contentType, byteSize: result.byteSize, mcpConnectionId: row.grantId, ...(session ? { workSessionId: row.workSessionId, authorizingPersonId: current.grant.userId } : {}) } });
|
||||
return result;
|
||||
return { attachment: result, publication };
|
||||
});
|
||||
res.json({ attachment: pick(attachment, attachmentFields) });
|
||||
res.json({ attachment: pick(published.attachment, attachmentFields), ...(published.publication ? { publication: published.publication } : {}) });
|
||||
});
|
||||
router.get(prefix + "/download", async (req, res) => {
|
||||
const { row, principal } = await authorize(req.query.ticket, false);
|
||||
@@ -181,9 +197,20 @@ export function createPublicMcpTransfers(db: Db, oauth: PublicMcpOAuth, api: Api
|
||||
await tx.update(mcpAttachmentUploads).set({ expiresAt: new Date(Date.now() + lifetime), cleanedAt: null }).where(eq(mcpAttachmentUploads.id, found.id));
|
||||
return found;
|
||||
});
|
||||
if (upload.attachmentId) return { attachment: pick(await attachmentAccess(p, upload.attachmentId, undefined, sessionId), attachmentFields), status: "completed" };
|
||||
if (upload.attachmentId) {
|
||||
const attachment = await attachmentAccess(p, upload.attachmentId, undefined, sessionId);
|
||||
const session = await workSession(p, sessionId, upload.taskId, true);
|
||||
const publication = session ? await readExternalDeliverablePublication(db, {
|
||||
companyId: upload.companyId, issueId: upload.taskId, ...session,
|
||||
uploadId: upload.id, workSessionId: sessionId!, attachmentId: attachment.id,
|
||||
}) : undefined;
|
||||
if (!session || publication) return { attachment: pick(attachment, attachmentFields), status: "completed",
|
||||
...(publication ? { publication } : {}) };
|
||||
// An older upload can lack publication proof. Verify the same bytes
|
||||
// through another bounded PUT; never manufacture a receipt from metadata.
|
||||
}
|
||||
return { ...await ticket(p, { uploadId: upload.id }, sessionId), method: "PUT", headers: { "Content-Type": a.contentType }, uploadId: upload.id, maxBytes: MAX_ATTACHMENT_BYTES,
|
||||
instructions: "Send the exact file bytes with the given Content-Type. Success saves the attachment and returns its ID; there is no completion tool. Do not share this temporary file credential. Reuse requestId after a lost response." };
|
||||
instructions: "Send the exact file bytes with the given Content-Type. Success saves the attachment and returns its ID. An agent upload also returns a verified publication receipt; cite its evidenceRef in your completion report. There is no separate upload completion tool. Do not share this temporary file credential. Reuse requestId after a lost response." };
|
||||
},
|
||||
async getDownloadUrl(p: McpPrincipal, attachmentId: string, sessionId?: string) {
|
||||
const attachment = await attachmentAccess(p, attachmentId, undefined, sessionId);
|
||||
|
||||
Reference in new issue
Block a user