mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix: offer Interrupt when queued-message steering is unavailable (#19450)
## Thinking Path > - Paperclip lets people manage work performed by AI agents. > - Task chat saves follow-up messages while an agent runs. > - A native runner can accept Steer only after an active provider turn exists. > - During preparation, or with an unsupported provider, the queue showed a disabled Steer button and no delivery action. > - The server also rejected Interrupt for ordinary native queues. > - This change offers an explicit Interrupt fallback and uses the existing cancellation and admission path. > - People can send their saved messages while Stop confirmation, cleanup, and execution gates remain enforced. ## Linked Issues or Issue Description Related: #15865 covers mixed legacy queues. #15030 covers steering acknowledgements. This change covers native queues without available steering. **What happened?** A queued message had a disabled Steer button while the native runner was still preparing. Clicking it did nothing. The server returned HTTP 409 for an explicit Interrupt request against the same native queue. **Expected behavior** Offer Steer when the provider can accept same-turn input. Otherwise offer Interrupt, labeled with its actual effect. Preserve saved messages until the previous execution stops and normal admission accepts the successor. **Steps to reproduce** 1. Start a task with the Paperclip Runner adapter. 2. Submit a follow-up while the run is preparing, before its provider turn opens. 3. Observe the disabled Steer button above the composer. 4. With this change, click Interrupt. The existing Stop path cancels the previous execution and submits the saved queue through normal admission. **Paperclip version or commit** Reproduced locally on master at523818ea9b. The branch is rebased on2f9d3d43c3. ## What Changed - Offer Interrupt for unavailable or unsupported native steering when the parent supplies an interrupt callback. - Keep Steer for an available provider turn. - Accept explicit Interrupt for a deferred native queue and record its run target. - Bind native continuation to the exact Interrupt queue, run, actor, and cancellation receipt. This lets one click send the messages after Stop completes. - Add component, task-thread, real PostgreSQL route, and continuation authorization regressions. - Close the route fixture database pool before disposing its PostgreSQL instance. - Document the fallback and its Stop and admission requirements. ## Verification - Red: two component cases failed because Interrupt was absent. Two route cases returned HTTP 409, `This queue does not support interruption`, before the repair. - Red: stronger delivery assertions also caught `send_predates_stop` after the first repair. The admission fix covers both the freshness exception and later eligibility check. - Green: both route regressions now wait for preparation settlement, then deliver both saved messages exactly once without a second click. The complete route suite passes 119 tests. The complete continuation suite passes 376 tests, including wrong actor, queue, target, and missing-receipt cases. - UI: task-thread and queued-message suites pass 394 tests, with one existing skip. The issue-page suite also passes (the earlier three-suite run passed 644 tests before the final three UI cases). - Token gates and `git diff --check` pass. - Full workspace build and low-memory typecheck pass. The initial typecheck caught a missing `membershipRole` in an unrelated fixture; the rebase picked up the correction already on master, and the complete typecheck then passed. - The full local `pnpm test:run` passed all six chat shards (1,931 tests). It was stopped during general-server validation after the final commit passed all PR CI gates. The complete local suite is not claimed as a pass. - Earlier route-suite teardown timed out with an open database pool. Closing that pool fixed teardown; the complete route suite then passed. - All final-head CI gates pass: 52 successful checks and two intentional skips. Greptile is 5/5 on the same head with no unresolved review threads. ## Risks Interrupt stops the previous execution and starts a successor; it is a different action from same-turn steering. The UI labels that action explicitly. Cleanup or execution gates can keep the messages queued. Existing company access, board permissions, queue revision, run target, cancellation proof, and admission checks remain in place. No schema migration is needed. ## Model Used OpenAI Codex (GPT-6), with reasoning, tool use, and code execution. The exact model ID and context window are not exposed by this client. ## 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:
1 parent
275b334fc5
commit
70d23612dd
8 files changed
+117
-25
No files matched your search
@@ -2609,6 +2609,13 @@ They read only the queued run's runtime mode. Mutation locks retain complete
|
||||
wake and run rows, so edits, ordering, interruption, and saved execution inputs
|
||||
keep their existing write contract.
|
||||
|
||||
The task composer offers Steer when the native session has an active provider
|
||||
turn that accepts steering. If steering is unavailable or unsupported, it offers
|
||||
an explicit Interrupt action instead. Interrupt records the board user's intent,
|
||||
requests cancellation, and delivers the saved queue through normal admission after
|
||||
the previous execution stops. Preparation and cleanup can keep messages queued;
|
||||
the action does not bypass Stop confirmation or execution gates.
|
||||
|
||||
Run retrieval and session state are in `server/src/services/heartbeat/run-state.ts`.
|
||||
It owns bounded run projections, database encoding checks, task session reads and
|
||||
writes, explicit resumes, session compaction, and usage/billing helpers.
|
||||
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
issueComments,
|
||||
issueThreadInteractions,
|
||||
issueRecoveryActions,
|
||||
nativeRunFinalizations,
|
||||
issues,
|
||||
runIdentityContexts,
|
||||
} from "@paperclipai/db";
|
||||
@@ -78,6 +79,7 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await db?.$client.end();
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
@@ -814,14 +816,12 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
|
||||
expect(await db.select().from(heartbeatRuns)).toHaveLength(3);
|
||||
});
|
||||
|
||||
it.each(["stale revision", "native run", "different issue"] as const)(
|
||||
it.each(["stale revision", "different issue"] as const)(
|
||||
"rejects queued interruption for a %s without stopping the run",
|
||||
async (scenario) => {
|
||||
const seeded = await seedQueue();
|
||||
if (scenario !== "native run") {
|
||||
await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, seeded.agentId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
}
|
||||
await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, seeded.agentId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
const client = app(seeded.companyId);
|
||||
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
||||
if (scenario === "different issue") {
|
||||
@@ -835,6 +835,54 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["preparing", "native"] as const)(
|
||||
"accepts an explicit interrupt fallback for a %s runner and preserves queued messages",
|
||||
async (runtime) => {
|
||||
const seeded = await seedQueue();
|
||||
if (runtime === "preparing") {
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null,
|
||||
executionStage: "preparing",
|
||||
runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
||||
}).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
} else {
|
||||
await db.update(heartbeatRuns).set({ nativeIssueId: seeded.issueId })
|
||||
.where(eq(heartbeatRuns.id, seeded.runId));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: seeded.companyId,
|
||||
issueId: seeded.issueId, runId: seeded.runId, phase: "observed" });
|
||||
}
|
||||
// Hold the successor in the ordinary queue without launching a provider.
|
||||
await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } })
|
||||
.where(eq(agents.id, seeded.agentId));
|
||||
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
||||
status: "running", contextSnapshot: { issueId: randomUUID() } });
|
||||
const client = app(seeded.companyId);
|
||||
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
||||
expect(queue.body).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" });
|
||||
const response = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({
|
||||
queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision,
|
||||
});
|
||||
expect(response.status, JSON.stringify(response.body)).toBe(200);
|
||||
const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
||||
expect(waiting.payload?.queuedCommentInterrupt).toMatchObject({ actorId: "queue-owner", targetRunId: seeded.runId });
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, seeded.issueId))).toHaveLength(2);
|
||||
expect(waiting.status).toBe("deferred_issue_execution");
|
||||
// The synthetic preparation owner now confirms that its unwind finished.
|
||||
// No provider ran; until this receipt exists, cancellation must hold input.
|
||||
const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
expect(stopped.status).toBe("cancelled");
|
||||
await db.update(heartbeatRuns).set({ resultJson: { ...stopped.resultJson,
|
||||
startupPreparationSettledAt: new Date().toISOString() } }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
||||
expect(wake.status, JSON.stringify(wake)).toBe("coalesced");
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
||||
expect(successor.status).toBe("queued");
|
||||
expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds);
|
||||
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
||||
expect(await db.select().from(heartbeatRuns)).toHaveLength(3);
|
||||
},
|
||||
);
|
||||
|
||||
it("does not accept interruption authority from an agent wake payload", async () => {
|
||||
const seeded = await seedQueue();
|
||||
await db.update(agents).set({ adapterType: "claude_local",
|
||||
|
||||
@@ -17152,13 +17152,10 @@ export function issueRoutes(
|
||||
allowStoppedTarget: true,
|
||||
});
|
||||
assertQueueMutationTarget({ queue: locked.queue, queueId: req.body.queueId, revision: req.body.revision });
|
||||
const freshResponse = locked.queue.entries.some(entry => entry.source?.requiresFreshSession);
|
||||
// The UI offers Send whenever no turn is running. The issue's run may
|
||||
// still be waiting to start or to retry: it has no turn to interrupt,
|
||||
// so save the click and deliver the queue as soon as that run stops.
|
||||
const previousRunNotRunning = !locked.activeRun || locked.activeRun.status !== "running";
|
||||
const stoppedNativeQueue = locked.queue.protocol === "paperclip_runner_v1" && previousRunNotRunning;
|
||||
if ((locked.queue.protocol !== "legacy" && !freshResponse && !stoppedNativeQueue) || locked.state !== "deferred" ||
|
||||
// Explicit Interrupt is also the fallback when a native turn cannot
|
||||
// accept steering. Cancellation and ordinary admission still own Stop
|
||||
// confirmation, cleanup, and delivery; this click only records intent.
|
||||
if (locked.state !== "deferred" ||
|
||||
!locked.queue.entries.length ||
|
||||
(locked.activeRun && locked.activeRun.agentId !== locked.wake.agentId)) {
|
||||
throw conflict("This queue does not support interruption");
|
||||
@@ -17172,7 +17169,7 @@ export function issueRoutes(
|
||||
await tx.update(agentWakeupRequests).set({
|
||||
payload: { ...readObject(locked.wake.payload), queuedCommentInterrupt: {
|
||||
actorId: actor.actorId, requestedAt: new Date().toISOString(),
|
||||
...(freshResponse && locked.queue.protocol === "paperclip_runner_v1" && locked.activeRun?.status === "running"
|
||||
...(locked.queue.protocol === "paperclip_runner_v1" && locked.activeRun?.status === "running"
|
||||
? { targetRunId: locked.activeRun.id } : {}),
|
||||
} },
|
||||
updatedAt: new Date(),
|
||||
|
||||
@@ -545,9 +545,9 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0])
|
||||
.toMatchObject({ result: interaction.result, resolvedByUserId: "board" });
|
||||
});
|
||||
it.each(["matching", "wrong_queue", "wrong_actor", "wrong_target", "missing_receipt", "ordinary_response"])("accepts only the exact live native plan Interrupt receipt: %s", async kind => {
|
||||
it.each(["matching", "wrong_queue", "wrong_actor", "wrong_target", "missing_receipt", "ordinary_response", "comment"])("accepts only the exact live native queue Interrupt receipt: %s", async kind => {
|
||||
const f = await seed(), queueId = randomUUID();
|
||||
const [interaction] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
const [interaction] = kind === "comment" ? [] : await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
sourceRunId: f.sourceRunId, createdByAgentId: f.agentId, kind: "request_confirmation", status: "accepted",
|
||||
resolvedByUserId: "board", resolvedAt: new Date("2026-09-11T09:00:00Z"),
|
||||
payload: { version: 1, prompt: "Approve this plan?" }, result: { version: 1, outcome: "accepted" },
|
||||
@@ -559,15 +559,18 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
} }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
|
||||
source: "on_demand", reason: "issue_commented", status: "deferred_issue_execution",
|
||||
payload: { issueId: f.issueId, mutation: "interaction", interactionId: interaction.id, interactionStatus: "accepted",
|
||||
_paperclipWakeContext: { forceFreshSession: kind !== "ordinary_response" },
|
||||
payload: { issueId: f.issueId,
|
||||
...(kind === "comment" ? { commentId: f.commentId } : {
|
||||
mutation: "interaction", interactionId: interaction!.id, interactionStatus: "accepted" }),
|
||||
_paperclipWakeContext: kind === "comment" ? { wakeCommentIds: [f.commentId] }
|
||||
: { forceFreshSession: kind !== "ordinary_response" },
|
||||
queuedCommentInterrupt: { actorId: "board", requestedAt: "2026-09-11T09:30:00Z",
|
||||
targetRunId: kind === "wrong_target" ? randomUUID() : f.sourceRunId },
|
||||
},
|
||||
});
|
||||
const result = await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db,
|
||||
commentId: null, queuedCommentInterruptId: queueId, dryRun: true }));
|
||||
if (kind === "matching") expect(result).toMatchObject({ previousRunId: f.sourceRunId });
|
||||
commentId: kind === "comment" ? f.commentId : null, queuedCommentInterruptId: queueId, dryRun: true }));
|
||||
if (["matching", "ordinary_response", "comment"].includes(kind)) expect(result).toMatchObject({ previousRunId: f.sourceRunId });
|
||||
else expect(result).toBeNull();
|
||||
});
|
||||
|
||||
|
||||
@@ -351,7 +351,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
}
|
||||
if (!historicalAdapter && agent.adapterType === "paperclip_runner") return null;
|
||||
}
|
||||
const interruptedByThisClick = response?.source.requiresFreshSession === true &&
|
||||
const interruptedByThisClick = queuedInterrupt &&
|
||||
interruptIntent?.targetRunId === run.id && run.resultJson?.operatorInterrupted === true &&
|
||||
run.resultJson?.interruptionSource === "issue_comment_interrupt" && run.resultJson?.interruptedIssueId === issueId &&
|
||||
run.resultJson?.interruptedByActorType === "user" && run.resultJson?.interruptedByActorId === actorId &&
|
||||
@@ -390,7 +390,8 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
!(queuedRequest && run.runtimeMode === "native" &&
|
||||
(run.status !== "cancelled" || authorizedAt > stoppedAt)) &&
|
||||
!(queuedRequest && cancelledStartup && authorizedAt > run.finishedAt!) &&
|
||||
!(nativeQueuedSend && (run.runtimeMode === "native" || cancelledStartup) && authorizedAt > run.finishedAt!) &&
|
||||
!(nativeQueuedSend && (run.runtimeMode === "native" || cancelledStartup) &&
|
||||
(authorizedAt > run.finishedAt! || interruptedByThisClick)) &&
|
||||
!(queuedInterrupt && response?.source.requiresFreshSession && run.runtimeMode === "native")) return null;
|
||||
// A saved native message sent after Stop is a new user direction. Older
|
||||
// input cannot restart a cancelled run merely because it is still queued.
|
||||
|
||||
@@ -4849,7 +4849,12 @@ describe("TaskChatThread Paperclip Runner queue", () => {
|
||||
expect(occurrenceCount(queuedComment.body)).toBe(1);
|
||||
});
|
||||
|
||||
it.each(["run-1", null])("offers Interrupt for a live legacy queue and Send for a stopped queue (%s)", (targetRunId) => {
|
||||
it.each([
|
||||
{ targetRunId: "run-1", protocol: "legacy", steeringDisposition: "unsupported" },
|
||||
{ targetRunId: null, protocol: "legacy", steeringDisposition: "unsupported" },
|
||||
{ targetRunId: "run-1", protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" },
|
||||
{ targetRunId: "run-1", protocol: "paperclip_runner_v1", steeringDisposition: "unsupported" },
|
||||
] as const)("offers Interrupt for a live $protocol queue with $steeringDisposition steering and Send for a stopped queue ($targetRunId)", ({ targetRunId, protocol, steeringDisposition }) => {
|
||||
const onInterruptQueued = vi.fn(async () => {});
|
||||
renderThread({
|
||||
comments: [
|
||||
@@ -4864,8 +4869,8 @@ describe("TaskChatThread Paperclip Runner queue", () => {
|
||||
queuedCommentQueue: {
|
||||
...queue,
|
||||
targetRunId,
|
||||
protocol: "legacy",
|
||||
steeringDisposition: "unsupported",
|
||||
protocol,
|
||||
steeringDisposition,
|
||||
},
|
||||
onEditQueuedComment: async () => {},
|
||||
onReorderQueuedComments: async () => {},
|
||||
|
||||
@@ -267,6 +267,36 @@ describe("TaskChatQueuedMessages", () => {
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it.each(["temporarily_unavailable", "unsupported"] as const)(
|
||||
"offers an explicit interrupt fallback when native steering is %s",
|
||||
async (steeringDisposition) => {
|
||||
const props = render({
|
||||
queue: { ...queue, steeringDisposition },
|
||||
onInterrupt: vi.fn().mockResolvedValue(undefined),
|
||||
});
|
||||
const interrupt = container.querySelector<HTMLButtonElement>(
|
||||
'[data-testid="task-chat-queued-interrupt-comment-1"]',
|
||||
);
|
||||
expect(interrupt).not.toBeNull();
|
||||
expect(interrupt!.disabled).toBe(false);
|
||||
expect(interrupt!.textContent).toContain("Interrupt");
|
||||
await act(async () => { interrupt!.click(); });
|
||||
expect(props.onInterrupt).toHaveBeenCalledOnce();
|
||||
expect(props.onSteer).not.toHaveBeenCalled();
|
||||
expect(container.querySelector('[data-testid="task-chat-queued-messages"]')).toBeNull();
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps same-turn steering when the provider is available, even with an interrupt callback", async () => {
|
||||
const props = render({ onInterrupt: vi.fn().mockResolvedValue(undefined) });
|
||||
expect(container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')).toBeNull();
|
||||
await act(async () => {
|
||||
container.querySelector<HTMLButtonElement>('[data-testid="task-chat-queued-steer-comment-1"]')!.click();
|
||||
});
|
||||
expect(props.onSteer).toHaveBeenCalledWith("comment-1", "rev-1");
|
||||
expect(props.onInterrupt).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("waits for authoritative discard acknowledgement before removing the row", async () => {
|
||||
const acknowledgement = deferred<void>();
|
||||
render({ onDiscard: vi.fn().mockReturnValue(acknowledgement.promise) });
|
||||
|
||||
@@ -156,7 +156,8 @@ function SortableQueuedMessage({
|
||||
{label}
|
||||
</span>
|
||||
|
||||
{(queue.targetRunId || queue.state === "deferred") && (!queue.targetRunId || queue.protocol === "legacy" || entry.source?.requiresFreshSession ? (
|
||||
{(queue.targetRunId || queue.state === "deferred") && (!queue.targetRunId || queue.protocol === "legacy" || entry.source?.requiresFreshSession ||
|
||||
(queue.steeringDisposition !== "available" && onInterrupt) ? (
|
||||
<button
|
||||
type="button"
|
||||
onClick={onInterrupt}
|
||||
|
||||
Reference in new issue
Block a user