mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
fix: repair MCP validation and cancelled execution recovery (#14951)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The tool gateway gives agents access to connected services. Recovery controls what happens when a run stops. > - Generated tool names can exceed the provider limit after the MCP client adds its prefix. > - The same invalid definition can fail each automatic retry. A cancelled run can also hold saved messages without showing its cause. > - This pull request bounds tool names, stops configuration retries, and retains cancellation evidence. > - It shows the stopped run and admits saved input only after the existing safety checks pass. > - The benefit is a clear recovery path that preserves operator Stop and prevents duplicate message delivery. ## Linked Issues or Issue Description **What happened?** A long connected MCP tool name makes the provider reject the entire request. Automatic recovery repeats the invalid request. Separately, unexpected legacy cancellations can leave saved input behind a recovery hold. The notice does not identify the stopped run or its cause. **Expected behavior** Complete MCP names fit the provider limit. Tool-definition errors require configuration repair. Cancelled runs retain their source and reason. The recovery notice shows the cause and saved-message count. Verified unexpected cancellations can start a fresh turn through the existing admission checks. **Steps to reproduce** 1. Assign an App gallery connection with a long application key and tool name to a Claude agent. 2. Start a run. The provider rejects a name over 128 characters, including its MCP prefix. 3. For cancellation recovery, stop a legacy provider turn without an operator Stop request and send a user message while the recovery hold is active. 4. Inspect the recovery notice and the deferred message queue. **Paperclip version or commit** Rebased onto master at `cf8ad63c806685bfd7c48e3ed4a919d61a7c55f1`. **Deployment mode** Hosted or self-hosted server with legacy Claude or Codex execution. Related public work: - Refs #14017. That PR caps name segments. This PR preserves existing short names and uses stable hash aliases for long complete names. It also covers classification and recovery. - Refs #4510. That PR adds a cancellation-source column. This PR records bounded evidence in the existing run result, without a migration. - Refs #12552 and #4506. Those PRs suppress recovery after operator cancellation. This PR preserves operator intent and uses the existing continuation gates. ## What Changed - Bound gateway names with the full provider prefix in the 128-character budget. Retain the original upstream tool name for dispatch and permissions. - Classify invalid tool definitions as configuration failures before diagnostic redaction. Stop automatic retries and continuation attempts for that error code. - Persist cancellation source, expectedness, initiator, reason, and time. Preserve recorded Stop intent when adapter results arrive. Report unexpected started cancellations with closed diagnostic labels. - Show the run cause, saved-message count, and Inspect run link. Offer Continue for eligible unexpected cancellations. Require verified provider stop, empty tool inventory, ownership, and the existing pause, budget, approval, and dependency gates. Use the existing queue for single delivery. - Add regression coverage and update the execution, MCP gateway, and run-log documentation. ## Verification - `pnpm -r typecheck` and `pnpm build` passed. - `pnpm check:token-gates` passed. - Ran `pnpm test:run` and completed its workspace and serialized groups. Initial resource and timing failures passed on isolated reruns. All 149 serialized route suites passed. - Reran the changed server, adapter, and UI suites after the rebase. Coverage includes long-name upstream dispatch, configuration retry suppression, cancellation evidence retention, privacy labels, oversized run projection, and concurrent saved-message delivery. - `pnpm test:e2e tests/e2e/legacy-failure-continuation.spec.ts` passed all six browser scenarios. The recovery notice shows the run cause and inspection link, and each recovery entry point reaches one new response. - Added database-backed checks for active, removed, paused, unavailable, and disabled chat connections. The final continuation and recovery-notice suites passed 167 tests. Externally bound chats hide board Continue and show a usable next action. - All 55 GitHub checks passed on `42afbf1371dcaeb72646e3d8f65c19ff7cddf8de`. Two unrelated Storybook jobs were skipped by their normal conditions. Greptile reviewed that commit at 5/5 with no findings and no open review threads. ## Risks - Long tool names change to aliases. Existing short names stay compatible. The original connection and upstream name remain the dispatch authority. - Invalid tool definitions no longer get automatic retries. An operator must repair the configuration before a new attempt. - Continuation changes apply only to positively identified unexpected legacy cancellations with complete empty tool inventory. Operator Stop, unknown historical cancellations, outstanding tools, and unverified provider termination keep their holds. - No database migration. The added projection fields are optional. Cancellation reason and initiator IDs remain local run evidence; Sentry receives only closed source and initiator-type labels and expectedness. ## Model Used - OpenAI GPT-6 through Codex, with reasoning, repository editing, shell execution, and GitHub tool use. The runtime does not expose the exact model variant or context-window size. ## 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
7d59de6113
commit
7a52dcdc74
32 files changed
+640
-48
No files matched your search
@@ -196,4 +196,26 @@ describeEmbeddedPostgres("direct adapter native-runner isolation", () => {
|
||||
expect(nativeRows.every((rows) => rows.length === 0)).toBe(true);
|
||||
},
|
||||
);
|
||||
|
||||
it("retains an unsolicited provider cancellation through finalization", async () => {
|
||||
const companyId = randomUUID(), agentId = randomUUID();
|
||||
execute.mockImplementation(async (context) => {
|
||||
expect(context.signal?.aborted).toBe(false);
|
||||
return { exitCode: 1, signal: null, timedOut: false, errorMessage: "ACP turn cancelled",
|
||||
resultJson: { status: "cancelled", acpToolInventoryComplete: true, acpPendingToolCount: 0 } };
|
||||
});
|
||||
await db.insert(companies).values({ id: companyId, name: "Provider stop",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "responsible-user" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Claude", role: "engineer", status: "idle",
|
||||
adapterType: "claude_local", adapterConfig: {}, runtimeConfig: {}, permissions: {} });
|
||||
const queued = await heartbeat.invoke(agentId, "on_demand", {}, "manual");
|
||||
const finished = await waitForRunToFinish(heartbeat, queued!.id);
|
||||
expect(finished).toMatchObject({ status: "cancelled", runtimeMode: "legacy", resultJson: {
|
||||
cancellation: { source: "provider", expected: false, initiator: { type: "provider" },
|
||||
reason: expect.stringContaining("Provider cancelled execution") },
|
||||
acpToolInventoryComplete: true, acpPendingToolCount: 0,
|
||||
} });
|
||||
expect(execute).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
@@ -288,6 +288,12 @@ describeEmbeddedPostgres("heartbeat list", () => {
|
||||
privateSyncMetadata: oversizedNestedPayload,
|
||||
},
|
||||
workspaceRestoreFailure: "restore_unsafe_archive",
|
||||
cancellation: { source: "provider", expected: false, initiator: { type: "provider" },
|
||||
reason: "Provider cancelled execution ".repeat(50), recordedAt: "2026-10-02T15:00:00.000Z",
|
||||
privateMetadata: oversizedNestedPayload },
|
||||
acpToolInventoryComplete: true,
|
||||
acpPendingToolCount: 0,
|
||||
errorFamily: "configuration",
|
||||
finalResponseRecorded: true,
|
||||
executionBeforeRestore: { errorCode: "model_error", exitCode: 2, timedOut: false },
|
||||
},
|
||||
@@ -312,6 +318,11 @@ describeEmbeddedPostgres("heartbeat list", () => {
|
||||
storageWarning: "Agent storage is full. Runs can continue.".repeat(50).slice(0, 1024),
|
||||
},
|
||||
workspaceRestoreFailure: "restore_unsafe_archive",
|
||||
cancellation: { source: "provider", expected: false, initiator: { type: "provider" },
|
||||
reason: "Provider cancelled execution ".repeat(50).slice(0, 512), recordedAt: "2026-10-02T15:00:00.000Z" },
|
||||
acpToolInventoryComplete: true,
|
||||
acpPendingToolCount: 0,
|
||||
errorFamily: "configuration",
|
||||
finalResponseRecorded: true,
|
||||
executionBeforeRestore: { errorCode: "model_error", exitCode: 2, timedOut: false },
|
||||
});
|
||||
@@ -319,6 +330,7 @@ describeEmbeddedPostgres("heartbeat list", () => {
|
||||
expect((result?.stdout as string).length).toBeLessThan(oversizedStdout.length);
|
||||
expect(result).not.toHaveProperty("nestedHuge");
|
||||
expect(result?.instructionSave).not.toHaveProperty("privateSyncMetadata");
|
||||
expect(result?.cancellation).not.toHaveProperty("privateMetadata");
|
||||
expect(result?.terminalSessionFailure).not.toHaveProperty("privateMetadata");
|
||||
const diagnostic = result?.terminalSessionFailure as { details: string };
|
||||
expect(diagnostic.details).toContain("[truncated for run retrieval; full text in run error/transcript]");
|
||||
|
||||
@@ -251,6 +251,16 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
|
||||
}
|
||||
|
||||
|
||||
it("never schedules invalid provider definitions even when a caller supplies a retry policy", async () => {
|
||||
const runId = randomUUID(), companyId = randomUUID(), agentId = randomUUID();
|
||||
const now = new Date("2026-04-20T12:00:00.000Z");
|
||||
await seedRetryFixture({ runId, companyId, agentId, now, errorCode: "provider_tool_definition_invalid",
|
||||
errorFamily: "transient_upstream" });
|
||||
expect(await heartbeat.scheduleBoundedRetry(runId, { now, retryReason: "transient_failure", maxAttempts: 9 }))
|
||||
.toMatchObject({ outcome: "not_scheduled", reason: expect.stringContaining("Repair") });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId))).toHaveLength(0);
|
||||
});
|
||||
|
||||
it.each(["restore_unsafe_archive", "restore_lock_timeout"])("keeps the existing retry budget for %s", async (classification) => {
|
||||
const runId = randomUUID(), companyId = randomUUID(), agentId = randomUUID();
|
||||
const now = new Date("2026-04-20T12:00:00.000Z");
|
||||
|
||||
@@ -2077,6 +2077,31 @@ rl.on("line", (line) => {
|
||||
]));
|
||||
});
|
||||
|
||||
it("dispatches a bounded alias to the original long upstream tool name", async () => {
|
||||
const company = await createCompany(db);
|
||||
const agent = await createAgent(db, company.id);
|
||||
const { run } = await createIssueAndRun(db, company.id, agent.id);
|
||||
const upstreamName = "file_download_batch_exports_count_rows_create";
|
||||
const fake = await startFakeRemoteMcpServer((request) => {
|
||||
expect(request.body).toMatchObject({ method: "tools/call", params: { name: upstreamName, arguments: {} } });
|
||||
return { body: { jsonrpc: "2.0", id: request.body?.id, result: { content: [{ type: "text", text: "ok" }] } } };
|
||||
});
|
||||
try {
|
||||
const { catalogEntry } = await createRemoteMcpTool(db, company.id, {
|
||||
applicationKey: `app-gallery:posthog:${randomUUID()}`, toolName: upstreamName, url: fake.url,
|
||||
});
|
||||
await allowAllToolsForAgent(db, company.id, agent.id);
|
||||
const gateway = createTestToolGatewayService(db);
|
||||
const session = await gateway.createSession({ companyId: company.id, agentId: agent.id, runId: run.id });
|
||||
const tool = (await gateway.listToolsForSession(session.token)).find((entry) => entry.catalogEntryId === catalogEntry.id)!;
|
||||
expect(`mcp__paperclip-assigned__${tool.name}`.length).toBeLessThanOrEqual(128);
|
||||
expect((await gateway.listToolsForSession(session.token)).find((entry) => entry.catalogEntryId === catalogEntry.id)?.name).toBe(tool.name);
|
||||
await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: {} });
|
||||
} finally {
|
||||
await fake.close();
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
["local_trusted", { deploymentMode: "local_trusted" as const, deploymentExposure: "private" as const }],
|
||||
["authenticated/private", { deploymentMode: "authenticated" as const, deploymentExposure: "private" as const }],
|
||||
|
||||
@@ -13,6 +13,7 @@ import { assertAiConnectionCreateAccess, canInstallSharedAiConnectionForNewAgent
|
||||
import { isAiConnectionCompatible } from "@paperclipai/shared";
|
||||
import { applyConnectorSkills, resolveConnectorAssignments, annotateConnectorSkills, isConnectorSkill } from "../services/connector-runtime.js";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
import { canContinueCancelledRun } from "../services/run-cancellation.js";
|
||||
import { paperclipRunnerTransitionConfig, normalizeLegacyRunnerProvider, isPaperclipRunnerProvider } from "@paperclipai/adapter-utils";
|
||||
import { executionProjectionForRun, executionProjectionsForRuns } from "../services/execution-projection.js";
|
||||
import { selectDashboardRunIds } from "../services/dashboard-run-selection.js";
|
||||
@@ -6005,8 +6006,8 @@ export function agentRoutes(
|
||||
) {
|
||||
throw notFound("Failed run not found");
|
||||
}
|
||||
if (!["failed", "timed_out"].includes(failedRun.status)) {
|
||||
throw conflict("Only a failed run can be retried.");
|
||||
if (!["failed", "timed_out"].includes(failedRun.status) && !canContinueCancelledRun(failedRun)) {
|
||||
throw conflict("Only a failed run or verified unexpected cancellation can start a new attempt.");
|
||||
}
|
||||
if (failedRun.runtimeMode === "native" && failedRun.errorCode === "native_session_cleanup_quarantined") {
|
||||
throw conflict("The stopped native session requires cleanup and reconciliation before a new attempt.", {
|
||||
@@ -6018,6 +6019,10 @@ export function agentRoutes(
|
||||
typeof failedContext.issueId === "string"
|
||||
? failedContext.issueId
|
||||
: null;
|
||||
if (failedRun.status === "cancelled" && (!issueId ||
|
||||
(await getExecutionBlocker(db, agent.companyId, issueId))?.runId !== failedRun.id)) {
|
||||
throw conflict("The stopped run no longer owns this task's recovery hold.");
|
||||
}
|
||||
if (issueId) {
|
||||
const issue = await issueService(db).getById(issueId);
|
||||
if (!issue || issue.companyId !== agent.companyId) throw notFound("Task not found");
|
||||
@@ -6073,6 +6078,9 @@ export function agentRoutes(
|
||||
.then((rows) => rows[0])
|
||||
: null;
|
||||
if (chatBinding) {
|
||||
if (failedRun.status === "cancelled") {
|
||||
throw conflict("Send a new chat message to continue this stopped conversation.");
|
||||
}
|
||||
if (!options.chatRunRetries || !req.actor.userId) {
|
||||
throw conflict("Chat retry authorization is unavailable.", {
|
||||
code: "chat_failed_run_retry_requires_authorized_context",
|
||||
|
||||
@@ -126,7 +126,7 @@ export function captureException(error: unknown): void {
|
||||
}
|
||||
|
||||
/** The run status values that mark a run as a genuine terminal failure. */
|
||||
export type RunFailureStatus = "failed" | "timed_out";
|
||||
export type RunFailureStatus = "failed" | "timed_out" | "cancelled";
|
||||
|
||||
/**
|
||||
* The diagnostic values `captureRunFailure` sends with a terminal-failure
|
||||
|
||||
@@ -261,6 +261,25 @@ describeEmbeddedPostgres("reportRunFailure", () => {
|
||||
|
||||
expect(mockCaptureRunFailure).not.toHaveBeenCalled();
|
||||
});
|
||||
it.each(["provider", "unknown"])("reports an unexpected started cancellation from %s", async source => {
|
||||
await seedCompanyAndAgent();
|
||||
const run = buildRun({ status: "cancelled", startedAt: new Date(0), finishedAt: new Date(1000),
|
||||
resultJson: { cancellation: { source, expected: false, initiator: { type: "provider", id: "private-actor" },
|
||||
reason: "private-reason", recordedAt: new Date(1000).toISOString() } } });
|
||||
await reportRunFailure(db, run);
|
||||
expect(mockCaptureRunFailure).toHaveBeenCalledOnce();
|
||||
expect(mockCaptureRunFailure.mock.calls[0][0]).toMatchObject({ runStatus: "cancelled",
|
||||
diagnostics: { execution: { cancellationSource: source, cancellationExpected: false } } });
|
||||
expect(JSON.stringify(mockCaptureRunFailure.mock.calls)).not.toMatch(/private-actor|private-reason/);
|
||||
});
|
||||
it("does not report an operator's Stop as a failure", async () => {
|
||||
await seedCompanyAndAgent();
|
||||
await reportRunFailure(db, buildRun({ status: "cancelled", startedAt: new Date(0), resultJson: {
|
||||
cancellation: { source: "operator", expected: true, initiator: { type: "user", id: "board" },
|
||||
reason: "Stop", recordedAt: new Date().toISOString() },
|
||||
} }));
|
||||
expect(mockCaptureRunFailure).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("sends agents.adapterType from the loaded agent row", async () => {
|
||||
await seedCompanyAndAgent({ adapterType: "claude_managed" });
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { and, desc, eq, gt, inArray, not, or, sql } from "drizzle-orm";
|
||||
import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js";
|
||||
import { z } from "zod";
|
||||
import { heartbeatRuns, issueComments, issues, issueRecoveryActions, type Db } from "@paperclipai/db";
|
||||
import { agentWakeupRequests, chatConversations, chatEndpoints, heartbeatRuns, issueComments, issues, issueRecoveryActions, toolConnections, type Db } from "@paperclipai/db";
|
||||
import { canContinueCancelledRun, readRunCancellation } from "./run-cancellation.js";
|
||||
import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js";
|
||||
import { EXECUTION_RECONCILIATION_CAUSES, type ExecutionBlocker } from "@paperclipai/shared";
|
||||
|
||||
/** Resolved recovery bookkeeping can still carry an effective no-replay hold. */
|
||||
@@ -16,7 +18,8 @@ export function executionBlockerPredicate() {
|
||||
|
||||
export async function getExecutionBlocker(db: Db, companyId: string, issueId: string, options?: { conversationResetCommentId?: string | null }): Promise<ExecutionBlocker | null> {
|
||||
const [conversation] = await db.select({ agentId: issues.conversationAgentId,
|
||||
boundaryId: issues.conversationBoundaryCommentId }).from(issues).where(and(
|
||||
boundaryId: issues.conversationBoundaryCommentId, status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId }).from(issues).where(and(
|
||||
eq(issues.companyId, companyId), eq(issues.id, issueId),
|
||||
)).limit(1);
|
||||
// Resetting model context cannot make an unsafe workspace safe. This hold
|
||||
@@ -53,9 +56,56 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st
|
||||
if (!action) return null;
|
||||
const parsedRunId = z.string().guid().safeParse(action.evidence.runId ?? action.evidence.sourceRunId);
|
||||
const runId = parsedRunId.success ? parsedRunId.data : null;
|
||||
const [run] = runId ? await db.select({ agentId: heartbeatRuns.agentId }).from(heartbeatRuns).where(and(
|
||||
const [run] = runId ? await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId),
|
||||
)).limit(1) : [];
|
||||
const queue = await db.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
));
|
||||
const savedIds = [...new Set(queue.flatMap(entry => {
|
||||
const context = entry.payload?._paperclipWakeContext as Record<string, unknown> | undefined;
|
||||
return [...queuedCommentIdsFromWakePayload(entry.payload), entry.payload?.commentId, context?.wakeCommentId]
|
||||
.filter((id): id is string => z.string().guid().safeParse(id).success);
|
||||
}))];
|
||||
const saved = savedIds.length ? await db.select({ id: issueComments.id }).from(issueComments).where(and(
|
||||
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
||||
inArray(issueComments.id, savedIds), eq(issueComments.authorType, "user"), sql`${issueComments.deletedAt} is null`,
|
||||
sql`not exists (select 1 from ${heartbeatRuns} consumed where consumed.company_id = ${companyId}
|
||||
and consumed.context_snapshot->>'issueId' = ${issueId}
|
||||
and not (consumed.status = 'cancelled' and consumed.started_at is null)
|
||||
and (consumed.context_snapshot->>'wakeCommentId' = ${issueComments.id}::text
|
||||
or consumed.context_snapshot->'wakeCommentIds' @> jsonb_build_array(${issueComments.id}::text)))`,
|
||||
)) : [];
|
||||
const cancellation = run ? readRunCancellation(run.resultJson) : null;
|
||||
const runError = cancellation?.reason ?? (run?.status === "cancelled"
|
||||
? "Execution was cancelled; its source was not recorded."
|
||||
: run?.error);
|
||||
const eligibleContinuation = Boolean(run && conversation?.assigneeAgentId === run.agentId &&
|
||||
!["done", "cancelled"].includes(conversation.status) && canContinueCancelledRun(run));
|
||||
const [chatBinding] = eligibleContinuation ? await db.select({
|
||||
state: chatConversations.state, endpointStatus: chatEndpoints.status,
|
||||
connectionStatus: toolConnections.status, connectionEnabled: toolConnections.enabled,
|
||||
}).from(chatConversations).leftJoin(chatEndpoints, and(
|
||||
eq(chatEndpoints.companyId, companyId), eq(chatEndpoints.id, chatConversations.endpointId),
|
||||
)).leftJoin(toolConnections, and(
|
||||
eq(toolConnections.companyId, companyId), eq(toolConnections.id, chatEndpoints.connectionId),
|
||||
)).where(and(
|
||||
eq(chatConversations.companyId, companyId), eq(chatConversations.issueId, issueId),
|
||||
)).limit(1) : [];
|
||||
let nextAction = action.nextAction;
|
||||
if (chatBinding) {
|
||||
if (chatBinding.state === "endpoint_removed" || !chatBinding.endpointStatus || chatBinding.endpointStatus === "archived") {
|
||||
nextAction = "This chat connection was removed. Inspect the stopped run and create a new task to continue the work.";
|
||||
} else if (chatBinding.state === "unavailable" || !["active", "verifying"].includes(chatBinding.endpointStatus) ||
|
||||
chatBinding.connectionStatus !== "active" || !chatBinding.connectionEnabled) {
|
||||
nextAction = "This chat connection is unavailable. Restore access in Apps or create a new task to continue the work.";
|
||||
} else {
|
||||
nextAction = "Send a new chat message to continue this conversation.";
|
||||
}
|
||||
} else if (run?.status === "cancelled" && !eligibleContinuation && action.cause === "legacy_execution_requires_reconciliation") {
|
||||
nextAction += " Inspect the run before sending a new message to request continuation.";
|
||||
}
|
||||
|
||||
return {
|
||||
recoveryActionId: action.id,
|
||||
@@ -63,6 +113,10 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st
|
||||
// A stopped reviewer can differ from the task owner who receives the work back.
|
||||
agentId: run?.agentId ?? null,
|
||||
cause: action.cause,
|
||||
nextAction: action.nextAction,
|
||||
nextAction,
|
||||
runStatus: run?.status ?? null,
|
||||
runError: runError?.slice(0, 1024) ?? null,
|
||||
canContinue: eligibleContinuation && !chatBinding,
|
||||
savedMessageCount: saved.length,
|
||||
};
|
||||
}
|
||||
@@ -12,7 +12,7 @@ import { join } from "node:path";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { beforeAll, afterAll, describe, it, expect, vi } from "vitest";
|
||||
import {
|
||||
approvals, issueApprovals, issueThreadInteractions,
|
||||
approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections,
|
||||
agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions,
|
||||
issues, nativeRunFinalizations, nativeRunResults, completionContracts, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers,
|
||||
} from "@paperclipai/db";
|
||||
@@ -326,12 +326,18 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
else await expect(result).rejects.toThrow("continuation_user_authorization_missing");
|
||||
});
|
||||
|
||||
it.each(["valid", "wrong_actor", "consumed", "discarded", "operator_stop", "already_delivered", "earlier_delivered", "unstarted_cancelled", "unstarted_cancelled_metadata", "foreign_queue"])("validates automatic saved-message delivery: %s", async kind => {
|
||||
it.each(["valid", "provider_cancel", "pending_tool", "unknown_cancel", "wrong_actor", "consumed", "discarded", "operator_stop", "already_delivered", "earlier_delivered", "unstarted_cancelled", "unstarted_cancelled_metadata", "foreign_queue"])("validates automatic saved-message delivery: %s", async kind => {
|
||||
const f = await seed();
|
||||
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null,
|
||||
status: kind === "operator_stop" ? "cancelled" : "failed",
|
||||
status: ["operator_stop", "provider_cancel", "pending_tool", "unknown_cancel"].includes(kind) ? "cancelled" : "failed",
|
||||
startedAt: new Date("2026-09-11T08:00:00Z"),
|
||||
...(["provider_cancel", "pending_tool"].includes(kind) ? { resultJson: {
|
||||
acpToolInventoryComplete: true, acpPendingToolCount: kind === "pending_tool" ? 1 : 0,
|
||||
cancellation: { source: "provider", expected: false, initiator: { type: "provider" },
|
||||
reason: "Provider cancelled execution", recordedAt: new Date().toISOString() },
|
||||
} } : {}),
|
||||
contextSnapshot: { issueId: f.issueId, ...(kind === "already_delivered" ? { wakeCommentIds: [f.commentId] } : {}) },
|
||||
}).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
@@ -367,7 +373,7 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
return admitExplicitNativeContinuation({ ...f, actorId: kind === "wrong_actor" ? "someone-else" : f.actorId,
|
||||
db: tx as unknown as typeof db, queuedCommentRequestId: queueId, dryRun: true });
|
||||
});
|
||||
if (kind === "valid" || kind.startsWith("unstarted_cancelled")) expect(result).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId });
|
||||
if (kind === "valid" || kind === "provider_cancel" || kind.startsWith("unstarted_cancelled")) expect(result).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId });
|
||||
else expect(result).toBeNull();
|
||||
});
|
||||
|
||||
@@ -379,6 +385,72 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
return result;
|
||||
});
|
||||
|
||||
it.each(["active", "removed", "unavailable", "paused", "disabled"])("shows usable recovery guidance for external chat cancellations: %s", async kind => {
|
||||
const f = await seed();
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "cancelled", nativeIssueId: null,
|
||||
startedAt: new Date("2026-09-11T08:00:00Z"), resultJson: {
|
||||
acpToolInventoryComplete: true, acpPendingToolCount: 0,
|
||||
cancellation: { source: "provider", expected: false, initiator: { type: "provider" },
|
||||
reason: "Provider cancelled execution", recordedAt: new Date().toISOString() },
|
||||
},
|
||||
}).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ canContinue: true });
|
||||
const applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID();
|
||||
await db.insert(toolApplications).values({ id: applicationId, companyId: f.companyId, name: "Slack", type: "chat" });
|
||||
await db.insert(toolConnections).values({ id: connectionId, companyId: f.companyId, applicationId,
|
||||
name: "Slack", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel", status: "active", enabled: kind !== "disabled" });
|
||||
await db.insert(chatEndpoints).values({ id: endpointId, companyId: f.companyId, connectionId,
|
||||
provider: "slack", publicId: endpointId, assignedAgentId: f.agentId,
|
||||
status: kind === "removed" ? "archived" : kind === "paused" ? "paused" : "active" });
|
||||
await db.insert(chatConversations).values({ companyId: f.companyId, endpointId, issueId: f.issueId,
|
||||
externalConversationId: "channel", externalLabel: "#recovery",
|
||||
state: kind === "removed" ? "endpoint_removed" : kind === "unavailable" ? "unavailable" : "active" });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({
|
||||
runId: f.sourceRunId, canContinue: false,
|
||||
nextAction: kind === "active" ? "Send a new chat message to continue this conversation."
|
||||
: kind === "removed" ? "This chat connection was removed. Inspect the stopped run and create a new task to continue the work."
|
||||
: "This chat connection is unavailable. Restore access in Apps or create a new task to continue the work.",
|
||||
});
|
||||
});
|
||||
|
||||
it("delivers saved messages once after provider cancellation, preserving source history", async () => {
|
||||
const f = await seed();
|
||||
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
|
||||
// Hold agent capacity so the test observes durable admission without launching a provider.
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
const resultJson = { status: "cancelled", acpToolInventoryComplete: true, acpPendingToolCount: 0,
|
||||
cancellation: { source: "provider", expected: false, initiator: { type: "provider" },
|
||||
reason: "Provider cancelled execution", recordedAt: new Date().toISOString() },
|
||||
completedAction: "preserve-this-record" };
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "cancelled", nativeIssueId: null,
|
||||
startedAt: new Date("2026-09-11T08:00:00Z"), resultJson,
|
||||
}).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
const queueId = randomUUID();
|
||||
await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
|
||||
source: "automation", reason: "issue_commented", status: "deferred_issue_execution",
|
||||
requestedByActorType: "user", requestedByActorId: f.actorId, updatedAt: new Date(0),
|
||||
payload: { issueId: f.issueId, commentId: f.commentId, executionWait: { reason: "execution_recovery" },
|
||||
_paperclipWakeContext: { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentIds: [f.commentId], wakeCommentId: f.commentId } },
|
||||
});
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ savedMessageCount: 1, canContinue: true });
|
||||
await Promise.all([heartbeatService(db).resumeExecutionWaitComments(), heartbeatService(db).resumeExecutionWaitComments()]);
|
||||
await heartbeatService(db).resumeExecutionWaitComments();
|
||||
const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId, wakeCommentIds: [f.commentId] });
|
||||
const [queue] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId));
|
||||
expect(queue).toMatchObject({ status: "coalesced", runId: runs[0].id });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
expect(source.resultJson).toEqual(resultJson);
|
||||
});
|
||||
|
||||
it.each(["verified", "unproven", "changed", "dry_run", "retry", "duplicate"])(
|
||||
"requires exact local cleanup for a new turn after worker loss (%s)", async mode => {
|
||||
const f = await seed();
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
|
||||
import { canContinueCancelledRun } from "./run-cancellation.js";
|
||||
import { readQueuedInteractionResponse } from "./queued-interaction-response.js";
|
||||
import { isCancelledNativeStartup } from "./cancelled-native-startup.js";
|
||||
import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession } from "./native-local-process-stop.js";
|
||||
@@ -171,7 +172,8 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
!(queuedInterrupt && response?.source.requiresFreshSession && run.runtimeMode === "native")) return null;
|
||||
// Saved input is a request for a new turn, never permission to undo an
|
||||
// operator Stop or redeliver a message already consumed by this run.
|
||||
if (queuedRequest && !queuedInterrupt && ((run.status === "cancelled" && !unusedAdmission) ||
|
||||
if (retry && run.status === "cancelled" && !canContinueCancelledRun(run)) return blocked("cancelled_by_operator", "Inspect the stopped run and send a new message to continue.");
|
||||
if (queuedRequest && !queuedInterrupt && ((run.status === "cancelled" && !unusedAdmission && !canContinueCancelledRun(run)) ||
|
||||
run.contextSnapshot?.wakeCommentId === commentId ||
|
||||
(Array.isArray(run.contextSnapshot?.wakeCommentIds) && run.contextSnapshot.wakeCommentIds.includes(commentId)))) return null;
|
||||
if (legacyUserTurn) {
|
||||
|
||||
@@ -5,6 +5,8 @@ import { AgentDirectoryReuseInvalidatedError, isAgentDirectoryCopy } from "./age
|
||||
|
||||
import type { PaperclipTurnContext } from "@paperclipai/adapter-utils/server-utils";
|
||||
import { restoreNativeWorkspaceBestEffort } from "./native-runtime/native-workspace-best-effort.js";
|
||||
import { ASSIGNED_MCP_SERVER_NAME } from "./mcp-tool-names.js";
|
||||
import { cancellationResultJson, canContinueCancelledRun, readRunCancellation, requestedRunCancellation } from "./run-cancellation.js";
|
||||
import {
|
||||
withNativeWorkspaceFinalizationOwnership,
|
||||
NativeWorkspaceFinalizationBusyError,
|
||||
@@ -3495,6 +3497,27 @@ const heartbeatRunSafeResultJsonColumn = sql<Record<string, unknown> | null>`
|
||||
'workspaceRestoreFailure', case when ${heartbeatRuns.resultJson} ->> 'workspaceRestoreFailure'
|
||||
in ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')
|
||||
then ${heartbeatRuns.resultJson} -> 'workspaceRestoreFailure' end,
|
||||
'cancellation', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'cancellation') = 'object'
|
||||
then jsonb_strip_nulls(jsonb_build_object(
|
||||
'source', case when ${heartbeatRuns.resultJson} #>> '{cancellation,source}'
|
||||
in ('operator', 'queued_message', 'shutdown', 'provider', 'transport', 'control_plane', 'unknown')
|
||||
then ${heartbeatRuns.resultJson} #> '{cancellation,source}' end,
|
||||
'expected', case when jsonb_typeof(${heartbeatRuns.resultJson} #> '{cancellation,expected}') = 'boolean'
|
||||
then ${heartbeatRuns.resultJson} #> '{cancellation,expected}' end,
|
||||
'initiator', jsonb_strip_nulls(jsonb_build_object(
|
||||
'type', case when ${heartbeatRuns.resultJson} #>> '{cancellation,initiator,type}' in ('user', 'agent', 'system', 'provider')
|
||||
then ${heartbeatRuns.resultJson} #> '{cancellation,initiator,type}' end,
|
||||
'id', left(${heartbeatRuns.resultJson} #>> '{cancellation,initiator,id}', 128)
|
||||
)),
|
||||
'reason', left(${heartbeatRuns.resultJson} #>> '{cancellation,reason}', 512),
|
||||
'recordedAt', left(${heartbeatRuns.resultJson} #>> '{cancellation,recordedAt}', 64)
|
||||
)) end,
|
||||
'acpToolInventoryComplete', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete') = 'boolean'
|
||||
then ${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete' end,
|
||||
'acpPendingToolCount', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'acpPendingToolCount') = 'number'
|
||||
and length(${heartbeatRuns.resultJson} ->> 'acpPendingToolCount') < 16
|
||||
then ${heartbeatRuns.resultJson} -> 'acpPendingToolCount' end,
|
||||
'errorFamily', left(${heartbeatRuns.resultJson} ->> 'errorFamily', 32),
|
||||
'finalResponseRecorded', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'finalResponseRecorded') = 'boolean'
|
||||
then ${heartbeatRuns.resultJson} -> 'finalResponseRecorded' end,
|
||||
'executionBeforeRestore', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'executionBeforeRestore') = 'object'
|
||||
@@ -4872,7 +4895,7 @@ export async function buildPaperclipRuntimeMcpServers(input: {
|
||||
|
||||
return [
|
||||
{
|
||||
name: "paperclip-assigned",
|
||||
name: ASSIGNED_MCP_SERVER_NAME,
|
||||
url: `${paperclipApiBaseUrl()}/mcp/gateways/${gateway!.gatewayPublicId}`,
|
||||
token: token.token,
|
||||
connectionId: `assignment:${assignmentDigest}`,
|
||||
@@ -10604,6 +10627,12 @@ export function heartbeatService(
|
||||
if (!sourceId || !isUuidLike(sourceId)) continue;
|
||||
const run = await getRun(sourceId);
|
||||
if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue;
|
||||
if (canContinueCancelledRun(run)) {
|
||||
await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => {
|
||||
logger.warn({ err, runId: run.id }, "failed to resume saved input after provider cancellation");
|
||||
});
|
||||
continue;
|
||||
}
|
||||
await resumeRemoteStopComments(run, wake.id).catch(err => {
|
||||
logger.warn({ err, runId: run.id }, "failed to resume saved execution-wait message");
|
||||
});
|
||||
@@ -12857,6 +12886,9 @@ export function heartbeatService(
|
||||
|
||||
// Cancelling a queued run that never acquired provider execution is
|
||||
// positive bootstrap evidence. It must not hold unrelated queued messages.
|
||||
if (previousStatus && (status === "cancelled" || status === "interrupted")) {
|
||||
patch = { ...patch, resultJson: cancellationResultJson(previousStatus, status, patch?.resultJson, patch?.errorCode, patch?.error) };
|
||||
}
|
||||
if (
|
||||
status === "cancelled" &&
|
||||
previousStatus?.status === "queued" &&
|
||||
@@ -12941,6 +12973,10 @@ export function heartbeatService(
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (previousStatus && (status === "cancelled" || status === "interrupted")) {
|
||||
patch = { ...patch, resultJson: cancellationResultJson(previousStatus, status, patch?.resultJson, patch?.errorCode, patch?.error) };
|
||||
}
|
||||
|
||||
// Cancelling a queued run that never acquired provider execution is
|
||||
// positive bootstrap evidence. It must not hold unrelated queued messages.
|
||||
if (
|
||||
@@ -15357,6 +15393,11 @@ export function heartbeatService(
|
||||
delayMs?: number;
|
||||
},
|
||||
) {
|
||||
if (run.errorCode === "provider_tool_definition_invalid") {
|
||||
return { outcome: "not_scheduled" as const,
|
||||
reason: "Repair the invalid tool definitions before starting a new attempt.",
|
||||
issueId: readNonEmptyString(run.contextSnapshot?.issueId) };
|
||||
}
|
||||
if (Array.isArray(run.contextSnapshot?.chatCompletionDeliveryIds) &&
|
||||
run.contextSnapshot.chatCompletionDeliveryIds.some(id => typeof id === "string")) {
|
||||
return { outcome: "not_scheduled" as const, reason: "The completion outbox owns this reply's retry budget and publication identity.",
|
||||
@@ -25255,6 +25296,8 @@ export function heartbeatService(
|
||||
: "failed";
|
||||
} else if (adapterResult.timedOut) {
|
||||
outcome = "timed_out";
|
||||
} else if (adapterResult.resultJson?.status === "cancelled") {
|
||||
outcome = "cancelled";
|
||||
} else if (
|
||||
(adapterResult.exitCode ?? 0) === 0 &&
|
||||
!adapterResult.errorMessage &&
|
||||
@@ -25287,7 +25330,7 @@ export function heartbeatService(
|
||||
const normalizedUsage = sessionUsageResolution.normalizedUsage;
|
||||
const runErrorMessage =
|
||||
outcome === "cancelled"
|
||||
? (latestRun?.error ?? adapterResult.errorMessage ?? "Cancelled")
|
||||
? redactCurrentUserText(latestRun?.error ?? adapterResult.errorMessage ?? "Cancelled", currentUserRedactionOptions)
|
||||
: outcome === "succeeded"
|
||||
? null
|
||||
: redactCurrentUserText(
|
||||
@@ -25402,7 +25445,7 @@ export function heartbeatService(
|
||||
} as Record<string, unknown>)
|
||||
: null;
|
||||
|
||||
const persistedResultJson = mergeHeartbeatRunResultJson(
|
||||
const persistedResultJson = cancellationResultJson(latestRun ?? run, outcome, mergeHeartbeatRunResultJson(
|
||||
mergeRunStopMetadataForAgent(agent, outcome, {
|
||||
resultJson: mergeAdapterRecoveryMetadata({
|
||||
resultJson: {
|
||||
@@ -25422,7 +25465,7 @@ export function heartbeatService(
|
||||
errorMessage: runErrorMessage,
|
||||
}),
|
||||
adapterResult.summary ?? null,
|
||||
);
|
||||
), runErrorCode, runErrorMessage);
|
||||
|
||||
const finalRunPatch: Partial<typeof heartbeatRuns.$inferInsert> = {
|
||||
finishedAt: new Date(),
|
||||
@@ -25467,6 +25510,7 @@ export function heartbeatService(
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
...finalRunPatch,
|
||||
resultJson: cancellationResultJson(persistedRunWrite.run, status, finalRunPatch.resultJson, runErrorCode, runErrorMessage),
|
||||
finishedAt:
|
||||
persistedRunWrite.run.finishedAt ?? finalRunPatch.finishedAt,
|
||||
updatedAt: new Date(),
|
||||
@@ -25519,6 +25563,7 @@ export function heartbeatService(
|
||||
payload: {
|
||||
status,
|
||||
exitCode: adapterResult.exitCode,
|
||||
...(readRunCancellation(finalizedRun.resultJson) ? { cancellation: readRunCancellation(finalizedRun.resultJson) } : {}),
|
||||
},
|
||||
});
|
||||
try {
|
||||
@@ -26811,7 +26856,7 @@ export function heartbeatService(
|
||||
if (opts.requestedByActorType !== "user" || !opts.requestedByActorId ||
|
||||
reason !== "retry_failed_run" || source !== "on_demand" || triggerDetail !== "manual" ||
|
||||
!failed || failed.companyId !== agent.companyId || failed.agentId !== agentId ||
|
||||
!["failed", "timed_out"].includes(failed.status) ||
|
||||
(!["failed", "timed_out"].includes(failed.status) && !canContinueCancelledRun(failed)) ||
|
||||
(failed.nativeIssueId ?? readNonEmptyString(failed.contextSnapshot?.issueId)) !== issueId) {
|
||||
throw conflict("The selected failed run cannot be retried for this task.");
|
||||
}
|
||||
@@ -29386,6 +29431,7 @@ export function heartbeatService(
|
||||
return run;
|
||||
const agent = await getAgent(run.agentId);
|
||||
const errorCode = options.errorCode ?? "cancelled";
|
||||
const cancellation = requestedRunCancellation(options.resultJson ?? {}, reason);
|
||||
|
||||
const pendingProcessCancellation = processRunCancellationSettlements.get(
|
||||
run.id,
|
||||
@@ -29415,6 +29461,7 @@ export function heartbeatService(
|
||||
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ||
|
||||
${JSON.stringify(options.errorCode === "issue_reassigned" && options.resultJson?.reassignmentStopConfirmed === true
|
||||
? { reassignmentStopRequested: true } : {})}::jsonb ||
|
||||
${JSON.stringify({ cancellation })}::jsonb ||
|
||||
jsonb_build_object('startupCancellation', jsonb_build_object(
|
||||
'requestedAt', ${new Date().toISOString()}::text,
|
||||
'beforeNativeSelection', ${heartbeatRuns.runtimeMode} = 'legacy'
|
||||
@@ -29428,7 +29475,7 @@ export function heartbeatService(
|
||||
if (!fenced) return getRun(runId);
|
||||
run = fenced;
|
||||
}
|
||||
const resultJson = agent
|
||||
const resultJson = { ...(agent
|
||||
? {
|
||||
...mergeRunStopMetadataForAgent(agent, "cancelled", {
|
||||
resultJson: parseObject(run.resultJson),
|
||||
@@ -29437,7 +29484,7 @@ export function heartbeatService(
|
||||
}),
|
||||
...(options.resultJson ?? {}),
|
||||
}
|
||||
: options.resultJson;
|
||||
: options.resultJson), cancellation };
|
||||
|
||||
try {
|
||||
let releaseProcessCancellation: (() => void) | undefined;
|
||||
@@ -29639,7 +29686,7 @@ export function heartbeatService(
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: options.eventMessage ?? "run cancelled",
|
||||
...(options.eventPayload ? { payload: options.eventPayload } : {}),
|
||||
payload: { ...options.eventPayload, cancellation: readRunCancellation(cancelled.resultJson) },
|
||||
});
|
||||
await releaseIssueExecutionAndPromote(cancelled, {
|
||||
suppressImmediateRecovery: options.suppressImmediateRecovery,
|
||||
@@ -29682,6 +29729,9 @@ export function heartbeatService(
|
||||
continue;
|
||||
}
|
||||
if (run.runtimeMode === "native") {
|
||||
await db.update(heartbeatRuns).set({ resultJson:
|
||||
sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ cancellation: requestedRunCancellation({}, reason) })}::jsonb`,
|
||||
}).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, run.status)));
|
||||
await cancelHeartbeatNativeRun({
|
||||
db,
|
||||
runId: run.id,
|
||||
@@ -29699,15 +29749,13 @@ export function heartbeatService(
|
||||
finishedAt: new Date(),
|
||||
error: reason,
|
||||
errorCode,
|
||||
...(agent
|
||||
? {
|
||||
resultJson: mergeRunStopMetadataForAgent(agent, "cancelled", {
|
||||
resultJson: persistedCancellationResult,
|
||||
errorCode,
|
||||
errorMessage: reason,
|
||||
}),
|
||||
}
|
||||
: {}),
|
||||
resultJson: {
|
||||
...persistedCancellationResult,
|
||||
...(agent ? mergeRunStopMetadataForAgent(agent, "cancelled", {
|
||||
resultJson: persistedCancellationResult, errorCode, errorMessage: reason,
|
||||
}) : {}),
|
||||
cancellation: readRunCancellation(persistedCancellationResult) ?? requestedRunCancellation({}, reason),
|
||||
},
|
||||
});
|
||||
|
||||
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { ASSIGNED_MCP_SERVER_NAME, boundedMcpToolName } from "./mcp-tool-names.js";
|
||||
|
||||
describe("provider MCP name budget", () => {
|
||||
it("keeps short names compatible", () => {
|
||||
expect(boundedMcpToolName("mcp.drive-12345678:read-file", ["connection", "read_file"]))
|
||||
.toBe("mcp.drive-12345678:read-file");
|
||||
});
|
||||
it("bounds the complete provider name and retains unique, stable aliases", () => {
|
||||
const name = `mcp.app-gallery-posthog-${"a".repeat(40)}-cb60822c:file-download-batch-exports-count-rows-create`;
|
||||
const first = boundedMcpToolName(name, ["connection-one", "file_download_batch_exports_count_rows_create"]);
|
||||
expect(`mcp__${ASSIGNED_MCP_SERVER_NAME}__${first}`.length).toBeLessThanOrEqual(128);
|
||||
expect(first).toContain("file-download");
|
||||
expect(boundedMcpToolName(name, ["connection-one", "file_download_batch_exports_count_rows_create"])).toBe(first);
|
||||
expect(boundedMcpToolName(name, ["connection-two", "file_download_batch_exports_count_rows_create"])).not.toBe(first);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,17 @@
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
// Claude includes this namespace in its 128-character tool-name limit.
|
||||
// Gateway punctuation is normalized one-for-one by the MCP client.
|
||||
export const ASSIGNED_MCP_SERVER_NAME = "paperclip-assigned";
|
||||
export const MAX_GATEWAY_TOOL_NAME_LENGTH = 128 - `mcp__${ASSIGNED_MCP_SERVER_NAME}__`.length;
|
||||
|
||||
/** Preserve existing short names; aliases still resolve through gateway metadata. */
|
||||
export function boundedMcpToolName(name: string, identity: readonly string[]): string {
|
||||
if (name.length <= MAX_GATEWAY_TOOL_NAME_LENGTH) return name;
|
||||
const hash = createHash("sha256").update(JSON.stringify(identity)).digest("hex").slice(0, 16);
|
||||
const separator = name.indexOf(":");
|
||||
const namespace = name.slice(0, separator < 0 ? name.length : separator).slice(0, 32);
|
||||
const tool = separator < 0 ? "tool" : name.slice(separator + 1);
|
||||
const budget = MAX_GATEWAY_TOOL_NAME_LENGTH - namespace.length - hash.length - 2;
|
||||
return `${namespace}:${tool.slice(0, budget)}-${hash}`;
|
||||
}
|
||||
@@ -172,4 +172,10 @@ describe("classifyAdapterFailureForRecovery", () => {
|
||||
resultJson: null,
|
||||
})).toBeNull();
|
||||
});
|
||||
it("holds invalid provider definitions for repair even with stale transient metadata", () => {
|
||||
const run = { errorCode: "provider_tool_definition_invalid", error: "Tool name is too long.",
|
||||
resultJson: { errorFamily: "transient_upstream" } };
|
||||
expect(classifyAdapterFailureForRecovery(run)).toEqual({ kind: "configuration_incomplete" });
|
||||
expect(classifyContinuationFailure(run as never)).toMatchObject({ kind: "non_retryable", maxAttempts: 0 });
|
||||
});
|
||||
});
|
||||
@@ -490,6 +490,7 @@ const TRANSIENT_INFRA_CONTINUATION_ERROR_CODES = new Set<string>([
|
||||
]);
|
||||
|
||||
const NON_RETRYABLE_CONTINUATION_ERROR_CODES = new Set<string>([
|
||||
"provider_tool_definition_invalid",
|
||||
"adapter_engine_unavailable",
|
||||
"agent_not_invokable",
|
||||
"agent_not_found",
|
||||
@@ -624,7 +625,7 @@ export function classifyAdapterFailureForRecovery(
|
||||
): AdapterFailureRecoveryClassification {
|
||||
// An engine prerequisite cannot be repaired by asking the same unavailable
|
||||
// engine to retry. Use the existing configuration-blocker path.
|
||||
if (latestRun.errorCode === "adapter_engine_unavailable") {
|
||||
if (latestRun.errorCode === "adapter_engine_unavailable" || latestRun.errorCode === "provider_tool_definition_invalid") {
|
||||
return { kind: "configuration_incomplete" };
|
||||
}
|
||||
if (
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { cancellationResultJson, canContinueCancelledRun, isUnexpectedRunCancellation, requestedRunCancellation } from "./run-cancellation.js";
|
||||
|
||||
describe("cancellation provenance", () => {
|
||||
const run = { status: "cancelled", runtimeMode: "legacy" as const, startedAt: new Date(0), finishedAt: new Date(1),
|
||||
error: "Cancelled", errorCode: "cancelled", resultJson: null };
|
||||
it("preserves operator intent over adapter completion metadata", () => {
|
||||
const cancellation = requestedRunCancellation({ cancelledByUserId: "board", cancelledByActorType: "user" }, "Stop");
|
||||
const resultJson = cancellationResultJson({ ...run, resultJson: { cancellation } }, "cancelled", { status: "cancelled" });
|
||||
expect(resultJson?.cancellation).toEqual(cancellation);
|
||||
expect(isUnexpectedRunCancellation({ ...run, resultJson })).toBe(false);
|
||||
expect(canContinueCancelledRun({ ...run, resultJson })).toBe(false);
|
||||
});
|
||||
it("records queue interruption and shutdown as expected system actions", () => {
|
||||
expect(requestedRunCancellation({ queuedCommentInterruptQueueId: "queue", cancelledByUserId: "board" }, "Send messages"))
|
||||
.toMatchObject({ source: "queued_message", expected: true, initiator: { type: "user", id: "board" } });
|
||||
expect(cancellationResultJson(run, "interrupted", {}, "server_shutdown_interrupted")?.cancellation)
|
||||
.toMatchObject({ source: "shutdown", expected: true });
|
||||
expect(requestedRunCancellation({ queuedCommentInterruptQueueId: "queue", interruptedByActorType: "user", interruptedByActorId: "board" }, "Send messages"))
|
||||
.toMatchObject({ source: "queued_message", initiator: { type: "user", id: "board" } });
|
||||
});
|
||||
it("requires positive provider evidence and a complete empty tool inventory for saved input", () => {
|
||||
const resultJson = cancellationResultJson(run, "cancelled", { status: "cancelled", acpToolInventoryComplete: true, acpPendingToolCount: 0 });
|
||||
expect(resultJson?.cancellation).toMatchObject({ source: "provider", expected: false });
|
||||
expect(canContinueCancelledRun({ ...run, resultJson })).toBe(true);
|
||||
expect(canContinueCancelledRun({ ...run, resultJson: { ...resultJson, acpPendingToolCount: 1 } })).toBe(false);
|
||||
expect(canContinueCancelledRun({ ...run, resultJson: { ...resultJson, acpToolInventoryComplete: false } })).toBe(false);
|
||||
expect(canContinueCancelledRun({ ...run, resultJson: {} })).toBe(false);
|
||||
expect(isUnexpectedRunCancellation(run)).toBe(true);
|
||||
expect(isUnexpectedRunCancellation({ ...run, resultJson: {
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false },
|
||||
} })).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,84 @@
|
||||
import type { heartbeatRuns } from "@paperclipai/db";
|
||||
|
||||
type Run = Pick<typeof heartbeatRuns.$inferSelect, "status" | "startedAt" | "finishedAt" | "runtimeMode" | "errorCode" | "error" | "resultJson">;
|
||||
const SOURCES = ["operator", "queued_message", "shutdown", "provider", "transport", "control_plane", "unknown"] as const;
|
||||
export type CancellationSource = typeof SOURCES[number];
|
||||
export interface RunCancellation {
|
||||
source: CancellationSource;
|
||||
expected: boolean;
|
||||
initiator: { type: "user" | "agent" | "system" | "provider"; id?: string };
|
||||
reason: string;
|
||||
recordedAt: string;
|
||||
}
|
||||
|
||||
/** Server-owned local evidence. Sentry receives only the closed labels below. */
|
||||
export function readRunCancellation(result: Run["resultJson"] | undefined): RunCancellation | null {
|
||||
const value = result?.cancellation;
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
|
||||
const entry = value as Record<string, unknown>;
|
||||
const initiator = entry.initiator as Record<string, unknown> | undefined;
|
||||
if (!SOURCES.includes(entry.source as CancellationSource) || typeof entry.expected !== "boolean" ||
|
||||
!initiator || !["user", "agent", "system", "provider"].includes(initiator.type as string) ||
|
||||
typeof entry.reason !== "string" || typeof entry.recordedAt !== "string") return null;
|
||||
return { source: entry.source as CancellationSource, expected: entry.expected,
|
||||
initiator: { type: initiator.type as RunCancellation["initiator"]["type"],
|
||||
...(typeof initiator.id === "string" ? { id: initiator.id } : {}) },
|
||||
reason: entry.reason.slice(0, 512), recordedAt: entry.recordedAt };
|
||||
}
|
||||
|
||||
export function requestedRunCancellation(result: Run["resultJson"], reason: string): RunCancellation {
|
||||
const queued = typeof result?.queuedCommentInterruptQueueId === "string";
|
||||
const actorType = result?.cancelledByActorType ?? result?.interruptedByActorType;
|
||||
const user = actorType === "user" || actorType === "board" || typeof result?.cancelledByUserId === "string";
|
||||
const agent = actorType === "agent";
|
||||
const actorId = result?.cancelledByUserId ?? result?.cancelledByAgentId ?? result?.interruptedByActorId;
|
||||
const source: CancellationSource = queued ? "queued_message" : user ? "operator" : "control_plane";
|
||||
return { source, expected: true, initiator: {
|
||||
type: user ? "user" : agent ? "agent" : "system",
|
||||
...(typeof actorId === "string" ? { id: actorId } : {}),
|
||||
}, reason: reason.slice(0, 512), recordedAt: new Date().toISOString() };
|
||||
}
|
||||
|
||||
/** Preserve recorded stop intent when adapter completion replaces resultJson. */
|
||||
export function cancellationResultJson(run: Run, status: string, patch?: Run["resultJson"], errorCode?: string | null, error?: string | null): Run["resultJson"] {
|
||||
if (status !== "cancelled" && status !== "interrupted") return patch ?? run.resultJson;
|
||||
const result = { ...run.resultJson, ...patch };
|
||||
const previous = readRunCancellation(run.resultJson);
|
||||
let cancellation = previous?.expected ? previous : readRunCancellation(patch) ?? previous;
|
||||
const code = errorCode ?? run.errorCode;
|
||||
if (!cancellation) {
|
||||
if (code === "server_shutdown_interrupted") {
|
||||
cancellation = { source: "shutdown", expected: true, initiator: { type: "system" },
|
||||
reason: "Server shutdown interrupted execution", recordedAt: new Date().toISOString() };
|
||||
} else if (result.cancelledByActorType || result.cancelledByUserId || result.interruptedByActorType || result.queuedCommentInterruptQueueId) {
|
||||
cancellation = requestedRunCancellation(result, error ?? run.error ?? "Execution stopped");
|
||||
} else {
|
||||
const source = code === "duplex_channel_lost" ? "transport" : result.status === "cancelled" ? "provider" : "unknown";
|
||||
const description = source === "provider" ? "Provider cancelled execution"
|
||||
: source === "transport" ? "Connection to the execution provider was lost"
|
||||
: "Execution stopped without a recorded cancellation request";
|
||||
const detail = error ?? run.error;
|
||||
cancellation = { source, expected: false, initiator: { type: source === "provider" ? "provider" : "system" },
|
||||
reason: (!detail || /^cancelled\.?$/i.test(detail) ? description : `${description}: ${detail}`).slice(0, 512),
|
||||
recordedAt: new Date().toISOString() };
|
||||
}
|
||||
}
|
||||
return { ...result, cancellation };
|
||||
}
|
||||
|
||||
/** Eligibility only; admission still verifies process/lease shutdown and gates.
|
||||
* Historical ambiguous cancellations and outstanding tools stay held.
|
||||
*/
|
||||
export function canContinueCancelledRun(run: Run): boolean {
|
||||
const cancellation = readRunCancellation(run.resultJson);
|
||||
return run.status === "cancelled" && run.runtimeMode === "legacy" && Boolean(run.startedAt && run.finishedAt) &&
|
||||
cancellation?.expected === false && ["provider", "transport"].includes(cancellation.source) &&
|
||||
run.resultJson?.acpToolInventoryComplete === true && run.resultJson?.acpPendingToolCount === 0;
|
||||
}
|
||||
|
||||
export function isUnexpectedRunCancellation(run: Run): boolean {
|
||||
const bootstrap = run.resultJson?.executionRecovery as Record<string, unknown> | undefined;
|
||||
return run.status === "cancelled" && Boolean(run.startedAt) &&
|
||||
!(bootstrap?.kind === "bootstrap" && bootstrap.providerWorkStarted === false) &&
|
||||
readRunCancellation(run.resultJson)?.expected !== true;
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { heartbeatRuns } from "@paperclipai/db";
|
||||
import { readRunCancellation } from "./run-cancellation.js";
|
||||
import { WORKSPACE_RESTORE_FAILURE_CODES } from "@paperclipai/shared";
|
||||
import { redactDiagnosticText } from "@paperclipai/adapter-utils/command-redaction";
|
||||
import { redactCurrentUserText } from "../log-redaction.js";
|
||||
@@ -143,6 +144,15 @@ export function collectRunFailureDiagnostics(run: Run, options: RunFailureReport
|
||||
if (Number.isFinite(durationMs) && durationMs >= 0) execution.durationMs = durationMs;
|
||||
}
|
||||
const result = run.resultJson;
|
||||
const cancellation = readRunCancellation(result);
|
||||
if (cancellation) {
|
||||
execution.cancellationSource = cancellation.source;
|
||||
execution.cancellationExpected = cancellation.expected;
|
||||
execution.cancellationInitiatorType = cancellation.initiator.type;
|
||||
} else if (run.status === "cancelled") {
|
||||
execution.cancellationSource = "unknown";
|
||||
execution.cancellationExpected = false;
|
||||
}
|
||||
Object.assign(execution, scalars(result, [
|
||||
"mode", "stopReason", "timeoutFired", "timeoutSource", "timeoutConfigured",
|
||||
"effectiveTimeoutSec", "errorFamily",
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
type RunFailureReportOptions,
|
||||
} from "./run-failure-diagnostics.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { isUnexpectedRunCancellation } from "./run-cancellation.js";
|
||||
|
||||
type HeartbeatRun = typeof heartbeatRuns.$inferSelect;
|
||||
|
||||
@@ -30,7 +31,7 @@ const pendingRunFailureReports = new Set<Promise<void>>();
|
||||
const PENDING_REPORT_DRAIN_TIMEOUT_MS = 5_000;
|
||||
|
||||
function isRunFailureStatus(status: string): status is RunFailureStatus {
|
||||
return status === "failed" || status === "timed_out";
|
||||
return status === "failed" || status === "timed_out" || status === "cancelled";
|
||||
}
|
||||
|
||||
function readTaskId(run: HeartbeatRun): string | null {
|
||||
@@ -41,7 +42,7 @@ function readTaskId(run: HeartbeatRun): string | null {
|
||||
|
||||
/**
|
||||
* Report a terminal run failure to Sentry. Returns at once for any status
|
||||
* other than `failed` and `timed_out`. Never throws — a Sentry failure or a
|
||||
* other than failures and unexpected started cancellations. Never throws — a Sentry failure or a
|
||||
* database read failure must not change the caller's control flow.
|
||||
*
|
||||
* Call this beside the caller's own terminal-status write, with
|
||||
@@ -52,6 +53,7 @@ function readTaskId(run: HeartbeatRun): string | null {
|
||||
*/
|
||||
export function reportRunFailure(db: Db, run: HeartbeatRun, options: RunFailureReportOptions = {}): Promise<void> {
|
||||
if (!isRunFailureStatus(run.status)) return Promise.resolve();
|
||||
if (run.status === "cancelled" && !isUnexpectedRunCancellation(run)) return Promise.resolve();
|
||||
const runStatus = run.status;
|
||||
const report = captureTerminalRunFailure(db, run, runStatus, options);
|
||||
pendingRunFailureReports.add(report);
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { isInsufficientConnectionScope, INSUFFICIENT_CONNECTION_SCOPE_MESSAGE } from "./connection-permission-errors.js";
|
||||
import { boundedMcpToolName } from "./mcp-tool-names.js";
|
||||
import { browserUseService } from "./browser-use.js";
|
||||
import { isBrowserUseConnection } from "./browser-use-client.js";
|
||||
import { COGNEE_STDIO_TEMPLATE, cogneeCloudUrl, callCogneeCloud } from "./cognee-connection.js";
|
||||
@@ -1384,10 +1385,14 @@ export function createToolGatewayService(
|
||||
);
|
||||
}
|
||||
const baseName = baseNames[index]!;
|
||||
const gatewayToolName =
|
||||
const unboundedGatewayToolName =
|
||||
baseNameCounts.get(baseName)! > 1
|
||||
? `${baseName}-${shortStableId(catalogEntry.id)}`
|
||||
: baseName;
|
||||
const gatewayToolName = boundedMcpToolName(unboundedGatewayToolName, [
|
||||
connection.id, application.id, catalogEntry.toolName,
|
||||
...(baseNameCounts.get(baseName)! > 1 ? [catalogEntry.id] : []),
|
||||
]);
|
||||
const applicationKey = application.applicationKey ?? null;
|
||||
const inputSchema = projectedConnectionToolInputSchema(
|
||||
connection,
|
||||
|
||||
Reference in new issue
Block a user