mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
fix(grok): settle false file holds through inspected recovery
Preserve stopped-run fences and no-replay while allowing verified new user work. Honor disabled workspace access at the dispatch-failure retention owner. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
1146a08135
commit
8a2bbf09bc
9 files changed
+250
-16
No files matched your search
@@ -65,7 +65,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st
|
||||
const [run] = runId ? await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId),
|
||||
)).limit(1) : [];
|
||||
const withoutWorkspace = Boolean(run && nativeRunHasNoWorkspaceAccess(run));
|
||||
const withoutWorkspace = Boolean(run?.driverKind === "grok_bot_mcp" && nativeRunHasNoWorkspaceAccess(run));
|
||||
const workspaceRepairRequired = Boolean(restoreHold && !withoutWorkspace);
|
||||
const queue = await db.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
@@ -120,12 +120,16 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st
|
||||
nextAction += " Inspect the run before sending a new message to request continuation.";
|
||||
}
|
||||
|
||||
if (withoutWorkspace && (restoreHold || run?.status === "cancelled")) {
|
||||
if (withoutWorkspace && (restoreHold || run?.status === "cancelled") && !run?.resultJson?.externalWorkspaceInspection) {
|
||||
nextAction = run?.driverKind === "grok_bot_mcp"
|
||||
? "Inspect this Bot in Grok Bot, then clear the recovery hold and verify the connection in the agent's Runtime settings before assigning more work."
|
||||
: "Inspect the stopped work in the external agent and its connection before requesting continuation.";
|
||||
}
|
||||
|
||||
if (run?.resultJson?.externalWorkspaceInspection && !restoreHold) {
|
||||
nextAction = "Verify the Grok connection, then send a new message to continue. The stopped work will not be replayed.";
|
||||
}
|
||||
|
||||
return {
|
||||
recoveryActionId: action.id,
|
||||
runId,
|
||||
|
||||
@@ -10,6 +10,8 @@ import { scheduleAgentLifecycle } from "../agent-lifecycle.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
import { deliverGrokRoutineWake, validateGrokRoutineCredential, type GrokRoutineDelivery, type GrokRoutineWake } from "./webhook.js";
|
||||
|
||||
import { grokWorkspaceInspectionCandidates, inspectGrokWorkspaceObligations } from "./workspace-inspection.js";
|
||||
|
||||
const digest = (value: string) => createHash("sha256").update(value).digest("hex");
|
||||
const fail = (message: string) => new McpOAuthError("access_denied", message, 409);
|
||||
export const grokWakeAcknowledgementSchema = z.object({ correlationId: z.uuid(), bindingId: z.uuid(),
|
||||
@@ -44,11 +46,12 @@ export function grokRoutineService(db: Db, deliver: Delivery = deliverGrokRoutin
|
||||
const [uncertain] = await db.select({ id: dispatches.id }).from(dispatches).innerJoin(routines, eq(routines.id, dispatches.routineId))
|
||||
.where(and(eq(dispatches.companyId, companyId), eq(routines.agentId, agentId),
|
||||
or(eq(dispatches.status, "unknown"), and(eq(dispatches.status, "sending"), lte(dispatches.updatedAt, new Date(Date.now() - 30_000))), and(eq(dispatches.status, "accepted"), lte(dispatches.expiresAt, new Date()))))).limit(1);
|
||||
return !!held || !!uncertain;
|
||||
return !!held || !!uncertain || (await grokWorkspaceInspectionCandidates(db, companyId, agentId)).length > 0;
|
||||
}
|
||||
|
||||
async function recover(companyId: string, agentId: string, operatorId: string) {
|
||||
await authorizeExternalBotOperator(db, companyId, operatorId);
|
||||
await inspectGrokWorkspaceObligations(db, companyId, agentId, operatorId);
|
||||
await db.transaction(async tx => {
|
||||
// All routine writers serialize on the binding before dispatch/routine rows.
|
||||
// Include retired bindings so recovery cannot bypass an old activity hold.
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
import { and, eq, inArray, isNotNull, or, sql } from "drizzle-orm";
|
||||
import { dotRunnerAssignments, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { authorizeExternalBotOperator } from "../external-bot-authority.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
import { McpOAuthError } from "../public-mcp/oauth.js";
|
||||
import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
|
||||
import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js";
|
||||
|
||||
/** Old Runner synchronization descriptors could incorrectly require a file
|
||||
* export for a Bot explicitly admitted without workspace access. Keep the old
|
||||
* execution fenced and its no-replay hold; inspection only retires that false
|
||||
* file obligation. Native Runner cleanup never certifies global Grok exit. */
|
||||
export async function grokWorkspaceInspectionCandidates(db: Db, companyId: string, agentId: string) {
|
||||
const runs = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId),
|
||||
eq(heartbeatRuns.driverKind, "grok_bot_mcp"), isNotNull(heartbeatRuns.finishedAt),
|
||||
inArray(heartbeatRuns.status, ["cancelled", "failed", "timed_out", "interrupted"]),
|
||||
sql`${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'`,
|
||||
));
|
||||
return runs.filter(run => nativeRunHasNoWorkspaceAccess(run) && hasRequiredWorkspaceRecovery(run.resultJson));
|
||||
}
|
||||
|
||||
export async function inspectGrokWorkspaceObligations(db: Db, companyId: string, agentId: string, operatorId: string) {
|
||||
for (const selected of await grokWorkspaceInspectionCandidates(db, companyId, agentId)) {
|
||||
// Independent task transactions keep the native task/coordinator/run lock
|
||||
// order out of the routine writer's binding/routine lock order. A partial
|
||||
// inspection grants no authority and never schedules work.
|
||||
await db.transaction(async tx => {
|
||||
const database = tx as unknown as Db;
|
||||
await authorizeExternalBotOperator(database, companyId, operatorId);
|
||||
const issueId = selected.nativeIssueId ?? selected.issueId;
|
||||
if (!issueId) return;
|
||||
const [task] = await tx.select().from(issues).where(and(
|
||||
eq(issues.companyId, companyId), eq(issues.id, issueId),
|
||||
)).for("update");
|
||||
const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, selected.id),
|
||||
eq(nativeRunFinalizations.issueId, issueId),
|
||||
)).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.id, selected.id),
|
||||
)).for("update");
|
||||
if (!run || !hasRequiredWorkspaceRecovery(run.resultJson)) return;
|
||||
const fail = () => new McpOAuthError("access_denied", "The stopped task still needs verified Runner cleanup before inspection can clear its file obligation.", 409);
|
||||
if (!task || !nativeRunHasNoWorkspaceAccess(run) || run.driverKind !== "grok_bot_mcp"
|
||||
|| !run.finishedAt || !["cancelled", "failed", "timed_out", "interrupted"].includes(run.status)
|
||||
|| !coordinator || coordinator.phase !== "terminal_failure" || coordinator.leaseOwner
|
||||
|| coordinator.failureDetail?.successorRunId) throw fail();
|
||||
const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
|
||||
or(eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.issueId, issueId)),
|
||||
)).limit(1);
|
||||
const work = await tx.select().from(dotRunnerAssignments).where(and(
|
||||
eq(dotRunnerAssignments.companyId, companyId), eq(dotRunnerAssignments.agentId, agentId),
|
||||
eq(dotRunnerAssignments.runId, run.id),
|
||||
));
|
||||
const leases = await tx.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
));
|
||||
if (active || !work.length || work.some(row => row.status !== "fenced") || !leases.length
|
||||
|| !leases.every(lease => lease.issueId === issueId && hasRemoteTerminationReceipt(lease))) throw fail();
|
||||
const actions = await tx.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
or(sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
|
||||
sql`${issueRecoveryActions.evidence}->>'sourceRunId' = ${run.id}`),
|
||||
)).for("update");
|
||||
const inspection = { schema: "paperclip.grok-workspace-inspection.v1", runId: run.id,
|
||||
inspectedBy: operatorId, inspectedAt: new Date().toISOString(), workspaceAccess: "none",
|
||||
remoteTerminationConfirmed: false, runnerLeaseIds: leases.map(lease => lease.id),
|
||||
previousWorkspaceRestoreFailure: run.resultJson?.workspaceRestoreFailure,
|
||||
previousWorkspaceRestoreRecovery: run.resultJson?.workspaceRestoreRecovery };
|
||||
const retireFileObligation = (value: Record<string, unknown>) => {
|
||||
const { workspaceRestoreFailure: _failure, workspaceRestoreRecovery: _recovery, ...rest } = value;
|
||||
return { ...rest, externalWorkspaceInspection: inspection };
|
||||
};
|
||||
await tx.update(heartbeatRuns).set({ resultJson: retireFileObligation(run.resultJson!) }).where(eq(heartbeatRuns.id, run.id));
|
||||
for (const action of actions) {
|
||||
await tx.update(issueRecoveryActions).set({ evidence: retireFileObligation(action.evidence), updatedAt: new Date(),
|
||||
nextAction: "Inspect Grok and verify its connection before sending a new task message. The old work will not be replayed.",
|
||||
}).where(eq(issueRecoveryActions.id, action.id));
|
||||
}
|
||||
await logActivity(database, { companyId, actorType: "user", actorId: operatorId,
|
||||
action: "grok.workspace_obligation_inspected", entityType: "issue", entityId: issueId,
|
||||
agentId, runId: run.id, details: inspection });
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -56,9 +56,9 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () =
|
||||
vi.mocked(executePaperclipNativeSession).mockResolvedValue({ exitCode: 0, signal: null, timedOut: false });
|
||||
});
|
||||
|
||||
async function fixture(native = false) {
|
||||
async function fixture(native = false, provider: "codex" | "grok_bot" | "openai_dot" = "codex") {
|
||||
const [company] = await db.insert(companies).values({ name: "Runtime execution", issuePrefix: "EXE" }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Executing agent", status: "running", adapterType: native ? "paperclip_runner" : "process", adapterConfig: { provider: "codex" } }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Executing agent", status: "running", adapterType: native ? "paperclip_runner" : "process", adapterConfig: provider === "codex" ? { provider } : { provider, lifecycleMode: "per_turn", allowUnmeteredProvider: true, [provider === "grok_bot" ? "externalBindingId" : "dotBindingId"]: randomUUID() } }).returning();
|
||||
const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Complete this task", assigneeAgentId: agent.id }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, issueId: issue.id, scopeKind: "issue", status: "running", runtimeMode: native ? "native" : "legacy" }).returning();
|
||||
const state = createHeartbeatRunState(db);
|
||||
@@ -68,14 +68,15 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () =
|
||||
const nativeExecution = native ? buildNativeExecutionInput({
|
||||
companyId: company.id, agentId: agent.id, runId: run.id, issue,
|
||||
taskPrompt: issue.title, workspace: { id: run.id, cwd: home, repoUrl: null, repoRef: null, branchName: null },
|
||||
normalizedSessionId: null, provider: "codex", runtimeContext: nativeRuntimeContextFixture(),
|
||||
normalizedSessionId: null, provider, runtimeContext: nativeRuntimeContextFixture(),
|
||||
dotBinding: { bindingId: randomUUID(), bindingGeneration: 1, companyId: company.id, agentId: agent.id, acceptByUnixMs: Date.now() + 60_000, expiresAtUnixMs: Date.now() + 120_000 },
|
||||
completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", contract: { revision: "1", objective: issue.title, criteria: [{ id: "output", requirement: issue.title }] } },
|
||||
}) : null;
|
||||
const input: HeartbeatRuntimeExecutionInput = {
|
||||
run, agent, options: {},
|
||||
task: { issueRef: issue, issueContext: null, context: {}, issueId: issue.id, taskKey: issue.id, executionContinuation: null },
|
||||
runtime: { nativeExecution, nativeRunnerInstanceId: native ? randomUUID() : null, getNativeFreshSessionHandoff: undefined,
|
||||
nativeRuntimeResolution: resolveHeartbeatNativeRuntimeMode({ persisted: run, enabled: native, runtimeConfig: {}, adapterConfig: agent.adapterConfig, agent, issue, target: null, workspaceId: null }),
|
||||
nativeRuntimeResolution: resolveHeartbeatNativeRuntimeMode({ persisted: run, enabled: native, grokEnabled: true, dotEnabled: true, runtimeConfig: {}, adapterConfig: agent.adapterConfig, agent, issue, target: null, workspaceId: null }),
|
||||
adapter, runtimeForAdapter: { taskKey: issue.id, sessionId: null, sessionParams: null, sessionDisplayId: null }, getFreshSessionHandoff: undefined, runOptions: {},
|
||||
},
|
||||
workspace: { persistedExecutionWorkspace: null, executionWorkspace: { baseCwd: home, cwd: home, source: "agent_home", projectId: null, workspaceId: null, repoUrl: null, repoRef: null, strategy: "project_primary", branchName: null, worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false },
|
||||
@@ -234,8 +235,8 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () =
|
||||
}]);
|
||||
});
|
||||
|
||||
it("retains an unexported native source before provider-error teardown", async () => {
|
||||
const { input, run } = await fixture(true);
|
||||
it.each(["codex", "grok_bot", "openai_dot"] as const)("retains user files only when %s can access the workspace after a dispatch failure", async provider => {
|
||||
const { input, run } = await fixture(true, provider);
|
||||
input.workspace.nativeWorkspaceSync = {
|
||||
mode: "host_current",
|
||||
reference: { schema: "paperclip.native-workspace-sync/v3", state: "prepared",
|
||||
@@ -246,7 +247,8 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () =
|
||||
const error = new Error("runner_remote_process_identity_unavailable");
|
||||
vi.mocked(executePaperclipNativeSession).mockRejectedValueOnce(error);
|
||||
input.instructions.releaseInstructionCopy = vi.fn(async () => {
|
||||
expect(input.effects.onProviderResourceDisposition).toHaveBeenCalledWith("stop_and_retain");
|
||||
if (provider === "codex") expect(input.effects.onProviderResourceDisposition).toHaveBeenCalledWith("stop_and_retain");
|
||||
else expect(input.effects.onProviderResourceDisposition).not.toHaveBeenCalledWith("stop_and_retain");
|
||||
});
|
||||
await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error);
|
||||
expect(input.workspace.nativeWorkspaceSync.restoreWorkspace).not.toHaveBeenCalled();
|
||||
|
||||
@@ -994,7 +994,8 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
|
||||
input.effects.onNativeOwnershipHeld();
|
||||
throw adapterErr;
|
||||
}
|
||||
if (nativeRuntimeResolution.kind === "native" && nativeWorkspaceSync?.reference.state === "prepared") {
|
||||
if (nativeRuntimeResolution.kind === "native" && nativeWorkspaceSync?.reference.state === "prepared"
|
||||
&& !(nativeExecution && "access" in nativeExecution.workspace && nativeExecution.workspace.access === "none")) {
|
||||
input.effects.onProviderResourceDisposition("stop_and_retain");
|
||||
}
|
||||
await db
|
||||
|
||||
@@ -109,7 +109,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
expect(await readLease(f.lease.id)).toMatchObject({ leasePolicy: "retain_on_failure" });
|
||||
});
|
||||
|
||||
it.each(["grok_bot", "openai_dot"] as const)("keeps historical %s inspection holds without offering file repair for a workspace-less run", async provider => {
|
||||
it.each(["grok_bot", "openai_dot"] as const)("keeps historical %s inspection holds with an available recovery path", async provider => {
|
||||
const f = await seed();
|
||||
// Reproduce a hold persisted by an older cancellation path, then attach its admitted contract.
|
||||
await preserveNativeFailedWorkspace(db, f.run);
|
||||
@@ -120,9 +120,11 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const blocker = await getExecutionBlocker(db, f.companyId, f.issueId);
|
||||
expect(blocker).toMatchObject({ runId: f.runId, canRetry: false, canContinue: false,
|
||||
workspaceRepairRequired: false, canRecordWorkspaceRepair: false });
|
||||
expect(blocker?.nextAction).not.toContain("files");
|
||||
expect(blocker?.nextAction).toContain(provider === "grok_bot" ? "Runtime" : "Inspect");
|
||||
workspaceRepairRequired: provider !== "grok_bot", canRecordWorkspaceRepair: provider !== "grok_bot" });
|
||||
if (provider === "grok_bot") {
|
||||
expect(blocker?.nextAction).not.toContain("files");
|
||||
expect(blocker?.nextAction).toContain("Runtime");
|
||||
}
|
||||
expect(await actions(f.issueId)).toHaveLength(1);
|
||||
});
|
||||
|
||||
|
||||
Reference in new issue
Block a user