mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-06 10:48:12 +02:00
fix: stop remote Grok runs before continuing queued messages (#14100)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The execution service owns each run and saves messages sent while it runs. > - Interrupt must stop the current executor before it delivers those messages. > - Remote Grok commands did not register the host cancellation control. > - A cancelled task run could still write Done and prevent queue recovery. > - This pull request connects remote cancellation and revokes cancelled run writes. > - Saved input can use the existing queue admission rules after verified cleanup. ## Linked Issues or Issue Description **What happened?** Interrupting a queued message marked a remote Grok run cancelled before its sandbox stopped. The old run could still post a reply and mark the task Done. Its saved follow-up remained deferred behind execution recovery. **Expected behavior** Stop revokes run write authority and waits for verified termination. Saved messages remain durable and enter one successor through normal admission after cleanup. **Steps to reproduce** 1. Run a task with `grok_local` in a remote sandbox. 2. Send a follow-up and use Interrupt while the command runs. 3. Let the old command attempt a task status update after cancellation. 4. Observe the task disposition and the saved message queue. **Paperclip version or commit** The gap is present in master at `d3e0f0a238`. **Deployment mode** Authenticated server with a Daytona sandbox. Related work: #14028 and #14046 handle bounded continuation. #13291 covers infrastructure interruption and verified remote cleanup. #13332 addresses atomic recovery holds. This change handles direct Grok operator cancellation and stale task writes. ## What Changed - Register remote Grok cancellation before preparation. Keep command ownership until the host confirms sandbox termination. - Reuse the sandbox cancellation boundary for the direct CLI invocation. Reject fresh attempts after cancellation and preserve workspace restore failure evidence. - Reject writes from cancelled task JWTs and runs with a pending stop. Preserve diagnostic reads and existing conversation error codes. - Recheck run authority under a database lock before task updates and interaction responses commit. - Preserve authorized handoffs that stop their own run. Only the server-issued stop receipt for that request permits the final task update. - Add tests for hung commands, unverified stops, early cancellation, copy-back failures, late Done, late interaction responses, authorized handoffs, exact lease receipts, and one queue successor across concurrent restart sweeps. - Document the cancellation and write-authority contract. ## Verification - Targeted adapter, cancellation-boundary, authentication, queued-message, interaction-service, and activity-route tests passed. The expanded run passed 214 tests; one new test had an incomplete fixture. After correcting the fixture, all 8 selected follow-up cases passed. - `pnpm -r typecheck`: passed on `179c86caf1bf0d89914a503d46e24af7e4b8c557`. - `pnpm build`: passed on the same commit. - `pnpm test:run`: the general-server group completed with 13,521 passed, 99 skipped, and 18 failed tests. It then stopped, so the remaining local groups did not run. Five Slack, email, and wake-batching failures passed on focused reruns after correcting the local environment. The remaining 13 failures reproduce as `EACCES` on rename in unchanged skill-cache code on macOS. Two custom-image suite setup hooks also failed to start embedded PostgreSQL after the machine exhausted shared-memory slots; all 31 tests in that file passed on rerun after the local resource issue was resolved. CI covers all test groups. - CI: 53 checks passed and 2 were skipped on the latest commit, including the aggregate verification gate. The last server shard passed on its single rerun after a preview-server startup timeout. The affected file also passed locally with 28 passed and 3 skipped. - Greptile: 5/5 on the latest commit. Both review threads are resolved. - No live deployment or staging task mutation has been performed. ## Risks - Stopping the sandbox can prevent file copy-back. The result preserves workspace restore failure evidence; termination does not imply restored files. - If provider termination fails, the adapter keeps ownership of its outstanding command and does not acknowledge Stop. - The write restriction now applies to ordinary cancelled tasks. Reads remain allowed. Task and interaction checks add a shared run-row lock to agent mutations. An exact server-issued receipt permits the task request that stopped its own run to complete its handoff. - Existing terminal tasks are not reopened automatically. An operator must correct a historical late Done before its saved queue can continue. - No schema migration or UI change. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository inspection, code execution, and test tools. The precise backend revision 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 #` 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:
1 parent
01d9a12185
commit
b2e9e82f05
13 files changed
+376
-13
No files matched your search
@@ -8,10 +8,11 @@ export function cancellableSandboxStartup(ctx: AdapterExecutionContext) {
|
||||
const signal = ctx.signal;
|
||||
const stop = ctx.stopRemoteStartup;
|
||||
if (!signal || !stop || target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner) {
|
||||
return { context: ctx, finish: async () => {} };
|
||||
return { context: ctx, stopAcknowledged: () => false, finish: async () => {} };
|
||||
}
|
||||
let armed = true;
|
||||
let stopping: Promise<void> | undefined;
|
||||
let stopAcknowledged = false;
|
||||
const inFlight = new Set<Promise<unknown>>();
|
||||
let rejectStopped!: (error: unknown) => void;
|
||||
const stopped = new Promise<never>((_, reject) => { rejectStopped = reject; });
|
||||
@@ -21,7 +22,10 @@ export function cancellableSandboxStartup(ctx: AdapterExecutionContext) {
|
||||
if (stopping) return;
|
||||
stopping = Promise.resolve().then(stop);
|
||||
void stopping.then(
|
||||
() => rejectStopped(signal.reason ?? new Error("Stopped during sandbox startup")),
|
||||
() => {
|
||||
stopAcknowledged = true;
|
||||
rejectStopped(signal.reason ?? new Error("Sandbox execution stopped"));
|
||||
},
|
||||
// Without proof, the original operation still owns its resources. Do
|
||||
// not abandon it or release credentials while it could be running.
|
||||
() => {},
|
||||
@@ -78,6 +82,7 @@ export function cancellableSandboxStartup(ctx: AdapterExecutionContext) {
|
||||
};
|
||||
return {
|
||||
context: { ...ctx, executionTarget: { ...target, runner } },
|
||||
stopAcknowledged: () => stopAcknowledged,
|
||||
async finish() {
|
||||
signal.removeEventListener("abort", onAbort);
|
||||
armed = false;
|
||||
|
||||
@@ -199,7 +199,7 @@ export interface AdapterExecutionContext {
|
||||
signal?: AbortSignal;
|
||||
/** Opt in to signal-based cancellation before starting provider work. */
|
||||
onCancellationReady?: () => Promise<void>;
|
||||
/** Host-owned stop of this run's sandbox during setup. Resolves only after
|
||||
/** Host-owned stop of this run's sandbox during setup or direct CLI execution. Resolves only after
|
||||
* provider termination is verified; never accepts an agent-selected lease. */
|
||||
stopRemoteStartup?: () => Promise<void>;
|
||||
/** Server-owned, actor-attributed snapshot also rendered by legacy wake prompts. */
|
||||
|
||||
@@ -62,8 +62,8 @@ vi.mock("@paperclipai/adapter-utils/execution-target", () => ({
|
||||
(mocks.ensureRuntimeInstalledMock as (...args: unknown[]) => unknown)(...args),
|
||||
prepareAdapterExecutionTargetRuntime: (...args: unknown[]) =>
|
||||
(mocks.prepareRuntimeMock as (...args: unknown[]) => unknown)(...args),
|
||||
readAdapterExecutionTarget: () =>
|
||||
mocks.state.isRemote ? { kind: "remote", transport: "ssh" } : { kind: "local" },
|
||||
readAdapterExecutionTarget: (input: { executionTarget?: unknown }) => input.executionTarget ??
|
||||
(mocks.state.isRemote ? { kind: "remote", transport: "ssh" } : { kind: "local" }),
|
||||
resolveAdapterExecutionTargetCommandForLogs: (...args: unknown[]) =>
|
||||
(mocks.resolveCommandForLogsMock as (...args: unknown[]) => unknown)(...args),
|
||||
resolveAdapterExecutionTargetTimeoutSec: (_target: unknown, timeoutSec: number) => timeoutSec,
|
||||
@@ -128,6 +128,8 @@ function makeRestoreWorkspace(
|
||||
|
||||
function makeSuccessfulRunResult(overrides: Partial<{ sessionId: string }> = {}) {
|
||||
return {
|
||||
pid: null,
|
||||
startedAt: new Date().toISOString(),
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
@@ -187,6 +189,94 @@ describe("grok_local execute", () => {
|
||||
await Promise.all(tempRoots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
async function cancellableContext(stop: () => Promise<void>) {
|
||||
const ctx = await makeCtx("remote-cancellation", await makeTempRoot());
|
||||
const controller = new AbortController();
|
||||
const remoteExecute = vi.fn(async () => makeSuccessfulRunResult());
|
||||
ctx.signal = controller.signal;
|
||||
ctx.stopRemoteStartup = vi.fn(stop);
|
||||
ctx.onCancellationReady = vi.fn(async () => {});
|
||||
ctx.executionTarget = { kind: "remote", transport: "sandbox", providerKey: "daytona", remoteCwd: "/remote/workspace",
|
||||
runner: { execute: remoteExecute } };
|
||||
remoteState.isRemote = true;
|
||||
runProcessMock.mockImplementation(async (_runId, target) => {
|
||||
expect(ctx.onCancellationReady).toHaveBeenCalledOnce();
|
||||
return target.runner.execute({ command: "grok" });
|
||||
});
|
||||
return { ctx, controller, remoteExecute };
|
||||
}
|
||||
|
||||
it("settles remote cancellation only after the sandbox stop receipt, even if the command RPC hangs", async () => {
|
||||
let confirmStop!: () => void;
|
||||
const receipt = new Promise<void>(resolve => { confirmStop = resolve; });
|
||||
const f = await cancellableContext(() => receipt);
|
||||
f.remoteExecute.mockImplementation(() => new Promise(() => {}));
|
||||
let settled = false;
|
||||
const execution = execute(f.ctx).finally(() => { settled = true; });
|
||||
await vi.waitFor(() => expect(f.remoteExecute).toHaveBeenCalledOnce());
|
||||
f.controller.abort(new Error("Interrupted to send queued messages"));
|
||||
await vi.waitFor(() => expect(f.ctx.stopRemoteStartup).toHaveBeenCalledOnce());
|
||||
expect(settled).toBe(false);
|
||||
confirmStop();
|
||||
expect(await execution).toMatchObject({ errorCode: "cancelled",
|
||||
resultJson: { executionCancellation: { state: "acknowledged" } } });
|
||||
expect(runProcessMock).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("keeps ownership of the command when remote termination cannot be verified", async () => {
|
||||
const f = await cancellableContext(async () => { throw new Error("stop unverified"); });
|
||||
let finishCommand!: (result: ReturnType<typeof makeSuccessfulRunResult>) => void;
|
||||
f.remoteExecute.mockImplementation(() => new Promise(resolve => { finishCommand = resolve; }));
|
||||
let settled = false;
|
||||
const execution = execute(f.ctx).catch(error => error).finally(() => { settled = true; });
|
||||
await vi.waitFor(() => expect(f.remoteExecute).toHaveBeenCalledOnce());
|
||||
f.controller.abort();
|
||||
await vi.waitFor(() => expect(f.ctx.stopRemoteStartup).toHaveBeenCalledOnce());
|
||||
expect(settled).toBe(false);
|
||||
finishCommand(makeSuccessfulRunResult());
|
||||
expect(await execution).toEqual(new Error("stop unverified"));
|
||||
expect(runProcessMock).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("does not start Grok when cancellation was requested before registration", async () => {
|
||||
const f = await cancellableContext(async () => {});
|
||||
f.ctx.onCancellationReady = vi.fn(async () => { f.controller.abort(); });
|
||||
expect(await execute(f.ctx)).toMatchObject({ errorCode: "cancelled",
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false },
|
||||
resultJson: { executionCancellation: { state: "acknowledged" } } });
|
||||
expect(runProcessMock).not.toHaveBeenCalled();
|
||||
expect(prepareRuntimeMock).not.toHaveBeenCalled();
|
||||
expect(f.ctx.stopRemoteStartup).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("does not acknowledge an early cancellation when its acquired sandbox cannot stop", async () => {
|
||||
const f = await cancellableContext(async () => { throw new Error("stop unverified"); });
|
||||
f.ctx.onCancellationReady = vi.fn(async () => { f.controller.abort(); });
|
||||
await expect(execute(f.ctx)).rejects.toThrow("stop unverified");
|
||||
expect(runProcessMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("retains workspace recovery evidence when a confirmed stop prevents copy-back", async () => {
|
||||
const f = await cancellableContext(async () => {});
|
||||
prepareRuntimeMock.mockImplementationOnce(async () => ({ workspaceRemoteDir: "/remote/workspace", assetDirs: {},
|
||||
restoreWorkspace: async () => { throw new Error("sandbox stopped during restore"); },
|
||||
}));
|
||||
f.remoteExecute.mockImplementation(() => new Promise(() => {}));
|
||||
const execution = execute(f.ctx);
|
||||
await vi.waitFor(() => expect(f.remoteExecute).toHaveBeenCalledOnce());
|
||||
f.controller.abort();
|
||||
const result = await execution;
|
||||
expect(result.resultJson?.executionCancellation).toMatchObject({ state: "acknowledged" });
|
||||
expect(result.resultJson?.workspaceRestoreFailure).toBeTruthy();
|
||||
});
|
||||
|
||||
it("does not stop the sandbox after a normal completed turn", async () => {
|
||||
const f = await cancellableContext(async () => {});
|
||||
expect(await execute(f.ctx)).toMatchObject({ exitCode: 0 });
|
||||
f.controller.abort();
|
||||
expect(f.ctx.stopRemoteStartup).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("stages Grok-native instructions and skills into the workspace for the run and cleans them up afterward", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const instructionsPath = path.join(root, "managed", "AGENTS.md");
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { withWorkspaceRestore } from "@paperclipai/adapter-utils/workspace-restore-result";
|
||||
import { cancellableSandboxStartup } from "@paperclipai/adapter-utils/acpx-engine/startup-cancellation";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
@@ -196,6 +197,55 @@ function resolveBillingType(env: Record<string, string>): "api" | "subscription"
|
||||
}
|
||||
|
||||
export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExecutionResult> {
|
||||
const target = ctx.executionTarget;
|
||||
if (!ctx.signal || !ctx.stopRemoteStartup || target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner) {
|
||||
return executeTurn(ctx);
|
||||
}
|
||||
|
||||
// Direct remote commands have no host child process to kill. Register before
|
||||
// setup and retain ownership until the host verifies this sandbox has stopped.
|
||||
await ctx.onCancellationReady?.();
|
||||
const cancelled = (result?: AdapterExecutionResult): AdapterExecutionResult => ({
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
...result,
|
||||
errorCode: "cancelled",
|
||||
errorMessage: "Grok execution was cancelled",
|
||||
resultJson: {
|
||||
...result?.resultJson,
|
||||
executionCancellation: { state: "acknowledged", acknowledgedAt: new Date().toISOString() },
|
||||
},
|
||||
});
|
||||
if (ctx.signal.aborted) {
|
||||
// The host may already have acquired a lease before adapter registration.
|
||||
await ctx.stopRemoteStartup();
|
||||
return { ...cancelled(), executionRecovery: { kind: "bootstrap", providerWorkStarted: false } };
|
||||
}
|
||||
// Keep the existing setup boundary armed for the whole direct CLI invocation:
|
||||
// unlike ACP adapters, Grok has no turn-level cancellation protocol.
|
||||
const cancellation = cancellableSandboxStartup(ctx);
|
||||
let result: AdapterExecutionResult | undefined;
|
||||
let failure: unknown;
|
||||
let failed = false;
|
||||
try {
|
||||
result = await executeTurn(cancellation.context);
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
failed = true;
|
||||
}
|
||||
try {
|
||||
await cancellation.finish();
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
failed = true;
|
||||
}
|
||||
if (cancellation.stopAcknowledged()) return cancelled(result);
|
||||
if (failed) throw failure;
|
||||
return result!;
|
||||
}
|
||||
|
||||
async function executeTurn(ctx: AdapterExecutionContext): Promise<AdapterExecutionResult> {
|
||||
const { runId, agent, runtime, config, context, onLog, onMeta, onSpawn, authToken } = ctx;
|
||||
const executionTarget = readAdapterExecutionTarget({
|
||||
executionTarget: ctx.executionTarget,
|
||||
@@ -543,6 +593,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
};
|
||||
|
||||
const runAttempt = async (resumeSessionId: string | null) => {
|
||||
ctx.signal?.throwIfAborted();
|
||||
const prompt = joinPromptSections([
|
||||
selectInitialCommunicationGuidance(context, { resumedSession: Boolean(resumeSessionId) }),
|
||||
basePrompt,
|
||||
@@ -655,6 +706,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
};
|
||||
|
||||
const initial = await runAttempt(sessionId);
|
||||
ctx.signal?.throwIfAborted();
|
||||
if (
|
||||
sessionId &&
|
||||
!initial.proc.timedOut &&
|
||||
|
||||
Reference in new issue
Block a user