From d0b67bfe71277f5a5ad21c7a7b6cb7af09e2ccca Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 16 Sep 2026 14:44:36 -0500 Subject: [PATCH] feat: queue approvals and answers during active runs (#13539) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Users guide running agents through messages, questions, and approval cards. > - Messages already wait in a queue when an agent is running. > - Card responses did not appear in that queue. Some question answers also steered a later run without a user click. > - A fast approval could invalidate the agent's review handoff and cause it to stop its own run. > - This pull request gives card responses the same queue controls and preserves the exact response during delivery. > - Users can wait for completion or explicitly send the response with Interrupt or Steer. ## Linked Issues or Issue Description Refs #13517, which is merged. This PR targets master and adds queued interaction responses on top of the onboarding changes. Related continuation work: #10519 and #12866. **What happened?** Accepting a proposal while its source run was active left a saved response outside the message queue. The agent could then lose its review path, reassign the task, and cancel itself. Answers to older questions could also steer another active turn without a click. **Expected behavior** Save the response immediately. Queue its continuation behind the active run. Deliver it after completion, or when the user explicitly chooses Interrupt or Steer. Preserve approval revisions and answer choices. **Steps to reproduce** 1. Let an agent publish a confirmation card while its run is still active. 2. Accept the card before the agent finishes its review handoff. 3. Inspect the message queue and the task's next run. **Paperclip version or commit** Reproduced on da8a3876c with the onboarding changes from #13517. **Deployment mode** Local development from source. The fix covers legacy adapters and native Runner turns. ## What Changed - Project resolved cards into the existing queue as immutable responses. Keep answers and exact approval revisions. - Require an explicit click to steer a response into a compatible native turn. Use Interrupt when a fresh session is required. - Preserve typed response context through interruption, cleanup waits, and normal queue promotion. Keep the direct answer channel for a provider blocked on its original question request. - Accept the source run's review handoff after its card resolves. Reject stale agent reassignment that would orphan a queued response. - Add deterministic regression tests and an `accept-while-running` case to the first-task suite. Require recorded timestamp overlap before that case can pass. - Keep the first-task skill name out of user-facing messages. ## Verification - Red-green: the original route failed the queue regression; the changed route passes it. - Focused server/UI tests: 139 passed, including 64 queue-route tests. - Runner harness unit tests: 314 passed. - Server, UI, and Runner E2E typechecks passed. UI token gates passed. - Full repository typecheck and build passed. Server typecheck passed again after review fixes. - Review regressions: 165 queue/reopen route tests, 53 wake admission tests, and 18 run identity tests passed. Approval acknowledgement recovery and both message/approval arrival orders are covered. - Full local test run: 12,401 passed; three new admission regressions ran against a cached pre-fix module. A fresh run of that entire suite passed (53 tests). The complete CI suite passed on the final commit. - Previous-head CI at `c28e2ef12`: 32 checks passed and 2 optional Storybook checks skipped. Every server/workspace/browser shard, Runner verification, build, typecheck/release registry, canary, policy, and security check passed. Greptile: 5/5, no unresolved threads. Earlier interrupted CI workers were replaced by this fresh complete run. - After integrating the updated parent: 314 harness tests, 119 queue/admission tests, 44 onboarding/question-delivery tests, and 13 native recovery tests passed locally. Full repository typecheck and build passed. - Clarified the skill wording preference: routine replies describe the action without announcing the internal skill; direct questions and permission/security/execution disclosures remain truthful. - The paid `accept-while-running` scenario is registered for all four local first-task profiles. It has not been run against a model in this change. - Rebased onto the merged parent at `11921075a`; the resulting tree exactly matches the locally verified integration tree. Final-head CI on `b53054807` passed: 54 successful checks, 2 optional Storybook checks skipped, no failed checks. Every new server/browser shard, aggregate verify/e2e gate, Runner, typecheck, build, canary, and security check passed on the first attempt. Greptile reviewed this exact head at 5/5 with no unresolved threads. ## Risks - Responses now wait instead of implicitly steering another active turn. A provider blocked on the original question still receives its answer directly. - Approval receipts cannot be edited, discarded, or reordered as comments. This preserves the recorded decision. - Interruption must still prove that the prior execution stopped. The tests cover cleanup waits and duplicate delivery. - The new paid overlap case can be unexercised if the model finishes before the click lands. It cannot pass without evidence of overlap. - No database migration is required. This repairs the existing approvals and execution controls; it does not implement the roadmap's work-stream queues. ## Model Used OpenAI GPT-6 through Codex. The exact deployed model ID and context-window size were not exposed in this session. Capabilities used: agentic reasoning, repository inspection, code editing, terminal commands, and test execution. ## 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 --- doc/execution-semantics.md | 30 ++ packages/shared/src/types/issue.ts | 7 + .../issue-comment-reopen-routes.test.ts | 4 + .../issue-execution-policy-routes.test.ts | 3 + .../issue-queued-comments-routes.test.ts | 299 +++++++++++++++++- .../question-response-delivery.test.ts | 95 ++---- .../wake-queue/application/use-cases.test.ts | 20 ++ .../wake-queue/application/use-cases.ts | 11 +- .../first-task/skills/first-task/SKILL.md | 6 +- server/src/routes/issues.ts | 68 +++- .../services/explicit-native-continuation.ts | 22 +- server/src/services/heartbeat.ts | 40 ++- .../services/question-response-delivery.ts | 52 +-- .../services/queued-interaction-response.ts | 67 ++++ server/src/services/run-identity.ts | 20 +- tests/runner-e2e/README.md | 17 +- tests/runner-e2e/catalog.test.ts | 14 +- tests/runner-e2e/catalog.ts | 2 +- tests/runner-e2e/first-task-cases.ts | 1 + tests/runner-e2e/first-task-flow.ts | 19 +- tests/runner-e2e/first-task-scoring.ts | 14 + tests/runner-e2e/first-task.test.ts | 25 +- .../task-chat/TaskChatQueuedMessages.test.tsx | 21 ++ .../task-chat/TaskChatQueuedMessages.tsx | 13 +- ui/src/lib/issue-queued-comment-queue.test.ts | 6 + ui/src/lib/issue-queued-comment-queue.ts | 7 + 26 files changed, 694 insertions(+), 189 deletions(-) create mode 100644 server/src/services/queued-interaction-response.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 1df3a5f6236..bc3e58f96fd 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1291,3 +1291,33 @@ permission again or copy Connect / Not now into a generic question. A generic question does not start setup. The real connection card keeps user identity, access grants, the decision, and continuation together. This guidance does not approve a connection or bypass its normal user decision. + +## Responses submitted during an active run + +A confirmation, checkbox confirmation, or question answer is new conversation +input. Resolving the card records the decision immediately; it does not implicitly +interrupt or steer an agent that is doing work. Its typed continuation wake waits +behind the issue's active execution and appears in the message queue. + +The queue projects the original resolved interaction as an immutable response. +It keeps the selected answers and accepted document revision; it does not create +an editable comment that could silently change what was approved. Ordinary +messages retain their existing edit, discard, and reorder behavior. + +- Normal run completion promotes the saved response once. The restart scan also + finds stranded interaction receipts after the issue execution lock is released. +- **Steer** explicitly delivers the saved response to a compatible native turn. + The acknowledgement consumes the receipt, so a retry cannot create a second + delivery. It retains the existing run's execution identity. +- **Interrupt** stops a legacy turn and starts a continuation with the typed + response. Native plan approvals that require a fresh session use Interrupt too; + steering cannot turn a planning session into an execution session. Existing + process-stop, environment-cleanup, ownership, and recovery gates still apply. +- If the provider is blocked on the original native question request, answering + resolves that tool request directly. It must not wait behind the blocked turn. + +An agent may finish its review handoff after the user has already answered its +card. A resolved card from that same source run, or its queued continuation, is a +valid live path. A stale agent handback to a human cannot cancel the run and orphan +a queued response. This does not make an old resolved card a review path for a +later run, or prevent an explicit board reassignment. diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index 348ce8ed888..dab04ce8b6e 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -974,6 +974,13 @@ export type IssueQueuedCommentSteeringDisposition = | "temporarily_unavailable"; export interface IssueQueuedCommentEntry { + /** Immutable response projected from its durable interaction receipt. */ + source?: { + kind: "interaction"; + interactionId: string; + interactionKind: string; + requiresFreshSession?: boolean; + }; comment: IssueComment; position: number; canEdit: boolean; diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index eb1dd5799d1..362cbcd23b7 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -121,6 +121,10 @@ vi.mock("../telemetry.js", () => ({ getTelemetryClient: vi.fn(() => ({ track: vi.fn() })), })); +vi.mock("../services/queued-interaction-response.js", () => ({ + hasQueuedInteractionResponse: vi.fn(async () => false), +})); + vi.mock("../services/access.js", () => ({ accessService: () => mockAccessService, })); diff --git a/server/src/__tests__/issue-execution-policy-routes.test.ts b/server/src/__tests__/issue-execution-policy-routes.test.ts index 4478356f784..abc52ad4139 100644 --- a/server/src/__tests__/issue-execution-policy-routes.test.ts +++ b/server/src/__tests__/issue-execution-policy-routes.test.ts @@ -74,6 +74,9 @@ const mockRunnerGoalService = vi.hoisted(() => ({ })); function registerModuleMocks() { + vi.doMock("../services/queued-interaction-response.js", () => ({ + hasQueuedInteractionResponse: vi.fn(async () => false), + })); vi.doMock("../services/runner-goals.js", () => ({ runnerGoalService: () => mockRunnerGoalService, RunnerGoalActionError: class RunnerGoalActionError extends Error {}, diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 801c829ac99..9d1f9708b54 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -1,3 +1,5 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { runningProcesses } from "../adapters/index.js"; import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; @@ -13,7 +15,9 @@ import { companySkills, createDb, heartbeatRuns, + heartbeatRunEvents, issueComments, + issueThreadInteractions, issueRecoveryActions, issues, runIdentityContexts, @@ -53,7 +57,14 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { db = createDb(tempDb.connectionString); }, 30_000); + const testProcesses = new Map(); + afterEach(async () => { + for (const [id, child] of testProcesses) { + runningProcesses.delete(id); + child.kill(); + } + testProcesses.clear(); // Each case owns the entire disposable database. Clear the full company // graph, including attribution rows and constraints added by migrations. await db.execute(sql`TRUNCATE TABLE companies CASCADE`); @@ -63,12 +74,13 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { await tempDb?.cleanup(); }); - function app(companyId: string, userId = "queue-owner") { + function app(companyId: string, userId = "queue-owner", agentActor?: { agentId: string; runId: string }) { const testApp = express(); testApp.use(express.json()); testApp.use((req, _res, next) => { (req as any).actor = { - type: "board", + type: agentActor ? "agent" : "board", + ...(agentActor ? { ...agentActor, companyId } : {}), source: "session", userId, companyIds: [companyId], @@ -175,6 +187,289 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { return { companyId, agentId, issueId, runId, wakeId, commentIds }; } + it.each((["request_confirmation", "request_checkbox_confirmation", "ask_user_questions"] as const) + .flatMap(kind => (["legacy", "native"] as const).map(runtime => ({ kind, runtime }))))( + "queues $kind resolution while its $runtime source run is still running", + async ({ kind, runtime }) => { + const seeded = await seedQueue(); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + if (runtime === "legacy") { + 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 child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + testProcesses.set(seeded.runId, child); + runningProcesses.set(seeded.runId, { child, graceSec: 1, processGroupId: null }); + const interactionId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId, + kind, status: "pending", createdByAgentId: seeded.agentId, sourceRunId: seeded.runId, + continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only", + effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit", + payload: kind === "ask_user_questions" + ? { version: 1, questions: [{ id: "runtime", prompt: "Which runtime?", selectionMode: "single", required: true, + options: [{ id: "node", label: "Node.js" }, { id: "python", label: "Python" }] }] } + : kind === "request_checkbox_confirmation" + ? { version: 1, prompt: "Accept scope", options: [{ id: "scope", label: "Build the app" }] } + : { version: 1, prompt: "Build the app?", detailsMarkdown: "Create exactly one child task." }, + }); + const client = app(seeded.companyId); + await request(client).post(`/api/issues/${seeded.issueId}/interactions/${interactionId}/${kind === "ask_user_questions" ? "respond" : "accept"}`) + .send(kind === "ask_user_questions" ? { answers: [{ questionId: "runtime", optionIds: ["node"] }] } + : kind === "request_checkbox_confirmation" ? { selectedOptionIds: ["scope"] } : {}).expect(200); + await vi.waitFor(async () => { + const receipts = await db.select().from(agentWakeupRequests); + expect(receipts.some(row => row.status === "deferred_issue_execution")).toBe(true); + }); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries).toHaveLength(1); + expect(queue.body.entries[0]).toMatchObject({ + source: { kind: "interaction", interactionId, interactionKind: kind }, + canEdit: false, canDiscard: false, + }); + expect(queue.body.targetRunId).toBe(seeded.runId); + expect(queue.body.entries[0].comment.body).toContain(kind === "ask_user_questions" ? "Node.js" : "Accepted"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0]!.status).toBe("running"); + const wakes = await db.select().from(agentWakeupRequests); + expect(wakes).toHaveLength(1); + expect(wakes[0]).toMatchObject({ status: "deferred_issue_execution", runId: null }); + }, + ); + + async function seedResponseQueue(native = false) { + const seeded = await seedQueue(); + const interactionId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId, + kind: "request_confirmation", status: "accepted", createdByAgentId: seeded.agentId, + sourceRunId: seeded.runId, resolvedByUserId: "queue-owner", resolvedAt: new Date(), + continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only", + effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit", + title: "Build the app", payload: { version: 1, prompt: "Build the app?" }, + result: { decision: "accepted" }, + }); + const context = { issueId: seeded.issueId, interactionId, interactionKind: "request_confirmation", + interactionStatus: "accepted", planReviewInteraction: { id: interactionId, + acceptedTargetRevision: { documentId: "plan-1", revisionId: "revision-1" } } }; + await db.update(agentWakeupRequests).set({ payload: { issueId: seeded.issueId, + mutation: "interaction", interactionId, interactionStatus: "accepted", _paperclipWakeContext: context }, + }).where(eq(agentWakeupRequests.id, seeded.wakeId)); + if (!native) { + await db.update(agents).set({ adapterType: "claude_local", + runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } }, + }).where(eq(agents.id, seeded.agentId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId)); + } + return { ...seeded, interactionId, context }; + } + + it("keeps messages and approvals in separate durable queues in either arrival order", async () => { + const seeded = await seedResponseQueue(); + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + testProcesses.set(seeded.runId, child); + runningProcesses.set(seeded.runId, { child, graceSec: 1, processGroupId: null }); + const client = app(seeded.companyId); + await request(client).post(`/api/issues/${seeded.issueId}/comments`) + .send({ body: "Keep this message too" }).expect(201); + await vi.waitFor(async () => expect(await db.select().from(agentWakeupRequests)).toHaveLength(2)); + let wakes = await db.select().from(agentWakeupRequests); + expect(wakes.find(w => w.id === seeded.wakeId)?.payload).not.toHaveProperty("commentId"); + // A second resolved card must not overwrite either existing receipt. + const nextId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: nextId, companyId: seeded.companyId, issueId: seeded.issueId, + kind: "request_confirmation", status: "pending", createdByAgentId: seeded.agentId, + sourceRunId: seeded.runId, continuationPolicy: "wake_assignee", + requestedResolverPolicy: "human_only", effectiveResolverPolicy: "human_only", + resolverPolicyProvenance: "explicit", payload: { version: 1, prompt: "Second approval?" }, + }); + await request(client).post(`/api/issues/${seeded.issueId}/interactions/${nextId}/accept`).send({}).expect(200); + wakes = await db.select().from(agentWakeupRequests); + expect(wakes).toHaveLength(3); + expect(wakes.filter(w => w.payload?.mutation === "interaction").map(w => w.payload?.interactionId).sort()) + .toEqual([seeded.interactionId, nextId].sort()); + expect(wakes.find(w => w.payload?.commentId)?.payload?._paperclipWakeContext).not.toHaveProperty("interactionId"); + }); + + it("recovers an approval acknowledged before the steering transaction failed without redelivery", async () => { + const seeded = await seedResponseQueue(true); + await seedDispatchIdentity(seeded); + // Reconstruct the durable state after acknowledgement and HTTP rollback. + const [pending] = await db.insert(runIdentityContexts).values({ + companyId: seeded.companyId, runId: seeded.runId, revision: 2, + cause: "steering", correlationId: `interaction:${seeded.interactionId}`, + messageId: seeded.interactionId, responsibleUserId: "queue-owner", status: "pending", + }).returning(); + await db.insert(heartbeatRunEvents).values({ + companyId: seeded.companyId, runId: seeded.runId, agentId: seeded.agentId, + seq: 1, eventType: "item.completed", sourceEventId: randomUUID(), + payload: { prpEvent: { turnId: "approval-ack", itemId: `approval-ack:steer:${seeded.interactionId}`, + payload: { kind: "steering_acknowledgement" } } }, + }); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + const body = { queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }; + const endpoint = `/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`; + const count = steerNativeSessionMock.mock.calls.length; + await request(client).post(endpoint).send(body).expect(200); + expect(steerNativeSessionMock.mock.calls).toHaveLength(count); + expect((await db.select().from(runIdentityContexts).where(eq(runIdentityContexts.id, pending.id)))[0].status).toBe("accepted"); + }); + + it("interrupts an active legacy run and delivers the exact approval once", async () => { + const seeded = await seedResponseQueue(); + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + testProcesses.set(seeded.runId, child); + runningProcesses.set(seeded.runId, { child, graceSec: 1, processGroupId: null }); + await db.update(heartbeatRuns).set({ processPid: child.pid }).where(eq(heartbeatRuns.id, seeded.runId)); + // Occupy another task to inspect admission without invoking a paid provider. + 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); + await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`) + .send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }).expect(200); + runningProcesses.delete(seeded.runId); + await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(["queued", "coalesced"]).toContain(wake.status); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)); + expect(successor.status).toBe("queued"); + expect(successor.contextSnapshot).toMatchObject(seeded.context); + await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId); + expect(await db.select().from(heartbeatRuns)).toHaveLength(3); + }); + + it("steers a saved approval only on click and acknowledges retries once", async () => { + const seeded = await seedResponseQueue(true); + await seedDispatchIdentity(seeded); + steerNativeSessionMock.mockResolvedValueOnce({ turnId: "approval-turn" }); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + const body = { queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }; + const endpoint = `/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`; + await request(client).post(endpoint).send(body).expect(200); + const delivered = steerNativeSessionMock.mock.calls.at(-1)![0]; + expect(delivered.message).toContain(seeded.interactionId); + expect(delivered.message).toContain('"status": "accepted"'); + expect((await db.select().from(runIdentityContexts).where(eq(runIdentityContexts.messageId, seeded.interactionId)))[0]) + .toMatchObject({ status: "accepted", responsibleUserId: "queue-owner", cause: "steering" }); + const callCount = steerNativeSessionMock.mock.calls.length; + await request(client).post(endpoint).send(body).expect(200); + expect(steerNativeSessionMock.mock.calls).toHaveLength(callCount); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status).toBe("cancelled"); + expect(await db.select().from(issueComments)).toHaveLength(2); // no forged approval comment + }); + + it.each([false, true])("lets the source run finish review after acceptance, before wake persistence: %s", async (beforeWake) => { + const seeded = await seedResponseQueue(); + if (beforeWake) await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + const client = app(seeded.companyId, "queue-owner", { agentId: seeded.agentId, runId: seeded.runId }); + const result = await request(client).patch(`/api/issues/${seeded.issueId}`).send({ status: "in_review", reviewInteractionId: seeded.interactionId }); + expect(result.status, JSON.stringify(result.body)).toBe(200); + const [task] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(task.assigneeAgentId).toBe(seeded.agentId); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running"); + }); + + it.each(["legacy", "native"] as const)("resumes a queued plan approval across a stopped %s run only after cleanup", async (runtime) => { + const seeded = await seedResponseQueue(runtime === "native"); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, seeded.agentId)); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + await db.update(agentWakeupRequests).set({ payload: { ...receipt.payload, + _paperclipWakeContext: { ...seeded.context, forceFreshSession: true } }, + }).where(eq(agentWakeupRequests.id, seeded.wakeId)); + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), errorCode: "process_lost", processPid: process.pid }) + .where(eq(heartbeatRuns.id, seeded.runId)); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId)); + await db.insert(issueRecoveryActions).values({ companyId: seeded.companyId, sourceIssueId: seeded.issueId, + kind: "active_run_watchdog", cause: runtime === "legacy" ? "legacy_execution_requires_reconciliation" : "native_continuation_requires_reconciliation", + fingerprint: seeded.runId, status: "resolved", outcome: "blocked", nextAction: "Inspect stopped execution", + evidence: { runId: seeded.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }); + 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); + await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`) + .send({ queueId: seeded.wakeId, targetRunId: null, revision: queue.body.revision }).expect(200); + const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(waiting.status).toBe("deferred_issue_execution"); + expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" }); + await db.update(heartbeatRuns).set({ processPid: 999999999 }).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).toBe("coalesced"); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)); + expect(successor.contextSnapshot).toMatchObject({ ...seeded.context, forceFreshSession: true, previousRunId: seeded.runId }); + await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId); + expect(await db.select().from(heartbeatRuns)).toHaveLength(3); + }); + + it("keeps queued approval content immutable through comment mutation endpoints", async () => { + const seeded = await seedResponseQueue(); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + const body = { queueId: seeded.wakeId, revision: queue.body.revision }; + await request(client).patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}`) + .send({ ...body, body: "Reject instead" }).expect(409); + await request(client).delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}`).send(body).expect(409); + await request(client).put(`/api/issues/${seeded.issueId}/queued-comments/order`) + .send({ ...body, orderedCommentIds: [seeded.interactionId] }).expect(409); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interactionId)))[0].status).toBe("accepted"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status).toBe("deferred_issue_execution"); + }); + + it("does not let a viewer steer someone else's approval", async () => { + const seeded = await seedResponseQueue(true); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + await db.update(companyMemberships).set({ membershipRole: "viewer" }).where(eq(companyMemberships.principalId, "queue-owner")); + const calls = steerNativeSessionMock.mock.calls.length; + await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`) + .send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }).expect(403); + expect(steerNativeSessionMock.mock.calls).toHaveLength(calls); + }); + + it("does not steer a plan approval that requires a fresh execution session", async () => { + const seeded = await seedResponseQueue(true); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + await db.update(agentWakeupRequests).set({ payload: { ...wake.payload, + _paperclipWakeContext: { ...seeded.context, forceFreshSession: true } } }).where(eq(agentWakeupRequests.id, seeded.wakeId)); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries[0].source.requiresFreshSession).toBe(true); + const calls = steerNativeSessionMock.mock.calls.length; + const result = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`) + .send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }).expect(409); + expect(result.body.details.code).toBe("queued_response_requires_fresh_session"); + expect(steerNativeSessionMock.mock.calls).toHaveLength(calls); + }); + + it("rejects stale agent handback without cancelling the accepted proposal's run", async () => { + const seeded = await seedResponseQueue(); + await db.update(issues).set({ createdByUserId: "queue-owner" }).where(eq(issues.id, seeded.issueId)); + const client = app(seeded.companyId, "queue-owner", { agentId: seeded.agentId, runId: seeded.runId }); + const result = await request(client).patch(`/api/issues/${seeded.issueId}`) + .send({ status: "in_review", assigneeAgentId: null, assigneeUserId: "queue-owner" }).expect(409); + expect(result.body.details.code).toBe("interaction_response_queued"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running"); + }); + + it("promotes an approval after normal completion without a click", async () => { + const seeded = await seedResponseQueue(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, seeded.runId)); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId)); + await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId, + status: "running", contextSnapshot: { issueId: randomUUID() } }); + await heartbeatService(db).resumeQueuedRuns(); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(wake.status).toBe("queued"); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)); + expect(successor.contextSnapshot).toMatchObject(seeded.context); + await heartbeatService(db).resumeQueuedRuns(); + expect(await db.select().from(heartbeatRuns)).toHaveLength(3); + }); + it.each(["stale revision", "native run", "different issue"] as const)( "rejects queued interruption for a %s without stopping the run", async (scenario) => { diff --git a/server/src/__tests__/question-response-delivery.test.ts b/server/src/__tests__/question-response-delivery.test.ts index 84428b9c29e..d1088780c25 100644 --- a/server/src/__tests__/question-response-delivery.test.ts +++ b/server/src/__tests__/question-response-delivery.test.ts @@ -287,84 +287,27 @@ describeEmbeddedPostgres("question response delivery", () => { }; } - it("persists the receipt atomically and steers exactly once into a running successor", async () => { + it("queues an answer to a running successor instead of automatically steering", async () => { const seeded = await seed({ successorStatus: "running" }); - const newerRunId = randomUUID(); - await db.insert(heartbeatRuns).values({ - id: newerRunId, - companyId: seeded.companyId, - agentId: seeded.agentId, - invocationSource: "manual", - status: "running", - runtimeMode: "native", - driverKind: "codex", - contextSnapshot: { issueId: seeded.issueId }, - startedAt: new Date(), - }); - await db - .update(issues) - .set({ executionRunId: seeded.successorRunId }) - .where(eq(issues.id, seeded.issueId)); - const persistedBeforeDelivery = await db - .select() - .from(issueQuestionResponseDeliveries) - .where( - eq( - issueQuestionResponseDeliveries.interactionId, - seeded.interaction.id, - ), - ) - .then((rows) => rows[0]); - expect(persistedBeforeDelivery).toMatchObject({ - status: "pending", - correlationId: `question-response:${seeded.interaction.id}`, - sourceRunId: seeded.sourceRunId, - }); - - const steer = vi.fn().mockResolvedValue({ turnId: "turn-successor" }); - const wakeup = vi.fn(); - const service = questionResponseDeliveryService(db, { - heartbeat: { wakeup } as never, - steer, + const steer = vi.fn().mockResolvedValue({ turnId: "unexpected" }); + const wakeup = vi.fn().mockImplementation(async () => { + await db.insert(agentWakeupRequests).values({ companyId: seeded.companyId, agentId: seeded.agentId, + source: "automation", reason: "issue_commented", status: "deferred_issue_execution", + idempotencyKey: `question-response:${seeded.interaction.id}`, + payload: { issueId: seeded.issueId, mutation: "interaction", interactionId: seeded.interaction.id, + interactionStatus: "answered" } }); + return null; }); + const service = questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer }); const first = await service.deliver(seeded.interaction.id); const second = await service.deliver(seeded.interaction.id); - - expect(first).toMatchObject({ - status: "delivered", - mode: "steered", - targetRunId: seeded.successorRunId, - targetTurnId: "turn-successor", - duplicate: false, - }); - expect(second).toMatchObject({ mode: "steered", duplicate: true }); - expect(steer).toHaveBeenCalledTimes(1); - expect(steer).toHaveBeenCalledWith( - expect.objectContaining({ - runId: seeded.successorRunId, - correlationId: `question-response:${seeded.interaction.id}`, - message: expect.stringContaining("- Runtime — Which runtime?: Node.js"), - }), - ); - expect(wakeup).not.toHaveBeenCalled(); - - const [delivery] = await db.select().from(issueQuestionResponseDeliveries); - expect(delivery).toMatchObject({ - status: "delivered", - deliveryMode: "steered", - targetRunId: seeded.successorRunId, - targetTurnId: "turn-successor", - attemptCount: 1, - }); - const deliveryEvents = await db - .select() - .from(activityLog) - .where(eq(activityLog.action, "issue.question_response_delivered")); - expect(deliveryEvents).toHaveLength(1); - expect(JSON.stringify(deliveryEvents[0]?.details)).not.toContain( - "Internal API", - ); - expect(JSON.stringify(deliveryEvents[0]?.details)).not.toContain("Node.js"); + expect(steer).not.toHaveBeenCalled(); + expect(wakeup).toHaveBeenCalledTimes(1); + expect(wakeup).toHaveBeenCalledWith(seeded.agentId, expect.objectContaining({ + payload: expect.objectContaining({ interactionId: seeded.interaction.id, interactionStatus: "answered" }), + })); + expect(first).toMatchObject({ status: "fallback_queued", mode: "wake_fallback" }); + expect(second).toMatchObject({ mode: "wake_fallback", duplicate: true }); }); it.each(["native", "legacy"] as const)( @@ -1236,7 +1179,7 @@ describeEmbeddedPostgres("question response delivery", () => { }, ); - it("falls back once when successor steering is unsupported", async () => { + it("queues once without probing successor steering", async () => { const seeded = await seed({ successorStatus: "running" }); const fallbackRunId = randomUUID(); const steer = vi.fn().mockRejectedValue( @@ -1272,7 +1215,7 @@ describeEmbeddedPostgres("question response delivery", () => { targetRunId: fallbackRunId, }); expect(second?.duplicate).toBe(true); - expect(steer).toHaveBeenCalledTimes(1); + expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); }); diff --git a/server/src/modules/wake-queue/application/use-cases.test.ts b/server/src/modules/wake-queue/application/use-cases.test.ts index 07c3e49a584..332cbb77a7c 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -825,6 +825,26 @@ describe("admitWakeBehindIssueExecution", () => { }, ); + it.each(["incoming", "queued", "deferred"])("does not coalesce a %s interaction with comments", async (location) => { + const writer = createFakeAdmissionWriter(); + const reader = createFakeAdmissionReader({ + isSameExecutionAgent: vi.fn(async () => location !== "deferred"), + findExistingDeferredWake: vi.fn(async () => location === "deferred" ? { + id: "existing", payload: { interactionId: "approval" }, + deferredContext: {}, coalescedCount: 0, + } : null), + }); + const admit = createAdmitWakeBehindIssueExecution({ reader, writer, helpers: createFakeAdmissionHelpers() }); + await admit(SCOPE, admissionInput({ + contextSnapshot: location === "incoming" ? { interactionId: "approval" } : {}, + activeExecutionRun: { ...ACTIVE_EXECUTION_RUN, status: "queued", + contextSnapshot: location === "queued" ? { interactionId: "approval" } : {} }, + })); + expect(writer.coalesceIntoActiveExecutionRun).not.toHaveBeenCalled(); + expect(writer.mergeIntoExistingDeferredWake).not.toHaveBeenCalled(); + expect(writer.insertNewDeferredWake).toHaveBeenCalledOnce(); + }); + it("partitions durable admission by the exact actor before considering a deferred merge", async () => { const durableReceipt = { id: "other-actor-receipt", diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 9341f3b5a15..fa24d22b52b 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -11,6 +11,7 @@ import { isConfigurationIncompleteFailedRun, isWorkspaceValidationFailedRun, readNonEmptyString, + parseObject, } from "../domain/values.js"; import type { AdmitWakeBehindIssueExecutionResult, @@ -822,8 +823,12 @@ export function createAdmitWakeBehindIssueExecution(deps: { requestedByActorType: input.requestedByActorType, requestedByActorId: input.requestedByActorId, })); + // Each resolved card has its own immutable decision and continuation. Never + // merge it with comments or another card, in either arrival order. + const interactionReceipt = Boolean(input.contextSnapshot.interactionId || input.payload?.interactionId); const decision = decideWakeAdmission({ - allowRunCoalescing: input.allowRunCoalescing, + allowRunCoalescing: interactionReceipt || parseObject(input.activeExecutionRun.contextSnapshot).interactionId + ? false : input.allowRunCoalescing, sameDurableActor, isSameExecutionAgent, shouldDeferFollowupWake, @@ -865,7 +870,7 @@ export function createAdmitWakeBehindIssueExecution(deps: { // existing deferred wake, so the coalesce path (the common path) never // pays for this query. const existingDeferred = - input.allowRunCoalescing === false + (interactionReceipt || input.allowRunCoalescing === false) ? null : await deps.reader.findExistingDeferredWake(scope, { companyId: input.companyId, @@ -881,7 +886,7 @@ export function createAdmitWakeBehindIssueExecution(deps: { : {}), }); - if (existingDeferred) { + if (existingDeferred && !existingDeferred.payload?.interactionId && !existingDeferred.deferredContext.interactionId) { const mergedDeferredContext = deps.helpers.mergeCoalescedContextSnapshot( existingDeferred.deferredContext, input.contextSnapshot, diff --git a/server/src/onboarding-assets/first-task/skills/first-task/SKILL.md b/server/src/onboarding-assets/first-task/skills/first-task/SKILL.md index f15ffff4fbd..b3fe0d98e4a 100644 --- a/server/src/onboarding-assets/first-task/skills/first-task/SKILL.md +++ b/server/src/onboarding-assets/first-task/skills/first-task/SKILL.md @@ -11,7 +11,11 @@ description: > Use this workflow only for the onboarding task that invokes `/first-task`, including later replies and approval wakes on that same task. Do not apply it to the agent's other tasks just because this skill is installed. Follow it -without announcing the skill invocation to the user. +without announcing the first-task skill in routine messages, cards, or documents. +Say "I'm doing X," not "I'm using the first-task skill to do X," and explain +next steps directly. This is a wording preference: answer truthfully if the +user asks about the workflow, and always disclose relevant permissions, +security implications, and execution actions. This is the user's first task in Paperclip. Your job is to understand what they want and propose a path forward. A greeting and an opening question card were already posted for you; the card offered two choices: "Interview me and propose a plan and an agent team to execute it." (option `interview`) or "I have a task in mind" (option `task`, with a text field). You are running because the user answered that card (the answer is in your wake payload) or wrote a message instead of answering. Don't re-introduce yourself and don't post the opening card again. diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 7f11fbdadcf..e3a2cd38dbd 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1,3 +1,4 @@ +import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse } from "../services/queued-interaction-response.js"; import { deliverConversationComments, isConversation } from "../services/agent-conversations.js"; import { issueRecoveryActionReadModel } from "../services/issue-recovery-actions.js"; import { getExecutionBlocker } from "../services/execution-blocker.js"; @@ -2702,7 +2703,7 @@ async function queueResolvedInteractionContinuationWakeup(input: { publication.idempotencyKey === `interaction:${input.interaction.id}:${publication.endpointId}`, ); - void input.heartbeat + await input.heartbeat .wakeup(input.issue.assigneeAgentId, { source: "automation", triggerDetail: "system", @@ -4616,6 +4617,7 @@ export function issueRoutes( companyId: string; status: string; assigneeUserId?: string | null; + assigneeAgentId?: string | null; executionState?: unknown; monitorNextCheckAt?: Date | null; }; @@ -4641,6 +4643,21 @@ export function issueRoutes( const pendingInteractions = interactions.filter( (interaction) => interaction.status === "pending", ); + // Acceptance can commit just before its continuation wake is persisted. + // The source run may still be finishing its original review handoff. The + // resolved card from that exact run is evidence of the in-flight response; + // an old card from an earlier run is not a new review path. + const resolvedSourceResponse = input.actorType === "agent" && input.actorRunId + ? interactions.find(interaction => + interaction.sourceRunId === input.actorRunId && + interaction.createdByAgentId === input.actorAgentId && + (!input.reviewInteractionId || interaction.id === input.reviewInteractionId) && + ["accepted", "answered", "rejected"].includes(interaction.status) && + (interaction.continuationPolicy === "wake_assignee" || + (interaction.continuationPolicy === "wake_assignee_on_accept" && + ["accepted", "answered"].includes(interaction.status)))) + : undefined; + if (resolvedSourceResponse) return null; if (input.reviewInteractionId) { const designatedReviewConfirmation = pendingInteractions.find( (interaction) => @@ -4704,6 +4721,7 @@ export function issueRoutes( return null; if (pendingInteractions.length > 0) return null; + if (await hasQueuedInteractionResponse(db, input.existing.companyId, input.existing.id, input.existing.assigneeAgentId)) return null; const approvals = await issueApprovalsSvc.listApprovalsForIssue( input.existing.id, @@ -6939,7 +6957,7 @@ export function issueRoutes( for (const wake of rows) { if ( readObject(wake.payload).issueId !== issue.id || - queuedCommentIdsFromWakePayload(wake.payload).length === 0 + (queuedCommentIdsFromWakePayload(wake.payload).length === 0 && !queuedInteractionId(wake.payload)) ) continue; if (wake.status === "deferred_issue_execution") { @@ -6970,6 +6988,10 @@ export function issueRoutes( issueId: string, wake: IssueQueueWake | null, ) { + if (wake && queuedInteractionId(wake.payload)) { + const response = await readQueuedInteractionResponse(executor as Db, wake.companyId, issueId, wake.payload); + return response ? [response.comment] : []; + } const ids = queuedCommentIdsFromWakePayload(wake?.payload); if (ids.length === 0) return []; const rows = await executor @@ -6981,7 +7003,7 @@ export function issueRoutes( const byId = new Map(rows.map((row) => [row.id, row])); return ids.flatMap((id) => { const row = byId.get(id); - return row && !row.deletedAt ? [row] : []; + return row && !row.deletedAt ? [{ ...row, authorType: row.authorType ?? (row.authorAgentId ? "agent" as const : "user" as const) }] : []; }); } @@ -7031,7 +7053,7 @@ export function issueRoutes( .then((state) => state.disposition) .catch(() => "temporarily_unavailable" as const)); const wait = queueState?.state === "deferred" ? readObject(readObject(wake?.payload).executionWait) : {}; - return buildQueuedCommentQueueSnapshot({ + const queue = buildQueuedCommentQueueSnapshot({ issueId: input.issue.id, executionWait: typeof wait.reason === "string" && typeof wait.message === "string" ? { reason: wait.reason, message: wait.message } : null, @@ -7044,6 +7066,12 @@ export function issueRoutes( actorType: input.actor.actorType, actorId: input.actor.actorId, }); + if (wake && queuedInteractionId(wake.payload)) { + const response = await readQueuedInteractionResponse(input.executor as Db, input.issue.companyId, input.issue.id, wake.payload); + if (response?.source.requiresFreshSession) queue.steeringDisposition = "unsupported"; + queue.entries = response ? [{ comment: response.comment, source: response.source, position: 0, canEdit: false, canDiscard: false }] : []; + } + return queue; } function assertQueueMutationTarget(input: { @@ -7106,7 +7134,7 @@ export function issueRoutes( if ( !wake || readObject(wake.payload).issueId !== input.issue.id || - queuedCommentIdsFromWakePayload(wake.payload).length === 0 + (queuedCommentIdsFromWakePayload(wake.payload).length === 0 && !queuedInteractionId(wake.payload)) ) { throw conflict("The queued message is no longer pending", { code: "queued_comment_not_pending", @@ -13340,6 +13368,13 @@ export function issueRoutes( !!existing.createdByUserId && nextAssigneeUserId === existing.createdByUserId; + if (assigneeWillChange && actor.actorType === "agent" && + existing.assigneeAgentId === actor.agentId && updateFields.status === "in_review" && + await hasQueuedInteractionResponse(db, existing.companyId, existing.id, existing.assigneeAgentId)) { + throw conflict("The user already responded. Keep the current assignee so the queued response can continue after this run.", { + code: "interaction_response_queued", + }); + } if (assigneeWillChange && !transition.workflowControlledAssignment) { if (!isAgentReturningIssueToCreator) { await assertCanAssignTasks(req, existing.companyId, { @@ -15383,10 +15418,11 @@ export function issueRoutes( allowStoppedTarget: true, }); assertQueueMutationTarget({ queue: locked.queue, queueId: req.body.queueId, revision: req.body.revision }); - if (locked.queue.protocol !== "legacy" || locked.state !== "deferred" || + const freshResponse = locked.queue.entries.some(entry => entry.source?.requiresFreshSession); + if ((locked.queue.protocol !== "legacy" && !freshResponse) || locked.state !== "deferred" || !locked.queue.entries.length || (locked.activeRun && locked.activeRun.agentId !== locked.wake.agentId)) { - throw conflict("This queue does not support legacy interruption"); + throw conflict("This queue does not support interruption"); } if (locked.activeRun && locked.activeRun.status !== "running" && !["succeeded", "failed", "timed_out", "interrupted", "cancelled"].includes(locked.activeRun.status)) { @@ -15445,12 +15481,19 @@ export function issueRoutes( ); if (!issue) return; if (issue.conversationAgentId) throw conflict("Conversation messages are processed in order at turn boundaries"); + const decision = await decideIssueAccess(req, issue, "issue:comment"); + if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); const actor = getActorInfo(req); + const responseWake = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, req.body.queueId), eq(agentWakeupRequests.companyId, issue.companyId), + )).then(rows => rows[0]); + const response = responseWake ? await readQueuedInteractionResponse(db, issue.companyId, issue.id, responseWake.payload) : null; const steeringIdentity = await reserveSteeredIdentity(db, { companyId: issue.companyId, runId: req.body.targetRunId, issueId: issue.id, messageId: commentId, + source: response?.comment.id === commentId ? "interaction" : "comment", }); let steeringDeliveryAttempted = false; let acknowledgedTurnId: string | null = null; @@ -15584,11 +15627,16 @@ export function issueRoutes( }); } + if (entry.source?.requiresFreshSession) { + throw conflict("This approval needs a fresh turn. Interrupt or wait for the current turn to finish.", { + code: "queued_response_requires_fresh_session", + }); + } steeringDeliveryAttempted = true; const acknowledgement = - (steeringIdentity - ? await storedSteeringAcknowledgement(tx, steeringIdentity) - : null) ?? + (await storedSteeringAcknowledgement(tx, steeringIdentity ?? { + companyId: issue.companyId, runId: locked.activeRun.id, messageId: commentId, + })) ?? (await steerNativeSession({ runId: locked.activeRun.id, message: entry.comment.body, diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index fecb224ca00..b9cff216ac5 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,3 +1,4 @@ +import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; import { isCancelledNativeStartup } from "./cancelled-native-startup.js"; import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; @@ -83,7 +84,7 @@ export async function admitExplicitNativeContinuation(input: { if (input.actorType !== "user" || !actorId) return null; const retry = input.reason === "retry_failed_run" && z.string().guid().safeParse(input.failedRunId).success; - if (!retry && (!commentId || !z.string().guid().safeParse(commentId).success || + if (!retry && !input.queuedCommentInterruptId && (!commentId || !z.string().guid().safeParse(commentId).success || !["issue_commented", "issue_reopened_via_comment"].includes(input.reason ?? ""))) return null; const [task] = await db.select().from(issues).where(and( eq(issues.companyId, companyId), eq(issues.id, issueId), @@ -96,8 +97,10 @@ export async function admitExplicitNativeContinuation(input: { sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`, )) : []; - const queuedInterrupt = Boolean(interruptQueue && commentId && - queuedCommentIdsFromWakePayload(interruptQueue.payload).includes(commentId)); + const response = interruptQueue + ? await readQueuedInteractionResponse(db, companyId, issueId, interruptQueue.payload) : null; + const queuedInterrupt = Boolean(interruptQueue && (response || (commentId && + queuedCommentIdsFromWakePayload(interruptQueue.payload).includes(commentId)))); if (input.queuedCommentInterruptId && !queuedInterrupt) return null; const [savedQueue] = input.queuedCommentRequestId ? await db.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.id, input.queuedCommentRequestId), @@ -114,17 +117,17 @@ export async function admitExplicitNativeContinuation(input: { const undelivered = await undeliveredLegacyUserCommentIds(db, companyId, issueId, agentId, ids); if (undelivered.length !== ids.length) return null; } - const [comment] = retry ? [] : await db.select().from(issueComments).where(and( + const [comment] = retry || response ? [] : await db.select().from(issueComments).where(and( eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), eq(issueComments.id, commentId!), eq(issueComments.authorType, "user"), queuedInterrupt ? undefined : eq(issueComments.authorUserId, actorId), isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt), )); - if (!retry && !comment?.body.trim()) return null; - const authorizedAt = comment?.createdAt ?? new Date(); + if (!retry && !response && !comment?.body.trim()) return null; + const authorizedAt = response?.comment.createdAt ?? comment?.createdAt ?? new Date(); const [agent] = await db.select().from(agents).where(and(eq(agents.companyId, companyId), eq(agents.id, agentId))); if (!agent || (!isConversationAdapter(agent.adapterType) && agent.adapterType !== "paperclip_runner")) return null; - if (queuedInterrupt && !isConversationAdapter(agent.adapterType)) return null; + if (queuedInterrupt && !isConversationAdapter(agent.adapterType) && !response?.source.requiresFreshSession) return null; const actions = await db.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), executionBlockerPredicate(), @@ -162,7 +165,8 @@ export async function admitExplicitNativeContinuation(input: { const legacyUserTurn = run.runtimeMode === "legacy" && action.cause === "legacy_execution_requires_reconciliation" && isConversationAdapter(agent.adapterType); - if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission) return null; + if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission && + !(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) || @@ -236,7 +240,7 @@ export async function admitExplicitNativeContinuation(input: { context: { previousRunId: previous.id, wakeCommentId: commentId }, summary: null, exposeLowTrustRaw: false }); if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) }; - const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}), + const authorization = { actorId, commentId, ...(response ? { interactionId: response.source.interactionId } : {}), ...(retry ? { failedRunId: input.failedRunId } : {}), ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}), ...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId, previousRunId: previous.id, recordedAt: new Date().toISOString() }; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 3efdbbf5cd8..abb5961d5f2 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; import { AGENT_CHAT_DIRECTIVE, conversationReplay, isConversation, isConversationExecutionWake, isWaitingConversation, prepareConversationTurn, settleConversationTurn } from "./agent-conversations.js"; import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js"; import { hasAcknowledgedNativeStopIntent, isAcknowledgedNativeStop, acknowledgedNativeStopExecutionHasStopped } from "./acknowledged-native-stop.js"; @@ -10256,7 +10257,13 @@ export function heartbeatService( let actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId); let commentIds = queuedCommentIdsFromWakePayload(payload); const issueId = readNonEmptyString(payload.issueId); - if (!issueId || !commentIds.length || wake.idempotencyKey?.startsWith("chat-inbound:")) return; + if (!issueId || wake.idempotencyKey?.startsWith("chat-inbound:")) return; + const response = await readQueuedInteractionResponse(db, companyId, issueId, payload); + if (!commentIds.length && !response) return; + // Resolved cards are immutable input. Only an explicit Interrupt click can + // authorize continuation across a stopped execution; ordinary completion + // uses normal deferred-wake promotion. + if (response && !interrupted) return; if (!interrupted) { commentIds = await undeliveredLegacyUserCommentIds(db, companyId, issueId, wake.agentId, commentIds); if (!commentIds.length) return; @@ -10272,7 +10279,7 @@ export function heartbeatService( } if (!actorId) return; const agent = await getAgent(wake.agentId); - if (!agent || agent.companyId !== companyId || agent.adapterType === "paperclip_runner") return; + if (!agent || agent.companyId !== companyId || (agent.adapterType === "paperclip_runner" && !response?.source.requiresFreshSession)) return; const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, wake.agentId), @@ -10294,7 +10301,8 @@ export function heartbeatService( sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`, )); if (!task || task.assigneeAgentId !== wake.agentId || ["done", "cancelled"].includes(task.status) || - !current || !queuedCommentIdsFromWakePayload(current.payload).length) return null; + !current || (!queuedCommentIdsFromWakePayload(current.payload).length && + !await readQueuedInteractionResponse(tx as unknown as Db, companyId, issueId, current.payload))) return null; const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, wake.agentId), @@ -10338,13 +10346,16 @@ export function heartbeatService( companyId, runId: sourceRun.id, actorId, reason: "queued_comment_interrupt", } }); } - const deliveryPayload = withQueuedCommentIdsInWakePayload(payload, commentIds); + const deliveryPayload = response ? { ...payload } : withQueuedCommentIdsInWakePayload(payload, commentIds); delete deliveryPayload.queuedCommentInterrupt; await enqueueWakeup(wake.agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", - payload: deliveryPayload, contextSnapshot: withQueuedCommentIdsInRunContext({ - issueId, triggeredBy: "board", actorId, responsibleUserId: actorId, - }, commentIds), + payload: deliveryPayload, contextSnapshot: response + ? { ...parseObject(payload._paperclipWakeContext), issueId, triggeredBy: "board", actorId, + responsibleUserId: actorId } + : withQueuedCommentIdsInRunContext({ + issueId, triggeredBy: "board", actorId, responsibleUserId: actorId, + }, commentIds), requestedByActorType: "user", requestedByActorId: actorId, ...(interrupted ? { queuedCommentInterruptId: queueId } : { queuedCommentRequestId: queueId }), issueStateGuard: { assigneeAgentId: wake.agentId, statuses: ["todo", "in_progress", "in_review", "blocked"] }, @@ -19185,13 +19196,16 @@ export function heartbeatService( .innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active"))) .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(issues.executionRunId), - sql`jsonb_typeof(${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}') = 'array'`, - sql`${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`, + or(and( + sql`jsonb_typeof(${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}') = 'array'`, + sql`${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`, + ), sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`), sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`, cutoff ? gte(agentWakeupRequests.requestedAt, cutoff) : undefined)) .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); for (const { wake } of strandedQueues) { - if (!queuedCommentIdsFromWakePayload(wake.payload).length) continue; + if (!queuedCommentIdsFromWakePayload(wake.payload).length && + !await readQueuedInteractionResponse(db, wake.companyId, String(wake.payload?.issueId), wake.payload)) continue; const [latest] = await db.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, wake.companyId), eq(heartbeatRuns.agentId, wake.agentId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${String(wake.payload?.issueId)}`, @@ -26401,13 +26415,15 @@ export function heartbeatService( )); // The issue lock serializes cleanup callbacks and periodic workers. // An adopted, discarded, or edited receipt is no longer authority. - if (!pending || !wakeCommentId || !queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) { + if (!pending || (!(wakeCommentId && queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) && + !(opts.queuedCommentInterruptId && await readQueuedInteractionResponse(tx as unknown as Db, + agent.companyId, issueId, pending.payload)))) { return { kind: "deferred" as const }; } if (opts.queuedCommentRequestId) { const ids = await undeliveredLegacyUserCommentIds(tx as unknown as Db, agent.companyId, issueId, agentId, queuedCommentIdsFromWakePayload(pending.payload)); - if (!ids.includes(wakeCommentId)) return { kind: "deferred" as const }; + if (!wakeCommentId || !ids.includes(wakeCommentId)) return { kind: "deferred" as const }; pending.payload = withQueuedCommentIdsInWakePayload(parseObject(pending.payload), ids); await tx.update(agentWakeupRequests).set({ payload: pending.payload }).where(and( eq(agentWakeupRequests.id, pending.id), eq(agentWakeupRequests.companyId, agent.companyId), diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index 2f83677b9c7..5db1e051508 100644 --- a/server/src/services/question-response-delivery.ts +++ b/server/src/services/question-response-delivery.ts @@ -82,7 +82,7 @@ export interface QuestionResponseDeliveryOutcome { export interface QuestionResponseDeliveryServiceOptions { heartbeat: Heartbeat; - /** Optional native steering seam. Direct adapters use the durable wake fallback. */ + /** Kept for caller compatibility. Answers never implicitly steer an active turn. */ steer?: QuestionResponseSteer; /** Resolve the original in-flight native input request before considering a continuation run. */ resolveNativeQuestion?: NativeQuestionResponseResolver; @@ -92,18 +92,6 @@ export interface QuestionResponseDeliveryServiceOptions { claimRefreshMs?: number; } -function readSteeringErrorCode(error: unknown): string { - if ( - error && - typeof error === "object" && - "code" in error && - typeof error.code === "string" - ) { - return error.code; - } - return "steering_rejected"; -} - function record(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? (value as Record) @@ -431,7 +419,6 @@ export function questionResponseDeliveryService( db: Db, options: QuestionResponseDeliveryServiceOptions, ) { - const steer = options.steer; const resolveNativeQuestion = options.resolveNativeQuestion; const now = options.now ?? (() => new Date()); const claimStaleMs = Math.max( @@ -1152,39 +1139,10 @@ export function questionResponseDeliveryService( } } - let steeringErrorCode: string | null = null; - if ( - !externalChatBoundary && - successorRunning?.runtimeMode === "native" && - steer - ) { - try { - const acknowledgement = await withClaimLease(claimed, () => - steer({ - runId: successorRunning.id, - message: formatQuestionResponseSteeringMessage(envelope), - correlationId: claimed.correlationId, - }), - ); - return recordTerminal({ - delivery: claimed, - interaction, - status: "delivered", - mode: "steered", - targetRunId: successorRunning.id, - targetTurnId: acknowledgement.turnId, - adapter: successorRunning.driverKind ?? adapter, - }); - } catch (error) { - if (error instanceof DeliveryClaimUnavailableError) - return terminalOutcome(interactionId); - steeringErrorCode = readSteeringErrorCode(error); - } - } else if (successorRunning) { - steeringErrorCode = externalChatBoundary - ? "steering_external_chat_context_incompatible" - : "steering_unsupported"; - } + // An answer to an older question is new input, not implicit permission to + // steer another active turn. Only the queue's explicit Steer action delivers it. + const steeringErrorCode = successorRunning && externalChatBoundary + ? "steering_external_chat_context_incompatible" : null; const actor = actorForInteraction(interaction); // This is a new, migration-fenced namespace. The partial unique index on diff --git a/server/src/services/queued-interaction-response.ts b/server/src/services/queued-interaction-response.ts new file mode 100644 index 00000000000..ec0794a80b5 --- /dev/null +++ b/server/src/services/queued-interaction-response.ts @@ -0,0 +1,67 @@ +import { and, eq, inArray, sql } from "drizzle-orm"; +import { agentWakeupRequests, issueThreadInteractions, type Db } from "@paperclipai/db"; +import type { IssueComment, IssueQueuedCommentEntry } from "@paperclipai/shared"; + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +/** Interaction receipts keep their typed payload; they never become editable comments. */ +export function queuedInteractionId(payload: unknown): string | null { + const p = record(payload); + return p.mutation === "interaction" && typeof p.interactionId === "string" + && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(p.interactionId) + && ["accepted", "answered", "rejected"].includes(String(p.interactionStatus)) + && p.externalChatContinuation !== true + ? p.interactionId : null; +} + +export async function readQueuedInteractionResponse(db: Db, companyId: string, issueId: string, payload: unknown) { + const id = queuedInteractionId(payload); + if (!id) return null; + const [interaction] = await db.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, id), eq(issueThreadInteractions.companyId, companyId), + eq(issueThreadInteractions.issueId, issueId), + inArray(issueThreadInteractions.status, ["accepted", "answered", "rejected"]), + )); + if (!interaction || interaction.status !== record(payload).interactionStatus) return null; + const p = record(interaction.payload); + const result = record(interaction.result); + const label = interaction.status === "accepted" ? "Accepted" : interaction.status === "rejected" ? "Rejected" : "Answered"; + const heading = `${label}: ${interaction.title ?? p.prompt ?? "Questions"}`; + // Preserve the question labels, selected options, custom answers, approval + // target and exact revision. This is also the explicit steering message. + const body = `${heading}\n\n${JSON.stringify({ + interactionId: interaction.id, kind: interaction.kind, status: interaction.status, + payload: interaction.payload, result, + }, null, 2)}`; + const comment: IssueComment = { + id: interaction.id, companyId, issueId, authorType: interaction.resolvedByUserId ? "user" : "agent", + authorUserId: interaction.resolvedByUserId, authorAgentId: interaction.resolvedByAgentId, + body, presentation: null, metadata: null, + createdAt: interaction.resolvedAt ?? interaction.updatedAt, updatedAt: interaction.updatedAt, + }; + const source: NonNullable = { + kind: "interaction", interactionId: interaction.id, interactionKind: interaction.kind, + requiresFreshSession: record(record(payload)._paperclipWakeContext).forceFreshSession === true, + }; + return { comment, source }; +} + +/** A saved answer is a live continuation, even though its card is no longer pending. */ +export async function hasQueuedInteractionResponse(db: Db, companyId: string, issueId: string, agentId?: string | null) { + if (!agentId) return false; + const [row] = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests) + .innerJoin(issueThreadInteractions, and( + eq(issueThreadInteractions.companyId, agentWakeupRequests.companyId), + sql`${issueThreadInteractions.id}::text = ${agentWakeupRequests.payload}->>'interactionId'`, + eq(issueThreadInteractions.issueId, issueId), + inArray(issueThreadInteractions.status, ["accepted", "answered", "rejected"]), + )).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId), + inArray(agentWakeupRequests.status, ["deferred_issue_execution", "queued"]), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`, + )).limit(1); + return Boolean(row); +} diff --git a/server/src/services/run-identity.ts b/server/src/services/run-identity.ts index 2bfe88ad828..dfc5f5bb896 100644 --- a/server/src/services/run-identity.ts +++ b/server/src/services/run-identity.ts @@ -280,6 +280,7 @@ export async function prepareSteeredIdentity( runId: string; messageId: string; issueId: string; + source?: "comment" | "interaction"; }, ) { const [run] = await executor @@ -291,7 +292,7 @@ export async function prepareSteeredIdentity( eq(heartbeatRuns.companyId, input.companyId), ), ); - const [comment] = await executor + const [comment] = input.source === "interaction" ? [] : await executor .select() .from(issueComments) .where( @@ -301,7 +302,14 @@ export async function prepareSteeredIdentity( eq(issueComments.issueId, input.issueId), ), ); - if (!run || !comment?.authorUserId) + const [interaction] = input.source === "interaction" ? await executor.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, input.messageId), + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + inArray(issueThreadInteractions.status, ["accepted", "answered", "rejected"]), + )) : []; + const responsibleUserId = interaction?.resolvedByUserId ?? comment?.authorUserId; + if (!run || !responsibleUserId) throw forbidden("Steering requires an authenticated message author"); if ( run.status !== "running" || @@ -313,11 +321,11 @@ export async function prepareSteeredIdentity( return append(executor, { companyId: input.companyId, runId: input.runId, - responsibleUserId: comment.authorUserId, - messageId: comment.id, + responsibleUserId, + messageId: input.messageId, parentContextId: run.activeIdentityContextId, cause: "steering", - correlationId: `message:${comment.id}`, + correlationId: `${input.source === "interaction" ? "interaction" : "message"}:${input.messageId}`, status: "pending", }); } @@ -513,7 +521,7 @@ export async function reconcileSteeredIdentity( /** Only events validated and persisted by the native control-plane transport count. */ export async function storedSteeringAcknowledgement( executor: Pick, - context: RunIdentityContext, + context: Pick, ) { if (!context.messageId) return null; const [receipt] = await executor diff --git a/tests/runner-e2e/README.md b/tests/runner-e2e/README.md index f9c7f70884c..8111948856a 100644 --- a/tests/runner-e2e/README.md +++ b/tests/runner-e2e/README.md @@ -546,7 +546,7 @@ Revision delivery checks exclude preserved originals by their content hash, even `first-task` is a suite in the main Runner E2E catalog. A full `pnpm test:e2e:runner -- --all` run (or an unfiltered full GitHub Actions campaign) -includes its 48 executions alongside the other suites in one shared dashboard, +includes its 52 executions alongside the other suites in one shared dashboard, campaign result bundle, and history entry. Suite/profile selectors narrow that same harness; they do not invoke a separate onboarding reporting program. @@ -558,8 +558,8 @@ select the Codex or Claude adapter family; **the production onboarding model default is retained**, even when it differs from that profile's normal harness model. Configured and provider-observed model identities are reported separately. -There are twelve cases on `legacy-codex`, `legacy-claude`, `runner-codex`, and -`runner-acpx-claude`, local only (48 cells). Native profiles complete the same +There are thirteen cases on `legacy-codex`, `legacy-claude`, `runner-codex`, and +`runner-acpx-claude`, local only (52 cells). Native profiles complete the same production wizard using their legacy provider, then change only the agent's runtime configuration via the public API before its first task. The wizard does not currently offer native Runner. Persona, managed instructions, skills, seeded @@ -576,6 +576,7 @@ path exists. Legacy setup is labeled `production-wizard`. | --- | --- | | `interview-first-response` | `interview-plan-accept` | | `clear-task-first-response` | `task-card-accept` | +| | `accept-while-running` | | `ambiguous-task-first-response` | `task-reply-accept` | | `plain-message-first-response` | `clarify-propose-accept` | | `plan-first-response` | `revise-accept` | @@ -685,3 +686,13 @@ identities and instruction hashes before comparing; dirty working trees are explicitly marked. Do not pool results with mismatched configurations or treat infrastructure failures as behavioral successes. Daytona, simulated-user models and prompt optimization are intentionally deferred. + + +`accept-while-running` clicks a confirmation as soon as its source run exposes +one, without the usual wait for that run to settle. It retains the normal +acceptance, child-task, duplicate-work, and durable-output checks. The additional +`accepted-while-running` matcher compares the persisted card resolution time +with the source run's start and finish times. If the model finishes before the +click lands, the case is unexercised, never a passing concurrency regression. +Provider-free route tests also hold a real child process open to exercise this +interleaving deterministically for confirmations, checkbox approvals, and answers. diff --git a/tests/runner-e2e/catalog.test.ts b/tests/runner-e2e/catalog.test.ts index 1a35aae22c6..3bb4fc55960 100644 --- a/tests/runner-e2e/catalog.test.ts +++ b/tests/runner-e2e/catalog.test.ts @@ -41,10 +41,10 @@ describe("runner E2E catalog", () => { expect(localIntegrityTasks).toHaveLength(2); expect(openRouterBreadthTasks).toHaveLength(3); expect(runnerSuites.map((suite) => suite.expectedMatrixSize)).toEqual([ - 35, 48, 24, 42, 14, 10, 2, + 35, 52, 24, 42, 14, 10, 2, ]); - expect(validateRunnerCatalog()).toHaveLength(175); - expect(new Set(runnerMatrix.map((entry) => entry.id)).size).toBe(175); + expect(validateRunnerCatalog()).toHaveLength(179); + expect(new Set(runnerMatrix.map((entry) => entry.id)).size).toBe(179); expect( runnerMatrix.filter((entry) => entry.suite.id === "core-compatibility"), ).toHaveLength(42); @@ -68,7 +68,7 @@ describe("runner E2E catalog", () => { (total, execution) => total + execution.task.expectedRunCount, 0, ), - ).toBe(288); + ).toBe(300); expect( runnerTasks.find((task) => task.id === "plan-revise-accept") ?.attemptTimeoutMs, @@ -560,10 +560,10 @@ describe("runner E2E selectors", () => { const jobs = buildMatrixJobs( selectRunnerExecutions(parseRunnerSelectors(["--all"])), ); - expect(jobs).toHaveLength(140); + expect(jobs).toHaveLength(144); expect(jobs.filter((job) => job.needsDaytona)).toHaveLength(23); - expect(jobs.filter((job) => !job.needsDaytona)).toHaveLength(117); - expect(new Set(jobs.map((job) => job.executionId)).size).toBe(140); + expect(jobs.filter((job) => !job.needsDaytona)).toHaveLength(121); + expect(new Set(jobs.map((job) => job.executionId)).size).toBe(144); expect( jobs.find( (job) => diff --git a/tests/runner-e2e/catalog.ts b/tests/runner-e2e/catalog.ts index da8270d4102..fd2cdb0773a 100644 --- a/tests/runner-e2e/catalog.ts +++ b/tests/runner-e2e/catalog.ts @@ -907,7 +907,7 @@ export const runnerSuites: readonly RunnerSuiteFixture[] = [ description: "Production onboarding, first replies, approval, and durable task execution.", groups: ["onboarding"], profiles: runnerProfiles.filter(profile => ["legacy-codex", "legacy-claude", "runner-codex", "runner-acpx-claude"].includes(profile.id)), - environments: [localEnvironment], tasks: firstTaskTasks, expectedMatrixSize: 48, + environments: [localEnvironment], tasks: firstTaskTasks, expectedMatrixSize: 52, definitionMetadata: { version: 3, credentialPersistenceCheck: false, questionChoiceMinimum: 2, nativeSetup: "post-onboarding-runtime-switch", productionInstructions: true, qualityGrading: "informational" }, }, { diff --git a/tests/runner-e2e/first-task-cases.ts b/tests/runner-e2e/first-task-cases.ts index 9bead960238..8244f0ead12 100644 --- a/tests/runner-e2e/first-task-cases.ts +++ b/tests/runner-e2e/first-task-cases.ts @@ -24,6 +24,7 @@ export const FIRST_TASK_CASES = [ ], ["interview-plan-accept", "Interview, plan, and acceptance", "interview", 3], ["task-card-accept", "Subtask accepted through a card", "task", 3], + ["accept-while-running", "Accept a proposal while its agent is still running", "task", 3], ["task-reply-accept", "Subtask accepted in conversation", "task", 3], [ "clarify-propose-accept", diff --git a/tests/runner-e2e/first-task-flow.ts b/tests/runner-e2e/first-task-flow.ts index dfe923162a6..ed5f5d89897 100644 --- a/tests/runner-e2e/first-task-flow.ts +++ b/tests/runner-e2e/first-task-flow.ts @@ -430,10 +430,23 @@ export async function runFirstTaskFlow(input: { }) .last() .click(); - await settle(before); + if (scenario.id === "accept-while-running") { + await pollUntil({ + label: "approval card published before the source run finishes", + deadlineAt: Math.min(deadlineAt, Date.now() + 300_000), intervalMs: 100, + load: async () => ({ interactions: await api.get(`/api/issues/${issue.id}/interactions`), runs: await allRuns() }), + accept: ({ interactions, runs }) => interactions.some(i => i.status === "pending" && + ["request_confirmation", "request_checkbox_confirmation"].includes(i.kind) && + runs.some(r => r.id === i.sourceRunId && r.status === "running")), + reject: ({ runs }) => runs.some(r => !before.has(r.id)) && !activeRuns(runs).length + ? "Acceptance overlap was not exercised: source run finished before a live approval card was observed" : undefined, + }); + } else await settle(before); await snapshot("response"); } - await input.capture( + // Screenshots can take longer than the source turn's final handoff. In the + // overlap case, accept first and retain the response checkpoint as evidence. + if (scenario.id !== "accept-while-running") await input.capture( "first-task-response", "First onboarding response", "first-task-response.png", @@ -508,7 +521,7 @@ export async function runFirstTaskFlow(input: { assertBeforeAcceptance(); if (scenario.id === "reject-no-execution") await turn(scenario.rejection, "rejected"); - else if (scenario.id === "task-card-accept") { + else if (["task-card-accept", "accept-while-running"].includes(scenario.id)) { const pending = ( await api.get(`/api/issues/${issue.id}/interactions`) ).find( diff --git a/tests/runner-e2e/first-task-scoring.ts b/tests/runner-e2e/first-task-scoring.ts index 1ccb68fc4d7..79531d3aec7 100644 --- a/tests/runner-e2e/first-task-scoring.ts +++ b/tests/runner-e2e/first-task-scoring.ts @@ -335,6 +335,20 @@ export function gradeFirstTask(e: FirstTaskEvidence): FirstTaskCheck[] { [last.id], ); } + if (e.caseId === "accept-while-running") { + const accepted = approval?.interactions.find(i => i.status === "accepted" && + ["request_confirmation", "request_checkbox_confirmation"].includes(i.kind)); + const source = last.runs.find(r => r.id === accepted?.sourceRunId); + const acceptedAt = Date.parse(accepted?.resolvedAt ?? ""); + const startedAt = Date.parse(source?.startedAt ?? ""); + const finishedAt = Date.parse(source?.finishedAt ?? ""); + const overlapped = Number.isFinite(acceptedAt) && Number.isFinite(startedAt) && + Number.isFinite(finishedAt) && startedAt <= acceptedAt && acceptedAt < finishedAt; + add("accepted-while-running", overlapped, + "The persisted approval resolution must fall inside its source run's actual execution interval", + approval ? [approval.id, last.id] : [last.id], + overlapped ? undefined : "The recording did not prove acceptance during the source run; the concurrency regression was not exercised."); + } if (!scenario.firstResponseOnly && e.caseId !== "reject-no-execution") { const notReached = approval ? undefined diff --git a/tests/runner-e2e/first-task.test.ts b/tests/runner-e2e/first-task.test.ts index 06d3e7ac5bf..9f7dc5d86dc 100644 --- a/tests/runner-e2e/first-task.test.ts +++ b/tests/runner-e2e/first-task.test.ts @@ -777,10 +777,10 @@ Accept the card above and I write it. This task stays in review until then.`; }, ); - it("selects exactly 48 local cells with one worker by default", () => { + it("selects exactly 52 local cells with one worker by default", () => { const options = parseRunnerSelectors(["--suite", "first-task"]); const cells = selectRunnerExecutions(options); - expect(cells).toHaveLength(48); + expect(cells).toHaveLength(52); expect(options.maxParallel).toBe(1); expect(new Set(cells.map((c) => c.profile.id))).toEqual( new Set([ @@ -812,7 +812,7 @@ Accept the card above and I write it. This task stays in review until then.`; ).toHaveLength(1); }); it("keeps fixed facts and stable case identities for separate campaigns", () => { - expect(FIRST_TASK_CASES).toHaveLength(12); + expect(FIRST_TASK_CASES).toHaveLength(13); expect( FIRST_TASK_CASES.map((c) => firstTaskScenario(c[0], "same")), ).toEqual(FIRST_TASK_CASES.map((c) => firstTaskScenario(c[0], "same"))); @@ -980,7 +980,7 @@ describe("first-task informational judging and reporting", () => { const all = selectRunnerExecutions(parseRunnerSelectors(["--all"])); expect( all.filter((execution) => execution.suite.id === "first-task"), - ).toHaveLength(48); + ).toHaveLength(52); const core = all.find( (execution) => execution.suite.id === "core-compatibility", )!; @@ -1199,3 +1199,20 @@ describe("first-task informational judging and reporting", () => { } }); }); + + +describe("accept-while-running overlap evidence", () => { + it.each([true, false])("requires persisted execution overlap: %s", (overlap) => { + const e = recording("accept-while-running"); + const accepted = e.checkpoints.find(c => c.phase === "accepted")!; + accepted.interactions.push({ id: "approval", kind: "request_confirmation", status: "accepted", + sourceRunId: "parent-run", resolvedAt: "2026-09-15T00:02:00Z", result: { outcome: "accepted" } }); + const last = e.checkpoints.at(-1)!; + Object.assign(last.runs.find(r => r.id === "parent-run")!, { + startedAt: "2026-09-15T00:01:00Z", finishedAt: overlap ? "2026-09-15T00:02:01Z" : "2026-09-15T00:01:59Z", + }); + const check = gradeFirstTask(e).find(c => c.id === "accepted-while-running")!; + expect(check.passed).toBe(overlap); + expect(Boolean(check.notReached)).toBe(!overlap); + }); +}); diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx index 929020b1311..8595554a717 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx @@ -103,6 +103,27 @@ describe("TaskChatQueuedMessages", () => { expect(container.textContent).toContain("Second queued message"); }); + it.each(["legacy", "native", "native-plan"] as const)("shows a read-only response and sends it only on click for %s", async (runtime) => { + const entry = { ...queue.entries[0], canEdit: false, canDiscard: false, + comment: { ...queue.entries[0].comment, body: 'Accepted: Build the app\n\n{"revision": "v1"}' }, + source: { kind: "interaction" as const, interactionId: "confirmation-1", interactionKind: "request_confirmation", + requiresFreshSession: runtime === "native-plan" } }; + const props = render({ queue: { ...queue, entries: [entry], + protocol: runtime === "legacy" ? "legacy" : "paperclip_runner_v1" }, + onInterrupt: vi.fn().mockResolvedValue(undefined) }); + expect(props.onSteer).not.toHaveBeenCalled(); + expect(props.onInterrupt).not.toHaveBeenCalled(); + expect(container.textContent).toContain("Accepted: Build the app"); + expect(container.textContent).not.toContain('"revision"'); + expect(container.querySelector('[aria-label^="Reorder"]')?.disabled).toBe(true); + expect(container.querySelector('[data-testid^="task-chat-queued-discard-"]')?.disabled).toBe(true); + expect(reorderQueuedMessageEntries([entry, queue.entries[1]], "comment-1", "comment-2")).toBeNull(); + const action = runtime === "native" ? "steer" : "interrupt"; + await act(async () => { container.querySelector(`[data-testid="task-chat-queued-${action}-comment-1"]`)!.click(); }); + if (action === "steer") expect(props.onSteer).toHaveBeenCalledWith("comment-1", "rev-1"); + else expect(props.onInterrupt).toHaveBeenCalledOnce(); + }); + it("reorders the complete queue and rewrites contiguous positions", () => { const next = reorderQueuedMessageEntries( queue.entries, diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx index 2562430f75e..2df4b95c99b 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx @@ -43,6 +43,7 @@ export function reorderQueuedMessageEntries( activeId: string, overId: string, ) { + if (entries.some((entry) => entry.source?.kind === "interaction")) return null; const from = entries.findIndex((entry) => entry.comment.id === activeId); const to = entries.findIndex((entry) => entry.comment.id === overId); if (from < 0 || to < 0 || from === to) return null; @@ -94,9 +95,11 @@ function SortableQueuedMessage({ onInterrupt?: () => void; onDiscard: () => void; }) { + const immutableResponse = entry.source?.kind === "interaction"; + const label = immutableResponse ? entry.comment.body.split("\n")[0] : entry.comment.body; const sortable = useSortable({ id: entry.comment.id, - disabled: queueMutationDisabled, + disabled: queueMutationDisabled || immutableResponse, }); const style = { transform: CSS.Transform.toString(sortable.transform), @@ -127,7 +130,7 @@ function SortableQueuedMessage({ ref={sortable.setActivatorNodeRef} {...sortable.attributes} {...sortable.listeners} - disabled={queueMutationDisabled} + disabled={queueMutationDisabled || immutableResponse} aria-label={`Reorder queued message: ${entry.comment.body}`} className="flex h-7 w-7 shrink-0 cursor-grab items-center justify-center rounded-md text-muted-foreground/70 transition-colors hover:bg-accent hover:text-foreground active:cursor-grabbing disabled:cursor-default disabled:opacity-40" > @@ -139,10 +142,10 @@ function SortableQueuedMessage({ aria-hidden /> - {entry.comment.body} + {label} - {queue.protocol === "legacy" ? ( + {queue.protocol === "legacy" || entry.source?.requiresFreshSession ? (